05/18 Enhanced codes and functionalities 5
This commit is contained in:
@@ -0,0 +1,388 @@
|
||||
"""
|
||||
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
|
||||
from datetime import datetime, timezone
|
||||
|
||||
import webauthn
|
||||
from webauthn.helpers.structs import (
|
||||
AuthenticatorSelectionCriteria,
|
||||
ResidentKeyRequirement,
|
||||
UserVerificationRequirement,
|
||||
AuthenticatorAttachment,
|
||||
)
|
||||
from webauthn.helpers.exceptions import InvalidCBORData, InvalidAuthenticatorResponse
|
||||
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__)
|
||||
|
||||
# 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).
|
||||
"""
|
||||
user = db.session.get(User, g.current_user_id)
|
||||
if not user:
|
||||
return jsonify({'error': 'User not found'}), 404
|
||||
|
||||
# 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,
|
||||
user_verification=UserVerificationRequirement.PREFERRED,
|
||||
authenticator_attachment=AuthenticatorAttachment.PLATFORM,
|
||||
),
|
||||
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(),
|
||||
require_user_verification=False,
|
||||
)
|
||||
except (InvalidCBORData, InvalidAuthenticatorResponse, Exception) as e:
|
||||
return jsonify({'error': f'Registration verification failed: {str(e)}'}), 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,
|
||||
user_verification=UserVerificationRequirement.PREFERRED,
|
||||
)
|
||||
|
||||
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,
|
||||
require_user_verification=False,
|
||||
)
|
||||
except Exception as e:
|
||||
AuditLog.log(
|
||||
user_id=user.id,
|
||||
action='webauthn.auth_failed',
|
||||
resource_type='webauthn_credential',
|
||||
resource_id=credential.id,
|
||||
detail=f'Passkey authentication failed: {str(e)[:200]}',
|
||||
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)
|
||||
|
||||
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
|
||||
Reference in New Issue
Block a user