api/identity_routes.py

"""D2 identity/account API — deterministic, torch-free, dormant unless enabled.

Mounted only when ``AMANPAY_D2_ENABLED=1`` (see ``api/main.py``). The runtime (durable store,
services) is built during application startup and injected here; nothing in this module opens
a database, connects to a bucket, or runs at import time.

Security boundaries enforced here:
* per-user resources are bound to the SESSION COOKIE subject, never a client-supplied id;
* state-changing requests require a matching ``X-CSRF-Token``;
* operator routes require ``role == 'operator'`` (deterministic RBAC);
* sensitive actions require a recent passkey step-up;
* responses are safe projections — no public-key bytes, token/invitation hashes, raw
  challenges, storage paths or bucket keys ever leave the server;
* errors are normalized reason tokens (see :mod:`amanpay.identity.errors`) and never disclose
  whether an account exists.
"""

from __future__ import annotations

import secrets
import time
from typing import Optional

from fastapi import APIRouter, Body, Request, Response
from fastapi.responses import JSONResponse

from amanpay.identity.errors import D2Error, status_for
from amanpay.identity.models import Credential, Session, User
from amanpay.identity.sessions import SESSION_COOKIE

# Reuse the existing shared KV-backed token-bucket limiter.
from api.auth import rate_limit

router = APIRouter(prefix="/identity/v1", tags=["identity"])

# The runtime is injected at startup (or by tests). None => D2 not ready (503).
_RUNTIME = None


def set_runtime(runtime) -> None:
    global _RUNTIME
    _RUNTIME = runtime


def get_runtime():
    if _RUNTIME is None:
        raise D2Error("not_found")                         # generic when D2 not ready
    return _RUNTIME


def install_error_handler(app) -> None:
    """Register the D2Error → normalized-HTTP translator on a FastAPI app."""
    @app.exception_handler(D2Error)
    async def _d2_error_handler(_request: Request, exc: D2Error):   # noqa: ANN202
        token = str(exc)
        return JSONResponse(status_code=status_for(token), content={"detail": token})


# --------------------------------------------------------------------- helpers
def _identity():
    return get_runtime().identity


def _operator_service():
    return get_runtime().operator


def _cookie_token(request: Request) -> Optional[str]:
    return request.cookies.get(SESSION_COOKIE)


def _set_session_cookie(response: Response, issued) -> None:
    attrs = _identity().sessions.cookie_attributes()
    response.set_cookie(SESSION_COOKIE, issued.token, httponly=attrs["httponly"],
                        secure=attrs["secure"], samesite=attrs["samesite"],
                        path=attrs["path"], max_age=attrs["max_age"])


def _clear_session_cookie(response: Response) -> None:
    response.delete_cookie(SESSION_COOKIE, path="/")


def _authed(request: Request) -> tuple[Session, User]:
    return _identity().authenticate_request(_cookie_token(request))


def _require_csrf(request: Request, session: Session) -> None:
    _identity().require_csrf(session, request.headers.get("x-csrf-token"))


def _account_view(ident, user: User) -> dict:
    return {
        "id": user.id,
        "public_handle": user.public_handle,
        "display_alias": user.display_alias,
        "role": user.role,
        "status": user.status,
        "preferred_language": user.preferred_language,
        "profile_generation": user.profile_generation,
        "passkey_count": ident.repo.count_active_credentials(user.id),
    }


def _passkey_view(c: Credential) -> dict:
    return {
        "id": c.id,                                        # opaque row id (NOT the raw cred id)
        "nickname": c.nickname,
        "device_type": c.device_type,
        "backup_eligible": c.backup_eligible,
        "backup_state": c.backup_state,
        "transports": [t for t in c.transports.split(",") if t],
        "created_at": c.created_at,
        "last_used_at": c.last_used_at,
        "revoked": c.revoked_at is not None,
    }


def _session_view(s: Session, current_id: str) -> dict:
    return {
        "id": s.id,
        "created_at": s.created_at,
        "last_seen_at": s.last_seen_at,
        "device_nickname": s.device_nickname,
        "current": s.id == current_id,
    }


