added the check of the transfer write permission (#17938)

nshestakov · nshestakov · commit c680e73c5281 · 2025-05-06T05:30:21.000Z
diff --git a/ydb/core/protos/replication.proto b/ydb/core/protos/replication.proto
@@ -96,6 +96,7 @@ message TReplicationConfig {
         }
         optional TTarget Target = 1;
         optional TBatchingSettings Batching = 2;
+        optional string RunAsUser = 3;
     }
 
     optional TConnectionParams SrcConnectionParams = 1;
@@ -258,6 +259,7 @@ message TTransferWriterSettings {
     optional NKikimrProto.TPathID PathId = 1;
     optional string TransformLambda = 2;
     optional TBatchingSettings Batching = 3;
+    optional string RunAsUser = 4;
 }
 
 message TRunWorkerCommand {
diff --git a/ydb/core/tx/replication/controller/event_util.cpp b/ydb/core/tx/replication/controller/event_util.cpp
@@ -60,6 +60,7 @@ THolder<TEvService::TEvRunWorker> MakeRunWorkerEv(
             dstPathId.ToProto(writerSettings.MutablePathId());
             writerSettings.SetTransformLambda(p->GetTransformLambda());
             writerSettings.MutableBatching()->CopyFrom(batchingSettings);
+            writerSettings.SetRunAsUser(p->GetRunAsUser());
             break;
         }
     }
diff --git a/ydb/core/tx/replication/controller/schema.h b/ydb/core/tx/replication/controller/schema.h
@@ -48,9 +48,10 @@ struct TControllerSchema: NIceDb::Schema {
         };
         struct Issue: Column<9, NScheme::NTypeIds::Utf8> {};
         struct TransformLambda: Column<10, NScheme::NTypeIds::Utf8> {};
+        struct RunAsUser: Column<11, NScheme::NTypeIds::Utf8> {};
 
         using TKey = TableKey<ReplicationId, Id>;
