paskia/paskia/fastapi/admin.py

import logging
from uuid import UUID

from fastapi import Body, FastAPI, HTTPException, Query, Request, Response
from fastapi.responses import JSONResponse

from paskia import aaguid as aaguid_mod
from paskia import db
from paskia.authsession import EXPIRES, reset_expires
from paskia.db import Org as OrgDC
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 (
    hostutil,
    permutil,
    querysafe,
    vitedev,
)
from paskia.util.apistructs import (
    ApiAaguidInfo,
    ApiCreateLinkResponse,
    ApiOrgResponse,
    ApiPermission,
    ApiUser,
    ApiUserDetail,
    ApiUserSession,
    ApiUuidResponse,
    format_datetime,
)
from paskia.util.hostutil import normalize_host

app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)


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.exception_handler(ValueError)
async def value_error_handler(_request, exc: ValueError):  # pragma: no cover - simple
    return JSONResponse(status_code=400, content={"detail": str(exc)})


@app.exception_handler(authz.AuthException)
async def auth_exception_handler(_request, exc: authz.AuthException):
    """Handle AuthException with auth info for UI."""
    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"})


@app.get("/")
async def adminapp(request: Request, auth=AUTH_COOKIE):
    return Response(*await vitedev.read("/auth/admin/index.html"))


# -------------------- Organizations --------------------


@app.get("/orgs")
async def admin_list_orgs(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 = 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=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({o.uuid: org_to_dict(o) for o in orgs})


@app.post("/orgs")
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("/orgs/{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("/orgs/{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("/orgs/{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("/orgs/{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"}


# -------------------- Roles --------------------


@app.post("/orgs/{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.patch("/roles/{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("/roles/{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("/roles/{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("/roles/{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"}


# -------------------- Users --------------------


@app.post("/orgs/{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)))


@app.patch("/users/{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("/users/{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=format_datetime(expiry),
            token_type=token_type,
        )
    )


@app.get("/users/{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 = [
        ApiUserSession.from_db(
            s,
            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,
        )
    )


@app.patch("/users/{user_uuid}/display-name")
async def admin_update_user_display_name(
    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"
        )
    new_name = (payload.get("display_name") or "").strip()
    if not new_name:
        raise HTTPException(status_code=400, detail="display_name required")
    if len(new_name) > 64:
        raise HTTPException(status_code=400, detail="display_name too long")
    db.update_user_display_name(user_uuid, new_name, ctx=ctx)
    return {"status": "ok"}


@app.delete("/users/{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("/users/{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("/users/{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"
        )

    target_session = db.data().sessions.get(session_id)
    if not target_session or target_session.user_uuid != user_uuid:
        raise HTTPException(status_code=404, detail="Session not found")

    db.delete_session(session_id, ctx=ctx, action="admin:delete_session")

    # Check if admin terminated their own session
    current_terminated = session_id == auth
    return {"status": "ok", "current_session_terminated": current_terminated}


# -------------------- Permissions (global) --------------------


def _validate_permission_domain(domain: str | None) -> None:
    """Validate that domain is rp_id or a subdomain of it."""
    if domain is None:
        return

    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}' or its subdomain")


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 = 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

        # Domain matches current host
        if host_without_port and domain == host_without_port:
            return

    # No accessible auth:admin permission would remain
    raise ValueError(
        "Cannot set this domain restriction: it would lock you out of admin access. "
        "Ensure at least one auth:admin permission remains accessible from your 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 = 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 after deletion
    for p in admin_perms:
        # Skip the permission being deleted
        if str(p.uuid) == perm_uuid:
            continue

        # No domain restriction = accessible from anywhere
        if p.domain is None:
            return

        # Domain matches current host
        if host_without_port and p.domain == host_without_port:
            return

    # No accessible auth:admin permission would remain
    raise ValueError(
        "Cannot delete this permission: it would lock you out of admin access. "
        "Ensure at least one auth:admin permission remains accessible from your current host."
    )


@app.get("/permissions")
async def admin_list_permissions(request: Request, auth=AUTH_COOKIE):
    ctx = await authz.verify(
        auth,
        ["auth:admin", "auth:org:admin"],
        match=permutil.has_any,
        host=request.headers.get("host"),
    )
    perms = db.data().permissions.values() if master_admin(ctx) else ctx.org.permissions
    return MsgspecResponse({p.uuid: ApiPermission.from_db(p) for p in perms})


@app.post("/permissions")
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")
async def admin_update_permission(
    request: Request,
    auth=AUTH_COOKIE,
    permission_uuid: UUID = Query(...),
    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")
async def admin_delete_permission(
    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,
        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"}