# ================================================================= status/health
@router.get("/status")
def status() -> dict:
    """Safe, secret-free D2 status for the UI/version surface."""
    rt = _RUNTIME
    if rt is None:
        return {"enabled": False, "ready": False}
    rr = rt.storage.db.runtime_report()
    from amanpay.identity.config import is_demo_open_enroll
    return {"enabled": True, "ready": True, "persistent": rt.persistent,
            "rp_id": rt.config.rp_id, "origin": rt.config.origin,
            "journal_mode": rr.get("journal_mode"), "wal_status": rr.get("wal_status"),
            "schema_version": rt.storage.db.schema_version(),
            "backup_retention_hours": rt.config.backup_retention_hours,
            "demo_open_enroll": is_demo_open_enroll()}


# ===================================================================== enrollment
@router.post("/invitations/validate")
def validate_invitation(request: Request, body: dict = Body(...)) -> dict:
    rate_limit(request, "d2_invite_validate", capacity=10, refill_per_sec=0.2)
    ident = _identity()
    vi = ident.invites.validate(str(body.get("code", "")))
    tenant = ident.repo.get_tenant(vi.tenant_id)
    return {"valid": True, "role": vi.role, "purpose": vi.purpose,
            "tenant": {"name_en": tenant.name_en, "name_ar": tenant.name_ar} if tenant else {}}


@router.post("/enrollment/options")
def enrollment_options(request: Request, body: dict = Body(...)) -> Response:
    rate_limit(request, "d2_enroll", capacity=6, refill_per_sec=0.1)
    opts = _identity().begin_enrollment(
        code=str(body.get("code", "")), public_handle=str(body.get("public_handle", "")),
        display_alias=str(body.get("display_alias", "")),
        preferred_language=str(body.get("preferred_language", "en")),
        consent_accepted=bool(body.get("consent_accepted", False)))
    return Response(content=opts, media_type="application/json")


@router.post("/demo/enrollment/options")
def demo_enrollment_options(request: Request, body: dict = Body(default={})) -> Response:
    """One-click, code-free enrollment into the DEMO tenant (gated by AMANPAY_DEMO_OPEN_ENROLL).

    The server mints a single-use invite itself so a visitor can create a passkey without an
    operator code. Synthetic/demo only — 404 when the flag is off, so it is inert by default.
    """
    from amanpay.identity.config import demo_tenant_slug, is_demo_open_enroll
    if not is_demo_open_enroll():
        raise D2Error("not_found")
    rate_limit(request, "d2_demo_enroll", capacity=6, refill_per_sec=0.1)
    ident = _identity()
    tenant = ident.repo.get_tenant_by_slug(demo_tenant_slug())
    if tenant is None:
        raise D2Error("not_found")
    role = str(body.get("role", "customer"))
    if role not in ("customer", "operator"):
        role = "customer"
    _inv, code = ident.invites.create(tenant_id=tenant.id, role=role, ttl_seconds=600,
                                      created_by="demo")
    opts = ident.begin_enrollment(
        code=code, public_handle=str(body.get("public_handle", "")),
        display_alias=str(body.get("display_alias", "")),
        preferred_language=str(body.get("preferred_language", "en")), consent_accepted=True)
    return Response(content=opts, media_type="application/json")


