feat(mfa): #64 WebAuthn complet — enregistrement/clé/verify login + gestion clés dans Sécurité (backend+frontend+i18n+tests)

This commit is contained in:
2026-09-07 23:55:35 -04:00
parent 7042307955
commit ab795ec9e0
9 changed files with 1029 additions and 12 deletions
+6 -1
View File
@@ -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
View File
@@ -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", [])),
}
+184
View File
@@ -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))
+1
View File
@@ -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
View File
@@ -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");
}
});
});
}
+14 -2
View File
@@ -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."
}
+14 -2
View File
@@ -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."
}
+12
View File
@@ -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;
+332
View File
@@ -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