security: add mode to restrict local user administration to admins (#14494)

Feature flag `enable_strict_user_management` restricts administration of local users and groups to subjects with administration access level. Administration access level belongs to cluster admins (members of the `administration_allowed_sids`) and also, if enabled, to database admins (owners of a database).

Feature flag `enable_database_admin` enables database admins as a concept.

Also allow admins to change ownership of the schema objects.

12 files changed, 1061 insertions, 153 deletions

diff --git a/ydb/core/kqp/ut/scheme/kqp_scheme_ut.cpp b/ydb/core/kqp/ut/scheme/kqp_scheme_ut.cpp
index 6fe2f3a677..4a5dd79c83 100644
--- a/ydb/core/kqp/ut/scheme/kqp_scheme_ut.cpp
+++ b/ydb/core/kqp/ut/scheme/kqp_scheme_ut.cpp
@@ -2823,7 +2823,7 @@ Y_UNIT_TEST_SUITE(KqpScheme) {
             auto result = userSession.AlterTable("/Root/SecondaryKeys/Index/indexImplTable", tableSettings).ExtractValueSync();
             UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::UNAUTHORIZED, result.GetIssues().ToString());
             UNIT_ASSERT_STRING_CONTAINS(result.GetIssues().ToString(),
-                "Error: Access denied for user@builtin to path Root/SecondaryKeys/Index/indexImplTable"
+                "Error: Access denied for user@builtin on path Root/SecondaryKeys/Index/indexImplTable"
             );
         }
         // grant necessary permission
diff --git a/ydb/core/kqp/workload_service/actors/scheme_actors.cpp b/ydb/core/kqp/workload_service/actors/scheme_actors.cpp
index e7e5e7b4c0..ef1e7c3dce 100644
--- a/ydb/core/kqp/workload_service/actors/scheme_actors.cpp
+++ b/ydb/core/kqp/workload_service/actors/scheme_actors.cpp
@@ -325,16 +325,19 @@ public:
 protected:
     void StartRequest() override {
         LOG_D("Start pool creating");
+        const auto& database = DatabaseIdToDatabase(DatabaseId);
+
         auto event = std::make_unique<TEvTxUserProxy::TEvProposeTransaction>();
 
         auto& schemeTx = *event->Record.MutableTransaction()->MutableModifyScheme();
-        schemeTx.SetWorkingDir(JoinPath({DatabaseIdToDatabase(DatabaseId), ".metadata/workload_manager/pools"}));
+        schemeTx.SetWorkingDir(JoinPath({database, ".metadata/workload_manager/pools"}));
         schemeTx.SetOperationType(NKikimrSchemeOp::ESchemeOpCreateResourcePool);
         schemeTx.SetInternal(true);
 
         BuildCreatePoolRequest(*schemeTx.MutableCreateResourcePool());
         BuildModifyAclRequest(*schemeTx.MutableModifyACL());
 
+        event->Record.SetDatabaseName(database);
         if (UserToken) {
             event->Record.SetUserToken(UserToken->SerializeAsString());
         }
diff --git a/ydb/core/protos/feature_flags.proto b/ydb/core/protos/feature_flags.proto
index b36662fef0..6739bea08e 100644
--- a/ydb/core/protos/feature_flags.proto
+++ b/ydb/core/protos/feature_flags.proto
@@ -190,4 +190,7 @@ message TFeatureFlags {
     optional bool EnableColumnStore = 165 [default = false];
     optional bool EnableStrictAclCheck = 166 [default = false];
     optional bool DatabaseYamlConfigAllowed = 167 [default = false];
+    // deny non-administrators the privilege of administering local users and groups
+    optional bool EnableStrictUserManagement = 168 [default = false];
+    optional bool EnableDatabaseAdmin = 169 [default = false];
 }
diff --git a/ydb/core/testlib/basics/feature_flags.h b/ydb/core/testlib/basics/feature_flags.h
index d42e3c23ed..8ffc35b923 100644
--- a/ydb/core/testlib/basics/feature_flags.h
+++ b/ydb/core/testlib/basics/feature_flags.h
@@ -73,6 +73,8 @@ public:
     FEATURE_FLAG_SETTER(EnableFollowerStats)
     FEATURE_FLAG_SETTER(EnableExportChecksums)
     FEATURE_FLAG_SETTER(EnableTopicTransfer)
+    FEATURE_FLAG_SETTER(EnableStrictUserManagement)
+    FEATURE_FLAG_SETTER(EnableDatabaseAdmin)
 
     #undef FEATURE_FLAG_SETTER
 };
