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

API to use msgspec structs as well.

Browse Source
This commit is contained in:
Leo Vasanko
2026-01-28 18:31:45 +00:00
parent b79f367495
commit 7614d0e8d9
5 changed files with 118 additions and 232 deletions
Download Patch File Download Diff File Expand all files Collapse all files
paskia/aaguid/__init__.py
+5 -4
View File
@@ -10,6 +10,7 @@ This module provides functionality to:
import json
from collections.abc import Iterable
from importlib.resources import files
from uuid import UUID
__ALL__ = ["AAGUID", "filter"]
@@ -18,15 +19,15 @@ AAGUID_FILE = files("paskia") / "aaguid" / "combined_aaguid.json"
AAGUID: dict[str, dict] = json.loads(AAGUID_FILE.read_text(encoding="utf-8"))
def filter(aaguids: Iterable[str]) -> dict[str, dict]:
def filter(aaguids: Iterable[UUID]) -> dict[str, dict]:
"""
Get AAGUID information only for the provided set of AAGUIDs.
Args:
aaguids: Set of AAGUID strings that the user has credentials for
aaguids: Iterable of AAGUIDs (UUIDs) that the user has credentials for
Returns:
Dictionary mapping AAGUID to authenticator information for only
Dictionary mapping AAGUID string to authenticator information for only
the AAGUIDs that the user has and that we have data for
"""
return {aaguid: AAGUID[aaguid] for aaguid in aaguids if aaguid in AAGUID}
return {(s := str(a)): AAGUID[s] for a in aaguids if (s := str(a)) in AAGUID}
paskia/db/structs.py
+14 -7
View File
@@ -28,7 +28,8 @@ class Permission(msgspec.Struct, dict=True, omit_defaults=True):
orgs: dict[UUID, bool] = {} # org_uuid -> True (which orgs can grant this)
def __post_init__(self):
self.uuid: UUID = _UUID_UNSET # Convenience field, not serialized
if getattr(self, "uuid", _UUID_UNSET) == _UUID_UNSET:
self.uuid: UUID = _UUID_UNSET
@property
def org_set(self) -> set[UUID]:
@@ -67,7 +68,8 @@ class Org(msgspec.Struct, dict=True):
display_name: str
def __post_init__(self):
self.uuid: UUID = _UUID_UNSET # Convenience field, not serialized
if getattr(self, "uuid", _UUID_UNSET) == _UUID_UNSET:
self.uuid: UUID = _UUID_UNSET
@property
def roles(self) -> list[Role]:
@@ -100,7 +102,8 @@ class Role(msgspec.Struct, dict=True, omit_defaults=True):
permissions: dict[UUID, bool] = {} # permission_uuid -> True
def __post_init__(self):
self.uuid: UUID = _UUID_UNSET # Convenience field, not serialized
if getattr(self, "uuid", _UUID_UNSET) == _UUID_UNSET:
self.uuid: UUID = _UUID_UNSET
@property
def permission_set(self) -> set[UUID]:
@@ -159,7 +162,8 @@ class User(msgspec.Struct, dict=True):
visits: int = 0
def __post_init__(self):
self.uuid: UUID = _UUID_UNSET # Convenience field, not serialized
if getattr(self, "uuid", _UUID_UNSET) == _UUID_UNSET:
self.uuid: UUID = _UUID_UNSET
@property
def role(self) -> Role:
@@ -222,7 +226,8 @@ class Credential(msgspec.Struct, dict=True):
last_verified: datetime | None = None
def __post_init__(self):
self.uuid: UUID = _UUID_UNSET # Convenience field, not serialized
if getattr(self, "uuid", _UUID_UNSET) == _UUID_UNSET:
self.uuid: UUID = _UUID_UNSET
@property
def user(self) -> User:
@@ -279,7 +284,8 @@ class Session(msgspec.Struct, dict=True):
expiry: datetime
def __post_init__(self):
self.key: str = "" # Convenience field, not serialized
if not getattr(self, "key", ""):
self.key: str = ""
@property
def user(self) -> User:
@@ -338,7 +344,8 @@ class ResetToken(msgspec.Struct, dict=True):
token_type: str
def __post_init__(self):
self.key: bytes = b"" # Convenience field, not serialized
if not getattr(self, "key", b""):
self.key: bytes = b""
@property
def user(self) -> User:
paskia/fastapi/admin.py
+50 -131
View File
@@ -1,5 +1,4 @@
import logging
from datetime import UTC
from uuid import UUID
from fastapi import Body, FastAPI, HTTPException, Query, Request, Response
@@ -13,6 +12,7 @@ from paskia.db import Permission as PermDC
from paskia.db import Role as RoleDC
from paskia.db import User as UserDC
from paskia.fastapi import authz
from paskia.fastapi.response import MsgspecResponse
from paskia.fastapi.session import AUTH_COOKIE
from paskia.globals import passkey
from paskia.util import (
@@ -20,9 +20,9 @@ from paskia.util import (
passphrase,
permutil,
querysafe,
useragent,
vitedev,
)
from paskia.util.apistructs import ApiPermission, ApiSession, format_datetime
from paskia.util.hostutil import normalize_host
app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
@@ -83,34 +83,34 @@ async def admin_list_orgs(request: Request, auth=AUTH_COOKIE):
# Org admins can only see their own organization
orgs = [o for o in orgs if o.uuid == ctx.org.uuid]
def role_to_dict(r):
return {
"uuid": str(r.uuid),
"org": str(r.org_uuid),
"display_name": r.display_name,
"permissions": list(r.permissions.keys()),
}
async def org_to_dict(o):
def org_to_dict(o):
users = db.get_organization_users(o.uuid)
return {
"uuid": str(o.uuid),
"uuid": o.uuid,
"display_name": o.display_name,
"permissions": {p.uuid for p in o.permissions},
"roles": [role_to_dict(r) for r in o.roles],
"roles": [
{
"uuid": r.uuid,
"org": r.org_uuid,
"display_name": r.display_name,
"permissions": list(r.permissions.keys()),
}
for r in o.roles
],
"users": [
{
"uuid": str(u.uuid),
"uuid": u.uuid,
"display_name": u.display_name,
"role": role_name,
"visits": u.visits,
"last_seen": u.last_seen.isoformat() if u.last_seen else None,
"last_seen": u.last_seen,
}
for (u, role_name) in users
],
}
return [await org_to_dict(o) for o in orgs]
return MsgspecResponse([org_to_dict(o) for o in orgs])
@app.post("/orgs")
@@ -552,11 +552,7 @@ async def admin_create_user_registration_link(
url = hostutil.reset_link_url(token)
return {
"url": url,
"expires": (
expiry.astimezone(UTC).isoformat().replace("+00:00", "Z")
if expiry.tzinfo
else expiry.replace(tzinfo=UTC).isoformat().replace("+00:00", "Z")
),
"expires": format_datetime(expiry),
}
@@ -584,108 +580,39 @@ async def admin_get_user_detail(
status_code=403, detail="Insufficient permissions", mode="forbidden"
)
user = db.data().users.get(user_uuid)
user_creds = user.credentials
creds: list[dict] = []
aaguids: set[str] = set()
for c in user_creds:
aaguid_str = str(c.aaguid)
aaguids.add(aaguid_str)
creds.append(
{
"credential": str(c.uuid),
"aaguid": aaguid_str,
"created_at": (
c.created_at.astimezone(UTC).isoformat().replace("+00:00", "Z")
if c.created_at.tzinfo
else c.created_at.replace(tzinfo=UTC)
.isoformat()
.replace("+00:00", "Z")
),
"last_used": (
c.last_used.astimezone(UTC).isoformat().replace("+00:00", "Z")
if c.last_used and c.last_used.tzinfo
else (
c.last_used.replace(tzinfo=UTC)
.isoformat()
.replace("+00:00", "Z")
if c.last_used
else None
)
),
"last_verified": (
c.last_verified.astimezone(UTC).isoformat().replace("+00:00", "Z")
if c.last_verified and c.last_verified.tzinfo
else (
c.last_verified.replace(tzinfo=UTC)
.isoformat()
.replace("+00:00", "Z")
if c.last_verified
else None
)
normalized_host = hostutil.normalize_host(request.headers.get("host"))
return MsgspecResponse(
{
"display_name": user.display_name,
"org": {"display_name": user_org.display_name},
"role": role_name,
"visits": user.visits,
"created_at": user.created_at,
"last_seen": user.last_seen,
"credentials": [
{
"credential": c.uuid,
"aaguid": c.aaguid,
"created_at": c.created_at,
"last_used": c.last_used,
"last_verified": c.last_verified,
"sign_count": c.sign_count,
}
for c in user.credentials
],
"aaguid_info": aaguid_mod.filter(c.aaguid for c in user.credentials),
"sessions": [
ApiSession.from_db(
s,
current_key=auth,
normalized_host=normalized_host,
expires_delta=EXPIRES,
)
if c.last_verified
else None,
"sign_count": c.sign_count,
}
)
aaguid_info = aaguid_mod.filter(aaguids)
# Get sessions for the user
normalized_request_host = hostutil.normalize_host(request.headers.get("host"))
session_records = user.sessions
current_session_key = auth
sessions_payload: list[dict] = []
for entry in session_records:
renewed = entry.expiry - EXPIRES
sessions_payload.append(
{
"id": entry.key,
"credential": str(entry.credential),
"host": entry.host,
"ip": entry.ip,
"user_agent": useragent.compact_user_agent(entry.user_agent),
"last_renewed": (
renewed.astimezone(UTC).isoformat().replace("+00:00", "Z")
if renewed.tzinfo
else renewed.replace(tzinfo=UTC).isoformat().replace("+00:00", "Z")
),
"is_current": entry.key == current_session_key,
"is_current_host": bool(
normalized_request_host
and entry.host
and entry.host == normalized_request_host
),
}
)
return {
"display_name": user.display_name,
"org": {"display_name": user_org.display_name},
"role": role_name,
"visits": user.visits,
"created_at": (
user.created_at.astimezone(UTC).isoformat().replace("+00:00", "Z")
if user.created_at and user.created_at.tzinfo
else (
user.created_at.replace(tzinfo=UTC).isoformat().replace("+00:00", "Z")
if user.created_at
else None
)
),
"last_seen": (
user.last_seen.astimezone(UTC).isoformat().replace("+00:00", "Z")
if user.last_seen and user.last_seen.tzinfo
else (
user.last_seen.replace(tzinfo=UTC).isoformat().replace("+00:00", "Z")
if user.last_seen
else None
)
),
"credentials": creds,
"aaguid_info": aaguid_info,
"sessions": sessions_payload,
}
for s in user.sessions
],
}
)
@app.patch("/orgs/{org_uuid}/users/{user_uuid}/display-name")
@@ -789,14 +716,6 @@ async def admin_delete_user_session(
# -------------------- Permissions (global) --------------------
def _perm_to_dict(p):
"""Convert Permission to dict, omitting domain if None."""
d = {"uuid": str(p.uuid), "scope": p.scope, "display_name": p.display_name}
if p.domain is not None:
d["domain"] = p.domain
return d
def _validate_permission_domain(domain: str | None) -> None:
"""Validate that domain is rp_id or a subdomain of it."""
if domain is None:
@@ -888,7 +807,7 @@ async def admin_list_permissions(request: Request, auth=AUTH_COOKIE):
host=request.headers.get("host"),
)
perms = db.data().permissions.values() if master_admin(ctx) else ctx.org.permissions
return [_perm_to_dict(p) for p in perms]
return MsgspecResponse([ApiPermission.from_db(p) for p in perms])
@app.post("/permissions")
paskia/fastapi/api.py
+15 -10
View File
@@ -20,6 +20,7 @@ from paskia.authsession import (
refresh_session_token,
)
from paskia.fastapi import authz, session, user
from paskia.fastapi.response import MsgspecResponse
from paskia.fastapi.session import AUTH_COOKIE, AUTH_COOKIE_NAME
from paskia.globals import passkey as global_passkey
from paskia.util import hostutil, htmlutil, passphrase, userinfo, vitedev
@@ -105,11 +106,13 @@ async def validate_token(
raise authz.AuthException(
status_code=401, detail="Session expired", mode="login"
)
return {
"valid": True,
"renewed": renewed,
"ctx": userinfo.format_session_context(ctx),
}
return MsgspecResponse(
{
"valid": True,
"renewed": renewed,
"ctx": userinfo.build_session_context(ctx),
}
)
@app.get("/token-info")
@@ -235,11 +238,13 @@ async def api_user_info(
if not ctx:
raise HTTPException(401, "Session expired")
return await userinfo.format_user_info(
user_uuid=ctx.user.uuid,
auth=auth,
session_record=ctx.session,
request_host=request.headers.get("host"),
return MsgspecResponse(
await userinfo.build_user_info(
user_uuid=ctx.user.uuid,
auth=auth,
session_record=ctx.session,
request_host=request.headers.get("host"),
)
)
paskia/util/userinfo.py
+34 -80
View File
@@ -1,106 +1,60 @@
"""User information formatting and retrieval logic."""
from datetime import UTC
from paskia import aaguid, db
from paskia.authsession import EXPIRES
from paskia.db import SessionContext
from paskia.util import hostutil, permutil, useragent
from paskia.util import hostutil, permutil
from paskia.util.apistructs import ApiSession
def _format_datetime(dt):
"""Format a datetime object to ISO 8601 string with UTC timezone."""
if dt is None:
return None
if dt.tzinfo:
return dt.astimezone(UTC).isoformat().replace("+00:00", "Z")
else:
return dt.replace(tzinfo=UTC).isoformat().replace("+00:00", "Z")
def format_session_context(ctx: SessionContext) -> dict:
"""Format SessionContext for JSON response."""
def build_session_context(ctx: SessionContext) -> dict:
"""Build session context dict from SessionContext."""
return {
"user": {
"uuid": str(ctx.user.uuid),
"display_name": ctx.user.display_name,
},
"org": {
"uuid": str(ctx.org.uuid),
"display_name": ctx.org.display_name,
},
"role": {
"uuid": str(ctx.role.uuid),
"display_name": ctx.role.display_name,
},
"user": {"uuid": ctx.user.uuid, "display_name": ctx.user.display_name},
"org": {"uuid": ctx.org.uuid, "display_name": ctx.org.display_name},
"role": {"uuid": ctx.role.uuid, "display_name": ctx.role.display_name},
"permissions": [p.scope for p in ctx.permissions],
}
async def format_user_info(
async def build_user_info(
*,
user_uuid,
auth: str,
session_record,
request_host: str | None,
) -> dict:
"""Format complete user information for authenticated users."""
"""Build user info dict for authenticated users."""
ctx = await permutil.session_context(auth, request_host)
# Fetch and format credentials
user = db.data().users[user_uuid]
user_credentials = user.credentials
credentials: list[dict] = []
user_aaguids: set[str] = set()
normalized_host = hostutil.normalize_host(request_host)
for c in user_credentials:
aaguid_str = str(c.aaguid)
user_aaguids.add(aaguid_str)
credentials.append(
credentials = sorted(user.credentials, key=lambda c: c.created_at)
return {
"ctx": build_session_context(ctx),
"created_at": ctx.user.created_at,
"last_seen": ctx.user.last_seen,
"visits": ctx.user.visits,
"credentials": [
{
"credential": str(c.uuid),
"aaguid": aaguid_str,
"created_at": _format_datetime(c.created_at),
"last_used": _format_datetime(c.last_used),
"last_verified": _format_datetime(c.last_verified),
"credential": c.uuid,
"aaguid": c.aaguid,
"created_at": c.created_at,
"last_used": c.last_used,
"last_verified": c.last_verified,
"sign_count": c.sign_count,
"is_current_session": session_record.credential == c.uuid,
}
)
credentials.sort(key=lambda cred: cred["created_at"])
aaguid_info = aaguid.filter(user_aaguids)
# Format sessions
normalized_request_host = hostutil.normalize_host(request_host)
session_records = user.sessions
current_session_key = auth
sessions_payload: list[dict] = []
for entry in session_records:
sessions_payload.append(
{
"id": entry.key,
"credential": str(entry.credential),
"host": entry.host,
"ip": entry.ip,
"user_agent": useragent.compact_user_agent(entry.user_agent),
"last_renewed": _format_datetime(entry.expiry - EXPIRES),
"is_current": entry.key == current_session_key,
"is_current_host": bool(
normalized_request_host
and entry.host
and entry.host == normalized_request_host
),
}
)
return {
"ctx": format_session_context(ctx),
"created_at": _format_datetime(ctx.user.created_at),
"last_seen": _format_datetime(ctx.user.last_seen),
"visits": ctx.user.visits,
"credentials": credentials,
"aaguid_info": aaguid_info,
"sessions": sessions_payload,
for c in credentials
],
"aaguid_info": aaguid.filter(c.aaguid for c in credentials),
"sessions": [
ApiSession.from_db(
s,
current_key=auth,
normalized_host=normalized_host,
expires_delta=EXPIRES,
)
for s in user.sessions
],
}