"""Private installer and import-boundary checks with synthetic material only.""" import importlib.util import io import json import os import shutil import subprocess import sys import tarfile from pathlib import Path import pytest ROOT = Path(__file__).resolve().parents[1] PACKAGING = ROOT / "plugins/xgrids-k1/packaging" SYNTHETIC_KEY = b"11111111-2222-3333-4444-555555555555" @pytest.fixture def modules(monkeypatch): monkeypatch.syspath_prepend(str(PACKAGING)) from importlib import import_module credential = import_module("credential_install") runtime = import_module("runtime_payload") spec = importlib.util.spec_from_file_location( "k1_package_builder_test", PACKAGING / "build_deb.py" ) builder = importlib.util.module_from_spec(spec) spec.loader.exec_module(builder) return credential, runtime, builder @pytest.fixture def root_metadata(monkeypatch, tmp_path, modules): credential, _, _ = modules original_stat = Path.stat original_fstat = os.fstat def owned(info): values = list(info) values[4] = 0 return os.stat_result(values) def file_stat(path, *args, **kwargs): info = original_stat(path, *args, **kwargs) return owned(info) if path.is_relative_to(tmp_path) else info monkeypatch.setattr(Path, "stat", file_stat) monkeypatch.setattr(credential.os, "fstat", lambda fd: owned(original_fstat(fd))) monkeypatch.setattr(credential.os, "geteuid", lambda: 0) return credential def test_private_material_is_encrypted_on_each_host_and_reinstall_is_idempotent( root_metadata, tmp_path, ): credential = root_metadata calls = [] def runner(args, **kwargs): calls.append(args) assert SYNTHETIC_KEY.decode() not in str(args) if args[1] == "encrypt": assert kwargs["input"] == SYNTHETIC_KEY Path(args[-1]).write_bytes(b"synthetic-host-encrypted-material") return subprocess.CompletedProcess(args, 0) return subprocess.CompletedProcess(args, 0, stdout=SYNTHETIC_KEY) store = tmp_path / "credentials" assert credential.install(bytearray(SYNTHETIC_KEY), root=store, runner=runner) == "installed" before = (store / "k1-application").stat().st_ino assert credential.install(bytearray(SYNTHETIC_KEY), root=store, runner=runner) == "unchanged" assert (store / "k1-application").stat().st_ino == before assert (store / "k1-application").stat().st_mode & 0o777 == 0o600 assert [args[1] for args in calls] == ["encrypt", "decrypt"] def test_different_material_never_rotates_an_existing_installation(root_metadata, tmp_path): store = tmp_path / "credentials" store.mkdir(mode=0o700) target = store / "k1-application" target.write_bytes(b"old-encrypted-credential") target.chmod(0o600) def runner(args, **kwargs): assert args[1] == "decrypt" return subprocess.CompletedProcess(args, 0, stdout=b"a-different-private-application-value") with pytest.raises(ValueError, match="explicit rotation") as error: root_metadata.install(bytearray(SYNTHETIC_KEY), root=store, runner=runner) assert SYNTHETIC_KEY.decode() not in str(error.value) assert target.read_bytes() == b"old-encrypted-credential" @pytest.mark.parametrize("bad", [b"", b"short", b"x" * 1025, b"\x00" * 36, b" " * 36]) def test_bad_material_fails_before_system_tools(modules, tmp_path, bad): credential, _, _ = modules def forbidden(*args, **kwargs): pytest.fail("Invalid material reached an operating-system command") with pytest.raises(ValueError): credential.install(bytearray(bad), root=tmp_path / "absent", runner=forbidden) assert not (tmp_path / "absent").exists() def test_bundle_rejects_symlink_and_public_permissions(root_metadata, tmp_path): target = tmp_path / "profile" target.write_bytes(SYNTHETIC_KEY) target.chmod(0o644) with pytest.raises(ValueError): root_metadata.read_bundle(target) target.chmod(0o600) link = tmp_path / "link" link.symlink_to(target) with pytest.raises(OSError): root_metadata.read_bundle(link) assert root_metadata.read_bundle(target) == SYNTHETIC_KEY def test_private_release_contains_material_only_in_root_private_member( modules, monkeypatch, tmp_path ): _, _, builder = modules packaging = tmp_path / "packaging" shutil.copytree(PACKAGING, packaging, ignore=shutil.ignore_patterns("__pycache__")) (packaging / "k1-bundle.json").write_text('{"wheels": []}') monkeypatch.setattr(builder, "PACKAGING", packaging) public = builder.payload(tmp_path) private = builder.payload(tmp_path, authority=bytearray(SYNTHETIC_KEY)) assert all(SYNTHETIC_KEY not in data for _, data, _ in public) secret_members = [(name, mode) for name, data, mode in private if SYNTHETIC_KEY in data] assert secret_members == [("usr/share/mission-core-node/k1/private-application-key", 0o600)] provenance = next(data for name, data, _ in private if name.endswith("provenance.json")) metadata = json.loads(provenance) assert metadata["application_material_included"] is True assert not any("private-application-key" in key for key in metadata["files"]) assert metadata["acceptance_scope"] == "owner-controlled-k1-fw-3.0.2-ubuntu-24.04-amd64" destination = tmp_path / "private.deb" builder.build(destination, tmp_path, authority=bytearray(SYNTHETIC_KEY)) assert destination.stat().st_mode & 0o777 == 0o600 original = destination.read_bytes() with pytest.raises(FileExistsError): builder.build(destination, tmp_path, authority=bytearray(SYNTHETIC_KEY)) assert destination.read_bytes() == original position = 8 members = {} while position < len(original): header = original[position : position + 60] length = int(header[48:58]) members[header[:16].decode().strip().rstrip("/")] = original[ position + 60 : position + 60 + length ] position += 60 + length + length % 2 with tarfile.open(fileobj=io.BytesIO(members["control.tar.gz"]), mode="r:gz") as archive: control = archive.extractfile("control").read().decode() assert "Depends: mission-core-node (>= 0.8.10)" in control assert "Replaces: mission-core-node (<< 0.8.0)" in control def test_onboard_imports_run_from_declared_payload_without_core_checkout(modules, tmp_path): _, runtime, builder = modules stage = tmp_path / "installed" for path in runtime.files(): target = stage / path.relative_to(ROOT) target.parent.mkdir(parents=True, exist_ok=True) shutil.copyfile(path, target) for relative in builder.RESOURCES: target = stage / relative target.parent.mkdir(parents=True, exist_ok=True) shutil.copyfile(ROOT / relative, target) script = """ import asyncio, pathlib, sys stage = pathlib.Path(sys.argv[1]) sys.path.insert(0, str(stage / "src")) from k1link.device_plugins.xgrids_k1.node_bridge import NodeBridge assert "k1link.laboratory.execution" not in sys.modules assert "k1link.device_plugins.xgrids_k1.legacy_api" not in sys.modules assert "k1link.compute.jobs" not in sys.modules import k1link assert pathlib.Path(k1link.__file__).is_relative_to(stage) bridge = NodeBridge(stage) async def check(): try: value = await bridge.state() assert value["phase"] == "idle" assert value["candidates"] == [] for name, module in list(sys.modules.items()): if name.startswith("k1link.") and getattr(module, "__file__", None): assert pathlib.Path(module.__file__).is_relative_to(stage), name finally: bridge.service.close() asyncio.run(check()) """ env = dict(os.environ, MISSIONCORE_DATA_DIR=str(tmp_path / "state")) result = subprocess.run( [sys.executable, "-I", "-c", script, str(stage)], cwd=tmp_path, env=env, capture_output=True, text=True, timeout=40, ) assert result.returncode == 0, result.stderr def test_lab_lazy_exports_keep_the_public_objects_and_do_not_load_on_leaf_import(tmp_path): script = """ import importlib, sys import k1link.laboratory as lab from k1link.laboratory.canonical_recorded_migration import matches_historical_recorded_projection assert "k1link.laboratory.execution" not in sys.modules assert set(lab.__all__) == set(lab._EXPORTS) for name in lab.__all__: assert getattr(lab, name) is getattr(importlib.import_module(lab._EXPORTS[name]), name) """ result = subprocess.run( [sys.executable, "-c", script], cwd=tmp_path, capture_output=True, text=True, timeout=30 ) assert result.returncode == 0, result.stderr