[1637] api/permissions: Allow the root user to query their own permissions

CHANGELOG.md

@@ -27,6 +27,7 @@ The types of changes are:
 * Fixed wording on identity verification modal in the Privacy Center [#1674](https://github.com/ethyca/fides/pull/1674)
 * Update system fides_key tooltip text [#1533](https://github.com/ethyca/fides/pull/1685)
 * Removed local storage parsing that is redundant with redux-persist. [#1678](https://github.com/ethyca/fides/pull/1678)
+* Allow users to query their own permissions, including root user. [#1698](https://github.com/ethyca/fides/pull/1698)
 
 ### Security
 

src/fides/api/ops/api/v1/endpoints/user_permission_endpoints.py

@@ -10,6 +10,7 @@
 from fides.api.ops.api import deps
 from fides.api.ops.api.v1 import urn_registry as urls
 from fides.api.ops.api.v1.scope_registry import (
+    SCOPE_REGISTRY,
     USER_PERMISSION_CREATE,
     USER_PERMISSION_READ,
     USER_PERMISSION_UPDATE,
@@ -26,7 +27,9 @@
     oauth2_scheme,
     verify_oauth_client,
 )
+from fides.ctl.core.config import get_config
 
+CONFIG = get_config()
 logger = logging.getLogger(__name__)
 router = APIRouter(tags=["User Permissions"], prefix=V1_URL_PREFIX)
 
@@ -97,15 +100,28 @@ async def get_user_permissions(
     current_user: FidesUser = Depends(get_current_user),
     user_id: str,
 ) -> FidesUserPermissions:
-    validate_user_id(db, user_id)
+    # A user is able to retrieve their own permissions.
     if current_user.id == user_id:
+        # The root user is a special case because they aren't persisted in the database.
+        if current_user.id == CONFIG.security.oauth_root_client_id:
+            logger.info("Created FidesUserPermission for root user")
+            return FidesUserPermissions(
+                id=CONFIG.security.oauth_root_client_id,
+                user_id=CONFIG.security.oauth_root_client_id,
+                scopes=SCOPE_REGISTRY,
+            )
+
         logger.info("Retrieved FidesUserPermission record for current user")
-    else:
-        await verify_oauth_client(
-            security_scopes=SecurityScopes([USER_PERMISSION_READ]),
-            authorization=authorization,
-            db=db,
-        )
-        logger.info("Retrieved FidesUserPermission record")
+        return FidesUserPermissions.get_by(db, field="user_id", value=current_user.id)
+
+    # To look up the permissions of another user, that user must exist and the current user must
+    # have permission to read users.
+    validate_user_id(db, user_id)
+    await verify_oauth_client(
+        security_scopes=SecurityScopes([USER_PERMISSION_READ]),
+        authorization=authorization,
+        db=db,
+    )
 
+    logger.info("Retrieved FidesUserPermission record")
     return FidesUserPermissions.get_by(db, field="user_id", value=user_id)

src/fides/api/ops/util/oauth_util.py

@@ -44,12 +44,20 @@ async def get_current_user(
     authorization: str = Security(oauth2_scheme),
     db: Session = Depends(get_db),
 ) -> FidesUser:
-    """A wrapper around verify_oauth_client that returns that client's user if one exsits."""
+    """A wrapper around verify_oauth_client that returns that client's user if one exists."""
     client = await verify_oauth_client(
         security_scopes=security_scopes,
         authorization=authorization,
         db=db,
     )
+
+    if client.id == CONFIG.security.oauth_root_client_id:
+        return FidesUser(
+            id=CONFIG.security.oauth_root_client_id,
+            username=CONFIG.security.root_username,
+            created_at=datetime.utcnow(),
+        )
+
     return client.user
 
 

tests/ops/api/v1/endpoints/test_user_permission_endpoints.py

@@ -16,6 +16,7 @@
 from fides.api.ops.api.v1.scope_registry import (
     PRIVACY_REQUEST_READ,
     SAAS_CONFIG_READ,
+    SCOPE_REGISTRY,
     USER_PERMISSION_CREATE,
     USER_PERMISSION_READ,
     USER_PERMISSION_UPDATE,
@@ -302,3 +303,36 @@ def test_get_current_user_permissions(self, db, api_client, auth_user) -> None:
         assert response_body["id"] == permissions.id
         assert response_body["user_id"] == auth_user.id
         assert response_body["scopes"] == scopes
+
+    def test_get_current_root_user_permissions(
+        self, api_client, oauth_root_client, root_auth_header
+    ):
+        response = api_client.get(
+            f"{V1_URL_PREFIX}/user/{oauth_root_client.id}/permission",
+            headers=root_auth_header,
+        )
+        response_body = response.json()
+        assert HTTP_200_OK == response.status_code
+        assert response_body["id"] == oauth_root_client.id
+        assert response_body["user_id"] == oauth_root_client.id
+        assert response_body["scopes"] == SCOPE_REGISTRY
+
+    def test_get_root_user_permissions_by_non_root_user(
+        self, db, api_client, oauth_root_client, auth_user
+    ):
+        # Even with user read permissions, the root user can't be queried.
+        scopes = [USER_PERMISSION_READ]
+        ClientDetail.create_client_and_secret(
+            db,
+            CONFIG.security.oauth_client_id_length_bytes,
+            CONFIG.security.oauth_client_secret_length_bytes,
+            scopes=scopes,
+            user_id=auth_user.id,
+        )
+        auth_header = generate_auth_header_for_user(auth_user, scopes)
+
+        response = api_client.get(
+            f"{V1_URL_PREFIX}/user/{oauth_root_client.user_id}/permission",
+            headers=auth_header,
+        )
+        assert HTTP_404_NOT_FOUND == response.status_code

tests/ops/conftest.py

@@ -154,6 +154,29 @@ def oauth_client(db: Session) -> Generator:
     yield client
 
 
+@pytest.fixture(scope="function")
+def oauth_root_client(db: Session) -> ClientDetail:
+    """Return the configured root client (never persisted)"""
+    return ClientDetail.get(
+        db,
+        object_id=CONFIG.security.oauth_root_client_id,
+        config=CONFIG,
+        scopes=SCOPE_REGISTRY,
+    )
+
+
+@pytest.fixture(scope="function")
+def root_auth_header(oauth_root_client: ClientDetail) -> Dict[str, str]:
+    """Return an auth header for the root client"""
+    payload = {
+        JWE_PAYLOAD_SCOPES: oauth_root_client.scopes,
+        JWE_PAYLOAD_CLIENT_ID: oauth_root_client.id,
+        JWE_ISSUED_AT: datetime.now().isoformat(),
+    }
+    jwe = generate_jwe(json.dumps(payload), CONFIG.security.app_encryption_key)
+    return {"Authorization": "Bearer " + jwe}
+
+
 def generate_auth_header_for_user(user, scopes) -> Dict[str, str]:
     payload = {
         JWE_PAYLOAD_SCOPES: scopes,