"""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 from install_support import load_installer 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_installer() 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 release_record(self, directory, version="1.2.3"): return { "tag_name": f"v{version}", "draft": False, "prerelease": False, "assets": [{"name": path.name, "size": path.stat().st_size, "browser_download_url": f"https://git.dever.cc/shemic/dever/releases/download/v{version}/{path.name}"} for path in sorted(directory.iterdir())], } 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 = [] record = self.release_record(assets_path) with mock.patch.object(installer, "download", return_value=io.BytesIO(json.dumps(record).encode())): version, attachments = installer.resolve_release("latest") self.assertEqual(version, "1.2.3") def download(address): self.assertTrue(address.startswith( f"https://git.dever.cc/shemic/dever/releases/download/v1.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, self.platform, self.catalog, self.public, attachments) 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, self.platform, self.catalog, self.public, attachments) 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, 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, 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, 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_pins_latest_or_exact_release_before_downloading_matching_assets(self): assets_path = self.root / "assets" self.create_assets(assets_path) record = self.release_record(assets_path) for requested, endpoint in (("latest", "latest"), ("1.2.3", "tags/v1.2.3")): destination = self.directory(requested) addresses = [] def download(address): addresses.append(address) if address == f"{installer.RELEASES}/{endpoint}": return io.BytesIO(json.dumps(record).encode()) return (assets_path / address.rsplit("/", 1)[-1]).open("rb") with mock.patch.object(installer, "download", side_effect=download): selected, attachments = installer.resolve_release(requested) installer.metadata(destination, None, attachments, self.platform) authenticated, _, _ = installer.authenticate(destination, self.public, self.scratch, self.platform, selected) self.assertEqual(authenticated, "1.2.3") self.assertEqual(addresses, [f"{installer.RELEASES}/{endpoint}"] + [ f"https://git.dever.cc/shemic/dever/releases/download/v1.2.3/dever-{self.platform}.{suffix}" for suffix in ("manifest.json", "manifest.sig") ]) def test_release_metadata_rejects_unpublished_ambiguous_and_foreign_assets(self): directory = self.root / "assets" self.create_assets(directory) valid = self.release_record(directory) mutations = [ lambda record: record.update(draft=True), lambda record: record.update(prerelease=True), lambda record: record.update(tag_name="v01.2.3"), lambda record: record.update(tag_name="v1.2.4"), lambda record: record["assets"].append(record["assets"][0]), lambda record: record["assets"][0].update(size=True), lambda record: record["assets"][0].update(browser_download_url="https://evil.example/asset"), lambda record: record.update(assets=[]), lambda record: record.update(assets=[record["assets"][0]] * 513), ] for mutate in mutations: record = json.loads(json.dumps(valid)) mutate(record) with self.subTest(record=record["tag_name"]), \ mock.patch.object(installer, "download", return_value=io.BytesIO(json.dumps(record).encode())), \ self.assertRaises(ValueError): installer.resolve_release("1.2.3") with mock.patch.object(installer, "MAX_METADATA", 10), \ mock.patch.object(installer, "download", return_value=io.BytesIO(b" " * 11)), \ self.assertRaisesRegex(ValueError, "大小限制"): installer.resolve_release("latest") def test_latest_tag_must_match_signed_manifest_and_required_assets_must_exist(self): directory = self.root / "assets" self.create_assets(directory) record = self.release_record(directory, "1.2.4") with mock.patch.object(installer, "download", return_value=io.BytesIO(json.dumps(record).encode())): selected, attachments = installer.resolve_release("latest") with self.assertRaisesRegex(ValueError, "版本与请求不匹配"): installer.authenticate(self.release, self.public, self.scratch, self.platform, selected) with self.assertRaisesRegex(ValueError, "缺少附件"): installer.asset(attachments, self.platform, "missing.tar.zst") with self.assertRaisesRegex(ValueError, "大小限制"): installer.asset(attachments, self.platform, "manifest.json", 1) def test_metadata_rejects_truncated_api_attachment(self): name = f"dever-{self.platform}.manifest.json" attachments = {name: {"address": "https://git.dever.cc/asset", "bytes": 20}} with mock.patch.object(installer, "download", return_value=io.BytesIO(b"short")), \ self.assertRaisesRegex(ValueError, "长度不匹配"): installer.metadata(self.directory("truncated"), None, attachments, self.platform) def test_redirects_remain_on_the_official_https_site(self): handler = installer.OfficialRedirect() current = "https://git.dever.cc/shemic/dever/releases/download/v1.2.3/a" request = installer.urllib.request.Request(current) allowed = "https://git.dever.cc/attachments/fixture" 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_requires_trusted_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. 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 = (ROOT / "install").read_text() image = self.directory("image") arguments = ["/bin/bash", "-s", "--", "--release", str(self.release), "--system-root", str(image), "--version", "1.2.3", "--no-start"] untrusted = subprocess.run(arguments, input=entry, env={}, text=True, capture_output=True, timeout=20) self.assertNotEqual(untrusted.returncode, 0) self.assertIn("签名校验失败", untrusted.stderr) self.assertEqual(list(image.iterdir()), []) arguments.extend(["--trusted-key", str(self.public)]) 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()