mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-10-24 04:33:17 +00:00
Keep API diagnostic credentials out of command arguments
Use a private header file and an authenticated read-only summary rather than passing tokens in argv to a public health check. Exercise both header formats, file permissions, preserved content and failing HTTP auth errors with synthetic local fixtures. Contract-Neutral: API authentication, routes, scopes and CSRF enforcement are unchanged; this updates existing usage guidance and executable documentation checks. Change-source: pulse-maintainer
This commit is contained in:
@@ -0,0 +1,148 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Exercise the API guide's credential-file commands with synthetic secrets.
|
||||
|
||||
The HTTP tests need guest-local loopback; run via pulse-worker-source-proof.
|
||||
No real Pulse instance, credentials or external network is used.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from contextlib import contextmanager
|
||||
from http.server import BaseHTTPRequestHandler, HTTPServer
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
import re
|
||||
import shutil
|
||||
import stat
|
||||
import subprocess
|
||||
import tempfile
|
||||
import threading
|
||||
import unittest
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
DOC = ROOT / "docs/API.md"
|
||||
TEST_TOKEN = "synthetic-doc-test-token"
|
||||
|
||||
|
||||
def auth_section() -> str:
|
||||
return DOC.read_text(encoding="utf-8").split("## 🔐 Authentication\n", 1)[1].split("\n## ", 1)[0]
|
||||
|
||||
|
||||
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
|
||||
|
||||
|
||||
@contextmanager
|
||||
def recording_server(status: int = 200):
|
||||
requests = []
|
||||
|
||||
class Handler(BaseHTTPRequestHandler):
|
||||
def do_GET(self):
|
||||
requests.append((self.path, dict(self.headers)))
|
||||
self.send_response(status)
|
||||
self.end_headers()
|
||||
self.wfile.write(b'{"fixture":true}\n')
|
||||
|
||||
def log_message(self, *_args):
|
||||
pass
|
||||
|
||||
server = HTTPServer(("127.0.0.1", 0), Handler)
|
||||
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
||||
thread.start()
|
||||
try:
|
||||
yield server.server_address[1], requests
|
||||
finally:
|
||||
server.shutdown()
|
||||
server.server_close()
|
||||
thread.join(timeout=5)
|
||||
|
||||
|
||||
class APIAuthDocsTest(unittest.TestCase):
|
||||
def test_credentials_are_not_documented_as_command_arguments(self):
|
||||
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("HTTPS", section)
|
||||
self.assertIn("does not verify authentication", section)
|
||||
self.assertIn("monitoring:read", section)
|
||||
self.assertIn("Do not paste tokens", section)
|
||||
|
||||
def test_shipped_copy_is_identical(self):
|
||||
self.assertEqual(DOC.read_bytes(), (ROOT / "frontend-modern/public/docs/API.md").read_bytes())
|
||||
|
||||
def test_preparation_protects_new_and_existing_header_files(self):
|
||||
preparation, _ = commands()
|
||||
with tempfile.TemporaryDirectory() as temporary:
|
||||
home = Path(temporary)
|
||||
tools = home / "tools"
|
||||
tools.mkdir()
|
||||
# The editor must be interactive in the real recipe; this fixture
|
||||
# verifies its filename without putting a synthetic token in argv.
|
||||
editor = tools / "vi"
|
||||
editor.write_text('#!/bin/sh\n[ "$#" = 1 ] && [ "$1" = "$HOME/.config/pulse/api-header" ]\n')
|
||||
editor.chmod(0o700)
|
||||
env = dict(os.environ, HOME=str(home), PATH=f"{tools}:{os.environ['PATH']}")
|
||||
header = home / ".config/pulse/api-header"
|
||||
for existing in (False, True):
|
||||
if existing:
|
||||
header.write_text(f"X-API-Token: {TEST_TOKEN}\n")
|
||||
header.chmod(0o644)
|
||||
subprocess.run(["bash", "-eu", "-c", preparation], env=env, check=True)
|
||||
self.assertEqual(stat.S_IMODE(header.stat().st_mode), 0o600)
|
||||
self.assertEqual(header.read_text(), f"X-API-Token: {TEST_TOKEN}\n" if existing else "")
|
||||
|
||||
def run_documented_request(self, home: Path, header_text: str, port: int):
|
||||
_, request = commands()
|
||||
header = home / ".config/pulse/api-header"
|
||||
header.parent.mkdir(parents=True, exist_ok=True)
|
||||
header.write_text(header_text + "\n")
|
||||
header.chmod(0o600)
|
||||
real_curl = shutil.which("curl")
|
||||
self.assertIsNotNone(real_curl, "curl is required to exercise the documented command")
|
||||
tools = home / "tools"
|
||||
tools.mkdir(exist_ok=True)
|
||||
recorder = tools / "curl"
|
||||
recorder.write_text(
|
||||
"#!/usr/bin/env python3\nimport json, os, sys\n"
|
||||
"from pathlib import Path\n"
|
||||
"Path(os.environ['ARGV_RECEIPT']).write_text(json.dumps(sys.argv[1:]))\n"
|
||||
"os.execv(os.environ['REAL_CURL'], [os.environ['REAL_CURL'], *sys.argv[1:]])\n"
|
||||
)
|
||||
recorder.chmod(0o700)
|
||||
receipt = home / "argv.json"
|
||||
env = dict(os.environ, HOME=str(home), PATH=f"{tools}:{os.environ['PATH']}",
|
||||
REAL_CURL=real_curl, ARGV_RECEIPT=str(receipt))
|
||||
request = request.replace("http://127.0.0.1:7655", f"http://127.0.0.1:{port}")
|
||||
result = subprocess.run(["bash", "-eu", "-c", request], env=env, capture_output=True, timeout=10)
|
||||
argv = json.loads(receipt.read_text())
|
||||
self.assertNotIn(TEST_TOKEN, " ".join(argv))
|
||||
self.assertIn("@" + str(header), argv)
|
||||
self.assertNotIn(b"synthetic-doc-test-token", result.stdout + result.stderr)
|
||||
return result
|
||||
|
||||
def test_header_file_sends_each_supported_header_without_exposing_argv(self):
|
||||
with tempfile.TemporaryDirectory() as temporary, recording_server() as (port, requests):
|
||||
for key, value in (("X-API-Token", TEST_TOKEN), ("Authorization", "Bearer " + TEST_TOKEN)):
|
||||
with self.subTest(header=key):
|
||||
result = self.run_documented_request(Path(temporary), f"{key}: {value}", port)
|
||||
self.assertEqual(result.returncode, 0, result.stderr.decode())
|
||||
self.assertEqual(requests[-1][0], "/api/state/summary")
|
||||
self.assertEqual(requests[-1][1][key], value)
|
||||
|
||||
def test_http_auth_errors_fail_instead_of_looking_successful(self):
|
||||
for status in (401, 403):
|
||||
with self.subTest(status=status), tempfile.TemporaryDirectory() as temporary:
|
||||
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)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user