From 4e53164c71e276a8892db28bf88ed71e80c63502 Mon Sep 17 00:00:00 2001
From: Sven Krieger <37476281+svkrieger@users.noreply.github.com>
Date: Thu, 9 Jan 2025 11:06:00 +0100
Subject: [PATCH] Implement shadow user creation by org manager (#4113)

* Add username and origin to user create message

* Add uaa client for creating shadow users

* Enable UAA shadow user creation for POST /v3/users

* Add error handling in case UAA is unavailable

* Disallow using origin 'uaa' when creating user by username

* Allow org managers to create users by username

* Add config flag

* Adjust error handling and add unit tests

* Implement shadow user creation in /v3/roles endpoint

* Allow empty array in config for uaa->clients

* Prevent calling UAA twice when creating new user through /v3/users

* Add unit test for scenario feature off but admin can still create shadow user using username and origin

* First check whether user exists in UAA before trying to create in /v3/users

* First check whether user exists in UAA before trying to create in /v3/roles

* Improve user-create-message error messages
---
 app/actions/user_create.rb                    |  18 +-
 app/controllers/v3/roles_controller.rb        |  27 +-
 app/controllers/v3/users_controller.rb        |  21 +-
 app/messages/user_create_message.rb           |  22 +-
 app/models/runtime/user.rb                    |  10 +
 .../config_schemas/base/api_schema.rb         |  11 +-
 lib/cloud_controller/dependency_locator.rb    |  13 +
 lib/cloud_controller/permissions.rb           |   4 +
 lib/cloud_controller/uaa/uaa_client.rb        |  11 +
 spec/request/roles_spec.rb                    | 544 +++++++++++++-----
 spec/request/users_spec.rb                    | 220 ++++++-
 spec/unit/actions/user_create_spec.rb         |  59 +-
 .../dependency_locator_spec.rb                |  42 ++
 spec/unit/lib/uaa/uaa_client_spec.rb          |  55 ++
 .../unit/messages/user_create_message_spec.rb |  91 ++-
 spec/unit/models/runtime/user_spec.rb         |  69 +++
 16 files changed, 1059 insertions(+), 158 deletions(-)

diff --git a/app/actions/user_create.rb b/app/actions/user_create.rb
index aabbec36e6b..b9d14b41b61 100644
--- a/app/actions/user_create.rb
+++ b/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
diff --git a/app/controllers/v3/roles_controller.rb b/app/controllers/v3/roles_controller.rb
index e3b8e9b33bf..6d6eae3b184 100644
--- a/app/controllers/v3/roles_controller.rb
+++ b/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
diff --git a/app/controllers/v3/users_controller.rb b/app/controllers/v3/users_controller.rb
index 045c76af2e7..bc6ccb925a4 100644
--- a/app/controllers/v3/users_controller.rb
+++ b/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
diff --git a/app/messages/user_create_message.rb b/app/messages/user_create_message.rb
index 0e6be93580c..34d0cd97f85 100644
--- a/app/messages/user_create_message.rb
+++ b/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
diff --git a/app/models/runtime/user.rb b/app/models/runtime/user.rb
index e6ec1773bca..212ce236089 100644
--- a/app/models/runtime/user.rb
+++ b/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
diff --git a/lib/cloud_controller/config_schemas/base/api_schema.rb b/lib/cloud_controller/config_schemas/base/api_schema.rb
index 37584dfda7c..6a0a224270f 100644
--- a/lib/cloud_controller/config_schemas/base/api_schema.rb
+++ b/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
diff --git a/lib/cloud_controller/dependency_locator.rb b/lib/cloud_controller/dependency_locator.rb
index 8bc203a0f3d..e9d5bb1976d 100644
--- a/lib/cloud_controller/dependency_locator.rb
+++ b/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?
 
diff --git a/lib/cloud_controller/permissions.rb b/lib/cloud_controller/permissions.rb
index 59e41b778aa..c77c72ba32d 100644
--- a/lib/cloud_controller/permissions.rb
+++ b/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
diff --git a/lib/cloud_controller/uaa/uaa_client.rb b/lib/cloud_controller/uaa/uaa_client.rb
index 6da7b3c4b8c..d997a2277db 100644
--- a/lib/cloud_controller/uaa/uaa_client.rb
+++ b/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
diff --git a/spec/request/roles_spec.rb b/spec/request/roles_spec.rb
index ba81bdfd409..fde2dc455be 100644
--- a/spec/request/roles_spec.rb
+++ b/spec/request/roles_spec.rb
@@ -13,13 +13,14 @@
   let(:uaa_client) { instance_double(VCAP::CloudController::UaaClient) }
 
   before do
-    allow(CloudController::DependencyLocator.instance).to receive(:uaa_username_lookup_client).and_return(uaa_client)
+    allow(CloudController::DependencyLocator.instance).to receive_messages(uaa_username_lookup_client: uaa_client, uaa_shadow_user_creation_client: uaa_client)
     allow(uaa_client).to receive(:usernames_for_ids).with([user_with_role.guid]).and_return(
       { user_with_role.guid => 'mona' }
     )
     allow(uaa_client).to receive(:usernames_for_ids).with([user_unaffiliated.guid]).and_return(
       { user_with_role.guid => 'bob_unaffiliated' }
     )
+    allow(uaa_client).to receive(:create_shadow_user)
   end
 
   describe 'POST /v3/roles' do
@@ -88,6 +89,19 @@
 
       it_behaves_like 'permissions for single object endpoint', ALL_PERMISSIONS
 
+      context 'when "allow_user_creation_by_org_manager" is enabled' do
+        before do
+          TestConfig.override(allow_user_creation_by_org_manager: true)
+        end
+
+        it_behaves_like 'permissions for single object endpoint', ALL_PERMISSIONS
+
+        it 'does not create a uaa shadow user' do
+          post '/v3/roles', params.to_json, admin_header
+          expect(uaa_client).not_to have_received(:create_shadow_user)
+        end
+      end
+
       context 'when user is invalid' do
         let(:params) do
           {
@@ -163,7 +177,7 @@
       end
     end
 
-    context 'creating a organization role' do
+    context 'creating a organization role by guid' do
       let(:params) do
         {
           type: 'organization_auditor',
@@ -215,12 +229,21 @@
         h
       end
 
-      before do
-        org.add_user(user_with_role)
-      end
-
       it_behaves_like 'permissions for single object endpoint', ALL_PERMISSIONS
 
+      context 'when "allow_user_creation_by_org_manager" is enabled' do
+        before do
+          TestConfig.override(allow_user_creation_by_org_manager: true)
+        end
+
+        it_behaves_like 'permissions for single object endpoint', ALL_PERMISSIONS
+
+        it 'does not create a uaa shadow user' do
+          post '/v3/roles', params.to_json, admin_header
+          expect(uaa_client).not_to have_received(:create_shadow_user)
+        end
+      end
+
       context 'when organization is invalid' do
         let(:params) do
           {
@@ -330,24 +353,24 @@
       end
     end
 
-    context 'creating a role by username' do
-      let(:params) do
-        {
-          type: 'space_auditor',
-          relationships: {
-            user: {
-              data: {
-                username: 'uuu'
+    describe 'creating a role by username' do
+      context 'when space role' do
+        let(:params) do
+          {
+            type: 'space_auditor',
+            relationships: {
+              user: {
+                data: {
+                  username: 'uuu'
+                }
+              },
+              space: {
+                data: { guid: space.guid }
               }
-            },
-            space: {
-              data: { guid: space.guid }
             }
           }
-        }
-      end
+        end
 
-      context 'when the user exists in a single origin' do
         let(:expected_response) do
           {
             guid: UUID_REGEX,
@@ -390,49 +413,153 @@
           h
         end
 
-        before do
-          allow(uaa_client).to receive(:origins_for_username).with('uuu').and_return(['uaa'])
-          allow(uaa_client).to receive(:id_for_username).with('uuu', origin: 'uaa').and_return(user_with_role.guid)
-          org.add_user(user_with_role)
+        context 'when the user exists in a single origin' do
+          before do
+            allow(uaa_client).to receive(:origins_for_username).with('uuu').and_return(['uaa'])
+            allow(uaa_client).to receive(:id_for_username).with('uuu', origin: 'uaa').and_return(user_with_role.guid)
+            org.add_user(user_with_role)
+          end
+
+          it_behaves_like 'permissions for single object endpoint', ALL_PERMISSIONS
         end
 
-        it_behaves_like 'permissions for single object endpoint', ALL_PERMISSIONS
-      end
+        context 'when there are multiple users with the same username' do
+          before do
+            allow(uaa_client).to receive(:origins_for_username).with('uuu').and_return(%w[uaa ldap okta])
+            allow(uaa_client).to receive(:id_for_username).with('uuu', origin: 'uaa').and_return(user_with_role.guid)
+          end
 
-      context 'when there are multiple users with the same username' do
-        before do
-          allow(uaa_client).to receive(:origins_for_username).with('uuu').and_return(%w[uaa ldap okta])
-          allow(uaa_client).to receive(:id_for_username).with('uuu', origin: 'uaa').and_return(user_with_role.guid)
+          it 'returns a 422 with a helpful message' do
+            post '/v3/roles', params.to_json, admin_header
+
+            expect(last_response).to have_status_code(422)
+            expect(last_response).to have_error_message(
+              "User with username 'uuu' exists in the following origins: ldap, okta, uaa. Specify an origin to disambiguate."
+            )
+          end
         end
 
-        it 'returns a 422 with a helpful message' do
-          post '/v3/roles', params.to_json, admin_header
+        context 'when there is no user with the given username' do
+          before do
+            allow(uaa_client).to receive(:origins_for_username).with('uuu').and_return([])
+            allow(uaa_client).to receive(:id_for_username).with('uuu', origin: nil).and_return(nil)
+          end
 
-          expect(last_response).to have_status_code(422)
-          expect(last_response).to have_error_message(
-            "User with username 'uuu' exists in the following origins: ldap, okta, uaa. Specify an origin to disambiguate."
-          )
+          it 'returns a 422 with a helpful message' do
+            post '/v3/roles', params.to_json, admin_header
+
+            expect(last_response).to have_status_code(422)
+            expect(last_response).to have_error_message(
+              "Users cannot be assigned roles in a space if they do not have a role in that space's organization."
+            )
+          end
+        end
+
+        context 'when "allow_user_creation_by_org_manager" is enabled' do
+          before do
+            TestConfig.override(allow_user_creation_by_org_manager: true)
+            allow(uaa_client).to receive(:origins_for_username).with('uuu').and_return(['uaa'])
+            allow(uaa_client).to receive(:id_for_username).with('uuu', origin: 'uaa').and_return(user_with_role.guid)
+            org.add_user(user_with_role)
+          end
+
+          it_behaves_like 'permissions for single object endpoint', ALL_PERMISSIONS
+
+          it 'does not create a uaa shadow user' do
+            post '/v3/roles', params.to_json, admin_header
+            expect(last_response).to have_status_code(201)
+            expect(parsed_response).to match_json_response(expected_response)
+            expect(uaa_client).not_to have_received(:create_shadow_user)
+          end
         end
       end
 
-      context 'when there is no user with the given username' do
-        before do
-          allow(uaa_client).to receive(:origins_for_username).with('uuu').and_return([])
-          allow(uaa_client).to receive(:id_for_username).with('uuu', origin: nil).and_return(nil)
+      context 'when org role' do
+        let(:params) do
+          {
+            type: 'organization_auditor',
+            relationships: {
+              user: {
+                data: {
+                  username: 'uuu'
+                }
+              },
+              organization: {
+                data: { guid: org.guid }
+              }
+            }
+          }
+        end
+
+        let(:expected_response) do
+          {
+            guid: UUID_REGEX,
+            created_at: iso8601,
+            updated_at: iso8601,
+            type: 'organization_auditor',
+            relationships: {
+              user: {
+                data: { guid: user_with_role.guid }
+              },
+              space: {
+                data: nil
+              },
+              organization: {
+                data: { guid: org.guid }
+              }
+            },
+            links: {
+              self: { href: %r{#{Regexp.escape(link_prefix)}/v3/roles/#{UUID_REGEX}} },
+              user: { href: %r{#{Regexp.escape(link_prefix)}/v3/users/#{user_with_role.guid}} },
+              organization: { href: %r{#{Regexp.escape(link_prefix)}/v3/organizations/#{org.guid}} }
+            }
+          }
+        end
+
+        let(:expected_codes_and_responses) do
+          h = Hash.new(code: 403)
+          h['admin'] = {
+            code: 201,
+            response_object: expected_response
+          }
+          h['org_manager'] = {
+            code: 201,
+            response_object: expected_response
+          }
+          h
+        end
+
+        context 'when the user exists in a single origin' do
+          before do
+            allow(uaa_client).to receive(:origins_for_username).with('uuu').and_return(['uaa'])
+            allow(uaa_client).to receive(:id_for_username).with('uuu', origin: 'uaa').and_return(user_with_role.guid)
+          end
+
+          it_behaves_like 'permissions for single object endpoint', ALL_PERMISSIONS
         end
 
-        context 'for a space role' do
+        context 'when there are multiple users with the same username' do
+          before do
+            allow(uaa_client).to receive(:origins_for_username).with('uuu').and_return(%w[uaa ldap okta])
+            allow(uaa_client).to receive(:id_for_username).with('uuu', origin: 'uaa').and_return(user_with_role.guid)
+          end
+
           it 'returns a 422 with a helpful message' do
             post '/v3/roles', params.to_json, admin_header
 
             expect(last_response).to have_status_code(422)
             expect(last_response).to have_error_message(
-              "Users cannot be assigned roles in a space if they do not have a role in that space's organization."
+              "User with username 'uuu' exists in the following origins: ldap, okta, uaa. Specify an origin to disambiguate."
             )
           end
         end
 
-        context 'for an org role' do
+        context 'when there is no user with the given username' do
+          before do
+            allow(uaa_client).to receive(:origins_for_username).with('uuu').and_return([])
+            allow(uaa_client).to receive(:id_for_username).with('uuu', origin: nil).and_return(nil)
+          end
+
           let(:params) do
             {
               type: 'organization_auditor',
@@ -458,6 +585,23 @@
             )
           end
         end
+
+        context 'when "allow_user_creation_by_org_manager" is enabled' do
+          before do
+            TestConfig.override(allow_user_creation_by_org_manager: true)
+            allow(uaa_client).to receive(:origins_for_username).with('uuu').and_return(['uaa'])
+            allow(uaa_client).to receive(:id_for_username).with('uuu', origin: 'uaa').and_return(user_with_role.guid)
+          end
+
+          it_behaves_like 'permissions for single object endpoint', ALL_PERMISSIONS
+
+          it 'does not create a uaa shadow user' do
+            post '/v3/roles', params.to_json, admin_header
+            expect(last_response).to have_status_code(201)
+            expect(parsed_response).to match_json_response(expected_response)
+            expect(uaa_client).not_to have_received(:create_shadow_user)
+          end
+        end
       end
     end
 
@@ -564,52 +708,6 @@
         it_behaves_like 'permissions for single object endpoint', ALL_PERMISSIONS
       end
 
-      context 'when there is no user with the given username and origin' do
-        before do
-          allow(uaa_client).to receive(:origins_for_username).with('uuu').and_return(['something-else'])
-          allow(uaa_client).to receive(:id_for_username).with('uuu', origin: 'okta').and_return(nil)
-        end
-
-        context 'for a space role' do
-          it 'returns a 422 with a helpful message' do
-            post '/v3/roles', params.to_json, admin_header
-
-            expect(last_response).to have_status_code(422)
-            expect(last_response).to have_error_message(
-              "Users cannot be assigned roles in a space if they do not have a role in that space's organization."
-            )
-          end
-        end
-
-        context 'for an org role' do
-          let(:params) do
-            {
-              type: 'organization_auditor',
-              relationships: {
-                user: {
-                  data: {
-                    username: 'uuu',
-                    origin: 'okta'
-                  }
-                },
-                organization: {
-                  data: { guid: org.guid }
-                }
-              }
-            }
-          end
-
-          it 'returns a 422 with a helpful message' do
-            post '/v3/roles', params.to_json, admin_header
-
-            expect(last_response).to have_status_code(422)
-            expect(last_response).to have_error_message(
-              "No user exists with the username 'uuu' and origin 'okta'."
-            )
-          end
-        end
-      end
-
       context 'when UAA is unavailable' do
         before do
           allow(uaa_client).to receive(:id_for_username).and_raise(VCAP::CloudController::UaaUnavailable)
@@ -629,18 +727,21 @@
       # a space role must also have at least an org user role
       context 'when the user is unaffiliated' do
         before do
-          allow(uaa_client).to receive(:origins_for_username).with('bob_unaffiliated').and_return(['uaa'])
-          allow(uaa_client).to receive(:id_for_username).with('bob_unaffiliated', origin: 'uaa').and_return(user_unaffiliated.guid)
+          allow(uaa_client).to receive(:origins_for_username).with('bob_unaffiliated').and_return([origin])
+          allow(uaa_client).to receive(:id_for_username).with('bob_unaffiliated', origin:).and_return(user_unaffiliated.guid)
           allow(uaa_client).to receive(:usernames_for_ids).with([user_unaffiliated.guid]).and_return({ user_unaffiliated.guid => 'bob_unaffiliated' })
         end
 
+        let(:origin) { 'uaa' }
+
         let(:params) do
           {
             type: 'organization_auditor',
             relationships: {
               user: {
                 data: {
-                  username: 'bob_unaffiliated'
+                  username: 'bob_unaffiliated',
+                  origin: origin
                 }
               },
               organization: {
@@ -708,88 +809,243 @@
 
           it_behaves_like 'permissions for single object endpoint', ALL_PERMISSIONS
         end
+
+        context 'when "allow_user_creation_by_org_manager" is enabled' do
+          before do
+            TestConfig.override(allow_user_creation_by_org_manager: true)
+            allow(uaa_client).to receive(:create_shadow_user).with('bob_unaffiliated', origin).and_return({ 'id' => user_unaffiliated.guid })
+            allow(uaa_client).to receive(:ids_for_usernames_and_origins).and_return([])
+          end
+
+          it 'does not call create_shadow_user' do
+            post '/v3/roles', params.to_json, admin_header
+            expect(last_response).to have_status_code(201)
+            expect(parsed_response).to match_json_response(expected_response)
+            expect(uaa_client).not_to have_received(:create_shadow_user)
+          end
+
+          context 'with origin different from uaa' do
+            let(:origin) { 'idp.local' }
+
+            it_behaves_like 'permissions for single object endpoint', ALL_PERMISSIONS
+
+            it 'calls create_shadow_user and retrieves the guid of the user from uaa' do
+              post '/v3/roles', params.to_json, admin_header
+
+              expect(last_response).to have_status_code(201)
+              expect(parsed_response).to match_json_response(expected_response)
+
+              expect(uaa_client).to have_received(:ids_for_usernames_and_origins)
+              expect(uaa_client).to have_received(:create_shadow_user)
+            end
+
+            context 'user already exists in UAA' do
+              before do
+                allow(uaa_client).to receive(:ids_for_usernames_and_origins).and_return([user_unaffiliated.guid])
+              end
+
+              it 'retrieves the id from UAA and does not create a shadow user' do
+                post '/v3/roles', params.to_json, admin_header
+
+                expect(last_response).to have_status_code(201)
+                expect(parsed_response).to match_json_response(expected_response)
+
+                expect(uaa_client).to have_received(:ids_for_usernames_and_origins)
+                expect(uaa_client).not_to have_received(:create_shadow_user)
+              end
+            end
+          end
+        end
       end
     end
 
-    context 'creating a role for a user that does not exist' do
-      let(:expected_response) do
-        {
-          guid: UUID_REGEX,
-          created_at: iso8601,
-          updated_at: iso8601,
-          type: 'organization_auditor',
-          relationships: {
-            user: {
-              data: { guid: 'a-new-user-guid' }
-            },
-            space: {
-              data: nil
-            },
-            organization: {
-              data: { guid: org.guid }
-            }
-          },
-          links: {
-            self: { href: %r{#{Regexp.escape(link_prefix)}/v3/roles/#{UUID_REGEX}} },
-            user: { href: %r{#{Regexp.escape(link_prefix)}/v3/users/a-new-user-guid} },
-            organization: { href: %r{#{Regexp.escape(link_prefix)}/v3/organizations/#{org.guid}} }
-          }
-        }
-      end
+    context 'creating a role for a user that does not exist in cc' do
+      let(:origin) { 'uaa' }
 
       before do
         allow(uaa_client).to receive(:usernames_for_ids).with(['a-new-user-guid']).and_return({ 'a-new-user-guid' => 'a-new-user-name' })
-        allow(uaa_client).to receive(:id_for_username).with('a-new-user-name', origin: 'uaa').and_return('a-new-user-guid')
+        allow(uaa_client).to receive(:id_for_username).with('a-new-user-name', origin:).and_return('a-new-user-guid')
         allow(uaa_client).to receive(:users_for_ids).with(['a-new-user-guid']).and_return({ 'a-new-user-guid' => { 'username' => 'a-new-user-name' } })
+        allow(uaa_client).to receive(:origins_for_username).with('a-new-user-name').and_return([origin])
       end
 
-      context 'by user guid' do
-        let(:params) do
+      context 'when the request is for a org role' do
+        let(:expected_response) do
           {
+            guid: UUID_REGEX,
+            created_at: iso8601,
+            updated_at: iso8601,
             type: 'organization_auditor',
             relationships: {
               user: {
                 data: { guid: 'a-new-user-guid' }
               },
+              space: {
+                data: nil
+              },
               organization: {
                 data: { guid: org.guid }
               }
+            },
+            links: {
+              self: { href: %r{#{Regexp.escape(link_prefix)}/v3/roles/#{UUID_REGEX}} },
+              user: { href: %r{#{Regexp.escape(link_prefix)}/v3/users/a-new-user-guid} },
+              organization: { href: %r{#{Regexp.escape(link_prefix)}/v3/organizations/#{org.guid}} }
             }
           }
         end
 
-        it 'creates the user and the role' do
-          expect(VCAP::CloudController::User.where(guid: 'a-new-user-guid').empty?).to be true
-          post '/v3/roles', params.to_json, admin_headers
+        let(:expected_codes_and_responses) do
+          h = Hash.new(
+            code: 403
+          )
+          h['admin'] = {
+            code: 201,
+            response_object: expected_response
+          }
+          h['org_manager'] = {
+            code: 201,
+            response_object: expected_response
+          }
+          h
+        end
+
+        context 'by user guid' do
+          let(:params) do
+            {
+              type: 'organization_auditor',
+              relationships: {
+                user: {
+                  data: { guid: 'a-new-user-guid' }
+                },
+                organization: {
+                  data: { guid: org.guid }
+                }
+              }
+            }
+          end
+
+          it_behaves_like 'permissions for single object endpoint', ALL_PERMISSIONS
+
+          it 'creates the user and the role' do
+            expect(VCAP::CloudController::User.where(guid: 'a-new-user-guid').empty?).to be true
+            post '/v3/roles', params.to_json, admin_headers
+
+            expect(last_response).to have_status_code(201)
+            expect(parsed_response).to match_json_response(expected_response)
+            expect(VCAP::CloudController::User.where(guid: 'a-new-user-guid').empty?).to be false
+          end
 
-          expect(last_response).to have_status_code(201)
-          expect(parsed_response).to match_json_response(expected_response)
-          expect(VCAP::CloudController::User.where(guid: 'a-new-user-guid').empty?).to be false
+          context 'when "allow_user_creation_by_org_manager" is enabled' do
+            before do
+              TestConfig.override(allow_user_creation_by_org_manager: true)
+            end
+
+            it 'does not call create_shadow_user' do
+              post '/v3/roles', params.to_json, admin_header
+              expect(last_response).to have_status_code(201)
+              expect(parsed_response).to match_json_response(expected_response)
+              expect(uaa_client).not_to have_received(:create_shadow_user)
+            end
+          end
         end
-      end
 
-      context 'by user name' do
-        let(:params) do
-          {
-            type: 'organization_auditor',
-            relationships: {
-              user: {
-                data: { username: 'a-new-user-name', origin: 'uaa' }
-              },
-              organization: {
-                data: { guid: org.guid }
+        context 'by user name' do
+          let(:params) do
+            {
+              type: 'organization_auditor',
+              relationships: {
+                user: {
+                  data: { username: 'a-new-user-name' }
+                },
+                organization: {
+                  data: { guid: org.guid }
+                }
               }
             }
-          }
+          end
+
+          it_behaves_like 'permissions for single object endpoint', ALL_PERMISSIONS
+
+          it 'creates the user and the role' do
+            expect(VCAP::CloudController::User.where(guid: 'a-new-user-guid').empty?).to be true
+            post '/v3/roles', params.to_json, admin_headers
+
+            expect(last_response).to have_status_code(201)
+            expect(parsed_response).to match_json_response(expected_response)
+            expect(VCAP::CloudController::User.where(guid: 'a-new-user-guid').empty?).to be false
+          end
+
+          context 'when "allow_user_creation_by_org_manager" is enabled' do
+            before do
+              TestConfig.override(allow_user_creation_by_org_manager: true)
+            end
+
+            it 'does not call create_shadow_user' do
+              post '/v3/roles', params.to_json, admin_header
+              expect(last_response).to have_status_code(201)
+              expect(parsed_response).to match_json_response(expected_response)
+              expect(uaa_client).not_to have_received(:create_shadow_user)
+            end
+          end
         end
 
-        it 'creates the user and the role' do
-          expect(VCAP::CloudController::User.where(guid: 'a-new-user-guid').empty?).to be true
-          post '/v3/roles', params.to_json, admin_headers
+        context 'by user name and origin' do
+          before do
+            allow(uaa_client).to receive(:ids_for_usernames_and_origins).and_return([])
+          end
+
+          let(:params) do
+            {
+              type: 'organization_auditor',
+              relationships: {
+                user: {
+                  data: { username: 'a-new-user-name', origin: origin }
+                },
+                organization: {
+                  data: { guid: org.guid }
+                }
+              }
+            }
+          end
+
+          it_behaves_like 'permissions for single object endpoint', ALL_PERMISSIONS
+
+          it 'creates the user and the role' do
+            expect(VCAP::CloudController::User.where(guid: 'a-new-user-guid').empty?).to be true
+            post '/v3/roles', params.to_json, admin_headers
 
-          expect(last_response).to have_status_code(201)
-          expect(parsed_response).to match_json_response(expected_response)
-          expect(VCAP::CloudController::User.where(guid: 'a-new-user-guid').empty?).to be false
+            expect(last_response).to have_status_code(201)
+            expect(parsed_response).to match_json_response(expected_response)
+            expect(VCAP::CloudController::User.where(guid: 'a-new-user-guid').empty?).to be false
+          end
+
+          context 'when "allow_user_creation_by_org_manager" is enabled' do
+            before do
+              TestConfig.override(allow_user_creation_by_org_manager: true)
+              allow(uaa_client).to receive(:create_shadow_user).and_return({ 'id' => 'a-new-user-guid' })
+            end
+
+            it 'does not create a uaa shadow user' do
+              post '/v3/roles', params.to_json, admin_header
+              expect(last_response).to have_status_code(201)
+              expect(parsed_response).to match_json_response(expected_response)
+              expect(uaa_client).not_to have_received(:create_shadow_user)
+            end
+
+            context 'origin is different from uaa' do
+              let(:origin) { 'idp.local' }
+
+              it_behaves_like 'permissions for single object endpoint', ALL_PERMISSIONS
+
+              it 'calls create_shadow_user and retrieves the guid of the user from uaa' do
+                post '/v3/roles', params.to_json, admin_header
+                expect(last_response).to have_status_code(201)
+                expect(parsed_response).to match_json_response(expected_response)
+                expect(uaa_client).to have_received(:create_shadow_user)
+              end
+            end
+          end
         end
       end
 
diff --git a/spec/request/users_spec.rb b/spec/request/users_spec.rb
index f6927dd6313..9da5256fb74 100644
--- a/spec/request/users_spec.rb
+++ b/spec/request/users_spec.rb
@@ -662,6 +662,55 @@
       end
 
       it_behaves_like 'permissions for single object endpoint', ALL_PERMISSIONS
+
+      context 'using username and origin' do
+        let(:params) do
+          {
+            username: 'some-user',
+            origin: 'idp.local'
+          }
+        end
+
+        let(:user_guid) { 'new-user-guid' }
+
+        let(:api_call) { ->(user_headers) { post '/v3/users', params.to_json, user_headers } }
+
+        let(:user_json) do
+          {
+            guid: user_guid,
+            created_at: iso8601,
+            updated_at: iso8601,
+            username: params[:username],
+            presentation_name: params[:username],
+            origin: params[:origin],
+            metadata: {
+              labels: {},
+              annotations: {}
+            },
+            links: {
+              self: { href: %r{#{Regexp.escape(link_prefix)}/v3/users/#{user_guid}} }
+            }
+          }
+        end
+
+        let(:expected_codes_and_responses) do
+          h = Hash.new(
+            code: 403
+          )
+          h['admin'] = {
+            code: 201,
+            response_object: user_json
+          }
+          h
+        end
+
+        before do
+          allow(CloudController::DependencyLocator.instance).to receive_messages(uaa_shadow_user_creation_client: uaa_client, uaa_username_lookup_client: uaa_client)
+          allow(uaa_client).to receive_messages(create_shadow_user: { 'id' => user_guid }, ids_for_usernames_and_origins: [])
+        end
+
+        it_behaves_like 'permissions for single object endpoint', ALL_PERMISSIONS
+      end
     end
 
     describe 'when creating a user that exists in uaa' do
@@ -720,7 +769,7 @@
         let(:uaa_client_id) { 'cc_routing' }
 
         before do
-          allow(uaa_client).to receive_messages(users_for_ids: {}, get_clients: [{ client_id: uaa_client_id }])
+          allow(uaa_client).to receive_messages(users_for_ids: {})
         end
 
         let(:api_call) { ->(user_headers) { post '/v3/users', params.to_json, user_headers } }
@@ -814,6 +863,175 @@
         end
       end
     end
+
+    context 'when "allow_user_creation_by_org_manager" is enabled' do
+      before do
+        TestConfig.override(allow_user_creation_by_org_manager: true)
+        allow(CloudController::DependencyLocator.instance).to receive_messages(uaa_shadow_user_creation_client: uaa_client, uaa_username_lookup_client: uaa_client)
+      end
+
+      describe 'when creating a user by guid' do
+        before do
+          allow(uaa_client).to receive(:users_for_ids).and_return({})
+        end
+
+        let(:api_call) { ->(user_headers) { post '/v3/users', params.to_json, user_headers } }
+
+        let(:user_json) do
+          {
+            guid: params[:guid],
+            created_at: iso8601,
+            updated_at: iso8601,
+            username: nil,
+            presentation_name: params[:guid],
+            origin: nil,
+            metadata: {
+              labels: {},
+              annotations: {}
+            },
+            links: {
+              self: { href: %r{#{Regexp.escape(link_prefix)}/v3/users/#{params[:guid]}} }
+            }
+          }
+        end
+
+        let(:expected_codes_and_responses) do
+          h = Hash.new(
+            code: 403
+          )
+          h['admin'] = {
+            code: 201,
+            response_object: user_json
+          }
+          h
+        end
+
+        it_behaves_like 'permissions for single object endpoint', ALL_PERMISSIONS
+      end
+
+      describe 'when creating a user by username and origin' do
+        let(:params) do
+          {
+            username: 'some-user',
+            origin: 'idp.local'
+          }
+        end
+        let(:user_guid) { 'new-user-guid' }
+
+        let(:api_call) { ->(user_headers) { post '/v3/users', params.to_json, user_headers } }
+
+        let(:user_json) do
+          {
+            guid: user_guid,
+            created_at: iso8601,
+            updated_at: iso8601,
+            username: params[:username],
+            presentation_name: params[:username],
+            origin: params[:origin],
+            metadata: {
+              labels: {},
+              annotations: {}
+            },
+            links: {
+              self: { href: %r{#{Regexp.escape(link_prefix)}/v3/users/#{user_guid}} }
+            }
+          }
+        end
+
+        let(:expected_codes_and_responses) do
+          h = Hash.new(
+            code: 403
+          )
+          h['admin'] = {
+            code: 201,
+            response_object: user_json
+          }
+          h['org_manager'] = {
+            code: 201,
+            response_object: user_json
+          }
+          h
+        end
+
+        before do
+          allow(uaa_client).to receive_messages(create_shadow_user: { 'id' => user_guid }, ids_for_usernames_and_origins: [])
+        end
+
+        it_behaves_like 'permissions for single object endpoint', ALL_PERMISSIONS
+
+        it 'constructs the response with the given information without calling the UAA again' do
+          post '/v3/users', params.to_json, admin_header
+
+          expect(last_response).to have_status_code(201)
+          expect(parsed_response).to match_json_response(user_json)
+
+          expect(uaa_client).not_to have_received(:users_for_ids)
+        end
+
+        context 'when user already exists in UAA' do
+          before do
+            allow(uaa_client).to receive(:ids_for_usernames_and_origins).and_return([user_guid])
+          end
+
+          it 'does not try to create a shadow user' do
+            post '/v3/users', params.to_json, admin_header
+
+            expect(last_response).to have_status_code(201)
+            expect(parsed_response).to match_json_response(user_json)
+
+            expect(uaa_client).not_to have_received(:create_shadow_user)
+          end
+        end
+      end
+
+      context 'when parameters are invalid' do
+        let(:params) do
+          {
+            guid: user_guid,
+            username: 'some-user',
+            origin: 'idp.local'
+          }
+        end
+        let(:user_guid) { 'new-user-guid' }
+
+        let(:api_call) { ->(user_headers) { post '/v3/users', params.to_json, user_headers } }
+
+        let(:user_json) do
+          {
+            guid: user_guid,
+            created_at: iso8601,
+            updated_at: iso8601,
+            username: params[:username],
+            presentation_name: params[:username],
+            origin: params[:origin],
+            metadata: {
+              labels: {},
+              annotations: {}
+            },
+            links: {
+              self: { href: %r{#{Regexp.escape(link_prefix)}/v3/users/#{user_guid}} }
+            }
+          }
+        end
+
+        let(:expected_codes_and_responses) do
+          h = Hash.new(
+            code: 403
+          )
+          h['admin'] = {
+            code: 422,
+            response_object: user_json
+          }
+          h['org_manager'] = {
+            code: 422,
+            response_object: user_json
+          }
+          h
+        end
+
+        it_behaves_like 'permissions for single object endpoint', ALL_PERMISSIONS
+      end
+    end
   end
 
   describe 'PATCH /v3/users/:guid' do
diff --git a/spec/unit/actions/user_create_spec.rb b/spec/unit/actions/user_create_spec.rb
index 8036e976444..6d182d97549 100644
--- a/spec/unit/actions/user_create_spec.rb
+++ b/spec/unit/actions/user_create_spec.rb
@@ -55,12 +55,16 @@ module VCAP::CloudController
       end
 
       describe 'creating users' do
-        context 'when creating a UAA user' do
+        before do
+          allow(User).to receive_messages(create_uaa_shadow_user: { 'id' => guid }, get_user_id_by_username_and_origin: nil)
+        end
+
+        context 'when creating a UAA user by guid' do
           let(:message) do
             UserCreateMessage.new({ guid:, metadata: })
           end
 
-          it 'creates a user' do
+          it 'creates a user in ccdb' do
             created_user = nil
             expect do
               created_user = subject.create(message:)
@@ -73,6 +77,57 @@ module VCAP::CloudController
             )
             expect(created_user).to have_annotations({ key_name: 'anno', value: 'tations' })
           end
+
+          it 'does not create a shadow user in uaa' do
+            subject.create(message:)
+            expect(User).not_to have_received(:create_uaa_shadow_user)
+          end
+        end
+
+        context 'when creating a UAA user by username and origin' do
+          let(:message) do
+            UserCreateMessage.new({ username:, origin:, metadata: })
+          end
+
+          it 'creates a user in ccdb' do
+            created_user = nil
+            expect do
+              created_user = subject.create(message:)
+            end.to change(User, :count).by(1)
+
+            expect(created_user.guid).to eq guid
+            expect(created_user).to have_labels(
+              { prefix: nil, key_name: 'release', value: 'stable' },
+              { prefix: 'seriouseats.com', key_name: 'potato', value: 'mashed' }
+            )
+            expect(created_user).to have_annotations({ key_name: 'anno', value: 'tations' })
+          end
+
+          it 'creates a shadow user in uaa' do
+            subject.create(message:)
+            expect(User).to have_received(:create_uaa_shadow_user).with(username, origin)
+          end
+
+          context 'when user already exists in UAA' do
+            before do
+              allow(User).to receive(:get_user_id_by_username_and_origin).and_return('some-user-guid')
+            end
+
+            it 'does not try to create a shadow user' do
+              subject.create(message:)
+              expect(User).not_to have_received(:create_uaa_shadow_user)
+            end
+          end
+
+          context 'when an UaaUnavailable error is raised' do
+            before do
+              allow(User).to receive(:create_uaa_shadow_user).and_raise(UaaUnavailable)
+            end
+
+            it 'raises the error' do
+              expect { subject.create(message:) }.to raise_error(UaaUnavailable)
+            end
+          end
         end
 
         context 'when creating a UAA client' do
diff --git a/spec/unit/lib/cloud_controller/dependency_locator_spec.rb b/spec/unit/lib/cloud_controller/dependency_locator_spec.rb
index 223d4965f1c..ff579e30e5e 100644
--- a/spec/unit/lib/cloud_controller/dependency_locator_spec.rb
+++ b/spec/unit/lib/cloud_controller/dependency_locator_spec.rb
@@ -568,4 +568,46 @@
       locator.bbs_task_client
     end
   end
+
+  describe 'uaa_shadow_user_creation_client' do
+    before do
+      TestConfig.override(
+        uaa: {
+          internal_url: 'https:/uaa.local',
+          ca_file: '/var/vcap/uaa.cert',
+          clients: [
+            {
+              'name' => 'cloud_controller_shadow_user_creation',
+              'id' => 'shadow_user_creation_client_id',
+              'secret' => 'shadow_user_creation_client_secret'
+            }
+          ]
+        }
+      )
+    end
+
+    it 'creates a new UAA client' do
+      client = locator.uaa_shadow_user_creation_client
+      expect(client).to be_an_instance_of(VCAP::CloudController::UaaClient)
+      expect(client.uaa_target).to eq('https:/uaa.local')
+      expect(client.ca_file).to eq('/var/vcap/uaa.cert')
+      expect(client.client_id).to eq('shadow_user_creation_client_id')
+      expect(client.secret).to eq('shadow_user_creation_client_secret')
+    end
+
+    context 'when the client does not exist in config' do
+      before do
+        TestConfig.override(
+          uaa: {
+            internal_url: 'https:/uaa.local',
+            ca_file: '/var/vcap/uaa.cert'
+          }
+        )
+      end
+
+      it 'returns nil' do
+        expect(locator.uaa_shadow_user_creation_client).to be_nil
+      end
+    end
+  end
 end
diff --git a/spec/unit/lib/uaa/uaa_client_spec.rb b/spec/unit/lib/uaa/uaa_client_spec.rb
index 98712c39857..04a23122020 100644
--- a/spec/unit/lib/uaa/uaa_client_spec.rb
+++ b/spec/unit/lib/uaa/uaa_client_spec.rb
@@ -469,6 +469,61 @@ module VCAP::CloudController
       end
     end
 
+    describe '#create_shadow_user' do
+      context 'when user does not exist in uaa' do
+        before do
+          scim = instance_double(CF::UAA::Scim)
+          allow(scim).to receive(:add).and_return({ 'id' => '7b5fe025-fe6b-4f82-af6e-2534523233a6', 'username' => 'test-user@idp.local', 'origin' => 'idp.local' })
+          allow(uaa_client).to receive_messages(scim:)
+        end
+
+        it 'creates the user and returns the id' do
+          shadow_user = uaa_client.create_shadow_user('test-user@idp.local', 'idp.local')
+          expect(shadow_user['id']).to eq('7b5fe025-fe6b-4f82-af6e-2534523233a6')
+          expect(shadow_user['username']).to eq('test-user@idp.local')
+          expect(shadow_user['origin']).to eq('idp.local')
+        end
+      end
+
+      context 'when user already exists in uaa' do
+        before do
+          scim = instance_double(CF::UAA::Scim)
+          allow(scim).to receive(:add).and_raise(
+            CF::UAA::TargetError.new({
+                                       'user_id' => 'ea2824df-b2b5-4444-a18f-b4c7b227a184',
+                                       'error_description' => 'Username already in use: test-user@idp.local',
+                                       'verified' => true,
+                                       'active' => true,
+                                       'error' => 'scim_resource_already_exists',
+                                       'message' => 'Username already in use: test-user@idp.local'
+                                     })
+          )
+          allow(uaa_client).to receive_messages(scim:)
+        end
+
+        it 'catches the exception and returns the id' do
+          shadow_user = uaa_client.create_shadow_user('test-user@idp.local', 'idp.local')
+          expect(shadow_user['id']).to eq('ea2824df-b2b5-4444-a18f-b4c7b227a184')
+        end
+      end
+
+      context 'when something goes wrong communicating with uaa' do
+        let(:uaa_error) { CF::UAA::UAAError.new('some error') }
+        let(:mock_logger) { double(:steno_logger, error: nil) }
+
+        before do
+          scim = instance_double(CF::UAA::Scim)
+          allow(scim).to receive(:add).and_raise(uaa_error)
+          allow(uaa_client).to receive_messages(scim: scim, logger: mock_logger)
+        end
+
+        it 'logs the error and raises UaaUnavailable' do
+          expect { uaa_client.create_shadow_user('test-user@idp.local', 'idp.local') }.to raise_error(UaaUnavailable)
+          expect(mock_logger).to have_received(:error).with("UAA request for creating a user failed: #{uaa_error.inspect}")
+        end
+      end
+    end
+
     describe '#id_for_username' do
       let(:username) { 'user@example.com' }
 
diff --git a/spec/unit/messages/user_create_message_spec.rb b/spec/unit/messages/user_create_message_spec.rb
index b3be8249a82..40427497f9b 100644
--- a/spec/unit/messages/user_create_message_spec.rb
+++ b/spec/unit/messages/user_create_message_spec.rb
@@ -27,7 +27,35 @@ module VCAP::CloudController
 
         it 'is not valid' do
           expect(subject).not_to be_valid
-          expect(subject.errors[:guid]).to include('must be between 1 and 200 characters')
+          expect(subject.errors[:guid]).to include("or 'username' and 'origin' must be provided")
+        end
+      end
+
+      context 'when guid and username is provided' do
+        let(:params) do
+          {
+            username: 'meow',
+            guid: 'some-user-guid'
+          }
+        end
+
+        it 'is not valid' do
+          expect(subject).not_to be_valid
+          expect(subject.errors[:username]).to include("cannot be provided with 'guid'")
+        end
+      end
+
+      context 'when guid and origin is provided' do
+        let(:params) do
+          {
+            origin: 'meow',
+            guid: 'some-user-guid'
+          }
+        end
+
+        it 'is not valid' do
+          expect(subject).not_to be_valid
+          expect(subject.errors[:origin]).to include("cannot be provided with 'guid'")
         end
       end
 
@@ -45,7 +73,7 @@ module VCAP::CloudController
         end
       end
 
-      context 'guid' do
+      describe 'guid' do
         context 'when not a string' do
           let(:params) do
             { guid: 5 }
@@ -75,6 +103,65 @@ module VCAP::CloudController
           end
         end
       end
+
+      describe 'username' do
+        context 'when not a string' do
+          let(:params) do
+            { username: 5 }
+          end
+
+          it 'is not valid' do
+            expect(subject).not_to be_valid
+            expect(subject.errors[:username]).to include('must be a string')
+          end
+        end
+
+        context 'when origin is missing' do
+          let(:params) do
+            { username: 5 }
+          end
+
+          it 'is not valid' do
+            expect(subject).not_to be_valid
+            expect(subject.errors[:username]).to include("can only be provided together with 'origin'")
+          end
+        end
+      end
+
+      describe 'origin' do
+        context 'when not a string' do
+          let(:params) do
+            { origin: 5 }
+          end
+
+          it 'is not valid' do
+            expect(subject).not_to be_valid
+            expect(subject.errors[:origin]).to include('must be a string')
+          end
+        end
+
+        context 'when username is missing' do
+          let(:params) do
+            { origin: 5 }
+          end
+
+          it 'is not valid' do
+            expect(subject).not_to be_valid
+            expect(subject.errors[:origin]).to include("can only be provided together with 'username'")
+          end
+        end
+
+        context 'when equal to "uaa"' do
+          let(:params) do
+            { origin: 'uaa' }
+          end
+
+          it 'is not valid' do
+            expect(subject).not_to be_valid
+            expect(subject.errors[:origin]).to include("cannot be 'uaa' when creating a user by username")
+          end
+        end
+      end
     end
   end
 end
diff --git a/spec/unit/models/runtime/user_spec.rb b/spec/unit/models/runtime/user_spec.rb
index 2bc17938dfb..ad10cdcf0d2 100644
--- a/spec/unit/models/runtime/user_spec.rb
+++ b/spec/unit/models/runtime/user_spec.rb
@@ -549,5 +549,74 @@ module VCAP::CloudController
         expect(billing_manager_result).to contain_exactly(org_billing_manager.id, other_user4.id)
       end
     end
+
+    describe '#create_uaa_shadow_user' do
+      let(:uaa_client) { instance_double(UaaClient) }
+
+      before do
+        allow(CloudController::DependencyLocator.instance).to receive(:uaa_shadow_user_creation_client).and_return(uaa_client)
+      end
+
+      context 'when uaa client returns shadow user info' do
+        before do
+          allow(uaa_client).to receive(:create_shadow_user).and_return({ 'id' => 'shadow-user-id' })
+        end
+
+        it 'returns shadow user info' do
+          shadow_user = User.create_uaa_shadow_user('shadow-user', 'idp.local')
+          expect(shadow_user['id']).to eq('shadow-user-id')
+        end
+      end
+
+      context 'when uaa client raises UaaUnavailable' do
+        before do
+          allow(uaa_client).to receive(:create_shadow_user).and_raise(UaaUnavailable)
+        end
+
+        it 'raises UaaUnavailable' do
+          expect { User.create_uaa_shadow_user('shadow-user', 'idp.local') }.to raise_error(UaaUnavailable)
+        end
+      end
+    end
+
+    describe '#get_user_id_by_username_and_origin' do
+      let(:uaa_client) { instance_double(UaaClient) }
+
+      before do
+        allow(CloudController::DependencyLocator.instance).to receive(:uaa_username_lookup_client).and_return(uaa_client)
+      end
+
+      context 'when uaa client returns user id' do
+        before do
+          allow(uaa_client).to receive(:ids_for_usernames_and_origins).and_return(['some-user-id'])
+        end
+
+        it 'returns user id' do
+          user_id = User.get_user_id_by_username_and_origin(['shadow-user'], ['idp.local'])
+          expect(user_id).to eq('some-user-id')
+        end
+      end
+
+      context 'when uaa client returns empty array' do
+        before do
+          allow(uaa_client).to receive(:ids_for_usernames_and_origins).and_return([])
+        end
+
+        it 'returns nil' do
+          user_id = User.get_user_id_by_username_and_origin(['shadow-user'], ['idp.local'])
+          expect(user_id).to be_nil
+        end
+      end
+
+      context 'when uaa client raises UaaUnavailable' do
+        before do
+          allow(uaa_client).to receive(:ids_for_usernames_and_origins).and_raise(UaaUnavailable)
+        end
+
+        it 'raises UaaUnavailable' do
+          expect { User.get_user_id_by_username_and_origin(['shadow-user'], ['idp.local']) }.to raise_error(UaaUnavailable)
+        end
+      end
+    end
   end
 end