#!/usr/bin/env python3 """Execute the administration guides with synthetic credentials and loopback. Run with pulse-worker-source-proof; these fixtures do not perform real role, tenant or audit operations. The Go documentation tests check the actual handlers. """ from __future__ import annotations import json from pathlib import Path import re import tempfile import unittest from test_api_auth_docs import (ROOT, TEST_TOKEN, exercise_curl, recording_server, private_recording_server, request_helper) DOCS = ROOT / "docs" GUIDES = ("RBAC", "AUDIT_LOGGING", "MULTI_TENANT") EXPECTED = { "RBAC": ( ("POST", "/api/admin/roles", { "id": "alert-manager", "name": "Alert Manager", "description": "Can view and manage alerts", "permissions": [{"action": "read", "resource": "alerts"}, {"action": "write", "resource": "alerts"}, {"action": "read", "resource": "nodes"}], }), ("GET", "/api/admin/roles", None), ("PUT", "/api/admin/roles/alert-manager", { "name": "Alert Manager", "description": "Updated description", "permissions": [{"action": "read", "resource": "alerts"}, {"action": "write", "resource": "alerts"}, {"action": "read", "resource": "nodes"}, {"action": "read", "resource": "ai"}], }), ("DELETE", "/api/admin/roles/alert-manager", None), ("GET", "/api/admin/users", None), ("PUT", "/api/admin/users/jane/roles", {"roleIds": ["alert-manager", "viewer"]}), ("PUT", "/api/admin/users/jane/roles", {"roleIds": []}), ("DELETE", "/api/admin/users/jane", None), ), "AUDIT_LOGGING": ( ("GET", "/api/audit?limit=50", None), ("GET", "/api/audit?limit=50&event=login&startTime=2026-01-01T00:00:00Z&endTime=2026-01-31T23:59:59Z&success=false", None), ("GET", "/api/audit/summary", None), ("GET", "/api/audit/export?event=login&startTime=2026-01-01T00:00:00Z&endTime=2026-01-31T23:59:59Z", None), ("GET", "/api/audit/6b3c9c3c-9a2f-4b3c-9a3b-3d0e8c5c5d45/verify", None), ), "MULTI_TENANT": ( ("GET", "/api/orgs/production-datacenter/members", None), ("GET", "/api/orgs/production-datacenter/shares/incoming", None), ), } def recipes(name: str) -> list[str]: requests = [] for block in re.findall(r"```bash\n(.*?)```", (DOCS / f"{name}.md").read_text(), re.DOTALL): # The audit read examples share a fence but are independent operations. for step in re.split(r"\n(?=# )", block): if "curl " in step or "pulse_api " in step: requests.append(step) return requests def executable_recipe(name: str, step: str) -> str: # Use the actual shared helper, not a test-only safe client. Legacy curl # snippets remain executable so the parent control can expose their leak. if "pulse_api " in step: return request_helper() + "\n" + step return step class AdminDocsTest(unittest.TestCase): def test_shell_recipes_never_expose_credentials_or_override_tls(self): for name in GUIDES: with self.subTest(guide=name): steps = recipes(name) self.assertEqual(len(steps), len(EXPECTED[name])) for step in steps: self.assertNotRegex(step, r"(?:\b\w*TOKEN=|Authorization:|X-API-Token:|Bearer\s|--cookie\b)") self.assertNotRegex(step, r"(?:--insecure|--verbose|--trace\S*|--location|\s-k\b)") self.assertIn("pulse_api ", step) self.assertNotIn("curl ", step) if re.search(r"(?:--request|pulse_api) (POST|PUT)", step): self.assertIn("--data-binary @-", request_helper() if "pulse_api " in step else step) self.assertIn("<<'JSON'", step) def test_guides_explain_authority_and_sensitive_output(self): for name in GUIDES: guide = (DOCS / f"{name}.md").read_text() with self.subTest(guide=name): self.assertIn("API.md#-authentication", guide) self.assertIn("HTTPS", guide) self.assertIn("certificate verification", guide) rbac = (DOCS / "RBAC.md").read_text() for boundary in ("full-access (`*`)", "bound to an authorised administrator", "cannot be modified or deleted", "self-escalation", "complete list", "URL-encode"): self.assertIn(boundary, rbac) audit = (DOCS / "AUDIT_LOGGING.md").read_text() for boundary in ("audit:read", "audit_logging", "private directory", "redacted excerpt", "does not reuse filters", "in the same Bash session", "not an installed Pulse command", "not list pagination", "reject malformed or empty supplied", "before reading events", "not `success` or list", "partial file", "does not follow redirects or retry"): self.assertIn(boundary, audit) org = (DOCS / "MULTI_TENANT.md").read_text() for boundary in ("session-based user authentication", "403 session_required", "CSRF", "settings:read", "organization binding", "pending invitation", "accept the share", "accessRole"): self.assertIn(boundary, org) self.assertNotRegex(org, r"\| `PATCH` \|") def test_shipped_copies_are_identical(self): for name in (*GUIDES, "API"): with self.subTest(guide=name): self.assertEqual((DOCS / f"{name}.md").read_bytes(), (ROOT / "frontend-modern/public/docs" / f"{name}.md").read_bytes()) def test_all_recipes_send_exact_methods_paths_and_bodies_with_each_header(self): with recording_server() as (port, requests): for name, operations in EXPECTED.items(): steps = recipes(name) self.assertEqual(len(steps), len(operations)) for key, value in (("X-API-Token", TEST_TOKEN), ("Authorization", "Bearer " + TEST_TOKEN)): for step, (method, path, body) in zip(steps, operations): with self.subTest(guide=name, method=method, path=path, header=key), tempfile.TemporaryDirectory(prefix="admin docs ") as temporary: before = len(requests) result = exercise_curl(self, Path(temporary), f"{key}: {value}", port, executable_recipe(name, step)) self.assertEqual(result.returncode, 0, result.stderr.decode()) self.assertEqual(len(requests), before + 1, "a copied step sends exactly one request") sent_path, headers, sent_method, sent_body = requests[-1] self.assertEqual((sent_method, sent_path), (method, path)) self.assertEqual(headers[key], value) self.assertNotIn("X-Curlrc-Injected", headers) self.assertNotIn("Authorization" if key == "X-API-Token" else "X-API-Token", headers) if body is None: self.assertEqual(sent_body, b"") else: self.assertEqual(headers["Content-Type"], "application/json") self.assertEqual(json.loads(sent_body), body) def test_every_recipe_surfaces_auth_and_license_errors_without_next_steps(self): for status in (401, 402, 403): with recording_server(status) as (port, requests): for name in GUIDES: for step in recipes(name): with self.subTest(status=status, guide=name), tempfile.TemporaryDirectory() as temporary: before = len(requests) result = exercise_curl(self, Path(temporary), f"X-API-Token: {TEST_TOKEN}", port, executable_recipe(name, step)) self.assertEqual(result.returncode, 22, result.stderr.decode()) self.assertEqual(len(requests), before + 1) self.assertNotIn(b"Saved private export", result.stdout) # Every guide preserves the error in a private response file. responses = list((Path(temporary) / ".config/pulse").glob("api-response.*")) self.assertEqual(len(responses), 1) output = responses[0].read_bytes() self.assertNotIn(b'{"fixture":true}', result.stdout) self.assertIn(b'{"fixture":true}', output) def test_audit_reads_keep_private_success_bodies_in_new_owner_only_files(self): self.assertEqual(len(recipes("AUDIT_LOGGING")), len(EXPECTED["AUDIT_LOGGING"])) with private_recording_server() as (port, requests), tempfile.TemporaryDirectory() as temporary: home = Path(temporary) for header in (f"X-API-Token: {TEST_TOKEN}", f"Authorization: Bearer {TEST_TOKEN}"): for step, (method, path, _) in zip(recipes("AUDIT_LOGGING"), EXPECTED["AUDIT_LOGGING"]): with self.subTest(path=path, header=header.split(":")[0]): before = len(requests) result = exercise_curl(self, home, header, port, executable_recipe("AUDIT_LOGGING", step), private_response=True) self.assertEqual(result.returncode, 0, result.stderr.decode()) self.assertEqual(len(requests), before + 1) self.assertEqual((requests[-1][2], requests[-1][0]), (method, path)) self.assertIn(b"HTTP 200", result.stdout) # One already-retained response plus ten distinct new responses. self.assertEqual(len(list((home / ".config/pulse").glob("api-response.*"))), 11) def test_audit_reads_keep_private_http_error_bodies_and_fail(self): self.assertEqual(len(recipes("AUDIT_LOGGING")), len(EXPECTED["AUDIT_LOGGING"])) for status in (400, 401, 402, 403, 500, 503): with private_recording_server(status) as (port, requests): for step in recipes("AUDIT_LOGGING"): with self.subTest(status=status, step=step), tempfile.TemporaryDirectory() as temporary: before = len(requests) result = exercise_curl(self, Path(temporary), f"X-API-Token: {TEST_TOKEN}", port, executable_recipe("AUDIT_LOGGING", step), private_response=True) self.assertEqual(result.returncode, 22, result.stderr.decode()) self.assertEqual(len(requests), before + 1) self.assertIn(f"HTTP {status}".encode(), result.stdout) def test_audit_reads_refuse_redirect_success_and_incomplete_transfers(self): self.assertEqual(len(recipes("AUDIT_LOGGING")), len(EXPECTED["AUDIT_LOGGING"])) for status, partial, expected_exit in ((302, False, 1), (200, True, 18)): with private_recording_server(status, partial) as (port, requests): for step in recipes("AUDIT_LOGGING"): with self.subTest(status=status, partial=partial, step=step), tempfile.TemporaryDirectory() as temporary: before = len(requests) result = exercise_curl(self, Path(temporary), f"X-API-Token: {TEST_TOKEN}", port, executable_recipe("AUDIT_LOGGING", step), private_response=True) self.assertEqual(result.returncode, expected_exit, result.stderr.decode()) self.assertEqual(len(requests), before + 1, "must not follow redirects or retry partial reads") def test_rbac_guidance_preserves_private_and_uncertain_change_boundaries(self): guide = (DOCS / "RBAC.md").read_text() for boundary in ("in the same Bash session", "not an installed Pulse command", "owner-only file", "redacted", "five-second", "twenty-second", "not a script to run from top to bottom", "administrator recovery path", "empty role list remove access", "does not follow redirects or retry", "after the change was applied", "inspect the current state", "HTTP success", "upstream IdP account"): self.assertIn(boundary, guide) def test_rbac_operations_keep_private_success_bodies_in_new_owner_only_files(self): steps = recipes("RBAC") self.assertEqual(len(steps), len(EXPECTED["RBAC"])) with private_recording_server() as (port, requests), tempfile.TemporaryDirectory(prefix="rbac docs ") as temporary: home = Path(temporary) for header in (f"X-API-Token: {TEST_TOKEN}", f"Authorization: Bearer {TEST_TOKEN}"): for step, (method, path, _) in zip(steps, EXPECTED["RBAC"]): with self.subTest(method=method, path=path, header=header.split(":")[0]): before = len(requests) result = exercise_curl(self, home, header, port, executable_recipe("RBAC", step), private_response=True) self.assertEqual(result.returncode, 0, result.stderr.decode()) self.assertEqual(len(requests), before + 1) self.assertEqual((requests[-1][2], requests[-1][0]), (method, path)) self.assertIn(b"HTTP 200", result.stdout) self.assertEqual(len(list((home / ".config/pulse").glob("api-response.*"))), 17) def test_rbac_operations_keep_private_http_errors_and_fail_without_retry(self): for status in (400, 401, 402, 403, 500, 503): with private_recording_server(status) as (port, requests): for step in recipes("RBAC"): with self.subTest(status=status, step=step), tempfile.TemporaryDirectory() as temporary: before = len(requests) result = exercise_curl(self, Path(temporary), f"X-API-Token: {TEST_TOKEN}", port, executable_recipe("RBAC", step), private_response=True) self.assertEqual(result.returncode, 22, result.stderr.decode()) self.assertEqual(len(requests), before + 1) self.assertIn(f"HTTP {status}".encode(), result.stdout) def test_rbac_operations_refuse_redirect_success_and_uncertain_transfer_retry(self): for status, partial, expected_exit in ((302, False, 1), (200, True, 18)): with private_recording_server(status, partial) as (port, requests): for step in recipes("RBAC"): with self.subTest(status=status, partial=partial, step=step), tempfile.TemporaryDirectory() as temporary: before = len(requests) result = exercise_curl(self, Path(temporary), f"X-API-Token: {TEST_TOKEN}", port, executable_recipe("RBAC", step), private_response=True) self.assertEqual(result.returncode, expected_exit, result.stderr.decode()) self.assertEqual(len(requests), before + 1, "an uncertain mutation must not be repeated or redirected") def test_org_guidance_preserves_private_bounded_reads_and_session_mutations(self): guide = (DOCS / "MULTI_TENANT.md").read_text() for boundary in ("in the same Bash session", "not an installed Pulse command", "five-second", "twenty-second", "does not follow redirects or retry", "owner-only file", "redacted", "partial file", "not a complete list", "HTTP 401/402/403", "do not", "widen a token", "mutations token-authenticated", "signed-in UI", "CSRF"): self.assertIn(boundary, guide) for step in recipes("MULTI_TENANT"): self.assertRegex(step, r"^pulse_api GET /api/orgs/") self.assertNotIn("curl ", step) def test_org_reads_keep_private_success_bodies_in_new_owner_only_files(self): steps = recipes("MULTI_TENANT") self.assertEqual(len(steps), len(EXPECTED["MULTI_TENANT"])) with private_recording_server() as (port, requests), tempfile.TemporaryDirectory(prefix="org docs ") as temporary: home = Path(temporary) for header in (f"X-API-Token: {TEST_TOKEN}", f"Authorization: Bearer {TEST_TOKEN}"): for step, (method, path, _) in zip(steps, EXPECTED["MULTI_TENANT"]): with self.subTest(path=path, header=header.split(":")[0]): before = len(requests) result = exercise_curl(self, home, header, port, executable_recipe("MULTI_TENANT", step), private_response=True) self.assertEqual(result.returncode, 0, result.stderr.decode()) self.assertEqual(len(requests), before + 1) self.assertEqual((requests[-1][2], requests[-1][0]), (method, path)) self.assertEqual(requests[-1][3], b"", "organization reads send no mutation body") self.assertIn(b"HTTP 200", result.stdout) # Four reads retain four new files without overwriting the old response. self.assertEqual(len(list((home / ".config/pulse").glob("api-response.*"))), 5) def test_org_reads_keep_private_http_errors_and_fail_without_retry(self): for status in (400, 401, 402, 403, 500, 503): with private_recording_server(status) as (port, requests): for step in recipes("MULTI_TENANT"): with self.subTest(status=status, step=step), tempfile.TemporaryDirectory() as temporary: before = len(requests) result = exercise_curl(self, Path(temporary), f"X-API-Token: {TEST_TOKEN}", port, executable_recipe("MULTI_TENANT", step), private_response=True) self.assertEqual(result.returncode, 22, result.stderr.decode()) self.assertEqual(len(requests), before + 1) self.assertIn(f"HTTP {status}".encode(), result.stdout) def test_org_reads_refuse_redirect_success_and_incomplete_transfer_retry(self): for status, partial, expected_exit in ((302, False, 1), (200, True, 18)): with private_recording_server(status, partial) as (port, requests): for step in recipes("MULTI_TENANT"): with self.subTest(status=status, partial=partial, step=step), tempfile.TemporaryDirectory() as temporary: before = len(requests) result = exercise_curl(self, Path(temporary), f"X-API-Token: {TEST_TOKEN}", port, executable_recipe("MULTI_TENANT", step), private_response=True) self.assertEqual(result.returncode, expected_exit, result.stderr.decode()) self.assertEqual(len(requests), before + 1, "do not follow redirects or retry partial reads") if __name__ == "__main__": unittest.main()