mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-10-24 20:52:51 +00:00
pulse-triage[bot]
04429cb42c
Replace token arguments, an ignored API_TOKENS bootstrap and plaintext DaemonSet credentials with scoped UI-created tokens, private files and a pre-created Secret. Route hosted installation to the existing private-file flow and retain HTTP errors and encoded resource identities in metrics queries. Exercise copied recipes, failure stops and actual curl requests with synthetic credentials. Contract-Neutral: Documentation-only private-file setup and metrics access guidance; no runtime, API, entitlement or deployment contract delta. Change-source: pulse-maintainer
218 lines
11 KiB
Python
218 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""Exercise copied setup/metrics recipes with synthetic credentials.
|
|
|
|
Run with pulse-worker-source-proof --workspace scripts. The kubectl recorder
|
|
checks local argv, pipe payloads and stop behaviour, not a real cluster. Curl
|
|
uses guest-local listeners; no real Pulse instance or token is involved.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import stat
|
|
import subprocess
|
|
import tempfile
|
|
import unittest
|
|
from urllib.parse import parse_qs, urlsplit
|
|
|
|
from test_api_auth_docs import ROOT, TEST_TOKEN, exercise_curl, recording_server
|
|
|
|
|
|
GUIDES = ("KUBERNETES", "METRICS_HISTORY", "CLOUD")
|
|
|
|
|
|
def guide(name: str) -> str:
|
|
return (ROOT / "docs" / f"{name}.md").read_text(encoding="utf-8")
|
|
|
|
|
|
def blocks(name: str, language: str = "bash") -> list[str]:
|
|
return re.findall(r"```" + language + r"\n(.*?)```", guide(name), re.DOTALL)
|
|
|
|
|
|
def recipe(name: str, needle: str) -> str:
|
|
matches = [block for block in blocks(name) if needle in block]
|
|
if len(matches) != 1:
|
|
raise AssertionError(f"expected one {name} recipe containing {needle!r}")
|
|
return matches[0]
|
|
|
|
|
|
KUBECTL_RECORDER = '''#!/usr/bin/env python3
|
|
import base64, json, os, sys
|
|
from pathlib import Path
|
|
args = sys.argv[1:]
|
|
assert os.environ['FIXTURE_TOKEN'] not in ' '.join(args)
|
|
assert not any(k in os.environ for k in ('PULSE_TOKEN', 'API_TOKENS'))
|
|
if args[:2] == ['create', 'namespace']:
|
|
phase = 'namespace-create'
|
|
assert args == ['create', 'namespace', 'pulse', '--dry-run=client', '-o', 'yaml']
|
|
document = {'kind': 'Namespace', 'metadata': {'name': 'pulse'}}
|
|
elif args[:5] == ['-n', 'pulse', 'create', 'secret', 'generic']:
|
|
phase = 'secret-create'
|
|
assert args[5] == 'pulse-agent-env'
|
|
assert args[7:] == ['--dry-run=client', '-o', 'yaml']
|
|
assert args[6].startswith('--from-file=PULSE_TOKEN=')
|
|
path = Path(args[6].split('=', 2)[2])
|
|
assert path.read_text().strip() == os.environ['FIXTURE_TOKEN']
|
|
assert path.stat().st_mode & 0o777 == 0o600
|
|
document = {'apiVersion': 'v1', 'kind': 'Secret',
|
|
'metadata': {'name': 'pulse-agent-env', 'namespace': 'pulse'},
|
|
'data': {'PULSE_TOKEN': base64.b64encode(path.read_bytes()).decode()}}
|
|
else:
|
|
assert args == ['apply', '--server-side', '--field-manager=pulse-agent-setup', '-f', '-']
|
|
payload = sys.stdin.read()
|
|
document = json.loads(payload) if payload else None
|
|
phase = 'secret-apply' if document and document['kind'] == 'Secret' else 'namespace-apply'
|
|
with Path(os.environ['KUBECTL_RECEIPT']).open('a') as output:
|
|
output.write(json.dumps({'phase': phase, 'argv': args, 'document': document}) + '\\n')
|
|
if phase == os.environ.get('FAIL_PHASE'):
|
|
sys.exit(23)
|
|
if args[0] == 'apply':
|
|
if document is None:
|
|
sys.exit(24)
|
|
print(document['kind'].lower() + '/' + document['metadata']['name'] + ' serverside-applied')
|
|
else:
|
|
print(json.dumps(document))
|
|
'''
|
|
|
|
|
|
class SetupAccessDocsTest(unittest.TestCase):
|
|
def test_mirrors_and_secret_free_shell_recipes(self):
|
|
for name in GUIDES:
|
|
with self.subTest(guide=name):
|
|
self.assertEqual((ROOT / "docs" / f"{name}.md").read_bytes(),
|
|
(ROOT / "frontend-modern/public/docs" / f"{name}.md").read_bytes())
|
|
shell = "\n".join(blocks(name))
|
|
self.assertNotRegex(shell, r"(?:export\s+PULSE_TOKEN|--from-literal=\w*TOKEN|--token(?:\s|=)|X-API-Token:|Bearer\s)")
|
|
self.assertNotRegex(shell, r"(?:--insecure|--verbose|--trace\S*|\s-k\b|curl[^`]*\|\s*bash)")
|
|
for block in blocks(name):
|
|
subprocess.run(["bash", "-n", "-c", block], check=True, capture_output=True)
|
|
|
|
def test_private_file_preparation_preserves_existing_token(self):
|
|
command = recipe("KUBERNETES", 'vi "$HOME/.config/pulse/kubernetes-agent-token"')
|
|
with tempfile.TemporaryDirectory(prefix="setup docs ") as temporary:
|
|
home = Path(temporary)
|
|
tools = home / "tools"
|
|
tools.mkdir()
|
|
editor = tools / "vi"
|
|
editor.write_text('#!/bin/sh\n[ "$#" = 1 ] && [ -f "$1" ]\n')
|
|
editor.chmod(0o700)
|
|
env = {**os.environ, "HOME": str(home), "PATH": f"{tools}:{os.environ['PATH']}"}
|
|
token = home / ".config/pulse/kubernetes-agent-token"
|
|
for existing in (False, True):
|
|
if existing:
|
|
token.write_text(TEST_TOKEN + "\n")
|
|
token.chmod(0o644)
|
|
token.parent.chmod(0o755)
|
|
result = subprocess.run(["bash", "-eu", "-c", command], env=env, capture_output=True, timeout=10)
|
|
self.assertEqual(result.returncode, 0, result.stderr.decode())
|
|
self.assertEqual(stat.S_IMODE(token.stat().st_mode), 0o600)
|
|
self.assertEqual(stat.S_IMODE(token.parent.stat().st_mode), 0o700)
|
|
self.assertEqual(token.read_text(), TEST_TOKEN + "\n" if existing else "")
|
|
self.assertNotIn(TEST_TOKEN.encode(), result.stdout + result.stderr)
|
|
|
|
def run_secret(self, home: Path, fail: str = "", empty: bool = False):
|
|
private = home / ".config/pulse"
|
|
private.mkdir(parents=True, mode=0o700)
|
|
token = private / "kubernetes-agent-token"
|
|
token.write_text("" if empty else TEST_TOKEN + "\n")
|
|
token.chmod(0o600)
|
|
tools = home / "tools"
|
|
tools.mkdir()
|
|
recorder = tools / "kubectl"
|
|
recorder.write_text(KUBECTL_RECORDER)
|
|
recorder.chmod(0o700)
|
|
receipt = home / "kubectl.jsonl"
|
|
env = {**os.environ, "HOME": str(home), "PATH": f"{tools}:{os.environ['PATH']}",
|
|
"FIXTURE_TOKEN": TEST_TOKEN, "KUBECTL_RECEIPT": str(receipt), "FAIL_PHASE": fail}
|
|
for key in ("PULSE_TOKEN", "API_TOKENS"):
|
|
env.pop(key, None)
|
|
result = subprocess.run(["bash", "-c", recipe("KUBERNETES", "--from-file=PULSE_TOKEN=")],
|
|
env=env, capture_output=True, timeout=10)
|
|
records = [json.loads(line) for line in receipt.read_text().splitlines()] if receipt.exists() else []
|
|
self.assertNotIn(TEST_TOKEN.encode(), result.stdout + result.stderr)
|
|
self.assertNotIn(base64.b64encode((TEST_TOKEN + "\n").encode()), result.stdout + result.stderr)
|
|
return result, records
|
|
|
|
def test_secret_passes_only_file_path_in_argv_and_data_over_stdin(self):
|
|
with tempfile.TemporaryDirectory(prefix="setup docs ") as temporary:
|
|
result, records = self.run_secret(Path(temporary))
|
|
self.assertEqual(result.returncode, 0, result.stderr.decode())
|
|
self.assertEqual(sorted(r["phase"] for r in records),
|
|
sorted(("namespace-create", "namespace-apply", "secret-create", "secret-apply")))
|
|
created = next(r["document"] for r in records if r["phase"] == "secret-create")
|
|
applied = next(r["document"] for r in records if r["phase"] == "secret-apply")
|
|
self.assertEqual(created, applied)
|
|
self.assertEqual(base64.b64decode(applied["data"]["PULSE_TOKEN"]), (TEST_TOKEN + "\n").encode())
|
|
self.assertNotIn("annotations", applied["metadata"])
|
|
self.assertNotIn(TEST_TOKEN, json.dumps([r["argv"] for r in records]))
|
|
|
|
def test_pipeline_failure_and_empty_file_stop_provisioning(self):
|
|
for phase in ("namespace-create", "namespace-apply", "secret-create", "secret-apply", ""):
|
|
with self.subTest(phase=phase), tempfile.TemporaryDirectory() as temporary:
|
|
result, records = self.run_secret(Path(temporary), fail=phase, empty=not phase)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
if phase.startswith("namespace"):
|
|
self.assertFalse(any(r["phase"].startswith("secret") for r in records))
|
|
if not phase:
|
|
self.assertEqual(records, [])
|
|
|
|
def test_daemonset_and_helm_reference_the_same_scoped_secret(self):
|
|
manifest = next(block for block in blocks("KUBERNETES", "yaml") if "kind: DaemonSet" in block)
|
|
self.assertRegex(manifest, r"name: PULSE_TOKEN\n\s+valueFrom:\n\s+secretKeyRef:\n\s+name: pulse-agent-env\n\s+key: PULSE_TOKEN")
|
|
self.assertNotIn("YOUR_API_TOKEN", manifest)
|
|
collector = recipe("KUBERNETES", "openShift.kubernetesAgent.enabled=true")
|
|
self.assertIn("agent.secretEnv.name=pulse-agent-env", collector)
|
|
self.assertIn("agent.secretEnv.keys[0]=PULSE_TOKEN", collector)
|
|
self.assertNotIn("server.secretEnv", collector)
|
|
chart = (ROOT / "deploy/helm/pulse/templates/agent.yaml").read_text()
|
|
for key in (".Values.agent.secretEnv", '"valueFrom"', '"secretKeyRef"', '"pulse.agentSecretName"'):
|
|
self.assertIn(key, chart)
|
|
content = " ".join(guide("KUBERNETES").split())
|
|
for boundary in ("kubernetes:report", "agent:report", "Do not reuse an administrator token",
|
|
"does not provision API tokens", "not inherently encrypted", "RBAC",
|
|
"restarting the agent pods", "forcing the conflict"):
|
|
self.assertIn(boundary, content)
|
|
|
|
def test_cloud_uses_existing_private_file_installation_not_server_installer(self):
|
|
content = " ".join(guide("CLOUD").split())
|
|
for boundary in ("UNIFIED_AGENT.md#private-file-installation-linux-macos-and-nas", "--token-file",
|
|
"host being monitored", "Windows", "HTTPS", "stop if the download fails",
|
|
"top-level GitHub server installer", "Keep TLS verification enabled"):
|
|
self.assertIn(boundary, content)
|
|
|
|
|
|
class MetricsAccessDocsHTTPTest(unittest.TestCase):
|
|
def test_history_query_with_both_headers_and_hostile_curl_defaults(self):
|
|
request = recipe("METRICS_HISTORY", "curl ")
|
|
expected_id = "pve 1:node/1:100&other=1"
|
|
request = request.replace("pve1:node1:100", expected_id)
|
|
with recording_server() as (port, records):
|
|
for key, value in (("X-API-Token", TEST_TOKEN), ("Authorization", "Bearer " + TEST_TOKEN)):
|
|
with self.subTest(header=key), tempfile.TemporaryDirectory() as temporary:
|
|
result = exercise_curl(self, Path(temporary), f"{key}: {value}", port, request)
|
|
self.assertEqual(result.returncode, 0, result.stderr.decode())
|
|
path, headers, method, body = records[-1]
|
|
self.assertEqual(urlsplit(path).path, "/api/metrics-store/history")
|
|
self.assertEqual(parse_qs(urlsplit(path).query), {
|
|
"resourceType": ["vm"], "resourceId": [expected_id], "range": ["7d"], "metric": ["cpu"]})
|
|
self.assertEqual((method, body), ("GET", b""))
|
|
self.assertEqual(headers[key], value)
|
|
self.assertNotIn("X-Curlrc-Injected", headers)
|
|
|
|
def test_http_auth_license_and_server_errors_fail(self):
|
|
for status in (401, 402, 403, 500):
|
|
with self.subTest(status=status), recording_server(status) as (port, records), tempfile.TemporaryDirectory() as temporary:
|
|
result = exercise_curl(self, Path(temporary), f"X-API-Token: {TEST_TOKEN}", port,
|
|
recipe("METRICS_HISTORY", "curl "))
|
|
self.assertEqual(result.returncode, 22, result.stderr.decode())
|
|
self.assertEqual(len(records), 1)
|
|
self.assertIn(b'{"fixture":true}', result.stdout)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|