from __future__ import annotations import copy import hashlib import importlib.util import json import os import sys from pathlib import Path import pytest ROOT = Path(__file__).resolve().parents[2] STORE_PATH = ROOT / "ops" / "gitea-deploy-broker" / "mcp_hub_release_store.py" def _load_store(): spec = importlib.util.spec_from_file_location("mcp_hub_release_store_test", STORE_PATH) assert spec and spec.loader module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) return module @pytest.fixture def store(): return _load_store() def _volume(store, **overrides): value = { "Name": store.VOLUME_NAME, "Driver": "local", "Scope": "local", "Mountpoint": store.VOLUME_HOST_ROOT.as_posix(), "Options": None, "Labels": dict(store.EXPECTED_LABELS), } return {**value, **overrides} def test_fixed_volume_contract_rejects_unexpected_options_labels_and_root(store): assert store.validate_volume_metadata(_volume(store), "/var/lib/docker") is None bad_documents = [ _volume(store, Name="other-volume"), _volume(store, Driver="local-persist"), _volume(store, Scope="global"), _volume(store, Mountpoint="/srv/example/appdata/releases"), _volume(store, Options={"o": "bind", "device": "/tmp"}), _volume(store, Labels={**store.EXPECTED_LABELS, "attacker": "extra"}), _volume(store, Labels={"itworx.managed-by": "deployment-controller"}), ] for document in bad_documents: with pytest.raises(store.ReleaseStoreError): store.validate_volume_metadata(document, "/var/lib/docker") with pytest.raises(store.ReleaseStoreError): store.validate_volume_metadata(_volume(store), "/var/lib/docker-alt") def test_release_store_fails_closed_when_acl_metadata_cannot_be_inspected(store, monkeypatch): def unavailable(_fd): raise OSError("filesystem ACL metadata is unavailable") monkeypatch.setattr(store.os, "listxattr", unavailable, raising=False) with pytest.raises(store.ReleaseStoreError, match="cannot inspect filesystem ACLs"): store._require_no_acl(1) @pytest.mark.skipif( os.name == "nt" or not hasattr(os, "geteuid") or os.geteuid() != 0, reason="Hub state reader verifies root-owned fixed storage", ) @pytest.mark.parametrize("state_mode", [0o700, 0o755]) def test_fixed_hub_state_reader_is_bounded_and_rejects_symlink_or_missing_state( store, tmp_path, monkeypatch, state_mode ): controller_root = tmp_path / "controller" state_root = controller_root / "state" controller_root.mkdir(mode=0o700) controller_root.chmod(0o700) monkeypatch.setattr(store, "CONTROLLER_STATE_PARENT", controller_root) monkeypatch.setattr(store, "CONTROLLER_STATE_DIRECTORY", state_root) assert store.read_hub_state_bytes() is None with pytest.raises(store.ReleaseStoreError): store.read_hub_state_bytes(missing_ok=False) state_root.mkdir(mode=state_mode) state_root.chmod(state_mode) state_file = state_root / store.HUB_STATE_FILENAME state_file.write_bytes(b'{"status":"success"}') state_file.chmod(0o600) assert store.read_hub_state_bytes(missing_ok=False) == b'{"status":"success"}' attacker_root = tmp_path / "attacker" attacker_root.mkdir(mode=0o700) attacker_state = attacker_root / store.HUB_STATE_FILENAME attacker_state.write_bytes(b'{"status":"attacker"}') attacker_state.chmod(0o600) state_file.unlink() state_file.symlink_to(attacker_state) with pytest.raises(store.ReleaseStoreError, match="unavailable or untrusted"): store.read_hub_state_bytes() assert attacker_state.read_bytes() == b'{"status":"attacker"}' @pytest.mark.skipif( os.name == "nt" or not hasattr(os, "geteuid") or os.geteuid() != 0, reason="Hub state reader verifies root-owned fixed storage", ) @pytest.mark.parametrize("state_file_mode", [0o640, 0o644, 0o660, 0o664, 0o666]) def test_fixed_hub_state_reader_rejects_other_file_modes( store, tmp_path, monkeypatch, state_file_mode ): controller_root = tmp_path / "controller" state_root = controller_root / "state" state_root.mkdir(parents=True, mode=0o700) controller_root.chmod(0o700) state_file = state_root / store.HUB_STATE_FILENAME state_file.write_bytes(b'{"status":"success"}') state_file.chmod(state_file_mode) monkeypatch.setattr(store, "CONTROLLER_STATE_PARENT", controller_root) monkeypatch.setattr(store, "CONTROLLER_STATE_DIRECTORY", state_root) with pytest.raises(store.ReleaseStoreError, match="unavailable or untrusted"): store.read_hub_state_bytes(missing_ok=False) @pytest.mark.skipif( os.name == "nt" or not hasattr(os, "geteuid") or os.geteuid() != 0, reason="external backup owner checks require root-owned POSIX fixtures", ) def test_descriptor_helpers_verify_an_exact_external_owner(store, tmp_path): expected_owner = (99, 100) backup_root = tmp_path / "backups" backup_root.mkdir(mode=0o700) backup_root.chmod(0o700) os.chown(backup_root, *expected_owner) root_fd = os.open(backup_root, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) attempt_fd = None try: attempt_fd = store.open_child_directory( root_fd, "attempt", create=True, mode=0o700, expected_owner=expected_owner, ) artifact_fd = os.open( "manifest.json", os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=attempt_fd, ) try: os.write(artifact_fd, b"private manifest") os.fchmod(artifact_fd, 0o600) os.fchown(artifact_fd, *expected_owner) finally: os.close(artifact_fd) assert store.read_at( attempt_fd, "manifest.json", max_bytes=64, expected_mode=0o600, expected_owner=expected_owner, ) == b"private manifest" digest, size = store.sha256_at( attempt_fd, "manifest.json", max_bytes=64, expected_mode=0o600, expected_owner=expected_owner, ) assert digest == hashlib.sha256(b"private manifest").hexdigest() assert size == len(b"private manifest") with pytest.raises(store.ReleaseStoreError, match="metadata is unsafe"): store.read_at( attempt_fd, "manifest.json", max_bytes=64, expected_mode=0o600, expected_owner=(0, 0), ) finally: if attempt_fd is not None: os.close(attempt_fd) os.close(root_fd) @pytest.mark.skipif( os.name == "nt" or not hasattr(os, "geteuid") or os.geteuid() != 0, reason="Hub state reader verifies root-owned fixed storage", ) @pytest.mark.parametrize("state_mode", [0o775, 0o777, 0o750]) def test_fixed_hub_state_reader_rejects_unapproved_directory_modes( store, tmp_path, monkeypatch, state_mode ): controller = tmp_path / "controller" state = controller / "state" state.mkdir(parents=True) controller.chmod(0o755) state.chmod(state_mode) state_file = state / store.HUB_STATE_FILENAME state_file.write_bytes(b'{"status":"success"}') state_file.chmod(0o600) monkeypatch.setattr(store, "CONTROLLER_STATE_PARENT", controller) monkeypatch.setattr(store, "CONTROLLER_STATE_DIRECTORY", state) with pytest.raises(store.ReleaseStoreError, match="unavailable or untrusted"): store.read_hub_state_bytes() @pytest.mark.skipif( os.name == "nt" or not hasattr(os, "geteuid") or os.geteuid() != 0, reason="Hub state reader path-swap test requires root-owned POSIX fixtures", ) def test_fixed_hub_state_reader_rejects_parent_swap_during_read(store, tmp_path, monkeypatch): controller_root = tmp_path / "controller" state_root = controller_root / "state" state_root.mkdir(parents=True, mode=0o700) controller_root.chmod(0o700) state_file = state_root / store.HUB_STATE_FILENAME state_file.write_bytes(b"trusted") state_file.chmod(0o600) attacker_root = tmp_path / "attacker" attacker_state_root = attacker_root / "state" attacker_state_root.mkdir(parents=True, mode=0o700) attacker_file = attacker_state_root / store.HUB_STATE_FILENAME attacker_file.write_bytes(b"attacker") attacker_file.chmod(0o600) monkeypatch.setattr(store, "CONTROLLER_STATE_PARENT", controller_root) monkeypatch.setattr(store, "CONTROLLER_STATE_DIRECTORY", state_root) original_read_at = store.read_at def read_then_swap(*args, **kwargs): result = original_read_at(*args, **kwargs) controller_root.rename(tmp_path / "controller.saved") controller_root.symlink_to(attacker_root, target_is_directory=True) return result monkeypatch.setattr(store, "read_at", read_then_swap) with pytest.raises(store.ReleaseStoreError, match="unavailable or untrusted"): store.read_hub_state_bytes() assert attacker_file.read_bytes() == b"attacker" _RENDERED_2A948_SERVICES = ( "control-api", "connector-appops", "connector-gitea", "connector-ingestion", "connector-knowledge", "connector-mobilityops", "connector-projectbrain", "connector-stitch", "connector-unraid", "contextforge", "dataplane-router", "docker-read-broker", "ingestion-clamd", "postgres", "redis", "web", "worker", ) _RENDERED_2A948_RELEASE_BINDS = ( ("connector-gitea", "services/runtime/connector_gitea.py", "/opt/hub/services/runtime/connector_gitea.py", "file"), ("contextforge", "infrastructure/images/contextforge/entrypoint.sh", "/opt/itworx/contextforge-entrypoint.sh", "file"), ("contextforge", "infrastructure/images/contextforge/runtime-contract.json", "/opt/itworx/contextforge-runtime-contract.json", "file"), ("contextforge", "scripts/contextforge_runtime_contract.py", "/opt/itworx/contextforge-runtime-contract.py", "file"), ("control-api", "catalog", "/opt/hub/catalog", "directory"), ("dataplane-router", "catalog", "/opt/hub/catalog", "directory"), ("dataplane-router", "services/runtime/dataplane.py", "/opt/hub/services/runtime/dataplane.py", "file"), ("postgres", "infrastructure/postgres/init", "/docker-entrypoint-initdb.d", "directory"), ("postgres", "infrastructure/images/postgres/entrypoint.sh", "/opt/itworx/postgres-entrypoint.sh", "file"), ("redis", "infrastructure/images/redis/entrypoint.sh", "/opt/itworx/redis-entrypoint.sh", "file"), ("redis", "infrastructure/images/redis/healthcheck.sh", "/opt/itworx/redis-healthcheck.sh", "file"), ("web", "services/runtime/web.py", "/opt/hub/services/runtime/web.py", "file"), ("worker", "catalog", "/opt/hub/catalog", "directory"), ("worker", "services/runtime/state_sync.py", "/opt/hub/services/runtime/state_sync.py", "file"), ) def _rendered_2a948_model(store, revision): """Mount projection captured from protected Hub main commit 2a948b17.""" services = {name: {"volumes": []} for name in _RENDERED_2A948_SERVICES} for service, suffix, target, _kind in _RENDERED_2A948_RELEASE_BINDS: services[service]["volumes"].append( { "type": "bind", "source": store.logical_release_path(revision) + "/" + suffix, "target": target, "read_only": True, "mode": "ro", } ) # These are rendered non-release binds and must remain unchanged. services["docker-read-broker"]["volumes"].append( { "type": "bind", "source": "/var/run/docker.sock", "target": "/var/run/docker.sock", "read_only": True, } ) services["postgres"]["volumes"].append( {"type": "volume", "source": "itworx-mcp-hub-postgres", "target": "/var/lib/postgresql/data"} ) return {"services": services} def test_exact_rendered_2a948_four_file_model_translates_all_release_binds_only(store): revision = "a" * 40 model = _rendered_2a948_model(store, revision) store.validate_release_mount_inventory(model, revision, source_kind="logical") translated = store.translate_release_mounts(model, revision) store.validate_release_mount_inventory(translated, revision, source_kind="host") assert store.RELEASE_BIND_CONTRACT == _RENDERED_2A948_RELEASE_BINDS assert len(_RENDERED_2A948_RELEASE_BINDS) == 14 for service, suffix, target, _kind in _RENDERED_2A948_RELEASE_BINDS: original = next( mount for mount in model["services"][service]["volumes"] if mount.get("target") == target ) observed = next( mount for mount in translated["services"][service]["volumes"] if mount.get("target") == target ) assert original["source"] == f"{store.logical_release_path(revision)}/{suffix}" assert observed["source"] == f"{store.host_release_path(revision)}/{suffix}" assert observed["read_only"] is True assert translated["services"]["docker-read-broker"]["volumes"] == model["services"]["docker-read-broker"]["volumes"] assert translated["services"]["postgres"]["volumes"][-1] == model["services"]["postgres"]["volumes"][-1] assert model["services"]["control-api"]["volumes"][0]["source"].startswith( store.logical_release_path(revision) ) mutations = [] missing = copy.deepcopy(model) missing["services"]["redis"]["volumes"].pop(0) mutations.append(missing) writable = copy.deepcopy(model) writable["services"]["worker"]["volumes"][0]["read_only"] = False mutations.append(writable) wrong_service = copy.deepcopy(model) wrong_service["services"]["contextforge"]["volumes"][0]["target"] = "/tmp/entrypoint.sh" mutations.append(wrong_service) undeclared_leaf = copy.deepcopy(model) undeclared_leaf["services"]["control-api"]["volumes"].append( { "type": "bind", "source": f"{store.logical_release_path(revision)}/scripts", "target": "/opt/hub/scripts", "read_only": True, } ) mutations.append(undeclared_leaf) release_root = copy.deepcopy(model) release_root["services"]["control-api"]["volumes"][0]["source"] = store.logical_release_path(revision) mutations.append(release_root) volume_ancestor = copy.deepcopy(model) volume_ancestor["services"]["control-api"]["volumes"][0]["source"] = store.DOCKER_ROOT.as_posix() mutations.append(volume_ancestor) for unsafe in mutations: with pytest.raises(store.ReleaseStoreError): store.translate_release_mounts(unsafe, revision) def test_private_overlay_contains_only_fixed_volume_leaf_sources(store): revision = "b" * 40 content = json.loads(store.compose_override_bytes(revision)) text = json.dumps(content) assert "${" not in text assert "secret" not in text.lower() mounts = [ (service, mount) for service, config in content["services"].items() for mount in config["volumes"] ] assert len(mounts) == len(_RENDERED_2A948_RELEASE_BINDS) for service, suffix, target, _kind in _RENDERED_2A948_RELEASE_BINDS: mount = next(item for name, item in mounts if name == service and item["target"] == target) assert mount["source"] == f"{store.host_release_path(revision)}/{suffix}" assert mount["type"] == "bind" assert mount["read_only"] is True assert mount["bind"] == {"create_host_path": False} assert "privileged" not in text and "/controller-hub-releases:" not in text @pytest.mark.skipif( os.name == "nt" or not hasattr(os, "geteuid") or os.geteuid() != 0, reason="release-leaf identity checks require root-owned POSIX fixtures", ) def test_release_leaf_verifier_allows_only_regular_files_and_declared_directories(store, tmp_path, monkeypatch): revision = "c" * 40 volume_root = tmp_path / "volume" releases = volume_root / store.RELEASES_DIRECTORY release = releases / revision release.mkdir(parents=True, mode=0o700) releases.chmod(0o700) unique: dict[str, str] = {} for _service, suffix, _target, kind in store.RELEASE_BIND_CONTRACT: unique[suffix] = kind for suffix, kind in unique.items(): target = release.joinpath(*suffix.split("/")) if kind == "directory": target.mkdir(parents=True, mode=0o700, exist_ok=True) else: target.parent.mkdir(parents=True, mode=0o700, exist_ok=True) target.write_bytes(b"protected") target.chmod(0o444) for path in sorted(release.rglob("*"), reverse=True): if path.is_dir(): path.chmod(0o555) release.chmod(0o555) pinned = store.PinnedDirectory.open(volume_root, check_ancestors=False) fixed = store.FixedReleaseStore(host=pinned, logical=pinned) monkeypatch.setattr(fixed, "assert_current", lambda: None) try: store.verify_release_bind_leaves(fixed, revision) bad_file = release / "infrastructure" / "images" / "redis" / "entrypoint.sh" bad_file.parent.chmod(0o755) bad_file.unlink() outside = tmp_path / "outside.sh" outside.write_text("attacker", encoding="utf-8") bad_file.symlink_to(outside) bad_file.parent.chmod(0o555) with pytest.raises(store.ReleaseStoreError, match="symlink"): store.verify_release_bind_leaves(fixed, revision) finally: pinned.close() @pytest.mark.skipif(os.name == "nt", reason="descriptor identity checks require POSIX openat") def test_pinned_store_refuses_ancestor_symlink_swap_before_path_readback(store, tmp_path): parent = tmp_path / "shared" parent.mkdir(mode=0o700) original = parent / "controller" original.mkdir(mode=0o700) marker = original / "state.json" marker.write_bytes(b"old-state") marker.chmod(0o600) pinned = store.PinnedDirectory.open(original, check_ancestors=False) attacker = tmp_path / "attacker" attacker.mkdir(mode=0o700) (attacker / "state.json").write_bytes(b"attacker-state") moved = parent / "controller.saved" try: original.rename(moved) original.symlink_to(attacker, target_is_directory=True) with pytest.raises(store.ReleaseStoreError, match="path identity changed"): pinned.assert_current(original) assert store.read_at(pinned.fd, "state.json", max_bytes=64) == b"old-state" assert (attacker / "state.json").read_bytes() == b"attacker-state" finally: pinned.close() @pytest.mark.skipif(os.name == "nt", reason="atomic dirfd replacement requires POSIX openat") def test_pinned_atomic_replace_uses_original_inode_and_detects_swapped_path(store, tmp_path): parent = tmp_path / "shared" parent.mkdir(mode=0o700) original = parent / "controller" original.mkdir(mode=0o700) pinned = store.PinnedDirectory.open(original, check_ancestors=False) attacker = tmp_path / "attacker" attacker.mkdir(mode=0o700) def swap_path(): original.rename(parent / "controller.saved") original.symlink_to(attacker, target_is_directory=True) try: with pytest.raises(store.ReleaseStoreError, match="path identity changed"): store.atomic_replace_at( pinned.fd, "state.json", b"new-state", expected=None, after_replace=swap_path, path_for_identity_check=original, ) with pytest.raises(FileNotFoundError): store.read_at(pinned.fd, "state.json", max_bytes=64) assert not (attacker / "state.json").exists() finally: pinned.close() @pytest.mark.skipif(os.name == "nt", reason="atomic dirfd replacement requires POSIX openat") def test_pinned_atomic_replace_restores_previous_bytes_after_ancestor_swap(store, tmp_path): parent = tmp_path / "shared" parent.mkdir(mode=0o700) original = parent / "controller" original.mkdir(mode=0o700) state = original / "state.json" state.write_bytes(b"old-state") state.chmod(0o600) pinned = store.PinnedDirectory.open(original, check_ancestors=False) attacker = tmp_path / "attacker" attacker.mkdir(mode=0o700) def swap_path(): original.rename(parent / "controller.saved") original.symlink_to(attacker, target_is_directory=True) try: with pytest.raises(store.ReleaseStoreError, match="path identity changed"): store.atomic_replace_at( pinned.fd, "state.json", b"candidate-state", expected=b"old-state", path_for_identity_check=original, after_replace=swap_path, ) assert store.read_at(pinned.fd, "state.json", max_bytes=64) == b"old-state" assert not (attacker / "state.json").exists() finally: pinned.close()