Merge 'Perform full connection set-up for CertificateAuthorization in process_startup()' from Pavel Emelyanov

The code responds ealry with READY message, but lack some necessary set up, namely:

* update_scheduling_group(): without it, the connection runs under the default scheduling group instead of the one mapped to the user's service level.

* on_connection_ready(): without it, the connection never releases its slot in the uninitialized-connections concurrency semaphore (acquired at connection creation), leaking one unit per cert-authenticated connection for the lifetime of the connection.

* _authenticating = false / _ready = true: without them, system.clients reports connection_stage = AUTHENTICATING forever instead of READY (not critical, but not nice either)

The PR fixes it and adds a regression test, that (for sanity) also covers AllowAll and Password authrticators

Fixes SCYLLADB-1226

Present since 2025.1, probably worth backporting

Closes scylladb/scylladb#29220

* github.com:scylladb/scylladb:
  transport: fix process_startup cert-auth path missing connection-ready setup
  transport: test that connection_stage is READY after auth via all process_startup paths

auth/certificate_authenticator.cc

@@ -14,6 +14,7 @@
 #include <fmt/ranges.h>
 
 #include "utils/to_string.hh"
+#include "utils/error_injection.hh"
 #include "data_dictionary/data_dictionary.hh"
 #include "cql3/query_processor.hh"
 #include "db/config.hh"
@@ -105,6 +106,9 @@ auth::authentication_option_set auth::certificate_authenticator::alterable_optio
 }
 
 future<std::optional<auth::authenticated_user>> auth::certificate_authenticator::authenticate(session_dn_func f) const {
+    if (auto user = utils::get_local_injector().inject_parameter("transport_early_auth_bypass")) {
+        co_return auth::authenticated_user{sstring(*user)};
+    }
     if (!f) {
         co_return std::nullopt;
     }

test/cluster/auth_cluster/test_connection_stage.py

@@ -0,0 +1,100 @@
+#
+# Copyright (C) 2026-present ScyllaDB
+#
+# SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
+#
+
+"""
+Tests that system.clients shows connection_stage = 'READY' after a successful
+authentication, regardless of which auth path was taken in process_startup:
+
+  1. No-auth path (AllowAllAuthenticator): require_authentication() = false,
+     connection is immediately marked ready.
+
+  2. SASL/password path (PasswordAuthenticator): client sends AUTH_RESPONSE,
+     process_auth_response() marks the connection ready.
+
+  3. Certificate-bypass path (CertificateAuthenticator with
+     transport_early_auth_bypass injection): authenticate(session_dn_func)
+     returns a user immediately, process_startup() marks the connection ready
+     without a SASL round-trip.
+
+     This is the path introduced by commit 20e9619bb1 that was missing the
+     _ready = true / update_scheduling_group() / on_connection_ready() calls.
+"""
+
+import logging
+import time
+import pytest
+from cassandra.auth import PlainTextAuthProvider
+from test.pylib.manager_client import ManagerClient
+from test.pylib.util import wait_for_cql_and_get_hosts, wait_for
+
+
+logger = logging.getLogger(__name__)
+
+
+def make_server_config(auth_type: str) -> dict:
+    """Return the scylla config dict for a given auth type."""
+    if auth_type == "allow_all":
+        return {
+            "authenticator": "AllowAllAuthenticator",
+            "authorizer": "AllowAllAuthorizer",
+        }
+    if auth_type == "password":
+        return {
+            "authenticator": "PasswordAuthenticator",
+            "authorizer": "CassandraAuthorizer",
+        }
+    if auth_type == "cert_bypass":
+        # CertificateAuthenticator with transport_early_auth_bypass active
+        # from the very first connection (including the server-add readiness
+        # check), so no TLS socket is required.  The injection makes
+        # certificate_authenticator::authenticate() return the cassandra
+        # superuser immediately, bypassing the TLS certificate check.
+        return {
+            "authenticator": "CertificateAuthenticator",
+            "authorizer": "CassandraAuthorizer",
+            # Minimal valid config; regex is never reached due to injection.
+            "auth_certificate_role_queries": [{"source": "SUBJECT", "query": "CN=(.+)"}],
+            "error_injections_at_startup": [
+                {"name": "transport_early_auth_bypass", "value": "cassandra"},
+            ],
+        }
+    raise ValueError(f"Unknown auth_type: {auth_type!r}")
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("auth_type", [
+    "allow_all",
+    "password",
+    "cert_bypass",
+])
+@pytest.mark.skip_mode("release", reason="error injections are not supported in release mode")
+async def test_connection_stage_ready_after_auth(manager: ManagerClient, auth_type: str) -> None:
+    """After a successful authentication via any code path in process_startup,
+    the connection must be reported as READY in system.clients."""
+    server = await manager.server_add(config=make_server_config(auth_type))
+
+    # CertificateAuthenticator bypassed by injection returns a user without a
+    # SASL challenge, so the driver must not attempt a password handshake.
+    auth_provider = PlainTextAuthProvider(username="cassandra", password="cassandra") \
+        if auth_type == "password" else None
+
+    await manager.driver_connect(server=server, auth_provider=auth_provider)
+    cql = manager.get_cql()
+    await wait_for_cql_and_get_hosts(cql, [server], time.time() + 60)
+
+    async def all_connections_ready():
+        rows = list(cql.execute(
+            "SELECT connection_stage FROM system.clients WHERE client_type = 'cql' ALLOW FILTERING"
+        ))
+        if not rows:
+            return None
+        if any(r.connection_stage != "READY" for r in rows):
+            return None
+        return rows
+
+    rows = await wait_for(all_connections_ready, time.time() + 30)
+    assert rows, "No CQL connections found in system.clients"
+    logger.info("auth_type=%s: all %d connection(s) are READY", auth_type, len(rows))

transport/server.cc

@@ -1405,6 +1405,10 @@ future<std::unique_ptr<cql_server::response>> cql_server::connection::process_st
             client_state.set_login(std::move(*opt_user));
             co_await client_state.check_user_can_login();
             client_state.maybe_update_per_service_level_params();
+            update_scheduling_group();
+            _authenticating = false;
+            _ready = true;
+            on_connection_ready();
             res = make_ready(stream, trace_state);
         } else {
             res = make_autheticate(stream, a.qualified_java_name(), trace_state);