Add assets/permissions/{identifier} endpoint that can be used to request a list of all permissions pertaining to that asset (#655)

Add endpoint to show permission, use Keycloak API more directly

Author: PGijsbers

docs/using/upload.md

@@ -201,6 +201,16 @@ their submission. An example body of the `POST` request to `/reviews` could look
 ```
 The comment may be up to 1800 characters long, so detailed feedback can be given.
 
+## Sharing Access
+By default, items are private when they are in draft or under submission.
+When items are published, they are publicly readable.
+However, you are the only person with the ability to edit or remove them.
+
+You can allow others to see assets which are still private, or give users administrator
+or write permissions for individual assets. For more information, see the `/assets/permissions`
+endpoints.
+
+
 ## Notes
 Assets registered by users will automatically be associated with the "AIoD" platform,
 indicating that it's a direct registration. Users cannot associate their registered

src/authentication.py

@@ -26,7 +26,7 @@
 from dotenv import load_dotenv
 from fastapi import HTTPException, Security, status
 from fastapi.security import OpenIdConnect
-from keycloak import KeycloakOpenID, KeycloakAdmin
+from keycloak import KeycloakOpenID, KeycloakAdmin, KeycloakGetError
 
 from config import KEYCLOAK_CONFIG
 
@@ -177,16 +177,24 @@ async def get_user_or_raise(token=Security(oidc)) -> KeycloakUser:
 
 
 def get_user_by_username(username: str) -> KeycloakUser | None:
-    """Gets the keycloak user by its username. `user.roles` will always be empty."""
-    users = keycloak_api().get_users(query={"username": username, "exact": True})
-    if not users:
-        return None
-
-    if len(users) > 1:
-        raise NotImplementedError(
-            f"Multiple users with username {username} found, expected behavior undefined."
+    """Gets the keycloak user by their username. `user.roles` will always be empty."""
+    if sub := keycloak_api().get_user_id(username):
+        return KeycloakUser(
+            name=username,
+            roles=set(),  # Not included with the call
+            _subject_identifier=sub,
         )
-    user = users[0]
+    return None
+
+
+def get_user_by_sub(sub: str) -> KeycloakUser | None:
+    """Gets the keycloak user by their sub. `user.roles` will always be empty."""
+    try:
+        user = keycloak_api().get_user(sub)
+    except KeycloakGetError as e:
+        if e.error_message == "User not found":
+            return None
+        raise
     return KeycloakUser(
         name=user.get("username"),
         roles=set(),  # Not included with the call

src/routers/asset_router.py

@@ -1,9 +1,16 @@
 import re
 from http import HTTPStatus
 from fastapi import APIRouter, Depends, HTTPException, Body
-from sqlmodel import Session
+from sqlmodel import Session, select
+import logging
 
-from authentication import KeycloakUser, get_user_or_none, get_user_or_raise, get_user_by_username
+from authentication import (
+    KeycloakUser,
+    get_user_or_none,
+    get_user_or_raise,
+    get_user_by_username,
+    get_user_by_sub,
+)
 from database.authorization import user_can_administer, set_permission, Permission, register_user
 from database.session import get_session
 from database.model.helper_functions import get_asset_by_identifier
@@ -12,6 +19,8 @@
 from database.authorization import user_can_read, PermissionType
 from versioning import Version
 
+logger = logging.getLogger(__file__)
+
 
 def create(url_prefix: str = "", version: Version = Version.LATEST) -> APIRouter:
     router = APIRouter()
@@ -76,6 +85,34 @@ def add_or_update_permission(
                 session.delete(permission)
                 session.commit()
 
+    @router.get(
+        "/assets/permissions/{identifier}",
+        tags=["Assets"],
+        description="Show the permissions for this asset. Requires admin rights of the asset.",
+    )
+    def show_permission(
+        identifier: str,
+        session: Session = Depends(get_session),
+        current_user: KeycloakUser = Depends(get_user_or_raise),
+    ):
+        _, resource = get_asset_by_identifier(identifier, session)
+        if not user_can_administer(current_user, resource.aiod_entry):
+            raise HTTPException(
+                status_code=HTTPStatus.FORBIDDEN,
+                detail=f"You are not allowed to see permissions for asset {identifier}.",
+            )
+
+        permissions = select(Permission).where(
+            Permission.aiod_entry_identifier == resource.aiod_entry.identifier
+        )
+        users = []
+        for permission in session.scalars(permissions).all():
+            if (user := get_user_by_sub(permission.user_identifier)) is not None:
+                users.append({"name": user.name, "permission": permission.type_})
+            else:
+                logger.warning(f"Could not find user for sub {permission.user_identifier}.")
+        return users
+
     @router.get(
         f"/assets/{{identifier}}",
         tags=["Assets"],

src/tests/resources/authentication/query_alice.json

@@ -0,0 +1 @@
+[{"id": "alice000-0000-0000-0000-000000000000", "username": "Alice", "emailVerified": "False", "createdTimestamp": 1739270119298, "enabled": "True", "totp": "False", "disableableCredentialTypes": [], "requiredActions": [], "notBefore": 0, "access": {"manageGroupMembership": "False", "view": "True", "mapRoles": "False", "impersonate": "False", "manage": "False"}}]

src/tests/test_asset_router.py

@@ -3,10 +3,12 @@
 
 import pytest
 from http import HTTPStatus
+
+from sqlmodel import select
 from starlette.testclient import TestClient
 import responses
 
-from database.authorization import register_user, PermissionType, Permission
+from database.authorization import register_user, PermissionType, Permission, set_permission
 from database.model.agent.organisation import Organisation
 from database.session import DbSession
 
@@ -80,7 +82,7 @@ def test_add_permission_by_name(
         )
         request_mock.add(
             responses.GET,
-            "http://keycloak:8080/aiod-auth/admin/realms/aiod/users?username=Bob&exact=True&max=100&first=0",
+            "http://keycloak:8080/aiod-auth/admin/realms/aiod/users?username=bob&max=1&exact=True",
             json=users_response,
         )
 
@@ -96,3 +98,57 @@ def test_add_permission_by_name(
             permission = session.get(Permission, {"aiod_entry_identifier": 1 , "user_identifier": BOB._subject_identifier})
         assert permission is not None
         assert permission.type_ == PermissionType.WRITE
+
+
+def test_show_permission(
+        client: TestClient,
+        publication: Publication,
+):
+    identifier = register_asset(publication, owner=ALICE)
+    with DbSession() as session:
+        register_user(BOB, session)
+        publication = session.scalar(select(Publication))
+        set_permission(BOB, publication.aiod_entry, session, type_=PermissionType.WRITE)
+        session.commit()
+
+    cached_api_token = path_test_resources() / "authentication" / "admin_connect.json"
+    with cached_api_token.open("r") as f:
+        connect_response = json.load(f)
+
+    cached_user_response = path_test_resources() / "authentication" / "query_alice.json"
+    with cached_user_response.open("r") as f:
+        alice_response = json.load(f)
+        alice_response = alice_response[0]  # default json returns "multiple" user response
+
+    cached_user_response = path_test_resources() / "authentication" / "query_bob.json"
+    with cached_user_response.open("r") as f:
+        bob_response = json.load(f)
+        bob_response = bob_response[0]  # default json returns "multiple" user response
+
+    # The second time around the cached KeycloakAdmin is used, so the connect is not called
+    with responses.RequestsMock(assert_all_requests_are_fired=False) as request_mock:
+        request_mock.add(
+            responses.POST,
+            "http://keycloak:8080/aiod-auth/realms/aiod/protocol/openid-connect/token",
+            json=connect_response,
+        )
+        request_mock.add(
+            responses.GET,
+            "http://keycloak:8080/aiod-auth/admin/realms/aiod/users/bob00000-0000-0000-0000-000000000000",
+            json=bob_response,
+        )
+
+        request_mock.add(
+            responses.GET,
+            "http://keycloak:8080/aiod-auth/admin/realms/aiod/users/alice000-0000-0000-0000-000000000000",
+            json=alice_response,
+        )
+
+        with logged_in_user(ALICE):
+            response = client.get(
+                f"/assets/permissions/{identifier}",
+                headers={"Authorization": "fake-token"},
+            )
+        assert response.status_code == HTTPStatus.OK
+        server_permissions = {p["name"]: p["permission"] for p in response.json()}
+        assert server_permissions == {"Alice": "admin", "Bob": "write"}