Antalya 26.6: Forward auth tokens to data lake catalogs

docs/en/engines/database-engines/datalake.md

@@ -61,6 +61,12 @@ The following settings are supported:
 | `dlf_access_key_id`     | Access key ID for DLF access                                                                  |
 | `dlf_access_key_secret` | Access key Secret for DLF access                                                              |
 | `namespaces`            | Comma-separated list of namespaces, implemented for catalog types: `rest`, `glue` and `unity` |
+| `oauth_forward_user_token` | Authenticate to the catalog as the user running the query instead of as the shared service identity. Iceberg REST and Glue only. See [Forwarding the user's identity to the catalog](#user-token-forwarding) |
+| `oauth_token_exchange_uri` | Empty (the default) forwards the user's token unchanged; non-empty performs an RFC 8693 token exchange at this URL first |
+| `oauth_subject_token_type` | RFC 8693 `subject_token_type` of the forwarded token. Default `urn:ietf:params:oauth:token-type:access_token` |
+| `oauth_requested_token_type` | RFC 8693 `requested_token_type`; empty omits the field. Default `urn:ietf:params:oauth:token-type:access_token` |
+| `oauth_forward_actor_token` | Send the service principal's own token as the RFC 8693 `actor_token`. Default `0`. See [Delegation with an actor token](#user-token-forwarding-actor-token) |
+| `oauth_user_token_cache_ttl` | Maximum lifetime (in seconds) of a cached exchanged session token; `0` disables caching. Default `300` |
 
 ## Examples {#examples}
 
@@ -86,6 +92,115 @@ SELECT count() from database_name.table_name;
 ```
 To authenticate without sharing a client secret, set `onelake_bearer_token` to a pre-obtained bearer token (scoped to `https://storage.azure.com`) instead of `onelake_client_id`/`onelake_client_secret`. ClickHouse does not refresh the token, so the database must be recreated after it expires.
 
+## Forwarding the user's identity to the catalog {#user-token-forwarding}
+
+Set `oauth_forward_user_token = 1` to use the querying user's token for catalog authentication.
+This requires:
+
+- the server-level [`enable_token_forwarding`](/operations/server-configuration-parameters/settings#enable_token_forwarding)
+  setting, which is `false` by default;
+- `catalog_type = 'rest'` or `catalog_type = 'glue'`;
+- token authentication through an `Authorization: Bearer` HTTP header or `--jwt` for the native
+  protocol. See [Token-based authentication](/en/operations/external-authenticators/oauth).
+
+:::danger Restrict `CREATE DATABASE`
+Database creators choose the endpoints that receive users' tokens. Grant `CREATE DATABASE` only
+to trusted users and restrict `remote_url_allow_hosts`, which applies to catalog and token-exchange
+endpoints.
+:::
+
+### Passthrough: the default {#user-token-forwarding-passthrough}
+
+For Iceberg REST catalogs, `oauth_forward_user_token = 1` forwards the user's bearer token
+unchanged. No token endpoint or client credentials are required:
+
+```sql
+CREATE DATABASE demo
+ENGINE = DataLakeCatalog('http://lakekeeper:8181/catalog')
+SETTINGS
+    catalog_type = 'rest',
+    warehouse = 'demo',
+    oauth_forward_user_token = 1;
+```
+
+The token's audience must cover both ClickHouse and the catalog. With Keycloak, add an audience
+mapper to the ClickHouse client to include the catalog's audience.
+
+### Token exchange: opt-in {#user-token-forwarding-exchange}
+
+Set `oauth_token_exchange_uri` to exchange the user's token using
+[RFC 8693](https://www.rfc-editor.org/rfc/rfc8693) before presenting it to the REST catalog.
+Use an IdP token endpoint that issues tokens with an audience the catalog accepts:
+
+```sql
+CREATE DATABASE demo
+ENGINE = DataLakeCatalog('http://lakekeeper:8181/catalog')
+SETTINGS
+    catalog_type = 'rest',
+    warehouse = 'demo',
+    catalog_credential = 'clickhouse:<client-secret>',
+    auth_scope = 'lakekeeper',
+    oauth_forward_user_token = 1,
+    oauth_token_exchange_uri = 'http://keycloak:8080/realms/demo/protocol/openid-connect/token';
+```
+
+The exchange requires `catalog_credential`; its `client_id` and `client_secret` are sent in the
+form body. `auth_scope` supplies the exchange `scope`. Override its Polaris-specific default,
+`PRINCIPAL_ROLE:ALL`, for other providers.
+
+A catalog's `/v1/oauth/tokens` endpoint can also be used if supported. The Iceberg REST
+specification deprecates this endpoint, and some catalogs, including Lakekeeper, do not implement it.
+
+### Delegation with an actor token {#user-token-forwarding-actor-token}
+
+Set `oauth_forward_actor_token = 1` to include the service principal's token as the RFC 8693
+`actor_token`. This requires `oauth_token_exchange_uri` and an endpoint that supports delegation
+and can validate the actor token.
+
+ClickHouse obtains the actor token through a `client_credentials` grant using `catalog_credential`
+at `oauth_server_uri`, or the catalog's `/v1/oauth/tokens` endpoint if that setting is empty.
+The token is cached until expiry and used only for exchanges. If obtaining it fails, the query fails.
+
+### Glue {#user-token-forwarding-glue}
+
+For Glue, ClickHouse exchanges the user's token through AWS STS `AssumeRoleWithWebIdentity`
+for temporary credentials of `aws_role_arn`. These credentials sign Glue and S3 requests using
+SigV4. The ClickHouse user name supplies `RoleSessionName` for CloudTrail auditing.
+
+```sql
+CREATE DATABASE glue_db
+ENGINE = DataLakeCatalog
+SETTINGS
+    catalog_type = 'glue',
+    region = 'us-east-1',
+    aws_role_arn = 'arn:aws:iam::123456789012:role/data-lake-reader',
+    oauth_forward_user_token = 1;
+```
+
+Register the token issuer as an IAM OIDC identity provider and configure the role's trust policy
+to accept the users' tokens, including their `aud` and `sub` claims.
+
+- `aws_role_arn` is required.
+- `aws_access_key_id`, `aws_secret_access_key`, and RFC 8693 exchange settings are rejected.
+- The AWS STS endpoint is determined by `region`.
+- Users receive the assumed role's permissions. Use separate roles or session-tag policies to
+  distinguish access. ClickHouse does not implement IAM Identity Center trusted identity propagation.
+
+### Scope and limitations {#user-token-forwarding-scope}
+
+- Forwarding covers catalog listings, table metadata, and write operations (`INSERT`, `ALTER`,
+  mutations, `DROP TABLE`, and snapshot expiry).
+- Vended storage credentials and Glue sessions are cached separately for each user token.
+- Requests without a user token fail with `CATALOG_USER_TOKEN_NOT_AVAILABLE`; ClickHouse does not
+  fall back to the service identity. Catalog listings may suppress these errors and appear empty,
+  depending on `database_datalake_require_metadata_access`.
+- With `object_storage_cluster`, workers receive table-scoped storage credentials over the
+  interserver channel. Configure `interserver_https_port` or a cluster `<secret>` for cluster reads.
+- For Iceberg REST, rotate `catalog_credential` with `ALTER DATABASE ... MODIFY SETTING`,
+  authenticated with a user token. ClickHouse validates the new credentials and reloads the catalog
+  configuration before applying the change, then invalidates cached session and storage credentials.
+  Glue settings cannot be altered.
+
 ## Namespace filter {#namespace}
 
 By default, ClickHouse reads tables from all namespaces available in the catalog. You can limit this behavior using the `namespaces` database setting. The value should be a comma‑separated list of namespaces that are allowed to be read.

docs/en/operations/external-authenticators/tokens.md

@@ -300,6 +300,35 @@ To reduce number of requests to IdP, tokens are cached internally for a maximum
 If token expires sooner than `token_cache_lifetime`, then cache entry for this token will only be valid while token is valid.
 If token lifetime is longer than `token_cache_lifetime`, cache entry for this token will be valid for `token_cache_lifetime`. 
 
+## Forwarding the token to external services {#token-forwarding}
+
+By default, the authenticated token is not retained in the session for forwarding. Set
+`enable_token_forwarding` to `1` in `config.xml` to retain it for external-service authentication:
+
+```xml
+<enable_token_forwarding>1</enable_token_forwarding>
+```
+
+The setting is hot-reloadable and defaults to `false`. To use it with Iceberg REST or Glue, enable
+`oauth_forward_user_token` on the [`DataLakeCatalog`](/engines/database-engines/datalakecatalog)
+database. See [catalog token forwarding](/engines/database-engines/datalakecatalog#user-token-forwarding)
+for configuration and examples.
+
+:::danger Restrict `CREATE DATABASE`
+Database creators choose the endpoints that receive users' tokens. Grant `CREATE DATABASE` only
+to trusted users and restrict `remote_url_allow_hosts` for catalog and token-exchange endpoints.
+:::
+
+Only the session's verified token is forwarded. It is not inherited by `EXECUTE AS` or
+`DEFINER` views, forwarded to other ClickHouse nodes, or persisted to disk.
+
+HTTP requests authenticate separately, so a rotated token takes effect on the next request.
+Native TCP sessions must reconnect with the new token.
+
+With `async_insert = 1`, each queued batch retains its token until the flush completes, even with
+`wait_for_async_insert = 0`. Different tokens use separate batches; rotation does not change a
+queued batch's token.
+
 ## Enabling token authentication for a user in `users.xml` {#enabling-jwt-auth-in-users-xml}
 
 In order to enable token-based authentication for the user, specify `jwt` section instead of `password` or other similar sections in the user definition.

src/Access/AccessControl.cpp

@@ -295,6 +295,7 @@ void AccessControl::setupFromMainConfig(const Poco::Util::AbstractConfiguration
     setPasswordComplexityRulesFromConfig(config_);
 
     setTokenAuthEnabled(config_.getBool("enable_token_auth", true));
+    setTokenForwardingEnabled(config_.getBool("enable_token_forwarding", false));
 
     setBcryptWorkfactor(config_.getInt("bcrypt_workfactor", 12));
 
@@ -705,6 +706,7 @@ void AccessControl::setExternalAuthenticatorsConfig(const Poco::Util::AbstractCo
     /// value in place -- operators who toggle token auth off in response to an
     /// IdP outage or a credential leak would see no effect until restart.
     setTokenAuthEnabled(config.getBool("enable_token_auth", true));
+    setTokenForwardingEnabled(config.getBool("enable_token_forwarding", false));
     external_authenticators->setConfiguration(config, getLogger(), token_http_timeouts, isTokenAuthEnabled());
 }
 
@@ -994,4 +996,14 @@ bool AccessControl::isTokenAuthEnabled() const
 {
     return enable_token_auth;
 }
+
+void AccessControl::setTokenForwardingEnabled(bool enable)
+{
+    enable_token_forwarding = enable;
+}
+
+bool AccessControl::isTokenForwardingEnabled() const
+{
+    return enable_token_forwarding;
+}
 }

src/Access/AccessControl.h

@@ -283,6 +283,9 @@ class AccessControl : public MultipleAccessStorage
     void setTokenAuthEnabled(bool enable);
     bool isTokenAuthEnabled() const;
 
+    void setTokenForwardingEnabled(bool enable);
+    bool isTokenForwardingEnabled() const;
+
 private:
     class ContextAccessCache;
     class CustomSettingsPrefixes;
@@ -320,6 +323,7 @@ class AccessControl : public MultipleAccessStorage
     std::atomic_bool enable_read_write_grants = false;
     std::atomic_bool allow_impersonate_user = false;
     std::atomic_bool enable_token_auth = true;
+    std::atomic_bool enable_token_forwarding = false;
 };
 
 }

src/Access/ForwardedAuthToken.cpp

@@ -0,0 +1,18 @@
+#include <Access/ForwardedAuthToken.h>
+
+#include <Access/Credentials.h>
+#include <Common/SipHash.h>
+
+namespace DB
+{
+
+ForwardedAuthTokenPtr makeForwardedAuthToken(const TokenCredentials & credentials, const String & principal)
+{
+    auto result = std::make_shared<ForwardedAuthToken>();
+    result->token = credentials.getToken();
+    result->fingerprint = getSipHash128AsHexString(sipHash128(result->token.data(), result->token.size()));
+    result->principal = principal;
+    return result;
+}
+
+}

src/Access/ForwardedAuthToken.h

@@ -0,0 +1,25 @@
+#pragma once
+
+#include <base/types.h>
+
+#include <memory>
+
+namespace DB
+{
+
+class TokenCredentials;
+
+struct ForwardedAuthToken
+{
+    String token;
+    /// Use the token fingerprint so rotation cannot reuse credentials cached for the previous token.
+    String fingerprint;
+    String principal;
+};
+
+using ForwardedAuthTokenPtr = std::shared_ptr<const ForwardedAuthToken>;
+
+/// `principal` must be the canonical `AuthResult::user_name`, not the name the client sent.
+ForwardedAuthTokenPtr makeForwardedAuthToken(const TokenCredentials & credentials, const String & principal);
+
+}

src/Common/CurrentMetrics.cpp

@@ -372,6 +372,10 @@
     M(DNSAddressesCacheSize, "Number of cached DNS addresses") \
     M(MarkCacheBytes, "Total size of mark cache in bytes") \
     M(MarkCacheFiles, "Total number of mark files cached in the mark cache") \
+    M(DataLakeCatalogUserTokenCacheBytes, "Total size in bytes of the per-user session tokens exchanged for data lake catalog access") \
+    M(DataLakeCatalogUserTokenCacheEntries, "Total number of per-user session tokens exchanged for data lake catalog access") \
+    M(DataLakeCatalogUserClientCacheBytes, "Total size in bytes of the per-user catalog clients built from forwarded user tokens") \
+    M(DataLakeCatalogUserClientCacheEntries, "Total number of per-user catalog clients built from forwarded user tokens") \
     M(UniqueKeyIndexCacheBytes, "Total size of UNIQUE KEY index cache in bytes") \
     M(UniqueKeyIndexCacheEntries, "Total number of UNIQUE KEY index blocks cached") \
     M(DeleteBitmapCacheBytes, "Total size of the UNIQUE KEY delete-bitmap cache in bytes") \

src/Common/ErrorCodes.cpp

@@ -659,6 +659,7 @@
     M(777, MEMORY_RESERVATION_KILLED) \
     M(778, MEMORY_RESERVATION_FAILED) \
     M(779, CATALOG_NAMESPACE_DISABLED) \
+    M(780, CATALOG_USER_TOKEN_NOT_AVAILABLE) \
 \
     M(900, DISTRIBUTED_CACHE_ERROR) \
     M(901, CANNOT_USE_DISTRIBUTED_CACHE) \

src/Common/FormUrlEncode.cpp

@@ -0,0 +1,15 @@
+#include <Common/FormUrlEncode.h>
+
+#include <Poco/URI.h>
+
+namespace DB
+{
+
+std::string formUrlEncode(const std::string & value)
+{
+    std::string encoded;
+    Poco::URI::encode(value, "!$&'()*+,;=:@/?", encoded);
+    return encoded;
+}
+
+}

src/Common/FormUrlEncode.h

@@ -0,0 +1,11 @@
+#pragma once
+
+#include <string>
+
+namespace DB
+{
+
+/// `Poco::URI::encode` leaves form delimiters unescaped unless they are explicitly reserved.
+std::string formUrlEncode(const std::string & value);
+
+}

src/Common/ProfileEvents.cpp

@@ -1734,6 +1734,10 @@ The server successfully detected this situation and will download merged part fr
     M(ObjectStorageListObjectsCachePrefixMatchHits, "Number of times object storage list objects operation miss the cache using prefix matching.", ValueType::Number) \
     M(DataLakeRestCatalogCredentialsVended, "Number of table metadata requests to REST catalog asking to vend storage credentials.", ValueType::Number) \
     M(DataLakeRestCatalogCredentialsCacheHits, "Number of table metadata requests to REST catalog reusing cached storage credentials.", ValueType::Number) \
+    M(DataLakeRestCatalogTokenExchange, "Number of RFC 8693 token exchanges performed to obtain a session token for the querying user.", ValueType::Number) \
+    M(DataLakeRestCatalogTokenExchangeMicroseconds, "Total time of RFC 8693 token exchanges.", ValueType::Microseconds) \
+    M(DataLakeRestCatalogTokenExchangeFailures, "Number of RFC 8693 token exchanges that failed.", ValueType::Number) \
+    M(DataLakeRestCatalogUserTokenCacheHits, "Number of times a previously exchanged per-user session token was reused.", ValueType::Number) \
     \
     M(DataLakeRestCatalogLoadConfig, "Number of 'load config' requests to Iceberg REST catalog.", ValueType::Number) \
     M(DataLakeRestCatalogLoadConfigMicroseconds, "Total time of 'load config' requests to Iceberg REST catalog.", ValueType::Microseconds) \
@@ -1772,6 +1776,10 @@ The server successfully detected this situation and will download merged part fr
     M(DataLakeGlueCatalogUpdateTableMicroseconds, "Total time of 'update table' requests to Iceberg Glue catalog.", ValueType::Microseconds) \
     M(DataLakeGlueCatalogDropTable, "Number of 'drop table' requests to Iceberg Glue catalog.", ValueType::Number) \
     M(DataLakeGlueCatalogDropTableMicroseconds, "Total time of 'drop table' requests to Iceberg Glue catalog.", ValueType::Microseconds) \
+    M(DataLakeGlueCatalogAssumeRoleWithWebIdentity, "Number of AWS STS `AssumeRoleWithWebIdentity` calls made to turn a forwarded user token into credentials for the Glue catalog.", ValueType::Number) \
+    M(DataLakeGlueCatalogAssumeRoleWithWebIdentityMicroseconds, "Total time of AWS STS `AssumeRoleWithWebIdentity` calls made for forwarded user tokens.", ValueType::Microseconds) \
+    M(DataLakeGlueCatalogAssumeRoleWithWebIdentityFailures, "Number of AWS STS `AssumeRoleWithWebIdentity` calls that returned no credentials.", ValueType::Number) \
+    M(DataLakeGlueCatalogUserClientCacheHits, "Number of times a Glue client built for a forwarded user token was reused.", ValueType::Number) \
     \
     M(DataLakeUnityCatalogGetTables, "Number of 'get tables' requests to Iceberg Unity catalog.", ValueType::Number) \
     M(DataLakeUnityCatalogGetTablesMicroseconds, "Total time of 'get tables' requests to Iceberg Unity catalog.", ValueType::Microseconds) \

src/Core/ServerSettings.cpp

@@ -890,6 +890,16 @@ namespace
 
     Default value: `true` (token authentication is enabled).
     )", 0) \
+    DECLARE(Bool, enable_token_forwarding, false, R"(
+    Retain authenticated bearer tokens for forwarding through the `DataLakeCatalog`
+    setting `oauth_forward_user_token`. Supports Iceberg REST and AWS STS for Glue.
+    When disabled, tokens are not retained in sessions for forwarding.
+
+    Database creators choose the endpoints that receive users' tokens. Grant `CREATE DATABASE`
+    only to trusted users and restrict `remote_url_allow_hosts`.
+
+    Default value: `false`.
+    )", 0) \
     DECLARE(UInt64, concurrent_threads_soft_limit_num, 0, R"(
     The maximum number of query processing threads, excluding threads for retrieving data from remote servers, allowed to run all queries. This is not a hard limit. In case if the limit is reached the query will still get at least one thread to run. Query can upscale to desired number of threads during execution if more threads become available.
 

src/Databases/DataLake/Common.cpp

@@ -1,5 +1,7 @@
 #include <Databases/DataLake/Common.h>
 
+#include <Interpreters/Context.h>
+
 #include <DataTypes/DataTypeArray.h>
 #include <DataTypes/DataTypeDate.h>
 #include <DataTypes/DataTypeDateTime64.h>
@@ -110,6 +112,13 @@ DB::DataTypePtr getType(const String & type_name, bool nullable, DB::ContextPtr
                     : DB::Iceberg::IcebergSchemaProcessor::getSimpleType(name, context);
 }
 
+DB::ForwardedAuthTokenPtr getForwardedAuthToken(const DB::ContextPtr & context)
+{
+    if (!context)
+        return {};
+    return context->getForwardedAuthToken();
+}
+
 std::pair<std::string, std::string> parseTableName(const std::string & name)
 {
     auto pos = name.rfind('.');

src/Databases/DataLake/Common.h

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <Access/ForwardedAuthToken.h>
 #include <Core/NamesAndTypes.h>
 #include <Core/Types.h>
 #include <Interpreters/Context_fwd.h>
@@ -19,4 +20,6 @@ DB::DataTypePtr getType(const String & type_name, bool nullable, DB::ContextPtr
 /// `E` is a table name.
 std::pair<std::string, std::string> parseTableName(const std::string & name);
 
+DB::ForwardedAuthTokenPtr getForwardedAuthToken(const DB::ContextPtr & context);
+
 }