@router.post("/demo/quick-session")
def demo_quick_session(request: Request, response: Response, body: dict = Body(default={})) -> dict:
    """No-passkey DEMO login (gated by AMANPAY_DEMO_OPEN_ENROLL) — works everywhere, incl. the HF
    embed iframe and any device where WebAuthn is unavailable.

    Creates a passkey-LESS demo account in the demo tenant and issues a session directly, then
    provisions a funded wallet (customer) / claims the demo merchants (operator). This deliberately
    relaxes the passkey requirement FOR THE DEMO TENANT ONLY; it is 404 when the flag is off and is
    clearly a synthetic, no-real-money convenience. Passkey enrollment remains available (and is
    the real security story) on the direct Space origin.
    """
    from amanpay.identity.config import demo_tenant_slug, is_demo_open_enroll
    from amanpay.identity.handles import normalize_handle
    if not is_demo_open_enroll():
        raise D2Error("not_found")
    rate_limit(request, "d2_demo_quick", capacity=8, refill_per_sec=0.2)
    ident = _identity()
    tenant = ident.repo.get_tenant_by_slug(demo_tenant_slug())
    if tenant is None:
        raise D2Error("not_found")
    role = str(body.get("role", "customer"))
    if role not in ("customer", "operator"):
        role = "customer"
    suffix = secrets.token_hex(3)
    handle = ("ops" if role == "operator" else "guest") + suffix
    created = ident.repo.create_user(
        tenant_id=tenant.id, webauthn_user_handle=f"demo-nopk-{handle}",
        public_handle=handle, public_handle_norm=normalize_handle(handle),
        display_alias=(str(body.get("display_alias", "")) or handle.title()),
        preferred_language=str(body.get("preferred_language", "en")), role=role, status="active")
    ident.repo.set_user_status(created.id, "active")
    user = ident.repo.get_user(created.id)
    _provision_demo_wallet(user)
    issued = ident.sessions.create(tenant_id=tenant.id, user_id=user.id, last_auth_at=time.time(),
                                   device_nickname="demo (no passkey)",
                                   user_agent=request.headers.get("user-agent"))
    _set_session_cookie(response, issued)
    return {"account": _account_view(ident, user), "csrf_token": issued.csrf_token,
            "no_passkey": True}


@router.post("/enrollment/verify")
def enrollment_verify(request: Request, response: Response, body: dict = Body(...)) -> dict:
    rate_limit(request, "d2_enroll_verify", capacity=6, refill_per_sec=0.1)
    ident = _identity()
    res = ident.complete_enrollment(credential=body.get("credential", body),
                                    user_agent=request.headers.get("user-agent"))
    _provision_demo_wallet(res.user)
    _set_session_cookie(response, res.issued)
    return {"account": _account_view(ident, res.user), "csrf_token": res.issued.csrf_token}


def _provision_demo_wallet(user) -> None:
    """Post-enroll demo provisioning (best-effort, gated by AMANPAY_DEMO_OPEN_ENROLL).

    Only runs when demo open-enroll is on, D3 is ready, and the new user is in the demo tenant:
    a customer gets a funded simulated wallet; an operator takes ownership of the demo merchants
    (so refunds + ATM confirmation work). A failure here never blocks enrollment.
    """
    from amanpay.identity.config import demo_tenant_slug, is_demo_open_enroll
    if not is_demo_open_enroll():
        return
    try:
        import api.finance_routes as fr
        d3 = fr._RUNTIME
        if d3 is None:
            return
        demo_tenant = _identity().repo.get_tenant_by_slug(demo_tenant_slug())
        if demo_tenant is None or user.tenant_id != demo_tenant.id:
            return
        if user.role == "customer":
            from amanpay.demo import provision_demo_customer
            provision_demo_customer(d3, user.tenant_id, user.id)
        elif user.role == "operator":
            from amanpay.demo import claim_demo_merchants
            claim_demo_merchants(d3, user.tenant_id, user.id)
    except Exception:                                  # never block enrollment on provisioning
        pass


# ===================================================================== login
@router.post("/authentication/options")
def authentication_options(request: Request, body: dict = Body(default={})) -> Response:
    rate_limit(request, "d2_login", capacity=10, refill_per_sec=0.3)
    opts = _identity().begin_login(public_handle=(body or {}).get("public_handle"))
    return Response(content=opts, media_type="application/json")


@router.post("/authentication/verify")
def authentication_verify(request: Request, response: Response, body: dict = Body(...)) -> dict:
    rate_limit(request, "d2_login_verify", capacity=10, refill_per_sec=0.3)
    ident = _identity()
    res = ident.complete_login(credential=body.get("credential", body),
                               user_agent=request.headers.get("user-agent"))
    _set_session_cookie(response, res.issued)
    return {"account": _account_view(ident, res.user), "csrf_token": res.issued.csrf_token}


