diff --git a/.env.example b/.env.example index 2e95f74..ff67e47 100644 --- a/.env.example +++ b/.env.example @@ -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 diff --git a/backend/auth/router.py b/backend/auth/router.py index 2375605..fa1b893 100644 --- a/backend/auth/router.py +++ b/backend/auth/router.py @@ -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", [])), } diff --git a/backend/auth/webauthn_mfa.py b/backend/auth/webauthn_mfa.py new file mode 100644 index 0000000..b802f8b --- /dev/null +++ b/backend/auth/webauthn_mfa.py @@ -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": , "public_key": , "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)) diff --git a/backend/requirements.txt b/backend/requirements.txt index a8421ec..eee76e2 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -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 diff --git a/frontend/js/auth.js b/frontend/js/auth.js index 49102c1..aaa182d 100644 --- a/frontend/js/auth.js +++ b/frontend/js/auth.js @@ -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 = ` +
🔑
+

${t('mfa.title')}

+

${t('mfa.webauthn_prompt')}

+ + +
+ + +
+ `; + 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 = `
🔐

${t('mfa.title')}

@@ -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 + ? `` + : `

${t("mfa.webauthn_none")}

`; + + section.innerHTML = ` +

${t("mfa.webauthn_title")}

+

${t("mfa.webauthn_desc")}

+ ${listHtml} +
+ +
+ +
+ `; + + 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"); + } + }); + }); } diff --git a/frontend/locales/en.json b/frontend/locales/en.json index 127644f..b57b823 100644 --- a/frontend/locales/en.json +++ b/frontend/locales/en.json @@ -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." } diff --git a/frontend/locales/fr.json b/frontend/locales/fr.json index fd5e234..17a0ae8 100644 --- a/frontend/locales/fr.json +++ b/frontend/locales/fr.json @@ -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." } diff --git a/frontend/style.css b/frontend/style.css index 4f490a4..316e7d8 100644 --- a/frontend/style.css +++ b/frontend/style.css @@ -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; diff --git a/tests/test_webauthn.py b/tests/test_webauthn.py new file mode 100644 index 0000000..4c5c997 --- /dev/null +++ b/tests/test_webauthn.py @@ -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