-        using TColumns = TableColumns<ReplicationId, Id, Kind, SrcPath, DstPath, DstState, DstPathOwnerId, DstPathLocalId, Issue, TransformLambda>;
+        using TColumns = TableColumns<ReplicationId, Id, Kind, SrcPath, DstPath, DstState, DstPathOwnerId, DstPathLocalId, Issue, TransformLambda, RunAsUser>;
     };
 
     struct SrcStreams: Table<4> {
diff --git a/ydb/core/tx/replication/controller/target_discoverer.cpp b/ydb/core/tx/replication/controller/target_discoverer.cpp
@@ -183,7 +183,8 @@ class TTargetDiscoverer: public TActorBootstrapped<TTargetDiscoverer> {
             const auto& targetConf = Config.GetTransferSpecific().GetTarget();
 
             const auto& target = ToAdd.emplace_back(TReplication::ETargetKind::Transfer,
-                std::make_shared<TTargetTransfer::TTransferConfig>(path.first, path.second, targetConf.GetTransformLambda()));
+                std::make_shared<TTargetTransfer::TTransferConfig>(path.first, path.second, targetConf.GetTransformLambda(),
+                    Config.GetTransferSpecific().GetRunAsUser()));
             LOG_I("Add target"
                 << ": srcPath# " << target.Config->GetSrcPath()
                 << ", dstPath# " << target.Config->GetDstPath()
diff --git a/ydb/core/tx/replication/controller/target_transfer.cpp b/ydb/core/tx/replication/controller/target_transfer.cpp
@@ -17,7 +17,8 @@ void TTargetTransfer::UpdateConfig(const NKikimrReplication::TReplicationConfig&
     Config = std::make_shared<TTargetTransfer::TTransferConfig>(
         GetConfig()->GetSrcPath(),
         GetConfig()->GetDstPath(),
-        t.GetTransformLambda());
+        t.GetTransformLambda(),
+        cfg.GetTransferSpecific().GetRunAsUser());
 }
 
 void TTargetTransfer::Progress(const TActorContext& ctx) {
@@ -55,14 +56,19 @@ TString TTargetTransfer::GetStreamPath() const {
     return CanonizePath(GetSrcPath());
 }
 
-TTargetTransfer::TTransferConfig::TTransferConfig(const TString& srcPath, const TString& dstPath, const TString& transformLambda)
+TTargetTransfer::TTransferConfig::TTransferConfig(const TString& srcPath, const TString& dstPath, const TString& transformLambda, const TString& runAsUser)
     : TConfigBase(ETargetKind::Transfer, srcPath, dstPath)
     , TransformLambda(transformLambda)
+    , RunAsUser(runAsUser)
 {
 }
 
 const TString& TTargetTransfer::TTransferConfig::GetTransformLambda() const {
     return TransformLambda;
 }
 
+const TString& TTargetTransfer::TTransferConfig::GetRunAsUser() const {
+    return RunAsUser;
+}
+
 }
diff --git a/ydb/core/tx/replication/controller/target_transfer.h b/ydb/core/tx/replication/controller/target_transfer.h
@@ -9,12 +9,14 @@ class TTargetTransfer: public TTargetWithStream {
     struct TTransferConfig : public TConfigBase {
         using TPtr = std::shared_ptr<TTransferConfig>;
 
-        TTransferConfig(const TString& srcPath, const TString& dstPath, const TString& transformLambda);
+        TTransferConfig(const TString& srcPath, const TString& dstPath, const TString& transformLambda, const TString& runAsUser);
         
         const TString& GetTransformLambda() const;
+        const TString& GetRunAsUser() const;
         
     private:
         TString TransformLambda;
+        TString RunAsUser;
     };
         
     explicit TTargetTransfer(TReplication* replication,
diff --git a/ydb/core/tx/replication/controller/tx_discovery_targets_result.cpp b/ydb/core/tx/replication/controller/tx_discovery_targets_result.cpp
@@ -46,15 +46,18 @@ class TController::TTxDiscoveryTargetsResult: public TTxBase {
                 const auto tid = Replication->AddTarget(target.Kind, target.Config);
                 
                 TString transformLambda;
+                TString runAsUser;
                 if (auto p = std::dynamic_pointer_cast<const TTargetTransfer::TTransferConfig>(target.Config)) {
                     transformLambda = p->GetTransformLambda();
+                    runAsUser = p->GetRunAsUser();
                 }
 
                 db.Table<Schema::Targets>().Key(rid, tid).Update(
                     NIceDb::TUpdate<Schema::Targets::Kind>(target.Kind),
                     NIceDb::TUpdate<Schema::Targets::SrcPath>(target.Config->GetSrcPath()),
                     NIceDb::TUpdate<Schema::Targets::DstPath>(target.Config->GetDstPath()),
-                    NIceDb::TUpdate<Schema::Targets::TransformLambda>(transformLambda)
+                    NIceDb::TUpdate<Schema::Targets::TransformLambda>(transformLambda),
+                    NIceDb::TUpdate<Schema::Targets::RunAsUser>(runAsUser)
                 );
 
                 CLOG_N(ctx, "Add target"
diff --git a/ydb/core/tx/replication/controller/tx_init.cpp b/ydb/core/tx/replication/controller/tx_init.cpp
@@ -88,6 +88,7 @@ class TController::TTxInit: public TTxBase {
                 rowset.GetValue<Schema::Targets::DstPathLocalId>()
             );
             const auto transformLambda = rowset.GetValue<Schema::Targets::TransformLambda>();
+            const auto runAsUser = rowset.GetValue<Schema::Targets::RunAsUser>();
 
             auto replication = Self->Find(rid);
             Y_VERIFY_S(replication, "Unknown replication: " << rid);
@@ -103,7 +104,7 @@ class TController::TTxInit: public TTxBase {
                     break;
 
                 case TReplication::ETargetKind::Transfer:
-                    config = std::make_shared<TTargetTransfer::TTransferConfig>(srcPath, dstPath, transformLambda);
+                    config = std::make_shared<TTargetTransfer::TTransferConfig>(srcPath, dstPath, transformLambda, runAsUser);
                     break;
             }
 
diff --git a/ydb/core/tx/replication/service/service.cpp b/ydb/core/tx/replication/service/service.cpp
@@ -432,9 +432,10 @@ class TReplicationService: public TActorBootstrapped<TReplicationService> {
             transformLambda = writerSettings.GetTransformLambda(),
             compilationService = *CompilationService,
             batchingSettings = writerSettings.GetBatching(),
-            transferWriterFactory = transferWriterFactory
+            transferWriterFactory = transferWriterFactory,
+            runAsUser = writerSettings.GetRunAsUser()
         ]() {
-            return transferWriterFactory->Create({transformLambda, tablePathId, compilationService, batchingSettings});
+            return transferWriterFactory->Create({transformLambda, tablePathId, compilationService, batchingSettings, runAsUser});
         };
     }
 
diff --git a/ydb/core/tx/replication/service/transfer_writer_factory.h b/ydb/core/tx/replication/service/transfer_writer_factory.h
@@ -21,6 +21,7 @@ class ITransferWriterFactory {
         const TPathId& TablePathId;
         const TActorId& CompileServiceId;
         const NKikimrReplication::TBatchingSettings& BatchingSettings;
+        const TString& RunAsUser;
     };
 
     virtual IActor* Create(const Parameters& parameters) const = 0;
diff --git a/ydb/core/tx/schemeshard/schemeshard__operation_create_replication.cpp b/ydb/core/tx/schemeshard/schemeshard__operation_create_replication.cpp
@@ -18,6 +18,7 @@ namespace {
 struct IStrategy {
     virtual TPathElement::EPathType GetPathType() const = 0;
     virtual bool Validate(TProposeResponse& result, const NKikimrSchemeOp::TReplicationDescription& desc, const TOperationContext& context) const = 0;
+    virtual void Proccess(NKikimrReplication::TReplicationConfig& config, const TString& owner) const = 0;
 };
 
 struct TReplicationStrategy : public IStrategy {
@@ -37,6 +38,9 @@ struct TReplicationStrategy : public IStrategy {
 
         return false;
     }
+
+    void Proccess(NKikimrReplication::TReplicationConfig&, const TString&) const override {
+    }
 };
 
 struct TTransferStrategy : public IStrategy {
@@ -90,6 +94,10 @@ struct TTransferStrategy : public IStrategy {
 
         return false;
     }
+
+    void Proccess(NKikimrReplication::TReplicationConfig& config, const TString& owner) const override {
+        config.MutableTransferSpecific()->SetRunAsUser(owner);
+    }
 };
 
 static constexpr TReplicationStrategy ReplicationStrategy;
@@ -435,6 +443,8 @@ class TCreateReplication: public TSubOperation {
             desc.MutableConfig()->MutableConsistencySettings()->MutableRow();
         }
 
+        Strategy->Proccess(*desc.MutableConfig(), owner);
+
         desc.MutableState()->MutableStandBy();
         auto replication = TReplicationInfo::Create(std::move(desc));
         context.SS->Replications[path->PathId] = replication;
diff --git a/ydb/tests/functional/replication/utils.h b/ydb/tests/functional/replication/utils.h
@@ -163,7 +163,16 @@ struct TConfig {
 
 struct MainTestCase {
 
-    MainTestCase()
+    static auto CreateDriverConfig(std::string connectionString, std::optional<std::string> user) {
+        auto config = TDriverConfig(connectionString);
+        if (user) {
+            config.SetAuthToken(TStringBuilder() << user.value() << "@builtin");
+        }
+        // config.SetLog(std::unique_ptr<TLogBackend>(CreateLogBackend("cerr", ELogPriority::TLOG_INFO).Release()))
+        return config;
+    }
+
+    MainTestCase(std::optional<std::string> user = std::nullopt)
         : Id(RandomNumber<size_t>())
         , ConnectionString(GetEnv("YDB_ENDPOINT") + "/?database=" + GetEnv("YDB_DATABASE"))
         , TopicName(TStringBuilder() << "Topic_" << Id)
@@ -172,7 +181,7 @@ struct MainTestCase {
         , TableName(TStringBuilder() << "Table_" << Id)
         , ReplicationName(TStringBuilder() << "Replication_" << Id)
         , TransferName(TStringBuilder() << "Transfer_" << Id)
-        , Driver(TDriverConfig(ConnectionString)/*.SetLog(std::unique_ptr<TLogBackend>(CreateLogBackend("cerr", ELogPriority::TLOG_INFO).Release()))*/)
+        , Driver(CreateDriverConfig(ConnectionString, user))
         , TableClient(Driver)
         , Session(TableClient.GetSession().GetValueSync().GetSession())
         , TopicClient(Driver)
@@ -224,6 +233,28 @@ struct MainTestCase {
         return ExecuteQuery(Sprintf(query.data(), SourceTableName.data()));
     }
 
+    void Grant(const std::string& object, const std::string& username, const std::vector<std::string>& permissions) {
+        TStringBuilder sql;
+        sql << "GRANT ";
+        for (size_t i = 0; i < permissions.size(); ++i) {
+            if (i) {
+                sql << ", ";
+            }
+            if ("ALL" == permissions[i]) {
+                sql << "ALL";
+            } else {
+                sql << "'" << permissions[i] << "'";
+            }
+        }
+        sql << " ON `/local";
+        if (!object.empty()) {
+            sql << "/" << object;
+        }
+        sql << "` TO `" << username << "@builtin`";
+
+        ExecuteDDL(sql);
+    }
+
     void CreateTable(const std::string& tableDDL) {
         ExecuteDDL(Sprintf(tableDDL.data(), TableName.data()));
     }
@@ -278,11 +309,12 @@ struct MainTestCase {
     }
 
     struct CreateTransferSettings {
-        std::optional<std::string> TopicName = std::nullopt;
-        std::optional<std::string> ConsumerName = std::nullopt;
+        std::optional<std::string> TopicName;
+        std::optional<std::string> ConsumerName;
         std::optional<TDuration> FlushInterval = TDuration::Seconds(1);
         std::optional<ui64> BatchSizeBytes = 8_MB;
-        std::optional<std::string> ExpectedError = std::nullopt;
+        std::optional<std::string> ExpectedError;
+        std::optional<std::string> Username;
 
         CreateTransferSettings() {};
 
@@ -311,7 +343,13 @@ struct MainTestCase {
             result.ExpectedError = expected;
             return result;
         }
-    };
+
+        static CreateTransferSettings WithUsername(const TString& username) {
+            CreateTransferSettings result;
+            result.Username = username;
+            return result;
+        }
+};
 
     void CreateTransfer(const std::string& lambda, const CreateTransferSettings& settings = CreateTransferSettings()) {
         TStringBuilder sb;
@@ -324,6 +362,9 @@ struct MainTestCase {
         if (settings.BatchSizeBytes) {
             sb << ", BATCH_SIZE_BYTES = " << *settings.BatchSizeBytes << Endl;
         }
+        if (settings.Username) {
+            sb << ", TOKEN = '" << *settings.Username << "@builtin'" << Endl;
+        }
 
         TString topicName = settings.TopicName.value_or(TopicName);
 
@@ -587,8 +628,12 @@ struct MainTestCase {
                 return result;
             }
     
-            UNIT_ASSERT_C(i, "Unable to wait transfer state. Expected: " << expected << ", actual: " << result.GetState());
-            Sleep(TDuration::Seconds(1));
+            std::string issues;
+            if (result.GetState() == TReplicationDescription::EState::Error) {
+                issues = result.GetErrorState().GetIssues().ToOneLineString();
+            }
+    
+            UNIT_ASSERT_C(i, "Unable to wait transfer state. Expected: " << expected << ", actual: " << result.GetState() << ", " << issues);            Sleep(TDuration::Seconds(1));
         }
 
         Y_UNREACHABLE();

Original file line number	Diff line number	Diff line change
`@@ -96,6 +96,7 @@ message TReplicationConfig {`
`96`	`96`	`}`
`97`	`97`	`optional TTarget Target = 1;`
`98`	`98`	`optional TBatchingSettings Batching = 2;`
	`99`	`+ optional string RunAsUser = 3;`
`99`	`100`	`}`
`100`	`101`
`101`	`102`	`optional TConnectionParams SrcConnectionParams = 1;`
`@@ -258,6 +259,7 @@ message TTransferWriterSettings {`
`258`	`259`	`optional NKikimrProto.TPathID PathId = 1;`
`259`	`260`	`optional string TransformLambda = 2;`
`260`	`261`	`optional TBatchingSettings Batching = 3;`
	`262`	`+ optional string RunAsUser = 4;`
`261`	`263`	`}`
`262`	`264`
`263`	`265`	`message TRunWorkerCommand {`
Original file line number	Diff line number	Diff line change
`@@ -60,6 +60,7 @@ THolder<TEvService::TEvRunWorker> MakeRunWorkerEv(`
`60`	`60`	`dstPathId.ToProto(writerSettings.MutablePathId());`
`61`	`61`	`writerSettings.SetTransformLambda(p->GetTransformLambda());`
`62`	`62`	`writerSettings.MutableBatching()->CopyFrom(batchingSettings);`
	`63`	`+ writerSettings.SetRunAsUser(p->GetRunAsUser());`
`63`	`64`	`break;`
`64`	`65`	`}`
`65`	`66`	`}`
Original file line number	Diff line number	Diff line change
`@@ -17,7 +17,8 @@ void TTargetTransfer::UpdateConfig(const NKikimrReplication::TReplicationConfig&`
`17`	`17`	`Config = std::make_shared<TTargetTransfer::TTransferConfig>(`
`18`	`18`	`GetConfig()->GetSrcPath(),`
`19`	`19`	`GetConfig()->GetDstPath(),`
`20`		`- t.GetTransformLambda());`
	`20`	`+ t.GetTransformLambda(),`
	`21`	`+ cfg.GetTransferSpecific().GetRunAsUser());`
`21`	`22`	`}`
`22`	`23`
`23`	`24`	`void TTargetTransfer::Progress(const TActorContext& ctx) {`
`@@ -55,14 +56,19 @@ TString TTargetTransfer::GetStreamPath() const {`
`55`	`56`	`return CanonizePath(GetSrcPath());`
`56`	`57`	`}`
`57`	`58`
`58`		`-TTargetTransfer::TTransferConfig::TTransferConfig(const TString& srcPath, const TString& dstPath, const TString& transformLambda)`
	`59`	`+TTargetTransfer::TTransferConfig::TTransferConfig(const TString& srcPath, const TString& dstPath, const TString& transformLambda, const TString& runAsUser)`
`59`	`60`	`: TConfigBase(ETargetKind::Transfer, srcPath, dstPath)`
`60`	`61`	`, TransformLambda(transformLambda)`
	`62`	`+ , RunAsUser(runAsUser)`
`61`	`63`	`{`
`62`	`64`	`}`
`63`	`65`
`64`	`66`	`const TString& TTargetTransfer::TTransferConfig::GetTransformLambda() const {`
`65`	`67`	`return TransformLambda;`
`66`	`68`	`}`
`67`	`69`
	`70`	`+const TString& TTargetTransfer::TTransferConfig::GetRunAsUser() const {`
	`71`	`+ return RunAsUser;`
	`72`	`+}`
	`73`	`+`
`68`	`74`	`}`