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

Implement full ORM. Various other cleanup.

Browse Source
This commit is contained in:
LeoVasanko
2026-01-28 17:19:56 +00:00
parent c8d659b5ca
commit 7329223784
19 changed files with 377 additions and 327 deletions
Download Patch File Download Diff File Expand all files Collapse all files
paskia/authsession.py
+3 -3
View File
@@ -8,7 +8,7 @@ independent of any web framework:
- Credential management - Credential management
""" """
from datetime import datetime, timezone from datetime import UTC, datetime
from typing import TYPE_CHECKING from typing import TYPE_CHECKING
from uuid import UUID from uuid import UUID
@@ -23,11 +23,11 @@ EXPIRES = SESSION_LIFETIME
def expires() -> datetime: def expires() -> datetime:
return datetime.now(timezone.utc) + EXPIRES return datetime.now(UTC) + EXPIRES
def reset_expires() -> datetime: def reset_expires() -> datetime:
return datetime.now(timezone.utc) + RESET_LIFETIME return datetime.now(UTC) + RESET_LIFETIME
def get_reset(token: str) -> "ResetToken": def get_reset(token: str) -> "ResetToken":
paskia/db/background.py
+3 -3
View File
@@ -6,7 +6,7 @@ Periodically flushes pending changes to disk and cleans up expired items.
import asyncio import asyncio
import logging import logging
from datetime import datetime, timezone from datetime import UTC, datetime
from paskia.db.operations import _store, cleanup_expired from paskia.db.operations import _store, cleanup_expired
@@ -33,7 +33,7 @@ async def _background_loop():
cleanup_expired() cleanup_expired()
await flush() await flush()
last_cleanup = datetime.now(timezone.utc) last_cleanup = datetime.now(UTC)
while True: while True:
try: try:
@@ -42,7 +42,7 @@ async def _background_loop():
await flush() await flush()
# Run cleanup periodically # Run cleanup periodically
now = datetime.now(timezone.utc) now = datetime.now(UTC)
if (now - last_cleanup).total_seconds() >= CLEANUP_INTERVAL: if (now - last_cleanup).total_seconds() >= CLEANUP_INTERVAL:
cleanup_expired() cleanup_expired()
await flush() # Flush cleanup changes await flush() # Flush cleanup changes
paskia/db/jsonl.py
+2 -2
View File
@@ -10,7 +10,7 @@ import logging
import sys import sys
from collections import deque from collections import deque
from contextlib import contextmanager from contextlib import contextmanager
from datetime import datetime, timezone from datetime import UTC, datetime
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
from uuid import UUID from uuid import UUID
@@ -97,7 +97,7 @@ def create_change_record(
) -> _ChangeRecord: ) -> _ChangeRecord:
"""Create a change record for persistence.""" """Create a change record for persistence."""
return _ChangeRecord( return _ChangeRecord(
ts=datetime.now(timezone.utc), ts=datetime.now(UTC),
a=action, a=action,
u=user, u=user,
diff=diff, diff=diff,
paskia/db/operations.py
+98 -104
View File
@@ -10,7 +10,7 @@ import hashlib
import logging import logging
import os import os
import secrets import secrets
from datetime import datetime, timezone from datetime import UTC, datetime
from uuid import UUID from uuid import UUID
import uuid7 import uuid7
@@ -68,21 +68,18 @@ def get_user_organization(user_uuid: UUID) -> tuple[Org, str]:
Raises ValueError if user not found. Raises ValueError if user not found.
Call sites: Call sites:
- Get user's organization when updating user role (admin.py:493) - update_user_role_in_organization: org only
- Get user's organization for user credential listing (admin.py:530) - admin_create_user_registration_link: org only
- Get user's organization for user details API (admin.py:579) - admin_get_user_detail: org and role
- Get user's organization for updating user display name (admin.py:721) - admin_update_user_display_name: org only
- Get user's organization for deleting user credential (admin.py:754) - admin_delete_user_credential: org only
- Get user's organization for deleting user session (admin.py:783) - admin_delete_user_session: org only
""" """
if user_uuid not in _db.users: if user_uuid not in _db.users:
raise ValueError(f"User {user_uuid} not found") raise ValueError(f"User {user_uuid} not found")
role_uuid = _db.users[user_uuid].role user = _db.users[user_uuid]
if role_uuid not in _db.roles: role = user.role
raise ValueError(f"Role {role_uuid} not found") return role.org, role.display_name
role_data = _db.roles[role_uuid]
org_uuid = role_data.org
return _db.orgs[org_uuid], role_data.display_name
def get_organization_users(org_uuid: UUID) -> list[tuple[User, str]]: def get_organization_users(org_uuid: UUID) -> list[tuple[User, str]]:
@@ -90,10 +87,8 @@ def get_organization_users(org_uuid: UUID) -> list[tuple[User, str]]:
Returns list of (User, role_display_name) tuples. Returns list of (User, role_display_name) tuples.
""" """
role_map = { org = _db.orgs[org_uuid]
rid: r.display_name for rid, r in _db.roles.items() if r.org == org_uuid return [(u, u.role.display_name) for role in org.roles for u in role.users]
}
return [(u, role_map[u.role]) for u in _db.users.values() if u.role in role_map]
def get_user_credential_ids(user_uuid: UUID) -> list[bytes]: def get_user_credential_ids(user_uuid: UUID) -> list[bytes]:
@@ -101,7 +96,8 @@ def get_user_credential_ids(user_uuid: UUID) -> list[bytes]:
Returns empty list if user has no credentials. Returns empty list if user has no credentials.
""" """
return [c.credential_id for c in _db.credentials.values() if c.user == user_uuid] assert user_uuid
return [c.credential_id for c in _db.users[user_uuid].credentials]
def _reset_key(passphrase: str) -> bytes: def _reset_key(passphrase: str) -> bytes:
@@ -177,9 +173,9 @@ def create_org(org: Org, *, ctx: SessionContext | None = None) -> None:
if org.uuid in _db.orgs: if org.uuid in _db.orgs:
raise ValueError(f"Organization {org.uuid} already exists") raise ValueError(f"Organization {org.uuid} already exists")
with _db.transaction("admin:create_org", ctx): with _db.transaction("admin:create_org", ctx):
new_org = Org(display_name=org.display_name) new_org = Org.create(display_name=org.display_name)
_db.orgs[org.uuid] = new_org
new_org.uuid = org.uuid new_org.uuid = org.uuid
_db.orgs[org.uuid] = new_org
# Create Administration role with org admin permission # Create Administration role with org admin permission
admin_role_uuid = uuid7.create() admin_role_uuid = uuid7.create()
@@ -191,7 +187,7 @@ def create_org(org: Org, *, ctx: SessionContext | None = None) -> None:
break break
role_permissions = {org_admin_perm_uuid: True} if org_admin_perm_uuid else {} role_permissions = {org_admin_perm_uuid: True} if org_admin_perm_uuid else {}
admin_role = Role( admin_role = Role(
org=org.uuid, org_uuid=org.uuid,
display_name="Administration", display_name="Administration",
permissions=role_permissions, permissions=role_permissions,
) )
@@ -217,17 +213,15 @@ def delete_org(uuid: UUID, *, ctx: SessionContext | None = None) -> None:
if uuid not in _db.orgs: if uuid not in _db.orgs:
raise ValueError(f"Organization {uuid} not found") raise ValueError(f"Organization {uuid} not found")
with _db.transaction("admin:delete_org", ctx): with _db.transaction("admin:delete_org", ctx):
org = _db.orgs[uuid]
# Remove org from all permissions # Remove org from all permissions
for p in _db.permissions.values(): for p in _db.permissions.values():
p.orgs.pop(uuid, None) p.orgs.pop(uuid, None)
# Delete roles in this org # Delete roles in this org and their users
role_uuids = [rid for rid, r in _db.roles.items() if r.org == uuid] for role in org.roles:
for rid in role_uuids: for user in role.users:
del _db.roles[rid] del _db.users[user.uuid]
# Delete users with those roles del _db.roles[role.uuid]
user_uuids = [uid for uid, u in _db.users.items() if u.role in role_uuids]
for uid in user_uuids:
del _db.users[uid]
del _db.orgs[uuid] del _db.orgs[uuid]
@@ -269,8 +263,8 @@ def create_role(role: Role, *, ctx: SessionContext | None = None) -> None:
"""Create a new role.""" """Create a new role."""
if role.uuid in _db.roles: if role.uuid in _db.roles:
raise ValueError(f"Role {role.uuid} already exists") raise ValueError(f"Role {role.uuid} already exists")
if role.org not in _db.orgs: if role.org_uuid not in _db.orgs:
raise ValueError(f"Organization {role.org} not found") raise ValueError(f"Organization {role.org_uuid} not found")
with _db.transaction("admin:create_role", ctx): with _db.transaction("admin:create_role", ctx):
_db.roles[role.uuid] = role _db.roles[role.uuid] = role
@@ -321,7 +315,8 @@ def delete_role(uuid: UUID, *, ctx: SessionContext | None = None) -> None:
if uuid not in _db.roles: if uuid not in _db.roles:
raise ValueError(f"Role {uuid} not found") raise ValueError(f"Role {uuid} not found")
# Check no users have this role # Check no users have this role
if any(u.role == uuid for u in _db.users.values()): role = _db.roles[uuid]
if role.users:
raise ValueError(f"Cannot delete role {uuid}: users still assigned") raise ValueError(f"Cannot delete role {uuid}: users still assigned")
with _db.transaction("admin:delete_role", ctx): with _db.transaction("admin:delete_role", ctx):
del _db.roles[uuid] del _db.roles[uuid]
@@ -331,8 +326,8 @@ def create_user(new_user: User, *, ctx: SessionContext | None = None) -> None:
"""Create a new user.""" """Create a new user."""
if new_user.uuid in _db.users: if new_user.uuid in _db.users:
raise ValueError(f"User {new_user.uuid} already exists") raise ValueError(f"User {new_user.uuid} already exists")
if new_user.role not in _db.roles: if new_user.role_uuid not in _db.roles:
raise ValueError(f"Role {new_user.role} not found") raise ValueError(f"Role {new_user.role_uuid} not found")
with _db.transaction("admin:create_user", ctx): with _db.transaction("admin:create_user", ctx):
_db.users[new_user.uuid] = new_user _db.users[new_user.uuid] = new_user
@@ -369,7 +364,7 @@ def update_user_role(
if role_uuid not in _db.roles: if role_uuid not in _db.roles:
raise ValueError(f"Role {role_uuid} not found") raise ValueError(f"Role {role_uuid} not found")
with _db.transaction("admin:update_user_role", ctx): with _db.transaction("admin:update_user_role", ctx):
_db.users[uuid].role = role_uuid _db.users[uuid].role_uuid = role_uuid
def update_user_role_in_organization( def update_user_role_in_organization(
@@ -381,39 +376,35 @@ def update_user_role_in_organization(
"""Update user's role by role name within their current organization.""" """Update user's role by role name within their current organization."""
if user_uuid not in _db.users: if user_uuid not in _db.users:
raise ValueError(f"User {user_uuid} not found") raise ValueError(f"User {user_uuid} not found")
current_role_uuid = _db.users[user_uuid].role user = _db.users[user_uuid]
if current_role_uuid not in _db.roles: org = user.org
raise ValueError("Current role not found")
org_uuid = _db.roles[current_role_uuid].org
# Find role by name in the same org # Find role by name in the same org
new_role_uuid = None new_role_uuid = None
for rid, r in _db.roles.items(): for r in org.roles:
if r.org == org_uuid and r.display_name == role_name: if r.display_name == role_name:
new_role_uuid = rid new_role_uuid = r.uuid
break break
if new_role_uuid is None: if new_role_uuid is None:
raise ValueError(f"Role '{role_name}' not found in organization") raise ValueError(f"Role '{role_name}' not found in organization")
with _db.transaction("admin:update_user_role", ctx): with _db.transaction("admin:update_user_role", ctx):
_db.users[user_uuid].role = new_role_uuid _db.users[user_uuid].role_uuid = new_role_uuid
def delete_user(uuid: UUID, *, ctx: SessionContext | None = None) -> None: def delete_user(uuid: UUID, *, ctx: SessionContext | None = None) -> None:
"""Delete user and their credentials/sessions.""" """Delete user and their credentials/sessions."""
if uuid not in _db.users: if uuid not in _db.users:
raise ValueError(f"User {uuid} not found") raise ValueError(f"User {uuid} not found")
user = _db.users[uuid]
with _db.transaction("admin:delete_user", ctx): with _db.transaction("admin:delete_user", ctx):
# Delete credentials # Delete credentials
cred_uuids = [cid for cid, c in _db.credentials.items() if c.user == uuid] for cred in user.credentials:
for cid in cred_uuids: del _db.credentials[cred.uuid]
del _db.credentials[cid]
# Delete sessions # Delete sessions
sess_keys = [k for k, s in _db.sessions.items() if s.user == uuid] for sess in user.sessions:
for k in sess_keys: del _db.sessions[sess.key]
del _db.sessions[k]
# Delete reset tokens # Delete reset tokens
token_keys = [k for k, t in _db.reset_tokens.items() if t.user == uuid] for token in user.reset_tokens:
for k in token_keys: del _db.reset_tokens[token.key]
del _db.reset_tokens[k]
del _db.users[uuid] del _db.users[uuid]
@@ -421,8 +412,8 @@ def create_credential(cred: Credential, *, ctx: SessionContext | None = None) ->
"""Create a new credential.""" """Create a new credential."""
if cred.uuid in _db.credentials: if cred.uuid in _db.credentials:
raise ValueError(f"Credential {cred.uuid} already exists") raise ValueError(f"Credential {cred.uuid} already exists")
if cred.user not in _db.users: if cred.user_uuid not in _db.users:
raise ValueError(f"User {cred.user} not found") raise ValueError(f"User {cred.user_uuid} not found")
with _db.transaction("create_credential", ctx): with _db.transaction("create_credential", ctx):
_db.credentials[cred.uuid] = cred _db.credentials[cred.uuid] = cred
@@ -455,20 +446,19 @@ def delete_credential(
""" """
if uuid not in _db.credentials: if uuid not in _db.credentials:
raise ValueError(f"Credential {uuid} not found") raise ValueError(f"Credential {uuid} not found")
cred = _db.credentials[uuid]
if user_uuid is not None: if user_uuid is not None:
cred_user = _db.credentials[uuid].user if cred.user_uuid != user_uuid:
if cred_user != user_uuid:
raise ValueError(f"Credential {uuid} does not belong to user {user_uuid}") raise ValueError(f"Credential {uuid} does not belong to user {user_uuid}")
with _db.transaction("delete_credential", ctx): with _db.transaction("delete_credential", ctx):
# Delete all sessions using this credential # Delete all sessions using this credential
keys = [k for k, s in _db.sessions.items() if s.credential == uuid] for sess in cred.sessions:
for k in keys: print(sess, repr(sess.key))
del _db.sessions[k] del _db.sessions[sess.key]
del _db.credentials[uuid] del _db.credentials[uuid]
def create_session( def create_session(
key: str,
user_uuid: UUID, user_uuid: UUID,
credential_uuid: UUID, credential_uuid: UUID,
host: str, host: str,
@@ -477,23 +467,25 @@ def create_session(
expiry: datetime, expiry: datetime,
*, *,
ctx: SessionContext | None = None, ctx: SessionContext | None = None,
) -> None: ) -> str:
"""Create a new session.""" """Create a new session. Returns the session key."""
if key in _db.sessions:
raise ValueError("Session already exists")
if user_uuid not in _db.users: if user_uuid not in _db.users:
raise ValueError(f"User {user_uuid} not found") raise ValueError(f"User {user_uuid} not found")
if credential_uuid not in _db.credentials: if credential_uuid not in _db.credentials:
raise ValueError(f"Credential {credential_uuid} not found") raise ValueError(f"Credential {credential_uuid} not found")
session = Session.create(
user=user_uuid,
credential=credential_uuid,
host=host,
ip=ip,
user_agent=user_agent,
expiry=expiry,
)
if session.key in _db.sessions:
raise ValueError("Session already exists")
with _db.transaction("create_session", ctx): with _db.transaction("create_session", ctx):
_db.sessions[key] = Session( _db.sessions[session.key] = session
user=user_uuid, return session.key
credential=credential_uuid,
host=host,
ip=ip,
user_agent=user_agent,
expiry=expiry,
)
def update_session( def update_session(
@@ -547,10 +539,12 @@ def delete_sessions_for_user(
For user logout-all, pass ctx of the user's session. For user logout-all, pass ctx of the user's session.
For admin bulk termination, pass admin's ctx. For admin bulk termination, pass admin's ctx.
""" """
user = _db.users.get(user_uuid)
if not user:
return
with _db.transaction("admin:delete_sessions_for_user", ctx): with _db.transaction("admin:delete_sessions_for_user", ctx):
keys = [k for k, s in _db.sessions.items() if s.user == user_uuid] for sess in user.sessions:
for k in keys: del _db.sessions[sess.key]
del _db.sessions[k]
def create_reset_token( def create_reset_token(
@@ -575,7 +569,7 @@ def create_reset_token(
raise ValueError(f"User {user_uuid} not found") raise ValueError(f"User {user_uuid} not found")
with _db.transaction("create_reset_token", ctx): with _db.transaction("create_reset_token", ctx):
_db.reset_tokens[key] = ResetToken( _db.reset_tokens[key] = ResetToken(
user=user_uuid, expiry=expiry, token_type=token_type user_uuid=user_uuid, expiry=expiry, token_type=token_type
) )
@@ -594,7 +588,7 @@ def delete_reset_token(key: bytes, *, ctx: SessionContext | None = None) -> None
def cleanup_expired() -> int: def cleanup_expired() -> int:
"""Remove expired sessions and reset tokens. Returns count removed.""" """Remove expired sessions and reset tokens. Returns count removed."""
now = datetime.now(timezone.utc) now = datetime.now(UTC)
count = 0 count = 0
with _db.transaction("expiry"): with _db.transaction("expiry"):
expired_sessions = [k for k, s in _db.sessions.items() if s.expiry < now] expired_sessions = [k for k, s in _db.sessions.items() if s.expiry < now]
@@ -639,13 +633,20 @@ def login(
""" """
if isinstance(user_uuid, str): if isinstance(user_uuid, str):
user_uuid = UUID(user_uuid) user_uuid = UUID(user_uuid)
now = datetime.now(timezone.utc) now = datetime.now(UTC)
if user_uuid not in _db.users: if user_uuid not in _db.users:
raise ValueError(f"User {user_uuid} not found") raise ValueError(f"User {user_uuid} not found")
if credential_uuid not in _db.credentials: if credential_uuid not in _db.credentials:
raise ValueError(f"Credential {credential_uuid} not found") raise ValueError(f"Credential {credential_uuid} not found")
session_key = _create_token() session = Session.create(
user=user_uuid,
credential=credential_uuid,
host=host,
ip=ip,
user_agent=user_agent,
expiry=expiry,
)
user_str = str(user_uuid) user_str = str(user_uuid)
with _db.transaction("login", user=user_str): with _db.transaction("login", user=user_str):
# Update user # Update user
@@ -655,15 +656,8 @@ def login(
_db.credentials[credential_uuid].sign_count = sign_count _db.credentials[credential_uuid].sign_count = sign_count
_db.credentials[credential_uuid].last_used = now _db.credentials[credential_uuid].last_used = now
# Create session # Create session
_db.sessions[session_key] = Session( _db.sessions[session.key] = session
user=user_uuid, return session.key
credential=credential_uuid,
host=host,
ip=ip,
user_agent=user_agent,
expiry=expiry,
)
return session_key
def create_credential_session( def create_credential_session(
@@ -686,13 +680,20 @@ def create_credential_session(
Returns the generated session token. Returns the generated session token.
""" """
now = datetime.now(timezone.utc) now = datetime.now(UTC)
expiry = now + SESSION_LIFETIME expiry = now + SESSION_LIFETIME
session_key = _create_token()
if user_uuid not in _db.users: if user_uuid not in _db.users:
raise ValueError(f"User {user_uuid} not found") raise ValueError(f"User {user_uuid} not found")
session = Session.create(
user=user_uuid,
credential=credential.uuid,
host=host,
ip=ip,
user_agent=user_agent,
expiry=expiry,
)
user_str = str(user_uuid) user_str = str(user_uuid)
with _db.transaction("create_credential_session", user=user_str): with _db.transaction("create_credential_session", user=user_str):
# Update display name if provided # Update display name if provided
@@ -703,20 +704,13 @@ def create_credential_session(
_db.credentials[credential.uuid] = credential _db.credentials[credential.uuid] = credential
# Create session # Create session
_db.sessions[session_key] = Session( _db.sessions[session.key] = session
user=user_uuid,
credential=credential.uuid,
host=host,
ip=ip,
user_agent=user_agent,
expiry=expiry,
)
# Delete reset token if provided # Delete reset token if provided
if reset_key: if reset_key:
if reset_key in _db.reset_tokens: if reset_key in _db.reset_tokens:
del _db.reset_tokens[reset_key] del _db.reset_tokens[reset_key]
return session_key return session.key
# ------------------------------------------------------------------------- # -------------------------------------------------------------------------
@@ -775,7 +769,7 @@ def bootstrap(
reset_expiry = reset_expires() reset_expiry = reset_expires()
reset_key = _reset_key(reset_passphrase) reset_key = _reset_key(reset_passphrase)
now = datetime.now(timezone.utc) now = datetime.now(UTC)
with _db.transaction("bootstrap"): with _db.transaction("bootstrap"):
# Create auth:admin permission # Create auth:admin permission
@@ -797,13 +791,13 @@ def bootstrap(
_db.permissions[perm_org_admin_uuid] = perm_org_admin _db.permissions[perm_org_admin_uuid] = perm_org_admin
# Create organization # Create organization
new_org = Org(display_name=org_name) new_org = Org.create(display_name=org_name)
new_org.uuid = org_uuid new_org.uuid = org_uuid
_db.orgs[org_uuid] = new_org _db.orgs[org_uuid] = new_org
# Create Administration role with both permissions # Create Administration role with both permissions
admin_role = Role( admin_role = Role(
org=org_uuid, org_uuid=org_uuid,
display_name="Administration", display_name="Administration",
permissions={perm_admin_uuid: True, perm_org_admin_uuid: True}, permissions={perm_admin_uuid: True, perm_org_admin_uuid: True},
) )
@@ -813,7 +807,7 @@ def bootstrap(
# Create admin user # Create admin user
admin_user = User( admin_user = User(
display_name=admin_name, display_name=admin_name,
role=role_uuid, role_uuid=role_uuid,
created_at=now, created_at=now,
last_seen=None, last_seen=None,
visits=0, visits=0,
@@ -823,7 +817,7 @@ def bootstrap(
# Create reset token # Create reset token
_db.reset_tokens[reset_key] = ResetToken( _db.reset_tokens[reset_key] = ResetToken(
user=user_uuid, user_uuid=user_uuid,
expiry=reset_expiry, expiry=reset_expiry,
token_type="admin bootstrap", token_type="admin bootstrap",
) )
paskia/db/structs.py
+166 -46
View File
@@ -1,9 +1,13 @@
from datetime import datetime, timezone from __future__ import annotations
import secrets
from datetime import UTC, datetime
from uuid import UUID from uuid import UUID
import msgspec import msgspec
import uuid7 import uuid7
from paskia import db
from paskia.util.hostutil import normalize_host from paskia.util.hostutil import normalize_host
# Sentinel for uuid fields before they are set by create() or DB post init # Sentinel for uuid fields before they are set by create() or DB post init
@@ -31,13 +35,22 @@ class Permission(msgspec.Struct, dict=True, omit_defaults=True):
"""Get orgs that can grant this permission as a set.""" """Get orgs that can grant this permission as a set."""
return set(self.orgs.keys()) return set(self.orgs.keys())
@property
def orgs_list(self) -> list[Org]:
"""Get list of Org objects that can grant this permission."""
return [
db.data().orgs[org_uuid]
for org_uuid in self.orgs.keys()
if org_uuid in db.data().orgs
]
@classmethod @classmethod
def create( def create(
cls, cls,
scope: str, scope: str,
display_name: str, display_name: str,
domain: str | None = None, domain: str | None = None,
) -> "Permission": ) -> Permission:
"""Create a new Permission with auto-generated uuid7.""" """Create a new Permission with auto-generated uuid7."""
perm = cls( perm = cls(
scope=scope, scope=scope,
@@ -48,15 +61,41 @@ class Permission(msgspec.Struct, dict=True, omit_defaults=True):
return perm return perm
class Org(msgspec.Struct, dict=True):
"""Organization data structure."""
display_name: str
def __post_init__(self):
self.uuid: UUID = _UUID_UNSET # Convenience field, not serialized
@property
def roles(self) -> list[Role]:
"""Get all roles that belong to this organization."""
return [r for r in db.data().roles.values() if r.org_uuid == self.uuid]
@property
def permissions(self) -> list[Permission]:
"""Get all permissions that this organization can grant."""
return [p for p in db.data().permissions.values() if self.uuid in p.orgs]
@classmethod
def create(cls, display_name: str) -> Org:
"""Create a new Org with auto-generated uuid7."""
org = cls(display_name=display_name)
org.uuid = uuid7.create()
return org
class Role(msgspec.Struct, dict=True, omit_defaults=True): class Role(msgspec.Struct, dict=True, omit_defaults=True):
"""Role data structure. """Role data structure.
Mutable fields: display_name, permissions Mutable fields: display_name, permissions
Immutable fields: org (set at creation, never modified) Immutable fields: org_uuid (set at creation, never modified)
uuid is generated at creation. uuid is generated at creation.
""" """
org: UUID org_uuid: UUID = msgspec.field(name="org")
display_name: str display_name: str
permissions: dict[UUID, bool] = {} # permission_uuid -> True permissions: dict[UUID, bool] = {} # permission_uuid -> True
@@ -68,16 +107,36 @@ class Role(msgspec.Struct, dict=True, omit_defaults=True):
"""Get permissions as a set of UUIDs.""" """Get permissions as a set of UUIDs."""
return set(self.permissions.keys()) return set(self.permissions.keys())
@property
def permissions_list(self) -> list[Permission]:
"""Get list of Permission objects for this role."""
return [
db.data().permissions[perm_uuid]
for perm_uuid in self.permissions.keys()
if perm_uuid in db.data().permissions
]
@property
def org(self) -> Org:
"""Get the organization object this role belongs to."""
return db.data().orgs[self.org_uuid]
@property
def users(self) -> list[User]:
"""Get all users that have this role."""
return [u for u in db.data().users.values() if u.role_uuid == self.uuid]
@classmethod @classmethod
def create( def create(
cls, cls,
org: UUID, org: UUID | Org,
display_name: str, display_name: str,
permissions: set[UUID] | None = None, permissions: set[UUID] | None = None,
) -> "Role": ) -> Role:
"""Create a new Role with auto-generated uuid7.""" """Create a new Role with auto-generated uuid7."""
org_uuid = org if isinstance(org, UUID) else org.uuid
role = cls( role = cls(
org=org, org_uuid=org_uuid,
display_name=display_name, display_name=display_name,
permissions={p: True for p in (permissions or set())}, permissions={p: True for p in (permissions or set())},
) )
@@ -85,32 +144,16 @@ class Role(msgspec.Struct, dict=True, omit_defaults=True):
return role return role
class Org(msgspec.Struct, dict=True):
"""Organization data structure."""
display_name: str
def __post_init__(self):
self.uuid: UUID = _UUID_UNSET # Convenience field, not serialized
@classmethod
def create(cls, display_name: str) -> "Org":
"""Create a new Org with auto-generated uuid7."""
org = cls(display_name=display_name)
org.uuid = uuid7.create()
return org
class User(msgspec.Struct, dict=True): class User(msgspec.Struct, dict=True):
"""User data structure. """User data structure.
Mutable fields: display_name, role, last_seen, visits Mutable fields: display_name, role_uuid, last_seen, visits
Immutable fields: created_at (set at creation, never modified) Immutable fields: created_at (set at creation, never modified)
uuid is derived from created_at using uuid7. uuid is derived from created_at using uuid7.
""" """
display_name: str display_name: str
role: UUID role_uuid: UUID = msgspec.field(name="role")
created_at: datetime created_at: datetime
last_seen: datetime | None = None last_seen: datetime | None = None
visits: int = 0 visits: int = 0
@@ -118,19 +161,44 @@ class User(msgspec.Struct, dict=True):
def __post_init__(self): def __post_init__(self):
self.uuid: UUID = _UUID_UNSET # Convenience field, not serialized self.uuid: UUID = _UUID_UNSET # Convenience field, not serialized
@property
def role(self) -> Role:
"""Get the role object this user has."""
return db.data().roles[self.role_uuid]
@property
def org(self) -> Org:
"""Get the organization this user belongs to (via role)."""
return self.role.org
@property
def credentials(self) -> list[Credential]:
"""Get all credentials for this user."""
return [c for c in db.data().credentials.values() if c.user_uuid == self.uuid]
@property
def sessions(self) -> list[Session]:
"""Get all sessions for this user."""
return [s for s in db.data().sessions.values() if s.user_uuid == self.uuid]
@property
def reset_tokens(self) -> list[ResetToken]:
"""Get all reset tokens for this user."""
return [t for t in db.data().reset_tokens.values() if t.user_uuid == self.uuid]
@classmethod @classmethod
def create( def create(
cls, cls,
display_name: str, display_name: str,
role: UUID, role: UUID | Role,
created_at: datetime | None = None, created_at: datetime | None = None,
) -> "User": ) -> User:
"""Create a new User with auto-generated uuid7.""" """Create a new User with auto-generated uuid7."""
role_uuid = role if isinstance(role, UUID) else role.uuid
user = cls( user = cls(
display_name=display_name, display_name=display_name,
role=role, role_uuid=role_uuid,
created_at=created_at or datetime.now(timezone.utc), created_at=created_at or datetime.now(UTC),
) )
user.uuid = uuid7.create(user.created_at) user.uuid = uuid7.create(user.created_at)
return user return user
@@ -145,7 +213,7 @@ class Credential(msgspec.Struct, dict=True):
""" """
credential_id: bytes # Long binary ID from the authenticator credential_id: bytes # Long binary ID from the authenticator
user: UUID user_uuid: UUID = msgspec.field(name="user")
aaguid: UUID aaguid: UUID
public_key: bytes public_key: bytes
sign_count: int sign_count: int
@@ -156,21 +224,34 @@ class Credential(msgspec.Struct, dict=True):
def __post_init__(self): def __post_init__(self):
self.uuid: UUID = _UUID_UNSET # Convenience field, not serialized self.uuid: UUID = _UUID_UNSET # Convenience field, not serialized
@property
def user(self) -> User:
"""Get the User object for this credential."""
return db.data().users[self.user_uuid]
@property
def sessions(self) -> list[Session]:
"""Get all sessions using this credential."""
return [
s for s in db.data().sessions.values() if s.credential_uuid == self.uuid
]
@classmethod @classmethod
def create( def create(
cls, cls,
credential_id: bytes, credential_id: bytes,
user: UUID, user: UUID | User,
aaguid: UUID, aaguid: UUID,
public_key: bytes, public_key: bytes,
sign_count: int, sign_count: int,
created_at: datetime | None = None, created_at: datetime | None = None,
) -> "Credential": ) -> Credential:
"""Create a new Credential with auto-generated uuid7.""" """Create a new Credential with auto-generated uuid7."""
now = created_at or datetime.now(timezone.utc) user_uuid = user if isinstance(user, UUID) else user.uuid
now = created_at or datetime.now(UTC)
cred = cls( cred = cls(
credential_id=credential_id, credential_id=credential_id,
user=user, user_uuid=user_uuid,
aaguid=aaguid, aaguid=aaguid,
public_key=public_key, public_key=public_key,
sign_count=sign_count, sign_count=sign_count,
@@ -186,12 +267,12 @@ class Session(msgspec.Struct, dict=True):
"""Session data structure. """Session data structure.
Mutable fields: expiry (updated on session refresh) Mutable fields: expiry (updated on session refresh)
Immutable fields: user, credential, host, ip, user_agent Immutable fields: user_uuid, credential_uuid, host, ip, user_agent
key is stored in the dict key, not in the struct. key is stored in the dict key, not in the struct.
""" """
user: UUID user_uuid: UUID = msgspec.field(name="user")
credential: UUID credential_uuid: UUID = msgspec.field(name="credential")
host: str host: str
ip: str ip: str
user_agent: str user_agent: str
@@ -200,6 +281,16 @@ class Session(msgspec.Struct, dict=True):
def __post_init__(self): def __post_init__(self):
self.key: str = "" # Convenience field, not serialized self.key: str = "" # Convenience field, not serialized
@property
def user(self) -> User:
"""Get the User object for this session."""
return db.data().users[self.user_uuid]
@property
def credential(self) -> Credential:
"""Get the Credential object for this session."""
return db.data().credentials[self.credential_uuid]
def metadata(self) -> dict: def metadata(self) -> dict:
"""Return session metadata for backwards compatibility.""" """Return session metadata for backwards compatibility."""
return { return {
@@ -208,6 +299,32 @@ class Session(msgspec.Struct, dict=True):
"expiry": self.expiry.isoformat(), "expiry": self.expiry.isoformat(),
} }
@classmethod
def create(
cls,
user: UUID | User,
credential: UUID | Credential,
host: str,
ip: str,
user_agent: str,
expiry: datetime,
) -> Session:
"""Create a new Session with auto-generated key."""
user_uuid = user if isinstance(user, UUID) else user.uuid
credential_uuid = (
credential if isinstance(credential, UUID) else credential.uuid
)
session = cls(
user_uuid=user_uuid,
credential_uuid=credential_uuid,
host=host,
ip=ip,
user_agent=user_agent,
expiry=expiry,
)
session.key = secrets.token_urlsafe(12)
return session
class ResetToken(msgspec.Struct, dict=True): class ResetToken(msgspec.Struct, dict=True):
"""Reset/device-addition token data structure. """Reset/device-addition token data structure.
@@ -216,13 +333,18 @@ class ResetToken(msgspec.Struct, dict=True):
key is stored in the dict key, not in the struct. key is stored in the dict key, not in the struct.
""" """
user: UUID user_uuid: UUID = msgspec.field(name="user")
expiry: datetime expiry: datetime
token_type: str token_type: str
def __post_init__(self): def __post_init__(self):
self.key: bytes = b"" # Convenience field, not serialized self.key: bytes = b"" # Convenience field, not serialized
@property
def user(self) -> User:
"""Get the User object for this reset token."""
return db.data().users[self.user_uuid]
class SessionContext(msgspec.Struct): class SessionContext(msgspec.Struct):
session: Session session: Session
@@ -296,18 +418,16 @@ class DB(msgspec.Struct, dict=True, omit_defaults=False):
return None return None
try: try:
user = self.users[s.user] user = s.user
role = self.roles[user.role] role = user.role
org = self.orgs[role.org] org = role.org
credential = self.credentials[s.credential] credential = s.credential
except KeyError: except KeyError:
return None return None
# Effective permissions: role's permissions that the org can grant # Effective permissions: role's permissions that the org can grant
# Also filter by domain if host is provided # Also filter by domain if host is provided
org_perm_uuids = { org_perm_uuids = {p.uuid for p in org.permissions}
pid for pid, p in self.permissions.items() if org.uuid in p.orgs
}
normalized_host = normalize_host(host) normalized_host = normalize_host(host)
host_without_port = ( host_without_port = (
normalized_host.rsplit(":", 1)[0] if normalized_host else None normalized_host.rsplit(":", 1)[0] if normalized_host else None
paskia/fastapi/admin.py
+44 -86
View File
@@ -1,5 +1,5 @@
import logging import logging
from datetime import timezone from datetime import UTC
from uuid import UUID from uuid import UUID
from fastapi import Body, FastAPI, HTTPException, Query, Request, Response from fastapi import Body, FastAPI, HTTPException, Query, Request, Response
@@ -28,38 +28,18 @@ from paskia.util.hostutil import normalize_host
app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None) app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
def is_global_admin(ctx) -> bool: def master_admin(ctx) -> bool:
"""Check if user has global admin permission.""" return any(p.scope == "auth:admin" for p in ctx.permissions)
effective_scopes = (
{p.scope for p in (ctx.permissions or [])}
if ctx.permissions def org_admin(ctx, org_uuid: UUID) -> bool:
else set(ctx.role.permissions or []) return ctx.org.uuid == org_uuid and any(
p.scope == "auth:org:admin" for p in ctx.permissions
) )
return "auth:admin" in effective_scopes
def is_org_admin(ctx, org_uuid: UUID | None = None) -> bool:
"""Check if user has org admin permission.
If org_uuid is provided, checks if user is admin of that specific org.
If org_uuid is None, checks if user is admin of their own org.
"""
effective_scopes = (
{p.scope for p in (ctx.permissions or [])}
if ctx.permissions
else set(ctx.role.permissions or [])
)
if "auth:org:admin" not in effective_scopes:
return False
if org_uuid is None:
return True
# User must belong to the target org (via their role)
return ctx.org.uuid == org_uuid
def can_manage_org(ctx, org_uuid: UUID) -> bool: def can_manage_org(ctx, org_uuid: UUID) -> bool:
"""Check if user can manage the specified organization.""" return master_admin(ctx) or org_admin(ctx, org_uuid)
return is_global_admin(ctx) or is_org_admin(ctx, org_uuid)
@app.exception_handler(ValueError) @app.exception_handler(ValueError)
@@ -99,14 +79,14 @@ async def admin_list_orgs(request: Request, auth=AUTH_COOKIE):
host=request.headers.get("host"), host=request.headers.get("host"),
) )
orgs = list(db.data().orgs.values()) orgs = list(db.data().orgs.values())
if not is_global_admin(ctx): if not master_admin(ctx):
# Org admins can only see their own organization # Org admins can only see their own organization
orgs = [o for o in orgs if o.uuid == ctx.org.uuid] orgs = [o for o in orgs if o.uuid == ctx.org.uuid]
def role_to_dict(r): def role_to_dict(r):
return { return {
"uuid": str(r.uuid), "uuid": str(r.uuid),
"org": str(r.org), "org": str(r.org_uuid),
"display_name": r.display_name, "display_name": r.display_name,
"permissions": list(r.permissions.keys()), "permissions": list(r.permissions.keys()),
} }
@@ -116,12 +96,8 @@ async def admin_list_orgs(request: Request, auth=AUTH_COOKIE):
return { return {
"uuid": str(o.uuid), "uuid": str(o.uuid),
"display_name": o.display_name, "display_name": o.display_name,
"permissions": { "permissions": {p.uuid for p in o.permissions},
pid for pid, p in db.data().permissions.items() if o.uuid in p.orgs "roles": [role_to_dict(r) for r in o.roles],
},
"roles": [
role_to_dict(r) for r in db.data().roles.values() if r.org == o.uuid
],
"users": [ "users": [
{ {
"uuid": str(u.uuid), "uuid": str(u.uuid),
@@ -283,7 +259,8 @@ async def admin_create_role(
perms = payload.get("permissions") or [] perms = payload.get("permissions") or []
if org_uuid not in db.data().orgs: if org_uuid not in db.data().orgs:
raise HTTPException(status_code=404, detail="Organization not found") 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} org = db.data().orgs[org_uuid]
grantable = {p.uuid for p in org.permissions}
# Normalize permission IDs to UUIDs # Normalize permission IDs to UUIDs
permission_uuids: set[UUID] = set() permission_uuids: set[UUID] = set()
@@ -324,7 +301,7 @@ async def admin_update_role_name(
status_code=403, detail="Insufficient permissions", mode="forbidden" status_code=403, detail="Insufficient permissions", mode="forbidden"
) )
role = db.data().roles.get(role_uuid) role = db.data().roles.get(role_uuid)
if not role or role.org != org_uuid: if not role or role.org_uuid != org_uuid:
raise HTTPException(status_code=404, detail="Role not found in organization") raise HTTPException(status_code=404, detail="Role not found in organization")
display_name = payload.get("display_name") display_name = payload.get("display_name")
@@ -356,7 +333,7 @@ async def admin_add_role_permission(
) )
role = db.data().roles.get(role_uuid) role = db.data().roles.get(role_uuid)
if not role or role.org != org_uuid: if not role or role.org_uuid != org_uuid:
raise HTTPException(status_code=404, detail="Role not found in organization") raise HTTPException(status_code=404, detail="Role not found in organization")
# Verify permission exists and org can grant it # Verify permission exists and org can grant it
@@ -391,7 +368,7 @@ async def admin_remove_role_permission(
) )
role = db.data().roles.get(role_uuid) role = db.data().roles.get(role_uuid)
if not role or role.org != org_uuid: if not role or role.org_uuid != org_uuid:
raise HTTPException(status_code=404, detail="Role not found in organization") raise HTTPException(status_code=404, detail="Role not found in organization")
# Sanity check: prevent admin from removing their own access # Sanity check: prevent admin from removing their own access
@@ -432,7 +409,7 @@ async def admin_delete_role(
status_code=403, detail="Insufficient permissions", mode="forbidden" status_code=403, detail="Insufficient permissions", mode="forbidden"
) )
role = db.data().roles.get(role_uuid) role = db.data().roles.get(role_uuid)
if not role or role.org != org_uuid: if not role or role.org_uuid != org_uuid:
raise HTTPException(status_code=404, detail="Role not found in organization") raise HTTPException(status_code=404, detail="Role not found in organization")
# Sanity check: prevent admin from deleting their own role # Sanity check: prevent admin from deleting their own role
@@ -468,8 +445,11 @@ async def admin_create_user(
if not display_name or not role_name: if not display_name or not role_name:
raise ValueError("display_name and role are required") raise ValueError("display_name and role are required")
roles = [r for r in db.data().roles.values() if r.org == org_uuid] org = db.data().orgs[org_uuid]
role_obj = next((r for r in roles if r.display_name == role_name), None) role_obj = next(
(r for r in org.roles if r.display_name == role_name),
None,
)
if not role_obj: if not role_obj:
raise ValueError("Role not found in organization") raise ValueError("Role not found in organization")
user = UserDC.create( user = UserDC.create(
@@ -507,7 +487,7 @@ async def admin_update_user_role(
raise ValueError("User not found") raise ValueError("User not found")
if user_org.uuid != org_uuid: if user_org.uuid != org_uuid:
raise ValueError("User does not belong to this organization") raise ValueError("User does not belong to this organization")
roles = [r for r in db.data().roles.values() if r.org == org_uuid] roles = user_org.roles
if not any(r.display_name == new_role for r in roles): if not any(r.display_name == new_role for r in roles):
raise ValueError("Role not found in organization") raise ValueError("Role not found in organization")
@@ -573,9 +553,9 @@ async def admin_create_user_registration_link(
return { return {
"url": url, "url": url,
"expires": ( "expires": (
expiry.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") expiry.astimezone(UTC).isoformat().replace("+00:00", "Z")
if expiry.tzinfo if expiry.tzinfo
else expiry.replace(tzinfo=timezone.utc).isoformat().replace("+00:00", "Z") else expiry.replace(tzinfo=UTC).isoformat().replace("+00:00", "Z")
), ),
} }
@@ -604,7 +584,7 @@ async def admin_get_user_detail(
status_code=403, detail="Insufficient permissions", mode="forbidden" status_code=403, detail="Insufficient permissions", mode="forbidden"
) )
user = db.data().users.get(user_uuid) user = db.data().users.get(user_uuid)
user_creds = [c for c in db.data().credentials.values() if c.user == user_uuid] user_creds = user.credentials
creds: list[dict] = [] creds: list[dict] = []
aaguids: set[str] = set() aaguids: set[str] = set()
for c in user_creds: for c in user_creds:
@@ -615,21 +595,17 @@ async def admin_get_user_detail(
"credential": str(c.uuid), "credential": str(c.uuid),
"aaguid": aaguid_str, "aaguid": aaguid_str,
"created_at": ( "created_at": (
c.created_at.astimezone(timezone.utc) c.created_at.astimezone(UTC).isoformat().replace("+00:00", "Z")
.isoformat()
.replace("+00:00", "Z")
if c.created_at.tzinfo if c.created_at.tzinfo
else c.created_at.replace(tzinfo=timezone.utc) else c.created_at.replace(tzinfo=UTC)
.isoformat() .isoformat()
.replace("+00:00", "Z") .replace("+00:00", "Z")
), ),
"last_used": ( "last_used": (
c.last_used.astimezone(timezone.utc) c.last_used.astimezone(UTC).isoformat().replace("+00:00", "Z")
.isoformat()
.replace("+00:00", "Z")
if c.last_used and c.last_used.tzinfo if c.last_used and c.last_used.tzinfo
else ( else (
c.last_used.replace(tzinfo=timezone.utc) c.last_used.replace(tzinfo=UTC)
.isoformat() .isoformat()
.replace("+00:00", "Z") .replace("+00:00", "Z")
if c.last_used if c.last_used
@@ -637,12 +613,10 @@ async def admin_get_user_detail(
) )
), ),
"last_verified": ( "last_verified": (
c.last_verified.astimezone(timezone.utc) c.last_verified.astimezone(UTC).isoformat().replace("+00:00", "Z")
.isoformat()
.replace("+00:00", "Z")
if c.last_verified and c.last_verified.tzinfo if c.last_verified and c.last_verified.tzinfo
else ( else (
c.last_verified.replace(tzinfo=timezone.utc) c.last_verified.replace(tzinfo=UTC)
.isoformat() .isoformat()
.replace("+00:00", "Z") .replace("+00:00", "Z")
if c.last_verified if c.last_verified
@@ -659,7 +633,7 @@ async def admin_get_user_detail(
# Get sessions for the user # Get sessions for the user
normalized_request_host = hostutil.normalize_host(request.headers.get("host")) normalized_request_host = hostutil.normalize_host(request.headers.get("host"))
session_records = [s for s in db.data().sessions.values() if s.user == user_uuid] session_records = user.sessions
current_session_key = auth current_session_key = auth
sessions_payload: list[dict] = [] sessions_payload: list[dict] = []
for entry in session_records: for entry in session_records:
@@ -672,11 +646,9 @@ async def admin_get_user_detail(
"ip": entry.ip, "ip": entry.ip,
"user_agent": useragent.compact_user_agent(entry.user_agent), "user_agent": useragent.compact_user_agent(entry.user_agent),
"last_renewed": ( "last_renewed": (
renewed.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") renewed.astimezone(UTC).isoformat().replace("+00:00", "Z")
if renewed.tzinfo if renewed.tzinfo
else renewed.replace(tzinfo=timezone.utc) else renewed.replace(tzinfo=UTC).isoformat().replace("+00:00", "Z")
.isoformat()
.replace("+00:00", "Z")
), ),
"is_current": entry.key == current_session_key, "is_current": entry.key == current_session_key,
"is_current_host": bool( "is_current_host": bool(
@@ -693,23 +665,19 @@ async def admin_get_user_detail(
"role": role_name, "role": role_name,
"visits": user.visits, "visits": user.visits,
"created_at": ( "created_at": (
user.created_at.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") user.created_at.astimezone(UTC).isoformat().replace("+00:00", "Z")
if user.created_at and user.created_at.tzinfo if user.created_at and user.created_at.tzinfo
else ( else (
user.created_at.replace(tzinfo=timezone.utc) user.created_at.replace(tzinfo=UTC).isoformat().replace("+00:00", "Z")
.isoformat()
.replace("+00:00", "Z")
if user.created_at if user.created_at
else None else None
) )
), ),
"last_seen": ( "last_seen": (
user.last_seen.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") user.last_seen.astimezone(UTC).isoformat().replace("+00:00", "Z")
if user.last_seen and user.last_seen.tzinfo if user.last_seen and user.last_seen.tzinfo
else ( else (
user.last_seen.replace(tzinfo=timezone.utc) user.last_seen.replace(tzinfo=UTC).isoformat().replace("+00:00", "Z")
.isoformat()
.replace("+00:00", "Z")
if user.last_seen if user.last_seen
else None else None
) )
@@ -808,7 +776,7 @@ async def admin_delete_user_session(
) )
target_session = db.data().sessions.get(session_id) target_session = db.data().sessions.get(session_id)
if not target_session or target_session.user != user_uuid: if not target_session or target_session.user_uuid != user_uuid:
raise HTTPException(status_code=404, detail="Session not found") raise HTTPException(status_code=404, detail="Session not found")
db.delete_session(session_id, ctx=ctx) db.delete_session(session_id, ctx=ctx)
@@ -919,18 +887,8 @@ async def admin_list_permissions(request: Request, auth=AUTH_COOKIE):
match=permutil.has_any, match=permutil.has_any,
host=request.headers.get("host"), host=request.headers.get("host"),
) )
perms = list(db.data().permissions.values()) perms = db.data().permissions.values() if master_admin(ctx) else ctx.org.permissions
return [_perm_to_dict(p) for p in perms]
# Global admins see all permissions
if is_global_admin(ctx):
return [_perm_to_dict(p) for p in perms]
# Org admins only see permissions their org can grant (by 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]
@app.post("/permissions") @app.post("/permissions")
paskia/fastapi/api.py
+5 -7
View File
@@ -1,6 +1,6 @@
import logging import logging
from contextlib import suppress from contextlib import suppress
from datetime import datetime, timedelta, timezone from datetime import UTC, datetime, timedelta
from fastapi import ( from fastapi import (
Depends, Depends,
@@ -90,7 +90,7 @@ async def validate_token(
raise raise
renewed = False renewed = False
if auth: if auth:
consumed = EXPIRES - (ctx.session.expiry - datetime.now(timezone.utc)) consumed = EXPIRES - (ctx.session.expiry - datetime.now(UTC))
if not timedelta(0) < consumed < _REFRESH_INTERVAL: if not timedelta(0) < consumed < _REFRESH_INTERVAL:
try: try:
refresh_session_token( refresh_session_token(
@@ -123,7 +123,7 @@ async def token_info(credentials=Depends(bearer_auth)):
except ValueError as e: except ValueError as e:
raise HTTPException(401, str(e)) raise HTTPException(401, str(e))
u = db.data().users.get(reset_token.user) u = reset_token.user
return { return {
"token_type": reset_token.token_type, "token_type": reset_token.token_type,
"display_name": u.display_name, "display_name": u.display_name,
@@ -170,11 +170,9 @@ async def forward_authentication(
"Remote-Role": str(ctx.role.uuid), "Remote-Role": str(ctx.role.uuid),
"Remote-Role-Name": ctx.role.display_name, "Remote-Role-Name": ctx.role.display_name,
"Remote-Session-Expires": ( "Remote-Session-Expires": (
ctx.session.expiry.astimezone(timezone.utc) ctx.session.expiry.astimezone(UTC).isoformat().replace("+00:00", "Z")
.isoformat()
.replace("+00:00", "Z")
if ctx.session.expiry.tzinfo if ctx.session.expiry.tzinfo
else ctx.session.expiry.replace(tzinfo=timezone.utc) else ctx.session.expiry.replace(tzinfo=UTC)
.isoformat() .isoformat()
.replace("+00:00", "Z") .replace("+00:00", "Z")
), ),
paskia/fastapi/remote.py
+4 -4
View File
@@ -324,7 +324,7 @@ async def websocket_remote_auth_permit(ws: WebSocket):
token_str = passphrase.generate() token_str = passphrase.generate()
expiry = expires() expiry = expires()
db.create_reset_token( db.create_reset_token(
user_uuid=cred.user, user_uuid=cred.user_uuid,
passphrase=token_str, passphrase=token_str,
expiry=expiry, expiry=expiry,
token_type="device addition", token_type="device addition",
@@ -333,7 +333,7 @@ async def websocket_remote_auth_permit(ws: WebSocket):
# Also create a session so the device is logged in # Also create a session so the device is logged in
normalized_host = hostutil.normalize_host(request.host) normalized_host = hostutil.normalize_host(request.host)
session_token = db.login( session_token = db.login(
user_uuid=cred.user, user_uuid=cred.user_uuid,
credential_uuid=cred.uuid, credential_uuid=cred.uuid,
sign_count=new_sign_count, sign_count=new_sign_count,
host=normalized_host, host=normalized_host,
@@ -346,7 +346,7 @@ async def websocket_remote_auth_permit(ws: WebSocket):
normalized_host = hostutil.normalize_host(request.host) normalized_host = hostutil.normalize_host(request.host)
session_token = db.login( session_token = db.login(
user_uuid=cred.user, user_uuid=cred.user_uuid,
credential_uuid=cred.uuid, credential_uuid=cred.uuid,
sign_count=new_sign_count, sign_count=new_sign_count,
host=normalized_host, host=normalized_host,
@@ -359,7 +359,7 @@ async def websocket_remote_auth_permit(ws: WebSocket):
completed = await remoteauth.instance.complete_request( completed = await remoteauth.instance.complete_request(
token=request.key, token=request.key,
session_token=session_token, session_token=session_token,
user_uuid=cred.user, user_uuid=cred.user_uuid,
credential_uuid=cred.uuid, credential_uuid=cred.uuid,
reset_token=reset_token, reset_token=reset_token,
) )
paskia/fastapi/user.py
+4 -4
View File
@@ -1,4 +1,4 @@
from datetime import timezone from datetime import UTC
from uuid import UUID from uuid import UUID
from fastapi import ( from fastapi import (
@@ -91,7 +91,7 @@ async def api_delete_session(
) )
target_session = db.data().sessions.get(session_id) target_session = db.data().sessions.get(session_id)
if not target_session or target_session.user != ctx.user.uuid: if not target_session or target_session.user_uuid != ctx.user.uuid:
raise HTTPException(status_code=404, detail="Session not found") raise HTTPException(status_code=404, detail="Session not found")
db.delete_session(session_id, ctx=ctx) db.delete_session(session_id, ctx=ctx)
@@ -141,8 +141,8 @@ async def api_create_link(
"message": "Registration link generated successfully", "message": "Registration link generated successfully",
"url": url, "url": url,
"expires": ( "expires": (
expiry.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") expiry.astimezone(UTC).isoformat().replace("+00:00", "Z")
if expiry.tzinfo if expiry.tzinfo
else expiry.replace(tzinfo=timezone.utc).isoformat().replace("+00:00", "Z") else expiry.replace(tzinfo=UTC).isoformat().replace("+00:00", "Z")
), ),
} }
paskia/fastapi/ws.py
+5 -5
View File
@@ -38,11 +38,11 @@ async def websocket_register_add(
f"The reset link for {passkey.instance.rp_name} is invalid or has expired" f"The reset link for {passkey.instance.rp_name} is invalid or has expired"
) )
s = get_reset(reset) s = get_reset(reset)
user_uuid = s.user user_uuid = s.user_uuid
else: else:
# Require recent authentication for adding a new passkey # Require recent authentication for adding a new passkey
ctx = await authz.verify(auth, perm=[], host=host, max_age="5m") ctx = await authz.verify(auth, perm=[], host=host, max_age="5m")
user_uuid = ctx.session.user user_uuid = ctx.session.user_uuid
s = ctx.session s = ctx.session
# Get user information and determine effective user_name for this registration # Get user information and determine effective user_name for this registration
@@ -99,7 +99,7 @@ async def websocket_authenticate(ws: WebSocket, auth=AUTH_COOKIE):
cred, new_sign_count = await authenticate_chat(ws, origin, credential_ids) cred, new_sign_count = await authenticate_chat(ws, origin, credential_ids)
# If reauth mode, verify the credential belongs to the session's user # If reauth mode, verify the credential belongs to the session's user
if session_user_uuid and cred.user != session_user_uuid: if session_user_uuid and cred.user_uuid != session_user_uuid:
raise ValueError("This passkey belongs to a different account") raise ValueError("This passkey belongs to a different account")
# Create session and update user/credential in a single transaction # Create session and update user/credential in a single transaction
@@ -114,7 +114,7 @@ async def websocket_authenticate(ws: WebSocket, auth=AUTH_COOKIE):
raise ValueError(f"Host must be the same as or a subdomain of {rp_id}") raise ValueError(f"Host must be the same as or a subdomain of {rp_id}")
token = db.login( token = db.login(
user_uuid=cred.user, user_uuid=cred.user_uuid,
credential_uuid=cred.uuid, credential_uuid=cred.uuid,
sign_count=new_sign_count, sign_count=new_sign_count,
host=normalized_host, host=normalized_host,
@@ -125,7 +125,7 @@ async def websocket_authenticate(ws: WebSocket, auth=AUTH_COOKIE):
await ws.send_json( await ws.send_json(
{ {
"user": str(cred.user), "user": str(cred.user_uuid),
"session_token": token, "session_token": token,
} }
) )
paskia/migrate/__init__.py
+6 -6
View File
@@ -14,7 +14,7 @@ Or via the CLI entry point (if installed):
import argparse import argparse
import asyncio import asyncio
import re import re
from datetime import datetime, timezone from datetime import UTC, datetime
from uuid import UUID from uuid import UUID
import base64url import base64url
@@ -173,7 +173,7 @@ async def migrate_from_sql(
new_user = User( new_user = User(
display_name=legacy_user.display_name, display_name=legacy_user.display_name,
role=legacy_user.role_uuid, role=legacy_user.role_uuid,
created_at=legacy_user.created_at or datetime.now(timezone.utc), created_at=legacy_user.created_at or datetime.now(UTC),
last_seen=legacy_user.last_seen, last_seen=legacy_user.last_seen,
visits=legacy_user.visits, visits=legacy_user.visits,
) )
@@ -190,7 +190,7 @@ async def migrate_from_sql(
cred_key: UUID = legacy_cred.uuid cred_key: UUID = legacy_cred.uuid
new_cred = Credential( new_cred = Credential(
credential_id=legacy_cred.credential_id, credential_id=legacy_cred.credential_id,
user=legacy_cred.user_uuid, user_uuid=legacy_cred.user_uuid,
aaguid=legacy_cred.aaguid, aaguid=legacy_cred.aaguid,
public_key=legacy_cred.public_key, public_key=legacy_cred.public_key,
sign_count=legacy_cred.sign_count, sign_count=legacy_cred.sign_count,
@@ -217,8 +217,8 @@ async def migrate_from_sql(
# Already in new format or unknown - try to use as-is # Already in new format or unknown - try to use as-is
session_key = base64url.enc(old_key[:12]) session_key = base64url.enc(old_key[:12])
db.sessions[session_key] = Session( db.sessions[session_key] = Session(
user=sess.user_uuid, user_uuid=sess.user_uuid,
credential=sess.credential_uuid, credential_uuid=sess.credential_uuid,
host=sess.host, host=sess.host,
ip=sess.ip, ip=sess.ip,
user_agent=sess.user_agent, user_agent=sess.user_agent,
@@ -241,7 +241,7 @@ async def migrate_from_sql(
# Already in new format or unknown - truncate to 9 bytes # Already in new format or unknown - truncate to 9 bytes
token_key = old_key[:9] token_key = old_key[:9]
db.reset_tokens[token_key] = ResetToken( db.reset_tokens[token_key] = ResetToken(
user=token.user_uuid, user_uuid=token.user_uuid,
expiry=token.expiry, expiry=token.expiry,
token_type=token.token_type, token_type=token.token_type,
) )
paskia/migrate/sql.py
+7 -7
View File
@@ -9,7 +9,7 @@ DO NOT use this module for new code. Use paskia.db instead.
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
from dataclasses import dataclass from dataclasses import dataclass
from datetime import datetime, timezone from datetime import UTC, datetime
from uuid import UUID from uuid import UUID
from sqlalchemy import ( from sqlalchemy import (
@@ -112,8 +112,8 @@ def _normalize_dt(value: datetime | None) -> datetime | None:
if value is None: if value is None:
return None return None
if value.tzinfo is None: if value.tzinfo is None:
return value.replace(tzinfo=timezone.utc) return value.replace(tzinfo=UTC)
return value.astimezone(timezone.utc) return value.astimezone(UTC)
class Base(DeclarativeBase): class Base(DeclarativeBase):
@@ -172,7 +172,7 @@ class UserModel(Base):
LargeBinary(16), ForeignKey("roles.uuid", ondelete="CASCADE"), nullable=False LargeBinary(16), ForeignKey("roles.uuid", ondelete="CASCADE"), nullable=False
) )
created_at: Mapped[datetime] = mapped_column( created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), default=lambda: datetime.now(timezone.utc) DateTime(timezone=True), default=lambda: datetime.now(UTC)
) )
last_seen: Mapped[datetime | None] = mapped_column( last_seen: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), nullable=True DateTime(timezone=True), nullable=True
@@ -195,7 +195,7 @@ class UserModel(Base):
uuid=user.uuid.bytes, uuid=user.uuid.bytes,
display_name=user.display_name, display_name=user.display_name,
role_uuid=user.role_uuid.bytes, role_uuid=user.role_uuid.bytes,
created_at=user.created_at or datetime.now(timezone.utc), created_at=user.created_at or datetime.now(UTC),
last_seen=user.last_seen, last_seen=user.last_seen,
visits=user.visits, visits=user.visits,
) )
@@ -215,7 +215,7 @@ class CredentialModel(Base):
public_key: Mapped[bytes] = mapped_column(BLOB, nullable=False) public_key: Mapped[bytes] = mapped_column(BLOB, nullable=False)
sign_count: Mapped[int] = mapped_column(Integer, nullable=False) sign_count: Mapped[int] = mapped_column(Integer, nullable=False)
created_at: Mapped[datetime] = mapped_column( created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), default=lambda: datetime.now(timezone.utc) DateTime(timezone=True), default=lambda: datetime.now(UTC)
) )
last_used: Mapped[datetime | None] = mapped_column( last_used: Mapped[datetime | None] = mapped_column(
DateTime(timezone=True), nullable=True DateTime(timezone=True), nullable=True
@@ -255,7 +255,7 @@ class SessionModel(Base):
user_agent: Mapped[str] = mapped_column(String(512), nullable=False) user_agent: Mapped[str] = mapped_column(String(512), nullable=False)
renewed: Mapped[datetime] = mapped_column( renewed: Mapped[datetime] = mapped_column(
DateTime(timezone=True), DateTime(timezone=True),
default=lambda: datetime.now(timezone.utc), default=lambda: datetime.now(UTC),
nullable=False, nullable=False,
) )
paskia/remoteauth.py
+6 -6
View File
@@ -19,9 +19,9 @@ The first 3 words of the token serve as the pairing code for manual entry.
import asyncio import asyncio
import logging import logging
from collections.abc import Callable
from dataclasses import dataclass from dataclasses import dataclass
from datetime import datetime, timedelta, timezone from datetime import UTC, datetime, timedelta
from typing import Callable
from uuid import UUID from uuid import UUID
from paskia.util import passphrase, pow from paskia.util import passphrase, pow
@@ -94,7 +94,7 @@ class RemoteAuthManager:
async def _cleanup_expired(self): async def _cleanup_expired(self):
"""Remove expired requests and notify waiting clients.""" """Remove expired requests and notify waiting clients."""
now = datetime.now(timezone.utc) now = datetime.now(UTC)
expired_keys = [] expired_keys = []
async with self._lock: async with self._lock:
for key, req in self._requests.items(): for key, req in self._requests.items():
@@ -123,7 +123,7 @@ class RemoteAuthManager:
Returns: Returns:
(code, expiry) - The 3-word passphrase code and expiration time (code, expiry) - The 3-word passphrase code and expiration time
""" """
now = datetime.now(timezone.utc) now = datetime.now(UTC)
expiry = now + REMOTE_AUTH_LIFETIME expiry = now + REMOTE_AUTH_LIFETIME
async with self._lock: async with self._lock:
@@ -160,7 +160,7 @@ class RemoteAuthManager:
req = self._requests.get(normalized) req = self._requests.get(normalized)
if req is None: if req is None:
return None return None
now = datetime.now(timezone.utc) now = datetime.now(UTC)
if now > req.created_at + REMOTE_AUTH_LIFETIME: if now > req.created_at + REMOTE_AUTH_LIFETIME:
# Expired # Expired
del self._requests[normalized] del self._requests[normalized]
@@ -331,7 +331,7 @@ class RemoteAuthManager:
req = self._requests.get(token) req = self._requests.get(token)
if req is None: if req is None:
return None return None
now = datetime.now(timezone.utc) now = datetime.now(UTC)
if now > req.created_at + REMOTE_AUTH_LIFETIME: if now > req.created_at + REMOTE_AUTH_LIFETIME:
del self._requests[token] del self._requests[token]
return None return None
paskia/util/sessionutil.py
+2 -2
View File
@@ -1,6 +1,6 @@
"""Utility functions for session validation and checking.""" """Utility functions for session validation and checking."""
from datetime import datetime, timezone from datetime import UTC, datetime
from paskia.authsession import EXPIRES from paskia.authsession import EXPIRES
from paskia.db import SessionContext from paskia.db import SessionContext
@@ -34,5 +34,5 @@ def check_session_age(ctx: SessionContext, max_age: str | None) -> bool:
else: else:
auth_time = ctx.session.expiry - EXPIRES auth_time = ctx.session.expiry - EXPIRES
time_since_auth = datetime.now(timezone.utc) - auth_time time_since_auth = datetime.now(UTC) - auth_time
return time_since_auth <= max_age_delta return time_since_auth <= max_age_delta
paskia/util/userinfo.py
+6 -7
View File
@@ -1,6 +1,6 @@
"""User information formatting and retrieval logic.""" """User information formatting and retrieval logic."""
from datetime import timezone from datetime import UTC
from paskia import aaguid, db from paskia import aaguid, db
from paskia.authsession import EXPIRES from paskia.authsession import EXPIRES
@@ -13,9 +13,9 @@ def _format_datetime(dt):
if dt is None: if dt is None:
return None return None
if dt.tzinfo: if dt.tzinfo:
return dt.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") return dt.astimezone(UTC).isoformat().replace("+00:00", "Z")
else: else:
return dt.replace(tzinfo=timezone.utc).isoformat().replace("+00:00", "Z") return dt.replace(tzinfo=UTC).isoformat().replace("+00:00", "Z")
def format_session_context(ctx: SessionContext) -> dict: def format_session_context(ctx: SessionContext) -> dict:
@@ -48,9 +48,8 @@ async def format_user_info(
ctx = await permutil.session_context(auth, request_host) ctx = await permutil.session_context(auth, request_host)
# Fetch and format credentials # Fetch and format credentials
user_credentials = [ user = db.data().users[user_uuid]
c for c in db.data().credentials.values() if c.user == user_uuid user_credentials = user.credentials
]
credentials: list[dict] = [] credentials: list[dict] = []
user_aaguids: set[str] = set() user_aaguids: set[str] = set()
@@ -74,7 +73,7 @@ async def format_user_info(
# Format sessions # Format sessions
normalized_request_host = hostutil.normalize_host(request_host) normalized_request_host = hostutil.normalize_host(request_host)
session_records = [s for s in db.data().sessions.values() if s.user == user_uuid] session_records = user.sessions
current_session_key = auth current_session_key = auth
sessions_payload: list[dict] = [] sessions_payload: list[dict] = []
pyproject.toml
-4
View File
@@ -74,10 +74,6 @@ filterwarnings = [
"ignore::DeprecationWarning", "ignore::DeprecationWarning",
] ]
[tool.ruff]
target-version = "py39"
line-length = 88
[tool.ruff.lint] [tool.ruff.lint]
select = ["E", "F", "I", "N", "W", "UP", "PLC0415"] select = ["E", "F", "I", "N", "W", "UP", "PLC0415"]
ignore = ["E501"] # Line too long ignore = ["E501"] # Line too long
tests/conftest.py
+3 -9
View File
@@ -38,7 +38,7 @@ from paskia.db import (
create_user, create_user,
) )
from paskia.db.jsonl import JsonlStore from paskia.db.jsonl import JsonlStore
from paskia.db.operations import DB, _create_token from paskia.db.operations import DB
from paskia.fastapi.mainapp import app from paskia.fastapi.mainapp import app
from paskia.fastapi.session import AUTH_COOKIE_NAME from paskia.fastapi.session import AUTH_COOKIE_NAME
from paskia.sansio import Passkey from paskia.sansio import Passkey
@@ -193,17 +193,14 @@ async def session_token(
test_db: DB, test_user: User, test_credential: Credential test_db: DB, test_user: User, test_credential: Credential
) -> str: ) -> str:
"""Create a session for the admin user and return the token.""" """Create a session for the admin user and return the token."""
token = _create_token() return create_session(
create_session(
user_uuid=test_user.uuid, user_uuid=test_user.uuid,
credential_uuid=test_credential.uuid, credential_uuid=test_credential.uuid,
key=token,
host="localhost:4401", host="localhost:4401",
ip="127.0.0.1", ip="127.0.0.1",
user_agent="pytest", user_agent="pytest",
expiry=expires(), expiry=expires(),
) )
return token
@pytest_asyncio.fixture(scope="function") @pytest_asyncio.fixture(scope="function")
@@ -211,17 +208,14 @@ async def regular_session_token(
test_db: DB, regular_user: User, regular_credential: Credential test_db: DB, regular_user: User, regular_credential: Credential
) -> str: ) -> str:
"""Create a session for a regular user and return the token.""" """Create a session for a regular user and return the token."""
token = _create_token() return create_session(
create_session(
user_uuid=regular_user.uuid, user_uuid=regular_user.uuid,
credential_uuid=regular_credential.uuid, credential_uuid=regular_credential.uuid,
key=token,
host="localhost:4401", host="localhost:4401",
ip="127.0.0.1", ip="127.0.0.1",
user_agent="pytest", user_agent="pytest",
expiry=expires(), expiry=expires(),
) )
return token
@pytest_asyncio.fixture(scope="function") @pytest_asyncio.fixture(scope="function")
tests/test_admin.py
+8 -15
View File
@@ -12,7 +12,8 @@ These tests cover:
""" """
import os import os
from datetime import datetime, timezone import secrets
from datetime import UTC, datetime
from uuid import UUID from uuid import UUID
import httpx import httpx
@@ -36,7 +37,7 @@ from paskia.db import (
create_session, create_session,
create_user, create_user,
) )
from paskia.db.operations import DB, _create_token from paskia.db.operations import DB
from tests.conftest import auth_headers from tests.conftest import auth_headers
# -------------------- Additional Fixtures -------------------- # -------------------- Additional Fixtures --------------------
@@ -97,17 +98,14 @@ async def second_org_session_token(
test_db: DB, second_org_user: User, second_org_credential: Credential test_db: DB, second_org_user: User, second_org_credential: Credential
) -> str: ) -> str:
"""Create a session for the second org admin user.""" """Create a session for the second org admin user."""
token = _create_token() return create_session(
create_session(
user_uuid=second_org_user.uuid, user_uuid=second_org_user.uuid,
credential_uuid=second_org_credential.uuid, credential_uuid=second_org_credential.uuid,
key=token,
host="localhost:4401", host="localhost:4401",
ip="127.0.0.1", ip="127.0.0.1",
user_agent="pytest", user_agent="pytest",
expiry=expires(), expiry=expires(),
) )
return token
@pytest_asyncio.fixture(scope="function") @pytest_asyncio.fixture(scope="function")
@@ -132,7 +130,7 @@ async def org_admin_user(test_db: DB, org_admin_role: Role) -> User:
role=org_admin_role.uuid, role=org_admin_role.uuid,
) )
user.visits = 5 user.visits = 5
user.last_seen = datetime.now(timezone.utc) user.last_seen = datetime.now(UTC)
create_user(user) create_user(user)
return user return user
@@ -157,17 +155,14 @@ async def org_admin_session_token(
test_db: DB, org_admin_user: User, org_admin_credential: Credential test_db: DB, org_admin_user: User, org_admin_credential: Credential
) -> str: ) -> str:
"""Create a session for the org admin user.""" """Create a session for the org admin user."""
token = _create_token() return create_session(
create_session(
user_uuid=org_admin_user.uuid, user_uuid=org_admin_user.uuid,
credential_uuid=org_admin_credential.uuid, credential_uuid=org_admin_credential.uuid,
key=token,
host="localhost:4401", host="localhost:4401",
ip="127.0.0.1", ip="127.0.0.1",
user_agent="pytest", user_agent="pytest",
expiry=expires(), expiry=expires(),
) )
return token
@pytest_asyncio.fixture(scope="function") @pytest_asyncio.fixture(scope="function")
@@ -1170,11 +1165,9 @@ class TestAdminSessions:
): ):
"""Admin should be able to delete a user's session.""" """Admin should be able to delete a user's session."""
# Create an additional session to delete # Create an additional session to delete
extra_token = _create_token() extra_token = create_session(
create_session(
user_uuid=test_user.uuid, user_uuid=test_user.uuid,
credential_uuid=test_credential.uuid, credential_uuid=test_credential.uuid,
key=extra_token,
host="other.host:4401", host="other.host:4401",
ip="192.168.1.1", ip="192.168.1.1",
user_agent="other-agent", user_agent="other-agent",
@@ -1255,7 +1248,7 @@ class TestAdminSessions:
): ):
"""Deleting non-existent session should fail.""" """Deleting non-existent session should fail."""
# Use a valid format but non-existent key # Use a valid format but non-existent key
fake_token = _create_token() fake_token = secrets.token_urlsafe(12)
response = await client.delete( response = await client.delete(
f"/auth/api/admin/orgs/{test_org.uuid}/users/{test_user.uuid}/sessions/{fake_token}", f"/auth/api/admin/orgs/{test_org.uuid}/users/{test_user.uuid}/sessions/{fake_token}",
headers={**auth_headers(session_token), "Host": "localhost:4401"}, headers={**auth_headers(session_token), "Host": "localhost:4401"},
tests/test_api.py
+5 -7
View File
@@ -10,14 +10,14 @@ These tests cover:
- /auth/api/set-session - Set session from bearer token - /auth/api/set-session - Set session from bearer token
""" """
from datetime import datetime, timedelta, timezone import secrets
from datetime import UTC, datetime, timedelta
import httpx import httpx
import pytest import pytest
from paskia.authsession import EXPIRES from paskia.authsession import EXPIRES
from paskia.db import create_session, delete_session from paskia.db import create_session, delete_session
from paskia.db.operations import _create_token
from paskia.util.passphrase import generate from paskia.util.passphrase import generate
from tests.conftest import auth_headers from tests.conftest import auth_headers
@@ -503,7 +503,7 @@ class TestValidateSessionRefresh:
"""Validate should handle session expiry during refresh attempt.""" """Validate should handle session expiry during refresh attempt."""
# Create a token but don't create a session for it # Create a token but don't create a session for it
token = _create_token() token = secrets.token_urlsafe(12)
response = await client.post( response = await client.post(
"/auth/api/validate", "/auth/api/validate",
headers={**auth_headers(token), "Host": "localhost:4401"}, headers={**auth_headers(token), "Host": "localhost:4401"},
@@ -522,12 +522,10 @@ class TestValidateSessionRefresh:
"""Validate should return 401 if session disappears during refresh.""" """Validate should return 401 if session disappears during refresh."""
# Create a session with an old expiry time to trigger refresh # Create a session with an old expiry time to trigger refresh
token = _create_token() old_expiry = datetime.now(UTC) + EXPIRES - timedelta(minutes=10)
old_expiry = datetime.now(timezone.utc) + EXPIRES - timedelta(minutes=10) token = create_session(
create_session(
user_uuid=test_user.uuid, user_uuid=test_user.uuid,
credential_uuid=test_credential.uuid, credential_uuid=test_credential.uuid,
key=token,
host="localhost:4401", host="localhost:4401",
ip="127.0.0.1", ip="127.0.0.1",
user_agent="pytest", user_agent="pytest",