"""Server-side Harbor immutability contracts for Hermes agent releases.""" from __future__ import annotations import importlib.util import json from pathlib import Path import pytest import yaml REPO_ROOT = Path(__file__).resolve().parents[2] SCRIPT = ( REPO_ROOT / "services/harbor/scripts/harbor_hermes_agent_immutability_ensure.py" ) def _load_module(): spec = importlib.util.spec_from_file_location("harbor_immutability", SCRIPT) 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 def _rule(module, *, rule_id: int = 17, disabled: bool = False) -> dict: return {"id": rule_id, **module.EXPECTED_RULE, "disabled": disabled} def _robot(module, *, immutable: bool = False, extra_immutable: bool = False) -> dict: access = [ {"resource": "repository", "action": "pull", "effect": "allow"}, {"resource": "repository", "action": "push", "effect": "allow"}, ] if immutable: access.append({"resource": "immutable-tag", "action": "list"}) if extra_immutable: access.append({"resource": "immutable-tag", "action": "delete"}) return { "id": 41, "name": module.PUBLISH_ROBOT, "description": "Jenkins publisher", "level": "system", "duration": -1, "editable": True, "disable": False, "permissions": [ { "kind": "project", "namespace": "other", "access": [{"resource": "repository", "action": "pull"}], }, {"kind": "project", "namespace": "bstein", "access": access}, ], } def _count(value: int) -> dict[str, str]: return {"X-Total-Count": str(value)} class FakeClient: """Small deterministic Harbor API fake.""" def __init__(self, responses): self.origin = "https://registry.bstein.dev/api/v2.0" self.responses = list(responses) self.calls = [] def request(self, method, path, payload=None): self.calls.append((method, path, payload)) return self.responses.pop(0) def test_flux_tracks_exact_immutable_rule_before_jenkins() -> None: """The Harbor rule is a reviewed prerequisite, not a Jenkins preflight only.""" harbor = yaml.safe_load( ( REPO_ROOT / "clusters/atlas/flux-system/applications/harbor/kustomization.yaml" ).read_text() ) jenkins = yaml.safe_load( ( REPO_ROOT / "clusters/atlas/flux-system/applications/jenkins/kustomization.yaml" ).read_text() ) check = { "apiVersion": "batch/v1", "kind": "Job", "name": "harbor-hermes-agent-immutability-ensure-1", "namespace": "harbor", } assert check in harbor["spec"]["healthChecks"] assert "harbor" in {item["name"] for item in jenkins["spec"]["dependsOn"]} def test_policy_job_uses_runtime_vault_secret_and_hardened_pinned_image() -> None: """No Harbor credential is committed or retained in a mutable workload.""" job = yaml.safe_load( (REPO_ROOT / "services/harbor/hermes-agent-immutability-job.yaml").read_text() ) template = job["spec"]["template"] annotations = template["metadata"]["annotations"] assert annotations["vault.hashicorp.com/role"] == "harbor-policy-bootstrap" assert ( annotations["vault.hashicorp.com/agent-inject-secret-harbor-admin-password"] == "kv/data/atlas/harbor/harbor-core" ) pod = template["spec"] assert pod["serviceAccountName"] == "harbor-policy-bootstrap" assert pod["enableServiceLinks"] is False container = pod["containers"][0] assert "@sha256:" in container["image"] security = container["securityContext"] assert security["runAsNonRoot"] is True assert security["readOnlyRootFilesystem"] is True assert security["allowPrivilegeEscalation"] is False assert security["capabilities"]["drop"] == ["ALL"] vault = ( REPO_ROOT / "services/vault/scripts/vault_k8s_auth_configure.sh" ).read_text() policy = vault.split("harbor_policy_bootstrap_policy='", 1)[1].split("'", 1)[0] assert 'path "kv/data/atlas/harbor/harbor-core"' in policy assert 'capabilities = ["read"]' in policy assert "*" not in policy assert 'bound_service_account_names="harbor-policy-bootstrap"' in vault script = SCRIPT.read_text() assert 'PUBLISH_ROBOT = "robot$jenkins-pipelines"' in script assert '{"resource": "immutable-tag", "action": "list"}' in script assert 'client.request(\n "PUT", f"/robots/{robot_id}", payload' in script assert '"secret"' not in script.split("payload = {", 1)[1].split("}", 1)[0] def test_rule_contract_is_exact_repository_and_unique_build_tags() -> None: """Unrelated bstein repositories and ordinary Hermes tags stay mutable.""" module = _load_module() assert module.EXPECTED_RULE == { "disabled": False, "action": "immutable", "template": "immutable_template", "tag_selectors": [ { "kind": "doublestar", "decoration": "matches", "pattern": "git-*-build-*", } ], "scope_selectors": { "repository": [ { "kind": "doublestar", "decoration": "repoMatches", "pattern": "hermes-agent", } ] }, } def test_existing_exact_rule_is_idempotent_without_mutation() -> None: """A rerun only validates the exact enabled rule.""" module = _load_module() body = json.dumps([_rule(module)]).encode() client = FakeClient([(200, body, _count(1))]) assert module.ensure_rule(client) == 17 assert [call[0] for call in client.calls] == ["GET"] def test_create_requires_201_exact_location_and_verified_reread() -> None: """Policy bootstrap fails closed until Harbor returns the exact persisted rule.""" module = _load_module() body = json.dumps([_rule(module, rule_id=23)]).encode() client = FakeClient( [ (200, b"[]", _count(0)), ( 201, b"", {"Location": "/api/v2.0/projects/bstein/immutabletagrules/23"}, ), (200, body, _count(1)), ] ) assert module.ensure_rule(client) == 23 assert client.calls[1] == ( "POST", "/projects/bstein/immutabletagrules", module.EXPECTED_RULE, ) @pytest.mark.parametrize( "responses,match", [ ([(200, b"[]", _count(0)), (200, b"", {})], "HTTP 200"), ( [ (200, b"[]", _count(0)), (201, b"", {"Location": "https://evil.invalid/1"}), ], "Location", ), ], ) def test_create_rejects_noncanonical_responses(responses, match: str) -> None: """Proxy success pages and foreign locations can never count as enforcement.""" module = _load_module() with pytest.raises(RuntimeError, match=match): module.ensure_rule(FakeClient(responses)) def test_readiness_failures_are_distinct_from_policy_rejections() -> None: """The tracked Job may wait for Harbor without retrying an auth denial.""" module = _load_module() with pytest.raises(module.HarborUnavailable): module.ensure_rule(FakeClient([(503, b"", {})])) with pytest.raises(RuntimeError, match="HTTP 403") as exc: module.ensure_rule(FakeClient([(403, b"", {})])) assert not isinstance(exc.value, module.HarborUnavailable) def test_disabled_or_duplicate_exact_scope_fails_closed() -> None: """Bootstrap never silently edits a conflicting security policy.""" module = _load_module() disabled = json.dumps([_rule(module, disabled=True)]).encode() with pytest.raises(RuntimeError, match="not enabled"): module.ensure_rule(FakeClient([(200, disabled, _count(1))])) duplicate = json.dumps([_rule(module), _rule(module, rule_id=18)]).encode() with pytest.raises(RuntimeError, match="multiple"): module.ensure_rule(FakeClient([(200, duplicate, _count(2))])) def test_policy_lists_reject_missing_or_truncated_count_evidence() -> None: """A hidden second page can never produce a duplicate rule or robot update.""" module = _load_module() body = json.dumps([_rule(module)]).encode() for headers in ({}, _count(2)): with pytest.raises(RuntimeError, match="count|truncated"): module.ensure_rule(FakeClient([(200, body, headers)])) robot_body = json.dumps( [{"id": 41, "name": module.PUBLISH_ROBOT}] ).encode() with pytest.raises(RuntimeError, match="truncated"): module.ensure_publisher_can_read_rule( FakeClient([(200, robot_body, _count(2))]) ) def test_publisher_policy_adds_only_read_access_and_preserves_robot() -> None: """Bootstrap preserves every existing scope and never touches robot secret state.""" module = _load_module() original = _robot(module) persisted = _robot(module, immutable=True) client = FakeClient( [ ( 200, json.dumps([{"id": 41, "name": module.PUBLISH_ROBOT}]).encode(), _count(1), ), (200, json.dumps(original).encode(), {}), (200, b"", {}), (200, json.dumps(persisted).encode(), {}), ] ) assert module.ensure_publisher_can_read_rule(client) == 41 method, path, payload = client.calls[2] assert (method, path) == ("PUT", "/robots/41") assert "secret" not in payload assert payload["permissions"][0] == original["permissions"][0] assert payload["permissions"][1]["access"][-1] == { "resource": "immutable-tag", "action": "list", } def test_publisher_policy_is_idempotent_and_rejects_broad_access() -> None: """An exact read policy is stable; mutation-capable immutable access is blocked.""" module = _load_module() exact = _robot(module, immutable=True) client = FakeClient( [ ( 200, json.dumps([{"id": 41, "name": module.PUBLISH_ROBOT}]).encode(), _count(1), ), (200, json.dumps(exact).encode(), {}), ] ) assert module.ensure_publisher_can_read_rule(client) == 41 assert [call[0] for call in client.calls] == ["GET", "GET"] broad = _robot(module, immutable=True, extra_immutable=True) client = FakeClient( [ ( 200, json.dumps([{"id": 41, "name": module.PUBLISH_ROBOT}]).encode(), _count(1), ), (200, json.dumps(broad).encode(), {}), ] ) with pytest.raises(RuntimeError, match="broader"): module.ensure_publisher_can_read_rule(client) def test_harbor_client_rejects_plaintext_or_ambiguous_origins() -> None: """Runtime admin credentials are never sent over HTTP or a URL with query state.""" module = _load_module() for origin in ( "http://registry.bstein.dev/api/v2.0", "https://registry.bstein.dev/api/v2.0?next=evil", "https://other.invalid/api/v2.0", "", ): with pytest.raises(ValueError): module.HarborClient(origin, "admin", "secret")