NAS-130428 / 24.10 / Add auditing for API key CRUD operations (#14131)

Generate an audit trail for changes to API keys.

src/middlewared/middlewared/plugins/api_key.py

@@ -52,7 +52,7 @@ async def item_extend(self, item):
         item.pop("key")
         return item
 
-    @api_method(ApiKeyCreateArgs, ApiKeyCreateResult)
+    @api_method(ApiKeyCreateArgs, ApiKeyCreateResult, audit='Create API key', audit_extended=lambda data: data['name'])
     async def do_create(self, data: dict) -> dict:
         """
         Creates API Key.
@@ -76,8 +76,8 @@ async def do_create(self, data: dict) -> dict:
 
         return self._serve(data, key)
 
-    @api_method(ApiKeyUpdateArgs, ApiKeyUpdateResult)
-    async def do_update(self, id_: int, data: dict) -> dict:
+    @api_method(ApiKeyUpdateArgs, ApiKeyUpdateResult, audit='Update API key', audit_callback=True)
+    async def do_update(self, audit_callback: callable, id_: int, data: dict) -> dict:
         """
         Update API Key `id`.
 
@@ -86,6 +86,7 @@ async def do_update(self, id_: int, data: dict) -> dict:
         reset = data.pop("reset", False)
 
         old = await self.get_instance(id_)
+        audit_callback(old['name'])
         new = old.copy()
 
         new.update(data)
@@ -108,11 +109,14 @@ async def do_update(self, id_: int, data: dict) -> dict:
 
         return self._serve(await self.get_instance(id_), key)
 
-    @api_method(ApiKeyDeleteArgs, ApiKeyDeleteResult)
-    async def do_delete(self, id_: int) -> Literal[True]:
+    @api_method(ApiKeyDeleteArgs, ApiKeyDeleteResult, audit='Delete API key', audit_callback=True)
+    async def do_delete(self, audit_callback: callable, id_: int) -> Literal[True]:
         """
         Delete API Key `id`.
         """
+        name = (await self.get_instance(id_))['name']
+        audit_callback(name)
+
         response = await self.middleware.call(
             "datastore.delete",
             self._config.datastore,

tests/api2/test_audit_api_key.py

@@ -0,0 +1,34 @@
+from middlewared.test.integration.utils import call
+from middlewared.test.integration.utils.audit import expect_audit_method_calls
+
+API_KEY_NAME = 'AUDIT_API_KEY'
+
+
+def test_api_key_audit():
+    payload = {'name': API_KEY_NAME, 'allowlist': [{'resource': '*', 'method': '*'}]}
+    payload2 = {'allowlist': []}
+    audit_id = None
+
+    try:
+        with expect_audit_method_calls([{
+            'method': 'api_key.create',
+            'params': [payload],
+            'description': f'Create API key {API_KEY_NAME}',
+        }]):
+            api_key_id = call('api_key.create', payload)['id']
+
+        with expect_audit_method_calls([{
+            'method': 'api_key.update',
+            'params': [api_key_id, payload2],
+            'description': f'Update API key {API_KEY_NAME}',
+        }]):
+            call('api_key.update', api_key_id, payload2)
+
+    finally:
+        if audit_id:
+            with expect_audit_method_calls([{
+                'method': 'api_key.delete',
+                'params': [api_key_id],
+                'description': f'Delete API key {API_KEY_NAME}',
+            }]):
+                call('api_key.delete', api_key_id)