hermes: add daemonless agent image release lane

ci/Jenkinsfile.hermes-agent-image

@@ -82,6 +82,7 @@ spec:
         allowPrivilegeEscalation: false
         capabilities:
           drop: ["ALL"]
+          add: ["CHOWN", "FOWNER", "DAC_OVERRIDE", "SETGID", "SETUID"]
         privileged: false
         runAsUser: 0
         seccompProfile:
@@ -203,6 +204,7 @@ spec:
                 --destination="${destination}" \
                 --digest-file="${WORKSPACE}/build/hermes-agent.digest" \
                 --image-name-tag-with-digest-file="${WORKSPACE}/build/hermes-agent.image" \
+                --build-arg=HERMES_KANIKO_HEREDOC_COMPAT=1 \
                 --label="org.opencontainers.image.revision=${EXPECTED_SOURCE_REVISION}" \
                 --cleanup \
                 --push-retry=3

ci/scripts/hermes_image_release.py

@@ -20,6 +20,8 @@ DEFAULT_IMAGE = "registry.bstein.dev/bstein/hermes-agent"
 HARBOR_API_ORIGIN = "https://registry.bstein.dev/api/v2.0"
 HARBOR_PROJECT = "bstein"
 HARBOR_REPOSITORY = "hermes-agent"
+IMMUTABLE_REPOSITORY_PATTERN = "hermes-agent"
+IMMUTABLE_TAG_PATTERN = "git-*-build-*"
 DIGEST_PATTERN = re.compile(r"^sha256:[0-9a-f]{64}$")
 REVISION_PATTERN = re.compile(r"^[0-9a-f]{40}$")
 BUILD_PATTERN = re.compile(r"^[1-9][0-9]*$")
