sr: add basic auth support to the schema registry

src/v/config/rest_authn_endpoint.cc

@@ -44,6 +44,15 @@ std::ostream& operator<<(std::ostream& os, const rest_authn_endpoint& ep) {
     return os;
 }
 
+rest_authn_method get_authn_method(
+  const std::vector<rest_authn_endpoint>& endpoints, size_t listener_idx) {
+    if (listener_idx >= endpoints.size()) {
+        return rest_authn_method::none;
+    }
+
+    return endpoints[listener_idx].authn_method.value_or(
+      rest_authn_method::none);
+}
 } // namespace config
 
 namespace YAML {

src/v/config/rest_authn_endpoint.h

@@ -24,6 +24,7 @@
 #include <iosfwd>
 #include <optional>
 #include <string>
+#include <vector>
 
 namespace config {
 
@@ -51,6 +52,17 @@ struct rest_authn_endpoint {
     operator<<(std::ostream& os, const rest_authn_endpoint& ep);
 };
 
+// A helper method that searches for the listener within a vector of
+// rest_authn_endpoints. Returns the authn method if the address is found.
+// Returns none type otherwise
+rest_authn_method get_authn_method(
+  const std::vector<rest_authn_endpoint>& endpoints, size_t listener_idx);
+
+struct always_true : public binding<bool> {
+    always_true()
+      : binding<bool>(true) {}
+};
+
 namespace detail {
 
 template<>

src/v/pandaproxy/auth_utils.h

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2022 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 "config/rest_authn_endpoint.h"
+#include "pandaproxy/types.h"
+#include "redpanda/request_auth.h"
+
+#include <seastar/http/request.hh>
+
+#include <vector>
+
+namespace pandaproxy {
+// Authenticate the request when HTTP Basic Auth is enabled and return the
+// authenticated user. Otherwise return the default user.
+inline credential_t maybe_authenticate_request(
+  config::rest_authn_method authn_method,
+  request_authenticator& authenticator,
+  const ss::httpd::request& req) {
+    credential_t user;
+
+    if (authn_method == config::rest_authn_method::http_basic) {
+        // Will throw 400 & 401 if auth fails
+        auto auth_result = authenticator.authenticate(req);
+        // Will throw 403 if user enabled HTTP Basic Auth but
+        // did not give the authorization header.
+        auth_result.require_authenticated();
+        user = credential_t{
+          auth_result.get_username(), auth_result.get_password()};
+    }
+
+    return user;
+}
+} // namespace pandaproxy

src/v/pandaproxy/rest/configuration.cc

@@ -83,15 +83,4 @@ configuration::configuration()
           }
           return msg;
       }) {}
-
-config::rest_authn_method configuration::authn_method(size_t listener_idx) {
-    const auto& pp_api = pandaproxy_api.value();
-
-    if (listener_idx < 0 || listener_idx >= pp_api.size()) {
-        return config::rest_authn_method::none;
-    }
-
-    return pp_api[listener_idx].authn_method.value_or(
-      config::rest_authn_method::none);
-}
 } // namespace pandaproxy::rest

src/v/pandaproxy/rest/configuration.h

@@ -41,12 +41,6 @@ struct configuration final : public config::config_store {
 
     configuration();
     explicit configuration(const YAML::Node& cfg);
-
-    // A helper method that searches for the listener within
-    // the list of pandaproxy_api endpoints.
-    // Returns the authn method if the address is found.
-    // Returns none type otherwise
-    config::rest_authn_method authn_method(size_t listener_idx);
 };
 
 } // namespace pandaproxy::rest

src/v/pandaproxy/rest/proxy.cc

@@ -9,16 +9,15 @@
 
 #include "pandaproxy/rest/proxy.h"
 
-#include "config/rest_authn_endpoint.h"
 #include "net/unresolved_address.h"
 #include "pandaproxy/api/api-doc/rest.json.h"
+#include "pandaproxy/auth_utils.h"
 #include "pandaproxy/logger.h"
 #include "pandaproxy/parsing/exceptions.h"
 #include "pandaproxy/parsing/from_chars.h"
 #include "pandaproxy/rest/configuration.h"
 #include "pandaproxy/rest/handlers.h"
 #include "pandaproxy/sharded_client_cache.h"
-#include "pandaproxy/types.h"
 #include "utils/gate_guard.h"
 
 #include <seastar/core/future-util.hh>
