hermes: isolate Atlas SCM write authority

clusters/atlas/flux-system/applications/hermes-observer-rbac/kustomization.yaml

@@ -14,5 +14,3 @@ spec:
     namespace: flux-system
   wait: true
   timeout: 5m
-  dependsOn:
-    - name: hermes

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

@@ -61,3 +61,4 @@ spec:
     - name: longhorn
     - name: vault
     - name: hermes-scm-broker
+    - name: hermes-observer-rbac

services/gitea/atlas-identity-bootstrap-job.yaml

@@ -2,7 +2,7 @@
 apiVersion: batch/v1
 kind: Job
 metadata:
-  name: gitea-atlas-identity-bootstrap-4
+  name: gitea-atlas-identity-bootstrap-5
   namespace: gitea
   labels:
     app.kubernetes.io/name: gitea
@@ -34,6 +34,20 @@ spec:
       nodeSelector:
         node-role.kubernetes.io/worker: "true"
         hardware: rpi5
+      initContainers:
+        - name: python-runtime
+          image: python@sha256:6d43704baacd1bfbe7c295d7f13079d5d8104ed33568873133f8fc69980419df
+          command: [/bin/sh, -ec]
+          args: ["cp -R /usr/local/. /python/"]
+          securityContext:
+            allowPrivilegeEscalation: false
+            capabilities: {drop: [ALL]}
+            readOnlyRootFilesystem: true
+            runAsNonRoot: true
+            runAsUser: 1000
+            runAsGroup: 1000
+          volumeMounts:
+            - {name: python-runtime, mountPath: /python}
       containers:
         - name: bootstrap
           image: gitea/gitea:1.23
@@ -80,6 +94,7 @@ spec:
             - name: bootstrap
               mountPath: /opt/bootstrap
               readOnly: true
+            - {name: python-runtime, mountPath: /opt/python, readOnly: true}
       volumes:
         - name: data
           persistentVolumeClaim:
@@ -88,3 +103,6 @@ spec:
           configMap:
             name: gitea-atlas-identity-bootstrap
             defaultMode: 0555
+        - name: python-runtime
+          emptyDir:
+            sizeLimit: 128Mi

services/gitea/kustomization.yaml

@@ -15,6 +15,7 @@ configMapGenerator:
   - name: gitea-atlas-identity-bootstrap
     files:
       - gitea_atlas_identity_ensure.sh=scripts/gitea_atlas_identity_ensure.sh
+      - gitea_branch_protection_check.py=scripts/gitea_branch_protection_check.py
 
 generatorOptions:
   disableNameSuffixHash: true

services/gitea/scripts/gitea_atlas_identity_ensure.sh

@@ -277,7 +277,6 @@ ensure_default_branch_protection() {
   repository=$1
   repo_file=$(mktemp)
   protections_file=$(mktemp)
-  protection_file=$(mktemp)
   response_file=$(mktemp)
 
   status=$(api_request GET "/repos/${organization}/${repository}" '' "${repo_file}")
@@ -290,44 +289,28 @@ ensure_default_branch_protection() {
 
   status=$(api_request GET "/repos/${organization}/${repository}/branch_protections" '' "${protections_file}")
   expect_status "${status}" 200 "list ${repository} branch protections"
-  tr -d '[:space:]' <"${protections_file}" | sed 's/},{/}\
-{/g' | grep -F "\"rule_name\":\"${default_branch}\"" >"${protection_file}" || true
-  matching_rules=$(grep -c . "${protection_file}" || true)
-  [ "${matching_rules}" -le 1 ] || die "${repository} has duplicate ${default_branch} protection rules"
+  if ! protection_state=$(
+    /opt/python/bin/python3 /opt/bootstrap/gitea_branch_protection_check.py \
+      "${protections_file}" "${default_branch}" "${protected_reviewer}"
+  ); then
+    die "${repository} ${default_branch} protection differs from the human-review policy"
+  fi
 
-  if [ "${matching_rules}" -eq 0 ]; then
+  if [ "${protection_state}" = ABSENT ]; then
     payload="{\"rule_name\":\"${default_branch}\",\"enable_push\":true,\"enable_push_whitelist\":true,\"push_whitelist_usernames\":[\"${protected_reviewer}\"],\"push_whitelist_teams\":[],\"push_whitelist_deploy_keys\":false,\"enable_force_push\":false,\"enable_force_push_allowlist\":false,\"force_push_allowlist_usernames\":[],\"force_push_allowlist_teams\":[],\"force_push_allowlist_deploy_keys\":false,\"enable_merge_whitelist\":true,\"merge_whitelist_usernames\":[\"${protected_reviewer}\"],\"merge_whitelist_teams\":[],\"enable_approvals_whitelist\":true,\"approvals_whitelist_username\":[\"${protected_reviewer}\"],\"approvals_whitelist_teams\":[],\"required_approvals\":1,\"block_on_rejected_reviews\":true,\"block_on_outdated_branch\":true,\"dismiss_stale_approvals\":true,\"block_admin_merge_override\":true}"
     status=$(api_request POST "/repos/${organization}/${repository}/branch_protections" "${payload}" "${response_file}")
     expect_status "${status}" 201 "protect ${repository} ${default_branch}"
     status=$(api_request GET "/repos/${organization}/${repository}/branch_protections" '' "${protections_file}")
     expect_status "${status}" 200 "read back ${repository} branch protections"
-    tr -d '[:space:]' <"${protections_file}" | sed 's/},{/}\
-{/g' | grep -F "\"rule_name\":\"${default_branch}\"" >"${protection_file}" || true
+    protection_state=$(
+      /opt/python/bin/python3 /opt/bootstrap/gitea_branch_protection_check.py \
+        "${protections_file}" "${default_branch}" "${protected_reviewer}"
+    ) || die "${repository} ${default_branch} protection differs from the human-review policy"
   fi
+  [ "${protection_state}" = PRESENT ] || \
+    die "${repository} ${default_branch} protection did not become effective"
 
-  compact=$(tr -d '[:space:]' <"${protection_file}")
-  for required in \
-    "\"rule_name\":\"${default_branch}\"" \
-    '"enable_push":true' \
-    '"enable_push_whitelist":true' \
-    "\"push_whitelist_usernames\":[\"${protected_reviewer}\"]" \
-    '"push_whitelist_deploy_keys":false' \
-    '"enable_force_push":false' \
-    '"enable_merge_whitelist":true' \
-    "\"merge_whitelist_usernames\":[\"${protected_reviewer}\"]" \
-    '"enable_approvals_whitelist":true' \
-    "\"approvals_whitelist_username\":[\"${protected_reviewer}\"]" \
-    '"required_approvals":1' \
-    '"block_on_rejected_reviews":true' \
-    '"block_on_outdated_branch":true' \
-    '"dismiss_stale_approvals":true' \
-    '"block_admin_merge_override":true'
-  do
-    printf '%s' "${compact}" | grep -Fq "${required}" || \
-      die "${repository} ${default_branch} protection differs from the human-review policy"
-  done
-
-  rm -f "${repo_file}" "${protections_file}" "${protection_file}" "${response_file}"
+  rm -f "${repo_file}" "${protections_file}" "${response_file}"
 }
 
 wait_for_gitea

services/gitea/scripts/gitea_branch_protection_check.py

@@ -0,0 +1,138 @@
+#!/usr/bin/env python3
+"""Validate the first effective Gitea protection for an Atlas primary branch."""
+
+from __future__ import annotations
+
+import argparse
+import fnmatch
+import json
+import os
+import stat
+from pathlib import Path
+
+MAX_INPUT = 1024 * 1024
+MAX_RULES = 100
+MAX_RULE_NAME = 255
+
+
+class PolicyError(RuntimeError):
+    """The protection response is ambiguous or violates human review policy."""
+
+
+def _read_bounded(path: Path) -> bytes:
+    descriptor = os.open(path, os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0))
+    try:
+        metadata = os.fstat(descriptor)
+        if not stat.S_ISREG(metadata.st_mode) or metadata.st_size > MAX_INPUT:
+            raise PolicyError("branch protection response exceeds the safe limit")
+        value = os.read(descriptor, MAX_INPUT + 1)
+    finally:
+        os.close(descriptor)
+    if len(value) != metadata.st_size:
+        raise PolicyError("branch protection response changed while reading")
+    return value
+
+
+def _matches(pattern: str, branch: str) -> bool:
+    if not pattern.isascii() or not 1 <= len(pattern) <= MAX_RULE_NAME:
+        raise PolicyError("branch protection rule name is invalid")
+    if pattern.count("[") != pattern.count("]") or "\\" in pattern:
+        raise PolicyError("branch protection glob is ambiguous")
+    # Gitea matches ordinary names case-insensitively and compiles wildcard
+    # names as globs with '/' as a separator. Primary branches have no slash,
+    # so fnmatchcase is equivalent for the only two targets managed here.
+    patterns = [pattern]
+    while any("{" in item or "}" in item for item in patterns):
+        expanded: list[str] = []
+        for item in patterns:
+            if "{" not in item and "}" not in item:
+                expanded.append(item)
+                continue
+            start = item.find("{")
+            end = item.find("}", start + 1)
+            if start < 0 or end < 0 or "{" in item[start + 1 : end]:
+                raise PolicyError("branch protection glob is ambiguous")
+            choices = item[start + 1 : end].split(",")
+            if len(choices) < 2 or any(not choice for choice in choices):
+                raise PolicyError("branch protection glob is ambiguous")
+            expanded.extend(item[:start] + choice + item[end + 1 :] for choice in choices)
+        patterns = expanded
+        if len(patterns) > 32:
+            raise PolicyError("branch protection glob expands beyond the safe limit")
+    return any(
+        fnmatch.fnmatchcase(branch.casefold(), item.casefold()) for item in patterns
+    )
+
+
+def _required(reviewer: str) -> dict[str, object]:
+    return {
+        "enable_push": True,
+        "enable_push_whitelist": True,
+        "push_whitelist_usernames": [reviewer],
+        "push_whitelist_deploy_keys": False,
+        "enable_force_push": False,
+        "enable_merge_whitelist": True,
+        "merge_whitelist_usernames": [reviewer],
+        "enable_approvals_whitelist": True,
+        "approvals_whitelist_username": [reviewer],
+        "required_approvals": 1,
+        "block_on_rejected_reviews": True,
+        "block_on_outdated_branch": True,
+        "dismiss_stale_approvals": True,
+        "block_admin_merge_override": True,
+    }
+
+
+def evaluate(value: bytes, branch: str, reviewer: str) -> str:
+    """Return PRESENT/ABSENT or fail on ambiguous/effective policy drift."""
+    if len(value) > MAX_INPUT or branch not in {"main", "master"}:
+        raise PolicyError("branch protection input is invalid")
+    data = json.loads(value)
+    if not isinstance(data, list) or len(data) > MAX_RULES:
+        raise PolicyError("branch protection response is invalid")
+    ordered: list[tuple[int, int, dict[str, object]]] = []
+    priorities: set[int] = set()
+    for index, raw in enumerate(data):
+        if not isinstance(raw, dict):
+            raise PolicyError("branch protection entry is invalid")
+        priority = raw.get("priority")
+        rule_name = raw.get("rule_name")
+        if (
+            not isinstance(priority, int)
+            or isinstance(priority, bool)
+            or not 1 <= priority <= 1_000_000
+            or priority in priorities
+            or not isinstance(rule_name, str)
+        ):
+            raise PolicyError("branch protection priority is ambiguous")
+        priorities.add(priority)
+        if _matches(rule_name, branch):
+            ordered.append((priority, index, raw))
+    if not ordered:
+        return "ABSENT"
+    effective = min(ordered)[2]
+    for key, expected in _required(reviewer).items():
+        if effective.get(key) != expected:
+            raise PolicyError(
+                f"effective {branch} protection differs from human-review policy"
+            )
+    return "PRESENT"
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("response", type=Path)
+    parser.add_argument("branch", choices=("main", "master"))
+    parser.add_argument("reviewer")
+    args = parser.parse_args()
+    try:
+        value = _read_bounded(args.response)
+        print(evaluate(value, args.branch, args.reviewer))
+    except (OSError, json.JSONDecodeError, PolicyError) as exc:
+        print(f"branch protection check failed: {exc}", file=__import__("sys").stderr)
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

