creating CreateSecuredServiceAccountCredentialsOverTokenAccessorFactory

Added ISecuredServiceAccountCredentialsFactory in TKqpProxyService.

12 files changed, 40 insertions, 8 deletions

diff --git a/ydb/core/driver_lib/run/kikimr_services_initializers.cpp b/ydb/core/driver_lib/run/kikimr_services_initializers.cpp
index e6805b6dda..6407a597df 100644
--- a/ydb/core/driver_lib/run/kikimr_services_initializers.cpp
+++ b/ydb/core/driver_lib/run/kikimr_services_initializers.cpp
@@ -2105,7 +2105,7 @@ void TKqpServiceInitializer::InitializeServices(NActors::TActorSystemSetup* setu
         GlobalObjects.AddGlobalObject(std::make_shared<NYql::NLog::YqlLoggerScope>(
             new NYql::NLog::TTlsLogBackend(new TNullLogBackend())));
 
-        auto proxy = NKqp::CreateKqpProxyService(Config.GetLogConfig(), Config.GetTableServiceConfig(),
+        auto proxy = NKqp::CreateKqpProxyService(Config.GetLogConfig(), Config.GetTableServiceConfig(), Config.GetAuthConfig().GetTokenAccessorConfig(),
             std::move(settings), Factories->QueryReplayBackendFactory, std::move(kqpProxySharedResources));
         setup->LocalServices.push_back(std::make_pair(
             NKqp::MakeKqpProxyID(NodeId),
diff --git a/ydb/core/kqp/compute_actor/kqp_compute_actor.cpp b/ydb/core/kqp/compute_actor/kqp_compute_actor.cpp
index 7e56569c05..97e6c30fa0 100644
--- a/ydb/core/kqp/compute_actor/kqp_compute_actor.cpp
+++ b/ydb/core/kqp/compute_actor/kqp_compute_actor.cpp
@@ -55,12 +55,12 @@ TComputationNodeFactory GetKqpActorComputeFactory(TKqpScanComputeContext* comput
 
 namespace NKqp {
 
-NYql::NDq::IDqAsyncIoFactory::TPtr CreateKqpAsyncIoFactory(TIntrusivePtr<TKqpCounters> counters, const NYql::IHTTPGateway::TPtr& httpGateway) {
+NYql::NDq::IDqAsyncIoFactory::TPtr CreateKqpAsyncIoFactory(TIntrusivePtr<TKqpCounters> counters, const NYql::IHTTPGateway::TPtr& httpGateway, const NYql::ISecuredServiceAccountCredentialsFactory::TPtr& credentialsFactory) {
     auto factory = MakeIntrusive<NYql::NDq::TDqAsyncIoFactory>();
     RegisterStreamLookupActorFactory(*factory, counters);
     RegisterKqpReadActor(*factory, counters);
-    RegisterS3ReadActorFactory(*factory, nullptr, httpGateway);
-    RegisterS3WriteActorFactory(*factory, nullptr, httpGateway);
+    RegisterS3ReadActorFactory(*factory, credentialsFactory, httpGateway);
+    RegisterS3WriteActorFactory(*factory, credentialsFactory, httpGateway);
     RegisterSequencerActorFactory(*factory, counters);
     return factory;
 }
diff --git a/ydb/core/kqp/compute_actor/kqp_compute_actor.h b/ydb/core/kqp/compute_actor/kqp_compute_actor.h
index e2b44eeff8..1911a336e6 100644
--- a/ydb/core/kqp/compute_actor/kqp_compute_actor.h
+++ b/ydb/core/kqp/compute_actor/kqp_compute_actor.h
@@ -4,6 +4,7 @@
 #include <ydb/library/yql/dq/actors/compute/dq_compute_actor.h>
 #include <ydb/library/yql/dq/actors/compute/dq_compute_actor_async_io_factory.h>
 #include <ydb/library/yql/providers/common/http_gateway/yql_http_gateway.h>
+#include <ydb/library/yql/providers/common/token_accessor/client/factory.h>
 #include <ydb/core/scheme/scheme_tabledefs.h>
 
 namespace NKikimr {
@@ -58,7 +59,7 @@ IActor* CreateKqpScanFetcher(const NKikimrKqp::TKqpSnapshot& snapshot, std::vect
     const NKikimrTxDataShard::TKqpTransaction::TScanTaskMeta& meta, const NYql::NDq::TComputeRuntimeSettings& settings,
     const ui64 txId, const TShardsScanningPolicy& shardsScanningPolicy, TIntrusivePtr<TKqpCounters> counters, NWilson::TTraceId traceId);
 
-NYql::NDq::IDqAsyncIoFactory::TPtr CreateKqpAsyncIoFactory(TIntrusivePtr<TKqpCounters> counters, const NYql::IHTTPGateway::TPtr& httpGateway);
+NYql::NDq::IDqAsyncIoFactory::TPtr CreateKqpAsyncIoFactory(TIntrusivePtr<TKqpCounters> counters, const NYql::IHTTPGateway::TPtr& httpGateway, const NYql::ISecuredServiceAccountCredentialsFactory::TPtr& credentialsFactory);
 
 } // namespace NKqp
 } // namespace NKikimr
diff --git a/ydb/core/kqp/node_service/kqp_node_ut.cpp b/ydb/core/kqp/node_service/kqp_node_ut.cpp
index 6b2479aeeb..d55ae12d64 100644
--- a/ydb/core/kqp/node_service/kqp_node_ut.cpp
+++ b/ydb/core/kqp/node_service/kqp_node_ut.cpp
@@ -180,7 +180,7 @@ public:
         WaitForBootstrap();
 
         auto httpGateway = NYql::IHTTPGateway::Make();
-        auto asyncIoFactory = CreateKqpAsyncIoFactory(KqpCounters, httpGateway);
+        auto asyncIoFactory = CreateKqpAsyncIoFactory(KqpCounters, httpGateway, nullptr);
         auto kqpNode = CreateKqpNodeService(config, KqpCounters, CompFactory.Get(), asyncIoFactory);
         KqpNodeActorId = Runtime->Register(kqpNode);
         Runtime->EnableScheduleForActor(KqpNodeActorId, true);
diff --git a/ydb/core/kqp/proxy_service/CMakeLists.darwin-x86_64.txt b/ydb/core/kqp/proxy_service/CMakeLists.darwin-x86_64.txt
index 2ce9ff4dd0..e962a6a7b5 100644
--- a/ydb/core/kqp/proxy_service/CMakeLists.darwin-x86_64.txt
+++ b/ydb/core/kqp/proxy_service/CMakeLists.darwin-x86_64.txt
@@ -16,6 +16,7 @@ target_link_libraries(core-kqp-proxy_service PUBLIC
   contrib-libs-cxxsupp
   yutil
   cpp-actors-core
+  cpp-actors-http
   cpp-protobuf-json
   ydb-core-actorlib_impl
   ydb-core-base
diff --git a/ydb/core/kqp/proxy_service/CMakeLists.linux-aarch64.txt b/ydb/core/kqp/proxy_service/CMakeLists.linux-aarch64.txt
index fc1b20b171..ae463c9f90 100644
--- a/ydb/core/kqp/proxy_service/CMakeLists.linux-aarch64.txt
+++ b/ydb/core/kqp/proxy_service/CMakeLists.linux-aarch64.txt
@@ -17,6 +17,7 @@ target_link_libraries(core-kqp-proxy_service PUBLIC
   contrib-libs-cxxsupp
   yutil
   cpp-actors-core
+  cpp-actors-http
   cpp-protobuf-json
   ydb-core-actorlib_impl
   ydb-core-base
diff --git a/ydb/core/kqp/proxy_service/CMakeLists.linux-x86_64.txt b/ydb/core/kqp/proxy_service/CMakeLists.linux-x86_64.txt
index fc1b20b171..ae463c9f90 100644
--- a/ydb/core/kqp/proxy_service/CMakeLists.linux-x86_64.txt
+++ b/ydb/core/kqp/proxy_service/CMakeLists.linux-x86_64.txt
@@ -17,6 +17,7 @@ target_link_libraries(core-kqp-proxy_service PUBLIC
   contrib-libs-cxxsupp
   yutil
   cpp-actors-core
+  cpp-actors-http
   cpp-protobuf-json
   ydb-core-actorlib_impl
   ydb-core-base
diff --git a/ydb/core/kqp/proxy_service/CMakeLists.windows-x86_64.txt b/ydb/core/kqp/proxy_service/CMakeLists.windows-x86_64.txt
index 2ce9ff4dd0..e962a6a7b5 100644
--- a/ydb/core/kqp/proxy_service/CMakeLists.windows-x86_64.txt
+++ b/ydb/core/kqp/proxy_service/CMakeLists.windows-x86_64.txt
@@ -16,6 +16,7 @@ target_link_libraries(core-kqp-proxy_service PUBLIC
   contrib-libs-cxxsupp
   yutil
   cpp-actors-core
+  cpp-actors-http
   cpp-protobuf-json
   ydb-core-actorlib_impl
   ydb-core-base
diff --git a/ydb/core/kqp/proxy_service/kqp_proxy_service.cpp b/ydb/core/kqp/proxy_service/kqp_proxy_service.cpp
index 5b871f4f62..7a55322303 100644
--- a/ydb/core/kqp/proxy_service/kqp_proxy_service.cpp
+++ b/ydb/core/kqp/proxy_service/kqp_proxy_service.cpp
@@ -31,6 +31,7 @@
 #include <library/cpp/actors/core/interconnect.h>
 #include <library/cpp/actors/core/hfunc.h>
 #include <library/cpp/actors/core/log.h>
+#include <library/cpp/actors/http/http.h>
 #include <library/cpp/actors/interconnect/interconnect.h>
 #include <library/cpp/lwtrace/mon/mon_lwtrace.h>
 #include <library/cpp/monlib/service/pages/templates.h>
@@ -104,6 +105,14 @@ TString EncodeSessionId(ui32 nodeId, const TString& id) {
     return NOperationId::ProtoToString(opId);
 }
 
+bool isGrpcProtocol(const TString& endpoint) {
+    TStringBuf scheme;
+    TStringBuf host;
+    TStringBuf uri;
+    NHttp::CrackURL(endpoint, scheme, host, uri);
+    return scheme == "grpcs";
+}
+
 class TKqpProxyService : public TActorBootstrapped<TKqpProxyService> {
     struct TEvPrivate {
         enum EEv {
@@ -158,11 +167,13 @@ public:
 
     TKqpProxyService(const NKikimrConfig::TLogConfig& logConfig,
         const NKikimrConfig::TTableServiceConfig& tableServiceConfig,
+        const NKikimrProto::TTokenAccessorConfig& tokenAccessorConfig,
         TVector<NKikimrKqp::TKqpSetting>&& settings,
         std::shared_ptr<IQueryReplayBackendFactory> queryReplayFactory,
         std::shared_ptr<TKqpProxySharedResources>&& kqpProxySharedResources)
         : LogConfig(logConfig)
         , TableServiceConfig(tableServiceConfig)
+        , TokenAccessorConfig(tokenAccessorConfig)
         , KqpSettings(std::make_shared<const TKqpSettings>(std::move(settings)))
         , QueryReplayFactory(std::move(queryReplayFactory))
         , HttpGateway(NYql::IHTTPGateway::Make()) // TODO: pass config and counters
@@ -172,9 +183,19 @@ public:
     {}
 
     void Bootstrap() {
+        if (TokenAccessorConfig.GetEnabled()) {
+            TString caContent;
+            if (const auto& path = TokenAccessorConfig.GetSslCaCert()) {
+                caContent = TUnbufferedFileInput(path).ReadAll();
+            }
+
+            const TString endpoint = TokenAccessorConfig.GetEndpoint();
+            CredentialsFactory = NYql::CreateSecuredServiceAccountCredentialsOverTokenAccessorFactory(endpoint, isGrpcProtocol(endpoint), caContent, TokenAccessorConfig.GetConnectionPoolSize());
+        }
+
         NLwTraceMonPage::ProbeRegistry().AddProbesList(LWTRACE_GET_PROBES(KQP_PROVIDER));
         Counters = MakeIntrusive<TKqpCounters>(AppData()->Counters, &TlsActivationContext->AsActorContext());
-        AsyncIoFactory = CreateKqpAsyncIoFactory(Counters, HttpGateway);
+        AsyncIoFactory = CreateKqpAsyncIoFactory(Counters, HttpGateway, CredentialsFactory);
         ModuleResolverState = MakeIntrusive<TModuleResolverState>();
 
         LocalSessions = std::make_unique<TLocalSessionsRegistry>(AppData()->RandomProvider);
@@ -1473,7 +1494,9 @@ private:
 private:
     NKikimrConfig::TLogConfig LogConfig;
     NKikimrConfig::TTableServiceConfig TableServiceConfig;
+    NKikimrProto::TTokenAccessorConfig TokenAccessorConfig;
     TKqpSettings::TConstPtr KqpSettings;
+    NYql::ISecuredServiceAccountCredentialsFactory::TPtr CredentialsFactory;
     std::shared_ptr<IQueryReplayBackendFactory> QueryReplayFactory;
     NYql::IHTTPGateway::TPtr HttpGateway;
 
@@ -1522,11 +1545,12 @@ private:
 
 IActor* CreateKqpProxyService(const NKikimrConfig::TLogConfig& logConfig,
     const NKikimrConfig::TTableServiceConfig& tableServiceConfig,
+    const NKikimrProto::TTokenAccessorConfig& tokenAccessorConfig,
     TVector<NKikimrKqp::TKqpSetting>&& settings,
     std::shared_ptr<IQueryReplayBackendFactory> queryReplayFactory,
     std::shared_ptr<TKqpProxySharedResources> kqpProxySharedResources)
 {
-    return new TKqpProxyService(logConfig, tableServiceConfig, std::move(settings),
+    return new TKqpProxyService(logConfig, tableServiceConfig, tokenAccessorConfig, std::move(settings),
         std::move(queryReplayFactory),std::move(kqpProxySharedResources));
 }
 
diff --git a/ydb/core/kqp/proxy_service/kqp_proxy_service.h b/ydb/core/kqp/proxy_service/kqp_proxy_service.h
index 654617b624..868bab3975 100644
--- a/ydb/core/kqp/proxy_service/kqp_proxy_service.h
+++ b/ydb/core/kqp/proxy_service/kqp_proxy_service.h
@@ -49,6 +49,7 @@ TPeerStats CalcPeerStats(const TVector<NKikimrKqp::TKqpProxyNodeResources>& data
 
 IActor* CreateKqpProxyService(const NKikimrConfig::TLogConfig& logConfig,
     const NKikimrConfig::TTableServiceConfig& tableServiceConfig,
+    const NKikimrProto::TTokenAccessorConfig& tokenAccessorConfig,
     TVector<NKikimrKqp::TKqpSetting>&& settings,
     std::shared_ptr<IQueryReplayBackendFactory> queryReplayFactory,
     std::shared_ptr<TKqpProxySharedResources> kqpProxySharedResources);
diff --git a/ydb/core/kqp/proxy_service/ya.make b/ydb/core/kqp/proxy_service/ya.make
index 96ae9140e6..eba98d2c09 100644
--- a/ydb/core/kqp/proxy_service/ya.make
+++ b/ydb/core/kqp/proxy_service/ya.make
@@ -8,6 +8,7 @@ SRCS(
 
 PEERDIR(
     library/cpp/actors/core
+    library/cpp/actors/http
     library/cpp/protobuf/json
     ydb/core/actorlib_impl
     ydb/core/base
diff --git a/ydb/core/testlib/test_client.cpp b/ydb/core/testlib/test_client.cpp
index 01caed4bdc..1d39ebfcfb 100644
--- a/ydb/core/testlib/test_client.cpp
+++ b/ydb/core/testlib/test_client.cpp
@@ -802,6 +802,7 @@ namespace Tests {
 
             IActor* kqpProxyService = NKqp::CreateKqpProxyService(Settings->AppConfig.GetLogConfig(),
                                                                   Settings->AppConfig.GetTableServiceConfig(),
+                                                                  Settings->AppConfig.GetAuthConfig().GetTokenAccessorConfig(),
                                                                   TVector<NKikimrKqp::TKqpSetting>(Settings->KqpSettings),
                                                                   nullptr, std::move(kqpProxySharedResources));
             TActorId kqpProxyServiceId = Runtime->Register(kqpProxyService, nodeIdx);