Implement shadow user creation by org manager

app/actions/user_create.rb

@@ -4,10 +4,20 @@ class Error < StandardError
     end
 
     def create(message:)
-      user = User.create(guid: message.guid)
+      if message.username && message.origin
+        existing_user_guid = User.get_user_id_by_username_and_origin(message.username, message.origin)
+
+        shadow_user = User.create_uaa_shadow_user(message.username, message.origin) unless existing_user_guid
+
+        user_guid = existing_user_guid || shadow_user['id']
+      else
+        user_guid = message.guid
+      end
+      user = User.create(guid: user_guid)
       User.db.transaction do
         MetadataUpdate.update(user, message)
       end
+
       user
     rescue Sequel::ValidationFailed => e
       validation_error!(message, e)
@@ -16,7 +26,11 @@ def create(message:)
     private
 
     def validation_error!(message, error)
-      error!("User with guid '#{message.guid}' already exists.") if error.errors.on(:guid)&.any? { |e| [:unique].include?(e) }
+      error!("User with guid '#{message.guid}' already exists.") if message.guid && error.errors.on(:guid)&.any? { |e| [:unique].include?(e) }
+
+      if !message.guid && error.errors.on(:guid)&.any? { |e| [:unique].include?(e) }
+        error!("User with username '#{message.username}' and origin '#{message.origin}' already exists.")
+      end
 
       error!(error.message)
     end

app/controllers/v3/roles_controller.rb

@@ -114,7 +114,11 @@ def create_org_role(message)
     unauthorized! unless permission_queryer.can_write_to_active_org?(org.id)
     suspended! unless permission_queryer.is_org_active?(org.id)
 
-    user_guid = message.user_guid || lookup_user_guid_in_uaa(message.username, message.user_origin)
+    user_guid = if message.username && message.user_origin && message.user_origin != 'uaa' && org_managers_can_create_users?
+                  create_or_get_uaa_user(message)
+                else
+                  message.user_guid || lookup_user_guid_in_uaa(message.username, message.user_origin)
+                end
 
     user = User.first(guid: user_guid) || create_cc_user(user_guid)
 
@@ -140,6 +144,23 @@ def create_cc_user(user_guid)
     UserCreate.new.create(message:)
   end
 
+  def create_or_get_uaa_user(message)
+    user_create_message = UserCreateMessage.new(username: message.username, origin: message.user_origin)
+    unprocessable!(user_create_message.errors.full_messages) unless user_create_message.valid?
+
+    existing_user_id = get_uaa_user_id(user_create_message)
+    user = create_uaa_shadow_user(user_create_message) unless existing_user_id
+    existing_user_id || user['id']
+  end
+
+  def get_uaa_user_id(message)
+    User.get_user_id_by_username_and_origin(message.username, message.origin)
+  end
+
+  def create_uaa_shadow_user(message)
+    User.create_uaa_shadow_user(message.username, message.origin)
+  end
+
   def readable_users
     current_user.readable_users(permission_queryer.can_read_globally?)
   end
@@ -203,4 +224,8 @@ def lookup_user_guid_in_uaa(username, given_origin, creating_space_role: false)
   def uaa_username_lookup_client
     CloudController::DependencyLocator.instance.uaa_username_lookup_client
   end
+
+  def org_managers_can_create_users?
+    VCAP::CloudController::Config.config.get(:allow_user_creation_by_org_manager) && FeatureFlag.raise_unless_enabled!(:set_roles_by_username)
+  end
 end

app/controllers/v3/users_controller.rb

@@ -38,13 +38,24 @@ def show
   end
 
   def create
-    unauthorized! unless permission_queryer.can_write_globally?
-
     message = UserCreateMessage.new(hashed_params[:body])
+    unauthorized! unless permission_queryer.can_write_globally? || org_managers_can_create_users?
     unprocessable!(message.errors.full_messages) unless message.valid?
+
+    # prevent org_managers from creating users by guid
+    unauthorized! if !permission_queryer.can_write_globally? && !(!message.guid && org_managers_can_create_users?)
+
     user = UserCreate.new.create(message:)
 
-    render status: :created, json: Presenters::V3::UserPresenter.new(user, uaa_users: User.uaa_users_info([user.guid]))
+    if message.username && message.origin
+      render status: :created,
+             json: Presenters::V3::UserPresenter.new(user,
+                                                     uaa_users: { user.guid => { 'username' => message.username, 'id' => user.guid, 'origin' => message.origin } })
+    else
+      render status: :created, json: Presenters::V3::UserPresenter.new(user, uaa_users: User.uaa_users_info([user.guid]))
+    end
+  rescue VCAP::CloudController::UaaUnavailable
+    raise CloudController::Errors::ApiError.new_from_details('UaaUnavailable')
   rescue UserCreate::Error => e
     unprocessable!(e)
   end
