#!/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", "runtime-reconciliation", "backhaul", "relay", "core-channel", ): 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", "runtime-reconciliation", "backhaul", "relay", "core-channel", ): 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_core_channel_plan_is_exact_and_keeps_tracker_ingress_closed(self): with tempfile.TemporaryDirectory(prefix="nodedc-vps-channel-plan-") as directory: inbox = Path(directory) / "inbox" inbox.mkdir() result = self.build( inbox, "core-channel", "device-edge-vps-core-channel-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": "accepted-foundation-closed-channel"}, ), 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=core-channel", rendered) self.assertIn( "predecessor=accepted-foundation-closed-channel", rendered, ) self.assertIn( "public_core_channel=155.212.211.15:8443/tcp:tls13-mtls-h2", rendered, ) self.assertIn("tracker_tcp_9921=closed", rendered) self.assertIn("public_b2_ingress=disabled", rendered) self.assertIn( "peer_trust=preprovisioned-pinned-self-signed-core-certificate+fingerprint", rendered, ) self.assertIn("command_transport=disabled", rendered) self.assertIn("gelios=untouched", rendered) def test_runtime_reconciliation_plan_is_exact_and_opens_no_port(self): with tempfile.TemporaryDirectory(prefix="nodedc-vps-reconcile-plan-") as directory: inbox = Path(directory) / "inbox" inbox.mkdir() result = self.build( inbox, "runtime-reconciliation", "device-edge-vps-runtime-reconciliation-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": ( "failed-core-channel-001-rollback-runtime-mode-drift" ), }, ), 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=runtime-reconciliation", rendered) self.assertIn( "runtime_reconciliation=exact-known-binaries:0644=>0755", rendered, ) self.assertIn("public_core_channel=disabled", rendered) self.assertIn("tracker_tcp_9921=closed", rendered) def test_publish_payload_preserves_unselected_executable_modes(self): with tempfile.TemporaryDirectory(prefix="nodedc-vps-publish-scope-") as directory: root = Path(directory) live = root / "live" payload = root / "payload" runtime = live / "runtime/node/bin/node" marker = payload / "deployment/reconciliation.json" runtime.parent.mkdir(parents=True) marker.parent.mkdir(parents=True) runtime.write_bytes(b"runtime-binary") runtime.chmod(0o755) marker.write_text("{}\n", encoding="utf-8") old_live = RUNNER.LIVE_ROOT RUNNER.LIVE_ROOT = live try: with patch.object(RUNNER.os, "chown"): RUNNER.publish_payload(payload, ("deployment/reconciliation.json",)) finally: RUNNER.LIVE_ROOT = old_live self.assertEqual(runtime.stat().st_mode & 0o777, 0o755) self.assertEqual( (live / "deployment/reconciliation.json").stat().st_mode & 0o777, 0o644, ) def test_source_baseline_is_pinned_to_the_exact_accepted_predecessor(self): with tempfile.TemporaryDirectory(prefix="nodedc-vps-baseline-") as directory: journal = Path(directory) / "applied.jsonl" old_journal = RUNNER.APPLIED_JOURNAL RUNNER.APPLIED_JOURNAL = journal try: journal.write_text(json.dumps({ "patch": "device-edge-vps-foundation-20260806-003", "phase": "foundation", "sha256": ( "1be852f144e9f0fea32af70bebd07a2607b6a1818825094bd4c1b4062064716a" ), "status": "ok", }) + "\n", encoding="utf-8") expected = RUNNER.phase_file_sha256("foundation") self.assertEqual( expected["vps/config/00-nodedc-b2-vps.conf"], "cc94d0579f85d0af9746b9ce760bc72980f4a22fb59027e1f5f9c7bf3aaebd64", ) self.assertEqual( expected["vps/config/nftables-foundation.conf"], "4d44f902d8d98d1aa8506fca9d9582e700f6424def2b1d667ab2cd5a5ee84934", ) self.assertEqual( expected["deployment/device-edge-vps-foundation-v1.json"], "317c98b42520fff3238275908482de7aa611b4ee41c6b1f8062abd2730ab072a", ) journal.write_text(json.dumps({ "patch": "device-edge-vps-foundation-20260806-003", "phase": "foundation", "sha256": "0" * 64, "status": "ok", }) + "\n", encoding="utf-8") unexpected = RUNNER.phase_file_sha256("foundation") self.assertEqual( unexpected["vps/config/00-nodedc-b2-vps.conf"], RUNNER.PHASE_FILE_SHA256[ "foundation" ]["vps/config/00-nodedc-b2-vps.conf"], ) finally: RUNNER.APPLIED_JOURNAL = old_journal 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() channel = (source_root / "vps/config/nftables-core-channel.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() channel_unit = ( source_root / "vps/systemd/nodedc-device-edge-channel.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("tcp dport 8443", channel) self.assertNotIn("tcp dport 9921", channel) 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) self.assertIn("User=nodedc-channel", channel_unit) self.assertIn( "ExecStart=/opt/nodedc-b2-vps/runtime/node/bin/node " "/opt/nodedc-b2-vps/services/device-edge-channel/src/server.mjs", channel_unit, ) self.assertIn("MemoryDenyWriteExecute=no", channel_unit) self.assertNotIn("--jitless", channel_unit) self.assertIn("MemoryMax=128M", channel_unit) self.assertIn("MemorySwapMax=0", channel_unit) self.assertIn("CPUQuota=50%", channel_unit) self.assertIn("TasksMax=64", channel_unit) self.assertIn("LimitNOFILE=1024", channel_unit) self.assertNotIn("LocalForward", channel_unit) self.assertNotIn("DEVICE_EDGE_RELAY_UPSTREAM", channel_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", ) def test_backup_restore_preserves_core_channel_source_trust_and_firewall(self): with tempfile.TemporaryDirectory(prefix="nodedc-vps-channel-backup-") as directory: root = Path(directory) live = root / "live" backups = root / "backups" nft = root / "etc/nftables.conf" channel_unit = root / "etc/nodedc-device-edge-channel.service" trust = root / "state/channel-trust" backups.mkdir() nft.parent.mkdir(parents=True) trust.mkdir(parents=True) nft.write_text("foundation-firewall\n", encoding="utf-8") channel_unit.parent.mkdir(parents=True, exist_ok=True) channel_unit.write_text("old-channel-unit\n", encoding="utf-8") (trust / "runtime.json").write_text("old-runtime\n", encoding="utf-8") for relative in RUNNER.CORE_CHANNEL_ENTRIES: target = live / relative if relative.endswith("/src"): target.mkdir(parents=True) (target / "server.mjs").write_text("old-channel-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_unit = RUNNER.CHANNEL_UNIT old_trust = RUNNER.CHANNEL_TRUST_ROOT RUNNER.LIVE_ROOT = live RUNNER.BACKUP_ROOT = backups RUNNER.NFTABLES_CONFIG = nft RUNNER.CHANNEL_UNIT = channel_unit RUNNER.CHANNEL_TRUST_ROOT = trust 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( "channel-unit", "core-channel", ) nft.write_text("candidate-firewall\n", encoding="utf-8") channel_unit.write_text("candidate-channel-unit\n", encoding="utf-8") (trust / "runtime.json").write_text("candidate-runtime\n", encoding="utf-8") (live / "services/device-edge-channel/src/server.mjs").write_text( "candidate-channel-source\n", encoding="utf-8", ) RUNNER.restore_backup(backup, "core-channel") finally: RUNNER.LIVE_ROOT = old_live RUNNER.BACKUP_ROOT = old_backups RUNNER.NFTABLES_CONFIG = old_nft RUNNER.CHANNEL_UNIT = old_unit RUNNER.CHANNEL_TRUST_ROOT = old_trust self.assertEqual(nft.read_text(), "foundation-firewall\n") self.assertEqual(channel_unit.read_text(), "old-channel-unit\n") self.assertEqual((trust / "runtime.json").read_text(), "old-runtime\n") self.assertEqual( (live / "services/device-edge-channel/src/server.mjs").read_text(), "old-channel-source\n", ) if __name__ == "__main__": unittest.main(verbosity=2)