From e8247a2c7f41d89dd94fe2a2267b6becbd22db1f Mon Sep 17 00:00:00 2001 From: Leo Vasanko Date: Tue, 27 Jan 2026 20:01:17 +0000 Subject: [PATCH] Remove most remaining DB getters. Add ws auth chat helper function to avoid repetition, along with the existing register chat in wschat.py. --- paskia/bootstrap.py | 5 +- paskia/db/__init__.py | 23 +----- paskia/db/operations.py | 147 +++------------------------------------ paskia/db/structs.py | 9 +-- paskia/fastapi/admin.py | 69 +++++++++--------- paskia/fastapi/api.py | 2 +- paskia/fastapi/remote.py | 40 +++-------- paskia/fastapi/ws.py | 58 +++------------ paskia/fastapi/wschat.py | 58 +++++++++++++++ paskia/util/userinfo.py | 4 +- tests/test_admin.py | 16 ++--- 11 files changed, 144 insertions(+), 287 deletions(-) create mode 100644 paskia/fastapi/wschat.py diff --git a/paskia/bootstrap.py b/paskia/bootstrap.py index 0ae0335..41c3cb2 100644 --- a/paskia/bootstrap.py +++ b/paskia/bootstrap.py @@ -69,9 +69,8 @@ async def check_admin_credentials() -> bool: # Check first admin user for credentials admin_user = admin_users[0] - credentials = db.get_credentials_by_user_uuid(admin_user.uuid) - if not credentials: + if not db.get_user_credential_ids(admin_user.uuid): # Admin exists but has no credentials, create reset link from paskia import authsession from paskia.util import passphrase @@ -101,7 +100,7 @@ async def bootstrap_if_needed() -> bool: bool: True if bootstrapping was performed, False if system was already set up """ # Check if the admin permission exists - if it does, system is already bootstrapped - if db.get_permission_by_scope("auth:admin"): + if any(p.scope == "auth:admin" for p in db.data().permissions.values()): # Permission exists, system is already bootstrapped # Check if admin needs credentials (only for already-bootstrapped systems) await check_admin_credentials() diff --git a/paskia/db/__init__.py b/paskia/db/__init__.py index 9c56795..ddebe67 100644 --- a/paskia/db/__init__.py +++ b/paskia/db/__init__.py @@ -46,21 +46,12 @@ from paskia.db.operations import ( delete_session, delete_sessions_for_user, delete_user, - get_credential_by_id, - get_credentials_by_user_uuid, - get_organization, get_organization_users, - get_permission, - get_permission_by_scope, get_reset_token, - get_role, - get_roles_by_organization, get_session_context, - get_user_by_uuid, + get_user_credential_ids, get_user_organization, init, - list_organizations, - list_permissions, login, remove_permission_from_organization, remove_permission_from_role, @@ -122,21 +113,11 @@ __all__ = [ "build_session", "build_user", # Read ops - "get_credential_by_id", - "get_credentials_by_user_uuid", - "get_organization", "get_organization_users", - "get_permission", - "get_permission_by_scope", "get_reset_token", - "get_role", - "get_roles_by_organization", "get_session_context", - "get_user_by_uuid", + "get_user_credential_ids", "get_user_organization", - "list_organizations", - "list_permissions", - "list_permissions", # Write ops "add_permission_to_organization", "add_permission_to_role", diff --git a/paskia/db/operations.py b/paskia/db/operations.py index 422f6e4..c194911 100644 --- a/paskia/db/operations.py +++ b/paskia/db/operations.py @@ -53,123 +53,6 @@ async def init(*args, **kwargs): # ------------------------------------------------------------------------- -def get_permission(uuid: UUID) -> Permission | None: - """Get permission by UUID. - - Call sites: - - Normalize permission IDs to UUIDs when creating a role (admin.py:277) - - Verify permission exists when adding to role (admin.py:349) - - Check permission scope when removing from role to prevent losing admin access (admin.py:385) - - Get permission to check scope for admin access check (admin.py:392) - - Check if new role has admin permissions when user changes own role (admin.py:509) - - Get permission for updating its details (admin.py:977) - - Get permission for renaming its scope (admin.py:1031) - - Get permission to check scope before deleting (admin.py:1071) - """ - return _db.permissions.get(uuid) - - -def get_permission_by_scope(scope: str) -> Permission | None: - """Get permission by scope identifier. - - Call sites: - - Check if system is already bootstrapped by looking for auth:admin permission (bootstrap.py:113) - """ - for p in _db.permissions.values(): - if p.scope == scope: - return p - return None - - -def get_permissions_by_scope(scope: str) -> list[Permission]: - """Get all permissions with the given scope. - - Since scopes are not unique, this returns all matching permissions. - Use this for scope-based permission checking. - """ - return [p for p in _db.permissions.values() if p.scope == scope] - - -def list_permissions() -> list[Permission]: - """List all permissions. - - Call sites: - - List permissions during migration to identify org-specific admin permissions (migrate/__init__.py:84) - - List permissions to delete organization-specific permissions when deleting org (admin.py:193) - - List permissions to check admin permissions when updating permission domain (admin.py:847) - - List permissions to check admin permissions when deleting permission (admin.py:882) - - Admin API endpoint to list permissions (admin.py:914) - """ - return list(_db.permissions.values()) - - -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) - """ - return _db.orgs.get(uuid) - - -def list_organizations() -> list[Org]: - """List all organizations. - - Call sites: - - List organizations during migration (migrate/__init__.py:131) - - Admin API endpoint to list organizations (admin.py:94) - """ - return list(_db.orgs.values()) - - -def get_organization_users(org_uuid: UUID) -> list[tuple[User, str]]: - """Get all users in an organization with their role names. - - Call sites: - - Get users for each organization in the admin list orgs API (admin.py:108) - - 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) - """ - role_map = { - rid: r.display_name for rid, r in _db.roles.items() if r.org == org_uuid - } - return [(u, role_map[u.role]) for u in _db.users.values() if u.role in role_map] - - -def get_role(uuid: UUID) -> Role | None: - """Get role by UUID. - - Call sites: - - Get role to update its display name (admin.py:312) - - Get role to add permission to it (admin.py:344) - - Get role to remove permission from it (admin.py:380) - - Get role to delete it (admin.py:421) - """ - return _db.roles.get(uuid) - - -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) - """ - return [r for r in _db.roles.values() if r.org == org_uuid] - - -def get_user_by_uuid(uuid: UUID) -> User | None: - """Get user by UUID. - - Call sites: - - Get user for WebAuthn credential registration (ws.py:68) - - Get user from reset token for registration info (api.py:127) - - Get user for listing user credentials in admin API (admin.py:594) - """ - return _db.users.get(uuid) - - def get_user_organization(user_uuid: UUID) -> tuple[Org, str]: """Get the organization a user belongs to and their role name. @@ -193,31 +76,23 @@ def get_user_organization(user_uuid: UUID) -> tuple[Org, str]: return _db.orgs[org_uuid], role_data.display_name -def get_credential_by_id(credential_id: bytes) -> Credential | None: - """Get credential by credential_id (the authenticator's ID). +def get_organization_users(org_uuid: UUID) -> list[tuple[User, str]]: + """Get all users in an organization with their role names. - Call sites: - - Get credential by ID for WebAuthn authentication (ws.py:132) - - Get credential by ID for remote authentication (remote.py:325) + Returns list of (User, role_display_name) tuples. """ - for c in _db.credentials.values(): - if c.credential_id == credential_id: - return c - return None + role_map = { + rid: r.display_name for rid, r in _db.roles.items() if r.org == org_uuid + } + return [(u, role_map[u.role]) for u in _db.users.values() if u.role in role_map] -def get_credentials_by_user_uuid(user_uuid: UUID) -> list[Credential]: - """Get all credentials for a user. +def get_user_credential_ids(user_uuid: UUID) -> list[bytes]: + """Get credential IDs for a user (for WebAuthn exclude lists). - Call sites: - - Get credentials for user during registration to exclude existing ones (ws.py:74) - - Get credentials for session user during reauth to restrict to user's credentials (ws.py:117) - - Get credentials to check if user has existing ones for reset token type (admin.py:548) - - Get credentials for user details API (admin.py:595) - - Get credentials to check if admin user has credentials (bootstrap.py:81) - - Get credentials for user info formatting (userinfo.py:51) + Returns empty list if user has no credentials. """ - return [c for c in _db.credentials.values() if c.user == user_uuid] + return [c.credential_id for c in _db.credentials.values() if c.user == user_uuid] def _reset_key(passphrase: str) -> bytes: diff --git a/paskia/db/structs.py b/paskia/db/structs.py index 78e8605..578c2f0 100644 --- a/paskia/db/structs.py +++ b/paskia/db/structs.py @@ -1,4 +1,4 @@ -from datetime import datetime +from datetime import datetime, timezone from uuid import UUID import msgspec @@ -76,13 +76,11 @@ class Org(msgspec.Struct, dict=True, omit_defaults=True): @classmethod def create(cls, display_name: str) -> "Org": """Create a new Org with auto-generated uuid7.""" - from datetime import timezone - org = cls( display_name=display_name, created_at=datetime.now(timezone.utc), ) - org.uuid = uuid7.create() + org.uuid = uuid7.create(org.created_at) return org @@ -104,7 +102,6 @@ class User(msgspec.Struct, dict=True): created_at: datetime | None = None, ) -> "User": """Create a new User with auto-generated uuid7.""" - from datetime import timezone user = cls( display_name=display_name, @@ -139,8 +136,6 @@ class Credential(msgspec.Struct, dict=True): created_at: datetime | None = None, ) -> "Credential": """Create a new Credential with auto-generated uuid7.""" - from datetime import timezone - now = created_at or datetime.now(timezone.utc) cred = cls( credential_id=credential_id, diff --git a/paskia/fastapi/admin.py b/paskia/fastapi/admin.py index 5bc2672..e7f0a5d 100644 --- a/paskia/fastapi/admin.py +++ b/paskia/fastapi/admin.py @@ -91,7 +91,7 @@ async def admin_list_orgs(request: Request, auth=AUTH_COOKIE): match=permutil.has_any, host=request.headers.get("host"), ) - orgs = db.list_organizations() + orgs = list(db.data().orgs.values()) if not is_global_admin(ctx): # Org admins can only see their own organization orgs = [o for o in orgs if o.uuid == ctx.org.uuid] @@ -194,7 +194,7 @@ async def admin_delete_org(org_uuid: UUID, request: Request, auth=AUTH_COOKIE): # Delete organization-specific permissions org_perm_pattern = f"org:{str(org_uuid).lower()}" - all_permissions = db.list_permissions() + all_permissions = list(db.data().permissions.values()) for perm in all_permissions: perm_scope_lower = perm.scope.lower() # Check if permission contains "org:{uuid}" separated by colons or at boundaries @@ -226,7 +226,10 @@ async def admin_add_org_permission( permission_uuid = UUID(permission_id) except ValueError: # It's a scope - look up the UUID - perm = db.get_permission_by_scope(permission_id) + perm = next( + (p for p in db.data().permissions.values() if p.scope == permission_id), + None, + ) if not perm: raise HTTPException(status_code=404, detail="Permission not found") permission_uuid = perm.uuid @@ -251,13 +254,16 @@ async def admin_remove_org_permission( permission_uuid = UUID(permission_id) except ValueError: # It's a scope - look up the UUID - perm = db.get_permission_by_scope(permission_id) + perm = next( + (p for p in db.data().permissions.values() if p.scope == permission_id), + None, + ) 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 - perm = db.get_permission(permission_uuid) + perm = db.data().permissions.get(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) @@ -294,15 +300,14 @@ async def admin_create_role( display_name = payload.get("display_name") or "New Role" perms = payload.get("permissions") or [] - org = db.get_organization(org_uuid) - if not org: + if org_uuid not in db.data().orgs: raise HTTPException(status_code=404, detail="Organization not found") grantable = {pid for pid, p in db.data().permissions.items() if org_uuid in p.orgs} # Normalize permission IDs to UUIDs permission_uuids: set[UUID] = set() for pid in perms: - perm = db.get_permission(UUID(pid)) + perm = db.data().permissions.get(UUID(pid)) if not perm: raise ValueError(f"Permission {pid} not found") if perm.uuid not in grantable: @@ -337,8 +342,8 @@ async def admin_update_role_name( raise authz.AuthException( status_code=403, detail="Insufficient permissions", mode="forbidden" ) - role = db.get_role(role_uuid) - if role.org != org_uuid: + role = db.data().roles.get(role_uuid) + if not role or role.org != org_uuid: raise HTTPException(status_code=404, detail="Role not found in organization") display_name = payload.get("display_name") @@ -369,12 +374,12 @@ async def admin_add_role_permission( status_code=403, detail="Insufficient permissions", mode="forbidden" ) - role = db.get_role(role_uuid) - if role.org != org_uuid: + role = db.data().roles.get(role_uuid) + if not role or role.org != org_uuid: raise HTTPException(status_code=404, detail="Role not found in organization") # Verify permission exists and org can grant it - perm = db.get_permission(permission_uuid) + perm = db.data().permissions.get(permission_uuid) if not perm: raise HTTPException(status_code=404, detail="Permission not found") if org_uuid not in perm.orgs: @@ -404,19 +409,19 @@ async def admin_remove_role_permission( status_code=403, detail="Insufficient permissions", mode="forbidden" ) - role = db.get_role(role_uuid) - if role.org != org_uuid: + role = db.data().roles.get(role_uuid) + if not role or role.org != org_uuid: raise HTTPException(status_code=404, detail="Role not found in organization") # Sanity check: prevent admin from removing their own access - perm = db.get_permission(permission_uuid) + perm = db.data().permissions.get(permission_uuid) if ctx.org.uuid == org_uuid and ctx.role.uuid == role_uuid: if perm and perm.scope in ["auth:admin", "auth:org:admin"]: # Check if removing this permission would leave no admin access remaining_perms = role.permission_set - {permission_uuid} has_admin = False for rp_uuid in remaining_perms: - rp = db.get_permission(rp_uuid) + rp = db.data().permissions.get(rp_uuid) if rp and rp.scope in ["auth:admin", "auth:org:admin"]: has_admin = True break @@ -445,8 +450,8 @@ async def admin_delete_role( raise authz.AuthException( status_code=403, detail="Insufficient permissions", mode="forbidden" ) - role = db.get_role(role_uuid) - if role.org != org_uuid: + role = db.data().roles.get(role_uuid) + if not role or role.org != org_uuid: raise HTTPException(status_code=404, detail="Role not found in organization") # Sanity check: prevent admin from deleting their own role @@ -483,7 +488,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(org_uuid) + roles = [r for r in db.data().roles.values() if r.org == 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") @@ -522,7 +527,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(org_uuid) + roles = [r for r in db.data().roles.values() if r.org == org_uuid] if not any(r.display_name == new_role for r in roles): raise ValueError("Role not found in organization") @@ -533,7 +538,7 @@ async def admin_update_user_role( # Check if any permission in the new role is an admin permission has_admin_access = False for perm_uuid in new_role_obj.permissions: - perm = db.get_permission(perm_uuid) + perm = db.data().permissions.get(perm_uuid) if perm and perm.scope in ["auth:admin", "auth:org:admin"]: has_admin_access = True break @@ -572,8 +577,8 @@ async def admin_create_user_registration_link( ) # Check if user has existing credentials - credentials = db.get_credentials_by_user_uuid(user_uuid) - token_type = "user registration" if not credentials else "account recovery" + has_credentials = db.get_user_credential_ids(user_uuid) + token_type = "user registration" if not has_credentials else "account recovery" token = passphrase.generate() expiry = reset_expires() @@ -618,8 +623,8 @@ async def admin_get_user_detail( raise authz.AuthException( status_code=403, detail="Insufficient permissions", mode="forbidden" ) - user = db.get_user_by_uuid(user_uuid) - user_creds = db.get_credentials_by_user_uuid(user_uuid) + user = db.data().users.get(user_uuid) + user_creds = [c for c in db.data().credentials.values() if c.user == user_uuid] creds: list[dict] = [] aaguids: set[str] = set() for c in user_creds: @@ -871,7 +876,7 @@ def _check_admin_lockout( host_without_port = normalized_host.rsplit(":", 1)[0] if normalized_host else None # Get all auth:admin permissions - all_perms = db.list_permissions() + all_perms = list(db.data().permissions.values()) admin_perms = [p for p in all_perms if p.scope == "auth:admin"] # Check if at least one auth:admin would remain accessible @@ -906,7 +911,7 @@ def _check_admin_lockout_on_delete(perm_uuid: str, current_host: str | None) -> host_without_port = normalized_host.rsplit(":", 1)[0] if normalized_host else None # Get all auth:admin permissions - all_perms = db.list_permissions() + all_perms = list(db.data().permissions.values()) admin_perms = [p for p in all_perms if p.scope == "auth:admin"] # Check if at least one auth:admin would remain accessible after deletion @@ -938,7 +943,7 @@ async def admin_list_permissions(request: Request, auth=AUTH_COOKIE): match=permutil.has_any, host=request.headers.get("host"), ) - perms = db.list_permissions() + perms = list(db.data().permissions.values()) # Global admins see all permissions if is_global_admin(ctx): @@ -997,7 +1002,7 @@ async def admin_update_permission( ) # Get existing permission - perm = db.get_permission(permission_uuid) + perm = db.data().permissions.get(permission_uuid) # Update fields that were provided new_scope = scope if scope is not None else perm.scope @@ -1045,7 +1050,7 @@ async def admin_rename_permission( raise ValueError("new_scope required") # Sanity check: prevent renaming critical permissions - perm = db.get_permission(permission_uuid) + perm = db.data().permissions.get(permission_uuid) if perm.scope == "auth:admin": raise ValueError("Cannot rename the master admin permission") @@ -1086,7 +1091,7 @@ async def admin_delete_permission( ) # Get the permission to check its scope - perm = db.get_permission(permission_uuid) + perm = db.data().permissions.get(permission_uuid) # Sanity check: prevent deleting critical permissions if it would lock out admin if perm.scope == "auth:admin": diff --git a/paskia/fastapi/api.py b/paskia/fastapi/api.py index 2d717af..5bb37da 100644 --- a/paskia/fastapi/api.py +++ b/paskia/fastapi/api.py @@ -124,7 +124,7 @@ async def token_info(credentials=Depends(bearer_auth)): except ValueError as e: raise HTTPException(401, str(e)) - u = db.get_user_by_uuid(reset_token.user) + u = db.data().users.get(reset_token.user) return { "token_type": reset_token.token_type, "display_name": u.display_name, diff --git a/paskia/fastapi/remote.py b/paskia/fastapi/remote.py index b64a1ba..f387fbc 100644 --- a/paskia/fastapi/remote.py +++ b/paskia/fastapi/remote.py @@ -17,8 +17,8 @@ from fastapi import FastAPI, WebSocket, WebSocketDisconnect from paskia import db, remoteauth from paskia.fastapi.session import infodict +from paskia.fastapi.wschat import authenticate_chat from paskia.fastapi.wsutil import validate_origin, websocket_error_handler -from paskia.globals import passkey from paskia.util import passphrase, pow # Create a FastAPI subapp for remote auth WebSocket endpoints @@ -311,30 +311,10 @@ async def websocket_remote_auth_permit(ws: WebSocket): # Handle authenticate request (no PoW needed - already validated during lookup) if msg.get("authenticate") and request is not None: - # Generate authentication options - options, webauthn_challenge = passkey.instance.auth_generate_options( - credential_ids=None - ) - await ws.send_json({"optionsJSON": options}) - - # Wait for WebAuthn response - credential = passkey.instance.auth_parse(await ws.receive_json()) - - # Fetch and verify credential - try: - stored_cred = db.get_credential_by_id(credential.raw_id) - except ValueError: - raise ValueError( - f"This passkey is no longer registered with {passkey.instance.rp_name}" - ) - - # Verify the credential - passkey.instance.auth_verify( - credential, webauthn_challenge, stored_cred, origin - ) + cred = await authenticate_chat(ws, origin) # Create a session for the REQUESTING device - assert stored_cred.uuid is not None + assert cred.uuid is not None session_token = None reset_token = None @@ -347,7 +327,7 @@ async def websocket_remote_auth_permit(ws: WebSocket): token_str = passphrase.generate() expiry = expires() db.create_reset_token( - user_uuid=stored_cred.user, + user_uuid=cred.user, passphrase=token_str, expiry=expiry, token_type="device addition", @@ -356,8 +336,8 @@ async def websocket_remote_auth_permit(ws: WebSocket): # Also create a session so the device is logged in normalized_host = hostutil.normalize_host(request.host) session_token = db.login( - user_uuid=stored_cred.user, - credential=stored_cred, + user_uuid=cred.user, + credential=cred, host=normalized_host, ip=request.ip, user_agent=request.user_agent, @@ -370,8 +350,8 @@ async def websocket_remote_auth_permit(ws: WebSocket): normalized_host = hostutil.normalize_host(request.host) session_token = db.login( - user_uuid=stored_cred.user, - credential=stored_cred, + user_uuid=cred.user, + credential=cred, host=normalized_host, ip=request.ip, user_agent=request.user_agent, @@ -382,8 +362,8 @@ async def websocket_remote_auth_permit(ws: WebSocket): completed = await remoteauth.instance.complete_request( token=request.key, session_token=session_token, - user_uuid=stored_cred.user, - credential_uuid=stored_cred.uuid, + user_uuid=cred.user, + credential_uuid=cred.uuid, reset_token=reset_token, ) diff --git a/paskia/fastapi/ws.py b/paskia/fastapi/ws.py index de45265..deb8e3d 100644 --- a/paskia/fastapi/ws.py +++ b/paskia/fastapi/ws.py @@ -1,11 +1,10 @@ -from uuid import UUID - from fastapi import FastAPI, WebSocket from paskia import db from paskia.authsession import expires, get_reset, get_session from paskia.fastapi import authz, remote from paskia.fastapi.session import AUTH_COOKIE, infodict +from paskia.fastapi.wschat import authenticate_chat, register_chat from paskia.fastapi.wsutil import validate_origin, websocket_error_handler from paskia.globals import passkey from paskia.util import hostutil, passphrase @@ -17,24 +16,6 @@ app = FastAPI() app.mount("/remote-auth", remote.app) -async def register_chat( - ws: WebSocket, - user_uuid: UUID, - user_name: str, - origin: str, - credential_ids: list[bytes] | None = None, -): - """Generate registration options and send them to the client.""" - options, challenge = passkey.instance.reg_generate_options( - user_id=user_uuid, - user_name=user_name, - credential_ids=credential_ids, - ) - await ws.send_json({"optionsJSON": options}) - response = await ws.receive_json() - return passkey.instance.reg_verify(response, challenge, user_uuid, origin=origin) - - @app.websocket("/register") @websocket_error_handler async def websocket_register_add( @@ -65,14 +46,13 @@ async def websocket_register_add( s = ctx.session # Get user information and determine effective user_name for this registration - user = db.get_user_by_uuid(user_uuid) + user = db.data().users.get(user_uuid) user_name = user.display_name if name is not None: stripped = name.strip() if stripped: user_name = stripped - credentials = db.get_credentials_by_user_uuid(user_uuid) - credential_ids = [c.credential_id for c in credentials] if credentials else None + credential_ids = db.get_user_credential_ids(user_uuid) or None # WebAuthn registration credential = await register_chat(ws, user_uuid, user_name, origin, credential_ids) @@ -114,36 +94,18 @@ async def websocket_authenticate(ws: WebSocket, auth=AUTH_COOKIE): try: session = await get_session(auth, host=host) session_user_uuid = session.user - credentials = db.get_credentials_by_user_uuid(session_user_uuid) - credential_ids = ( - [c.credential_id for c in credentials] if credentials else None - ) + credential_ids = db.get_user_credential_ids(session_user_uuid) or None except ValueError: pass # Invalid/expired session - allow normal authentication - options, challenge = passkey.instance.auth_generate_options( - credential_ids=credential_ids - ) - await ws.send_json({"optionsJSON": options}) - # Wait for the client to use his authenticator to authenticate - credential = passkey.instance.auth_parse(await ws.receive_json()) - # Fetch from the database by credential ID - try: - stored_cred = db.get_credential_by_id(credential.raw_id) - except ValueError: - raise ValueError( - f"This passkey is no longer registered with {passkey.instance.rp_name}" - ) + cred = await authenticate_chat(ws, origin, credential_ids) # If reauth mode, verify the credential belongs to the session's user - if session_user_uuid and stored_cred.user != session_user_uuid: + if session_user_uuid and cred.user != session_user_uuid: raise ValueError("This passkey belongs to a different account") - # Verify the credential matches the stored data - passkey.instance.auth_verify(credential, challenge, stored_cred, origin) - # Create session and update user/credential in a single transaction - assert stored_cred.uuid is not None + assert cred.uuid is not None metadata = infodict(ws, "auth") normalized_host = hostutil.normalize_host(host) if not normalized_host: @@ -154,8 +116,8 @@ async def websocket_authenticate(ws: WebSocket, auth=AUTH_COOKIE): raise ValueError(f"Host must be the same as or a subdomain of {rp_id}") token = db.login( - user_uuid=stored_cred.user, - credential=stored_cred, + user_uuid=cred.user, + credential=cred, host=normalized_host, ip=metadata.get("ip") or "", user_agent=metadata.get("user_agent") or "", @@ -164,7 +126,7 @@ async def websocket_authenticate(ws: WebSocket, auth=AUTH_COOKIE): await ws.send_json( { - "user": str(stored_cred.user), + "user": str(cred.user), "session_token": token, } ) diff --git a/paskia/fastapi/wschat.py b/paskia/fastapi/wschat.py new file mode 100644 index 0000000..5443264 --- /dev/null +++ b/paskia/fastapi/wschat.py @@ -0,0 +1,58 @@ +""" +WebSocket chat functions for WebAuthn registration and authentication flows. +""" + +from uuid import UUID + +from fastapi import WebSocket + +from paskia import db +from paskia.db import Credential +from paskia.globals import passkey + + +async def register_chat( + ws: WebSocket, + user_uuid: UUID, + user_name: str, + origin: str, + credential_ids: list[bytes] | None = None, +): + """Run WebAuthn registration flow and return the verified credential.""" + options, challenge = passkey.instance.reg_generate_options( + user_id=user_uuid, + user_name=user_name, + credential_ids=credential_ids, + ) + await ws.send_json({"optionsJSON": options}) + response = await ws.receive_json() + return passkey.instance.reg_verify(response, challenge, user_uuid, origin=origin) + + +async def authenticate_chat( + ws: WebSocket, + origin: str, + credential_ids: list[bytes] | None = None, +) -> Credential: + """Run WebAuthn authentication flow and return the verified credential.""" + options, challenge = passkey.instance.auth_generate_options( + credential_ids=credential_ids + ) + await ws.send_json({"optionsJSON": options}) + authcred = passkey.instance.auth_parse(await ws.receive_json()) + + cred = next( + ( + c + for c in db.data().credentials.values() + if c.credential_id == authcred.raw_id + ), + None, + ) + if not cred: + raise ValueError( + f"This passkey is no longer registered with {passkey.instance.rp_name}" + ) + + passkey.instance.auth_verify(authcred, challenge, cred, origin) + return cred diff --git a/paskia/util/userinfo.py b/paskia/util/userinfo.py index 52e1f1f..6f6c833 100644 --- a/paskia/util/userinfo.py +++ b/paskia/util/userinfo.py @@ -48,7 +48,9 @@ async def format_user_info( ctx = await permutil.session_context(auth, request_host) # Fetch and format credentials - user_credentials = db.get_credentials_by_user_uuid(user_uuid) + user_credentials = [ + c for c in db.data().credentials.values() if c.user == user_uuid + ] credentials: list[dict] = [] user_aaguids: set[str] = set() diff --git a/tests/test_admin.py b/tests/test_admin.py index eb839d8..57a5ba4 100644 --- a/tests/test_admin.py +++ b/tests/test_admin.py @@ -1423,9 +1423,9 @@ class TestAdminPermissions: ): """Cannot rename the auth:admin permission.""" # Get the auth:admin permission - from paskia.db import list_permissions + from paskia import db - perms = list_permissions() + perms = list(db.data().permissions.values()) admin_perm = next(p for p in perms if p.scope == "auth:admin") response = await client.post( @@ -1478,9 +1478,9 @@ class TestAdminPermissions: ): """Cannot delete the only auth:admin permission (would lock out admin).""" # Get the auth:admin permission - from paskia.db import list_permissions + from paskia import db - perms = list_permissions() + perms = list(db.data().permissions.values()) admin_perm = next(p for p in perms if p.scope == "auth:admin") response = await client.delete( @@ -1503,9 +1503,9 @@ class TestAdminPermissions: create_permission(perm2) # Get the original auth:admin permission (the one created in setup) - from paskia.db import list_permissions + from paskia import db - perms = list_permissions() + perms = list(db.data().permissions.values()) admin_perms = [p for p in perms if p.scope == "auth:admin"] # Delete the first one (not the one we just created) original_admin_perm = next(p for p in admin_perms if p.uuid != perm2.uuid) @@ -1536,9 +1536,9 @@ class TestAdminPermissions: # Cannot delete the original one because the remaining one is not accessible # Get the original auth:admin permission - from paskia.db import list_permissions + from paskia import db - perms = list_permissions() + perms = list(db.data().permissions.values()) admin_perms = [p for p in perms if p.scope == "auth:admin" and p.domain is None] original_admin_perm = admin_perms[0] # The one without domain