395 lines
21 KiB
Python
395 lines
21 KiB
Python
import hashlib
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import struct
|
|
import tarfile
|
|
import tempfile
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
import package_linux as pack
|
|
from embeddinggemma2 import runtime_provenance as provenance
|
|
import stage_extraction_worker as extraction
|
|
import stage_speech_worker as speech
|
|
|
|
|
|
def elf(path, machine=62):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
data = bytearray(64)
|
|
data[:7] = b"\x7fELF\x02\x01\x01"
|
|
struct.pack_into("<HH", data, 16, 3, machine)
|
|
path.write_bytes(data)
|
|
|
|
|
|
class LinuxPackageTests(unittest.TestCase):
|
|
def speech_fixture(self, fixture):
|
|
root = fixture[0]
|
|
runtime = root / 'speech-runtime'
|
|
runtime.mkdir()
|
|
for name in speech.INPUTS:
|
|
path = speech.input_path(root, name)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(speech.input_path(speech.ROOT, name).read_bytes())
|
|
files = []
|
|
for name in ('searchi-speech-worker', 'libwhisper.so.1', 'libggml-cpu-x64.so', 'libggml-vulkan.so'):
|
|
path = runtime / name
|
|
elf(path)
|
|
files.append({'path': name, 'bytes': path.stat().st_size, 'sha256': pack.digest(path)})
|
|
(runtime / 'runtime-manifest.json').write_text(json.dumps({'schema': 1,
|
|
'upstream_commit': speech.UPSTREAM, 'device': 'auto', 'distribution': True,
|
|
'source_hashes': {name: pack.digest(speech.input_path(root, name)) for name in speech.INPUTS},
|
|
'files': files}))
|
|
for name in speech.NOTICES:
|
|
notice = root / 'assets/licenses' / name
|
|
notice.parent.mkdir(parents=True, exist_ok=True)
|
|
notice.write_text('Approved test notice ' + name)
|
|
(runtime / name).write_bytes(notice.read_bytes())
|
|
return runtime
|
|
|
|
def test_speech_archive_has_isolated_verified_files_modes_and_instructions(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
fixture = self.fixture(directory)
|
|
runtime = self.speech_fixture(fixture)
|
|
with mock.patch.object(pack, 'dynamic', side_effect=self.metadata), \
|
|
mock.patch.object(speech, 'verify_linux_closure') as closure:
|
|
pack.package(*fixture, speech_runtime=runtime)
|
|
closure.assert_called_once_with(runtime)
|
|
first = fixture[-1].read_bytes()
|
|
pack.package(*fixture, speech_runtime=runtime)
|
|
self.assertEqual(first, fixture[-1].read_bytes())
|
|
with tarfile.open(fixture[-1]) as archive:
|
|
manifest = json.load(archive.extractfile('PACKAGE.json'))
|
|
self.assertTrue(manifest['features']['local_speech'])
|
|
self.assertEqual(manifest['features']['speech_status'], 'experimental')
|
|
self.assertEqual(archive.getmember('speech-runtime/searchi-speech-worker').mode, 0o755)
|
|
for name in (*speech.NOTICES, 'runtime-manifest.json'):
|
|
self.assertEqual(archive.getmember('speech-runtime/' + name).mode, 0o644)
|
|
for record in manifest['files']:
|
|
data = archive.extractfile(record['path']).read()
|
|
self.assertEqual(record['sha256'], hashlib.sha256(data).hexdigest())
|
|
instructions = archive.extractfile('START-HERE.txt').read().decode()
|
|
self.assertIn('setup --speech --model-dir ./speech-models', instructions)
|
|
self.assertIn('--speech-worker ./speech-runtime/searchi-speech-worker', instructions)
|
|
self.assertIn('--speech-model-dir ./speech-models', instructions)
|
|
|
|
def test_invalid_speech_runtime_preserves_previous_archive(self):
|
|
for mutation in ('cpu', 'tampered', 'notice', 'closure'):
|
|
with self.subTest(mutation=mutation), tempfile.TemporaryDirectory() as directory:
|
|
fixture = self.fixture(directory)
|
|
runtime = self.speech_fixture(fixture)
|
|
fixture[-1].write_bytes(b'previous archive')
|
|
if mutation == 'cpu':
|
|
path = runtime / 'runtime-manifest.json'
|
|
manifest = json.loads(path.read_text())
|
|
manifest.update(device='cpu', distribution=False)
|
|
path.write_text(json.dumps(manifest))
|
|
elif mutation == 'tampered':
|
|
(runtime / 'libwhisper.so.1').write_bytes(b'tampered')
|
|
elif mutation == 'notice':
|
|
(runtime / speech.NOTICES[0]).unlink()
|
|
with mock.patch.object(pack, 'dynamic', side_effect=self.metadata), \
|
|
mock.patch.object(speech, 'verify_linux_closure',
|
|
side_effect=ValueError('Invalid ELF closure') if mutation == 'closure' else None):
|
|
with self.assertRaises(ValueError):
|
|
pack.package(*fixture, speech_runtime=runtime)
|
|
self.assertEqual(fixture[-1].read_bytes(), b'previous archive')
|
|
self.assertEqual(list(fixture[-1].parent.glob('*.partial')), [])
|
|
|
|
def test_extraction_missing_stale_tampered_or_wrong_architecture_preserves_output(self):
|
|
for mutation in ("missing", "inventory_missing", "stale", "tampered", "architecture"):
|
|
with self.subTest(mutation=mutation), tempfile.TemporaryDirectory() as directory:
|
|
fixture = self.fixture(directory)
|
|
root, _, runtime, _, output = fixture
|
|
output.write_bytes(b"previous archive")
|
|
helper = runtime / "searchi-extract-worker"
|
|
if mutation == "missing":
|
|
helper.unlink()
|
|
elif mutation == "inventory_missing":
|
|
(runtime / "extraction-worker.json").unlink()
|
|
elif mutation == "stale":
|
|
source = root / "crates/searchi-extract-worker/src/main.rs"
|
|
source.write_bytes(source.read_bytes() + b"\n// changed\n")
|
|
elif mutation == "tampered":
|
|
helper.write_bytes(helper.read_bytes() + b"changed")
|
|
else:
|
|
elf(helper, machine=183)
|
|
inventory = runtime / "extraction-worker.json"
|
|
manifest = json.loads(inventory.read_text())
|
|
manifest["file"] = extraction.identity(helper)[2]
|
|
inventory.write_text(json.dumps(manifest))
|
|
with self.assertRaises(ValueError):
|
|
self.run_package(fixture)
|
|
self.assertEqual(output.read_bytes(), b"previous archive")
|
|
self.assertEqual(list(output.parent.glob("*.partial")), [])
|
|
|
|
def test_extraction_dependency_closure_and_search_paths_preserve_output(self):
|
|
for metadata in ({"NEEDED": ["libunpackaged.so.1"]},
|
|
{"NEEDED": ["libvulkan.so.1"]},
|
|
{"RUNPATH": ["$ORIGIN"]}, {"RPATH": ["/build/lib"]}):
|
|
with self.subTest(metadata=metadata), tempfile.TemporaryDirectory() as directory:
|
|
fixture = self.fixture(directory)
|
|
fixture[-1].write_bytes(b"previous archive")
|
|
def dependencies(path):
|
|
return metadata if path.name == "searchi-extract-worker" else self.metadata(path)
|
|
with mock.patch.object(pack, "dynamic", side_effect=dependencies), self.assertRaisesRegex(ValueError, "Extraction helper"):
|
|
pack.package(*fixture)
|
|
self.assertEqual(fixture[-1].read_bytes(), b"previous archive")
|
|
self.assertEqual(list(fixture[-1].parent.glob("*.partial")), [])
|
|
|
|
def fixture(self, directory):
|
|
root = Path(directory)
|
|
for name in provenance.INPUTS:
|
|
path = root / name
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes((provenance.ROOT / name).read_bytes())
|
|
for name in extraction.fingerprint()["inputs"]:
|
|
path = root / name
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes((extraction.ROOT / name).read_bytes())
|
|
runtime, media = root / "runtime", root / "media"
|
|
executable = root / "build/searchi"
|
|
elf(executable)
|
|
(root / "LICENSE").write_text("application MIT notice")
|
|
license = root / ".searchi/references/llama.cpp/LICENSE"
|
|
license.parent.mkdir(parents=True)
|
|
license.write_text("llama MIT notice")
|
|
names = ["libggml.so.0", "libggml-base.so.0", "libllama.so.0", "libmtmd.so.0", "libggml-cpu-x64.so", "libggml-vulkan.so"]
|
|
elf(runtime / "searchi-gemma-worker")
|
|
helper = root / "build/searchi-extract-worker"
|
|
elf(helper)
|
|
extraction.stage(helper, extraction.fingerprint(root), runtime, root)
|
|
records = []
|
|
for name in names:
|
|
elf(runtime / name)
|
|
records.append({"path": name, "bytes": 64, "sha256": pack.digest(runtime / name)})
|
|
(runtime / "runtime-libraries.json").write_text(json.dumps({"platform": "linux", "worker_sha256": pack.digest(runtime / "searchi-gemma-worker"), "build_provenance": provenance.fingerprint(root), "libraries": records}))
|
|
media.mkdir()
|
|
elf(media / "bin/ffmpeg")
|
|
elf(media / "bin/ffprobe")
|
|
notice = media / "licenses/upstream/LICENSE.txt"
|
|
notice.parent.mkdir(parents=True)
|
|
notice.write_text("LGPL notice")
|
|
archive = media / "verified.tar.xz"
|
|
with tarfile.open(archive, "w:xz") as output:
|
|
for name, path in [("upstream/bin/ffmpeg", media / "bin/ffmpeg"), ("upstream/bin/ffprobe", media / "bin/ffprobe"), ("upstream/LICENSE.txt", notice)]:
|
|
output.add(path, arcname=name)
|
|
manifest = root / "tools/media-runtime/manifest.json"
|
|
manifest.parent.mkdir(parents=True)
|
|
manifest.write_text(json.dumps({"provider": "test-fixture", "release": "fixture", "platforms": {"linux-x86_64": {"archive": archive.name, "bytes": archive.stat().st_size, "sha256": pack.digest(archive)}}}))
|
|
return root, executable, runtime, media, root / "candidate.tar.gz"
|
|
|
|
@staticmethod
|
|
def metadata(path):
|
|
if path.name == "searchi-gemma-worker" or path.name.startswith("lib"):
|
|
metadata = {"RUNPATH": ["$ORIGIN"], "NEEDED": ["libc.so.6"]}
|
|
if path.name in {"libggml.so.0", "libggml-base.so.0", "libllama.so.0", "libmtmd.so.0"}:
|
|
metadata["SONAME"] = [path.name]
|
|
if path.name == "libggml-vulkan.so":
|
|
metadata["NEEDED"].append("libvulkan.so.1")
|
|
return metadata
|
|
return {"NEEDED": ["libc.so.6"]}
|
|
|
|
def run_package(self, fixture):
|
|
with mock.patch.object(pack, "dynamic", side_effect=self.metadata):
|
|
return pack.package(*fixture)
|
|
|
|
def test_archive_integrity_modes_no_userdata_and_reproducibility(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
fixture = self.fixture(directory)
|
|
root, executable, runtime, media, output = fixture
|
|
(runtime / "settings.json").write_text("private")
|
|
(media / "index.sqlite").write_text("private")
|
|
self.run_package(fixture)
|
|
first = output.read_bytes()
|
|
self.run_package(fixture)
|
|
self.assertEqual(first, output.read_bytes())
|
|
with tarfile.open(output) as archive:
|
|
self.assertNotIn("settings.json", archive.getnames())
|
|
manifest = json.load(archive.extractfile("PACKAGE.json"))
|
|
for record in manifest["files"]:
|
|
member = archive.getmember(record["path"])
|
|
data = archive.extractfile(member).read()
|
|
self.assertEqual(record["sha256"], hashlib.sha256(data).hexdigest())
|
|
self.assertEqual(member.uid, 0)
|
|
self.assertEqual(member.mtime, 0)
|
|
self.assertEqual(archive.getmember("searchi").mode, 0o755)
|
|
self.assertEqual(archive.getmember("searchi-extract-worker").mode, 0o755)
|
|
self.assertEqual(archive.getmember("extraction-worker.json").mode, 0o644)
|
|
self.assertEqual(archive.getmember("media-runtime/bin/ffmpeg").mode, 0o755)
|
|
self.assertEqual(archive.getmember("LICENSE.txt").mode, 0o644)
|
|
|
|
def test_tampered_runtime_preserves_previous_archive(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
fixture = self.fixture(directory)
|
|
self.run_package(fixture)
|
|
previous = fixture[-1].read_bytes()
|
|
(fixture[2] / "libggml-cpu-x64.so").write_bytes(b"tampered")
|
|
with self.assertRaisesRegex(ValueError, "integrity mismatch"):
|
|
self.run_package(fixture)
|
|
self.assertEqual(previous, fixture[-1].read_bytes())
|
|
|
|
def test_xz_preserves_members_checksums_modes_and_reproducibility(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
fixture = self.fixture(directory)
|
|
self.run_package(fixture)
|
|
with tarfile.open(fixture[-1]) as archive:
|
|
original = {m.name: (archive.extractfile(m).read(), m.mode) for m in archive}
|
|
with mock.patch.object(pack, "dynamic", side_effect=self.metadata):
|
|
pack.package(*fixture, compression="xz")
|
|
first = fixture[-1].read_bytes()
|
|
pack.package(*fixture, compression="xz")
|
|
self.assertEqual(first, fixture[-1].read_bytes())
|
|
self.assertTrue(first.startswith(b"\xfd7zXZ\x00"))
|
|
with tarfile.open(fixture[-1]) as archive:
|
|
compressed = {m.name: (archive.extractfile(m).read(), m.mode) for m in archive}
|
|
self.assertEqual(original, compressed)
|
|
|
|
def test_old_or_stale_build_provenance_preserves_previous_archive(self):
|
|
for mutation in ("missing", "source_changed"):
|
|
with self.subTest(mutation=mutation), tempfile.TemporaryDirectory() as directory:
|
|
fixture = self.fixture(directory)
|
|
fixture[-1].write_bytes(b"previous archive")
|
|
if mutation == "missing":
|
|
inventory = fixture[2] / "runtime-libraries.json"
|
|
manifest = json.loads(inventory.read_text())
|
|
del manifest["build_provenance"]
|
|
inventory.write_text(json.dumps(manifest))
|
|
else:
|
|
(fixture[0] / provenance.INPUTS[0]).write_bytes(b"changed worker source")
|
|
with self.assertRaisesRegex(ValueError, "provenance"):
|
|
self.run_package(fixture)
|
|
self.assertEqual(fixture[-1].read_bytes(), b"previous archive")
|
|
|
|
def test_media_helper_tamper_is_rejected_against_pinned_archive(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
fixture = self.fixture(directory)
|
|
(fixture[3] / "bin/ffmpeg").write_bytes(b"tampered")
|
|
with self.assertRaisesRegex(ValueError, "differs from pinned archive"):
|
|
self.run_package(fixture)
|
|
|
|
def test_media_archive_provenance_tamper_rejected(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
fixture = self.fixture(directory)
|
|
with (fixture[3] / "verified.tar.xz").open("ab") as stream:
|
|
stream.write(b"tampered")
|
|
with self.assertRaisesRegex(ValueError, "provenance mismatch"):
|
|
self.run_package(fixture)
|
|
|
|
def test_mixed_architecture_rejected(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
fixture = self.fixture(directory)
|
|
elf(fixture[1], machine=183)
|
|
with self.assertRaisesRegex(ValueError, "x86-64 ELF"):
|
|
self.run_package(fixture)
|
|
|
|
def test_inventory_traversal_rejected(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
fixture = self.fixture(directory)
|
|
path = fixture[2] / "runtime-libraries.json"
|
|
manifest = json.loads(path.read_text())
|
|
manifest["libraries"][0]["path"] = "../private.so"
|
|
path.write_text(json.dumps(manifest))
|
|
with self.assertRaisesRegex(ValueError, "Unsafe package path"):
|
|
self.run_package(fixture)
|
|
|
|
def test_missing_baseline_cpu_rejected(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
fixture = self.fixture(directory)
|
|
path = fixture[2] / "runtime-libraries.json"
|
|
manifest = json.loads(path.read_text())
|
|
manifest["libraries"] = [r for r in manifest["libraries"] if r["path"] != "libggml-cpu-x64.so"]
|
|
path.write_text(json.dumps(manifest))
|
|
with self.assertRaisesRegex(ValueError, "lacks baseline"):
|
|
self.run_package(fixture)
|
|
|
|
def test_worker_cannot_link_vulkan_loader(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
fixture = self.fixture(directory)
|
|
def metadata(path):
|
|
result = self.metadata(path)
|
|
if path.name == "searchi-gemma-worker":
|
|
result["NEEDED"].append("libvulkan.so.1")
|
|
return result
|
|
with mock.patch.object(pack, "dynamic", side_effect=metadata), self.assertRaisesRegex(ValueError, "Unpackaged runtime"):
|
|
pack.package(*fixture)
|
|
|
|
def test_external_runtime_search_path_rejected(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
fixture = self.fixture(directory)
|
|
with mock.patch.object(pack, "dynamic", return_value={"RUNPATH": ["/build/lib"]}), self.assertRaisesRegex(ValueError, "origin-only"):
|
|
pack.package(*fixture)
|
|
|
|
def test_glibc_vector_math_is_declared_but_unknown_cli_dependency_rejected(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
fixture = self.fixture(directory)
|
|
def vector_math(path):
|
|
result = self.metadata(path)
|
|
if path.name == "searchi":
|
|
result["NEEDED"].append("libmvec.so.1")
|
|
return result
|
|
with mock.patch.object(pack, "dynamic", side_effect=vector_math):
|
|
pack.package(*fixture)
|
|
with tarfile.open(fixture[-1]) as archive:
|
|
manifest = json.load(archive.extractfile("PACKAGE.json"))
|
|
self.assertIn("libmvec.so.1", manifest["system_libraries"])
|
|
previous = fixture[-1].read_bytes()
|
|
def unknown(path):
|
|
result = vector_math(path)
|
|
if path.name == "searchi":
|
|
result["NEEDED"].append("libunpackaged.so.1")
|
|
return result
|
|
with mock.patch.object(pack, "dynamic", side_effect=unknown), self.assertRaisesRegex(ValueError, "unexpected native dependency"):
|
|
pack.package(*fixture)
|
|
self.assertEqual(previous, fixture[-1].read_bytes())
|
|
|
|
def test_readelf_parser_is_nonexecuting_and_preserves_dependencies(self):
|
|
with mock.patch.object(pack.subprocess, "check_output", return_value="(NEEDED) Shared library: [libc.so.6]\n(RUNPATH) Library runpath: [$ORIGIN]\n") as process:
|
|
self.assertEqual(pack.dynamic(Path("candidate")), {"NEEDED": ["libc.so.6"], "RUNPATH": ["$ORIGIN"]})
|
|
self.assertEqual(process.call_args.args[0], ["readelf", "--dynamic", "candidate"])
|
|
|
|
def test_missing_pinned_media_notice_rejected(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
fixture = self.fixture(directory)
|
|
(fixture[3] / "licenses/upstream/LICENSE.txt").unlink()
|
|
with self.assertRaisesRegex(ValueError, "Missing or oversized"):
|
|
self.run_package(fixture)
|
|
|
|
def test_output_cannot_overwrite_cli(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root, executable, runtime, media, _ = self.fixture(directory)
|
|
original = executable.read_bytes()
|
|
with self.assertRaisesRegex(ValueError, "overwrite an input"):
|
|
self.run_package((root, executable, runtime, media, executable))
|
|
self.assertEqual(original, executable.read_bytes())
|
|
|
|
def test_changed_source_during_packaging_preserves_previous_archive(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
fixture = self.fixture(directory)
|
|
self.run_package(fixture)
|
|
previous = fixture[-1].read_bytes()
|
|
original_add = pack.add
|
|
def mutate(archive, name, data, mode):
|
|
original_add(archive, name, data, mode)
|
|
if name == "LICENSE.txt":
|
|
fixture[1].write_bytes(b"changed after preflight")
|
|
with mock.patch.object(pack, "add", side_effect=mutate), self.assertRaisesRegex(ValueError, "Source changed"):
|
|
self.run_package(fixture)
|
|
self.assertEqual(previous, fixture[-1].read_bytes())
|
|
|
|
def test_unapproved_plugin_rejected_even_with_valid_hash(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
fixture = self.fixture(directory)
|
|
runtime = fixture[2]
|
|
elf(runtime / "libarbitrary.so")
|
|
inventory = runtime / "runtime-libraries.json"
|
|
manifest = json.loads(inventory.read_text())
|
|
manifest["libraries"].append({"path": "libarbitrary.so", "bytes": 64, "sha256": pack.digest(runtime / "libarbitrary.so")})
|
|
inventory.write_text(json.dumps(manifest))
|
|
with self.assertRaisesRegex(ValueError, "Unapproved runtime library"):
|
|
self.run_package(fixture)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|