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

Finalize database API class merge.

Browse Source
This commit is contained in:
Leo Vasanko
2026-01-27 03:00:18 +00:00
parent 73e4d41f04
commit 637b1b95a8
7 changed files with 78 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files
paskia/db/operations.py
+23 -39
View File
@@ -30,6 +30,7 @@ from paskia.db.jsonl import (
)
from paskia.db.structs import (
Credential,
DatabaseData,
Org,
Permission,
ResetToken,
@@ -37,9 +38,6 @@ from paskia.db.structs import (
Session,
SessionContext,
User,
_DatabaseData,
_OrgData,
_RoleData,
)
from paskia.util.passphrase import is_well_formed as _is_passphrase
@@ -47,7 +45,7 @@ _logger = logging.getLogger(__name__)
# msgspec encoder/decoder
_json_encoder = msgspec.json.Encoder()
_json_decoder = msgspec.json.Decoder(_DatabaseData)
_json_decoder = msgspec.json.Decoder(DatabaseData)
class DB:
@@ -59,7 +57,7 @@ class DB:
def __init__(self, db_path: str = DB_PATH_DEFAULT):
self.db_path = Path(db_path)
self._data = _DatabaseData(
self._data = DatabaseData(
permissions={},
orgs={},
roles={},
@@ -179,27 +177,19 @@ def build_user(uuid: UUID) -> User:
def build_role(uuid: UUID) -> Role:
r = _db._data.roles[uuid]
role = Role(
org=r.org,
display_name=r.display_name,
permissions=[str(pid) for pid in r.permissions.keys()],
)
role.uuid = uuid
return role
r.uuid = uuid
return r
def build_org(uuid: UUID, include_roles: bool = False) -> Org:
o = _db._data.orgs[uuid]
perm_uuids = [
str(pid) for pid, p in _db._data.permissions.items() if uuid in p.orgs
]
org = Org(display_name=o.display_name, permissions=perm_uuids)
org.uuid = uuid
o.uuid = uuid
o.permissions = {pid for pid, p in _db._data.permissions.items() if uuid in p.orgs}
if include_roles:
org.roles = [
o.roles = [
build_role(rid) for rid, r in _db._data.roles.items() if r.org == uuid
]
return org
return o
def build_credential(uuid: UUID) -> Credential:
@@ -460,15 +450,14 @@ 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 = set(org.permissions) # Set of permission UUID strings
org_perm_uuids = org.permissions # set[UUID] computed by build_org
normalized_host = normalize_host(host)
host_without_port = normalized_host.rsplit(":", 1)[0] if normalized_host else None
effective_perms = []
for perm_uuid_str in role.permissions:
if perm_uuid_str not in org_perm_uuids:
for perm_uuid in role.permission_set:
if perm_uuid not in org_perm_uuids:
continue
perm_uuid = UUID(perm_uuid_str)
if perm_uuid not in _db._data.permissions:
continue
p = _db._data.permissions[perm_uuid]
@@ -560,16 +549,9 @@ def create_organization(org: Org, *, ctx: SessionContext | None = None) -> None:
if org.uuid in _db._data.orgs:
raise ValueError(f"Organization {org.uuid} already exists")
with _db.transaction("Created organization", ctx):
_db._data.orgs[org.uuid] = _OrgData(
_db._data.orgs[org.uuid] = Org(
display_name=org.display_name, created_at=datetime.now(timezone.utc)
)
# Grant listed permissions to this org (org.permissions contains UUIDs now)
for perm_uuid_str in org.permissions:
perm_uuid = (
UUID(perm_uuid_str) if isinstance(perm_uuid_str, str) else perm_uuid_str
)
if perm_uuid in _db._data.permissions:
_db._data.permissions[perm_uuid].orgs[org.uuid] = True
# Create Administration role with org admin permission
import uuid7
@@ -581,11 +563,13 @@ def create_organization(org: Org, *, ctx: SessionContext | None = None) -> None:
org_admin_perm_uuid = pid
break
role_permissions = {org_admin_perm_uuid: True} if org_admin_perm_uuid else {}
_db._data.roles[admin_role_uuid] = _RoleData(
admin_role = Role(
org=org.uuid,
display_name="Administration",
permissions=role_permissions,
)
admin_role.uuid = admin_role_uuid
_db._data.roles[admin_role_uuid] = admin_role
def update_organization_name(
@@ -699,11 +683,7 @@ def create_role(role: Role, *, ctx: SessionContext | None = None) -> None:
if role.org not in _db._data.orgs:
raise ValueError(f"Organization {role.org} not found")
with _db.transaction("Created role", ctx):
_db._data.roles[role.uuid] = _RoleData(
org=role.org,
display_name=role.display_name,
permissions={UUID(pid): True for pid in role.permissions},
)
_db._data.roles[role.uuid] = role
def update_role_name(
@@ -1263,17 +1243,21 @@ def bootstrap(
_db._data.permissions[perm_org_admin_uuid] = perm_org_admin
# Create organization
_db._data.orgs[org_uuid] = _OrgData(
new_org = Org(
display_name=org_name,
created_at=now,
)
new_org.uuid = org_uuid
_db._data.orgs[org_uuid] = new_org
# Create Administration role with both permissions
_db._data.roles[role_uuid] = _RoleData(
admin_role = Role(
org=org_uuid,
display_name="Administration",
permissions={perm_admin_uuid: True, perm_org_admin_uuid: True},
)
admin_role.uuid = role_uuid
_db._data.roles[role_uuid] = admin_role
# Create admin user
admin_user = User(
paskia/db/structs.py
+24 -28
View File
@@ -14,6 +14,11 @@ class Permission(msgspec.Struct, dict=True, omit_defaults=True):
def __post_init__(self):
self.uuid: UUID | None = None # Convenience field, not serialized
@property
def org_set(self) -> set[UUID]:
"""Get orgs that can grant this permission as a set."""
return set(self.orgs.keys())
@classmethod
def create(
cls,
@@ -31,49 +36,51 @@ class Permission(msgspec.Struct, dict=True, omit_defaults=True):
return perm
class Role(msgspec.Struct, dict=True):
class Role(msgspec.Struct, dict=True, omit_defaults=True):
org: UUID
display_name: str
permissions: list[str] = [] # permission UUIDs this role grants
permissions: dict[UUID, bool] = {} # permission_uuid -> True
def __post_init__(self):
self.uuid: UUID | None = None # Convenience field, not serialized
@property
def permission_set(self) -> set[UUID]:
"""Get permissions as a set of UUIDs."""
return set(self.permissions.keys())
@classmethod
def create(
cls,
org: UUID,
display_name: str,
permissions: list[str] | None = None,
permissions: set[UUID] | None = None,
) -> "Role":
"""Create a new Role with auto-generated uuid7."""
role = cls(
org=org,
display_name=display_name,
permissions=permissions or [],
permissions={p: True for p in (permissions or set())},
)
role.uuid = uuid7.create()
return role
class Org(msgspec.Struct, dict=True):
class Org(msgspec.Struct, dict=True, omit_defaults=True):
display_name: str
permissions: list[str] = [] # permission UUIDs this org can grant
roles: list[Role] = [] # roles belonging to this org
created_at: datetime | None = None
def __post_init__(self):
self.uuid: UUID | None = None # Convenience field, not serialized
@classmethod
def create(
cls,
display_name: str,
permissions: list[str] | None = None,
) -> "Org":
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,
permissions=permissions or [],
created_at=datetime.now(timezone.utc),
)
org.uuid = uuid7.create()
return org
@@ -188,25 +195,14 @@ class SessionContext(msgspec.Struct):
# -------------------------------------------------------------------------
# Internal storage types (different structure for efficient storage)
# Database storage structure
# -------------------------------------------------------------------------
class _OrgData(msgspec.Struct):
display_name: str
created_at: datetime | None = None
class _RoleData(msgspec.Struct):
org: UUID
display_name: str
permissions: dict[UUID, bool] = {} # permission_uuid -> True
class _DatabaseData(msgspec.Struct, omit_defaults=True):
class DatabaseData(msgspec.Struct, omit_defaults=True):
permissions: dict[UUID, Permission]
orgs: dict[UUID, _OrgData]
roles: dict[UUID, _RoleData]
orgs: dict[UUID, Org]
roles: dict[UUID, Role]
users: dict[UUID, User]
credentials: dict[UUID, Credential]
sessions: dict[str, Session]
paskia/fastapi/admin.py
+12 -12
View File
@@ -137,8 +137,11 @@ async def admin_create_org(
display_name = payload.get("display_name") or "New Organization"
permissions = payload.get("permissions") or []
org = OrgDC.create(display_name=display_name, permissions=permissions)
org = OrgDC.create(display_name=display_name)
db.create_organization(org, ctx=ctx)
# Grant requested permissions to the new org
for perm in permissions:
db.add_permission_to_organization(str(org.uuid), perm)
return {"uuid": str(org.uuid)}
@@ -266,18 +269,17 @@ 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))
grantable = set(org.permissions or [])
grantable = org.permissions # set[UUID] computed by build_org
# Normalize permission IDs to UUIDs
permission_uuids = []
permission_uuids: set[UUID] = set()
for pid in perms:
perm = db.get_permission(pid)
if not perm:
raise ValueError(f"Permission {pid} not found")
perm_uuid_str = str(perm.uuid)
if perm_uuid_str not in grantable:
if perm.uuid not in grantable:
raise ValueError(f"Permission not grantable by org: {pid}")
permission_uuids.append(perm_uuid_str)
permission_uuids.add(perm.uuid)
role = RoleDC.create(
org=org_uuid,
@@ -348,7 +350,7 @@ async def admin_add_role_permission(
if not perm:
raise HTTPException(status_code=404, detail="Permission not found")
org = db.get_organization(str(org_uuid))
if str(permission_uuid) not in org.permissions:
if permission_uuid not in org.permissions:
raise ValueError("Permission not grantable by organization")
db.add_permission_to_role(role_uuid, permission_uuid, ctx=ctx)
@@ -380,13 +382,11 @@ async def admin_remove_role_permission(
raise HTTPException(status_code=404, detail="Role not found in organization")
# Sanity check: prevent admin from removing their own access
# Find auth:admin and auth:org:admin permission UUIDs
perm_uuid_str = str(permission_uuid)
perm = db.get_permission(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 = set(role.permissions) - {perm_uuid_str}
remaining_perms = role.permission_set - {permission_uuid}
has_admin = False
for rp_uuid in remaining_perms:
rp = db.get_permission(rp_uuid)
@@ -918,8 +918,8 @@ 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 = set(ctx.org.permissions or [])
filtered_perms = [p for p in perms if str(p.uuid) in grantable]
grantable = ctx.org.permissions # set[UUID]
filtered_perms = [p for p in perms if p.uuid in grantable]
return [_perm_to_dict(p) for p in filtered_perms]
paskia/fastapi/api.py
+4 -3
View File
@@ -157,9 +157,10 @@ async def forward_authentication(
ctx = await authz.verify(
auth, perm, host=request.headers.get("host"), max_age=max_age
)
role_permissions = set(ctx.role.permissions or [])
if ctx.permissions:
role_permissions.update(permission.scope for permission in ctx.permissions)
# Build permission scopes for Remote-Groups header
role_permissions = (
{p.scope for p in ctx.permissions} if ctx.permissions else set()
)
remote_headers: dict[str, str] = {
"Remote-User": str(ctx.user.uuid),
paskia/migrate/__init__.py
+8 -6
View File
@@ -56,12 +56,12 @@ async def migrate_from_sql(
from paskia.db.operations import DB as JSONDB
from paskia.db.structs import (
Credential,
Org,
Permission,
ResetToken,
Role,
Session,
User,
_OrgData,
_RoleData,
)
# Initialize source SQL database
@@ -131,9 +131,9 @@ async def migrate_from_sql(
orgs = await sql_db.list_organizations()
for org in orgs:
org_key: UUID = org.uuid
json_db._data.orgs[org_key] = _OrgData(
display_name=org.display_name,
)
new_org = Org(display_name=org.display_name)
new_org.uuid = org_key
json_db._data.orgs[org_key] = new_org
# Update permissions to allow this org to grant them (by UUID)
for old_perm_id in org.permissions:
perm_uuid = perm_id_to_uuid.get(old_perm_id)
@@ -154,11 +154,13 @@ async def migrate_from_sql(
perm_uuid = perm_id_to_uuid.get(old_perm_id)
if perm_uuid:
new_permissions[perm_uuid] = True
json_db._data.roles[role_key] = _RoleData(
new_role = Role(
org=role.org_uuid,
display_name=role.display_name,
permissions=new_permissions,
)
new_role.uuid = role_key
json_db._data.roles[role_key] = new_role
role_count += 1
print(f" Migrated {role_count} roles")
tests/conftest.py
+4 -7
View File
@@ -13,13 +13,11 @@ import asyncio
import os
import tempfile
from collections.abc import AsyncGenerator
from datetime import datetime, timezone
from uuid import UUID
import httpx
import pytest
import pytest_asyncio
import uuid7
from paskia import globals as paskia_globals
from paskia.authsession import expires
@@ -84,11 +82,10 @@ async def passkey_instance() -> Passkey:
@pytest_asyncio.fixture(scope="function")
async def test_org(test_db: DB, admin_permission: Permission) -> Org:
"""Create a test organization with admin permission."""
org = Org.create(
display_name="Test Organization",
permissions=[str(admin_permission.uuid)], # Org can grant this permission
)
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))
return org
@@ -121,7 +118,7 @@ async def test_role(
role = Role.create(
org=test_org.uuid,
display_name="Test Admin Role",
permissions=[str(admin_permission.uuid), str(org_admin_permission.uuid)],
permissions={admin_permission.uuid, org_admin_permission.uuid},
)
create_role(role)
return role
tests/test_admin.py
+3 -3
View File
@@ -58,7 +58,7 @@ async def second_org_role(
role = Role.create(
org=second_org.uuid,
display_name="Second Org Admin Role",
permissions=[str(admin_permission.uuid)],
permissions={admin_permission.uuid},
)
create_role(role)
return role
@@ -117,7 +117,7 @@ async def org_admin_role(
role = Role.create(
org=test_org.uuid,
display_name="Org Admin Role",
permissions=[str(org_admin_permission.uuid)],
permissions={org_admin_permission.uuid},
)
create_role(role)
return role
@@ -397,7 +397,7 @@ class TestAdminOrganizations:
client: httpx.AsyncClient,
session_token: str,
test_db: DB,
):
):
"""Admin should be able to delete another organization."""
# Create org to delete
org_to_delete = Org.create(