@@ -34,19 +33,12 @@ auto wrap(Handler h) {
     return [h{std::move(h)}](
              server::request_t rq,
              server::reply_t rp) -> ss::future<server::reply_t> {
-        rq.authn_method = rq.service().config().authn_method(
+        rq.authn_method = config::get_authn_method(
+          rq.service().config().pandaproxy_api.value(),
           rq.req->get_listener_idx());
 
-        if (rq.authn_method == config::rest_authn_method::http_basic) {
-            // Will throw 400 & 401 if auth fails
-            auto auth_result = rq.service().authenticator().authenticate(
-              *rq.req);
-            // Will throw 403 if user enabled HTTP Basic Auth but
-            // did not give the authorization header.
-            auth_result.require_authenticated();
-            rq.user = credential_t{
-              auth_result.get_username(), auth_result.get_password()};
-        }
+        rq.user = maybe_authenticate_request(
+          rq.authn_method, rq.service().authenticator(), *rq.req);
 
         return h(std::move(rq), std::move(rp));
     };
@@ -90,11 +82,6 @@ server::routes_t get_proxy_routes() {
     return routes;
 }
 
-struct always_true : public config::binding<bool> {
-    always_true()
-      : config::binding<bool>(true) {}
-};
-
 proxy::proxy(
   const YAML::Node& config,
   ss::smp_service_group smp_sg,
@@ -115,7 +102,7 @@ proxy::proxy(
       "/definitions",
       _ctx,
       json::serialization_format::application_json)
-  , _auth{always_true(), controller} {}
+  , _auth{config::always_true(), controller} {}
 
 ss::future<> proxy::start() {
     _server.routes(get_proxy_routes());

src/v/pandaproxy/schema_registry/service.cc

@@ -17,6 +17,7 @@
 #include "kafka/server/handlers/topics/types.h"
 #include "model/fundamental.h"
 #include "pandaproxy/api/api-doc/schema_registry.json.h"
+#include "pandaproxy/auth_utils.h"
 #include "pandaproxy/error.h"
 #include "pandaproxy/logger.h"
 #include "pandaproxy/schema_registry/configuration.h"
@@ -30,6 +31,7 @@
 #include <seastar/core/memory.hh>
 #include <seastar/core/std-coroutine.hh>
 #include <seastar/http/api_docs.hh>
+#include <seastar/http/exception.hh>
 
 namespace pandaproxy::schema_registry {
 
@@ -41,6 +43,13 @@ auto wrap(ss::gate& g, one_shot& os, Handler h) {
              server::request_t rq,
              server::reply_t rp) -> ss::future<server::reply_t> {
         auto h{_h};
+
+        rq.authn_method = config::get_authn_method(
+          rq.service().config().schema_registry_api.value(),
+          rq.req->get_listener_idx());
+        rq.user = maybe_authenticate_request(
+          rq.authn_method, rq.service().authenticator(), *rq.req);
+
         auto units = co_await os();
         auto guard = gate_guard(g);
         co_return co_await h(std::move(rq), std::move(rp));
@@ -228,7 +237,8 @@ service::service(
   , _store(store)
   , _writer(sequencer)
   , _controller(controller)
-  , _ensure_started{[this]() { return do_start(); }} {}
+  , _ensure_started{[this]() { return do_start(); }}
+  , _auth{config::always_true(), controller.get()} {}
 
 ss::future<> service::start() {
     static std::vector<model::broker_endpoint> not_advertised{};

src/v/pandaproxy/schema_registry/service.h

@@ -17,6 +17,7 @@
 #include "pandaproxy/schema_registry/sharded_store.h"
 #include "pandaproxy/schema_registry/util.h"
 #include "pandaproxy/server.h"
+#include "redpanda/request_auth.h"
 #include "seastarx.h"
 
 #include <seastar/core/future.hh>
@@ -51,6 +52,7 @@ class service {
     ss::sharded<kafka::client::client>& client() { return _client; }
     seq_writer& writer() { return _writer.local(); }
     sharded_store& schema_store() { return _store; }
+    request_authenticator& authenticator() { return _auth; }
 
 private:
     ss::future<> do_start();
@@ -67,6 +69,7 @@ class service {
     std::unique_ptr<cluster::controller>& _controller;
 
     one_shot _ensure_started;
+    request_authenticator _auth;
 };
 
 } // namespace pandaproxy::schema_registry

tests/rptest/services/redpanda.py

@@ -374,6 +374,7 @@ def __init__(self):
         self.tls_provider: Optional[TLSProvider] = None
         self.require_client_auth: bool = True
         self.pp_authn_method: Optional[str] = None
+        self.sr_authn_method: Optional[str] = None
 
         # The rules to extract principal from mtls
         self.principal_mapping_rules = self.__DEFAULT_PRINCIPAL_MAPPING_RULES
@@ -1592,7 +1593,8 @@ def write_node_conf_file(self, node, override_cfg_params=None):
                            superuser=self._superuser,
                            sasl_enabled=self.sasl_enabled(),
                            endpoint_authn_method=self.endpoint_authn_method(),
-                           pp_authn_method=self.pp_authn_method())
+                           pp_authn_method=self._security.pp_authn_method,
+                           sr_authn_method=self._security.sr_authn_method)
 
         if override_cfg_params or self._extra_node_conf[node]:
             doc = yaml.full_load(conf)

tests/rptest/services/templates/redpanda.yaml

@@ -85,8 +85,27 @@ schema_registry:
   schema_registry_api:
     address: "{{node.account.hostname}}"
     port: 8081
+    {% if sr_authn_method %}
+    authentication_method: {{ sr_authn_method }}
+    {% endif %}
 
   api_doc_dir: {{root}}/usr/share/redpanda/proxy-api-doc
+
+schema_registry_client:
+  # kafka-api compatible brokers
+  brokers:
+  {% for node in nodes.values() %}
+    - address: "{{node.account.hostname}}"
+      port: 9092
+  {% endfor %}
+
+  retries: 10
+  retry_base_backoff_ms: 10
+  {% if sasl_enabled %}
+  sasl_mechanism: {{superuser[2]}}
+  scram_username: {{superuser[0]}}
+  scram_password: {{superuser[1]}}
+  {% endif %}
 {% endif %}
 
 rpk: