Refactor to separate admin app modules to subapps, required trailing slashes and plural changes on some of the URLs.

docs/API.md

@@ -40,6 +40,38 @@ Normally only used via admin panel, requires auth admin permissions and can modi
 
 E.g. Org admin cannot see anything of the other orgs that he has no admin access to. Master admin `auth:admin` can see everything and create and manage orgs.
 
+| Method | Path | Used for | Notes |
+|---:|---|---|---|
+| GET | `/auth/api/admin/info` | Admin overview | Returns orgs, permissions, OIDC clients info |
+| POST | `/auth/api/admin/permissions/` | Create permission | Body: JSON with scope, display_name, domain |
+| PATCH | `/auth/api/admin/permissions/{uuid}` | Update permission | Query params: display_name, scope, domain |
+| DELETE | `/auth/api/admin/permissions/{uuid}` | Delete permission | |
+| POST | `/auth/api/admin/orgs/` | Create organization | Body: JSON with display_name, permissions |
+| GET | `/auth/api/admin/orgs/{uuid}` | Get organization details | |
+| PATCH | `/auth/api/admin/orgs/{uuid}` | Update organization | Body: JSON with display_name |
+| DELETE | `/auth/api/admin/orgs/{uuid}` | Delete organization | |
+| POST | `/auth/api/admin/orgs/{uuid}/users` | Create user in org | Body: JSON with display_name, role_uuid |
+| POST | `/auth/api/admin/orgs/{uuid}/roles` | Create role in org | Body: JSON with display_name, permissions |
+| POST | `/auth/api/admin/orgs/{uuid}/permission` | Grant permission to org | Query param: permission_uuid |
+| DELETE | `/auth/api/admin/orgs/{uuid}/permission` | Revoke permission from org | Query param: permission_uuid |
+| PATCH | `/auth/api/admin/roles/{uuid}` | Update role | Body: JSON with display_name |
+| POST | `/auth/api/admin/roles/{uuid}/permissions/{uuid}` | Add permission to role | |
+| DELETE | `/auth/api/admin/roles/{uuid}/permissions/{uuid}` | Remove permission from role | |
+| DELETE | `/auth/api/admin/roles/{uuid}` | Delete role | |
+| PATCH | `/auth/api/admin/users/{uuid}/role` | Update user role | Body: JSON with role_uuid |
+| PATCH | `/auth/api/admin/users/{uuid}/info` | Update user info | Body: JSON with display_name |
+| GET | `/auth/api/admin/users/{uuid}` | Get user details | |
+| DELETE | `/auth/api/admin/users/{uuid}` | Delete user | |
+| POST | `/auth/api/admin/users/{uuid}/create-link` | Create device add link | |
+| DELETE | `/auth/api/admin/users/{uuid}/credentials/{uuid}` | Delete user credential | |
+| DELETE | `/auth/api/admin/users/{uuid}/sessions/{key}` | Delete user session | |
+| POST | `/auth/api/admin/oidc-clients/` | Create OIDC client | Body: JSON with client_name, redirect_uris |
+| PATCH | `/auth/api/admin/oidc-clients/{uuid}` | Update OIDC client | Body: JSON with client_name, redirect_uris |
+| PATCH | `/auth/api/admin/oidc-clients/{uuid}/reset-secret` | Reset client secret | |
+| DELETE | `/auth/api/admin/oidc-clients/{uuid}` | Delete OIDC client | |
+| GET | `/auth/api/admin/server-config/` | Get server config | Returns rp_name, auth_host, origins |
+| PATCH | `/auth/api/admin/server-config/` | Update server config | Body: JSON with rp_name, auth_host, origins |
+
 ### WebSockets: `/auth/ws/*`
 
 | Path | Used for | Notes |

paskia/fastapi/admin/__init__.py

@@ -0,0 +1,3 @@
+from paskia.fastapi.admin.adminapp import app
+
+__all__ = ["app"]

paskia/fastapi/admin/adminapp.py

@@ -0,0 +1,116 @@
+from uuid import UUID
+
+from fastapi import FastAPI, Request
+
+from paskia import db
+from paskia.fastapi import authz
+from paskia.fastapi.admin import (
+    oidc_clients,
+    orgs,
+    permissions,
+    roles,
+    server_config,
+    users,
+)
+from paskia.fastapi.admin.errors import install_error_handlers
+from paskia.fastapi.front import frontend
+from paskia.fastapi.response import MsgspecResponse
+from paskia.fastapi.session import AUTH_COOKIE
+from paskia.util import (
+    permutil,
+    vitedev,
+)
+from paskia.util.apistructs import (
+    ApiAdminInfo,
+    ApiOidcClient,
+    ApiOrg,
+    ApiOrgResponse,
+    ApiPermission,
+)
+
+app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
+
+install_error_handlers(app)
+app.mount("/oidc-clients", oidc_clients.app)
+app.mount("/orgs", orgs.app)
+app.mount("/roles", roles.app)
+app.mount("/users", users.app)
+app.mount("/permissions", permissions.app)
+app.mount("/server-config", server_config.app)
+
+
+def master_admin(ctx) -> bool:
+    return any(p.scope == "auth:admin" for p in ctx.permissions)
+
+
+def org_admin(ctx, org_uuid: UUID) -> bool:
+    return ctx.org.uuid == org_uuid and any(
+        p.scope == "auth:org:admin" for p in ctx.permissions
+    )
+
+
+def can_manage_org(ctx, org_uuid: UUID) -> bool:
+    return master_admin(ctx) or org_admin(ctx, org_uuid)
+
+
+@app.get("/")
+async def adminapp(request: Request, auth=AUTH_COOKIE):
+    return await vitedev.handle(request, frontend, "/auth/admin/")
+
+
+@app.get("/info")
+async def admin_info(request: Request, auth=AUTH_COOKIE):
+    ctx = await authz.verify(
+        auth,
+        ["auth:admin", "auth:org:admin"],
+        match=permutil.has_any,
+        host=request.headers.get("host"),
+    )
+
+    # Orgs
+    orgs = list(db.data().orgs.values())
+    if not master_admin(ctx):
+        # Org admins can only see their own organization
+        orgs = [o for o in orgs if o.uuid == ctx.org.uuid]
+
+    def org_to_dict(o):
+        roles = o.roles
+        return ApiOrgResponse(
+            org=ApiOrg.from_db(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},
+        )
+
+    orgs_dict = {o.uuid: org_to_dict(o) for o in orgs}
+
+    # Permissions
+    perms = db.data().permissions.values() if master_admin(ctx) else ctx.org.permissions
+    perms_dict = {p.uuid: ApiPermission.from_db(p) for p in perms}
+
+    # OIDC Clients (master admin only)
+    oidc_clients_dict = {}
+    if master_admin(ctx):
+        clients = sorted(db.data().oidc.clients.values(), key=lambda c: c.uuid)
+        sessions = db.data().sessions
+        # Count active sessions per client
+        client_session_counts = {}
+        for session in sessions.values():
+            if session.client_uuid:
+                client_session_counts[session.client_uuid] = (
+                    client_session_counts.get(session.client_uuid, 0) + 1
+                )
+        oidc_clients_dict = {
+            client.uuid: ApiOidcClient.from_db(
+                client, client_session_counts.get(client.uuid, 0)
+            )
+            for client in clients
+        }
+
+    return MsgspecResponse(
+        ApiAdminInfo(
+            orgs=orgs_dict,
+            permissions=perms_dict,
+            oidc_clients=oidc_clients_dict,
+        )
+    )

paskia/fastapi/admin/errors.py

@@ -0,0 +1,30 @@
+"""Shared exception handlers for admin sub-apps."""
+
+import logging
+
+from fastapi import FastAPI
+from fastapi.responses import JSONResponse
+
+from paskia.fastapi import authz
+
+
+def install_error_handlers(app: FastAPI) -> None:
+    """Register standard exception handlers on *app*."""
+
+    @app.exception_handler(ValueError)
+    async def value_error_handler(_request, exc: ValueError):
+        return JSONResponse(status_code=400, content={"detail": str(exc)})
+
+    @app.exception_handler(authz.AuthException)
+    async def auth_exception_handler(_request, exc: authz.AuthException):
+        return JSONResponse(
+            status_code=exc.status_code,
+            content=await authz.auth_error_content(exc),
+        )
+
+    @app.exception_handler(Exception)
+    async def general_exception_handler(_request, exc: Exception):  # pragma: no cover
+        logging.exception("Unhandled exception in admin app")
+        return JSONResponse(
+            status_code=500, content={"detail": "Internal server error"}
+        )

paskia/fastapi/admin/oidc_clients.py

@@ -0,0 +1,237 @@
+from uuid import UUID
+
+from fastapi import Body, FastAPI, HTTPException, Request
+
+from paskia import db
+from paskia.db.operations import _UNSET
+from paskia.db.structs import Client
+from paskia.fastapi import authz
+from paskia.fastapi.admin.errors import install_error_handlers
+from paskia.fastapi.session import AUTH_COOKIE
+from paskia.util import permutil
+
+app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
+
+install_error_handlers(app)
+
+
+def master_admin(ctx) -> bool:
+    return any(p.scope == "auth:admin" for p in ctx.permissions)
+
+
+@app.post("/")
+async def admin_create_oidc_client(
+    request: Request,
+    payload: dict = Body(...),
+    auth=AUTH_COOKIE,
+):
+    """Create a new OIDC client (master admin only)."""
+    ctx = await authz.verify(
+        auth,
+        ["auth:admin"],
+        host=request.headers.get("host"),
+        match=permutil.has_all,
+        max_age="5m",
+    )
+    if not master_admin(ctx):
+        raise authz.AuthException(
+            status_code=403,
+            detail="Only master admin can manage OIDC clients",
+            mode="forbidden",
+        )
+
+    # Client ID and secret hash are generated client-side
+    client_id = payload.get("client_id", "").strip()
+    secret_hash_hex = payload.get("secret_hash", "").strip()
+    name = payload.get("name", "").strip()
+    redirect_uris = payload.get("redirect_uris", [])
+    backchannel_logout_uri = payload.get("backchannel_logout_uri")
+    if isinstance(backchannel_logout_uri, str):
+        backchannel_logout_uri = backchannel_logout_uri.strip() or None
+
+    if not client_id or not secret_hash_hex:
+        raise ValueError("client_id and secret_hash are required")
+
+    try:
+        client_uuid = UUID(client_id)
+    except (ValueError, AttributeError):
+        raise ValueError("client_id must be a valid UUID")
+
+    try:
+        secret_hash = bytes.fromhex(secret_hash_hex)
+    except ValueError:
+        raise ValueError("secret_hash must be a hex-encoded SHA-256 hash")
+    if len(secret_hash) != 32:
+        raise ValueError("secret_hash must be a SHA-256 hash (32 bytes)")
+
+    if not isinstance(redirect_uris, list):
+        raise ValueError("redirect_uris must be a list")
+
+    # Validate redirect URIs
+    for uri in redirect_uris:
+        if not isinstance(uri, str) or not uri.startswith("http"):
+            raise ValueError(f"Invalid redirect URI: {uri}")
+
+    if backchannel_logout_uri and not backchannel_logout_uri.startswith("http"):
+        raise ValueError("backchannel_logout_uri must be an HTTP(S) URL")
+
+    client = Client(
+        client_secret_hash=secret_hash,
+        name=name,
+        redirect_uris=redirect_uris,
+        backchannel_logout_uri=backchannel_logout_uri,
+    )
+    client.uuid = client_uuid
+
+    db.create_oid_client(client, ctx=ctx)
+
+    return {"status": "ok", "client_id": str(client.uuid)}
+
+
+@app.patch("/{client_uuid}")
+async def admin_update_oidc_client(
+    client_uuid: UUID,
+    request: Request,
+    payload: dict = Body(...),
+    auth=AUTH_COOKIE,
+):
+    """Update an OIDC client's name and redirect URIs (master admin only)."""
+    ctx = await authz.verify(
+        auth,
+        ["auth:admin"],
+        host=request.headers.get("host"),
+        match=permutil.has_all,
+        max_age="5m",
+    )
+    if not master_admin(ctx):
+        raise authz.AuthException(
+            status_code=403,
+            detail="Only master admin can manage OIDC clients",
+            mode="forbidden",
+        )
+
+    name = payload.get("name", "").strip() if "name" in payload else None
+    redirect_uris = payload.get("redirect_uris") if "redirect_uris" in payload else None
+    secret_hash_hex = (
+        payload.get("secret_hash", "").strip() if "secret_hash" in payload else None
+    )
+    backchannel_logout_uri = (
+        payload.get("backchannel_logout_uri")
+        if "backchannel_logout_uri" in payload
+        else _UNSET
+    )
+    if isinstance(backchannel_logout_uri, str):
+        backchannel_logout_uri = backchannel_logout_uri.strip() or None
+
+    if name is not None and not name:
+        raise ValueError("Client name cannot be empty")
+
+    if redirect_uris is not None:
+        if not isinstance(redirect_uris, list):
+            raise ValueError("redirect_uris must be a list")
+        # Validate redirect URIs
+        for uri in redirect_uris:
+            if not isinstance(uri, str) or not uri.startswith("http"):
+                raise ValueError(f"Invalid redirect URI: {uri}")
+
+    if (
+        backchannel_logout_uri is not _UNSET
+        and backchannel_logout_uri
+        and not backchannel_logout_uri.startswith("http")
+    ):
+        raise ValueError("backchannel_logout_uri must be an HTTP(S) URL")
+
+    secret_hash = None
+    if secret_hash_hex:
+        try:
+            secret_hash = bytes.fromhex(secret_hash_hex)
+        except ValueError:
+            raise ValueError("secret_hash must be a hex-encoded SHA-256 hash")
+        if len(secret_hash) != 32:
+            raise ValueError("secret_hash must be a SHA-256 hash (32 bytes)")
+
+    try:
+        db.update_oid_client(
+            client_uuid,
+            name=name,
+            redirect_uris=redirect_uris,
+            secret_hash=secret_hash,
+            backchannel_logout_uri=backchannel_logout_uri,
+            ctx=ctx,
+        )
+    except ValueError as e:
+        raise HTTPException(status_code=404, detail=str(e))
+
+    return {"status": "ok"}
+
+
+@app.post("/{client_uuid}/reset-secret")
+async def admin_reset_oidc_client_secret(
+    client_uuid: UUID,
+    request: Request,
+    payload: dict = Body(...),
+    auth=AUTH_COOKIE,
+):
+    """Reset an OIDC client's secret (master admin only).
+
+    The new secret is generated client-side; only the SHA-256 hash is sent.
+    """
+    ctx = await authz.verify(
+        auth,
+        ["auth:admin"],
+        host=request.headers.get("host"),
+        match=permutil.has_all,
+        max_age="5m",
+    )
+    if not master_admin(ctx):
+        raise authz.AuthException(
+            status_code=403,
+            detail="Only master admin can manage OIDC clients",
+            mode="forbidden",
+        )
+
+    secret_hash_hex = payload.get("secret_hash", "").strip()
+    if not secret_hash_hex:
+        raise ValueError("secret_hash is required")
+    try:
+        secret_hash = bytes.fromhex(secret_hash_hex)
+    except ValueError:
+        raise ValueError("secret_hash must be a hex-encoded SHA-256 hash")
+    if len(secret_hash) != 32:
+        raise ValueError("secret_hash must be a SHA-256 hash (32 bytes)")
+
+    try:
+        db.reset_oid_client_secret(client_uuid, secret_hash, ctx=ctx)
+    except ValueError as e:
+        raise HTTPException(status_code=404, detail=str(e))
+
+    return {"status": "ok"}
+
+
+@app.delete("/{client_uuid}")
+async def admin_delete_oidc_client(
+    client_uuid: UUID,
+    request: Request,
+    auth=AUTH_COOKIE,
+):
+    """Delete an OIDC client (master admin only)."""
+    ctx = await authz.verify(
+        auth,
+        ["auth:admin"],
+        host=request.headers.get("host"),
+        match=permutil.has_all,
+        max_age="5m",
+    )
+    if not master_admin(ctx):
+        raise authz.AuthException(
+            status_code=403,
+            detail="Only master admin can manage OIDC clients",
+            mode="forbidden",
+        )
+
+    try:
+        db.delete_oid_client(client_uuid, ctx=ctx)
+    except ValueError as e:
+        raise HTTPException(status_code=404, detail=str(e))
+
+    return {"status": "ok"}

