admin: implement OIDC session revocation in v2 API

Implement RevokeOidcSessions RPC to refresh OIDC keys and revoke all
active OIDC sessions across the cluster. The receiving broker
refreshes OIDC keys on all local shards, revokes OIDC SASL
sessions with kafka::server, and broadcasts the request to all
other nodes in the cluster, with each broker performing local
key refresh and session revocation. The originating node aggregates
errors across all brokers and returns an aggregated error if any
node fails, otherwise success.

Add comprehensive test coverage validating session revocation with
various authentication scenarios. This enables administrators to
immediately invalidate all OIDC sessions when security policies
change or credentials are compromised.

Author: nguyen-andrew

src/v/redpanda/admin/services/BUILD

@@ -59,6 +59,7 @@ redpanda_cc_library(
     deps = [
         "//proto/redpanda/core/admin/v2:security_redpanda_proto",
         "//src/v/cluster",
+        "//src/v/kafka/server",
         "//src/v/redpanda/admin/proxy:client",
         "//src/v/serde/protobuf:rpc",
         "@seastar",

src/v/redpanda/admin/services/security.cc

@@ -11,6 +11,7 @@
 
 #include "redpanda/admin/services/security.h"
 
+#include "kafka/server/server.h"
 #include "redpanda/admin/proxy/context.h"
 #include "security/oidc_authenticator.h"
 #include "security/oidc_service.h"
@@ -26,9 +27,12 @@ ss::logger securitylog{"admin_api_server/security_service"};
 } // namespace
 
 security_service_impl::security_service_impl(
-  admin::proxy::client proxy_client, cluster::controller* controller)
+  admin::proxy::client proxy_client,
+  cluster::controller* controller,
+  ss::sharded<kafka::server>& kafka_server)
   : _proxy_client(std::move(proxy_client))
-  , _controller(controller) {}
+  , _controller(controller)
+  , _kafka_server(kafka_server) {}
 
 seastar::future<proto::admin::resolve_oidc_identity_response>
 security_service_impl::resolve_oidc_identity(
@@ -108,9 +112,33 @@ security_service_impl::refresh_oidc_keys(
 
 seastar::future<proto::admin::revoke_oidc_sessions_response>
 security_service_impl::revoke_oidc_sessions(
-  serde::pb::rpc::context, proto::admin::revoke_oidc_sessions_request) {
-    throw serde::pb::rpc::unimplemented_exception(
-      "revoke_oidc_sessions is not implemented");
+  serde::pb::rpc::context ctx, proto::admin::revoke_oidc_sessions_request) {
+    vlog(securitylog.debug, "Refreshing OIDC keys and revoking OIDC sessions");
+
+    co_await _controller->get_oidc_service().invoke_on_all(
+      [](security::oidc::service& s) { return s.refresh_keys(); });
+
+    co_await _kafka_server.invoke_on_all([](kafka::server& ks) {
+        return ks.revoke_credentials(security::oidc::sasl_authenticator::name);
+    });
+
+    if (!proxy::is_proxied(ctx)) {
+        vlog(securitylog.debug, "Broadcasting request to other nodes");
+
+        auto clients = _proxy_client.make_clients_for_other_nodes<
+          proto::admin::security_service_client>();
+
+        for (auto& client_pair : clients) {
+            auto& [node_id, client] = client_pair;
+            vlog(
+              securitylog.trace,
+              "Proxying revoke_oidc_sessions to {}",
+              node_id);
+            co_await client.revoke_oidc_sessions(ctx, {});
+        }
+    }
+
+    co_return proto::admin::revoke_oidc_sessions_response{};
 }
 
 } // namespace admin

src/v/redpanda/admin/services/security.h

@@ -12,6 +12,7 @@
 #pragma once
 
 #include "cluster/controller.h"
+#include "kafka/server/fwd.h"
 #include "proto/redpanda/core/admin/v2/security.proto.h"
 #include "redpanda/admin/proxy/client.h"
 
@@ -20,7 +21,9 @@ namespace admin {
 class security_service_impl : public proto::admin::security_service {
 public:
     security_service_impl(
-      admin::proxy::client proxy_client, cluster::controller* controller);
+      admin::proxy::client proxy_client,
+      cluster::controller* controller,
+      ss::sharded<kafka::server>& kafka_server);
 
     seastar::future<proto::admin::resolve_oidc_identity_response>
       resolve_oidc_identity(
@@ -37,6 +40,7 @@ class security_service_impl : public proto::admin::security_service {
 private:
     admin::proxy::client _proxy_client;
     cluster::controller* _controller;
+    ss::sharded<kafka::server>& _kafka_server;
 };
 
 } // namespace admin

src/v/redpanda/application_admin.cc

@@ -108,7 +108,7 @@ void application::configure_admin_server(model::node_id node_id) {
               create_client(), &_cluster_link_service, &metadata_cache));
           s.add_service(
             std::make_unique<admin::security_service_impl>(
-              create_client(), controller.get()));
+              create_client(), controller.get(), _kafka_server.ref()));
       })
       .get();
 }

