tegwick
c8e07615c3
All checks were successful
Map uncaught HTTPError from LLDAP login to a structured dependency_unavailable response, add /readyz as the provisioner-to-directory preflight, keep /healthz as process liveness, and run the contract in CI. Auth rejection is not retried during cooldown. NK-WP-0036-T05 remains in progress until the immutable image is published, pinned with /readyz, and one native login/create/password-setup journey is verified. Assistant: grok Assistant-Session: 01a09dc6-3f0e-78f1-a884-c8c703c24ddf
336 lines
13 KiB
Python
336 lines
13 KiB
Python
"""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))
|