PassKeeper/app/routes/webauthn.py

"""
app/routes/webauthn.py — Passkey (WebAuthn) registration and authentication.

Zero-knowledge design:
  WebAuthn handles *server authentication* only. The vault key is still derived
  from the master password client-side — WebAuthn does not weaken the ZK model.

Flows:
  Registration (settings page, user must already be logged in):
    POST /api/webauthn/register/begin    → returns PublicKeyCredentialCreationOptions JSON
    POST /api/webauthn/register/complete → verifies attestation, stores credential

  Authentication (login page, passwordless):
    POST /api/webauthn/authenticate/begin    → returns PublicKeyCredentialRequestOptions JSON
    POST /api/webauthn/authenticate/complete → verifies assertion, returns tokens + enc_key_salt
      └── Client still prompts for master password to derive vault key.

  Management:
    GET    /api/webauthn/credentials        → list user's registered passkeys
    DELETE /api/webauthn/credentials/<id>   → remove a passkey
    PATCH  /api/webauthn/credentials/<id>   → rename a passkey

Challenge storage:
  Challenges are stored in the *Flask session* (server-side cookie) which is
  signed with SECRET_KEY. They expire when the browser session ends or after
  a configurable TTL. This is safe across Gunicorn workers because the cookie
  is stored client-side (signed, not encrypted — the challenge is not secret).
"""
import json
import logging
from datetime import datetime, timezone

import webauthn
from webauthn.helpers.structs import (
    AuthenticatorSelectionCriteria,
    ResidentKeyRequirement,
    UserVerificationRequirement,
    AuthenticatorAttachment,
)
from webauthn.helpers.exceptions import (
    InvalidCBORData,
    InvalidRegistrationResponse,
)
from flask import Blueprint, request, jsonify, g, session, current_app

from app import db, limiter, client_ip
from app.models.user import User
from app.models.webauthn_credential import WebAuthnCredential
from app.models.audit_log import AuditLog
from app.services.auth_service import require_jwt, generate_tokens

webauthn_bp = Blueprint('webauthn', __name__)

_log = logging.getLogger(__name__)

# Session key for the pending challenge bytes.
_REG_CHALLENGE_KEY = 'webauthn_reg_challenge'
_AUTH_CHALLENGE_KEY = 'webauthn_auth_challenge'
_AUTH_USER_ID_KEY = 'webauthn_auth_user_id'


def _rp_id():
    return current_app.config['WEBAUTHN_RP_ID']


def _rp_name():
    return current_app.config['WEBAUTHN_RP_NAME']


def _origins():
    return current_app.config['WEBAUTHN_ORIGINS']


# ── Registration ──────────────────────────────────────────────────────────────

@webauthn_bp.route('/register/begin', methods=['POST'])
@limiter.limit('10 per minute')
@require_jwt
def register_begin():
    """
    Generate PublicKeyCredentialCreationOptions for passkey registration.
    Requires an active JWT session (user must be logged in).

    Optional JSON body:
      attachment: "platform" (default) | "cross-platform"
        "platform"       → device biometrics (Touch ID, Face ID, Windows Hello)
        "cross-platform" → roaming authenticators (YubiKey, phone-as-key via QR)
    """
    user = db.session.get(User, g.current_user_id)
    if not user:
        return jsonify({'error': 'User not found'}), 404

    data = request.get_json(silent=True) or {}
    attachment_str = (data.get('attachment') or 'platform').strip().lower()

    if attachment_str == 'cross-platform':
        authenticator_attachment = AuthenticatorAttachment.CROSS_PLATFORM
    else:
        authenticator_attachment = AuthenticatorAttachment.PLATFORM

    # Collect existing credential IDs to exclude (prevent re-registering same key).
    existing = WebAuthnCredential.query.filter_by(user_id=user.id).all()
    exclude_credentials = [
        webauthn.helpers.structs.PublicKeyCredentialDescriptor(
            id=webauthn.base64url_to_bytes(cred.credential_id),
            transports=cred.get_transports() or [],
        )
        for cred in existing
    ]

    options = webauthn.generate_registration_options(
        rp_id=_rp_id(),
        rp_name=_rp_name(),
        user_id=str(user.id).encode(),
        user_name=user.email,
        user_display_name=user.email,
        authenticator_selection=AuthenticatorSelectionCriteria(
            resident_key=ResidentKeyRequirement.PREFERRED,
            # REQUIRED, not PREFERRED. A passkey here replaces BOTH the password
            # and the TOTP second factor, so the authenticator must actually
            # verify the human (biometric or PIN) rather than merely prove it is
            # present. Under PREFERRED an authenticator is free to skip that,
            # which reduced a full login to possession of an unlocked device.
            user_verification=UserVerificationRequirement.REQUIRED,
            authenticator_attachment=authenticator_attachment,
        ),
        exclude_credentials=exclude_credentials,
    )

    # Store challenge in signed session for verification in /complete.
    session[_REG_CHALLENGE_KEY] = webauthn.options_to_json(options)

    return jsonify(json.loads(webauthn.options_to_json(options))), 200


