350 lines
15 KiB
Python
350 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
import hashlib
|
|
import importlib.machinery
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import tarfile
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
|
|
SCRIPT_DIR = Path(__file__).resolve().parent
|
|
BUILDER = SCRIPT_DIR / "build-device-edge-vps-artifact.mjs"
|
|
RUNNER_PATH = SCRIPT_DIR / "nodedc-b2-vps-deploy"
|
|
DEFAULT_RUNTIME_CACHE = Path(
|
|
os.environ.get("NODEDC_DEVICE_EDGE_VPS_RUNTIME_DIR", "/tmp")
|
|
)
|
|
|
|
|
|
def load_runner():
|
|
loader = importlib.machinery.SourceFileLoader(
|
|
"nodedc_b2_vps_runner_under_test",
|
|
str(RUNNER_PATH),
|
|
)
|
|
spec = importlib.util.spec_from_loader(loader.name, loader)
|
|
module = importlib.util.module_from_spec(spec)
|
|
loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
RUNNER = load_runner()
|
|
|
|
|
|
class DeviceEdgeVpsArtifactTest(unittest.TestCase):
|
|
def build(self, artifact_dir, phase, patch_id, runtime_cache=None):
|
|
environment = os.environ.copy()
|
|
environment["NODEDC_DEPLOY_ARTIFACT_DIR"] = str(artifact_dir)
|
|
environment["NODEDC_DEVICE_EDGE_VPS_RUNTIME_DIR"] = str(
|
|
runtime_cache or DEFAULT_RUNTIME_CACHE
|
|
)
|
|
if phase in {"backhaul", "relay"}:
|
|
environment["NODEDC_ALLOW_SUPERSEDED_TRANSPORT"] = "test-only"
|
|
return subprocess.run(
|
|
["node", str(BUILDER), phase, patch_id],
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
env=environment,
|
|
)
|
|
|
|
def test_superseded_transport_builds_fail_closed_by_default(self):
|
|
environment = os.environ.copy()
|
|
environment.pop("NODEDC_ALLOW_SUPERSEDED_TRANSPORT", None)
|
|
with tempfile.TemporaryDirectory(prefix="nodedc-vps-frozen-") as directory:
|
|
environment["NODEDC_DEPLOY_ARTIFACT_DIR"] = directory
|
|
for phase in ("backhaul", "relay"):
|
|
with self.subTest(phase=phase):
|
|
result = subprocess.run(
|
|
[
|
|
"node",
|
|
str(BUILDER),
|
|
phase,
|
|
f"device-edge-vps-{phase}-frozen-001",
|
|
],
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
env=environment,
|
|
)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn(
|
|
"vps_initiated_transport_frozen:ADR-0001",
|
|
result.stderr,
|
|
)
|
|
|
|
def test_runner_rejects_superseded_transport_before_host_preflight(self):
|
|
for phase in ("backhaul", "relay"):
|
|
with self.subTest(phase=phase), self.assertRaises(RUNNER.DeployError):
|
|
RUNNER.preflight({"phase": phase})
|
|
|
|
def require_runtime_cache(self):
|
|
for name, digest in (
|
|
(RUNNER.NODE_ARCHIVE, RUNNER.NODE_ARCHIVE_SHA256),
|
|
(RUNNER.TAILSCALE_ARCHIVE, RUNNER.TAILSCALE_ARCHIVE_SHA256),
|
|
):
|
|
path = DEFAULT_RUNTIME_CACHE / name
|
|
self.assertTrue(path.is_file(), f"missing runtime fixture: {path}")
|
|
self.assertEqual(hashlib.sha256(path.read_bytes()).hexdigest(), digest)
|
|
|
|
def test_builders_are_deterministic_narrow_and_secret_free(self):
|
|
self.require_runtime_cache()
|
|
for phase in ("foundation", "backhaul", "relay"):
|
|
with self.subTest(phase=phase), tempfile.TemporaryDirectory(
|
|
prefix=f"nodedc-vps-{phase}-"
|
|
) as directory:
|
|
root = Path(directory)
|
|
patch_id = f"device-edge-vps-{phase}-unit-001"
|
|
first = self.build(root, phase, patch_id)
|
|
self.assertEqual(first.returncode, 0, first.stderr)
|
|
first_result = json.loads(first.stdout)
|
|
first_bytes = Path(first_result["artifact"]).read_bytes()
|
|
second = self.build(root, phase, patch_id)
|
|
self.assertEqual(second.returncode, 0, second.stderr)
|
|
second_result = json.loads(second.stdout)
|
|
second_bytes = Path(second_result["artifact"]).read_bytes()
|
|
|
|
self.assertEqual(first_bytes, second_bytes)
|
|
self.assertEqual(first_result["sha256"], second_result["sha256"])
|
|
self.assertEqual(
|
|
first_result["sha256"],
|
|
hashlib.sha256(first_bytes).hexdigest(),
|
|
)
|
|
self.assertEqual(first_result["entries"], list(RUNNER.PHASE_ENTRIES[phase]))
|
|
|
|
with tarfile.open(first_result["artifact"], "r:gz") as archive:
|
|
members = archive.getmembers()
|
|
names = {member.name for member in members}
|
|
payload = b"\n".join(
|
|
archive.extractfile(member).read()
|
|
for member in members
|
|
if member.isfile() and member.size < 2 * 1024 * 1024
|
|
)
|
|
self.assertIn("manifest.env", names)
|
|
self.assertIn("files.txt", names)
|
|
self.assertFalse(any(
|
|
"/secrets/" in name
|
|
or "/keys/" in name
|
|
or "/trust/" in name
|
|
or "/runtime/" in name
|
|
or "/node_modules/" in name
|
|
or Path(name).name.startswith(".env")
|
|
for name in names
|
|
))
|
|
self.assertNotIn(b"PRIVATE KEY", payload)
|
|
self.assertNotIn(b"TS_AUTHKEY", payload)
|
|
|
|
def test_foundation_builder_rejects_modified_runtime_archive(self):
|
|
self.require_runtime_cache()
|
|
with tempfile.TemporaryDirectory(prefix="nodedc-vps-corrupt-") as directory:
|
|
cache = Path(directory) / "cache"
|
|
artifacts = Path(directory) / "artifacts"
|
|
cache.mkdir()
|
|
for name in (RUNNER.NODE_ARCHIVE, RUNNER.TAILSCALE_ARCHIVE):
|
|
(cache / name).write_bytes((DEFAULT_RUNTIME_CACHE / name).read_bytes())
|
|
with (cache / RUNNER.NODE_ARCHIVE).open("ab") as handle:
|
|
handle.write(b"corrupt")
|
|
result = self.build(
|
|
artifacts,
|
|
"foundation",
|
|
"device-edge-vps-foundation-corrupt-001",
|
|
runtime_cache=cache,
|
|
)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("runtime_digest_mismatch", result.stderr)
|
|
|
|
def test_runner_loads_each_exact_phase(self):
|
|
self.require_runtime_cache()
|
|
with tempfile.TemporaryDirectory(prefix="nodedc-vps-load-") as directory:
|
|
inbox = Path(directory) / "inbox"
|
|
inbox.mkdir()
|
|
old_inbox = RUNNER.INBOX_ROOT
|
|
RUNNER.INBOX_ROOT = inbox
|
|
try:
|
|
for phase in ("foundation", "backhaul", "relay"):
|
|
result = self.build(
|
|
inbox,
|
|
phase,
|
|
f"device-edge-vps-{phase}-load-001",
|
|
)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
artifact = Path(json.loads(result.stdout)["artifact"])
|
|
extraction = Path(directory) / f"extract-{phase}"
|
|
extraction.mkdir()
|
|
loaded = RUNNER.load_artifact(artifact, extraction)
|
|
self.assertEqual(loaded["phase"], phase)
|
|
self.assertEqual(loaded["entries"], RUNNER.PHASE_ENTRIES[phase])
|
|
self.assertEqual(
|
|
loaded["sha256"],
|
|
hashlib.sha256(artifact.read_bytes()).hexdigest(),
|
|
)
|
|
finally:
|
|
RUNNER.INBOX_ROOT = old_inbox
|
|
|
|
def test_plan_is_exact_and_never_claims_dns_or_b2_mutation(self):
|
|
self.require_runtime_cache()
|
|
with tempfile.TemporaryDirectory(prefix="nodedc-vps-plan-") as directory:
|
|
inbox = Path(directory) / "inbox"
|
|
inbox.mkdir()
|
|
result = self.build(
|
|
inbox,
|
|
"foundation",
|
|
"device-edge-vps-foundation-plan-001",
|
|
)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
artifact = Path(json.loads(result.stdout)["artifact"])
|
|
old_inbox = RUNNER.INBOX_ROOT
|
|
RUNNER.INBOX_ROOT = inbox
|
|
try:
|
|
with patch.object(RUNNER, "assert_root"), patch.object(
|
|
RUNNER,
|
|
"preflight",
|
|
return_value={"predecessor": "unit-predecessor"},
|
|
), patch("builtins.print") as output:
|
|
RUNNER.plan_artifact(str(artifact))
|
|
finally:
|
|
RUNNER.INBOX_ROOT = old_inbox
|
|
rendered = "\n".join(
|
|
" ".join(str(arg) for arg in call.args)
|
|
for call in output.call_args_list
|
|
)
|
|
self.assertIn("phase=foundation", rendered)
|
|
self.assertIn("public_b2_ingress=disabled", rendered)
|
|
self.assertIn("dns=unchanged", rendered)
|
|
self.assertIn("b2_routes=unchanged", rendered)
|
|
self.assertIn("command_transport=disabled", rendered)
|
|
|
|
def test_units_and_firewalls_keep_the_required_boundaries(self):
|
|
source_root = SCRIPT_DIR.parent.parent / "device-plane"
|
|
foundation = (source_root / "vps/config/nftables-foundation.conf").read_text()
|
|
relay = (source_root / "vps/config/nftables-relay.conf").read_text()
|
|
sshd = (source_root / "vps/config/00-nodedc-b2-vps.conf").read_text()
|
|
backhaul = (source_root / "vps/config/backhaul_ssh_config").read_text()
|
|
tailscale_unit = (
|
|
source_root / "vps/systemd/nodedc-b2-tailscaled.service"
|
|
).read_text()
|
|
relay_unit = (source_root / "vps/systemd/nodedc-b2-relay.service").read_text()
|
|
backhaul_unit = (
|
|
source_root / "vps/systemd/nodedc-b2-backhaul.service"
|
|
).read_text()
|
|
|
|
self.assertIn("policy drop", foundation)
|
|
self.assertIn("tcp dport 22", foundation)
|
|
self.assertNotIn("tcp dport 9921", foundation)
|
|
self.assertIn("tcp dport 9921", relay)
|
|
self.assertIn("PasswordAuthentication no", sshd)
|
|
self.assertIn("AllowTcpForwarding no", sshd)
|
|
self.assertIn("StrictHostKeyChecking yes", backhaul)
|
|
self.assertIn("ProxyCommand /usr/bin/nc -X 5 -x 127.0.0.1:1055", backhaul)
|
|
self.assertIn("AF_NETLINK", tailscale_unit)
|
|
self.assertIn("User=nodedc-edge", tailscale_unit)
|
|
self.assertIn("StateDirectoryMode=0700", tailscale_unit)
|
|
self.assertIn("User=nodedc-backhaul", backhaul_unit)
|
|
self.assertNotIn("User=nodedc-edge", backhaul_unit)
|
|
self.assertIn("User=nodedc-relay", relay_unit)
|
|
self.assertNotIn("User=nodedc-edge", relay_unit)
|
|
self.assertIn("DEVICE_EDGE_RELAY_SOURCE_POLICY=public-ipv4-only", relay_unit)
|
|
self.assertIn("MemoryMax=192M", relay_unit)
|
|
|
|
def test_runner_has_registered_rollback_and_no_generic_latest(self):
|
|
source = RUNNER_PATH.read_text(encoding="utf-8")
|
|
self.assertIn("def rollback(", source)
|
|
self.assertIn('TAILSCALE_REQUIRED_TAG = "tag:device-edge-vps"', source)
|
|
self.assertIn("assign_backhaul_trust", source)
|
|
self.assertIn("deploy-ok patch=", source)
|
|
self.assertNotIn("apply-latest", source)
|
|
self.assertNotIn("compose down", source)
|
|
self.assertNotIn("docker system prune", source)
|
|
|
|
def test_executable_preflight_accepts_a_valid_alternatives_symlink(self):
|
|
with tempfile.TemporaryDirectory(prefix="nodedc-vps-tool-") as directory:
|
|
root = Path(directory)
|
|
target = root / "netcat.openbsd"
|
|
target.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8")
|
|
target.chmod(0o755)
|
|
command = root / "nc"
|
|
command.symlink_to(target.name)
|
|
|
|
self.assertEqual(
|
|
RUNNER.assert_executable_command_path(command, "test command"),
|
|
target.resolve(),
|
|
)
|
|
|
|
def test_executable_preflight_rejects_a_broken_symlink(self):
|
|
with tempfile.TemporaryDirectory(prefix="nodedc-vps-tool-") as directory:
|
|
command = Path(directory) / "nc"
|
|
command.symlink_to("missing-netcat")
|
|
with self.assertRaises(RUNNER.DeployError):
|
|
RUNNER.assert_executable_command_path(command, "test command")
|
|
|
|
def test_backup_restore_preserves_the_exact_relay_partition(self):
|
|
with tempfile.TemporaryDirectory(prefix="nodedc-vps-backup-") as directory:
|
|
root = Path(directory)
|
|
live = root / "live"
|
|
backups = root / "backups"
|
|
nft = root / "etc/nftables.conf"
|
|
relay_unit = root / "etc/nodedc-b2-relay.service"
|
|
backups.mkdir()
|
|
nft.parent.mkdir(parents=True)
|
|
nft.write_text("foundation-firewall\n", encoding="utf-8")
|
|
relay_unit.write_text("old-unit\n", encoding="utf-8")
|
|
for relative in RUNNER.RELAY_ENTRIES:
|
|
target = live / relative
|
|
if relative.endswith("/src"):
|
|
target.mkdir(parents=True)
|
|
(target / "server.mjs").write_text("old-source\n", encoding="utf-8")
|
|
else:
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.write_text(f"old:{relative}\n", encoding="utf-8")
|
|
|
|
old_live = RUNNER.LIVE_ROOT
|
|
old_backups = RUNNER.BACKUP_ROOT
|
|
old_nft = RUNNER.NFTABLES_CONFIG
|
|
old_relay_unit = RUNNER.RELAY_UNIT
|
|
RUNNER.LIVE_ROOT = live
|
|
RUNNER.BACKUP_ROOT = backups
|
|
RUNNER.NFTABLES_CONFIG = nft
|
|
RUNNER.RELAY_UNIT = relay_unit
|
|
completed = subprocess.CompletedProcess([], 0, "table inet old {}\n", "")
|
|
try:
|
|
with patch.object(RUNNER, "run", return_value=completed), patch.object(
|
|
RUNNER,
|
|
"service_active",
|
|
return_value=False,
|
|
), patch.object(
|
|
RUNNER,
|
|
"systemctl",
|
|
return_value=completed,
|
|
), patch.object(
|
|
RUNNER,
|
|
"user_exists",
|
|
return_value=False,
|
|
):
|
|
_backup_id, backup = RUNNER.create_backup("relay-unit", "relay")
|
|
nft.write_text("candidate-firewall\n", encoding="utf-8")
|
|
relay_unit.write_text("candidate-unit\n", encoding="utf-8")
|
|
(live / "services/device-edge-relay/src/server.mjs").write_text(
|
|
"candidate-source\n",
|
|
encoding="utf-8",
|
|
)
|
|
RUNNER.restore_backup(backup, "relay")
|
|
finally:
|
|
RUNNER.LIVE_ROOT = old_live
|
|
RUNNER.BACKUP_ROOT = old_backups
|
|
RUNNER.NFTABLES_CONFIG = old_nft
|
|
RUNNER.RELAY_UNIT = old_relay_unit
|
|
|
|
self.assertEqual(nft.read_text(), "foundation-firewall\n")
|
|
self.assertEqual(relay_unit.read_text(), "old-unit\n")
|
|
self.assertEqual(
|
|
(live / "services/device-edge-relay/src/server.mjs").read_text(),
|
|
"old-source\n",
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|