@@ -101,7 +103,8 @@ def _artifact_response(
     auth = base64.b64encode(f"{username}:{password}".encode()).decode("ascii")
     request = urllib.request.Request(
         f"{HARBOR_API_ORIGIN}/projects/{HARBOR_PROJECT}/repositories/"
-        f"{HARBOR_REPOSITORY}/artifacts/{encoded_tag}",
+        f"{HARBOR_REPOSITORY}/artifacts/{encoded_tag}"
+        "?with_immutable_status=true",
         headers={"Accept": "application/json", "Authorization": f"Basic {auth}"},
         method="GET",
     )
@@ -112,6 +115,126 @@ def _artifact_response(
         return int(response.status), body
 
 
+def _immutable_rules_response(
+    *,
+    username: str,
+    password: str,
+    opener: Callable[[urllib.request.Request, int], Any] = _registry_request,
+) -> tuple[int, bytes, dict[str, str]]:
+    """Read the project policy with the same least-privilege publish identity."""
+    if not username or not password:
+        raise RuntimeError("Harbor credentials are empty")
+    auth = base64.b64encode(f"{username}:{password}".encode()).decode("ascii")
+    request = urllib.request.Request(
+        f"{HARBOR_API_ORIGIN}/projects/{HARBOR_PROJECT}/immutabletagrules"
+        "?page=1&page_size=100",
+        headers={"Accept": "application/json", "Authorization": f"Basic {auth}"},
+        method="GET",
+    )
+    with opener(request, 20) as response:
+        body = response.read(1_048_577)
+        if len(body) > 1_048_576:
+            raise RuntimeError("Harbor immutable rule response exceeded the size limit")
+        return int(response.status), body, dict(response.headers)
+
+
+def _require_complete_rule_page(
+    rules: list[dict[str, Any]], headers: dict[str, str]
+) -> None:
+    """Require proof that the bounded first page contains every rule."""
+    raw_total = next(
+        (value for key, value in headers.items() if key.lower() == "x-total-count"),
+        None,
+    )
+    if raw_total is None or not str(raw_total).isdecimal():
+        raise RuntimeError("Harbor immutable rule list omitted a valid total count")
+    if int(raw_total) != len(rules):
+        raise RuntimeError("Harbor immutable rule list was truncated")
+
+
+def _normalized_immutable_rule(rule: dict[str, Any]) -> dict[str, Any]:
+    """Select only fields that bind the server-side build-tag policy."""
+    return {
+        "disabled": bool(rule.get("disabled", False)),
+        "action": rule.get("action"),
+        "template": rule.get("template"),
+        "tag_selectors": [
+            {
+                "kind": item.get("kind"),
+                "decoration": item.get("decoration"),
+                "pattern": item.get("pattern"),
+            }
+            for item in rule.get("tag_selectors") or []
+            if isinstance(item, dict)
+        ],
+        "scope_selectors": {
+            "repository": [
+                {
+                    "kind": item.get("kind"),
+                    "decoration": item.get("decoration"),
+                    "pattern": item.get("pattern"),
+                }
+                for item in (rule.get("scope_selectors") or {}).get(
+                    "repository", []
+                )
+                if isinstance(item, dict)
+            ]
+        },
+    }
+
+
+def verify_immutable_policy(
+    *,
+    username: str,
+    password: str,
+    opener: Callable[[urllib.request.Request, int], Any] = _registry_request,
+) -> None:
+    """Fail closed before build unless the exact Harbor rule is active."""
+    status, body, headers = _immutable_rules_response(
+        username=username, password=password, opener=opener
+    )
+    if status != 200:
+        raise RuntimeError(f"Harbor immutable policy preflight returned HTTP {status}")
+    try:
+        rules = json.loads(body.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise RuntimeError("Harbor returned invalid immutable rule JSON") from exc
+    if not isinstance(rules, list) or not all(isinstance(item, dict) for item in rules):
+        raise RuntimeError("Harbor immutable rule list has an invalid shape")
+    _require_complete_rule_page(rules, headers)
+    expected = {
+        "disabled": False,
+        "action": "immutable",
+        "template": "immutable_template",
+        "tag_selectors": [
+            {
+                "kind": "doublestar",
+                "decoration": "matches",
+                "pattern": IMMUTABLE_TAG_PATTERN,
+            }
+        ],
+        "scope_selectors": {
+            "repository": [
+                {
+                    "kind": "doublestar",
+                    "decoration": "repoMatches",
+                    "pattern": IMMUTABLE_REPOSITORY_PATTERN,
+                }
+            ]
+        },
+    }
+    matches = [
+        _normalized_immutable_rule(item)
+        for item in rules
+        if _normalized_immutable_rule(item)["tag_selectors"]
+        == expected["tag_selectors"]
+        and _normalized_immutable_rule(item)["scope_selectors"]
+        == expected["scope_selectors"]
+    ]
+    if matches != [expected]:
+        raise RuntimeError("Harbor immutable build-tag policy is absent or not exact")
+
+
 def assert_tag_absent(
     destination: str,
     *,
@@ -155,13 +278,15 @@ def verify_registry_digest(
     if harbor_digest != digest:
         raise RuntimeError("Harbor digest does not match Kaniko evidence")
     expected_tag = destination.rsplit(":", 1)[1]
-    tag_names = {
-        str(item.get("name"))
+    matching_tags = [
+        item
         for item in artifact.get("tags") or []
-        if isinstance(item, dict) and item.get("name")
-    }
-    if expected_tag not in tag_names:
+        if isinstance(item, dict) and item.get("name") == expected_tag
+    ]
+    if len(matching_tags) != 1:
         raise RuntimeError("Harbor artifact does not contain the expected tag")
+    if matching_tags[0].get("immutable") is not True:
+        raise RuntimeError("Harbor did not enforce the expected tag as immutable")
 
 
 def render_kustomization(source: str, digest: str, image: str = DEFAULT_IMAGE) -> str:
@@ -274,6 +399,7 @@ def main() -> int:
     validate_destination(args.destination, args.source_revision, args.build_number)
     username, password = _credentials()
     if args.command == "assert-absent":
+        verify_immutable_policy(username=username, password=password)
         assert_tag_absent(args.destination, username=username, password=password)
         return 0
 

clusters/atlas/flux-system/applications/harbor/kustomization.yaml

@@ -15,7 +15,13 @@ spec:
     kind: GitRepository
     name: flux-system
     namespace: flux-system
-  wait: false
+  wait: true
+  timeout: 10m
+  healthChecks:
+    - apiVersion: batch/v1
+      kind: Job
+      name: harbor-hermes-agent-immutability-ensure-1
+      namespace: harbor
   dependsOn:
     - name: core
     - name: longhorn

clusters/atlas/flux-system/applications/jenkins/kustomization.yaml

@@ -17,7 +17,8 @@ spec:
   targetNamespace: jenkins
   dependsOn:
     - name: helm
-    - name: vault
+    - name: harbor
+    - name: vault-hermes-jenkins-token-seed
   healthChecks:
     - apiVersion: apps/v1
       kind: Deployment

clusters/atlas/flux-system/applications/kustomization.yaml

@@ -4,6 +4,7 @@ kind: Kustomization
 resources:
   - gitea/kustomization.yaml
   - vault/kustomization.yaml
+  - vault-hermes-jenkins-token-seed/kustomization.yaml
   - vaultwarden/kustomization.yaml
   - comms/kustomization.yaml
   - crypto/kustomization.yaml

clusters/atlas/flux-system/applications/vault-hermes-jenkins-token-seed/kustomization.yaml

@@ -0,0 +1,26 @@
+# clusters/atlas/flux-system/applications/vault-hermes-jenkins-token-seed/kustomization.yaml
+apiVersion: kustomize.toolkit.fluxcd.io/v1
+kind: Kustomization
+metadata:
+  name: vault-hermes-jenkins-token-seed
+  namespace: flux-system
+  annotations:
+    kustomize.toolkit.fluxcd.io/ssa: IfNotPresent
+spec:
+  interval: 10m
+  sourceRef:
+    kind: GitRepository
+    name: flux-system
+    namespace: flux-system
+  path: ./services/vault-hermes-jenkins-token-seed
+  targetNamespace: vault
+  prune: true
+  wait: true
+  timeout: 5m
+  dependsOn:
+    - name: vault
+  healthChecks:
+    - apiVersion: batch/v1
+      kind: Job
+      name: vault-hermes-jenkins-build-token-seed-2
+      namespace: vault

clusters/atlas/flux-system/applications/vault/kustomization.yaml

@@ -19,7 +19,7 @@ spec:
   healthChecks:
     - apiVersion: batch/v1
       kind: Job
-      name: vault-hermes-jenkins-build-token-seed-1
+      name: vault-k8s-auth-hermes-9
       namespace: vault
   dependsOn:
     - name: longhorn

dockerfiles/Dockerfile.hermes-agent

@@ -1433,6 +1433,16 @@ NODE
 
 COPY dockerfiles/hermes-session-migrate.py /opt/hermes/bin/hermes-session-migrate
 
+ARG HERMES_KANIKO_HEREDOC_COMPAT=0
+COPY dockerfiles/Dockerfile.hermes-agent /tmp/hermes-agent.Dockerfile
+COPY dockerfiles/hermes-kaniko-heredoc-runner.py /tmp/hermes-kaniko-heredoc-runner.py
+RUN case "${HERMES_KANIKO_HEREDOC_COMPAT}" in 0|1) ;; *) exit 2 ;; esac \
+    && if [ "${HERMES_KANIKO_HEREDOC_COMPAT}" = 1 ]; then \
+      python /tmp/hermes-kaniko-heredoc-runner.py \
+        --dockerfile /tmp/hermes-agent.Dockerfile; \
+    fi \
+    && rm -f /tmp/hermes-agent.Dockerfile /tmp/hermes-kaniko-heredoc-runner.py
+
 RUN cd /opt/hermes/web \
     && npm run build \
     && grep -Fq 'await api.getSessions(1, 0' src/pages/ChatPage.tsx \

dockerfiles/Dockerfile.hermes-agent.dockerignore

@@ -4,3 +4,5 @@
 !dockerfiles/hermes-public-extract/**
 !dockerfiles/hermes-session-activity-panel.tsx
 !dockerfiles/hermes-session-migrate.py
+!dockerfiles/Dockerfile.hermes-agent
+!dockerfiles/hermes-kaniko-heredoc-runner.py

dockerfiles/hermes-kaniko-heredoc-runner.py

@@ -0,0 +1,79 @@
+#!/usr/bin/env python3
+"""Replay the reviewed Hermes Dockerfile heredocs for pinned Kaniko.
+
+Kaniko v1.24 parses shell heredoc bodies into ``RunCommand.Files`` but its RUN
+implementation executes only ``CmdLine``. The interpreters therefore receive
+empty stdin and exit successfully. Docker and BuildKit execute these blocks
+normally, so this compatibility runner is enabled only by the Kaniko pipeline.
+"""
+
+from __future__ import annotations
+
+import argparse
+import subprocess
+from pathlib import Path
+
+
+MAX_DOCKERFILE_BYTES = 2_000_000
+BLOCKS = (
+    ("RUN node <<'NODE'", "NODE", ("node",)),
+    ("RUN python - <<'PY'", "PY", ("python", "-")),
+)
+EXPECTED_COMMANDS = ("node", *("python",) * 7, "node")
+
+
+def extract_blocks(source: str) -> list[tuple[tuple[str, ...], str]]:
+    """Return only exact supported RUN heredocs in their reviewed order."""
+    markers = {start: (end, command) for start, end, command in BLOCKS}
+    lines = source.splitlines()
+    blocks: list[tuple[tuple[str, ...], str]] = []
+    index = 0
+    while index < len(lines):
+        marker = markers.get(lines[index])
+        if marker is None:
+            index += 1
+            continue
+        end, command = marker
+        body_start = index + 1
+        index = body_start
+        while index < len(lines) and lines[index] != end:
+            index += 1
+        if index == len(lines):
+            raise ValueError(f"unterminated {command[0]} heredoc")
+        body = "\n".join(lines[body_start:index]) + "\n"
+        if not body.strip():
+            raise ValueError(f"empty {command[0]} heredoc")
+        blocks.append((command, body))
+        index += 1
+
+    commands = tuple(command[0] for command, _body in blocks)
+    if commands != EXPECTED_COMMANDS:
+        raise ValueError(
+            "Hermes Dockerfile heredoc contract changed: "
+            f"expected {EXPECTED_COMMANDS!r}, received {commands!r}"
+        )
+    return blocks
+
+
+def replay(dockerfile: Path) -> None:
+    """Execute each exact heredoc in an isolated interpreter process."""
+    size = dockerfile.stat().st_size
+    if size < 1 or size > MAX_DOCKERFILE_BYTES:
+        raise ValueError("Hermes Dockerfile size is outside the reviewed boundary")
+    source = dockerfile.read_text(encoding="utf-8")
+    for index, (command, body) in enumerate(extract_blocks(source), start=1):
+        print(f"replaying reviewed Dockerfile heredoc {index}/{len(EXPECTED_COMMANDS)}")
+        subprocess.run(command, input=body, text=True, check=True)
+
+
+def main() -> int:
+    """Parse the one explicit Dockerfile path and replay its reviewed patches."""
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--dockerfile", required=True, type=Path)
+    args = parser.parse_args()
+    replay(args.dockerfile)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

services/harbor/hermes-agent-immutability-job.yaml

@@ -0,0 +1,79 @@
+# services/harbor/hermes-agent-immutability-job.yaml
+apiVersion: batch/v1
+kind: Job
+metadata:
+  name: harbor-hermes-agent-immutability-ensure-1
+  namespace: harbor
+spec:
+  backoffLimit: 2
+  activeDeadlineSeconds: 600
+  template:
+    metadata:
+      annotations:
+        vault.hashicorp.com/agent-inject: "true"
+        vault.hashicorp.com/agent-pre-populate-only: "true"
+        vault.hashicorp.com/agent-run-as-user: "65532"
+        vault.hashicorp.com/agent-run-as-group: "65532"
+        vault.hashicorp.com/role: harbor-policy-bootstrap
+        vault.hashicorp.com/agent-inject-secret-harbor-admin-password: kv/data/atlas/harbor/harbor-core
+        vault.hashicorp.com/agent-inject-template-harbor-admin-password: |
+          {{- with secret "kv/data/atlas/harbor/harbor-core" -}}
+          {{ .Data.data.harbor_admin_password }}
+          {{- end -}}
+    spec:
+      serviceAccountName: harbor-policy-bootstrap
+      enableServiceLinks: false
+      restartPolicy: Never
+      nodeSelector:
+        hardware: rpi5
+        kubernetes.io/arch: arm64
+        node-role.kubernetes.io/worker: "true"
+      affinity:
+        nodeAffinity:
+          requiredDuringSchedulingIgnoredDuringExecution:
+            nodeSelectorTerms:
+              - matchExpressions:
+                  - key: kubernetes.io/hostname
+                    operator: NotIn
+                    values: [titan-04, titan-14, titan-18, titan-19, titan-24]
+      securityContext:
+        fsGroup: 65532
+        fsGroupChangePolicy: OnRootMismatch
+        seccompProfile:
+          type: RuntimeDefault
+      containers:
+        - name: ensure
+          image: docker.io/library/python@sha256:efcdfa6a6b2fd2afb9c7dfa9a5b288a6f68338b5cfdebe6b637d986067d85757
+          imagePullPolicy: IfNotPresent
+          command: [python3, /scripts/harbor_hermes_agent_immutability_ensure.py]
+          env:
+            - name: HARBOR_API_ORIGIN
+              value: https://registry.bstein.dev/api/v2.0
+            - name: HARBOR_ADMIN_PASSWORD_FILE
+              value: /vault/secrets/harbor-admin-password
+          securityContext:
+            allowPrivilegeEscalation: false
+            capabilities:
+              drop: ["ALL"]
+            readOnlyRootFilesystem: true
+            runAsGroup: 65532
+            runAsNonRoot: true
+            runAsUser: 65532
+            seccompProfile:
+              type: RuntimeDefault
+          volumeMounts:
+            - name: scripts
+              mountPath: /scripts
+              readOnly: true
+            - name: tmp
+              mountPath: /tmp
+          resources:
+            requests: {cpu: 25m, memory: 32Mi}
+            limits: {cpu: 250m, memory: 128Mi}
+      volumes:
+        - name: scripts
+          configMap:
+            name: harbor-hermes-agent-immutability-script
+            defaultMode: 0555
+        - name: tmp
+          emptyDir: {}

services/harbor/kustomization.yaml

@@ -12,9 +12,14 @@ resources:
   - certificate.yaml
   - helmrelease.yaml
   - vault-sync-deployment.yaml
+  - policy-bootstrap-serviceaccount.yaml
+  - hermes-agent-immutability-job.yaml
   - bootstrap-jobs/cassandra-registry-ensure-job.yaml
   - image.yaml
 configMapGenerator:
   - name: harbor-vault-entrypoint
     files:
       - scripts/vault-entrypoint.sh
+  - name: harbor-hermes-agent-immutability-script
+    files:
+      - harbor_hermes_agent_immutability_ensure.py=scripts/harbor_hermes_agent_immutability_ensure.py

services/harbor/policy-bootstrap-serviceaccount.yaml

@@ -0,0 +1,7 @@
+# services/harbor/policy-bootstrap-serviceaccount.yaml
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: harbor-policy-bootstrap
+  namespace: harbor
+automountServiceAccountToken: true

services/harbor/scripts/harbor_hermes_agent_immutability_ensure.py

@@ -0,0 +1,378 @@
+#!/usr/bin/env python3
+"""Create and verify the narrowly scoped Hermes agent immutable-tag rule."""
+
+from __future__ import annotations
+
+import base64
+import json
+import os
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+from pathlib import Path
+from typing import Any
+
+
+PROJECT = "bstein"
+REPOSITORY_PATTERN = "hermes-agent"
+TAG_PATTERN = "git-*-build-*"
+PUBLISH_ROBOT = "robot$jenkins-pipelines"
+EXPECTED_ORIGIN = "https://registry.bstein.dev/api/v2.0"
+MAX_RESPONSE = 1_048_576
+TRANSIENT_STATUSES = {429, 502, 503, 504}
+EXPECTED_RULE = {
+    "disabled": False,
+    "action": "immutable",
+    "template": "immutable_template",
+    "tag_selectors": [
+        {
+            "kind": "doublestar",
+            "decoration": "matches",
+            "pattern": TAG_PATTERN,
+        }
+    ],
+    "scope_selectors": {
+        "repository": [
+            {
+                "kind": "doublestar",
+                "decoration": "repoMatches",
+                "pattern": REPOSITORY_PATTERN,
+            }
+        ]
+    },
+}
+
+
+class NoRedirect(urllib.request.HTTPRedirectHandler):
+    """Prevent Basic credentials from following an unexpected redirect."""
+
+    def redirect_request(self, _request, _file, _code, _message, _headers, _url):
+        return None
+
+
+class HarborUnavailable(RuntimeError):
+    """Harbor is not ready yet, rather than returning a policy decision."""
+
+
+def normalized_rule(rule: dict[str, Any]) -> dict[str, Any]:
+    """Return only the immutable contract fields Harbor must preserve."""
+    return {
+        "disabled": bool(rule.get("disabled", False)),
+        "action": rule.get("action"),
+        "template": rule.get("template"),
+        "tag_selectors": [
+            {
+                "kind": selector.get("kind"),
+                "decoration": selector.get("decoration"),
+                "pattern": selector.get("pattern"),
+            }
+            for selector in rule.get("tag_selectors") or []
+            if isinstance(selector, dict)
+        ],
+        "scope_selectors": {
+            "repository": [
+                {
+                    "kind": selector.get("kind"),
+                    "decoration": selector.get("decoration"),
+                    "pattern": selector.get("pattern"),
+                }
+                for selector in (rule.get("scope_selectors") or {}).get(
+                    "repository", []
+                )
+                if isinstance(selector, dict)
+            ]
+        },
+    }
+
+
+def targets_hermes_builds(rule: dict[str, Any]) -> bool:
+    """Detect a rule that claims this exact repository and tag selector."""
+    normalized = normalized_rule(rule)
+    return (
+        normalized["tag_selectors"] == EXPECTED_RULE["tag_selectors"]
+        and normalized["scope_selectors"] == EXPECTED_RULE["scope_selectors"]
+    )
+
+
+class HarborClient:
+    """Bounded same-origin client for Harbor's immutable-tag API."""
+
+    def __init__(self, origin: str, username: str, password: str) -> None:
+        normalized_origin = origin.rstrip("/")
+        if normalized_origin != EXPECTED_ORIGIN:
+            raise ValueError("Harbor API origin is not the pinned production API")
+        self.origin = normalized_origin
+        token = base64.b64encode(f"{username}:{password}".encode()).decode("ascii")
+        self.headers = {"Authorization": f"Basic {token}"}
+        self.opener = urllib.request.build_opener(NoRedirect())
+
+    def request(
+        self, method: str, path: str, payload: dict[str, Any] | None = None
+    ) -> tuple[int, bytes, dict[str, str]]:
+        """Issue one request, returning even non-2xx responses for strict checks."""
+        data = None
+        headers = dict(self.headers)
+        if payload is not None:
+            data = json.dumps(payload, separators=(",", ":")).encode()
+            headers["Content-Type"] = "application/json"
+        request = urllib.request.Request(
+            f"{self.origin}{path}", data=data, headers=headers, method=method
+        )
+        try:
+            response = self.opener.open(request, timeout=20)
+        except urllib.error.HTTPError as exc:
+            response = exc
+        except (urllib.error.URLError, TimeoutError) as exc:
+            raise HarborUnavailable("Harbor policy API is unavailable") from exc
+        with response:
+            body = response.read(MAX_RESPONSE + 1)
+            if len(body) > MAX_RESPONSE:
+                raise RuntimeError("Harbor response exceeded the size limit")
+            return int(response.status), body, dict(response.headers)
+
+
+def list_rules(client: HarborClient) -> list[dict[str, Any]]:
+    """Read and validate the complete small rule set for the project."""
+    path = f"/projects/{PROJECT}/immutabletagrules?page=1&page_size=100"
+    status, body, headers = client.request("GET", path)
+    if status in TRANSIENT_STATUSES:
+        raise HarborUnavailable(f"Harbor immutable rule list returned HTTP {status}")
+    if status != 200:
+        raise RuntimeError(f"Harbor immutable rule list returned HTTP {status}")
+    values = _json_list(body, "immutable rule")
+    _require_complete_page(values, headers, "immutable rule")
+    return values
+
+
+def _json_list(body: bytes, label: str) -> list[dict[str, Any]]:
+    """Decode one bounded Harbor list without accepting a partial shape."""
+    try:
+        value = json.loads(body.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise RuntimeError(f"Harbor returned invalid {label} JSON") from exc
+    if not isinstance(value, list) or not all(isinstance(item, dict) for item in value):
+        raise RuntimeError(f"Harbor {label} list has an invalid shape")
+    return value
+
+
+def _require_complete_page(
+    values: list[dict[str, Any]], headers: dict[str, str], label: str
+) -> None:
+    """Reject a truncated first page or a proxy that strips count evidence."""
+    raw_total = next(
+        (value for key, value in headers.items() if key.lower() == "x-total-count"),
+        None,
+    )
+    if raw_total is None or not str(raw_total).isdecimal():
+        raise RuntimeError(f"Harbor {label} list omitted a valid total count")
+    if int(raw_total) != len(values):
+        raise RuntimeError(f"Harbor {label} list was truncated")
+
+
+def _allowed(access: dict[str, Any], resource: str, action: str) -> bool:
+    """Match one positive robot permission, accepting Harbor's default effect."""
+    return (
+        access.get("resource") == resource
+        and access.get("action") == action
+        and access.get("effect") in (None, "", "allow")
+    )
+
+
+def _publisher_scope(robot: dict[str, Any]) -> tuple[dict[str, Any], list[dict[str, Any]]]:
+    """Return the existing bstein scope after validating its push boundary."""
+    permissions = robot.get("permissions")
+    if not isinstance(permissions, list) or not all(
+        isinstance(item, dict) for item in permissions
+    ):
+        raise RuntimeError("Jenkins Harbor robot permissions have an invalid shape")
+    matches = [
+        item
+        for item in permissions
+        if item.get("kind") == "project" and item.get("namespace") == PROJECT
+    ]
+    if len(matches) != 1:
+        raise RuntimeError("Jenkins Harbor robot must have one existing bstein scope")
+    access = matches[0].get("access")
+    if not isinstance(access, list) or not all(isinstance(item, dict) for item in access):
+        raise RuntimeError("Jenkins Harbor robot bstein access has an invalid shape")
+    for required in (("repository", "pull"), ("repository", "push")):
+        if not any(_allowed(item, *required) for item in access):
+            raise RuntimeError("Jenkins Harbor robot lacks its existing pull/push boundary")
+    immutable_access = [
+        item for item in access if item.get("resource") == "immutable-tag"
+    ]
+    if immutable_access and not (
+        len(immutable_access) == 1
+        and _allowed(immutable_access[0], "immutable-tag", "list")
+    ):
+        raise RuntimeError("Jenkins Harbor robot has broader immutable-tag access")
+    return matches[0], access
+
+
+def _verified_publisher(robot: dict[str, Any]) -> bool:
+    """Check the exact identity and least-privilege policy after persistence."""
+    if (
+        robot.get("name") != PUBLISH_ROBOT
+        or robot.get("level") != "system"
+        or robot.get("editable") is not True
+        or robot.get("disable") is not False
+    ):
+        return False
+    try:
+        _scope, access = _publisher_scope(robot)
+    except RuntimeError:
+        return False
+    return any(_allowed(item, "immutable-tag", "list") for item in access)
+
+
+def ensure_publisher_can_read_rule(client: HarborClient) -> int:
+    """Grant only immutable-tag:list to the existing Jenkins push robot."""
+    status, body, headers = client.request("GET", "/robots?page=1&page_size=100")
+    if status in TRANSIENT_STATUSES:
+        raise HarborUnavailable(f"Harbor robot list returned HTTP {status}")
+    if status != 200:
+        raise RuntimeError(f"Harbor robot list returned HTTP {status}")
+    robots = _json_list(body, "robot")
+    _require_complete_page(robots, headers, "robot")
+    matches = [
+        item
+        for item in robots
+        if item.get("name") == PUBLISH_ROBOT
+    ]
+    if len(matches) != 1:
+        raise RuntimeError("expected exactly one Jenkins Harbor publisher robot")
+    robot_id = matches[0].get("id")
+    if not isinstance(robot_id, int) or robot_id < 1:
+        raise RuntimeError("Jenkins Harbor robot omitted a valid ID")
+    status, body, _headers = client.request("GET", f"/robots/{robot_id}")
+    if status in TRANSIENT_STATUSES:
+        raise HarborUnavailable(f"Harbor robot read returned HTTP {status}")
+    if status != 200:
+        raise RuntimeError(f"Harbor robot read returned HTTP {status}")
+    try:
+        robot = json.loads(body.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise RuntimeError("Harbor returned invalid robot JSON") from exc
+    if not isinstance(robot, dict):
+        raise RuntimeError("Harbor robot response has an invalid shape")
+    if (
+        robot.get("name") != PUBLISH_ROBOT
+        or robot.get("level") != "system"
+        or robot.get("editable") is not True
+        or robot.get("disable") is not False
+    ):
+        raise RuntimeError("Jenkins Harbor publisher identity is not active and exact")
+    scope, access = _publisher_scope(robot)
+    if any(_allowed(item, "immutable-tag", "list") for item in access):
+        return robot_id
+
+    # Harbor updates permissions through PUT /robots/{id}; its separate PATCH
+    # endpoint is the only secret-rotation operation. Preserve every current
+    # scope and field while adding the one read-only action.
+    access.append({"resource": "immutable-tag", "action": "list"})
+    payload = {
+        "name": robot["name"],
+        "description": robot.get("description") or "",
+        "level": robot["level"],
+        "disable": robot["disable"],
+        "permissions": robot["permissions"],
+    }
+    duration = robot.get("duration")
+    if duration is not None:
+        if not isinstance(duration, int):
+            raise RuntimeError("Jenkins Harbor robot duration has an invalid shape")
+        payload["duration"] = duration
+    # Keep the validated object reference live in the preserved permission list.
+    if scope.get("access") is not access:
+        raise RuntimeError("Jenkins Harbor robot permission normalization drifted")
+    status, _body, _headers = client.request(
+        "PUT", f"/robots/{robot_id}", payload
+    )
+    if status in TRANSIENT_STATUSES:
+        raise HarborUnavailable(f"Harbor robot update returned HTTP {status}")
+    if status != 200:
+        raise RuntimeError(f"Harbor robot policy update returned HTTP {status}")
+
+    for attempt in range(1, 6):
+        status, body, _headers = client.request("GET", f"/robots/{robot_id}")
+        if status == 200:
+            try:
+                persisted = json.loads(body.decode("utf-8"))
+            except (UnicodeDecodeError, json.JSONDecodeError):
+                persisted = None
+            if isinstance(persisted, dict) and _verified_publisher(persisted):
+                return robot_id
+        if attempt < 5:
+            time.sleep(attempt)
+    raise RuntimeError("Jenkins Harbor robot policy did not verify exactly")
+
+
+def ensure_rule(client: HarborClient) -> int:
+    """Create once, or validate the one exact enabled rule already present."""
+    rules = list_rules(client)
+    matches = [rule for rule in rules if targets_hermes_builds(rule)]
+    if len(matches) > 1:
+        raise RuntimeError("multiple Hermes agent immutable rules exist")
+    if matches:
+        if normalized_rule(matches[0]) != EXPECTED_RULE:
+            raise RuntimeError("Hermes agent immutable rule exists but is not enabled/exact")
+        rule_id = matches[0].get("id")
+        if not isinstance(rule_id, int) or rule_id < 1:
+            raise RuntimeError("Harbor immutable rule omitted a valid ID")
+        return rule_id
+
+    path = f"/projects/{PROJECT}/immutabletagrules"
+    status, _body, headers = client.request("POST", path, EXPECTED_RULE)
+    if status in TRANSIENT_STATUSES:
+        raise HarborUnavailable(f"Harbor immutable rule create returned HTTP {status}")
+    if status != 201:
+        raise RuntimeError(f"Harbor immutable rule create returned HTTP {status}")
+    location = headers.get("Location") or headers.get("location") or ""
+    api_path = urllib.parse.urlsplit(client.origin).path.rstrip("/")
+    expected_prefix = f"{api_path}{path}/"
+    if not location.startswith(expected_prefix):
+        raise RuntimeError("Harbor immutable rule create omitted the exact Location")
+    suffix = location[len(expected_prefix) :]
+    if not suffix.isdecimal() or int(suffix) < 1:
+        raise RuntimeError("Harbor immutable rule Location has an invalid ID")
+
+    # Harbor persists synchronously, but bounded retries distinguish a stale
+    # read from accepting an unverified policy.
+    for attempt in range(1, 6):
+        matches = [rule for rule in list_rules(client) if targets_hermes_builds(rule)]
+        if len(matches) == 1 and normalized_rule(matches[0]) == EXPECTED_RULE:
+            rule_id = matches[0].get("id")
+            if rule_id == int(suffix):
+                return rule_id
+        if attempt < 5:
+            time.sleep(attempt)
+    raise RuntimeError("created Harbor immutable rule did not verify exactly")
+
+
+def main() -> int:
+    """Load the runtime-only admin credential and enforce the tracked policy."""
+    origin = os.environ.get("HARBOR_API_ORIGIN", "")
+    password_file = Path(os.environ.get("HARBOR_ADMIN_PASSWORD_FILE", ""))
+    password = password_file.read_text(encoding="utf-8").strip()
+    if not password:
+        raise RuntimeError("Harbor admin password is empty")
+    client = HarborClient(origin, "admin", password)
+    for attempt in range(1, 13):
+        try:
+            rule_id = ensure_rule(client)
+            robot_id = ensure_publisher_can_read_rule(client)
+            break
+        except HarborUnavailable:
+            if attempt == 12:
+                raise
+            time.sleep(min(attempt * 2, 15))
+    print(
+        "Hermes agent immutable build-tag rule is active "
+        f"(id={rule_id}); publisher preflight access is active (robot={robot_id})"
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

services/hermes/NOTES.md

@@ -110,6 +110,29 @@ privileged container, or writable Git credential. It accepts only an exact
 40-character revision that is both the checked-out commit and current
 `atlas/titan-iac` `main`, so a human must merge the source PR first.
 
+Kaniko runs as UID 0 because it must unpack an image root filesystem and enter
+that filesystem for Dockerfile `RUN` instructions. Its capability set is the
+minimum proven by an exact ARM64 no-push build of this Dockerfile:
+`CHOWN`, `FOWNER`, `DAC_OVERRIDE`, `SETGID`, and `SETUID`. `SETGID` is required
+because pinned Kaniko applies the base image's supplementary group list before
+each Dockerfile `RUN`; `SETUID` lets apt drop privileges to its `_apt` account
+while downloading package indexes. It still has no privilege
+escalation, service-account token, host path, daemon socket, or Docker/BuildKit
+TCP endpoint and uses the runtime-default seccomp profile. This is residual
+root-in-the-build-pod risk, bounded to disposable `emptyDir` storage and a
+human-reviewed `main` revision; Dockerfile changes require the same scrutiny as
+executable cluster code.
+
+Pinned Kaniko parses Dockerfile `RUN` heredocs but does not materialize their
+inline files when executing a command. The Jenkins lane therefore enables a
+bounded compatibility replay after copying the reviewed Dockerfile and runner
+from the same checked-out commit. The runner accepts only the exact nine
+Node/Python heredocs in their expected order and launches each in a separate
+interpreter process. Docker and BuildKit keep their native behavior because the
+compatibility argument defaults off. The final TypeScript build, Python compile,
+and source assertions remain mandatory, so an omitted or drifted replay fails
+before any image can pass release verification.
+
 From agent.hermes, trigger that one fixed job with:
 
 ```sh
@@ -131,12 +154,30 @@ denied, and the token cannot select, configure, read, or administer another
 Jenkins job. The pipeline independently rejects any commit that is not the
 current `origin/main`, preserving the human merge/review boundary.
 
-The tracked rollout order is deliberate: the revisioned Vault seed Job must
-complete before the Vault Kustomization becomes Ready, Jenkins depends on
-Vault, and Hermes depends on Jenkins. The seeder uses KV-v2 CAS create-only
-semantics. It never changes an existing token, never fills a missing field in
-an existing secret, and fails closed when it cannot distinguish absence from a
-read error. The recurring Vault configuration job verifies the same invariant.
+Harbor independently enforces an enabled immutable-tag rule scoped to only the
+`bstein/hermes-agent` repository and `git-*-build-*` tags. A second manifest PUT,
+retag, or deletion is rejected by Harbor even if a caller bypasses the Jenkins
+preflight. A revisioned, Flux-tracked policy Job creates or verifies that exact
+rule with a runtime-only Vault credential; it refuses to alter a conflicting
+rule and retries only explicit transport/readiness failures while Harbor
+starts. The same Job grants the existing Jenkins publisher only
+`immutable-tag:list` on `bstein`, preserving its other project scopes and never
+calling Harbor's separate secret-rotation endpoint. The pipeline uses that
+read-only permission to require the exact enabled rule before Kaniko starts.
+That check also makes an already-running Jenkins controller fail closed during
+rollout; Jenkins cannot become Ready on the new revision before the policy Job
+succeeds.
+
+The tracked rollout order is deliberate: the revisioned Vault role Job must
+complete before the Vault Kustomization becomes Ready. A separate Flux
+Kustomization then runs the seed Job under the dedicated
+`hermes-jenkins-token-seed` identity, which can only create/read the exact
+`kv/data/atlas/hermes/developer-jenkins` path and request random bytes. Jenkins
+depends on both that seed and the Harbor policy Job; Hermes depends on Jenkins.
+The seeder uses KV-v2 CAS create-only semantics. It never changes an existing
+token, never fills a missing field in an existing secret, and fails closed when
+it cannot distinguish absence from a read error. The recurring Vault
+configuration job reconciles the same narrow role and policy.
 
 Rotate this fixed-job token only as a coordinated operator action:
 

services/vault-hermes-jenkins-token-seed/job.yaml

@@ -1,15 +1,15 @@
-# services/vault/hermes-jenkins-build-token-seed-job.yaml
-# Revisioned prerequisite: seed the job-scoped token before Jenkins/Hermes roll.
+# services/vault-hermes-jenkins-token-seed/job.yaml
 apiVersion: batch/v1
 kind: Job
 metadata:
-  name: vault-hermes-jenkins-build-token-seed-1
+  name: vault-hermes-jenkins-build-token-seed-2
   namespace: vault
 spec:
   backoffLimit: 2
   template:
     spec:
-      serviceAccountName: vault-admin
+      serviceAccountName: hermes-jenkins-token-seed
+      enableServiceLinks: false
       restartPolicy: Never
       nodeSelector:
         hardware: rpi5
@@ -23,25 +23,46 @@ spec:
                   - key: kubernetes.io/hostname
                     operator: NotIn
                     values: [titan-04, titan-14, titan-18, titan-19, titan-24]
+      securityContext:
+        fsGroup: 1000
+        fsGroupChangePolicy: OnRootMismatch
+        seccompProfile:
+          type: RuntimeDefault
       containers:
         - name: seed
-          image: hashicorp/vault:1.21.4
+          image: docker.io/hashicorp/vault@sha256:4e33b126a59c0c333b76fb4e894722462659a6bec7c48c9ee8cea56fccfd2569
           imagePullPolicy: IfNotPresent
           command: [sh, /scripts/vault_hermes_jenkins_build_token_ensure.sh]
           env:
+            - name: HOME
+              value: /tmp
             - name: VAULT_ADDR
               value: http://vault.vault.svc.cluster.local:8200
             - name: VAULT_K8S_ROLE
-              value: vault-admin
+              value: hermes-jenkins-token-seed
+          securityContext:
+            allowPrivilegeEscalation: false
+            capabilities:
+              drop: ["ALL"]
+            readOnlyRootFilesystem: true
+            runAsGroup: 1000
+            runAsNonRoot: true
+            runAsUser: 100
+            seccompProfile:
+              type: RuntimeDefault
           volumeMounts:
             - name: scripts
               mountPath: /scripts
               readOnly: true
+            - name: tmp
+              mountPath: /tmp
           resources:
             requests: {cpu: 25m, memory: 32Mi}
             limits: {cpu: 250m, memory: 128Mi}
       volumes:
         - name: scripts
           configMap:
-            name: vault-k8s-auth-config-script
+            name: vault-hermes-jenkins-token-seed-script
             defaultMode: 0555
+        - name: tmp
+          emptyDir: {}

services/vault-hermes-jenkins-token-seed/kustomization.yaml

@@ -0,0 +1,13 @@
+# services/vault-hermes-jenkins-token-seed/kustomization.yaml
+apiVersion: kustomize.config.k8s.io/v1beta1
+kind: Kustomization
+namespace: vault
+resources:
+  - serviceaccount.yaml
+  - job.yaml
+generatorOptions:
+  disableNameSuffixHash: true
+configMapGenerator:
+  - name: vault-hermes-jenkins-token-seed-script
+    files:
+      - vault_hermes_jenkins_build_token_ensure.sh=scripts/vault_hermes_jenkins_build_token_ensure.sh

services/vault-hermes-jenkins-token-seed/scripts/vault_hermes_jenkins_build_token_ensure.sh

@@ -1,11 +1,14 @@
 #!/usr/bin/env sh
 set -eu
 
-secret_path="kv/atlas/hermes/developer-jenkins"
+secret_api_path="kv/data/atlas/hermes/developer-jenkins"
 jwt_file="${VAULT_K8S_JWT_FILE:-/var/run/secrets/kubernetes.io/serviceaccount/token}"
-vault_role="${VAULT_K8S_ROLE:-vault-admin}"
+vault_role="${VAULT_K8S_ROLE:-hermes-jenkins-token-seed}"
+payload_file="${TMPDIR:-/tmp}/hermes-jenkins-token.json"
 
 log() { printf '[hermes-jenkins-token] %s\n' "$*" >&2; }
+cleanup() { rm -f "${payload_file}"; }
+trap cleanup EXIT HUP INT TERM
 
 retry_command() {
   attempt=1
@@ -44,32 +47,25 @@ ensure_token() {
   export VAULT_TOKEN
 }
 
-# Return 0 when the field exists, 10 when the secret is absent, 11 when the
-# secret exists without the field, and 12 for a persistent read error.
+# Return 0 for the complete field, 10 when absent, 11 when present but
+# incomplete, and 12 for a persistent authorization/transport failure.
 read_build_token() {
   attempt=1
   while [ "${attempt}" -le 5 ]; do
     set +e
-    secret_error="$(vault kv get "${secret_path}" 2>&1 >/dev/null)"
+    secret_json="$(vault read -format=json "${secret_api_path}" 2>&1)"
     secret_status=$?
     set -e
     if [ "${secret_status}" -eq 0 ]; then
-      set +e
-      build_token="$(vault kv get -field=build_token "${secret_path}" 2>&1)"
-      field_status=$?
-      set -e
-      if [ "${field_status}" -eq 0 ] && [ -n "${build_token}" ]; then
+      if printf '%s' "${secret_json}" | grep -Eq '"build_token"[[:space:]]*:[[:space:]]*"[0-9a-f]{64}"'; then
+        unset secret_json
         return 0
       fi
-      if printf '%s' "${build_token}" | grep -q 'No value found'; then
-        unset build_token
-        return 11
-      fi
-      if [ "${field_status}" -eq 0 ]; then
-        unset build_token
-        return 11
-      fi
-    elif printf '%s' "${secret_error}" | grep -Eq 'Code: 404|No value found at'; then
+      unset secret_json
+      return 11
+    fi
+    if printf '%s' "${secret_json}" | grep -Eq 'Code: 404|No value found at'; then
+      unset secret_json
       return 10
     fi
     if [ "${attempt}" -lt 5 ]; then
@@ -77,7 +73,7 @@ read_build_token() {
     fi
     attempt=$((attempt + 1))
   done
-  unset secret_error build_token
+  unset secret_json
   return 12
 }
 
@@ -90,12 +86,11 @@ else
 fi
 case "${read_status}" in
   0)
-    unset build_token
     log "job-scoped token already present; no write performed"
     exit 0
     ;;
   11)
-    log "secret exists without build_token; refusing to overwrite existing fields"
+    log "secret exists without a valid build_token; refusing to overwrite existing fields"
     exit 1
     ;;
   12)
@@ -114,33 +109,30 @@ new_token="$(retry_command vault write -field=random_bytes \
     log "Vault random token generation failed after retries"
     exit 1
   }
-if [ -z "${new_token}" ]; then
-  log "Vault returned an empty random token"
+if ! printf '%s' "${new_token}" | grep -Eq '^[0-9a-f]{64}$'; then
+  unset new_token
+  log "Vault returned an invalid random token"
   exit 1
 fi
 
+umask 077
+printf '{"options":{"cas":0},"data":{"build_token":"%s"}}\n' \
+  "${new_token}" > "${payload_file}"
+unset new_token
+
 set +e
-create_error="$(printf '%s' "${new_token}" | \
-  vault kv put -cas=0 "${secret_path}" build_token=- 2>&1 >/dev/null)"
+create_error="$(vault write "${secret_api_path}" @"${payload_file}" 2>&1 >/dev/null)"
 create_status=$?
 set -e
-unset new_token
+cleanup
 if [ "${create_status}" -eq 0 ]; then
   log "job-scoped token created with KV-v2 CAS create-only semantics"
   exit 0
 fi
 
-# A simultaneous revisioned Job/Cron run may win the create. Accept that race
-# only if a fresh retried read finds the complete field; never rotate it.
 if printf '%s' "${create_error}" | grep -qi 'check-and-set'; then
   unset create_error
   if read_build_token; then
-    read_status=0
-  else
-    read_status=$?
-  fi
-  if [ "${read_status}" -eq 0 ]; then
-    unset build_token
     log "another seeder won CAS; existing token preserved"
     exit 0
   fi

services/vault-hermes-jenkins-token-seed/serviceaccount.yaml

@@ -0,0 +1,7 @@
+# services/vault-hermes-jenkins-token-seed/serviceaccount.yaml
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: hermes-jenkins-token-seed
+  namespace: vault
+automountServiceAccountToken: true

services/vault/hermes-auth-role-bootstrap-job.yaml

@@ -3,7 +3,7 @@
 apiVersion: batch/v1
 kind: Job
 metadata:
-  name: vault-k8s-auth-hermes-8
+  name: vault-k8s-auth-hermes-9
   namespace: vault
 spec:
   backoffLimit: 2
@@ -15,16 +15,26 @@ spec:
         hardware: rpi5
         kubernetes.io/arch: arm64
         node-role.kubernetes.io/worker: "true"
+      affinity:
+        nodeAffinity:
+          requiredDuringSchedulingIgnoredDuringExecution:
+            nodeSelectorTerms:
+              - matchExpressions:
+                  - key: kubernetes.io/hostname
+                    operator: NotIn
+                    values: [titan-04, titan-14, titan-18, titan-19, titan-24]
       containers:
         - name: configure-k8s-auth
-          image: hashicorp/vault:1.21.4
+          image: docker.io/hashicorp/vault@sha256:4e33b126a59c0c333b76fb4e894722462659a6bec7c48c9ee8cea56fccfd2569
           imagePullPolicy: IfNotPresent
           command:
             - sh
             - /scripts/vault_k8s_auth_configure.sh
           env:
+            - name: HOME
+              value: /tmp
             - name: VAULT_ADDR
-              value: http://10.43.57.249:8200
+              value: http://vault.vault.svc.cluster.local:8200
             - name: VAULT_K8S_ROLE
               value: vault-admin
             - name: VAULT_K8S_TOKEN_REVIEWER_JWT_FILE
@@ -38,6 +48,18 @@ spec:
             - name: token-reviewer
               mountPath: /var/run/secrets/vault-token-reviewer
               readOnly: true
+            - name: tmp
+              mountPath: /tmp
+          securityContext:
+            allowPrivilegeEscalation: false
+            capabilities:
+              drop: ["ALL"]
+            readOnlyRootFilesystem: true
+            runAsGroup: 1000
+            runAsNonRoot: true
+            runAsUser: 100
+            seccompProfile:
+              type: RuntimeDefault
           resources:
             requests:
               cpu: 25m
@@ -53,3 +75,5 @@ spec:
         - name: token-reviewer
           secret:
             secretName: vault-admin-token-reviewer
+        - name: tmp
+          emptyDir: {}

services/vault/kustomization.yaml

@@ -12,7 +12,6 @@ resources:
   - statefulset.yaml
   - k8s-auth-config-cronjob.yaml
   - hermes-auth-role-bootstrap-job.yaml
-  - hermes-jenkins-build-token-seed-job.yaml
   - oidc-config-cronjob.yaml
   - service.yaml
   - certificate.yaml
@@ -28,7 +27,6 @@ configMapGenerator:
   - name: vault-k8s-auth-config-script
     files:
       - vault_k8s_auth_configure.sh=scripts/vault_k8s_auth_configure.sh
-      - vault_hermes_jenkins_build_token_ensure.sh=scripts/vault_hermes_jenkins_build_token_ensure.sh
   - name: vault-entrypoint
     files:
       - vault-entrypoint.sh=scripts/vault-entrypoint.sh

services/vault/scripts/vault_k8s_auth_configure.sh

@@ -235,6 +235,18 @@ write_policy_and_role "mailu-mailserver" "mailu-mailserver" "mailu-vault-sync" \
   "mailu/* shared/postmark-relay shared/harbor-pull" ""
 write_policy_and_role "harbor" "harbor" "harbor-vault-sync" \
   "harbor/* shared/harbor-pull" "hermes/developer-harbor"
+harbor_policy_bootstrap_policy='
+path "kv/data/atlas/harbor/harbor-core" {
+  capabilities = ["read"]
+}
+'
+write_raw_policy "harbor-policy-bootstrap" "${harbor_policy_bootstrap_policy}"
+log "writing role harbor-policy-bootstrap"
+vault_cmd write "auth/kubernetes/role/harbor-policy-bootstrap" \
+  bound_service_account_names="harbor-policy-bootstrap" \
+  bound_service_account_namespaces="harbor" \
+  policies="harbor-policy-bootstrap" \
+  ttl="${role_ttl}"
 write_policy_and_role "nextcloud" "nextcloud" "nextcloud-vault" \
   "nextcloud/* shared/keycloak-admin shared/postmark-relay" ""
 write_policy_and_role "comms" "comms" "comms-vault,atlasbot" \
@@ -261,6 +273,21 @@ write_policy_and_role "hermes-agent" "hermes" "hermes-agent" \
   "hermes/agent-oidc hermes/agent-tokens hermes/chat-telegram hermes/developer-keycloak hermes/developer-gitea hermes/developer-harbor hermes/developer-jenkins hermes/developer-ssh" ""
 write_policy_and_role "hermes-switchyard" "hermes" "hermes-switchyard" \
   "hermes/chat-telegram" ""
+hermes_jenkins_token_seed_policy='
+path "kv/data/atlas/hermes/developer-jenkins" {
+  capabilities = ["create", "read"]
+}
+path "sys/tools/random/32" {
+  capabilities = ["update"]
+}
+'
+write_raw_policy "hermes-jenkins-token-seed" "${hermes_jenkins_token_seed_policy}"
+log "writing role hermes-jenkins-token-seed"
+vault_cmd write "auth/kubernetes/role/hermes-jenkins-token-seed" \
+  bound_service_account_names="hermes-jenkins-token-seed" \
+  bound_service_account_namespaces="vault" \
+  policies="hermes-jenkins-token-seed" \
+  ttl="${role_ttl}"
 write_policy_and_role "hermes-credential-sync" "hermes" "hermes-agent" \
   "" "hermes/agent-tokens"
 write_policy_and_role "hermes-node-ssh" "hermes" "hermes-node-ssh-access" \
@@ -308,10 +335,6 @@ write_policy_and_role "postgres" "postgres" "postgres-vault" \
 write_policy_and_role "vault" "vault" "vault" \
   "vault/*" ""
 
-# This verifier is also run by a revisioned prerequisite Job. It creates only
-# when the entire KV-v2 secret is absent and never rotates an existing value.
-/scripts/vault_hermes_jenkins_build_token_ensure.sh
-
 write_policy_and_role "sso-secrets" "sso" "mas-secrets-ensure" \
   "shared/keycloak-admin shared/postmark-relay maintenance/metis-ssh-keys" \
   "harbor/harbor-oidc vault/vault-oidc-config comms/synapse-oidc logging/oauth2-proxy-logs-oidc finance/actual-oidc maintenance/metis-oidc maintenance/soteria-oidc maintenance/metis-ssh-keys veles/veles-oidc cassandra/cassandra-oidc gitea/gitea-veles-oidc gitea/gitea-cassandra-oidc hermes/chat-oidc hermes/chat-telegram hermes/agent-oidc hermes/triage-oidc hermes/developer-keycloak" \

testing/tests/test_hermes_image_builder.py

@@ -16,6 +16,7 @@ REPO_ROOT = Path(__file__).resolve().parents[2]
 PIPELINE_PATH = REPO_ROOT / "ci/Jenkinsfile.hermes-agent-image"
 RELEASE_SCRIPT = REPO_ROOT / "ci/scripts/hermes_image_release.py"
 TRIGGER_SCRIPT = REPO_ROOT / "services/hermes/scripts/jenkins_image_build_trigger.py"
+HEREDOC_RUNNER = REPO_ROOT / "dockerfiles/hermes-kaniko-heredoc-runner.py"
 
 
 def _load_release_module():
@@ -38,14 +39,24 @@ def _load_trigger_module():
     return module
 
 
+def _load_heredoc_runner():
+    spec = importlib.util.spec_from_file_location(
+        "hermes_kaniko_heredoc_runner", HEREDOC_RUNNER
+    )
+    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 _pod_spec() -> dict:
     source = PIPELINE_PATH.read_text(encoding="utf-8")
     pod_yaml = source.split('yaml """', 1)[1].split('"""', 1)[0]
     return yaml.safe_load(pod_yaml)["spec"]
 
 
-def test_builder_pod_is_daemonless_and_kernel_unprivileged() -> None:
-    """The builder must not gain host, daemon, service-token, or Linux privileges."""
+def test_builder_pod_is_daemonless_and_kernel_bounded() -> None:
+    """The builder gets only proven build caps, never host, daemon, or K8s access."""
     source = PIPELINE_PATH.read_text(encoding="utf-8")
     spec = _pod_spec()
     assert spec["automountServiceAccountToken"] is False
@@ -62,12 +73,21 @@ def test_builder_pod_is_daemonless_and_kernel_unprivileged() -> None:
         "gcr.io/kaniko-project/executor@sha256:"
         "c3109d5926a997b100c4343944e06c6b30a6804b2f9abe0994d3de6ef92b028e"
     )
+    assert kaniko["securityContext"]["capabilities"]["add"] == [
+        "CHOWN",
+        "FOWNER",
+        "DAC_OVERRIDE",
+        "SETGID",
+        "SETUID",
+    ]
     for container in containers.values():
         security = container["securityContext"]
         assert security["allowPrivilegeEscalation"] is False
         assert security["capabilities"]["drop"] == ["ALL"]
         assert security["seccompProfile"]["type"] == "RuntimeDefault"
         assert security.get("privileged", False) is False
+    assert "add" not in containers["jnlp"]["securityContext"]["capabilities"]
+    assert "add" not in containers["python"]["securityContext"]["capabilities"]
 
 
 def test_builder_is_restricted_to_healthy_rpi5_capacity() -> None:
@@ -107,6 +127,58 @@ def test_pipeline_requires_reviewed_main_and_runtime_credentials() -> None:
     assert "--destination=" in source
     assert "assert-absent" in source
     assert "git-${actual_revision}-build-${BUILD_NUMBER}" in source
+    assert source.count("--build-arg=HERMES_KANIKO_HEREDOC_COMPAT=1") == 1
+
+
+def test_kaniko_replays_only_the_exact_reviewed_heredoc_contract() -> None:
+    """Pinned Kaniko's ignored inline files are replayed in exact source order."""
+    module = _load_heredoc_runner()
+    dockerfile = (REPO_ROOT / "dockerfiles/Dockerfile.hermes-agent").read_text()
+    blocks = module.extract_blocks(dockerfile)
+    assert tuple(command[0] for command, _body in blocks) == (
+        "node",
+        "python",
+        "python",
+        "python",
+        "python",
+        "python",
+        "python",
+        "python",
+        "node",
+    )
+    assert "session message total" in blocks[-1][1]
+    compat = dockerfile.split("ARG HERMES_KANIKO_HEREDOC_COMPAT=0", 1)[1]
+    assert "case \"${HERMES_KANIKO_HEREDOC_COMPAT}\" in 0|1)" in compat
+    assert "python /tmp/hermes-kaniko-heredoc-runner.py" in compat
+    ignored = (
+        REPO_ROOT / "dockerfiles/Dockerfile.hermes-agent.dockerignore"
+    ).read_text()
+    assert "!dockerfiles/Dockerfile.hermes-agent" in ignored
+    assert "!dockerfiles/hermes-kaniko-heredoc-runner.py" in ignored
+
+
+def test_kaniko_heredoc_runner_rejects_drift_and_executes_separately(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    """Missing blocks fail closed and each reviewed body gets a fresh process."""
+    module = _load_heredoc_runner()
+    source = (REPO_ROOT / "dockerfiles/Dockerfile.hermes-agent").read_text()
+    with pytest.raises(ValueError, match="contract changed"):
+        module.extract_blocks(source.replace("RUN node <<'NODE'", "RUN node", 1))
+
+    calls = []
+
+    def run(command, **kwargs):
+        calls.append((command, kwargs))
+
+    monkeypatch.setattr(module.subprocess, "run", run)
+    dockerfile = tmp_path / "Dockerfile"
+    dockerfile.write_text(source, encoding="utf-8")
+    module.replay(dockerfile)
+    assert len(calls) == 9
+    assert all(call[1]["check"] is True for call in calls)
+    assert all(call[1]["text"] is True for call in calls)
+    assert all(call[1]["input"].endswith("\n") for call in calls)
 
 
 def test_jenkins_job_is_manual_and_reads_pipeline_from_main() -> None:
@@ -212,7 +284,8 @@ def test_job_token_is_generated_and_injected_only_at_runtime() -> None:
         REPO_ROOT / "services/vault/scripts/vault_k8s_auth_configure.sh"
     ).read_text(encoding="utf-8")
     seeder = (
-        REPO_ROOT / "services/vault/scripts/vault_hermes_jenkins_build_token_ensure.sh"
+        REPO_ROOT
+        / "services/vault-hermes-jenkins-token-seed/scripts/vault_hermes_jenkins_build_token_ensure.sh"
     ).read_text(encoding="utf-8")
     jenkins = (REPO_ROOT / "services/jenkins/deployment.yaml").read_text(
         encoding="utf-8"
@@ -224,10 +297,10 @@ def test_job_token_is_generated_and_injected_only_at_runtime() -> None:
         encoding="utf-8"
     )
     assert "build-token-root:365.v717f8685a_09e" in plugins
-    assert "vault_hermes_jenkins_build_token_ensure.sh" in vault
+    assert 'write_raw_policy "hermes-jenkins-token-seed"' in vault
     assert "sys/tools/random/32 format=hex" in seeder
-    assert "kv put -cas=0" in seeder
-    assert "kv/atlas/hermes/developer-jenkins" in seeder
+    assert '"options":{"cas":0}' in seeder
+    assert "kv/data/atlas/hermes/developer-jenkins" in seeder
     assert "HERMES_AGENT_IMAGE_BUILD_TOKEN={{ .Data.data.build_token }}" in jenkins
     assert "agent-inject-secret-jenkins-image-build-token" in agent
     assert '"jenkins-image-build-token"' in stage

testing/tests/test_hermes_image_builder_adversarial.py

@@ -120,6 +120,74 @@ def test_registry_preflight_rejects_existing_tag_and_auth_failures() -> None:
             )
 
 
+def test_registry_preflight_requires_exact_server_immutable_policy() -> None:
+    """The already-running controller cannot publish before Flux installs the rule."""
+    module = _load(RELEASE_SCRIPT, "hermes_image_release_policy_preflight")
+    expected = {
+        "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",
+                }
+            ]
+        },
+    }
+    captured = {}
+
+    def exact(request, timeout):
+        captured["url"] = request.full_url
+        captured["timeout"] = timeout
+        return Response(
+            200,
+            {"X-Total-Count": "1"},
+            body=json.dumps([{"id": 9, **expected}]).encode(),
+        )
+
+    module.verify_immutable_policy(username="robot", password="private", opener=exact)
+    assert captured["url"].endswith(
+        "/projects/bstein/immutabletagrules?page=1&page_size=100"
+    )
+    assert captured["timeout"] == 20
+
+    for status, body, headers, message in (
+        (403, b"", {}, "HTTP 403"),
+        (200, b"[]", {"X-Total-Count": "0"}, "absent"),
+        (
+            200,
+            json.dumps([{**expected, "disabled": True}]).encode(),
+            {"X-Total-Count": "1"},
+            "not exact",
+        ),
+        (200, b"not-json", {}, "invalid"),
+        (
+            200,
+            json.dumps([expected]).encode(),
+            {"X-Total-Count": "2"},
+            "truncated",
+        ),
+    ):
+        with pytest.raises(RuntimeError, match=message):
+            module.verify_immutable_policy(
+                username="robot",
+                password="private",
+                opener=lambda *_args, code=status, value=body, evidence=headers: Response(
+                    code, evidence, body=value
+                ),
+            )
+
+
 def test_harbor_client_fails_closed_on_input_size_auth_and_json(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
@@ -238,13 +306,53 @@ def test_registry_verification_accepts_exact_pushed_digest() -> None:
             body=json.dumps(
                 {
                     "digest": digest,
-                    "tags": [{"name": destination.rsplit(":", 1)[1]}],
+                    "tags": [
+                        {
+                            "name": destination.rsplit(":", 1)[1],
+                            "immutable": True,
+                        }
+                    ],
                 }
             ).encode("utf-8"),
         ),
     )
 
 
+def test_registry_verification_rejects_tag_without_server_immutability() -> None:
+    """Digest evidence is insufficient unless Harbor reports the build tag immutable."""
+    module = _load(RELEASE_SCRIPT, "hermes_image_release_registry_mutable")
+    digest = "sha256:" + "7" * 64
+    destination = f"{module.DEFAULT_IMAGE}:git-{'8' * 40}-build-19"
+    captured = {}
+
+    def opener(request, _timeout):
+        captured["url"] = request.full_url
+        return Response(
+            200,
+            body=json.dumps(
+                {
+                    "digest": digest,
+                    "tags": [
+                        {
+                            "name": destination.rsplit(":", 1)[1],
+                            "immutable": False,
+                        }
+                    ],
+                }
+            ).encode(),
+        )
+
+    with pytest.raises(RuntimeError, match="immutable"):
+        module.verify_registry_digest(
+            destination,
+            digest,
+            username="robot",
+            password="private",
+            opener=opener,
+        )
+    assert captured["url"].endswith("?with_immutable_status=true")
+
+
 def test_release_main_preflight_uses_fixed_credentials_and_destination(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
@@ -260,7 +368,11 @@ def test_release_main_preflight_uses_fixed_credentials_and_destination(
         captured["destination"] = value
         captured["credentials"] = credentials
 
+    def verify_policy(**credentials):
+        captured["policy_credentials"] = credentials
+
     monkeypatch.setattr(module, "assert_tag_absent", assert_absent)
+    monkeypatch.setattr(module, "verify_immutable_policy", verify_policy)
     monkeypatch.setattr(
         "sys.argv",
         [
@@ -278,6 +390,7 @@ def test_release_main_preflight_uses_fixed_credentials_and_destination(
     assert captured == {
         "destination": destination,
         "credentials": {"username": "robot", "password": "private"},
+        "policy_credentials": {"username": "robot", "password": "private"},
     }
 
 

testing/tests/test_hermes_image_builder_harbor.py

@@ -0,0 +1,325 @@
+"""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")

testing/tests/test_hermes_image_builder_vault.py

@@ -13,7 +13,10 @@ import yaml
 
 REPO_ROOT = Path(__file__).resolve().parents[2]
 VAULT_CONFIG = REPO_ROOT / "services/vault/scripts/vault_k8s_auth_configure.sh"
-SEEDER = REPO_ROOT / "services/vault/scripts/vault_hermes_jenkins_build_token_ensure.sh"
+SEEDER = (
+    REPO_ROOT
+    / "services/vault-hermes-jenkins-token-seed/scripts/vault_hermes_jenkins_build_token_ensure.sh"
+)
 
 
 def test_developer_jenkins_access_is_read_only_and_bound_to_two_consumers() -> None:
@@ -45,19 +48,31 @@ def test_flux_orders_seed_then_jenkins_then_hermes() -> None:
     """The fixed token must be Ready before either consumer is rolled."""
     app_root = REPO_ROOT / "clusters/atlas/flux-system/applications"
     vault = yaml.safe_load((app_root / "vault/kustomization.yaml").read_text())
+    seed = yaml.safe_load(
+        (app_root / "vault-hermes-jenkins-token-seed/kustomization.yaml").read_text()
+    )
     jenkins = yaml.safe_load((app_root / "jenkins/kustomization.yaml").read_text())
     hermes = yaml.safe_load((app_root / "hermes/kustomization.yaml").read_text())
+    role_check = {
+        "apiVersion": "batch/v1",
+        "kind": "Job",
+        "name": "vault-k8s-auth-hermes-9",
+        "namespace": "vault",
+    }
     seed_check = {
         "apiVersion": "batch/v1",
         "kind": "Job",
-        "name": "vault-hermes-jenkins-build-token-seed-1",
+        "name": "vault-hermes-jenkins-build-token-seed-2",
         "namespace": "vault",
     }
-    assert seed_check in vault["spec"]["healthChecks"]
+    assert role_check in vault["spec"]["healthChecks"]
+    assert {item["name"] for item in seed["spec"]["dependsOn"]} == {"vault"}
+    assert seed_check in seed["spec"]["healthChecks"]
     assert jenkins["spec"]["suspend"] is False
     assert {item["name"] for item in jenkins["spec"]["dependsOn"]} >= {
         "helm",
-        "vault",
+        "harbor",
+        "vault-hermes-jenkins-token-seed",
     }
     assert "jenkins" in {item["name"] for item in hermes["spec"]["dependsOn"]}
 
@@ -66,12 +81,13 @@ def test_seed_job_is_revisioned_bounded_and_tracks_fail_closed_script() -> None:
     """The prerequisite is a tracked one-shot on healthy non-reserved capacity."""
     job = yaml.safe_load(
         (
-            REPO_ROOT / "services/vault/hermes-jenkins-build-token-seed-job.yaml"
+            REPO_ROOT / "services/vault-hermes-jenkins-token-seed/job.yaml"
         ).read_text()
     )
-    assert job["metadata"]["name"] == "vault-hermes-jenkins-build-token-seed-1"
+    assert job["metadata"]["name"] == "vault-hermes-jenkins-build-token-seed-2"
     pod = job["spec"]["template"]["spec"]
-    assert pod["serviceAccountName"] == "vault-admin"
+    assert pod["serviceAccountName"] == "hermes-jenkins-token-seed"
+    assert pod["enableServiceLinks"] is False
     assert pod["restartPolicy"] == "Never"
     assert pod["nodeSelector"]["hardware"] == "rpi5"
     expression = pod["affinity"]["nodeAffinity"][
@@ -84,11 +100,33 @@ def test_seed_job_is_revisioned_bounded_and_tracks_fail_closed_script() -> None:
         "titan-19",
         "titan-24",
     }
+    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"]
     source = SEEDER.read_text(encoding="utf-8")
-    assert "kv put -cas=0" in source
-    assert "build_token=-" in source
+    assert '"options":{"cas":0}' in source
+    assert "kv/data/atlas/hermes/developer-jenkins" in source
     assert "kv patch" not in source
     assert "kv delete" not in source
+    assert SEEDER.name not in VAULT_CONFIG.read_text(encoding="utf-8")
+
+
+def test_seed_vault_policy_is_exact_create_read_and_rng_only() -> None:
+    """The seed identity cannot rotate, delete, list, or touch unrelated KV."""
+    source = VAULT_CONFIG.read_text(encoding="utf-8")
+    policy = source.split("hermes_jenkins_token_seed_policy='", 1)[1].split("'", 1)[0]
+    assert 'path "kv/data/atlas/hermes/developer-jenkins"' in policy
+    assert 'capabilities = ["create", "read"]' in policy
+    assert 'path "sys/tools/random/32"' in policy
+    assert 'capabilities = ["update"]' in policy
+    for forbidden in ("delete", "patch", "list", 'kv/data/atlas/hermes/*'):
+        assert forbidden not in policy
+    assert 'bound_service_account_names="hermes-jenkins-token-seed"' in source
+    assert 'bound_service_account_namespaces="vault"' in source
 
 
 def _fake_vault_tools(tmp_path: Path) -> tuple[Path, Path, Path]:
@@ -103,12 +141,13 @@ def _fake_vault_tools(tmp_path: Path) -> tuple[Path, Path, Path]:
 set -eu
 printf '%s\n' "$*" >> "$FAKE_CALL_LOG"
 scenario="$FAKE_SCENARIO"
-if [ "$1" = "write" ]; then
-  printf '%s\n' generated-token-value
+if [ "$1" = "write" ] && [ "$2" = "-field=random_bytes" ] && [ "$3" = "sys/tools/random/32" ]; then
+  printf '%064d\n' 0
   exit 0
 fi
-if [ "$1" = "kv" ] && [ "$2" = "put" ]; then
-  cat > "$FAKE_STDIN_CAPTURE"
+if [ "$1" = "write" ] && [ "$2" = "kv/data/atlas/hermes/developer-jenkins" ]; then
+  payload="${3#@}"
+  cp "$payload" "$FAKE_STDIN_CAPTURE"
   if [ "$scenario" = "cas-race" ]; then
     : > "$FAKE_WINNER"
     echo 'check-and-set parameter did not match' >&2
@@ -116,21 +155,14 @@ if [ "$1" = "kv" ] && [ "$2" = "put" ]; then
   fi
   exit 0
 fi
-if [ "$1" = "kv" ] && [ "$2" = "get" ]; then
-  if [ "${3:-}" = "-field=build_token" ]; then
-    case "$scenario" in
-      existing|cas-race)
-        printf '%s\n' existing-token
-        exit 0
-        ;;
-      missing-field)
-        echo 'No value found at kv/atlas/hermes/developer-jenkins' >&2
-        exit 2
-        ;;
-    esac
-  fi
+if [ "$1" = "read" ] && [ "$2" = "-format=json" ]; then
   case "$scenario" in
-    existing|missing-field)
+    existing)
+      printf '{"data":{"data":{"build_token":"%064d"}}}\n' 1
+      exit 0
+      ;;
+    missing-field)
+      printf '{"data":{"data":{"other":"preserve-me"}}}\n'
       exit 0
       ;;
     absent)
@@ -138,7 +170,10 @@ if [ "$1" = "kv" ] && [ "$2" = "get" ]; then
       exit 2
       ;;
     cas-race)
-      if [ -f "$FAKE_WINNER" ]; then exit 0; fi
+      if [ -f "$FAKE_WINNER" ]; then
+        printf '{"data":{"data":{"build_token":"%064d"}}}\n' 2
+        exit 0
+      fi
       echo 'Code: 404' >&2
       exit 2
       ;;
@@ -182,6 +217,7 @@ def _run_seeder(
         "FAKE_STDIN_CAPTURE": str(stdin_capture),
         "FAKE_WINNER": str(tmp_path / "winner"),
         "FAKE_COUNT": str(tmp_path / "count"),
+        "TMPDIR": str(tmp_path),
     }
     result = subprocess.run(
         ["sh", str(SEEDER)],
@@ -201,7 +237,7 @@ def test_seeder_never_overwrites_existing_or_ambiguous_state(
     """Existing fields and persistent read failures can never become a put."""
     result, calls = _run_seeder(tmp_path, scenario)
     assert (result.returncode == 0) is (scenario == "existing")
-    assert "kv put" not in calls
+    assert "write kv/data/atlas/hermes/developer-jenkins" not in calls
     assert "sys/tools/random" not in calls
 
 
@@ -211,10 +247,13 @@ def test_seeder_uses_single_cas_create_and_stdin_for_absent_secret(
     """A confirmed 404 produces one create-only write without a token argument."""
     result, calls = _run_seeder(tmp_path, "absent")
     assert result.returncode == 0, result.stderr
-    assert calls.count("kv put") == 1
-    assert "kv put -cas=0 kv/atlas/hermes/developer-jenkins build_token=-" in calls
-    assert "generated-token-value" not in calls
-    assert (tmp_path / "stdin").read_text(encoding="utf-8") == "generated-token-value"
+    assert calls.count("write kv/data/atlas/hermes/developer-jenkins") == 1
+    assert "@" in calls
+    assert "0000000000000000000000000000000000000000000000000000000000000000" not in calls
+    payload = (tmp_path / "stdin").read_text(encoding="utf-8")
+    assert '"options":{"cas":0}' in payload
+    assert '"build_token":"' in payload
+    assert not (tmp_path / "hermes-jenkins-token.json").exists()
 
 
 def test_seeder_retries_reads_and_accepts_only_a_verified_cas_winner(
@@ -223,11 +262,11 @@ def test_seeder_retries_reads_and_accepts_only_a_verified_cas_winner(
     """Transient reads retry; a competing create is accepted only after reread."""
     transient, transient_calls = _run_seeder(tmp_path / "transient", "transient")
     assert transient.returncode == 0, transient.stderr
-    assert transient_calls.count("kv get kv/atlas/hermes/developer-jenkins") == 3
-    assert transient_calls.count("kv put") == 1
+    assert transient_calls.count("read -format=json kv/data/atlas/hermes/developer-jenkins") == 3
+    assert transient_calls.count("write kv/data/atlas/hermes/developer-jenkins") == 1
 
     race, race_calls = _run_seeder(tmp_path / "race", "cas-race")
     assert race.returncode == 0, race.stderr
-    assert race_calls.count("kv put") == 1
-    assert "kv get -field=build_token" in race_calls
+    assert race_calls.count("write kv/data/atlas/hermes/developer-jenkins") == 1
+    assert race_calls.count("read -format=json kv/data/atlas/hermes/developer-jenkins") == 2
     assert "another seeder won CAS" in race.stderr