paskia/fastapi/admin/orgs.py

@@ -0,0 +1,234 @@
+from uuid import UUID
+
+from fastapi import Body, FastAPI, HTTPException, Query, Request
+
+from paskia import db
+from paskia.db import Org as OrgDC
+from paskia.db import Role as RoleDC
+from paskia.db import User as UserDC
+from paskia.fastapi import authz
+from paskia.fastapi.admin.errors import install_error_handlers
+from paskia.fastapi.response import MsgspecResponse
+from paskia.fastapi.session import AUTH_COOKIE
+from paskia.util import permutil
+from paskia.util.apistructs import ApiUuidResponse
+
+app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
+
+install_error_handlers(app)
+
+
+def master_admin(ctx) -> bool:
+    return any(p.scope == "auth:admin" for p in ctx.permissions)
+
+
+def org_admin(ctx, org_uuid: UUID) -> bool:
+    return ctx.org.uuid == org_uuid and any(
+        p.scope == "auth:org:admin" for p in ctx.permissions
+    )
+
+
+def can_manage_org(ctx, org_uuid: UUID) -> bool:
+    return master_admin(ctx) or org_admin(ctx, org_uuid)
+
+
+@app.post("/")
+async def admin_create_org(
+    request: Request, payload: dict = Body(...), auth=AUTH_COOKIE
+):
+    ctx = await authz.verify(
+        auth, ["auth:admin"], host=request.headers.get("host"), match=permutil.has_all
+    )
+
+    display_name = payload.get("display_name") or "New Organization"
+    permissions = payload.get("permissions") or []
+    org = OrgDC.create(display_name=display_name)
+    db.create_org(org, ctx=ctx)
+    # Grant requested permissions to the new org
+    for perm in permissions:
+        db.add_permission_to_org(str(org.uuid), perm, ctx=ctx)
+
+    return MsgspecResponse(ApiUuidResponse(uuid=str(org.uuid)))
+
+
+@app.patch("/{org_uuid}")
+async def admin_update_org_name(
+    org_uuid: UUID,
+    request: Request,
+    payload: dict = Body(...),
+    auth=AUTH_COOKIE,
+):
+    """Update organization display name only."""
+    ctx = await authz.verify(
+        auth,
+        ["auth:admin", "auth:org:admin"],
+        match=permutil.has_any,
+        host=request.headers.get("host"),
+    )
+    if not can_manage_org(ctx, org_uuid):
+        raise authz.AuthException(
+            status_code=403, detail="Insufficient permissions", mode="forbidden"
+        )
+    display_name = payload.get("display_name")
+    if not display_name:
+        raise ValueError("display_name is required")
+
+    db.update_org_name(org_uuid, display_name, ctx=ctx)
+    return {"status": "ok"}
+
+
+@app.delete("/{org_uuid}")
+async def admin_delete_org(org_uuid: UUID, request: Request, auth=AUTH_COOKIE):
+    ctx = await authz.verify(
+        auth,
+        ["auth:admin", "auth:org:admin"],
+        match=permutil.has_any,
+        host=request.headers.get("host"),
+        max_age="5m",
+    )
+    if not can_manage_org(ctx, org_uuid):
+        raise authz.AuthException(
+            status_code=403, detail="Insufficient permissions", mode="forbidden"
+        )
+    if ctx.org.uuid == org_uuid:
+        raise ValueError("Cannot delete the organization you belong to")
+
+    # Delete organization-specific permissions
+    org_perm_pattern = f"org:{str(org_uuid).lower()}"
+    all_permissions = list(db.data().permissions.values())
+    for perm in all_permissions:
+        perm_scope_lower = perm.scope.lower()
+        # Check if permission contains "org:{uuid}" separated by colons or at boundaries
+        if (
+            f":{org_perm_pattern}:" in perm_scope_lower
+            or perm_scope_lower.startswith(f"{org_perm_pattern}:")
+            or perm_scope_lower.endswith(f":{org_perm_pattern}")
+            or perm_scope_lower == org_perm_pattern
+        ):
+            db.delete_permission(perm.uuid, ctx=ctx)
+
+    db.delete_org(org_uuid, ctx=ctx)
+    return {"status": "ok"}
+
+
+@app.post("/{org_uuid}/permission")
+async def admin_add_org_permission(
+    org_uuid: UUID,
+    request: Request,
+    permission_uuid: UUID = Query(...),
+    auth=AUTH_COOKIE,
+):
+    ctx = await authz.verify(
+        auth, ["auth:admin"], host=request.headers.get("host"), match=permutil.has_all
+    )
+
+    db.add_permission_to_org(org_uuid, permission_uuid, ctx=ctx)
+    return {"status": "ok"}
+
+
+@app.delete("/{org_uuid}/permission")
+async def admin_remove_org_permission(
+    org_uuid: UUID,
+    request: Request,
+    permission_uuid: UUID = Query(...),
+    auth=AUTH_COOKIE,
+):
+    ctx = await authz.verify(
+        auth, ["auth:admin"], host=request.headers.get("host"), match=permutil.has_all
+    )
+
+    db.remove_permission_from_org(org_uuid, permission_uuid, ctx=ctx)
+
+    # Guard rail: prevent removing auth:admin from your own org if it would lock you out
+    perm = db.data().permissions.get(permission_uuid)
+    if perm and perm.scope == "auth:admin" and ctx.org.uuid == org_uuid:
+        # Check if any other org grants auth:admin that we're a member of
+        # (we only know our current org, so this effectively means we can't remove it from our own org)
+        raise ValueError(
+            "Cannot remove auth:admin from your own organization. "
+            "This would lock you out of admin access."
+        )
+
+    db.remove_permission_from_org(org_uuid, permission_uuid, ctx=ctx)
+    return {"status": "ok"}
+
+
+@app.post("/{org_uuid}/roles")
+async def admin_create_role(
+    org_uuid: UUID,
+    request: Request,
+    payload: dict = Body(...),
+    auth=AUTH_COOKIE,
+):
+    ctx = await authz.verify(
+        auth,
+        ["auth:admin", "auth:org:admin"],
+        match=permutil.has_any,
+        host=request.headers.get("host"),
+    )
+    if not can_manage_org(ctx, org_uuid):
+        raise authz.AuthException(
+            status_code=403, detail="Insufficient permissions", mode="forbidden"
+        )
+
+    display_name = payload.get("display_name") or "New Role"
+    perms = payload.get("permissions") or []
+    if org_uuid not in db.data().orgs:
+        raise HTTPException(status_code=404, detail="Organization not found")
+    org = db.data().orgs[org_uuid]
+    grantable = {p.uuid for p in org.permissions}
+
+    # Normalize permission IDs to UUIDs
+    permission_uuids: set[UUID] = set()
+    for pid in perms:
+        perm = db.data().permissions.get(UUID(pid))
+        if not perm:
+            raise ValueError(f"Permission {pid} not found")
+        if perm.uuid not in grantable:
+            raise ValueError(f"Permission not grantable by org: {pid}")
+        permission_uuids.add(perm.uuid)
+
+    role = RoleDC.create(
+        org=org_uuid,
+        display_name=display_name,
+        permissions=permission_uuids,
+    )
+    db.create_role(role, ctx=ctx)
+    return MsgspecResponse(ApiUuidResponse(uuid=str(role.uuid)))
+
+
+@app.post("/{org_uuid}/users")
+async def admin_create_user(
+    org_uuid: UUID,
+    request: Request,
+    payload: dict = Body(...),
+    auth=AUTH_COOKIE,
+):
+    ctx = await authz.verify(
+        auth,
+        ["auth:admin", "auth:org:admin"],
+        match=permutil.has_any,
+        host=request.headers.get("host"),
+    )
+    if not can_manage_org(ctx, org_uuid):
+        raise authz.AuthException(
+            status_code=403, detail="Insufficient permissions", mode="forbidden"
+        )
+    display_name = payload.get("display_name")
+    role_name = payload.get("role")
+    if not display_name or not role_name:
+        raise ValueError("display_name and role are required")
+
+    org = db.data().orgs[org_uuid]
+    role_obj = next(
+        (r for r in org.roles if r.display_name == role_name),
+        None,
+    )
+    if not role_obj:
+        raise ValueError("Role not found in organization")
+    user = UserDC.create(
+        display_name=display_name,
+        role=role_obj.uuid,
+    )
+    db.create_user(user, ctx=ctx)
+    return MsgspecResponse(ApiUuidResponse(uuid=str(user.uuid)))

