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

"""WebAuthn Passkey authentication endpoints."""
import json
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_registration_options,
    verify_registration_response,
    generate_authentication_options,
    verify_authentication_response
)
from webauthn.helpers.structs import PublicKeyCredentialDescriptor, UserVerificationRequirement
from webauthn.helpers import bytes_to_base64url, base64url_to_bytes, options_to_json
import auth
import config

router = APIRouter()
limiter = Limiter(key_func=get_remote_address)

# 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:
        raise HTTPException(status_code=400, detail=str(e))

    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()),
    })
    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:
        raise HTTPException(status_code=400, detail=str(e))

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

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

@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}