@webauthn_bp.route('/register/complete', methods=['POST'])
@limiter.limit('10 per minute')
@require_jwt
def register_complete():
    """
    Verify the authenticator's attestation response and persist the credential.
    """
    user = db.session.get(User, g.current_user_id)
    if not user:
        return jsonify({'error': 'User not found'}), 404

    options_json = session.pop(_REG_CHALLENGE_KEY, None)
    if not options_json:
        return jsonify({'error': 'No pending registration challenge. Call /register/begin first.'}), 400

    data = request.get_json(silent=True) or {}
    credential_name = (data.get('name') or 'Passkey').strip()[:128]

    # Remove 'name' before passing to py-webauthn — it doesn't expect this field.
    credential_data = {k: v for k, v in data.items() if k != 'name'}

    try:
        import json as _json
        options_obj = _json.loads(options_json)
        expected_challenge = webauthn.base64url_to_bytes(options_obj['challenge'])

        verification = webauthn.verify_registration_response(
            credential=credential_data,
            expected_challenge=expected_challenge,
            expected_rp_id=_rp_id(),
            expected_origin=_origins(),
            # Reject a credential created without user verification — otherwise
            # the REQUIRED hint above is only a request, not a guarantee.
            require_user_verification=True,
        )
    except (InvalidCBORData, InvalidRegistrationResponse) as e:
        # Never echo str(e) to the client: py-webauthn messages quote raw
        # attestation internals. Log the detail, return a generic message.
        _log.warning('[PassKeeper] passkey registration rejected: %s', e)
        return jsonify({'error': 'Could not verify this passkey. Please try again.'}), 400
    except Exception:
        _log.exception('[PassKeeper] passkey registration failed unexpectedly')
        return jsonify({'error': 'Could not verify this passkey. Please try again.'}), 400

    # Persist the new credential.
    import base64
    cred = WebAuthnCredential(
        user_id=user.id,
        credential_id=base64.urlsafe_b64encode(
            verification.credential_id
        ).rstrip(b'=').decode(),
        public_key=base64.urlsafe_b64encode(
            verification.credential_public_key
        ).rstrip(b'=').decode(),
        sign_count=verification.sign_count,
        aaguid=str(verification.aaguid) if verification.aaguid else None,
        name=credential_name,
    )
    # Store transport hints if available.
    if hasattr(verification, 'credential_device_type'):
        pass  # transport not exposed directly — set via request data if present
    transports = data.get('response', {}).get('transports', [])
    if transports:
        cred.set_transports(transports)

    db.session.add(cred)
    db.session.flush()

    AuditLog.log(
        user_id=user.id,
        action='webauthn.register',
        resource_type='webauthn_credential',
        resource_id=cred.id,
        detail=f'Registered passkey "{credential_name}" (id={cred.id})',
        ip_address=client_ip(),
    )
    db.session.commit()

    return jsonify({'message': 'Passkey registered successfully', 'credential': cred.to_dict()}), 201


# ── Authentication ────────────────────────────────────────────────────────────

@webauthn_bp.route('/authenticate/begin', methods=['POST'])
@limiter.limit('20 per minute')
def authenticate_begin():
    """
    Generate PublicKeyCredentialRequestOptions for passkey authentication.

    Accepts optional 'email' in the request body to pre-filter credentials.
    If email is omitted, all credentials for discoverable-credential flow are allowed.
    """
    data = request.get_json(silent=True) or {}
    email = (data.get('email') or '').strip().lower()

    allow_credentials = []
    user_id = None

    if email:
        user = User.query.filter_by(email=email).first()
        if user:
            user_id = user.id
            creds = WebAuthnCredential.query.filter_by(user_id=user.id).all()
            allow_credentials = [
                webauthn.helpers.structs.PublicKeyCredentialDescriptor(
                    id=webauthn.base64url_to_bytes(c.credential_id),
                    transports=c.get_transports() or [],
                )
                for c in creds
            ]

    options = webauthn.generate_authentication_options(
        rp_id=_rp_id(),
        allow_credentials=allow_credentials,
        # See register_begin — this assertion stands in for password + MFA.
        user_verification=UserVerificationRequirement.REQUIRED,
    )

    session[_AUTH_CHALLENGE_KEY] = webauthn.options_to_json(options)
    if user_id:
        session[_AUTH_USER_ID_KEY] = user_id

    return jsonify(json.loads(webauthn.options_to_json(options))), 200


