security/oidc: Support oidc for kafka api

Signed-off-by: Ben Pope <[email protected]>

Author: BenPope

src/v/config/configuration.cc

@@ -1110,7 +1110,8 @@ configuration::configuration()
   , sasl_mechanisms(
       *this,
       "sasl_mechanisms",
-      "A list of supported SASL mechanisms. `SCRAM` and `GSSAPI` are allowed.",
+      "A list of supported SASL mechanisms. `SCRAM`, `GSSAPI`, and "
+      "`OAUTHBEARER` are allowed.",
       {.needs_restart = needs_restart::no, .visibility = visibility::user},
       {"SCRAM"},
       validate_sasl_mechanisms)

src/v/config/validators.cc

@@ -109,7 +109,7 @@ std::optional<ss::sstring> validate_client_groups_byte_rate_quota(
 std::optional<ss::sstring>
 validate_sasl_mechanisms(const std::vector<ss::sstring>& mechanisms) {
     static const absl::flat_hash_set<std::string_view> supported{
-      "GSSAPI", "SCRAM"};
+      "GSSAPI", "SCRAM", "OAUTHBEARER"};
 
     // Validate results
     for (const auto& m : mechanisms) {

src/v/kafka/server/server.cc

@@ -52,6 +52,7 @@
 #include "security/exceptions.h"
 #include "security/gssapi_authenticator.h"
 #include "security/mtls.h"
+#include "security/oidc_authenticator.h"
 #include "security/scram_algorithm.h"
 #include "security/scram_authenticator.h"
 #include "ssx/future-util.h"
@@ -567,6 +568,18 @@ ss::future<response_ptr> sasl_handshake_handler::handle(
         }
     }
 
+    if (supports(security::oidc::sasl_authenticator::name)) {
+        supported_sasl_mechanisms.emplace_back(
+          security::oidc::sasl_authenticator::name);
+
+        if (
+          request.data.mechanism == security::oidc::sasl_authenticator::name) {
+            ctx.sasl()->set_mechanism(
+              std::make_unique<security::oidc::sasl_authenticator>(
+                ctx.connection()->server().oidc_service().local()));
+        }
+    }
+
     if (!ctx.sasl()->has_mechanism()) {
         error = error_code::unsupported_sasl_mechanism;
     }

tests/rptest/tests/redpanda_oauth_test.py

@@ -13,8 +13,6 @@
 from rptest.services.redpanda import LoggingConfig, SecurityConfig, make_redpanda_service
 from rptest.services.keycloak import KeycloakService
 from rptest.services.cluster import cluster
-from rptest.util import expect_exception
-from confluent_kafka import KafkaException
 
 CLIENT_ID = 'myapp'
 TOKEN_AUDIENCE = 'account'
@@ -55,10 +53,15 @@ def __init__(self,
         security.enable_sasl = True
         security.sasl_mechanisms = sasl_mechanisms
 
-        self.redpanda = make_redpanda_service(test_context,
-                                              num_brokers,
-                                              security=security,
-                                              log_config=log_config)
+        self.redpanda = make_redpanda_service(
+            test_context,
+            num_brokers,
+            extra_rp_conf={
+                "oidc_discovery_url": self.keycloak.get_discovery_url(kc_node),
+                "oidc_token_audience": TOKEN_AUDIENCE,
+            },
+            security=security,
+            log_config=log_config)
 
         self.su_username, self.su_password, self.su_algorithm = self.redpanda.SUPERUSER_CREDENTIALS
 
@@ -112,5 +115,4 @@ def test_init(self):
         # metadata requests to behave nicely.
         producer.poll(0.0)
 
-        with expect_exception(KafkaException, lambda _: True):
-            producer.list_topics(timeout=5)
+        producer.list_topics(timeout=5)