paskia/fastapi/admin/permissions.py

@@ -0,0 +1,215 @@
+from uuid import UUID
+
+from fastapi import Body, FastAPI, Query, Request
+
+from paskia import db
+from paskia.db import Permission as PermDC
+from paskia.fastapi import authz
+from paskia.fastapi.admin.errors import install_error_handlers
+from paskia.fastapi.session import AUTH_COOKIE
+from paskia.globals import passkey
+from paskia.util import hostutil, permutil, querysafe
+
+app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
+
+install_error_handlers(app)
+
+
+def _validate_permission_domain(domain: str | None) -> None:
+    """Validate that domain is rp_id, a subdomain of it, or an OIDC client UUID."""
+    if domain is None:
+        return
+
+    # Allow OIDC client UUIDs (used for groups claim)
+    try:
+        client_uuid = UUID(domain)
+        if client_uuid in db.data().oidc.clients:
+            return
+    except ValueError:
+        pass
+
+    rp_id = passkey.instance.rp_id
+    if domain == rp_id or domain.endswith(f".{rp_id}"):
+        return
+    raise ValueError(
+        f"Domain '{domain}' must be '{rp_id}', its subdomain, or an OIDC client UUID"
+    )
+
+
+def _check_admin_lockout(
+    perm_uuid: str, new_domain: str | None, current_host: str | None
+) -> None:
+    """Check if setting domain on auth:admin would lock out the admin.
+
+    Raises ValueError if this change would result in no auth:admin permissions
+    being accessible from the current host.
+    """
+
+    normalized_host = hostutil.normalize_host(current_host)
+    host_without_port = normalized_host.rsplit(":", 1)[0] if normalized_host else None
+
+    # Get all auth:admin permissions
+    all_perms = list(db.data().permissions.values())
+    admin_perms = [p for p in all_perms if p.scope == "auth:admin"]
+
+    # Check if at least one auth:admin would remain accessible
+    for p in admin_perms:
+        # If this is the permission being modified, use the new domain
+        domain = new_domain if str(p.uuid) == perm_uuid else p.domain
+
+        # No domain restriction = accessible from anywhere
+        if domain is None:
+            return
+
+        # Check if domain matches current host
+        if domain == normalized_host or domain == host_without_port:
+            return
+
+        # Check if domain is a subdomain of current host or vice versa
+        if normalized_host and normalized_host.endswith(f".{domain}"):
+            return
+        if host_without_port and host_without_port.endswith(f".{domain}"):
+            return
+
+    raise ValueError(
+        f"Setting domain '{new_domain}' on auth:admin permission would lock you out of "
+        f"admin access from current host '{current_host}'"
+    )
+
+
+def _check_admin_lockout_on_delete(perm_uuid: str, current_host: str | None) -> None:
+    """Check if deleting an auth:admin permission would lock out the admin.
+
+    Raises ValueError if this deletion would result in no auth:admin permissions
+    being accessible from the current host.
+    """
+    normalized_host = hostutil.normalize_host(current_host)
+    host_without_port = normalized_host.rsplit(":", 1)[0] if normalized_host else None
+
+    # Get all auth:admin permissions except the one being deleted
+    all_perms = list(db.data().permissions.values())
+    admin_perms = [
+        p for p in all_perms if p.scope == "auth:admin" and str(p.uuid) != perm_uuid
+    ]
+
+    # Check if at least one auth:admin would remain accessible
+    for p in admin_perms:
+        domain = p.domain
+
+        # No domain restriction = accessible from anywhere
+        if domain is None:
+            return
+
+        # Check if domain matches current host
+        if domain == normalized_host or domain == host_without_port:
+            return
+
+        # Check if domain is a subdomain of current host or vice versa
+        if normalized_host and normalized_host.endswith(f".{domain}"):
+            return
+        if host_without_port and host_without_port.endswith(f".{domain}"):
+            return
+
+    raise ValueError(
+        f"Deleting this auth:admin permission would lock you out of "
+        f"admin access from current host '{current_host}'"
+    )
+
+
+@app.post("/")
+async def admin_create_permission(
+    request: Request,
+    payload: dict = Body(...),
+    auth=AUTH_COOKIE,
+):
+    ctx = await authz.verify(
+        auth,
+        ["auth:admin"],
+        host=request.headers.get("host"),
+        match=permutil.has_all,
+        max_age="5m",
+    )
+
+    scope = payload.get("scope") or payload.get(
+        "id"
+    )  # Support both for backwards compat
+    display_name = payload.get("display_name")
+    domain = payload.get("domain") or None  # Treat empty string as None
+    if not scope or not display_name:
+        raise ValueError("scope and display_name are required")
+    querysafe.assert_safe(scope, field="scope")
+    _validate_permission_domain(domain)
+    db.create_permission(
+        PermDC.create(scope=scope, display_name=display_name, domain=domain),
+        ctx=ctx,
+    )
+    return {"status": "ok"}
+
+
+@app.patch("/{permission_uuid}")
+async def admin_update_permission(
+    permission_uuid: UUID,
+    request: Request,
+    auth=AUTH_COOKIE,
+    display_name: str | None = Query(None),
+    scope: str | None = Query(None),
+    domain: str | None = Query(None),
+):
+    ctx = await authz.verify(
+        auth, ["auth:admin"], host=request.headers.get("host"), match=permutil.has_all
+    )
+
+    # Get existing permission
+    perm = db.data().permissions.get(permission_uuid)
+
+    # Update fields that were provided
+    new_scope = scope if scope is not None else perm.scope
+    new_display_name = display_name if display_name is not None else perm.display_name
+    domain_value = domain if domain else None
+
+    # Sanity check: prevent changing the auth:admin permission scope
+    if perm.scope == "auth:admin" and new_scope != "auth:admin":
+        raise ValueError("Cannot rename the master admin permission")
+
+    if not new_display_name:
+        raise ValueError("display_name is required")
+    querysafe.assert_safe(new_scope, field="scope")
+    _validate_permission_domain(domain_value)
+
+    # Safety check: prevent admin lockout when setting domain on auth:admin
+    if perm.scope == "auth:admin" or new_scope == "auth:admin":
+        _check_admin_lockout(str(perm.uuid), domain_value, request.headers.get("host"))
+
+    db.update_permission(
+        uuid=perm.uuid,
+        scope=new_scope,
+        display_name=new_display_name,
+        domain=domain_value,
+        ctx=ctx,
+    )
+    return {"status": "ok"}
+
+
+@app.delete("/{permission_uuid}")
+async def admin_delete_permission(
+    permission_uuid: UUID,
+    request: Request,
+    auth=AUTH_COOKIE,
+):
+    ctx = await authz.verify(
+        auth,
+        ["auth:admin"],
+        host=request.headers.get("host"),
+        match=permutil.has_all,
+        max_age="5m",
+    )
+
+    # Get the permission to check its scope
+    perm = db.data().permissions.get(permission_uuid)
+
+    # Sanity check: prevent deleting critical permissions if it would lock out admin
+    if perm.scope == "auth:admin":
+        _check_admin_lockout_on_delete(str(perm.uuid), request.headers.get("host"))
+
+    db.delete_permission(permission_uuid, ctx=ctx)
+    return {"status": "ok"}

