mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-10-25 13:10:03 +00:00
pulse-triage[bot]
5d748986b3
Admit only the manual helper's exact new curl envelope in the existing closed transport fixture. Require a signed emitted-helper round trip and reject altered deadlines, retries and TLS flags through the registered installer suite; do not widen the adapter or execute a native journey. Change-source: pulse-maintainer
806 lines
50 KiB
Python
806 lines
50 KiB
Python
#!/usr/bin/env python3
|
|
"""Discriminating controls for the native CI recipe; no host manager/root use."""
|
|
import contextlib
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
SPEC = importlib.util.spec_from_file_location("native_updater", ROOT / "scripts/installtests/auto_update_systemd.py")
|
|
native = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(native)
|
|
WORKFLOW = ROOT / ".github/workflows/build-and-test.yml"
|
|
STEP = "Check current-source unattended recovery on native systemd"
|
|
FIXTURE_COMMANDS = ("bash", "tar", "chmod", "mv", "timeout", "cmp")
|
|
LINUX_CONTROL = "test_connected_signed_archive_and_recovery_controls_without_host_manager"
|
|
LINUX_ONLY_REASON = "Linux server ELF execution; portable controls remain required on macOS"
|
|
|
|
|
|
def selected_fixture_commands():
|
|
# The containing native job has already supplied its Linux-only tools.
|
|
# Capture only these executables, not the whole ambient PATH or environment.
|
|
commands = {name: shutil.which(name) for name in FIXTURE_COMMANDS}
|
|
for name, executable in commands.items():
|
|
if executable is None:
|
|
raise ValueError("missing installer fixture tool: " + name)
|
|
return commands
|
|
|
|
|
|
def recipe(source):
|
|
job = re.search(r"(?ms)^ frontend:\n(.*?)(?=^ [a-z][a-z0-9-]*:|\Z)", source)[1]
|
|
steps = dict(re.findall(r"(?ms)^ - name: ([^\n]+)\n(.*?)(?=^ - name:|\Z)", job))
|
|
order = list(steps)
|
|
assert order.index("Verify backend with real frontend embed") < order.index(STEP) < order.index("Require frontend dependency audit")
|
|
step = steps[STEP]
|
|
assert not re.search(r"^ (if|continue-on-error|working-directory):", step, re.M)
|
|
assert "runs-on: ubuntu-24.04" in job and "timeout-minutes: 30" in job
|
|
assert "persist-credentials: false" in steps["Checkout repository"]
|
|
assert "EXPECTED_SOURCE_SHA: ${{ github.sha }}" in step
|
|
shell = re.search(r"(?ms)^ run: \|\n((?: [^\n]*\n)+)", step)[1]
|
|
shell = "\n".join(line[10:] for line in shell.splitlines()) + "\n"
|
|
for command in ('test "$(git rev-parse HEAD)" = "$EXPECTED_SOURCE_SHA"',
|
|
"git diff --quiet HEAD",
|
|
"diff -qr --no-dereference frontend-modern/dist internal/api/frontend-modern/dist",
|
|
"-ldflags='-X main.Version=v0.0.1'", "-ldflags='-X main.Version=v0.0.2'",
|
|
"go build -mod=readonly -p=2", "sudo env -i",
|
|
"GITHUB_ACTIONS=true RUNNER_ENVIRONMENT=github-hosted GITHUB_REPOSITORY=rcourtman/Pulse",
|
|
"python3 scripts/installtests/auto_update_systemd.py",
|
|
'--receipt "$RUNNER_TEMP/pulse-native-updater.json"'):
|
|
assert command in shell, command
|
|
assert "GH_TOKEN" not in shell and "secrets." not in step
|
|
retention = steps["Retain bounded native updater receipt"]
|
|
assert "if: always()" in retention and "retention-days: 7" in retention
|
|
assert "path: ${{ runner.temp }}/pulse-native-updater.json" in retention
|
|
assert "*.log" not in retention and "pulse-native-before" not in retention
|
|
return shell
|
|
|
|
|
|
class NativeContracts(unittest.TestCase):
|
|
def test_fixture_command_selection_refuses_missing_tools_without_fallback(self):
|
|
for missing in FIXTURE_COMMANDS:
|
|
with self.subTest(missing=missing), patch.object(shutil, "which", side_effect=
|
|
lambda name: None if name == missing else "/supplied/" + name):
|
|
with self.assertRaisesRegex(ValueError, "missing installer fixture tool: " + missing):
|
|
selected_fixture_commands()
|
|
|
|
def test_definition_boundary_is_exact_and_rejects_drift(self):
|
|
source = (ROOT / "install.sh").read_text()
|
|
definitions = native.installer_definitions(source)
|
|
self.assertEqual(definitions, source[:source.index(native.PARSE_MARKER)])
|
|
for changed in (source.replace(native.PARSE_MARKER, "# moved\n"),
|
|
source.replace(native.PARSE_MARKER, native.PARSE_MARKER * 2),
|
|
source.replace(native.PARSE_MARKER, native.PARSE_MARKER + "echo different entry\n")):
|
|
with self.assertRaises(ValueError): native.installer_definitions(changed)
|
|
|
|
def test_host_admission_requires_fresh_public_vm_not_worker_or_container(self):
|
|
env = dict(GITHUB_ACTIONS="true", RUNNER_ENVIRONMENT="github-hosted", GITHUB_REPOSITORY="rcourtman/Pulse")
|
|
with patch.object(Path, "exists", return_value=False):
|
|
native.hosted_admission(env, 0, "a" * 40)
|
|
for changed, uid, sha in ((env, 1000, "a" * 40), ({}, 0, "a" * 40),
|
|
({**env, "RUNNER_ENVIRONMENT": "self-hosted"}, 0, "a" * 40),
|
|
({**env, "GITHUB_REPOSITORY": "other/repo"}, 0, "a" * 40),
|
|
(env, 0, "main")):
|
|
with self.assertRaises(ValueError): native.hosted_admission(changed, uid, sha)
|
|
with patch.object(Path, "exists", return_value=True):
|
|
with self.assertRaises(ValueError): native.hosted_admission(env, 0, "a" * 40)
|
|
|
|
def test_transport_and_private_allocator_have_no_online_fallback(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory); (root / "tmp").mkdir()
|
|
script = root / "curl"; script.write_text(native.TRANSPORT); script.chmod(0o755)
|
|
(root / "installer").write_text("signed fixture\n")
|
|
(root / "installer.sshsig").write_text("signature sidecar\n")
|
|
env = {**os.environ, "PULSE_NATIVE_ROOT": str(root)}
|
|
for tag in ("latest/download", "download/v0.0.1", "download/v0.0.2"):
|
|
for suffix in ("", ".sshsig"):
|
|
result = subprocess.run([str(script), "-fsSL", f"https://github.com/rcourtman/Pulse/releases/{tag}/install.sh{suffix}"], env=env, capture_output=True)
|
|
self.assertEqual(result.returncode, 0)
|
|
self.assertEqual(result.stdout, (root / ("installer" + suffix)).read_bytes())
|
|
for url in ("https://example.com/install.sh", "https://github.com/rcourtman/Pulse/releases/download/v6.5.0/install.sh"):
|
|
result = subprocess.run([str(script), "-fsSL", url, "-o", str(root / "unexpected")], env=env, capture_output=True)
|
|
self.assertEqual(result.returncode, 97)
|
|
self.assertFalse((root / "unexpected").exists())
|
|
allocator = root / "mktemp"; allocator.write_text(native.TRANSPORT); allocator.chmod(0o755)
|
|
result = subprocess.run([str(allocator), "-d", "/tmp/pulse-backup.XXXXXX"], env=env, capture_output=True, check=True)
|
|
self.assertEqual(Path(result.stdout.decode().strip()).parent, root / "tmp")
|
|
|
|
def test_transport_accepts_actual_bounded_downloader_and_rejects_option_drift(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
script = root / "curl"; script.write_text(native.TRANSPORT); script.chmod(0o755)
|
|
payload = b"private signed fixture bytes\n"
|
|
(root / "installer").write_bytes(payload)
|
|
(root / "installer.sshsig").write_bytes(b"private signature sidecar\n")
|
|
env = {**os.environ, "PULSE_NATIVE_ROOT": str(root),
|
|
"PATH": str(root) + os.pathsep + os.environ["PATH"]}
|
|
url = "https://github.com/rcourtman/Pulse/releases/download/v0.0.2/install.sh"
|
|
shell = ('source "$1"; download_update_asset "$2" "$3"')
|
|
for suffix in ("", ".sshsig"):
|
|
target = root / "downloaded"
|
|
result = subprocess.run(["bash", "-c", shell, "fixture",
|
|
str(ROOT / "scripts/pulse-auto-update.sh"), url + suffix, str(target)],
|
|
env=env, capture_output=True)
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertEqual(target.read_bytes(), (root / ("installer" + suffix)).read_bytes())
|
|
target.unlink()
|
|
flags = ["--disable", "-fsSL", "--connect-timeout", "5", "--max-time", "60",
|
|
"--proto", "=https", "--proto-redir", "=https"]
|
|
for changed in (flags[1:], [*flags, "--insecure"],
|
|
[arg if arg != "60" else "61" for arg in flags],
|
|
[arg if arg != "=https" else "=all" for arg in flags],
|
|
[*flags, "https://example.com/install.sh"],
|
|
[*flags, "--max-time", "60"]):
|
|
with self.subTest(flags=changed):
|
|
result = subprocess.run([str(script), *changed, url, "-o", str(root / "refused")],
|
|
env=env, capture_output=True)
|
|
self.assertEqual(result.returncode, 97)
|
|
self.assertFalse((root / "refused").exists())
|
|
|
|
def test_transport_accepts_emitted_manual_helper_and_rejects_option_drift(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
for leaf in ("tools", "tmp", "config"):
|
|
(root / leaf).mkdir(mode=0o700)
|
|
for name in ("curl", "mktemp"):
|
|
path = root / "tools" / name
|
|
path.write_text(native.TRANSPORT)
|
|
path.chmod(0o755)
|
|
installer = root / "installer"
|
|
installer.write_text('#!/bin/bash\nprintf "%s\\n" "$@" > "$PULSE_NATIVE_ROOT/executed"\n')
|
|
key = root / "key"
|
|
subprocess.run(["ssh-keygen", "-q", "-t", "ed25519", "-N", "", "-f", str(key)],
|
|
check=True, capture_output=True)
|
|
subprocess.run(["ssh-keygen", "-q", "-Y", "sign", "-f", str(key),
|
|
"-n", "pulse-install", str(installer)], check=True, capture_output=True)
|
|
Path(str(installer) + ".sig").rename(root / "installer.sshsig")
|
|
helper = root / "update"
|
|
env = {**os.environ, "PULSE_NATIVE_ROOT": str(root),
|
|
"FIXTURE_KEY": (root / "key.pub").read_text().strip(),
|
|
"PULSE_INSTALL_DIR": str(root), "PULSE_CONFIG_DIR": str(root / "config"),
|
|
"PULSE_UPDATE_HELPER_PATH": str(helper),
|
|
"PULSE_PROFILE_PATH": str(root / "profile"),
|
|
"PULSE_BASHRC_PATH": str(root / "bashrc"),
|
|
"PATH": str(root / "tools") + os.pathsep + os.environ["PATH"]}
|
|
generated = subprocess.run(["bash", "-c", 'source "$1"; '
|
|
'PINNED_RELEASE_SSH_PUBLIC_KEY="$FIXTURE_KEY"; setup_update_command',
|
|
"fixture", str(ROOT / "install.sh")], env=env, capture_output=True)
|
|
self.assertEqual(generated.returncode, 0, generated.stdout + generated.stderr)
|
|
result = subprocess.run(["bash", str(helper), "--version", "v0.0.1"],
|
|
env=env, capture_output=True, timeout=10)
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertEqual((root / "executed").read_text(), "--version\nv0.0.1\n")
|
|
self.assertEqual(list((root / "tmp").iterdir()), [])
|
|
flags = ["-q", "-fsSL", "--connect-timeout", "10", "--max-time", "60"]
|
|
url = "https://github.com/rcourtman/Pulse/releases/latest/download/install.sh"
|
|
for changed in (flags[1:], [*flags, "--insecure"], [*flags, "--retry", "3"],
|
|
[arg if arg != "60" else "61" for arg in flags],
|
|
[arg if arg != "10" else "11" for arg in flags]):
|
|
with self.subTest(flags=changed):
|
|
target = root / "refused"
|
|
result = subprocess.run([str(root / "tools/curl"), *changed, url, "-o", str(target)],
|
|
env=env, capture_output=True, timeout=10)
|
|
self.assertEqual(result.returncode, 97)
|
|
self.assertFalse(target.exists())
|
|
|
|
def test_real_verifier_accepts_fixture_and_refuses_tampering(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
lab = native.NativeLab(root, ROOT, root / "a", root / "b", "a" * 40)
|
|
(root / "signer").mkdir()
|
|
lab.run("ssh-keygen", "-q", "-t", "ed25519", "-N", "", "-f", str(root / "signer/key"))
|
|
installer = root / "installer"; installer.write_text(native.INSTALLER)
|
|
lab.sign(installer)
|
|
shell = ('source "$PULSE_NATIVE_UPDATER"; '
|
|
'PINNED_RELEASE_SSH_PUBLIC_KEY=$(cat "$PULSE_NATIVE_ROOT/signer/key.pub"); '
|
|
'verify_release_signature "$PULSE_NATIVE_ROOT/installer" "$PULSE_NATIVE_ROOT/installer.sshsig" fixture')
|
|
self.assertEqual(lab.shell(shell).returncode, 0)
|
|
installer.write_text(native.INSTALLER + "# tampered\n")
|
|
result = subprocess.run(["/bin/bash", "-c", shell], env=lab.env, capture_output=True)
|
|
self.assertEqual(result.returncode, 1)
|
|
self.assertIn(b"Cryptographic signature verification failed", result.stdout)
|
|
|
|
def test_signer_failure_retains_operation_and_metadata_not_private_output(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
(root / "signer").mkdir(mode=0o700)
|
|
lab = native.NativeLab(root, ROOT, root / "a", root / "b", "a" * 40)
|
|
lab.signer_command("generate-ephemeral-key", "-q", "-t", "ed25519", "-N", "", "-f",
|
|
str(root / "signer/key"))
|
|
target = root / "installer"
|
|
target.write_text("closed synthetic fixture, not a release\n")
|
|
# A real OpenSSH refusal, not a simulated exit code or key read.
|
|
(root / "signer/key").chmod(0o644)
|
|
with self.assertRaisesRegex(RuntimeError, "native command failed"):
|
|
lab.sign(target)
|
|
failure = lab.receipt["command_failure"]
|
|
self.assertEqual(failure["operation"], "sign-installer")
|
|
self.assertEqual(failure["observed_exit"], 255)
|
|
self.assertIn("private-key-permissions", failure["observed_diagnostics"])
|
|
self.assertEqual(failure["signer_key_metadata"], {
|
|
"present": True, "regular_file": True, "mode": "0644", "owned_by_executor": True,
|
|
})
|
|
self.assertFalse(Path(str(target) + ".sshsig").exists())
|
|
retained = json.dumps(lab.receipt)
|
|
self.assertNotIn("PRIVATE KEY", retained)
|
|
self.assertNotIn(str(root), retained)
|
|
self.assertNotIn(lab._last_output.decode(), retained)
|
|
|
|
def test_prepare_owns_generated_signer_mode_before_first_real_signature(self):
|
|
# Reproduce the exact observed mode, not its unknown generation cause.
|
|
# Run real OpenSSH signing before stopping ahead of account/unit writes.
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
base = Path(directory)
|
|
before, after = base / "a", base / "b"
|
|
before.write_bytes(b"before\n"); after.write_bytes(b"after\n")
|
|
lab = native.NativeLab(base / "lab", ROOT, before, after, "a" * 40)
|
|
lab.unit = base / "absent.service"
|
|
original_run = lab.run
|
|
def confined_run(*args, **kwargs):
|
|
if args[0] in (str(before), str(after)) and args[1:] == ("--version",):
|
|
tag = native.TAGS[0 if args[0] == str(before) else 1]
|
|
return subprocess.CompletedProcess(args, 0, ("Pulse " + tag + "\n").encode())
|
|
if args[0] != "ssh-keygen":
|
|
raise AssertionError("unexpected service/account operation")
|
|
result = original_run(*args, **kwargs)
|
|
if args[1] == "-q":
|
|
(lab.root / "signer/key").chmod(0o644)
|
|
return result
|
|
def first_signature(archive):
|
|
self.assertEqual((lab.root / "signer/key").stat().st_mode & 0o777, 0o600)
|
|
native.NativeLab.sign(lab, archive)
|
|
self.assertTrue(Path(str(archive) + ".sshsig").is_file())
|
|
raise RuntimeError("stop before account/unit mutation")
|
|
try:
|
|
with patch.object(lab, "state", return_value={"LoadState": "not-found"}), \
|
|
patch.object(lab, "run", side_effect=confined_run), \
|
|
patch.object(lab, "sign", side_effect=first_signature):
|
|
with self.assertRaisesRegex(RuntimeError, "stop before account/unit mutation"):
|
|
lab.prepare()
|
|
self.assertFalse(lab.owned_unit)
|
|
self.assertFalse(lab.owned_user)
|
|
finally:
|
|
lab.cleanup()
|
|
self.assertTrue(lab.receipt["cleanup"]["complete"])
|
|
|
|
def test_signer_creation_refuses_adoption_wrong_type_owner_or_failed_mode(self):
|
|
for mutation in ("existing", "symlink", "generated-symlink", "directory", "owner", "chmod", "mode-unchanged"):
|
|
with self.subTest(mutation=mutation), tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory); (root / "signer").mkdir(mode=0o700)
|
|
key = root / "signer/key"
|
|
marker = root / "keep"; marker.write_text("not a key; not ours\n")
|
|
lab = native.NativeLab(root, ROOT, root / "a", root / "b", "a" * 40)
|
|
if mutation == "existing": key.write_text("not a key; not ours\n")
|
|
if mutation == "symlink": key.symlink_to(marker)
|
|
def generate(*args):
|
|
if mutation == "generated-symlink": key.symlink_to(marker)
|
|
elif mutation == "directory": key.mkdir()
|
|
else: key.write_text("mode/type fixture; not a key\n")
|
|
with patch.object(lab, "signer_command", side_effect=generate) as command, \
|
|
patch.object(native.os, "geteuid", return_value=-1 if mutation == "owner" else os.getuid()), \
|
|
patch.object(Path, "chmod", side_effect=OSError("mode change unavailable") if mutation == "chmod" else None):
|
|
with self.assertRaises((ValueError, OSError)):
|
|
lab.create_signer()
|
|
if mutation in ("existing", "symlink"): command.assert_not_called()
|
|
else: command.assert_called_once()
|
|
self.assertEqual(marker.read_text(), "not a key; not ours\n")
|
|
|
|
def test_signer_timeout_does_not_attribute_stale_output_to_new_command(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
lab = native.NativeLab(root, ROOT, root / "a", root / "b", "a" * 40)
|
|
lab._last_output = b"UNPROTECTED PRIVATE KEY FILE from an earlier command"
|
|
def timeout(*args):
|
|
lab.receipt["command_failure"] = {"command": "ssh-keygen", "deadline_exceeded": True,
|
|
"observed_exit": -9}
|
|
raise RuntimeError("native command deadline exceeded")
|
|
with patch.object(lab, "run", side_effect=timeout), self.assertRaises(RuntimeError):
|
|
lab.signer_command("generate-ephemeral-key", "-q")
|
|
failure = lab.receipt["command_failure"]
|
|
self.assertEqual(failure["operation"], "generate-ephemeral-key")
|
|
self.assertEqual(failure["signer_key_metadata"], {"present": False})
|
|
self.assertNotIn("observed_diagnostics", failure)
|
|
|
|
def make_phase(self, directory, mutation=""):
|
|
root = Path(directory)
|
|
for name in ("install/bin", "data", "tmp"):
|
|
(root / name).mkdir(parents=True)
|
|
before, after = root / "a", root / "b"
|
|
before.write_bytes(b"baseline executable\n"); after.write_bytes(b"changed executable\n")
|
|
(root / "install/bin/pulse").write_bytes(after.read_bytes())
|
|
(root / "install/VERSION").write_text("v0.0.2\n")
|
|
(root / "data/sentinel").write_text("original\n")
|
|
lab = native.NativeLab(root, ROOT, before, after, "a" * 40)
|
|
lab.baseline_data = {"sentinel": native.digest(root / "data/sentinel")}
|
|
lab.baseline_mode = 0o644
|
|
lab.receipt["binaries"] = {"v0.0.1": native.digest(before), "v0.0.2": native.digest(after)}
|
|
states = [{"LoadState": "loaded", "ActiveState": "inactive", "UnitFileState": "disabled", "MainPID": "0"}] * 2
|
|
if mutation == "mask": states[1] = {**states[1], "LoadState": "masked", "UnitFileState": "masked"}
|
|
if mutation == "activity": states[1] = {**states[1], "ActiveState": "active", "MainPID": "123"}
|
|
lab.state = unittest.mock.Mock(side_effect=states)
|
|
def run(*args, **kwargs):
|
|
if mutation == "persistent": (root / "data/sentinel").write_text("changed\n")
|
|
if mutation == "binary": (root / "install/bin/pulse").write_text("different\n")
|
|
if mutation == "version": (root / "install/VERSION").write_text("v0.0.1\n")
|
|
if mutation == "backup": (root / "tmp/pulse-backup.extra").mkdir()
|
|
if mutation == "staging": (root / "install/bin/pulse.rollback.leak").write_text("leaked\n")
|
|
return subprocess.CompletedProcess(args, 0, b"fixture command, not native execution\n")
|
|
lab.run = run
|
|
return lab
|
|
|
|
def test_phase_cannot_accept_wrong_bytes_state_mask_persistence_or_backup(self):
|
|
for mutation in ("", "binary", "version", "mask", "activity", "persistent", "backup", "staging"):
|
|
with self.subTest(mutation=mutation), tempfile.TemporaryDirectory() as directory:
|
|
lab = self.make_phase(directory, mutation)
|
|
if mutation:
|
|
with self.assertRaises(RuntimeError): lab.phase("forward", "v0.0.2", active="inactive")
|
|
self.assertEqual(lab.receipt["phases"], [])
|
|
else:
|
|
lab.phase("forward", "v0.0.2", active="inactive")
|
|
self.assertTrue(lab.receipt["phases"][0]["passed"])
|
|
self.assertEqual(lab.operation, "forward")
|
|
|
|
def test_managed_process_identity_follows_served_readiness_and_rejects_residuals(self):
|
|
for mutation in ("", "process-bytes", "stopped-after-readiness", "inactive-pid"):
|
|
with self.subTest(mutation=mutation), tempfile.TemporaryDirectory() as directory:
|
|
lab = self.make_phase(directory)
|
|
state = dict(LoadState="loaded", ActiveState="active", UnitFileState="disabled", MainPID="1")
|
|
observed = {**state, "MainPID": "2"}
|
|
if mutation == "stopped-after-readiness":
|
|
observed.update(ActiveState="inactive", MainPID="0")
|
|
if mutation == "inactive-pid":
|
|
lab.state = unittest.mock.Mock(return_value={**state, "ActiveState": "inactive"})
|
|
with self.assertRaises(RuntimeError): lab.assert_runtime("v0.0.2", "inactive")
|
|
continue
|
|
lab.state = unittest.mock.Mock(side_effect=[state, observed, observed])
|
|
order = []
|
|
def response(url, **kwargs):
|
|
endpoint = url.rsplit("/", 1)[-1]
|
|
order.append(endpoint)
|
|
return io.BytesIO(json.dumps({"status": "healthy"} if endpoint == "health"
|
|
else {"version": "v0.0.2"}).encode())
|
|
actual_digest = native.digest
|
|
def process_digest(path):
|
|
if str(path).startswith("/proc/"):
|
|
order.append("process")
|
|
self.assertEqual(str(path), "/proc/2/exe")
|
|
return "wrong" if mutation == "process-bytes" else lab.receipt["binaries"]["v0.0.2"]
|
|
return actual_digest(path)
|
|
with patch.object(native.urllib.request, "urlopen", side_effect=response), patch.object(native, "digest", side_effect=process_digest):
|
|
if mutation:
|
|
with self.assertRaises(RuntimeError): lab.assert_runtime("v0.0.2", "active")
|
|
else:
|
|
self.assertEqual(lab.assert_runtime("v0.0.2", "active"), observed)
|
|
self.assertEqual(order[:2], ["health", "version"])
|
|
self.assertEqual(order[2:], [] if mutation == "stopped-after-readiness" else ["process"])
|
|
|
|
def test_runtime_readback_rejects_post_comparison_drift(self):
|
|
# Portable manager/API observations, never native acceptance. Keep the
|
|
# final oracle independent of the updater's own executable verifier.
|
|
mutations = ("", "pid", "activity", "load", "enablement", "installed",
|
|
"equally-wrong-bytes", "missing-process", "failed-readback")
|
|
for mutation in mutations:
|
|
with self.subTest(mutation=mutation), tempfile.TemporaryDirectory() as directory:
|
|
lab = self.make_phase(directory)
|
|
state = dict(LoadState="loaded", ActiveState="active", UnitFileState="disabled", MainPID="2")
|
|
final = state.copy()
|
|
changes = {
|
|
"pid": {"MainPID": "3"},
|
|
"activity": {"ActiveState": "inactive", "MainPID": "0"},
|
|
"load": {"LoadState": "not-found"},
|
|
"enablement": {"UnitFileState": "masked"},
|
|
}
|
|
final.update(changes.get(mutation, {}))
|
|
end = OSError("confined manager read failed") if mutation == "failed-readback" else final
|
|
lab.state = unittest.mock.Mock(side_effect=[state, state, end])
|
|
binary = lab.root / "install/bin/pulse"
|
|
expected = lab.receipt["binaries"]["v0.0.2"]
|
|
original_digest = native.digest
|
|
def process_digest(path):
|
|
if str(path) == "/proc/2/exe":
|
|
if mutation == "missing-process":
|
|
raise FileNotFoundError("confined missing process")
|
|
if mutation in ("installed", "equally-wrong-bytes"):
|
|
binary.write_bytes(b"changed during executable readback\n")
|
|
return original_digest(binary) if mutation == "equally-wrong-bytes" else expected
|
|
return original_digest(path)
|
|
def response(url, **kwargs):
|
|
return io.BytesIO(json.dumps({"status": "healthy"} if url.endswith("/health")
|
|
else {"version": "v0.0.2"}).encode())
|
|
with patch.object(native, "digest", side_effect=process_digest), \
|
|
patch.object(native.urllib.request, "urlopen", side_effect=response), \
|
|
patch.object(lab, "run", side_effect=AssertionError("readback cannot control a service")):
|
|
if mutation:
|
|
with self.assertRaises((RuntimeError, OSError)):
|
|
lab.assert_runtime("v0.0.2", "active")
|
|
self.assertEqual(lab.receipt["phases"], [])
|
|
else:
|
|
self.assertEqual(lab.assert_runtime("v0.0.2", "active"), state)
|
|
self.assertEqual(lab.state.call_count, 3)
|
|
|
|
# Refuse malformed or non-active manager identity before constructing
|
|
# a process path, even when a digest double would otherwise match.
|
|
for pid, load in (("0", "loaded"), ("-2", "loaded"), ("02", "loaded"),
|
|
("2/../1", "loaded"), ("unknown", "loaded"), ("2", "masked")):
|
|
with self.subTest(pid=pid, load=load), tempfile.TemporaryDirectory() as directory:
|
|
lab = self.make_phase(directory)
|
|
state = dict(LoadState=load, ActiveState="active", UnitFileState="disabled", MainPID=pid)
|
|
lab.state = unittest.mock.Mock(return_value=state)
|
|
original_digest = native.digest
|
|
def no_process_digest(path):
|
|
if str(path).startswith("/proc/"):
|
|
raise AssertionError("invalid manager identity reached process-byte read")
|
|
return original_digest(path)
|
|
def response(url, **kwargs):
|
|
return io.BytesIO(json.dumps({"status": "healthy"} if url.endswith("/health")
|
|
else {"version": "v0.0.2"}).encode())
|
|
with patch.object(native, "digest", side_effect=no_process_digest), \
|
|
patch.object(native.urllib.request, "urlopen", side_effect=response):
|
|
with self.assertRaises(RuntimeError): lab.assert_runtime("v0.0.2", "active")
|
|
|
|
def test_cleanup_never_deletes_preexisting_unowned_scratch(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory); marker = root / "prior"; marker.write_text("not ours\n")
|
|
lab = native.NativeLab(root, ROOT, root / "a", root / "b", "a" * 40)
|
|
with patch.object(lab, "run", side_effect=AssertionError("unowned manager operation")):
|
|
lab.cleanup()
|
|
self.assertTrue(marker.exists())
|
|
self.assertFalse(lab.receipt["cleanup"]["owned_fixture_removed"])
|
|
|
|
def test_prepare_allocates_scratch_before_child_home_initialisation(self):
|
|
# Model a child runtime that eagerly initialises HOME. This catches the
|
|
# macOS control's FileExistsError without pretending to run systemd.
|
|
for load in ("not-found", "loaded"):
|
|
with self.subTest(load=load), tempfile.TemporaryDirectory() as directory:
|
|
base = Path(directory)
|
|
before, after = base / "a", base / "b"
|
|
before.write_bytes(b"before\n"); after.write_bytes(b"after\n")
|
|
lab = native.NativeLab(base / "lab", ROOT, before, after, "a" * 40)
|
|
lab.unit = base / "absent.service"
|
|
def child_observation():
|
|
lab.root.mkdir(exist_ok=True)
|
|
(lab.root / "child-home-marker").write_text("runtime initialised HOME\n")
|
|
return {"LoadState": load}
|
|
def bounded_command(*args, **kwargs):
|
|
if args[0] in (str(before), str(after)) and args[1:] == ("--version",):
|
|
tag = native.TAGS[0 if args[0] == str(before) else 1]
|
|
return subprocess.CompletedProcess(args, 0, ("Pulse " + tag + "\n").encode())
|
|
raise RuntimeError("stop before signer or service mutation")
|
|
with patch.object(lab, "state", side_effect=child_observation) as observation, \
|
|
patch.object(lab, "run", side_effect=bounded_command):
|
|
expected = ValueError if load == "loaded" else RuntimeError
|
|
with self.assertRaises(expected):
|
|
lab.prepare()
|
|
observation.assert_called_once()
|
|
self.assertTrue(lab.owned_root)
|
|
self.assertTrue((lab.root / "child-home-marker").exists())
|
|
self.assertFalse(lab.owned_unit)
|
|
self.assertFalse(lab.owned_user)
|
|
lab.cleanup()
|
|
self.assertFalse(lab.root.exists())
|
|
self.assertTrue(lab.receipt["cleanup"]["complete"])
|
|
|
|
def test_prepare_refuses_existing_directory_or_symlink_before_child(self):
|
|
for symlink in (False, True):
|
|
with self.subTest(symlink=symlink), tempfile.TemporaryDirectory() as directory:
|
|
base = Path(directory); prior = base / "prior"; prior.mkdir()
|
|
marker = prior / "keep"; marker.write_text("not ours\n")
|
|
root = base / "lab" if symlink else prior
|
|
if symlink:
|
|
root.symlink_to(prior, target_is_directory=True)
|
|
lab = native.NativeLab(root, ROOT, base / "a", base / "b", "a" * 40)
|
|
lab.unit = base / "absent.service"
|
|
with patch.object(lab, "state", side_effect=AssertionError("unowned child")) as observation:
|
|
with self.assertRaises(FileExistsError):
|
|
lab.prepare()
|
|
observation.assert_not_called()
|
|
self.assertFalse(lab.owned_root)
|
|
lab.cleanup()
|
|
self.assertEqual(marker.read_text(), "not ours\n")
|
|
self.assertTrue(root.exists())
|
|
self.assertFalse(lab.receipt["cleanup"]["owned_fixture_removed"])
|
|
|
|
def run_recipe(self, failure="", mismatch=False):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory); tools = root / "tools"; tools.mkdir()
|
|
for name in ("frontend-modern/dist", "internal/api/frontend-modern/dist"):
|
|
(root / name).mkdir(parents=True); (root / name / "index.html").write_text("real-embed fixture\n")
|
|
trace = root / "trace"
|
|
fake = '''#!/usr/bin/python3
|
|
import json, os, pathlib, sys
|
|
args = sys.argv[1:]; name = pathlib.Path(sys.argv[0]).name
|
|
with pathlib.Path(os.environ["TRACE"]).open("a") as out: out.write(json.dumps([name, *args])+"\\n")
|
|
if name == "git":
|
|
if args == ["rev-parse", "HEAD"]: print("b"*40 if os.environ["MISMATCH"] == "true" else "a"*40)
|
|
elif args == ["diff", "--quiet", "HEAD"]:
|
|
if os.environ["FAIL_BUILD"] == "source-modified": sys.exit(49)
|
|
else: sys.exit(97)
|
|
if name == "go":
|
|
if args[args.index("-o")+1].endswith(os.environ["FAIL_BUILD"]) and os.environ["FAIL_BUILD"]: sys.exit(43)
|
|
pathlib.Path(args[args.index("-o")+1]).write_text("compiler double, not a product build\\n")
|
|
if name == "sudo":
|
|
if os.environ["FAIL_BUILD"] == "native": sys.exit(45)
|
|
pathlib.Path(os.environ["RUNNER_TEMP"]+"/pulse-native-updater.json").write_text("{}\\n")
|
|
'''
|
|
for name in ("git", "go", "sudo"):
|
|
path = tools / name; path.write_text(fake); path.chmod(0o755)
|
|
env = {**os.environ, "PATH": str(tools) + ":" + os.environ["PATH"], "TRACE": str(trace),
|
|
"RUNNER_TEMP": str(root), "EXPECTED_SOURCE_SHA": "a" * 40,
|
|
"FAIL_BUILD": failure, "MISMATCH": "true" if mismatch else "false"}
|
|
result = subprocess.run(["/bin/bash", "-c", recipe(WORKFLOW.read_text())], env=env, cwd=root, capture_output=True)
|
|
return result, [json.loads(line) for line in trace.read_text().splitlines()], (root / "pulse-native-updater.json").exists()
|
|
|
|
def test_exact_workflow_recipe_is_scrubbed_and_propagates_all_failures(self):
|
|
result, calls, receipt = self.run_recipe()
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertEqual([call[0] for call in calls], ["git", "git", "go", "go", "sudo"])
|
|
self.assertTrue(receipt)
|
|
self.assertEqual(calls[-1][1:3], ["env", "-i"])
|
|
for failure, code, count in (("source-modified", 49, 2), ("before", 43, 3), ("after", 43, 4), ("native", 45, 5)):
|
|
result, calls, receipt = self.run_recipe(failure)
|
|
self.assertEqual(result.returncode, code)
|
|
self.assertEqual(len(calls), count)
|
|
self.assertFalse(receipt)
|
|
result, calls, receipt = self.run_recipe(mismatch=True)
|
|
self.assertEqual(result.returncode, 1)
|
|
self.assertEqual([call[0] for call in calls], ["git"])
|
|
self.assertFalse(receipt)
|
|
|
|
def test_connected_linux_control_is_required_without_macho_or_validator_bypass(self):
|
|
connected = getattr(NativeContracts, LINUX_CONTROL)
|
|
self.assertEqual(getattr(connected, "__unittest_skip__", False), not sys.platform.startswith("linux"))
|
|
# Import just the definitions as each platform would. No fixture is run
|
|
# under a false platform identity, and no production function is changed.
|
|
for platform in ("linux", "darwin"):
|
|
spec = importlib.util.spec_from_file_location("platform_contracts", Path(__file__))
|
|
module = importlib.util.module_from_spec(spec)
|
|
with patch.object(sys, "platform", platform): spec.loader.exec_module(module)
|
|
control = getattr(module.NativeContracts, LINUX_CONTROL)
|
|
self.assertEqual(getattr(control, "__unittest_skip__", False), platform != "linux")
|
|
for name in unittest.defaultTestLoader.getTestCaseNames(module.NativeContracts):
|
|
if name != LINUX_CONTROL:
|
|
self.assertFalse(getattr(getattr(module.NativeContracts, name), "__unittest_skip__", False), name)
|
|
|
|
@unittest.skipUnless(sys.platform.startswith("linux"), LINUX_ONLY_REASON)
|
|
def test_connected_signed_archive_and_recovery_controls_without_host_manager(self):
|
|
# Same orchestration and production shell functions, real signatures and
|
|
# ELF staging. This test replaces the manager, its process-byte readback,
|
|
# account/chown and unit path. /proc/123/exe maps only to the executable
|
|
# snapshot made by that simulated manager's start. Real owned /proc
|
|
# checks run separately; this is not native/real-Pulse acceptance.
|
|
manager = r'''#!/usr/bin/python3
|
|
import os, pathlib, shutil, sys
|
|
root = pathlib.Path(os.environ["PULSE_NATIVE_ROOT"])
|
|
# Tool startup must have an existing owned home, not create the lab before
|
|
# NativeLab has admitted and claimed its scratch root (as on macOS).
|
|
if not pathlib.Path(os.environ["HOME"]).is_dir(): sys.exit(96)
|
|
unit = root / "pulse-source-updater.service"
|
|
args = sys.argv[1:]
|
|
if any(".service" in a and a != "pulse-source-updater.service" for a in args): sys.exit(97)
|
|
load = "masked" if unit.is_symlink() else ("loaded" if unit.exists() else "not-found")
|
|
statefile = root / "manager-state"
|
|
active = statefile.read_text().strip() if statefile.exists() and load != "not-found" else "inactive"
|
|
values = dict(LoadState=load, ActiveState=active, UnitFileState="masked" if load == "masked" else "disabled", MainPID="123" if active == "active" else "0")
|
|
if args[0] == "show":
|
|
properties = [p for a in args if a.startswith("--property=") for p in a.split("=", 1)[1].split(",")]
|
|
if not properties or len(properties) != len(set(properties)) or any(p not in values for p in properties): sys.exit(97)
|
|
for p in properties: print(values[p] if "--value" in args else p+"="+values[p])
|
|
elif args[0] in ("start", "stop"):
|
|
if args[0] == "start" and load != "loaded": sys.exit(1)
|
|
statefile.write_text("active" if args[0] == "start" else "inactive")
|
|
process_bytes = root / "mock-managed-executable"
|
|
if args[0] == "start": shutil.copyfile(root / "install/bin/pulse", process_bytes)
|
|
else: process_bytes.unlink(missing_ok=True)
|
|
if args[0] == "start" and not (root / "data/.encryption.key").exists():
|
|
(root / "data/.encryption.key").write_text("synthetic persistence marker, not a key\n")
|
|
elif args[0] == "mask": unit.symlink_to("/dev/null")
|
|
elif args[0] == "unmask": unit.unlink()
|
|
elif args[0] not in ("daemon-reload", "reset-failed"): sys.exit(97)
|
|
'''
|
|
class FixtureLab(native.NativeLab):
|
|
def run(self, *args, **kwargs):
|
|
if args[0] == "/bin/bash":
|
|
args = (str(tools / "bash"), *args[1:])
|
|
return super().run(*args, **kwargs)
|
|
|
|
def assert_runtime(self, tag, active):
|
|
if native.digest(self.root / "install/bin/pulse") != self.receipt["binaries"][tag]:
|
|
raise RuntimeError("fixture executable mismatch")
|
|
if (self.root / "install/VERSION").read_text() != tag + "\n":
|
|
raise RuntimeError("fixture version mismatch")
|
|
state = self.state()
|
|
if state["ActiveState"] != active: raise RuntimeError("fixture activity mismatch")
|
|
return state
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
base = Path(directory); tools = base / "tools"; tools.mkdir()
|
|
selected = selected_fixture_commands()
|
|
tool_trace = base / "selected-tools.jsonl"
|
|
for name, executable in selected.items():
|
|
# Record only tool identity, never shell arguments or key/log
|
|
# bytes. Execute the actual supplied GNU utility; only the
|
|
# manager double's fixed process path is redirected below.
|
|
tool = tools / name
|
|
tool.write_text(f'''#!{sys.executable}
|
|
import json, os, pathlib, sys
|
|
with pathlib.Path({str(tool_trace)!r}).open('a') as receipt:
|
|
receipt.write(json.dumps({name!r}) + '\\n')
|
|
arguments = sys.argv[1:]
|
|
if {name!r} == 'cmp' and '/proc/123/exe' in arguments:
|
|
arguments = [str(pathlib.Path(os.environ['PULSE_NATIVE_ROOT']) / 'mock-managed-executable')
|
|
if arg == '/proc/123/exe' else arg for arg in arguments]
|
|
os.execv({executable!r}, [{executable!r}, *arguments])
|
|
''')
|
|
tool.chmod(0o700)
|
|
path = tools / "systemctl"; path.write_text(manager); path.chmod(0o755)
|
|
account = r'''#!/usr/bin/python3
|
|
import os, pathlib, sys
|
|
marker = pathlib.Path(os.environ["PULSE_NATIVE_ROOT"]) / "fixture-user"
|
|
if pathlib.Path(sys.argv[0]).name == "id":
|
|
if sys.argv[1:] != ["-u", "pulse"]: sys.exit(97)
|
|
if not marker.exists(): sys.exit(1)
|
|
print("1234")
|
|
elif pathlib.Path(sys.argv[0]).name == "useradd":
|
|
if sys.argv[1:] != ["--system", "--home-dir", str(marker.parent / "install"), "--shell", "/bin/false", "pulse"]: sys.exit(97)
|
|
marker.write_text("owned confined fixture account\n")
|
|
elif pathlib.Path(sys.argv[0]).name == "userdel":
|
|
if sys.argv[1:] != ["pulse"]: sys.exit(97)
|
|
marker.unlink()
|
|
elif pathlib.Path(sys.argv[0]).name == "chown":
|
|
args = [arg for arg in sys.argv[1:] if arg != "-R"]
|
|
if len(args) < 2 or args[0] != "pulse:pulse": sys.exit(97)
|
|
if any(not pathlib.Path(p).is_relative_to(marker.parent) for p in args[1:]): sys.exit(97)
|
|
else: sys.exit(97)
|
|
'''
|
|
for name in ("id", "useradd", "userdel", "chown"):
|
|
tool = tools / name; tool.write_text(account); tool.chmod(0o755)
|
|
program = base / "fixture.go"
|
|
program.write_text('package main\nimport "fmt"\nvar Version string\nfunc main(){fmt.Println("Pulse " + Version)}\n')
|
|
binaries = []
|
|
for tag in native.TAGS:
|
|
binary = base / tag
|
|
subprocess.run(["go", "build", "-o", str(binary), "-ldflags=-X main.Version=" + tag, str(program)],
|
|
check=True, capture_output=True, timeout=60)
|
|
binaries.append(binary)
|
|
lab = FixtureLab(base / "lab", ROOT, *binaries, "a" * 40)
|
|
lab.unit = lab.root / "pulse-source-updater.service"
|
|
lab.env["HOME"] = str(base)
|
|
lab.env["PATH"] = str(lab.root / "tools") + ":" + str(tools) + ":/usr/sbin:/usr/bin:/sbin:/bin"
|
|
original_write = Path.write_text
|
|
def relocate(path, data, *args, **kwargs):
|
|
if path.name == "installer-definitions":
|
|
before = "cat > /etc/systemd/system/$SERVICE_NAME.service << EOF"
|
|
self.assertEqual(data.count(before), 1)
|
|
append = "cat >> /etc/systemd/system/$SERVICE_NAME.service << EOF"
|
|
self.assertEqual(data.count(append), 2)
|
|
data = data.replace(before, 'cat > "$PULSE_NATIVE_ROOT/pulse-source-updater.service" << EOF')
|
|
data = data.replace(append, 'cat >> "$PULSE_NATIVE_ROOT/pulse-source-updater.service" << EOF')
|
|
return original_write(path, data, *args, **kwargs)
|
|
try:
|
|
with patch.object(Path, "write_text", new=relocate), patch.object(native.time, "sleep"):
|
|
try:
|
|
lab.prepare()
|
|
self.assertEqual((lab.root / "data/.env").read_text(),
|
|
"PULSE_TELEMETRY=false\nBIND_ADDRESS=127.0.0.1\n")
|
|
lab.exercise()
|
|
except RuntimeError as error:
|
|
self.fail(str(error) + "\nConfined fixture output:\n" + lab._last_output.decode())
|
|
self.assertEqual([p["name"] for p in lab.receipt["phases"]],
|
|
["forward", "manual-reverse", "failed-active", "failed-inactive",
|
|
"failed-masked", "bad-signature", "incomplete-restore"])
|
|
self.assertTrue(lab.receipt["phases"][-1]["backup_retained"])
|
|
used = [json.loads(line) for line in tool_trace.read_text().splitlines()]
|
|
self.assertEqual(set(used), set(FIXTURE_COMMANDS),
|
|
"nested shell lost a containing-job fixture tool")
|
|
finally:
|
|
lab.cleanup()
|
|
self.assertTrue(lab.receipt["cleanup"]["complete"])
|
|
self.assertFalse(lab.root.exists())
|
|
|
|
def test_workflow_cannot_hide_or_rebind_native_step(self):
|
|
source = WORKFLOW.read_text()
|
|
for changed in (source.replace("- name: " + STEP + "\n", "- name: " + STEP + "\n continue-on-error: true\n"),
|
|
source.replace("sudo env -i", "sudo env"),
|
|
source.replace("main.Version=v0.0.2", "main.Version=v0.0.1"),
|
|
source.replace("EXPECTED_SOURCE_SHA: ${{ github.sha }}", "EXPECTED_SOURCE_SHA: main"),
|
|
source.replace("runs-on: ubuntu-24.04", "runs-on: self-hosted")):
|
|
with self.assertRaises(AssertionError): recipe(changed)
|
|
|
|
def run_receipt(self, primary, cleanup_fails):
|
|
# Execute the actual receipt producer, with real unprivileged failing
|
|
# children. Substitute only native admission and service/account work.
|
|
# These controls cannot provide a native transaction or cleanup proof.
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
receipt = root / "receipt.json"
|
|
lab = native.NativeLab(root, ROOT, root / "a", root / "b", "a" * 40)
|
|
owned_cleanup = lab.cleanup
|
|
def prepare():
|
|
lab.operation = "baseline-install"
|
|
if primary == "command":
|
|
lab.run("/bin/bash", "-c", "printf 'private-transaction-fixture'; exit 23")
|
|
if primary == "timeout":
|
|
lab.run(sys.executable, "-c", "import time; time.sleep(60)", timeout=0.05)
|
|
if primary == "assertion":
|
|
raise RuntimeError("confined persistence assertion")
|
|
def exercise():
|
|
lab.receipt["phases"] = [{"name": "confined-control"}] * 7
|
|
def cleanup():
|
|
lab.operation = "cleanup"
|
|
if cleanup_fails:
|
|
lab.run(sys.executable, "-c", "print('private-cleanup-fixture'); raise SystemExit(31)")
|
|
owned_cleanup() # Owns nothing; touches no manager.
|
|
argv = ["native-receipt-control", "--before", "unused-before", "--after", "unused-after",
|
|
"--receipt", str(receipt)]
|
|
with patch.dict(os.environ, {"EXPECTED_SOURCE_SHA": "a" * 40, "RUNNER_TEMP": str(root)}), \
|
|
patch.object(sys, "argv", argv), patch.object(native, "hosted_admission"), \
|
|
patch.object(native, "NativeLab", return_value=lab), \
|
|
patch.object(lab, "prepare", side_effect=prepare), \
|
|
patch.object(lab, "exercise", side_effect=exercise), \
|
|
patch.object(lab, "cleanup", side_effect=cleanup), \
|
|
contextlib.redirect_stdout(io.StringIO()):
|
|
if primary or cleanup_fails:
|
|
with self.assertRaises(SystemExit): native.main()
|
|
else:
|
|
native.main()
|
|
result = json.loads(receipt.read_text())
|
|
for private in ("private-transaction-fixture", "private-cleanup-fixture", str(root)):
|
|
self.assertNotIn(private, json.dumps(result))
|
|
return result
|
|
|
|
def test_primary_command_survives_successful_or_failed_cleanup(self):
|
|
for cleanup_fails in (False, True):
|
|
with self.subTest(cleanup_fails=cleanup_fails):
|
|
result = self.run_receipt("command", cleanup_fails)
|
|
self.assertEqual(result["command_failure"]["observed_exit"], 23)
|
|
self.assertEqual(result["command_failure"]["command"], "bash")
|
|
self.assertEqual(result["command_failure"]["operation"], "baseline-install")
|
|
self.assertEqual(result["failure"]["operation"], "baseline-install")
|
|
self.assertEqual(result["phases"], [])
|
|
self.assertEqual(result["cleanup"]["complete"], not cleanup_fails)
|
|
if cleanup_fails:
|
|
self.assertEqual(result["cleanup"]["command_failure"]["observed_exit"], 31)
|
|
self.assertEqual(result["cleanup"]["command_failure"]["operation"], "cleanup")
|
|
else:
|
|
self.assertNotIn("command_failure", result["cleanup"])
|
|
|
|
def test_primary_timeout_is_not_replaced_by_cleanup_exit(self):
|
|
result = self.run_receipt("timeout", True)
|
|
self.assertTrue(result["command_failure"]["deadline_exceeded"])
|
|
self.assertLess(result["command_failure"]["observed_exit"], 0)
|
|
self.assertEqual(result["command_failure"]["operation"], "baseline-install")
|
|
self.assertNotIn("output_sha256", result["command_failure"])
|
|
self.assertEqual(result["cleanup"]["command_failure"]["observed_exit"], 31)
|
|
self.assertEqual(result["phases"], [])
|
|
|
|
def test_assertion_and_cleanup_only_errors_do_not_invent_primary_command(self):
|
|
for primary in ("assertion", ""):
|
|
with self.subTest(primary=primary):
|
|
result = self.run_receipt(primary, True)
|
|
self.assertNotIn("command_failure", result)
|
|
self.assertEqual(result["cleanup"]["command_failure"]["observed_exit"], 31)
|
|
self.assertFalse(result["cleanup"]["complete"])
|
|
if primary:
|
|
self.assertEqual(result["failure"]["operation"], "baseline-install")
|
|
self.assertEqual(result["phases"], [])
|
|
else:
|
|
self.assertNotIn("failure", result)
|
|
|
|
def test_successful_receipt_retains_no_failure_record(self):
|
|
result = self.run_receipt("", False)
|
|
self.assertNotIn("failure", result)
|
|
self.assertNotIn("command_failure", result)
|
|
self.assertNotIn("command_failure", result["cleanup"])
|
|
self.assertTrue(result["cleanup"]["complete"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|