security/credential_store: Make iteration more explicit

Currently all users of the store may want to filter out principals
that are of type ephemeral_user.

In order to prevent bugs such as serializing ephemeral users to disk,
require a predicate when iterating the store.

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

src/v/cluster/security_manager.cc

@@ -151,20 +151,13 @@ ss::future<>
 security_manager::fill_snapshot(controller_snapshot& controller_snap) const {
     auto& snapshot = controller_snap.security;
 
-    for (const auto& cred : _credentials.local()) {
+    // Ephemeral credentials must not be stored in the snapshot.
+    auto creds = _credentials.local().range(
+      security::credential_store::is_not_ephemeral);
+    for (const auto& cred : creds) {
         ss::visit(cred.second, [&](security::scram_credential scram) {
-            // Regular users may not have a defined principal. For example,
-            // those users created from HTTP POST security/users (See
-            // parse_scram_credential() in the Admin server). Therefore, add
-            // these users into the snapshot.
-            bool is_ephemeral_user
-              = scram.principal().has_value()
-                && scram.principal()->type()
-                     == security::principal_type::ephemeral_user;
-            if (!is_ephemeral_user) {
-                snapshot.user_credentials.push_back(user_and_credential{
-                  security::credential_user{cred.first}, std::move(scram)});
-            }
+            snapshot.user_credentials.push_back(user_and_credential{
+              security::credential_user{cred.first}, std::move(scram)});
         });
         co_await ss::coroutine::maybe_yield();
     }

src/v/redpanda/admin_server.cc

@@ -128,6 +128,8 @@
 #include <algorithm>
 #include <charconv>
 #include <chrono>
+#include <functional>
+#include <iterator>
 #include <limits>
 #include <memory>
 #include <numeric>
@@ -1963,21 +1965,17 @@ void admin_server::register_security_routes() {
       [this](std::unique_ptr<ss::http::request> req) {
           bool include_ephemeral = req->get_query_param("include_ephemeral")
                                    == "true";
-          constexpr auto is_ephemeral =
-            [](security::credential_store::credential_types const& t) {
-                return ss::visit(t, [](security::scram_credential const& c) {
-                    return c.principal().has_value()
-                           && c.principal().value().type()
-                                == security::principal_type::ephemeral_user;
-                });
-            };
 
-          std::vector<ss::sstring> users;
-          for (const auto& [user, type] :
-               _controller->get_credential_store().local()) {
-              if (include_ephemeral || !is_ephemeral(type)) {
-                  users.push_back(user());
-              }
+          auto pred = [include_ephemeral](auto const& c) {
+              return include_ephemeral
+                     || security::credential_store::is_not_ephemeral(c);
+          };
+          auto creds = _controller->get_credential_store().local().range(pred);
+
+          std::vector<ss::sstring> users{};
+          users.reserve(std::distance(creds.begin(), creds.end()));
+          for (const auto& [user, type] : creds) {
+              users.push_back(user());
           }
           return ss::make_ready_future<ss::json::json_return_type>(
             std::move(users));

src/v/security/credential_store.h

@@ -13,7 +13,10 @@
 #include "security/scram_credential.h"
 #include "security/types.h"
 
+#include <seastar/util/variant_utils.hh>
+
 #include <absl/container/node_hash_map.h>
+#include <boost/range/adaptor/filtered.hpp>
 
 namespace security {
 
@@ -63,9 +66,25 @@ class credential_store {
         return _credentials.contains(name);
     }
 
-    const_iterator begin() const { return _credentials.cbegin(); }
-    const_iterator end() const { return _credentials.cend(); }
+    // Ephemeral credentials often require careful handling; they must not
+    // be serialized to disk, and in the general case, should not be displayed
+    // to users.
+    static constexpr auto is_not_ephemeral =
+      [](security::credential_store::container_type::value_type const& t) {
+          return ss::visit(t.second, [](security::scram_credential const& c) {
+              return !c.principal().has_value()
+                     || c.principal().value().type()
+                          != security::principal_type::ephemeral_user;
+          });
+      };
 
+    // Retrieve a list of credentials that satisfy the predicate.
+    //
+    // E.g.:
+    // _creds.range(credential_store::is_not_ephemeral);
+    auto range(auto pred) {
+        return boost::adaptors::filter(_credentials, std::move(pred));
+    }
     void clear() { _credentials.clear(); }
 
 private: