508 lines
26 KiB
Python
508 lines
26 KiB
Python
"""Offline install/asset contracts; bootstrap is mocked, never installed globally."""
|
|
|
|
import argparse
|
|
import contextlib
|
|
import hashlib
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tarfile
|
|
import tempfile
|
|
import unittest
|
|
import zipfile
|
|
from unittest import mock
|
|
|
|
from install_support import load_installer
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
ZSTD = Path(os.environ.get("DEVER_TEST_ZSTD", "/usr/bin/zstd"))
|
|
|
|
|
|
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, ZSTD, complete=False)
|
|
|
|
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([str(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):
|
|
base_only = self.root / "base-only"
|
|
self.create_assets(base_only)
|
|
manifest_path = self.release / "manifest.json"
|
|
manifest = json.loads(manifest_path.read_text())
|
|
expected = {}
|
|
for ecosystem in ("pip", "npm", "go"):
|
|
name = f"runtime/{ecosystem}/{self.platform}/runtime.pack"
|
|
path = self.release / name
|
|
path.parent.mkdir(parents=True)
|
|
contents = f"optional {ecosystem} bytes".encode()
|
|
path.write_bytes(contents)
|
|
entry = {"path": name, "bytes": len(contents), "sha256": hashlib.sha256(contents).hexdigest()}
|
|
manifest["extensions"].append({"kind": {"type": "runtime", "ecosystem": ecosystem},
|
|
"target": self.platform, "artifacts": [entry]})
|
|
expected[ecosystem] = {name: contents}
|
|
manifest_path.write_text(json.dumps(manifest))
|
|
self.sign_manifest()
|
|
self.authenticate()
|
|
output = self.root / "split"
|
|
self.create_assets(output)
|
|
base_name = f"dever-{self.platform}.tar.zst"
|
|
self.assertEqual(self.archive_payloads(output / base_name), self.payloads)
|
|
self.assertEqual((output / base_name).read_bytes(), (base_only / base_name).read_bytes())
|
|
for ecosystem, payloads in expected.items():
|
|
extension = output / f"dever-{self.platform}.ext-runtime-{ecosystem}-{self.platform}.tar.zst"
|
|
self.assertEqual(self.archive_payloads(extension), payloads)
|
|
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_complete_release_requires_all_extensions_and_matching_editor(self):
|
|
manifest = json.loads((self.release / "manifest.json").read_text())
|
|
with self.assertRaisesRegex(ValueError, "complete release is missing"):
|
|
assets.create(self.release, self.root / "incomplete", ZSTD)
|
|
self.assertFalse((self.root / "incomplete").exists())
|
|
with self.assertRaisesRegex(ValueError, "complete release is missing"):
|
|
assets.verify_complete(manifest, None)
|
|
for target in ("linux-x86_64", "linux-aarch64"):
|
|
for ecosystem in ("pip", "npm", "go"):
|
|
manifest["extensions"].append({"kind": {"type": "runtime", "ecosystem": ecosystem},
|
|
"target": target, "artifacts": []})
|
|
for ecosystem in ("pip", "npm"):
|
|
manifest["extensions"].append({"kind": {"type": "build", "ecosystem": ecosystem},
|
|
"target": self.platform, "artifacts": []})
|
|
manifest["extensions"].append({"kind": {"type": "target"}, "target": "linux-aarch64",
|
|
"artifacts": []})
|
|
with self.assertRaisesRegex(ValueError, "editor VSIX"):
|
|
assets.verify_complete(manifest, None)
|
|
editor = self.root / "editor.vsix"
|
|
package = {"name": "dever-language", "publisher": "shemic", "version": manifest["version"]}
|
|
with zipfile.ZipFile(editor, "w") as archive:
|
|
archive.writestr("extension/package.json", json.dumps(package))
|
|
assets.verify_complete(manifest, editor)
|
|
for index in range(len(manifest["extensions"])):
|
|
incomplete = {**manifest, "extensions": manifest["extensions"][:index] + manifest["extensions"][index + 1:]}
|
|
with self.assertRaisesRegex(ValueError, "complete release is missing"):
|
|
assets.verify_complete(incomplete, editor)
|
|
package["version"] = "9.9.9"
|
|
with zipfile.ZipFile(editor, "w") as archive:
|
|
archive.writestr("extension/package.json", json.dumps(package))
|
|
with self.assertRaisesRegex(ValueError, "identity/version"):
|
|
assets.verify_complete(manifest, editor)
|
|
|
|
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()
|