@@ -21,6 +21,8 @@ def create(message:, shared_organizations: [])
                end
 
       domain.router_group_guid = message.router_group_guid
+      domain.enforce_route_policies = message.enforce_route_policies || false
+      domain.route_policies_scope = message.route_policies_scope
 
       Domain.db.transaction do
         domain.save
 
@@ -0,0 +1,47 @@
+module VCAP::CloudController
+  class RoutePolicyCreate
+    class Error < StandardError
+    end
+
+    def create(route:, message:)
+      validate_scope!(route.domain, message.source)
+
+      RoutePolicy.db.transaction do
+        # Lock the parent route row to serialize concurrent creates.
+        # SELECT ... FOR UPDATE on an empty policies table acquires no row locks,
+        # so two concurrent transactions can both read [] and both pass cf:any
+        # exclusivity validation. Locking the route row (which always exists)
+        # ensures they serialize regardless of how many policies currently exist.
+        Route.where(id: route.id).for_update.first or raise Error.new("Route '#{route.guid}' not found.")
+
+        existing_policies = RoutePolicy.where(route_id: route.id).all
+        validate_source_exclusivity(existing_policies, message.source)
+
+        RoutePolicy.create(
+          source: message.source,
+          route_id: route.id
+        )
+      end
+    rescue Sequel::UniqueConstraintViolation
+      raise Error.new("A route policy with source '#{message.source}' already exists for this route.")
+    end
+
+    private
+
+    def validate_scope!(domain, source)
+      return unless domain.route_policies_scope == 'space'
+      return unless source.start_with?('cf:org:')
+
+      raise Error.new("Source '#{source}' is not allowed: domain's route_policies_scope is 'space'.")
+    end
+
+    def validate_source_exclusivity(locked_policies, source)
+      existing_sources = locked_policies.map(&:source)
+
+      # Enforce cf:any exclusivity: if new policy is cf:any, reject if route already has any policies;
+      # if route already has a cf:any policy, reject new policies.
+      raise Error.new("Cannot add 'cf:any' source when other route policies already exist for this route.") if source == 'cf:any' && existing_sources.any?
+      raise Error.new("Cannot add source '#{source}': route already has a 'cf:any' policy.") if existing_sources.include?('cf:any')
+    end
+  end
+end
@@ -0,0 +1,184 @@
+require 'messages/route_policy_create_message'
+require 'messages/route_policy_update_message'
+require 'messages/route_policies_list_message'
+require 'messages/route_policy_show_message'
+require 'presenters/v3/route_policy_presenter'
+require 'decorators/include_route_policy_source_decorator'
+require 'decorators/include_route_policy_route_decorator'
+require 'actions/route_policy_create'
+require 'repositories/route_policy_event_repository'
+require 'fetchers/label_selector_query_generator'
+
+class RoutePoliciesController < ApplicationController
+  def index
+    message = RoutePoliciesListMessage.from_params(query_params)
+    invalid_param!(message.errors.full_messages) unless message.valid?
+
+    dataset = build_dataset(message)
+
+    decorators = []
+    decorators << IncludeRoutePolicySourceDecorator if IncludeRoutePolicySourceDecorator.match?(message.include)
+    decorators << IncludeRoutePolicyRouteDecorator if IncludeRoutePolicyRouteDecorator.match?(message.include)
+
+    render status: :ok, json: Presenters::V3::PaginatedListPresenter.new(
+      presenter: Presenters::V3::RoutePolicyPresenter,
+      paginated_result: SequelPaginator.new.get_page(dataset, message.try(:pagination_options)),
+      path: '/v3/route_policies',
+      message: message,
+      decorators: decorators
+    )
+  end
+
+  def show
+    message = RoutePolicyShowMessage.from_params(query_params)
+    unprocessable!(message.errors.full_messages) unless message.valid?
+
+    route_policy = VCAP::CloudController::RoutePolicy.find(guid: hashed_params[:guid])
+    resource_not_found!(:route_policy) unless route_policy
+
+    route = route_policy.route
+    resource_not_found!(:route_policy) unless route && permission_queryer.can_read_route_policy_from_space?(route.space.id, route.space.organization_id)
+
+    decorators = []
+    decorators << IncludeRoutePolicySourceDecorator if IncludeRoutePolicySourceDecorator.match?(message.include)
+    decorators << IncludeRoutePolicyRouteDecorator if IncludeRoutePolicyRouteDecorator.match?(message.include)
+
+    render status: :ok, json: Presenters::V3::RoutePolicyPresenter.new(route_policy, decorators: decorators)
+  end
+
+  def create
+    message = RoutePolicyCreateMessage.new(hashed_params[:body])
+    unprocessable!(message.errors.full_messages) unless message.valid?
+
+    route = find_and_authorize_route(message.route_guid)
+    validate_route_domain(route)
+
+    route_policy = VCAP::CloudController::RoutePolicyCreate.new.create(route: route, message: message)
+
+    route_policy_event_repository.record_route_policy_create(
+      route_policy,
+      user_audit_info,
+      { 'source' => message.source, 'route_guid' => message.route_guid }
+    )
+
+    render status: :created, json: Presenters::V3::RoutePolicyPresenter.new(route_policy)
+  rescue VCAP::CloudController::RoutePolicyCreate::Error => e
+    unprocessable!(e.message)
+  end
+
+  def update
+    route_policy = VCAP::CloudController::RoutePolicy.find(guid: hashed_params[:guid])
+    resource_not_found!(:route_policy) unless route_policy
+
+    find_and_authorize_route_for_policy(route_policy)
+
+    message = RoutePolicyUpdateMessage.new(hashed_params[:body])
+    unprocessable!(message.errors.full_messages) unless message.valid?
+
+    VCAP::CloudController::MetadataUpdate.update(route_policy, message)
+
+    route_policy_event_repository.record_route_policy_update(
+      route_policy.reload,
+      user_audit_info,
+      message.audit_hash
+    )
+
+    render status: :ok, json: Presenters::V3::RoutePolicyPresenter.new(route_policy.reload)
+  end
+
+  def destroy
+    route_policy = VCAP::CloudController::RoutePolicy.find(guid: hashed_params[:guid])
+    resource_not_found!(:route_policy) unless route_policy
+
+    find_and_authorize_route_for_policy(route_policy)
+
+    route_policy.destroy
+
+    route_policy_event_repository.record_route_policy_delete(
+      route_policy,
+      user_audit_info
+    )
+    head :no_content
+  end
+
+  private
+
+  def route_policy_event_repository
+    @route_policy_event_repository ||= Repositories::RoutePolicyEventRepository.new
+  end
+
+  def find_and_authorize_route(route_guid)
+    route = VCAP::CloudController::Route.find(guid: route_guid)
+    resource_not_found!(:route) unless route && permission_queryer.can_read_from_space?(route.space.id, route.space.organization_id)
+    unauthorized! unless permission_queryer.can_write_to_active_space?(route.space.id)
+    suspended! unless permission_queryer.is_space_active?(route.space.id)
+    route
+  end
+
+  def find_and_authorize_route_for_policy(route_policy)
+    route = route_policy.route
+    resource_not_found!(:route_policy) unless route && permission_queryer.can_read_route_policy_from_space?(route.space.id, route.space.organization_id)
+    unauthorized! unless permission_queryer.can_write_to_active_space?(route.space.id)
+    suspended! unless permission_queryer.is_space_active?(route.space.id)
+  end
+
+  def validate_route_domain(route)
+    if route.domain.internal?
+      unprocessable!('Cannot create route policies for routes on internal domains. Internal routes use container-to-container networking and bypass GoRouter.')
+    end
+    return if route.domain.enforce_route_policies
+
+    unprocessable!("Cannot create route policies for route '#{route.guid}': the route's domain does not have enforce_route_policies enabled.")
+  end
+
+  def build_dataset(message)
+    dataset = VCAP::CloudController::RoutePolicy.dataset
+
+    if permission_queryer.can_read_globally?
+      readable_route_ids = VCAP::CloudController::Route.select(:id)
+    else
+      readable_space_ids = permission_queryer.readable_route_policies_spaces_query.select(:id)
+      readable_route_ids = VCAP::CloudController::Route.where(space_id: readable_space_ids).select(:id)
+    end
+
+    dataset = dataset.where(route_id: readable_route_ids)
+
+    # Join routes at most once when either route_guids or space_guids is requested
+    if message.requested?(:route_guids) || message.requested?(:space_guids)
+      dataset = dataset.
+                join(:routes, id: :route_id).
+                select_all(:route_policies)
+
+      dataset = dataset.where(Sequel[:routes][:guid] => message.route_guids) if message.requested?(:route_guids)
+
+      dataset = dataset.where(Sequel[:routes][:space_id] => VCAP::CloudController::Space.where(guid: message.space_guids).select(:id)) if message.requested?(:space_guids)
+    end
+
+    dataset = dataset.where(guid: message.guids) if message.requested?(:guids)
+
+    if message.requested?(:sources)
+      conditions = message.sources.map do |src|
+        if src == 'cf:any'
+          Sequel.&(source_type: 'any', source_guid: '')
+        else
+          m = src.match(/\Acf:(app|space|org):([0-9a-f-]+)\z/)
+          Sequel.&(source_type: m[1], source_guid: m[2])
+        end
+      end
+      dataset = dataset.where(Sequel.|(*conditions))
+    end
+
+    dataset = dataset.where(source_guid: message.source_guids) if message.requested?(:source_guids)
+
+    if message.requested?(:label_selector)
+      dataset = VCAP::CloudController::LabelSelectorQueryGenerator.add_selector_queries(
+        label_klass: VCAP::CloudController::RoutePolicyLabelModel,
+        resource_dataset: dataset,
+        requirements: message.requirements,
+        resource_klass: VCAP::CloudController::RoutePolicy
+      )
+    end
+
+    dataset
+  end
+end
@@ -8,6 +8,7 @@
 require 'messages/route_update_destinations_message'
 require 'actions/update_route_destinations'
 require 'decorators/include_route_domain_decorator'