paskia/fastapi/admin/roles.py

@@ -0,0 +1,160 @@
+from uuid import UUID
+
+from fastapi import Body, FastAPI, HTTPException, Request
+
+from paskia import db
+from paskia.fastapi import authz
+from paskia.fastapi.admin.errors import install_error_handlers
+from paskia.fastapi.session import AUTH_COOKIE
+from paskia.util import permutil
+
+app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
+
+install_error_handlers(app)
+
+
+def master_admin(ctx) -> bool:
+    return any(p.scope == "auth:admin" for p in ctx.permissions)
+
+
+def org_admin(ctx, org_uuid: UUID) -> bool:
+    return ctx.org.uuid == org_uuid and any(
+        p.scope == "auth:org:admin" for p in ctx.permissions
+    )
+
+
+def can_manage_org(ctx, org_uuid: UUID) -> bool:
+    return master_admin(ctx) or org_admin(ctx, org_uuid)
+
+
+@app.patch("/{role_uuid}")
+async def admin_update_role_name(
+    role_uuid: UUID,
+    request: Request,
+    payload: dict = Body(...),
+    auth=AUTH_COOKIE,
+):
+    """Update role display name only."""
+    role = db.data().roles.get(role_uuid)
+    if not role:
+        raise HTTPException(status_code=404, detail="Role not found")
+    ctx = await authz.verify(
+        auth,
+        ["auth:admin", "auth:org:admin"],
+        match=permutil.has_any,
+        host=request.headers.get("host"),
+    )
+    if not can_manage_org(ctx, role.org_uuid):
+        raise authz.AuthException(
+            status_code=403, detail="Insufficient permissions", mode="forbidden"
+        )
+
+    display_name = payload.get("display_name")
+    if not display_name:
+        raise ValueError("display_name is required")
+
+    db.update_role_name(role_uuid, display_name, ctx=ctx)
+    return {"status": "ok"}
+
+
+@app.post("/{role_uuid}/permissions/{permission_uuid}")
+async def admin_add_role_permission(
+    role_uuid: UUID,
+    permission_uuid: UUID,
+    request: Request,
+    auth=AUTH_COOKIE,
+):
+    """Add a permission to a role (intent-based API)."""
+    role = db.data().roles.get(role_uuid)
+    if not role:
+        raise HTTPException(status_code=404, detail="Role not found")
+    ctx = await authz.verify(
+        auth,
+        ["auth:admin", "auth:org:admin"],
+        match=permutil.has_any,
+        host=request.headers.get("host"),
+    )
+    if not can_manage_org(ctx, role.org_uuid):
+        raise authz.AuthException(
+            status_code=403, detail="Insufficient permissions", mode="forbidden"
+        )
+
+    # Verify permission exists and org can grant it
+    perm = db.data().permissions.get(permission_uuid)
+    if not perm:
+        raise HTTPException(status_code=404, detail="Permission not found")
+    if role.org_uuid not in perm.orgs:
+        raise ValueError("Permission not grantable by organization")
+
+    db.add_permission_to_role(role_uuid, permission_uuid, ctx=ctx)
+    return {"status": "ok"}
+
+
+@app.delete("/{role_uuid}/permissions/{permission_uuid}")
+async def admin_remove_role_permission(
+    role_uuid: UUID,
+    permission_uuid: UUID,
+    request: Request,
+    auth=AUTH_COOKIE,
+):
+    """Remove a permission from a role (intent-based API)."""
+    role = db.data().roles.get(role_uuid)
+    if not role:
+        raise HTTPException(status_code=404, detail="Role not found")
+    ctx = await authz.verify(
+        auth,
+        ["auth:admin", "auth:org:admin"],
+        match=permutil.has_any,
+        host=request.headers.get("host"),
+    )
+    if not can_manage_org(ctx, role.org_uuid):
+        raise authz.AuthException(
+            status_code=403, detail="Insufficient permissions", mode="forbidden"
+        )
+
+    # Sanity check: prevent admin from removing their own access
+    perm = db.data().permissions.get(permission_uuid)
+    if ctx.org.uuid == role.org_uuid and ctx.role.uuid == role_uuid:
+        if perm and perm.scope in ["auth:admin", "auth:org:admin"]:
+            # Check if removing this permission would leave no admin access
+            remaining_perms = role.permission_set - {permission_uuid}
+            has_admin = False
+            for rp_uuid in remaining_perms:
+                rp = db.data().permissions.get(rp_uuid)
+                if rp and rp.scope in ["auth:admin", "auth:org:admin"]:
+                    has_admin = True
+                    break
+            if not has_admin:
+                raise ValueError("Cannot remove your own admin permissions")
+
+    db.remove_permission_from_role(role_uuid, permission_uuid, ctx=ctx)
+    return {"status": "ok"}
+
+
+@app.delete("/{role_uuid}")
+async def admin_delete_role(
+    role_uuid: UUID,
+    request: Request,
+    auth=AUTH_COOKIE,
+):
+    role = db.data().roles.get(role_uuid)
+    if not role:
+        raise HTTPException(status_code=404, detail="Role not found")
+    ctx = await authz.verify(
+        auth,
+        ["auth:admin", "auth:org:admin"],
+        match=permutil.has_any,
+        host=request.headers.get("host"),
+        max_age="5m",
+    )
+    if not can_manage_org(ctx, role.org_uuid):
+        raise authz.AuthException(
+            status_code=403, detail="Insufficient permissions", mode="forbidden"
+        )
+
+    # Sanity check: prevent admin from deleting their own role
+    if ctx.role.uuid == role_uuid:
+        raise ValueError("Cannot delete your own role")
+
+    db.delete_role(role_uuid, ctx=ctx)
+    return {"status": "ok"}

