pandaproxy: Support SCRAM-SHA-512 for basic_auth

src/v/pandaproxy/auth_utils.h

@@ -34,7 +34,9 @@ inline credential_t maybe_authenticate_request(
         // did not give the authorization header.
         auth_result.require_authenticated();
         user = credential_t{
-          auth_result.get_username(), auth_result.get_password()};
+          auth_result.get_username(),
+          auth_result.get_password(),
+          auth_result.get_sasl_mechanism()};
     }
 
     return user;

src/v/pandaproxy/kafka_client_cache.cc

@@ -54,7 +54,7 @@ client_ptr kafka_client_cache::make_client(
     // Set the principal when the request is using HTTP Basic AuthN
     if (authn_method == config::rest_authn_method::http_basic) {
         // Need to specify type or else bad any_cast runtime error
-        cfg.sasl_mechanism.set_value(ss::sstring{"SCRAM-SHA-256"});
+        cfg.sasl_mechanism.set_value(std::move(user.sasl_mechanism));
         cfg.scram_username.set_value(std::move(user.name));
         cfg.scram_password.set_value(std::move(user.pass));
     }

src/v/pandaproxy/server.h

@@ -258,7 +258,9 @@ class auth_ctx_server : public ctx_server<service_t> {
                 // did not give the authorization header.
                 auth_result.require_authenticated();
                 user = credential_t{
-                  auth_result.get_username(), auth_result.get_password()};
+                  auth_result.get_username(),
+                  auth_result.get_password(),
+                  auth_result.get_sasl_mechanism()};
             }
         }
 

src/v/pandaproxy/test/kafka_client_cache.cc

@@ -13,6 +13,7 @@
 #include "config/rest_authn_endpoint.h"
 #include "kafka/client/configuration.h"
 #include "pandaproxy/types.h"
+#include "security/scram_authenticator.h"
 #include "ssx/future-util.h"
 
 #include <seastar/core/gate.hh>
