CORE-14880: Migrate "OIDC" Admin API v1 Security Endpoints to v2 ConnectRPC API

proto/redpanda/core/admin/v2/BUILD

@@ -96,3 +96,25 @@ redpanda_proto_library(
         "@abseil-cpp//absl/time:time",
     ],
 )
+
+proto_library(
+    name = "security_proto",
+    srcs = ["security.proto"],
+    visibility = ["//visibility:public"],
+    deps = [
+        "//proto/redpanda/core/pbgen:options_proto",
+        "//proto/redpanda/core/pbgen:rpc_proto",
+        "@protobuf//:timestamp_proto",
+    ],
+)
+
+redpanda_proto_library(
+    name = "security_redpanda_proto",
+    protos = [
+        ":security_proto",
+    ],
+    visibility = ["//visibility:public"],
+    deps = [
+        "@abseil-cpp//absl/time:time",
+    ],
+)

proto/redpanda/core/admin/v2/security.proto

@@ -0,0 +1,96 @@
+// Copyright 2025 Redpanda Data, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+syntax = "proto3";
+
+package redpanda.core.admin.v2;
+
+import "proto/redpanda/core/pbgen/options.proto";
+import "proto/redpanda/core/pbgen/rpc.proto";
+import "google/protobuf/timestamp.proto";
+
+option (pbgen.cpp_namespace) = "proto::admin";
+
+// The SecurityService provides security-related operations.
+service SecurityService {
+    // ResolveOidcIdentity
+    //
+    // Validate an `Authorization` header `Bearer` token and return the mapped
+    // principal and token expiry time.
+    rpc ResolveOidcIdentity(ResolveOidcIdentityRequest)
+        returns (ResolveOidcIdentityResponse) {
+        option (pbgen.rpc) = {
+            authz: USER
+        };
+    }
+
+    // RefreshOidcKeys
+    //
+    // Refresh cached OpenID Connect (OIDC) signing keys across the cluster so
+    // newly issued or rotated keys are recognized.
+    rpc RefreshOidcKeys(RefreshOidcKeysRequest)
+        returns (RefreshOidcKeysResponse) {
+        option (pbgen.rpc) = {
+            authz: SUPERUSER
+        };
+    }
+
+    // RevokeOidcSessions
+    //
+    // Refresh cached OpenID Connect (OIDC) signing keys and revoke all active
+    // OIDC sessions across the cluster. Clients authenticated with
+    // `OAUTHBEARER` must re-authenticate.
+    rpc RevokeOidcSessions(RevokeOidcSessionsRequest)
+        returns (RevokeOidcSessionsResponse) {
+        option (pbgen.rpc) = {
+            authz: SUPERUSER
+        };
+    }
+}
+
+// =============================================
+/* Resources */
+// =============================================
+
+// =============================================
+/* RPC Requests and Responses */
+// =============================================
+
+// ResolveOidcIdentityRequest is the request for the ResolveOidcIdentity RPC.
+message ResolveOidcIdentityRequest {}
+
+// ResolveOidcIdentityResponse is the response from the ResolveOidcIdentity RPC.
+message ResolveOidcIdentityResponse {
+    // The principal resolved from the OIDC token.
+    string principal = 1;
+
+    // The timestamp of the token's expiry.
+    google.protobuf.Timestamp expire = 2;
+}
+
+// RefreshOidcKeysRequest is the request for the RefreshOidcKeys RPC.
+message RefreshOidcKeysRequest {}
+
+// RefreshOidcKeysResponse is the response from the RefreshOidcKeys RPC.
+message RefreshOidcKeysResponse {}
+
+// RevokeOidcSessionsRequest is the request for the RevokeOidcSessions RPC.
+message RevokeOidcSessionsRequest {}
+
+// RevokeOidcSessionsResponse is the response from the RevokeOidcSessions RPC.
+message RevokeOidcSessionsResponse {}
+
+// =============================================
+/* Other Messages */
+// =============================================

src/v/redpanda/BUILD

@@ -78,6 +78,7 @@ redpanda_cc_library(
         "//src/v/redpanda/admin/proxy:client",
         "//src/v/redpanda/admin/proxy:service",
         "//src/v/redpanda/admin/services:cluster",
+        "//src/v/redpanda/admin/services:security",
         "//src/v/redpanda/admin/services/datalake",
         "//src/v/redpanda/admin/services/internal:breakglass",
         "//src/v/redpanda/admin/services/internal:debug",

src/v/redpanda/admin/services/BUILD

@@ -46,3 +46,22 @@ redpanda_cc_library(
         "@seastar",
     ],
 )
