Implement org-owned service accounts with space entitlements and workload identity

app/actions/app_assign_service_account.rb

@@ -0,0 +1,78 @@
+require 'repositories/service_account_event_repository'
+
+module VCAP::CloudController
+  class AppAssignServiceAccount
+    class Error < StandardError; end
+    class Unauthorized < Error; end
+    class Conflict < Error; end
+
+    def initialize(permissions, actor: nil)
+      @permissions = permissions
+      @actor = actor
+    end
+
+    def assign(app, account, provision: false)
+      job = nil
+      app.db.transaction do
+        app.lock!
+        raise Unauthorized.new('not authorized to update the app') unless @permissions.can_write_to_active_space?(app.space.id)
+
+        validate_binding_change!(app, account)
+
+        if account
+          account.lock!
+          raise Conflict.new('organization service account prefix is unset') unless account.organization.service_account_prefix
+          raise Conflict.new('account must be entitled to the app space in the same organization') unless currently_entitled?(account, app.space)
+
+          validate_ready!(account, provision)
+          raise Conflict.new('another service account is already assigned; unbind first') if app.service_account_guid && app.service_account_guid != account.guid
+
+          job = provision_account(account) unless account.status == 'ready'
+        end
+
+        unless app.service_account_guid == account&.guid
+          previous_guid = app.service_account_guid
+          app.update(service_account: account)
+          record_assignment(app, account, previous_guid) if @actor
+        end
+      end
+      provision ? job : app
+    end
+
+    private
+
+    def currently_entitled?(account, space)
+      account.organization_id == space.organization_id && account.db[:service_account_space_entitlements].
+        where(service_account_guid: account.guid, space_guid: space.guid).for_update.first
+    end
+
+    def validate_binding_change!(app, account)
+      return unless app.service_account_guid && app.service_account_guid != account&.guid && app.service_bindings_dataset.exclude(service_account_guid: nil).any?
+
+      raise Conflict.new('remove service bindings before changing the service account')
+    end
+
+    def record_assignment(app, account, previous_guid)
+      Repositories::ServiceAccountEventRepository.record(
+        app, account ? 'assign' : 'unassign', @actor,
+        service_account_guid: account&.guid || previous_guid
+      )
+    end
+
+    def validate_ready!(account, provision)
+      raise Conflict.new('service account is not ready or enabled') unless account.enabled && (account.status == 'ready' || provision)
+
+      if PollableJobModel.where(resource_guid: account.guid, resource_type: 'service_account', state: %w[PROCESSING POLLING]).exclude(operation: 'service_account.provision').any?
+        raise Conflict.new('service account lifecycle operation is in progress')
+      end
+    end
+
+    def provision_account(account)
+      job = PollableJobModel.where(resource_guid: account.guid, operation: 'service_account.provision', state: %w[PROCESSING POLLING]).for_update.first
+      return job if job
+
+      account.update(status: 'reconciling')
+      Jobs::Enqueuer.new(queue: Jobs::Queues.generic).enqueue_pollable(Jobs::V3::ServiceAccountProvision.new(account.guid))
+    end
+  end
+end

app/actions/organization_create.rb

@@ -14,6 +14,7 @@ def create(message)
       Organization.db.transaction do
         org = VCAP::CloudController::Organization.create(
           name: message.name,
+          service_account_prefix: message.service_account_prefix,
           status: message.suspended ? Organization::SUSPENDED : Organization::ACTIVE
         )
 
@@ -29,6 +30,7 @@ def create(message)
     private
 
     def validation_error!(error, message)
+      error!("Service account prefix '#{message.service_account_prefix}' is already reserved.") if error.errors.on(:service_account_prefix)&.include?(:unique)
       error!("Organization '#{message.name}' already exists.") if error.errors.on(:name)&.include?(:unique)
       error!(error.message)
     end

app/actions/organization_delete.rb

@@ -1,22 +1,32 @@
 require 'actions/space_delete'
 require 'actions/label_delete'
 require 'actions/annotation_delete'
+require 'actions/organization_service_account_cleanup'
 
 module VCAP::CloudController
   class OrganizationDelete
