nas-tools/dashboard/backend/routers/passkey.py

"""WebAuthn Passkey authentication endpoints."""

import json
import logging
import time
from datetime import timedelta

from fastapi import APIRouter, Depends, HTTPException, Request, Response
from fastapi.responses import JSONResponse
from slowapi import Limiter
from slowapi.util import get_remote_address
from webauthn import (
    generate_authentication_options,
    generate_registration_options,
    verify_authentication_response,
    verify_registration_response,
)
from webauthn.helpers import base64url_to_bytes, bytes_to_base64url, options_to_json
from webauthn.helpers.structs import PublicKeyCredentialDescriptor, UserVerificationRequirement

import auth_service as auth
import config

router = APIRouter()
limiter = Limiter(key_func=get_remote_address)
logger = logging.getLogger(__name__)

# In-memory challenge store with TTL
_challenges = {}


def _store_challenge(challenge: bytes):
    """Store a WebAuthn challenge with timestamp for TTL tracking."""
    _challenges[challenge] = time.time()
    # Clean up expired challenges (>5 minutes old)
    now = time.time()
    expired = [k for k, v in _challenges.items() if now - v > 300]
    for k in expired:
        _challenges.pop(k, None)


def _get_challenge(client_data_b64: str) -> bytes:
    """Extract and validate challenge from clientDataJSON."""
    if not client_data_b64:
        raise HTTPException(status_code=400, detail="Missing clientDataJSON")
    try:
        data = json.loads(base64url_to_bytes(client_data_b64).decode("utf-8"))
        chal_bytes = base64url_to_bytes(data.get("challenge", ""))
    except Exception:
        raise HTTPException(status_code=400, detail="Invalid clientDataJSON")

    entry = _challenges.pop(chal_bytes, None)
    if not entry or time.time() - entry > 300:
        raise HTTPException(status_code=400, detail="Challenge expired or invalid")
    return chal_bytes


@router.post("/passkey/register/options")
async def passkey_register_options(current_user=Depends(auth.get_current_user)):
    """Generate WebAuthn registration options for adding a new passkey."""
    existing = auth.load_passkey_credentials()
    exclude = [PublicKeyCredentialDescriptor(id=base64url_to_bytes(c["credential_id"])) for c in existing]
    options = generate_registration_options(
        rp_id=config.WEBAUTHN_RP_ID,
        rp_name="NAS Dashboard",
        user_id=current_user.username.encode(),
        user_name=current_user.username,
        user_display_name=current_user.username,
        exclude_credentials=exclude,
    )
    _store_challenge(options.challenge)
    return JSONResponse(content=json.loads(options_to_json(options)))


@router.post("/passkey/register/verify")
async def passkey_register_verify(request: Request, current_user=Depends(auth.get_current_user)):
    """Verify and save a new passkey registration."""
    body = await request.json()
    challenge = _get_challenge(body.get("response", {}).get("clientDataJSON", ""))
    try:
        verification = verify_registration_response(
            credential=body,
            expected_challenge=challenge,
            expected_rp_id=config.WEBAUTHN_RP_ID,
            expected_origin=config.WEBAUTHN_ORIGINS,
        )
    except Exception as e:
        logger.exception("Passkey registration verification failed")
        raise HTTPException(status_code=400, detail="Invalid passkey registration")

    auth.save_passkey_credential(
        {
            "credential_id": bytes_to_base64url(verification.credential_id),
            "public_key": bytes_to_base64url(verification.credential_public_key),
            "sign_count": verification.sign_count,
            "name": body.get("name", "Passkey"),
            "created_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
            "username": current_user.username,
            "role": current_user.role,
        }
    )
    return {"ok": True}


@router.post("/passkey/login/options")
@limiter.limit("10/minute")
async def passkey_login_options(request: Request):
    """Generate WebAuthn authentication options for passkey login."""
    creds = auth.load_passkey_credentials()
    if not creds:
        raise HTTPException(status_code=404, detail="No passkeys registered")

    allow = [PublicKeyCredentialDescriptor(id=base64url_to_bytes(c["credential_id"])) for c in creds]
    options = generate_authentication_options(
        rp_id=config.WEBAUTHN_RP_ID,
        allow_credentials=allow,
        user_verification=UserVerificationRequirement.PREFERRED,
    )
    _store_challenge(options.challenge)
    return JSONResponse(content=json.loads(options_to_json(options)))


@router.post("/passkey/login/verify")
@limiter.limit("10/minute")
async def passkey_login_verify(request: Request, response: Response):
    """Verify passkey authentication and issue tokens."""
    body = await request.json()
    challenge = _get_challenge(body.get("response", {}).get("clientDataJSON", ""))
    creds = auth.load_passkey_credentials()
    cred_id_b64 = body.get("id", "")
    matched = next((c for c in creds if c["credential_id"] == cred_id_b64), None)
    if not matched:
        raise HTTPException(status_code=400, detail="Unknown credential")

    try:
        verification = verify_authentication_response(
            credential=body,
            expected_challenge=challenge,
            expected_rp_id=config.WEBAUTHN_RP_ID,
            expected_origin=config.WEBAUTHN_ORIGINS,
            credential_public_key=base64url_to_bytes(matched["public_key"]),
            credential_current_sign_count=matched["sign_count"],
        )
    except Exception as e:
        logger.exception("Passkey authentication verification failed")
        raise HTTPException(status_code=400, detail="Invalid passkey authentication")

    # Update sign count
    matched["sign_count"] = verification.new_sign_count
    data = auth._load_auth_data()
    data["passkey_credentials"] = creds
    auth._save_auth_data(data)

    # Use stored username and role from passkey credential
    username = matched.get("username", config.ADMIN_USER)
    role = matched.get("role", "admin")

    access_token = auth.create_access_token(
        data={"sub": username, "role": role},
        expires_delta=timedelta(minutes=config.ACCESS_TOKEN_EXPIRE_MINUTES),
    )
    refresh_token = auth.create_refresh_token(data={"sub": username, "role": role})
    auth.set_auth_cookies(response, access_token, refresh_token)
    return {
        "access_token": access_token,
        "refresh_token": refresh_token,
        "token_type": "bearer",
        "user": username,
        "role": role,
    }


@router.get("/passkey/list")
async def passkey_list(current_user=Depends(auth.get_current_user)):
    """List all registered passkeys for the current user."""
    creds = auth.load_passkey_credentials()
    return {
        "passkeys": [
            {"id": c["credential_id"], "name": c.get("name", "Passkey"), "created_at": c.get("created_at", "")}
            for c in creds
        ]
    }


@router.post("/passkey/delete")
async def passkey_delete(request: Request, current_user=Depends(auth.get_current_user)):
    """Delete a specific passkey by credential ID."""
    body = await request.json()
    cred_id = body.get("id")
    data = auth._load_auth_data()
    creds = data.get("passkey_credentials", [])
    data["passkey_credentials"] = [c for c in creds if c["credential_id"] != cred_id]
    auth._save_auth_data(data)
    return {"ok": True}


@router.post("/passkey/clear")
async def passkey_clear(current_user=Depends(auth.get_current_user)):
    """Clear all passkeys for the current user."""
    auth.clear_passkey_credentials()
    return {"ok": True}