paskia/fastapi/admin/server_config.py

@@ -0,0 +1,87 @@
+from fastapi import Body, FastAPI, HTTPException, Request
+
+from paskia import db
+from paskia.db.structs import Config
+from paskia.fastapi import authz
+from paskia.fastapi.admin.errors import install_error_handlers
+from paskia.fastapi.session import AUTH_COOKIE
+from paskia.globals import passkey
+from paskia.sansio import Passkey
+from paskia.util import hostutil
+
+app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
+
+install_error_handlers(app)
+
+
+@app.get("/")
+async def admin_get_server_config(
+    request: Request,
+    auth=AUTH_COOKIE,
+):
+    """Get current server configuration (master admin only)."""
+    await authz.verify(auth, ["auth:admin"], host=request.headers.get("host"))
+    pk = passkey.instance
+    config = db.data().config
+    return {
+        "rp_name": pk.rp_name,
+        "auth_host": config.auth_host or "",
+        "origins": list(pk.allowed_origins) if pk.allowed_origins else [],
+    }
+
+
+@app.patch("/")
+async def admin_update_server_config(
+    request: Request,
+    payload: dict = Body(...),
+    auth=AUTH_COOKIE,
+):
+    """Update server configuration (master admin only).
+
+    Updates rp_name, auth_host, and origins in both the runtime Passkey
+    instance and the persisted database config.
+    """
+    await authz.verify(
+        auth, ["auth:admin"], host=request.headers.get("host"), max_age="5m"
+    )
+    config = db.data().config
+    pk = passkey.instance
+
+    rp_name = payload.get("rp_name", "").strip() or None
+    auth_host = payload.get("auth_host", "").strip() or None
+    raw_origins = payload.get("origins", [])
+    origins = [
+        hostutil.normalize_origin(o.strip()) for o in raw_origins if o.strip()
+    ] or None
+
+    # Normalize auth_host and origins (matching CLI startup behavior)
+    if auth_host:
+        try:
+            hostutil.validate_auth_host(auth_host, config.rp_id)
+        except ValueError as e:
+            raise HTTPException(status_code=400, detail=str(e))
+    auth_host, origins = hostutil.normalize_auth_host_and_origins(auth_host, origins)
+
+    # Validate origins against the current rp_id
+    if origins:
+        for o in origins:
+            Passkey(rp_id=config.rp_id, origins=[o])  # validates or raises
+
+    # Update runtime Passkey instance
+    pk.rp_name = rp_name or config.rp_id
+    pk.allowed_origins = set(origins) if origins else None
+
+    # Persist to database
+    new_config = Config(
+        rp_id=config.rp_id,
+        rp_name=rp_name,
+        auth_host=auth_host,
+        origins=origins,
+        listen=config.listen,
+    )
+    await db.update_config(new_config)
+
+    # Reload hostutil cached config so auth_host changes take effect
+    hostutil.reload_config()
+
+    return {"status": "ok"}

paskia/fastapi/admin/users.py

