audit: Do not audit ephemeral users

Ephemeral users are used by auditing and schema registry.  There is no
reason to audit the auditing ephemeral user as that does not correspond
to a user derived action.

For schema registry, there are already audit events tracked when the
user access SR via the SR API.  It isn't necessary to audit that the SR
client has performed an action.

Signed-off-by: Michael Boquard <michael@redpanda.com>

Author: michael-redpanda

src/v/security/audit/audit_log_manager.cc

@@ -1081,12 +1081,21 @@ audit_log_manager::should_enqueue_audit_event(
     return should_enqueue_audit_event();
 }
 
+namespace {
+bool is_ignored_ephemeral_user(const security::acl_principal& principal) {
+    return principal == security::audit_principal
+           || principal == security::schema_registry_principal;
+}
+} // namespace
+
 std::optional<audit_log_manager::audit_event_passthrough>
 audit_log_manager::should_enqueue_audit_event(
   event_type type,
   const security::acl_principal& principal,
   ignore_enabled_events ignore_events) const {
-    if (_audit_excluded_principals.contains(principal)) {
+    if (
+      _audit_excluded_principals.contains(principal)
+      || is_ignored_ephemeral_user(principal)) {
         return std::make_optional(audit_event_passthrough::yes);
     }
 

tests/rptest/tests/audit_log_test.py

@@ -1612,6 +1612,27 @@ def test_authn_messages(self):
         assert len(
             records) == 1, f"Expected only one record got {len(records)}"
 
+    @skip_fips_mode
+    @cluster(num_nodes=5)
+    def test_no_ephemeral_user(self):
+        """
+        Verifies that ephemeral users do not generate audit messages
+        """
+        self.setup_cluster()
+
+        user_rpk = self.get_rpk()
+
+        _ = user_rpk.list_topics()
+
+        start_time = time.time()
+
+        # Read all records that have the audit log user for two seconds - should not get any records
+        records = self.read_all_from_audit_log(partial(self.authn_filter_function, self.kafka_rpc_service_name, "__auditing", 99, "SASL-SCRAM"),
+                                               lambda records: time.time() >= start_time + 2)
+        
+        assert len(records) == 0, f'Expected 0 records: {records}'
+
+
     @skip_fips_mode
     @cluster(num_nodes=5)
     def test_authn_failure_messages(self):