#!/usr/bin/env python3 """Isolated native-updater integration tests: real shell/tar/OpenSSL, fake network/procd. No root, containers, network access or writes to the host's /etc are needed. """ import hashlib import io import json import os from pathlib import Path import shutil import subprocess import sys import tarfile import tempfile import unittest ROOT = Path(__file__).resolve().parents[1] UPDATER = ROOT / "packaging/openwrt/install.sh" SHIM = r''' import json, os, pathlib, shutil, sys, time name = pathlib.Path(sys.argv[0]).name base = pathlib.Path(os.environ['FIXTURE']) if name == 'curl': args = sys.argv[1:] url = next(x for x in args if x.startswith('https://test.invalid/')) package_api = '/api/packages/justme/generic/moshiro-openwrt/' if package_api in url: destination = base / 'published' / url.split(package_api, 1)[1] if '--request' in args: code = 204 if destination.exists() else 404 destination.unlink(missing_ok=True) print(code, end='') else: destination.parent.mkdir(parents=True, exist_ok=True) shutil.copyfile(args[args.index('--upload-file') + 1], destination) print(201, end='') sys.exit(0) source = base / 'feed' / url.split('/feed/', 1)[1] if not source.is_file(): sys.exit(22) shutil.copyfile(source, args[args.index('-o') + 1]) elif name == 'jsonfilter': args = sys.argv[1:] data = json.loads(pathlib.Path(args[args.index('-i') + 1]).read_text() if '-i' in args else sys.stdin.read()) query = args[args.index('-e') + 1] value = data['moshiro-agent']['instances']['default']['pid'] if query.startswith('@[') else data.get(query[2:]) if value is None: sys.exit(1) print(value) elif name == 'ubus': print(json.dumps({'moshiro-agent': {'instances': {'default': {'pid': 1234}}}})) elif name == 'service': with (base / 'restarts').open('a') as f: f.write('restart\n') if os.environ.get('HEALTH') != 'good': sys.exit(0) manifest = json.loads((base / 'feed/dev/manifest-linux-musl-arm64.json').read_text()) runtime = base / 'runtime' runtime.mkdir(exist_ok=True) (runtime / 'health.json').write_text(json.dumps({'version': manifest['version'], 'pid': 1234, 'at': int(time.time())})) ''' class UpdaterTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory(prefix="moshiro-openwrt-", dir="/tmp/opencode" if Path("/tmp/opencode").is_dir() else None) self.base = Path(self.temp.name) self.root = self.base / "installation" self.state = self.base / "state" self.runtime = self.base / "runtime" self.bin = self.base / "bin" for path in (self.state, self.runtime, self.bin, self.base / "feed/dev"): path.mkdir(parents=True) self.old = "a" * 64 (self.root / "versions" / self.old).mkdir(parents=True) (self.root / "current").symlink_to(f"versions/{self.old}") self.key = self.base / "key.pem" subprocess.run(["openssl", "ecparam", "-name", "prime256v1", "-genkey", "-noout", "-out", str(self.key)], check=True) subprocess.run(["openssl", "pkey", "-in", str(self.key), "-pubout", "-out", str(self.state / "release-key.pem")], check=True) for name in ("curl", "jsonfilter", "ubus", "service"): path = self.bin / name path.write_text(f"#!{sys.executable}\n" + SHIM) path.chmod(0o755) self.conf = self.base / "config" self.conf.write_text("FEED='https://test.invalid/feed'\nCHANNEL=dev\nRID=linux-musl-arm64\nPOLICY=auto\nINTERVAL=30\nHEALTH_TIMEOUT=2\nHEALTH_OBSERVE=0\n") self.env = dict(os.environ, PATH=str(self.bin) + ":" + os.environ["PATH"], FIXTURE=str(self.base), HEALTH="good", MOSHIRO_NATIVE_TEST="1", MOSHIRO_NATIVE_ROOT=str(self.root), MOSHIRO_NATIVE_CONF=str(self.conf), MOSHIRO_NATIVE_STATE=str(self.state), MOSHIRO_NATIVE_STATUS=str(self.base / "status"), MOSHIRO_NATIVE_RUNTIME=str(self.runtime), MOSHIRO_NATIVE_SERVICE=str(self.bin / "service")) self.publish() def tearDown(self): self.temp.cleanup() def publish(self, *, seq=100, version="1.0.1-dev.1", members=None, changes=None): payload = io.BytesIO() members = members or {name: b"executable\n" for name in ("moshiro-agent", "fleet-enroll", "openwrt.sh", "agent.init", "updater.init")} with tarfile.open(fileobj=payload, mode="w:gz") as archive: for name, content in members.items(): entry = tarfile.TarInfo(name) entry.size = len(content) archive.addfile(entry, io.BytesIO(content)) data = payload.getvalue() self.hash = hashlib.sha256(data).hexdigest() package = f"agent-linux-musl-arm64-{self.hash}.tar.gz" version_dir = self.base / "feed" / version version_dir.mkdir(exist_ok=True) (version_dir / package).write_bytes(data) self.manifest = dict(schema=1, component="agent", channel="dev", rid="linux-musl-arm64", version=version, package=package, sha256=self.hash, size=len(data), seq=seq) self.manifest.update(changes or {}) self.sign() def sign(self): path = self.base / "feed/dev/manifest-linux-musl-arm64.json" path.write_text(json.dumps(self.manifest, separators=(",", ":"))) sig = subprocess.run(["openssl", "dgst", "-sha256", "-sign", str(self.key), str(path)], capture_output=True, check=True).stdout import base64 path.with_suffix(".sig").write_bytes(base64.b64encode(sig)) def run_check(self, success=True): result = subprocess.run(["sh", str(UPDATER), "check"], env=self.env, capture_output=True, text=True) if success: self.assertEqual(result.returncode, 0, result.stdout + result.stderr) else: self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr) return result def current(self): return os.readlink(self.root / "current") def status(self): return json.loads((self.base / "status/status.json").read_text())["state"] def test_healthy_update_switches_atomically_and_records_sequence(self): self.run_check() self.assertEqual(self.current(), f"versions/{self.hash}") self.assertEqual((self.state / "dev.seq").read_text().strip(), "100") self.assertEqual(self.status(), "up_to_date") self.assertTrue((self.root / "versions" / self.old).is_dir()) self.assertFalse((self.state / "pending").exists()) self.run_check() # Idempotent; no second restart. self.assertEqual((self.base / "restarts").read_text().count("restart"), 1) def test_ci_publisher_produces_a_signed_package_the_updater_can_install(self): binaries = self.base / "binaries" binaries.mkdir() for name in ("moshiro-agent", "fleet-enroll"): (binaries / name).write_bytes(b"native executable fixture\n") env = dict(self.env, CI_COMMIT_TAG="", CI_PIPELINE_EVENT="push", CI_COMMIT_BRANCH="main", CI_COMMIT_MESSAGE="[CI] fixture", GITEA_BASE="https://test.invalid", GITEA_TOKEN="test-token", RELEASE_SIGNING_KEY=self.key.read_text()) result = subprocess.run(["bash", str(ROOT / "ci/publish-openwrt.sh"), "dev", "linux-musl-arm64", str(binaries)], env=env, capture_output=True, text=True) self.assertEqual(result.returncode, 0, result.stdout + result.stderr) shutil.copytree(self.base / "published", self.base / "feed", dirs_exist_ok=True) metadata = json.loads((self.base / "published/dev/manifest-linux-musl-arm64.json").read_text()) self.run_check() self.assertEqual(self.current(), "versions/" + metadata["sha256"]) self.assertEqual(self.status(), "up_to_date") def test_unhealthy_update_rolls_back_and_is_not_retried(self): self.env["HEALTH"] = "bad" self.run_check(False) self.assertEqual(self.current(), f"versions/{self.old}") self.assertEqual(self.status(), "rolled_back") self.run_check() self.assertEqual((self.base / "restarts").read_text().count("restart"), 2) def test_tampered_manifest_never_changes_current(self): path = self.base / "feed/dev/manifest-linux-musl-arm64.json" path.write_text(path.read_text().replace('"seq":100', '"seq":101')) self.run_check(False) self.assertEqual(self.status(), "invalid") self.assertEqual(self.current(), f"versions/{self.old}") def test_wrong_signing_key_is_rejected(self): other_key = self.base / "other.pem" subprocess.run(["openssl", "ecparam", "-name", "prime256v1", "-genkey", "-noout", "-out", str(other_key)], check=True) subprocess.run(["openssl", "pkey", "-in", str(other_key), "-pubout", "-out", str(self.state / "release-key.pem")], check=True) self.run_check(False) self.assertEqual(self.current(), f"versions/{self.old}") self.assertEqual(self.status(), "invalid") def test_tampered_archive_never_changes_current(self): path = self.base / "feed" / self.manifest["version"] / self.manifest["package"] data = path.read_bytes() path.write_bytes(b"X" + data[1:]) self.run_check(False) self.assertEqual(self.current(), f"versions/{self.old}") self.assertFalse((self.state / "dev.seq").exists()) def test_wrong_architecture_channel_schema_and_url_are_rejected(self): for changes in ({"rid": "linux-musl-x64"}, {"channel": "release"}, {"schema": 2}, {"package": "../../evil"}, {"version": ".."}, {"sha256": "0" * 63}, {"seq": "999999999999999999999999999999"}): with self.subTest(changes=changes): self.publish(changes=changes) self.run_check(False) self.assertEqual(self.current(), f"versions/{self.old}") def test_archive_path_traversal_is_rejected_before_extraction(self): self.publish(members={"../../escape": b"bad"}) self.run_check(False) self.assertFalse((self.base / "escape").exists()) self.assertEqual(self.current(), f"versions/{self.old}") def test_old_sequence_does_not_replace_current(self): (self.state / "dev.seq").write_text("101\n") self.run_check() self.assertEqual(self.current(), f"versions/{self.old}") self.assertFalse((self.base / "restarts").exists()) def test_manual_policy_requires_request_and_consumes_it(self): self.conf.write_text(self.conf.read_text().replace("POLICY=auto", "POLICY=manual")) self.run_check() self.assertEqual(self.status(), "available") self.assertEqual(self.current(), f"versions/{self.old}") (self.runtime / "apply.request").write_text("untrusted content is never executed") self.run_check() self.assertEqual(self.current(), f"versions/{self.hash}") self.assertFalse((self.runtime / "apply.request").exists()) def test_interrupted_update_recovers_previous_version_before_fetching(self): candidate = self.root / "versions" / self.hash candidate.mkdir() (self.root / "current").unlink() (self.root / "current").symlink_to(f"versions/{self.hash}") (self.state / "pending").write_text(f"versions/{self.old}\n") (self.state / "dev.seq").write_text("100\n") self.run_check() self.assertEqual(self.current(), f"versions/{self.old}") self.assertEqual(self.status(), "rolled_back") self.assertFalse((self.state / "pending").exists()) if __name__ == "__main__": unittest.main(verbosity=2)