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

DB cleanup: removed get_permission_organizations and build_org. Using db.data() for read access at call sites.

Browse Source
This commit is contained in:
LeoVasanko
2026-01-27 18:23:17 +00:00
parent 7530d7a710
commit 31f40d874c
5 changed files with 55 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
paskia/bootstrap.py
+6 -4
View File
@@ -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:
paskia/db/__init__.py
+4 -8
View File
@@ -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",
paskia/db/operations.py
+5 -31
View File
@@ -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
paskia/fastapi/admin.py
+13 -6
View File
@@ -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]
paskia/fastapi/reset.py
+27 -17
View File
@@ -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,