From e2a4143543785cd115f198f8227dba7c7a48ac8a Mon Sep 17 00:00:00 2001 From: Leo Vasanko Date: Tue, 27 Jan 2026 16:24:02 +0000 Subject: [PATCH] Consistently use UUID type in APIs instead of UUID str as option. --- paskia/db/operations.py | 142 +++++++++------------------------------- paskia/fastapi/admin.py | 42 +++++++++--- paskia/sansio.py | 1 - tests/conftest.py | 4 +- tests/test_admin.py | 2 +- 5 files changed, 65 insertions(+), 126 deletions(-) diff --git a/paskia/db/operations.py b/paskia/db/operations.py index 6233352..5ad1fc7 100644 --- a/paskia/db/operations.py +++ b/paskia/db/operations.py @@ -284,15 +284,13 @@ def get_permission_organizations(scope: str) -> list[Org]: return [] -def get_organization(uuid: str | UUID) -> Org | None: +def get_organization(uuid: UUID) -> Org | None: """Get organization by UUID. Call sites: - Get organization when creating a role to check grantable permissions (admin.py:271) - Get organization when adding permission to role to check if org can grant it (admin.py:352) """ - if isinstance(uuid, str): - uuid = UUID(uuid) return build_org(uuid, include_roles=True) if uuid in _db._data.orgs else None @@ -306,7 +304,7 @@ def list_organizations() -> list[Org]: return [build_org(uuid, include_roles=True) for uuid in _db._data.orgs] -def get_organization_users(org_uuid: str | UUID) -> list[tuple[User, str]]: +def get_organization_users(org_uuid: UUID) -> list[tuple[User, str]]: """Get all users in an organization with their role names. Call sites: @@ -314,8 +312,6 @@ def get_organization_users(org_uuid: str | UUID) -> list[tuple[User, str]]: - Get users from organizations with auth:admin for reset targets (reset.py:31,42,58) - Get users from organization to check if admin has credentials (bootstrap.py:73) """ - if isinstance(org_uuid, str): - org_uuid = UUID(org_uuid) role_map = { rid: r.display_name for rid, r in _db._data.roles.items() if r.org == org_uuid } @@ -326,7 +322,7 @@ def get_organization_users(org_uuid: str | UUID) -> list[tuple[User, str]]: ] -def get_role(uuid: str | UUID) -> Role | None: +def get_role(uuid: UUID) -> Role | None: """Get role by UUID. Call sites: @@ -335,24 +331,20 @@ def get_role(uuid: str | UUID) -> Role | None: - Get role to remove permission from it (admin.py:380) - Get role to delete it (admin.py:421) """ - if isinstance(uuid, str): - uuid = UUID(uuid) return build_role(uuid) if uuid in _db._data.roles else None -def get_roles_by_organization(org_uuid: str | UUID) -> list[Role]: +def get_roles_by_organization(org_uuid: UUID) -> list[Role]: """Get all roles in an organization. Call sites: - Get roles by organization when creating a user to find the role by name (admin.py:459) - Get roles by organization when updating user role to validate the new role name (admin.py:498) """ - if isinstance(org_uuid, str): - org_uuid = UUID(org_uuid) return [build_role(rid) for rid, r in _db._data.roles.items() if r.org == org_uuid] -def get_user_by_uuid(uuid: str | UUID) -> User | None: +def get_user_by_uuid(uuid: UUID) -> User | None: """Get user by UUID. Call sites: @@ -360,12 +352,10 @@ def get_user_by_uuid(uuid: str | UUID) -> User | None: - Get user from reset token for registration info (api.py:127) - Get user for listing user credentials in admin API (admin.py:594) """ - if isinstance(uuid, str): - uuid = UUID(uuid) return build_user(uuid) if uuid in _db._data.users else None -def get_user_organization(user_uuid: str | UUID) -> tuple[Org, str]: +def get_user_organization(user_uuid: UUID) -> tuple[Org, str]: """Get the organization a user belongs to and their role name. Raises ValueError if user not found. @@ -378,8 +368,6 @@ def get_user_organization(user_uuid: str | UUID) -> tuple[Org, str]: - Get user's organization for deleting user credential (admin.py:754) - Get user's organization for deleting user session (admin.py:783) """ - if isinstance(user_uuid, str): - user_uuid = UUID(user_uuid) if user_uuid not in _db._data.users: raise ValueError(f"User {user_uuid} not found") role_uuid = _db._data.users[user_uuid].role @@ -403,7 +391,7 @@ def get_credential_by_id(credential_id: bytes) -> Credential | None: return None -def get_credentials_by_user_uuid(user_uuid: str | UUID) -> list[Credential]: +def get_credentials_by_user_uuid(user_uuid: UUID) -> list[Credential]: """Get all credentials for a user. Call sites: @@ -414,8 +402,6 @@ def get_credentials_by_user_uuid(user_uuid: str | UUID) -> list[Credential]: - Get credentials to check if admin user has credentials (bootstrap.py:81) - Get credentials for user info formatting (userinfo.py:51) """ - if isinstance(user_uuid, str): - user_uuid = UUID(user_uuid) return [ build_credential(cid) for cid, c in _db._data.credentials.items() @@ -440,15 +426,13 @@ def get_session(key: str) -> Session | None: return build_session(key) -def list_sessions_for_user(user_uuid: str | UUID) -> list[Session]: +def list_sessions_for_user(user_uuid: UUID) -> list[Session]: """Get all active sessions for a user. Call sites: - List sessions for user info (userinfo.py:75) - List sessions for user details API (admin.py:651) """ - if isinstance(user_uuid, str): - user_uuid = UUID(user_uuid) now = datetime.now(timezone.utc) return [ build_session(key) @@ -664,24 +648,20 @@ def create_organization(org: Org, *, ctx: SessionContext | None = None) -> None: def update_organization_name( - uuid: str | UUID, + uuid: UUID, display_name: str, *, ctx: SessionContext | None = None, ) -> None: """Update organization display name.""" - if isinstance(uuid, str): - uuid = UUID(uuid) if uuid not in _db._data.orgs: raise ValueError(f"Organization {uuid} not found") with _db.transaction("Renamed organization", ctx): _db._data.orgs[uuid].display_name = display_name -def delete_organization(uuid: str | UUID, *, ctx: SessionContext | None = None) -> None: +def delete_organization(uuid: UUID, *, ctx: SessionContext | None = None) -> None: """Delete organization and all its roles/users.""" - if isinstance(uuid, str): - uuid = UUID(uuid) if uuid not in _db._data.orgs: raise ValueError(f"Organization {uuid} not found") with _db.transaction("Deleted organization", ctx): @@ -700,32 +680,15 @@ def delete_organization(uuid: str | UUID, *, ctx: SessionContext | None = None) def add_permission_to_organization( - org_uuid: str | UUID, - permission_id: str | UUID, + org_uuid: UUID, + permission_uuid: UUID, *, ctx: SessionContext | None = None, ) -> None: """Grant a permission to an organization by UUID.""" - if isinstance(org_uuid, str): - org_uuid = UUID(org_uuid) if org_uuid not in _db._data.orgs: raise ValueError(f"Organization {org_uuid} not found") - # Convert permission_id to UUID - if isinstance(permission_id, str): - try: - permission_uuid = UUID(permission_id) - except ValueError: - # It's a scope - look up the UUID (backwards compat) - for pid, p in _db._data.permissions.items(): - if p.scope == permission_id: - permission_uuid = pid - break - else: - raise ValueError(f"Permission {permission_id} not found") - else: - permission_uuid = permission_id - if permission_uuid not in _db._data.permissions: raise ValueError(f"Permission {permission_uuid} not found") @@ -734,32 +697,15 @@ def add_permission_to_organization( def remove_permission_from_organization( - org_uuid: str | UUID, - permission_id: str | UUID, + org_uuid: UUID, + permission_uuid: UUID, *, ctx: SessionContext | None = None, ) -> None: """Remove a permission from an organization by UUID.""" - if isinstance(org_uuid, str): - org_uuid = UUID(org_uuid) if org_uuid not in _db._data.orgs: raise ValueError(f"Organization {org_uuid} not found") - # Convert permission_id to UUID - if isinstance(permission_id, str): - try: - permission_uuid = UUID(permission_id) - except ValueError: - # It's a scope - look up the UUID (backwards compat) - for pid, p in _db._data.permissions.items(): - if p.scope == permission_id: - permission_uuid = pid - break - else: - return # Permission not found, silently return - else: - permission_uuid = permission_id - if permission_uuid not in _db._data.permissions: return # Permission not found, silently return @@ -778,14 +724,12 @@ def create_role(role: Role, *, ctx: SessionContext | None = None) -> None: def update_role_name( - uuid: str | UUID, + uuid: UUID, display_name: str, *, ctx: SessionContext | None = None, ) -> None: """Update role display name.""" - if isinstance(uuid, str): - uuid = UUID(uuid) if uuid not in _db._data.roles: raise ValueError(f"Role {uuid} not found") with _db.transaction("Renamed role", ctx): @@ -793,16 +737,12 @@ def update_role_name( def add_permission_to_role( - role_uuid: str | UUID, - permission_uuid: str | UUID, + role_uuid: UUID, + permission_uuid: UUID, *, ctx: SessionContext | None = None, ) -> None: """Add permission to role by UUID.""" - if isinstance(role_uuid, str): - role_uuid = UUID(role_uuid) - if isinstance(permission_uuid, str): - permission_uuid = UUID(permission_uuid) if role_uuid not in _db._data.roles: raise ValueError(f"Role {role_uuid} not found") if permission_uuid not in _db._data.permissions: @@ -812,26 +752,20 @@ def add_permission_to_role( def remove_permission_from_role( - role_uuid: str | UUID, - permission_uuid: str | UUID, + role_uuid: UUID, + permission_uuid: UUID, *, ctx: SessionContext | None = None, ) -> None: """Remove permission from role by UUID.""" - if isinstance(role_uuid, str): - role_uuid = UUID(role_uuid) - if isinstance(permission_uuid, str): - permission_uuid = UUID(permission_uuid) if role_uuid not in _db._data.roles: raise ValueError(f"Role {role_uuid} not found") with _db.transaction("Revoked role permission", ctx): _db._data.roles[role_uuid].permissions.pop(permission_uuid, None) -def delete_role(uuid: str | UUID, *, ctx: SessionContext | None = None) -> None: +def delete_role(uuid: UUID, *, ctx: SessionContext | None = None) -> None: """Delete a role.""" - if isinstance(uuid, str): - uuid = UUID(uuid) if uuid not in _db._data.roles: raise ValueError(f"Role {uuid} not found") # Check no users have this role @@ -852,7 +786,7 @@ def create_user(new_user: User, *, ctx: SessionContext | None = None) -> None: def update_user_display_name( - uuid: str | UUID, + uuid: UUID, display_name: str, *, ctx: SessionContext | None = None, @@ -873,16 +807,12 @@ def update_user_display_name( def update_user_role( - uuid: str | UUID, - role_uuid: str | UUID, + uuid: UUID, + role_uuid: UUID, *, ctx: SessionContext | None = None, ) -> None: """Update user's role.""" - if isinstance(uuid, str): - uuid = UUID(uuid) - if isinstance(role_uuid, str): - role_uuid = UUID(role_uuid) if uuid not in _db._data.users: raise ValueError(f"User {uuid} not found") if role_uuid not in _db._data.roles: @@ -892,14 +822,12 @@ def update_user_role( def update_user_role_in_organization( - user_uuid: str | UUID, + user_uuid: UUID, role_name: str, *, ctx: SessionContext | None = None, ) -> None: """Update user's role by role name within their current organization.""" - if isinstance(user_uuid, str): - user_uuid = UUID(user_uuid) if user_uuid not in _db._data.users: raise ValueError(f"User {user_uuid} not found") current_role_uuid = _db._data.users[user_uuid].role @@ -918,10 +846,8 @@ def update_user_role_in_organization( _db._data.users[user_uuid].role = new_role_uuid -def delete_user(uuid: str | UUID, *, ctx: SessionContext | None = None) -> None: +def delete_user(uuid: UUID, *, ctx: SessionContext | None = None) -> None: """Delete user and their credentials/sessions.""" - if isinstance(uuid, str): - uuid = UUID(uuid) if uuid not in _db._data.users: raise ValueError(f"User {uuid} not found") with _db.transaction("Deleted user", ctx): @@ -951,15 +877,13 @@ def create_credential(cred: Credential, *, ctx: SessionContext | None = None) -> def update_credential_sign_count( - uuid: str | UUID, + uuid: UUID, sign_count: int, last_used: datetime | None = None, *, ctx: SessionContext | None = None, ) -> None: """Update credential sign count and last_used.""" - if isinstance(uuid, str): - uuid = UUID(uuid) if uuid not in _db._data.credentials: raise ValueError(f"Credential {uuid} not found") with _db.transaction("Updated credential", ctx): @@ -969,8 +893,8 @@ def update_credential_sign_count( def delete_credential( - uuid: str | UUID, - user_uuid: str | UUID | None = None, + uuid: UUID, + user_uuid: UUID | None = None, *, ctx: SessionContext | None = None, ) -> None: @@ -978,13 +902,9 @@ def delete_credential( If user_uuid is provided, validates that the credential belongs to that user. """ - if isinstance(uuid, str): - uuid = UUID(uuid) if uuid not in _db._data.credentials: raise ValueError(f"Credential {uuid} not found") if user_uuid is not None: - if isinstance(user_uuid, str): - user_uuid = UUID(user_uuid) cred_user = _db._data.credentials[uuid].user if cred_user != user_uuid: raise ValueError(f"Credential {uuid} does not belong to user {user_uuid}") @@ -1069,15 +989,13 @@ def delete_session(key: str, *, ctx: SessionContext | None = None) -> None: def delete_sessions_for_user( - user_uuid: str | UUID, *, ctx: SessionContext | None = None + user_uuid: UUID, *, ctx: SessionContext | None = None ) -> None: """Delete all sessions for a user. For logout-all (user deleting own sessions), ctx can be None and user is derived from user_uuid. For admin operations, ctx should be provided. """ - if isinstance(user_uuid, str): - user_uuid = UUID(user_uuid) # For self-service, derive user from the user_uuid param user_str = str(user_uuid) if not ctx else None with _db.transaction("Deleted user sessions", ctx, user=user_str): @@ -1154,7 +1072,7 @@ def _create_token() -> str: def login( - user_uuid: str | UUID, + user_uuid: UUID, credential: Credential, host: str | None, ip: str | None, diff --git a/paskia/fastapi/admin.py b/paskia/fastapi/admin.py index 7a38ad6..64c0c7f 100644 --- a/paskia/fastapi/admin.py +++ b/paskia/fastapi/admin.py @@ -105,7 +105,7 @@ async def admin_list_orgs(request: Request, auth=AUTH_COOKIE): } async def org_to_dict(o): - users = db.get_organization_users(str(o.uuid)) + users = db.get_organization_users(o.uuid) return { "uuid": str(o.uuid), "display_name": o.display_name, @@ -209,30 +209,52 @@ async def admin_delete_org(org_uuid: UUID, request: Request, auth=AUTH_COOKIE): @app.post("/orgs/{org_uuid}/permission") async def admin_add_org_permission( org_uuid: UUID, - permission_id: str, request: Request, + permission_id: str = Query(...), auth=AUTH_COOKIE, ): ctx = await authz.verify( auth, ["auth:admin"], host=request.headers.get("host"), match=permutil.has_all ) - db.add_permission_to_organization(str(org_uuid), permission_id, ctx=ctx) + + # Convert permission_id to UUID + try: + permission_uuid = UUID(permission_id) + except ValueError: + # It's a scope - look up the UUID + perm = db.get_permission_by_scope(permission_id) + if not perm: + raise HTTPException(status_code=404, detail="Permission not found") + permission_uuid = perm.uuid + + db.add_permission_to_organization(org_uuid, permission_uuid, ctx=ctx) return {"status": "ok"} @app.delete("/orgs/{org_uuid}/permission") async def admin_remove_org_permission( org_uuid: UUID, - permission_id: str, request: Request, + permission_id: str = Query(...), auth=AUTH_COOKIE, ): ctx = await authz.verify( auth, ["auth:admin"], host=request.headers.get("host"), match=permutil.has_all ) + # Convert permission_id to UUID + try: + permission_uuid = UUID(permission_id) + except ValueError: + # It's a scope - look up the UUID + perm = db.get_permission_by_scope(permission_id) + if not perm: + raise HTTPException(status_code=404, detail="Permission not found") + permission_uuid = perm.uuid + # Guard rail: prevent removing auth:admin from your own org if it would lock you out - if permission_id == "auth:admin" and ctx.org.uuid == org_uuid: + perm = db.get_permission(permission_uuid) + if perm and perm.scope == "auth:admin" and ctx.org.uuid == org_uuid: # Check if any other org grants auth:admin that we're a member of # (we only know our current org, so this effectively means we can't remove it from our own org) raise ValueError( @@ -240,7 +262,7 @@ async def admin_remove_org_permission( "This would lock you out of admin access." ) - db.remove_permission_from_organization(str(org_uuid), permission_id, ctx=ctx) + db.remove_permission_from_organization(org_uuid, permission_uuid, ctx=ctx) return {"status": "ok"} @@ -268,7 +290,7 @@ async def admin_create_role( display_name = payload.get("display_name") or "New Role" perms = payload.get("permissions") or [] - org = db.get_organization(str(org_uuid)) + org = db.get_organization(org_uuid) grantable = org.permissions # set[UUID] computed by build_org # Normalize permission IDs to UUIDs @@ -349,7 +371,7 @@ async def admin_add_role_permission( perm = db.get_permission(permission_uuid) if not perm: raise HTTPException(status_code=404, detail="Permission not found") - org = db.get_organization(str(org_uuid)) + org = db.get_organization(org_uuid) if permission_uuid not in org.permissions: raise ValueError("Permission not grantable by organization") @@ -456,7 +478,7 @@ async def admin_create_user( raise ValueError("display_name and role are required") from ..db import User as UserDC - roles = db.get_roles_by_organization(str(org_uuid)) + roles = db.get_roles_by_organization(org_uuid) role_obj = next((r for r in roles if r.display_name == role_name), None) if not role_obj: raise ValueError("Role not found in organization") @@ -495,7 +517,7 @@ async def admin_update_user_role( raise ValueError("User not found") if user_org.uuid != org_uuid: raise ValueError("User does not belong to this organization") - roles = db.get_roles_by_organization(str(org_uuid)) + roles = db.get_roles_by_organization(org_uuid) if not any(r.display_name == new_role for r in roles): raise ValueError("Role not found in organization") diff --git a/paskia/sansio.py b/paskia/sansio.py index 4032d0e..f2ac7c8 100644 --- a/paskia/sansio.py +++ b/paskia/sansio.py @@ -12,7 +12,6 @@ from datetime import datetime, timezone from urllib.parse import urlparse from uuid import UUID -import uuid7 from webauthn import ( generate_authentication_options, generate_registration_options, diff --git a/tests/conftest.py b/tests/conftest.py index 7767f91..517b2af 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -85,7 +85,7 @@ async def test_org(test_db: DB, admin_permission: Permission) -> Org: org = Org.create(display_name="Test Organization") create_organization(org) # Grant admin permission to this org - add_permission_to_organization(str(org.uuid), str(admin_permission.uuid)) + add_permission_to_organization(org.uuid, admin_permission.uuid) return org @@ -103,7 +103,7 @@ async def org_admin_permission(test_db: DB, test_org: Org) -> Permission: perm = Permission.create(scope="auth:org:admin", display_name="Organization Admin") create_permission(perm) # Make it grantable by the org - add_permission_to_organization(str(test_org.uuid), "auth:org:admin") + add_permission_to_organization(test_org.uuid, perm.uuid) return perm diff --git a/tests/test_admin.py b/tests/test_admin.py index 5ac70c6..eb839d8 100644 --- a/tests/test_admin.py +++ b/tests/test_admin.py @@ -176,7 +176,7 @@ async def grantable_permission(test_db: DB, test_org: Org) -> Permission: perm = Permission.create(scope="test:grantable:perm", display_name="Grantable Perm") create_permission(perm) # Add to org's grantable permissions - add_permission_to_organization(str(test_org.uuid), perm.scope) + add_permission_to_organization(test_org.uuid, perm.uuid) return perm