services/hermes-scm-broker/deployment.yaml

@@ -73,6 +73,7 @@ spec:
             httpGet: {path: /healthz, port: http}
             initialDelaySeconds: 15
             periodSeconds: 30
+            failureThreshold: 10
           securityContext:
             allowPrivilegeEscalation: false
             capabilities:

services/hermes/kustomization.yaml

@@ -106,6 +106,8 @@ configMapGenerator:
     namespace: hermes
     files:
       - node_account_hardening.py=scripts/node_account_hardening.py
+      - node_account_audit.py=scripts/node_account_audit.py
+      - node_account_io.py=scripts/node_account_io.py
     options:
       disableNameSuffixHash: true
   - name: hermes-auto-router-plugin

services/hermes/node-ssh-access.yaml

@@ -45,13 +45,14 @@ spec:
         - operator: Exists
       containers:
         - name: key-reconciler
-          image: registry.bstein.dev/bstein/hermes-agent
+          # The pinned index supports the cluster's amd64 and arm64 nodes.
+          image: python@sha256:6d43704baacd1bfbe7c295d7f13079d5d8104ed33568873133f8fc69980419df
           imagePullPolicy: IfNotPresent
           command: [/bin/sh, -ec]
           args:
             - |
               while true; do
-                /opt/hermes/.venv/bin/python \
+                /usr/local/bin/python \
                   /opt/node-hardener/node_account_hardening.py \
                   --public-key-file /vault/secrets/node-ssh-public-key
                 sleep 300
@@ -68,6 +69,9 @@ spec:
               mountPath: /host-home
             - name: host-etc
               mountPath: /host-etc
+            - name: host-polkit-share
+              mountPath: /host-polkit-share
+              readOnly: true
             - name: host-k3s
               mountPath: /host-k3s
             - name: host-kubelet
@@ -100,6 +104,10 @@ spec:
           hostPath:
             path: /etc
             type: Directory
+        - name: host-polkit-share
+          hostPath:
+            path: /usr/share/polkit-1
+            type: Directory
         - name: host-k3s
           hostPath:
             path: /var/lib/rancher/k3s

services/hermes/scm-common/kustomization.yaml

@@ -7,6 +7,8 @@ configMapGenerator:
       - gitea_api.py=scripts/gitea_api.py
       - gitea_api_policy.py=scripts/gitea_api_policy.py
       - scm_broker.py=scripts/scm_broker.py
+      - scm_broker_io.py=scripts/scm_broker_io.py
+      - scm_broker_server.py=scripts/scm_broker_server.py
       - scm_broker_client.py=scripts/scm_broker_client.py
     options:
       disableNameSuffixHash: true

services/hermes/scm-common/scripts/scm_broker.py

@@ -7,10 +7,14 @@ import argparse
 import base64
 import json
 import re
+import socket
+import tempfile
+import time
 import urllib.error
 import urllib.parse
 import urllib.request
