[v23.1.x] admin: Exclude ephemeral users from list_users

src/v/redpanda/admin/api-doc/security.json

@@ -11,6 +11,14 @@
     "get": {
         "summary": "List users",
         "operationId": "list_users",
+        "parameters": [
+            {
+                "name": "include_ephemeral",
+                "in": "query",
+                "required": false,
+                "type": "boolean"
+            }
+        ],
         "responses": {
             "200": {
                 "description": "List users"

src/v/redpanda/admin_server.cc

@@ -77,9 +77,12 @@
 #include "redpanda/admin/api-doc/transaction.json.h"
 #include "redpanda/admin/api-doc/usage.json.h"
 #include "rpc/errc.h"
+#include "security/acl.h"
 #include "security/credential_store.h"
 #include "security/scram_algorithm.h"
 #include "security/scram_authenticator.h"
+#include "security/scram_credential.h"
+#include "ssx/future-util.h"
 #include "ssx/metrics.h"
 #include "utils/string_switch.h"
 #include "vlog.h"
@@ -97,6 +100,7 @@
 #include <seastar/http/reply.hh>
 #include <seastar/http/request.hh>
 #include <seastar/util/log.hh>
+#include <seastar/util/variant_utils.hh>
 
 #include <boost/algorithm/string/classification.hpp>
 #include <boost/algorithm/string/split.hpp>
@@ -1757,11 +1761,24 @@ void admin_server::register_security_routes() {
 
     register_route<superuser>(
       ss::httpd::security_json::list_users,
-      [this](std::unique_ptr<ss::httpd::request>) {
+      [this](std::unique_ptr<ss::httpd::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, _] :
+          for (const auto& [user, type] :
                _controller->get_credential_store().local()) {
-              users.push_back(user());
+              if (include_ephemeral || !is_ephemeral(type)) {
+                  users.push_back(user());
+              }
           }
           return ss::make_ready_future<ss::json::json_return_type>(
             std::move(users));

src/v/security/scram_credential.h

@@ -47,7 +47,7 @@ class scram_credential
     const bytes& server_key() const { return _server_key; }
     const bytes& stored_key() const { return _stored_key; }
     int iterations() const { return _iterations; }
-    const std::optional<acl_principal>& principal() { return _principal; }
+    const std::optional<acl_principal>& principal() const { return _principal; }
 
     bool operator==(const scram_credential&) const = default;
 

tests/rptest/services/admin.py

@@ -682,8 +682,12 @@ def update_user(self, username, password, algorithm):
                           algorithm=algorithm,
                       ))
 
-    def list_users(self, node=None):
-        return self._request("get", "security/users", node=node).json()
+    def list_users(self, node=None, include_ephemeral: Optional[bool] = False):
+        params = None if include_ephemeral is None else {
+            "include_ephemeral": f"{include_ephemeral}".lower()
+        }
+        return self._request("get", "security/users", node=node,
+                             params=params).json()
 
     def partition_transfer_leadership(self,
                                       namespace,

tests/rptest/tests/admin_api_auth_test.py

@@ -7,11 +7,14 @@
 # the Business Source License, use of this software will be governed
 # by the Apache License, Version 2.0
 
+import requests
+
 from rptest.services.admin import Admin
 from rptest.tests.redpanda_test import RedpandaTest
+from rptest.tests.pandaproxy_test import PandaProxyEndpoints
 from rptest.clients.rpk import RpkTool
 from rptest.services.cluster import cluster
-from rptest.services.redpanda import SaslCredentials
+from rptest.services.redpanda import SaslCredentials, SecurityConfig
 from rptest.util import expect_exception, expect_http_error
 
 from ducktape.utils.util import wait_until
@@ -171,3 +174,36 @@ def test_combined_request(self):
             "superusers":
             [self.redpanda.SUPERUSER_CREDENTIALS.username, ALICE.username]
         })
+
+
+class AdminApiListUsersTest(PandaProxyEndpoints):
+    def __init__(self, context):
+        security = SecurityConfig()
+        security.kafka_enable_authorization = True
+        security.endpoint_authn_method = 'sasl'
+        security.auto_auth = True
+
+        super(AdminApiListUsersTest, self).__init__(context, security=security)
+
+        self.superuser = self.redpanda.SUPERUSER_CREDENTIALS
+        self.superuser_admin = Admin(self.redpanda,
+                                     auth=(self.superuser.username,
+                                           self.superuser.password))
+
+    @cluster(num_nodes=3)
+    def test_list_users(self):
+        # Create ephemeral users for each pandaproxy instance
+        pp_hosts = [node.account.hostname for node in self.redpanda.nodes]
+        for host in pp_hosts:
+            res = requests.get(f"http://{host}:8082/status/ready")
+            assert res.status_code == requests.codes.ok
+
+        users = self.superuser_admin.list_users()
+        ephemeral_users = self.superuser_admin.list_users(
+            include_ephemeral=True)
+
+        self.logger.debug(
+            f"users: {users}\n:ephemeral_users: {ephemeral_users}\npp_hosts: {pp_hosts}"
+        )
+        assert len(pp_hosts) > 0
+        assert len(ephemeral_users) - len(users) == len(pp_hosts)