security: memoize scram password validation

validate_scram_credential now consults a shard-local
scram_credential_cache, skipping the salted password derivation for
repeat authentications. HTTP Basic auth (pandaproxy, schema registry,
admin API) and SASL/PLAIN pay that derivation on every request; in
production, unbatched HTTP produce through pandaproxy at a few hundred
requests per second saturated a core on it. Cache hits skip the
derivation entirely; the uncached path is unchanged.

The scram_credential_cache_enabled cluster config (default true) is an
operational kill switch.

Author: nguyen-andrew

src/v/config/configuration.cc

@@ -4178,6 +4178,16 @@ configuration::configuration()
       },
       std::vector<ss::sstring>{"BASIC"},
       validate_http_authn_mechanisms)
+  , scram_credential_cache_enabled(
+      *this,
+      "scram_credential_cache_enabled",
+      "Whether to cache SCRAM password validation results for authentication "
+      "paths that receive a plaintext password on every request (HTTP Basic "
+      "authentication and SASL/PLAIN). When enabled, repeat authentications "
+      "skip the salted password derivation, which costs thousands of HMAC "
+      "operations per validation.",
+      {.needs_restart = needs_restart::no, .visibility = visibility::tunable},
+      false)
   , enable_mpx_extensions(
       *this,
       "enable_mpx_extensions",

src/v/config/configuration.h

@@ -712,6 +712,7 @@ struct configuration final : public config_store {
 
     // HTTP Authentication
     enterprise<property<std::vector<ss::sstring>>> http_authentication;
+    property<bool> scram_credential_cache_enabled;
 
     // MPX
     property<bool> enable_mpx_extensions;

src/v/security/scram_authenticator.cc

@@ -11,10 +11,12 @@
 #include "security/scram_authenticator.h"
 
 #include "base/vlog.h"
+#include "config/configuration.h"
 #include "random/secure_generators.h"
 #include "security/credential_store.h"
 #include "security/errc.h"
 #include "security/logger.h"
+#include "security/scram_credential_cache.h"
 
 namespace security {
 
@@ -148,22 +150,68 @@ scram_authenticator<T>::authenticate(bytes auth_bytes) {
 template class scram_authenticator<scram_sha256>;
 template class scram_authenticator<scram_sha512>;
 
+namespace {
+
+// Matches a password against a stored credential by rederiving its stored
+// key. When a cache is given, the derivation is memoized; a null cache
+// derives every time (the plain, un-memoized path).
+template<typename scram>
+bool match_password(
+  scram_credential_cache* cache,
+  const scram_credential& cred,
+  const credential_password& password) {
+    constexpr auto mech = scram_mechanism_traits<scram>::algorithm;
+    if (cache != nullptr) {
+        auto cached = cache->get(
+          mech, password, cred.salt(), cred.iterations());
+        if (cached.has_value()) {
+            return *cached == cred.stored_key();
+        }
+    }
+    auto stored_key = scram::derive_stored_key(
+      password(), cred.salt(), cred.iterations());
+    auto valid = stored_key == cred.stored_key();
+    if (cache != nullptr) {
+        cache->put(
+          mech,
+          password,
+          cred.salt(),
+          cred.iterations(),
+          std::move(stored_key));
+    }
+    return valid;
+}
+} // namespace
+
+namespace detail {
+
 std::optional<std::string_view> validate_scram_credential(
-  const scram_credential& cred, const credential_password& password) {
+  const scram_credential& cred,
+  const credential_password& password,
+  scram_credential_cache* cache) {
     std::optional<std::string_view> sasl_mechanism;
     if (
-      cred.stored_key().size() == security::scram_sha256::key_size
-      && security::scram_sha256::validate_password(
-        password, cred.stored_key(), cred.salt(), cred.iterations())) {
-        sasl_mechanism = security::scram_sha256_authenticator::name;
+      cred.stored_key().size() == scram_sha256::key_size
+      && match_password<scram_sha256>(cache, cred, password)) {
+        sasl_mechanism = scram_sha256_authenticator::name;
     } else if (
-      cred.stored_key().size() == security::scram_sha512::key_size
-      && security::scram_sha512::validate_password(
-        password, cred.stored_key(), cred.salt(), cred.iterations())) {
-        sasl_mechanism = security::scram_sha512_authenticator::name;
+      cred.stored_key().size() == scram_sha512::key_size
+      && match_password<scram_sha512>(cache, cred, password)) {
+        sasl_mechanism = scram_sha512_authenticator::name;
     }
 
     return sasl_mechanism;
 }
+} // namespace detail
+
+std::optional<std::string_view> validate_scram_credential(
+  const scram_credential& cred, const credential_password& password) {
+    if (config::shard_local_cfg().scram_credential_cache_enabled()) {
+        static thread_local scram_credential_cache cache{
+          scram_credential_cache::default_capacity};
+        return detail::validate_scram_credential(cred, password, &cache);
+    }
+    return detail::validate_scram_credential(cred, password, nullptr);
+}
 
 } // namespace security

src/v/security/scram_authenticator.h

@@ -79,11 +79,13 @@ class scram_authenticator final : public sasl_mechanism {
 template<>
 struct scram_mechanism_traits<scram_sha256> {
     static constexpr const char* name = "SCRAM-SHA-256";
+    static constexpr scram_algorithm_t algorithm = scram_algorithm_t::sha256;
 };
 
 template<>
 struct scram_mechanism_traits<scram_sha512> {
     static constexpr const char* name = "SCRAM-SHA-512";
+    static constexpr scram_algorithm_t algorithm = scram_algorithm_t::sha512;
 };
 
 struct scram_sha256_authenticator {
@@ -98,7 +100,23 @@ struct scram_sha512_authenticator {
       = scram_mechanism_traits<scram_sha512>::name;
 };
 
+class scram_credential_cache;
+
+/// Validates a plaintext password against a stored SCRAM credential, for
+/// non-SCRAM authentication such as HTTP Basic auth and SASL/PLAIN. Returns
+/// the credential's SASL mechanism name on success.
+///
+/// Consults the shard's scram_credential_cache to skip the expensive salted
+/// password derivation for recently validated (password, credential) pairs,
+/// unless disabled via the scram_credential_cache_enabled config.
 std::optional<std::string_view> validate_scram_credential(
   const scram_credential& cred, const credential_password& password);
 
+namespace detail {
+/// As above, against a caller-provided cache.
+std::optional<std::string_view> validate_scram_credential(
+  const scram_credential& cred,
+  const credential_password& password,
+  scram_credential_cache* cache);
+} // namespace detail
 } // namespace security

src/v/security/tests/scram_credential_cache_test.cc

@@ -11,6 +11,8 @@
 #define BOOST_TEST_MODULE security
 
 #include "bytes/bytes.h"
+#include "security/scram_algorithm.h"
+#include "security/scram_authenticator.h"
 #include "security/scram_credential_cache.h"
 #include "security/types.h"
 #include "test_utils/random_bytes.h"
@@ -130,4 +132,65 @@ BOOST_AUTO_TEST_CASE(cache_tiny_capacity_does_not_crash) {
     }
 }
 
+BOOST_AUTO_TEST_CASE(cached_validate_scram_credential) {
+    scram_credential_cache cache(16);
+    auto cred = scram_sha256::make_credentials(
+      password()(), scram_sha256::min_iterations);
+
+    auto mech = detail::validate_scram_credential(cred, password(), &cache);
+    BOOST_REQUIRE(mech.has_value());
+    BOOST_REQUIRE_EQUAL(*mech, scram_sha256_authenticator::name);
+    BOOST_REQUIRE_EQUAL(cache.stats().hits, 0);
+
+    // Repeat validation is served from the cache with the same result.
+    mech = detail::validate_scram_credential(cred, password(), &cache);
+    BOOST_REQUIRE(mech.has_value());
+    BOOST_REQUIRE_EQUAL(*mech, scram_sha256_authenticator::name);
+    BOOST_REQUIRE_EQUAL(cache.stats().hits, 1);
+
+    // A wrong password fails, and keeps failing once its derivation is
+    // cached.
+    const credential_password wrong{"wrong-password"};
+    BOOST_REQUIRE(
+      !detail::validate_scram_credential(cred, wrong, &cache).has_value());
+    BOOST_REQUIRE(
+      !detail::validate_scram_credential(cred, wrong, &cache).has_value());
+}
+
+BOOST_AUTO_TEST_CASE(cached_validate_scram_credential_sha512) {
+    scram_credential_cache cache(16);
+    auto cred = scram_sha512::make_credentials(
+      password()(), scram_sha512::min_iterations);
+
+    for (int i = 0; i < 2; ++i) {
+        auto mech = detail::validate_scram_credential(cred, password(), &cache);
+        BOOST_REQUIRE(mech.has_value());
+        BOOST_REQUIRE_EQUAL(*mech, scram_sha512_authenticator::name);
+    }
+    BOOST_REQUIRE_EQUAL(cache.stats().hits, 1);
+}
+
+BOOST_AUTO_TEST_CASE(cached_validate_scram_credential_password_change) {
+    scram_credential_cache cache(16);
+    auto cred = scram_sha256::make_credentials(
+      password()(), scram_sha256::min_iterations);
+
+    // Warm the cache with the old credential.
+    BOOST_REQUIRE(detail::validate_scram_credential(cred, password(), &cache));
+    BOOST_REQUIRE(detail::validate_scram_credential(cred, password(), &cache));
+
+    // The user's password is changed: the new credential gets a fresh salt.
+    const credential_password new_password{"brand-new-password"};
+    auto new_cred = scram_sha256::make_credentials(
+      new_password(), scram_sha256::min_iterations);
+
+    // The stale cache entries do not interfere in either direction.
+    BOOST_REQUIRE(
+      detail::validate_scram_credential(new_cred, new_password, &cache)
+        .has_value());
+    BOOST_REQUIRE(
+      !detail::validate_scram_credential(new_cred, password(), &cache)
+         .has_value());
+}
+
 } // namespace security