feat(mfa): #64 WebAuthn complet — enregistrement/clé/verify login + gestion clés dans Sécurité (backend+frontend+i18n+tests)
This commit is contained in:
+6
-1
@@ -34,9 +34,14 @@ OBSIGATE_ADMIN_PASSWORD=chab30
|
||||
# OBSIGATE_BACKUP_DIR=.obsigate-backup
|
||||
|
||||
# PDF (ROADMAP #74)
|
||||
# OBSIGATE_PDF_MAX_SIZE_MB=50 # PDFs more volumineux = texte non indexé
|
||||
# OBSIGATE_PDF_MAX_SIZE_MB=50 # PDFs plus volumineux = texte non indexé
|
||||
# OBSIGATE_PDF_EXTRACT_TIMEOUT=30 # secondes avant abandon de l'extraction
|
||||
|
||||
# WebAuthn / MFA (ROADMAP #64) — nécessaire hors localhost
|
||||
# OBSIGATE_WEBAUTHN_RP_ID=obsigate.example.com
|
||||
# OBSIGATE_WEBAUTHN_RP_NAME=ObsiGate
|
||||
# OBSIGATE_WEBAUTHN_ORIGINS=https://obsigate.example.com
|
||||
|
||||
# ── AI Provider Configuration ──
|
||||
# Définir au moins un provider pour activer les fonctionnalités AI dans l'éditeur
|
||||
|
||||
|
||||
+198
-5
@@ -5,7 +5,7 @@
|
||||
import logging
|
||||
import re
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, Request, Response
|
||||
from fastapi import APIRouter, Body, Depends, HTTPException, Request, Response
|
||||
from pydantic import BaseModel, validator
|
||||
|
||||
from backend.ratelimit import is_rate_limited
|
||||
@@ -147,12 +147,13 @@ async def login(body: LoginRequest, response: Response, request: Request):
|
||||
# Success — clear rate limits
|
||||
rl_record_success(client_ip)
|
||||
|
||||
# If MFA is enabled, don't issue token yet — require TOTP verification
|
||||
if user.get("mfa_enabled") and user.get("mfa_secret"):
|
||||
logger.info(f"User '{body.username}' login deferred — MFA required")
|
||||
# If MFA is enabled, don't issue token yet — require second factor
|
||||
if user.get("mfa_enabled"):
|
||||
method = "totp" if user.get("mfa_secret") else _preferred_mfa_method(user)
|
||||
logger.info(f"User '{body.username}' login deferred — MFA required ({method})")
|
||||
return {
|
||||
"mfa_required": True,
|
||||
"mfa_method": "totp",
|
||||
"mfa_method": method,
|
||||
"username": body.username,
|
||||
"remember_me": body.remember_me,
|
||||
}
|
||||
@@ -434,6 +435,196 @@ async def mfa_totp_disable(
|
||||
return {"mfa_enabled": False}
|
||||
|
||||
|
||||
# ── WebAuthn endpoints (ROADMAP #64) ─────────────────────────────────
|
||||
|
||||
def _preferred_mfa_method(user: dict) -> str:
|
||||
"""Which second factor to offer at login: webauthn when keys exist, else totp."""
|
||||
if user.get("webauthn_credentials"):
|
||||
return "webauthn"
|
||||
return "totp"
|
||||
|
||||
|
||||
class WebauthnRegisterRequest(BaseModel):
|
||||
credential: dict
|
||||
label: str = ""
|
||||
|
||||
|
||||
class WebauthnVerifyRequest(BaseModel):
|
||||
username: str
|
||||
credential: dict
|
||||
remember_me: bool = False
|
||||
|
||||
|
||||
class WebauthnRemoveRequest(BaseModel):
|
||||
credential_id: str
|
||||
password: str
|
||||
|
||||
|
||||
@router.post("/mfa/webauthn/register/options")
|
||||
async def mfa_webauthn_register_options(current_user=Depends(require_auth)):
|
||||
"""Start WebAuthn key enrolment — returns publicKey creation options for the browser."""
|
||||
from .webauthn_mfa import begin_registration
|
||||
|
||||
options = begin_registration(current_user["username"],
|
||||
current_user.get("display_name", ""))
|
||||
return {"options": options}
|
||||
|
||||
|
||||
@router.post("/mfa/webauthn/register")
|
||||
async def mfa_webauthn_register(
|
||||
req: WebauthnRegisterRequest,
|
||||
current_user=Depends(require_auth),
|
||||
):
|
||||
"""Verify the created credential, store it, and enable MFA if not already on.
|
||||
|
||||
Returns recovery codes when MFA is newly enabled (they were never issued).
|
||||
"""
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from .user_store import get_user, update_user
|
||||
from .webauthn_mfa import complete_registration
|
||||
|
||||
user = get_user(current_user["username"])
|
||||
try:
|
||||
record = complete_registration(current_user["username"], req.credential,
|
||||
label=req.label)
|
||||
except ValueError as e:
|
||||
raise HTTPException(400, str(e))
|
||||
except Exception as e:
|
||||
logger.warning(f"WebAuthn registration failed for {current_user['username']}: {e}")
|
||||
raise HTTPException(400, "Validation du credential WebAuthn échouée")
|
||||
|
||||
record["registered_at"] = datetime.now(timezone.utc).isoformat()
|
||||
creds = list(user.get("webauthn_credentials", []))
|
||||
creds = [c for c in creds if c.get("credential_id") != record["credential_id"]]
|
||||
creds.append(record)
|
||||
|
||||
updates: dict = {"webauthn_credentials": creds}
|
||||
issued_recovery: list[str] = []
|
||||
if not user.get("mfa_enabled"):
|
||||
issued_recovery = generate_recovery_codes()
|
||||
updates.update({
|
||||
"mfa_enabled": True,
|
||||
"mfa_method": "webauthn",
|
||||
"mfa_recovery_codes": [hash_recovery_code(c) for c in issued_recovery],
|
||||
})
|
||||
update_user(current_user["username"], updates)
|
||||
|
||||
logger.info(f"WebAuthn credential registered for user '{current_user['username']}' "
|
||||
f"({record['label']})")
|
||||
return {
|
||||
"ok": True,
|
||||
"credentials": user_credentials_response(creds),
|
||||
"mfa_enabled": True,
|
||||
"recovery_codes": issued_recovery,
|
||||
}
|
||||
|
||||
|
||||
def user_credentials_response(creds: list[dict]) -> list[dict]:
|
||||
from .webauthn_mfa import credentials_for_api
|
||||
return credentials_for_api(creds)
|
||||
|
||||
|
||||
@router.get("/mfa/webauthn/credentials")
|
||||
async def mfa_webauthn_list(current_user=Depends(require_auth)):
|
||||
from .user_store import get_user
|
||||
user = get_user(current_user["username"])
|
||||
return {"credentials": user_credentials_response(user.get("webauthn_credentials", []))}
|
||||
|
||||
|
||||
@router.post("/mfa/webauthn/credentials/remove")
|
||||
async def mfa_webauthn_remove(
|
||||
req: WebauthnRemoveRequest,
|
||||
current_user=Depends(require_auth),
|
||||
):
|
||||
"""Remove a WebAuthn key. Requires password. Disables MFA if no second factor remains."""
|
||||
from .user_store import get_user, update_user
|
||||
from .webauthn_mfa import clear_pending
|
||||
|
||||
user = get_user(current_user["username"])
|
||||
if not verify_password(req.password, user["password_hash"]):
|
||||
raise HTTPException(400, "Mot de passe incorrect")
|
||||
|
||||
creds = [c for c in user.get("webauthn_credentials", [])
|
||||
if c.get("credential_id") != req.credential_id]
|
||||
if len(creds) == len(user.get("webauthn_credentials", [])):
|
||||
raise HTTPException(404, "Credential inconnu")
|
||||
|
||||
updates: dict = {"webauthn_credentials": creds}
|
||||
if not creds and not user.get("mfa_secret"):
|
||||
updates.update({"mfa_enabled": False, "mfa_method": None, "mfa_recovery_codes": []})
|
||||
elif not creds and user.get("mfa_secret"):
|
||||
updates["mfa_method"] = "totp"
|
||||
update_user(current_user["username"], updates)
|
||||
clear_pending(current_user["username"])
|
||||
return {"ok": True, "credentials": user_credentials_response(creds),
|
||||
"mfa_enabled": bool(updates.get("mfa_enabled", user.get("mfa_enabled"))) and bool(creds or user.get("mfa_secret"))}
|
||||
|
||||
|
||||
@router.post("/mfa/webauthn/options")
|
||||
async def mfa_webauthn_login_options(body: dict = Body(...)):
|
||||
"""Unauthenticated: begin the login assertion for a user with registered keys.
|
||||
|
||||
Returns null options (mfa_method 'totp') when the user has no WebAuthn keys.
|
||||
"""
|
||||
username = str(body.get("username", ""))
|
||||
user = get_user(username)
|
||||
if not user or not user.get("mfa_enabled"):
|
||||
raise HTTPException(400, "MFA non activé pour cet utilisateur")
|
||||
|
||||
creds = user.get("webauthn_credentials", [])
|
||||
if not creds:
|
||||
return {"mfa_method": "totp", "options": None}
|
||||
|
||||
from .webauthn_mfa import begin_authentication
|
||||
options = begin_authentication(username, creds)
|
||||
if options is None:
|
||||
return {"mfa_method": "totp", "options": None}
|
||||
return {"mfa_method": "webauthn", "options": options}
|
||||
|
||||
|
||||
@router.post("/mfa/webauthn/verify")
|
||||
async def mfa_webauthn_verify(
|
||||
body: WebauthnVerifyRequest,
|
||||
response: Response,
|
||||
request: Request,
|
||||
):
|
||||
"""Unauthenticated: verify the WebAuthn assertion and issue JWT tokens."""
|
||||
from .user_store import get_user, update_user
|
||||
from .webauthn_mfa import complete_authentication
|
||||
|
||||
user = get_user(body.username)
|
||||
if not user:
|
||||
hash_password("dummy_timing_protection")
|
||||
raise HTTPException(401, "Identifiants invalides")
|
||||
if not user.get("mfa_enabled"):
|
||||
raise HTTPException(400, "MFA non activé pour cet utilisateur")
|
||||
|
||||
creds = user.get("webauthn_credentials", [])
|
||||
try:
|
||||
credential_id = body.credential.get("id", "")
|
||||
stored = next((c for c in creds if c.get("credential_id") == credential_id), None)
|
||||
if stored is None:
|
||||
raise ValueError("Credential non enregistré")
|
||||
new_count = complete_authentication(body.username, body.credential, stored)
|
||||
except ValueError as e:
|
||||
raise HTTPException(401, str(e))
|
||||
except Exception as e:
|
||||
logger.warning(f"WebAuthn verification failed for {body.username}: {e}")
|
||||
raise HTTPException(401, "Vérification WebAuthn échouée")
|
||||
|
||||
updated = [dict(c) for c in creds]
|
||||
for c in updated:
|
||||
if c.get("credential_id") == body.credential.get("id"):
|
||||
c["sign_count"] = new_count
|
||||
update_user(body.username, {"webauthn_credentials": updated})
|
||||
|
||||
client_ip = request.client.host if request.client else "unknown"
|
||||
rl_record_success(client_ip)
|
||||
logger.info(f"User '{body.username}' logged in via WebAuthn")
|
||||
return _issue_tokens(user, body.username, body.remember_me, response)
|
||||
|
||||
|
||||
@router.get("/mfa/status")
|
||||
async def mfa_status(current_user=Depends(require_auth)):
|
||||
"""Return current user's MFA status."""
|
||||
@@ -442,6 +633,8 @@ async def mfa_status(current_user=Depends(require_auth)):
|
||||
return {
|
||||
"mfa_enabled": user.get("mfa_enabled", False),
|
||||
"mfa_method": user.get("mfa_method"),
|
||||
"totp_enabled": bool(user.get("mfa_secret")),
|
||||
"webauthn_credentials": len(user.get("webauthn_credentials", [])),
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,184 @@
|
||||
# backend/auth/webauthn_mfa.py
|
||||
# WebAuthn support for MFA (ROADMAP #64): security keys / platform biometrics.
|
||||
# Thin wrapper over the `webauthn` library with an in-memory challenge store.
|
||||
#
|
||||
# Credentials are persisted in users.json under "webauthn_credentials":
|
||||
# [{ "credential_id": <b64url>, "public_key": <b64url>, "sign_count": int,
|
||||
# "transports": [...], "label": str, "registered_at": iso }]
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
import secrets
|
||||
import time
|
||||
from typing import Any
|
||||
|
||||
from webauthn import (
|
||||
generate_authentication_options,
|
||||
generate_registration_options,
|
||||
options_to_json,
|
||||
verify_authentication_response,
|
||||
verify_registration_response,
|
||||
)
|
||||
from webauthn.helpers import (
|
||||
base64url_to_bytes,
|
||||
bytes_to_base64url,
|
||||
parse_authentication_credential_json,
|
||||
parse_registration_credential_json,
|
||||
)
|
||||
from webauthn.helpers.structs import (
|
||||
AuthenticatorSelectionCriteria,
|
||||
ResidentKeyRequirement,
|
||||
UserVerificationRequirement,
|
||||
)
|
||||
|
||||
logger = logging.getLogger("obsigate.auth.webauthn")
|
||||
|
||||
# Challenge lifetime: clients have 3 minutes to complete the ceremony.
|
||||
CHALLENGE_TTL_SECONDS = 180
|
||||
|
||||
# In-memory pending challenges: key -> (challenge_bytes, expires_at)
|
||||
_pending: dict[str, tuple[bytes, float]] = {}
|
||||
|
||||
|
||||
def rp_id() -> str:
|
||||
return os.environ.get("OBSIGATE_WEBAUTHN_RP_ID", "localhost")
|
||||
|
||||
|
||||
def rp_name() -> str:
|
||||
return os.environ.get("OBSIGATE_WEBAUTHN_RP_NAME", "ObsiGate")
|
||||
|
||||
|
||||
def expected_origins() -> list[str]:
|
||||
raw = os.environ.get("OBSIGATE_WEBAUTHN_ORIGINS", "http://localhost")
|
||||
return [o.strip() for o in raw.split(",") if o.strip()]
|
||||
|
||||
|
||||
def _prune_expired() -> None:
|
||||
now = time.time()
|
||||
for key in [k for k, (_, exp) in _pending.items() if exp < now]:
|
||||
_pending.pop(key, None)
|
||||
|
||||
|
||||
def _store_challenge(key: str) -> bytes:
|
||||
_prune_expired()
|
||||
challenge = secrets.token_bytes(32)
|
||||
_pending[key] = (challenge, time.time() + CHALLENGE_TTL_SECONDS)
|
||||
return challenge
|
||||
|
||||
|
||||
def _take_challenge(key: str) -> bytes | None:
|
||||
"""Pop a challenge (single-use). Returns None if missing/expired."""
|
||||
_prune_expired()
|
||||
entry = _pending.pop(key, None)
|
||||
return entry[0] if entry else None
|
||||
|
||||
|
||||
def clear_pending(username: str) -> None:
|
||||
"""Drop all pending challenges for a user (e.g. after enable/disable)."""
|
||||
for key in [k for k in _pending if k.startswith(f"{username}:")]:
|
||||
_pending.pop(key, None)
|
||||
|
||||
|
||||
# ── Registration (enrol a key in settings) ─────────────────────────────
|
||||
|
||||
def begin_registration(username: str, display_name: str) -> dict:
|
||||
options = generate_registration_options(
|
||||
rp_id=rp_id(),
|
||||
rp_name=rp_name(),
|
||||
user_name=username,
|
||||
user_display_name=display_name or username,
|
||||
challenge=_store_challenge(f"{username}:register"),
|
||||
authenticator_selection=AuthenticatorSelectionCriteria(
|
||||
resident_key=ResidentKeyRequirement.PREFERRED,
|
||||
user_verification=UserVerificationRequirement.PREFERRED,
|
||||
),
|
||||
)
|
||||
return _finalize_options(options)
|
||||
|
||||
|
||||
def complete_registration(username: str, credential_json: dict[str, Any],
|
||||
label: str = "") -> dict:
|
||||
challenge = _take_challenge(f"{username}:register")
|
||||
if challenge is None:
|
||||
raise ValueError("Session d'enregistrement expirée — recommencez")
|
||||
|
||||
credential = parse_registration_credential_json(credential_json)
|
||||
verification = verify_registration_response(
|
||||
credential=credential,
|
||||
expected_challenge=challenge,
|
||||
expected_rp_id=rp_id(),
|
||||
expected_origin=expected_origins(),
|
||||
)
|
||||
|
||||
transports = credential.response.transports or []
|
||||
label = (label or str(credential_json.get("label") or "")).strip() or "Security key"
|
||||
record = {
|
||||
"credential_id": bytes_to_base64url(verification.credential_id),
|
||||
"public_key": bytes_to_base64url(verification.credential_public_key),
|
||||
"sign_count": int(verification.sign_count),
|
||||
"transports": [str(t) for t in transports],
|
||||
"label": label[:60],
|
||||
}
|
||||
return record
|
||||
|
||||
|
||||
# ── Authentication (assertion at login) ────────────────────────────────
|
||||
|
||||
def begin_authentication(username: str, credentials: list[dict]) -> dict | None:
|
||||
if not credentials:
|
||||
return None
|
||||
from webauthn.helpers.structs import PublicKeyCredentialDescriptor
|
||||
|
||||
allow = [
|
||||
PublicKeyCredentialDescriptor(id=base64url_to_bytes(c["credential_id"]))
|
||||
for c in credentials
|
||||
]
|
||||
options = generate_authentication_options(
|
||||
rp_id=rp_id(),
|
||||
challenge=_store_challenge(f"{username}:login"),
|
||||
allow_credentials=allow,
|
||||
)
|
||||
return _finalize_options(options)
|
||||
|
||||
|
||||
def complete_authentication(
|
||||
username: str,
|
||||
credential_json: dict[str, Any],
|
||||
stored: dict,
|
||||
) -> int:
|
||||
"""Verify an assertion. Returns the new sign_count. Raises ValueError on failure."""
|
||||
challenge = _take_challenge(f"{username}:login")
|
||||
if challenge is None:
|
||||
raise ValueError("Session expirée — rechargez la page")
|
||||
|
||||
credential = parse_authentication_credential_json(credential_json)
|
||||
verification = verify_authentication_response(
|
||||
credential=credential,
|
||||
expected_challenge=challenge,
|
||||
expected_rp_id=rp_id(),
|
||||
expected_origin=expected_origins(),
|
||||
credential_public_key=base64url_to_bytes(stored["public_key"]),
|
||||
credential_current_sign_count=int(stored.get("sign_count", 0)),
|
||||
)
|
||||
return int(verification.new_sign_count)
|
||||
|
||||
|
||||
def credentials_for_api(credentials: list[dict]) -> list[dict]:
|
||||
"""Sanitized view for the settings UI (no public keys)."""
|
||||
return [
|
||||
{
|
||||
"credential_id": c.get("credential_id"),
|
||||
"label": c.get("label", "Security key"),
|
||||
"transports": c.get("transports", []),
|
||||
"registered_at": c.get("registered_at"),
|
||||
}
|
||||
for c in credentials
|
||||
]
|
||||
|
||||
|
||||
def _finalize_options(options) -> dict:
|
||||
import json
|
||||
|
||||
return json.loads(options_to_json(options))
|
||||
@@ -14,4 +14,5 @@ weasyprint>=60.0
|
||||
httpx>=0.27.0
|
||||
pypdf>=4.0
|
||||
pyotp>=2.10.0
|
||||
webauthn==2.6.0
|
||||
psutil>=5.9
|
||||
|
||||
+268
-2
@@ -179,6 +179,57 @@ const AuthManager = {
|
||||
return data.user;
|
||||
},
|
||||
|
||||
// ── WebAuthn (ROADMAP #64) ─────────────────────────────────────────
|
||||
|
||||
async webauthnLoginOptions(username) {
|
||||
const resp = await fetch("/api/auth/mfa/webauthn/options", {
|
||||
method: "POST",
|
||||
headers: { "Content-Type": "application/json" },
|
||||
credentials: "include",
|
||||
body: JSON.stringify({ username }),
|
||||
});
|
||||
if (!resp.ok) throw new Error((await resp.json()).detail || "WebAuthn indisponible");
|
||||
return await resp.json();
|
||||
},
|
||||
|
||||
async verifyWebauthn(username, credential, rememberMe) {
|
||||
const response = await fetch("/api/auth/mfa/webauthn/verify", {
|
||||
method: "POST",
|
||||
headers: { "Content-Type": "application/json" },
|
||||
credentials: "include",
|
||||
body: JSON.stringify({ username, credential, remember_me: rememberMe || false }),
|
||||
});
|
||||
if (!response.ok) {
|
||||
const err = await response.json();
|
||||
throw new Error(err.detail || "Vérification WebAuthn échouée");
|
||||
}
|
||||
const data = await response.json();
|
||||
this.saveToken(data);
|
||||
return data.user;
|
||||
},
|
||||
|
||||
async webauthnRegisterOptions() {
|
||||
return await api("/api/auth/mfa/webauthn/register/options", { method: "POST" });
|
||||
},
|
||||
|
||||
async webauthnRegister(credential, label) {
|
||||
return await api("/api/auth/mfa/webauthn/register", {
|
||||
method: "POST",
|
||||
body: JSON.stringify({ credential, label }),
|
||||
});
|
||||
},
|
||||
|
||||
async webauthnCredentials() {
|
||||
return await api("/api/auth/mfa/webauthn/credentials");
|
||||
},
|
||||
|
||||
async webauthnRemove(credentialId, password) {
|
||||
return await api("/api/auth/mfa/webauthn/credentials/remove", {
|
||||
method: "POST",
|
||||
body: JSON.stringify({ credential_id: credentialId, password }),
|
||||
});
|
||||
},
|
||||
|
||||
// ── MFA Setup API calls ──────────────────────────────────────────
|
||||
|
||||
async getMfaStatus() {
|
||||
@@ -376,7 +427,112 @@ async function _onLoginSuccess() {
|
||||
// MFA Challenge UI (TOTP code input during login)
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
function showMfaChallenge(username, rememberMe, loginBtn, loginErrorEl) {
|
||||
// base64url <-> ArrayBuffer helpers for WebAuthn (ROADMAP #64)
|
||||
function _b64urlToBuf(s) {
|
||||
const pad = "=".repeat((4 - (s.length % 4)) % 4);
|
||||
const b = (s + pad).replace(/-/g, "+").replace(/_/g, "/");
|
||||
const raw = atob(b);
|
||||
return Uint8Array.from(raw, (c) => c.charCodeAt(0));
|
||||
}
|
||||
|
||||
function _bufToB64url(buf) {
|
||||
const bytes = new Uint8Array(buf);
|
||||
let s = "";
|
||||
for (const c of bytes) s += String.fromCharCode(c);
|
||||
return btoa(s).replace(/\+/g, "-").replace(/\//g, "_").replace(/=+$/, "");
|
||||
}
|
||||
|
||||
async function runWebauthnCeremony(optionsJson) {
|
||||
const pkOptions = {
|
||||
challenge: _b64urlToBuf(optionsJson.challenge),
|
||||
timeout: optionsJson.timeout || 60000,
|
||||
rpId: optionsJson.rpId,
|
||||
userVerification: optionsJson.userVerification || "preferred",
|
||||
allowCredentials: (optionsJson.allowCredentials || []).map((c) => ({
|
||||
type: c.type || "public-key",
|
||||
id: _b64urlToBuf(c.id),
|
||||
transports: c.transports,
|
||||
})),
|
||||
};
|
||||
const cred = await navigator.credentials.get({ publicKey: pkOptions });
|
||||
if (!cred) throw new Error(t("mfa.webauthn_cancelled"));
|
||||
return {
|
||||
id: cred.id,
|
||||
rawId: _bufToB64url(cred.rawId),
|
||||
type: cred.type,
|
||||
response: {
|
||||
clientDataJSON: _bufToB64url(cred.response.clientDataJSON),
|
||||
authenticatorData: _bufToB64url(cred.response.authenticatorData),
|
||||
signature: _bufToB64url(cred.response.signature),
|
||||
userHandle: cred.response.userHandle ? _bufToB64url(cred.response.userHandle) : "",
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
async function runWebauthnRegistration() {
|
||||
const resp = await AuthManager.webauthnRegisterOptions();
|
||||
const o = resp.options;
|
||||
const pkOptions = {
|
||||
rp: { id: o.rpId || o.rp.id, name: o.rp.name },
|
||||
challenge: _b64urlToBuf(o.challenge),
|
||||
user: {
|
||||
id: _b64urlToBuf(o.user.id),
|
||||
name: o.user.name,
|
||||
displayName: o.user.displayName,
|
||||
},
|
||||
pubKeyCredParams: o.pubKeyCredParams,
|
||||
timeout: o.timeout || 60000,
|
||||
authenticatorSelection: o.authenticatorSelection || undefined,
|
||||
attestation: "none",
|
||||
};
|
||||
const cred = await navigator.credentials.create({ publicKey: pkOptions });
|
||||
if (!cred) throw new Error(t("mfa.webauthn_cancelled"));
|
||||
const credential = {
|
||||
id: cred.id,
|
||||
rawId: _bufToB64url(cred.rawId),
|
||||
type: cred.type,
|
||||
response: {
|
||||
clientDataJSON: _bufToB64url(cred.response.clientDataJSON),
|
||||
attestationObject: _bufToB64url(cred.response.attestationObject),
|
||||
},
|
||||
};
|
||||
if (cred.response.getTransports) {
|
||||
credential.response.transports = Array.from(cred.response.getTransports());
|
||||
}
|
||||
return credential;
|
||||
}
|
||||
|
||||
function _startWebauthnLogin(mfaSection, username, rememberMe) {
|
||||
const btn = mfaSection.querySelector("#mfa-webauthn-btn");
|
||||
const errorEl = mfaSection.querySelector("#mfa-error");
|
||||
if (btn) btn.disabled = true;
|
||||
if (errorEl) errorEl.classList.add("hidden");
|
||||
|
||||
(async () => {
|
||||
try {
|
||||
const resp = await AuthManager.webauthnLoginOptions(username);
|
||||
if (resp.mfa_method !== "webauthn" || !resp.options) {
|
||||
throw new Error(t("mfa.webauthn_no_key"));
|
||||
}
|
||||
const credential = await runWebauthnCeremony(resp.options);
|
||||
await AuthManager.verifyWebauthn(username, credential, rememberMe);
|
||||
mfaSection.classList.add("hidden");
|
||||
const loginForm = document.getElementById("login-form");
|
||||
if (loginForm) loginForm.classList.remove("hidden");
|
||||
await _onLoginSuccess();
|
||||
} catch (err) {
|
||||
if (errorEl) {
|
||||
errorEl.textContent = err.message || String(err);
|
||||
errorEl.classList.remove("hidden");
|
||||
}
|
||||
} finally {
|
||||
if (btn) btn.disabled = false;
|
||||
}
|
||||
})();
|
||||
}
|
||||
|
||||
|
||||
function showMfaChallenge(username, rememberMe, loginBtn, loginErrorEl, mfaMethod) {
|
||||
const loginBox = document.querySelector(".login-box");
|
||||
if (!loginBox) return;
|
||||
|
||||
@@ -393,6 +549,37 @@ function showMfaChallenge(username, rememberMe, loginBtn, loginErrorEl) {
|
||||
loginBox.appendChild(mfaSection);
|
||||
}
|
||||
|
||||
// WebAuthn second factor: key prompt instead of TOTP code input
|
||||
if (mfaMethod === "webauthn") {
|
||||
mfaSection.innerHTML = `
|
||||
<div class="mfa-icon">🔑</div>
|
||||
<h3>${t('mfa.title')}</h3>
|
||||
<p class="mfa-subtitle">${t('mfa.webauthn_prompt')}</p>
|
||||
<p class="mfa-error hidden" id="mfa-error"></p>
|
||||
<button type="button" class="btn-login" id="mfa-webauthn-btn">
|
||||
<span class="btn-text">${t('mfa.webauthn_btn')}</span>
|
||||
</button>
|
||||
<div class="mfa-actions">
|
||||
<button type="button" class="mfa-link-btn" id="mfa-use-recovery">${t('mfa.use_recovery')}</button>
|
||||
<button type="button" class="mfa-link-btn" id="mfa-back-login">${t('mfa.back_to_login')}</button>
|
||||
</div>
|
||||
`;
|
||||
mfaSection.classList.remove("hidden");
|
||||
document.getElementById("mfa-webauthn-btn").addEventListener("click", () => {
|
||||
_startWebauthnLogin(mfaSection, username, rememberMe);
|
||||
});
|
||||
document.getElementById("mfa-use-recovery").addEventListener("click", () => {
|
||||
showRecoveryChallenge(username, rememberMe, loginForm, mfaSection);
|
||||
});
|
||||
document.getElementById("mfa-back-login").addEventListener("click", () => {
|
||||
mfaSection.classList.add("hidden");
|
||||
if (loginForm) loginForm.classList.remove("hidden");
|
||||
});
|
||||
// Auto-start the ceremony — the browser shows its own dialog
|
||||
_startWebauthnLogin(mfaSection, username, rememberMe);
|
||||
return;
|
||||
}
|
||||
|
||||
mfaSection.innerHTML = `
|
||||
<div class="mfa-icon">🔐</div>
|
||||
<h3>${t('mfa.title')}</h3>
|
||||
@@ -565,7 +752,7 @@ function initLoginForm() {
|
||||
const result = await AuthManager.login(username, password, rememberMe);
|
||||
// Check if MFA is required
|
||||
if (result && result.mfa_required) {
|
||||
showMfaChallenge(result.username, rememberMe, btn, errorEl);
|
||||
showMfaChallenge(result.username, rememberMe, btn, errorEl, result.mfa_method);
|
||||
return;
|
||||
}
|
||||
// Normal login success
|
||||
@@ -858,6 +1045,85 @@ async function initMfaSettings() {
|
||||
_startMfaSetup();
|
||||
});
|
||||
}
|
||||
|
||||
// WebAuthn security keys section (ROADMAP #64)
|
||||
_renderWebauthnSection(area);
|
||||
}
|
||||
|
||||
|
||||
async function _renderWebauthnSection(container) {
|
||||
if (!container || !window.PublicKeyCredential) return;
|
||||
|
||||
let keys = [];
|
||||
try {
|
||||
const resp = await AuthManager.webauthnCredentials();
|
||||
keys = resp.credentials || [];
|
||||
} catch (e) {
|
||||
return; // auth disabled or endpoint unreachable — hide section
|
||||
}
|
||||
|
||||
let section = document.getElementById("webauthn-settings");
|
||||
if (!section) {
|
||||
section = document.createElement("div");
|
||||
section.id = "webauthn-settings";
|
||||
section.className = "webauthn-settings";
|
||||
container.appendChild(section);
|
||||
}
|
||||
|
||||
const listHtml = keys.length
|
||||
? `<ul class="webauthn-key-list">${keys.map((k) => `
|
||||
<li class="webauthn-key-item">
|
||||
<span class="webauthn-key-label">🔑 ${k.label || "Security key"}</span>
|
||||
<span class="webauthn-key-meta">${(k.transports || []).join(", ") || "—"}</span>
|
||||
<button class="config-btn-secondary config-btn-sm webauthn-key-remove"
|
||||
data-id="${k.credential_id}">${t("mfa.webauthn_remove")}</button>
|
||||
</li>`).join("")}</ul>`
|
||||
: `<p class="mfa-info-text">${t("mfa.webauthn_none")}</p>`;
|
||||
|
||||
section.innerHTML = `
|
||||
<h4 class="webauthn-title">${t("mfa.webauthn_title")}</h4>
|
||||
<p class="mfa-info-text">${t("mfa.webauthn_desc")}</p>
|
||||
${listHtml}
|
||||
<div class="mfa-recovery-actions">
|
||||
<button class="config-btn-primary" id="webauthn-add-btn">${t("mfa.webauthn_add")}</button>
|
||||
</div>
|
||||
<p class="mfa-error hidden" id="webauthn-error"></p>
|
||||
<div id="webauthn-flow-area"></div>
|
||||
`;
|
||||
|
||||
const errEl = section.querySelector("#webauthn-error");
|
||||
document.getElementById("webauthn-add-btn").addEventListener("click", async () => {
|
||||
errEl.classList.add("hidden");
|
||||
try {
|
||||
const credential = await runWebauthnRegistration();
|
||||
const label = prompt(t("mfa.webauthn_label_prompt"), "Ma clé");
|
||||
const result = await AuthManager.webauthnRegister(credential, label || "Security key");
|
||||
if (result.recovery_codes && result.recovery_codes.length) {
|
||||
_showRecoveryCodes(result.recovery_codes);
|
||||
} else {
|
||||
showToast(t("mfa.webauthn_added"), "success");
|
||||
}
|
||||
initMfaSettings();
|
||||
} catch (err) {
|
||||
errEl.textContent = err.message || String(err);
|
||||
errEl.classList.remove("hidden");
|
||||
}
|
||||
});
|
||||
|
||||
section.querySelectorAll(".webauthn-key-remove").forEach((btn) => {
|
||||
btn.addEventListener("click", async () => {
|
||||
const password = prompt(t("mfa.webauthn_remove_confirm"));
|
||||
if (password === null) return;
|
||||
try {
|
||||
await AuthManager.webauthnRemove(btn.dataset.id, password);
|
||||
showToast(t("mfa.webauthn_removed"), "success");
|
||||
initMfaSettings();
|
||||
} catch (err) {
|
||||
errEl.textContent = err.message || String(err);
|
||||
errEl.classList.remove("hidden");
|
||||
}
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -1581,7 +1581,6 @@
|
||||
"mfa.disable_confirm_btn": "Disable 2FA",
|
||||
"mfa.disabled_success": "2FA has been disabled.",
|
||||
"mfa.fill_all_fields": "Please fill in all fields.",
|
||||
|
||||
"bookslm.title": "BooksLM",
|
||||
"bookslm.files_indexed": "{count} files indexed",
|
||||
"bookslm.chars_loaded": "{chars} chars loaded",
|
||||
@@ -1601,5 +1600,18 @@
|
||||
"palette.bookslm_open": "BooksLM: Open for current directory",
|
||||
"palette.bookslm_new": "BooksLM: New conversation",
|
||||
"fab.open": "Open AI assistant",
|
||||
"fab.close": "Close AI assistant"
|
||||
"fab.close": "Close AI assistant",
|
||||
"mfa.webauthn_title": "Security keys (WebAuthn)",
|
||||
"mfa.webauthn_desc": "Authenticate with a physical key (YubiKey) or your device biometrics (Windows Hello, Touch ID).",
|
||||
"mfa.webauthn_none": "No registered keys.",
|
||||
"mfa.webauthn_add": "Add a security key",
|
||||
"mfa.webauthn_label_prompt": "Key name (e.g. Pocket YubiKey)",
|
||||
"mfa.webauthn_added": "Security key registered.",
|
||||
"mfa.webauthn_removed": "Security key removed.",
|
||||
"mfa.webauthn_remove": "Remove",
|
||||
"mfa.webauthn_remove_confirm": "Enter your password to remove this key:",
|
||||
"mfa.webauthn_prompt": "Present your security key or confirm with Windows Hello.",
|
||||
"mfa.webauthn_btn": "Verify with my key",
|
||||
"mfa.webauthn_cancelled": "WebAuthn ceremony cancelled.",
|
||||
"mfa.webauthn_no_key": "No security key registered for this account."
|
||||
}
|
||||
|
||||
@@ -1581,7 +1581,6 @@
|
||||
"mfa.disable_confirm_btn": "Désactiver la 2FA",
|
||||
"mfa.disabled_success": "La 2FA a été désactivée.",
|
||||
"mfa.fill_all_fields": "Veuillez remplir tous les champs.",
|
||||
|
||||
"bookslm.title": "BooksLM",
|
||||
"bookslm.files_indexed": "{count} fichiers indexés",
|
||||
"bookslm.chars_loaded": "{chars} caractères chargés",
|
||||
@@ -1601,5 +1600,18 @@
|
||||
"palette.bookslm_open": "BooksLM: Ouvrir pour le répertoire courant",
|
||||
"palette.bookslm_new": "BooksLM: Nouvelle conversation",
|
||||
"fab.open": "Ouvrir l'assistant AI",
|
||||
"fab.close": "Fermer l'assistant AI"
|
||||
"fab.close": "Fermer l'assistant AI",
|
||||
"mfa.webauthn_title": "Clés de sécurité (WebAuthn)",
|
||||
"mfa.webauthn_desc": "Authentifiez-vous avec une clé physique (YubiKey) ou la biométrie de votre appareil (Windows Hello, Touch ID).",
|
||||
"mfa.webauthn_none": "Aucune clé enregistrée.",
|
||||
"mfa.webauthn_add": "Ajouter une clé de sécurité",
|
||||
"mfa.webauthn_label_prompt": "Nom de la clé (ex : YubiKey de poche)",
|
||||
"mfa.webauthn_added": "Clé de sécurité enregistrée.",
|
||||
"mfa.webauthn_removed": "Clé de sécurité supprimée.",
|
||||
"mfa.webauthn_remove": "Retirer",
|
||||
"mfa.webauthn_remove_confirm": "Entrez votre mot de passe pour retirer cette clé :",
|
||||
"mfa.webauthn_prompt": "Présentez votre clé de sécurité ou confirmez avec Windows Hello.",
|
||||
"mfa.webauthn_btn": "Valider avec ma clé",
|
||||
"mfa.webauthn_cancelled": "Cérémonie WebAuthn annulée.",
|
||||
"mfa.webauthn_no_key": "Aucune clé de sécurité enregistrée pour ce compte."
|
||||
}
|
||||
|
||||
@@ -9000,6 +9000,18 @@ body.popup-mode .content-area {
|
||||
}
|
||||
.mfa-recovery-actions { display: flex; gap: 8px; margin-top: 12px; }
|
||||
|
||||
/* WebAuthn security keys (ROADMAP #64) */
|
||||
.webauthn-settings { margin-top: 20px; padding-top: 16px; border-top: 1px solid var(--border, #333); }
|
||||
.webauthn-title { margin: 0 0 6px; color: var(--text, #fff); font-size: 0.95rem; }
|
||||
.webauthn-key-list { list-style: none; margin: 10px 0; padding: 0; display: flex; flex-direction: column; gap: 6px; }
|
||||
.webauthn-key-item {
|
||||
display: flex; align-items: center; gap: 10px; padding: 8px 10px;
|
||||
background: var(--surface2, #1a1a2e); border-radius: 6px;
|
||||
}
|
||||
.webauthn-key-label { flex: 1; font-size: 0.9rem; color: var(--text, #fff); }
|
||||
.webauthn-key-meta { font-size: 0.75rem; color: var(--text-muted, #888); }
|
||||
.config-btn-sm { padding: 4px 10px; font-size: 0.78rem; }
|
||||
|
||||
/* MFA Disable card */
|
||||
.mfa-disable-card {
|
||||
padding: 16px; border-radius: 10px;
|
||||
|
||||
@@ -0,0 +1,332 @@
|
||||
# tests/test_webauthn.py
|
||||
# WebAuthn MFA tests (ROADMAP #64).
|
||||
# Uses a virtual authenticator (EC P-256, packed-free 'none' attestation, raw
|
||||
# CBOR via cbor2) to exercise the real verification path end-to-end.
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import struct
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
import cbor2
|
||||
import pytest
|
||||
from cryptography.hazmat.primitives.asymmetric import ec
|
||||
from webauthn.helpers import bytes_to_base64url
|
||||
|
||||
|
||||
def _b64url(data: bytes) -> str:
|
||||
return bytes_to_base64url(data)
|
||||
|
||||
|
||||
class VirtualAuthenticator:
|
||||
"""Minimal WebAuthn authenticator: generates a P-256 key, produces
|
||||
'none'-attestation registration responses and ES256 assertion responses."""
|
||||
|
||||
RP_ID = "localhost"
|
||||
ORIGIN = "http://localhost"
|
||||
|
||||
def __init__(self):
|
||||
self.key = ec.generate_private_key(ec.SECP256R1())
|
||||
self.credential_id = os.urandom(32)
|
||||
self.sign_count = 0
|
||||
|
||||
# ── COSE public key (ES256) ──
|
||||
def _cose_key(self) -> bytes:
|
||||
pub = self.key.public_key().public_numbers()
|
||||
x = pub.x.to_bytes(32, "big")
|
||||
y = pub.y.to_bytes(32, "big")
|
||||
return cbor2.dumps({1: 2, 3: -7, -1: 1, -2: x, -3: y}, canonical=True)
|
||||
|
||||
def _rp_id_hash(self) -> bytes:
|
||||
return hashlib.sha256(self.RP_ID.encode()).digest()
|
||||
|
||||
def _client_data(self, typ: str, challenge_b64: str) -> bytes:
|
||||
return json.dumps({
|
||||
"type": typ,
|
||||
"challenge": challenge_b64,
|
||||
"origin": self.ORIGIN,
|
||||
"crossOrigin": False,
|
||||
}).encode()
|
||||
|
||||
def make_registration(self, options: dict) -> dict:
|
||||
challenge = options["challenge"]
|
||||
auth_data = bytearray(self._rp_id_hash())
|
||||
auth_data += bytes([0x41]) # UP + AT
|
||||
auth_data += struct.pack(">I", 0)
|
||||
aaguid = b"\x00" * 16
|
||||
auth_data += aaguid
|
||||
auth_data += struct.pack(">H", len(self.credential_id))
|
||||
auth_data += self.credential_id
|
||||
auth_data += self._cose_key()
|
||||
|
||||
attestation_object = cbor2.dumps(
|
||||
{"fmt": "none", "attStmt": {}, "authData": bytes(auth_data)},
|
||||
canonical=True,
|
||||
)
|
||||
client_data = self._client_data("webauthn.create", challenge)
|
||||
return {
|
||||
"id": _b64url(self.credential_id),
|
||||
"rawId": _b64url(self.credential_id),
|
||||
"type": "public-key",
|
||||
"response": {
|
||||
"clientDataJSON": _b64url(client_data),
|
||||
"attestationObject": _b64url(attestation_object),
|
||||
},
|
||||
}
|
||||
|
||||
def make_assertion(self, options: dict) -> dict:
|
||||
challenge = options["challenge"]
|
||||
auth_data = bytearray(self._rp_id_hash())
|
||||
auth_data += bytes([0x01]) # UP
|
||||
self.sign_count += 1
|
||||
auth_data += struct.pack(">I", self.sign_count)
|
||||
|
||||
client_data = self._client_data("webauthn.get", challenge)
|
||||
signed = bytes(auth_data) + hashlib.sha256(client_data).digest()
|
||||
# WebAuthn spec: ECDSA signatures are ASN.1 DER (not raw r||s like U2F)
|
||||
der_sig = self.key.sign(signed, ec.ECDSA(hashes.SHA256()))
|
||||
|
||||
return {
|
||||
"id": _b64url(self.credential_id),
|
||||
"rawId": _b64url(self.credential_id),
|
||||
"type": "public-key",
|
||||
"response": {
|
||||
"clientDataJSON": _b64url(client_data),
|
||||
"authenticatorData": _b64url(bytes(auth_data)),
|
||||
"signature": _b64url(der_sig),
|
||||
"userHandle": "",
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
from cryptography.hazmat.primitives import hashes # noqa: E402 (used above)
|
||||
|
||||
|
||||
# ── Unit tests: webauthn_mfa module ──────────────────────────────────
|
||||
|
||||
class TestWebauthnModule:
|
||||
def test_rp_config_defaults(self, monkeypatch):
|
||||
import backend.auth.webauthn_mfa as w
|
||||
|
||||
monkeypatch.delenv("OBSIGATE_WEBAUTHN_RP_ID", raising=False)
|
||||
assert w.rp_id() == "localhost"
|
||||
monkeypatch.setenv("OBSIGATE_WEBAUTHN_RP_ID", "obs.example.com")
|
||||
assert w.rp_id() == "obs.example.com"
|
||||
|
||||
def test_challenge_is_single_use(self):
|
||||
import backend.auth.webauthn_mfa as w
|
||||
|
||||
w._pending.clear()
|
||||
ch = w._store_challenge("u1:register")
|
||||
assert isinstance(ch, bytes) and len(ch) == 32
|
||||
assert w._take_challenge("u1:register") == ch
|
||||
assert w._take_challenge("u1:register") is None # popped
|
||||
|
||||
def test_take_challenge_expired(self):
|
||||
import time as _t
|
||||
|
||||
import backend.auth.webauthn_mfa as w
|
||||
|
||||
w._pending.clear()
|
||||
w._store_challenge("u2:register")
|
||||
key = "u2:register"
|
||||
ch, _ = w._pending[key]
|
||||
w._pending[key] = (ch, _t.time() - 1)
|
||||
assert w._take_challenge(key) is None
|
||||
|
||||
def test_full_registration_and_authentication_roundtrip(self):
|
||||
from webauthn import (
|
||||
generate_authentication_options,
|
||||
generate_registration_options,
|
||||
options_to_json,
|
||||
)
|
||||
|
||||
import backend.auth.webauthn_mfa as w
|
||||
|
||||
w._pending.clear()
|
||||
auth = VirtualAuthenticator()
|
||||
|
||||
reg_opts = generate_registration_options(
|
||||
rp_id="localhost", rp_name="ObsiGate",
|
||||
user_name="alice", user_id=b"1", user_display_name="Alice",
|
||||
challenge=w._store_challenge("alice:register"),
|
||||
)
|
||||
cred = auth.make_registration(json.loads(options_to_json(reg_opts)))
|
||||
verified = w.complete_registration("alice", cred)
|
||||
assert verified["credential_id"] == cred["id"]
|
||||
assert verified["public_key"]
|
||||
|
||||
auth_opts = generate_authentication_options(
|
||||
rp_id="localhost",
|
||||
challenge=w._store_challenge("alice:login"),
|
||||
)
|
||||
assertion = auth.make_assertion(json.loads(options_to_json(auth_opts)))
|
||||
new_count = w.complete_authentication(
|
||||
"alice", assertion,
|
||||
{"public_key": verified["public_key"], "sign_count": 0})
|
||||
assert new_count == 1
|
||||
|
||||
|
||||
# ── Integration: API endpoints ───────────────────────────────────────
|
||||
|
||||
@pytest.fixture
|
||||
def wa_client(monkeypatch):
|
||||
"""Auth-enabled client with a user, WebAuthn RP configured for localhost."""
|
||||
tmp = Path(tempfile.mkdtemp())
|
||||
data_dir = tmp / "data"
|
||||
data_dir.mkdir()
|
||||
|
||||
from backend.auth.password import hash_password
|
||||
|
||||
users = {"version": 1, "users": {"testuser": {
|
||||
"id": "t-1", "username": "testuser", "display_name": "Test",
|
||||
"password_hash": hash_password("TestPass123!"), "role": "admin",
|
||||
"vaults": ["*"], "active": True,
|
||||
}}}
|
||||
(data_dir / "users.json").write_text(json.dumps(users), encoding="utf-8")
|
||||
|
||||
monkeypatch.setattr("backend.auth.user_store.USERS_FILE", data_dir / "users.json")
|
||||
monkeypatch.setenv("OBSIGATE_WEBAUTHN_RP_ID", "localhost")
|
||||
monkeypatch.setenv("OBSIGATE_WEBAUTHN_ORIGINS", "http://localhost")
|
||||
|
||||
os.environ["VAULT_1_NAME"] = "TestVault"
|
||||
os.environ["VAULT_1_PATH"] = os.path.abspath("test-vault")
|
||||
os.environ["OBSIGATE_AUTH_ENABLED"] = "true"
|
||||
os.environ["OBSIGATE_WATCHER_ENABLED"] = "false"
|
||||
|
||||
import backend.main
|
||||
backend.main._load_config = lambda: {"watcher_enabled": False}
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
client = TestClient(backend.main.app)
|
||||
yield client
|
||||
client.close()
|
||||
shutil.rmtree(str(tmp), ignore_errors=True)
|
||||
for k in ["VAULT_1_NAME", "VAULT_1_PATH", "OBSIGATE_AUTH_ENABLED",
|
||||
"OBSIGATE_WATCHER_ENABLED"]:
|
||||
os.environ.pop(k, None)
|
||||
|
||||
|
||||
def _login_headers(client):
|
||||
r = client.post("/api/auth/login",
|
||||
json={"username": "testuser", "password": "TestPass123!"})
|
||||
token = r.json()["access_token"]
|
||||
return {"Authorization": f"Bearer {token}"}
|
||||
|
||||
|
||||
class TestWebauthnApi:
|
||||
def test_register_requires_auth(self, wa_client):
|
||||
r = wa_client.post("/api/auth/mfa/webauthn/register/options")
|
||||
assert r.status_code == 401
|
||||
|
||||
def test_registration_flow_enables_mfa(self, wa_client):
|
||||
headers = _login_headers(wa_client)
|
||||
r = wa_client.post("/api/auth/mfa/webauthn/register/options", headers=headers)
|
||||
assert r.status_code == 200
|
||||
options = r.json()["options"]
|
||||
|
||||
auth = VirtualAuthenticator()
|
||||
cred = auth.make_registration(options)
|
||||
r2 = wa_client.post("/api/auth/mfa/webauthn/register", headers=headers,
|
||||
json={"credential": cred, "label": "YubiKey 5"})
|
||||
assert r2.status_code == 200, r2.text
|
||||
body = r2.json()
|
||||
assert body["mfa_enabled"] is True
|
||||
assert len(body["recovery_codes"]) == 8
|
||||
assert body["credentials"][0]["label"] == "YubiKey 5"
|
||||
|
||||
# status reflects webauthn
|
||||
r3 = wa_client.get("/api/auth/mfa/status", headers=headers)
|
||||
st = r3.json()
|
||||
assert st["mfa_enabled"] is True
|
||||
assert st["webauthn_credentials"] == 1
|
||||
assert st["totp_enabled"] is False
|
||||
|
||||
def test_login_with_webauthn_assertion(self, wa_client):
|
||||
headers = _login_headers(wa_client)
|
||||
options = wa_client.post("/api/auth/mfa/webauthn/register/options",
|
||||
headers=headers).json()["options"]
|
||||
auth = VirtualAuthenticator()
|
||||
cred = auth.make_registration(options)
|
||||
wa_client.post("/api/auth/mfa/webauthn/register", headers=headers,
|
||||
json={"credential": cred, "label": "Key"})
|
||||
|
||||
# Fresh login → MFA required via webauthn
|
||||
r = wa_client.post("/api/auth/login",
|
||||
json={"username": "testuser", "password": "TestPass123!"})
|
||||
body = r.json()
|
||||
assert body["mfa_required"] is True
|
||||
assert body["mfa_method"] == "webauthn"
|
||||
|
||||
opts_r = wa_client.post("/api/auth/mfa/webauthn/options",
|
||||
json={"username": "testuser"})
|
||||
assert opts_r.status_code == 200
|
||||
assertion = auth.make_assertion(opts_r.json()["options"])
|
||||
v = wa_client.post("/api/auth/mfa/webauthn/verify",
|
||||
json={"username": "testuser", "credential": assertion})
|
||||
assert v.status_code == 200, v.text
|
||||
assert "access_token" in v.json()
|
||||
|
||||
def test_login_with_wrong_credential_rejected(self, wa_client):
|
||||
headers = _login_headers(wa_client)
|
||||
options = wa_client.post("/api/auth/mfa/webauthn/register/options",
|
||||
headers=headers).json()["options"]
|
||||
auth = VirtualAuthenticator()
|
||||
cred = auth.make_registration(options)
|
||||
wa_client.post("/api/auth/mfa/webauthn/register", headers=headers,
|
||||
json={"credential": cred, "label": "Key"})
|
||||
|
||||
opts_r = wa_client.post("/api/auth/mfa/webauthn/options",
|
||||
json={"username": "testuser"})
|
||||
# Impostor key signs the challenge
|
||||
impostor = VirtualAuthenticator()
|
||||
bad = impostor.make_assertion(opts_r.json()["options"])
|
||||
v = wa_client.post("/api/auth/mfa/webauthn/verify",
|
||||
json={"username": "testuser", "credential": bad})
|
||||
assert v.status_code == 401
|
||||
|
||||
def test_challenge_single_use(self, wa_client):
|
||||
headers = _login_headers(wa_client)
|
||||
options = wa_client.post("/api/auth/mfa/webauthn/register/options",
|
||||
headers=headers).json()["options"]
|
||||
auth = VirtualAuthenticator()
|
||||
cred = auth.make_registration(options)
|
||||
wa_client.post("/api/auth/mfa/webauthn/register", headers=headers,
|
||||
json={"credential": cred, "label": "K"})
|
||||
|
||||
opts_r = wa_client.post("/api/auth/mfa/webauthn/options",
|
||||
json={"username": "testuser"})
|
||||
assertion = auth.make_assertion(opts_r.json()["options"])
|
||||
v1 = wa_client.post("/api/auth/mfa/webauthn/verify",
|
||||
json={"username": "testuser", "credential": assertion})
|
||||
assert v1.status_code == 200
|
||||
# replay the same credential → challenge already consumed
|
||||
v2 = wa_client.post("/api/auth/mfa/webauthn/verify",
|
||||
json={"username": "testuser", "credential": assertion})
|
||||
assert v2.status_code == 401
|
||||
|
||||
def test_remove_key_disables_mfa(self, wa_client):
|
||||
headers = _login_headers(wa_client)
|
||||
options = wa_client.post("/api/auth/mfa/webauthn/register/options",
|
||||
headers=headers).json()["options"]
|
||||
auth = VirtualAuthenticator()
|
||||
cred = auth.make_registration(options)
|
||||
wa_client.post("/api/auth/mfa/webauthn/register", headers=headers,
|
||||
json={"credential": cred, "label": "K"})
|
||||
cred_id = cred["id"]
|
||||
|
||||
r = wa_client.post("/api/auth/mfa/webauthn/credentials/remove",
|
||||
headers=headers,
|
||||
json={"credential_id": cred_id, "password": "wrong"})
|
||||
assert r.status_code == 400
|
||||
|
||||
r2 = wa_client.post("/api/auth/mfa/webauthn/credentials/remove",
|
||||
headers=headers,
|
||||
json={"credential_id": cred_id, "password": "TestPass123!"})
|
||||
assert r2.status_code == 200
|
||||
st = wa_client.get("/api/auth/mfa/status", headers=headers).json()
|
||||
assert st["mfa_enabled"] is False
|
||||
Reference in New Issue
Block a user