370 lines
14 KiB
Python
370 lines
14 KiB
Python
"""Tests for MFA (Multi-Factor Authentication) — TOTP + recovery codes."""
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import pyotp
|
|
import pytest
|
|
|
|
from backend.auth.mfa import (
|
|
TOTP_ISSUER,
|
|
generate_qr_uri,
|
|
generate_recovery_codes,
|
|
generate_secret,
|
|
hash_recovery_code,
|
|
verify_recovery_code,
|
|
verify_totp,
|
|
)
|
|
|
|
|
|
# ── Unit tests: TOTP ───────────────────────────────────────────────────
|
|
|
|
class TestTotpGeneration:
|
|
def test_generate_secret_returns_base32(self):
|
|
secret = generate_secret()
|
|
assert isinstance(secret, str)
|
|
assert len(secret) >= 16
|
|
import base64
|
|
base64.b32decode(secret, casefold=True)
|
|
|
|
def test_generate_secret_unique(self):
|
|
secrets = {generate_secret() for _ in range(100)}
|
|
assert len(secrets) == 100
|
|
|
|
def test_generate_qr_uri_format(self):
|
|
secret = generate_secret()
|
|
uri = generate_qr_uri(secret, "testuser")
|
|
assert uri.startswith("otpauth://totp/")
|
|
assert "ObsiGate" in uri
|
|
assert "testuser" in uri
|
|
assert "secret=" in uri
|
|
|
|
def test_generate_qr_uri_custom_issuer(self):
|
|
secret = generate_secret()
|
|
uri = generate_qr_uri(secret, "testuser", issuer="CustomIssuer")
|
|
assert "CustomIssuer" in uri
|
|
|
|
def test_verify_totp_valid(self):
|
|
secret = generate_secret()
|
|
totp = pyotp.TOTP(secret)
|
|
code = totp.now()
|
|
assert verify_totp(secret, code) is True
|
|
|
|
def test_verify_totp_invalid(self):
|
|
secret = generate_secret()
|
|
assert verify_totp(secret, "000000") is False
|
|
|
|
def test_verify_totp_wrong_secret(self):
|
|
secret1 = generate_secret()
|
|
secret2 = generate_secret()
|
|
totp1 = pyotp.TOTP(secret1)
|
|
code = totp1.now()
|
|
assert verify_totp(secret2, code) is False
|
|
|
|
def test_verify_totp_window_tolerance(self):
|
|
secret = generate_secret()
|
|
totp = pyotp.TOTP(secret)
|
|
current_time = time.time()
|
|
code = totp.at(int(current_time))
|
|
assert verify_totp(secret, str(code).zfill(6)) is True
|
|
|
|
|
|
# ── Unit tests: Recovery codes ─────────────────────────────────────────
|
|
|
|
class TestRecoveryCodes:
|
|
def test_generate_count(self):
|
|
codes = generate_recovery_codes()
|
|
assert len(codes) == 8
|
|
|
|
def test_generate_custom_count(self):
|
|
codes = generate_recovery_codes(n=12)
|
|
assert len(codes) == 12
|
|
|
|
def test_code_format(self):
|
|
codes = generate_recovery_codes()
|
|
for code in codes:
|
|
assert len(code) == 9 # XXXX-XXXX
|
|
assert code[4] == "-"
|
|
assert code[:4].isalnum()
|
|
assert code[5:].isalnum()
|
|
|
|
def test_codes_unique(self):
|
|
codes = generate_recovery_codes(n=20)
|
|
assert len(set(codes)) == 20
|
|
|
|
def test_codes_uppercase(self):
|
|
codes = generate_recovery_codes()
|
|
for code in codes:
|
|
assert code == code.upper()
|
|
|
|
def test_hash_recovery_code_deterministic(self):
|
|
code = "ABCD-1234"
|
|
h1 = hash_recovery_code(code)
|
|
h2 = hash_recovery_code(code)
|
|
assert h1 == h2
|
|
|
|
def test_hash_case_insensitive(self):
|
|
h1 = hash_recovery_code("abcd-1234")
|
|
h2 = hash_recovery_code("ABCD-1234")
|
|
assert h1 == h2
|
|
|
|
def test_hash_different_for_different_codes(self):
|
|
h1 = hash_recovery_code("AAAA-AAAA")
|
|
h2 = hash_recovery_code("BBBB-BBBB")
|
|
assert h1 != h2
|
|
|
|
def test_verify_recovery_code_match(self):
|
|
codes = generate_recovery_codes()
|
|
hashed = [hash_recovery_code(c) for c in codes]
|
|
for i, code in enumerate(codes):
|
|
idx = verify_recovery_code(code, hashed)
|
|
assert idx == i
|
|
|
|
def test_verify_recovery_code_case_insensitive(self):
|
|
codes = ["ABCD-1234"]
|
|
hashed = [hash_recovery_code(c) for c in codes]
|
|
assert verify_recovery_code("abcd-1234", hashed) == 0
|
|
|
|
def test_verify_recovery_code_invalid(self):
|
|
codes = generate_recovery_codes()
|
|
hashed = [hash_recovery_code(c) for c in codes]
|
|
assert verify_recovery_code("ZZZZ-ZZZZ", hashed) is None
|
|
|
|
def test_verify_recovery_code_empty_list(self):
|
|
assert verify_recovery_code("AAAA-AAAA", []) is None
|
|
|
|
def test_recovery_code_single_use(self):
|
|
codes = generate_recovery_codes(n=3)
|
|
hashed = [hash_recovery_code(c) for c in codes]
|
|
idx = verify_recovery_code(codes[0], hashed)
|
|
assert idx == 0
|
|
hashed.pop(idx)
|
|
assert verify_recovery_code(codes[0], hashed) is None
|
|
idx2 = verify_recovery_code(codes[1], hashed)
|
|
assert idx2 == 0
|
|
|
|
|
|
# ── Integration tests: MFA API endpoints ───────────────────────────────
|
|
|
|
@pytest.fixture
|
|
def mfa_client():
|
|
"""Create a TestClient with auth enabled, isolated temp data."""
|
|
tmp = Path(tempfile.mkdtemp())
|
|
data_dir = tmp / "data"
|
|
data_dir.mkdir()
|
|
|
|
from backend.auth.password import hash_password
|
|
pw_hash = hash_password("TestPass123!")
|
|
users = {
|
|
"version": 1,
|
|
"users": {
|
|
"testuser": {
|
|
"id": "testuser-1",
|
|
"username": "testuser",
|
|
"display_name": "Test User",
|
|
"password_hash": pw_hash,
|
|
"role": "admin",
|
|
"vaults": ["*"],
|
|
"active": True,
|
|
"created_at": "2026-01-01T00:00:00",
|
|
}
|
|
}
|
|
}
|
|
(data_dir / "users.json").write_text(json.dumps(users), encoding="utf-8")
|
|
|
|
src_secret = Path("data/secret.key")
|
|
if src_secret.exists():
|
|
shutil.copy2(str(src_secret), str(data_dir / "secret.key"))
|
|
|
|
orig_cwd = os.getcwd()
|
|
test_vault_path = os.path.abspath("test-vault")
|
|
os.chdir(str(tmp))
|
|
|
|
os.environ["VAULT_1_NAME"] = "TestVault"
|
|
os.environ["VAULT_1_PATH"] = test_vault_path
|
|
os.environ["OBSIGATE_AUTH_ENABLED"] = "true"
|
|
os.environ["OBSIGATE_ADMIN_USER"] = "testuser"
|
|
os.environ["OBSIGATE_ADMIN_PASSWORD"] = "TestPass123!"
|
|
os.environ["OBSIGATE_WATCHER_ENABLED"] = "false"
|
|
|
|
import backend.main
|
|
backend.main._load_config = lambda: {"watcher_enabled": False}
|
|
|
|
from backend.main import app
|
|
from backend.indexer import build_index, index
|
|
for key in list(index.keys()):
|
|
del index[key]
|
|
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
loop.run_until_complete(build_index())
|
|
|
|
from backend.search import init_inverted_index
|
|
init_inverted_index()
|
|
|
|
from fastapi.testclient import TestClient
|
|
client = TestClient(app)
|
|
yield client
|
|
|
|
if hasattr(client, 'close'):
|
|
client.close()
|
|
loop.run_until_complete(asyncio.sleep(0))
|
|
|
|
os.chdir(orig_cwd)
|
|
shutil.rmtree(str(tmp), ignore_errors=True)
|
|
for k in ["VAULT_1_NAME", "VAULT_1_PATH", "OBSIGATE_AUTH_ENABLED",
|
|
"OBSIGATE_ADMIN_USER", "OBSIGATE_ADMIN_PASSWORD", "OBSIGATE_WATCHER_ENABLED"]:
|
|
os.environ.pop(k, None)
|
|
|
|
|
|
def _login(client, username="testuser", password="TestPass123!"):
|
|
resp = client.post("/api/auth/login", json={"username": username, "password": password})
|
|
return resp.json().get("access_token"), resp
|
|
|
|
|
|
def _auth_headers(token):
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
class TestMfaApiEndpoints:
|
|
def test_mfa_status_initially_disabled(self, mfa_client):
|
|
token, _ = _login(mfa_client)
|
|
resp = mfa_client.get("/api/auth/mfa/status", headers=_auth_headers(token))
|
|
assert resp.status_code == 200
|
|
assert resp.json()["mfa_enabled"] is False
|
|
|
|
def test_mfa_setup_generates_secret(self, mfa_client):
|
|
token, _ = _login(mfa_client)
|
|
resp = mfa_client.post("/api/auth/mfa/totp/setup", headers=_auth_headers(token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "secret" in data
|
|
assert "otpauth_uri" in data
|
|
assert "otpauth://totp/" in data["otpauth_uri"]
|
|
assert len(data["secret"]) >= 16
|
|
|
|
def test_mfa_setup_returns_local_qr_data_url(self, mfa_client):
|
|
"""BUG-068: the setup response carries a CSP-safe local QR code.
|
|
|
|
The previous client used an https://api.qrserver.com image, blocked by
|
|
the CSP (img-src 'self' data: blob:) — the QR never displayed — and
|
|
leaking the otpauth URI to a third party.
|
|
"""
|
|
token, _ = _login(mfa_client)
|
|
resp = mfa_client.post("/api/auth/mfa/totp/setup", headers=_auth_headers(token))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
qr_data_url = data.get("qr_data_url")
|
|
assert qr_data_url, "setup must return a local qr_data_url"
|
|
assert qr_data_url.startswith("data:image/svg+xml"), qr_data_url[:60]
|
|
assert "qrserver.com" not in qr_data_url
|
|
|
|
def test_mfa_enable_flow(self, mfa_client):
|
|
token, _ = _login(mfa_client)
|
|
headers = _auth_headers(token)
|
|
# Setup
|
|
setup_resp = mfa_client.post("/api/auth/mfa/totp/setup", headers=headers)
|
|
secret = setup_resp.json()["secret"]
|
|
totp = pyotp.TOTP(secret)
|
|
# Enable
|
|
enable_resp = mfa_client.post("/api/auth/mfa/totp/enable", headers=headers, json={
|
|
"code": totp.now(),
|
|
})
|
|
assert enable_resp.status_code == 200
|
|
data = enable_resp.json()
|
|
assert "recovery_codes" in data
|
|
assert len(data["recovery_codes"]) == 8
|
|
assert data["mfa_enabled"] is True
|
|
|
|
def test_mfa_enable_invalid_code(self, mfa_client):
|
|
token, _ = _login(mfa_client)
|
|
headers = _auth_headers(token)
|
|
mfa_client.post("/api/auth/mfa/totp/setup", headers=headers)
|
|
resp = mfa_client.post("/api/auth/mfa/totp/enable", headers=headers, json={
|
|
"code": "000000",
|
|
})
|
|
assert resp.status_code in (400, 401)
|
|
|
|
def test_mfa_login_defers_to_totp(self, mfa_client):
|
|
token, _ = _login(mfa_client)
|
|
headers = _auth_headers(token)
|
|
# Enable MFA
|
|
setup_resp = mfa_client.post("/api/auth/mfa/totp/setup", headers=headers)
|
|
secret = setup_resp.json()["secret"]
|
|
totp = pyotp.TOTP(secret)
|
|
mfa_client.post("/api/auth/mfa/totp/enable", headers=headers, json={"code": totp.now()})
|
|
# Login again — should require MFA
|
|
login_resp = mfa_client.post("/api/auth/login", json={
|
|
"username": "testuser", "password": "TestPass123!",
|
|
})
|
|
assert login_resp.status_code == 200
|
|
data = login_resp.json()
|
|
assert data.get("mfa_required") is True
|
|
assert data.get("mfa_method") == "totp"
|
|
assert "access_token" not in data
|
|
|
|
def test_mfa_verify_issues_token(self, mfa_client):
|
|
token, _ = _login(mfa_client)
|
|
headers = _auth_headers(token)
|
|
setup_resp = mfa_client.post("/api/auth/mfa/totp/setup", headers=headers)
|
|
secret = setup_resp.json()["secret"]
|
|
totp = pyotp.TOTP(secret)
|
|
mfa_client.post("/api/auth/mfa/totp/enable", headers=headers, json={"code": totp.now()})
|
|
# Login → MFA challenge
|
|
mfa_client.post("/api/auth/login", json={"username": "testuser", "password": "TestPass123!"})
|
|
# Verify TOTP
|
|
verify_resp = mfa_client.post("/api/auth/mfa/totp/verify", json={
|
|
"username": "testuser", "code": totp.now(),
|
|
})
|
|
assert verify_resp.status_code == 200
|
|
data = verify_resp.json()
|
|
assert "access_token" in data
|
|
assert data["token_type"] == "bearer"
|
|
|
|
def test_mfa_recovery_login(self, mfa_client):
|
|
token, _ = _login(mfa_client)
|
|
headers = _auth_headers(token)
|
|
setup_resp = mfa_client.post("/api/auth/mfa/totp/setup", headers=headers)
|
|
secret = setup_resp.json()["secret"]
|
|
totp = pyotp.TOTP(secret)
|
|
enable_resp = mfa_client.post("/api/auth/mfa/totp/enable", headers=headers, json={"code": totp.now()})
|
|
recovery_codes = enable_resp.json()["recovery_codes"]
|
|
# Login → MFA challenge
|
|
mfa_client.post("/api/auth/login", json={"username": "testuser", "password": "TestPass123!"})
|
|
# Use recovery code
|
|
recover_resp = mfa_client.post("/api/auth/mfa/recovery", json={
|
|
"username": "testuser", "recovery_code": recovery_codes[0],
|
|
})
|
|
assert recover_resp.status_code == 200
|
|
assert "access_token" in recover_resp.json()
|
|
|
|
def test_mfa_disable_flow(self, mfa_client):
|
|
token, _ = _login(mfa_client)
|
|
headers = _auth_headers(token)
|
|
setup_resp = mfa_client.post("/api/auth/mfa/totp/setup", headers=headers)
|
|
secret = setup_resp.json()["secret"]
|
|
totp = pyotp.TOTP(secret)
|
|
mfa_client.post("/api/auth/mfa/totp/enable", headers=headers, json={"code": totp.now()})
|
|
# Disable
|
|
disable_resp = mfa_client.post("/api/auth/mfa/totp/disable", headers=headers, json={
|
|
"password": "TestPass123!", "code": totp.now(),
|
|
})
|
|
assert disable_resp.status_code == 200
|
|
# Verify disabled
|
|
status_resp = mfa_client.get("/api/auth/mfa/status", headers=headers)
|
|
assert status_resp.json()["mfa_enabled"] is False
|
|
|
|
def test_login_without_mfa_still_works(self, mfa_client):
|
|
login_resp = mfa_client.post("/api/auth/login", json={
|
|
"username": "testuser", "password": "TestPass123!",
|
|
})
|
|
assert login_resp.status_code == 200
|
|
data = login_resp.json()
|
|
assert "access_token" in data
|
|
assert data.get("mfa_required") is None
|