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

API cleanup: msgspec structs for all JSON responses, aaguid icon normalization, User kw_only, migration v4.

Browse Source
This commit is contained in:
Leo Vasanko
2026-02-17 20:16:55 +00:00
parent eafbdce4f8
commit aae4117a38
11 changed files with 298 additions and 189 deletions
Download Patch File Download Diff File Expand all files Collapse all files
paskia/aaguid/__init__.py
+13 -1
View File
@@ -30,4 +30,16 @@ def filter(aaguids: Iterable[UUID]) -> dict[str, dict]:
Dictionary mapping AAGUID string to authenticator information for only
the AAGUIDs that the user has and that we have data for
"""
return {(s := str(a)): AAGUID[s] for a in aaguids if (s := str(a)) in AAGUID}
result = {}
for a in aaguids:
s = str(a)
if s in AAGUID:
info = AAGUID[s].copy()
# Rename icon_light to icon
if "icon_light" in info:
info["icon"] = info.pop("icon_light")
# If icons are the same, set dark to None to save space
if info.get("icon") == info.get("icon_dark"):
info["icon_dark"] = None
result[s] = info
return result
paskia/db/__init__.py
+1 -1
View File
@@ -48,6 +48,7 @@ from paskia.db.operations import (
delete_session,
delete_sessions_for_user,
delete_user,
is_username_taken,
login,
oidc_login,
remove_permission_from_org,
@@ -64,7 +65,6 @@ from paskia.db.operations import (
update_user_display_name,
update_user_info,
update_user_role,
is_username_taken,
)
from paskia.db.structs import (
DB,
paskia/db/bootstrap.py
+1
View File
@@ -102,6 +102,7 @@ def bootstrap(
created_at=now,
last_seen=None,
visits=0,
theme="",
)
admin_user.uuid = user_uuid
admin_user.store()
paskia/db/migrations.py
+6
View File
@@ -24,6 +24,12 @@ def migrate_v2(d: dict, *, rp_id: str = "localhost") -> None:
def migrate_v3(d: dict, **kwargs) -> None:
"""Ensure all users have visits field."""
for user_data in d["users"].values():
user_data.setdefault("visits", 0)
def migrate_v4(d: dict, **kwargs) -> None:
"""OpenID Connect support and hardened session keys."""
d["oid_clients"] = {}
d["sessions"] = {
paskia/db/structs.py
+6 -3
View File
@@ -195,7 +195,7 @@ class Role(msgspec.Struct, dict=True, omit_defaults=True):
return role
class User(msgspec.Struct, dict=True, omit_defaults=True):
class User(msgspec.Struct, dict=True, omit_defaults=True, kw_only=True):
"""User data structure.
Mutable fields: display_name, role_uuid, last_seen, visits, theme, email, preferred_username
@@ -206,9 +206,9 @@ class User(msgspec.Struct, dict=True, omit_defaults=True):
display_name: str
role_uuid: UUID = msgspec.field(name="role")
created_at: datetime
visits: int
last_seen: datetime | None = None
visits: int = 0
theme: str = "" # "" or "auto" = OS default, "light", "dark"
theme: str = ""
email: str | None = None # OIDC email claim
preferred_username: str | None = None # OIDC preferred_username claim
telephone: str | None = None # Telephone number
@@ -278,6 +278,9 @@ class User(msgspec.Struct, dict=True, omit_defaults=True):
display_name=display_name,
role_uuid=role_uuid,
created_at=created_at or datetime.now(UTC),
last_seen=None,
visits=0,
theme="",
)
user.uuid = uuid7.create(user.created_at)
return user
paskia/fastapi/admin.py
+46 -66
View File
@@ -24,7 +24,16 @@ from paskia.util import (
querysafe,
vitedev,
)
from paskia.util.apistructs import ApiPermission, ApiSession, format_datetime
from paskia.util.apistructs import (
ApiAaguidInfo,
ApiCreateLinkResponse,
ApiOrgResponse,
ApiPermission,
ApiUser,
ApiUserDetail,
ApiUserSession,
ApiUuidResponse,
)
from paskia.util.hostutil import normalize_host
app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
@@ -86,34 +95,15 @@ async def admin_list_orgs(request: Request, auth=AUTH_COOKIE):
orgs = [o for o in orgs if o.uuid == ctx.org.uuid]
def org_to_dict(o):
return {
"uuid": o.uuid,
"display_name": o.display_name,
"permissions": {p.uuid for p in o.permissions},
"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": u.uuid,
"display_name": u.display_name,
"role": r.display_name,
"role_uuid": u.role_uuid,
"visits": u.visits,
"last_seen": u.last_seen,
}
for r in o.roles
for u in r.users
],
}
roles = o.roles
return ApiOrgResponse(
org=o,
permissions={p.uuid: p for p in o.permissions},
roles={r.uuid: r for r in roles},
users={u.uuid: u for r in roles for u in r.users},
)
return MsgspecResponse([org_to_dict(o) for o in orgs])
return MsgspecResponse({o.uuid: org_to_dict(o) for o in orgs})
@app.post("/orgs")
@@ -132,7 +122,7 @@ async def admin_create_org(
for perm in permissions:
db.add_permission_to_org(str(org.uuid), perm, ctx=ctx)
return {"uuid": str(org.uuid)}
return MsgspecResponse(ApiUuidResponse(uuid=str(org.uuid)))
@app.patch("/orgs/{org_uuid}")
@@ -281,7 +271,7 @@ async def admin_create_role(
permissions=permission_uuids,
)
db.create_role(role, ctx=ctx)
return {"uuid": str(role.uuid)}
return MsgspecResponse(ApiUuidResponse(uuid=str(role.uuid)))
@app.patch("/roles/{role_uuid}")
@@ -454,7 +444,7 @@ async def admin_create_user(
role=role_obj.uuid,
)
db.create_user(user, ctx=ctx)
return {"uuid": str(user.uuid)}
return MsgspecResponse(ApiUuidResponse(uuid=str(user.uuid)))
@app.patch("/users/{user_uuid}/role")
@@ -541,11 +531,13 @@ async def admin_create_user_registration_link(
ctx=ctx,
)
url = hostutil.reset_link_url(token)
return {
"url": url,
"expires": format_datetime(expiry),
"token_type": token_type,
}
return MsgspecResponse(
ApiCreateLinkResponse(
url=url,
expires=expiry,
token_type=token_type,
)
)
@app.get("/users/{user_uuid}")
@@ -556,7 +548,6 @@ async def admin_get_user_detail(
):
try:
user = db.data().users[user_uuid]
role_name = user.role.display_name
except KeyError:
raise HTTPException(status_code=404, detail="User not found")
ctx = await authz.verify(
@@ -571,39 +562,28 @@ async def admin_get_user_detail(
)
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,
"email": user.email,
"preferred_username": user.preferred_username,
"telephone": user.telephone,
"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(
sessions = [
ApiUserSession.from_db(
s,
current_key=ctx.session.key,
current_key=auth,
normalized_host=normalized_host,
expires_delta=EXPIRES,
)
for s in user.sessions
],
}
]
return MsgspecResponse(
ApiUserDetail(
user=ApiUser.from_db(user),
credentials={c.uuid: c for c in user.credentials},
aaguid_info={
k: ApiAaguidInfo(**v)
for k, v in aaguid_mod.filter(
c.aaguid for c in user.credentials
).items()
},
sessions=sessions,
)
)
@@ -852,7 +832,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 MsgspecResponse([ApiPermission.from_db(p) for p in perms])
return MsgspecResponse({p.uuid: ApiPermission.from_db(p) for p in perms})
@app.post("/permissions")
paskia/fastapi/api.py
+24 -18
View File
@@ -21,6 +21,7 @@ from paskia.fastapi.response import MsgspecResponse
from paskia.fastapi.session import AUTH_COOKIE, AUTH_COOKIE_NAME, get_client_ip
from paskia.globals import passkey as global_passkey
from paskia.util import hostutil, htmlutil, passphrase, userinfo, vitedev
from paskia.util.apistructs import ApiSettings, ApiTokenInfo, ApiValidateResponse
bearer_auth = HTTPBearer(auto_error=False)
@@ -100,11 +101,11 @@ async def validate_token(
session.set_session_cookie(response, auth)
renewed = True
return MsgspecResponse(
{
"valid": True,
"renewed": renewed,
"ctx": userinfo.build_session_context(ctx),
}
ApiValidateResponse(
valid=True,
renewed=renewed,
ctx=userinfo.build_session_context(ctx),
)
)
@@ -189,15 +190,17 @@ async def forward_authentication(
async def get_settings():
pk = global_passkey.instance
base_path = hostutil.ui_base_path()
return {
"rp_id": pk.rp_id,
"rp_name": pk.rp_name,
"ui_base_path": base_path,
"auth_host": hostutil.dedicated_auth_host(),
"auth_site_url": hostutil.auth_site_url(),
"session_cookie": AUTH_COOKIE_NAME,
"version": __version__,
}
return MsgspecResponse(
ApiSettings(
rp_id=pk.rp_id,
rp_name=pk.rp_name,
ui_base_path=base_path,
auth_host=hostutil.dedicated_auth_host(),
auth_site_url=hostutil.auth_site_url(),
session_cookie=AUTH_COOKIE_NAME,
version=__version__,
)
)
@app.post("/user-info")
@@ -223,6 +226,7 @@ async def api_user_info(
auth=auth,
session_record=ctx.session,
request_host=request.headers.get("host"),
ctx=ctx,
)
)
@@ -241,10 +245,12 @@ async def token_info(credentials=Depends(bearer_auth)):
raise HTTPException(401, str(e))
u = reset_token.user
return {
"token_type": reset_token.token_type,
"display_name": u.display_name,
}
return MsgspecResponse(
ApiTokenInfo(
token_type=reset_token.token_type,
display_name=u.display_name,
)
)
@app.post("/logout")
paskia/fastapi/mainapp.py
+1 -1
View File
@@ -9,10 +9,10 @@ from fastapi.responses import FileResponse, RedirectResponse
from fastapi_vue import Frontend
from paskia import authcode, globals
from paskia.__main__ import DEVMODE
from paskia.db import start_background, stop_background
from paskia.db.logging import configure_db_logging
from paskia.fastapi import admin, api, auth_host, oid, ws
from paskia.__main__ import DEVMODE
from paskia.fastapi.logging import AccessLogMiddleware, configure_access_logging
from paskia.fastapi.session import AUTH_COOKIE
from paskia.util import hostutil, passphrase, vitedev
paskia/fastapi/user.py
+10 -11
View File
@@ -1,4 +1,3 @@
from datetime import UTC
from uuid import UUID
import base64url
@@ -17,8 +16,10 @@ from paskia.authsession import (
expires,
)
from paskia.fastapi import authz, session
from paskia.fastapi.response import MsgspecResponse
from paskia.fastapi.session import AUTH_COOKIE
from paskia.util import hostutil
from paskia.util.apistructs import ApiCreateLinkResponse
app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
@@ -207,13 +208,11 @@ async def api_create_link(
ctx=ctx,
)
url = hostutil.reset_link_url(token)
return {
"message": "Registration link generated successfully",
"url": url,
"expires": (
expiry.astimezone(UTC).isoformat().replace("+00:00", "Z")
if expiry.tzinfo
else expiry.replace(tzinfo=UTC).isoformat().replace("+00:00", "Z")
),
"token_type": "device addition",
}
return MsgspecResponse(
ApiCreateLinkResponse(
message="Registration link generated successfully",
url=url,
expires=expiry,
token_type="device addition",
)
)
paskia/util/apistructs.py
+131 -33
View File
@@ -1,36 +1,20 @@
from __future__ import annotations
"""API response utilities using msgspec for JSON serialization.
msgspec handles UUID and datetime conversion automatically.
API structs inherit from db structs with kw_only=True to add uuid/key fields.
"""
from datetime import UTC, datetime
from datetime import datetime
from uuid import UUID
import msgspec
from paskia import db
from paskia.db.structs import Org, Permission, Role, User
from paskia.db.structs import Credential, Org, Permission, Role, User
from paskia.util import useragent
def _utc_datetime(dt: datetime | None) -> datetime | None:
"""Convert datetime to UTC, handling both aware and naive datetimes."""
if dt is None:
return None
if dt.tzinfo:
return dt.astimezone(UTC)
return dt.replace(tzinfo=UTC)
def format_datetime(dt: datetime | None) -> str | None:
"""Format a datetime to ISO 8601 string with Z suffix for UTC."""
if dt is None:
return None
utc_dt = _utc_datetime(dt)
return utc_dt.isoformat().replace("+00:00", "Z") if utc_dt else None
# -------------------------------------------------------------------------
# API structs - inherit from db structs, add uuid for serialization
# -------------------------------------------------------------------------
@@ -42,7 +26,7 @@ class ApiUser(User, kw_only=True):
uuid: UUID
@classmethod
def from_db(cls, u: User) -> "ApiUser":
def from_db(cls, u: User) -> ApiUser:
return cls(uuid=u.uuid, **msgspec.structs.asdict(u))
@@ -52,7 +36,7 @@ class ApiOrg(Org, kw_only=True):
uuid: UUID
@classmethod
def from_db(cls, o: Org) -> "ApiOrg":
def from_db(cls, o: Org) -> ApiOrg:
return cls(uuid=o.uuid, **msgspec.structs.asdict(o))
@@ -62,28 +46,42 @@ class ApiRole(Role, kw_only=True):
uuid: UUID
@classmethod
def from_db(cls, r: Role) -> "ApiRole":
def from_db(cls, r: Role) -> ApiRole:
return cls(uuid=r.uuid, **msgspec.structs.asdict(r))
class ApiPermission(Permission, kw_only=True):
"""Permission with uuid serialized."""
class ApiPermission(msgspec.Struct, kw_only=True):
"""Permission for API responses, without org details."""
uuid: UUID
scope: str
display_name: str
domain: str | None = None
@classmethod
def from_db(cls, p: Permission) -> "ApiPermission":
return cls(uuid=p.uuid, **msgspec.structs.asdict(p))
def from_db(cls, p: Permission) -> ApiPermission:
return cls(
scope=p.scope,
display_name=p.display_name,
domain=p.domain,
)
class ApiSession(msgspec.Struct, omit_defaults=True):
"""Session for API responses with computed fields."""
class ApiAaguidInfo(msgspec.Struct, kw_only=True, omit_defaults=True):
"""AAGUID information for authenticators."""
name: str
icon: str | None = None
icon_dark: str | None = None
class ApiUserSession(msgspec.Struct, omit_defaults=True):
"""Session for user info responses with computed fields."""
id: str
credential_uuid: UUID = msgspec.field(name="credential")
host: str
ip: str
user_agent: str
expiry: datetime
last_renewed: datetime
is_current: bool = False
is_current_host: bool = False
@@ -98,17 +96,17 @@ class ApiSession(msgspec.Struct, omit_defaults=True):
current_key: str,
normalized_host: str | None,
expires_delta, # timedelta
) -> "ApiSession":
) -> ApiUserSession:
client_name = None
if s.client_uuid:
c = db.data().oid_clients.get(s.client_uuid)
client_name = c.name if c else str(s.client_uuid)
return cls(
id=s.key,
credential_uuid=s.credential_uuid,
host=s.host,
ip=s.ip,
user_agent=useragent.compact_user_agent(s.user_agent),
expiry=s.expiry,
last_renewed=s.expiry - expires_delta,
is_current=s.key == current_key,
is_current_host=not s.client_uuid
@@ -116,3 +114,103 @@ class ApiSession(msgspec.Struct, omit_defaults=True):
client_uuid=s.client_uuid,
client_name=client_name,
)
class ApiUserDetail(msgspec.Struct, kw_only=True):
"""User detail response with credentials and sessions."""
user: ApiUser
credentials: dict[UUID, Credential]
aaguid_info: dict[str, ApiAaguidInfo]
sessions: list[ApiUserSession]
permissions: dict[UUID, ApiPermission] = {}
# -------------------------------------------------------------------------
# Nested API structs for org response - without uuid
# -------------------------------------------------------------------------
class ApiOrgResponse(msgspec.Struct, kw_only=True):
"""Org response containing Org with roles and users as UUID-keyed dicts."""
org: Org
permissions: dict[UUID, Permission]
roles: dict[UUID, Role]
users: dict[UUID, User]
class ApiSettings(msgspec.Struct):
"""Settings response struct."""
rp_id: str
rp_name: str
ui_base_path: str
auth_host: str | None
auth_site_url: str
session_cookie: str
version: str
class ApiTokenInfo(msgspec.Struct):
"""Token info response struct."""
token_type: str
display_name: str
class ApiUuidResponse(msgspec.Struct):
"""Response struct for creation endpoints returning a UUID."""
uuid: str
class ApiCreateLinkResponse(msgspec.Struct):
"""Response struct for create-link endpoints."""
url: str
expires: datetime
token_type: str
message: str | None = None
class ApiUserContext(msgspec.Struct, omit_defaults=True):
"""User context for session validation."""
uuid: UUID
display_name: str
theme: str = ""
email: str | None = None
preferred_username: str | None = None
telephone: str | None = None
class ApiOrgContext(msgspec.Struct):
"""Org context for session validation."""
uuid: UUID
display_name: str
class ApiRoleContext(msgspec.Struct):
"""Role context for session validation."""
uuid: UUID
display_name: str
class ApiSessionContext(msgspec.Struct):
"""Session context struct."""
user: ApiUserContext
org: ApiOrgContext
role: ApiRoleContext
permissions: list[str]
class ApiValidateResponse(msgspec.Struct):
"""Response struct for validate endpoint."""
valid: bool
renewed: bool
ctx: ApiSessionContext
paskia/util/userinfo.py
+50 -46
View File
@@ -3,27 +3,38 @@
from paskia import aaguid, db
from paskia.authsession import EXPIRES
from paskia.db import SessionContext
from paskia.util import hostutil, permutil
from paskia.util.apistructs import ApiSession
from paskia.util import hostutil
from paskia.util.apistructs import (
ApiAaguidInfo,
ApiOrgContext,
ApiPermission,
ApiRoleContext,
ApiSessionContext,
ApiUser,
ApiUserContext,
ApiUserDetail,
ApiUserSession,
)
def build_session_context(ctx: SessionContext) -> dict:
"""Build session context dict from SessionContext."""
result = {
"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],
}
if ctx.user.theme:
result["user"]["theme"] = ctx.user.theme
if ctx.user.email:
result["user"]["email"] = ctx.user.email
if ctx.user.preferred_username:
result["user"]["preferred_username"] = ctx.user.preferred_username
if ctx.user.telephone:
result["user"]["telephone"] = ctx.user.telephone
return result
def build_session_context(ctx: SessionContext) -> ApiSessionContext:
"""Build session context struct from SessionContext."""
user = ApiUserContext(
uuid=ctx.user.uuid,
display_name=ctx.user.display_name,
theme=ctx.user.theme,
email=ctx.user.email,
preferred_username=ctx.user.preferred_username,
telephone=ctx.user.telephone,
)
org = ApiOrgContext(uuid=ctx.org.uuid, display_name=ctx.org.display_name)
role = ApiRoleContext(uuid=ctx.role.uuid, display_name=ctx.role.display_name)
return ApiSessionContext(
user=user,
org=org,
role=role,
permissions=[p.scope for p in ctx.permissions],
)
async def build_user_info(
@@ -32,38 +43,31 @@ async def build_user_info(
auth: str,
session_record,
request_host: str | None,
) -> dict:
"""Build user info dict for authenticated users."""
ctx = await permutil.session_context(auth, request_host)
ctx: SessionContext | None = None,
) -> ApiUserDetail:
"""Build user info struct for authenticated users."""
user = db.data().users[user_uuid]
normalized_host = hostutil.normalize_host(request_host)
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": 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,
}
for c in credentials
],
"aaguid_info": aaguid.filter(c.aaguid for c in credentials),
"sessions": [
ApiSession.from_db(
sessions = [
ApiUserSession.from_db(
s,
current_key=ctx.session.key,
current_key=session_record.key,
normalized_host=normalized_host,
expires_delta=EXPIRES,
)
for s in user.sessions
],
}
]
return ApiUserDetail(
user=ApiUser.from_db(user),
credentials={c.uuid: c for c in user.credentials},
aaguid_info={
k: ApiAaguidInfo(**v)
for k, v in aaguid.filter(c.aaguid for c in user.credentials).items()
},
sessions=sessions,
permissions={p.uuid: ApiPermission.from_db(p) for p in ctx.permissions}
if ctx
else {},
)