#!/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)