tests/rptest/tests/redpanda_oauth_test.py

@@ -203,6 +203,20 @@ def get_idp_request_count(self, nodes: list[ClusterNode]):
             )
         return result["security_idp_latency_seconds_count"]
 
+    def get_sasl_session_revoked_total(self):
+        metrics = [
+            "kafka_rpc_sasl_session_revoked_total",
+        ]
+        samples = self.redpanda.metrics_samples(
+            metrics, self.redpanda.nodes, MetricsEndpoint.METRICS
+        )
+        result = {}
+        for k in samples.keys():
+            result[k] = result.get(k, 0) + sum(
+                [int(s.value) for s in samples[k].samples]
+            )
+        return result["kafka_rpc_sasl_session_revoked_total"]
+
 
 class RedpandaOIDCTestMethods(RedpandaOIDCTestBase):
     def __init__(self, test_context, **kwargs):
@@ -541,20 +555,6 @@ def test_admin_revoke(self):
 
         kc_node = self.keycloak.nodes[0]
 
-        def get_sasl_session_revoked_total():
-            metrics = [
-                "kafka_rpc_sasl_session_revoked_total",
-            ]
-            samples = self.redpanda.metrics_samples(
-                metrics, self.redpanda.nodes, MetricsEndpoint.METRICS
-            )
-            result = {}
-            for k in samples.keys():
-                result[k] = result.get(k, 0) + sum(
-                    [int(s.value) for s in samples[k].samples]
-                )
-            return result["kafka_rpc_sasl_session_revoked_total"]
-
         client_id = CLIENT_ID
         service_user_id = self.create_service_user(client_id)
         cfg = self.keycloak.generate_oauth_config(kc_node, client_id)
@@ -642,7 +642,7 @@ def has_group():
         t1 = threading.Thread(target=consume_one)
         t1.start()
 
-        revoked_total = get_sasl_session_revoked_total()
+        revoked_total = self.get_sasl_session_revoked_total()
 
         time.sleep(5)
 