@@ -91,4 +102,8 @@ def fetch_user_if_readable(desired_guid)
   def user_not_found!
     resource_not_found!(:user)
   end
+
+  def org_managers_can_create_users?
+    VCAP::CloudController::Config.config.get(:allow_user_creation_by_org_manager) && permission_queryer.is_org_manager?
+  end
 end

app/messages/user_create_message.rb

@@ -2,9 +2,27 @@
 
 module VCAP::CloudController
   class UserCreateMessage < MetadataBaseMessage
-    register_allowed_keys [:guid]
+    register_allowed_keys %i[guid origin username]
+
+    class UserCreateValidator < ActiveModel::Validator
+      def validate(record)
+        if record.guid
+          record.errors.add(:username, message: "cannot be provided with 'guid'") if record.username
+          record.errors.add(:origin, message: "cannot be provided with 'guid'") if record.origin
+        elsif record.username || record.origin
+          record.errors.add(:origin, message: "can only be provided together with 'username'") unless record.username
+          record.errors.add(:username, message: "can only be provided together with 'origin'") unless record.origin
+          record.errors.add(:origin, message: "cannot be 'uaa' when creating a user by username") unless record.origin != 'uaa'
+        else
+          record.errors.add(:guid, message: "or 'username' and 'origin' must be provided")
+        end
+      end
+    end
 
     validates_with NoAdditionalKeysValidator
-    validates :guid, guid: true
+    validates :guid, guid: true, allow_nil: true
+    validates :origin, string: true, allow_nil: true
+    validates :username, string: true, allow_nil: true
+    validates_with UserCreateValidator
   end
 end

app/models/runtime/user.rb

@@ -252,6 +252,16 @@ def self.uaa_users_info(user_guids)
       uaa_username_lookup_client.users_for_ids(user_guids)
     end
 
+    def self.get_user_id_by_username_and_origin(username, origin)
+      uaa_username_lookup_client = CloudController::DependencyLocator.instance.uaa_username_lookup_client
+      uaa_username_lookup_client.ids_for_usernames_and_origins([username], [origin]).first
+    end
+
+    def self.create_uaa_shadow_user(username, origin)
+      uaa_shadow_user_creation_client = CloudController::DependencyLocator.instance.uaa_shadow_user_creation_client
+      uaa_shadow_user_creation_client.create_shadow_user(username, origin)
+    end
+
     def self.user_visibility_filter(_)
       full_dataset_filter
     end

lib/cloud_controller/config_schemas/base/api_schema.rb

@@ -58,9 +58,18 @@ class ApiSchema < VCAP::Config
               optional(:ca_file) => String,
               :client_timeout => Integer,
               optional(:symmetric_secret) => String,
-              optional(:symmetric_secret2) => String
+              optional(:symmetric_secret2) => String,
+              optional(:clients) => enum([
+                {
+                  'name' => String,
+                  'id' => String,
+                  'secret' => String
+                }
+              ], NilClass)
             },
 
+            optional(:allow_user_creation_by_org_manager) => bool,
+
             logging: {
               level: String, # debug, info, etc.
               file: String, # Log file to use

lib/cloud_controller/dependency_locator.rb

@@ -288,6 +288,19 @@ def uaa_username_lookup_client
       )
     end
 
+    def uaa_shadow_user_creation_client
+      client = config.get(:uaa, :clients)&.find { |client_config| client_config['name'] == 'cloud_controller_shadow_user_creation' }
+
+      return unless client
+
+      UaaClient.new(
+        uaa_target: config.get(:uaa, :internal_url),
+        client_id: client['id'],
+        secret: client['secret'],
+        ca_file: config.get(:uaa, :ca_file)
+      )
+    end
+
     def routing_api_client
       return RoutingApi::DisabledClient.new if config.get(:routing_api).nil?
 

lib/cloud_controller/permissions.rb

@@ -97,6 +97,10 @@ def can_write_globally?
     roles.admin?
   end
 
+  def is_org_manager?
+    VCAP::CloudController::OrganizationManager.where(user_id: @user.id).any?
+  end
+
   def readable_org_guids
     readable_org_guids_query.select_map(:guid)
   end

lib/cloud_controller/uaa/uaa_client.rb

@@ -96,6 +96,17 @@ def origins_for_username(username)
       raise UaaUnavailable
     end
 
+    def create_shadow_user(username, origin)
+      with_cache_retry { scim.add(:user, { username: username, origin: origin, emails: [{ primary: true, value: username }] }) }
+    rescue CF::UAA::TargetError => e
+      raise e unless e.info['error'] == 'scim_resource_already_exists'
+
+      { 'id' => e.info['user_id'] }
+    rescue CF::UAA::UAAError => e
+      logger.error("UAA request for creating a user failed: #{e.inspect}")
+      raise UaaUnavailable
+    end
+
     def info
       CF::UAA::Info.new(uaa_target, uaa_connection_opts)
     end