flowdeck/app/routers/webauthn.py

"""FlowDeck — Passkeys / WebAuthn (v7.2.0).

Registration + passwordless login via the ``webauthn`` package (pinned in
requirements). Challenges live in a short-lived in-memory store (5 min,
single-process — same tradeoff as the SSE rooms). RP ID is derived from the
request host. See ``docs/V72_Enterprise_SCIM_2FA.md``.
"""
from __future__ import annotations

import secrets
import time

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

from app.auth.session import SessionManager
from app.db import get_conn

router = APIRouter(tags=["webauthn"], prefix="/auth/webauthn")

# key -> (challenge bytes, expires_at). key = f"reg:{user_id}" | f"login:{login}".
_challenges: dict[str, tuple[bytes, float]] = {}
_CHALLENGE_TTL = 300.0


def _require_lib():
    try:
        import webauthn  # noqa: F401
        return True
    except ImportError:
        return False


def _store_challenge(key: str, challenge: bytes) -> None:
    _challenges[key] = (challenge, time.time() + _CHALLENGE_TTL)


def _take_challenge(key: str) -> bytes | None:
    item = _challenges.pop(key, None)
    if not item:
        return None
    challenge, exp = item
    return challenge if exp > time.time() else None


def _rp(request: Request) -> tuple[str, str]:
    host = (request.url.hostname or "localhost").split(":")[0]
    return host, f"{request.url.scheme}://{request.headers.get('host', host)}"


def _session_user(request: Request) -> dict:
    user = SessionManager.decode_session(request.cookies.get("flowdeck_session", ""))
    if not user or not user.get("id"):
        raise HTTPException(401, "Authentication required")
    return user


@router.post("/register/begin")
def register_begin(request: Request):
    if not _require_lib():
        raise HTTPException(501, "WebAuthn library not installed")
    from webauthn import generate_registration_options, options_to_json
    user = _session_user(request)
    rp_id, _origin = _rp(request)
    with get_conn() as conn:
        existing = conn.execute("SELECT credential_id FROM webauthn_credentials WHERE user_id=?",
                                (user["id"],)).fetchall()
    from webauthn.helpers.structs import PublicKeyCredentialDescriptor
    exclude = [PublicKeyCredentialDescriptor(id=_b64url_to_bytes(r["credential_id"]))
               for r in existing]
    options = generate_registration_options(
        rp_id=rp_id, rp_name="FlowDeck", user_name=user.get("login", f"user{user['id']}"),
        user_id=str(user["id"]).encode(), exclude_credentials=exclude or None)
    _store_challenge(f"reg:{user['id']}", options.challenge)
    return JSONResponse(content=__import__("json").loads(options_to_json(options)))


@router.post("/register/finish")
def register_finish(request: Request, body: dict = Body(default={})):
    if not _require_lib():
        raise HTTPException(501, "WebAuthn library not installed")
    from webauthn import verify_registration_response
    user = _session_user(request)
    challenge = _take_challenge(f"reg:{user['id']}")
    if not challenge:
        raise HTTPException(400, "Challenge expired — begin again")
    rp_id, origin = _rp(request)
    try:
        verified = verify_registration_response(
            credential=body.get("credential") or {},
            expected_challenge=challenge, expected_rp_id=rp_id, expected_origin=origin,
            require_user_verification=False)
    except Exception as exc:  # noqa: BLE001 — invalid attestation → 400, never 500
        raise HTTPException(400, f"Registration rejected: {exc}") from None
    import base64
    cred_id = base64.urlsafe_b64encode(verified.credential_id).decode().rstrip("=")
    pubkey = base64.b64encode(bytes(verified.credential_public_key)).decode()
    with get_conn() as conn:
        try:
            cur = conn.execute(
                """INSERT INTO webauthn_credentials
                   (user_id, credential_id, public_key, sign_count, name)
                   VALUES (?,?,?,?,?)""",
                (user["id"], cred_id, pubkey, verified.sign_count,
                 str(body.get("name") or "Passkey")[:80]))
            conn.commit()
        except Exception:
            raise HTTPException(409, "Credential already registered") from None
        kid = cur.lastrowid
    return {"id": kid, "status": "registered"}


