NAS-130069 / 24.10 / Fix API key API

src/middlewared/middlewared/api/v25_04_0/api_key.py

@@ -2,9 +2,9 @@
 from typing import Literal, TypeAlias
 from typing_extensions import Annotated
 
-from pydantic import ConfigDict, StringConstraints
+from pydantic import StringConstraints
 
-from middlewared.api.base import BaseModel, Excluded, excluded_field, NonEmptyString, Private
+from middlewared.api.base import BaseModel, Excluded, excluded_field, ForUpdateMetaclass, NonEmptyString, Private
 
 
 HttpVerb: TypeAlias = Literal["GET", "POST", "PUT", "DELETE", "CALL", "SUBSCRIBE", "*"]
@@ -16,15 +16,16 @@ class AllowListItem(BaseModel):
 
 
 class ApiKeyEntry(BaseModel):
-    """Represents a record in the account.api_key table."""
-
     id: int
     name: Annotated[NonEmptyString, StringConstraints(max_length=200)]
-    key: Private[str]
     created_at: datetime
     allowlist: list[AllowListItem]
 
 
+class ApiKeyEntryWithKey(ApiKeyEntry):
+    key: Private[str]
+
+
 class ApiKeyCreate(ApiKeyEntry):
     id: Excluded = excluded_field()
     key: Excluded = excluded_field()
@@ -36,10 +37,10 @@ class ApiKeyCreateArgs(BaseModel):
 
 
 class ApiKeyCreateResult(BaseModel):
-    result: ApiKeyEntry
+    result: ApiKeyEntryWithKey
 
 
-class ApiKeyUpdate(ApiKeyCreate):
+class ApiKeyUpdate(ApiKeyCreate, metaclass=ForUpdateMetaclass):
     reset: bool
 
 
@@ -49,7 +50,7 @@ class ApiKeyUpdateArgs(BaseModel):
 
 
 class ApiKeyUpdateResult(BaseModel):
-    result: ApiKeyEntry
+    result: ApiKeyEntryWithKey
 
 
 class ApiKeyDeleteArgs(BaseModel):

src/middlewared/middlewared/plugins/api_key.py

@@ -47,6 +47,11 @@ class Config:
         datastore_extend = "api_key.item_extend"
         cli_namespace = "auth.api_key"
 
+    @private
+    async def item_extend(self, item):
+        item.pop("key")
+        return item
+
     @api_method(ApiKeyCreateArgs, ApiKeyCreateResult)
     async def do_create(self, data: dict) -> dict:
         """

tests/api2/test_api_key.py

@@ -1,8 +1,8 @@
-#!/usr/bin/env python3
-
 import contextlib
 import os
+
 import pytest
+
 import sys
 sys.path.append(os.getcwd())
 from functions import POST, GET, DELETE, SSH_TEST
@@ -31,7 +31,7 @@ def user():
         assert results.status_code == 200, results.text
 
 
-def test_root_api_key_websocket(request):
+def test_root_api_key_websocket():
     """We should be able to call a method with root API key using Websocket."""
     ip = truenas_server.ip
     with api_key([{"method": "*", "resource": "*"}]) as key:
@@ -53,7 +53,7 @@ def test_root_api_key_websocket(request):
             c.call("service.update", "cifs", {"enable": False})
 
 
-def test_allowed_api_key_websocket(request):
+def test_allowed_api_key_websocket():
     """We should be able to call a method with API key that allows that call using Websocket."""
     ip = truenas_server.ip
     with api_key([{"method": "CALL", "resource": "system.info"}]) as key:
@@ -64,7 +64,7 @@ def test_allowed_api_key_websocket(request):
         assert 'uptime' in str(results['stdout'])
 
 
-def test_denied_api_key_websocket(request):
+def test_denied_api_key_websocket():
     """We should not be able to call a method with API key that does not allow that call using Websocket."""
     ip = truenas_server.ip
     with api_key([{"method": "CALL", "resource": "system.info_"}]) as key:
@@ -74,10 +74,8 @@ def test_denied_api_key_websocket(request):
         assert results['result'] is False
 
 
-def test_denied_api_key_noauthz(request):
+def test_denied_api_key_noauthz():
     with api_key([{"method": "CALL", "resource": "system.info"}]) as key:
-        auth_token = None
-
         with client(auth=None) as c:
             assert c.call("auth.login_with_api_key", key)
 

tests/api2/test_api_key_crud.py

@@ -0,0 +1,49 @@
+import contextlib
+
+from middlewared.test.integration.utils import call, client
+
+
+@contextlib.contextmanager
+def api_key(allowlist):
+    key = call("api_key.create", {"name": "Test API Key", "allowlist": allowlist})
+    try:
+        yield key
+    finally:
+        call("api_key.delete", key["id"])
+
+
+def test_has_key_after_creation_but_not_read():
+    api_key = call("api_key.create", {"name": "Test", "allowlist": []})
+    try:
+        assert "key" in api_key
+
+        instance = call("api_key.get_instance", api_key["id"])
+        assert "key" not in instance
+
+        update = call("api_key.update", api_key["id"], {})
+        assert "key" not in update
+    finally:
+        call("api_key.delete", api_key["id"])
+
+
+def test_api_key_reset():
+    with api_key([]) as key:
+        with client(auth=None) as c:
+            assert c.call("auth.login_with_api_key", key["key"])
+
+        updated = call("api_key.update", key["id"], {"reset": True})
+
+        with client(auth=None) as c:
+            assert not c.call("auth.login_with_api_key", key["key"])
+
+        with client(auth=None) as c:
+            assert c.call("auth.login_with_api_key", updated["key"])
+
+
+def test_api_key_delete():
+    with api_key([]) as key:
+        with client(auth=None) as c:
+            assert c.call("auth.login_with_api_key", key["key"])
+
+    with client(auth=None) as c:
+        assert not c.call("auth.login_with_api_key", key["key"])