# 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][0] 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_login_options_enumeration_safe(self, wa_client): # Unknown user / no MFA -> always 200 with totp fallback, no 404/400 leak r = wa_client.post("/api/auth/mfa/webauthn/options", json={"username": "ghost-user"}) assert r.status_code == 200 assert r.json() == {"mfa_method": "totp", "options": None} 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 # ── BUG-070: relying party derived from the request ───────────────────── # # The old defaults (rp_id "localhost", origins ["http://localhost"]) rejected # every real access URL: "Unexpected client data origin # "http://localhost:2020", expected one of ['http://localhost']". def _fake_request(host, scheme="http", forwarded_host=None, forwarded_proto=None): from fastapi import Request headers = [(b"host", host.encode())] if forwarded_host is not None: headers.append((b"x-forwarded-host", forwarded_host.encode())) if forwarded_proto is not None: headers.append((b"x-forwarded-proto", forwarded_proto.encode())) return Request({ "type": "http", "method": "POST", "path": "/", "headers": headers, "scheme": scheme, "server": ("testserver", 80), "client": ("127.0.0.1", 5000), }) class TestRelyingPartyResolution: def test_defaults_without_request(self, monkeypatch): import backend.auth.webauthn_mfa as w monkeypatch.delenv("OBSIGATE_WEBAUTHN_RP_ID", raising=False) monkeypatch.delenv("OBSIGATE_WEBAUTHN_ORIGINS", raising=False) assert w.resolve_relying_party(None) == ("localhost", ["http://localhost"]) def test_derives_host_with_port(self, monkeypatch): """Exact BUG-070 report: http://localhost:2020 was rejected.""" import backend.auth.webauthn_mfa as w monkeypatch.delenv("OBSIGATE_WEBAUTHN_RP_ID", raising=False) monkeypatch.delenv("OBSIGATE_WEBAUTHN_ORIGINS", raising=False) rp, origins = w.resolve_relying_party(_fake_request("localhost:2020")) assert rp == "localhost" assert origins == ["http://localhost:2020"] def test_derives_ip_host(self, monkeypatch): import backend.auth.webauthn_mfa as w monkeypatch.delenv("OBSIGATE_WEBAUTHN_RP_ID", raising=False) monkeypatch.delenv("OBSIGATE_WEBAUTHN_ORIGINS", raising=False) rp, origins = w.resolve_relying_party(_fake_request("127.0.0.1:2020")) assert rp == "127.0.0.1" assert origins == ["http://127.0.0.1:2020"] def test_explicit_env_wins_over_request(self, monkeypatch): import backend.auth.webauthn_mfa as w monkeypatch.setenv("OBSIGATE_WEBAUTHN_RP_ID", "obs.example.com") monkeypatch.setenv("OBSIGATE_WEBAUTHN_ORIGINS", "https://obs.example.com, https://www.obs.example.com") rp, origins = w.resolve_relying_party(_fake_request("localhost:2020")) assert rp == "obs.example.com" assert origins == ["https://obs.example.com", "https://www.obs.example.com"] def test_forwarded_headers_require_trust(self, monkeypatch): import backend.auth.webauthn_mfa as w monkeypatch.delenv("OBSIGATE_WEBAUTHN_RP_ID", raising=False) monkeypatch.delenv("OBSIGATE_WEBAUTHN_ORIGINS", raising=False) monkeypatch.setenv("OBSIGATE_TRUST_PROXY", "false") req = _fake_request("internal:8080", scheme="http", forwarded_host="obs.example.com", forwarded_proto="https") assert w.resolve_relying_party(req) == ("internal", ["http://internal:8080"]) monkeypatch.setenv("OBSIGATE_TRUST_PROXY", "true") assert w.resolve_relying_party(req) == ("obs.example.com", ["https://obs.example.com"]) def test_hostname_only(self): import backend.auth.webauthn_mfa as w assert w._hostname_only("example.com:2020") == "example.com" assert w._hostname_only("example.com") == "example.com" assert w._hostname_only("[::1]:8080") == "::1" assert w._hostname_only("127.0.0.1:2020") == "127.0.0.1" def test_retry_after_reoptions_still_verifies(self): """A re-requested options call (double-click) must not kill the in-flight ceremony: "challenge was not expected challenge".""" import backend.auth.webauthn_mfa as w w._pending.clear() auth = VirtualAuthenticator() first = w._store_challenge("bob:register") w._store_challenge("bob:register") # second options call overwrites cred = auth.make_registration({"challenge": _b64url(first)}) rec = w.complete_registration("bob", cred, rp_id_override="localhost", origins_override=["http://localhost"]) assert rec["credential_id"] == cred["id"] def test_register_flow_without_env_config(self, wa_client, monkeypatch): """Full register + login roundtrip with no WEBAUTHN env at all: the relying party derives from the request (TestClient host).""" import backend.auth.webauthn_mfa as w monkeypatch.delenv("OBSIGATE_WEBAUTHN_RP_ID", raising=False) monkeypatch.delenv("OBSIGATE_WEBAUTHN_ORIGINS", raising=False) w._pending.clear() 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"] assert options["rp"]["id"] == "testserver" auth = VirtualAuthenticator() auth.RP_ID = "testserver" auth.ORIGIN = "http://testserver" cred = auth.make_registration(options) r2 = wa_client.post("/api/auth/mfa/webauthn/register", headers=headers, json={"credential": cred, "label": "Key"}) assert r2.status_code == 200, r2.text opts_r = wa_client.post("/api/auth/mfa/webauthn/options", json={"username": "testuser"}) 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()