@webauthn_bp.route('/authenticate/complete', methods=['POST'])
@limiter.limit('20 per minute')
def authenticate_complete():
    """
    Verify the authenticator's assertion response and issue JWT tokens.

    On success returns the same payload as /api/auth/login:
      { access_token, refresh_token, enc_key_salt }

    The client must still prompt for the master password to derive the vault key.
    MFA is not required after a successful WebAuthn assertion (the passkey IS the MFA).
    """
    import base64

    options_json = session.pop(_AUTH_CHALLENGE_KEY, None)
    stored_user_id = session.pop(_AUTH_USER_ID_KEY, None)

    if not options_json:
        return jsonify({'error': 'No pending authentication challenge. Call /authenticate/begin first.'}), 400

    data = request.get_json(silent=True) or {}

    # Identify the credential being used.
    raw_credential_id = data.get('id') or data.get('rawId', '')

    # Normalise to base64url without padding for DB lookup.
    cred_id_lookup = raw_credential_id.rstrip('=').replace('+', '-').replace('/', '_')

    credential = WebAuthnCredential.query.filter_by(
        credential_id=cred_id_lookup
    ).first()

    if not credential:
        return jsonify({'error': 'Passkey not recognised'}), 401

    user = db.session.get(User, credential.user_id)
    if not user:
        return jsonify({'error': 'User not found'}), 401

    # Guard: if we pre-filtered to a specific user, reject mismatches.
    if stored_user_id and credential.user_id != stored_user_id:
        return jsonify({'error': 'Credential does not match the requested user'}), 401

    try:
        import json as _json
        options_obj = _json.loads(options_json)
        expected_challenge = webauthn.base64url_to_bytes(options_obj['challenge'])

        verification = webauthn.verify_authentication_response(
            credential=data,
            expected_challenge=expected_challenge,
            expected_rp_id=_rp_id(),
            expected_origin=_origins(),
            credential_public_key=webauthn.base64url_to_bytes(credential.public_key),
            credential_current_sign_count=credential.sign_count,
            # Enforced, not merely requested: the assertion must carry the UV
            # flag or it is not sufficient to stand in for two factors.
            require_user_verification=True,
        )
    except Exception:
        _log.warning(
            '[PassKeeper] passkey assertion rejected for credential id=%s',
            credential.id, exc_info=True,
        )
        AuditLog.log(
            user_id=user.id,
            action='webauthn.auth_failed',
            resource_type='webauthn_credential',
            resource_id=credential.id,
            detail='Passkey authentication failed (assertion rejected)',
            ip_address=client_ip(),
        )
        db.session.commit()
        return jsonify({'error': 'Passkey authentication failed'}), 401

    # Update sign count (clone detection).
    credential.sign_count = verification.new_sign_count
    credential.last_used_at = datetime.now(timezone.utc).replace(tzinfo=None)

    # Issue tokens.
    tokens = generate_tokens(user.id, user.token_epoch)

    AuditLog.log(
        user_id=user.id,
        action='webauthn.auth_success',
        resource_type='webauthn_credential',
        resource_id=credential.id,
        detail=f'Passkey authentication succeeded (credential id={credential.id})',
        ip_address=client_ip(),
    )
    db.session.commit()

    return jsonify({
        'access_token': tokens['access_token'],
        'refresh_token': tokens['refresh_token'],
        'enc_key_salt': user.enc_key_salt,
        'webauthn': True,   # signals client to skip master-password server auth
    }), 200


# ── Credential management ─────────────────────────────────────────────────────

@webauthn_bp.route('/credentials', methods=['GET'])
@limiter.limit('30 per minute')
@require_jwt
def list_credentials():
    """List all passkeys registered by the current user."""
    creds = WebAuthnCredential.query.filter_by(user_id=g.current_user_id).all()
    return jsonify([c.to_dict() for c in creds]), 200


@webauthn_bp.route('/credentials/<int:cred_id>', methods=['PATCH'])
@limiter.limit('20 per minute')
@require_jwt
def rename_credential(cred_id):
    """Rename a passkey."""
    cred = WebAuthnCredential.query.filter_by(
        id=cred_id, user_id=g.current_user_id
    ).first()
    if not cred:
        return jsonify({'error': 'Credential not found'}), 404

    data = request.get_json(silent=True) or {}
    new_name = (data.get('name') or '').strip()[:128]
    if not new_name:
        return jsonify({'error': 'name is required'}), 400

    old_name = cred.name
    cred.name = new_name

    AuditLog.log(
        user_id=g.current_user_id,
        action='webauthn.rename',
        resource_type='webauthn_credential',
        resource_id=cred.id,
        detail=f'Renamed passkey (id={cred.id}) from "{old_name}" to "{new_name}"',
        ip_address=client_ip(),
    )
    db.session.commit()
    return jsonify(cred.to_dict()), 200


@webauthn_bp.route('/credentials/<int:cred_id>', methods=['DELETE'])
@limiter.limit('10 per minute')
@require_jwt
def delete_credential(cred_id):
    """Remove a registered passkey."""
    cred = WebAuthnCredential.query.filter_by(
        id=cred_id, user_id=g.current_user_id
    ).first()
    if not cred:
        return jsonify({'error': 'Credential not found'}), 404

    cred_name = cred.name
    db.session.delete(cred)
    db.session.flush()

    AuditLog.log(
        user_id=g.current_user_id,
        action='webauthn.delete',
        resource_type='webauthn_credential',
        resource_id=cred_id,
        detail=f'Deleted passkey "{cred_name}" (id={cred_id})',
        ip_address=client_ip(),
    )
    db.session.commit()
    return jsonify({'message': 'Passkey removed'}), 200