"""Offline install/asset contracts; bootstrap is mocked, never installed globally.""" import argparse import contextlib import hashlib import importlib.util import io import json from pathlib import Path import subprocess import tarfile import tempfile import unittest from unittest import mock ROOT = Path(__file__).resolve().parents[2] def load_module(name, path): spec = importlib.util.spec_from_file_location(name, path) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module installer = load_module("dever_installer", ROOT / "skills/scripts/install.py") assets = load_module("dever_release_assets", ROOT / "sdk/release-assets.py") class InstallerTests(unittest.TestCase): def setUp(self): self.temporary = tempfile.TemporaryDirectory(prefix="dever-installer-test-") self.addCleanup(self.temporary.cleanup) self.root = Path(self.temporary.name) self.release = self.directory("release") self.scratch = self.directory("scratch") self.platform = installer.platform_name() self.payloads = { "dever-core": b"fixture core, never executed\n", "bootstrap/dever": b"fixture launcher, never executed\n", "bootstrap/lib/libfixture.so": b"fixture private closure\n", "skills/dever-language/SKILL.md": b"fixture skill\n", } self.catalog = {} for name, content in self.payloads.items(): path = self.release / name path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(content) self.catalog[name] = { "path": name, "bytes": len(content), "sha256": hashlib.sha256(content).hexdigest(), } manifest = { "format": "dever-release-v2", "platform": self.platform, "version": "1.2.3", "artifacts": list(self.catalog.values()), "extensions": [], } (self.release / "manifest.json").write_text(json.dumps(manifest)) self.private = self.root / "test-key.pem" self.openssl("genpkey", "-algorithm", "ED25519", "-out", str(self.private)) public_der = self.openssl("pkey", "-in", str(self.private), "-pubout", "-outform", "DER") self.public = self.root / "test-key.pub" self.public.write_text(public_der[-32:].hex()) self.sign_manifest() def sign_manifest(self): signature = self.root / "signature.bin" self.openssl("pkeyutl", "-sign", "-inkey", str(self.private), "-rawin", "-in", str(self.release / "manifest.json"), "-out", str(signature)) (self.release / "manifest.sig").write_text(signature.read_bytes().hex()) def directory(self, name): path = self.root / name path.mkdir() return path def openssl(self, *arguments): return subprocess.run(["/usr/bin/openssl", *arguments], env={}, capture_output=True, check=True, timeout=15).stdout def authenticate(self, key=None): return installer.authenticate(self.release, key or self.public, self.scratch, self.platform, "1.2.3") def create_assets(self, output): return assets.create(self.release, output, Path("/usr/bin/zstd")) def archive_payloads(self, path): contents = subprocess.run(["/usr/bin/zstd", "-q", "-d", "-c", str(path)], env={}, capture_output=True, check=True, timeout=15).stdout with tarfile.open(fileobj=io.BytesIO(contents)) as archive: return {member.name: archive.extractfile(member).read() for member in archive} def test_signature_accepts_independent_key_and_rejects_wrong_key_or_changed_manifest(self): version, catalog, key = self.authenticate() self.assertEqual(version, "1.2.3") self.assertEqual(catalog, self.catalog) self.assertEqual(len(key), 32) wrong = self.root / "wrong.pub" wrong.write_text(bytes(32).hex()) with self.assertRaisesRegex(ValueError, "签名校验失败"): self.authenticate(wrong) with (self.release / "manifest.json").open("a") as stream: stream.write(" ") with self.assertRaisesRegex(ValueError, "签名校验失败"): self.authenticate() def test_asset_author_round_trip_is_deterministic_and_contains_only_signed_payload(self): first = self.root / "first" second = self.root / "second" (self.release / "unlisted").write_text("not published") names = self.create_assets(first) self.assertEqual(names, self.create_assets(second)) self.assertEqual(len(names), 5) for name in names: self.assertEqual((first / name).read_bytes(), (second / name).read_bytes()) actual = self.archive_payloads(first / f"dever-{self.platform}.tar.zst") self.assertEqual(actual, self.payloads) for name in ("bootstrap/dever", "bootstrap/lib/libfixture.so"): blob = first / f"dever-{self.platform}.blob-{self.catalog[name]['sha256']}" self.assertEqual(blob.read_bytes(), self.payloads[name]) with self.assertRaisesRegex(ValueError, "already exists"): self.create_assets(first) def test_extension_assets_are_separate_and_cannot_overlap_base(self): manifest_path = self.release / "manifest.json" manifest = json.loads(manifest_path.read_text()) name = f"runtime/pip/{self.platform}/runtime.pack" path = self.release / name path.parent.mkdir(parents=True) contents = b"optional Python bytes" path.write_bytes(contents) entry = {"path": name, "bytes": len(contents), "sha256": hashlib.sha256(contents).hexdigest()} manifest["extensions"] = [{"kind": {"type": "runtime", "ecosystem": "pip"}, "target": self.platform, "artifacts": [entry]}] manifest_path.write_text(json.dumps(manifest)) self.sign_manifest() self.authenticate() output = self.root / "split" self.create_assets(output) self.assertEqual(self.archive_payloads(output / f"dever-{self.platform}.tar.zst"), self.payloads) extension = output / f"dever-{self.platform}.ext-runtime-pip-{self.platform}.tar.zst" self.assertEqual(self.archive_payloads(extension), {name: contents}) manifest["extensions"][0]["artifacts"].append(self.catalog["dever-core"]) manifest_path.write_text(json.dumps(manifest)) with self.assertRaisesRegex(ValueError, "resource namespace"): self.create_assets(self.root / "overlap") manifest["extensions"] = [{"kind": {"type": "target"}, "target": self.platform, "artifacts": [self.catalog["dever-core"]]}] manifest_path.write_text(json.dumps(manifest)) with self.assertRaisesRegex(ValueError, "base package"): self.create_assets(self.root / "host-target") def test_signed_release_without_matching_skill_is_rejected(self): path = self.release / "manifest.json" manifest = json.loads(path.read_text()) manifest["artifacts"] = [entry for entry in manifest["artifacts"] if not entry["path"].startswith("skills/")] path.write_text(json.dumps(manifest)) self.sign_manifest() with self.assertRaisesRegex(ValueError, "配套 skill"): self.authenticate() def test_publish_preserves_concurrent_targets_and_owned_staging_on_failure(self): staging = self.directory("staging") (staging / "asset").write_bytes(b"owned release bytes") empty = self.directory("empty") populated = self.directory("populated") (populated / "existing").write_bytes(b"existing user bytes") file = self.root / "file" file.write_bytes(b"existing file bytes") symlink = self.root / "symlink" symlink.symlink_to(self.root / "missing") for destination in (empty, populated, file, symlink): with self.subTest(destination=destination.name), self.assertRaises(OSError): assets.publish(staging, destination) self.assertEqual((staging / "asset").read_bytes(), b"owned release bytes") self.assertEqual(list(empty.iterdir()), []) self.assertEqual((populated / "existing").read_bytes(), b"existing user bytes") self.assertEqual(file.read_bytes(), b"existing file bytes") self.assertEqual(symlink.readlink(), self.root / "missing") destination = self.root / "published" assets.publish(staging, destination) self.assertFalse(staging.exists()) self.assertEqual((destination / "asset").read_bytes(), b"owned release bytes") def test_create_cleans_only_its_staging_when_target_appears_during_preparation(self): actual_publish = assets.publish for populated in (False, True): output = self.root / ("race-populated" if populated else "race-empty") def publish(prepared, destination): destination.mkdir() if populated: (destination / "user-file").write_bytes(b"concurrent user bytes") actual_publish(prepared, destination) with mock.patch.object(assets, "publish", side_effect=publish), \ self.assertRaises(OSError): self.create_assets(output) self.assertTrue(output.is_dir()) if populated: self.assertEqual((output / "user-file").read_bytes(), b"concurrent user bytes") else: self.assertEqual(list(output.iterdir()), []) self.assertEqual(list(self.root.glob(".dever-release-assets-*")), []) def test_network_payload_executes_only_verified_helper_with_private_libraries(self): assets_path = self.root / "assets" self.create_assets(assets_path) calls = [] def download(address): self.assertTrue(address.startswith( f"https://git.dever.cc/shemic/dever/dist/1.2.3/dever-{self.platform}.")) return (assets_path / address.rsplit("/", 1)[-1]).open("rb") def extract(command, **kwargs): self.assertEqual(command[1], "--dever-extract") launcher = Path(command[0]) self.assertEqual(launcher.read_bytes(), self.payloads["bootstrap/dever"]) self.assertEqual((launcher.parent / "lib/libfixture.so").read_bytes(), self.payloads["bootstrap/lib/libfixture.so"]) self.assertEqual(launcher.stat().st_mode & 0o777, 0o755) configuration = json.loads(Path(command[2]).read_text()) self.assertEqual(configuration["trusted_key"], str(self.public)) self.assertEqual(kwargs["env"], {}) self.assertTrue(Path(configuration["archive"]).is_file()) calls.append(command) with mock.patch.object(installer, "download", side_effect=download), \ mock.patch.object(installer.subprocess, "run", side_effect=extract): installer.payload(self.directory("network"), None, "1.2.3", self.platform, self.catalog, self.public) self.assertEqual(len(calls), 1) self.assertEqual(list(self.root.glob(".extract-*")), []) blob = assets_path / f"dever-{self.platform}.blob-{self.catalog['bootstrap/dever']['sha256']}" for corrupt in (b"x" * len(self.payloads["bootstrap/dever"]), self.payloads["bootstrap/dever"] + b"extra"): blob.write_bytes(corrupt) with mock.patch.object(installer, "download", side_effect=download), \ mock.patch.object(installer.subprocess, "run") as execute, \ self.assertRaises(ValueError): installer.payload(self.release, None, "1.2.3", self.platform, self.catalog, self.public) execute.assert_not_called() def test_download_archive_is_bounded(self): with mock.patch.object(installer, "MAX_RELEASE", 4), \ mock.patch.object(installer, "download", return_value=io.BytesIO(b"12345")), \ self.assertRaisesRegex(ValueError, "超过"): installer.download_archive("unused", self.root / "oversized.zst") def test_local_payload_rejects_symlink_and_source_directory(self): copied = self.directory("copied") installer.payload(copied, self.release, "1.2.3", self.platform, self.catalog) self.assertEqual((copied / "dever-core").read_bytes(), self.payloads["dever-core"]) core = self.release / "dever-core" core.unlink() core.symlink_to(self.public) with self.assertRaisesRegex(ValueError, "符号链接"): installer.payload(self.directory("symlink"), self.release, "1.2.3", self.platform, self.catalog) with self.assertRaisesRegex(ValueError, "symlinks"): self.create_assets(self.root / "invalid-assets") core.unlink() core.mkdir() with self.assertRaisesRegex(ValueError, "类型"): installer.payload(self.directory("directory"), self.release, "1.2.3", self.platform, self.catalog) def test_image_install_authenticates_before_mock_bootstrap_and_cleans_owned_staging(self): image = self.directory("image") arguments = argparse.Namespace(version="1.2.3", system_root=image, release=self.release, trusted_key=self.public, no_start=False) actual_run = subprocess.run calls = [] def run(command, **kwargs): if command[0] == "/usr/bin/openssl": return actual_run(command, **kwargs) self.assertEqual(command[1], "--dever-install") scratch = Path(command[2]) setting = json.loads((scratch / "config/setting.json").read_text()) self.assertEqual(set(setting), {"release", "system_root", "trusted_key", "activate_service"}) self.assertEqual(setting["system_root"], str(image)) self.assertFalse(setting["activate_service"]) self.assertEqual(kwargs["env"], {}) self.assertFalse(Path(setting["trusted_key"]).is_relative_to(Path(setting["release"]))) calls.append(command) return subprocess.CompletedProcess(command, 0) with mock.patch.object(installer.os, "geteuid", return_value=0), \ mock.patch.object(installer.subprocess, "run", side_effect=run), \ contextlib.redirect_stdout(io.StringIO()): installer.install(arguments) self.assertEqual(len(calls), 1) self.assertEqual(list(image.iterdir()), []) (self.release / "dever-core").write_bytes(b"corrupt") with self.assertRaises(ValueError): installer.install(arguments) self.assertEqual(len(calls), 1) self.assertEqual(list(image.iterdir()), []) def test_metadata_uses_latest_or_exact_directory_and_never_a_github_fallback(self): for version in ("latest", "1.2.3"): destination = self.directory(version) addresses = [] def download(address): addresses.append(address) return io.BytesIO(b"metadata") with mock.patch.object(installer, "download", side_effect=download): installer.metadata(destination, None, version, self.platform) self.assertEqual(addresses, [ f"https://git.dever.cc/shemic/dever/dist/{version}/dever-{self.platform}.{suffix}" for suffix in ("manifest.json", "manifest.sig") ]) def test_redirects_remain_on_the_official_https_site(self): handler = installer.OfficialRedirect() current = "https://git.dever.cc/shemic/dever/dist/latest/a" request = installer.urllib.request.Request(current) allowed = "https://git.dever.cc/shemic/dever/dist/1.2.3/a" redirected = handler.redirect_request(request, None, 302, "Found", {}, allowed) self.assertEqual(redirected.full_url, allowed) for address in ( "http://git.dever.cc/a", "https://evil.example/a", "https://git.dever.cc.evil.example/a", "https://user@git.dever.cc/a", "https://git.dever.cc:444/a", "https://git.dever.cc/a#fragment", "https://github.com/a", "https://release-assets.githubusercontent.com/a"): with self.subTest(address=address), self.assertRaises(ValueError): handler.redirect_request(request, None, 302, "Found", {}, address) def test_prerequisites_reject_unsupported_hosts_before_download(self): with mock.patch.object(installer.os, "confstr", return_value="glibc 2.39"): installer.prerequisites("linux-x86_64") for libc in ("glibc 2.38", "musl 1.2", ""): with mock.patch.object(installer.os, "confstr", return_value=libc), \ self.assertRaisesRegex(ValueError, "glibc"): installer.prerequisites("linux-x86_64") with mock.patch.object(installer.os, "access", return_value=False), \ self.assertRaisesRegex(ValueError, "openssl"): installer.prerequisites("linux-x86_64") @unittest.skipUnless(installer.os.geteuid() == 0, "owned image installation requires root") def test_standalone_bash_uses_embedded_key_and_cleans_after_signature_failure(self): # Execute an independently signed fixture to observe the entire entry # and bootstrap call chain, without installing a machine or service. generator = load_module("dever_install_entry", ROOT / "sdk/install-entry.py") launcher = self.release / "bootstrap/dever" launcher.write_text('''#!/usr/bin/python3 import json from pathlib import Path import sys assert sys.argv[1] == "--dever-install" setting = json.loads((Path(sys.argv[2]) / "config/setting.json").read_text()) assert setting["activate_service"] is False (Path(setting["system_root"]) / "called.json").write_text(json.dumps(setting)) ''') content = launcher.read_bytes() manifest = json.loads((self.release / "manifest.json").read_text()) for artifact in manifest["artifacts"]: if artifact["path"] == "bootstrap/dever": artifact.update(bytes=len(content), sha256=hashlib.sha256(content).hexdigest()) (self.release / "manifest.json").write_text(json.dumps(manifest)) self.sign_manifest() entry = generator.render((ROOT / "skills/scripts/install.py").read_text(), self.public.read_text()) image = self.directory("image") arguments = ["/bin/bash", "-s", "--", "--release", str(self.release), "--system-root", str(image), "--version", "1.2.3", "--no-start"] completed = subprocess.run(arguments, input=entry, env={}, text=True, capture_output=True, timeout=20) self.assertEqual(completed.returncode, 0, completed.stderr) self.assertTrue((image / "called.json").is_file()) self.assertEqual(list(image.glob(".dever-install-*")), []) (image / "called.json").unlink() with (self.release / "manifest.json").open("a") as stream: stream.write(" ") rejected = subprocess.run(arguments, input=entry, env={}, text=True, capture_output=True, timeout=20) self.assertNotEqual(rejected.returncode, 0) self.assertIn("签名校验失败", rejected.stderr) self.assertEqual(list(image.iterdir()), []) if __name__ == "__main__": unittest.main()