"""Functional failure matrix through the real HTTP transport, no live credentials.""" import copy import importlib import json import subprocess import sys import threading from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import parse_qs, urlsplit import pytest ROOT = Path(__file__).resolve().parents[1] HELPERS = ROOT / "sso-mfa/k8s/privacyidea" sys.path.insert(0, str(HELPERS)) pi_api = importlib.import_module("pi_api") verify_mfa = importlib.import_module("verify_mfa") def result(value, **extra): return {"result": {"status": True, "value": value}, **extra} @pytest.fixture def target(): state = { "/auth": result({"token": "test-token"}), "/realm/": result({"coulomb": {"resolver": [{"name": "lldap-coulomb"}]}}), "/resolver/lldap-coulomb": result( { "lldap-coulomb": { "type": "ldapresolver", "data": { "TIMEOUT": "5", "CACHE_TIMEOUT": "120", "SIZELIMIT": "500", }, } } ), "/user/": result([{"username": "test-user", "resolver": "lldap-coulomb"}]), "/validate/check": result(True, detail={"serial": "TOTP-TEST", "type": "totp"}), "requests": [], } class Handler(BaseHTTPRequestHandler): def log_message(self, *_): pass def do_GET(self): self.respond() def do_POST(self): self.respond() def respond(self): parsed = urlsplit(self.path) body = self.rfile.read(int(self.headers.get("Content-Length", 0))) state["requests"].append( (self.command, self.path, dict(self.headers), body) ) # Reproduce the original server rejection at the request boundary. if self.command == "GET" and self.headers.get("Content-Type"): self.send_error(400) return if parsed.path == "/user/": assert parse_qs(parsed.query) == { "realm": ["coulomb"], "username": ["test-user"], } response = state.get(parsed.path) if response is None: self.send_error(403) return self.send_response(200) self.end_headers() self.wfile.write(json.dumps(response).encode()) server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() try: yield f"http://127.0.0.1:{server.server_port}", state finally: server.shutdown() server.server_close() thread.join() def run(url): return verify_mfa.verify( url, "test-user", "coulomb", "lldap-coulomb", "test-password", lambda: "test-otp", ) @pytest.mark.parametrize("field", ["TIMEOUT", "CACHE_TIMEOUT", "SIZELIMIT"]) def test_clear_and_restore_tuning(target, field): url, state = target data = state["/resolver/lldap-coulomb"]["result"]["value"]["lldap-coulomb"]["data"] original = data.pop(field) assert run(url) == {"result": "FAIL", "phase": "resolver-tuning"} assert all(urlsplit(r[1]).path != "/validate/check" for r in state["requests"]) data[field] = original assert run(url)["result"] == "PASS" @pytest.mark.parametrize( "route,replacement,phase", [ ("/auth", None, "authentication"), ("/auth", result({}), "authentication"), ("/realm/", result({"coulomb": {"resolver": []}}), "realm-binding"), ("/user/", result([]), "known-user-lookup"), ( "/user/", result([{"username": "other", "resolver": "lldap-coulomb"}]), "known-user-lookup", ), ( "/user/", result([{"username": "test-user", "resolver": "other"}]), "known-user-lookup", ), ("/user/", {"result": {"status": False, "value": []}}, "known-user-lookup"), ("/validate/check", result(False), "mfa-validation"), ("/validate/check", result(True), "mfa-validation"), ( "/validate/check", result(True, detail={"serial": "PASSWORD", "type": "spass"}), "mfa-validation", ), ("/validate/check", result(True, detail=None), "mfa-validation"), ], ) def test_failures_never_become_success(target, route, replacement, phase): url, state = target state[route] = copy.deepcopy(replacement) assert run(url) == {"result": "FAIL", "phase": phase} def test_transport_and_shell_adapter_share_bodyless_get_behavior(target): url, state = target assert run(url)["result"] == "PASS" proc = subprocess.run( [sys.executable, str(HELPERS / "pi_api.py"), "GET", url + "/realm/"], input="test-token\n", text=True, capture_output=True, check=False, ) assert proc.returncode == 0 assert json.loads(proc.stdout)["result"]["status"] is True for method, path, headers, body in state["requests"]: if method == "GET": assert "Content-Type" not in headers assert not body else: assert headers["Content-Type"] == "application/json" assert isinstance(json.loads(body), dict) def test_shell_entrypoint_requires_attendance_and_does_not_read_bundle(): proc = subprocess.run( ["bash", str(ROOT / "sso-mfa/k8s/verify-t06.sh"), "--user", "test-user"], stdin=subprocess.DEVNULL, capture_output=True, check=False, text=True, ) assert proc.returncode == 2 assert "attended terminal" in proc.stderr assert not proc.stdout @pytest.mark.parametrize("policy_ok", [True, False]) def test_bootstrap_uses_shared_transport_and_reports_full_result( target, tmp_path, policy_ok ): url, state = target for route in [ "/realm/coulomb", "/defaultrealm/coulomb", "/policy/totp-self-enrollment", "/policy/coulomb-friendly-token-labels", "/policy/mfa-passthru-phase1", ]: state[route] = result(True) if not policy_ok: state["/policy/totp-self-enrollment"] = None for folder, content in [ ("privacyidea", "PI_ADMIN_PASSWORD=test-password\n"), ("lldap", "LLDAP_LDAP_USER_PASS=test-bind-password\n"), ]: directory = tmp_path / folder directory.mkdir() (directory / "secrets.env").write_text(content) proc = subprocess.run( ["bash", str(HELPERS / "bootstrap-realm.sh"), str(tmp_path), url], capture_output=True, text=True, check=False, ) assert proc.returncode == (0 if policy_ok else 1) assert "Realm bootstrap: PASS=" in proc.stdout assert "FAIL=0" in proc.stdout if policy_ok else "FAIL=1" in proc.stdout for secret in ["test-token", "test-password", "test-bind-password"]: assert secret not in proc.stdout + proc.stderr assert any( method == "GET" and path == "/resolver/lldap-coulomb" for method, path, _, _ in state["requests"] )