mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-10-03 04:38:48 +00:00
603d4cad14
Use private header and passphrase files for copied API requests, check protected access instead of public health, and preserve actual transfer and lockout permissions. Replace unsafe legacy credential setup and key cleanup, explain deployment-specific recovery, and align existing mobile security guidance with retirement. Contract-Neutral: Existing security guidance, shipped mirrors and executable documentation fixtures only; authentication, encryption, scopes, entitlements and runtime behaviour are unchanged. Change-source: pulse-maintainer
161 lines
8.3 KiB
Python
161 lines
8.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Execute the canonical security guide with synthetic secrets and loopback.
|
|
|
|
Run via pulse-worker-source-proof. The fixture checks copied requests and private
|
|
files, not a deployed Pulse instance; existing Go tests check transfer authority,
|
|
encryption and lockout handlers.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import stat
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
|
|
from test_api_auth_docs import ROOT, TEST_TOKEN, exercise_curl, recording_server
|
|
|
|
|
|
DOC = ROOT / "SECURITY.md"
|
|
TEST_PASSPHRASE = "synthetic-security-export-passphrase"
|
|
EXPECTED = (
|
|
("GET", "/api/state/summary", None),
|
|
("POST", "/api/config/export", {"passphrase": TEST_PASSPHRASE}),
|
|
("GET", "/api/monitoring/scheduler/health", None),
|
|
("POST", "/api/security/reset-lockout", {"identifier": "username"}),
|
|
("POST", "/api/security/reset-lockout", {"identifier": "198.51.100.100"}),
|
|
)
|
|
|
|
|
|
def blocks() -> list[str]:
|
|
return re.findall(r"```bash\n(.*?)```", DOC.read_text(encoding="utf-8"), re.DOTALL)
|
|
|
|
|
|
def recipes() -> list[str]:
|
|
return [step for block in blocks() for step in re.split(r"\n(?=# )", block)
|
|
if "curl " in step]
|
|
|
|
|
|
def private_export_request(home: Path):
|
|
request = home / ".config/pulse/export-request.json"
|
|
request.parent.mkdir(parents=True, exist_ok=True)
|
|
request.write_text(json.dumps({"passphrase": TEST_PASSPHRASE}))
|
|
request.chmod(0o600)
|
|
|
|
|
|
class SecurityDocsTest(unittest.TestCase):
|
|
def test_shipped_copy_matches_canonical_guide(self):
|
|
self.assertEqual(DOC.read_bytes(),
|
|
(ROOT / "frontend-modern/public/docs/SECURITY.md").read_bytes())
|
|
|
|
def test_copied_requests_never_expose_credentials_or_weaken_tls(self):
|
|
requests = recipes()
|
|
self.assertEqual(len(requests), len(EXPECTED))
|
|
for request in requests:
|
|
self.assertNotRegex(request, r"(?:\b\w*TOKEN=|X-API-Token:|Authorization:|Bearer\s|--cookie\b)")
|
|
self.assertNotRegex(request, r"(?:--insecure|--verbose|--trace\S*|--location|\s-k\b)")
|
|
self.assertIn('curl --disable --fail-with-body --header "@$HOME/.config/pulse/api-header"', request)
|
|
if "/api/config/export" in request:
|
|
self.assertIn('--data-binary "@$HOME/.config/pulse/export-request.json"', request)
|
|
self.assertNotIn('"passphrase":', request)
|
|
elif "--request POST" in request:
|
|
self.assertIn("--data-binary @-", request)
|
|
self.assertIn("<<'JSON'", request)
|
|
shell = "\n".join(blocks())
|
|
self.assertNotRegex(shell, r"(?:--token\s|PULSE_AUTH_PASS=|API_TOKEN[S]?=|curl[^\n]*\|)")
|
|
self.assertNotIn("rm -rf", shell)
|
|
self.assertNotIn("authorized_keys", shell)
|
|
|
|
def test_export_preparation_protects_new_and_existing_files_without_erasing_them(self):
|
|
preparation = next(block for block in blocks() if 'vi "$HOME/.config/pulse/export-request.json"' in block)
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
home = Path(temporary)
|
|
tools = home / "tools"
|
|
tools.mkdir()
|
|
editor = tools / "vi"
|
|
editor.write_text('#!/bin/sh\n[ "$#" = 1 ] && [ "$1" = "$HOME/.config/pulse/export-request.json" ]\n')
|
|
editor.chmod(0o700)
|
|
env = dict(os.environ, HOME=str(home), PATH=f"{tools}:{os.environ['PATH']}")
|
|
request = home / ".config/pulse/export-request.json"
|
|
for existing in (False, True):
|
|
with self.subTest(existing=existing):
|
|
if existing:
|
|
private_export_request(home)
|
|
request.chmod(0o644)
|
|
request.parent.chmod(0o755)
|
|
subprocess.run(["bash", "-eu", "-c", preparation], env=env, check=True, timeout=10)
|
|
self.assertEqual(stat.S_IMODE(request.stat().st_mode), 0o600)
|
|
self.assertEqual(stat.S_IMODE(request.parent.stat().st_mode), 0o700)
|
|
self.assertEqual(request.read_text(), json.dumps({"passphrase": TEST_PASSPHRASE}) if existing else "")
|
|
|
|
def test_exact_methods_paths_and_bodies_with_both_supported_headers(self):
|
|
steps = recipes()
|
|
with recording_server() as (port, requests):
|
|
for key, value in (("X-API-Token", TEST_TOKEN), ("Authorization", "Bearer " + TEST_TOKEN)):
|
|
for step, (method, path, body) in zip(steps, EXPECTED):
|
|
with self.subTest(header=key, method=method, path=path), tempfile.TemporaryDirectory() as temporary:
|
|
home = Path(temporary)
|
|
private_export_request(home)
|
|
result = exercise_curl(self, home, f"{key}: {value}", port, step)
|
|
self.assertEqual(result.returncode, 0, result.stderr.decode())
|
|
self.assertEqual(requests[-1][0], path)
|
|
self.assertEqual(requests[-1][1][key], value)
|
|
self.assertNotIn("X-Curlrc-Injected", requests[-1][1])
|
|
self.assertEqual(requests[-1][2], method)
|
|
self.assertEqual(json.loads(requests[-1][3]) if body is not None else None, body)
|
|
self.assertNotIn(TEST_PASSPHRASE.encode(), result.stdout + result.stderr)
|
|
self.assertNotIn(TEST_PASSPHRASE, (home / "argv.json").read_text())
|
|
if path == "/api/config/export":
|
|
self.check_export_files(home, result, success=True)
|
|
|
|
def check_export_files(self, home: Path, result, *, success: bool):
|
|
directories = list(home.glob("pulse-export.*"))
|
|
self.assertEqual(len(directories), 1)
|
|
directory = directories[0]
|
|
output = directory / "config-export.json"
|
|
self.assertEqual(stat.S_IMODE(directory.stat().st_mode), 0o700)
|
|
self.assertEqual(stat.S_IMODE(output.stat().st_mode), 0o600)
|
|
# A recording server does not prove encryption. This only checks the
|
|
# exact response is saved privately, with truthful HTTP success/failure.
|
|
self.assertEqual(output.read_text(), '{"fixture":true}\n')
|
|
self.assertNotIn(TEST_TOKEN, output.read_text())
|
|
self.assertNotIn(TEST_PASSPHRASE, output.read_text())
|
|
if success:
|
|
self.assertIn(b"Export response saved", result.stdout)
|
|
else:
|
|
self.assertNotIn(b"Export response saved", result.stdout)
|
|
self.assertIn(b"Export failed; do not import", result.stderr)
|
|
|
|
def test_unauthorised_and_wrong_scope_responses_never_report_success(self):
|
|
for status in (401, 403, 500):
|
|
with self.subTest(status=status), recording_server(status) as (port, requests):
|
|
for step, (_, path, _) in zip(recipes(), EXPECTED):
|
|
with self.subTest(path=path), tempfile.TemporaryDirectory() as temporary:
|
|
home = Path(temporary)
|
|
private_export_request(home)
|
|
result = exercise_curl(self, home, f"X-API-Token: {TEST_TOKEN}", port, step)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertEqual(requests[-1][0], path)
|
|
if path == "/api/config/export":
|
|
self.check_export_files(home, result, success=False)
|
|
|
|
def test_guidance_retains_authority_and_credential_lifecycle_boundaries(self):
|
|
guide = " ".join(DOC.read_text().split())
|
|
for boundary in ("monitoring:read", "settings:read", "settings:write", "CSRF",
|
|
"organization-bound", "direct loopback", "export only",
|
|
"12 characters", "public", "separate from the export",
|
|
"entire configuration bundle", "31 March 2027", "no longer sold",
|
|
"does not reset a password", "deleting `.env` is not a universal reset",
|
|
"fingerprint", "Revoking the public key", "privileged inspection"):
|
|
self.assertTrue(boundary in guide, f"missing safety boundary: {boundary}")
|
|
self.assertIn("docs/UNIFIED_AGENT.md#private-file-installation-linux-macos-and-nas", guide)
|
|
self.assertIn("docs/TROUBLESHOOTING.md#i-forgot-my-password", guide)
|
|
self.assertNotIn("### What's **Not** Encrypted", guide)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|