"""Contained provisioner-to-directory contract: login, preflight, redacted failure.""" from __future__ import annotations import io import json import pathlib import sys import threading import unittest from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from urllib.error import HTTPError, URLError from urllib.parse import urlencode from urllib.request import Request, urlopen sys.path.insert(0, str(pathlib.Path(__file__).parents[1])) from password_setup import LLDAPPasswordSetter, PasswordSetupGrants from provisioner import DependencyFailure, LLDAPProvisioner, directory_login from server import Handler SECRET = "synthetic-only-directory-password" TOKEN = "synthetic-directory-token" class DirectoryState: def __init__(self) -> None: self.password = SECRET self.logins = 0 self.graphql = 0 self.users: dict[str, dict] = {} self.groups: dict[str, int] = {} self.next_group = 1 self.login_status = 200 self.login_body: dict | bytes = {"token": TOKEN} class FakeDirectory(BaseHTTPRequestHandler): state: DirectoryState def do_POST(self): raw = self.rfile.read(int(self.headers.get("Content-Length", "0"))) if self.path == "/auth/simple/login": self.state.logins += 1 payload = json.loads(raw) if self.state.login_status != 200: body = self.state.login_body if isinstance(body, dict): body = json.dumps(body).encode() self._raw(self.state.login_status, body) return if payload.get("username") != "admin" or payload.get("password") != self.state.password: leaked = json.dumps({"message": "invalid", "password": payload.get("password")}).encode() self._raw(401, leaked) return self._send(200, {"token": TOKEN}) return if self.path == "/api/graphql": self.state.graphql += 1 if self.headers.get("Authorization") != f"Bearer {TOKEN}": self._send(401, {"message": "unauthorized"}) return self._send(200, self._graphql(json.loads(raw))) return self._send(404, {"error": "not_found"}) def _graphql(self, body: dict) -> dict: query = str(body.get("query") or "") variables = body.get("variables") or {} if "createUser" in query: user_id = str(variables["id"]) self.state.users[user_id] = { "id": user_id, "email": variables["email"], "displayName": variables["display"], "groups": [], } return {"data": {"createUser": {"id": user_id}}} if "createGroup" in query: name = str(variables["name"]) group_id = self.state.next_group self.state.next_group += 1 self.state.groups[name] = group_id return {"data": {"createGroup": {"id": group_id, "displayName": name}}} if "addUserToGroup" in query: user = self.state.users[str(variables["userId"])] group_id = int(variables["groupId"]) name = next(n for n, i in self.state.groups.items() if i == group_id) user["groups"].append({"id": group_id, "displayName": name}) return {"data": {"addUserToGroup": {"ok": True}}} if "users {" in query and "groups {" in query: return { "data": { "users": [ {"id": user["id"], "email": user["email"], "displayName": user["displayName"]} for user in self.state.users.values() ], "groups": [ {"id": group_id, "displayName": name} for name, group_id in self.state.groups.items() ], } } if "groups { id }" in query: return {"data": {"groups": [{"id": group_id} for group_id in self.state.groups.values()]}} return {"data": {}} def _send(self, status: int, payload: dict) -> None: self._raw(status, json.dumps(payload).encode()) def _raw(self, status: int, body: bytes) -> None: self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def log_message(self, format, *args): return def serve(handler): server = ThreadingHTTPServer(("127.0.0.1", 0), handler) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() return server, thread, f"http://127.0.0.1:{server.server_port}" class DirectoryContractTests(unittest.TestCase): def setUp(self): self.state = DirectoryState() class Bound(FakeDirectory): state = self.state self.directory, self.directory_thread, self.directory_url = serve(Bound) self.provisioner = LLDAPProvisioner( base_url=self.directory_url, admin_password=SECRET, auth_rejected_cooldown=60, clock=lambda: self.now, ) self.now = 0.0 class App(Handler): service_token = "fixture-only-service-token" provisioner = self.provisioner password_setups = PasswordSetupGrants( public_url="https://kc.example/setup/password", setter=LLDAPPasswordSetter( base_url=self.directory_url, admin_password=SECRET, helper="/bin/true", runner=lambda *args, **kwargs: type("R", (), {"returncode": 0})(), ), ) self.app, self.app_thread, self.app_url = serve(App) def tearDown(self): for server, thread in ( (self.app, self.app_thread), (self.directory, self.directory_thread), ): server.shutdown() server.server_close() thread.join() def get(self, path: str): try: response = urlopen(self.app_url + path) except HTTPError as error: response = error with response: body = response.read().decode() return response.status, json.loads(body) if body.startswith("{") else body def provision(self): payload = { "user_id": "person-1", "tenant": "tenant:trial:demo", "primary_email": "person@example.test", "display_name": "Person", "idempotency_key": "fixture-long-idempotency", "correlation_id": "corr-fixture", "roles": ["tenant-admin"], } request = Request( self.app_url + "/v1/identities/provision", data=json.dumps(payload).encode(), headers={"Authorization": "Bearer fixture-only-service-token"}, ) try: response = urlopen(request) except HTTPError as error: response = error with response: return response.status, json.loads(response.read()) def test_preflight_covers_login_and_directory_read(self): self.assertEqual( {"status": "ready", "dependency": "directory"}, self.provisioner.preflight(), ) self.assertEqual(1, self.state.logins) self.assertEqual(1, self.state.graphql) def test_healthz_stays_ok_when_directory_rejects_bind(self): self.state.login_status = 401 self.state.login_body = {"message": "invalid", "password": SECRET} live_status, live = self.get("/healthz") ready_status, ready = self.get("/readyz") self.assertEqual(200, live_status) self.assertEqual({"status": "ok"}, live) self.assertEqual(503, ready_status) self.assertEqual( { "error": "dependency_unavailable", "dependency": "directory", "reason": "auth_rejected", }, ready, ) self.assertNotIn(SECRET, json.dumps(ready)) def test_rejected_bind_is_not_retried_during_cooldown(self): self.state.login_status = 401 with self.assertRaises(DependencyFailure) as failure: self.provisioner.preflight() self.assertEqual("auth_rejected", failure.exception.code) self.assertEqual(1, self.state.logins) self.now = 30.0 with self.assertRaises(DependencyFailure): self.provisioner.preflight() self.assertEqual(1, self.state.logins) self.now = 60.0 with self.assertRaises(DependencyFailure): self.provisioner.preflight() self.assertEqual(2, self.state.logins) def test_provision_returns_redacted_structured_failure_not_http_error(self): self.state.login_status = 401 self.state.login_body = {"message": "invalid", "password": SECRET} status, payload = self.provision() self.assertEqual(503, status) self.assertEqual("dependency_unavailable", payload["error"]) self.assertEqual("directory", payload["dependency"]) self.assertEqual("auth_rejected", payload["reason"]) serialized = json.dumps(payload) self.assertNotIn(SECRET, serialized) self.assertNotIn("HTTPError", serialized) self.assertNotIn("password", serialized) def test_native_provision_and_password_setup_journey(self): status, payload = self.provision() self.assertEqual(200, status) self.assertEqual("password_setup_required", payload["status"]) self.assertIn("uid=person,ou=people,dc=netkingdom,dc=local", payload["external_subject"]) self.assertIn("person", self.state.users) self.assertIn("tenant:trial:demo:users", self.state.groups) self.assertIn("tenant:trial:demo:admins", self.state.groups) token = payload["password_setup_url"].partition("token=")[2] request = Request( self.app_url + "/setup/password", data=urlencode({ "token": token, "password": "recipient-long-password", "confirmation": "recipient-long-password", }).encode(), headers={"Content-Type": "application/x-www-form-urlencoded"}, ) with urlopen(request) as response: body = response.read().decode() self.assertEqual(200, response.status) self.assertIn("Password set", body) self.assertNotIn(SECRET, body) self.assertNotIn("recipient-long-password", body) self.assertNotIn(token, body) class DirectoryLoginMappingTests(unittest.TestCase): def test_http_401_becomes_auth_rejected_without_body_or_cause(self): def opener(request, timeout=10): raise HTTPError( request.full_url, 401, "Unauthorized", hdrs=None, fp=io.BytesIO(json.dumps({"password": SECRET, "token": "leaked"}).encode()), ) with self.assertRaises(DependencyFailure) as failure: directory_login(base_url="http://directory", admin_password=SECRET, opener=opener) self.assertEqual("auth_rejected", failure.exception.code) self.assertIsNone(failure.exception.__cause__) self.assertNotIn(SECRET, str(failure.exception)) self.assertNotIn("leaked", str(failure.exception.payload())) self.assertEqual( { "error": "dependency_unavailable", "dependency": "directory", "reason": "auth_rejected", }, failure.exception.payload(), ) def test_unreachable_directory_is_structured_and_not_retried(self): calls = [] def opener(request, timeout=10): calls.append(timeout) raise URLError("connection refused") with self.assertRaises(DependencyFailure) as failure: directory_login(base_url="http://directory", admin_password=SECRET, opener=opener) self.assertEqual("unreachable", failure.exception.code) self.assertEqual([10], calls) def test_password_setter_rejects_bind_once(self): calls = [] def opener(request, timeout=10): calls.append(json.loads(request.data)["password"]) raise HTTPError(request.full_url, 401, "Unauthorized", hdrs=None, fp=io.BytesIO(b"{}")) setter = LLDAPPasswordSetter( base_url="http://directory", admin_password=SECRET, opener=opener, runner=lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("helper must not run")), ) with self.assertRaises(DependencyFailure) as failure: setter("person", "recipient-long-password") self.assertEqual("auth_rejected", failure.exception.code) self.assertEqual([SECRET], calls) self.assertNotIn(SECRET, str(failure.exception))