security: cache derived SCRAM stored key

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/BUILD

@@ -111,6 +111,7 @@ redpanda_cc_library(
         "role.cc",
         "scram_algorithm.cc",
         "scram_authenticator.cc",
+        "scram_credential_cache.cc",
     ],
     hdrs = [
         "acl.h",
@@ -142,6 +143,7 @@ redpanda_cc_library(
         "scram_algorithm.h",
         "scram_authenticator.h",
         "scram_credential.h",
+        "scram_credential_cache.h",
         "types.h",
     ],
     implementation_deps = [
@@ -186,6 +188,7 @@ redpanda_cc_library(
         "//src/v/strings:utf8",
         "//src/v/utils:absl_sstring_hash",
         "//src/v/utils:base64",
+        "//src/v/utils:chunked_kv_cache",
         "//src/v/utils:log_hist",
         "//src/v/utils:named_type",
         "//src/v/utils:to_string",

src/v/security/scram_algorithm.h

@@ -338,6 +338,18 @@ class scram_algorithm {
         return bytes(result.begin(), result.end());
     }
 
+    /**
+     * Computes the stored key for a plaintext password, as persisted in a
+     * scram_credential. This is the expensive part of non-SCRAM password
+     * validation: the salted password derivation runs `iterations` HMAC
+     * rounds.
+     */
+    static bytes derive_stored_key(
+      const ss::sstring& password, bytes_view salt, int iterations) {
+        auto salted_password = salt_password(password, salt, iterations);
+        return stored_key(client_key(salted_password));
+    }
+
     /**
      * For doing non-SCRAM authentication, such as HTTP basic auth over TLS,
      * using stored SCRAM credentials.
@@ -347,10 +359,8 @@ class scram_algorithm {
       bytes_view reference_stored_key,
       bytes_view salt,
       int iterations) {
-        auto salted_password = salt_password(password, salt, iterations);
-        auto clientkey = client_key(salted_password);
-        auto storedkey = stored_key(clientkey);
-        return storedkey == reference_stored_key;
+        return derive_stored_key(password, salt, iterations)
+               == reference_stored_key;
     }
 
 private:

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/scram_credential_cache.cc

@@ -0,0 +1,90 @@
+/*
+ * Copyright 2026 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 "security/scram_credential_cache.h"
+
+#include "bytes/random.h"
+#include "hashing/secure.h"
+
+#include <seastar/core/shared_ptr.hh>
+
+#include <algorithm>
+#include <bit>
+
+namespace security {
+
+namespace {
+// The S3-FIFO probationary queue is conventionally ~10% of the cache.
+constexpr size_t small_queue_ratio = 10;
+} // namespace
+
+scram_credential_cache::scram_credential_cache(size_t capacity)
+  : _digest_key(random_generators::get_crypto_bytes(digest_size))
+  , _cache([capacity]() -> decltype(_cache)::config {
+      auto small_size = std::max<size_t>(1, capacity / small_queue_ratio);
+      // s3_fifo::cache requires cache_size > small_size; derive cache_size
+      // from small_size so the invariant holds even for tiny capacities.
+      return {
+        .cache_size = std::max(capacity, small_size + 1),
+        .small_size = small_size};
+  }()) {}
+
+scram_credential_cache::key_t scram_credential_cache::make_key(
+  scram_algorithm_t mech,
+  const credential_password& password,
+  bytes_view salt,
+  int iterations) const {
+    hmac_sha256 mac(_digest_key);
+    // Unambiguous serialization: fixed-size fields and the password length
+    // come first, so no two distinct inputs produce the same byte stream.
+    mac.update(std::array<char, 1>{static_cast<char>(mech)});
+    mac.update(
+      std::bit_cast<std::array<char, sizeof(uint32_t)>>(
+        static_cast<uint32_t>(iterations)));
+    mac.update(
+      std::bit_cast<std::array<char, sizeof(uint32_t)>>(
+        static_cast<uint32_t>(password().size())));
+    mac.update(std::string_view{password()});
+    mac.update(salt);
+    return {.digest = mac.reset()};
+}
+
+std::optional<bytes> scram_credential_cache::get(
+  scram_algorithm_t mech,
+  const credential_password& password,
+  bytes_view salt,
+  int iterations) {
+    auto val = _cache.get_value(make_key(mech, password, salt, iterations));
+    if (!val) {
+        return std::nullopt;
+    }
+    return **val;
+}
+
+void scram_credential_cache::put(
+  scram_algorithm_t mech,
+  const credential_password& password,
+  bytes_view salt,
+  int iterations,
+  bytes stored_key) {
+    _cache.try_insert(
+      make_key(mech, password, salt, iterations),
+      ss::make_shared<bytes>(std::move(stored_key)));
+}
+
+scram_credential_cache::stats_t scram_credential_cache::stats() const {
+    auto s = _cache.stat();
+    return {
+      .hits = s.hit_count,
+      .misses = s.access_count - s.hit_count,
+      .size = s.index_size};
+}
+
+} // namespace security

src/v/security/scram_credential_cache.h

@@ -0,0 +1,96 @@
+/*
+ * Copyright 2026 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 "bytes/bytes.h"
+#include "security/scram_credential.h"
+#include "security/types.h"
+#include "utils/chunked_kv_cache.h"
+
+#include <array>
+#include <cstddef>
+#include <optional>
+
+namespace security {
+
+/// \brief Per-shard memoization of SCRAM password derivations.
+///
+/// Validating a plaintext password against a stored SCRAM credential
+/// rederives the credential's stored key (RFC 5802 `Hi`, at least 4096 HMAC
+/// rounds). HTTP Basic auth and SASL/PLAIN pay that cost on every
+/// authentication, which is expensive enough to saturate a core at a few
+/// thousand authentications per second. This cache memoizes the derivation:
+/// (algorithm, password, salt, iterations) -> stored key.
+///
+/// Entries are keyed by an HMAC (with a random per-instance key) of the
+/// inputs so plaintext passwords are not retained. The value is the stored
+/// key derived from the presented password: for a correct password this is
+/// the credential's stored key, already persisted in the credential store;
+/// an incorrect password is cached too (so repeated bad credentials stay
+/// cheap), and its derived value reveals nothing about the stored credential.
+/// Either way a memory disclosure yields no credential secret beyond what the
+/// store already holds. The memoized mapping is purely functional, so entries
+/// never go stale: updating a credential generates a fresh salt, which
+/// changes the cache key.
+class scram_credential_cache {
+public:
+    static constexpr size_t default_capacity = 1024;
+
+    struct stats_t {
+        size_t hits;
+        size_t misses;
+        size_t size;
+    };
+
+    explicit scram_credential_cache(size_t capacity);
+
+    /// Returns the memoized stored key for the derivation inputs, if any.
+    std::optional<bytes> get(
+      scram_algorithm_t mech,
+      const credential_password& password,
+      bytes_view salt,
+      int iterations);
+
+    /// Memoizes the stored key derived from the given inputs.
+    void put(
+      scram_algorithm_t mech,
+      const credential_password& password,
+      bytes_view salt,
+      int iterations,
+      bytes stored_key);
+
+    stats_t stats() const;
+
+private:
+    static constexpr size_t digest_size = 32;
+
+    struct key_t {
+        std::array<char, digest_size> digest;
+
+        bool operator==(const key_t&) const = default;
+
+        template<typename H>
+        friend H AbslHashValue(H h, const key_t& k) {
+            return H::combine(std::move(h), k.digest);
+        }
+    };
+
+    key_t make_key(
+      scram_algorithm_t mech,
+      const credential_password& password,
+      bytes_view salt,
+      int iterations) const;
+
+    bytes _digest_key;
+    utils::chunked_kv_cache<key_t, bytes> _cache;
+};
+
+} // namespace security