diff --git a/ydb/core/testlib/test_client.cpp b/ydb/core/testlib/test_client.cpp
index a693daeaaa..a19b0151ac 100644
--- a/ydb/core/testlib/test_client.cpp
+++ b/ydb/core/testlib/test_client.cpp
@@ -2679,6 +2679,9 @@ namespace Tests {
         TAutoPtr<NMsgBusProxy::TBusBlobStorageConfigRequest> request(new NMsgBusProxy::TBusBlobStorageConfigRequest());
         request->Record.MutableRequest()->AddCommand()->MutableDefineStoragePool()->CopyFrom(storagePool);
         request->Record.SetDomain(Domain);
+        if (SecurityToken) {
+            request->Record.SetSecurityToken(SecurityToken);
+        }
 
         TAutoPtr<NBus::TBusMessage> reply;
         NBus::EMessageStatus msgStatus = SendWhenReady(request, reply);
diff --git a/ydb/core/testlib/test_client.h b/ydb/core/testlib/test_client.h
index 666b41f84c..27f7133742 100644
--- a/ydb/core/testlib/test_client.h
+++ b/ydb/core/testlib/test_client.h
@@ -300,6 +300,7 @@ namespace Tests {
             FeatureFlags.SetEnableColumnStore(true);
         }
 
+        TServerSettings() = default;
         TServerSettings(const TServerSettings& settings) = default;
         TServerSettings& operator=(const TServerSettings& settings) = default;
     private:
diff --git a/ydb/core/tx/schemeshard/ut_helpers/test_env.h b/ydb/core/tx/schemeshard/ut_helpers/test_env.h
index 7a62d39a84..f14d6ed5f5 100644
--- a/ydb/core/tx/schemeshard/ut_helpers/test_env.h
+++ b/ydb/core/tx/schemeshard/ut_helpers/test_env.h
@@ -73,6 +73,8 @@ namespace NSchemeShardUT_Private {
         OPTION(std::optional<bool>, EnableTopicTransfer, std::nullopt);
         OPTION(bool, SetupKqpProxy, false);
         OPTION(bool, EnableStrictAclCheck, false);
+        OPTION(std::optional<bool>, EnableStrictUserManagement, std::nullopt);
+        OPTION(std::optional<bool>, EnableDatabaseAdmin, std::nullopt);
 
         #undef OPTION
     };
diff --git a/ydb/core/tx/tx_proxy/schemereq.cpp b/ydb/core/tx/tx_proxy/schemereq.cpp
index 58efcfd0d6..159933ede9 100644
--- a/ydb/core/tx/tx_proxy/schemereq.cpp
+++ b/ydb/core/tx/tx_proxy/schemereq.cpp
@@ -1,6 +1,7 @@
 #include "proxy.h"
 
 #include <ydb/core/base/appdata.h>
+#include <ydb/core/base/auth.h>
 #include <ydb/core/base/path.h>
 #include <ydb/core/base/tablet_pipe.h>
 #include <ydb/core/base/tx_processing.h>
@@ -20,6 +21,10 @@
 namespace NKikimr {
 namespace NTxProxy {
 
+TString GetUserSID(const std::optional<NACLib::TUserToken>& userToken) {
+    return (userToken ? userToken->GetUserSID() : "<empty>");
+}
+
 template<typename TDerived>
 struct TBaseSchemeReq: public TActorBootstrapped<TDerived> {
     using TBase = TActorBootstrapped<TDerived>;
@@ -37,16 +42,16 @@ struct TBaseSchemeReq: public TActorBootstrapped<TDerived> {
     TActorId PipeClient;
 
     struct TPathToResolve {
-        NKikimrSchemeOp::EOperationType OperationRelated;
+        const NKikimrSchemeOp::TModifyScheme& ModifyScheme;
+        ui32 RequireAccess = NACLib::EAccessRights::NoAccess;
+        bool AllowedByLevel = true;
 
+        // Params for NSchemeCache::TSchemeCacheNavigate::TEntry
         TVector<TString> Path;
-        bool RequiredRedirect = true;
-        ui32 RequiredAccess = NACLib::EAccessRights::NoAccess;
-
-        std::optional<NKikimrSchemeOp::TModifyACL> RequiredGrandAccess;
+        bool RequireRedirect = true;
 
-        TPathToResolve(NKikimrSchemeOp::EOperationType opType)
-            : OperationRelated(opType)
+        TPathToResolve(const NKikimrSchemeOp::TModifyScheme& modifyScheme)
+            : ModifyScheme(modifyScheme)
         {
         }
     };
@@ -54,6 +59,10 @@ struct TBaseSchemeReq: public TActorBootstrapped<TDerived> {
     TVector<TPathToResolve> ResolveForACL;
 
     std::optional<NACLib::TUserToken> UserToken;
+    bool CheckAdministrator = false;
+    bool CheckDatabaseAdministrator = false;
+    bool IsClusterAdministrator = false;
+    bool IsDatabaseAdministrator = false;
 
     TBaseSchemeReq(const TTxProxyServices &services, ui64 txid, TAutoPtr<TEvTxProxyReq::TEvSchemeRequest> request, const TIntrusivePtr<TTxProxyMon> &txProxyMon)
         : Services(services)
@@ -513,9 +522,6 @@ struct TBaseSchemeReq: public TActorBootstrapped<TDerived> {
                     break;
             }
         }
-        LOG_DEBUG_S(ctx, NKikimrServices::TX_PROXY, "Actor# " << ctx.SelfID.ToString() << " txid# " << TxId
-            << " SEND to# " << Source.ToString() << " Source " << result->ToString());
-
         if (result->Record.GetSchemeShardReason()) {
             auto issueStatus = NKikimrIssues::TIssuesIds::DEFAULT_ERROR;
             if (result->Record.GetSchemeShardStatus() == NKikimrScheme::EStatus::StatusPathDoesNotExist) {
@@ -524,6 +530,12 @@ struct TBaseSchemeReq: public TActorBootstrapped<TDerived> {
             auto issue = MakeIssue(std::move(issueStatus), result->Record.GetSchemeShardReason());
             NYql::IssueToMessage(issue, result->Record.AddIssues());
         }
+        if (result->Record.IssuesSize() > 0) {
+            LOG_ERROR_S(ctx, NKikimrServices::TX_PROXY, "Actor# " << ctx.SelfID.ToString() << " txid# " << TxId
+                << ", issues: " << result->Record.GetIssues());
+        }
+        LOG_DEBUG_S(ctx, NKikimrServices::TX_PROXY, "Actor# " << ctx.SelfID.ToString() << " txid# " << TxId
+            << " SEND to# " << Source.ToString() << " Source " << result->ToString());
         ctx.Send(Source, result);
     }
 
@@ -532,8 +544,45 @@ struct TBaseSchemeReq: public TActorBootstrapped<TDerived> {
         ReportStatus(status, nullptr, nullptr, ctx);
     }
 
-    void Bootstrap(const TActorContext&) {
+    void Bootstrap(const TActorContext& ctx) {
         ExtractUserToken();
+
+        CheckAdministrator = AppData()->FeatureFlags.GetEnableStrictUserManagement();
+        CheckDatabaseAdministrator = CheckAdministrator && AppData()->FeatureFlags.GetEnableDatabaseAdmin();
+
+        LOG_DEBUG_S(ctx, NKikimrServices::TX_PROXY, "Actor# " << ctx.SelfID.ToString() << " txid# " << TxId
+            << " Bootstrap,"
+            << " UserSID: " << GetUserSID(UserToken)
+            << " CheckAdministrator: " << CheckAdministrator
+            << " CheckDatabaseAdministrator: " << CheckDatabaseAdministrator
+        );
+
+        // Resolve database to get its owner and be able to detect if user is the database admin
+        if (UserToken) {
+            IsClusterAdministrator = NKikimr::IsAdministrator(AppData(), &UserToken.value());
+            LOG_DEBUG_S(ctx, NKikimrServices::TX_PROXY, "Actor# " << ctx.SelfID.ToString() << " txid# " << TxId
+                << " Bootstrap,"
+                << " UserSID: " << GetUserSID(UserToken)
+                << " IsClusterAdministrator: " << IsClusterAdministrator
+            );
+
+            // Cluster admin trumps database admin, database owner check is needed only for database admin.
+            if (!IsClusterAdministrator && CheckDatabaseAdministrator) {
+                auto request = MakeHolder<NSchemeCache::TSchemeCacheNavigate>();
+                request->DatabaseName = CanonizePath(GetRequestProto().GetDatabaseName());
+
+                auto& entry = request->ResultSet.emplace_back();
+                entry.Operation = NSchemeCache::TSchemeCacheNavigate::OpPath;
+                entry.Path = NKikimr::SplitPath(request->DatabaseName);
+
+                ctx.Send(Services.SchemeCache, new TEvTxProxySchemeCache::TEvNavigateKeySet(request.Release()));
+
+                static_cast<TDerived*>(this)->Become(&TDerived::StateWaitResolveDatabase);
+                return;
+            }
+        }
+
+        static_cast<TDerived*>(this)->Start(ctx);
     }
 
     void Die(const TActorContext &ctx) override {
@@ -556,7 +605,9 @@ struct TBaseSchemeReq: public TActorBootstrapped<TDerived> {
             const auto &partition = desc.GetPartitionConfig();
             if (partition.HasPartitioningPolicy() && partition.GetPartitioningPolicy().GetSizeToSplit() > 0) {
                 if (PartitionConfigHasExternalBlobsEnabled(partition)) {
-                    LOG_ERROR_S(ctx, NKikimrServices::TX_PROXY, "Actor#" << ctx.SelfID.ToString() << " txid# " << TxId << " must not use auto-split and external blobs simultaneously, path# " << path);
+                    LOG_ERROR_S(ctx, NKikimrServices::TX_PROXY, "Actor#" << ctx.SelfID.ToString() << " txid# " << TxId
+                        << " must not use auto-split and external blobs simultaneously, path# " << path
+                    );
                     return false;
                 }
             }
@@ -605,29 +656,29 @@ struct TBaseSchemeReq: public TActorBootstrapped<TDerived> {
         case NKikimrSchemeOp::ESchemeOpAlterSubDomain:
         case NKikimrSchemeOp::ESchemeOpAlterExtSubDomain:
         {
-            auto toResolve = TPathToResolve(pbModifyScheme.GetOperationType());
+            auto toResolve = TPathToResolve(pbModifyScheme);
             toResolve.Path = Merge(workingDir, SplitPath(GetPathNameForScheme(pbModifyScheme)));
-            toResolve.RequiredAccess = NACLib::EAccessRights::CreateDatabase | NACLib::EAccessRights::AlterSchema | accessToUserAttrs;
-            toResolve.RequiredRedirect = false;
+            toResolve.RequireAccess = NACLib::EAccessRights::CreateDatabase | NACLib::EAccessRights::AlterSchema | accessToUserAttrs;
+            toResolve.RequireRedirect = false;
             ResolveForACL.push_back(toResolve);
             break;
         }
         case NKikimrSchemeOp::ESchemeOpCreateSubDomain:
         case NKikimrSchemeOp::ESchemeOpCreateExtSubDomain:
         {
-            auto toResolve = TPathToResolve(pbModifyScheme.GetOperationType());
+            auto toResolve = TPathToResolve(pbModifyScheme);
             toResolve.Path = workingDir;
-            toResolve.RequiredAccess = NACLib::EAccessRights::CreateDatabase | accessToUserAttrs;
-            toResolve.RequiredRedirect = false;
+            toResolve.RequireAccess = NACLib::EAccessRights::CreateDatabase | accessToUserAttrs;
+            toResolve.RequireRedirect = false;
             ResolveForACL.push_back(toResolve);
             break;
         }
         case NKikimrSchemeOp::ESchemeOpAlterUserAttributes:
         {
-            auto toResolve = TPathToResolve(pbModifyScheme.GetOperationType());
+            auto toResolve = TPathToResolve(pbModifyScheme);
             toResolve.Path = Merge(workingDir, SplitPath(GetPathNameForScheme(pbModifyScheme)));
-            toResolve.RequiredAccess = NACLib::EAccessRights::WriteUserAttributes | accessToUserAttrs;
-            toResolve.RequiredRedirect = false;
+            toResolve.RequireAccess = NACLib::EAccessRights::WriteUserAttributes | accessToUserAttrs;
+            toResolve.RequireRedirect = false;
             ResolveForACL.push_back(toResolve);
             break;
         }
@@ -635,9 +686,9 @@ struct TBaseSchemeReq: public TActorBootstrapped<TDerived> {
             auto& path = pbModifyScheme.GetSplitMergeTablePartitions().GetTablePath();
             TString baseDir = ToString(ExtractParent(path)); // why baseDir?
 
-            auto toResolve = TPathToResolve(pbModifyScheme.GetOperationType());
+            auto toResolve = TPathToResolve(pbModifyScheme);
             toResolve.Path = SplitPath(baseDir);
-            toResolve.RequiredAccess = NACLib::EAccessRights::NoAccess; // why not?
+            toResolve.RequireAccess = NACLib::EAccessRights::NoAccess; // why not?
             ResolveForACL.push_back(toResolve);
             break;
         }
@@ -667,17 +718,17 @@ struct TBaseSchemeReq: public TActorBootstrapped<TDerived> {
         case NKikimrSchemeOp::ESchemeOpAlterResourcePool:
         case NKikimrSchemeOp::ESchemeOpAlterBackupCollection:
         {
-            auto toResolve = TPathToResolve(pbModifyScheme.GetOperationType());
+            auto toResolve = TPathToResolve(pbModifyScheme);
             toResolve.Path = Merge(workingDir, SplitPath(GetPathNameForScheme(pbModifyScheme)));
-            toResolve.RequiredAccess = NACLib::EAccessRights::AlterSchema | accessToUserAttrs;
+            toResolve.RequireAccess = NACLib::EAccessRights::AlterSchema | accessToUserAttrs;
             ResolveForACL.push_back(toResolve);
             break;
         }
         case NKikimrSchemeOp::ESchemeOpRestoreMultipleIncrementalBackups:
         {
-            auto toResolve = TPathToResolve(pbModifyScheme.GetOperationType());
+            auto toResolve = TPathToResolve(pbModifyScheme);
             toResolve.Path = SplitPath(GetPathNameForScheme(pbModifyScheme));
-            toResolve.RequiredAccess = NACLib::EAccessRights::AlterSchema | accessToUserAttrs;
+            toResolve.RequireAccess = NACLib::EAccessRights::AlterSchema | accessToUserAttrs;
             ResolveForACL.push_back(toResolve);
             break;
         }
@@ -702,48 +753,47 @@ struct TBaseSchemeReq: public TActorBootstrapped<TDerived> {
         case NKikimrSchemeOp::ESchemeOpDropResourcePool:
         case NKikimrSchemeOp::ESchemeOpDropBackupCollection:
         {
-            auto toResolve = TPathToResolve(pbModifyScheme.GetOperationType());
+            auto toResolve = TPathToResolve(pbModifyScheme);
             toResolve.Path = Merge(workingDir, SplitPath(GetPathNameForScheme(pbModifyScheme)));
-            toResolve.RequiredAccess = NACLib::EAccessRights::RemoveSchema;
+            toResolve.RequireAccess = NACLib::EAccessRights::RemoveSchema;
             ResolveForACL.push_back(toResolve);
             break;
         }
         case NKikimrSchemeOp::ESchemeOpDropSubDomain:
         case NKikimrSchemeOp::ESchemeOpForceDropSubDomain:
         case NKikimrSchemeOp::ESchemeOpForceDropExtSubDomain: {
-            auto toResolve = TPathToResolve(pbModifyScheme.GetOperationType());
+            auto toResolve = TPathToResolve(pbModifyScheme);
             toResolve.Path = Merge(workingDir, SplitPath(GetPathNameForScheme(pbModifyScheme)));
-            toResolve.RequiredAccess = NACLib::EAccessRights::DropDatabase;
-            toResolve.RequiredRedirect = false;
+            toResolve.RequireAccess = NACLib::EAccessRights::DropDatabase;
+            toResolve.RequireRedirect = false;
             ResolveForACL.push_back(toResolve);
             break;
         }
         case NKikimrSchemeOp::ESchemeOpForceDropUnsafe: {
-            auto toResolve = TPathToResolve(pbModifyScheme.GetOperationType());
+            auto toResolve = TPathToResolve(pbModifyScheme);
             toResolve.Path = Merge(workingDir, SplitPath(GetPathNameForScheme(pbModifyScheme)));
-            toResolve.RequiredAccess = NACLib::EAccessRights::DropDatabase | NACLib::EAccessRights::RemoveSchema;
-            toResolve.RequiredRedirect = false;
+            toResolve.RequireAccess = NACLib::EAccessRights::DropDatabase | NACLib::EAccessRights::RemoveSchema;
+            toResolve.RequireRedirect = false;
             ResolveForACL.push_back(toResolve);
             break;
         }
         case NKikimrSchemeOp::ESchemeOpModifyACL: {
-            auto toResolve = TPathToResolve(pbModifyScheme.GetOperationType());
+            auto toResolve = TPathToResolve(pbModifyScheme);
             toResolve.Path = Merge(workingDir, SplitPath(GetPathNameForScheme(pbModifyScheme)));
-            toResolve.RequiredAccess = NACLib::EAccessRights::GrantAccessRights | accessToUserAttrs;
-            toResolve.RequiredGrandAccess = pbModifyScheme.GetModifyACL();
+            toResolve.RequireAccess = NACLib::EAccessRights::GrantAccessRights | accessToUserAttrs;
             ResolveForACL.push_back(toResolve);
             break;
         }
         case NKikimrSchemeOp::ESchemeOpCreateTable: {
-            auto toResolve = TPathToResolve(pbModifyScheme.GetOperationType());
+            auto toResolve = TPathToResolve(pbModifyScheme);
             toResolve.Path = workingDir;
-            toResolve.RequiredAccess = NACLib::EAccessRights::CreateTable | accessToUserAttrs;
+            toResolve.RequireAccess = NACLib::EAccessRights::CreateTable | accessToUserAttrs;
             ResolveForACL.push_back(toResolve);
 
             if (pbModifyScheme.GetCreateTable().HasCopyFromTable()) {
-                auto toResolve = TPathToResolve(pbModifyScheme.GetOperationType());
+                auto toResolve = TPathToResolve(pbModifyScheme);
                 toResolve.Path = SplitPath(pbModifyScheme.GetCreateTable().GetCopyFromTable());
-                toResolve.RequiredAccess = NACLib::EAccessRights::SelectRow;
+                toResolve.RequireAccess = NACLib::EAccessRights::SelectRow;
                 ResolveForACL.push_back(toResolve);
             }
             break;
@@ -766,70 +816,70 @@ struct TBaseSchemeReq: public TActorBootstrapped<TDerived> {
         case NKikimrSchemeOp::ESchemeOpCreateResourcePool:
         case NKikimrSchemeOp::ESchemeOpCreateBackupCollection:
         {
-            auto toResolve = TPathToResolve(pbModifyScheme.GetOperationType());
+            auto toResolve = TPathToResolve(pbModifyScheme);
             toResolve.Path = workingDir;
-            toResolve.RequiredAccess = NACLib::EAccessRights::CreateTable | accessToUserAttrs;
+            toResolve.RequireAccess = NACLib::EAccessRights::CreateTable | accessToUserAttrs;
             ResolveForACL.push_back(toResolve);
             break;
         }
         case NKikimrSchemeOp::ESchemeOpCreateConsistentCopyTables: {
             for (auto& item: pbModifyScheme.GetCreateConsistentCopyTables().GetCopyTableDescriptions()) {
                 {
-                    auto toResolve = TPathToResolve(pbModifyScheme.GetOperationType());
+                    auto toResolve = TPathToResolve(pbModifyScheme);
                     toResolve.Path = SplitPath(item.GetSrcPath());
-                    toResolve.RequiredAccess = NACLib::EAccessRights::SelectRow;
+                    toResolve.RequireAccess = NACLib::EAccessRights::SelectRow;
                     ResolveForACL.push_back(toResolve);
                 }
                 {
-                    auto toResolve = TPathToResolve(pbModifyScheme.GetOperationType());
+                    auto toResolve = TPathToResolve(pbModifyScheme);
                     auto dstDir = ToString(ExtractParent(item.GetDstPath()));
                     toResolve.Path = SplitPath(dstDir);
-                    toResolve.RequiredAccess = NACLib::EAccessRights::CreateTable;
+                    toResolve.RequireAccess = NACLib::EAccessRights::CreateTable;
                     ResolveForACL.push_back(toResolve);
                 }
             }
             break;
         }
         case NKikimrSchemeOp::ESchemeOpBackupBackupCollection: {
-            auto toResolve = TPathToResolve(pbModifyScheme.GetOperationType());
+            auto toResolve = TPathToResolve(pbModifyScheme);
             toResolve.Path = workingDir;
             auto collectionPath = SplitPath(pbModifyScheme.GetBackupBackupCollection().GetName());
             std::move(collectionPath.begin(), collectionPath.end(), std::back_inserter(toResolve.Path));
-            toResolve.RequiredAccess = NACLib::EAccessRights::GenericWrite;
+            toResolve.RequireAccess = NACLib::EAccessRights::GenericWrite;
             ResolveForACL.push_back(toResolve);
             break;
         }
         case NKikimrSchemeOp::ESchemeOpBackupIncrementalBackupCollection: {
-            auto toResolve = TPathToResolve(pbModifyScheme.GetOperationType());
+            auto toResolve = TPathToResolve(pbModifyScheme);
             toResolve.Path = workingDir;
             auto collectionPath = SplitPath(pbModifyScheme.GetBackupIncrementalBackupCollection().GetName());
             std::move(collectionPath.begin(), collectionPath.end(), std::back_inserter(toResolve.Path));
-            toResolve.RequiredAccess = NACLib::EAccessRights::GenericWrite;
+            toResolve.RequireAccess = NACLib::EAccessRights::GenericWrite;
             ResolveForACL.push_back(toResolve);
             break;
         }
         case NKikimrSchemeOp::ESchemeOpRestoreBackupCollection: {
-            auto toResolve = TPathToResolve(pbModifyScheme.GetOperationType());
+            auto toResolve = TPathToResolve(pbModifyScheme);
             toResolve.Path = workingDir;
             auto collectionPath = SplitPath(pbModifyScheme.GetRestoreBackupCollection().GetName());
             std::move(collectionPath.begin(), collectionPath.end(), std::back_inserter(toResolve.Path));
-            toResolve.RequiredAccess = NACLib::EAccessRights::GenericWrite;
+            toResolve.RequireAccess = NACLib::EAccessRights::GenericWrite;
             ResolveForACL.push_back(toResolve);
             break;
         }
         case NKikimrSchemeOp::ESchemeOpMoveTable: {
             auto& descr = pbModifyScheme.GetMoveTable();
             {
-                auto toResolve = TPathToResolve(pbModifyScheme.GetOperationType());
+                auto toResolve = TPathToResolve(pbModifyScheme);
                 toResolve.Path = SplitPath(descr.GetSrcPath());
-                toResolve.RequiredAccess = NACLib::EAccessRights::SelectRow | NACLib::EAccessRights::RemoveSchema;
+                toResolve.RequireAccess = NACLib::EAccessRights::SelectRow | NACLib::EAccessRights::RemoveSchema;
                 ResolveForACL.push_back(toResolve);
             }
             {
-                auto toResolve = TPathToResolve(pbModifyScheme.GetOperationType());
+                auto toResolve = TPathToResolve(pbModifyScheme);
                 auto dstDir = ToString(ExtractParent(descr.GetDstPath()));
                 toResolve.Path = SplitPath(dstDir);
-                toResolve.RequiredAccess = NACLib::EAccessRights::CreateTable;
+                toResolve.RequireAccess = NACLib::EAccessRights::CreateTable;
                 ResolveForACL.push_back(toResolve);
             }
             break;
@@ -837,54 +887,50 @@ struct TBaseSchemeReq: public TActorBootstrapped<TDerived> {
         case NKikimrSchemeOp::ESchemeOpMoveIndex: {
             auto& descr = pbModifyScheme.GetMoveIndex();
             {
-                auto toResolve = TPathToResolve(pbModifyScheme.GetOperationType());
+                auto toResolve = TPathToResolve(pbModifyScheme);
                 toResolve.Path = SplitPath(descr.GetTablePath());
-                toResolve.RequiredAccess = NACLib::EAccessRights::AlterSchema;
+                toResolve.RequireAccess = NACLib::EAccessRights::AlterSchema;
                 ResolveForACL.push_back(toResolve);
             }
             break;
         }
         case NKikimrSchemeOp::ESchemeOpMkDir:
         {
-            auto toResolve = TPathToResolve(pbModifyScheme.GetOperationType());
+            auto toResolve = TPathToResolve(pbModifyScheme);
             toResolve.Path = workingDir;
-            toResolve.RequiredAccess = NACLib::EAccessRights::CreateDirectory | accessToUserAttrs;
+            toResolve.RequireAccess = NACLib::EAccessRights::CreateDirectory | accessToUserAttrs;
             ResolveForACL.push_back(toResolve);
             break;
         }
         case NKikimrSchemeOp::ESchemeOpCreatePersQueueGroup:
         {
-            auto toResolve = TPathToResolve(pbModifyScheme.GetOperationType());
+            auto toResolve = TPathToResolve(pbModifyScheme);
             toResolve.Path = workingDir;
-            toResolve.RequiredAccess = NACLib::EAccessRights::CreateQueue | accessToUserAttrs;
+            toResolve.RequireAccess = NACLib::EAccessRights::CreateQueue | accessToUserAttrs;
             ResolveForACL.push_back(toResolve);
             break;
         }
         case NKikimrSchemeOp::ESchemeOpAlterLogin:
         {
-            auto toResolve = TPathToResolve(pbModifyScheme.GetOperationType());
+            auto toResolve = TPathToResolve(pbModifyScheme);
             toResolve.Path = workingDir;
-            const auto& alter = pbModifyScheme.GetAlterLogin();
-
-            toResolve.RequiredAccess = (!IsChangeCanLoginOperation(alter) && IsSelfChangePasswordOperation(alter) ?
-                        NACLib::EAccessRights::NoAccess : 
-                        NACLib::EAccessRights::AlterSchema | accessToUserAttrs);
+            toResolve.RequireAccess = NACLib::EAccessRights::AlterSchema;
             ResolveForACL.push_back(toResolve);
             break;
         }
         case NKikimrSchemeOp::ESchemeOpMoveSequence: {
             auto& descr = pbModifyScheme.GetMoveSequence();
             {
-                auto toResolve = TPathToResolve(pbModifyScheme.GetOperationType());
+                auto toResolve = TPathToResolve(pbModifyScheme);
                 toResolve.Path = SplitPath(descr.GetSrcPath());
-                toResolve.RequiredAccess = NACLib::EAccessRights::RemoveSchema;
+                toResolve.RequireAccess = NACLib::EAccessRights::RemoveSchema;
                 ResolveForACL.push_back(toResolve);
             }
             {
-                auto toResolve = TPathToResolve(pbModifyScheme.GetOperationType());
+                auto toResolve = TPathToResolve(pbModifyScheme);
                 auto dstDir = ToString(ExtractParent(descr.GetDstPath()));
                 toResolve.Path = SplitPath(dstDir);
-                toResolve.RequiredAccess = NACLib::EAccessRights::CreateTable | accessToUserAttrs;
+                toResolve.RequireAccess = NACLib::EAccessRights::CreateTable | accessToUserAttrs;
                 ResolveForACL.push_back(toResolve);
             }
             break;
@@ -920,20 +966,6 @@ struct TBaseSchemeReq: public TActorBootstrapped<TDerived> {
         return true;
     }
 
-    bool IsChangeCanLoginOperation(const NKikimrSchemeOp::TAlterLogin& alterLogin) {
-        return alterLogin.GetAlterCase() == NKikimrSchemeOp::TAlterLogin::kModifyUser && alterLogin.GetModifyUser().HasCanLogin();
-    }
-
-    bool IsSelfChangePasswordOperation(const NKikimrSchemeOp::TAlterLogin& alterLogin) {
-        if (alterLogin.GetAlterCase() == NKikimrSchemeOp::TAlterLogin::kModifyUser && alterLogin.GetModifyUser().HasPassword()) {
-            if (UserToken) {
-                const auto& modifyUser = alterLogin.GetModifyUser();
-                return UserToken->GetUserSID() == modifyUser.GetUser();
-            }
-        }
-        return false;
-    }
-
     THolder<NSchemeCache::TSchemeCacheNavigate> ResolveRequestForACL() {
         if (!ResolveForACL) {
             return {};
@@ -952,7 +984,7 @@ struct TBaseSchemeReq: public TActorBootstrapped<TDerived> {
             NSchemeCache::TSchemeCacheNavigate::TEntry entry;
             entry.Operation = NSchemeCache::TSchemeCacheNavigate::OpPath;
             entry.Path = toReq.Path;
-            entry.RedirectRequired = toReq.RequiredRedirect;
+            entry.RedirectRequired = toReq.RequireRedirect;
             entry.SyncVersion = true;
             entry.ShowPrivatePath = true;
 
@@ -967,7 +999,7 @@ struct TBaseSchemeReq: public TActorBootstrapped<TDerived> {
             return resolveResult.DomainInfo->DomainKey.OwnerId;
         }
 
-        if (resolveTask.OperationRelated == NKikimrSchemeOp::ESchemeOpAlterUserAttributes) {
+        if (resolveTask.ModifyScheme.GetOperationType() == NKikimrSchemeOp::ESchemeOpAlterUserAttributes) {
             // ESchemeOpAlterUserAttributes applies on GSS when path is DB
             // but on GSS in other cases
             if (IsDB(resolveResult)) {
@@ -984,6 +1016,23 @@ struct TBaseSchemeReq: public TActorBootstrapped<TDerived> {
         }
     }
 
+    TString MakeAccessDeniedError(const TActorContext& ctx, const TVector<TString>& path, const TString& part) {
+        const TString msg = TStringBuilder() << "Access denied for " << GetUserSID(UserToken)
+            << " on path " << JoinPath(path)
+        ;
+        LOG_ERROR_S(ctx, NKikimrServices::TX_PROXY, "Actor# " << ctx.SelfID.ToString() << " txid# " << TxId
+            << ", " << msg << ", " << part
+        );
+        return msg;
+    }
+    TString MakeAccessDeniedError(const TActorContext& ctx, const TString& part) {
+        const TString msg = TStringBuilder() << "Access denied for " << GetUserSID(UserToken);
+        LOG_ERROR_S(ctx, NKikimrServices::TX_PROXY, "Actor# " << ctx.SelfID.ToString() << " txid# " << TxId
+            << ", " << msg << ", " << part
+        );
+        return msg;
+    }
+
     void InterpretResolveError(const NSchemeCache::TSchemeCacheNavigate* navigate, const TActorContext &ctx) {
         for (const auto& entry: navigate->ResultSet) {
             switch (entry.Status) {
@@ -992,13 +1041,9 @@ struct TBaseSchemeReq: public TActorBootstrapped<TDerived> {
 
             case NSchemeCache::TSchemeCacheNavigate::EStatus::AccessDenied: {
                 const ui32 access = NACLib::EAccessRights::DescribeSchema;
-                LOG_ERROR_S(ctx, NKikimrServices::TX_PROXY,
-                            "Access denied for " << (UserToken ? UserToken->GetUserSID() : "empty")
-                            << " with access " << NACLib::AccessRightsToString(access)
-                            << " to path " << JoinPath(entry.Path) << " because the base path");
-                const TString errString = TStringBuilder()
-                    << "Access denied for " << (UserToken ? UserToken->GetUserSID() : "empty")
-                    << " to path " << JoinPath(entry.Path);
+                const auto errString = MakeAccessDeniedError(ctx, entry.Path, TStringBuilder()
+                    << "with access " << NACLib::AccessRightsToString(access) << ": base path is inaccessible"
+                );
                 auto issue = MakeIssue(NKikimrIssues::TIssuesIds::ACCESS_DENIED, errString);
                 ReportStatus(TEvTxUserProxy::TEvProposeTransactionStatus::EStatus::AccessDenied, nullptr, &issue, ctx);
                 break;
@@ -1046,76 +1091,105 @@ struct TBaseSchemeReq: public TActorBootstrapped<TDerived> {
         }
     }
 
-    bool CheckACL(const NSchemeCache::TSchemeCacheNavigate::TResultSet& resolveSet, const TActorContext &ctx) {
+    bool CheckAccess(const NSchemeCache::TSchemeCacheNavigate::TResultSet& resolveSet, const TActorContext &ctx) {
+        const bool checkAdmin = (CheckAdministrator || CheckDatabaseAdministrator);
+        const bool isAdmin = (IsClusterAdministrator || IsDatabaseAdministrator);
+
         auto resolveIt = resolveSet.begin();
         auto requestIt = ResolveForACL.begin();
 
         while (resolveIt != resolveSet.end() && requestIt != ResolveForACL.end()) {
             const NSchemeCache::TSchemeCacheNavigate::TEntry& entry = *resolveIt;
             const TPathToResolve& request = *requestIt;
+            const auto& modifyScheme = request.ModifyScheme;
+
+            bool allowACLBypass = false;
+
+            // Check admin restrictions and special cases
+            if (modifyScheme.GetOperationType() == NKikimrSchemeOp::ESchemeOpAlterLogin) {
+                // User management allowed to any user or (if configured so) to admins only
+                if (checkAdmin && !isAdmin) {
+                    const auto errString = MakeAccessDeniedError(ctx, "attempt to manage user");
+                    auto issue = MakeIssue(NKikimrIssues::TIssuesIds::ACCESS_DENIED, errString);
+                    ReportStatus(TEvTxUserProxy::TEvProposeTransactionStatus::EStatus::AccessDenied, nullptr, &issue, ctx);
+                    return false;
+                }
+                allowACLBypass = checkAdmin && isAdmin;
+
+                // Any user can change their own password (but nothing else)
+                auto isUserChangesOwnPassword = [](const auto& modifyScheme, const NACLib::TSID& subjectSid) {
+                    const auto& alter = modifyScheme.GetAlterLogin();
+                    if (alter.GetAlterCase() == NKikimrSchemeOp::TAlterLogin::kModifyUser) {
+                        const auto& targetUser = alter.GetModifyUser();
+                        if (targetUser.HasPassword() && !targetUser.HasCanLogin()) {
+                            return (subjectSid == targetUser.GetUser());
+                        }
+                    }
+                    return false;
+                };
+                allowACLBypass = allowACLBypass || isUserChangesOwnPassword(modifyScheme, UserToken->GetUserSID());
+
+            } else if (modifyScheme.GetOperationType() == NKikimrSchemeOp::ESchemeOpModifyACL) {
+                // Only the owner of the schema object (path) can transfer their ownership away.
+                // Or admins (if configured so).
+                const auto& newOwner = modifyScheme.GetModifyACL().GetNewOwner();
+                if (!newOwner.empty()) {
+                    // That modifyACL is changing the owner
+                    auto isObjectOwner = [](const auto& userToken, const NACLib::TSID& owner) {
+                        return userToken->IsExist(owner);
+                    };
+                    const auto& owner = entry.Self->Info.GetOwner();
+                    const bool allow = (isAdmin || isObjectOwner(UserToken, owner));
+                    if (!allow) {
+                        const auto errString = MakeAccessDeniedError(ctx, entry.Path, TStringBuilder()
+                            << "attempt to change ownership"
+                            << " from " << owner
+                            << " to " << newOwner
+                        );
+                        auto issue = MakeIssue(NKikimrIssues::TIssuesIds::ACCESS_DENIED, errString);
+                        ReportStatus(TEvTxUserProxy::TEvProposeTransactionStatus::EStatus::AccessDenied, nullptr, &issue, ctx);
+                        return false;
+                    }
+                }
+            }
 
-            ui32 access = requestIt->RequiredAccess;
+            ui32 access = requestIt->RequireAccess;
 
             // request more rights if dst path is DB
-            if (request.OperationRelated == NKikimrSchemeOp::ESchemeOpAlterUserAttributes) {
+            if (modifyScheme.GetOperationType() == NKikimrSchemeOp::ESchemeOpAlterUserAttributes) {
                 if (IsDB(entry)) {
                     access |= NACLib::EAccessRights::GenericManage;
                 }
             }
 
-            if (access == NACLib::EAccessRights::NoAccess || !entry.SecurityObject) {
+            if (allowACLBypass || access == NACLib::EAccessRights::NoAccess || !entry.SecurityObject) {
                 ++resolveIt;
                 ++requestIt;
                 continue;
             }
 
             if (!entry.SecurityObject->CheckAccess(access, *UserToken)) {
-                LOG_ERROR_S(ctx, NKikimrServices::TX_PROXY,
-                            "Access denied for " << UserToken->GetUserSID()
-                            << " with access " << NACLib::AccessRightsToString(access)
-                            << " to path " << JoinPath(entry.Path));
-
-                const TString errString = TStringBuilder()
-                    << "Access denied for " << UserToken->GetUserSID()
-                    << " to path " << JoinPath(entry.Path);
+                const auto errString = MakeAccessDeniedError(ctx, entry.Path, TStringBuilder()
+                    << "with access " << NACLib::AccessRightsToString(access)
+                );
                 auto issue = MakeIssue(NKikimrIssues::TIssuesIds::ACCESS_DENIED, errString);
                 ReportStatus(TEvTxUserProxy::TEvProposeTransactionStatus::EStatus::AccessDenied, nullptr, &issue, ctx);
                 return false;
             }
 
-            if (request.OperationRelated == NKikimrSchemeOp::ESchemeOpModifyACL) {
-                const auto& modifyACL = *request.RequiredGrandAccess;
-                if (UserToken->IsExist(entry.SecurityObject->GetOwnerSID())) {
-                    ++resolveIt;
-                    ++requestIt;
-                    continue;
-                }
-
-                if (!modifyACL.GetNewOwner().empty()) {
-                    const TString errString = TStringBuilder()
-                        << "Access denied for " << UserToken->GetUserSID()
-                        << " to change ownership of " << JoinPath(entry.Path)
-                        << " to " << modifyACL.GetNewOwner();
-                    LOG_ERROR_S(ctx, NKikimrServices::TX_PROXY, errString);
-
-                    auto issue = MakeIssue(NKikimrIssues::TIssuesIds::ACCESS_DENIED, errString);
-                    ReportStatus(TEvTxUserProxy::TEvProposeTransactionStatus::EStatus::AccessDenied, nullptr, &issue, ctx);
-                    return false;
-                }
-
-                NACLib::TDiffACL diffACL(modifyACL.GetDiffACL());
-                if (!entry.SecurityObject->CheckGrantAccess(diffACL, *UserToken)) {
-                    LOG_ERROR_S(ctx, NKikimrServices::TX_PROXY,
-                                "Access denied for " << UserToken->GetUserSID()
-                                << " with diff ACL access " << NACLib::AccessRightsToString(NACLib::EAccessRights::GrantAccessRights)
-                                << " to path " << JoinPath(entry.Path));
-
-                    const TString errString = TStringBuilder()
-                        << "Access denied for " << UserToken->GetUserSID()
-                        << " to path " << JoinPath(entry.Path);
-                    auto issue = MakeIssue(NKikimrIssues::TIssuesIds::ACCESS_DENIED, errString);
-                    ReportStatus(TEvTxUserProxy::TEvProposeTransactionStatus::EStatus::AccessDenied, nullptr, &issue, ctx);
-                    return false;
+            if (modifyScheme.GetOperationType() == NKikimrSchemeOp::ESchemeOpModifyACL) {
+                const auto& modifyACL = modifyScheme.GetModifyACL();
+
+                if (!modifyACL.GetDiffACL().empty()) {
+                    NACLib::TDiffACL diffACL(modifyACL.GetDiffACL());
+                    if (!entry.SecurityObject->CheckGrantAccess(diffACL, *UserToken)) {
+                        const auto errString = MakeAccessDeniedError(ctx, entry.Path, TStringBuilder()
+                            << "with diff ACL access " << NACLib::AccessRightsToString(NACLib::EAccessRights::GrantAccessRights)
+                        );
+                        auto issue = MakeIssue(NKikimrIssues::TIssuesIds::ACCESS_DENIED, errString);
+                        ReportStatus(TEvTxUserProxy::TEvProposeTransactionStatus::EStatus::AccessDenied, nullptr, &issue, ctx);
+                        return false;
+                    }
                 }
             }
 
@@ -1200,6 +1274,48 @@ struct TBaseSchemeReq: public TActorBootstrapped<TDerived> {
         return Die(ctx);
     }
 
+    void HandleResolveDatabase(TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr &ev, const TActorContext &ctx) {
+        const NSchemeCache::TSchemeCacheNavigate& request = *ev->Get()->Request.Get();
+
+        LOG_DEBUG_S(ctx, NKikimrServices::TX_PROXY, "Actor# " << ctx.SelfID.ToString() << " txid# " << TxId
+            << " HandleResolveDatabase,"
+            << " ResultSet size: " << request.ResultSet.size()
+            << " ResultSet error count: " << request.ErrorCount
+        );
+
+        if (request.ResultSet.empty()) {
+            const TString msg = TStringBuilder() << "Error resolving database " << request.DatabaseName << ": no response";
+            LOG_ERROR_S(ctx, NKikimrServices::TX_PROXY, "Actor# " << ctx.SelfID.ToString() << " txid# " << TxId
+                << ", " << msg
+            );
+
+            TxProxyMon->ResolveKeySetWrongRequest->Inc();
+
+            const auto issue = MakeIssue(NKikimrIssues::TIssuesIds::GENERIC_RESOLVE_ERROR, msg);
+            ReportStatus(TEvTxUserProxy::TEvProposeTransactionStatus::EStatus::ResolveError, nullptr, &issue, ctx);
+            return Die(ctx);
+        }
+
+        if (request.ErrorCount > 0) {
+            InterpretResolveError(&request, ctx);
+            return Die(ctx);
+        }
+
+        const auto& database = request.ResultSet.front();
+        IsDatabaseAdministrator = NKikimr::IsDatabaseAdministrator(&UserToken.value(), database.Self->Info.GetOwner());
+
+        LOG_DEBUG_S(ctx, NKikimrServices::TX_PROXY, "Actor# " << ctx.SelfID.ToString() << " txid# " << TxId
+            << " HandleResolveDatabase,"
+            << " UserSID: " << GetUserSID(UserToken)
+            << " CheckAdministrator: " << CheckAdministrator
+            << " CheckDatabaseAdministrator: " << CheckDatabaseAdministrator
+            << " IsClusterAdministrator: " << IsClusterAdministrator
+            << " IsDatabaseAdministrator: " << IsDatabaseAdministrator
+        );
+
+        static_cast<TDerived*>(this)->Start(ctx);
+    }
+
     void Handle(TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr &ev, const TActorContext &ctx) {
         NSchemeCache::TSchemeCacheNavigate *navigate = ev->Get()->Request.Get();
 
@@ -1221,23 +1337,27 @@ struct TBaseSchemeReq: public TActorBootstrapped<TDerived> {
         Y_ABORT_UNLESS(!navigate->ResultSet.empty());
         Y_ABORT_UNLESS(navigate->ResultSet.size() == ResolveForACL.size());
 
-        ui64 shardToRequest = GetShardToRequest(*navigate->ResultSet.begin(), *ResolveForACL.begin());
-
-        auto request = MakeHolder<TEvSchemeShardPropose>(TxId, shardToRequest);
+        // Check user access level, permissions on scheme objects and other restrictions/permissions
         if (UserToken) {
-            request->Record.SetOwner(UserToken->GetUserSID());
-
-            if (!CheckACL(navigate->ResultSet, ctx)) {
+            if (!CheckAccess(navigate->ResultSet, ctx)) {
                 return Die(ctx);
             }
         }
 
+        // Check doc-api restrictions on operations
         if (IsDocApiRestricted(SchemeRequest->Ev->Get()->Record)) {
             if (!CheckDocApi(navigate->ResultSet, ctx)) {
-                    return Die(ctx);
+                return Die(ctx);
             }
         }
 
+        ui64 shardToRequest = GetShardToRequest(*navigate->ResultSet.begin(), *ResolveForACL.begin());
+        auto request = MakeHolder<TEvSchemeShardPropose>(TxId, shardToRequest);
+
+        if (UserToken) {
+            request->Record.SetOwner(UserToken->GetUserSID());
+        }
+
         request->Record.SetPeerName(GetRequestProto().GetPeerName());
         if (GetRequestEv().HasModifyScheme()) {
             request->Record.AddTransaction()->MergeFrom(GetModifyScheme());
@@ -1268,6 +1388,7 @@ struct TFlatSchemeReq : public TBaseSchemeReq<TFlatSchemeReq> {
     using TBase = TBaseSchemeReq<TFlatSchemeReq>;
 
     void Bootstrap(const TActorContext &ctx);
+    void Start(const TActorContext &ctx);
     void ProcessRequest(const TActorContext &ctx);
 
     void HandleWorkingDir(TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr &ev, const TActorContext &ctx);
@@ -1284,6 +1405,12 @@ struct TFlatSchemeReq : public TBaseSchemeReq<TFlatSchemeReq> {
         TBase::Die(ctx);
     }
 
+    STFUNC(StateWaitResolveDatabase) {
+        switch (ev->GetTypeRewrite()) {
+            HFunc(TEvTxProxySchemeCache::TEvNavigateKeySetResult, HandleResolveDatabase);
+        }
+    }
+
     STFUNC(StateWaitResolveWorkingDir) {
         switch (ev->GetTypeRewrite()) {
             HFunc(TEvTxProxySchemeCache::TEvNavigateKeySetResult, HandleWorkingDir);
@@ -1317,7 +1444,12 @@ void TFlatSchemeReq::Bootstrap(const TActorContext &ctx) {
     WallClockStarted = ctx.Now();
 
     TBase::Bootstrap(ctx);
+}
 
+void TFlatSchemeReq::Start(const TActorContext &ctx) {
+    //NOTE: split-merge operations here bypass access checks:
+    // - internal requests should not follow general rules
+    // - external requests are checked for admin rights elsewhere
     if (IsSplitMergeFromSchemeShard(GetModifyScheme())) {
         SendSplitMergePropose(ctx);
         Become(&TThis::StateWaitPrepare);
@@ -1390,7 +1522,9 @@ void TFlatSchemeReq::HandleWorkingDir(TEvTxProxySchemeCache::TEvNavigateKeySetRe
             << "Cannot resolve working dir"
             << " workingDir# " << (workingDir ? JoinPath(*workingDir) : "null")
             << " path# " << JoinPath(parts);
-        LOG_ERROR_S(ctx, NKikimrServices::TX_PROXY, errText);
+        LOG_ERROR_S(ctx, NKikimrServices::TX_PROXY, "Actor# " << ctx.SelfID.ToString() << " txid# " << TxId
+            << ", " << errText
+        );
 
         TxProxyMon->ResolveKeySetWrongRequest->Inc();
         const auto issue = MakeIssue(NKikimrIssues::TIssuesIds::GENERIC_RESOLVE_ERROR, errText);
@@ -1412,6 +1546,7 @@ struct TSchemeTransactionalReq : public TBaseSchemeReq<TSchemeTransactionalReq>
     using TBase = TBaseSchemeReq<TSchemeTransactionalReq>;
 
     void Bootstrap(const TActorContext &ctx);
+    void Start(const TActorContext &ctx);
 
     static constexpr NKikimrServices::TActivity::EType ActorActivityType() {
         return NKikimrServices::TActivity::TX_PROXY_SCHEMEREQ;
@@ -1425,6 +1560,12 @@ struct TSchemeTransactionalReq : public TBaseSchemeReq<TSchemeTransactionalReq>
         TBase::Die(ctx);
     }
 
+    STFUNC(StateWaitResolveDatabase) {
+        switch (ev->GetTypeRewrite()) {
+            HFunc(TEvTxProxySchemeCache::TEvNavigateKeySetResult, HandleResolveDatabase);
+        }
+    }
+
     STFUNC(StateWaitResolve) {
         switch (ev->GetTypeRewrite()) {
             HFunc(TEvTxProxySchemeCache::TEvNavigateKeySetResult, Handle);
@@ -1452,7 +1593,9 @@ void TSchemeTransactionalReq::Bootstrap(const TActorContext &ctx) {
     WallClockStarted = ctx.Now();
 
     TBase::Bootstrap(ctx);
+}
 
+void TSchemeTransactionalReq::Start(const TActorContext &ctx) {
     for(auto& scheme: GetModifications()) {
         if (!ExamineTables(scheme, ctx)) {
             ReportStatus(TEvTxUserProxy::TEvProposeTransactionStatus::EStatus::NotImplemented, ctx);
@@ -1478,6 +1621,7 @@ void TSchemeTransactionalReq::Bootstrap(const TActorContext &ctx) {
 
     LOG_DEBUG_S(ctx, NKikimrServices::TX_PROXY, "Actor# " << ctx.SelfID.ToString() << " txid# " << TxId << " TEvNavigateKeySet requested from SchemeCache");
     ctx.Send(Services.SchemeCache, new TEvTxProxySchemeCache::TEvNavigateKeySet(resolveRequest));
+
     Become(&TThis::StateWaitResolve);
     return;
 }
diff --git a/ydb/core/tx/tx_proxy/schemereq_ut.cpp b/ydb/core/tx/tx_proxy/schemereq_ut.cpp
new file mode 100644
index 0000000000..f252b52716
--- /dev/null
+++ b/ydb/core/tx/tx_proxy/schemereq_ut.cpp
@@ -0,0 +1,720 @@
+#include <library/cpp/testing/unittest/registar.h>
+
+#include <ydb-cpp-sdk/client/query/client.h>
+#include <ydb-cpp-sdk/client/scheme/scheme.h>
+#include <ydb-cpp-sdk/client/driver/driver.h>
+
+#include <ydb/core/base/path.h>
+#include <ydb/core/base/storage_pools.h>
+
+#include <ydb/core/testlib/test_client.h>
+
+#include <ydb/core/protos/subdomains.pb.h>
+#include <ydb/core/protos/console_tenant.pb.h>
+#include <ydb/core/grpc_services/base/base.h>
+#include <ydb/core/grpc_services/local_rpc/local_rpc.h>
+#include <ydb/public/api/grpc/ydb_auth_v1.grpc.pb.h>
+
+
+namespace NKikimr::NTxProxyUT {
+
+using namespace NYdb;
+
+// TTestEnv from proxy_ut_helpers.h does not fit for the tuning we need here.
+class TTestEnv {
+public:
+    TString RootToken;  // auth token of the superuser
+    TString RootPath;  // root database path
+
+    TPortManager PortManager;
+
+    Tests::TServerSettings::TPtr ServerSettings;
+    Tests::TServer::TPtr Server;
+    THolder<Tests::TClient> Client;
+    THolder<Tests::TTenants> Tenants;
+
+    TString Endpoint;
+    TDriverConfig DriverConfig;
+    THolder<TDriver> Driver;
+
+    Tests::TServer& GetTestServer() const {
+        return *Server;
+    }
+
+    Tests::TClient& GetTestClient() const {
+        return *Client;
+    }
+
+    Tests::TTenants& GetTestTenants() const {
+        return *Tenants;
+    }
+
+    TDriver& GetDriver() const {
+        return *Driver;
+    }
+
+    const TString& GetEndpoint() const {
+        return Endpoint;
+    }
+
+    const Tests::TServerSettings& GetSettings() const {
+        return *ServerSettings;
+    }
+
+    TTestEnv(const Tests::TServerSettings& settings, const TString rootToken) {
+        RootToken = rootToken;
+
+        auto mbusPort = PortManager.GetPort();
+        auto grpcPort = PortManager.GetPort();
+
+        Cerr << "Starting YDB, grpc: " << grpcPort << ", msgbus: " << mbusPort << Endl;
+
+        ServerSettings = new Tests::TServerSettings;
+        ServerSettings->Port = mbusPort;
+
+        // default settings
+        ServerSettings->AppConfig = std::make_shared<NKikimrConfig::TAppConfig>();
+        ServerSettings->AppConfig->MutableDomainsConfig()->MutableSecurityConfig()->AddAdministrationAllowedSIDs(RootToken);
+        ServerSettings->AuthConfig.SetUseBuiltinDomain(true);
+        ServerSettings->SetEnableMockOnSingleNode(false);
+
+        // settings possible override
+        // it's imperative that DomainName was without leading '/' -- is a name
+        ServerSettings->SetDomainName(ToString(ExtractDomain(settings.DomainName)));  // also creates storage pool for the root db
+        ServerSettings->SetNodeCount(settings.NodeCount);
+        ServerSettings->SetDynamicNodeCount(settings.DynamicNodeCount);
+        ServerSettings->SetUseRealThreads(settings.UseRealThreads);
+        ServerSettings->SetLoggerInitializer(settings.LoggerInitializer);
+
+        // feature flags
+        ServerSettings->SetFeatureFlags(settings.FeatureFlags);
+        ServerSettings->SetEnableAlterDatabaseCreateHiveFirst(true);
+
+        // additional storage pool type for use by tenant
+        ServerSettings->AddStoragePoolType("tenant-db");
+
+        // test server with default logging settings
+        Server = new Tests::TServer(ServerSettings);
+        Server->EnableGRpc(grpcPort);
+        {
+            auto& runtime = *Server->GetRuntime();
+            runtime.SetLogPriority(NKikimrServices::SCHEME_BOARD_REPLICA, NActors::NLog::PRI_ERROR);
+            runtime.SetLogPriority(NKikimrServices::SCHEME_BOARD_POPULATOR, NActors::NLog::PRI_ERROR);
+            runtime.SetLogPriority(NKikimrServices::SCHEME_BOARD_SUBSCRIBER, NActors::NLog::PRI_ERROR);
+            runtime.SetLogPriority(NKikimrServices::TX_PROXY_SCHEME_CACHE, NActors::NLog::PRI_ERROR);
+
+            runtime.SetLogPriority(NKikimrServices::TX_PROXY, NActors::NLog::PRI_DEBUG);
+        }
+
+        // test tenant control
+        Tenants = MakeHolder<Tests::TTenants>(Server);
+
+        // root database path
+        // it's imperative that RootPath was with leading '/' -- is a path
+        RootPath = CanonizePath(ServerSettings->DomainName);
+
+        // test client
+        Client = MakeHolder<Tests::TClient>(*ServerSettings);
+        Client->SetSecurityToken(RootToken);
+        Client->InitRootScheme();
+        Client->GrantConnect(RootToken);
+
+        // driver for actual grpc clients
+        Endpoint = "localhost:" + ToString(grpcPort);
+        DriverConfig = TDriverConfig()
+            .SetEndpoint(Endpoint)
+            .SetDatabase(RootPath)
+            .SetDiscoveryMode(EDiscoveryMode::Async)
+            .SetAuthToken(RootToken)
+        ;
+        Driver = MakeHolder<TDriver>(DriverConfig);
+    }
+
+    ~TTestEnv() {
+        Driver->Stop(true);
+    }
+
+    TStoragePools CreatePools(const TString& databaseName) {
+        TStoragePools result;
+        for (const auto& [kind, _]: ServerSettings->StoragePoolTypes) {
+            result.emplace_back(Client->CreateStoragePool(kind, databaseName), kind);
+        }
+        return result;
+    }
+};
+
+void CreateDatabase(TTestEnv& env, const TString& databaseName) {
+    NKikimrSubDomains::TSubDomainSettings subdomain;
+    subdomain.SetName(databaseName);
+    {
+        auto status = env.GetTestClient().CreateExtSubdomain(env.RootPath, subdomain);
+        UNIT_ASSERT_VALUES_EQUAL(status, NMsgBusProxy::MSTATUS_OK);
+    }
+    env.GetTestTenants().Run(JoinPath({env.RootPath, databaseName}), 1);
+    subdomain.SetExternalSchemeShard(true);
+    subdomain.SetPlanResolution(50);
+    subdomain.SetCoordinators(1);
+    subdomain.SetMediators(1);
+    subdomain.SetTimeCastBucketsPerMediator(2);
+    for (auto& pool : env.CreatePools(databaseName)) {
+        *subdomain.AddStoragePools() = pool;
+    }
+    {
+        auto status = env.GetTestClient().AlterExtSubdomain(env.RootPath, subdomain);
+        UNIT_ASSERT_VALUES_EQUAL(status, NMsgBusProxy::MSTATUS_OK);
+    }
+}
+
+TString LoginUser(TTestEnv& env, const TString& database, const TString& user, const TString& password) {
+    Ydb::Auth::LoginRequest request;
+    request.set_user(user);
+    request.set_password(password);
+
+    using TEvLoginRequest = NGRpcService::TGRpcRequestWrapperNoAuth<NGRpcService::TRpcServices::EvLogin, Ydb::Auth::LoginRequest, Ydb::Auth::LoginResponse>;
+
+    auto result = NRpcService::DoLocalRpc<TEvLoginRequest>(
+        std::move(request), database, {}, env.GetTestServer().GetRuntime()->GetActorSystem(0)
+    ).ExtractValueSync();
+
+    const auto& operation = result.operation();
+    UNIT_ASSERT_VALUES_EQUAL_C(operation.status(), Ydb::StatusIds::SUCCESS, operation.issues(0).message());
+    Ydb::Auth::LoginResult loginResult;
+    operation.result().UnpackTo(&loginResult);
+
+    return loginResult.token();
+}
+
+NYdb::NQuery::TQueryClient CreateQueryClient(const TTestEnv& env, const TString& token, const TString& database) {
+    NYdb::NQuery::TClientSettings settings;
+    settings.Database(database);
+    settings.AuthToken(token);
+    return NYdb::NQuery::TQueryClient(env.GetDriver(), settings);
+}
+
+NYdb::NScheme::TSchemeClient CreateSchemeClient(const TTestEnv& env, const TString& token) {
+    NYdb::TCommonClientSettings settings;
+    settings.AuthToken(token);
+    return NYdb::NScheme::TSchemeClient(env.GetDriver(), settings);
+}
+
+void CreateLocalUser(const TTestEnv& env, const TString& database, const TString& user) {
+    auto query = Sprintf(
+        R"(
+            CREATE USER %s PASSWORD 'passwd'
+        )",
+        user.c_str()
+    );
+    auto result = CreateQueryClient(env, env.RootToken, database).GetSession().GetValueSync().GetSession()
+        .ExecuteQuery(query,  NYdb::NQuery::TTxControl::NoTx()).ExtractValueSync();
+    UNIT_ASSERT_C(result.IsSuccess(), result.GetIssues().ToString());
+}
+
+void SetPermissions(const TTestEnv& env, const TString& path, const TString& targetSid, const std::vector<std::string>& permissions) {
+    auto client = CreateSchemeClient(env, env.RootToken);
+    auto modify = NYdb::NScheme::TModifyPermissionsSettings();
+    auto status = client.ModifyPermissions(path, modify.AddSetPermissions({targetSid, permissions}))
+        .ExtractValueSync();
+    UNIT_ASSERT_C(status.IsSuccess(), status.GetIssues().ToString());
+}
+
+void ChangeOwner(const TTestEnv& env, const TString& path, const TString& targetSid) {
+    auto client = CreateSchemeClient(env, env.RootToken);
+    auto modify = NYdb::NScheme::TModifyPermissionsSettings();
+    auto status = client.ModifyPermissions(path, modify.AddChangeOwner(targetSid))
+        .ExtractValueSync();
+    UNIT_ASSERT_C(status.IsSuccess(), status.GetIssues().ToString());
+}
+
+
+Y_UNIT_TEST_SUITE(SchemeReqAccess) {
+
+    enum class EAccessLevel {
+        User,
+        DatabaseAdmin,
+        ClusterAdmin,
+    };
+
+    const TString UserName = "ordinaryuser@builtin";
+    const TString DatabaseAdminName = "db_admin@builtin";
+    const TString ClusterAdminName = "cluster_admin@builtin";
+
+    TString BuiltinSubjectSid(EAccessLevel level) {
+        switch (level) {
+            case EAccessLevel::User:
+                return UserName;
+            case EAccessLevel::DatabaseAdmin:
+                return DatabaseAdminName;
+            case EAccessLevel::ClusterAdmin:
+                return ClusterAdminName;
+        }
+    }
+
+    const TString LocalUserName = "ordinaryuser";
+    const TString LocalDatabaseAdminName = "dbadmin";
+    const TString LocalClusterAdminName = "clusteradmin";
+
+    TString LocalSubjectSid(EAccessLevel level) {
+        switch (level) {
+            case EAccessLevel::User:
+                return LocalUserName;
+            case EAccessLevel::DatabaseAdmin:
+                return LocalDatabaseAdminName;
+            case EAccessLevel::ClusterAdmin:
+                return LocalClusterAdminName;
+        }
+    }
+
+    TString ToString(bool arg) {
+        return arg ? "true" : "false";
+    }
+
+    // dimensions:
+    // + EnforceUserTokenRequirement: true or false
+    // + EnableStrictUserManagement: true or false
+    // + EnableDatabaseAdmin: true or false
+    // - database: root or tenant
+    // + subject: user, db_admin, cluster_admin
+    // + subject: local or non-local
+    // - subject is admin directly or by group membership
+    // + subject permissions on database
+    // + protected operation: create|modify|drop user
+    // - protected operation: create|modify|drop group
+    // - protected operation: modify ACL - change owner
+    // - protected operation: modify ACL - change permissions
+
+    struct TAlterLoginTestCase {
+        TString Tag;
+        bool PrecreateTarget = false;
+        TString SqlStatement;
+        bool EnforceUserTokenRequirement = false;
+        EAccessLevel SubjectLevel;
+        std::vector<std::string> SubjectPermissions;
+        bool LocalSid = false;
+        bool EnableStrictUserManagement = false;
+        bool EnableDatabaseAdmin = false;
+        bool ExpectedResult;
+    };
+    void AlterLoginProtect_RootDB(NUnitTest::TTestContext&, const TAlterLoginTestCase params) {
+        auto settings = Tests::TServerSettings()
+            .SetNodeCount(1)
+            .SetDynamicNodeCount(1)
+            .SetEnableStrictUserManagement(params.EnableStrictUserManagement)
+            .SetEnableDatabaseAdmin(params.EnableDatabaseAdmin)
+            // .SetLoggerInitializer([](auto& runtime) {
+            //     runtime.SetLogPriority(NKikimrServices::FLAT_TX_SCHEMESHARD, NActors::NLog::PRI_INFO);
+            // })
+        ;
+        TTestEnv env(settings, /* rootToken*/ "root@builtin");
+
+        // Test context preparations
+
+        // Turn on mandatory authentication, if requested
+        env.GetTestServer().GetRuntime()->GetAppData().EnforceUserTokenRequirement = params.EnforceUserTokenRequirement;
+
+        // Create local user for the subject and obtain auth token, if requested
+        TString subjectSid;
+        TString subjectToken;
+        if (params.LocalSid) {
+            subjectSid = LocalSubjectSid(params.SubjectLevel);
+            CreateLocalUser(env, env.RootPath, subjectSid);
+            subjectToken = LoginUser(env, env.RootPath, subjectSid, "passwd");
+        } else {
+            subjectSid = subjectToken = BuiltinSubjectSid(params.SubjectLevel);
+        }
+
+        // Make subject a proper cluster admin, if requested
+        if (params.SubjectLevel == EAccessLevel::ClusterAdmin) {
+            env.GetTestServer().GetRuntime()->GetAppData().AdministrationAllowedSIDs.push_back(subjectSid);
+        }
+
+        // Give subject requested schema permissions
+        SetPermissions(env, env.RootPath, subjectSid, params.SubjectPermissions);
+
+        // Precreate target user, if requested
+        if (params.PrecreateTarget) {
+            CreateLocalUser(env, env.RootPath, "targetuser");
+        }
+
+        // Make subject a proper database admin (by transfer the database ownership to them), if requested
+        // This should be the last preparation step (as right after the transfer root@builtin will loose it privileges)
+        if (params.SubjectLevel == EAccessLevel::DatabaseAdmin) {
+            ChangeOwner(env, env.RootPath, subjectSid);
+        }
+
+        // Test body
+        {
+            auto client = CreateQueryClient(env, subjectToken, env.RootPath);
+            auto sessionResult = client.GetSession().ExtractValueSync();
+            UNIT_ASSERT_C(sessionResult.IsSuccess(), sessionResult.GetIssues().ToString());
+            auto session = sessionResult.GetSession();
+
+            // test body
+            auto result = session.ExecuteQuery(params.SqlStatement, NYdb::NQuery::TTxControl::NoTx()).ExtractValueSync();
+            UNIT_ASSERT_VALUES_EQUAL_C(result.IsSuccess(), params.ExpectedResult,
+                "query '" << params.SqlStatement << "'"
+                << ", subject " << subjectSid
+                << ", permissions '" << JoinSeq("|", params.SubjectPermissions) << "'"
+                << ", EnforceUserTokenRequirement " << ToString(params.EnforceUserTokenRequirement)
+                << ", EnableStrictUserManagement " << ToString(params.EnableStrictUserManagement)
+                << ", EnableDatabaseAdmin " << ToString(params.EnableDatabaseAdmin)
+                << ", expected result " << ToString(params.ExpectedResult)
+                << ", actual result " << ToString(result.IsSuccess()) << " '" << (result.IsSuccess() ? "" : result.GetIssues().ToString()) << "'"
+            );
+        }
+    }
+    static const std::vector<TAlterLoginTestCase> AlterLoginProtect_Tests = {
+        // CreateUser
+        // Cluster admin can always administer users, but require the same schema permissions as ordinary user (but why?).
+        { .Tag = "CreateUser", .SqlStatement = "CREATE USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::ClusterAdmin, .SubjectPermissions = {"ydb.database.connect", "ydb.granular.alter_schema"},
+            .EnableStrictUserManagement = false, .EnableDatabaseAdmin = false, .ExpectedResult = true
+        },
+        { .Tag = "CreateUser", .SqlStatement = "CREATE USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::ClusterAdmin, .SubjectPermissions = {"ydb.database.connect", "ydb.granular.alter_schema"},
+            .EnableStrictUserManagement = false, .EnableDatabaseAdmin = true, .ExpectedResult = true
+        },
+        { .Tag = "CreateUser", .SqlStatement = "CREATE USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::ClusterAdmin, .SubjectPermissions = {"ydb.database.connect", "ydb.granular.alter_schema"},
+            .EnableStrictUserManagement = true, .EnableDatabaseAdmin = false, .ExpectedResult = true
+        },
+        { .Tag = "CreateUser", .SqlStatement = "CREATE USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::ClusterAdmin, .SubjectPermissions = {"ydb.database.connect", "ydb.granular.alter_schema"},
+            .EnableStrictUserManagement = true, .EnableDatabaseAdmin = true, .ExpectedResult = true
+        },
+        { .Tag = "CreateUser", .SqlStatement = "CREATE USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::ClusterAdmin, .SubjectPermissions = {"ydb.database.connect"},
+            .EnableStrictUserManagement = false, .EnableDatabaseAdmin = false, .ExpectedResult = false
+        },
+        { .Tag = "CreateUser", .SqlStatement = "CREATE USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::ClusterAdmin, .SubjectPermissions = {"ydb.database.connect"},
+            .EnableStrictUserManagement = false, .EnableDatabaseAdmin = true, .ExpectedResult = false
+        },
+        { .Tag = "CreateUser", .SqlStatement = "CREATE USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::ClusterAdmin, .SubjectPermissions = {"ydb.database.connect"},
+            .EnableStrictUserManagement = true, .EnableDatabaseAdmin = false, .ExpectedResult = true
+        },
+        { .Tag = "CreateUser", .SqlStatement = "CREATE USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::ClusterAdmin, .SubjectPermissions = {"ydb.database.connect"},
+            .EnableStrictUserManagement = true, .EnableDatabaseAdmin = true, .ExpectedResult = true
+        },
+        // Database admin can administer users if EnableStrictUserManagement and EnableDatabaseAdmin are true.
+        // If not, database admin still can administer users as the owner of the database.
+        // In both cases it require no schema permissions except ydb.database.connect (why?).
+        { .Tag = "CreateUser", .SqlStatement = "CREATE USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::DatabaseAdmin, .SubjectPermissions = {"ydb.database.connect", "ydb.granular.alter_schema"},
+            .EnableStrictUserManagement = false, .EnableDatabaseAdmin = false, .ExpectedResult = true
+        },
+        { .Tag = "CreateUser", .SqlStatement = "CREATE USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::DatabaseAdmin, .SubjectPermissions = {"ydb.database.connect", "ydb.granular.alter_schema"},
+            .EnableStrictUserManagement = false, .EnableDatabaseAdmin = true, .ExpectedResult = true
+        },
+        { .Tag = "CreateUser", .SqlStatement = "CREATE USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::DatabaseAdmin, .SubjectPermissions = {"ydb.database.connect", "ydb.granular.alter_schema"},
+            .EnableStrictUserManagement = true, .EnableDatabaseAdmin = false, .ExpectedResult = false
+        },
+        { .Tag = "CreateUser", .SqlStatement = "CREATE USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::DatabaseAdmin, .SubjectPermissions = {"ydb.database.connect", "ydb.granular.alter_schema"},
+            .EnableStrictUserManagement = true, .EnableDatabaseAdmin = true, .ExpectedResult = true
+        },
+        { .Tag = "CreateUser", .SqlStatement = "CREATE USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::DatabaseAdmin, .SubjectPermissions = {"ydb.database.connect"},
+            .EnableStrictUserManagement = false, .EnableDatabaseAdmin = false, .ExpectedResult = true
+        },
+        { .Tag = "CreateUser", .SqlStatement = "CREATE USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::DatabaseAdmin, .SubjectPermissions = {"ydb.database.connect"},
+            .EnableStrictUserManagement = false, .EnableDatabaseAdmin = true, .ExpectedResult = true
+        },
+        { .Tag = "CreateUser", .SqlStatement = "CREATE USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::DatabaseAdmin, .SubjectPermissions = {"ydb.database.connect"},
+            .EnableStrictUserManagement = true, .EnableDatabaseAdmin = false, .ExpectedResult = false
+        },
+        { .Tag = "CreateUser", .SqlStatement = "CREATE USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::DatabaseAdmin, .SubjectPermissions = {"ydb.database.connect"},
+            .EnableStrictUserManagement = true, .EnableDatabaseAdmin = true, .ExpectedResult = true
+        },
+        // Ordinary user can create users only if EnableStrictUserManagement is false
+        // and ydb.granular.alter_schema is granted (besides ydb.database.connect).
+        { .Tag = "CreateUser", .SqlStatement = "CREATE USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::User, .SubjectPermissions = {"ydb.database.connect", "ydb.granular.alter_schema"},
+            .EnableStrictUserManagement = false, .EnableDatabaseAdmin = false, .ExpectedResult = true
+        },
+        { .Tag = "CreateUser", .SqlStatement = "CREATE USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::User, .SubjectPermissions = {"ydb.database.connect", "ydb.granular.alter_schema"},
+            .EnableStrictUserManagement = false, .EnableDatabaseAdmin = true, .ExpectedResult = true
+        },
+        { .Tag = "CreateUser", .SqlStatement = "CREATE USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::User, .SubjectPermissions = {"ydb.database.connect", "ydb.granular.alter_schema"},
+            .EnableStrictUserManagement = true, .EnableDatabaseAdmin = false, .ExpectedResult = false
+        },
+        { .Tag = "CreateUser", .SqlStatement = "CREATE USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::User, .SubjectPermissions = {"ydb.database.connect", "ydb.granular.alter_schema"},
+            .EnableStrictUserManagement = true, .EnableDatabaseAdmin = true, .ExpectedResult = false
+        },
+        { .Tag = "CreateUser", .SqlStatement = "CREATE USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::User, .SubjectPermissions = {"ydb.database.connect"},
+            .EnableStrictUserManagement = false, .EnableDatabaseAdmin = false, .ExpectedResult = false
+        },
+        { .Tag = "CreateUser", .SqlStatement = "CREATE USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::User, .SubjectPermissions = {"ydb.database.connect"},
+            .EnableStrictUserManagement = false, .EnableDatabaseAdmin = true, .ExpectedResult = false
+        },
+        { .Tag = "CreateUser", .SqlStatement = "CREATE USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::User, .SubjectPermissions = {"ydb.database.connect"},
+            .EnableStrictUserManagement = true, .EnableDatabaseAdmin = false, .ExpectedResult = false
+        },
+        { .Tag = "CreateUser", .SqlStatement = "CREATE USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::User, .SubjectPermissions = {"ydb.database.connect"},
+            .EnableStrictUserManagement = true, .EnableDatabaseAdmin = true, .ExpectedResult = false
+        },
+
+        // ModifyUser
+        // Cluster admin can always administer users, but require the same schema permissions as ordinary user (but why?).
+        { .Tag = "ModifyUser", .PrecreateTarget = true, .SqlStatement = "ALTER USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::ClusterAdmin, .SubjectPermissions = {"ydb.database.connect", "ydb.granular.alter_schema"},
+            .EnableStrictUserManagement = false, .EnableDatabaseAdmin = false, .ExpectedResult = true
+        },
+        { .Tag = "ModifyUser", .PrecreateTarget = true, .SqlStatement = "ALTER USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::ClusterAdmin, .SubjectPermissions = {"ydb.database.connect", "ydb.granular.alter_schema"},
+            .EnableStrictUserManagement = false, .EnableDatabaseAdmin = true, .ExpectedResult = true
+        },
+        { .Tag = "ModifyUser", .PrecreateTarget = true, .SqlStatement = "ALTER USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::ClusterAdmin, .SubjectPermissions = {"ydb.database.connect", "ydb.granular.alter_schema"},
+            .EnableStrictUserManagement = true, .EnableDatabaseAdmin = false, .ExpectedResult = true
+        },
+        { .Tag = "ModifyUser", .PrecreateTarget = true, .SqlStatement = "ALTER USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::ClusterAdmin, .SubjectPermissions = {"ydb.database.connect", "ydb.granular.alter_schema"},
+            .EnableStrictUserManagement = true, .EnableDatabaseAdmin = true, .ExpectedResult = true
+        },
+        { .Tag = "ModifyUser", .PrecreateTarget = true, .SqlStatement = "ALTER USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::ClusterAdmin, .SubjectPermissions = {"ydb.database.connect"},
+            .EnableStrictUserManagement = false, .EnableDatabaseAdmin = false, .ExpectedResult = false
+        },
+        { .Tag = "ModifyUser", .PrecreateTarget = true, .SqlStatement = "ALTER USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::ClusterAdmin, .SubjectPermissions = {"ydb.database.connect"},
+            .EnableStrictUserManagement = false, .EnableDatabaseAdmin = true, .ExpectedResult = false
+        },
+        { .Tag = "ModifyUser", .PrecreateTarget = true, .SqlStatement = "ALTER USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::ClusterAdmin, .SubjectPermissions = {"ydb.database.connect"},
+            .EnableStrictUserManagement = true, .EnableDatabaseAdmin = false, .ExpectedResult = true
+        },
+        { .Tag = "ModifyUser", .PrecreateTarget = true, .SqlStatement = "ALTER USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::ClusterAdmin, .SubjectPermissions = {"ydb.database.connect"},
+            .EnableStrictUserManagement = true, .EnableDatabaseAdmin = true, .ExpectedResult = true
+        },
+        // Database admin can administer users if EnableStrictUserManagement and EnableDatabaseAdmin are true.
+        // If not, database admin still can administer users as the owner of the database.
+        // In both cases it require no schema permissions except ydb.database.connect (why?).
+        { .Tag = "ModifyUser", .PrecreateTarget = true, .SqlStatement = "ALTER USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::DatabaseAdmin, .SubjectPermissions = {"ydb.database.connect", "ydb.granular.alter_schema"},
+            .EnableStrictUserManagement = false, .EnableDatabaseAdmin = false, .ExpectedResult = true
+        },
+        { .Tag = "ModifyUser", .PrecreateTarget = true, .SqlStatement = "ALTER USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::DatabaseAdmin, .SubjectPermissions = {"ydb.database.connect", "ydb.granular.alter_schema"},
+            .EnableStrictUserManagement = false, .EnableDatabaseAdmin = true, .ExpectedResult = true
+        },
+        { .Tag = "ModifyUser", .PrecreateTarget = true, .SqlStatement = "ALTER USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::DatabaseAdmin, .SubjectPermissions = {"ydb.database.connect", "ydb.granular.alter_schema"},
+            .EnableStrictUserManagement = true, .EnableDatabaseAdmin = false, .ExpectedResult = false
+        },
+        { .Tag = "ModifyUser", .PrecreateTarget = true, .SqlStatement = "ALTER USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::DatabaseAdmin, .SubjectPermissions = {"ydb.database.connect", "ydb.granular.alter_schema"},
+            .EnableStrictUserManagement = true, .EnableDatabaseAdmin = true, .ExpectedResult = true
+        },
+        { .Tag = "ModifyUser", .PrecreateTarget = true, .SqlStatement = "ALTER USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::DatabaseAdmin, .SubjectPermissions = {"ydb.database.connect"},
+            .EnableStrictUserManagement = false, .EnableDatabaseAdmin = false, .ExpectedResult = true
+        },
+        { .Tag = "ModifyUser", .PrecreateTarget = true, .SqlStatement = "ALTER USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::DatabaseAdmin, .SubjectPermissions = {"ydb.database.connect"},
+            .EnableStrictUserManagement = false, .EnableDatabaseAdmin = true, .ExpectedResult = true
+        },
+        { .Tag = "ModifyUser", .PrecreateTarget = true, .SqlStatement = "ALTER USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::DatabaseAdmin, .SubjectPermissions = {"ydb.database.connect"},
+            .EnableStrictUserManagement = true, .EnableDatabaseAdmin = false, .ExpectedResult = false
+        },
+        { .Tag = "ModifyUser", .PrecreateTarget = true, .SqlStatement = "ALTER USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::DatabaseAdmin, .SubjectPermissions = {"ydb.database.connect"},
+            .EnableStrictUserManagement = true, .EnableDatabaseAdmin = true, .ExpectedResult = true
+        },
+        // Ordinary user can create users only if EnableStrictUserManagement is false
+        // and ydb.granular.alter_schema is granted (besides ydb.database.connect).
+        { .Tag = "ModifyUser", .PrecreateTarget = true, .SqlStatement = "ALTER USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::User, .SubjectPermissions = {"ydb.database.connect", "ydb.granular.alter_schema"},
+            .EnableStrictUserManagement = false, .EnableDatabaseAdmin = false, .ExpectedResult = true
+        },
+        { .Tag = "ModifyUser", .PrecreateTarget = true, .SqlStatement = "ALTER USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::User, .SubjectPermissions = {"ydb.database.connect", "ydb.granular.alter_schema"},
+            .EnableStrictUserManagement = false, .EnableDatabaseAdmin = true, .ExpectedResult = true
+        },
+        { .Tag = "ModifyUser", .PrecreateTarget = true, .SqlStatement = "ALTER USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::User, .SubjectPermissions = {"ydb.database.connect", "ydb.granular.alter_schema"},
+            .EnableStrictUserManagement = true, .EnableDatabaseAdmin = false, .ExpectedResult = false
+        },
+        { .Tag = "ModifyUser", .PrecreateTarget = true, .SqlStatement = "ALTER USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::User, .SubjectPermissions = {"ydb.database.connect", "ydb.granular.alter_schema"},
+            .EnableStrictUserManagement = true, .EnableDatabaseAdmin = true, .ExpectedResult = false
+        },
+        { .Tag = "ModifyUser", .PrecreateTarget = true, .SqlStatement = "ALTER USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::User, .SubjectPermissions = {"ydb.database.connect"},
+            .EnableStrictUserManagement = false, .EnableDatabaseAdmin = false, .ExpectedResult = false
+        },
+        { .Tag = "ModifyUser", .PrecreateTarget = true, .SqlStatement = "ALTER USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::User, .SubjectPermissions = {"ydb.database.connect"},
+            .EnableStrictUserManagement = false, .EnableDatabaseAdmin = true, .ExpectedResult = false
+        },
+        { .Tag = "ModifyUser", .PrecreateTarget = true, .SqlStatement = "ALTER USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::User, .SubjectPermissions = {"ydb.database.connect"},
+            .EnableStrictUserManagement = true, .EnableDatabaseAdmin = false, .ExpectedResult = false
+        },
+        { .Tag = "ModifyUser", .PrecreateTarget = true, .SqlStatement = "ALTER USER targetuser PASSWORD 'passwd'",
+            .SubjectLevel = EAccessLevel::User, .SubjectPermissions = {"ydb.database.connect"},
+            .EnableStrictUserManagement = true, .EnableDatabaseAdmin = true, .ExpectedResult = false
+        },
+
+        // DropUser
+        // Cluster admin can always administer users, but require the same schema permissions as ordinary user (but why?).
+        { .Tag = "DropUser", .PrecreateTarget = true, .SqlStatement = "DROP USER targetuser",
+            .SubjectLevel = EAccessLevel::ClusterAdmin, .SubjectPermissions = {"ydb.database.connect", "ydb.granular.alter_schema"},
+            .EnableStrictUserManagement = false, .EnableDatabaseAdmin = false, .ExpectedResult = true
+        },
+        { .Tag = "DropUser", .PrecreateTarget = true, .SqlStatement = "DROP USER targetuser",
+            .SubjectLevel = EAccessLevel::ClusterAdmin, .SubjectPermissions = {"ydb.database.connect", "ydb.granular.alter_schema"},
+            .EnableStrictUserManagement = false, .EnableDatabaseAdmin = true, .ExpectedResult = true
+        },
+        { .Tag = "DropUser", .PrecreateTarget = true, .SqlStatement = "DROP USER targetuser",
+            .SubjectLevel = EAccessLevel::ClusterAdmin, .SubjectPermissions = {"ydb.database.connect", "ydb.granular.alter_schema"},
+            .EnableStrictUserManagement = true, .EnableDatabaseAdmin = false, .ExpectedResult = true
+        },
+        { .Tag = "DropUser", .PrecreateTarget = true, .SqlStatement = "DROP USER targetuser",
+            .SubjectLevel = EAccessLevel::ClusterAdmin, .SubjectPermissions = {"ydb.database.connect", "ydb.granular.alter_schema"},
+            .EnableStrictUserManagement = true, .EnableDatabaseAdmin = true, .ExpectedResult = true
+        },
+        { .Tag = "DropUser", .PrecreateTarget = true, .SqlStatement = "DROP USER targetuser",
+            .SubjectLevel = EAccessLevel::ClusterAdmin, .SubjectPermissions = {"ydb.database.connect"},
+            .EnableStrictUserManagement = false, .EnableDatabaseAdmin = false, .ExpectedResult = false
+        },
+        { .Tag = "DropUser", .PrecreateTarget = true, .SqlStatement = "DROP USER targetuser",
+            .SubjectLevel = EAccessLevel::ClusterAdmin, .SubjectPermissions = {"ydb.database.connect"},
+            .EnableStrictUserManagement = false, .EnableDatabaseAdmin = true, .ExpectedResult = false
+        },
+        { .Tag = "DropUser", .PrecreateTarget = true, .SqlStatement = "DROP USER targetuser",
+            .SubjectLevel = EAccessLevel::ClusterAdmin, .SubjectPermissions = {"ydb.database.connect"},
+            .EnableStrictUserManagement = true, .EnableDatabaseAdmin = false, .ExpectedResult = true
+        },
+        { .Tag = "DropUser", .PrecreateTarget = true, .SqlStatement = "DROP USER targetuser",
+            .SubjectLevel = EAccessLevel::ClusterAdmin, .SubjectPermissions = {"ydb.database.connect"},
+            .EnableStrictUserManagement = true, .EnableDatabaseAdmin = true, .ExpectedResult = true
+        },
+        // Database admin can administer users if EnableStrictUserManagement and EnableDatabaseAdmin are true.
+        // If not, database admin still can administer users as the owner of the database.
+        // In both cases it require no schema permissions except ydb.database.connect (why?).
+        { .Tag = "DropUser", .PrecreateTarget = true, .SqlStatement = "DROP USER targetuser",
+            .SubjectLevel = EAccessLevel::DatabaseAdmin, .SubjectPermissions = {"ydb.database.connect", "ydb.granular.alter_schema"},
+            .EnableStrictUserManagement = false, .EnableDatabaseAdmin = false, .ExpectedResult = true
+        },
+        { .Tag = "DropUser", .PrecreateTarget = true, .SqlStatement = "DROP USER targetuser",
+            .SubjectLevel = EAccessLevel::DatabaseAdmin, .SubjectPermissions = {"ydb.database.connect", "ydb.granular.alter_schema"},
+            .EnableStrictUserManagement = false, .EnableDatabaseAdmin = true, .ExpectedResult = true
+        },
+        { .Tag = "DropUser", .PrecreateTarget = true, .SqlStatement = "DROP USER targetuser",
+            .SubjectLevel = EAccessLevel::DatabaseAdmin, .SubjectPermissions = {"ydb.database.connect", "ydb.granular.alter_schema"},
+            .EnableStrictUserManagement = true, .EnableDatabaseAdmin = false, .ExpectedResult = false
+        },
+        { .Tag = "DropUser", .PrecreateTarget = true, .SqlStatement = "DROP USER targetuser",
+            .SubjectLevel = EAccessLevel::DatabaseAdmin, .SubjectPermissions = {"ydb.database.connect", "ydb.granular.alter_schema"},
+            .EnableStrictUserManagement = true, .EnableDatabaseAdmin = true, .ExpectedResult = true
+        },
+        { .Tag = "DropUser", .PrecreateTarget = true, .SqlStatement = "DROP USER targetuser",
+            .SubjectLevel = EAccessLevel::DatabaseAdmin, .SubjectPermissions = {"ydb.database.connect"},
+            .EnableStrictUserManagement = false, .EnableDatabaseAdmin = false, .ExpectedResult = true
+        },
+        { .Tag = "DropUser", .PrecreateTarget = true, .SqlStatement = "DROP USER targetuser",
+            .SubjectLevel = EAccessLevel::DatabaseAdmin, .SubjectPermissions = {"ydb.database.connect"},
+            .EnableStrictUserManagement = false, .EnableDatabaseAdmin = true, .ExpectedResult = true
+        },
+        { .Tag = "DropUser", .PrecreateTarget = true, .SqlStatement = "DROP USER targetuser",
+            .SubjectLevel = EAccessLevel::DatabaseAdmin, .SubjectPermissions = {"ydb.database.connect"},
+            .EnableStrictUserManagement = true, .EnableDatabaseAdmin = false, .ExpectedResult = false
+        },
+        { .Tag = "DropUser", .PrecreateTarget = true, .SqlStatement = "DROP USER targetuser",
+            .SubjectLevel = EAccessLevel::DatabaseAdmin, .SubjectPermissions = {"ydb.database.connect"},
+            .EnableStrictUserManagement = true, .EnableDatabaseAdmin = true, .ExpectedResult = true
+        },
+        // Ordinary user can create users only if EnableStrictUserManagement is false
+        // and ydb.granular.alter_schema is granted (besides ydb.database.connect).
+        { .Tag = "DropUser", .PrecreateTarget = true, .SqlStatement = "DROP USER targetuser",
+            .SubjectLevel = EAccessLevel::User, .SubjectPermissions = {"ydb.database.connect", "ydb.granular.alter_schema"},
+            .EnableStrictUserManagement = false, .EnableDatabaseAdmin = false, .ExpectedResult = true
+        },
+        { .Tag = "DropUser", .PrecreateTarget = true, .SqlStatement = "DROP USER targetuser",
+            .SubjectLevel = EAccessLevel::User, .SubjectPermissions = {"ydb.database.connect", "ydb.granular.alter_schema"},
+            .EnableStrictUserManagement = false, .EnableDatabaseAdmin = true, .ExpectedResult = true
+        },
+        { .Tag = "DropUser", .PrecreateTarget = true, .SqlStatement = "DROP USER targetuser",
+            .SubjectLevel = EAccessLevel::User, .SubjectPermissions = {"ydb.database.connect", "ydb.granular.alter_schema"},
+            .EnableStrictUserManagement = true, .EnableDatabaseAdmin = false, .ExpectedResult = false
+        },
+        { .Tag = "DropUser", .PrecreateTarget = true, .SqlStatement = "DROP USER targetuser",
+            .SubjectLevel = EAccessLevel::User, .SubjectPermissions = {"ydb.database.connect", "ydb.granular.alter_schema"},
+            .EnableStrictUserManagement = true, .EnableDatabaseAdmin = true, .ExpectedResult = false
+        },
+        { .Tag = "DropUser", .PrecreateTarget = true, .SqlStatement = "DROP USER targetuser",
+            .SubjectLevel = EAccessLevel::User, .SubjectPermissions = {"ydb.database.connect"},
+            .EnableStrictUserManagement = false, .EnableDatabaseAdmin = false, .ExpectedResult = false
+        },
+        { .Tag = "DropUser", .PrecreateTarget = true, .SqlStatement = "DROP USER targetuser",
+            .SubjectLevel = EAccessLevel::User, .SubjectPermissions = {"ydb.database.connect"},
+            .EnableStrictUserManagement = false, .EnableDatabaseAdmin = true, .ExpectedResult = false
+        },
+        { .Tag = "DropUser", .PrecreateTarget = true, .SqlStatement = "DROP USER targetuser",
+            .SubjectLevel = EAccessLevel::User, .SubjectPermissions = {"ydb.database.connect"},
+            .EnableStrictUserManagement = true, .EnableDatabaseAdmin = false, .ExpectedResult = false
+        },
+        { .Tag = "DropUser", .PrecreateTarget = true, .SqlStatement = "DROP USER targetuser",
+            .SubjectLevel = EAccessLevel::User, .SubjectPermissions = {"ydb.database.connect"},
+            .EnableStrictUserManagement = true, .EnableDatabaseAdmin = true, .ExpectedResult = false
+        },
+    };
+    struct TTestRegistration_AlterLoginProtect_RootDB {
+        TTestRegistration_AlterLoginProtect_RootDB() {
+            static std::vector<TString> TestNames;
+
+            for (size_t testId = 0; const auto& entry : AlterLoginProtect_Tests) {
+                TestNames.emplace_back(TStringBuilder() << "AlterLoginProtect-RootDB-NoAuth-BuiltinUser-" << entry.Tag << "-" << ++testId);
+                TCurrentTest::AddTest(TestNames.back().c_str(), std::bind(AlterLoginProtect_RootDB, std::placeholders::_1, entry), /*forceFork*/ false);
+            }
+
+            static auto testsWithAuth = AlterLoginProtect_Tests;
+            for (auto& entry : testsWithAuth) {
+                entry.EnforceUserTokenRequirement = true;
+            }
+            for (size_t testId = 0; const auto& entry : testsWithAuth) {
+                TestNames.emplace_back(TStringBuilder() << "AlterLoginProtect-RootDB-Auth-BuiltinUser-" << entry.Tag << "-" << ++testId);
+                TCurrentTest::AddTest(TestNames.back().c_str(), std::bind(AlterLoginProtect_RootDB, std::placeholders::_1, entry), /*forceFork*/ false);
+            }
+
+            static auto testsWithLocalSubject = AlterLoginProtect_Tests;
+            for (auto& entry : testsWithLocalSubject) {
+                entry.LocalSid = true;
+            }
+            for (size_t testId = 0; const auto& entry : testsWithLocalSubject) {
+                TestNames.emplace_back(TStringBuilder() << "AlterLoginProtect-RootDB-NoAuth-LocalUser-" << entry.Tag << "-" << ++testId);
+                TCurrentTest::AddTest(TestNames.back().c_str(), std::bind(AlterLoginProtect_RootDB, std::placeholders::_1, entry), /*forceFork*/ false);
+            }
+
+            static auto testsWithAuthAndLocalSubject = AlterLoginProtect_Tests;
+            for (auto& entry : testsWithAuthAndLocalSubject) {
+                entry.EnforceUserTokenRequirement = true;
+                entry.LocalSid = true;
+            }
+            for (size_t testId = 0; const auto& entry : testsWithAuthAndLocalSubject) {
+                TestNames.emplace_back(TStringBuilder() << "AlterLoginProtect-RootDB-Auth-LocalUser-" << entry.Tag << "-" << ++testId);
+                TCurrentTest::AddTest(TestNames.back().c_str(), std::bind(AlterLoginProtect_RootDB, std::placeholders::_1, entry), /*forceFork*/ false);
+            }
+        }
+    };
+    static TTestRegistration_AlterLoginProtect_RootDB testRegistration_AlterLoginProtect_RootDB;
+
+}
+
+}  // namespace NKikimr::NTxProxyUT
diff --git a/ydb/core/tx/tx_proxy/ut_schemereq/ya.make b/ydb/core/tx/tx_proxy/ut_schemereq/ya.make
new file mode 100644
index 0000000000..70413ac260
--- /dev/null
+++ b/ydb/core/tx/tx_proxy/ut_schemereq/ya.make
@@ -0,0 +1,28 @@
+UNITTEST_FOR(ydb/core/tx/tx_proxy)
+
+FORK_SUBTESTS()
+
+IF (WITH_VALGRIND)
+    SIZE(LARGE)
+    TAG(ya:fat)
+ELSE()
+    SIZE(MEDIUM)
+ENDIF()
+
+PEERDIR(
+    library/cpp/getopt
+    library/cpp/svnversion
+    library/cpp/testing/unittest
+    ydb/core/testlib/default
+    ydb/core/tx
+    yql/essentials/public/udf/service/exception_policy
+)
+
+YQL_LAST_ABI_VERSION()
+
+SRCS(
+    schemereq_ut.cpp
+)
+
+
+END()
diff --git a/ydb/core/tx/tx_proxy/ya.make b/ydb/core/tx/tx_proxy/ya.make
index 907eb06dc1..1aaf29809a 100644
--- a/ydb/core/tx/tx_proxy/ya.make
+++ b/ydb/core/tx/tx_proxy/ya.make
@@ -63,5 +63,6 @@ RECURSE_FOR_TESTS(
     ut_base_tenant
     ut_encrypted_storage
     ut_ext_tenant
+    ut_schemereq
     ut_storage_tenant
 )
diff --git a/ydb/library/table_creator/table_creator.cpp b/ydb/library/table_creator/table_creator.cpp
index 06c01c858d..dfc69c4342 100644
--- a/ydb/library/table_creator/table_creator.cpp
+++ b/ydb/library/table_creator/table_creator.cpp
@@ -89,6 +89,7 @@ public:
 
     void RunTableRequest() {
         auto request = MakeHolder<TEvTxUserProxy::TEvProposeTransaction>();
+        request->Record.SetDatabaseName(Database);
         NKikimrSchemeOp::TModifyScheme& modifyScheme = *request->Record.MutableTransaction()->MutableModifyScheme();
         auto pathComponents = SplitPath(Database);
         for (size_t i = 0; i < PathComponents.size() - 1; ++i) {
@@ -207,7 +208,7 @@ public:
                 // In the process of creating a database, errors of the form may occur -
                 // database doesn't have storage pools at all to create tablet
                 // channels to storage pool binding by profile id
-                // Also, this status is returned when column types mismatch - 
+                // Also, this status is returned when column types mismatch -
                 // need to fallback to rebuild column diff
                 } else if (ssStatus == NKikimrScheme::EStatus::StatusInvalidParameter) {
                     FallBack(true /* long delay */);