Antalya 26.3: fix interserver JWT

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<const AlwaysAllowCredentials *>(&credentials))
+        return true;
+
     if (const auto * token_credentials = dynamic_cast<const TokenCredentials *>(&credentials))
         return external_authenticators.checkTokenCredentials(*token_credentials);
 
@@ -570,17 +575,18 @@ std::optional<AuthResult> 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 `<token>` ahead of `<users.xml>` 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 `<token>` ahead of `<users.xml>`
+    /// 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<const AlwaysAllowCredentials *>(&credentials);
     const auto * token_credentials_ptr = dynamic_cast<const TokenCredentials *>(&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<AuthResult> TokenAccessStorage::authenticateImpl(
     auto id = memory_storage.find<User>(credentials.getUserName());
     UserPtr user = id ? memory_storage.read<User>(*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<User> new_user;
     if (!user)
     {
@@ -625,6 +620,38 @@ std::optional<AuthResult> 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<String> external_roles;

tests/integration/test_jwt_auth/configs/remote_servers.xml

@@ -0,0 +1,20 @@
+<clickhouse>
+    <interserver_http_port>9009</interserver_http_port>
+    <remote_servers>
+        <test_token_cluster>
+            <secret>my_secret</secret>
+            <shard>
+                <replica>
+                    <host>instance1</host>
+                    <port>9000</port>
+                </replica>
+            </shard>
+            <shard>
+                <replica>
+                    <host>instance2</host>
+                    <port>9000</port>
+                </replica>
+            </shard>
+        </test_token_cluster>
+    </remote_servers>
+</clickhouse>

tests/integration/test_jwt_auth/configs/users_interserver.xml

@@ -0,0 +1,15 @@
+<clickhouse>
+    <profiles>
+        <default>
+            <prefer_localhost_replica>0</prefer_localhost_replica>
+        </default>
+    </profiles>
+    <users>
+        <default remove="remove"/>
+        <common_user>
+            <password>qwerty</password>
+            <access_management>1</access_management>
+            <profile>default</profile>
+        </common_user>
+    </users>
+</clickhouse>

tests/integration/test_jwt_auth/configs/validators.xml

@@ -23,4 +23,12 @@
             <allow_no_expiration>true</allow_no_expiration>
         </remote_jwks_processor>
     </token_processors>
+
+    <user_directories>
+        <token>
+            <processor>single_key_processor</processor>
+            <default_profile>default</default_profile>
+            <roles_filter>^role_[a-z_]+$</roles_filter>
+        </token>
+    </user_directories>
 </clickhouse>

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 <secret>.
+
+Without AlwaysAllowCredentials support in TokenAccessStorage, the internal
+native hop authenticates with the JWT subject name alone, fails with
+"There is no user '<sub>' 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 <single_key_processor>.
+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")