+require 'decorators/include_route_policies_decorator'
 require 'presenters/v3/route_presenter'
 require 'presenters/v3/route_destinations_presenter'
 require 'presenters/v3/paginated_list_presenter'
@@ -45,6 +46,7 @@ def index
     decorators << IncludeRouteDomainDecorator if IncludeRouteDomainDecorator.match?(message.include)
     decorators << IncludeSpaceDecorator if IncludeSpaceDecorator.match?(message.include)
     decorators << IncludeOrganizationDecorator if IncludeOrganizationDecorator.match?(message.include)
+    decorators << IncludeRoutePoliciesDecorator if IncludeRoutePoliciesDecorator.match?(message.include)
 
     render status: :ok, json: Presenters::V3::PaginatedListPresenter.new(
       presenter: Presenters::V3::RoutePresenter,
@@ -63,6 +65,7 @@ def show
     decorators << IncludeRouteDomainDecorator if IncludeRouteDomainDecorator.match?(message.include)
     decorators << IncludeSpaceDecorator if IncludeSpaceDecorator.match?(message.include)
     decorators << IncludeOrganizationDecorator if IncludeOrganizationDecorator.match?(message.include)
+    decorators << IncludeRoutePoliciesDecorator if IncludeRoutePoliciesDecorator.match?(message.include)
 
     render status: :ok, json: Presenters::V3::RoutePresenter.new(
       route,
 
@@ -0,0 +1,19 @@
+module VCAP::CloudController
+  class IncludeRoutePoliciesDecorator
+    class << self
+      def match?(include)
+        include&.any? { |i| %w[route_policies].include?(i) }
+      end
+
+      def decorate(hash, routes)
+        hash[:included] ||= {}
+        route_ids = routes.map(&:id).uniq
+        route_policies = RoutePolicy.where(route_id: route_ids).
+                         eager(:route, :labels, :annotations).all
+
+        hash[:included][:route_policies] = route_policies.map { |rp| Presenters::V3::RoutePolicyPresenter.new(rp).to_hash }
+        hash
+      end
+    end
+  end
+end
@@ -0,0 +1,27 @@
+module VCAP::CloudController
+  class IncludeRoutePolicyRouteDecorator
+    # Handles `?include=route` for GET /v3/route_policies
+    # Includes the route resources associated with the route policies
+
+    def self.match?(include_params)
+      include_params&.include?('route')
+    end
+
+    def self.decorate(hash, route_policies)
+      hash[:included] ||= {}
+
+      # Collect all unique route IDs from route policies
+      route_ids = route_policies.map(&:route_id).uniq
+
+      # Fetch routes with their associations
+      routes = Route.where(id: route_ids).
+               order(:created_at, :guid).
+               eager(Presenters::V3::RoutePresenter.associated_resources).all
+
+      # Present routes
+      hash[:included][:routes] = routes.map { |route| Presenters::V3::RoutePresenter.new(route).to_hash }
+
+      hash
+    end
+  end
+end
@@ -0,0 +1,71 @@
+module VCAP::CloudController
+  class IncludeRoutePolicySourceDecorator
+    # Handles `?include=source` for GET /v3/route_policies
+    # Stale/missing resources (source GUIDs that no longer exist) are silently absent.
+    # Resources the current user cannot read are also silently absent.
+
+    def self.match?(include_params)
+      return false unless include_params
+
+      include_params.include?('source')
+    end
+
+    def self.decorate(hash, route_policies)
+      hash[:included] ||= {}
+
+      # Collect all GUIDs by type
+      app_guids   = []
+      space_guids = []
+      org_guids   = []
+
+      route_policies.each do |policy|
+        next if policy.source_type == 'any'
+
+        case policy.source_type
+        when 'app'   then app_guids   << policy.source_guid
+        when 'space' then space_guids << policy.source_guid
+        when 'org'   then org_guids   << policy.source_guid
+        end
+      end
+
+      permission_queryer = Permissions.new(SecurityContext.current_user)
+
+      # Fetch and present resources, filtering by what the current user can read
+      hash[:included][:apps] = fetch_and_present_apps(app_guids.uniq, permission_queryer)
+      hash[:included][:spaces] = fetch_and_present_spaces(space_guids.uniq, permission_queryer)
+      hash[:included][:organizations] = fetch_and_present_organizations(org_guids.uniq, permission_queryer)
+
+      hash
+    end
+
+    private_class_method def self.fetch_and_present_apps(guids, permission_queryer)
+      return [] if guids.empty?
+
+      apps = AppModel.where(guid: guids)
+      apps = apps.where(space_guid: permission_queryer.readable_space_guids_query) unless permission_queryer.can_read_globally?
+      apps.order(:created_at, :guid).
+        eager(Presenters::V3::AppPresenter.associated_resources).all.
+        map { |app| Presenters::V3::AppPresenter.new(app).to_hash }
+    end
+
+    private_class_method def self.fetch_and_present_spaces(guids, permission_queryer)
+      return [] if guids.empty?
+
+      spaces = Space.where(guid: guids)
+      spaces = spaces.where(guid: permission_queryer.readable_space_guids_query) unless permission_queryer.can_read_globally?
+      spaces.order(:created_at, :guid).
+        eager(Presenters::V3::SpacePresenter.associated_resources).all.
+        map { |space| Presenters::V3::SpacePresenter.new(space).to_hash }
+    end
+
+    private_class_method def self.fetch_and_present_organizations(guids, permission_queryer)
+      return [] if guids.empty?
+
+      orgs = Organization.where(guid: guids)
+      orgs = orgs.where(guid: permission_queryer.readable_org_guids_query) unless permission_queryer.can_read_globally?
+      orgs.order(:created_at, :guid).
+        eager(Presenters::V3::OrganizationPresenter.associated_resources).all.
+        map { |org| Presenters::V3::OrganizationPresenter.new(org).to_hash }
+    end
+  end
+end