@@ -0,0 +1,318 @@
+from uuid import UUID
+
+from fastapi import Body, FastAPI, HTTPException, Request
+
+from paskia import aaguid as aaguid_mod
+from paskia import db
+from paskia.authsession import reset_expires
+from paskia.fastapi import authz
+from paskia.fastapi.admin.errors import install_error_handlers
+from paskia.fastapi.response import MsgspecResponse
+from paskia.fastapi.session import AUTH_COOKIE
+from paskia.util import hostutil, permutil
+from paskia.util.apistructs import (
+    ApiAaguidInfo,
+    ApiCreateLinkResponse,
+    ApiOrg,
+    ApiRole,
+    ApiUser,
+    ApiUserDetail,
+    ApiUserSession,
+)
+
+app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
+
+install_error_handlers(app)
+
+
+def master_admin(ctx) -> bool:
+    return any(p.scope == "auth:admin" for p in ctx.permissions)
+
+
+def org_admin(ctx, org_uuid: UUID) -> bool:
+    return ctx.org.uuid == org_uuid and any(
+        p.scope == "auth:org:admin" for p in ctx.permissions
+    )
+
+
+def can_manage_org(ctx, org_uuid: UUID) -> bool:
+    return master_admin(ctx) or org_admin(ctx, org_uuid)
+
+
+@app.patch("/{user_uuid}/role")
+async def admin_update_user_role(
+    user_uuid: UUID,
+    request: Request,
+    payload: dict = Body(...),
+    auth=AUTH_COOKIE,
+):
+    try:
+        user = db.data().users[user_uuid]
+    except KeyError:
+        raise HTTPException(status_code=404, detail="User not found")
+    ctx = await authz.verify(
+        auth,
+        ["auth:admin", "auth:org:admin"],
+        match=permutil.has_any,
+        host=request.headers.get("host"),
+    )
+    if not can_manage_org(ctx, user.org.uuid):
+        raise authz.AuthException(
+            status_code=403, detail="Insufficient permissions", mode="forbidden"
+        )
+    role_uuid_str = payload.get("role_uuid")
+    if not role_uuid_str:
+        raise ValueError("role_uuid is required")
+    try:
+        new_role_uuid = UUID(role_uuid_str)
+    except (ValueError, TypeError):
+        raise ValueError("Invalid role UUID")
+    new_role = db.data().roles.get(new_role_uuid)
+    if not new_role or new_role.org_uuid != user.org.uuid:
+        raise ValueError("Role not found in organization")
+
+    # Sanity check: prevent admin from removing their own access
+    if ctx.user.uuid == user_uuid:
+        # Check if any permission in the new role is an admin permission
+        has_admin_access = False
+        for perm_uuid in new_role.permissions:
+            perm = db.data().permissions.get(perm_uuid)
+            if perm and perm.scope in ["auth:admin", "auth:org:admin"]:
+                has_admin_access = True
+                break
+        if not has_admin_access:
+            raise ValueError(
+                "Cannot change your own role to one without admin permissions"
+            )
+
+    db.update_user_role(user_uuid, new_role_uuid, ctx=ctx)
+    return {"status": "ok"}
+
+
+@app.post("/{user_uuid}/create-link")
+async def admin_create_user_registration_link(
+    user_uuid: UUID,
+    request: Request,
+    auth=AUTH_COOKIE,
+):
+    try:
+        user = db.data().users[user_uuid]
+    except KeyError:
+        raise HTTPException(status_code=404, detail="User not found")
+    ctx = await authz.verify(
+        auth,
+        ["auth:admin", "auth:org:admin"],
+        match=permutil.has_any,
+        host=request.headers.get("host"),
+        max_age="5m",
+    )
+    if not can_manage_org(ctx, user.org.uuid):
+        raise authz.AuthException(
+            status_code=403, detail="Insufficient permissions", mode="forbidden"
+        )
+
+    # Check if user has existing credentials
+    has_credentials = db.data().users[user_uuid].credential_ids
+    token_type = "user registration" if not has_credentials else "account recovery"
+
+    expiry = reset_expires()
+    token = db.create_reset_token(
+        user_uuid=user_uuid,
+        expiry=expiry,
+        token_type=token_type,
+        ctx=ctx,
+    )
+    url = hostutil.reset_link_url(token)
+    return MsgspecResponse(
+        ApiCreateLinkResponse(
+            url=url,
+            expires=expiry,
+            token_type=token_type,
+        )
+    )
+
+
+@app.get("/{user_uuid}")
+async def admin_get_user_detail(
+    user_uuid: UUID,
+    request: Request,
+    auth=AUTH_COOKIE,
+):
+    try:
+        user = db.data().users[user_uuid]
+    except KeyError:
+        raise HTTPException(status_code=404, detail="User not found")
+    ctx = await authz.verify(
+        auth,
+        ["auth:admin", "auth:org:admin"],
+        match=permutil.has_any,
+        host=request.headers.get("host"),
+    )
+    if not can_manage_org(ctx, user.org.uuid):
+        raise authz.AuthException(
+            status_code=403, detail="Insufficient permissions", mode="forbidden"
+        )
+    normalized_host = hostutil.normalize_host(request.headers.get("host"))
+
+    sessions = {
+        s.key: ApiUserSession.from_db(
+            s,
+            current_key=ctx.session.key,
+            normalized_host=normalized_host,
+        )
+        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,
+            org=ApiOrg.from_db(user.org),
+            role=ApiRole.from_db(user.role),
+        )
+    )
+
+
+@app.patch("/{user_uuid}/info")
+async def admin_update_user_info(
+    user_uuid: UUID,
+    request: Request,
+    payload: dict = Body(...),
+    auth=AUTH_COOKIE,
+):
+    """Update user profile info (display_name, email, preferred_username, telephone).
+
+    Pass only the fields you want to update. Use null to clear optional fields.
+    """
+    try:
+        user = db.data().users[user_uuid]
+    except KeyError:
+        raise HTTPException(status_code=404, detail="User not found")
+    ctx = await authz.verify(
+        auth,
+        ["auth:admin", "auth:org:admin"],
+        match=permutil.has_any,
+        host=request.headers.get("host"),
+    )
+    if not can_manage_org(ctx, user.org.uuid):
+        raise authz.AuthException(
+            status_code=403, detail="Insufficient permissions", mode="forbidden"
+        )
+
+    kwargs = {}
+    if "display_name" in payload:
+        name = (payload["display_name"] or "").strip()
+        if not name:
+            raise HTTPException(status_code=400, detail="display_name cannot be empty")
+        if len(name) > 64:
+            raise HTTPException(status_code=400, detail="display_name too long")
+        kwargs["display_name"] = name
+    if "email" in payload:
+        kwargs["email"] = payload["email"]
+    if "preferred_username" in payload:
+        kwargs["preferred_username"] = payload["preferred_username"]
+    if "telephone" in payload:
+        kwargs["telephone"] = payload["telephone"]
+
+    if not kwargs:
+        raise HTTPException(status_code=400, detail="No fields to update")
+
+    db.update_user_info(user_uuid, **kwargs, ctx=ctx)
+    return {"status": "ok"}
+
+
+@app.delete("/{user_uuid}")
+async def admin_delete_user(
+    user_uuid: UUID,
+    request: Request,
+    auth=AUTH_COOKIE,
+):
+    """Delete a user and all their credentials/sessions."""
+    try:
+        user = db.data().users[user_uuid]
+    except KeyError:
+        raise HTTPException(status_code=404, detail="User not found")
+    ctx = await authz.verify(
+        auth,
+        ["auth:admin", "auth:org:admin"],
+        match=permutil.has_any,
+        host=request.headers.get("host"),
+        max_age="5m",
+    )
+    if not can_manage_org(ctx, user.org.uuid):
+        raise authz.AuthException(
+            status_code=403, detail="Insufficient permissions", mode="forbidden"
+        )
+    # Prevent admin from deleting themselves
+    if ctx.user.uuid == user_uuid:
+        raise ValueError("Cannot delete your own account")
+    db.delete_user(user_uuid, ctx=ctx)
+    return {"status": "ok"}
+
+
+@app.delete("/{user_uuid}/credentials/{credential_uuid}")
+async def admin_delete_user_credential(
+    user_uuid: UUID,
+    credential_uuid: UUID,
+    request: Request,
+    auth=AUTH_COOKIE,
+):
+    try:
+        user = db.data().users[user_uuid]
+    except KeyError:
+        raise HTTPException(status_code=404, detail="User not found")
+    ctx = await authz.verify(
+        auth,
+        ["auth:admin", "auth:org:admin"],
+        match=permutil.has_any,
+        host=request.headers.get("host"),
+        max_age="5m",
+    )
+    if not can_manage_org(ctx, user.org.uuid):
+        raise authz.AuthException(
+            status_code=403, detail="Insufficient permissions", mode="forbidden"
+        )
+    db.delete_credential(credential_uuid, user_uuid, ctx=ctx)
+    return {"status": "ok"}
+
+
+@app.delete("/{user_uuid}/sessions/{session_id}")
+async def admin_delete_user_session(
+    user_uuid: UUID,
+    session_id: str,
+    request: Request,
+    auth=AUTH_COOKIE,
+):
+    try:
+        user = db.data().users[user_uuid]
+    except KeyError:
+        raise HTTPException(status_code=404, detail="User not found")
+    ctx = await authz.verify(
+        auth,
+        ["auth:admin", "auth:org:admin"],
+        match=permutil.has_any,
+        host=request.headers.get("host"),
+    )
+    if not can_manage_org(ctx, user.org.uuid):
+        raise authz.AuthException(
+            status_code=403, detail="Insufficient permissions", mode="forbidden"
+        )
+
+    session_key = session_id
+
+    target_session = db.data().sessions.get(session_key)
+    if not target_session or target_session.user_uuid != user_uuid:
+        raise HTTPException(status_code=404, detail="Session not found")
+
+    db.delete_session(session_key, ctx=ctx, action="admin:delete_session")
+
+    # Check if admin terminated their own session
+    current_terminated = session_key == ctx.session.key
+    return {"status": "ok", "current_session_terminated": current_terminated}