-from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from http.server import BaseHTTPRequestHandler
+from typing import BinaryIO
 
 from gitea_api import (
     CANONICAL_BASE_URL,
@@ -20,6 +24,8 @@ from gitea_api import (
     read_token,
 )
 from gitea_api_policy import _reject_forbidden, _validate_ref, _validate_repo
+from scm_broker_io import spool_response
+from scm_broker_server import BoundedThreadingHTTPServer
 
 BROKER_PORT = 9081
 GIT_USER = "hermes-automation"
@@ -27,6 +33,12 @@ MAX_CONTROL_BODY = 64 * 1024
 MAX_GIT_REQUEST = 128 * 1024 * 1024
 MAX_GIT_RESPONSE = 128 * 1024 * 1024
 MAX_PUSH_COMMANDS = 16
+MAX_HEADERS = 32
+MAX_HEADER_BYTES = 16 * 1024
+SPOOL_MEMORY_LIMIT = 1024 * 1024
+STREAM_CHUNK = 64 * 1024
+INBOUND_HEADER_TIMEOUT = 10.0
+INBOUND_BODY_TIMEOUT = 120.0
 ZERO_SHA = b"0" * 40
 GIT_PATH_RE = re.compile(
     r"/git/atlas/(?P<repo>[A-Za-z0-9][A-Za-z0-9._-]{0,99})\.git/"
@@ -39,7 +51,6 @@ FEATURE_REF_RE = re.compile(
 
 class RejectRedirect(urllib.request.HTTPRedirectHandler):
     """Reject every upstream redirect before credentials can be forwarded."""
-
     def redirect_request(self, req, fp, code, msg, headers, newurl):
         raise PolicyError("SCM upstream redirects are not allowed")
 
@@ -54,22 +65,105 @@ def _status(response: object) -> int | None:
     return value
 
 
-def _read_bounded(stream, maximum: int, length: int | None = None) -> bytes:
+def _read_bounded(
+    stream,
+    maximum: int,
+    length: int | None = None,
+    *,
+    deadline: float | None = None,
+    set_timeout=None,
+) -> bytes:
     if length is not None and not 0 <= length <= maximum:
         raise PolicyError("SCM request exceeds the safe size limit")
-    value = stream.read(maximum + 1 if length is None else length)
+    wanted = maximum + 1 if length is None else length
+    chunks: list[bytes] = []
+    total = 0
+    while total < wanted:
+        if deadline is not None:
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                raise PolicyError("SCM request body deadline exceeded")
+            if set_timeout is not None:
+                set_timeout(remaining)
+        chunk = stream.read(min(STREAM_CHUNK, wanted - total))
+        if not chunk:
+            break
+        chunks.append(chunk)
+        total += len(chunk)
+    value = b"".join(chunks)
     if len(value) > maximum or (length is not None and len(value) != length):
         raise PolicyError("SCM request exceeds the safe size limit")
     return value
 
 
+def _spool_bounded(
+    stream,
+    maximum: int,
+    length: int,
+    *,
+    token: str,
+    context: str,
+    deadline: float,
+    set_timeout=None,
+) -> tuple[BinaryIO, int]:
+    """Copy a fixed-length exchange through a bounded-memory disk spool."""
+    if not 0 <= length <= maximum:
+        raise PolicyError("SCM request exceeds the safe size limit")
+    spool = tempfile.SpooledTemporaryFile(max_size=SPOOL_MEMORY_LIMIT, dir="/tmp")
+    remaining = length
+    forms = _credential_forms(token)
+    carry = b""
+    try:
+        while remaining:
+            remaining_time = deadline - time.monotonic()
+            if remaining_time <= 0:
+                raise PolicyError("SCM request body deadline exceeded")
+            if set_timeout is not None:
+                set_timeout(remaining_time)
+            chunk = stream.read(min(STREAM_CHUNK, remaining))
+            if not chunk:
+                raise PolicyError("SCM request body ended early")
+            candidate = carry + chunk
+            if any(form in candidate for form in forms):
+                raise PolicyError(f"{context} contains runtime credential material")
+            width = max(len(form) for form in forms) - 1
+            carry = candidate[-width:] if width else b""
+            spool.write(chunk)
+            remaining -= len(chunk)
+        spool.seek(0)
+        return spool, length
+    except Exception:
+        spool.close()
+        raise
+
+
+def _spool_response(response, maximum: int, token: str) -> tuple[BinaryIO, int]:
+    return spool_response(
+        response,
+        maximum,
+        _credential_forms(token),
+        memory_limit=SPOOL_MEMORY_LIMIT,
+        chunk_size=STREAM_CHUNK,
+        deadline_seconds=INBOUND_BODY_TIMEOUT,
+    )
+
+
 def _load_json(handler: BaseHTTPRequestHandler) -> dict[str, object]:
     if handler.headers.get("Transfer-Encoding"):
         raise PolicyError("chunked broker control requests are not allowed")
     if handler.headers.get_content_type() != "application/json":
         raise PolicyError("broker control request must be JSON")
     length = _content_length(handler.headers, MAX_CONTROL_BODY)
-    value = json.loads(_read_bounded(handler.rfile, MAX_CONTROL_BODY, length))
+    deadline = time.monotonic() + INBOUND_BODY_TIMEOUT
+    value = json.loads(
+        _read_bounded(
+            handler.rfile,
+            MAX_CONTROL_BODY,
+            length,
+            deadline=deadline,
+            set_timeout=handler.connection.settimeout,
+        )
+    )
     if not isinstance(value, dict):
         raise PolicyError("broker control request must be an object")
     return value
@@ -133,8 +227,20 @@ def _reject_credential_bytes(value: bytes, token: str, context: str) -> None:
         raise PolicyError(f"{context} contains runtime credential material")
 
 
-def _validate_receive_pack(body: bytes, token: str) -> None:
+def _receive_prefix(body: bytes | BinaryIO) -> bytes:
+    if isinstance(body, bytes):
+        return body
+    position = body.tell()
+    try:
+        body.seek(0)
+        return body.read(64 * 1024)
+    finally:
+        body.seek(position)
+
+
+def _validate_receive_pack(body: bytes | BinaryIO, token: str) -> None:
     """Permit only creation of new, namespaced feature branches."""
+    body = _receive_prefix(body)
     _reject_credential_bytes(body, token, "Git request")
     position = 0
     commands = 0
@@ -175,16 +281,24 @@ def _upstream_git_request(
     target: str,
     *,
     method: str,
-    body: bytes | None,
+    body: bytes | BinaryIO | None,
     content_type: str | None,
     expected_type: str,
     token: str,
+    body_length: int | None = None,
+    stream_result: bool = False,
     opener=UPSTREAM_OPENER.open,
-) -> bytes:
+) -> bytes | tuple[BinaryIO, int]:
     credentials = base64.b64encode(f"{GIT_USER}:{token}".encode()).decode("ascii")
     headers = {"Accept": expected_type, "Authorization": f"Basic {credentials}"}
     if content_type:
         headers["Content-Type"] = content_type
+    if body is not None:
+        if body_length is None:
+            if not isinstance(body, bytes):
+                raise PolicyError("Git upstream request length is missing")
+            body_length = len(body)
+        headers["Content-Length"] = str(body_length)
     request = urllib.request.Request(
         CANONICAL_BASE_URL + target,
         data=body,
@@ -196,9 +310,13 @@ def _upstream_git_request(
             raise PolicyError("Git upstream returned an unexpected HTTP status")
         if response.headers.get_content_type() != expected_type:
             raise PolicyError("Git upstream returned an unexpected response type")
-        result = _read_bounded(response, MAX_GIT_RESPONSE)
-    _reject_credential_bytes(result, token, "Git upstream response")
-    return result
+        result, result_length = _spool_response(response, MAX_GIT_RESPONSE, token)
+    if stream_result:
+        return result, result_length
+    try:
+        return result.read()
+    finally:
+        result.close()
 
 
 class BrokerHandler(BaseHTTPRequestHandler):
@@ -207,6 +325,42 @@ class BrokerHandler(BaseHTTPRequestHandler):
     server_version = "HermesSCMBroker/1"
     sys_version = ""
 
+    def setup(self) -> None:
+        super().setup()
+        self.connection.settimeout(INBOUND_HEADER_TIMEOUT)
+
+    def _validate_headers(self) -> None:
+        items = list(self.headers.items())
+        if len(items) > MAX_HEADERS:
+            raise PolicyError("SCM request has too many headers")
+        total = 0
+        for name, value in items:
+            if not name.isascii() or not value.isascii():
+                raise PolicyError("SCM request headers must be ASCII")
+            total += len(name) + len(value) + 4
+            if any(ord(character) < 32 and character != "\t" for character in value):
+                raise PolicyError("SCM request header contains controls")
+        if total > MAX_HEADER_BYTES:
+            raise PolicyError("SCM request headers exceed the safe limit")
+        if len(self.headers.get_all("Content-Length", [])) > 1:
+            raise PolicyError("SCM request has duplicate Content-Length")
+
+    def _stream(self, status: int, content_type: str, body: BinaryIO, length: int) -> None:
+        self.send_response(status)
+        self.send_header("Content-Type", content_type)
+        self.send_header("Content-Length", str(length))
+        self.end_headers()
+        deadline = time.monotonic() + INBOUND_BODY_TIMEOUT
+        while True:
+            chunk = body.read(STREAM_CHUNK)
+            if not chunk:
+                break
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                raise PolicyError("SCM response write deadline exceeded")
+            self.connection.settimeout(remaining)
+            self.wfile.write(chunk)
+
     def log_message(self, _format: str, *_args: object) -> None:
         return
 
@@ -223,6 +377,7 @@ class BrokerHandler(BaseHTTPRequestHandler):
 
     def do_GET(self) -> None:
         try:
+            self._validate_headers()
             if self.path == "/healthz":
                 self._json(200, b'{"status":"ok"}')
                 return
@@ -232,29 +387,32 @@ class BrokerHandler(BaseHTTPRequestHandler):
             token = read_token()
             _reject_forbidden(repo, "repository", (token,))
             expected = f"application/x-{service}-advertisement"
-            body = _upstream_git_request(
+            streamed = _upstream_git_request(
                 f"/atlas/{repo}.git/info/refs?service={service}",
                 method="GET",
                 body=None,
                 content_type=None,
                 expected_type=expected,
                 token=token,
+                stream_result=True,
             )
-            self.send_response(200)
-            self.send_header("Content-Type", expected)
-            self.send_header("Content-Length", str(len(body)))
-            self.end_headers()
-            self.wfile.write(body)
-        except (OSError, PolicyError, urllib.error.URLError, ValueError):
+            body, length = streamed
+            try:
+                self._stream(200, expected, body, length)
+            finally:
+                body.close()
+        except (OSError, PolicyError, socket.timeout, urllib.error.URLError, ValueError):
             self._reject()
 
     def do_POST(self) -> None:
         try:
+            self._validate_headers()
+            self.connection.settimeout(INBOUND_BODY_TIMEOUT)
             if self.path in {"/v1/metadata", "/v1/drafts"}:
                 self._control()
             else:
                 self._git_rpc()
-        except (OSError, PolicyError, urllib.error.URLError, ValueError, json.JSONDecodeError):
+        except (OSError, PolicyError, socket.timeout, urllib.error.URLError, ValueError, json.JSONDecodeError):
             self._reject()
 
     def _control(self) -> None:
@@ -281,27 +439,38 @@ class BrokerHandler(BaseHTTPRequestHandler):
         if self.headers.get_content_type() != expected_request or self.headers.get("Transfer-Encoding"):
             raise PolicyError("Git RPC request type is invalid")
         length = _content_length(self.headers, MAX_GIT_REQUEST)
-        body = _read_bounded(self.rfile, MAX_GIT_REQUEST, length)
         token = read_token()
         _reject_forbidden(repo, "repository", (token,))
-        if service == "git-receive-pack":
-            _validate_receive_pack(body, token)
-        else:
-            _reject_credential_bytes(body, token, "Git request")
-        expected = f"application/x-{service}-result"
-        result = _upstream_git_request(
-            f"/atlas/{repo}.git/{service}",
-            method="POST",
-            body=body,
-            content_type=expected_request,
-            expected_type=expected,
+        body, body_length = _spool_bounded(
+            self.rfile,
+            MAX_GIT_REQUEST,
+            length,
             token=token,
+            context="Git request",
+            deadline=time.monotonic() + INBOUND_BODY_TIMEOUT,
+            set_timeout=self.connection.settimeout,
         )
-        self.send_response(200)
-        self.send_header("Content-Type", expected)
-        self.send_header("Content-Length", str(len(result)))
-        self.end_headers()
-        self.wfile.write(result)
+        try:
+            if service == "git-receive-pack":
+                _validate_receive_pack(body, token)
+            expected = f"application/x-{service}-result"
+            streamed = _upstream_git_request(
+                f"/atlas/{repo}.git/{service}",
+                method="POST",
+                body=body,
+                body_length=body_length,
+                content_type=expected_request,
+                expected_type=expected,
+                token=token,
+                stream_result=True,
+            )
+            result, result_length = streamed
+            try:
+                self._stream(200, expected, result, result_length)
+            finally:
+                result.close()
+        finally:
+            body.close()
 
 
 def main() -> int:
@@ -309,7 +478,7 @@ def main() -> int:
     parser.add_argument("--listen", default="0.0.0.0")
     parser.add_argument("--port", type=int, default=BROKER_PORT)
     args = parser.parse_args()
-    ThreadingHTTPServer((args.listen, args.port), BrokerHandler).serve_forever()
+    BoundedThreadingHTTPServer((args.listen, args.port), BrokerHandler).serve_forever()
     return 0
 
 

services/hermes/scm-common/scripts/scm_broker_io.py

@@ -0,0 +1,54 @@
+#!/usr/bin/env python3
+"""Bounded-memory stream helpers for the Hermes SCM broker."""
+
+from __future__ import annotations
+
+import tempfile
+import time
+from typing import BinaryIO
+
+from gitea_api_policy import PolicyError
+
+
+def spool_response(
+    response,
+    maximum: int,
+    forbidden: tuple[bytes, ...],
+    *,
+    memory_limit: int,
+    chunk_size: int,
+    deadline_seconds: float,
+) -> tuple[BinaryIO, int]:
+    """Spool an upstream response to disk while scanning chunk boundaries."""
+    spool = tempfile.SpooledTemporaryFile(max_size=memory_limit, dir="/tmp")
+    carry = b""
+    total = 0
+    deadline = time.monotonic() + deadline_seconds
+    try:
+        while True:
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                raise PolicyError("SCM upstream response deadline exceeded")
+            raw = getattr(getattr(response, "fp", None), "raw", None)
+            sock = getattr(raw, "_sock", raw)
+            if hasattr(sock, "settimeout"):
+                sock.settimeout(remaining)
+            chunk = response.read(chunk_size)
+            if not chunk:
+                break
+            total += len(chunk)
+            if total > maximum:
+                raise PolicyError("SCM upstream response exceeds the safe size limit")
+            candidate = carry + chunk
+            if any(value in candidate for value in forbidden):
+                raise PolicyError(
+                    "Git upstream response contains runtime credential material"
+                )
+            width = max(len(value) for value in forbidden) - 1
+            carry = candidate[-width:] if width else b""
+            spool.write(chunk)
+        spool.seek(0)
+        return spool, total
+    except Exception:
+        spool.close()
+        raise

services/hermes/scm-common/scripts/scm_broker_server.py

@@ -0,0 +1,44 @@
+#!/usr/bin/env python3
+"""Concurrency-bounded threaded HTTP server for the Hermes SCM broker."""
+
+from __future__ import annotations
+
+import threading
+from http.server import ThreadingHTTPServer
+
+MAX_CONCURRENT_REQUESTS = 2
+
+
+class BoundedThreadingHTTPServer(ThreadingHTTPServer):
+    """Cap active handlers so large Git exchanges cannot exhaust memory."""
+
+    daemon_threads = True
+    block_on_close = True
+    request_queue_size = 16
+
+    def __init__(self, *args, **kwargs):
+        self._slots = threading.BoundedSemaphore(MAX_CONCURRENT_REQUESTS)
+        super().__init__(*args, **kwargs)
+
+    def process_request(self, request, client_address) -> None:
+        if not self._slots.acquire(blocking=False):
+            try:
+                request.sendall(
+                    b"HTTP/1.1 503 Service Unavailable\r\n"
+                    b"Connection: close\r\nContent-Length: 0\r\n\r\n"
+                )
+            finally:
+                self.shutdown_request(request)
+            return
+        try:
+            super().process_request(request, client_address)
+        except Exception:
+            self._slots.release()
+            self.shutdown_request(request)
+            raise
+
+    def process_request_thread(self, request, client_address) -> None:
+        try:
+            super().process_request_thread(request, client_address)
+        finally:
+            self._slots.release()

services/hermes/scripts/node_account_audit.py

@@ -0,0 +1,138 @@
+#!/usr/bin/env python3
+"""Fail-closed supplementary-group and host privilege-policy audits."""
+
+from __future__ import annotations
+
+import stat
+import struct
+from pathlib import Path
+
+from node_account_io import HardeningError, read_regular
+
+
+def _members(field: str, context: str) -> set[str]:
+    if not field:
+        return set()
+    values = field.split(",")
+    if any(not value or value.strip() != value for value in values):
+        raise HardeningError(f"{context} has a malformed member list")
+    if len(values) != len(set(values)):
+        raise HardeningError(f"{context} has duplicate members")
+    return set(values)
+
+
+def audit_membership(
+    account: str, groups: list[list[str]], gshadow: list[list[str]]
+) -> None:
+    """Reject the dedicated account in any supplementary group/admin list."""
+    for record in groups:
+        if account in _members(record[3], f"group {record[0]}"):
+            raise HardeningError("dedicated Hermes account has supplementary group access")
+    for record in gshadow:
+        principals = _members(record[2], f"gshadow {record[0]} admins")
+        principals |= _members(record[3], f"gshadow {record[0]} members")
+        if account in principals:
+            raise HardeningError("dedicated Hermes account has gshadow group access")
+
+
+def _policy_files(host_etc: Path, polkit_share: Path, root_uid: int) -> list[Path]:
+    files: list[Path] = []
+    sudoers = host_etc / "sudoers"
+    if sudoers.exists():
+        files.append(sudoers)
+    roots = (
+        host_etc / "sudoers.d",
+        host_etc / "polkit-1/rules.d",
+        host_etc / "polkit-1/localauthority",
+        polkit_share / "rules.d",
+        polkit_share / "localauthority",
+    )
+    for root in roots:
+        if not root.exists():
+            continue
+        metadata = root.lstat()
+        if (
+            not stat.S_ISDIR(metadata.st_mode)
+            or stat.S_ISLNK(metadata.st_mode)
+            or metadata.st_uid != root_uid
+            or stat.S_IMODE(metadata.st_mode) & 0o022
+        ):
+            raise HardeningError("unsafe sudo/polkit policy directory")
+        files.extend(path for path in root.rglob("*") if path.is_file())
+    if len(files) > 256:
+        raise HardeningError("too many sudo/polkit policy files")
+    return sorted(files)
+
+
+def _audit_policy_metadata(snapshot, root_uid: int, account_uid: int) -> None:
+    if snapshot.uid != root_uid or snapshot.mode & 0o022:
+        raise HardeningError("sudo/polkit policy metadata permits unsafe mutation")
+    acl = dict(snapshot.xattrs).get("system.posix_acl_access", b"")
+    if not acl:
+        return
+    header = struct.Struct("<I")
+    entry = struct.Struct("<HHI")
+    if len(acl) < header.size or (len(acl) - header.size) % entry.size:
+        raise HardeningError("sudo/polkit policy ACL is malformed")
+    for offset in range(header.size, len(acl), entry.size):
+        tag, permissions, identifier = entry.unpack_from(acl, offset)
+        if tag == 0x02 and identifier == account_uid and permissions & 0o2:
+            raise HardeningError("sudo/polkit policy ACL grants Hermes write access")
+
+
+def audit_privilege_policies(
+    account: str,
+    account_uid: int,
+    root_uid: int,
+    host_etc: Path,
+    polkit_share: Path,
+) -> None:
+    """Reject direct, group, wildcard sudo, or root-equivalent polkit grants."""
+    dangerous_polkit = (
+        "org.freedesktop.policykit.exec",
+        "org.freedesktop.systemd1.manage-unit",
+        "org.freedesktop.udisks2.modify-device",
+        "org.freedesktop.packagekit",
+    )
+    nsswitch = host_etc / "nsswitch.conf"
+    if nsswitch.exists():
+        try:
+            nss_snapshot = read_regular(nsswitch, 64 * 1024)
+            _audit_policy_metadata(nss_snapshot, root_uid, account_uid)
+            text = nss_snapshot.value.decode("utf-8")
+        except UnicodeDecodeError as exc:
+            raise HardeningError("nsswitch policy is not UTF-8") from exc
+        for line in text.splitlines():
+            name, separator, sources = line.partition(":")
+            if name.strip() in {"group", "initgroups", "sudoers"} and separator:
+                active_sources = [item for item in sources.split() if not item.startswith("[")]
+                if not active_sources or any(
+                    item not in {"files", "systemd"} for item in active_sources
+                ):
+                    raise HardeningError("external group/sudo authority source is unsafe")
+    total = 0
+    for path in _policy_files(host_etc, polkit_share, root_uid):
+        snapshot = read_regular(path, 256 * 1024)
+        _audit_policy_metadata(snapshot, root_uid, account_uid)
+        value = snapshot.value
+        total += len(value)
+        if total > 2 * 1024 * 1024:
+            raise HardeningError("sudo/polkit policy input exceeds safe limit")
+        try:
+            text = value.decode("utf-8")
+        except UnicodeDecodeError as exc:
+            raise HardeningError("sudo/polkit policy is not UTF-8") from exc
+        active = "\n".join(
+            line for line in text.splitlines() if not line.lstrip().startswith("#")
+        )
+        if account in active or f"%{account}" in active:
+            raise HardeningError("dedicated Hermes account has sudo/polkit authority")
+        if path.name == "sudoers" or "sudoers.d" in path.parts:
+            for line in active.splitlines():
+                fields = line.split()
+                if fields and fields[0] in {"ALL", "%ALL"} and "=" in line:
+                    raise HardeningError("broad sudo authority includes Hermes")
+        broad = "unix-user:*" in active or "Identity=unix-user:*" in active
+        grants = "polkit.Result.YES" in active or "ResultActive=yes" in active
+        if broad and grants and any(item in active for item in dangerous_polkit):
+            raise HardeningError("root-equivalent broad polkit authority includes Hermes")

services/hermes/scripts/node_account_hardening.py

@@ -7,16 +7,30 @@ import argparse
 import base64
 import errno
 import os
+import re
+import shlex
 import stat
 import struct
 from pathlib import Path
 
+from node_account_audit import audit_membership, audit_privilege_policies
+from node_account_io import (
+    FileSnapshot,
+    HardeningError,
+    account_lock,
+    assert_unchanged,
+    atomic_write,
+    backup_once,
+    read_regular,
+)
+
 HOST_ETC = Path("/host-etc")
 HOST_HOME = Path("/host-home")
 HOST_K3S = Path("/host-k3s")
 HOST_KUBELET = Path("/host-kubelet")
 HOST_RUN_K3S = Path("/host-run-k3s")
 HOST_RUN_CONTAINERD = Path("/host-run-containerd")
+HOST_POLKIT_SHARE = Path("/host-polkit-share")
 ACCOUNT = "hermes-agent"
 ACCOUNT_UID = 1200
 ACCOUNT_GID = 1200
@@ -40,23 +54,9 @@ ACL_ENTRY = struct.Struct("<HHI")
 ACL_XATTR = "system.posix_acl_access"
 
 
-class HardeningError(RuntimeError):
-    """Raised before an unsafe or ambiguous host-account change."""
-
-
 def _read_regular(path: Path, maximum: int = MAX_ACCOUNT_FILE) -> tuple[bytes, os.stat_result]:
-    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
-    descriptor = os.open(path, flags)
-    try:
-        metadata = os.fstat(descriptor)
-        if not stat.S_ISREG(metadata.st_mode) or metadata.st_size > maximum:
-            raise HardeningError(f"unsafe regular file: {path.name}")
-        value = os.read(descriptor, maximum + 1)
-    finally:
-        os.close(descriptor)
-    if len(value) != metadata.st_size:
-        raise HardeningError(f"short file read: {path.name}")
-    return value, metadata
+    snapshot = read_regular(path, maximum)
+    return snapshot.value, path.stat(follow_symlinks=False)
 
 
 def _records(value: bytes, fields: int, name: str) -> list[list[str]]:
@@ -111,49 +111,6 @@ def _reconcile_record(
     return [*records, expected]
 
 
-def _backup_once(path: Path, value: bytes, metadata: os.stat_result) -> Path:
-    backup = path.with_name(path.name + ".hermes-boundary-backup")
-    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0)
-    try:
-        descriptor = os.open(backup, flags, stat.S_IMODE(metadata.st_mode))
-    except FileExistsError:
-        _read_regular(backup)
-        return backup
-    try:
-        if os.write(descriptor, value) != len(value):
-            raise HardeningError(f"short backup write: {path.name}")
-        os.fchown(descriptor, metadata.st_uid, metadata.st_gid)
-        os.fsync(descriptor)
-    finally:
-        os.close(descriptor)
-    return backup
-
-
-def _atomic_write(
-    path: Path, value: bytes, *, mode: int, uid: int, gid: int
-) -> None:
-    temporary = path.with_name(f".{path.name}.hermes-{os.getpid()}")
-    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0)
-    descriptor = os.open(temporary, flags, mode)
-    try:
-        if os.write(descriptor, value) != len(value):
-            raise HardeningError(f"short atomic write: {path.name}")
-        os.fchmod(descriptor, mode)
-        os.fchown(descriptor, uid, gid)
-        os.fsync(descriptor)
-    except Exception:
-        temporary.unlink(missing_ok=True)
-        raise
-    finally:
-        os.close(descriptor)
-    os.replace(temporary, path)
-    directory = os.open(path.parent, os.O_RDONLY)
-    try:
-        os.fsync(directory)
-    finally:
-        os.close(directory)
-
-
 def _reconcile_databases() -> None:
     expected = _expected_records()
     definitions = (
@@ -162,18 +119,24 @@ def _reconcile_databases() -> None:
         ("shadow", 9, 0),
         ("gshadow", 4, 0),
     )
-    planned = []
+    planned: list[tuple[Path, FileSnapshot, bytes]] = []
     account_present = False
+    parsed: dict[str, list[list[str]]] = {}
     for name, fields, identity_index in definitions:
         path = HOST_ETC / name
-        value, metadata = _read_regular(path)
-        records = _records(value, fields, name)
+        snapshot = read_regular(path, MAX_ACCOUNT_FILE)
+        records = _records(snapshot.value, fields, name)
+        parsed[name] = records
         updated = _reconcile_record(
             records, expected[name], identity_index=identity_index
         )
         if name == "passwd":
             account_present = expected[name] in records
-        planned.append((path, value, metadata, _encode(updated)))
+        planned.append((path, snapshot, _encode(updated)))
+    audit_membership(ACCOUNT, parsed["group"], parsed["gshadow"])
+    audit_privilege_policies(
+        ACCOUNT, ACCOUNT_UID, HOST_ROOT_UID, HOST_ETC, HOST_POLKIT_SHARE
+    )
     try:
         (HOST_HOME / ACCOUNT).lstat()
     except FileNotFoundError:
@@ -181,52 +144,85 @@ def _reconcile_databases() -> None:
     else:
         if not account_present:
             raise HardeningError("dedicated Hermes account home already exists")
-    for path, value, metadata, _updated in planned:
-        _backup_once(path, value, metadata)
+    for path, snapshot, _updated in planned:
+        assert_unchanged(path, snapshot, MAX_ACCOUNT_FILE)
+    for path, snapshot, _updated in planned:
+        backup_once(path, snapshot, MAX_ACCOUNT_FILE)
+    for path, snapshot, _updated in planned:
+        assert_unchanged(path, snapshot, MAX_ACCOUNT_FILE)
+    written: list[tuple[Path, FileSnapshot, bytes]] = []
     try:
-        for path, _old, metadata, updated in planned:
-            _atomic_write(
-                path,
-                updated,
-                mode=stat.S_IMODE(metadata.st_mode),
-                uid=metadata.st_uid,
-                gid=metadata.st_gid,
-            )
-        for path, _old, _metadata, _updated in planned:
+        for path, snapshot, updated in planned:
+            assert_unchanged(path, snapshot, MAX_ACCOUNT_FILE)
+            atomic_write(path, updated, snapshot)
+            written.append((path, snapshot, updated))
+        for path, _snapshot, _updated in planned:
             name = path.name
             fields = next(item[1] for item in definitions if item[0] == name)
             records = _records(_read_regular(path)[0], fields, name)
             if expected[name] not in records:
                 raise HardeningError("dedicated Hermes account validation failed")
     except Exception:
-        for path, old, metadata, _updated in planned:
-            _atomic_write(
-                path,
-                old,
-                mode=stat.S_IMODE(metadata.st_mode),
-                uid=metadata.st_uid,
-                gid=metadata.st_gid,
-            )
+        for path, snapshot, updated in reversed(written):
+            current = read_regular(path, MAX_ACCOUNT_FILE)
+            if (
+                current.value != updated
+                or current.mode != snapshot.mode
+                or current.uid != snapshot.uid
+                or current.gid != snapshot.gid
+                or current.xattrs != snapshot.xattrs
+            ):
+                raise HardeningError(
+                    f"concurrent host account change prevents rollback: {path.name}"
+                )
+            atomic_write(path, snapshot.value, snapshot)
         raise
 
 
-def _validated_public_key(path: Path) -> bytes:
+SUPPORTED_KEY_TYPES = {"ssh-ed25519", "ecdsa-sha2-nistp256", "ssh-rsa"}
+KEY_TYPE_RE = re.compile(r"(?:sk-)?(?:ssh|ecdsa)-[A-Za-z0-9@._+-]+\Z")
+
+
+def _key_identity(line: bytes) -> tuple[str, bytes] | None:
+    """Return OpenSSH key type/blob, ignoring options and comments."""
+    stripped = line.strip(b"\r\n")
+    if not stripped or stripped.lstrip().startswith(b"#"):
+        return None
+    try:
+        fields = shlex.split(stripped.decode("ascii"), posix=True)
+    except (UnicodeDecodeError, ValueError) as exc:
+        raise HardeningError("authorized key line is malformed") from exc
+    indexes = [index for index, field in enumerate(fields) if KEY_TYPE_RE.fullmatch(field)]
+    if len(indexes) != 1 or indexes[0] + 1 >= len(fields):
+        raise HardeningError("authorized key line has ambiguous key material")
+    key_type = fields[indexes[0]]
+    try:
+        blob = base64.b64decode(fields[indexes[0] + 1], validate=True)
+    except ValueError as exc:
+        raise HardeningError("authorized key payload is invalid") from exc
+    encoded_type = key_type.encode("ascii")
+    if len(blob) < 4 or int.from_bytes(blob[:4], "big") != len(encoded_type):
+        raise HardeningError("authorized key blob is malformed")
+    if blob[4 : 4 + len(encoded_type)] != encoded_type:
+        raise HardeningError("authorized key type does not match its blob")
+    return key_type, blob
+
+
+def _validated_public_key(path: Path) -> tuple[bytes, tuple[str, bytes]]:
     value, _ = _read_regular(path, 16 * 1024)
     line = value.strip()
     if b"\n" in line or b"\r" in line:
         raise HardeningError("Hermes public key must contain one line")
+    identity = _key_identity(line)
     fields = line.split()
-    if len(fields) not in {2, 3} or fields[0] not in {
-        b"ssh-ed25519",
-        b"ecdsa-sha2-nistp256",
-        b"ssh-rsa",
-    }:
+    if (
+        identity is None
+        or identity[0] not in SUPPORTED_KEY_TYPES
+        or len(fields) not in {2, 3}
+        or fields[0].decode("ascii") != identity[0]
+    ):
         raise HardeningError("Hermes public key format is unsupported")
-    try:
-        base64.b64decode(fields[1], validate=True)
-    except ValueError as exc:
-        raise HardeningError("Hermes public key payload is invalid") from exc
-    return line
+    return line, identity
 
 
 def _directory(path: Path, *, mode: int, uid: int, gid: int) -> None:
@@ -243,43 +239,78 @@ def _directory(path: Path, *, mode: int, uid: int, gid: int) -> None:
     path.chmod(mode)
 
 
-def _without_key(value: bytes, key: bytes) -> bytes:
-    return b"".join(
-        line
-        for line in value.splitlines(keepends=True)
-        if line.strip(b"\r\n") != key
-    )
+def _without_key(value: bytes, identity: tuple[str, bytes]) -> bytes:
+    kept = []
+    for line in value.splitlines(keepends=True):
+        if _key_identity(line) != identity:
+            kept.append(line)
+    return b"".join(kept)
 
 
 def _move_key(public_key: Path) -> None:
-    key = _validated_public_key(public_key)
+    key, identity = _validated_public_key(public_key)
+    target = HOST_HOME / ACCOUNT / ".ssh" / "authorized_keys"
+    legacy_plan: list[tuple[Path, FileSnapshot, bytes]] = []
+    for legacy in LEGACY_ACCOUNTS:
+        authorized = HOST_HOME / legacy / ".ssh" / "authorized_keys"
+        try:
+            snapshot = read_regular(authorized, MAX_AUTHORIZED_KEYS)
+        except FileNotFoundError:
+            continue
+        updated = _without_key(snapshot.value, identity)
+        if updated != snapshot.value:
+            legacy_plan.append((authorized, snapshot, updated))
+    if legacy_plan and target.exists():
+        target_snapshot = read_regular(target, MAX_AUTHORIZED_KEYS)
+        target_without_key = _without_key(target_snapshot.value, identity)
+        if target_without_key != target_snapshot.value:
+            backup_once(target, target_snapshot, MAX_AUTHORIZED_KEYS)
+            assert_unchanged(target, target_snapshot, MAX_AUTHORIZED_KEYS)
+            atomic_write(target, target_without_key, target_snapshot)
+            verified_target = read_regular(target, MAX_AUTHORIZED_KEYS).value
+            if _without_key(verified_target, identity) != verified_target:
+                raise HardeningError("preexisting Hermes authorization removal failed")
+    for authorized, snapshot, _updated in legacy_plan:
+        assert_unchanged(authorized, snapshot, MAX_AUTHORIZED_KEYS)
+        backup_once(authorized, snapshot, MAX_AUTHORIZED_KEYS)
+    for authorized, snapshot, updated in legacy_plan:
+        assert_unchanged(authorized, snapshot, MAX_AUTHORIZED_KEYS)
+        atomic_write(authorized, updated, snapshot)
+    for legacy in LEGACY_ACCOUNTS:
+        authorized = HOST_HOME / legacy / ".ssh" / "authorized_keys"
+        try:
+            value = read_regular(authorized, MAX_AUTHORIZED_KEYS).value
+        except FileNotFoundError:
+            continue
+        if _without_key(value, identity) != value:
+            raise HardeningError("legacy Hermes authorization key removal failed")
+
+    # Installation happens only after every legacy authorization is absent. A
+    # crash can therefore remove access temporarily, but can never duplicate
+    # the machine credential across privileged and unprivileged accounts.
     home = HOST_HOME / ACCOUNT
     ssh = home / ".ssh"
     _directory(home, mode=0o700, uid=ACCOUNT_UID, gid=ACCOUNT_GID)
     _directory(ssh, mode=0o700, uid=ACCOUNT_UID, gid=ACCOUNT_GID)
     target = ssh / "authorized_keys"
     if target.exists():
-        current, metadata = _read_regular(target, MAX_AUTHORIZED_KEYS)
-        if current != key + b"\n":
-            _backup_once(target, current, metadata)
-    _atomic_write(target, key + b"\n", mode=0o600, uid=ACCOUNT_UID, gid=ACCOUNT_GID)
-    for legacy in LEGACY_ACCOUNTS:
-        authorized = HOST_HOME / legacy / ".ssh" / "authorized_keys"
-        try:
-            value, metadata = _read_regular(authorized, MAX_AUTHORIZED_KEYS)
-        except FileNotFoundError:
-            continue
-        updated = _without_key(value, key)
-        if updated == value:
-            continue
-        _backup_once(authorized, value, metadata)
-        _atomic_write(
-            authorized,
-            updated,
-            mode=stat.S_IMODE(metadata.st_mode),
-            uid=metadata.st_uid,
-            gid=metadata.st_gid,
+        current = read_regular(target, MAX_AUTHORIZED_KEYS)
+        if current.value != key + b"\n":
+            backup_once(target, current, MAX_AUTHORIZED_KEYS)
+    else:
+        current = FileSnapshot(
+            value=b"",
+            device=0,
+            inode=0,
+            mode=0o600,
+            uid=ACCOUNT_UID,
+            gid=ACCOUNT_GID,
+            size=0,
+            mtime_ns=0,
+            ctime_ns=0,
+            xattrs=(),
         )
+    atomic_write(target, key + b"\n", current)
     if _read_regular(target, MAX_AUTHORIZED_KEYS)[0] != key + b"\n":
         raise HardeningError("dedicated Hermes authorized key validation failed")
 
@@ -358,11 +389,11 @@ def _validate_acl_backup(value: bytes) -> None:
         raise HardeningError("sensitive directory ACL backup is malformed")
 
 
-def _acl_backup_once(path: Path, value: bytes) -> None:
+def _acl_backup_once(path: Path, value: bytes, backup_name: str | None = None) -> None:
     """Persist the original ACL as a private, durable host recovery file."""
     root = HOST_ETC / "hermes-node-boundary"
     _directory(root, mode=0o700, uid=HOST_ROOT_UID, gid=HOST_ROOT_GID)
-    backup = root / f"{path.name}.acl"
+    backup = root / f"{backup_name or path.name}.acl"
     encoded = b"A" + value if value else b"N"
     try:
         existing, _metadata = _read_regular(backup, 64 * 1024)
@@ -402,14 +433,14 @@ def _acl_backup_once(path: Path, value: bytes) -> None:
         os.close(directory)
 
 
-def _deny_sensitive_root(path: Path) -> None:
+def _deny_sensitive_root(path: Path, backup_name: str | None = None) -> None:
     metadata = path.lstat()
     if not stat.S_ISDIR(metadata.st_mode) or stat.S_ISLNK(metadata.st_mode):
         raise HardeningError(f"unsafe sensitive directory: {path.name}")
     if metadata.st_uid != HOST_ROOT_UID:
         raise HardeningError(f"sensitive directory is not root-owned: {path.name}")
     current = _read_acl(path)
-    _acl_backup_once(path, current)
+    _acl_backup_once(path, current, backup_name)
     updated = _acl_with_deny(current, stat.S_IMODE(metadata.st_mode))
     try:
         os.setxattr(path, ACL_XATTR, updated, follow_symlinks=False)
@@ -432,15 +463,21 @@ def _deny_sensitive_root(path: Path) -> None:
 
 
 def _deny_sensitive_roots() -> None:
-    for path in (HOST_K3S, HOST_KUBELET, HOST_RUN_K3S, HOST_RUN_CONTAINERD):
-        _deny_sensitive_root(path)
+    for path, backup_name in (
+        (HOST_K3S, "var-lib-rancher-k3s"),
+        (HOST_KUBELET, "var-lib-kubelet"),
+        (HOST_RUN_K3S, "run-k3s"),
+        (HOST_RUN_CONTAINERD, "run-containerd"),
+    ):
+        _deny_sensitive_root(path, backup_name)
 
 
 def reconcile(public_key: Path) -> None:
     """Create the locked account and move only the Hermes authorization key."""
-    _reconcile_databases()
-    _deny_sensitive_roots()
-    _move_key(public_key)
+    with account_lock(HOST_ETC, expected_uid=HOST_ROOT_UID):
+        _reconcile_databases()
+        _deny_sensitive_roots()
+        _move_key(public_key)
 
 
 def main() -> int:

services/hermes/scripts/node_account_io.py

@@ -0,0 +1,205 @@
+#!/usr/bin/env python3
+"""Crash-safe, metadata-preserving host account database I/O."""
+
+from __future__ import annotations
+
+import fcntl
+import os
+import stat
+import time
+from contextlib import contextmanager
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Iterator
+
+MAX_XATTRS = 64
+MAX_XATTR_NAME = 255
+MAX_XATTR_VALUE = 256 * 1024
+MAX_XATTR_TOTAL = 1024 * 1024
+LOCK_TIMEOUT_SECONDS = 15.0
+
+
+class HardeningError(RuntimeError):
+    """Raised before an unsafe or ambiguous host-account change."""
+
+
+@dataclass(frozen=True)
+class FileSnapshot:
+    """Bounded file contents and all security-relevant inode metadata."""
+
+    value: bytes
+    device: int
+    inode: int
+    mode: int
+    uid: int
+    gid: int
+    size: int
+    mtime_ns: int
+    ctime_ns: int
+    xattrs: tuple[tuple[str, bytes], ...]
+
+
+def _stat_identity(item: os.stat_result) -> tuple[int, ...]:
+    return (
+        item.st_dev,
+        item.st_ino,
+        item.st_mode,
+        item.st_uid,
+        item.st_gid,
+        item.st_size,
+        item.st_mtime_ns,
+        item.st_ctime_ns,
+    )
+
+
+def _xattrs(path_or_fd: Path | int) -> tuple[tuple[str, bytes], ...]:
+    try:
+        names = sorted(os.listxattr(path_or_fd, follow_symlinks=False))
+    except (TypeError, ValueError):
+        names = sorted(os.listxattr(path_or_fd))
+    if len(names) > MAX_XATTRS:
+        raise HardeningError("host account file has too many extended attributes")
+    values: list[tuple[str, bytes]] = []
+    total = 0
+    for name in names:
+        if len(name.encode("utf-8")) > MAX_XATTR_NAME:
+            raise HardeningError("host account file xattr name is too long")
+        try:
+            value = os.getxattr(path_or_fd, name, follow_symlinks=False)
+        except (TypeError, ValueError):
+            value = os.getxattr(path_or_fd, name)
+        total += len(value)
+        if len(value) > MAX_XATTR_VALUE or total > MAX_XATTR_TOTAL:
+            raise HardeningError("host account file xattrs exceed the safe limit")
+        values.append((name, value))
+    return tuple(values)
+
+
+def _restore_xattrs(descriptor: int, expected: tuple[tuple[str, bytes], ...]) -> None:
+    expected_names = {name for name, _value in expected}
+    for name, _value in _xattrs(descriptor):
+        if name not in expected_names:
+            os.removexattr(descriptor, name)
+    for name, value in expected:
+        os.setxattr(descriptor, name, value)
+
+
+def read_regular(path: Path, maximum: int) -> FileSnapshot:
+    """Read one bounded, non-symlink regular file and capture its metadata."""
+    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
+    descriptor = os.open(path, flags)
+    try:
+        metadata = os.fstat(descriptor)
+        if not stat.S_ISREG(metadata.st_mode) or metadata.st_size > maximum:
+            raise HardeningError(f"unsafe regular file: {path.name}")
+        value = os.read(descriptor, maximum + 1)
+        attrs = _xattrs(descriptor)
+        after = os.fstat(descriptor)
+    finally:
+        os.close(descriptor)
+    if len(value) != metadata.st_size or _stat_identity(metadata) != _stat_identity(after):
+        raise HardeningError(f"short file read: {path.name}")
+    return FileSnapshot(
+        value=value,
+        device=metadata.st_dev,
+        inode=metadata.st_ino,
+        mode=stat.S_IMODE(metadata.st_mode),
+        uid=metadata.st_uid,
+        gid=metadata.st_gid,
+        size=metadata.st_size,
+        mtime_ns=metadata.st_mtime_ns,
+        ctime_ns=metadata.st_ctime_ns,
+        xattrs=attrs,
+    )
+
+
+def assert_unchanged(path: Path, expected: FileSnapshot, maximum: int) -> None:
+    """Fail if an account file changed since planning began."""
+    current = read_regular(path, maximum)
+    if current != expected:
+        raise HardeningError(f"concurrent host account change detected: {path.name}")
+
+
+def atomic_write(path: Path, value: bytes, metadata: FileSnapshot) -> None:
+    """Atomically replace a file while retaining ACLs, labels, and xattrs."""
+    temporary = path.with_name(f".{path.name}.hermes-{os.getpid()}")
+    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0)
+    descriptor = os.open(temporary, flags, metadata.mode)
+    try:
+        if os.write(descriptor, value) != len(value):
+            raise HardeningError(f"short atomic write: {path.name}")
+        os.fchmod(descriptor, metadata.mode)
+        os.fchown(descriptor, metadata.uid, metadata.gid)
+        _restore_xattrs(descriptor, metadata.xattrs)
+        os.fsync(descriptor)
+    except Exception:
+        temporary.unlink(missing_ok=True)
+        raise
+    finally:
+        os.close(descriptor)
+    os.replace(temporary, path)
+    directory = os.open(path.parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
+    try:
+        os.fsync(directory)
+    finally:
+        os.close(directory)
+
+
+def backup_once(path: Path, snapshot: FileSnapshot, maximum: int) -> Path:
+    """Create the first durable recovery copy without replacing an older one."""
+    backup = path.with_name(path.name + ".hermes-boundary-backup")
+    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0)
+    try:
+        descriptor = os.open(backup, flags, snapshot.mode)
+    except FileExistsError:
+        read_regular(backup, maximum)
+        return backup
+    try:
+        if os.write(descriptor, snapshot.value) != len(snapshot.value):
+            raise HardeningError(f"short backup write: {path.name}")
+        os.fchmod(descriptor, snapshot.mode)
+        os.fchown(descriptor, snapshot.uid, snapshot.gid)
+        _restore_xattrs(descriptor, snapshot.xattrs)
+        os.fsync(descriptor)
+    except Exception:
+        backup.unlink(missing_ok=True)
+        raise
+    finally:
+        os.close(descriptor)
+    directory = os.open(path.parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
+    try:
+        os.fsync(directory)
+    finally:
+        os.close(directory)
+    return backup
+
+
+@contextmanager
+def account_lock(host_etc: Path, *, expected_uid: int = 0) -> Iterator[None]:
+    """Hold the standard shadow-utils account lock for the full transaction."""
+    path = host_etc / ".pwd.lock"
+    flags = os.O_WRONLY | os.O_CREAT | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
+    descriptor = os.open(path, flags, 0o600)
+    try:
+        metadata = os.fstat(descriptor)
+        if (
+            not stat.S_ISREG(metadata.st_mode)
+            or metadata.st_uid != expected_uid
+            or stat.S_IMODE(metadata.st_mode) != 0o600
+        ):
+            raise HardeningError("unsafe standard account lock")
+        deadline = time.monotonic() + LOCK_TIMEOUT_SECONDS
+        while True:
+            try:
+                fcntl.lockf(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB)
+                break
+            except BlockingIOError as exc:
+                if time.monotonic() >= deadline:
+                    raise HardeningError("standard account lock is busy") from exc
+                time.sleep(0.05)
+        yield
+    finally:
+        try:
+            fcntl.lockf(descriptor, fcntl.LOCK_UN)
+        finally:
+            os.close(descriptor)

testing/tests/test_hermes_node_account_hardening.py

@@ -13,6 +13,21 @@ import yaml
 
 ROOT = Path(__file__).parents[2]
 SCRIPT = ROOT / "services/hermes/scripts/node_account_hardening.py"
+sys.path.insert(0, str(SCRIPT.parent))
+
+
+def _key(material: bytes, comment: str = "") -> str:
+    key_type = b"ssh-ed25519"
+    blob = len(key_type).to_bytes(4, "big") + key_type + material
+    suffix = f" {comment}" if comment else ""
+    return "ssh-ed25519 " + base64.b64encode(blob).decode() + suffix
+
+
+def _typed_key(key_type: str, material: bytes, comment: str = "") -> str:
+    encoded_type = key_type.encode()
+    blob = len(encoded_type).to_bytes(4, "big") + encoded_type + material
+    suffix = f" {comment}" if comment else ""
+    return f"{key_type} {base64.b64encode(blob).decode()}{suffix}"
 
 
 def _load():
@@ -58,8 +73,8 @@ def _fixture(tmp_path: Path, monkeypatch):
     }
     for name, value in originals.items():
         (host_etc / name).write_text(value, encoding="utf-8")
-    key = "ssh-ed25519 " + base64.b64encode(b"synthetic-hermes-key").decode()
-    other = "ssh-ed25519 " + base64.b64encode(b"human-operator-key").decode()
+    key = _key(b"synthetic-hermes-key")
+    other = _key(b"human-operator-key")
     for user in ("atlas", "oceanus"):
         ssh = host_home / user / ".ssh"
         ssh.mkdir(parents=True)
@@ -72,6 +87,8 @@ def _fixture(tmp_path: Path, monkeypatch):
     monkeypatch.setattr(module, "HOST_HOME", host_home)
     monkeypatch.setattr(module, "ACCOUNT_UID", os.getuid())
     monkeypatch.setattr(module, "ACCOUNT_GID", os.getgid())
+    monkeypatch.setattr(module, "HOST_ROOT_UID", os.getuid())
+    monkeypatch.setattr(module, "HOST_POLKIT_SHARE", tmp_path / "missing-polkit")
     # ACL behavior has its own executable test below. These account-database
     # tests must not depend on the host running pytest with CAP_FOWNER.
     monkeypatch.setattr(module, "_deny_sensitive_roots", lambda: None)
@@ -206,7 +223,12 @@ def test_flux_daemonset_reconciles_every_node_without_mutating_human_groups():
     assert "ACCOUNT_UID = 1200" in source
     assert "ACCOUNT_GID = 1200" in source
     assert 'LEGACY_ACCOUNTS = ("atlas", "oceanus")' in source
-    assert "supplementary" not in source.lower()
+    assert "audit_membership" in source
+    image = spec["containers"][0]["image"]
+    assert image == (
+        "python@sha256:6d43704baacd1bfbe7c295d7f13079d5d8104ed33568873133f8fc69980419df"
+    )
+    assert mounts["host-polkit-share"]["hostPath"]["path"] == "/usr/share/polkit-1"
 
 
 def test_sensitive_roots_get_explicit_zero_permission_acl_for_hermes(
@@ -270,9 +292,13 @@ def test_acl_parser_rejects_malformed_or_incomplete_values():
 
 def test_acl_failure_stops_before_installing_the_ssh_key(tmp_path: Path, monkeypatch):
     module = _load()
+    host_etc = tmp_path / "etc"
+    host_etc.mkdir()
+    monkeypatch.setattr(module, "HOST_ETC", host_etc)
+    monkeypatch.setattr(module, "HOST_ROOT_UID", os.getuid())
     public_key = tmp_path / "public-key"
     public_key.write_text(
-        "ssh-ed25519 " + base64.b64encode(b"synthetic-key").decode() + "\n",
+        _key(b"synthetic-key") + "\n",
         encoding="utf-8",
     )
     moved = False
@@ -303,3 +329,267 @@ def test_runtime_ssh_config_forces_dedicated_account_for_every_titan():
     )
     assert "User atlas" not in yaml.safe_dump(agent)
     assert "User oceanus" not in yaml.safe_dump(agent)
+
+
+@pytest.mark.parametrize("database,record", [
+    ("group", "disk:x:6:atlas,hermes-agent\n"),
+    ("gshadow", "disk:!:hermes-agent:atlas\n"),
+    ("gshadow", "disk:!::atlas,hermes-agent\n"),
+])
+def test_unexpected_group_or_gshadow_membership_fails_before_writes(
+    tmp_path: Path, monkeypatch, database: str, record: str
+):
+    module, originals, _key_value, _other, public_key = _fixture(tmp_path, monkeypatch)
+    path = module.HOST_ETC / database
+    path.write_text(
+        "\n".join(line for line in originals[database].splitlines() if not line.startswith("disk:"))
+        + "\n"
+        + record,
+        encoding="utf-8",
+    )
+    before = {name: (module.HOST_ETC / name).read_bytes() for name in originals}
+
+    with pytest.raises(module.HardeningError, match="group access"):
+        module.reconcile(public_key)
+
+    assert {name: (module.HOST_ETC / name).read_bytes() for name in originals} == before
+
+
+@pytest.mark.parametrize(
+    "relative,value",
+    [
+        ("sudoers.d/hermes", "hermes-agent ALL=(ALL:ALL) NOPASSWD: ALL\n"),
+        ("sudoers", "ALL ALL=(ALL:ALL) ALL\n"),
+        (
+            "polkit-1/rules.d/90-root.rules",
+            'if (subject.user == "hermes-agent") return polkit.Result.YES;\n',
+        ),
+        (
+            "polkit-1/rules.d/90-root.rules",
+            'unix-user:* org.freedesktop.policykit.exec polkit.Result.YES\n',
+        ),
+    ],
+)
+def test_sudo_and_root_equivalent_polkit_authority_fail_closed(
+    tmp_path: Path, monkeypatch, relative: str, value: str
+):
+    module, originals, _key_value, _other, public_key = _fixture(tmp_path, monkeypatch)
+    policy = module.HOST_ETC / relative
+    policy.parent.mkdir(parents=True, exist_ok=True)
+    policy.write_text(value, encoding="utf-8")
+
+    with pytest.raises(module.HardeningError, match="sudo|polkit"):
+        module.reconcile(public_key)
+
+    assert all((module.HOST_ETC / name).read_text() == original for name, original in originals.items())
+
+
+def test_external_group_authority_source_is_rejected(tmp_path: Path, monkeypatch):
+    module, originals, _key_value, _other, public_key = _fixture(tmp_path, monkeypatch)
+    (module.HOST_ETC / "nsswitch.conf").write_text("passwd: files\ngroup: files ldap\n")
+    with pytest.raises(module.HardeningError, match="external group"):
+        module.reconcile(public_key)
+    assert all((module.HOST_ETC / name).read_text() == original for name, original in originals.items())
+
+
+def test_writable_privilege_policy_is_rejected(tmp_path: Path, monkeypatch):
+    module, originals, _key_value, _other, public_key = _fixture(tmp_path, monkeypatch)
+    sudoers = module.HOST_ETC / "sudoers"
+    sudoers.write_text("root ALL=(ALL:ALL) ALL\n")
+    sudoers.chmod(0o666)
+    with pytest.raises(module.HardeningError, match="unsafe mutation"):
+        module.reconcile(public_key)
+    assert all((module.HOST_ETC / name).read_text() == original for name, original in originals.items())
+
+
+def test_key_identity_ignores_options_and_comments_and_removes_legacy_first(
+    tmp_path: Path, monkeypatch
+):
+    module, _originals, key, other, public_key = _fixture(tmp_path, monkeypatch)
+    key_fields = key.split()
+    decorated = f'restrict,command="echo denied" {key_fields[0]} {key_fields[1]} old-comment'
+    for user in module.LEGACY_ACCOUNTS:
+        path = module.HOST_HOME / user / ".ssh/authorized_keys"
+        path.write_text(f"{other} {user}\n{decorated}\n", encoding="utf-8")
+    writes: list[Path] = []
+    real_write = module.atomic_write
+
+    def recording_write(path, value, metadata):
+        writes.append(path)
+        real_write(path, value, metadata)
+
+    monkeypatch.setattr(module, "atomic_write", recording_write)
+    module.reconcile(public_key)
+
+    key_writes = [path for path in writes if path.name == "authorized_keys"]
+    assert key_writes[-1].parts[-3:] == (module.ACCOUNT, ".ssh", "authorized_keys")
+    assert all(module.ACCOUNT not in path.parts for path in key_writes[:-1])
+
+
+def test_unrelated_modern_human_key_type_is_preserved(tmp_path: Path, monkeypatch):
+    module, _originals, key, _other, public_key = _fixture(tmp_path, monkeypatch)
+    modern = _typed_key("sk-ssh-ed25519@openssh.com", b"human-security-key", "human")
+    for user in module.LEGACY_ACCOUNTS:
+        path = module.HOST_HOME / user / ".ssh/authorized_keys"
+        path.write_text(f"{modern}\n{key}\n", encoding="utf-8")
+
+    module.reconcile(public_key)
+
+    for user in module.LEGACY_ACCOUNTS:
+        path = module.HOST_HOME / user / ".ssh/authorized_keys"
+        assert path.read_text() == modern + "\n"
+
+
+def test_crash_during_legacy_key_removal_never_installs_duplicate_key(
+    tmp_path: Path, monkeypatch
+):
+    module, _originals, _key_value, _other, public_key = _fixture(tmp_path, monkeypatch)
+    real_write = module.atomic_write
+    key_writes = 0
+
+    def crash_on_second_key(path, value, metadata):
+        nonlocal key_writes
+        if path.name == "authorized_keys":
+            key_writes += 1
+            if key_writes == 2:
+                raise module.HardeningError("synthetic crash")
+        real_write(path, value, metadata)
+
+    monkeypatch.setattr(module, "atomic_write", crash_on_second_key)
+    with pytest.raises(module.HardeningError, match="synthetic crash"):
+        module.reconcile(public_key)
+    assert not (module.HOST_HOME / module.ACCOUNT / ".ssh/authorized_keys").exists()
+
+
+def test_preexisting_target_key_is_removed_before_legacy_cleanup_and_reinstalled(
+    tmp_path: Path, monkeypatch
+):
+    module, _originals, key, _other, public_key = _fixture(tmp_path, monkeypatch)
+    target = module.HOST_HOME / module.ACCOUNT / ".ssh/authorized_keys"
+    target.parent.mkdir(parents=True)
+    target.write_text(key + " existing\n")
+    writes: list[Path] = []
+    real_write = module.atomic_write
+
+    def recording_write(path, value, metadata):
+        if path.name == "authorized_keys":
+            writes.append(path)
+        real_write(path, value, metadata)
+
+    monkeypatch.setattr(module, "atomic_write", recording_write)
+    module._move_key(public_key)
+
+    assert writes[0] == target
+    assert writes[-1] == target
+    assert all(path != target for path in writes[1:-1])
+    assert target.read_text() == key + "\n"
+
+
+def test_account_database_write_preserves_mode_owner_and_xattrs(
+    tmp_path: Path, monkeypatch
+):
+    module, originals, _key_value, _other, public_key = _fixture(tmp_path, monkeypatch)
+    passwd = module.HOST_ETC / "passwd"
+    passwd.chmod(0o640)
+    os.setxattr(passwd, "user.hermes-test", b"preserve")
+    before = passwd.stat()
+
+    module.reconcile(public_key)
+
+    after = passwd.stat()
+    assert after.st_mode & 0o777 == 0o640
+    assert (after.st_uid, after.st_gid) == (before.st_uid, before.st_gid)
+    assert os.getxattr(passwd, "user.hermes-test") == b"preserve"
+    backup = module.HOST_ETC / "passwd.hermes-boundary-backup"
+    assert os.getxattr(backup, "user.hermes-test") == b"preserve"
+    assert backup.read_text() == originals["passwd"]
+
+
+def test_atomic_write_restores_acl_and_security_label_xattrs(tmp_path: Path, monkeypatch):
+    module = _load()
+    io_module = sys.modules[module.atomic_write.__module__]
+    target = tmp_path / "passwd"
+    target.write_bytes(b"old\n")
+    metadata = io_module.FileSnapshot(
+        value=b"old\n",
+        device=target.stat().st_dev,
+        inode=target.stat().st_ino,
+        mode=0o600,
+        uid=os.getuid(),
+        gid=os.getgid(),
+        size=4,
+        mtime_ns=target.stat().st_mtime_ns,
+        ctime_ns=target.stat().st_ctime_ns,
+        xattrs=(
+            ("system.posix_acl_access", b"synthetic-acl"),
+            ("security.selinux", b"synthetic-label"),
+            ("user.audit", b"synthetic-xattr"),
+        ),
+    )
+    restored: list[tuple[str, bytes]] = []
+    monkeypatch.setattr(
+        io_module.os,
+        "setxattr",
+        lambda _fd, name, value: restored.append((name, value)),
+    )
+
+    io_module.atomic_write(target, b"new\n", metadata)
+
+    assert target.read_bytes() == b"new\n"
+    assert restored == list(metadata.xattrs)
+
+
+def test_concurrent_database_change_is_detected_before_any_account_write(
+    tmp_path: Path, monkeypatch
+):
+    module, originals, _key_value, _other, public_key = _fixture(tmp_path, monkeypatch)
+    real_backup = module.backup_once
+    backup_count = 0
+    writes: list[Path] = []
+
+    def racing_backup(path, snapshot, maximum):
+        nonlocal backup_count
+        result = real_backup(path, snapshot, maximum)
+        backup_count += 1
+        if backup_count == 4:
+            group = module.HOST_ETC / "group"
+            group.write_text(originals["group"] + "race:x:4000:\n")
+        return result
+
+    monkeypatch.setattr(module, "backup_once", racing_backup)
+    monkeypatch.setattr(module, "atomic_write", lambda path, *_args: writes.append(path))
+    with pytest.raises(module.HardeningError, match="concurrent host account change"):
+        module.reconcile(public_key)
+    assert writes == []
+
+
+def test_flux_orders_observer_rbac_before_hermes_prunes_old_authority():
+    observer = yaml.safe_load(
+        (ROOT / "clusters/atlas/flux-system/applications/hermes-observer-rbac/kustomization.yaml").read_text()
+    )
+    hermes = yaml.safe_load(
+        (ROOT / "clusters/atlas/flux-system/applications/hermes/kustomization.yaml").read_text()
+    )
+    assert "dependsOn" not in observer["spec"]
+    assert {item["name"] for item in hermes["spec"]["dependsOn"]} >= {
+        "hermes-observer-rbac",
+        "hermes-scm-broker",
+    }
+
+
+def test_sensitive_root_acl_backups_have_unique_host_path_names(monkeypatch):
+    module = _load()
+    seen = []
+    monkeypatch.setattr(
+        module,
+        "_deny_sensitive_root",
+        lambda path, backup_name=None: seen.append((path, backup_name)),
+    )
+    module._deny_sensitive_roots()
+    assert [name for _path, name in seen] == [
+        "var-lib-rancher-k3s",
+        "var-lib-kubelet",
+        "run-k3s",
+        "run-containerd",
+    ]
+    assert len({name for _path, name in seen}) == 4

testing/tests/test_hermes_scm_broker.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import importlib.util
+import io
 import json
 import sys
 from email.message import Message
@@ -26,9 +27,19 @@ def _load(name: str):
     return module
 
 
+def _load_path(name: str, path: Path):
+    spec = importlib.util.spec_from_file_location(name, path)
+    assert spec and spec.loader
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
 class Response:
     def __init__(self, body: bytes, *, status: int = 200, content_type: str):
         self.body = body
+        self.stream = io.BytesIO(body)
         self.status = status
         self.headers = Message()
         self.headers["Content-Type"] = content_type
@@ -40,7 +51,7 @@ class Response:
         return False
 
     def read(self, limit=-1):
-        return self.body if limit < 0 else self.body[:limit]
+        return self.stream.read(limit)
 
 
 def _receive_command(old: bytes, new: bytes, ref: bytes) -> bytes:
@@ -216,6 +227,163 @@ def test_broker_content_length_is_canonical_and_bounded(raw: str):
         broker._content_length(headers, broker.MAX_GIT_REQUEST)
 
 
+def test_large_git_exchange_rolls_to_disk_and_scans_chunk_boundaries():
+    broker = _load("scm_broker")
+    token = "runtime-sentinel"
+    safe = b"x" * (broker.SPOOL_MEMORY_LIMIT + 1)
+    inbound_timeouts = []
+    spool, length = broker._spool_bounded(
+        io.BytesIO(safe),
+        broker.MAX_GIT_REQUEST,
+        len(safe),
+        token=token,
+        context="Git request",
+        deadline=broker.time.monotonic() + 30,
+        set_timeout=inbound_timeouts.append,
+    )
+    try:
+        assert length == len(safe)
+        assert spool._rolled is True
+        assert spool.read(4) == b"xxxx"
+        assert inbound_timeouts
+        assert all(0 < value <= 30 for value in inbound_timeouts)
+    finally:
+        spool.close()
+
+    crossing = b"x" * (broker.STREAM_CHUNK - 5) + token.encode() + b"tail"
+    with pytest.raises(broker.PolicyError, match="credential material"):
+        broker._spool_bounded(
+            io.BytesIO(crossing),
+            broker.MAX_GIT_REQUEST,
+            len(crossing),
+            token=token,
+            context="Git request",
+            deadline=broker.time.monotonic() + 30,
+        )
+
+
+def test_large_upstream_response_is_streamed_through_bounded_spool():
+    broker = _load("scm_broker")
+    body = b"z" * (broker.SPOOL_MEMORY_LIMIT + 1)
+    seen = []
+
+    def opener(request, **_kwargs):
+        seen.append(request)
+        assert request.get_header("Content-length") == "7"
+        assert request.data.read() == b"request"
+        return Response(body, content_type="application/x-git-upload-pack-result")
+
+    streamed = broker._upstream_git_request(
+        "/atlas/cassandra.git/git-upload-pack",
+        method="POST",
+        body=io.BytesIO(b"request"),
+        body_length=7,
+        content_type="application/x-git-upload-pack-request",
+        expected_type="application/x-git-upload-pack-result",
+        token="runtime-sentinel",
+        stream_result=True,
+        opener=opener,
+    )
+    spool, length = streamed
+    try:
+        assert length == len(body)
+        assert spool._rolled is True
+        assert spool.read(3) == b"zzz"
+        assert len(seen) == 1
+    finally:
+        spool.close()
+
+
+def test_body_deadline_fails_closed_before_reading():
+    broker = _load("scm_broker")
+    with pytest.raises(broker.PolicyError, match="deadline"):
+        broker._spool_bounded(
+            io.BytesIO(b"body"),
+            broker.MAX_GIT_REQUEST,
+            4,
+            token="runtime-sentinel",
+            context="Git request",
+            deadline=broker.time.monotonic() - 1,
+        )
+
+
+def test_upstream_stream_applies_absolute_socket_deadline():
+    broker = _load("scm_broker")
+
+    class DeadlineSocket:
+        def __init__(self):
+            self.values = []
+
+        def settimeout(self, value):
+            self.values.append(value)
+
+    response = Response(b"safe", content_type="application/x-git-upload-pack-result")
+    sock = DeadlineSocket()
+    response.fp = type("FP", (), {"raw": type("Raw", (), {"_sock": sock})()})()
+    spool, length = broker._spool_response(response, broker.MAX_GIT_RESPONSE, "sentinel")
+    try:
+        assert length == 4
+        assert sock.values
+        assert all(0 < value <= broker.INBOUND_BODY_TIMEOUT for value in sock.values)
+    finally:
+        spool.close()
+
+
+def test_broker_rejects_work_when_concurrency_slots_are_exhausted():
+    broker = _load("scm_broker")
+
+    class FakeSocket:
+        def __init__(self):
+            self.value = b""
+
+        def sendall(self, value):
+            self.value += value
+
+        def shutdown(self, _how):
+            return None
+
+        def close(self):
+            return None
+
+    server = broker.BoundedThreadingHTTPServer(("127.0.0.1", 0), broker.BrokerHandler)
+    try:
+        assert server._slots.acquire(blocking=False)
+        assert server._slots.acquire(blocking=False)
+        request = FakeSocket()
+        server.process_request(request, ("127.0.0.1", 1))
+        assert b"503 Service Unavailable" in request.value
+    finally:
+        server._slots.release()
+        server._slots.release()
+        server.server_close()
+
+
+def test_broker_bounds_header_count_size_and_duplicate_lengths():
+    broker = _load("scm_broker")
+
+    def validate(headers):
+        handler = object.__new__(broker.BrokerHandler)
+        handler.headers = headers
+        handler._validate_headers()
+
+    too_many = Message()
+    for index in range(broker.MAX_HEADERS + 1):
+        too_many[f"X-Test-{index}"] = "safe"
+    with pytest.raises(broker.PolicyError, match="too many headers"):
+        validate(too_many)
+
+    too_large = Message()
+    too_large["X-Test"] = "x" * broker.MAX_HEADER_BYTES
+    with pytest.raises(broker.PolicyError, match="safe limit"):
+        validate(too_large)
+
+    duplicate = Message()
+    duplicate["Content-Length"] = "1"
+    duplicate["Content-Length"] = "1"
+    with pytest.raises(broker.PolicyError, match="duplicate Content-Length"):
+        validate(duplicate)
+
+
 def _can_i(role: dict, api_group: str, resource: str, verb: str) -> bool:
     return any(
         (api_group in rule["apiGroups"] or "*" in rule["apiGroups"])
@@ -291,6 +459,13 @@ def test_flux_boundary_keeps_broker_secret_and_network_separate_from_agent():
         "hermes-scm-broker"
     )
     assert "gitea-token" in json.dumps(broker)
+    container = broker["spec"]["template"]["spec"]["containers"][0]
+    assert container["resources"]["limits"]["memory"] == "768Mi"
+    assert container["livenessProbe"]["failureThreshold"] == 10
+    tmp = next(
+        item for item in broker["spec"]["template"]["spec"]["volumes"] if item["name"] == "tmp"
+    )
+    assert tmp["emptyDir"]["sizeLimit"] == "1Gi"
 
     policy = yaml.safe_load(
         (ROOT / "services/hermes-scm-broker/networkpolicy.yaml").read_text(
@@ -350,3 +525,67 @@ def test_gitea_bootstrap_enforces_human_review_without_overwriting_drift():
     assert 'merge_whitelist_usernames\\\":[\\\"${protected_reviewer}\\\"]' in script
     assert "protection differs from the human-review policy" in script
     assert "api_request PATCH" not in script
+    job = yaml.safe_load(
+        (ROOT / "services/gitea/atlas-identity-bootstrap-job.yaml").read_text()
+    )
+    init = job["spec"]["template"]["spec"]["initContainers"][0]
+    assert init["image"] == (
+        "python@sha256:6d43704baacd1bfbe7c295d7f13079d5d8104ed33568873133f8fc69980419df"
+    )
+    assert init["securityContext"]["runAsNonRoot"] is True
+    assert init["securityContext"]["capabilities"]["drop"] == ["ALL"]
+    assert "/opt/python/bin/python3" in script
+
+
+def _protection(rule_name: str, priority: int, helper, **overrides):
+    value = {"rule_name": rule_name, "priority": priority, **helper._required("bstein")}
+    value.update(overrides)
+    return value
+
+
+def test_branch_protection_uses_first_effective_glob_by_priority():
+    helper = _load_path(
+        "branch_protection_test",
+        ROOT / "services/gitea/scripts/gitea_branch_protection_check.py",
+    )
+    earlier_drift = _protection("m*", 1, helper, required_approvals=0)
+    later_exact = _protection("main", 2, helper)
+
+    with pytest.raises(helper.PolicyError, match="effective main protection differs"):
+        helper.evaluate(json.dumps([later_exact, earlier_drift]).encode(), "main", "bstein")
+
+    earlier_drift["required_approvals"] = 1
+    assert (
+        helper.evaluate(json.dumps([later_exact, earlier_drift]).encode(), "main", "bstein")
+        == "PRESENT"
+    )
+    brace_drift = _protection("m{ain,aster}", 1, helper, required_approvals=0)
+    with pytest.raises(helper.PolicyError, match="effective master protection differs"):
+        helper.evaluate(json.dumps([brace_drift]).encode(), "master", "bstein")
+
+
+@pytest.mark.parametrize(
+    "rules",
+    [
+        [{"rule_name": "main", "priority": 1}, {"rule_name": "m*", "priority": 1}],
+        [{"rule_name": "[main", "priority": 1}],
+        [{"rule_name": "main", "priority": 0}],
+        [{"rule_name": "main"}],
+    ],
+)
+def test_branch_protection_rejects_ambiguous_rules_before_creation(rules):
+    helper = _load_path(
+        "branch_protection_adversarial_test",
+        ROOT / "services/gitea/scripts/gitea_branch_protection_check.py",
+    )
+    with pytest.raises(helper.PolicyError):
+        helper.evaluate(json.dumps(rules).encode(), "main", "bstein")
+
+
+def test_branch_protection_reports_absent_only_when_no_rule_matches():
+    helper = _load_path(
+        "branch_protection_absent_test",
+        ROOT / "services/gitea/scripts/gitea_branch_protection_check.py",
+    )
+    rules = [_protection("release/*", 1, helper)]
+    assert helper.evaluate(json.dumps(rules).encode(), "master", "bstein") == "ABSENT"