@router.post("/login/begin")
def login_begin(request: Request, body: dict = Body(default={})):
    if not _require_lib():
        raise HTTPException(501, "WebAuthn library not installed")
    from webauthn import generate_authentication_options, options_to_json
    login = (body.get("login") or "").strip()
    if not login:
        raise HTTPException(400, "login required")
    with get_conn() as conn:
        user = conn.execute("SELECT * FROM users WHERE login=?", (login,)).fetchone()
        if not user or not user["is_active"]:
            raise HTTPException(401, "Invalid credentials")
        creds = conn.execute("SELECT credential_id FROM webauthn_credentials WHERE user_id=?",
                             (user["id"],)).fetchall()
    if not creds:
        raise HTTPException(400, "No passkeys for this account")
    rp_id, _origin = _rp(request)
    from webauthn.helpers.structs import PublicKeyCredentialDescriptor
    allow = [PublicKeyCredentialDescriptor(id=_b64url_to_bytes(r["credential_id"]))
             for r in creds]
    options = generate_authentication_options(rp_id=rp_id, allow_credentials=allow)
    _store_challenge(f"login:{login}", options.challenge)
    return JSONResponse(content=__import__("json").loads(options_to_json(options)))


@router.post("/login/finish")
def login_finish(request: Request, body: dict = Body(default={})):
    if not _require_lib():
        raise HTTPException(501, "WebAuthn library not installed")
    from webauthn import verify_authentication_response
    login = (body.get("login") or "").strip()
    challenge = _take_challenge(f"login:{login}")
    if not login or not challenge:
        raise HTTPException(400, "Challenge expired — begin again")
    with get_conn() as conn:
        user = conn.execute("SELECT * FROM users WHERE login=?", (login,)).fetchone()
        if not user or not user["is_active"]:
            raise HTTPException(401, "Invalid credentials")
        stored = conn.execute("SELECT * FROM webauthn_credentials WHERE user_id=?",
                              (user["id"],)).fetchall()
    rp_id, origin = _rp(request)
    credential = body.get("credential") or {}
    cred_id = (credential.get("id") or "").rstrip("=")
    match = next((dict(r) for r in stored if r["credential_id"].rstrip("=") == cred_id), None)
    if not match:
        raise HTTPException(401, "Unknown credential")
    import base64
    try:
        verified = verify_authentication_response(
            credential=credential, expected_challenge=challenge,
            expected_origin=origin, expected_rp_id=rp_id,
            credential_public_key=base64.b64decode(match["public_key"]),
            credential_current_sign_count=match["sign_count"],
            require_user_verification=False)
    except Exception as exc:  # noqa: BLE001
        raise HTTPException(401, f"Authentication rejected: {exc}") from None
    with get_conn() as conn:
        conn.execute("UPDATE webauthn_credentials SET sign_count=? WHERE id=?",
                     (verified.new_sign_count, match["id"]))
        conn.execute("UPDATE users SET last_login=? WHERE id=?",
                     (str(time.time()), user["id"]))
        conn.commit()
        ud = dict(conn.execute("SELECT * FROM users WHERE id=?", (user["id"],)).fetchone())
    session = SessionManager.create_session(ud, request)
    response = JSONResponse({"status": "ok", "user": {"login": ud["login"]}})
    response.set_cookie("flowdeck_session", session, httponly=True,
                        max_age=86400 * 7, samesite="lax", path="/")
    return response


@router.get("/keys")
def list_keys(request: Request):
    user = _session_user(request)
    with get_conn() as conn:
        rows = conn.execute("SELECT id, name, sign_count, created_at FROM webauthn_credentials"
                            " WHERE user_id=? ORDER BY id", (user["id"],)).fetchall()
    return {"keys": [dict(r) for r in rows]}


@router.delete("/keys/{key_id}")
def delete_key(key_id: int, request: Request):
    user = _session_user(request)
    with get_conn() as conn:
        cur = conn.execute("DELETE FROM webauthn_credentials WHERE id=? AND user_id=?",
                           (key_id, user["id"]))
        conn.commit()
        if not cur.rowcount:
            raise HTTPException(404, "Key not found")
    return {"status": "deleted", "id": key_id}


def _b64url_to_bytes(data: str) -> bytes:
    import base64
    padded = data + "=" * (-len(data) % 4)
    return base64.urlsafe_b64decode(padded)


def reset_challenges() -> None:
    _challenges.clear()


__all__ = ["router", "reset_challenges", "secrets"]