Bind guide checks to private PBS and API response recipes

Repair deterministic governance and script-smoke failures against the reviewed guides without restoring public response output, literal credential commands or clobbering downloads. Exercise the actual shared API helper and independently verify private, unique response files, methods, bodies and hostile curl defaults. Contract-Neutral: Update test adapters for the existing reviewed credential-private guide contract; no runtime or guide change. Change-source: pulse-maintainer
2026-10-23 20:25:56 +00:00 · 2026-10-05 04:59:34 +01:00
parent 70bf9b8db9
commit 8ad902ed1f
3 changed files with 116 additions and 33 deletions
@@ -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://<pulse-ip>:7655/api/setup-script?type=pbs&host=https://<pbs-ip>:8007&pulse_url=http://<pulse-ip>:7655" | bash`,
 		`PULSE_SETUP_TOKEN=`,
 		`sudo env PULSE_SETUP_TOKEN=`,
 		`--insecure`,
+		`--location`,
+		`--trace`,
+		`--verbose`,
 		` -k`,
 		`curl -k`,
 		`--token "`,
@@ -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__":
@@ -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())