diff --git a/internal/repoctl/setup_bootstrap_docs_test.go b/internal/repoctl/setup_bootstrap_docs_test.go index 32ef86754a..064012da86 100644 --- a/internal/repoctl/setup_bootstrap_docs_test.go +++ b/internal/repoctl/setup_bootstrap_docs_test.go @@ -48,27 +48,50 @@ func TestSetupBootstrapDocsStayOnCanonicalArtifactContract(t *testing.T) { "Stop if the download fails", "Create a separate Pulse token in **API Access**", "enter an administrator root shell", - `chmod 700 "$HOME/.config/pulse"`, - `chmod 600 "$HOME/.config/pulse/pbs-agent-token"`, - `vi "$HOME/.config/pulse/pbs-agent-token"`, - `curl --fail --silent --show-error`, - `--output "$HOME/.config/pulse/pbs-agent-install.sh"`, - `https://pulse.example.com/install.sh`, - "successful download and inspection", - `bash "$HOME/.config/pulse/pbs-agent-install.sh"`, - `--token-file "$HOME/.config/pulse/pbs-agent-token"`, - `--enable-proxmox --proxmox-type pbs --enable-docker=false`, + "successful HTTP **200** download and inspection", "Do not bypass certificate checks", }) shellExamples := pbsShellExamples(pbsDoc) if shellExamples == "" { t.Fatal("PBS guide must retain executable shell examples") } + // Bind the guard to the private, non-clobbering recipes, not their old + // literal-path spelling. installtests also executes these copied commands + // against independent permission, transport and failure controls. + assertContainsAll(t, pbsRel, shellExamples, []string{ + `set -eu`, + `umask 077`, + `config_dir="$HOME/.config/pulse"`, + `credential_file="$config_dir/pbs-agent-token"`, + `[ -L "$HOME/.config" ]`, + `[ -L "$config_dir" ]`, + `[ -L "$credential_file" ]`, + `[ ! -f "$credential_file" ]`, + `chmod 700 "$config_dir"`, + `chmod 600 "$credential_file"`, + `vi "$credential_file"`, + `installer_file="$config_dir/pbs-agent-install.sh"`, + `[ -e "$installer_file" ] || [ -L "$installer_file" ]`, + `download_file=$(mktemp "$config_dir/pbs-agent-download.XXXXXX")`, + `curl --disable --fail --silent --show-error --proto '=https'`, + `--connect-timeout 5 --max-time 60 --output "$download_file"`, + `--write-out '%{http_code}' https://pulse.example.com/install.sh`, + `[ "$curl_exit" -eq 0 ] || exit "$curl_exit"`, + `[ "$status" = 200 ]`, + `mv -n "$download_file" "$installer_file"`, + `[ ! -e "$download_file" ]`, + `bash "$HOME/.config/pulse/pbs-agent-install.sh"`, + `--token-file "$HOME/.config/pulse/pbs-agent-token"`, + `--enable-proxmox --proxmox-type pbs --enable-docker=false`, + }) assertContainsNone(t, pbsRel, shellExamples, []string{ `curl -sSL "http://:7655/api/setup-script?type=pbs&host=https://:8007&pulse_url=http://:7655" | bash`, `PULSE_SETUP_TOKEN=`, `sudo env PULSE_SETUP_TOKEN=`, `--insecure`, + `--location`, + `--trace`, + `--verbose`, ` -k`, `curl -k`, `--token "`, diff --git a/scripts/tests/test_api_auth_docs.py b/scripts/tests/test_api_auth_docs.py index 6ec083d45d..e777d1eadb 100644 --- a/scripts/tests/test_api_auth_docs.py +++ b/scripts/tests/test_api_auth_docs.py @@ -24,6 +24,7 @@ import unittest ROOT = Path(__file__).resolve().parents[2] DOC = ROOT / "docs/API.md" TEST_TOKEN = "synthetic-doc-test-token" +PRIVATE_BODY = json.dumps({"fixture": True, "private": TEST_TOKEN}).encode() + b"\n" def auth_section() -> str: @@ -33,8 +34,8 @@ def auth_section() -> str: def commands() -> tuple[str, str]: blocks = re.findall(r"```bash\n(.*?)```", auth_section(), re.DOTALL) preparation = next(block for block in blocks if "umask 077" in block) - request = next(block for block in blocks if block.startswith("curl ")) - return preparation, request + helper = next(block for block in blocks if block.startswith("pulse_api() (")) + return preparation, helper def recipe(path: str) -> str: @@ -76,7 +77,7 @@ def recording_server(status: int = 200): requests.append((self.path, dict(self.headers), self.command, body)) self.send_response(status) self.end_headers() - self.wfile.write(b'{"fixture":true}\n') + self.wfile.write(PRIVATE_BODY) do_GET = record_request do_POST = record_request @@ -129,14 +130,31 @@ def exercise_curl(case: unittest.TestCase, home: Path, header_text: str, port: i for key in list(env): if key.lower().endswith("_proxy"): del env[key] - request = request.replace("http://127.0.0.1:7655", f"http://127.0.0.1:{port}") + # Readers define the documented helper before copying one call. Do not + # invent a replacement client or expect response bodies on stdout. + helper = commands()[1].replace("http://127.0.0.1:7655", f"http://127.0.0.1:{port}") + request = helper + "\n" + request + previous_responses = {path: path.read_bytes() for path in header.parent.glob("api-response.*")} result = subprocess.run(["bash", "-eu", "-c", request], env=env, capture_output=True, timeout=10) + case.assertTrue(receipt.exists(), "documented helper did not reach the curl fixture") argv = json.loads(receipt.read_text()) case.assertEqual(argv[0], "--disable", "curl defaults must be disabled by the first option") case.assertNotIn(TEST_TOKEN, " ".join(argv)) case.assertIn("@" + str(header), argv) case.assertNotIn(b"synthetic-doc-test-token", result.stdout + result.stderr) case.assertFalse(trace.exists(), "local curl configuration must not create a credential trace") + observation = re.fullmatch(rb"HTTP ([0-9]{3})\nPrivate response: ([^\n]+)\n", result.stdout) + case.assertIsNotNone(observation, "only status and the private response location should be printed") + response = Path(os.fsdecode(observation[2])) + case.assertEqual(response.parent, header.parent) + case.assertTrue(response.name.startswith("api-response.")) + case.assertNotIn(response, previous_responses, "each call must retain a new response file") + case.assertEqual(response.read_bytes(), PRIVATE_BODY) + case.assertEqual(stat.S_IMODE(response.stat().st_mode), 0o600) + case.assertEqual(stat.S_IMODE(header.parent.stat().st_mode), 0o700) + for path, body in previous_responses.items(): + case.assertEqual(path.read_bytes(), body, "later calls must not replace earlier evidence") + case.assertEqual(len(list(header.parent.glob("api-response.*"))), len(previous_responses) + 1) return result @@ -145,7 +163,8 @@ class APIAuthDocsTest(unittest.TestCase): section = auth_section() bash = "\n".join(re.findall(r"```bash\n(.*?)```", section, re.DOTALL)) self.assertNotRegex(bash, r"(?:X-API-Token:|Authorization:|Bearer\s)") - self.assertIn('--header "@$HOME/.config/pulse/api-header"', bash) + self.assertIn('auth_file="$auth_dir/api-header"', bash) + self.assertIn('--header "@$auth_file"', bash) self.assertIn("HTTPS", section) self.assertIn("does not verify authentication", section) self.assertIn("monitoring:read", section) @@ -156,13 +175,18 @@ class APIAuthDocsTest(unittest.TestCase): self.assertNotRegex(bash, r"(?:\b\w*TOKEN=|X-API-Token:|Authorization:|Bearer\s)") self.assertNotRegex(bash, r"(?:--insecure|--verbose|--trace\S*|--location|\s-k\b)") requests = [block for block in re.findall(r"```bash\n(.*?)```", DOC.read_text(), re.DOTALL) - if block.startswith("curl ")] + if block.startswith("pulse_api ")] self.assertEqual(len(requests), 8) + helper = commands()[1] + for control in ("curl --disable --fail-with-body ", '--header "@$auth_file"', + '--output "$result_file"', '--connect-timeout 5 --max-time 20', + 'result_file=$(mktemp "$auth_dir/api-response.XXXXXX")', + '[ "$curl_exit" -eq 0 ] || exit "$curl_exit"'): + self.assertIn(control, helper) for request in requests: - self.assertTrue(request.startswith("curl --disable --fail-with-body ")) - self.assertIn('--header "@$HOME/.config/pulse/api-header"', request) - if "--request POST" in request: - self.assertIn("--data-binary @-", request) + self.assertRegex(request, r"^pulse_api (?:GET|POST) ") + if request.startswith("pulse_api POST "): + self.assertIn("--data-binary @-", helper) self.assertIn("<<'JSON'", request) def test_action_guidance_preserves_scope_and_review_boundaries(self): @@ -251,7 +275,8 @@ class APIAuthDocsTest(unittest.TestCase): self.assertFalse((home / "editor-called").exists()) def run_documented_request(self, home: Path, header_text: str, port: int, request=None): - return exercise_curl(self, home, header_text, port, request if request is not None else commands()[1]) + return exercise_curl(self, home, header_text, port, + request if request is not None else recipe("/api/state/summary")) def test_header_file_sends_each_supported_header_without_exposing_argv(self): with tempfile.TemporaryDirectory() as temporary, recording_server() as (port, requests): @@ -294,7 +319,8 @@ class APIAuthDocsTest(unittest.TestCase): result = self.run_documented_request(Path(temporary), "X-API-Token: " + TEST_TOKEN, port, recipe(path)) self.assertEqual(result.returncode, 22, result.stderr.decode()) - self.assertIn(b'{"fixture":true}', result.stdout) + self.assertIn(f"HTTP {status}\n".encode(), result.stdout) + self.assertNotIn(PRIVATE_BODY, result.stdout + result.stderr) self.assertEqual(len(requests), before + 1) def test_http_auth_errors_fail_instead_of_looking_successful(self): @@ -303,7 +329,8 @@ class APIAuthDocsTest(unittest.TestCase): with recording_server(status) as (port, _): result = self.run_documented_request(Path(temporary), "X-API-Token: " + TEST_TOKEN, port) self.assertEqual(result.returncode, 22, result.stderr.decode()) - self.assertIn(b'{"fixture":true}', result.stdout) + self.assertIn(f"HTTP {status}\n".encode(), result.stdout) + self.assertNotIn(PRIVATE_BODY, result.stdout + result.stderr) if __name__ == "__main__": diff --git a/scripts/tests/test_pbs_docs.py b/scripts/tests/test_pbs_docs.py index 3e9f104fcb..b7bbff0f51 100644 --- a/scripts/tests/test_pbs_docs.py +++ b/scripts/tests/test_pbs_docs.py @@ -25,6 +25,7 @@ DOC = ROOT / "docs/PBS.md" PBS_TOKEN = "synthetic-pbs-secret" AGENT_TOKEN = "synthetic-agent-secret" AUTH_HEADER = f"PBSAPIToken=pulse-monitor@pbs!pulse-token:{PBS_TOKEN}" +PRIVATE_BODY = json.dumps({"data": [], "private": PBS_TOKEN}).encode() + b"\n" def blocks(): @@ -34,8 +35,8 @@ def blocks(): return re.findall(r"```bash\n(.*?)```", text, re.DOTALL) -def recipe(needle): - matches = [block for block in blocks() if needle in block] +def recipe(needle, also=""): + matches = [block for block in blocks() if needle in block and also in block] if len(matches) != 1: raise AssertionError(f"expected one recipe containing {needle!r}") return matches[0] @@ -78,7 +79,7 @@ def server(cert, key, status=200, installer=None, installer_path="/install.sh"): requests.append((self.path, dict(self.headers))) self.send_response(status) self.end_headers() - body = installer if self.path == installer_path else b'{"data":[]}\n' + body = installer if self.path == installer_path else PRIVATE_BODY self.wfile.write(body or b"fixture error\n") def log_message(self, *_args): @@ -169,7 +170,7 @@ class PBSDocsTest(unittest.TestCase): def test_private_preparation_preserves_existing_content(self): for name in ("pbs-agent-token", "pbs-header"): - command = recipe(f'vi "$HOME/.config/pulse/{name}"') + command = recipe(f'credential_file="$config_dir/{name}"', 'vi "$credential_file"') with self.subTest(name=name), tempfile.TemporaryDirectory() as temporary: home = Path(temporary) tools = home / "tools" @@ -208,23 +209,50 @@ class PBSDocsTest(unittest.TestCase): ) recorder.chmod(0o700) receipt = home / "argv.json" + trace = home / "curl-trace.txt" + (home / ".curlrc").write_text( + f'header = "X-Curlrc-Injected: yes"\nverbose\ntrace-ascii = "{trace}"\n' + ) env = fixture_environment(home) - env.update(PATH=f"{tools}:{os.environ['PATH']}", REAL_CURL=real_curl, ARGV_RECEIPT=str(receipt)) + env.update(PATH=f"{tools}:{os.environ['PATH']}", REAL_CURL=real_curl, ARGV_RECEIPT=str(receipt), + CURL_HOME=str(home), XDG_CONFIG_HOME=str(home / ".config")) command = command.replace("https://pbs.example.com:8007", f"https://{hostname}:{port}") command = command.replace("https://pulse.example.com", f"https://{hostname}:{port}") if ca is not None: shutil.copyfile(ca, private / "pbs-ca.pem") # This is the guide's optional, independently verified CA-file form. - command = command.replace("curl ", 'curl --cacert "$HOME/.config/pulse/pbs-ca.pem" ', 1) + command = command.replace("curl --disable ", + 'curl --disable --cacert "$HOME/.config/pulse/pbs-ca.pem" ', 1) + previous_responses = {path: path.read_bytes() for path in private.glob("pbs-response.*")} result = subprocess.run(["bash", "-eu", "-c", command], env=env, capture_output=True, timeout=20) + self.assertTrue(receipt.exists(), "documented recipe did not reach the curl fixture") argv = json.loads(receipt.read_text()) + self.assertEqual(argv[0], "--disable", "even the optional CA form must ignore curl defaults first") for secret in (PBS_TOKEN, AGENT_TOKEN): self.assertNotIn(secret, " ".join(argv)) self.assertNotIn(secret.encode(), result.stdout + result.stderr) self.assertNotIn("--insecure", argv) self.assertNotIn("-k", argv) + self.assertFalse(trace.exists(), "local curl configuration must not create a credential trace") + if "pbs-response.XXXXXX" in command: + observation = re.fullmatch(rb"Private response file: ([^\n]+)\nHTTP ([0-9]{3})\n", result.stdout) + self.assertIsNotNone(observation, "only the private response location and status should be printed") + response = Path(os.fsdecode(observation[1])) + self.assertEqual(response.parent, private) + self.assertTrue(response.name.startswith("pbs-response.")) + self.assertNotIn(response, previous_responses) + self.assertEqual(stat.S_IMODE(response.stat().st_mode), 0o600) + self.assertEqual(stat.S_IMODE(private.stat().st_mode), 0o700) + for path, body in previous_responses.items(): + self.assertEqual(path.read_bytes(), body, "later requests must preserve earlier evidence") + self.assertEqual(len(list(private.glob("pbs-response.*"))), len(previous_responses) + 1) return result, argv + def private_response(self, result): + location = re.fullmatch(rb"Private response file: ([^\n]+)\nHTTP ([0-9]{3})\n", result.stdout) + self.assertIsNotNone(location) + return Path(os.fsdecode(location[1])).read_bytes() + def test_datastore_request_transmits_auth_only_in_header(self): with tempfile.TemporaryDirectory() as temporary, server(self.cert, self.key) as (port, requests): result, argv = self.run_curl(Path(temporary), recipe("--fail-with-body"), port, self.cert) @@ -235,7 +263,9 @@ class PBSDocsTest(unittest.TestCase): self.assertEqual(requests, [("/api2/json/admin/datastore", { **requests[0][1], "Authorization": AUTH_HEADER, })]) - self.assertEqual(result.stdout, b'{"data":[]}\n') + self.assertNotIn("X-Curlrc-Injected", requests[0][1]) + self.assertEqual(self.private_response(result), PRIVATE_BODY) + self.assertTrue(result.stdout.endswith(b"HTTP 200\n")) def test_auth_errors_retain_body_and_fail(self): for status in (401, 403): @@ -244,7 +274,8 @@ class PBSDocsTest(unittest.TestCase): result, _ = self.run_curl(Path(temporary), recipe("--fail-with-body"), port, self.cert) self.assertEqual(result.returncode, 22, result.stderr.decode()) self.assertEqual(len(requests), 1) - self.assertEqual(result.stdout, b'{"data":[]}\n') + self.assertEqual(self.private_response(result), PRIVATE_BODY) + self.assertTrue(result.stdout.endswith(f"HTTP {status}\n".encode())) def test_tls_rejects_untrusted_wrong_certificate_and_wrong_hostname(self): for ca, hostname in ((None, "localhost"), (self.other_cert, "localhost"), (self.cert, "127.0.0.1")): @@ -272,7 +303,8 @@ PY ''' with tempfile.TemporaryDirectory() as temporary, server(self.cert, self.key, installer=installer) as (port, requests): home = Path(temporary) - result, _ = self.run_curl(home, recipe('--output "$HOME/.config/pulse/pbs-agent-install.sh"'), port, self.cert) + result, _ = self.run_curl(home, recipe('download_file=$(mktemp "$config_dir/pbs-agent-download.XXXXXX")'), + port, self.cert) self.assertEqual(result.returncode, 0, result.stderr.decode()) self.assertEqual(requests[0][0], "/install.sh") self.assertNotIn("Authorization", requests[0][1]) @@ -296,7 +328,8 @@ PY def test_failed_installer_download_is_nonzero(self): with tempfile.TemporaryDirectory() as temporary, server(self.cert, self.key, 403) as (port, _requests): home = Path(temporary) - result, _ = self.run_curl(home, recipe('--output "$HOME/.config/pulse/pbs-agent-install.sh"'), port, self.cert) + result, _ = self.run_curl(home, recipe('download_file=$(mktemp "$config_dir/pbs-agent-download.XXXXXX")'), + port, self.cert) self.assertEqual(result.returncode, 22, result.stderr.decode()) self.assertFalse((home / ".config/pulse/pbs-agent-install.sh").exists())