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
|
# OBSIGATE_BACKUP_DIR=.obsigate-backup
|
||||||
|
|
||||||
# PDF (ROADMAP #74)
|
# 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
|
# 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 ──
|
# ── AI Provider Configuration ──
|
||||||
# Définir au moins un provider pour activer les fonctionnalités AI dans l'éditeur
|
# Définir au moins un provider pour activer les fonctionnalités AI dans l'éditeur
|
||||||
|
|
||||||
|
|||||||
+198
-5
@@ -5,7 +5,7 @@
|
|||||||
import logging
|
import logging
|
||||||
import re
|
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 pydantic import BaseModel, validator
|
||||||
|
|
||||||
from backend.ratelimit import is_rate_limited
|
from backend.ratelimit import is_rate_limited
|
||||||
@@ -147,12 +147,13 @@ async def login(body: LoginRequest, response: Response, request: Request):
|
|||||||
# Success — clear rate limits
|
# Success — clear rate limits
|
||||||
rl_record_success(client_ip)
|
rl_record_success(client_ip)
|
||||||
|
|
||||||
# If MFA is enabled, don't issue token yet — require TOTP verification
|
# If MFA is enabled, don't issue token yet — require second factor
|
||||||
if user.get("mfa_enabled") and user.get("mfa_secret"):
|
if user.get("mfa_enabled"):
|
||||||
logger.info(f"User '{body.username}' login deferred — MFA required")
|
method = "totp" if user.get("mfa_secret") else _preferred_mfa_method(user)
|
||||||
|
logger.info(f"User '{body.username}' login deferred — MFA required ({method})")
|
||||||
return {
|
return {
|
||||||
"mfa_required": True,
|
"mfa_required": True,
|
||||||
"mfa_method": "totp",
|
"mfa_method": method,
|
||||||
"username": body.username,
|
"username": body.username,
|
||||||
"remember_me": body.remember_me,
|
"remember_me": body.remember_me,
|
||||||
}
|
}
|
||||||
@@ -434,6 +435,196 @@ async def mfa_totp_disable(
|
|||||||
return {"mfa_enabled": False}
|
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")
|
@router.get("/mfa/status")
|
||||||
async def mfa_status(current_user=Depends(require_auth)):
|
async def mfa_status(current_user=Depends(require_auth)):
|
||||||
"""Return current user's MFA status."""
|
"""Return current user's MFA status."""
|
||||||
@@ -442,6 +633,8 @@ async def mfa_status(current_user=Depends(require_auth)):
|
|||||||
return {
|
return {
|
||||||
"mfa_enabled": user.get("mfa_enabled", False),
|
"mfa_enabled": user.get("mfa_enabled", False),
|
||||||
"mfa_method": user.get("mfa_method"),
|
"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
|
httpx>=0.27.0
|
||||||
pypdf>=4.0
|
pypdf>=4.0
|
||||||
pyotp>=2.10.0
|
pyotp>=2.10.0
|
||||||
|
webauthn==2.6.0
|
||||||
psutil>=5.9
|
psutil>=5.9
|
||||||
|
|||||||
+268
-2
@@ -179,6 +179,57 @@ const AuthManager = {
|
|||||||
return data.user;
|
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 ──────────────────────────────────────────
|
// ── MFA Setup API calls ──────────────────────────────────────────
|
||||||
|
|
||||||
async getMfaStatus() {
|
async getMfaStatus() {
|
||||||
@@ -376,7 +427,112 @@ async function _onLoginSuccess() {
|
|||||||
// MFA Challenge UI (TOTP code input during login)
|
// 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");
|
const loginBox = document.querySelector(".login-box");
|
||||||
if (!loginBox) return;
|
if (!loginBox) return;
|
||||||
|
|
||||||
@@ -393,6 +549,37 @@ function showMfaChallenge(username, rememberMe, loginBtn, loginErrorEl) {
|
|||||||
loginBox.appendChild(mfaSection);
|
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 = `
|
mfaSection.innerHTML = `
|
||||||
<div class="mfa-icon">🔐</div>
|
<div class="mfa-icon">🔐</div>
|
||||||
<h3>${t('mfa.title')}</h3>
|
<h3>${t('mfa.title')}</h3>
|
||||||
@@ -565,7 +752,7 @@ function initLoginForm() {
|
|||||||
const result = await AuthManager.login(username, password, rememberMe);
|
const result = await AuthManager.login(username, password, rememberMe);
|
||||||
// Check if MFA is required
|
// Check if MFA is required
|
||||||
if (result && result.mfa_required) {
|
if (result && result.mfa_required) {
|
||||||
showMfaChallenge(result.username, rememberMe, btn, errorEl);
|
showMfaChallenge(result.username, rememberMe, btn, errorEl, result.mfa_method);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
// Normal login success
|
// Normal login success
|
||||||
@@ -858,6 +1045,85 @@ async function initMfaSettings() {
|
|||||||
_startMfaSetup();
|
_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.disable_confirm_btn": "Disable 2FA",
|
||||||
"mfa.disabled_success": "2FA has been disabled.",
|
"mfa.disabled_success": "2FA has been disabled.",
|
||||||
"mfa.fill_all_fields": "Please fill in all fields.",
|
"mfa.fill_all_fields": "Please fill in all fields.",
|
||||||
|
|
||||||
"bookslm.title": "BooksLM",
|
"bookslm.title": "BooksLM",
|
||||||
"bookslm.files_indexed": "{count} files indexed",
|
"bookslm.files_indexed": "{count} files indexed",
|
||||||
"bookslm.chars_loaded": "{chars} chars loaded",
|
"bookslm.chars_loaded": "{chars} chars loaded",
|
||||||
@@ -1601,5 +1600,18 @@
|
|||||||
"palette.bookslm_open": "BooksLM: Open for current directory",
|
"palette.bookslm_open": "BooksLM: Open for current directory",
|
||||||
"palette.bookslm_new": "BooksLM: New conversation",
|
"palette.bookslm_new": "BooksLM: New conversation",
|
||||||
"fab.open": "Open AI assistant",
|
"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.disable_confirm_btn": "Désactiver la 2FA",
|
||||||
"mfa.disabled_success": "La 2FA a été désactivée.",
|
"mfa.disabled_success": "La 2FA a été désactivée.",
|
||||||
"mfa.fill_all_fields": "Veuillez remplir tous les champs.",
|
"mfa.fill_all_fields": "Veuillez remplir tous les champs.",
|
||||||
|
|
||||||
"bookslm.title": "BooksLM",
|
"bookslm.title": "BooksLM",
|
||||||
"bookslm.files_indexed": "{count} fichiers indexés",
|
"bookslm.files_indexed": "{count} fichiers indexés",
|
||||||
"bookslm.chars_loaded": "{chars} caractères chargé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_open": "BooksLM: Ouvrir pour le répertoire courant",
|
||||||
"palette.bookslm_new": "BooksLM: Nouvelle conversation",
|
"palette.bookslm_new": "BooksLM: Nouvelle conversation",
|
||||||
"fab.open": "Ouvrir l'assistant AI",
|
"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; }
|
.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 */
|
||||||
.mfa-disable-card {
|
.mfa-disable-card {
|
||||||
padding: 16px; border-radius: 10px;
|
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