redpanda: Audit all admin api requests

- Immediately after logging an incoming request, the request will be
audited.

- If auditing is not enabled, or it is and the event type ::management
is not configured then the call to enqueue_audit_event just returns
true, otherwise the event is enqueued to be later pushed.

- If the event is to be enqueued but cannot because the queue is full,
the admin API will return a 5xx error code to the client.

- In the event authorization is enabled and the client fails to
authorize, a log to indicate this will now be printed.

- Futhermore the event will be added to the auditing system if auditing
is enabled and the ::management event type had been configured.

- If the above is true and the auditing event queue is full, then an
warning message is logged.

src/v/redpanda/admin_server.cc

@@ -90,6 +90,9 @@
 #include "rpc/rpc_utils.h"
 #include "security/acl.h"
 #include "security/audit/audit_log_manager.h"
+#include "security/audit/schemas/application_activity.h"
+#include "security/audit/schemas/utils.h"
+#include "security/audit/types.h"
 #include "security/credential_store.h"
 #include "security/scram_algorithm.h"
 #include "security/scram_authenticator.h"
@@ -116,6 +119,7 @@
 #include <seastar/core/with_scheduling_group.hh>
 #include <seastar/coroutine/maybe_yield.hh>
 #include <seastar/http/api_docs.hh>
+#include <seastar/http/exception.hh>
 #include <seastar/http/httpd.hh>
 #include <seastar/http/reply.hh>
 #include <seastar/http/request.hh>
@@ -586,6 +590,25 @@ ss::future<> admin_server::configure_listeners() {
     }
 }
 
+void admin_server::audit_authz(
+  ss::httpd::const_req req,
+  const request_auth_result& auth_result,
+  httpd_authorized authorized,
+  std::optional<std::string_view> reason) {
+    auto api_event = security::audit::make_api_activity_event(
+      req, auth_result, bool(authorized), reason);
+    auto success = _audit_mgr.local().enqueue_audit_event(
+      security::audit::event_type::management, std::move(api_event));
+    if (!success) {
+        vlog(
+          logger.error,
+          "Failed to audit authorization request for endpoint: {}",
+          req.format_url());
+        throw ss::httpd::server_error_exception(
+          "Failed to audit authorization request");
+    }
+}
+
 void admin_server::log_request(
   const ss::http::request& req, const request_auth_result& auth_state) const {
     vlog(

src/v/redpanda/admin_server.h

@@ -39,6 +39,7 @@
 #include <seastar/http/httpd.hh>
 #include <seastar/http/json_path.hh>
 #include <seastar/json/json_elements.hh>
+#include <seastar/util/bool_class.hh>
 #include <seastar/util/log.hh>
 
 #include <absl/container/flat_hash_map.h>
@@ -105,25 +106,40 @@ class admin_server {
     static constexpr auth_level user = auth_level::user;
     static constexpr auth_level superuser = auth_level::superuser;
 
+    using httpd_authorized = ss::bool_class<struct httpd_authorized>;
+    void audit_authz(
+      ss::httpd::const_req req,
+      const request_auth_result& auth_result,
+      httpd_authorized authorized,
+      std::optional<std::string_view> reason = std::nullopt);
+
     /**
      * Authenticate, and raise if `required_auth` is not met by
      * the credential (or pass if authentication is disabled).
      */
     template<auth_level required_auth>
     request_auth_result apply_auth(ss::httpd::const_req req) {
         auto auth_state = _auth.authenticate(req);
-        if constexpr (required_auth == auth_level::superuser) {
-            auth_state.require_superuser();
-        } else if constexpr (required_auth == auth_level::user) {
-            auth_state.require_authenticated();
-        } else if constexpr (required_auth == auth_level::publik) {
-            auth_state.pass();
-        } else {
-            static_assert(
-              utils::unsupported_value<required_auth>::value,
-              "Invalid auth_level");
+        try {
+            if constexpr (required_auth == auth_level::superuser) {
+                auth_state.require_superuser();
+            } else if constexpr (required_auth == auth_level::user) {
+                auth_state.require_authenticated();
+            } else if constexpr (required_auth == auth_level::publik) {
+                auth_state.pass();
+            } else {
+                static_assert(
+                  utils::unsupported_value<required_auth>::value,
+                  "Invalid auth_level");
+            }
+
+        } catch (const ss::httpd::base_exception& ex) {
+            audit_authz(req, auth_state, httpd_authorized::no, ex.what());
+            throw;
         }
 
+        audit_authz(req, auth_state, httpd_authorized::yes);
+
         return auth_state;
     }