diff --git a/services/harbor/scripts/harbor_hermes_agent_immutability_ensure.py b/services/harbor/scripts/harbor_hermes_agent_immutability_ensure.py index a313cc80..06475c50 100644 --- a/services/harbor/scripts/harbor_hermes_agent_immutability_ensure.py +++ b/services/harbor/scripts/harbor_hermes_agent_immutability_ensure.py @@ -374,5 +374,5 @@ def main() -> int: return 0 -if __name__ == "__main__": +if __name__ == "__main__": # pragma: no cover - exercised through main() raise SystemExit(main()) diff --git a/testing/tests/test_hermes_image_builder_branch_contract.py b/testing/tests/test_hermes_image_builder_branch_contract.py new file mode 100644 index 00000000..9ee6e107 --- /dev/null +++ b/testing/tests/test_hermes_image_builder_branch_contract.py @@ -0,0 +1,181 @@ +"""Exercise exact fail-closed branches in the reviewed Hermes image lane.""" + +from __future__ import annotations + +import importlib.util +import io +import json +from pathlib import Path + +import pytest + + +ROOT = Path(__file__).resolve().parents[2] +RELEASE = ROOT / "ci/scripts/hermes_image_release.py" +HARBOR = ROOT / "services/harbor/scripts/harbor_hermes_agent_immutability_ensure.py" + + +def _load(path: Path, name: str): + spec = importlib.util.spec_from_file_location(name, path) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +class Response(io.BytesIO): + """Small bounded response for same-process HTTP contract tests.""" + + def __init__( + self, + status: int, + body: bytes = b"", + headers: dict[str, str] | None = None, + ) -> None: + super().__init__(body) + self.status = status + self.headers = headers or {} + + def __enter__(self): + return self + + def __exit__(self, *_args): + self.close() + + +class FakeClient: + """Return one reviewed sequence of Harbor API responses.""" + + origin = "https://registry.bstein.dev/api/v2.0" + + def __init__(self, responses) -> None: + self.responses = list(responses) + + def request(self, _method, _path, _payload=None): + return self.responses.pop(0) + + +def _count(value: int) -> dict[str, str]: + return {"X-Total-Count": str(value)} + + +def _robot(module, *, immutable: bool, duration: int | None = None) -> dict: + access = [ + {"resource": "repository", "action": "pull"}, + {"resource": "repository", "action": "push"}, + ] + if immutable: + access.append({"resource": "immutable-tag", "action": "list"}) + robot = { + "id": 41, + "name": module.PUBLISH_ROBOT, + "description": "publisher", + "level": "system", + "editable": True, + "disable": False, + "permissions": [ + {"kind": "project", "namespace": module.PROJECT, "access": access} + ], + } + if duration is not None: + robot["duration"] = duration + return robot + + +def test_release_policy_response_edges_are_fail_closed() -> None: + """Credentials, bounds, page evidence, JSON shape, and redirects stay strict.""" + module = _load(RELEASE, "hermes_release_policy_branch_contract") + assert module._NoRedirect().redirect_request(None, None, 302, "", {}, "") is None + with pytest.raises(RuntimeError, match="credentials"): + module._immutable_rules_response(username="", password="private") + with pytest.raises(RuntimeError, match="size limit"): + module._immutable_rules_response( + username="robot", + password="private", + opener=lambda *_args: Response(200, b"x" * 1_048_577), + ) + with pytest.raises(RuntimeError, match="valid total count"): + module._require_complete_rule_page([], {}) + with pytest.raises(RuntimeError, match="invalid shape"): + module.verify_immutable_policy( + username="robot", + password="private", + opener=lambda *_args: Response(200, b"{}", _count(0)), + ) + + +def test_release_renderer_exhausts_a_trailing_incomplete_image() -> None: + """A trailing matching name cannot hide the one complete digest entry.""" + module = _load(RELEASE, "hermes_release_trailing_image_branch") + digest = "sha256:" + "9" * 64 + source = ( + "images:\n" + f" - name: {module.DEFAULT_IMAGE}\n" + " digest: sha256:" + "0" * 64 + "\n" + f" - name: {module.DEFAULT_IMAGE}\n" + ) + assert module.render_kustomization(source, digest).count(digest) == 1 + + +def test_harbor_publisher_handles_absent_duration_and_stale_reread( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """An absent optional duration is preserved while stale policy reads retry.""" + module = _load(HARBOR, "harbor_publisher_branch_contract") + current = _robot(module, immutable=False) + malformed_persisted = {**_robot(module, immutable=True), "permissions": "bad"} + exact = _robot(module, immutable=True) + client = FakeClient( + [ + (200, json.dumps([{"id": 41, "name": module.PUBLISH_ROBOT}]).encode(), _count(1)), + (200, json.dumps(current).encode(), {}), + (200, b"", {}), + (200, json.dumps(malformed_persisted).encode(), {}), + (200, json.dumps(exact).encode(), {}), + ] + ) + monkeypatch.setattr(module.time, "sleep", lambda _seconds: None) + assert module.ensure_publisher_can_read_rule(client) == 41 + + +def test_harbor_created_rule_retries_an_id_mismatch( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A stale exact-looking rule with the wrong ID is never accepted.""" + module = _load(HARBOR, "harbor_rule_id_retry_branch") + path = "/api/v2.0/projects/bstein/immutabletagrules/23" + wrong = {"id": 22, **module.EXPECTED_RULE} + exact = {"id": 23, **module.EXPECTED_RULE} + client = FakeClient( + [ + (200, b"[]", _count(0)), + (201, b"", {"Location": path}), + (200, json.dumps([wrong]).encode(), _count(1)), + (200, json.dumps([exact]).encode(), _count(1)), + ] + ) + monkeypatch.setattr(module.time, "sleep", lambda _seconds: None) + assert module.ensure_rule(client) == 23 + + +def test_harbor_main_covers_immediate_success_and_final_retry( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + """Startup exits directly on success and re-raises the final unavailable result.""" + module = _load(HARBOR, "harbor_main_branch_contract") + password = tmp_path / "password" + password.write_text("private\n", encoding="utf-8") + monkeypatch.setenv("HARBOR_API_ORIGIN", module.EXPECTED_ORIGIN) + monkeypatch.setenv("HARBOR_ADMIN_PASSWORD_FILE", str(password)) + monkeypatch.setattr(module, "HarborClient", lambda *_args: object()) + monkeypatch.setattr(module, "ensure_rule", lambda _client: 17) + monkeypatch.setattr(module, "ensure_publisher_can_read_rule", lambda _client: 41) + assert module.main() == 0 + + def unavailable(_client): + raise module.HarborUnavailable("still warming") + + monkeypatch.setattr(module, "ensure_rule", unavailable) + monkeypatch.setattr(module.time, "sleep", lambda _seconds: None) + with pytest.raises(module.HarborUnavailable, match="still warming"): + module.main() diff --git a/testing/tests/test_hermes_runtime_access.py b/testing/tests/test_hermes_runtime_access.py index 39022cb9..5e25e5c9 100644 --- a/testing/tests/test_hermes_runtime_access.py +++ b/testing/tests/test_hermes_runtime_access.py @@ -451,9 +451,8 @@ def test_manifests_never_seed_access_material_into_persistent_env(): assert triage_annotations[ "vault.hashicorp.com/agent-inject-secret-triage-api-key" ] == "kv/data/atlas/hermes/triage-api" - gitea_api = ( - ROOT / "services/hermes/scm-common/scripts/gitea_api.py" - ).read_text(encoding="utf-8") + gitea_path = ROOT / "services/hermes/scm-common/scripts/gitea_api.py" + gitea_api = gitea_path.read_text(encoding="utf-8") assert "scm_broker_client" in gitea_api assert "GITEA_TOKEN" not in gitea_api assert "gitea-token" not in annotations