@@ -60,9 +61,10 @@ struct test_client_cache : public kafka_client_cache {
 
 namespace pp = pandaproxy;
 
-SEASTAR_THREAD_TEST_CASE(cache_make_client) {
+SEASTAR_THREAD_TEST_CASE(cache_make_client_scram_sha_256) {
     pp::test_client_cache client_cache{10};
-    pp::credential_t user{"red", "panda"};
+    pp::credential_t user{
+      "red", "panda", security::scram_sha256_authenticator::name};
 
     {
         // Creating a client with no authn methods results in a kafka
@@ -87,9 +89,38 @@ SEASTAR_THREAD_TEST_CASE(cache_make_client) {
     }
 }
 
+SEASTAR_THREAD_TEST_CASE(cache_make_client_scram_sha_512) {
+    pp::test_client_cache client_cache{10};
+    pp::credential_t user{
+      "red", "panda", security::scram_sha512_authenticator::name};
+
+    {
+        // Creating a client with no authn methods results in a kafka
+        // client without a principal
+        pp::client_ptr client = client_cache.make_client(
+          user, config::rest_authn_method::none);
+        BOOST_TEST(client->config().sasl_mechanism.value() == "");
+        BOOST_TEST(client->config().scram_username.value() == "");
+        BOOST_TEST(client->config().scram_password.value() == "");
+    }
+
+    {
+        // Creating a client with http_basic authn type results
+        // in a kafka client with a principal
+        pp::client_ptr client = client_cache.make_client(
+          user, config::rest_authn_method::http_basic);
+        BOOST_TEST(
+          client->config().sasl_mechanism.value()
+          == ss::sstring{"SCRAM-SHA-512"});
+        BOOST_TEST(client->config().scram_username.value() == user.name);
+        BOOST_TEST(client->config().scram_password.value() == user.pass);
+    }
+}
+
 SEASTAR_THREAD_TEST_CASE(cache_fetch_or_insert) {
     size_t s = 1, max_s = 1;
-    pp::credential_t user{"red", "panda"};
+    pp::credential_t user{
+      "red", "panda", security::scram_sha256_authenticator::name};
     pp::test_client_cache client_cache{s};
     BOOST_TEST(client_cache.size() == 0);
     BOOST_TEST(client_cache.max_size() == max_s);
@@ -144,7 +175,8 @@ SEASTAR_THREAD_TEST_CASE(cache_fetch_or_insert) {
 }
 
 SEASTAR_THREAD_TEST_CASE(test_keep_alive) {
-    pp::credential_t user{"red", "panda"};
+    pp::credential_t user{
+      "red", "panda", security::scram_sha256_authenticator::name};
     size_t cache_max_size = 5;
     std::chrono::milliseconds keep_alive{50ms};
 

src/v/pandaproxy/types.h

@@ -54,11 +54,13 @@ struct timestamped_user {
 struct credential_t {
     ss::sstring name;
     ss::sstring pass;
+    ss::sstring sasl_mechanism;
 
     credential_t() = default;
-    credential_t(ss::sstring n, ss::sstring p)
+    credential_t(ss::sstring n, ss::sstring p, ss::sstring sasl_mechanism)
       : name{std::move(n)}
-      , pass{std::move(p)} {}
+      , pass{std::move(p)}
+      , sasl_mechanism{std::move(sasl_mechanism)} {}
 };
 
 } // namespace pandaproxy

src/v/utils/request_auth.cc

@@ -15,8 +15,11 @@
 #include "seastar/http/exception.hh"
 #include "security/credential_store.h"
 #include "security/scram_algorithm.h"
+#include "security/scram_authenticator.h"
 #include "vlog.h"
 
+#include <seastar/core/sstring.hh>
+
 static ss::logger logger{"request_auth"};
 
 request_authenticator::request_authenticator(
@@ -112,11 +115,23 @@ request_auth_result request_authenticator::do_authenticate(
               "Unauthorized", ss::http::reply::status_type::unauthorized);
         } else {
             const auto& cred = cred_opt.value();
-            bool is_valid = (
-              security::scram_sha256::validate_password(
-                password, cred.stored_key(), cred.salt(), cred.iterations())
-              || security::scram_sha512::validate_password(
-                password, cred.stored_key(), cred.salt(), cred.iterations()));
+            ss::sstring sasl_mechanism;
+            bool is_valid{false};
+            if (security::scram_sha256::validate_password(
+                  password,
+                  cred.stored_key(),
+                  cred.salt(),
+                  cred.iterations())) {
+                is_valid = true;
+                sasl_mechanism = security::scram_sha256_authenticator::name;
+            } else if (security::scram_sha512::validate_password(
+                         password,
+                         cred.stored_key(),
+                         cred.salt(),
+                         cred.iterations())) {
+                is_valid = true;
+                sasl_mechanism = security::scram_sha512_authenticator::name;
+            }
             if (!is_valid) {
                 // User found, password doesn't match
                 vlog(
@@ -134,6 +149,7 @@ request_auth_result request_authenticator::do_authenticate(
                 return request_auth_result(
                   std::move(username),
                   std::move(password),
+                  std::move(sasl_mechanism),
                   request_auth_result::superuser(superuser));
             }
         }

src/v/utils/request_auth.h

@@ -43,9 +43,11 @@ class [[nodiscard]] request_auth_result {
     request_auth_result(
       security::credential_user username,
       security::credential_password password,
+      ss::sstring sasl_mechanism,
       superuser is_superuser)
       : _username(std::move(username))
       , _password(std::move(password))
+      , _sasl_mechanism(std::move(sasl_mechanism))
       , _authenticated(true)
       , _superuser(is_superuser){};
 
@@ -77,10 +79,12 @@ class [[nodiscard]] request_auth_result {
 
     ss::sstring const& get_username() const { return _username; }
     ss::sstring const& get_password() const { return _password; }
+    ss::sstring const& get_sasl_mechanism() const { return _sasl_mechanism; }
 
 private:
     security::credential_user _username;
     security::credential_password _password;
+    ss::sstring _sasl_mechanism;
     bool _authenticated{false};
     bool _superuser{false};
     bool _checked{false};