From f41380bb6bfe5294d5640a7889c1cdef11132157 Mon Sep 17 00:00:00 2001 From: Andrey Zvonov <32552679+zvonand@users.noreply.github.com> Date: Fri, 24 Jul 2026 18:33:07 +0200 Subject: [PATCH] fix interserver jwt --- src/Access/TokenAccessStorage.cpp | 69 +++++--- .../test_jwt_auth/configs/remote_servers.xml | 20 +++ .../configs/users_interserver.xml | 15 ++ .../test_jwt_auth/configs/validators.xml | 8 + .../test_jwt_auth/test_interserver.py | 151 ++++++++++++++++++ 5 files changed, 242 insertions(+), 21 deletions(-) create mode 100644 tests/integration/test_jwt_auth/configs/remote_servers.xml create mode 100644 tests/integration/test_jwt_auth/configs/users_interserver.xml create mode 100644 tests/integration/test_jwt_auth/test_interserver.py diff --git a/src/Access/TokenAccessStorage.cpp b/src/Access/TokenAccessStorage.cpp index 5723d2d9d783..4142e4b91c36 100644 --- a/src/Access/TokenAccessStorage.cpp +++ b/src/Access/TokenAccessStorage.cpp @@ -407,6 +407,11 @@ bool TokenAccessStorage::areTokenCredentialsValidNoLock(const User & user, const if (credentials.getUserName() != user.getName()) return false; + /// Interserver hops authenticate with the cluster secret and then trust + /// `initial_user` via AlwaysAllowCredentials (see TCPHandler). Mirror LDAP. + if (typeid_cast(&credentials)) + return true; + if (const auto * token_credentials = dynamic_cast(&credentials)) return external_authenticators.checkTokenCredentials(*token_credentials); @@ -570,17 +575,18 @@ std::optional TokenAccessStorage::authenticateImpl( { std::unique_lock lock(mutex); - /// Reject mismatched credential types BEFORE the typeid_cast that would - /// throw a `LOGICAL_ERROR`. The reference-form `typeid_cast` is fatal on - /// mismatch, and `MultipleAccessStorage::authenticateImpl` does not catch - /// per-storage exceptions -- so a single Basic / SSL-cert / Kerberos / SSH - /// login attempt would propagate that exception out of the chain and abort - /// authentication for every later storage in `user_directories`. Concretely, - /// listing `` ahead of `` would lock out every Basic-auth - /// user. Return nullopt cleanly, matching the LDAP-side idiom in - /// `LDAPAccessStorage::areLDAPCredentialsValidNoLock`. + /// Accept TokenCredentials (normal JWT login) and AlwaysAllowCredentials + /// (interserver hop after cluster-secret verification). Reject every other + /// credential type BEFORE any reference-form `typeid_cast` that would throw + /// a `LOGICAL_ERROR`. `MultipleAccessStorage::authenticateImpl` does not + /// catch per-storage exceptions -- so a single Basic / SSL-cert / Kerberos / + /// SSH login attempt would abort authentication for every later storage in + /// `user_directories`. Concretely, listing `` ahead of `` + /// would lock out every Basic-auth user. Return nullopt cleanly, matching + /// the LDAP-side idiom in `LDAPAccessStorage::areLDAPCredentialsValidNoLock`. + const auto * always_allow_credentials = dynamic_cast(&credentials); const auto * token_credentials_ptr = dynamic_cast(&credentials); - if (!token_credentials_ptr) + if (!always_allow_credentials && !token_credentials_ptr) { if (throw_if_user_not_exists) throwNotFound(AccessEntityType::USER, credentials.getUserName(), getStorageName()); @@ -590,17 +596,6 @@ std::optional TokenAccessStorage::authenticateImpl( auto id = memory_storage.find(credentials.getUserName()); UserPtr user = id ? memory_storage.read(*id) : nullptr; - const auto & token_credentials = *token_credentials_ptr; - - if (!external_authenticators.checkTokenCredentials(token_credentials, provider_name)) - { - // Even though token itself may be valid (especially in case of a jwt token), authentication has just failed. - if (throw_if_user_not_exists) - throwNotFound(AccessEntityType::USER, credentials.getUserName(), getStorageName()); - - return {}; - } - std::shared_ptr new_user; if (!user) { @@ -625,6 +620,38 @@ std::optional TokenAccessStorage::authenticateImpl( if (!isAddressAllowed(*user, address)) throwAddressNotAllowed(address); + /// Interserver mode: TCPHandler already verified the cluster secret and is + /// trusting `client_info.initial_user`. + if (always_allow_credentials) + { + if (new_user) + { + /// TODO: mapped roles from the JWT are not available here without + /// the bearer token; grant common roles only. Session applies the + /// roles pushed by the initiator. + assignRolesNoLock(*new_user, /* external_roles= */ {}); + assignProfileNoLock(*new_user); + id = memory_storage.insert(new_user); + + lock.unlock(); + access_control.getChangesNotifier().sendNotifications(); + } + + if (id) + return AuthResult{ .user_id = *id, .authentication_data = AuthenticationData(AuthenticationType::JWT), .user_name = user->getName() }; + return std::nullopt; + } + + const auto & token_credentials = *token_credentials_ptr; + + if (!external_authenticators.checkTokenCredentials(token_credentials, provider_name)) + { + if (throw_if_user_not_exists) + throwNotFound(AccessEntityType::USER, credentials.getUserName(), getStorageName()); + + return {}; + } + /// Pipeline: incoming group --(roles_mapping)--> mapped name --(roles_filter)--> kept/dropped --(roles_transform)--> CH role name. /// Each stage is independent and optional; groups absent from `roles_mapping` pass through unchanged. std::set external_roles; diff --git a/tests/integration/test_jwt_auth/configs/remote_servers.xml b/tests/integration/test_jwt_auth/configs/remote_servers.xml new file mode 100644 index 000000000000..8c694dfc4f1a --- /dev/null +++ b/tests/integration/test_jwt_auth/configs/remote_servers.xml @@ -0,0 +1,20 @@ + + 9009 + + + my_secret + + + instance1 + 9000 + + + + + instance2 + 9000 + + + + + diff --git a/tests/integration/test_jwt_auth/configs/users_interserver.xml b/tests/integration/test_jwt_auth/configs/users_interserver.xml new file mode 100644 index 000000000000..aef005651442 --- /dev/null +++ b/tests/integration/test_jwt_auth/configs/users_interserver.xml @@ -0,0 +1,15 @@ + + + + 0 + + + + + + qwerty + 1 + default + + + diff --git a/tests/integration/test_jwt_auth/configs/validators.xml b/tests/integration/test_jwt_auth/configs/validators.xml index e6bb8a1d265e..1c3ad2ba2076 100644 --- a/tests/integration/test_jwt_auth/configs/validators.xml +++ b/tests/integration/test_jwt_auth/configs/validators.xml @@ -23,4 +23,12 @@ true + + + + single_key_processor + default + ^role_[a-z_]+$ + + diff --git a/tests/integration/test_jwt_auth/test_interserver.py b/tests/integration/test_jwt_auth/test_interserver.py new file mode 100644 index 000000000000..4709242b91e7 --- /dev/null +++ b/tests/integration/test_jwt_auth/test_interserver.py @@ -0,0 +1,151 @@ +""" +Regression: JWT / token-directory users must be able to run Distributed / +cluster() queries when remote_servers uses an interserver . + +Without AlwaysAllowCredentials support in TokenAccessStorage, the internal +native hop authenticates with the JWT subject name alone, fails with +"There is no user '' in token", and the initiator surfaces Code 210. + +The durable path mirrors LDAP: after the cluster secret is verified, the +receiving node trusts initial_user via AlwaysAllowCredentials and applies +roles pushed by push_external_roles_in_interserver_queries. +""" + +import jwt +import pytest + +from helpers.cluster import ClickHouseCluster + +# Matches configs/validators.xml . +SECRET = "my_secret" +ROLE_GROUP = "role_read" + +cluster = ClickHouseCluster(__file__) + +instance1 = cluster.add_instance( + "instance1", + main_configs=["configs/validators.xml", "configs/remote_servers.xml"], + user_configs=["configs/users_interserver.xml"], + macros={"shard": 1, "replica": "instance1"}, + stay_alive=True, +) + +instance2 = cluster.add_instance( + "instance2", + main_configs=["configs/validators.xml", "configs/remote_servers.xml"], + user_configs=["configs/users_interserver.xml"], + macros={"shard": 1, "replica": "instance2"}, + stay_alive=True, +) + + +@pytest.fixture(scope="module", autouse=True) +def started_cluster(): + try: + cluster.start() + yield cluster + finally: + cluster.shutdown() + + +def make_jwt(sub, groups): + return jwt.encode({"sub": sub, "groups": groups}, SECRET, algorithm="HS256") + + +def query_with_token(node, token, sql): + resp = node.http_request( + "", + method="POST", + data=sql, + headers={"Authorization": f"Bearer {token}"}, + ) + resp.raise_for_status() + return resp.text + + +def test_push_token_role_to_other_nodes(): + """ + instance1 maps JWT group -> role_read and pushes it over the interserver + hop. instance2 must honor the pushed role even though the JWT subject has + never authenticated there directly. + """ + for node in (instance1, instance2): + node.query("DROP TABLE IF EXISTS distributed_table SYNC", user="common_user", password="qwerty") + node.query("DROP TABLE IF EXISTS local_table SYNC", user="common_user", password="qwerty") + node.query("DROP ROLE IF EXISTS role_read", user="common_user", password="qwerty") + node.query("CREATE ROLE role_read", user="common_user", password="qwerty") + node.query("GRANT SELECT, REMOTE ON *.* TO role_read", user="common_user", password="qwerty") + node.query( + "CREATE TABLE IF NOT EXISTS local_table (id UInt32) ENGINE = MergeTree() ORDER BY id", + user="common_user", + password="qwerty", + ) + + instance2.query( + "INSERT INTO local_table VALUES (1), (2), (3)", + user="common_user", + password="qwerty", + ) + + instance1.query( + "CREATE TABLE IF NOT EXISTS distributed_table AS local_table " + "ENGINE = Distributed(test_token_cluster, default, local_table)", + user="common_user", + password="qwerty", + ) + + token = make_jwt("jwt_distributed_user", [ROLE_GROUP]) + + # Sanity: token auth + role mapping work locally on the initiator. + roles = query_with_token( + instance1, + token, + "SELECT role_name FROM system.current_roles ORDER BY role_name FORMAT TabSeparated", + ) + assert roles.strip() == "role_read" + + # Distributed query must succeed via the interserver secret hop with + # AlwaysAllowCredentials + pushed roles (prefer_localhost_replica=0). + result = query_with_token( + instance1, + token, + "SELECT sum(id) FROM distributed_table FORMAT TabSeparated", + ) + assert result.strip() == "6" + + for node in (instance1, instance2): + node.query("DROP TABLE IF EXISTS distributed_table SYNC", user="common_user", password="qwerty") + node.query("DROP TABLE IF EXISTS local_table SYNC", user="common_user", password="qwerty") + node.query("DROP ROLE IF EXISTS role_read", user="common_user", password="qwerty") + + +def test_cluster_table_function_with_token_user(): + """Same failure mode via cluster(), which Hybrid tables use for the native segment.""" + for node in (instance1, instance2): + node.query("DROP TABLE IF EXISTS local_table SYNC", user="common_user", password="qwerty") + node.query("DROP ROLE IF EXISTS role_read", user="common_user", password="qwerty") + node.query("CREATE ROLE role_read", user="common_user", password="qwerty") + node.query("GRANT SELECT, REMOTE ON *.* TO role_read", user="common_user", password="qwerty") + node.query( + "CREATE TABLE IF NOT EXISTS local_table (id UInt32) ENGINE = MergeTree() ORDER BY id", + user="common_user", + password="qwerty", + ) + + instance2.query( + "INSERT INTO local_table VALUES (10), (20)", + user="common_user", + password="qwerty", + ) + + token = make_jwt("jwt_cluster_user", [ROLE_GROUP]) + result = query_with_token( + instance1, + token, + "SELECT sum(id) FROM cluster('test_token_cluster', default.local_table) FORMAT TabSeparated", + ) + assert result.strip() == "30" + + for node in (instance1, instance2): + node.query("DROP TABLE IF EXISTS local_table SYNC", user="common_user", password="qwerty") + node.query("DROP ROLE IF EXISTS role_read", user="common_user", password="qwerty")