@@ -659,9 +659,142 @@ def has_group():
         self.redpanda.logger.debug("joined consumer thread")
 
         wait_until(
-            lambda: revoked_total < get_sasl_session_revoked_total(),
+            lambda: revoked_total < self.get_sasl_session_revoked_total(),
+            timeout_sec=10,
+            backoff_sec=1,
+        )
+
+        consumer.close()
+
+    @cluster(num_nodes=4)
+    def test_admin_v2_revoke_oidc_sessions(self):
+        FETCH_TIMEOUT_SEC = 10
+        GROUP_ID = "test_admin_revoke"
+
+        kc_node = self.keycloak.nodes[0]
+
+        client_id = CLIENT_ID
+        service_user_id = self.create_service_user(client_id)
+        cfg = self.keycloak.generate_oauth_config(kc_node, client_id)
+        token = self.get_client_credentials_token(cfg)
+
+        def request_revoke_oidc_sessions(
+            with_auth: bool,
+        ) -> security_pb2.RevokeOidcSessionsResponse:
+            admin_v2 = AdminV2(self.redpanda)
+            req = security_pb2.RevokeOidcSessionsRequest()
+            return admin_v2.security().revoke_oidc_sessions(
+                req,
+                extra_headers={"Authorization": f"Bearer {token['access_token']}"}
+                if with_auth
+                else None,
+            )
+
+        # At this point, admin API does not require auth and service_user_id is not a superuser
+        # Both calls should succeed
+        request_revoke_oidc_sessions(with_auth=False)
+        request_revoke_oidc_sessions(with_auth=True)
+
+        # Require Auth for Admin
+        self.redpanda.set_cluster_config({"admin_api_require_auth": True})
+
+        with expect_exception(
+            ConnectError,
+            lambda e: e.code == ConnectErrorCode.PERMISSION_DENIED,
+        ):
+            request_revoke_oidc_sessions(with_auth=False)
+
+        with expect_exception(
+            ConnectError,
+            lambda e: e.code == ConnectErrorCode.PERMISSION_DENIED,
+        ):
+            request_revoke_oidc_sessions(with_auth=True)
+
+        # Add service_user_id as a superuser
+        self.redpanda.set_cluster_config(
+            {
+                "superusers": [
+                    self.redpanda.SUPERUSER_CREDENTIALS.username,
+                    service_user_id,
+                ]
+            }
+        )
+
+        self.rpk.create_topic(EXAMPLE_TOPIC)
+        expected_topics = set([EXAMPLE_TOPIC])
+        wait_until(
+            lambda: set(self.rpk.list_topics()) == expected_topics,
+            timeout_sec=10,
+            err_msg=f"Expected topics: {expected_topics}, got: {self.rpk.list_topics()}",
+        )
+
+        cfg = self.keycloak.generate_oauth_config(kc_node, client_id)
+        k_client = PythonLibrdkafka(
+            self.redpanda,
+            algorithm="OAUTHBEARER",
+            oauth_config=cfg,
+            tls_cert=self.client_cert,
+        )
+
+        consumer = k_client.get_consumer(extra_config={"group.id": GROUP_ID})
+        producer = k_client.get_producer()
+
+        self.redpanda.logger.debug("starting producer")
+        producer.poll(1.0)
+        wait_until(
+            lambda: set(producer.list_topics(timeout=10).topics.keys())
+            == expected_topics,
+            timeout_sec=5,
+            err_msg=f"Producer topics do not match expected topics: {expected_topics}",
+        )
+
+        def consume_one():
+            self.redpanda.logger.debug("starting consumer")
+            rec = consumer.poll(FETCH_TIMEOUT_SEC)
+            self.redpanda.logger.debug(f"consumed: {rec}")
+            return rec
+
+        def has_group():
+            groups = self.rpk.group_describe(group=GROUP_ID, summary=True)
+            return groups.members == 1 and groups.state == "Stable"
+
+        self.redpanda.logger.debug("starting consumer.subscribe")
+        consumer.subscribe([EXAMPLE_TOPIC])
+        self.redpanda.logger.debug("consumer.subscribed")
+        rec = consumer.poll(1.0)
+        assert rec is None, f"Expected no record, got: {rec}"
+
+        wait_until(
+            has_group,
+            timeout_sec=30,
+            backoff_sec=1,
+            err_msg="Consumer group did not reach expected state",
+        )
+
+        t1 = threading.Thread(target=consume_one)
+        t1.start()
+
+        revoked_total = self.get_sasl_session_revoked_total()
+
+        time.sleep(5)
+
+        self.redpanda.logger.debug("starting final revoke")
+        request_revoke_oidc_sessions(with_auth=True)
+
+        self.redpanda.logger.debug("starting producer")
+        producer.produce(topic=EXAMPLE_TOPIC, key="bar", value="23")
+        producer.flush(timeout=5)
+        self.redpanda.logger.debug("produced 1")
+
+        self.redpanda.logger.debug("joining consumer thread")
+        t1.join()
+        self.redpanda.logger.debug("joined consumer thread")
+
+        wait_until(
+            lambda: revoked_total < self.get_sasl_session_revoked_total(),
             timeout_sec=10,
             backoff_sec=1,
+            err_msg="Expected sasl_session_revoked_total to increase after OIDC sessions revoke",
         )
 
         consumer.close()