#!/usr/bin/env python3 import base64 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 types import SimpleNamespace from unittest import mock SCRIPT_DIR = Path(__file__).resolve().parent BUILDER = SCRIPT_DIR / "build-device-plane-backhaul-target-artifact.mjs" RUNNER_PATH = SCRIPT_DIR / "nodedc-deploy" COMPOSE = ( SCRIPT_DIR.parent.parent / "device-plane/docker-compose.device-plane.backhaul-target.yml" ) PREDECESSOR_COMPOSE = ( SCRIPT_DIR.parent.parent / "device-plane/docker-compose.device-plane.yml" ) SSHD_CONFIG = ( SCRIPT_DIR.parent.parent / "device-plane/services/device-backhaul-target/sshd_config" ) def load_runner(): loader = importlib.machinery.SourceFileLoader( "nodedc_device_plane_backhaul_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() def valid_public_key(comment="nodedc-device-edge-backhaul"): blob = b"\x00\x00\x00\x0bssh-ed25519\x00\x00\x00\x20" + bytes(range(32)) return f"ssh-ed25519 {base64.b64encode(blob).decode()} {comment}\n" def healthy_inventory(): return { "schemaVersion": "nodedc.device-plane.runtime-inventory.v1", "composeProject": "nodedc-device-plane", "services": [ { "service": service, "containerId": character * 64, "imageId": f"sha256:{character * 64}", "status": "running", "running": True, "health": "healthy", "restartCount": 0, } for service, character in ( ("device-control-core", "a"), ("device-gateway", "b"), ("device-postgres", "c"), ) ], } class DevicePlaneBackhaulTargetArtifactTest(unittest.TestCase): def build(self, artifact_dir, patch_id): environment = os.environ.copy() environment["NODEDC_DEPLOY_ARTIFACT_DIR"] = str(artifact_dir) result = subprocess.run( ["node", str(BUILDER), patch_id], check=True, capture_output=True, text=True, env=environment, ) return json.loads(result.stdout) def test_artifact_is_exact_deterministic_and_contains_no_keys(self): with tempfile.TemporaryDirectory( prefix="nodedc-device-plane-backhaul-artifact-", ) as directory: target = Path(directory) first = self.build(target, "device-plane-backhaul-target-unit-001") first_bytes = Path(first["artifact"]).read_bytes() second = self.build(target, "device-plane-backhaul-target-unit-001") second_bytes = Path(second["artifact"]).read_bytes() self.assertEqual(first_bytes, second_bytes) self.assertEqual(first["sha256"], hashlib.sha256(first_bytes).hexdigest()) self.assertEqual( first["entries"], list(RUNNER.DEVICE_PLANE_BACKHAUL_TARGET_ENTRIES), ) self.assertEqual(first["services"], ["device-backhaul-target"]) with tarfile.open(first["artifact"], "r:gz") as archive: names = archive.getnames() files = archive.extractfile("files.txt").read().decode().splitlines() descriptor = json.loads( archive.extractfile( "payload/deployment/" "device-plane-backhaul-target-tailnet-serve-v1.json" ).read() ) self.assertEqual(files, first["entries"]) self.assertEqual( descriptor, RUNNER.expected_device_plane_backhaul_target_descriptor(), ) self.assertFalse(any( name.endswith((".key", ".pem", "authorized_keys")) for name in names )) def test_registry_selects_only_target_and_preserves_red_boundaries(self): entries = RUNNER.DEVICE_PLANE_BACKHAUL_TARGET_ENTRIES self.assertEqual( RUNNER.component_services("device-plane", entries), ("device-backhaul-target",), ) builds = RUNNER.component_builds("device-plane", entries) self.assertEqual(len(builds), 1) self.assertIn("services/device-backhaul-target/Dockerfile", builds[0][1]) self.assertIn(RUNNER.DEVICE_PLANE_BACKHAUL_TARGET_IMAGE, builds[0][1]) for path in ( RUNNER.DEVICE_PLANE_BACKHAUL_TARGET_COMPOSE_REL, "services/device-backhaul-target/Dockerfile", "services/device-backhaul-target/sshd_config", RUNNER.DEVICE_PLANE_BACKHAUL_TARGET_REL, ): self.assertTrue(RUNNER.allowed_payload_path("device-plane", path)) compose = COMPOSE.read_text(encoding="utf-8") sshd = SSHD_CONFIG.read_text(encoding="utf-8") self.assertIn("network_mode: host", compose) self.assertNotIn("0.0.0.0:2222", compose) self.assertIn('"127.0.0.1", "2222"', compose) self.assertIn("ListenAddress 127.0.0.1", sshd) self.assertIn("AllowTcpForwarding local", sshd) self.assertIn("PermitOpen 127.0.0.1:9921", sshd) self.assertIn("ForceCommand /bin/false", sshd) self.assertIn("PasswordAuthentication no", sshd) def test_preflight_requires_exact_applied_006_and_enrollment_key(self): with tempfile.TemporaryDirectory( prefix="nodedc-device-plane-backhaul-preflight-", ) as directory: work = Path(directory) built = self.build(work, "device-plane-backhaul-target-unit-002") extracted = work / "extracted" extracted.mkdir() _manifest, _entries, payload = RUNNER.load_artifact( Path(built["artifact"]), extracted, ) live = work / "live" live.mkdir() (live / "docker-compose.device-plane.yml").write_bytes( PREDECESSOR_COMPOSE.read_bytes() ) descriptor = live / RUNNER.DEVICE_PLANE_B2_DISCOVERY_INGRESS_REL descriptor.parent.mkdir(parents=True) descriptor.write_text( json.dumps(RUNNER.expected_device_plane_b2_discovery_ingress_descriptor()), encoding="utf-8", ) enrollment = work / "device-edge-backhaul.pub" enrollment.write_text(valid_public_key(), encoding="ascii") def has_patch(value): return value == RUNNER.DEVICE_PLANE_BACKHAUL_TARGET_PREDECESSOR_PATCH_ID def has_sha(value): return value == RUNNER.DEVICE_PLANE_BACKHAUL_TARGET_PREDECESSOR_ARTIFACT_SHA256 with ( mock.patch.object(RUNNER, "component_root", return_value=live), mock.patch.object( RUNNER, "DEVICE_PLANE_BACKHAUL_ENROLLMENT_PUBLIC_KEY_FILE", enrollment, ), mock.patch.object(RUNNER, "state_has_patch_id", side_effect=has_patch), mock.patch.object(RUNNER, "state_has_sha", side_effect=has_sha), mock.patch.object( RUNNER, "device_plane_service_container_ids", return_value=[], ), mock.patch.object( RUNNER, "device_plane_runtime_inventory", return_value=healthy_inventory(), ), mock.patch.object(RUNNER, "assert_loopback_tcp_port_open"), mock.patch.object( RUNNER, "validate_device_plane_backhaul_failed_evidence", return_value={ "backup": work / "failed-backup", "failedArtifact": work / "failed-artifact.tgz", }, ), mock.patch.object( RUNNER, "validate_device_plane_tailscale_cli", return_value={ "binary": str(RUNNER.DEVICE_PLANE_TAILSCALE), "uid": 1024, "gid": 1024, "binarySha256": "d" * 64, }, ), mock.patch.object( RUNNER, "validate_device_plane_tailscale_runtime", return_value={ "self": { "Online": True, "TailscaleIPs": ["100.109.216.21"], }, "serve": {}, }, ), ): accepted = RUNNER.validate_device_plane_backhaul_target_evidence(payload) self.assertEqual( accepted["mode"], "failed-backhaul-target-to-loopback-tailnet-serve", ) self.assertRegex(accepted["enrollmentPublicKeySha256"], r"^[a-f0-9]{64}$") self.assertEqual(accepted["tailscaleServeBefore"], {}) self.assertEqual(accepted["tailscaleCli"]["uid"], 1024) def test_registered_health_gate_checks_preserved_and_target_services(self): with mock.patch.object(RUNNER, "healthcheck_compose_service") as health: RUNNER.run_healthchecks( "device-plane", RUNNER.DEVICE_PLANE_BACKHAUL_TARGET_ENTRIES, (RUNNER.DEVICE_PLANE_BACKHAUL_TARGET_SERVICE,), ) self.assertEqual( [call.args[1] for call in health.call_args_list], [ "device-control-core", "device-gateway", "device-postgres", "device-backhaul-target", ], ) def test_candidate_rollback_removes_only_target_and_preserves_runtime(self): runtime = healthy_inventory() with ( mock.patch.object( RUNNER, "read_backup_path_list", side_effect=[[], list(RUNNER.DEVICE_PLANE_BACKHAUL_TARGET_ENTRIES)], ), mock.patch.object( RUNNER, "validate_backup_partition", return_value=(set(), set(RUNNER.DEVICE_PLANE_BACKHAUL_TARGET_ENTRIES)), ), mock.patch.object( RUNNER, "read_strict_json", side_effect=[runtime, {}], ), mock.patch.object( RUNNER, "disable_device_plane_tailscale_serve", ) as disable_serve, mock.patch.object(RUNNER, "stop_and_remove_compose_services") as stop, mock.patch.object(RUNNER, "restore_platform_overlay", return_value=3), mock.patch.object( RUNNER, "device_plane_service_container_ids", return_value=[], ), mock.patch.object( RUNNER, "device_plane_runtime_inventory", return_value=runtime, ), mock.patch.object( RUNNER, "validate_device_plane_tailscale_runtime", return_value={"serve": {}}, ), mock.patch.object(RUNNER, "assert_loopback_tcp_port_open") as port, ): result = RUNNER.rollback_device_plane_apply( Path("/unused/live"), Path("/unused/backup"), RUNNER.DEVICE_PLANE_BACKHAUL_TARGET_ENTRIES, "20260803-000000", True, (RUNNER.DEVICE_PLANE_BACKHAUL_TARGET_SERVICE,), ) self.assertEqual( result, "tailscale-serve-restored-source-restored-target-removed-" "preserved-runtime-unchanged:3", ) stop.assert_called_once_with( "device-plane", (RUNNER.DEVICE_PLANE_BACKHAUL_TARGET_SERVICE,), ) disable_serve.assert_called_once_with({}) port.assert_called_once_with(9921) def test_tailscale_serve_port_parser_rejects_funnel_and_nested_collision(self): clean = {"TCP": {"443": {"HTTPS": True}}} self.assertEqual( RUNNER.device_plane_tailscale_handlers_for_port(clean, 2222), [], ) active = { "TCP": {"2222": {"TCPForward": "127.0.0.1:2222"}}, "Foreground": { "session": { "TCP": {"443": {"HTTPS": True}}, }, }, } self.assertEqual( RUNNER.device_plane_tailscale_handlers_for_port(active, 2222), [((), {"TCPForward": "127.0.0.1:2222"})], ) self.assertFalse( RUNNER.device_plane_tailscale_funnel_uses_port(active, 2222) ) active["AllowFunnel"] = {"edge.example.ts.net:2222": True} self.assertTrue( RUNNER.device_plane_tailscale_funnel_uses_port(active, 2222) ) def test_runtime_activation_enables_private_tailscale_serve_after_health(self): calls = [] with ( mock.patch.object(RUNNER, "run_build", side_effect=lambda *a: calls.append("build")), mock.patch.object( RUNNER, "prepare_component_runtime", side_effect=lambda *a: calls.append("prepare"), ), mock.patch.object( RUNNER, "run_compose", side_effect=lambda *a: calls.append("compose"), ), mock.patch.object( RUNNER, "healthcheck_compose_service", side_effect=lambda *a: calls.append("health"), ), mock.patch.object( RUNNER, "enable_device_plane_tailscale_serve", side_effect=lambda *a: calls.append("serve"), ) as enable, ): RUNNER.run_device_plane_runtime_for_apply( RUNNER.DEVICE_PLANE_BACKHAUL_TARGET_ENTRIES, (RUNNER.DEVICE_PLANE_BACKHAUL_TARGET_SERVICE,), lambda: calls.append("started"), backhaul_serve_before={"TCP": {"443": {"HTTPS": True}}}, ) self.assertEqual( calls, ["build", "prepare", "started", "compose", "health", "serve"], ) enable.assert_called_once_with({"TCP": {"443": {"HTTPS": True}}}) def test_tailscale_serve_enable_and_disable_preserve_unrelated_routes(self): before = {"TCP": {"443": {"HTTPS": True}}} active = { "TCP": { "443": {"HTTPS": True}, "2222": {"TCPForward": "127.0.0.1:2222"}, }, } with ( mock.patch.object( RUNNER, "validate_device_plane_tailscale_runtime", side_effect=[{"serve": before}, {"serve": active}], ), mock.patch.object(RUNNER, "run_device_plane_tailscale") as run, ): result = RUNNER.enable_device_plane_tailscale_serve(before) self.assertEqual(result, active) self.assertEqual( run.call_args.args[0], [ "serve", "--bg", "--yes", "--tcp=2222", "tcp://127.0.0.1:2222", ], ) with ( mock.patch.object( RUNNER, "read_device_plane_tailscale_json", return_value=active, ), mock.patch.object( RUNNER, "validate_device_plane_tailscale_runtime", return_value={"serve": before}, ), mock.patch.object(RUNNER, "run_device_plane_tailscale") as run, ): changed = RUNNER.disable_device_plane_tailscale_serve(before) self.assertTrue(changed) self.assertEqual( run.call_args.args[0], [ "serve", "--tcp=2222", "off", ], ) def test_tailscale_cli_runs_as_official_package_account(self): context = { "binary": "/var/packages/Tailscale/target/bin/tailscale", "uid": 1051, "gid": 1051, "binarySha256": "e" * 64, } with ( mock.patch.object( RUNNER, "validate_device_plane_tailscale_cli", return_value=context, ), mock.patch.object(RUNNER.subprocess, "run") as run, mock.patch.object( RUNNER, "device_plane_tailscale_drop_privileges", return_value="drop-to-package-account", ) as drop, ): RUNNER.run_device_plane_tailscale( ["status", "--json"], check=False, capture_output=True, text=True, ) self.assertEqual( run.call_args.args[0], [context["binary"], "status", "--json"], ) self.assertEqual( run.call_args.kwargs["preexec_fn"], "drop-to-package-account", ) drop.assert_called_once_with(1051, 1051) def test_tailscale_cli_accepts_package_owned_binary_without_root_execution(self): privilege_path = mock.MagicMock() privilege_path.__str__.return_value = ( "/var/packages/Tailscale/conf/privilege" ) privilege_path.lstat.return_value = SimpleNamespace( st_mode=RUNNER.stat.S_IFREG | 0o644, st_uid=0, ) binary_path = mock.MagicMock() binary_path.__str__.return_value = ( "/var/packages/Tailscale/target/bin/tailscale" ) binary_path.lstat.return_value = SimpleNamespace( st_mode=RUNNER.stat.S_IFREG | 0o755, st_uid=1051, st_gid=1051, st_size=32 * 1024 * 1024, ) account = SimpleNamespace(pw_uid=1051, pw_gid=1051) group = SimpleNamespace(gr_gid=1051) help_result = SimpleNamespace( stdout="--tcp --bg --yes", stderr="", ) with ( mock.patch.object( RUNNER, "DEVICE_PLANE_TAILSCALE_PRIVILEGE", privilege_path, ), mock.patch.object( RUNNER, "DEVICE_PLANE_TAILSCALE", binary_path, ), mock.patch.object( RUNNER, "read_strict_json", return_value={ "defaults": {"run-as": "package"}, "username": "tailscale", "groupname": "tailscale", }, ), mock.patch.object(RUNNER.pwd, "getpwnam", return_value=account), mock.patch.object(RUNNER.grp, "getgrnam", return_value=group), mock.patch.object( RUNNER, "sha256_file", return_value="f" * 64, ), mock.patch.object( RUNNER.subprocess, "run", return_value=help_result, ) as run, ): context = RUNNER.validate_device_plane_tailscale_cli() self.assertEqual(context["uid"], 1051) self.assertEqual(context["gid"], 1051) self.assertEqual(context["binarySha256"], "f" * 64) self.assertTrue(callable(run.call_args.kwargs["preexec_fn"])) def test_failed_001_evidence_is_exact_and_terminal(self): with tempfile.TemporaryDirectory( prefix="nodedc-device-plane-backhaul-failed-evidence-", ) as directory: root = Path(directory) backups = root / "backups" failed = root / "failed" state = root / "state" tmp = root / "tmp" for path in (backups, failed, state, tmp): path.mkdir() backup = backups / RUNNER.DEVICE_PLANE_BACKHAUL_FAILED_BACKUP_ID backup.mkdir() backup_hashes = {} for name in ( "manifest.env", "files.txt", "source-before.tgz", "existing-files.txt", "missing-files.txt", "runtime-before.json", ): payload = f"fixture:{name}\n".encode() (backup / name).write_bytes(payload) backup_hashes[name] = hashlib.sha256(payload).hexdigest() stage = root / "failed-stage" payload = stage / "payload" service = payload / "services/device-backhaul-target" deployment = payload / "deployment" service.mkdir(parents=True) deployment.mkdir(parents=True) (stage / "manifest.env").write_text( "id=device-plane-backhaul-target-20260803-001\n" "component=device-plane\n" "type=app-overlay\n", encoding="utf-8", ) (stage / "files.txt").write_text( "\n".join(RUNNER.DEVICE_PLANE_BACKHAUL_FAILED_TARGET_ENTRIES) + "\n", encoding="utf-8", ) (payload / RUNNER.DEVICE_PLANE_BACKHAUL_TARGET_COMPOSE_REL).write_text( "services: {}\n", encoding="utf-8", ) (service / "Dockerfile").write_text( "FROM scratch\n", encoding="utf-8", ) (service / "sshd_config").write_text( "PasswordAuthentication no\n", encoding="utf-8", ) (payload / RUNNER.DEVICE_PLANE_BACKHAUL_FAILED_TARGET_REL).write_text( json.dumps( RUNNER.expected_failed_device_plane_backhaul_target_descriptor() ) + "\n", encoding="utf-8", ) failed_artifact = failed / RUNNER.DEVICE_PLANE_BACKHAUL_FAILED_ARTIFACT with tarfile.open(failed_artifact, "w:gz") as archive: for name in ("manifest.env", "files.txt", "payload"): archive.add(stage / name, arcname=name) failed_sha = hashlib.sha256(failed_artifact.read_bytes()).hexdigest() failed_state = state / "failed.jsonl" failed_state.write_text( json.dumps({ "artifact": RUNNER.DEVICE_PLANE_BACKHAUL_FAILED_ARTIFACT, "backup_id": RUNNER.DEVICE_PLANE_BACKHAUL_FAILED_BACKUP_ID, "component": "device-plane", "id": RUNNER.DEVICE_PLANE_BACKHAUL_FAILED_PATCH_ID, "message": RUNNER.DEVICE_PLANE_BACKHAUL_FAILED_MESSAGE, "rollback_status": ( "ok:device-plane-overlay:source-restored-target-removed-" "preserved-runtime-unchanged:3" ), "sha256": failed_sha, "started_apply": True, "status": "failed", }) + "\n", encoding="utf-8", ) with ( mock.patch.object(RUNNER, "BACKUPS_DIR", backups), mock.patch.object(RUNNER, "FAILED_DIR", failed), mock.patch.object(RUNNER, "FAILED_STATE_FILE", failed_state), mock.patch.object(RUNNER, "TMP_DIR", tmp), mock.patch.object( RUNNER, "DEVICE_PLANE_BACKHAUL_FAILED_BACKUP_SHA256", backup_hashes, ), mock.patch.object( RUNNER, "DEVICE_PLANE_BACKHAUL_FAILED_ARTIFACT_SHA256", failed_sha, ), ): evidence = RUNNER.validate_device_plane_backhaul_failed_evidence() self.assertEqual(evidence["backup"], backup) self.assertEqual(evidence["failedArtifact"], failed_artifact) if __name__ == "__main__": unittest.main()