From 837c263cd814c284495f57b49e415ea7161a310c Mon Sep 17 00:00:00 2001 From: Public Source Publisher Date: Thu, 8 Oct 2026 13:51:07 +0000 Subject: [PATCH] Publish curated Unraid-AppOps-Gateway source from aaf7d11821a86be65c91fead97a6a4758c4fbacd --- PUBLIC_SOURCE_EXPORT.md | 2 +- PUBLIC_SOURCE_MANIFEST.json | 23 +- backend/app/mcp_server.py | 32 + backend/app/services/deployment_broker.py | 68 +- .../test_controller_hub_release_volume.py | 402 ++++++- .../tests/test_hub_delivery_revalidation.py | 1014 +++++++++++++++++ ...test_projectbrain_delivery_revalidation.py | 193 +++- 7 files changed, 1674 insertions(+), 60 deletions(-) create mode 100644 backend/tests/test_hub_delivery_revalidation.py diff --git a/PUBLIC_SOURCE_EXPORT.md b/PUBLIC_SOURCE_EXPORT.md index 7138be6..84c087a 100644 --- a/PUBLIC_SOURCE_EXPORT.md +++ b/PUBLIC_SOURCE_EXPORT.md @@ -1,3 +1,3 @@ # Public source export -This source snapshot was generated from the private canonical repository at revision `21729207a161d87393cfb73e79582bc373aa19cf`. It contains no private Git history or operational evidence. Changes are published from the canonical repository. +This source snapshot was generated from the private canonical repository at revision `aaf7d11821a86be65c91fead97a6a4758c4fbacd`. It contains no private Git history or operational evidence. Changes are published from the canonical repository. diff --git a/PUBLIC_SOURCE_MANIFEST.json b/PUBLIC_SOURCE_MANIFEST.json index c198be9..be7a24d 100644 --- a/PUBLIC_SOURCE_MANIFEST.json +++ b/PUBLIC_SOURCE_MANIFEST.json @@ -1,6 +1,6 @@ { "schemaVersion": 1, - "sourceRevision": "21729207a161d87393cfb73e79582bc373aa19cf", + "sourceRevision": "aaf7d11821a86be65c91fead97a6a4758c4fbacd", "files": [ { "path": ".dockerignore", @@ -404,8 +404,8 @@ }, { "path": "backend/app/mcp_server.py", - "sha256": "24817f3e7566b441876b3e12c0bffe52cf596cf27d098e1f8579bce8a28c65ec", - "bytes": 47931 + "sha256": "feac22ee7a1c755f0fafd6890500b83442b5f13b51e6a6cde3b0227c5b5ab25b", + "bytes": 49536 }, { "path": "backend/app/runner_statistics_mcp.py", @@ -539,8 +539,8 @@ }, { "path": "backend/app/services/deployment_broker.py", - "sha256": "452e6a2c3316501a7828990cb9fe7f689edfd1fa07e68e20e612126cd387bc5f", - "bytes": 42391 + "sha256": "4370e327b0b608b46be446159f59b9975fb99381967282c9661dc1c7837f16bd", + "bytes": 45377 }, { "path": "backend/app/services/deployment_control_plane.py", @@ -969,8 +969,8 @@ }, { "path": "backend/tests/test_controller_hub_release_volume.py", - "sha256": "b629dadd5616d5afd5ff8ad33662d93da829424226f0bb36bda71d9c621049cf", - "bytes": 144585 + "sha256": "630ba534344dd11a499f32ed6aa53816f1cefd0537a4d69f278ed2a11fa68f6a", + "bytes": 160633 }, { "path": "backend/tests/test_controller_root_procedure_contract.py", @@ -1182,6 +1182,11 @@ "sha256": "7fbce0be7f6cc12b68f5b05eb3f7c9eb00478207f1c49ab36ef88915ee53eb82", "bytes": 9756 }, + { + "path": "backend/tests/test_hub_delivery_revalidation.py", + "sha256": "4913be4a0b04549d709c56af2fc88fa6743fca54581c69d3cc853277ab44f7a0", + "bytes": 45474 + }, { "path": "backend/tests/test_hub_permission_preparation.py", "sha256": "3ba8efe338a21b100e4bc20e4fbffe043eb57c411a4143c00332fdb368db5011", @@ -1389,8 +1394,8 @@ }, { "path": "backend/tests/test_projectbrain_delivery_revalidation.py", - "sha256": "579550ac68438da05d9472196961062f33364fe734fc8c4dfc8f6f912ab286c3", - "bytes": 50954 + "sha256": "d46946e8be179372c84adab6f56f57de27a4492db6d0edb73d75035a433c5b40", + "bytes": 57909 }, { "path": "backend/tests/test_prowlarr_connector.py", diff --git a/backend/app/mcp_server.py b/backend/app/mcp_server.py index eb8226c..67ef5df 100644 --- a/backend/app/mcp_server.py +++ b/backend/app/mcp_server.py @@ -792,6 +792,38 @@ def appops_projectbrain_delivery_revalidate( )) +@appops_mcp.tool(name="appops.hub_delivery.preflight", annotations=_READ_ONLY_TOOL) +@_tracked("appops.hub_delivery.preflight") +def appops_hub_delivery_preflight(revision: str) -> dict[str, Any]: + """Read fixed MCP Hub delivery evidence without importing or deploying a candidate.""" + if not isinstance(revision, str) or not re.fullmatch(r"[0-9a-f]{40}", revision): + raise ValueError("Hub delivery revision is invalid") + return _json(DeploymentBrokerClient().hub_delivery_operation( + "preflight", {"revision": revision}, + )) + + +@appops_mcp.tool(name="appops.hub_delivery.revalidate", annotations=_OPERATION_RUN_TOOL) +@_tracked("appops.hub_delivery.revalidate") +def appops_hub_delivery_revalidate( + revision: str, expected_preflight_sha256: str, idempotency_key: str, +) -> dict[str, Any]: + """Queue one fixed MCP Hub delivery only after its exact native evidence fingerprint is rechecked.""" + if (not isinstance(revision, str) or not re.fullmatch(r"[0-9a-f]{40}", revision) + or not isinstance(expected_preflight_sha256, str) + or not re.fullmatch(r"[0-9a-f]{64}", expected_preflight_sha256) + or not isinstance(idempotency_key, str) + or not re.fullmatch(r"[A-Za-z0-9._:-]{8,128}", idempotency_key)): + raise ValueError("Hub delivery revalidation contract is invalid") + return _json(DeploymentBrokerClient().hub_delivery_operation( + "revalidate", { + "revision": revision, + "expectedPreflightSha256": expected_preflight_sha256, + "idempotencyKey": idempotency_key, + }, + )) + + @appops_mcp.tool(name="server.operation.execute", annotations=_LEGACY_APPROVAL_EXECUTION_TOOL) @_tracked("server.operation.execute") def execute_operation( diff --git a/backend/app/services/deployment_broker.py b/backend/app/services/deployment_broker.py index 837455b..c6a0a50 100644 --- a/backend/app/services/deployment_broker.py +++ b/backend/app/services/deployment_broker.py @@ -1,9 +1,10 @@ """Client for the private, typed deployment broker surface. -The client only sends a canonical DeploymentPlanV1 to a configured controller. -It neither exposes a command field nor calls a Docker socket/Portainer exec -endpoint. The controller independently validates repository, revision and -runtime topology before it can mutate production. +The client sends a canonical DeploymentPlanV1 or a closed fixed-target +operation to a configured controller. It neither exposes a command field nor +calls a Docker socket/Portainer exec endpoint. The controller independently +validates repository, revision and runtime topology before it can mutate +production. """ from __future__ import annotations @@ -200,7 +201,15 @@ class DeploymentBrokerClient: except httpx.RequestError as exc: return {"status": "unreachable", "reason": redact_message(str(exc))} if response.status_code in {401, 403}: - return {"status": "unauthorized", "reason": "deployment_broker_authentication_failed"} + rejected = {"status": "unauthorized", "reason": "deployment_broker_authentication_failed"} + try: + auth_payload = response.json() + except ValueError: + auth_payload = None + if ("noMutation" in error_response_fields and isinstance(auth_payload, dict) + and isinstance(auth_payload.get("noMutation"), bool)): + rejected["noMutation"] = auth_payload["noMutation"] + return rejected try: decoded = response.json() except ValueError: @@ -316,6 +325,55 @@ class DeploymentBrokerClient: error_response_fields=frozenset({"noMutation"}), ) + def hub_delivery_operation(self, stage: str, payload: dict[str, Any]) -> dict[str, Any]: + """Use the fixed Hub signed-delivery gate without retrying uncertain requests.""" + if stage not in {"preflight", "revalidate"} or not isinstance(payload, dict): + raise ValueError("Hub delivery stage is invalid") + if stage == "preflight": + if (set(payload) != {"revision"} or not isinstance(payload.get("revision"), str) + or not re.fullmatch(r"[0-9a-f]{40}", payload["revision"])): + raise ValueError("Hub delivery preflight contract is invalid") + elif (set(payload) != {"revision", "expectedPreflightSha256", "idempotencyKey"} + or not isinstance(payload.get("revision"), str) + or not re.fullmatch(r"[0-9a-f]{40}", payload["revision"]) + or not isinstance(payload.get("expectedPreflightSha256"), str) + or not re.fullmatch(r"[0-9a-f]{64}", payload["expectedPreflightSha256"]) + or not isinstance(payload.get("idempotencyKey"), str) + or not re.fullmatch(r"[A-Za-z0-9._:-]{8,128}", payload["idempotencyKey"])): + raise ValueError("Hub delivery revalidation contract is invalid") + result = self._request( + "POST", f"/v1/appops/hub-delivery/{stage}", + secret=settings.mcp_token or "", payload=payload, + error_response_fields=frozenset({"noMutation"}), + ) + if stage == "revalidate" and ( + result.get("status") == "unreachable" or type(result.get("noMutation")) is not bool + ): + return { + "status": "unknown", + "reason": "hub_delivery_outcome_unknown", + "noMutation": False, + } + if stage == "preflight" and result.get("status") == "unreachable": + return { + "status": "unavailable", + "reason": "hub_delivery_preflight_unavailable", + "noMutation": True, + } + if stage == "preflight" and type(result.get("noMutation")) is not bool: + return { + "status": "unavailable", + "reason": "hub_delivery_preflight_response_invalid", + "noMutation": True, + } + if stage == "preflight" and result.get("noMutation") is not True: + return { + "status": "unknown", + "reason": "hub_delivery_preflight_mutation_unexpected", + "noMutation": False, + } + return result + def registered_legacy_operation(self, stage: str, payload: dict[str, Any]) -> dict[str, Any]: """Signed fixed-stage metadata adoption/recovery for existing registered targets.""" if stage not in {"observe", "baseline", "preflight", "adopt", "status", "rollback", "delivery"}: diff --git a/backend/tests/test_controller_hub_release_volume.py b/backend/tests/test_controller_hub_release_volume.py index 2aa3e58..2522a26 100644 --- a/backend/tests/test_controller_hub_release_volume.py +++ b/backend/tests/test_controller_hub_release_volume.py @@ -22,6 +22,7 @@ spec = importlib.util.spec_from_file_location( module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) REVISION = "a" * 40 +OLD_REVISION = "c" * 40 IMAGE_ID = "sha256:" + "b" * 64 ORIGINAL_COMPOSE = b"name: gitea-deploy-runners\nservices:\n deploy-control:\n image: immutable-controller\n" @@ -830,7 +831,7 @@ def test_compose_uses_protected_code_and_two_config_views(compose): "source": module._expected_revision_path(REVISION).as_posix(), "target": module.EXECUTABLE_TARGET, "read_only": True, - "bind": {"propagation": "rprivate"}, + "bind": {"propagation": "rslave"}, } in volumes assert { "type": "volume", @@ -849,7 +850,7 @@ def test_compose_uses_protected_code_and_two_config_views(compose): "source": module._expected_config_path().as_posix(), "target": module.CONFIG_TARGET, "read_only": True, - "bind": {"propagation": "rprivate"}, + "bind": {"propagation": "rslave"}, } in volumes assert len(volumes) == 5 volumes.remove(next(item for item in volumes if item["target"] == module.EXECUTABLE_TARGET)) @@ -899,7 +900,7 @@ def test_compose_migrates_legacy_config_alias_to_explicit_private_bind(compose): "source": module._expected_config_path().as_posix(), "target": module.CONFIG_TARGET, "read_only": True, - "bind": {"propagation": "rprivate"}, + "bind": {"propagation": "rslave"}, } ] @@ -924,8 +925,8 @@ def test_compose_accepts_empty_bind_normalization_and_canonicalizes_both_managed if mount.get("target") in {module.EXECUTABLE_TARGET, module.CONFIG_TARGET} ] assert {item["target"]: item["bind"] for item in managed} == { - module.EXECUTABLE_TARGET: {"propagation": "rprivate"}, - module.CONFIG_TARGET: {"propagation": "rprivate"}, + module.EXECUTABLE_TARGET: {"propagation": "rslave"}, + module.CONFIG_TARGET: {"propagation": "rslave"}, } @@ -947,7 +948,7 @@ def test_compose_normalizes_all_observed_live_generated_bind_defaults(compose): @pytest.mark.parametrize("bind", [ {"create_host_path": 1}, {"create_host_path": "true"}, {"create_host_path": False}, - {"create_host_path": True, "propagation": "rslave"}, + {"create_host_path": True, "propagation": "rshared"}, {"create_host_path": True, "unknown": True}, ]) def test_compose_rejects_unsafe_generated_docker_root_bind(compose, bind): @@ -960,7 +961,7 @@ def test_compose_rejects_unsafe_generated_docker_root_bind(compose, bind): @pytest.mark.parametrize("target", [module.EXECUTABLE_TARGET, module.CONFIG_TARGET]) -@pytest.mark.parametrize("private", [False, True]) +@pytest.mark.parametrize("private", [None, "rprivate", "rslave"]) def test_compose_normalizes_generated_create_host_path_for_fixed_readonly_mounts( compose, target, private ): @@ -968,7 +969,7 @@ def test_compose_normalizes_generated_create_host_path_for_fixed_readonly_mounts volumes = baseline["services"][module.SERVICE]["volumes"] bind = {"create_host_path": True} if private: - bind["propagation"] = "rprivate" + bind["propagation"] = private if target == module.EXECUTABLE_TARGET: mount = next(item for item in volumes if item["target"] == target) mount["bind"] = bind @@ -983,7 +984,7 @@ def test_compose_normalizes_generated_create_host_path_for_fixed_readonly_mounts if mount["target"] in {module.EXECUTABLE_TARGET, module.CONFIG_TARGET}] assert len(managed) == 2 assert all(mount["read_only"] is True for mount in managed) - assert all(mount["bind"] == {"propagation": "rprivate"} for mount in managed) + assert all(mount["bind"] == {"propagation": "rslave"} for mount in managed) assert rendered["services"]["runner"] == baseline["services"]["runner"] # Only captured source metadata is normalized. Activation stays canonical. module._bound_model(json.dumps(rendered).encode()) @@ -996,7 +997,7 @@ def test_compose_normalizes_generated_create_host_path_for_fixed_readonly_mounts @pytest.mark.parametrize("bind", [ {"create_host_path": 1}, {"create_host_path": "true"}, - {"create_host_path": True, "propagation": "rslave"}, + {"create_host_path": True, "propagation": "rshared"}, {"create_host_path": True, "selinux": "z"}, ]) def test_compose_rejects_unsafe_or_untyped_generated_bind_metadata(compose, target, bind): @@ -1016,12 +1017,12 @@ def test_compose_rejects_unsafe_or_untyped_generated_bind_metadata(compose, targ @pytest.mark.parametrize( ("target", "bind"), [ - (module.EXECUTABLE_TARGET, {"propagation": "rslave"}), + (module.EXECUTABLE_TARGET, {"propagation": "rshared"}), (module.EXECUTABLE_TARGET, {"propagation": "shared"}), - (module.EXECUTABLE_TARGET, {"propagation": "rprivate", "create_host_path": False}), - (module.CONFIG_TARGET, {"propagation": "rslave"}), + (module.EXECUTABLE_TARGET, {"propagation": "rslave", "create_host_path": False}), + (module.CONFIG_TARGET, {"propagation": "rshared"}), (module.CONFIG_TARGET, {"propagation": "shared"}), - (module.CONFIG_TARGET, {"propagation": "rprivate", "create_host_path": False}), + (module.CONFIG_TARGET, {"propagation": "rslave", "create_host_path": False}), ], ) def test_compose_rejects_noncanonical_controller_bind_propagation(compose, target, bind): @@ -1049,7 +1050,7 @@ def test_compose_accepts_already_canonical_controller_mounts_without_duplicates( baseline, _calls = compose volumes = baseline["services"][module.SERVICE]["volumes"] next(item for item in volumes if item["target"] == module.EXECUTABLE_TARGET)["bind"] = { - "propagation": "rprivate" + "propagation": "rslave" } volumes.extend( [ @@ -1070,7 +1071,7 @@ def test_compose_accepts_already_canonical_controller_mounts_without_duplicates( "source": module._expected_config_path().as_posix(), "target": module.CONFIG_TARGET, "read_only": True, - "bind": {"propagation": "rprivate"}, + "bind": {"propagation": "rslave"}, }, ] ) @@ -1101,8 +1102,129 @@ def test_activation_and_rollback_models_both_pin_managed_bind_propagation(compos if mount.get("target") in {module.EXECUTABLE_TARGET, module.CONFIG_TARGET} } assert set(managed) == {module.EXECUTABLE_TARGET, module.CONFIG_TARGET} - assert managed[module.EXECUTABLE_TARGET]["bind"] == {"propagation": "rprivate"} - assert managed[module.CONFIG_TARGET]["bind"] == {"propagation": "rprivate"} + assert managed[module.EXECUTABLE_TARGET]["bind"] == {"propagation": "rslave"} + assert managed[module.CONFIG_TARGET]["bind"] == {"propagation": "rslave"} + + +def _legacy_private_compose(revision): + value = json.loads(module.rendered_model(ORIGINAL_COMPOSE, revision, IMAGE_ID)) + for mount in value["services"][module.SERVICE]["volumes"]: + if mount.get("target") in {module.EXECUTABLE_TARGET, module.CONFIG_TARGET}: + mount["bind"] = {"propagation": "rprivate"} + return (json.dumps(value, sort_keys=True, indent=2) + "\n").encode() + + +def _projected_compose(captured): + value = json.loads(captured) + for mount in value["services"][module.SERVICE]["volumes"]: + if mount.get("target") in {module.EXECUTABLE_TARGET, module.CONFIG_TARGET}: + mount["bind"] = {"propagation": "rslave"} + assert mount["read_only"] is True + return (json.dumps(value, sort_keys=True, indent=2) + "\n").encode() + + +@pytest.mark.parametrize( + ("rollback", "current_state", "write_expected", "error"), + [ + (False, "raw_target", True, None), + (False, "projected_target", False, None), + (False, "before", True, None), + (True, "raw_target", True, None), + (True, "raw_after", True, None), + (True, "projected_after", True, None), + (True, "projected_target", False, None), + (True, "before", True, None), + (False, "unrecognized", False, "controller_volume_compose_transition_conflict"), + (True, "unrecognized", False, "controller_volume_compose_rollback_conflict"), + ], +) +def test_root_compose_projects_legacy_receipts_and_bounds_retries( + compose, monkeypatch, tmp_path, rollback, current_state, write_expected, error, +): + _baseline, _calls = compose + before = b"captured-source-compose" + after = _legacy_private_compose("c" * 40) + rollback_value = _legacy_private_compose("b" * 40) + values = {"compose-after.json": after, "compose-rollback.json": rollback_value} + raw_target = rollback_value if rollback else after + projected_target = _projected_compose(raw_target) + current_by_state = { + "raw_target": raw_target, + "projected_target": projected_target, + "raw_after": after, + "projected_after": _projected_compose(after), + "before": before, + "unrecognized": b"unrecognized-compose", + } + live_file = {"value": current_by_state[current_state], "mode": 0o640} + receipt = { + "composeBeforeSha256": module._hash(before), + "composeAfterSha256": module._hash(after), + "composeRollbackSha256": module._hash(rollback_value), + } + writes = [] + + monkeypatch.setattr(module, "_directory", lambda *_args: 23) + monkeypatch.setattr( + module, "_file_info_at", lambda *_args: (live_file["value"], live_file["mode"]) + ) + monkeypatch.setattr(module, "_read_backup", lambda *_args: b'{"composeMode":416}') + monkeypatch.setattr(module.os, "close", lambda _descriptor: None) + + def atomic_at(_parent, name, value, expected, *, private, mode): + assert name == "docker-compose.yml" + assert expected == live_file["value"] + assert private is False + assert mode == 0o640 + writes.append(value) + live_file["value"] = value + + monkeypatch.setattr(module, "_atomic_at", atomic_at) + backup = tmp_path / "backup" + if error: + with pytest.raises(module.VolumeTransitionError, match=error): + module._ensure_root_compose(backup, receipt, values, rollback=rollback) + assert writes == [] + assert live_file["value"] == current_by_state[current_state] + return + + module._ensure_root_compose(backup, receipt, values, rollback=rollback) + module._ensure_root_compose(backup, receipt, values, rollback=rollback) + assert len(writes) == int(write_expected) + assert live_file["value"] == projected_target + assert module._hash(after) == receipt["composeAfterSha256"] + assert module._hash(rollback_value) == receipt["composeRollbackSha256"] + assert values["compose-after.json"] == after + assert values["compose-rollback.json"] == rollback_value + + +@pytest.mark.parametrize("rollback", [False, True]) +def test_root_compose_projected_retry_still_checks_file_mode( + compose, monkeypatch, tmp_path, rollback, +): + _baseline, _calls = compose + after = module.rendered_model(ORIGINAL_COMPOSE, "c" * 40, IMAGE_ID) + rollback_value = module.rendered_model(ORIGINAL_COMPOSE, "b" * 40, IMAGE_ID) + projected = rollback_value if rollback else after + before = b"captured-source-compose" + receipt = { + "composeBeforeSha256": module._hash(before), + "composeAfterSha256": module._hash(after), + "composeRollbackSha256": module._hash(rollback_value), + } + values = {"compose-after.json": after, "compose-rollback.json": rollback_value} + + monkeypatch.setattr(module, "_directory", lambda *_args: 23) + monkeypatch.setattr(module, "_file_info_at", lambda *_args: (projected, 0o600)) + monkeypatch.setattr(module, "_read_backup", lambda *_args: b'{"composeMode":416}') + monkeypatch.setattr(module.os, "close", lambda _descriptor: None) + monkeypatch.setattr( + module, "_atomic_at", + lambda *_args, **_kwargs: pytest.fail("A projected retry must not rewrite the file"), + ) + + with pytest.raises(module.VolumeTransitionError, match="compose_mode_invalid"): + module._ensure_root_compose(tmp_path / "backup", receipt, values, rollback=rollback) def test_compose_allows_only_the_fixed_managed_appdata_volume_mount(compose): @@ -2304,6 +2426,70 @@ def test_copy_diagnostic_classifies_missing_passwd_user_without_retaining_user_v assert "fixture-user-name" not in diagnostic +@pytest.mark.parametrize('stderr,condition', [ + (b'cannot create /private/fixture-secret: No space left on device', 'storage_full'), + (b'Cannot connect to the Docker daemon at unix:///private/fixture-secret', 'daemon_unavailable'), +]) +def test_storage_command_diagnostics_never_retain_private_paths(stderr, condition): + assert module._command_stderr_condition(stderr, 'compose_up', max_bytes=1024) == condition + diagnostic = module._safe_command_diagnostic(('compose_up', 'exit_1', condition)) + assert diagnostic == f'compose_up_exit_1_{condition}' + assert 'fixture-secret' not in diagnostic + + +@pytest.mark.parametrize('read_only,free_bytes', [(True, 1024**3), (False, 0)]) +def test_stop_rejects_failed_storage_before_recording_intent_or_stopping(monkeypatch, read_only, free_bytes): + receipt = {'phase': 'prepared'} + monkeypatch.setattr(module, '_load_transaction', lambda _: (42, receipt, {})) + monkeypatch.setattr(module.os, 'close', lambda _: None) + monkeypatch.setattr(module.os, 'statvfs', lambda _: types.SimpleNamespace( + f_flag=1 if read_only else 0, f_bavail=free_bytes, f_frsize=1, + ), raising=False) + def forbidden(*args, **kwargs): + pytest.fail('Storage rejection must happen before any persisted stop intent or container stop') + monkeypatch.setattr(module, '_set_promotion_phase', forbidden) + monkeypatch.setattr(module, '_stop_current_controller', forbidden) + code = 'controller_volume_storage_read_only' if read_only else 'controller_volume_storage_insufficient' + with pytest.raises(module.VolumeTransitionError, match=code): + module.stop_promotion(Path('/unused-backup')) + assert receipt['phase'] == 'prepared' + + +@pytest.mark.parametrize('native_read_only_graph', [False, True]) +def test_stop_allows_healthy_storage_and_preserves_intent_before_stop(monkeypatch, native_read_only_graph): + receipt = {'phase': 'prepared'} + events = [] + monkeypatch.setattr(module, '_load_transaction', lambda _: (42, receipt, {})) + monkeypatch.setattr(module.os, 'close', lambda _: None) + def filesystem(path): + events.append(('storage', path)) + return types.SimpleNamespace(f_flag=1 if native_read_only_graph and path == module.DOCKER_ROOT else 0, f_bavail=1024**3, f_frsize=1) + monkeypatch.setattr(module.os, 'statvfs', filesystem, raising=False) + def persist(descriptor, value, phase): + events.append(('intent', phase)) + return {**value, 'phase': phase} + monkeypatch.setattr(module, '_set_promotion_phase', persist) + monkeypatch.setattr(module, '_stop_current_controller', lambda *a, **kw: events.append(('stop', None))) + module.stop_promotion(Path('/unused-backup')) + assert events == [('storage', module.ROOT), ('storage', module.DOCKER_ROOT), ('intent', 'stopped'), ('stop', None)] + + +def test_stop_rejects_full_graph_even_with_native_read_only_bind(monkeypatch): + receipt = {'phase': 'prepared'} + monkeypatch.setattr(module, '_load_transaction', lambda _: (42, receipt, {})) + monkeypatch.setattr(module.os, 'close', lambda _: None) + monkeypatch.setattr(module.os, 'statvfs', lambda p: types.SimpleNamespace( + f_flag=1 if p == module.DOCKER_ROOT else 0, + f_bavail=0 if p == module.DOCKER_ROOT else 1024**3, f_frsize=1, + ), raising=False) + def forbidden(*args, **kwargs): + pytest.fail('Full graph must not record a stop or interrupt the controller') + monkeypatch.setattr(module, '_set_promotion_phase', forbidden) + monkeypatch.setattr(module, '_stop_current_controller', forbidden) + with pytest.raises(module.VolumeTransitionError, match='controller_volume_storage_insufficient'): + module.stop_promotion(Path('/unused-backup')) + + def test_promoted_compose_actions_use_only_stdin_and_fixed_service(compose): baseline, calls = compose rendered = module.rendered_model(ORIGINAL_COMPOSE, REVISION, IMAGE_ID) @@ -2319,59 +2505,86 @@ def test_promoted_compose_actions_use_only_stdin_and_fixed_service(compose): assert baseline["name"] == "gitea-deploy-runners" -@pytest.mark.parametrize("phase", ["stopped", "activated"]) -def test_rollback_uses_no_dependencies_when_recreating_prior_controller( - compose, tmp_path, monkeypatch, phase, +@pytest.mark.parametrize( + ("operation", "phase"), + [("activate", "stopped"), ("rollback", "stopped"), ("rollback", "activated")], +) +def test_promotion_uses_projected_persistent_compose_without_dependencies( + compose, tmp_path, monkeypatch, operation, phase, ): _baseline, calls = compose previous = REVISION target = "c" * 40 - previous_compose = module.rendered_model(ORIGINAL_COMPOSE, previous, IMAGE_ID) - target_compose = module.rendered_model(ORIGINAL_COMPOSE, target, IMAGE_ID) + previous_compose = _legacy_private_compose(previous) + target_compose = _legacy_private_compose(target) receipt = { "phase": phase, "imageId": IMAGE_ID, "previousRevision": previous, "sourceRevision": target, + "composeBeforeSha256": module._hash(b"captured-source-compose"), + "composeAfterSha256": module._hash(target_compose), + "composeRollbackSha256": module._hash(previous_compose), } values = { - "controller-before.tar": module._release_tar_bytes( - previous, - ({"controller-release.json": (b"previous release", 0o644)}, {"": 0o755}), - ), - "controller-after.tar": b"unused by isolated rollback test", - "mirror-before.tar": b"unused by isolated rollback test", + "controller-before.tar": b"previous tree", + "controller-after.tar": b"target tree", + "mirror-before.tar": b"mirror snapshot", "compose-rollback.json": previous_compose, "compose-after.json": target_compose, } - current = { - "Mounts": [{ - "Destination": module.EXECUTABLE_TARGET, - "Source": module._expected_revision_path(previous).as_posix(), - }] - } + current = {"Mounts": [{ + "Destination": module.EXECUTABLE_TARGET, + "Source": module._expected_revision_path(target).as_posix(), + }]} + live_file = {"value": target_compose, "mode": 0o640} + monkeypatch.setattr(module, "_load_transaction", lambda _backup: (1, receipt, values)) monkeypatch.setattr(module.os, "close", lambda _descriptor: None) monkeypatch.setattr(module, "_controller_image_inspect", lambda *args, **kwargs: current) + monkeypatch.setattr(module, "_read_tar_tree", lambda _value: ({}, {"": 0o755})) monkeypatch.setattr(module, "_verify_release_leaf", lambda *args, **kwargs: True) - monkeypatch.setattr(module, "_stop_current_controller", lambda *args, **kwargs: None) monkeypatch.setattr(module, "_sync_controller_mirror", lambda *args, **kwargs: None) - monkeypatch.setattr(module, "_ensure_root_compose", lambda *args, **kwargs: None) + monkeypatch.setattr(module, "_stop_current_controller", lambda *args, **kwargs: None) monkeypatch.setattr(module, "_verify_compose_image", lambda *args, **kwargs: None) monkeypatch.setattr(module, "_verify_active_mount", lambda *args, **kwargs: None) monkeypatch.setattr(module, "_set_promotion_phase", lambda *args, **kwargs: None) + monkeypatch.setattr(module, "_directory", lambda *_args: 23) + monkeypatch.setattr( + module, "_file_info_at", lambda *_args: (live_file["value"], live_file["mode"]) + ) + monkeypatch.setattr(module, "_read_backup", lambda *_args: b'{"composeMode":416}') - module.rollback_promotion(tmp_path / "backup") + def atomic_at(_parent, _name, value, expected, **_kwargs): + assert expected == live_file["value"] + live_file["value"] = value + + monkeypatch.setattr(module, "_atomic_at", atomic_at) + calls.clear() + if operation == "activate": + module.activate_promotion(tmp_path / "backup") + raw_target = target_compose + else: + module.rollback_promotion(tmp_path / "backup") + raw_target = previous_compose + projected_target = _projected_compose(raw_target) + assert live_file["value"] == projected_target up_calls = [ - arguments for arguments, _data in calls - if arguments[:2] == ["docker", "--config"] and arguments[3] == "compose" and "up" in arguments + (arguments, data) for arguments, data in calls + if arguments[:2] == ["docker", "--config"] + and arguments[3] == "compose" and "up" in arguments ] assert len(up_calls) == 1 - up_index = up_calls[0].index("up") - assert up_calls[0][up_index:] == [ + up_index = up_calls[0][0].index("up") + assert up_calls[0][0][up_index:] == [ "up", "-d", "--force-recreate", "--pull", "never", "--no-deps", module.SERVICE ] + assert up_calls[0][1] == projected_target + assert module._hash(target_compose) == receipt["composeAfterSha256"] + assert module._hash(previous_compose) == receipt["composeRollbackSha256"] + assert values["compose-after.json"] == target_compose + assert values["compose-rollback.json"] == previous_compose def test_installer_has_no_shared_executable_stage_or_compose_path(): @@ -3261,3 +3474,108 @@ def test_config_snapshot_keeps_appops_legacy_state_contract_closed(damage): entries["appops-legacy-upgrade/unraid-appops-gateway.json"] = (b"{}", 0o644) with pytest.raises(module.VolumeTransitionError): module._config_tree(entries, directories) + + +@pytest.mark.parametrize("action", ["up", "stop"]) +def test_legacy_receipt_projection_preserves_evidence_and_readonly_mounts(compose, action): + _baseline, calls = compose + value = json.loads(module.rendered_model(ORIGINAL_COMPOSE, REVISION, IMAGE_ID)) + for mount in value["services"][module.SERVICE]["volumes"]: + if mount.get("target") in {module.EXECUTABLE_TARGET, module.CONFIG_TARGET}: + mount["bind"] = {"propagation": "rprivate"} + captured = json.dumps(value).encode() + original_hash = module._hash(captured) + with pytest.raises(module.VolumeTransitionError): + module._bound_model(captured) + module.compose_action(captured, action) + projected = json.loads(calls[-1][1]) + for mount in projected["services"][module.SERVICE]["volumes"]: + if mount.get("target") in {module.EXECUTABLE_TARGET, module.CONFIG_TARGET}: + assert mount["read_only"] is True + assert mount["bind"] == {"propagation": "rslave"} + mount["bind"] = {"propagation": "rprivate"} + assert projected == value + assert module._hash(captured) == original_hash + + +@pytest.mark.parametrize("damage", ["writable", "shared", "extra_option", "foreign_source"]) +def test_legacy_receipt_projection_rejects_unsafe_mount_before_docker(compose, damage): + _baseline, calls = compose + value = json.loads(module.rendered_model(ORIGINAL_COMPOSE, REVISION, IMAGE_ID)) + mount = next(x for x in value["services"][module.SERVICE]["volumes"] + if x.get("target") == module.EXECUTABLE_TARGET) + mount["bind"] = {"propagation": "rprivate"} + if damage == "writable": + mount["read_only"] = False + elif damage == "shared": + mount["bind"]["propagation"] = "rshared" + elif damage == "extra_option": + mount["bind"]["create_host_path"] = True + else: + mount["source"] = "/tmp/untrusted-controller" + calls.clear() + with pytest.raises(module.VolumeTransitionError): + module.compose_action(json.dumps(value).encode(), "up") + assert calls == [] + + +@pytest.mark.parametrize("phase", ["prepared", "stopped", "activated", "verified"]) +@pytest.mark.parametrize("failure", ["read_only", "full"]) +def test_forward_stop_rechecks_storage_after_intent_or_on_resume(monkeypatch, phase, failure): + receipt = {"phase": phase, "imageId": IMAGE_ID, + "previousRevision": OLD_REVISION, "sourceRevision": REVISION} + persisted = phase != "prepared" + events = [] + monkeypatch.setattr(module, "_load_transaction", lambda _: (42, receipt, { + "compose-rollback.json": b"old", "compose-after.json": b"new"})) + monkeypatch.setattr(module.os, "close", lambda _: None) + def persist(_fd, value, new_phase): + nonlocal persisted + persisted = True + return {**value, "phase": new_phase} + monkeypatch.setattr(module, "_set_promotion_phase", persist) + def filesystem(_path): + return types.SimpleNamespace( + f_flag=1 if persisted and failure == "read_only" else 0, + f_bavail=0 if persisted and failure == "full" else 1024**3, + f_frsize=1) + monkeypatch.setattr(module.os, "statvfs", filesystem, raising=False) + monkeypatch.setattr(module, "_controller_image_inspect", lambda *_a, **_kw: { + "State": {"Running": True}, "Mounts": [{"Destination": module.EXECUTABLE_TARGET, + "Source": module._expected_revision_path(OLD_REVISION).as_posix()}]}) + monkeypatch.setattr(module, "compose_action", lambda *args: events.append(args)) + code = "controller_volume_storage_read_only" if failure == "read_only" else "controller_volume_storage_insufficient" + with pytest.raises(module.VolumeTransitionError, match=code): + module.stop_promotion(Path("/unused-backup")) + assert events == [] + + +def test_resumed_forward_stop_of_stopped_controller_needs_no_storage(monkeypatch): + receipt = {"phase": "stopped", "imageId": IMAGE_ID, + "previousRevision": OLD_REVISION, "sourceRevision": REVISION} + monkeypatch.setattr(module, "_load_transaction", lambda _: (42, receipt, {})) + monkeypatch.setattr(module.os, "close", lambda _: None) + monkeypatch.setattr(module, "_controller_image_inspect", lambda *_a, **_kw: { + "State": {"Running": False}, "Mounts": [{"Destination": module.EXECUTABLE_TARGET, + "Source": module._expected_revision_path(OLD_REVISION).as_posix()}]}) + monkeypatch.setattr(module, "_require_writable_promotion_storage", + lambda: pytest.fail("Already stopped recovery must remain possible")) + monkeypatch.setattr(module, "compose_action", lambda *_: pytest.fail("No stop needed")) + module.stop_promotion(Path("/unused-backup")) + + +def test_rollback_stop_remains_available_without_forward_storage_gate(monkeypatch): + receipt = {"phase": "stopped", "imageId": IMAGE_ID, + "previousRevision": OLD_REVISION, "sourceRevision": REVISION} + events = [] + monkeypatch.setattr(module, "_load_transaction", lambda _: (42, receipt, { + "compose-rollback.json": b"old", "compose-after.json": b"new"})) + monkeypatch.setattr(module.os, "close", lambda _: None) + monkeypatch.setattr(module, "_controller_image_inspect", lambda *_a, **_kw: { + "State": {"Running": True}, "Mounts": [{"Destination": module.EXECUTABLE_TARGET, + "Source": module._expected_revision_path(REVISION).as_posix()}]}) + monkeypatch.setattr(module, "_require_writable_promotion_storage", + lambda: pytest.fail("Rollback must not inherit forward capacity gate")) + monkeypatch.setattr(module, "compose_action", lambda *args: events.append(args)) + module.rollback_promotion(Path("/unused-backup"), stop_only=True) + assert events == [(b"new", "stop")] diff --git a/backend/tests/test_hub_delivery_revalidation.py b/backend/tests/test_hub_delivery_revalidation.py new file mode 100644 index 0000000..3b2c0be --- /dev/null +++ b/backend/tests/test_hub_delivery_revalidation.py @@ -0,0 +1,1014 @@ +from __future__ import annotations + +import hashlib +import io +import json +import os +import sys +from contextlib import contextmanager, nullcontext +from email.message import Message +from types import SimpleNamespace + +import pytest +from test_deploy_broker import BROKER_ROOT, _load_module +from test_projectbrain_delivery_revalidation import _git + +REPOSITORY = "jens/itworx_mcp_hub" +REVISION = "0253e7acef01ad952266ff180ed434e22b986ddc" +PREFIX = "/v1/appops/hub-delivery/" +KEY = "hub-delivery-request-2026-10-08" + + +@pytest.fixture +def controller(monkeypatch): + module = _load_module("hub_delivery_revalidation_test", BROKER_ROOT / "status_server.py") + monkeypatch.setattr(module, "_self_upgrade_signature_is_valid", lambda *_a, **_k: True) + monkeypatch.setattr(module, "_self_upgrade_secret", lambda: b"hub-delivery-test-key-" + b"x" * 40) + monkeypatch.setattr(module, "_require_controller_release", lambda: {"artifactSha256": "f" * 64}) + return module + + +def invoke(module, stage: str, payload: dict[str, object], *, trusted: bool = True): + handler = module.Handler.__new__(module.Handler) + body = json.dumps(payload, separators=(",", ":")).encode() + handler.path = PREFIX + stage + handler.headers = Message() + handler.headers["Content-Length"] = str(len(body)) + handler.headers["Content-Type"] = "application/json" + handler.rfile = io.BytesIO(body) + handler._trusted_client = lambda: trusted + responses = [] + handler._send_json = lambda status, data: responses.append((status, json.loads(data))) + handler._single_header = lambda name: ( + "0" * 32 if name == "X-ITWorx-Deployment-Nonce" + else "0" if name in {"X-ITWorx-Deployment-Timestamp", "X-ITWorx-Deployment-Signature"} + else handler.headers.get(name) + ) + handler._handle_hub_delivery() + return responses[0] + + +def _snapshot(module, *, preflight_sha: str = "a" * 64): + return { + "status": "ready", + "repository": REPOSITORY, + "revision": REVISION, + "originalDelivery": "c" * 64, + "validationRunId": 22495, + "validationRunNumber": 22495, + "validationRunAttempt": 1, + "validationJobId": 39309, + "validationJobCount": 2, + "validationStepCount": 12, + "validationStatusId": 4, + "preflightSha256": preflight_sha, + "_semanticDigest": "c" * 64, + "_originalFileSha256": "b" * 64, + "_project": {"slug": "itworx-mcp-hub", "branch": "main"}, + "_validationPolicy": {"contexts": [module.HUB_DELIVERY_FULL_CONTEXT]}, + "_stateSha256": "1" * 64, + "_migrationIntentSha256": "2" * 64, + "_baselineSha256": "5" * 64, + "_profileSha256": "3" * 64, + "_controllerArtifactSha256": "f" * 64, + "_validationEvidenceSha256": "4" * 64, + } + + +def test_preflight_is_fixed_read_only_and_returns_bounded_evidence(controller, monkeypatch): + module = controller + calls = [] + monkeypatch.setattr(module, "_hub_delivery_mutation_lock", nullcontext) + monkeypatch.setattr(module, "_hub_delivery_snapshot", lambda revision: calls.append(revision) or _snapshot(module)) + monkeypatch.setattr( + module, + "_enqueue_hub_terminal_revalidation", + lambda *_a, **_k: pytest.fail("preflight must never enqueue"), + ) + + code, result = invoke(module, "preflight", {"revision": REVISION}) + + assert code == 200 + assert calls == [REVISION] + assert result == { + "status": "ready", + "repository": REPOSITORY, + "revision": REVISION, + "originalDelivery": "c" * 64, + "validationRunId": 22495, + "validationRunNumber": 22495, + "validationRunAttempt": 1, + "validationJobId": 39309, + "validationJobCount": 2, + "validationStepCount": 12, + "validationStatusId": 4, + "preflightSha256": "a" * 64, + "noMutation": True, + } + + +def test_preflight_auth_and_exact_contract_precede_all_evidence_reads(controller, monkeypatch): + module = controller + monkeypatch.setattr(module, "_hub_delivery_snapshot", lambda *_a: pytest.fail("unauthorized or malformed input read state")) + monkeypatch.setattr(module, "_hub_delivery_mutation_lock", lambda: pytest.fail("lock acquired for malformed input")) + monkeypatch.setattr(module, "_self_upgrade_signature_is_valid", lambda *_a, **_k: False) + assert invoke(module, "preflight", {"revision": REVISION})[0] == 403 + + monkeypatch.setattr(module, "_self_upgrade_signature_is_valid", lambda *_a, **_k: True) + assert invoke(module, "preflight", {"revision": REVISION, "repository": REPOSITORY})[0] == 400 + assert invoke(module, "preflight", {"revision": "A" * 40})[0] == 400 + + +def test_revalidate_recomputes_cas_before_one_enqueue(controller, monkeypatch): + module = controller + snap = _snapshot(module) + queued = [] + monkeypatch.setattr(module, "_hub_delivery_mutation_lock", nullcontext) + monkeypatch.setattr(module, "_hub_delivery_existing_revalidation", lambda *_a: None) + monkeypatch.setattr(module, "_hub_delivery_snapshot", lambda revision: snap) + monkeypatch.setattr(module, "_hub_delivery_signed_binding", lambda snapshot, key_sha: {"key": key_sha}) + monkeypatch.setattr( + module, + "_enqueue_hub_terminal_revalidation", + lambda revision, project, binding, original_sha: queued.append( + (revision, project, binding, original_sha) + ) or ("awaiting_validation", "c" * 64, True), + ) + + code, result = invoke(module, "revalidate", { + "revision": REVISION, + "expectedPreflightSha256": "a" * 64, + "idempotencyKey": KEY, + }) + + assert code == 202 + assert result["status"] == "awaiting_validation" + assert result["noMutation"] is False + assert queued == [(REVISION, snap["_project"], {"key": hashlib.sha256(KEY.encode()).hexdigest()}, "b" * 64)] + + +def test_revalidate_stale_cas_has_explicit_no_mutation_and_does_not_enqueue(controller, monkeypatch): + module = controller + monkeypatch.setattr(module, "_hub_delivery_mutation_lock", nullcontext) + monkeypatch.setattr(module, "_hub_delivery_existing_revalidation", lambda *_a: None) + monkeypatch.setattr(module, "_hub_delivery_snapshot", lambda revision: _snapshot(module)) + monkeypatch.setattr(module, "_enqueue_hub_terminal_revalidation", lambda *_a: pytest.fail("stale CAS enqueued")) + + code, result = invoke(module, "revalidate", { + "revision": REVISION, + "expectedPreflightSha256": "9" * 64, + "idempotencyKey": KEY, + }) + + assert code == 409 + assert result == {"status": "blocked", "reason": "hub_delivery_preflight_changed", "noMutation": True} + + +def test_revalidate_same_key_replays_progressed_sibling_without_new_snapshot(controller, monkeypatch): + module = controller + expected = {"status": "deployed", "delivery": "c" * 64} + monkeypatch.setattr(module, "_hub_delivery_mutation_lock", nullcontext) + monkeypatch.setattr(module, "_hub_delivery_existing_revalidation", lambda *args: expected) + monkeypatch.setattr(module, "_hub_delivery_snapshot", lambda *_a: pytest.fail("replay must not require a fresh snapshot")) + monkeypatch.setattr(module, "_enqueue_hub_terminal_revalidation", lambda *_a: pytest.fail("replay created another sibling")) + + code, result = invoke(module, "revalidate", { + "revision": REVISION, + "expectedPreflightSha256": "a" * 64, + "idempotencyKey": KEY, + }) + + assert code == 200 + assert result == {**expected, "repository": REPOSITORY, "revision": REVISION, "noMutation": True} + + +def test_revalidate_write_ambiguity_is_unknown_not_definite_failure(controller, monkeypatch): + module = controller + monkeypatch.setattr(module, "_hub_delivery_mutation_lock", nullcontext) + monkeypatch.setattr(module, "_hub_delivery_existing_revalidation", lambda *_a: None) + monkeypatch.setattr(module, "_hub_delivery_snapshot", lambda revision: _snapshot(module)) + monkeypatch.setattr(module, "_hub_delivery_signed_binding", lambda *_a: {"binding": "signed"}) + monkeypatch.setattr(module, "_enqueue_hub_terminal_revalidation", lambda *_a: (_ for _ in ()).throw(OSError("fsync failed"))) + + code, result = invoke(module, "revalidate", { + "revision": REVISION, + "expectedPreflightSha256": "a" * 64, + "idempotencyKey": KEY, + }) + + assert code == 503 + assert result == {"status": "unknown", "reason": "hub_delivery_enqueue_outcome_unknown", "noMutation": False} + + +def test_revalidate_wakeup_failure_after_durable_enqueue_is_reported_unknown(controller, monkeypatch): + module = controller + monkeypatch.setattr(module, "_hub_delivery_mutation_lock", nullcontext) + monkeypatch.setattr(module, "_hub_delivery_existing_revalidation", lambda *_a: None) + monkeypatch.setattr(module, "_hub_delivery_snapshot", lambda revision: _snapshot(module)) + monkeypatch.setattr(module, "_hub_delivery_signed_binding", lambda *_a: {"binding": "signed"}) + monkeypatch.setattr(module, "_enqueue_hub_terminal_revalidation", lambda *_a: ("awaiting_validation", "c" * 64, True)) + monkeypatch.setattr(module, "_webhook_wakeup", SimpleNamespace(set=lambda: (_ for _ in ()).throw(OSError("wake failed")))) + + code, result = invoke(module, "revalidate", { + "revision": REVISION, + "expectedPreflightSha256": "a" * 64, + "idempotencyKey": KEY, + }) + + assert code == 503 + assert result == {"status": "unknown", "reason": "hub_delivery_enqueue_outcome_unknown", "noMutation": False} + + +def test_hub_binding_is_hmac_bound_and_rejects_tampering(controller, monkeypatch): + module = controller + snapshot = _snapshot(module) + binding = module._hub_delivery_signed_binding(snapshot, hashlib.sha256(KEY.encode()).hexdigest()) + assert module._hub_delivery_binding_is_valid(binding, REPOSITORY, REVISION) + assert binding["schema"] == module.HUB_DELIVERY_BINDING_SCHEMA + assert not module._hub_delivery_binding_is_valid( + {**binding, "stateSha256": "9" * 64}, REPOSITORY, REVISION, + ) + assert not module._hub_delivery_binding_is_valid( + {**binding, "repository": "jens/other"}, REPOSITORY, REVISION, + ) + + +def test_worker_rejects_unbound_hub_revalidation_before_candidate_pipeline(controller): + module = controller + assert not module._hub_delivery_worker_admission( + {"schema": module._WEBHOOK_REVALIDATION_SCHEMA}, REPOSITORY, REVISION, + ) + + +def test_worker_rechecks_hmac_bound_snapshot_before_candidate_import(controller, monkeypatch): + module = controller + snapshot = _snapshot(module) + key_sha = hashlib.sha256(KEY.encode()).hexdigest() + binding = module._hub_delivery_signed_binding(snapshot, key_sha) + record = { + "schema": module._WEBHOOK_REVALIDATION_SCHEMA, + "semantic_sha256": binding["semanticSha256"], + "revalidation_of": binding["semanticSha256"], + "hub_delivery_revalidation": binding, + } + changed = {**snapshot, "preflightSha256": "9" * 64} + monkeypatch.setattr(module, "_hub_delivery_mutation_lock", nullcontext) + monkeypatch.setattr(module, "_hub_delivery_snapshot", lambda *_a: changed) + + assert not module._hub_delivery_worker_admission(record, REPOSITORY, REVISION) + + +def test_final_hub_admission_persists_deployment_start_under_one_lock(controller, monkeypatch, tmp_path): + module = controller + semantic = module._webhook_semantic_digest(7, "refs/heads/main", REVISION) + snapshot = {**_snapshot(module), "_semanticDigest": semantic} + binding = module._hub_delivery_signed_binding( + snapshot, hashlib.sha256(KEY.encode()).hexdigest(), + ) + record = { + "schema": module._WEBHOOK_REVALIDATION_SCHEMA, + "semantic_sha256": semantic, + "revalidation_of": semantic, + "hub_delivery_revalidation": binding, + "status": "validation_passed", + "audit": [], + } + marker = tmp_path / f"{semantic}.revalidation.json" + lock_depth = [] + persisted = [] + + @contextmanager + def mutation_lock(): + lock_depth.append(True) + try: + yield + finally: + lock_depth.pop() + + monkeypatch.setattr(module, "_hub_delivery_mutation_lock", mutation_lock) + monkeypatch.setattr(module, "_hub_delivery_snapshot", lambda _revision: snapshot) + monkeypatch.setattr( + module, "_atomic_inbox_update", + lambda _marker, updated: persisted.append((bool(lock_depth), json.loads(json.dumps(updated)))), + ) + + assert module._hub_delivery_finalize_worker_admission( + marker, record, REPOSITORY, REVISION, + ) + assert len(persisted) == 1 + assert persisted[0][0] + assert persisted[0][1]["status"] == "deploying" + assert persisted[0][1]["audit"][-1]["stage"] == "deployment_started" + + +@pytest.mark.parametrize( + "drift_field", + [ + "_stateSha256", + "_profileSha256", + "_controllerArtifactSha256", + "_validationEvidenceSha256", + "_originalFileSha256", + ], +) +def test_worker_rechecks_hub_binding_after_candidate_prep_before_deployment( + controller, monkeypatch, tmp_path, drift_field, +): + module = controller + semantic = module._webhook_semantic_digest(7, "refs/heads/main", REVISION) + snapshot = {**_snapshot(module), "_semanticDigest": semantic} + binding = module._hub_delivery_signed_binding( + snapshot, hashlib.sha256(KEY.encode()).hexdigest(), + ) + changed = { + **snapshot, + drift_field: "9" * 64, + "preflightSha256": "e" * 64, + } + snapshot_reads = 0 + preparation_calls = [] + deployment_calls = [] + persisted = [] + project = {"slug": "itworx-mcp-hub"} + record = { + "schema": module._WEBHOOK_REVALIDATION_SCHEMA, + "semantic_sha256": semantic, + "revalidation_of": semantic, + "body_sha256": "6" * 64, + "gitea_repository_id": 7, + "ref": "refs/heads/main", + "repository": REPOSITORY, + "revision": REVISION, + "status": "validation_passed", + "validation": {}, + "audit": [], + "hub_delivery_revalidation": binding, + } + marker = tmp_path / f"{semantic}.revalidation.json" + + def read_snapshot(_revision): + nonlocal snapshot_reads + snapshot_reads += 1 + return snapshot if snapshot_reads == 1 else changed + + monkeypatch.setattr(module, "_hub_delivery_mutation_lock", nullcontext) + monkeypatch.setattr(module, "_hub_delivery_snapshot", read_snapshot) + monkeypatch.setattr(module, "_load_projects", lambda: {REPOSITORY: project}) + monkeypatch.setattr(module, "_webhook_policy", lambda *_a: (7, "main")) + monkeypatch.setattr( + module, "_validation_policy", + lambda *_a: {"contexts": [module.HUB_DELIVERY_FULL_CONTEXT]}, + ) + monkeypatch.setattr(module, "_record_policy_for_processing", lambda *_a: True) + monkeypatch.setattr(module, "_is_current_branch_head", lambda *_a: True) + monkeypatch.setattr(module, "_supersede_older_deliveries", lambda *_a: None) + monkeypatch.setattr(module, "_read_inbox_record", lambda _marker: record) + monkeypatch.setattr( + module, "_atomic_inbox_update", + lambda _marker, updated: persisted.append(json.loads(json.dumps(updated))), + ) + monkeypatch.setattr( + module, "_prepare_hub_candidate_delivery", + lambda *_a: preparation_calls.append(True) or True, + ) + monkeypatch.setattr( + module, "_run_registered_deploy", + lambda *_a, **_kw: deployment_calls.append(True) or SimpleNamespace(returncode=0), + ) + + module._process_webhook_record(marker) + + assert snapshot_reads == 2 + assert preparation_calls == [True] + assert deployment_calls == [] + assert persisted[-1]["status"] == "rejected" + assert persisted[-1]["reason"] == "hub_delivery_revalidation_preflight_changed" + assert persisted[-1]["audit"][-1]["stage"] == "rejected" + + +def test_hub_native_full_validation_requires_real_successful_steps(controller, monkeypatch): + module = controller + run = { + "id": 22495, "run_number": 1180, "run_attempt": 1, + "created_at": "2026-10-08T00:00:00.000Z", "status": "queued", + "path": "managed-validation.yml@refs/heads/main", "event": "push", + "head_branch": "main", "head_sha": REVISION, + "repository": {"full_name": REPOSITORY}, + "head_repository": {"full_name": REPOSITORY}, "pull_requests": [], + } + full = { + "id": 39309, "name": "full", "run_id": 22495, "run_attempt": 1, + "head_sha": REVISION, "head_branch": "main", "status": "completed", + "conclusion": "success", + "steps": [ + {"number": 0, "name": module.HUB_DELIVERY_CHECKOUT_STEP, "status": "completed", "conclusion": "success"}, + {"number": 1, "name": module.HUB_DELIVERY_DOTNET_STEP, "status": "completed", "conclusion": "skipped"}, + {"number": 2, "name": module.HUB_DELIVERY_VALIDATION_STEP, "status": "completed", "conclusion": "success"}, + ], + } + release = { + "id": 39310, "name": "release-images", "run_id": 22495, + "run_attempt": 1, "head_sha": REVISION, "head_branch": "main", + "status": "queued", + } + jobs = [full, release] + status = { + "id": 4, + "context": module.HUB_DELIVERY_FULL_CONTEXT, + "status": "success", + "target_url": "/Jens/ITWorx_MCP_Hub/actions/runs/22495/jobs/39309", + "url": f"https://gitea.example/api/v1/repos/Jens/ITWorx_MCP_Hub/statuses/{REVISION}", + "updated_at": "2026-10-08T00:05:00.000Z", + } + statuses = [status] + fake_candidate = SimpleNamespace( + WORKFLOW_LIST_PATH="/repos/jens/itworx_mcp_hub/actions/workflows", + REPOSITORY_API="/repos/jens/itworx_mcp_hub", + RUN_PATH="managed-validation.yml@refs/heads/main", + _api_json=lambda _api, path, **_kwargs: ( + {"workflows": [{"id": "managed-validation.yml", "path": ".gitea/workflows/managed-validation.yml", "state": "active"}], "total_count": 1} + if path.endswith("/actions/workflows") else + {"protected": True, "commit": {"id": REVISION}} + ), + _fixed_workflow=lambda *_a: None, + _run_matches=lambda row, revision: ( + row.get("head_sha") == revision + and row.get("head_branch") == "main" + and row.get("event") == "push" + and row.get("path") == "managed-validation.yml@refs/heads/main" + and row.get("repository", {}).get("full_name") == REPOSITORY + and row.get("head_repository", {}).get("full_name") == REPOSITORY + and row.get("pull_requests") == [] + ), + _paged_rows=lambda _api, path, **_kwargs: [run] if path.endswith("/actions/runs") else jobs, + ) + monkeypatch.setitem(sys.modules, "hub_candidate_import", fake_candidate) + monkeypatch.setattr(module, "_hub_candidate_api_json", lambda path, **_k: statuses if "/statuses/" in path else {}) + monkeypatch.setattr(module, "_gitea_validation_settings", lambda: ("https://gitea.example/api/v1", b"token")) + + proof = module._hub_delivery_validation_evidence(REVISION) + assert proof["runId"] == 22495 + assert proof["runNumber"] == 1180 + assert proof["jobId"] == 39309 + assert proof["jobCount"] == 2 + assert proof["releaseJobId"] == 39310 + assert proof["stepCount"] == 3 + assert proof["statusId"] == 4 + + run["status"] = "requested" + release["status"] = "requested" + assert module._hub_delivery_validation_evidence(REVISION)["runNumber"] == 1180 + run["status"] = "in_progress" + release["status"] = "queued" + + full["steps"][2]["conclusion"] = "failure" + with pytest.raises(module.HubDeliveryRefused): + module._hub_delivery_validation_evidence(REVISION) + + full["steps"][2]["conclusion"] = "success" + full["run_attempt"] = True + with pytest.raises(module.HubDeliveryRefused): + module._hub_delivery_validation_evidence(REVISION) + full["run_attempt"] = 1 + del full["steps"][0] + with pytest.raises(module.HubDeliveryRefused): + module._hub_delivery_validation_evidence(REVISION) + + full["steps"].insert(0, { + "number": 0, "name": "Run actions/checkout@v4", "status": "completed", "conclusion": "success", + }) + with pytest.raises(module.HubDeliveryRefused): + module._hub_delivery_validation_evidence(REVISION) + + full["steps"][0]["name"] = module.HUB_DELIVERY_CHECKOUT_STEP + statuses.append({ + **status, + "id": 5, + "status": "failure", + "target_url": "/Jens/ITWorx_MCP_Hub/actions/runs/22496/jobs/39309", + "updated_at": "2026-10-08T00:06:00.000Z", + }) + with pytest.raises(module.HubDeliveryRefused): + module._hub_delivery_validation_evidence(REVISION) + + statuses[:] = [status] + jobs.pop() + with pytest.raises(module.HubDeliveryRefused): + module._hub_delivery_validation_evidence(REVISION) + jobs.append(release) + release["id"] = full["id"] + with pytest.raises(module.HubDeliveryRefused): + module._hub_delivery_validation_evidence(REVISION) + + +@pytest.mark.parametrize( + ("success_time", "failure_time", "success_id", "failure_id", "refused"), + [ + ("2026-10-08T00:05:00Z", "2026-10-08T00:05:00.500Z", 4, 5, True), + ("2026-10-08T00:05:00.500Z", "2026-10-08T00:05:00Z", 5, 4, False), + ("2026-10-08T00:05:00Z", "2026-10-08T00:05:00.000Z", 4, 5, True), + ("2026-10-08T00:05:00.000Z", "2026-10-08T00:05:00Z", 5, 4, False), + ("2026-10-08T00:05:00Z", "2026-02-30T00:05:00Z", 4, 5, True), + ], +) +def test_hub_required_status_uses_chronological_utc_order( + controller, monkeypatch, success_time, failure_time, success_id, failure_id, refused, +): + module = controller + source = f"https://gitea.example/api/v1/repos/Jens/ITWorx_MCP_Hub/statuses/{REVISION}" + statuses = [ + {"id": success_id, "context": module.HUB_DELIVERY_FULL_CONTEXT, + "status": "success", "url": source, + "target_url": "/Jens/ITWorx_MCP_Hub/actions/runs/22495/jobs/39309", + "updated_at": success_time}, + {"id": failure_id, "context": module.HUB_DELIVERY_FULL_CONTEXT, + "status": "failure", "url": source, + "target_url": "/Jens/ITWorx_MCP_Hub/actions/runs/22496/jobs/39311", + "updated_at": failure_time}, + ] + monkeypatch.setitem(sys.modules, "hub_candidate_import", SimpleNamespace( + REPOSITORY_API="/repos/jens/itworx_mcp_hub", + )) + monkeypatch.setattr(module, "_gitea_validation_settings", lambda: ("https://gitea.example/api/v1", b"token")) + monkeypatch.setattr(module, "_hub_candidate_api_json", lambda *_a, **_k: statuses) + if refused: + with pytest.raises(module.HubDeliveryRefused): + module._hub_delivery_full_status_evidence(REVISION, 22495, 39309) + else: + proof = module._hub_delivery_full_status_evidence(REVISION, 22495, 39309) + assert proof["id"] == success_id + assert proof["updatedAt"] == success_time + + +def test_hub_status_transport_allows_only_fixed_exact_sha_pages(controller, monkeypatch): + module = controller + requests = [] + monkeypatch.setattr( + module, "_hub_candidate_request", + lambda target, **_kwargs: requests.append(target) or object(), + ) + monkeypatch.setattr(module, "_hub_candidate_transport", lambda *_a, **_kwargs: b"[]") + + path = f"/repos/jens/itworx_mcp_hub/statuses/{REVISION}" + assert module._hub_candidate_api_json(path, query={"page": "1", "limit": "100"}) == [] + assert requests == [path + "?page=1&limit=100"] + with pytest.raises(module._HubCandidateTransportError): + module._hub_candidate_api_json(path, query={"page": "1", "limit": "100", "state": "success"}) + with pytest.raises(module._HubCandidateTransportError): + module._hub_candidate_api_json(f"/repos/jens/other/statuses/{REVISION}", query={"page": "1", "limit": "100"}) + + +def test_original_terminal_record_is_preserved_and_sibling_is_idempotent(controller, monkeypatch, tmp_path): + module = controller + repository_id = 314 + semantic = module._webhook_semantic_digest(repository_id, "refs/heads/main", REVISION) + original = { + "schema": module._WEBHOOK_SCHEMA, + "semantic_sha256": semantic, + "body_sha256": "d" * 64, + "delivery": "01234567-89ab-4cde-8fab-0123456789ab", + "gitea_repository_id": repository_id, + "ref": "refs/heads/main", + "repository": REPOSITORY, + "revision": REVISION, + "source": "gitea_webhook", + "status": "validation_timeout", + "reason": "validation_timeout", + "validation": {"validation_gate_version": 1}, + "audit": [{"at": 1, "stage": "received"}, {"at": 2, "stage": "validation_timeout"}], + "received_at": 1, + "updated_at": 2, + } + raw = module._hub_delivery_canonical(original) + b"\n" + original_path = tmp_path / f"{semantic}.json" + sibling_path = tmp_path / f"{semantic}.revalidation.json" + original_path.write_bytes(raw) + original_path.chmod(0o600) + tmp_path.chmod(0o700) + monkeypatch.setattr(module, "WEBHOOK_INBOX_ROOT", tmp_path) + project = {"slug": "itworx-mcp-hub", "branch": "main", "gitea_repository_id": repository_id} + monkeypatch.setattr(module, "_load_projects", lambda: {REPOSITORY: project}) + policy = {"validation_gate_version": 1, "contexts": [module.HUB_DELIVERY_FULL_CONTEXT], + "timeout_seconds": 60, "poll_interval_seconds": 2, "max_poll_interval_seconds": 30} + monkeypatch.setattr(module, "_webhook_policy", lambda *_a: (repository_id, "main", "hub-secret")) + monkeypatch.setattr(module, "_validation_policy", lambda *_a: policy) + monkeypatch.setattr(module, "_is_current_branch_head", lambda *_a: True) + monkeypatch.setattr(module, "_validation_snapshot_matches", lambda observed, expected: observed == {"validation_gate_version": 1}) + monkeypatch.setattr(module, "_webhook_wakeup", SimpleNamespace(set=lambda: None)) + binding_snapshot = { + "revision": REVISION, + "_semanticDigest": semantic, + "preflightSha256": "a" * 64, + "_originalFileSha256": hashlib.sha256(raw).hexdigest(), + "_stateSha256": "1" * 64, + "_migrationIntentSha256": "2" * 64, + "_baselineSha256": "5" * 64, + "_profileSha256": "3" * 64, + "_controllerArtifactSha256": "f" * 64, + "_validationEvidenceSha256": "4" * 64, + } + binding = module._hub_delivery_signed_binding( + binding_snapshot, hashlib.sha256(KEY.encode()).hexdigest(), + ) + + status, queued_semantic, created = module._enqueue_hub_terminal_revalidation( + REVISION, project, binding, hashlib.sha256(raw).hexdigest(), + ) + assert (status, queued_semantic, created) == ("awaiting_validation", semantic, True) + assert original_path.read_bytes() == raw + sibling = json.loads(sibling_path.read_text()) + assert sibling["hub_delivery_revalidation"] == binding + assert sibling["revalidation_of"] == semantic + + sibling["status"] = "deployed" + sibling_path.write_bytes(module._hub_delivery_canonical(sibling) + b"\n") + status, queued_semantic, created = module._enqueue_hub_terminal_revalidation( + REVISION, project, binding, hashlib.sha256(raw).hexdigest(), + ) + assert (status, queued_semantic, created) == ("deployed", semantic, False) + assert original_path.read_bytes() == raw + with pytest.raises(module.HubDeliveryRefused): + module._hub_delivery_existing_revalidation( + REVISION, hashlib.sha256(b"different-idempotency-key").hexdigest(), "a" * 64, + ) + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX inbox permission contract") +@pytest.mark.parametrize(("root_mode", "file_mode"), [(0o700, 0o644), (0o755, 0o600)]) +def test_hub_original_reader_refuses_public_fixture_permissions( + controller, monkeypatch, tmp_path, root_mode, file_mode, +): + module = controller + semantic = "c" * 64 + raw = b'{"fixture":true}\n' + original = tmp_path / f"{semantic}.json" + original.write_bytes(raw) + original.chmod(0o600) + tmp_path.chmod(0o700) + monkeypatch.setattr(module, "WEBHOOK_INBOX_ROOT", tmp_path) + assert module._hub_delivery_read_original(semantic) == ( + {"fixture": True}, hashlib.sha256(raw).hexdigest(), + ) + original.chmod(file_mode) + tmp_path.chmod(root_mode) + with pytest.raises(module.HubDeliveryRefused): + module._hub_delivery_read_original(semantic) + assert original.read_bytes() == raw + + +def test_controller_snapshot_binds_full_profile_state_and_intent(controller, monkeypatch): + module = controller + revision = REVISION + project = { + "slug": "itworx-mcp-hub", + "branch": "main", + "gitea_repository_id": 314, + "auto_apply": True, + "environment": {"APPOPS_SOURCE_REVISION": "{revision}"}, + } + policy = {"validation_gate_version": 1, "contexts": [module.HUB_DELIVERY_FULL_CONTEXT], + "timeout_seconds": 60, "poll_interval_seconds": 2, "max_poll_interval_seconds": 30} + semantic = module._webhook_semantic_digest(314, "refs/heads/main", revision) + original = {"schema": module._WEBHOOK_SCHEMA, "semantic_sha256": semantic, + "body_sha256": "d" * 64, "delivery": "01234567-89ab-4cde-8fab-0123456789ab", "gitea_repository_id": 314, + "ref": "refs/heads/main", "repository": REPOSITORY, "revision": revision, + "source": "gitea_webhook", "status": "validation_timeout", "reason": "validation_timeout", + "validation": { + "validation_gate_version": 1, + "required_contexts": [module.HUB_DELIVERY_FULL_CONTEXT], + "timeout_seconds": 60, + "poll_interval_seconds": 2, + "max_poll_interval_seconds": 30, + }, + "audit": [{"stage": "validation_timeout"}]} + original_bytes = module._hub_delivery_canonical(original) + b"\n" + original_sha = hashlib.sha256(original_bytes).hexdigest() + monkeypatch.setattr(module, "_load_projects", lambda: {REPOSITORY: project}) + monkeypatch.setattr(module, "_webhook_policy", lambda *_a: (314, "main", "secret")) + monkeypatch.setattr(module, "_validation_policy", lambda *_a: policy) + monkeypatch.setattr(module, "_is_current_branch_head", lambda *_a: True) + monkeypatch.setattr(module, "_hub_delivery_read_original", lambda *_a: (original, original_sha)) + monkeypatch.setattr(module, "_hub_delivery_validation_evidence", lambda *_a: { + "revision": revision, + "branch": "main", + "event": "push", + "workflowPath": "managed-validation.yml@refs/heads/main", + "validationContext": module.HUB_DELIVERY_FULL_CONTEXT, + "runId": 22495, + "runNumber": 1180, + "runAttempt": 1, + "jobId": 39309, + "releaseJobId": 39310, + "jobCount": 2, + "stepCount": 12, + "stepSummarySha256": "6" * 64, + "statusId": 4, + "statusUpdatedAt": "2026-10-08T00:05:00.000Z", + }) + monkeypatch.setattr(module._mcp_hub_release_store, "read_hub_state_bytes", lambda **_k: b"{\"state\":1}") + monkeypatch.setitem(sys.modules, "deploy", SimpleNamespace(_mcp_hub_read_migration_intent=lambda: None)) + monkeypatch.setattr(module, "_self_upgrade_secret", lambda: b"hub-test-key-" + b"x" * 40) + + first = module._hub_delivery_snapshot(revision) + project["environment"]["APPOPS_SOURCE_REVISION"] = "changed" + second = module._hub_delivery_snapshot(revision) + assert first["preflightSha256"] != second["preflightSha256"] + assert first["_originalFileSha256"] == original_sha + assert first["_stateSha256"] == hashlib.sha256(b"{\"state\":1}").hexdigest() + project["environment"]["APPOPS_SOURCE_REVISION"] = "{revision}" + original["audit"].append({"stage": "deployment_started"}) + with pytest.raises(module.HubDeliveryRefused): + module._hub_delivery_snapshot(revision) + + +def test_hub_delivery_client_and_mcp_tool_keep_closed_contract(monkeypatch): + from backend.app import mcp_server + from backend.app.services.deployment_broker import DeploymentBrokerClient, settings + + monkeypatch.setattr(settings, "mcp_token", "t" * 48) + client = DeploymentBrokerClient(base_url="https://controller.example", verify_tls=True) + requests = [] + monkeypatch.setattr( + client, + "_request", + lambda *args, **kwargs: requests.append((args, kwargs)) or {"status": "ready", "noMutation": True}, + ) + preflight_payload = {"revision": REVISION} + assert client.hub_delivery_operation("preflight", preflight_payload)["status"] == "ready" + assert requests[-1][0][:2] == ("POST", PREFIX + "preflight") + assert requests[-1][1]["payload"] == preflight_payload + revalidate_payload = { + "revision": REVISION, + "expectedPreflightSha256": "a" * 64, + "idempotencyKey": KEY, + } + assert client.hub_delivery_operation("revalidate", revalidate_payload)["status"] == "ready" + assert requests[-1][0][:2] == ("POST", PREFIX + "revalidate") + assert requests[-1][1]["payload"] == revalidate_payload + monkeypatch.setattr( + client, "_request", lambda *_a, **_k: {"status": "ready", "noMutation": False}, + ) + assert client.hub_delivery_operation("preflight", preflight_payload) == { + "status": "unknown", "reason": "hub_delivery_preflight_mutation_unexpected", "noMutation": False, + } + with pytest.raises(ValueError): + client.hub_delivery_operation("preflight", {**preflight_payload, "repository": REPOSITORY}) + with pytest.raises(ValueError): + client.hub_delivery_operation("revalidate", {**revalidate_payload, "idempotencyKey": "short"}) + + # A transport failure after an operation request is outcome-unknown. The + # fixed client makes one request and never retries it implicitly. + calls = [] + monkeypatch.setattr( + client, + "_request", + lambda *args, **kwargs: calls.append((args, kwargs)) or {"status": "unreachable"}, + ) + outcome = client.hub_delivery_operation("revalidate", revalidate_payload) + assert outcome == {"status": "unknown", "reason": "hub_delivery_outcome_unknown", "noMutation": False} + assert len(calls) == 1 + monkeypatch.setattr(client, "_request", lambda *_a, **_k: {"status": "server_error"}) + assert client.hub_delivery_operation("revalidate", revalidate_payload) == { + "status": "unknown", "reason": "hub_delivery_outcome_unknown", "noMutation": False, + } + + tool_calls = [] + monkeypatch.setattr( + mcp_server, + "DeploymentBrokerClient", + lambda: SimpleNamespace( + hub_delivery_operation=lambda stage, payload: tool_calls.append((stage, payload)) + or {"status": "ready", "noMutation": True}, + ), + ) + assert mcp_server.appops_hub_delivery_preflight(REVISION)["status"] == "ready" + with pytest.raises(ValueError): + mcp_server.appops_hub_delivery_revalidate(REVISION, "a" * 64, "short") + assert mcp_server.appops_hub_delivery_revalidate(REVISION, "a" * 64, KEY)["status"] == "ready" + assert tool_calls == [("preflight", preflight_payload), ("revalidate", revalidate_payload)] + + +def test_hub_deploy_gate_requires_the_exact_signed_deploying_sibling(controller, monkeypatch): + module = controller + project = { + "slug": module.HUB_DELIVERY_PROJECT_SLUG, + "branch": "main", + "gitea_repository_id": 7, + } + semantic = module._webhook_semantic_digest(7, "refs/heads/main", REVISION) + snapshot = {**_snapshot(module), "_semanticDigest": semantic} + binding = module._hub_delivery_signed_binding( + snapshot, hashlib.sha256(KEY.encode()).hexdigest(), + ) + attempt = { + "schema": module._WEBHOOK_REVALIDATION_SCHEMA, + "semantic_sha256": semantic, + "revalidation_of": semantic, + "repository": REPOSITORY, + "revision": REVISION, + "ref": "refs/heads/main", + "gitea_repository_id": 7, + "status": "deploying", + "audit": [{"stage": "deployment_started"}], + "hub_delivery_revalidation": binding, + } + monkeypatch.setattr(module, "_load_projects", lambda: {REPOSITORY: project}) + monkeypatch.setattr(module, "_webhook_policy", lambda *_a: (7, "main")) + monkeypatch.setattr(module, "_read_inbox_record", lambda marker: attempt) + monkeypatch.setattr(module, "_hub_delivery_snapshot", lambda _revision: snapshot) + + assert module._hub_delivery_deploy_gate(REPOSITORY, REVISION, "a" * 64) + assert not module._hub_delivery_deploy_gate("jens/other", REVISION, "a" * 64) + assert not module._hub_delivery_deploy_gate(REPOSITORY, REVISION, "short") + + attempt["status"] = "validation_passed" + assert not module._hub_delivery_deploy_gate(REPOSITORY, REVISION, "a" * 64) + attempt["status"] = "deploying" + snapshot["preflightSha256"] = "e" * 64 + assert not module._hub_delivery_deploy_gate(REPOSITORY, REVISION, "a" * 64) + + +def test_deploy_hub_preflight_binding_is_fixed_and_fails_closed(controller, monkeypatch): + deploy = _load_module("hub_delivery_locked_deploy_gate_test", BROKER_ROOT / "deploy.py") + calls = [] + gate = SimpleNamespace( + _hub_delivery_deploy_gate=lambda *args: calls.append(args) or True, + ) + monkeypatch.setitem(sys.modules, "status_server", gate) + + deploy._verify_hub_delivery_preflight(REPOSITORY, REVISION, "a" * 64) + assert calls == [(REPOSITORY, REVISION, "a" * 64)] + for repository, revision, fingerprint in ( + ("jens/other", REVISION, "a" * 64), + (REPOSITORY, "invalid", "a" * 64), + (REPOSITORY, REVISION, "invalid"), + ): + with pytest.raises(deploy.DeployError, match="preflight binding is invalid"): + deploy._verify_hub_delivery_preflight(repository, revision, fingerprint) + assert len(calls) == 1 + + gate._hub_delivery_deploy_gate = lambda *_a: False + with pytest.raises(deploy.DeployError, match="preflight changed before deployment"): + deploy._verify_hub_delivery_preflight(REPOSITORY, REVISION, "a" * 64) + + monkeypatch.setenv(deploy.HUB_DELIVERY_PREFLIGHT_ENV, "a" * 64) + monkeypatch.setattr(sys, "argv", ["deploy.py", "describe", "jens/other", REVISION]) + with pytest.raises(deploy.DeployError, match="preflight binding is invalid"): + deploy.main() + + +def test_run_registered_deploy_passes_only_the_fixed_hub_binding(controller, monkeypatch): + module = controller + captured = [] + monkeypatch.setattr(module, "_load_projects", lambda: {REPOSITORY: {"auto_apply": True}}) + monkeypatch.setenv("APPOPS_HUB_DELIVERY_PREFLIGHT_SHA256", "c" * 64) + monkeypatch.setenv("APPOPS_PROJECTBRAIN_DELIVERY_PREFLIGHT_SHA256", "d" * 64) + monkeypatch.setattr( + module.subprocess, "run", + lambda argv, **kwargs: captured.append((argv, kwargs)) or SimpleNamespace(returncode=0), + ) + + module._run_registered_deploy( + REPOSITORY, REVISION, hub_delivery_preflight_sha256="a" * 64, + ) + + argv, kwargs = captured[0] + assert argv[-3:] == ["deploy", REPOSITORY, REVISION] + assert kwargs["env"]["APPOPS_HUB_DELIVERY_PREFLIGHT_SHA256"] == "a" * 64 + assert "APPOPS_PROJECTBRAIN_DELIVERY_PREFLIGHT_SHA256" not in kwargs["env"] + with pytest.raises(ValueError, match="Hub delivery binding invalid"): + module._run_registered_deploy("jens/other", REVISION, hub_delivery_preflight_sha256="a" * 64) + with pytest.raises(ValueError, match="Hub delivery binding invalid"): + module._run_registered_deploy(REPOSITORY, REVISION, hub_delivery_preflight_sha256="short") + + +def test_deploy_hub_preflight_is_rechecked_at_both_locked_boundaries(monkeypatch, tmp_path): + deploy = _load_module("hub_delivery_two_boundary_deploy_test", BROKER_ROOT / "deploy.py") + repository_root = tmp_path / "repository" + repository_root.mkdir() + _git(repository_root, "init", "-b", "main") + _git(repository_root, "config", "user.email", "hub-delivery-test@example.invalid") + _git(repository_root, "config", "user.name", "Hub delivery test") + source_file = repository_root / "README.md" + source_file.write_text("prior live source\n", encoding="utf-8") + _git(repository_root, "add", ".") + _git(repository_root, "commit", "-m", "prior") + prior_revision = _git(repository_root, "rev-parse", "HEAD") + source_file.write_text("candidate source\n", encoding="utf-8") + _git(repository_root, "add", ".") + _git(repository_root, "commit", "-m", "candidate") + target_revision = _git(repository_root, "rev-parse", "HEAD") + repository_git_dir = repository_root / ".git" + source_dir = tmp_path / "source" + deploy.sync_source(repository_git_dir, prior_revision, source_dir, "itworx-mcp-hub") + + project = { + "slug": "itworx-mcp-hub", + "branch": "main", + "source_dir": str(source_dir), + "containers": ["itworx-mcp-hub"], + "compose_project": "itworx-mcp-hub", + "compose_files": ["docker-compose.yml"], + } + previous_state = { + "schema_version": 3, + "repository": REPOSITORY, + "live_revision": prior_revision, + "live_status": "healthy", + "live_legacy_provenance": False, + "live_artifacts": {"containers": {"hub": {"image_id": "sha256:" + "a" * 64}}}, + "live_receipt": {"revision": prior_revision, "testOnly": True}, + } + active_locks = [] + project_lock = deploy.STATE_ROOT / "locks" / "itworx-mcp-hub.lock" + + @contextmanager + def lock(path, _description): + active_locks.append(path) + try: + yield + finally: + active_locks.pop() + + monkeypatch.setattr(deploy, "exclusive_lock", lock) + monkeypatch.setattr(deploy, "exclusive_global_deploy_lock", lambda: lock( + deploy.GLOBAL_DEPLOY_LOCK, "global", + )) + monkeypatch.setattr(deploy, "load_projects", lambda: {REPOSITORY: dict(project)}) + monkeypatch.setattr(deploy, "effective_mcp_hub_project", lambda _repo, config: config) + monkeypatch.setattr(deploy, "repository_path", lambda *_a: repository_git_dir) + monkeypatch.setattr(deploy, "bind_appops_self_upgrade_recovery", lambda *_a: None) + monkeypatch.setattr(deploy, "bind_registered_legacy_request", lambda *_a: None) + monkeypatch.setattr(deploy, "mcp_hub_prepare_current_receipt_model", lambda *_a: None) + monkeypatch.setattr(deploy, "validate_commit", lambda *_a: target_revision) + monkeypatch.setattr(deploy, "deployment_metadata", lambda *_a: { + "revision": target_revision, + "tree": "f" * 40, + }) + monkeypatch.setattr(deploy, "validate_project_config", lambda *_a: None) + monkeypatch.setattr(deploy, "read_state", lambda _project: dict(previous_state)) + monkeypatch.setattr(deploy, "verified_current_receipt", lambda *_a: None) + monkeypatch.setattr(deploy, "validate_forward_revision", lambda *_a: None) + monkeypatch.setattr(deploy, "_mcp_hub_assert_no_unresolved_migration_intent", lambda *_a: None) + + source_syncs = [] + actual_sync_source = deploy.sync_source + + def record_sync(repo, revision, target, slug, clean_untracked_paths=None, *, exact=False): + source_syncs.append((revision, tuple(active_locks))) + return actual_sync_source( + repo, revision, target, slug, clean_untracked_paths, exact=exact, + ) + + monkeypatch.setattr(deploy, "sync_source", record_sync) + monkeypatch.setattr(deploy, "copy_environment", lambda *_a, **_k: None) + gate_calls = [] + gate_results = iter((True, False)) + + def gate(repository, revision, fingerprint): + gate_calls.append((repository, revision, fingerprint, tuple(active_locks))) + return next(gate_results) + + monkeypatch.setitem(sys.modules, "status_server", SimpleNamespace( + _hub_delivery_deploy_gate=gate, + )) + state_updates = [] + monkeypatch.setattr( + deploy, "write_state", + lambda repository, config, revision, status, message, **kwargs: state_updates.append({ + "repository": repository, + "project": config, + "revision": revision, + "status": status, + "message": message, + **kwargs, + }), + ) + monkeypatch.setattr( + deploy, "capture_compose_rollback", + lambda *_a: pytest.fail("production snapshot must not run after stale Hub proof"), + ) + monkeypatch.setattr( + deploy, "deploy_project", + lambda *_a: pytest.fail("production mutation must not run after stale Hub proof"), + ) + monkeypatch.delenv(deploy.PROJECTBRAIN_DELIVERY_PREFLIGHT_ENV, raising=False) + monkeypatch.setenv(deploy.HUB_DELIVERY_PREFLIGHT_ENV, "b" * 64) + monkeypatch.setattr(sys, "argv", ["deploy.py", "deploy", REPOSITORY, target_revision]) + + with pytest.raises(deploy.DeployError, match="preflight changed before deployment"): + deploy.main() + + assert gate_calls == [ + (REPOSITORY, target_revision, "b" * 64, (project_lock, deploy.GLOBAL_DEPLOY_LOCK)), + (REPOSITORY, target_revision, "b" * 64, (project_lock, deploy.GLOBAL_DEPLOY_LOCK)), + ] + assert source_syncs == [ + (target_revision, (project_lock,)), + (prior_revision, (project_lock, deploy.GLOBAL_DEPLOY_LOCK)), + ] + assert source_dir.joinpath("README.md").read_text(encoding="utf-8") == "prior live source\n" + assert len(state_updates) == 1 + assert state_updates[0]["status"] == "failed" + assert state_updates[0]["live_revision"] == prior_revision + assert state_updates[0]["live_status"] == "healthy" diff --git a/backend/tests/test_projectbrain_delivery_revalidation.py b/backend/tests/test_projectbrain_delivery_revalidation.py index fcd5a4f..e244cc0 100644 --- a/backend/tests/test_projectbrain_delivery_revalidation.py +++ b/backend/tests/test_projectbrain_delivery_revalidation.py @@ -5,6 +5,7 @@ import io import json import subprocess import sys +import threading import time import uuid from concurrent.futures import ThreadPoolExecutor @@ -186,7 +187,11 @@ def test_revalidate_rejects_stale_fingerprint_without_queueing(controller, monke ) assert code == 409 - assert result == {"status": "blocked", "reason": "projectbrain_delivery_preflight_changed"} + assert result == { + "status": "blocked", + "reason": "projectbrain_delivery_preflight_changed", + "noMutation": True, + } assert queued == [] @@ -876,6 +881,29 @@ def test_concurrent_enqueue_creates_one_additive_attempt_and_preserves_original( preflight, hashlib.sha256(b"pb-revalidate-concurrent-2026").hexdigest(), ) project = fixture["project"] + marker = fixture["inbox"] / f"{fixture['semantic']}.revalidation.json" + writer_opened = threading.Event() + allow_writer_to_continue = threading.Event() + gate_lock = threading.Lock() + gate_used = False + real_open = controller.os.open + + def gated_open(path, flags, mode=0o777): + nonlocal gate_used + descriptor = real_open(path, flags, mode) + name = Path(path).name + if name == marker.name or name.startswith(f".{marker.name}."): + with gate_lock: + pause = not gate_used + gate_used = True + if pause: + writer_opened.set() + if not allow_writer_to_continue.wait(timeout=5): + controller.os.close(descriptor) + raise TimeoutError("test writer gate timed out") + return descriptor + + monkeypatch.setattr(controller.os, "open", gated_open) def enqueue(): return controller._enqueue_terminal_revalidation( @@ -887,14 +915,173 @@ def test_concurrent_enqueue_creates_one_additive_attempt_and_preserves_original( ) with ThreadPoolExecutor(max_workers=2) as workers: - results = list(workers.map(lambda _index: enqueue(), range(2))) + first = workers.submit(enqueue) + assert writer_opened.wait(timeout=5) + try: + assert not marker.exists(), "an incomplete revalidation marker became visible" + second = workers.submit(enqueue) + assert second.result(timeout=5) == ("awaiting_validation", fixture["semantic"]) + finally: + allow_writer_to_continue.set() + assert first.result(timeout=5) == ("awaiting_validation", fixture["semantic"]) - assert results == [("awaiting_validation", fixture["semantic"])] * 2 assert fixture["original_path"].read_bytes() == fixture["original_bytes"] attempt_files = list(fixture["inbox"].glob("*.revalidation.json")) assert len(attempt_files) == 1 record = controller._read_inbox_record(attempt_files[0]) assert record["projectbrain_delivery_revalidation"] == binding + assert not list(fixture["inbox"].glob(".*.tmp")) + + +def test_new_inbox_record_ignores_stale_temporary_file(controller, tmp_path): + marker = tmp_path / "stale.json" + record = {"record": "complete"} + stale = marker.with_name( + f".{marker.name}.{controller.os.getpid()}.{controller.threading.get_ident()}.tmp" + ) + stale_bytes = b"stale temporary file from an earlier process\n" + stale.write_bytes(stale_bytes) + + controller._write_new_inbox_record(marker, record) + + assert json.loads(marker.read_text(encoding="utf-8")) == record + if controller.os.name != "nt": + assert marker.stat().st_mode & 0o777 == 0o600 + assert stale.read_bytes() == stale_bytes + assert set(tmp_path.iterdir()) == {marker, stale} + + +def test_new_inbox_record_collision_preserves_existing_bytes_and_cleans_temp( + controller, tmp_path, +): + marker = tmp_path / "existing.json" + original = b'{"existing":true}\n' + marker.write_bytes(original) + + with pytest.raises(FileExistsError): + controller._write_new_inbox_record(marker, {"replacement": True}) + + assert marker.read_bytes() == original + assert not list(tmp_path.glob(".*.tmp")) + + +def test_new_inbox_record_write_failure_cleans_temporary_file( + controller, monkeypatch, tmp_path, +): + marker = tmp_path / "failed.json" + real_fdopen = controller.os.fdopen + + class FailingWriter: + def __init__(self, handle): + self.handle = handle + + def __enter__(self): + self.handle.__enter__() + return self + + def __exit__(self, exc_type, exc, traceback): + return self.handle.__exit__(exc_type, exc, traceback) + + def write(self, _payload): + raise OSError("simulated inbox write failure") + + monkeypatch.setattr( + controller.os, + "fdopen", + lambda descriptor, mode: FailingWriter(real_fdopen(descriptor, mode)), + ) + + with pytest.raises(OSError, match="simulated inbox write failure"): + controller._write_new_inbox_record(marker, {"record": True}) + + assert not marker.exists() + assert not list(tmp_path.glob(".*.tmp")) + + +@pytest.mark.parametrize("failure_point", ["directory_fsync", "wakeup"]) +def test_publish_followup_failure_reports_unknown_and_same_key_replay_resolves( + controller, monkeypatch, tmp_path, failure_point, +): + fixture = _snapshot_fixture(controller, monkeypatch, tmp_path) + monkeypatch.setattr(controller, "_projectbrain_delivery_mutation_lock", _null_context) + preflight = controller._projectbrain_delivery_snapshot(fixture["revision"]) + idempotency_key = "pb-revalidate-publish-failure-2026" + payload = { + "revision": fixture["revision"], + "expectedPreflightSha256": preflight["preflightSha256"], + "idempotencyKey": idempotency_key, + } + real_fsync = controller._fsync_inbox_root + real_wakeup = controller._webhook_wakeup + + if failure_point == "directory_fsync": + def fail_fsync(): + raise OSError("simulated directory fsync failure") + + monkeypatch.setattr(controller, "_fsync_inbox_root", fail_fsync) + else: + def fail_wakeup(): + raise RuntimeError("simulated worker wakeup failure") + + monkeypatch.setattr(controller, "_webhook_wakeup", SimpleNamespace(set=fail_wakeup)) + + code, result = invoke(controller, "revalidate", payload) + + assert code == 503 + assert result == { + "status": "unknown", + "reason": "projectbrain_delivery_enqueue_outcome_unknown", + "noMutation": False, + } + assert fixture["original_path"].read_bytes() == fixture["original_bytes"] + marker = fixture["inbox"] / f"{fixture['semantic']}.revalidation.json" + record = controller._read_inbox_record(marker) + assert record is not None + assert record["schema"] == controller._WEBHOOK_REVALIDATION_SCHEMA + assert record["status"] == "awaiting_validation" + assert record["projectbrain_delivery_revalidation"]["idempotencyKeySha256"] == hashlib.sha256( + idempotency_key.encode("ascii"), + ).hexdigest() + assert not list(fixture["inbox"].glob(".*.tmp")) + + if failure_point == "directory_fsync": + monkeypatch.setattr(controller, "_fsync_inbox_root", real_fsync) + else: + monkeypatch.setattr(controller, "_webhook_wakeup", real_wakeup) + replay_code, replay = invoke(controller, "revalidate", payload) + + assert replay_code == 202 + assert replay["status"] == "awaiting_validation" + assert replay["delivery"] == fixture["semantic"] + assert fixture["original_path"].read_bytes() == fixture["original_bytes"] + + +def test_revalidation_preflight_refusal_reports_no_mutation( + controller, monkeypatch, tmp_path, +): + fixture = _snapshot_fixture(controller, monkeypatch, tmp_path) + monkeypatch.setattr(controller, "_projectbrain_delivery_mutation_lock", _null_context) + + def refuse_snapshot(_revision): + raise controller.ProjectBrainDeliveryRefused("projectbrain_delivery_preflight_changed") + + monkeypatch.setattr(controller, "_projectbrain_delivery_snapshot", refuse_snapshot) + payload = { + "revision": fixture["revision"], + "expectedPreflightSha256": "a" * 64, + "idempotencyKey": "pb-revalidate-preflight-refusal-2026", + } + + code, result = invoke(controller, "revalidate", payload) + + assert code == 409 + assert result == { + "status": "blocked", + "reason": "projectbrain_delivery_preflight_changed", + "noMutation": True, + } + assert fixture["original_path"].read_bytes() == fixture["original_bytes"] + assert not (fixture["inbox"] / f"{fixture['semantic']}.revalidation.json").exists() def test_generic_terminal_revalidation_cannot_occupy_projectbrain_sibling(