468 lines
19 KiB
Python
468 lines
19 KiB
Python
# 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()
|