diff --git a/backend/tests/test_delivery_io.py b/backend/tests/test_delivery_io.py index 73f3bae..eea3c4d 100644 --- a/backend/tests/test_delivery_io.py +++ b/backend/tests/test_delivery_io.py @@ -18,6 +18,8 @@ sys.path.insert(0, str(ROOT)) manifest = importlib.import_module("scripts.release.manifest") policy = importlib.import_module("scripts.release.policy") file_copy = importlib.import_module("scripts.release.file_copy") +factory_service = importlib.import_module("scripts.release.factory_service") +proxy_install = importlib.import_module("scripts.release.proxy_install") @pytest.fixture @@ -122,6 +124,41 @@ def test_failed_operation_cannot_keep_a_byte_proof(ordinary_bundle, monkeypatch) assert reads == first_reads * 2 +@pytest.mark.parametrize("operator", ["factory", "proxy"]) +def test_installer_refusal_reuses_only_its_own_complete_byte_proof( + ordinary_bundle, monkeypatch, tmp_path, operator +): + # Model the early operator guard with ordinary owned fixture bytes. No OS + # identity, installation, service or VM execution is being accepted here. + bundle, expected = ordinary_bundle + reads = observed_digests(monkeypatch) + manifest.verify(bundle, expected) + complete_read = list(reads) + reads.clear() + + def refuse(*args, **kwargs): + assert manifest.verify(bundle, expected) == manifest.verify(bundle, expected) + raise RuntimeError("ordinary installer refusal probe") + + module = factory_service if operator == "factory" else proxy_install + monkeypatch.setattr(module.os, "getuid", lambda: 0) + monkeypatch.setattr(module.os, "geteuid", lambda: 0) + monkeypatch.setattr(module.os, "getgid", lambda: 0) + monkeypatch.setattr(module.preflight, "protected", refuse) + source = tmp_path / "uncreated-installation-input" + arguments = [source, expected] + entry = module.install_unit if operator == "factory" else module.install_proxy + if operator == "proxy": + arguments.extend([tmp_path / "unread-certificate", tmp_path / "unread-key"]) + with pytest.raises(RuntimeError, match="ordinary installer refusal probe"): + entry(*arguments) + assert reads == complete_read + assert manifest._READ_SCOPE.get() is None + assert not source.exists() + manifest.verify(bundle, expected) + assert reads == complete_read * 2 + + @pytest.mark.parametrize("index", range(9)) @pytest.mark.parametrize("phase", ["reuse", "return"]) def test_bundle_identity_model_is_required_through_operation_return( diff --git a/scripts/release/factory_service.py b/scripts/release/factory_service.py index d505bc7..a5e4929 100644 --- a/scripts/release/factory_service.py +++ b/scripts/release/factory_service.py @@ -9,7 +9,7 @@ import stat import subprocess from pathlib import Path -from . import configuration, install, preflight, service_install, services +from . import configuration, install, manifest, preflight, service_install, services from .policy import DeliveryError, canonical, digest, fields, read_json, sha256 UNIT = "kf-sandbox-factory.service" @@ -167,6 +167,7 @@ def generate(output: Path) -> dict: payload = install.RELEASES / selected["release_id"] / "payload" for name in ( "factory_service.py", + "manifest.py", "services.py", "service_profiles.py", "layout.py", @@ -199,6 +200,7 @@ def generate(output: Path) -> dict: return result +@manifest.verified_operation def install_unit(source: Path, expected: str) -> dict: if os.getuid() != 0 or os.geteuid() != 0: raise DeliveryError("factory unit installation requires root operator") @@ -219,6 +221,7 @@ def install_unit(source: Path, expected: str) -> dict: payload = install.RELEASES / selected["release_id"] / "payload" for name in ( "factory_service.py", + "manifest.py", "services.py", "service_profiles.py", "layout.py", diff --git a/scripts/release/proxy_install.py b/scripts/release/proxy_install.py index e2024d4..a39b4fe 100644 --- a/scripts/release/proxy_install.py +++ b/scripts/release/proxy_install.py @@ -15,6 +15,7 @@ from . import ( configuration, install, journal, + manifest, preflight, proxy_runtime, proxy_service, @@ -308,6 +309,7 @@ def temporary_directories() -> None: os.chown(temporary_directory, proxy_service.UID, proxy_service.UID) +@manifest.verified_operation def install_proxy(source: Path, expected: str, certificate: Path, key: Path) -> dict: if os.getuid() != 0 or os.geteuid() != 0 or os.getgid() != 0: raise DeliveryError("root proxy installer required") @@ -325,6 +327,7 @@ def install_proxy(source: Path, expected: str, certificate: Path, key: Path) -> "proxy.py", "proxy_service.py", "proxy_install.py", + "manifest.py", "proxy_guard.py", "proxy_runtime.py", ):