sysview: add access for database admins (#15098)

Give database admins the same unlimited rights to view system views about users, groups, and their permissions as cluster admins have.

For the ordinary users:
- `.sys/auth_groups` and `.sys/auth_group_members` are closed
- `.sys/auth_users` is filtered to show only the user himself

Cluster admins and now database admins do not have those restrictions.

5 files changed, 207 insertions, 12 deletions

diff --git a/ydb/core/sys_view/auth/auth_scan_base.h b/ydb/core/sys_view/auth/auth_scan_base.h
index b6ce0a4a98..f315253020 100644
--- a/ydb/core/sys_view/auth/auth_scan_base.h
+++ b/ydb/core/sys_view/auth/auth_scan_base.h
@@ -84,7 +84,21 @@ protected:
     void ProceedToScan() override {
         TBase::Become(&TAuthScanBase::StateScan);
 
-        if (RequireUserAdministratorAccess && !IsAdministrator(AppData(), UserToken.Get())) {
+        //NOTE: here is the earliest point when Base::DatabaseOwner is already set
+        bool isClusterAdmin = IsAdministrator(AppData(), UserToken.Get());
+        bool isDatabaseAdmin = (AppData()->FeatureFlags.GetEnableDatabaseAdmin() && IsDatabaseAdministrator(UserToken.Get(), TBase::DatabaseOwner));
+        bool isAdmin = isClusterAdmin || isDatabaseAdmin;
+
+        LOG_DEBUG_S(TlsActivationContext->AsActorContext(), NKikimrServices::SYSTEM_VIEWS,
+            "ProceedToScan,"
+            << " tenant name: " << TBase::TenantName
+            << " tenant owner: " << TBase::DatabaseOwner
+            << " subject sid: " << (UserToken ? UserToken->GetUserSID() : "empty")
+            << " require admin access: " << RequireUserAdministratorAccess
+            << " is admin: " << isAdmin
+        );
+
+        if (RequireUserAdministratorAccess && !isAdmin) {
             TBase::ReplyErrorAndDie(Ydb::StatusIds::UNAUTHORIZED, TStringBuilder() << "Administrator access is required");
             return;
         }
@@ -128,7 +142,7 @@ protected:
                     last.Entry.Path.pop_back();
                     continue;
                 }
-                
+
                 NavigatePath(last.Entry.Path);
                 last.Entry.Path.pop_back();
                 return;
@@ -145,21 +159,21 @@ protected:
 
         Y_ABORT_UNLESS(request->ResultSet.size() == 1);
         auto& entry = request->ResultSet.back();
-        
+
         if (entry.Status != TNavigate::EStatus::Ok) {
-            TBase::ReplyErrorAndDie(Ydb::StatusIds::INTERNAL_ERROR, TStringBuilder() << 
+            TBase::ReplyErrorAndDie(Ydb::StatusIds::INTERNAL_ERROR, TStringBuilder() <<
                 "Failed to navigate " << CanonizePath(entry.Path) << ": " << entry.Status);
             return;
         }
 
         LOG_TRACE_S(ctx, NKikimrServices::SYSTEM_VIEWS,
             "Got navigate: " << request->ToString(*AppData()->TypeRegistry));
-        
+
         auto batch = MakeHolder<NKqp::TEvKqpCompute::TEvScanData>(TBase::ScanId);
 
         FillBatch(*batch, entry);
 
-        if (!RequireUserAdministratorAccess 
+        if (!RequireUserAdministratorAccess
                 && UserToken && !UserToken->GetSerializedToken().empty()
                 && entry.SecurityObject && !entry.SecurityObject->CheckAccess(NACLib::DescribeSchema, *UserToken)) {
             batch->Rows.clear();
@@ -196,7 +210,7 @@ protected:
 
         TBase::Send(MakeSchemeCacheID(), new TEvTxProxySchemeCache::TEvNavigateKeySet(request.Release()));
     }
-    
+
     // this method only skip foolproof useless paths
     // ignores from/to inclusive flags for simplicity
     // ignores some boundary cases for simplicity
diff --git a/ydb/core/sys_view/auth/users.cpp b/ydb/core/sys_view/auth/users.cpp
index 5aa8c3b337..11f31259ba 100644
--- a/ydb/core/sys_view/auth/users.cpp
+++ b/ydb/core/sys_view/auth/users.cpp
@@ -44,6 +44,12 @@ public:
 protected:
     void ProceedToScan() override {
         TBase::Become(&TUsersScan::StateScan);
+
+        //NOTE: here is the earliest point when Base::DatabaseOwner is already set
+        bool isClusterAdmin = IsAdministrator(AppData(), UserToken.Get());
+        bool isDatabaseAdmin = (AppData()->FeatureFlags.GetEnableDatabaseAdmin() && IsDatabaseAdministrator(UserToken.Get(), TBase::DatabaseOwner));
+        IsAdmin = isClusterAdmin || isDatabaseAdmin;
+
         if (TBase::AckReceived) {
             StartScan();
         }
@@ -97,9 +103,9 @@ protected:
         SortBatch(users, [](const auto* left, const auto* right) {
             return left->GetName() < right->GetName();
         });
-        
+
         TVector<TCell> cells(::Reserve(Columns.size()));
-        
+
         for (const auto* user : users) {
             for (auto& column : Columns) {
                 switch (column.Tag) {
@@ -156,7 +162,7 @@ protected:
 
 private:
     bool CanAccessUser(const TString& user) {
-        if (IsAdministrator(AppData(), UserToken.Get())) {
+        if (IsAdmin) {
             return true;
         }
 
@@ -165,6 +171,7 @@ private:
 
 private:
     const TIntrusiveConstPtr<NACLib::TUserToken> UserToken;
+    bool IsAdmin = false;
 };
 
 THolder<NActors::IActor> CreateUsersScan(const NActors::TActorId& ownerId, ui32 scanId, const TTableId& tableId,
diff --git a/ydb/core/sys_view/common/scan_actor_base_impl.h b/ydb/core/sys_view/common/scan_actor_base_impl.h
index c3a72b3ac8..c4da6c3e23 100644
--- a/ydb/core/sys_view/common/scan_actor_base_impl.h
+++ b/ydb/core/sys_view/common/scan_actor_base_impl.h
@@ -297,6 +297,8 @@ private:
         DomainKey = entry.DomainInfo->DomainKey;
 
         TenantName = CanonizePath(entry.Path);
+        DatabaseOwner = entry.Self->Info.GetOwner();
+        Y_ABORT_UNLESS(entry.Self->Info.GetOwner() == entry.SecurityObject->GetOwnerSID());
 
         TBase::Register(CreateTenantNodeEnumerationLookup(TBase::SelfId(), TenantName));
         TBase::Become(&TDerived::StateLookup);
@@ -313,9 +315,10 @@ private:
             "Scan prepared, actor: " << TBase::SelfId()
                 << ", schemeshard id: " << SchemeShardId
                 << ", hive id: " << HiveId
-                << ", tenant name: " << TenantName
+                << ", database: " << TenantName
+                << ", database owner: " << DatabaseOwner
                 << ", domain key: " << DomainKey
-                << ", tenant node count: " << TenantNodes.size());
+                << ", database node count: " << TenantNodes.size());
 
         ProceedToScan();
     }
@@ -370,6 +373,7 @@ protected:
     ui64 SchemeShardId = 0;
     TPathId DomainKey;
     TString TenantName;
+    NACLib::TSID DatabaseOwner;
     THashSet<ui32> TenantNodes;
     ui64 HiveId = 0;
     ui64 SysViewProcessorId = 0;
diff --git a/ydb/tests/functional/tenants/test_auth_system_views.py b/ydb/tests/functional/tenants/test_auth_system_views.py
new file mode 100644
index 0000000000..d231663e03
--- /dev/null
+++ b/ydb/tests/functional/tenants/test_auth_system_views.py
@@ -0,0 +1,169 @@
+# -*- coding: utf-8 -*-
+import logging
+import copy
+
+import pytest
+
+from ydb.tests.oss.ydb_sdk_import import ydb
+from ydb.tests.library.harness.util import LogLevels
+from ydb.tests.library.harness.kikimr_config import KikimrConfigGenerator
+
+
+logger = logging.getLogger(__name__)
+
+
+# local configuration for the ydb cluster (fetched by ydb_cluster_configuration fixture)
+CLUSTER_CONFIG = dict(
+    additional_log_configs={
+        # more logs
+        'TX_PROXY': LogLevels.DEBUG,
+        'GRPC_SERVER': LogLevels.INFO,
+        'GRPC_PROXY_NO_CONNECT_ACCESS': LogLevels.DEBUG,
+        'FLAT_TX_SCHEMESHARD': LogLevels.INFO,
+        'SYSTEM_VIEWS': LogLevels.DEBUG,
+        'TICKET_PARSER': LogLevels.DEBUG,
+        # less logs
+        'KQP_PROXY': LogLevels.CRIT,
+        'KQP_WORKER': LogLevels.CRIT,
+        'KQP_GATEWAY': LogLevels.CRIT,
+        'GRPC_PROXY': LogLevels.ERROR,
+        'TX_DATASHARD': LogLevels.ERROR,
+        'TX_PROXY_SCHEME_CACHE': LogLevels.ERROR,
+        'KQP_YQL': LogLevels.ERROR,
+        'KQP_SESSION': LogLevels.CRIT,
+        'KQP_COMPILE_ACTOR': LogLevels.CRIT,
+        'PERSQUEUE_CLUSTER_TRACKER': LogLevels.CRIT,
+        'SCHEME_BOARD_SUBSCRIBER': LogLevels.CRIT,
+        'SCHEME_BOARD_POPULATOR': LogLevels.CRIT,
+    },
+    extra_feature_flags=[
+        'enable_strict_acl_check',
+        'enable_strict_user_management',
+        'enable_database_admin',
+    ],
+    # do not clutter logs with resource pools auto creation
+    enable_resource_pools=False,
+    enforce_user_token_requirement=True,
+    # make all bootstrap and setup under this user
+    default_clusteradmin='root@builtin',
+)
+
+
+# ydb_fixtures.ydb_cluster_configuration local override
+@pytest.fixture(scope='module')
+def ydb_cluster_configuration():
+    conf = copy.deepcopy(CLUSTER_CONFIG)
+    return conf
+
+
+@pytest.fixture(scope='module')
+def ydb_configurator(ydb_cluster_configuration):
+    config_generator = KikimrConfigGenerator(**ydb_cluster_configuration)
+    config_generator.yaml_config['auth_config'] = {
+        'domain_login_only': False,
+    }
+    config_generator.yaml_config['domains_config']['disable_builtin_security'] = True
+    security_config = config_generator.yaml_config['domains_config']['security_config']
+    security_config['administration_allowed_sids'].append('clusteradmin')
+    return config_generator
+
+
+def stream_query_result(driver, query):
+    it = driver.table_client.scan_query(query)
+    result = []
+    error = None
+
+    while True:
+        try:
+            response = next(it)
+        except StopIteration:
+            break
+        except ydb.issues.Error as e:
+            error = e
+            break
+
+        for row in response.result_set.rows:
+            result.append(row)
+
+    return result, error
+
+
+@pytest.fixture(scope='function')
+def prepared_test_env(ydb_cluster, ydb_root, ydb_database, ydb_client):
+    cluster_admin = ydb.AuthTokenCredentials(ydb_cluster.config.default_clusteradmin)
+
+    # prepare root database
+    with ydb_client(ydb_root, credentials=cluster_admin) as driver:
+        pool = ydb.SessionPool(driver)
+        with pool.checkout() as session:
+            session.execute_scheme("create user clusteradmin password '1234'")
+            session.execute_scheme("create user clusteruser password '1234'")
+
+    # prepare tenant database
+    database_path = ydb_database
+    with ydb_client(database_path, credentials=cluster_admin) as driver:
+        pool = ydb.SessionPool(driver)
+        with pool.checkout() as session:
+            # add users
+            session.execute_scheme("create user dbadmin password '1234'")
+            session.execute_scheme("create user ordinaryuser password '1234'")
+
+            # add group and its members
+            session.execute_scheme('create group dbadmins')
+            session.execute_scheme('alter group dbadmins add user dbadmin')
+
+            # add required permissions on paths
+            session.execute_scheme(f'grant "ydb.generic.use" on `{database_path}` to clusteradmin')
+            # shouldn't be possible but we allow that for the admin
+            session.execute_scheme(f'grant "ydb.generic.use" on `{database_path}` to clusteruser')
+            session.execute_scheme(f'grant "ydb.generic.use" on `{database_path}` to dbadmin')
+            session.execute_scheme(f'grant "ydb.generic.use" on `{database_path}` to ordinaryuser')
+
+        # make dbadmin the real admin of the database
+        driver.scheme_client.modify_permissions(database_path, ydb.ModifyPermissionsSettings().change_owner('dbadmin'))
+
+    return database_path
+
+
+# python sdk does not legally expose login method, so that is a crude way
+# to get auth-token in exchange to credentials
+def login_user(endpoint, database, user, password):
+    driver_config = ydb.DriverConfig(endpoint, database)
+    credentials = ydb.StaticCredentials(driver_config, user, password)
+    return credentials._make_token_request()['access_token']
+
+
+@pytest.mark.parametrize('user,expected_access', [
+    ('clusteradmin', True),
+    ('clusteruser', False),
+    ('dbadmin', True),
+    ('ordinaryuser', False)
+])
+def test_tenant_auth_groups_access(ydb_endpoint, ydb_root, prepared_test_env, ydb_client, user, expected_access):
+    tenant_database = prepared_test_env
+
+    # user could be either from the root or tenant database,
+    # but they must obtain auth token by logging in the database they live in
+    login_database = ydb_root if user.startswith('cluster') else tenant_database
+    user_auth_token = login_user(ydb_endpoint, login_database, user, '1234')
+    credentials = ydb.AuthTokenCredentials(user_auth_token)
+
+    with ydb_client(tenant_database, credentials=credentials) as driver:
+        driver.wait()
+
+        result, error = stream_query_result(driver, f'SELECT * FROM `{tenant_database}/.sys/auth_groups`')
+
+        # logger.debug('AAA result row count %s, error %s', len(result), error)
+
+        if expected_access:
+            assert error is None
+            assert len(result) == 1
+            assert result[0]['Sid'] == 'dbadmins'
+
+        else:
+            assert error is not None, result
+
+            # FIXME: status should have been UNAUTHORIZED
+            # assert error.status == ydb.issues.StatusCode.UNAUTHORIZED
+            assert error.status == ydb.issues.StatusCode.INTERNAL_ERROR
+            assert 'Administrator access is required' in error.message
diff --git a/ydb/tests/functional/tenants/ya.make b/ydb/tests/functional/tenants/ya.make
index 88c25d9d27..1e12a21e94 100644
--- a/ydb/tests/functional/tenants/ya.make
+++ b/ydb/tests/functional/tenants/ya.make
@@ -11,6 +11,7 @@ TEST_SRCS(
     test_tenants.py
     test_storage_config.py
     test_system_views.py
+    test_auth_system_views.py
     test_publish_into_schemeboard_with_common_ssring.py
     test_users_groups_with_acl.py
 )