From 21c32a1c5e5e131e97422a8b7c414ff70ce934ac Mon Sep 17 00:00:00 2001
From: Ilnaz Nizametdinov <ilnaz@ydb.tech>
Date: Wed, 24 Jul 2024 21:23:10 +0300
Subject: [PATCH] Enable/disable ssl connections, return connection_string in
 API (#7040)

---
 ydb/core/grpc_services/rpc_replication.cpp    | 11 ++++
 ydb/core/kqp/host/kqp_gateway_proxy.cpp       |  1 +
 ydb/core/kqp/ut/scheme/kqp_scheme_ut.cpp      | 55 +++++++++++++++++++
 ydb/core/protos/replication.proto             |  1 +
 .../tx/replication/controller/replication.cpp |  7 ++-
 ydb/core/tx/replication/service/service.cpp   | 22 +++++---
 ydb/core/tx/replication/ut_helpers/test_env.h |  2 +-
 .../tx/replication/ydb_proxy/ydb_proxy.cpp    | 25 +++++----
 ydb/core/tx/replication/ydb_proxy/ydb_proxy.h |  6 +-
 .../api/protos/draft/ydb_replication.proto    |  2 +
 .../sdk/cpp/client/draft/ydb_replication.cpp  |  5 ++
 .../sdk/cpp/client/draft/ydb_replication.h    |  1 +
 12 files changed, 113 insertions(+), 25 deletions(-)

diff --git a/ydb/core/grpc_services/rpc_replication.cpp b/ydb/core/grpc_services/rpc_replication.cpp
index 83c28cc04845..8b75b72a08d0 100644
--- a/ydb/core/grpc_services/rpc_replication.cpp
+++ b/ydb/core/grpc_services/rpc_replication.cpp
@@ -11,6 +11,8 @@
 
 #include <google/protobuf/util/time_util.h>
 
+#include <util/string/builder.h>
+
 namespace NKikimr::NGRpcService {
 
 using namespace Ydb;
@@ -138,9 +140,18 @@ class TDescribeReplicationRPC: public TRpcSchemeRequestActor<TDescribeReplicatio
         return ReplyWithResult(Ydb::StatusIds::SUCCESS, Result, ctx);
     }
 
+    static TString BuildConnectionString(const NKikimrReplication::TConnectionParams& params) {
+        return TStringBuilder()
+            << (params.GetEnableSsl() ? "grpcs://" : "grpc://")
+            << params.GetEndpoint()
+            << "/?database=" << params.GetDatabase();
+    }
+
     static void ConvertConnectionParams(const NKikimrReplication::TConnectionParams& from, Ydb::Replication::ConnectionParams& to) {
         to.set_endpoint(from.GetEndpoint());
         to.set_database(from.GetDatabase());
+        to.set_enable_ssl(from.GetEnableSsl());
+        to.set_connection_string(BuildConnectionString(from));
 
         switch (from.GetCredentialsCase()) {
         case NKikimrReplication::TConnectionParams::kStaticCredentials:
diff --git a/ydb/core/kqp/host/kqp_gateway_proxy.cpp b/ydb/core/kqp/host/kqp_gateway_proxy.cpp
index 3c717ee29c7c..e5a84a6b7a08 100644
--- a/ydb/core/kqp/host/kqp_gateway_proxy.cpp
+++ b/ydb/core/kqp/host/kqp_gateway_proxy.cpp
@@ -1901,6 +1901,7 @@ class TKqpGatewayProxy : public IKikimrGateway {
                 const auto parseResult = NYdb::ParseConnectionString(*connectionString);
                 params.SetEndpoint(parseResult.Endpoint);
                 params.SetDatabase(parseResult.Database);
+                params.SetEnableSsl(parseResult.EnableSsl);
             }
             if (const auto& endpoint = settings.Settings.Endpoint) {
                 params.SetEndpoint(*endpoint);
diff --git a/ydb/core/kqp/ut/scheme/kqp_scheme_ut.cpp b/ydb/core/kqp/ut/scheme/kqp_scheme_ut.cpp
index b2b60631c079..124d7c46f7d1 100644
--- a/ydb/core/kqp/ut/scheme/kqp_scheme_ut.cpp
+++ b/ydb/core/kqp/ut/scheme/kqp_scheme_ut.cpp
@@ -6055,6 +6055,61 @@ Y_UNIT_TEST_SUITE(KqpScheme) {
         }
     }
 
+    void AsyncReplicationConnectionParams(TKikimrRunner& kikimr, const TString& connectionParam, bool ssl = false) {
+        using namespace NReplication;
+
+        auto repl = TReplicationClient(kikimr.GetDriver(), TCommonClientSettings().Database("/Root"));
+        auto db = kikimr.GetTableClient();
+        auto session = db.CreateSession().GetValueSync().GetSession();
+
+        {
+            auto query = R"(
+                --!syntax_v1
+                CREATE TABLE `/Root/table` (Key Uint64, Value String, PRIMARY KEY (Key));
+            )";
+
+            auto result = session.ExecuteSchemeQuery(query).GetValueSync();
+            UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
+        }
+        {
+            auto query = Sprintf(R"(
+                --!syntax_v1
+                CREATE ASYNC REPLICATION `/Root/replication` FOR
+                    `/Root/table` AS `/Root/replica`
+                WITH (
+                    %s, TOKEN = "root@builtin"
+                );
+            )", connectionParam.c_str());
+
+            const auto result = session.ExecuteSchemeQuery(query).GetValueSync();
+            UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
+        }
+        {
+            const auto result = repl.DescribeReplication("/Root/replication").ExtractValueSync();
+            UNIT_ASSERT_VALUES_EQUAL_C(result.GetStatus(), EStatus::SUCCESS, result.GetIssues().ToString());
+
+            const auto& params = result.GetReplicationDescription().GetConnectionParams();
+            UNIT_ASSERT_VALUES_EQUAL(params.GetDiscoveryEndpoint(), kikimr.GetEndpoint());
+            UNIT_ASSERT_VALUES_EQUAL(params.GetDatabase(), "/Root");
+            UNIT_ASSERT_VALUES_EQUAL(params.GetEnableSsl(), ssl);
+        }
+    }
+
+    Y_UNIT_TEST(AsyncReplicationConnectionString) {
+        TKikimrRunner kikimr;
+        AsyncReplicationConnectionParams(kikimr, Sprintf(R"(CONNECTION_STRING = "grpc://%s/?database=/Root")", kikimr.GetEndpoint().c_str()));
+    }
+
+    Y_UNIT_TEST(AsyncReplicationConnectionStringWithSsl) {
+        TKikimrRunner kikimr;
+        AsyncReplicationConnectionParams(kikimr, Sprintf(R"(CONNECTION_STRING = "grpcs://%s/?database=/Root")", kikimr.GetEndpoint().c_str()), true);
+    }
+
+    Y_UNIT_TEST(AsyncReplicationEndpointAndDatabase) {
+        TKikimrRunner kikimr;
+        AsyncReplicationConnectionParams(kikimr, Sprintf(R"(ENDPOINT = "%s", DATABASE = "/Root")", kikimr.GetEndpoint().c_str()));
+    }
+
     Y_UNIT_TEST(DisableResourcePools) {
         TKikimrRunner kikimr(TKikimrSettings().SetEnableResourcePools(false));
         auto db = kikimr.GetTableClient();
diff --git a/ydb/core/protos/replication.proto b/ydb/core/protos/replication.proto
index 31b2e2eaba28..63571ce6fa55 100644
--- a/ydb/core/protos/replication.proto
+++ b/ydb/core/protos/replication.proto
@@ -19,6 +19,7 @@ message TOAuthToken {
 message TConnectionParams {
     optional string Endpoint = 1;
     optional string Database = 2;
+    optional bool EnableSsl = 5;
     // credentials
     oneof Credentials {
         TStaticCredentials StaticCredentials = 3;
diff --git a/ydb/core/tx/replication/controller/replication.cpp b/ydb/core/tx/replication/controller/replication.cpp
index 983f89fa4eb2..9bd3a8606e1b 100644
--- a/ydb/core/tx/replication/controller/replication.cpp
+++ b/ydb/core/tx/replication/controller/replication.cpp
@@ -113,19 +113,22 @@ class TReplication::TImpl: public TLagProvider {
         if (!YdbProxy && !(State == EState::Removing && !Targets)) {
             THolder<IActor> ydbProxy;
             const auto& params = Config.GetSrcConnectionParams();
+            const auto& endpoint = params.GetEndpoint();
+            const auto& database = params.GetDatabase();
+            const bool ssl = params.GetEnableSsl();
 
             switch (params.GetCredentialsCase()) {
             case NKikimrReplication::TConnectionParams::kStaticCredentials:
                 if (!params.GetStaticCredentials().HasPassword()) {
                     return ResolveSecret(params.GetStaticCredentials().GetPasswordSecretName(), ctx);
                 }
-                ydbProxy.Reset(CreateYdbProxy(params.GetEndpoint(), params.GetDatabase(), params.GetStaticCredentials()));
+                ydbProxy.Reset(CreateYdbProxy(endpoint, database, ssl, params.GetStaticCredentials()));
                 break;
             case NKikimrReplication::TConnectionParams::kOAuthToken:
                 if (!params.GetOAuthToken().HasToken()) {
                     return ResolveSecret(params.GetOAuthToken().GetTokenSecretName(), ctx);
                 }
-                ydbProxy.Reset(CreateYdbProxy(params.GetEndpoint(), params.GetDatabase(), params.GetOAuthToken().GetToken()));
+                ydbProxy.Reset(CreateYdbProxy(endpoint, database, ssl, params.GetOAuthToken().GetToken()));
                 break;
             default:
                 ErrorState(TStringBuilder() << "Unexpected credentials: " << params.GetCredentialsCase());
diff --git a/ydb/core/tx/replication/service/service.cpp b/ydb/core/tx/replication/service/service.cpp
index a8ea5d64ef5f..0e399990b268 100644
--- a/ydb/core/tx/replication/service/service.cpp
+++ b/ydb/core/tx/replication/service/service.cpp
@@ -125,9 +125,9 @@ class TSessionInfo {
 
 }; // TSessionInfo
 
-struct TCredentialsKey: std::tuple<TString, TString, TString> {
-    explicit TCredentialsKey(const TString& endpoint, const TString& database, const TString& user)
-        : std::tuple<TString, TString, TString>(endpoint, database, user)
+struct TCredentialsKey: std::tuple<TString, TString, bool, TString> {
+    explicit TCredentialsKey(const TString& endpoint, const TString& database, bool ssl, const TString& user)
+        : std::tuple<TString, TString, bool, TString>(endpoint, database, ssl, user)
     {
     }
 
@@ -139,12 +139,20 @@ struct TCredentialsKey: std::tuple<TString, TString, TString> {
         return std::get<1>(*this);
     }
 
+    bool EnableSsl() const {
+        return std::get<2>(*this);
+    }
+
     static TCredentialsKey FromParams(const NKikimrReplication::TConnectionParams& params) {
+        const auto& endpoint = params.GetEndpoint();
+        const auto& database = params.GetDatabase();
+        const bool ssl = params.GetEnableSsl();
+
         switch (params.GetCredentialsCase()) {
         case NKikimrReplication::TConnectionParams::kStaticCredentials:
-            return TCredentialsKey(params.GetEndpoint(), params.GetDatabase(), params.GetStaticCredentials().GetUser());
+            return TCredentialsKey(endpoint, database, ssl, params.GetStaticCredentials().GetUser());
         case NKikimrReplication::TConnectionParams::kOAuthToken:
-            return TCredentialsKey(params.GetEndpoint(), params.GetDatabase(), params.GetOAuthToken().GetToken() /* TODO */);
+            return TCredentialsKey(endpoint, database, ssl, params.GetOAuthToken().GetToken());
         default:
             Y_ABORT("Unexpected credentials");
         }
@@ -155,7 +163,7 @@ struct TCredentialsKey: std::tuple<TString, TString, TString> {
 } // NKikimr::NReplication::NService
 
 template <>
-struct THash<NKikimr::NReplication::NService::TCredentialsKey> : THash<std::tuple<TString, TString, TString>> {};
+struct THash<NKikimr::NReplication::NService::TCredentialsKey> : THash<std::tuple<TString, TString, bool, TString>> {};
 
 namespace NKikimr::NReplication {
 
@@ -212,7 +220,7 @@ class TReplicationService: public TActorBootstrapped<TReplicationService> {
     const TActorId& GetOrCreateYdbProxy(TCredentialsKey&& key, Args&&... args) {
         auto it = YdbProxies.find(key);
         if (it == YdbProxies.end()) {
-            auto ydbProxy = Register(CreateYdbProxy(key.Endpoint(), key.Database(), std::forward<Args>(args)...));
+            auto ydbProxy = Register(CreateYdbProxy(key.Endpoint(), key.Database(), key.EnableSsl(), std::forward<Args>(args)...));
             auto res = YdbProxies.emplace(std::move(key), std::move(ydbProxy));
             Y_ABORT_UNLESS(res.second);
             it = res.first;
diff --git a/ydb/core/tx/replication/ut_helpers/test_env.h b/ydb/core/tx/replication/ut_helpers/test_env.h
index f0aec854575f..6920d0f2e921 100644
--- a/ydb/core/tx/replication/ut_helpers/test_env.h
+++ b/ydb/core/tx/replication/ut_helpers/test_env.h
@@ -32,7 +32,7 @@ class TEnv {
         Database = "/" + ToString(DomainName);
 
         YdbProxy = Server.GetRuntime()->Register(CreateYdbProxy(
-            Endpoint, UseDatabase ? Database : "", std::forward<Args>(args)...));
+            Endpoint, UseDatabase ? Database : "", false /* ssl */, std::forward<Args>(args)...));
         Sender = Server.GetRuntime()->AllocateEdgeActor();
     }
 
diff --git a/ydb/core/tx/replication/ydb_proxy/ydb_proxy.cpp b/ydb/core/tx/replication/ydb_proxy/ydb_proxy.cpp
index ccd65e9cc62e..a92aa13aec37 100644
--- a/ydb/core/tx/replication/ydb_proxy/ydb_proxy.cpp
+++ b/ydb/core/tx/replication/ydb_proxy/ydb_proxy.cpp
@@ -419,20 +419,21 @@ class TYdbProxy: public TBaseProxyActor<TYdbProxy> {
         Call<TEvYdbProxy::TEvCommitOffsetResponse>(ev, &TTopicClient::CommitOffset);
     }
 
-    static TCommonClientSettings MakeSettings(const TString& endpoint, const TString& database) {
+    static TCommonClientSettings MakeSettings(const TString& endpoint, const TString& database, bool ssl) {
         return TCommonClientSettings()
             .DiscoveryEndpoint(endpoint)
             .DiscoveryMode(EDiscoveryMode::Async)
-            .Database(database);
+            .Database(database)
+            .SslCredentials(ssl);
     }
 
-    static TCommonClientSettings MakeSettings(const TString& endpoint, const TString& database, const TString& token) {
-        return MakeSettings(endpoint, database)
+    static TCommonClientSettings MakeSettings(const TString& endpoint, const TString& database, bool ssl, const TString& token) {
+        return MakeSettings(endpoint, database, ssl)
             .AuthToken(token);
     }
 
-    static TCommonClientSettings MakeSettings(const TString& endpoint, const TString& database, const TStaticCredentials& credentials) {
-        return MakeSettings(endpoint, database)
+    static TCommonClientSettings MakeSettings(const TString& endpoint, const TString& database, bool ssl, const TStaticCredentials& credentials) {
+        return MakeSettings(endpoint, database, ssl)
             .CredentialsProviderFactory(CreateLoginCredentialsProviderFactory({
                 .User = credentials.GetUser(),
                 .Password = credentials.GetPassword(),
@@ -485,16 +486,16 @@ class TYdbProxy: public TBaseProxyActor<TYdbProxy> {
 
 }; // TYdbProxy
 
-IActor* CreateYdbProxy(const TString& endpoint, const TString& database) {
-    return new TYdbProxy(endpoint, database);
+IActor* CreateYdbProxy(const TString& endpoint, const TString& database, bool ssl) {
+    return new TYdbProxy(endpoint, database, ssl);
 }
 
-IActor* CreateYdbProxy(const TString& endpoint, const TString& database, const TString& token) {
-    return new TYdbProxy(endpoint, database, token);
+IActor* CreateYdbProxy(const TString& endpoint, const TString& database, bool ssl, const TString& token) {
+    return new TYdbProxy(endpoint, database, ssl, token);
 }
 
-IActor* CreateYdbProxy(const TString& endpoint, const TString& database, const TStaticCredentials& credentials) {
-    return new TYdbProxy(endpoint, database, credentials);
+IActor* CreateYdbProxy(const TString& endpoint, const TString& database, bool ssl, const TStaticCredentials& credentials) {
+    return new TYdbProxy(endpoint, database, ssl, credentials);
 }
 
 }
diff --git a/ydb/core/tx/replication/ydb_proxy/ydb_proxy.h b/ydb/core/tx/replication/ydb_proxy/ydb_proxy.h
index 8eba5e9a00a2..9b9c13244679 100644
--- a/ydb/core/tx/replication/ydb_proxy/ydb_proxy.h
+++ b/ydb/core/tx/replication/ydb_proxy/ydb_proxy.h
@@ -255,9 +255,9 @@ struct TEvYdbProxy {
 
 #pragma pop_macro("RemoveDirectory")
 
-IActor* CreateYdbProxy(const TString& endpoint, const TString& database);
-IActor* CreateYdbProxy(const TString& endpoint, const TString& database, const TString& token);
-IActor* CreateYdbProxy(const TString& endpoint, const TString& database,
+IActor* CreateYdbProxy(const TString& endpoint, const TString& database, bool ssl);
+IActor* CreateYdbProxy(const TString& endpoint, const TString& database, bool ssl, const TString& token);
+IActor* CreateYdbProxy(const TString& endpoint, const TString& database, bool ssl,
     const NKikimrReplication::TStaticCredentials& credentials);
 
 }
diff --git a/ydb/public/api/protos/draft/ydb_replication.proto b/ydb/public/api/protos/draft/ydb_replication.proto
index c1421a05e2ab..8695f025bc25 100644
--- a/ydb/public/api/protos/draft/ydb_replication.proto
+++ b/ydb/public/api/protos/draft/ydb_replication.proto
@@ -36,6 +36,8 @@ message ConnectionParams {
 
     string endpoint = 1;
     string database = 2;
+    bool enable_ssl = 5;
+    string connection_string = 6;
 
     oneof credentials {
         StaticCredentials static_credentials = 3;
diff --git a/ydb/public/sdk/cpp/client/draft/ydb_replication.cpp b/ydb/public/sdk/cpp/client/draft/ydb_replication.cpp
index 181a3f2d4171..bc0fe1f80fd9 100644
--- a/ydb/public/sdk/cpp/client/draft/ydb_replication.cpp
+++ b/ydb/public/sdk/cpp/client/draft/ydb_replication.cpp
@@ -19,6 +19,7 @@ namespace NReplication {
 TConnectionParams::TConnectionParams(const Ydb::Replication::ConnectionParams& params) {
     DiscoveryEndpoint(params.endpoint());
     Database(params.database());
+    SslCredentials(params.enable_ssl());
 
     switch (params.credentials_case()) {
     case Ydb::Replication::ConnectionParams::kStaticCredentials:
@@ -47,6 +48,10 @@ const TString& TConnectionParams::GetDatabase() const {
     return *Database_;
 }
 
+bool TConnectionParams::GetEnableSsl() const {
+    return SslCredentials_->IsEnabled;
+}
+
 TConnectionParams::ECredentials TConnectionParams::GetCredentials() const {
     return static_cast<ECredentials>(Credentials_.index());
 }
diff --git a/ydb/public/sdk/cpp/client/draft/ydb_replication.h b/ydb/public/sdk/cpp/client/draft/ydb_replication.h
index ee7db99f197b..ba63be422450 100644
--- a/ydb/public/sdk/cpp/client/draft/ydb_replication.h
+++ b/ydb/public/sdk/cpp/client/draft/ydb_replication.h
@@ -52,6 +52,7 @@ class TConnectionParams: private TCommonClientSettings {
 
     const TString& GetDiscoveryEndpoint() const;
     const TString& GetDatabase() const;
+    bool GetEnableSsl() const;
 
     ECredentials GetCredentials() const;
     const TStaticCredentials& GetStaticCredentials() const;