justme
016dbc2bbf
Use Agent for native, container, systemd and OpenWrt installations and signed updates. Remove retired Probe deployment templates and smoke scripts.
236 lines
12 KiB
Python
236 lines
12 KiB
Python
#!/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)
|