@router.get("/session")
def session_info(request: Request) -> dict:
    try:
        _sess, user = _authed(request)
    except D2Error:
        return {"authenticated": False}
    return {"authenticated": True, "account": _account_view(_identity(), user)}


@router.post("/logout")
def logout(request: Request, response: Response) -> dict:
    ident = _identity()
    sess, _user = _authed(request)
    _require_csrf(request, sess)
    ident.logout(sess)
    _clear_session_cookie(response)
    return {"ok": True}


# ===================================================================== recovery
@router.post("/recovery/options")
def recovery_options(request: Request, body: dict = Body(...)) -> Response:
    rate_limit(request, "d2_recovery", capacity=6, refill_per_sec=0.05)
    opts = _identity().begin_recovery(code=str(body.get("code", "")))
    return Response(content=opts, media_type="application/json")


@router.post("/recovery/verify")
def recovery_verify(request: Request, response: Response, body: dict = Body(...)) -> dict:
    ident = _identity()
    res = ident.complete_recovery(credential=body.get("credential", body),
                                  user_agent=request.headers.get("user-agent"))
    _set_session_cookie(response, res.issued)
    return {"account": _account_view(ident, res.user), "csrf_token": res.issued.csrf_token}


# ============================================================ authenticated account
@router.get("/account")
def get_account(request: Request) -> dict:
    ident = _identity()
    _sess, user = _authed(request)
    return {"account": _account_view(ident, user)}


@router.get("/passkeys")
def list_passkeys(request: Request) -> dict:
    ident = _identity()
    _sess, user = _authed(request)
    return {"passkeys": [_passkey_view(c) for c in ident.list_passkeys(user)]}


@router.post("/passkeys/options")
def add_passkey_options(request: Request) -> Response:
    ident = _identity()
    sess, user = _authed(request)
    _require_csrf(request, sess)
    opts = ident.begin_add_passkey(user, sess)
    return Response(content=opts, media_type="application/json")


@router.post("/passkeys/verify")
def add_passkey_verify(request: Request, body: dict = Body(...)) -> dict:
    ident = _identity()
    sess, user = _authed(request)
    _require_csrf(request, sess)
    cred = ident.complete_add_passkey(user, sess, credential=body.get("credential", body),
                                      nickname=str(body.get("nickname", "")))
    return {"passkey": _passkey_view(cred)}


@router.patch("/passkeys/{credential_id}")
def rename_passkey(request: Request, credential_id: str, body: dict = Body(...)) -> dict:
    ident = _identity()
    sess, user = _authed(request)
    _require_csrf(request, sess)
    ident.rename_passkey(user, credential_id, str(body.get("nickname", "")))
    return {"ok": True}


@router.delete("/passkeys/{credential_id}")
def revoke_passkey(request: Request, credential_id: str) -> dict:
    ident = _identity()
    sess, user = _authed(request)
    _require_csrf(request, sess)
    ident.revoke_passkey(user, sess, credential_id)
    return {"ok": True}


@router.get("/sessions")
def list_sessions(request: Request) -> dict:
    ident = _identity()
    sess, user = _authed(request)
    return {"sessions": [_session_view(s, sess.id) for s in ident.list_sessions(user)]}


@router.delete("/sessions/{session_id}")
def revoke_session(request: Request, session_id: str) -> dict:
    ident = _identity()
    sess, user = _authed(request)
    _require_csrf(request, sess)
    ident.revoke_session(user, sess, session_id)
    return {"ok": True}


@router.post("/sessions/revoke-others")
def revoke_other_sessions(request: Request) -> dict:
    ident = _identity()
    sess, user = _authed(request)
    _require_csrf(request, sess)
    ident.revoke_other_sessions(user, sess)
    return {"ok": True}


# --------------------------------------------------- step-up (refresh recent auth)
@router.post("/stepup/options")
def stepup_options(request: Request) -> Response:
    ident = _identity()
    sess, user = _authed(request)
    _require_csrf(request, sess)
    opts = ident.begin_stepup(user, sess)
    return Response(content=opts, media_type="application/json")


