feat(mfa): #64 WebAuthn complet — enregistrement/clé/verify login + gestion clés dans Sécurité (backend+frontend+i18n+tests)
This commit is contained in:
@@ -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