+
+redpanda_cc_library(
+    name = "security",
+    srcs = ["security.cc"],
+    hdrs = ["security.h"],
+    implementation_deps = [
+        "//src/v/redpanda/admin/proxy:context",
+        "//src/v/security",
+        "//src/v/security:request_auth",
+    ],
+    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

@@ -0,0 +1,144 @@
+/*
+ * Copyright 2025 Redpanda Data, Inc.
+ *
+ * Use of this software is governed by the Business Source License
+ * included in the file licenses/BSL.md
+ *
+ * As of the Change Date specified in that file, in accordance with
+ * the Business Source License, use of this software will be governed
+ * by the Apache License, Version 2.0
+ */
+
+#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"
+#include "security/request_auth.h"
+
+#include <seastar/core/coroutine.hh>
+
+namespace admin {
+
+namespace {
+// NOLINTNEXTLINE(*-non-const-global-variables,cert-err58-*)
+ss::logger securitylog{"admin_api_server/security_service"};
+} // namespace
+
+security_service_impl::security_service_impl(
+  admin::proxy::client proxy_client,
+  cluster::controller* controller,
+  ss::sharded<kafka::server>& kafka_server)
+  : _proxy_client(std::move(proxy_client))
+  , _controller(controller)
+  , _kafka_server(kafka_server) {}
+
+seastar::future<proto::admin::resolve_oidc_identity_response>
+security_service_impl::resolve_oidc_identity(
+  serde::pb::rpc::context ctx, proto::admin::resolve_oidc_identity_request) {
+    const auto* auth_result = ctx.get_optional_value<request_auth_result>();
+
+    if (auth_result == nullptr) {
+        vlog(securitylog.warn, "No request_auth_result found in context");
+        throw serde::pb::rpc::failed_precondition_exception(
+          "No authentication result found");
+    }
+
+    auto& sasl_mechanism = auth_result->get_sasl_mechanism();
+    if (sasl_mechanism != security::oidc::sasl_authenticator::name) {
+        vlog(
+          securitylog.warn, "SASL mechanism is not OIDC: {}", sasl_mechanism);
+        throw serde::pb::rpc::failed_precondition_exception(
+          "SASL mechanism is not OIDC");
+    }
+
+    proto::admin::resolve_oidc_identity_response resp;
+    resp.set_principal(ss::sstring(auth_result->get_username()));
+
+    const auto& bearer = auth_result->get_password();
+    if (!bearer.starts_with(authz_bearer_prefix)) {
+        vlog(securitylog.warn, "Invalid OIDC bearer token format: {}", bearer);
+        throw serde::pb::rpc::unauthenticated_exception(
+          "Invalid OIDC bearer token format");
+    }
+
+    security::oidc::authenticator auth{_controller->get_oidc_service().local()};
+    auto res = auth.authenticate(bearer.substr(authz_bearer_prefix.length()));
+
+    if (res.has_error() || !res.has_value()) {
+        vlog(
+          securitylog.warn,
+          "Failed to authenticate OIDC token: {}",
+          res.has_error() ? res.error().message() : "unknown");
+
+        throw serde::pb::rpc::unauthenticated_exception(
+          "Failed to authenticate OIDC token");
+    }
+
+    // Convert ss::lowres_system_clock::time_point to absl::Time
+    resp.set_expire(
+      absl::FromChrono(
+        std::chrono::system_clock::time_point{
+          res.assume_value().expiry.time_since_epoch()}));
+
+    co_return resp;
+}
+
+seastar::future<proto::admin::refresh_oidc_keys_response>
+security_service_impl::refresh_oidc_keys(
+  serde::pb::rpc::context ctx, proto::admin::refresh_oidc_keys_request) {
+    vlog(securitylog.debug, "Refreshing OIDC keys.");
+
+    co_await _controller->get_oidc_service().invoke_on_all(
+      [](security::oidc::service& s) { return s.refresh_keys(); });
+
+    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 refresh_oidc_keys to {}", node_id);
+            co_await client.refresh_oidc_keys(ctx, {});
+        }
+    }
+
+    co_return proto::admin::refresh_oidc_keys_response{};
+}
+
+seastar::future<proto::admin::revoke_oidc_sessions_response>
+security_service_impl::revoke_oidc_sessions(
+  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

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2025 Redpanda Data, Inc.
+ *
+ * Use of this software is governed by the Business Source License
+ * included in the file licenses/BSL.md
+ *
+ * As of the Change Date specified in that file, in accordance with
+ * the Business Source License, use of this software will be governed
+ * by the Apache License, Version 2.0
+ */
+
+#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"
+
+namespace admin {
+
+class security_service_impl : public proto::admin::security_service {
+public:
+    security_service_impl(
+      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(
+        serde::pb::rpc::context,
+        proto::admin::resolve_oidc_identity_request) override;
+    seastar::future<proto::admin::refresh_oidc_keys_response> refresh_oidc_keys(
+      serde::pb::rpc::context,
+      proto::admin::refresh_oidc_keys_request) override;
+    seastar::future<proto::admin::revoke_oidc_sessions_response>
+      revoke_oidc_sessions(
+        serde::pb::rpc::context,
+        proto::admin::revoke_oidc_sessions_request) override;
+
+private:
+    admin::proxy::client _proxy_client;
+    cluster::controller* _controller;
+    ss::sharded<kafka::server>& _kafka_server;
+};
+
+} // namespace admin

src/v/redpanda/application_admin.cc

@@ -18,6 +18,7 @@
 #include "redpanda/admin/services/internal/debug.h"
 #include "redpanda/admin/services/internal/metastore.h"
 #include "redpanda/admin/services/internal/shadow_link_internal.h"
+#include "redpanda/admin/services/security.h"
 #include "redpanda/admin/services/shadow_link/shadow_link.h"
 #include "redpanda/application.h"
 #include "resource_mgmt/memory_groups.h"
@@ -105,6 +106,9 @@ void application::configure_admin_server(model::node_id node_id) {
             std::make_unique<
               admin::internal::shadow_link_internal_service_impl>(
               create_client(), &_cluster_link_service, &metadata_cache));
+          s.add_service(
+            std::make_unique<admin::security_service_impl>(
+              create_client(), controller.get(), _kafka_server.ref()));
       })
       .get();
 }