Merge fdbc79b into 2764fa1

Author: qyryq

ydb/core/tx/scheme_board/cache.cpp

@@ -190,8 +190,8 @@ namespace {
             return entry.KeyDescription->SecurityObject;
         }
 
-        static ui32 GetAccess(const TNavigate::TEntry&) {
-            return NACLib::EAccessRights::DescribeSchema;
+        static ui32 GetAccess(const TNavigate::TEntry& entry) {
+            return entry.Access;
         }
 
         static ui32 GetAccess(const TResolve::TEntry& entry) {
@@ -296,12 +296,11 @@ namespace {
                         continue;
                     }
 
-                    const ui32 access = GetAccess(entry);
-                    if (!securityObject->CheckAccess(access, *Context->Request->UserToken)) {
+                    if (!securityObject->CheckAccess(entry.Access, *Context->Request->UserToken)) {
                         SBC_LOG_W("Access denied"
                             << ": self# " << this->SelfId()
                             << ", for# " << Context->Request->UserToken->GetUserSID()
-                            << ", access# " << NACLib::AccessRightsToString(access));
+                            << ", access# " << NACLib::AccessRightsToString(entry.Access));
 
                         SetErrorAndClear(
                             Context.Get(),

ydb/core/tx/scheme_cache/scheme_cache.h

@@ -258,6 +258,7 @@ struct TSchemeCacheNavigate {
 
         // in
         TVector<TString> Path;
+        ui32 Access = NACLib::DescribeSchema;
         TTableId TableId;
         ERequestType RequestType = ERequestType::ByPath;
         EOp Operation = OpUnknown;

ydb/public/sdk/cpp/client/ydb_topic/ut/describe_topic_ut.cpp

@@ -1,6 +1,9 @@
+#include "impl/trace_lazy.h"
 #include "ut_utils/topic_sdk_test_setup.h"
 
+#include <format>
 #include <ydb/library/persqueue/topic_parser_public/topic_parser.h>
+#include <ydb/public/api/protos/persqueue_error_codes_v1.pb.h>
 #include <ydb/public/sdk/cpp/client/ydb_topic/topic.h>
 #include <ydb/public/sdk/cpp/client/ydb_persqueue_public/persqueue.h>
 #include <ydb/public/sdk/cpp/client/ydb_topic/impl/common.h>
@@ -10,8 +13,6 @@
 #include <library/cpp/threading/future/future.h>
 #include <library/cpp/threading/future/async.h>
 
-#include <future>
-
 namespace NYdb::NTopic::NTests {
 
     Y_UNIT_TEST_SUITE(Describe) {
@@ -344,5 +345,78 @@ namespace NYdb::NTopic::NTests {
             DescribeConsumer(setup, client, false, false, true, true);
             DescribePartition(setup, client, false, false, true, true);
         }
+
+        TDescribePartitionResult RunPermissionTest(TTopicSdkTestSetup& setup, int userId, bool existingTopic, bool allowUpdateRow, bool allowDescribeSchema) {
+            TString authToken = TStringBuilder() << "x-user-" << userId << "@builtin";
+            Cerr << std::format("=== existingTopic={} allowUpdateRow={} allowDescribeSchema={} authToken={}\n",
+                                existingTopic, allowUpdateRow, allowDescribeSchema, std::string(authToken));
+
+            auto driverConfig = setup.MakeDriverConfig().SetAuthToken(authToken);
+            auto client = TTopicClient(TDriver(driverConfig));
+            auto settings = TDescribePartitionSettings().IncludeLocation(true);
+            i64 testPartitionId = 0;
+
+            NACLib::TDiffACL acl;
+            if (allowDescribeSchema) {
+                acl.AddAccess(NACLib::EAccessType::Allow, NACLib::DescribeSchema, authToken);
+            }
+            if (allowUpdateRow) {
+                acl.AddAccess(NACLib::EAccessType::Allow, NACLib::UpdateRow, authToken);
+            }
+            setup.GetServer().AnnoyingClient->ModifyACL("/Root", TEST_TOPIC, acl.SerializeAsString());
+
+            return client.DescribePartition(existingTopic ? TEST_TOPIC : "bad-topic", testPartitionId, settings).GetValueSync();
+        }
+
+        Y_UNIT_TEST(DescribePartitionPermissions) {
+            TTopicSdkTestSetup setup(TEST_CASE_NAME);
+            setup.GetServer().EnableLogs({NKikimrServices::TX_PROXY_SCHEME_CACHE, NKikimrServices::SCHEME_BOARD_SUBSCRIBER}, NActors::NLog::PRI_TRACE);
+
+            int userId = 0;
+
+            struct Expectation {
+                bool existingTopic;
+                bool allowUpdateRow;
+                bool allowDescribeSchema;
+                EStatus status;
+                NYql::TIssueCode issueCode;
+            };
+
+            std::vector<Expectation> expectations{
+                {0, 0, 0, EStatus::SCHEME_ERROR, Ydb::PersQueue::ErrorCode::ACCESS_DENIED},
+                {0, 0, 1, EStatus::SCHEME_ERROR, Ydb::PersQueue::ErrorCode::ACCESS_DENIED},
+                {0, 1, 0, EStatus::SCHEME_ERROR, Ydb::PersQueue::ErrorCode::ACCESS_DENIED},
+                {0, 1, 1, EStatus::SCHEME_ERROR, Ydb::PersQueue::ErrorCode::ACCESS_DENIED},
+                {1, 0, 0, EStatus::SCHEME_ERROR, Ydb::PersQueue::ErrorCode::ACCESS_DENIED},
+                {1, 0, 1, EStatus::SUCCESS, 0},
+                {1, 1, 0, EStatus::SUCCESS, 0},
+                {1, 1, 1, EStatus::SUCCESS, 0},
+            };
+
+            for (auto [existing, update, describe, status, issue] : expectations) {
+                auto result = RunPermissionTest(setup, userId++, existing, update, describe);
+                auto resultStatus = result.GetStatus();
+                auto line = TStringBuilder() << "status=" << resultStatus;
+                NYql::TIssueCode resultIssue = 0;
+                if (!result.GetIssues().Empty()) {
+                    resultIssue = result.GetIssues().begin()->GetCode();
+                    line << " issueCode=" << resultIssue;
+                }
+                line << " issues=" << result.GetIssues().ToOneLineString() << Endl;
+                Cerr << line;
+
+                if (resultStatus == EStatus::SUCCESS) {
+                    auto& p = result.GetPartitionDescription().GetPartition();
+                    TRACE_LAZY(setup.GetLog(), "PartitionDescription",
+                        TRACE_KV("partition_id", p.GetPartitionId()),
+                        TRACE_KV("is_active", p.GetActive()),
+                        TRACE_IF(p.GetPartitionLocation().Defined(),
+                            TRACE_KV("node_id", p.GetPartitionLocation()->GetNodeId()),
+                            TRACE_KV("generation", p.GetPartitionLocation()->GetGeneration())));
+                }
+                UNIT_ASSERT_EQUAL(resultStatus, status);
+                UNIT_ASSERT_EQUAL(resultIssue, issue);
+            }
+        }
     }
 }

ydb/public/sdk/cpp/client/ydb_topic/ut/local_partition_ut.cpp

@@ -610,6 +610,55 @@ namespace NYdb::NTopic::NTests {
             UNIT_ASSERT(expected.Matches(events));
         }
 
+        Y_UNIT_TEST(DirectWriteWithoutDescribeResourcesPermission) {
+            // It should be possible to use DirectWrite option with UpdateRow permission only.
+            // In this test we don't grant DescribeSchema permission and check that direct write still works.
+
+            auto existingTopic = GetTestParam("existing", "yes") == "yes";
+            auto allowUpdateRow = GetTestParam("update", "allow") == "allow";
+            auto allowDescribe = GetTestParam("describe") == "allow";
+            auto authToken = GetTestParam("token", "x-user-x@builtin");
+
+            auto setup = CreateSetup(TEST_CASE_NAME);
+            setup->GetServer().EnableLogs({NKikimrServices::TX_PROXY_SCHEME_CACHE}, NActors::NLog::PRI_TRACE);
+
+            {
+                // Add UpdateRow permission.
+                NACLib::TDiffACL acl;
+                if (allowUpdateRow) {
+                    acl.AddAccess(NACLib::EAccessType::Allow, NACLib::UpdateRow, authToken);
+                }
+                // DescribePartitionRequest should work without DescribeSchema permission.
+                if (allowDescribe) {
+                    acl.AddAccess(NACLib::EAccessType::Allow, NACLib::DescribeSchema, authToken);
+                }
+                setup->GetServer().AnnoyingClient->ModifyACL("/Root", TEST_TOPIC, acl.SerializeAsString());
+            }
+
+            TMockDiscoveryService discovery;
+            discovery.SetGoodEndpoints(*setup);
+            auto* tracingBackend = new TTracingBackend();
+            auto driverConfig = CreateConfig(*setup, discovery.GetDiscoveryAddr())
+                .SetLog(CreateCompositeLogBackend({new TStreamLogBackend(&Cerr), tracingBackend}))
+                .SetAuthToken(authToken);
+            TDriver driver(driverConfig);
+
+            auto clientSettings = TTopicClientSettings()
+                .Database("/Root");
+
+            TTopicClient client(driver, clientSettings);
+
+            auto sessionSettings = TWriteSessionSettings()
+                .Path(existingTopic ? TEST_TOPIC : "non-existent")
+                .ProducerId(TEST_MESSAGE_GROUP_ID)
+                .MessageGroupId(TEST_MESSAGE_GROUP_ID)
+                .DirectWriteToPartition(true);
+
+            auto writeSession = client.CreateSimpleBlockingWriteSession(sessionSettings);
+            UNIT_ASSERT(writeSession->Write("message"));
+            writeSession->Close();
+        }
+
         Y_UNIT_TEST(WithoutPartitionWithSplit) {
             auto setup = CreateSetupForSplitMerge(TEST_CASE_NAME);
             setup.CreateTopic(TEST_TOPIC, TEST_CONSUMER, 1, 100);

ydb/services/lib/actors/pq_schema_actor.h

@@ -134,6 +134,9 @@ namespace NKikimr::NGRpcProxy::V1 {
             navigateRequest->DatabaseName = CanonizePath(Database);
 
             NSchemeCache::TSchemeCacheNavigate::TEntry entry;
+            if (CheckAccessWithUpdateRowPermission) {
+                entry.Access = NACLib::EAccessRights::UpdateRow;
+            }
             entry.Path = NKikimr::SplitPath(GetTopicPath());
             entry.SyncVersion = true;
             entry.ShowPrivatePath = showPrivate;
@@ -199,6 +202,8 @@ namespace NKikimr::NGRpcProxy::V1 {
                 return static_cast<TDerived*>(this)->HandleCacheNavigateResponse(ev);
             }
             break;
+            case NSchemeCache::TSchemeCacheNavigate::EStatus::AccessDenied:
+                [[fallthrough]];
             case NSchemeCache::TSchemeCacheNavigate::EStatus::PathErrorUnknown: {
                 AddIssue(
                     FillIssue(
@@ -245,7 +250,7 @@ namespace NKikimr::NGRpcProxy::V1 {
             }
         }
 
-       
+
         void StateWork(TAutoPtr<IEventHandle>& ev) {
             switch (ev->GetTypeRewrite()) {
                 hFunc(TEvTxProxySchemeCache::TEvNavigateKeySetResult, Handle);
@@ -256,6 +261,7 @@ namespace NKikimr::NGRpcProxy::V1 {
 
     protected:
         bool IsDead = false;
+        bool CheckAccessWithUpdateRowPermission = false;
         const TString TopicPath;
         const TString Database;
     };

ydb/services/persqueue_v1/actors/schema_actors.cpp

@@ -1311,20 +1311,52 @@ TDescribePartitionActor::TDescribePartitionActor(NKikimr::NGRpcService::IRequest
 
 void TDescribePartitionActor::Bootstrap(const NActors::TActorContext& ctx)
 {
+    LOG_DEBUG_S(ctx, NKikimrServices::PQ_READ_PROXY, "TDescribePartitionActor" << ctx.SelfID.ToString() << ": Bootstrap");
+    CheckAccessWithUpdateRowPermission = true;
     TBase::Bootstrap(ctx);
     SendDescribeProposeRequest(ctx);
     Become(&TDescribePartitionActor::StateWork);
 }
 
 void TDescribePartitionActor::StateWork(TAutoPtr<IEventHandle>& ev) {
     switch (ev->GetTypeRewrite()) {
+        case TEvTxProxySchemeCache::TEvNavigateKeySetResult::EventType:
+            if (MaybeRequestWithDescribeSchema(ev)) {
+                break;
+            }
+            [[fallthrough]];
         default:
             if (!TDescribeTopicActorImpl::StateWork(ev, ActorContext())) {
                 TBase::StateWork(ev);
             };
     }
 }
 
+// Return true if the second request to SchemeCache has been sent,
+// i.e. we had already sent the first request checking the UpdateRow permission,
+// but the result was negative, so we try again, this time using the DescribeSchema permission.
+bool TDescribePartitionActor::MaybeRequestWithDescribeSchema(TAutoPtr<IEventHandle>& ev) {
+    if (!std::exchange(CheckAccessWithUpdateRowPermission, false)) {
+        // We've got a response to our second request.
+        return false;
+    }
+
+    auto evNav = *reinterpret_cast<typename TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr*>(&ev);
+    auto const& entries = evNav->Get()->Request.Get()->ResultSet;
+    Y_ABORT_UNLESS(entries.size() == 1);
+
+    if (entries.front().Status == NSchemeCache::TSchemeCacheNavigate::EStatus::Ok) {
+        // We do have access to the requested entity.
+        // Transfer ownership to the ev pointer, and let the base classes' StateWork methods handle the response.
+        ev = *reinterpret_cast<TAutoPtr<IEventHandle>*>(&evNav);
+        return false;
+    }
+
+    // We do not have the UpdateRow permission. Check if we're allowed to DescribeSchema.
+    SendDescribeProposeRequest(ActorContext());
+    return true;
+}
+
 void TDescribePartitionActor::HandleCacheNavigateResponse(
     TEvTxProxySchemeCache::TEvNavigateKeySetResult::TPtr& ev
 ) {

ydb/services/persqueue_v1/actors/schema_actors.h

@@ -71,7 +71,7 @@ struct TDescribeTopicActorSettings {
     TVector<ui32> Partitions;
     bool RequireStats = false;
     bool RequireLocation = false;
-    
+
     TDescribeTopicActorSettings()
         : Mode(EMode::DescribeTopic)
     {}
@@ -85,7 +85,7 @@ struct TDescribeTopicActorSettings {
         , RequireStats(requireStats)
         , RequireLocation(requireLocation)
     {}
-    
+
     static TDescribeTopicActorSettings DescribeTopic(bool requireStats, bool requireLocation, const TVector<ui32>& partitions = {}) {
         TDescribeTopicActorSettings res{EMode::DescribeTopic, requireStats, requireLocation};
         res.Partitions = partitions;
@@ -104,7 +104,7 @@ struct TDescribeTopicActorSettings {
         res.Partitions = partitions;
         return res;
     }
-    
+
     static TDescribeTopicActorSettings DescribePartitionSettings(ui32 partition, bool stats, bool location) {
         TDescribeTopicActorSettings res{EMode::DescribePartitions, stats, location};
         res.Partitions = {partition};
@@ -261,9 +261,11 @@ using TTabletInfo = TDescribeTopicActorImpl::TTabletInfo;
     void ApplyResponse(TTabletInfo& tabletInfo, NKikimr::TEvPersQueue::TEvStatusResponse::TPtr& ev, const TActorContext& ctx) override;
     void ApplyResponse(TTabletInfo& tabletInfo, NKikimr::TEvPersQueue::TEvReadSessionsInfoResponse::TPtr& ev, const TActorContext& ctx) override;
     bool ApplyResponse(TEvPersQueue::TEvGetPartitionsLocationResponse::TPtr& ev, const TActorContext& ctx) override;
-    
+
     virtual void Reply(const TActorContext& ctx) override;
 
+    bool MaybeRequestWithDescribeSchema(TAutoPtr<IEventHandle>& ev);
+
 private:
     TIntrusiveConstPtr<NSchemeCache::TSchemeCacheNavigate::TPQGroupInfo> PQGroupInfo;
     Ydb::Topic::DescribePartitionResult Result;