+    include OrganizationServiceAccountCleanup
+
     def initialize(space_deleter, user_audit_info)
       @space_deleter = space_deleter
       @user_audit_info = user_audit_info
     end
 
     def delete(org_dataset)
       org_dataset.each do |org|
+        shared_errors = shared_domain_errors(org)
+        return shared_errors unless shared_errors.empty?
+
+        mark_deleting_for_account_cleanup(org)
         errs = @space_deleter.delete(org.spaces_dataset)
         unless errs.empty?
           error_message = errs.map(&:message).join("\n\n")
           return [CloudController::Errors::ApiError.new_from_details('OrganizationDeletionFailed', org.name, error_message)]
         end
 
+        account_errors = delete_service_accounts(org)
+        return [CloudController::Errors::ApiError.new_from_details('OrganizationDeletionFailed', org.name, account_errors.map(&:message).join("\n\n"))] unless account_errors.empty?
+
         domains_to_unshare = []
         org.private_domains.each do |private_domain|
           if private_domain.owning_organization == org

app/actions/organization_quotas_create.rb

@@ -16,6 +16,7 @@ def create(message)
       QuotaDefinition.db.transaction do
         org_quota = VCAP::CloudController::QuotaDefinition.create(
           name: message.name,
+          total_service_accounts: message.total_service_accounts || QuotaDefinition::UNLIMITED,
 
           # Apps
           memory_limit: message.total_memory_in_mb || QuotaDefinition::DEFAULT_MEMORY_LIMIT,

app/actions/organization_quotas_update.rb

@@ -18,6 +18,7 @@ def self.update(quota, message, user_audit_info)
         quota.lock!
 
         quota.name = message.name if message.requested? :name
+        quota.total_service_accounts = message.total_service_accounts || QuotaDefinition::UNLIMITED if message.service_accounts_limits_message.requested?(:total_service_accounts)
 
         quota.memory_limit = memory_limit(message) if message.apps_limits_message.requested? :total_memory_in_mb
         quota.instance_memory_limit = instance_memory_limit(message) if message.apps_limits_message.requested? :per_process_memory_in_mb

app/actions/organization_service_account_cleanup.rb

@@ -0,0 +1,56 @@
+require 'actions/service_account_provision'
+require 'repositories/service_account_event_repository'
+
+module VCAP::CloudController
+  module OrganizationServiceAccountCleanup
+    private
+
+    def shared_domain_errors(org)
+      return [] unless ServiceAccountModel.where(organization_id: org.id).any?
+
+      domains = org.owned_private_domains_dataset.all.select(&:shared_with_any_orgs?)
+      return [] if domains.empty?
+
+      details = domains.map { |domain| "Domain '#{domain.name}' is shared with other organizations. Unshare before deleting." }.join("\n")
+      [CloudController::Errors::ApiError.new_from_details('OrganizationDeletionFailed', org.name, details)]
+    end
+
+    def mark_deleting_for_account_cleanup(org)
+      return unless ServiceAccountModel.where(organization_id: org.id).any?
+
+      org.db.transaction do
+        org.lock!
+        org.update(status: Organization::DELETING)
+      end
+    end
+
+    def delete_service_accounts(org)
+      accounts = ServiceAccountModel.where(organization_id: org.id).order(:guid).all
+      return [] if accounts.empty?
+
+      accounts.each_with_object([]) do |account, errors|
+        locally_deleted = false
+        account.db.transaction do
+          account.lock!
+          if PollableJobModel.where(resource_guid: account.guid, resource_type: 'service_account', state: %w[PROCESSING POLLING]).any?
+            raise ServiceAccountProvision::Conflict.new('service account operation is already in progress')
+          end
+
+          if account.status == 'reserved' && !account.in_use?
+            Repositories::ServiceAccountEventRepository.record(account, 'delete', @user_audit_info, { recursive: true })
+            account.destroy
+            locally_deleted = true
+          else
+            account.update(enabled: false, status: 'deleting')
+          end
+        end
+        next if locally_deleted
+
+        Repositories::ServiceAccountEventRepository.record(account, 'delete', @user_audit_info, { recursive: true })
+        CloudController::DependencyLocator.instance.service_account_provisioner.deprovision(account, delete: true)
+      rescue StandardError => e
+        errors << e
+      end
+    end
+  end
+end

app/actions/organization_update.rb

@@ -10,6 +10,7 @@ def initialize(user_audit_info)
     def update(org, message)
       org.db.transaction do
         org.lock!
+        org.service_account_prefix = message.service_account_prefix if message.requested?(:service_account_prefix)
         org.name = message.name if message.requested?(:name)
         LabelsUpdate.update(org, message.labels, OrganizationLabelModel)
         AnnotationsUpdate.update(org, message.annotations, OrganizationAnnotationModel)
@@ -29,6 +30,9 @@ def update(org, message)
     end
 
     def validation_error!(error, message)
+      prefix_errors = error.errors.on(:service_account_prefix)
+      error!("Service account prefix '#{message.service_account_prefix}' is already reserved.") if prefix_errors&.include?(:unique)
+      error!('Service account prefix is immutable once allocated.') if prefix_errors&.include?(:immutable)
       error!("Organization name '#{message.name}' is already taken.") if error.errors.on(:name)&.include?(:unique)
       error!(error.message)
     end

app/actions/process_restart.rb

@@ -14,7 +14,7 @@ def restart(process:, config:, stop_in_runtime:, revision: nil)
             runners(config).runner_for_process(process).stop
           end
 
-          process.update(state: ProcessModel::STARTED, revision: revision_to_set)
+          process.update(state: ProcessModel::STARTED, revision: revision_to_set, service_account_guid: process.app.service_account_guid, service_account_snapshot: true)
           runners(config).runner_for_process(process).start
         end
       end

app/actions/route_policy_create.rb

@@ -13,6 +13,7 @@ def create(route:, message:)
       validate_scope!(route.domain, message.source)
 
       route_policy = RoutePolicy.db.transaction do
+        lock_account_source!(message.source)
         # 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
@@ -45,6 +46,17 @@ def create(route:, message:)
 
     private
 
+    def lock_account_source!(source)
+      match = source.match(RoutePolicy::SERVICE_ACCOUNT_SOURCE_REGEX)
+      return unless match
+
+      account = ServiceAccountModel.first(identity_prefix: match[1], name: match[2])
+      raise Error.new('Service account not found') unless account
+
+      account.lock!
+      raise Error.new('Service account is being deleted') if account.status == 'deleting'
+    end
+
     def validate_scope!(domain, source)
       return unless domain.route_policies_scope == 'space'
       return unless source.start_with?('cf:org:')

app/actions/service_account_creation_budget.rb

@@ -0,0 +1,27 @@
+module VCAP::CloudController
+  class ServiceAccountCreationBudget
+    WINDOW = 7.days
+
+    def self.consume(principal, exempt:)
+      return yield if exempt
+
+      # Lock the authenticated principal, not a space: requests on different API
+      # instances and in different spaces must share one atomic budget.
+      principal.lock!
+      now = Time.now.utc
+      creations = ServiceAccountModel.db[:service_account_creations]
+      limit = Config.config.get(:service_account_creation_limit) || -1
+      # MySQL permission/authentication reads may have established an older
+      # repeatable-read snapshot before the principal lock was acquired.
+      used = creations.where(principal_guid: principal.guid).where { created_at > now - WINDOW }.select(:principal_guid).for_update.all.size
+      raise CloudController::Errors::V3::ApiError.new_from_details('ServiceAccountCreationLimitExceeded') if limit.between?(0, used)
+
+      result = yield
+      # Independent of accounts, tombstones and audit-event retention. Failed
+      # creations roll this back; successful deletions never refund the budget.
+      creations.insert(principal_guid: principal.guid, created_at: now) # rubocop:disable Rails/SkipsModelValidations -- Transactional creation ledger, not an ActiveRecord model.
+      creations.where(principal_guid: principal.guid).where { created_at <= now - WINDOW }.delete
+      result
+    end
+  end
+end