tests/test_admin.py

@@ -263,7 +263,7 @@ class TestAdminOrganizations:
     ):
         """Creating org without admin permission should fail."""
         response = await client.post(
-            "/auth/api/admin/orgs",
+            "/auth/api/admin/orgs/",
             json={"display_name": "New Org"},
             headers={
                 **auth_headers(regular_session_token),
@@ -278,7 +278,7 @@ class TestAdminOrganizations:
     ):
         """Admin should be able to create a new organization."""
         response = await client.post(
-            "/auth/api/admin/orgs",
+            "/auth/api/admin/orgs/",
             json={"display_name": "New Test Org", "permissions": []},
             headers={**auth_headers(session_token), "Host": "localhost:4401"},
         )
@@ -292,7 +292,7 @@ class TestAdminOrganizations:
     ):
         """Admin should be able to create org with default values."""
         response = await client.post(
-            "/auth/api/admin/orgs",
+            "/auth/api/admin/orgs/",
             json={},  # No display_name or permissions
             headers={**auth_headers(session_token), "Host": "localhost:4401"},
         )
@@ -1421,7 +1421,7 @@ class TestAdminPermissions:
     ):
         """Admin should be able to create new permissions."""
         response = await client.post(
-            "/auth/api/admin/permissions",
+            "/auth/api/admin/permissions/",
             json={"scope": "test:create:permission", "display_name": "Test Permission"},
             headers={**auth_headers(session_token), "Host": "localhost:4401"},
         )
@@ -1435,7 +1435,7 @@ class TestAdminPermissions:
     ):
         """Creating permission without required fields should fail."""
         response = await client.post(
-            "/auth/api/admin/permissions",
+            "/auth/api/admin/permissions/",
             json={},
             headers={**auth_headers(session_token), "Host": "localhost:4401"},
         )
@@ -1449,7 +1449,7 @@ class TestAdminPermissions:
     ):
         """Creating permission without admin should fail."""
         response = await client.post(
-            "/auth/api/admin/permissions",
+            "/auth/api/admin/permissions/",
             json={"scope": "test:forbidden", "display_name": "Forbidden"},
             headers={
                 **auth_headers(regular_session_token),
@@ -1468,7 +1468,7 @@ class TestAdminPermissions:
         create_permission(perm)
 
         response = await client.patch(
-            f"/auth/api/admin/permission?permission_uuid={perm.uuid}&display_name=Updated%20Name",
+            f"/auth/api/admin/permissions/{perm.uuid}?display_name=Updated%20Name",
             headers={**auth_headers(session_token), "Host": "localhost:4401"},
         )
         assert response.status_code == 200
@@ -1485,7 +1485,7 @@ class TestAdminPermissions:
         create_permission(perm)
 
         response = await client.patch(
-            f"/auth/api/admin/permission?permission_uuid={perm.uuid}&display_name=",
+            f"/auth/api/admin/permissions/{perm.uuid}?display_name=",
             headers={**auth_headers(session_token), "Host": "localhost:4401"},
         )
         assert response.status_code == 400
@@ -1502,7 +1502,7 @@ class TestAdminPermissions:
         create_permission(perm)
 
         response = await client.patch(
-            f"/auth/api/admin/permission?permission_uuid={perm.uuid}&scope=test:renamed2",
+            f"/auth/api/admin/permissions/{perm.uuid}?scope=test:renamed2",
             headers={**auth_headers(session_token), "Host": "localhost:4401"},
         )
         assert response.status_code == 200
@@ -1518,7 +1518,7 @@ class TestAdminPermissions:
         admin_perm = next(p for p in perms if p.scope == "auth:admin")
 
         response = await client.patch(
-            f"/auth/api/admin/permission?permission_uuid={admin_perm.uuid}&scope=auth:superadmin",
+            f"/auth/api/admin/permissions/{admin_perm.uuid}?scope=auth:superadmin",
             headers={**auth_headers(session_token), "Host": "localhost:4401"},
         )
         assert response.status_code == 400
@@ -1534,7 +1534,7 @@ class TestAdminPermissions:
         create_permission(perm)
 
         response = await client.patch(
-            f"/auth/api/admin/permission?permission_uuid={perm.uuid}&scope=test:renamed:withname&display_name=New%20Display%20Name",
+            f"/auth/api/admin/permissions/{perm.uuid}?scope=test:renamed:withname&display_name=New%20Display%20Name",
             headers={**auth_headers(session_token), "Host": "localhost:4401"},
         )
         assert response.status_code == 200
@@ -1549,7 +1549,7 @@ class TestAdminPermissions:
         create_permission(perm)
 
         response = await client.delete(
-            f"/auth/api/admin/permission?permission_uuid={perm.uuid}",
+            f"/auth/api/admin/permissions/{perm.uuid}",
             headers={**auth_headers(session_token), "Host": "localhost:4401"},
         )
         assert response.status_code == 200
@@ -1567,7 +1567,7 @@ class TestAdminPermissions:
         admin_perm = next(p for p in perms if p.scope == "auth:admin")
 
         response = await client.delete(
-            f"/auth/api/admin/permission?permission_uuid={admin_perm.uuid}",
+            f"/auth/api/admin/permissions/{admin_perm.uuid}",
             headers={**auth_headers(session_token), "Host": "localhost:4401"},
         )
         assert response.status_code == 400
@@ -1593,7 +1593,7 @@ class TestAdminPermissions:
 
         # Now we can delete the original one
         response = await client.delete(
-            f"/auth/api/admin/permission?permission_uuid={original_admin_perm.uuid}",
+            f"/auth/api/admin/permissions/{original_admin_perm.uuid}",
             headers={**auth_headers(session_token), "Host": "localhost:4401"},
         )
         assert response.status_code == 200
@@ -1622,7 +1622,7 @@ class TestAdminPermissions:
         original_admin_perm = admin_perms[0]  # The one without domain
 
         response = await client.delete(
-            f"/auth/api/admin/permission?permission_uuid={original_admin_perm.uuid}",
+            f"/auth/api/admin/permissions/{original_admin_perm.uuid}",
             headers={**auth_headers(session_token), "Host": "localhost:4401"},
         )
         assert response.status_code == 400