LeoVasanko/paskia
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

DB getter refactoring. Documented call sites. Added separate function for by-scope permission lookup.

Browse Source
This commit is contained in:
Leo Vasanko
2026-01-27 15:54:28 +00:00
parent 071a7b9e0b
commit bafd62a1b1
3 changed files with 205 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files
paskia/db/operations.py
+140 -49
View File
@@ -215,46 +215,69 @@ def build_reset_token(key: bytes) -> ResetToken:
# -------------------------------------------------------------------------
def get_permission(permission_id: str | UUID) -> Permission | None:
"""Get permission by UUID or scope.
def get_permission(uuid: UUID) -> Permission | None:
"""Get permission by UUID.
For backwards compatibility, this accepts either:
- A UUID string (the primary key)
- A scope string (searches for matching scope)
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)
"""
# First try as UUID key
if isinstance(permission_id, UUID):
if permission_id in _db._data.permissions:
return build_permission(permission_id)
else:
try:
uuid = UUID(permission_id)
if uuid in _db._data.permissions:
return build_permission(uuid)
except ValueError:
pass
# Fall back to scope search
for uuid, p in _db._data.permissions.items():
if p.scope == str(permission_id):
return build_permission(uuid)
if uuid in _db._data.permissions:
return build_permission(uuid)
return None
def get_permission_by_scope(scope: str) -> Permission | None:
"""Get permission by scope identifier."""
"""Get permission by scope identifier.
Call sites:
- Check if system is already bootstrapped by looking for auth:admin permission (bootstrap.py:113)
"""
for uuid, p in _db._data.permissions.items():
if p.scope == scope:
return build_permission(uuid)
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 [
build_permission(uuid)
for uuid, p in _db._data.permissions.items()
if p.scope == scope
]
def list_permissions() -> list[Permission]:
"""List all permissions."""
"""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 [build_permission(uuid) for uuid in _db._data.permissions]
def get_permission_organizations(scope: str) -> list[Org]:
"""Get organizations that can grant a permission scope."""
"""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._data.permissions.values():
if p.scope == scope:
return [build_org(org_uuid) for org_uuid in p.orgs]
@@ -262,19 +285,35 @@ def get_permission_organizations(scope: str) -> list[Org]:
def get_organization(uuid: str | UUID) -> Org | None:
"""Get organization by UUID."""
"""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
def list_organizations() -> list[Org]:
"""List all organizations."""
"""List all organizations.
Call sites:
- 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._data.orgs]
def get_organization_users(org_uuid: str | UUID) -> list[tuple[User, str]]:
"""Get all users in an organization with their role names."""
"""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)
"""
if isinstance(org_uuid, str):
org_uuid = UUID(org_uuid)
role_map = {
@@ -288,21 +327,39 @@ def get_organization_users(org_uuid: str | UUID) -> list[tuple[User, str]]:
def get_role(uuid: str | UUID) -> Role | None:
"""Get role by UUID."""
"""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)
"""
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]:
"""Get all roles in an organization."""
"""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:
"""Get user by UUID."""
"""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)
"""
if isinstance(uuid, str):
uuid = UUID(uuid)
return build_user(uuid) if uuid in _db._data.users else None
@@ -312,6 +369,14 @@ def get_user_organization(user_uuid: str | UUID) -> tuple[Org, str]:
"""Get the organization a user belongs to and their role name.
Raises ValueError if user not found.
Call sites:
- Get user's organization when updating user role (admin.py:493)
- Get user's organization for user credential listing (admin.py:530)
- Get user's organization for user details API (admin.py:579)
- Get user's organization for updating user display name (admin.py:721)
- 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)
@@ -326,7 +391,12 @@ def get_user_organization(user_uuid: str | UUID) -> tuple[Org, str]:
def get_credential_by_id(credential_id: bytes) -> Credential | None:
"""Get credential by credential_id (the authenticator's ID)."""
"""Get credential by credential_id (the authenticator's ID).
Call sites:
- Get credential by ID for WebAuthn authentication (ws.py:132)
- Get credential by ID for remote authentication (remote.py:325)
"""
for uuid, c in _db._data.credentials.items():
if c.credential_id == credential_id:
return build_credential(uuid)
@@ -334,7 +404,16 @@ def get_credential_by_id(credential_id: bytes) -> Credential | None:
def get_credentials_by_user_uuid(user_uuid: str | UUID) -> list[Credential]:
"""Get all credentials for a user."""
"""Get all credentials for a user.
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)
"""
if isinstance(user_uuid, str):
user_uuid = UUID(user_uuid)
return [
@@ -345,7 +424,14 @@ def get_credentials_by_user_uuid(user_uuid: str | UUID) -> list[Credential]:
def get_session(key: str) -> Session | None:
"""Get session by key."""
"""Get session by key.
Call sites:
- Get session to delete it (admin.py:799)
- Get session to validate token (authsession.py:45)
- Get session to refresh it (authsession.py:59)
- Get session to delete it in user API (user.py:94)
"""
if key not in _db._data.sessions:
return None
s = _db._data.sessions[key]
@@ -355,7 +441,12 @@ def get_session(key: str) -> Session | None:
def list_sessions_for_user(user_uuid: str | UUID) -> list[Session]:
"""Get all active sessions for a user."""
"""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)
@@ -378,7 +469,11 @@ def _reset_key(passphrase: str) -> bytes:
def get_reset_token(passphrase: str) -> ResetToken | None:
"""Get reset token by passphrase."""
"""Get reset token by passphrase.
Call sites:
- Get reset token to validate it (authsession.py:34)
"""
key = _reset_key(passphrase)
if key not in _db._data.reset_tokens:
return None
@@ -404,6 +499,10 @@ def get_session_context(
Returns:
SessionContext if valid, None if session not found, expired, or host mismatch
Call sites:
- Example usage in docstring (db/__init__.py:16)
- Get session context from auth token (util/permutil.py:43)
"""
from paskia.util.hostutil import normalize_host
@@ -500,7 +599,7 @@ def update_permission(perm: Permission, *, ctx: SessionContext | None = None) ->
def rename_permission(
old_scope: str,
uuid: UUID,
new_scope: str,
display_name: str,
domain: str | None = None,
@@ -512,26 +611,18 @@ def rename_permission(
Since roles reference permissions by UUID, no role updates are needed.
Note: Scopes do not need to be unique (same scope with different domains is valid).
"""
# Find permission by old scope
key = None
for pid, p in _db._data.permissions.items():
if p.scope == old_scope:
key = pid
break
if not key:
raise ValueError(f"Permission with scope '{old_scope}' not found")
if uuid not in _db._data.permissions:
raise ValueError(f"Permission {uuid} not found")
with _db.transaction("Renamed permission", ctx):
# Update the permission
_db._data.permissions[key].scope = new_scope
_db._data.permissions[key].display_name = display_name
_db._data.permissions[key].domain = domain
_db._data.permissions[uuid].scope = new_scope
_db._data.permissions[uuid].display_name = display_name
_db._data.permissions[uuid].domain = domain
def delete_permission(uuid: str | UUID, *, ctx: SessionContext | None = None) -> None:
def delete_permission(uuid: UUID, *, ctx: SessionContext | None = None) -> None:
"""Delete a permission and remove it from all roles."""
if isinstance(uuid, str):
uuid = UUID(uuid)
if uuid not in _db._data.permissions:
raise ValueError(f"Permission {uuid} not found")
with _db.transaction("Deleted permission", ctx):
paskia/fastapi/admin.py
+20 -31
View File
@@ -2,7 +2,7 @@ import logging
from datetime import timezone
from uuid import UUID
from fastapi import Body, FastAPI, HTTPException, Request, Response
from fastapi import Body, FastAPI, HTTPException, Query, Request, Response
from fastapi.responses import JSONResponse
from paskia import db
@@ -200,7 +200,7 @@ async def admin_delete_org(org_uuid: UUID, request: Request, auth=AUTH_COOKIE):
or perm_scope_lower.endswith(f":{org_perm_pattern}")
or perm_scope_lower == org_perm_pattern
):
db.delete_permission(str(perm.uuid), ctx=ctx)
db.delete_permission(perm.uuid, ctx=ctx)
db.delete_organization(org_uuid, ctx=ctx)
return {"status": "ok"}
@@ -274,7 +274,7 @@ async def admin_create_role(
# Normalize permission IDs to UUIDs
permission_uuids: set[UUID] = set()
for pid in perms:
perm = db.get_permission(pid)
perm = db.get_permission(UUID(pid))
if not perm:
raise ValueError(f"Permission {pid} not found")
if perm.uuid not in grantable:
@@ -958,23 +958,17 @@ async def admin_create_permission(
async def admin_update_permission(
request: Request,
auth=AUTH_COOKIE,
permission_uuid: str | None = None,
permission_id: str | None = None, # Backwards compat - treated as scope
display_name: str | None = None,
scope: str | None = None,
domain: str | None = None,
permission_uuid: UUID = Query(...),
display_name: str | None = Query(None),
scope: str | None = Query(None),
domain: str | None = Query(None),
):
ctx = await authz.verify(
auth, ["auth:admin"], host=request.headers.get("host"), match=permutil.has_all
)
# permission_uuid or permission_id (scope) to identify the permission
perm_identifier = permission_uuid or permission_id
if not perm_identifier:
raise ValueError("permission_uuid or permission_id required")
# Get existing permission
perm = db.get_permission(perm_identifier)
perm = db.get_permission(permission_uuid)
# Update fields that were provided
new_scope = scope if scope is not None else perm.scope
@@ -1003,32 +997,31 @@ async def admin_update_permission(
@app.post("/permission/rename")
@app.put("/permission/rename")
async def admin_rename_permission(
request: Request,
permission_uuid: UUID = Query(...),
payload: dict = Body(...),
auth=AUTH_COOKIE,
):
ctx = await authz.verify(
auth, ["auth:admin"], host=request.headers.get("host"), match=permutil.has_all
)
old_scope = payload.get("old_scope") or payload.get("old_id") # Support both
new_scope = payload.get("new_scope") or payload.get("new_id") # Support both
display_name = payload.get("display_name")
domain = payload.get(
"domain"
) # Can be None (not provided), empty string (clear), or value
if not old_scope or not new_scope:
raise ValueError("old_scope and new_scope required")
if not new_scope:
raise ValueError("new_scope required")
# Sanity check: prevent renaming critical permissions
if old_scope == "auth:admin":
perm = db.get_permission(permission_uuid)
if perm.scope == "auth:admin":
raise ValueError("Cannot rename the master admin permission")
querysafe.assert_safe(old_scope, field="old_scope")
querysafe.assert_safe(new_scope, field="new_scope")
# Get existing permission to preserve values not being changed
perm = db.get_permission(old_scope)
if display_name is None:
display_name = perm.display_name
# domain=None means "not provided, keep existing", domain="" means "clear it"
@@ -1043,16 +1036,17 @@ async def admin_rename_permission(
_check_admin_lockout(str(perm.uuid), domain_value, request.headers.get("host"))
# All current backends support rename_permission
db.rename_permission(old_scope, new_scope, display_name, domain_value, ctx=ctx)
db.rename_permission(
permission_uuid, new_scope, display_name, domain_value, ctx=ctx
)
return {"status": "ok"}
@app.delete("/permission")
async def admin_delete_permission(
request: Request,
permission_uuid: UUID = Query(...),
auth=AUTH_COOKIE,
permission_uuid: str | None = None,
permission_id: str | None = None, # Backwards compat - treated as scope
):
ctx = await authz.verify(
auth,
@@ -1062,17 +1056,12 @@ async def admin_delete_permission(
max_age="5m",
)
perm_identifier = permission_uuid or permission_id
if not perm_identifier:
raise ValueError("permission_uuid or permission_id required")
querysafe.assert_safe(perm_identifier, field="permission_id")
# Get the permission to check its scope
perm = db.get_permission(perm_identifier)
perm = db.get_permission(permission_uuid)
# Sanity check: prevent deleting critical permissions if it would lock out admin
if perm.scope == "auth:admin":
_check_admin_lockout_on_delete(str(perm.uuid), request.headers.get("host"))
db.delete_permission(str(perm.uuid), ctx=ctx)
db.delete_permission(permission_uuid, ctx=ctx)
return {"status": "ok"}
tests/test_admin.py
+45 -16
View File
@@ -564,7 +564,7 @@ class TestAdminRoles:
f"/auth/api/admin/orgs/{test_org.uuid}/roles",
json={
"display_name": "Role With Perms",
"permissions": [grantable_permission.scope],
"permissions": [str(grantable_permission.uuid)],
},
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
@@ -592,7 +592,7 @@ class TestAdminRoles:
f"/auth/api/admin/orgs/{test_org.uuid}/roles",
json={
"display_name": "Bad Role",
"permissions": ["test:not:grantable"],
"permissions": [str(perm.uuid)],
},
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
@@ -1360,7 +1360,7 @@ class TestAdminPermissions:
create_permission(perm)
response = await client.patch(
"/auth/api/admin/permission?permission_id=test:updateable&display_name=Updated%20Name",
f"/auth/api/admin/permission?permission_uuid={perm.uuid}&display_name=Updated%20Name",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 200
@@ -1377,7 +1377,7 @@ class TestAdminPermissions:
create_permission(perm)
response = await client.patch(
"/auth/api/admin/permission?permission_id=test:perm&display_name=",
f"/auth/api/admin/permission?permission_uuid={perm.uuid}&display_name=",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 400
@@ -1394,8 +1394,8 @@ class TestAdminPermissions:
create_permission(perm)
response = await client.post(
"/auth/api/admin/permission/rename",
json={"old_scope": "test:renameable2", "new_scope": "test:renamed2"},
f"/auth/api/admin/permission/rename?permission_uuid={perm.uuid}",
json={"new_scope": "test:renamed2"},
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 200
@@ -1410,18 +1410,27 @@ class TestAdminPermissions:
json={},
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 400
assert response.status_code == 422
data = response.json()
assert "required" in data["detail"]
assert any(
"required" in str(error) or "Field required" in str(error)
for error in data["detail"]
)
@pytest.mark.asyncio
async def test_rename_permission_auth_admin_fails(
self, client: httpx.AsyncClient, session_token: str
):
"""Cannot rename the auth:admin permission."""
# Get the auth:admin permission
from paskia.db import list_permissions
perms = list_permissions()
admin_perm = next(p for p in perms if p.scope == "auth:admin")
response = await client.post(
"/auth/api/admin/permission/rename",
json={"old_id": "auth:admin", "new_id": "auth:superadmin"},
f"/auth/api/admin/permission/rename?permission_uuid={admin_perm.uuid}",
json={"new_scope": "auth:superadmin"},
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 400
@@ -1437,9 +1446,8 @@ class TestAdminPermissions:
create_permission(perm)
response = await client.post(
"/auth/api/admin/permission/rename",
f"/auth/api/admin/permission/rename?permission_uuid={perm.uuid}",
json={
"old_scope": "test:rename:withname",
"new_scope": "test:renamed:withname",
"display_name": "New Display Name",
},
@@ -1457,7 +1465,7 @@ class TestAdminPermissions:
create_permission(perm)
response = await client.delete(
"/auth/api/admin/permission?permission_id=test:deleteable",
f"/auth/api/admin/permission?permission_uuid={perm.uuid}",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 200
@@ -1469,8 +1477,14 @@ class TestAdminPermissions:
self, client: httpx.AsyncClient, session_token: str
):
"""Cannot delete the only auth:admin permission (would lock out admin)."""
# Get the auth:admin permission
from paskia.db import list_permissions
perms = list_permissions()
admin_perm = next(p for p in perms if p.scope == "auth:admin")
response = await client.delete(
"/auth/api/admin/permission?permission_id=auth:admin",
f"/auth/api/admin/permission?permission_uuid={admin_perm.uuid}",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 400
@@ -1488,9 +1502,17 @@ class TestAdminPermissions:
perm2 = Permission.create(scope="auth:admin", display_name="Secondary Admin")
create_permission(perm2)
# Get the original auth:admin permission (the one created in setup)
from paskia.db import list_permissions
perms = list_permissions()
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)
# Now we can delete the original one
response = await client.delete(
"/auth/api/admin/permission?permission_id=auth:admin",
f"/auth/api/admin/permission?permission_uuid={original_admin_perm.uuid}",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 200
@@ -1513,8 +1535,15 @@ class TestAdminPermissions:
create_permission(perm2)
# Cannot delete the original one because the remaining one is not accessible
# Get the original auth:admin permission
from paskia.db import list_permissions
perms = list_permissions()
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
response = await client.delete(
"/auth/api/admin/permission?permission_id=auth:admin",
f"/auth/api/admin/permission?permission_uuid={original_admin_perm.uuid}",
headers={**auth_headers(session_token), "Host": "localhost:4401"},
)
assert response.status_code == 400