From 31f40d874cad9270ef12f79547537d3699e77636 Mon Sep 17 00:00:00 2001 From: Leo Vasanko Date: Tue, 27 Jan 2026 18:23:17 +0000 Subject: [PATCH] DB cleanup: removed get_permission_organizations and build_org. Using db.data() for read access at call sites. --- paskia/bootstrap.py | 10 ++++++---- paskia/db/__init__.py | 12 ++++------- paskia/db/operations.py | 36 +++++---------------------------- paskia/fastapi/admin.py | 19 ++++++++++++------ paskia/fastapi/reset.py | 44 +++++++++++++++++++++++++---------------- 5 files changed, 55 insertions(+), 66 deletions(-) diff --git a/paskia/bootstrap.py b/paskia/bootstrap.py index 4f09ffe..0ae0335 100644 --- a/paskia/bootstrap.py +++ b/paskia/bootstrap.py @@ -53,13 +53,15 @@ async def check_admin_credentials() -> bool: """ try: # Get permission organizations to find admin users - permission_orgs = db.get_permission_organizations("auth:admin") - - if not permission_orgs: + p = next( + (p for p in db.data().permissions.values() if p.scope == "auth:admin"), None + ) + if not p or not p.orgs: return False # Get users from the first organization with admin permission - org_users = db.get_organization_users(str(permission_orgs[0].uuid)) + first_org_uuid = next(iter(p.orgs)) + org_users = db.get_organization_users(first_org_uuid) admin_users = [user for user, role in org_users if role == "Administration"] if not admin_users: diff --git a/paskia/db/__init__.py b/paskia/db/__init__.py index ce870f1..a478c27 100644 --- a/paskia/db/__init__.py +++ b/paskia/db/__init__.py @@ -1,7 +1,7 @@ """ Database module for WebAuthn passkey authentication. -Read: Access db() directly, use build_* to convert to public structs. +Read: Access data() directly, use build_* to convert to public structs. CTX: get_session_context(key) returns SessionContext with effective permissions. Write: Functions validate and commit, or raise ValueError. @@ -9,7 +9,7 @@ Usage: from paskia import db # Read (after init) - user_data = db.db().users[user_uuid] + user_data = db.data().users[user_uuid] user = db.build_user(user_uuid) # Context @@ -29,7 +29,6 @@ from paskia.db.operations import ( add_permission_to_organization, add_permission_to_role, bootstrap, - build_org, cleanup_expired, create_credential, create_credential_session, @@ -53,7 +52,6 @@ from paskia.db.operations import ( get_organization_users, get_permission, get_permission_by_scope, - get_permission_organizations, get_reset_token, get_role, get_roles_by_organization, @@ -92,7 +90,7 @@ from paskia.db.structs import ( ) -def db() -> DB: +def data() -> DB: """Get the database instance for direct read access.""" from paskia.db.operations import _db @@ -111,7 +109,7 @@ __all__ = [ "SessionContext", "User", # Instance - "db", + "data", "init", # Background "start_background", @@ -120,7 +118,6 @@ __all__ = [ "stop_cleanup", # Builders "build_credential", - "build_org", "build_permission", "build_reset_token", "build_role", @@ -133,7 +130,6 @@ __all__ = [ "get_organization_users", "get_permission", "get_permission_by_scope", - "get_permission_organizations", "get_reset_token", "get_role", "get_roles_by_organization", diff --git a/paskia/db/operations.py b/paskia/db/operations.py index 5edd6c3..92203ec 100644 --- a/paskia/db/operations.py +++ b/paskia/db/operations.py @@ -48,19 +48,6 @@ async def init(*args, **kwargs): _db = _store.db -# ------------------------------------------------------------------------- -# Builders: Convert internal _*Data to public structs -# ------------------------------------------------------------------------- - - -def build_org(uuid: UUID, include_roles: bool = False) -> Org: - o = _db.orgs[uuid] - o.permissions = {pid for pid, p in _db.permissions.items() if uuid in p.orgs} - if include_roles: - o.roles = [_db.roles[rid] for rid, r in _db.roles.items() if r.org == uuid] - return o - - # ------------------------------------------------------------------------- # Read/lookup functions # ------------------------------------------------------------------------- @@ -116,19 +103,6 @@ def list_permissions() -> list[Permission]: return list(_db.permissions.values()) -def get_permission_organizations(scope: str) -> list[Org]: - """Get organizations that can grant a permission scope. - - Call sites: - - Get organizations that can grant auth:admin to find admin users (bootstrap.py:67) - - Get organizations with auth:admin permission to find admin users for reset targets (reset.py:29,40,55) - """ - for p in _db.permissions.values(): - if p.scope == scope: - return [build_org(org_uuid) for org_uuid in p.orgs] - return [] - - def get_organization(uuid: UUID) -> Org | None: """Get organization by UUID. @@ -136,7 +110,7 @@ def get_organization(uuid: UUID) -> Org | None: - 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 build_org(uuid, include_roles=True) if uuid in _db.orgs else None + return _db.orgs.get(uuid) def list_organizations() -> list[Org]: @@ -146,7 +120,7 @@ def list_organizations() -> list[Org]: - List organizations during migration (migrate/__init__.py:131) - Admin API endpoint to list organizations (admin.py:94) """ - return [build_org(uuid, include_roles=True) for uuid in _db.orgs] + return list(_db.orgs.values()) def get_organization_users(org_uuid: UUID) -> list[tuple[User, str]]: @@ -216,7 +190,7 @@ def get_user_organization(user_uuid: UUID) -> tuple[Org, str]: raise ValueError(f"Role {role_uuid} not found") role_data = _db.roles[role_uuid] org_uuid = role_data.org - return build_org(org_uuid, include_roles=True), role_data.display_name + return _db.orgs[org_uuid], role_data.display_name def get_credential_by_id(credential_id: bytes) -> Credential | None: @@ -346,7 +320,7 @@ def get_session_context( session = _db.sessions[session_key] user = _db.users[s.user] role = _db.roles[role_uuid] - org = build_org(org_uuid) + org = _db.orgs[org_uuid] # Credential must exist (sessions are cascade-deleted when credential is deleted) if s.credential not in _db.credentials: @@ -355,7 +329,7 @@ def get_session_context( # Effective permissions: role's permissions that the org can grant # Also filter by domain if host is provided - org_perm_uuids = org.permissions # set[UUID] computed by build_org + org_perm_uuids = {pid for pid, p in _db.permissions.items() if org_uuid in p.orgs} normalized_host = normalize_host(host) host_without_port = normalized_host.rsplit(":", 1)[0] if normalized_host else None diff --git a/paskia/fastapi/admin.py b/paskia/fastapi/admin.py index 64c0c7f..9abf079 100644 --- a/paskia/fastapi/admin.py +++ b/paskia/fastapi/admin.py @@ -109,8 +109,12 @@ async def admin_list_orgs(request: Request, auth=AUTH_COOKIE): return { "uuid": str(o.uuid), "display_name": o.display_name, - "permissions": o.permissions, - "roles": [role_to_dict(r) for r in o.roles], + "permissions": { + pid for pid, p in db.data().permissions.items() if o.uuid in p.orgs + }, + "roles": [ + role_to_dict(r) for r in db.data().roles.values() if r.org == o.uuid + ], "users": [ { "uuid": str(u.uuid), @@ -291,7 +295,9 @@ 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) - grantable = org.permissions # set[UUID] computed by build_org + if not org: + 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() @@ -371,8 +377,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(org_uuid) - if permission_uuid not in org.permissions: + if org_uuid not in perm.orgs: raise ValueError("Permission not grantable by organization") db.add_permission_to_role(role_uuid, permission_uuid, ctx=ctx) @@ -940,7 +945,9 @@ async def admin_list_permissions(request: Request, auth=AUTH_COOKIE): return [_perm_to_dict(p) for p in perms] # Org admins only see permissions their org can grant (by UUID) - grantable = ctx.org.permissions # set[UUID] + grantable = { + pid for pid, p in db.data().permissions.items() if ctx.org.uuid in p.orgs + } filtered_perms = [p for p in perms if p.uuid in grantable] return [_perm_to_dict(p) for p in filtered_perms] diff --git a/paskia/fastapi/reset.py b/paskia/fastapi/reset.py index baacd8e..0ad473d 100644 --- a/paskia/fastapi/reset.py +++ b/paskia/fastapi/reset.py @@ -16,7 +16,7 @@ import asyncio from uuid import UUID from paskia import authsession as _authsession -from paskia import db as _db +from paskia import db from paskia.util import hostutil, passphrase @@ -26,23 +26,30 @@ async def _resolve_targets(query: str | None): targets: list[tuple] = [] try: q_uuid = UUID(query) - perm_orgs = _db.get_permission_organizations("auth:admin") - for o in perm_orgs: - users = _db.get_organization_users(str(o.uuid)) - for u, role_name in users: - if u.uuid == q_uuid: - return [(u, role_name)] + p = next( + (p for p in db.data().permissions.values() if p.scope == "auth:admin"), + None, + ) + if p: + for org_uuid in p.orgs: + users = db.get_organization_users(org_uuid) + for u, role_name in users: + if u.uuid == q_uuid: + return [(u, role_name)] # UUID not found among admin orgs -> fall back to substring search (rare case) except ValueError: pass # Substring search needle = query.lower() - perm_orgs = _db.get_permission_organizations("auth:admin") - for o in perm_orgs: - users = _db.get_organization_users(str(o.uuid)) - for u, role_name in users: - if needle in (u.display_name or "").lower(): - targets.append((u, role_name)) + p = next( + (p for p in db.data().permissions.values() if p.scope == "auth:admin"), None + ) + if p: + for org_uuid in p.orgs: + users = db.get_organization_users(org_uuid) + for u, role_name in users: + if needle in (u.display_name or "").lower(): + targets.append((u, role_name)) # De-duplicate seen = set() deduped = [] @@ -52,10 +59,13 @@ async def _resolve_targets(query: str | None): deduped.append((u, role_name)) return deduped # No query -> master admin - perm_orgs = _db.get_permission_organizations("auth:admin") - if not perm_orgs: + p = next( + (p for p in db.data().permissions.values() if p.scope == "auth:admin"), None + ) + if not p or not p.orgs: return [] - users = _db.get_organization_users(str(perm_orgs[0].uuid)) + first_org_uuid = next(iter(p.orgs)) + users = db.get_organization_users(first_org_uuid) admin_users = [pair for pair in users if pair[1] == "Administration"] return admin_users[:1] @@ -63,7 +73,7 @@ async def _resolve_targets(query: str | None): async def _create_reset(user, role_name: str): token = passphrase.generate() expiry = _authsession.reset_expires() - _db.create_reset_token( + db.create_reset_token( passphrase=token, user_uuid=user.uuid, expiry=expiry,