@router.post("/stepup/verify")
def stepup_verify(request: Request, response: Response, body: dict = Body(...)) -> dict:
    ident = _identity()
    sess, user = _authed(request)
    _require_csrf(request, sess)
    issued = ident.complete_stepup(user, sess, credential=body.get("credential", body))
    _set_session_cookie(response, issued)
    return {"ok": True, "csrf_token": issued.csrf_token}


# ------------------------------------------------------------------- pause/delete
@router.post("/account/pause")
def pause_account(request: Request, response: Response) -> dict:
    ident = _identity()
    sess, user = _authed(request)
    _require_csrf(request, sess)
    ident.pause_account(user, sess)
    _clear_session_cookie(response)
    return {"ok": True, "status": "paused"}


@router.post("/account/delete/options")
def delete_options(request: Request) -> Response:
    ident = _identity()
    sess, user = _authed(request)
    _require_csrf(request, sess)
    opts = ident.begin_delete(user, sess)
    return Response(content=opts, media_type="application/json")


@router.post("/account/delete/confirm")
def delete_confirm(request: Request, response: Response, body: dict = Body(...)) -> dict:
    ident = _identity()
    sess, user = _authed(request)
    _require_csrf(request, sess)
    receipt = ident.complete_delete(user, sess, credential=body.get("credential", body))
    _clear_session_cookie(response)
    return receipt


# ===================================================================== operator
def _authed_operator(request: Request) -> tuple[Session, User]:
    sess, user = _authed(request)
    _operator_service().require_operator(user)
    return sess, user


@router.post("/operator/invitations")
def op_create_invitations(request: Request, body: dict = Body(...)) -> dict:
    op = _operator_service()
    sess, user = _authed_operator(request)
    _require_csrf(request, sess)
    pairs = op.create_invitations(user, role=str(body.get("role", "customer")),
                                  count=int(body.get("count", 1)),
                                  ttl_seconds=int(body.get("ttl_seconds", 86400)))
    # Raw codes are returned ONCE here and never stored in plaintext.
    return {"invitations": [{"id": iid, "code": code} for iid, code in pairs]}


@router.get("/operator/invitations")
def op_list_invitations(request: Request) -> dict:
    op = _operator_service()
    _sess, user = _authed_operator(request)
    return {"invitations": op.list_invitations(user)}


@router.delete("/operator/invitations/{invitation_id}")
def op_revoke_invitation(request: Request, invitation_id: str) -> dict:
    op = _operator_service()
    sess, user = _authed_operator(request)
    _require_csrf(request, sess)
    op.revoke_invitation(user, invitation_id)
    return {"ok": True}


@router.get("/operator/accounts")
def op_list_accounts(request: Request) -> dict:
    op = _operator_service()
    _sess, user = _authed_operator(request)
    return {"accounts": op.list_accounts(user)}


@router.get("/operator/accounts/{user_id}")
def op_account_detail(request: Request, user_id: str) -> dict:
    op = _operator_service()
    _sess, user = _authed_operator(request)
    return op.account_detail(user, user_id)


@router.post("/operator/accounts/{user_id}/pause")
def op_pause(request: Request, user_id: str) -> dict:
    op = _operator_service()
    sess, user = _authed_operator(request)
    _require_csrf(request, sess)
    op.pause_account(user, sess, user_id)
    return {"ok": True}


@router.post("/operator/accounts/{user_id}/reactivate")
def op_reactivate(request: Request, user_id: str) -> dict:
    op = _operator_service()
    sess, user = _authed_operator(request)
    _require_csrf(request, sess)
    op.reactivate_account(user, sess, user_id)
    return {"ok": True}


@router.post("/operator/accounts/{user_id}/recovery-invitation")
def op_recovery_invitation(request: Request, user_id: str) -> dict:
    op = _operator_service()
    sess, user = _authed_operator(request)
    _require_csrf(request, sess)
    _iid, code = op.issue_recovery_invitation(user, sess, user_id)
    return {"code": code}


@router.post("/operator/accounts/{user_id}/delete")
def op_delete(request: Request, user_id: str) -> dict:
    op = _operator_service()
    sess, user = _authed_operator(request)
    _require_csrf(request, sess)
    return op.delete_account(user, sess, user_id)