"""Unit tests for the MFA service layer (Issue #18, v1.6.0). These tests cover the pure-Python side of MFA — no DB, no FastAPI. """ from __future__ import annotations import asyncio import os import re import sys import time import pytest # Repo path setup (mirrors other tests in this folder). sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import pyotp # noqa: E402 from services import mfa_service # noqa: E402 # --------------------------------------------------------------------------- # TOTP # --------------------------------------------------------------------------- class TestTotpSecret: def test_secret_is_base32(self): secret = mfa_service.generate_totp_secret() # pyotp.random_base32() returns 32-character base32 strings. assert len(secret) == 32 assert re.fullmatch(r"[A-Z2-7]+", secret), "secret must be valid base32" def test_secrets_are_unique(self): secrets = {mfa_service.generate_totp_secret() for _ in range(50)} assert len(secrets) == 50 class TestVerifyTotp: def setup_method(self): self.secret = mfa_service.generate_totp_secret() self.totp = pyotp.TOTP(self.secret, digits=6, interval=30, digest="sha1") def test_happy_path(self): code = self.totp.now() ok, step = mfa_service.verify_totp_with_replay_guard(self.secret, code, None) assert ok is True assert step == int(time.time()) // 30 def test_invalid_code_format_rejected(self): ok, step = mfa_service.verify_totp_with_replay_guard(self.secret, "abc", None) assert ok is False and step is None ok, step = mfa_service.verify_totp_with_replay_guard(self.secret, "12345", None) assert ok is False and step is None def test_tolerance_minus_30s(self): now = int(time.time()) previous_step = (now // 30) - 1 prev_code = self.totp.at(previous_step * 30) ok, step = mfa_service.verify_totp_with_replay_guard(self.secret, prev_code, None) assert ok is True assert step == previous_step def test_tolerance_plus_30s(self): now = int(time.time()) next_step = (now // 30) + 1 next_code = self.totp.at(next_step * 30) ok, step = mfa_service.verify_totp_with_replay_guard(self.secret, next_code, None) assert ok is True assert step == next_step def test_replay_rejected(self): code = self.totp.now() ok, step = mfa_service.verify_totp_with_replay_guard(self.secret, code, None) assert ok is True # Submit again with the previously-consumed step: must be rejected. ok2, step2 = mfa_service.verify_totp_with_replay_guard(self.secret, code, step) assert ok2 is False assert step2 is None def test_wrong_code_rejected(self): ok, step = mfa_service.verify_totp_with_replay_guard(self.secret, "000000", None) assert ok is False assert step is None # --------------------------------------------------------------------------- # Fernet + key resolution # --------------------------------------------------------------------------- class TestFernet: def setup_method(self): mfa_service.reset_fernet_for_tests() def teardown_method(self): mfa_service.reset_fernet_for_tests() def test_encrypt_decrypt_roundtrip_with_env_key(self, monkeypatch): from cryptography.fernet import Fernet key = Fernet.generate_key().decode() monkeypatch.setenv("MFA_ENCRYPTION_KEY", key) mfa_service.reset_fernet_for_tests() secret = "JBSWY3DPEHPK3PXP" * 2 token = mfa_service.encrypt_secret(secret) assert token and token != secret recovered = mfa_service.decrypt_secret(token) assert recovered == secret def test_decrypt_invalid_token_returns_none(self, monkeypatch): from cryptography.fernet import Fernet monkeypatch.setenv("MFA_ENCRYPTION_KEY", Fernet.generate_key().decode()) mfa_service.reset_fernet_for_tests() assert mfa_service.decrypt_secret("not-a-valid-fernet-token") is None def test_hkdf_fallback_when_env_unset(self, monkeypatch, caplog): monkeypatch.delenv("MFA_ENCRYPTION_KEY", raising=False) mfa_service.reset_fernet_for_tests() with caplog.at_level("WARNING"): secret = "JBSWY3DPEHPK3PXPJBSWY3DPEHPK3PXP" token = mfa_service.encrypt_secret(secret) recovered = mfa_service.decrypt_secret(token) assert recovered == secret assert any("MFA_ENCRYPTION_KEY" in r.message for r in caplog.records), ( "expected a WARN log when falling back to SECRET_KEY derivation" ) # --------------------------------------------------------------------------- # Backup codes # --------------------------------------------------------------------------- class TestBackupCodes: def test_generate_count_and_format(self): codes = mfa_service.generate_backup_codes() assert len(codes) == 10 # 31-char alphabet: A-H J K M N P-Z 2-9 (excludes I, L, O, 0, 1). for code in codes: assert re.fullmatch(r"[A-HJKM-NP-Z2-9]{4}-[A-HJKM-NP-Z2-9]{4}", code), code def test_alphabet_excludes_confusing_characters(self): # Generate enough codes to virtually guarantee any forbidden char would surface. for _ in range(20): codes = mfa_service.generate_backup_codes() for code in codes: for ch in code.replace("-", ""): assert ch not in "0O1IL", f"forbidden char {ch!r} in {code!r}" def test_codes_are_unique(self): codes = mfa_service.generate_backup_codes() assert len(set(codes)) == len(codes) def test_normalize_strips_case_dash_space(self): assert mfa_service.normalize_backup_code("abcd-efgh") == "ABCDEFGH" assert mfa_service.normalize_backup_code(" ab cd-ef gh ") == "ABCDEFGH" assert mfa_service.normalize_backup_code("") == "" assert mfa_service.normalize_backup_code(None) == "" # type: ignore[arg-type] def test_hash_and_check_async(self): async def _run(): plain = mfa_service.generate_backup_codes()[:1] hashes = await mfa_service.hash_backup_codes(plain) assert len(hashes) == 1 assert await mfa_service.check_backup_code(plain[0], hashes[0]) is True assert await mfa_service.check_backup_code("WRONG-CODE", hashes[0]) is False # Case + dash normalization assert await mfa_service.check_backup_code(plain[0].lower(), hashes[0]) is True assert ( await mfa_service.check_backup_code(plain[0].replace("-", ""), hashes[0]) is True ) asyncio.run(_run()) # --------------------------------------------------------------------------- # otpauth URI # --------------------------------------------------------------------------- class TestOtpAuthUri: def test_uri_shape(self): uri = mfa_service.build_otpauth_uri("alice@example.com", "JBSWY3DPEHPK3PXP") assert uri.startswith("otpauth://totp/") assert "secret=JBSWY3DPEHPK3PXP" in uri assert "issuer=" in uri assert "algorithm=SHA1" in uri assert "digits=6" in uri assert "period=30" in uri def test_account_label_env_override(self, monkeypatch): monkeypatch.setenv("MFA_ACCOUNT_LABEL_DOMAIN", "ops.example.com") label = mfa_service.build_account_label("alice", hostname_hint="ignored.com") assert label == "alice@ops.example.com" def test_account_label_hostname_hint(self, monkeypatch): monkeypatch.delenv("MFA_ACCOUNT_LABEL_DOMAIN", raising=False) label = mfa_service.build_account_label("alice", hostname_hint="api.local") assert label == "alice@api.local" def test_account_label_fallback(self, monkeypatch): monkeypatch.delenv("MFA_ACCOUNT_LABEL_DOMAIN", raising=False) label = mfa_service.build_account_label("alice", hostname_hint=None) assert label == "alice@haproxy-openmanager" # --------------------------------------------------------------------------- # Challenge token # --------------------------------------------------------------------------- class TestChallengeToken: def test_length_and_uniqueness(self): tokens = {mfa_service.generate_challenge_token() for _ in range(50)} assert len(tokens) == 50 for t in tokens: assert len(t) == 64 assert re.fullmatch(r"[0-9a-f]{64}", t)