hermes: isolate Atlas SCM write authority

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

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

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

@@ -61,4 +61,3 @@ 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-5
+  name: gitea-atlas-identity-bootstrap-4
   namespace: gitea
   labels:
     app.kubernetes.io/name: gitea
@@ -34,20 +34,6 @@ 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
@@ -94,7 +80,6 @@ spec:
             - name: bootstrap
               mountPath: /opt/bootstrap
               readOnly: true
-            - {name: python-runtime, mountPath: /opt/python, readOnly: true}
       volumes:
         - name: data
           persistentVolumeClaim:
@@ -103,6 +88,3 @@ spec:
           configMap:
             name: gitea-atlas-identity-bootstrap
             defaultMode: 0555
-        - name: python-runtime
-          emptyDir:
-            sizeLimit: 128Mi

services/gitea/kustomization.yaml

@@ -15,7 +15,6 @@ 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,6 +277,7 @@ 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}")
@@ -289,28 +290,44 @@ ensure_default_branch_protection() {
 
   status=$(api_request GET "/repos/${organization}/${repository}/branch_protections" '' "${protections_file}")
   expect_status "${status}" 200 "list ${repository} branch protections"
-  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
+  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}" = ABSENT ]; then
+  if [ "${matching_rules}" -eq 0 ]; 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"
-    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"
+    tr -d '[:space:]' <"${protections_file}" | sed 's/},{/}\
+{/g' | grep -F "\"rule_name\":\"${default_branch}\"" >"${protection_file}" || true
   fi
-  [ "${protection_state}" = PRESENT ] || \
-    die "${repository} ${default_branch} protection did not become effective"
 
-  rm -f "${repo_file}" "${protections_file}" "${response_file}"
+  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}"
 }
 
 wait_for_gitea

services/gitea/scripts/gitea_branch_protection_check.py

@@ -1,138 +0,0 @@
-#!/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,7 +73,6 @@ spec:
             httpGet: {path: /healthz, port: http}
             initialDelaySeconds: 15
             periodSeconds: 30
-            failureThreshold: 10
           securityContext:
             allowPrivilegeEscalation: false
             capabilities:

services/hermes/kustomization.yaml

@@ -106,8 +106,6 @@ 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,14 +45,13 @@ spec:
         - operator: Exists
       containers:
         - name: key-reconciler
-          # The pinned index supports the cluster's amd64 and arm64 nodes.
-          image: python@sha256:6d43704baacd1bfbe7c295d7f13079d5d8104ed33568873133f8fc69980419df
+          image: registry.bstein.dev/bstein/hermes-agent
           imagePullPolicy: IfNotPresent
           command: [/bin/sh, -ec]
           args:
             - |
               while true; do
-                /usr/local/bin/python \
+                /opt/hermes/.venv/bin/python \
                   /opt/node-hardener/node_account_hardening.py \
                   --public-key-file /vault/secrets/node-ssh-public-key
                 sleep 300
@@ -69,9 +68,6 @@ 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
@@ -104,10 +100,6 @@ 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,8 +7,6 @@ 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,14 +7,10 @@ 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
-from typing import BinaryIO
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 
 from gitea_api import (
     CANONICAL_BASE_URL,
@@ -24,8 +20,6 @@ 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"
@@ -33,12 +27,6 @@ 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/"
@@ -51,6 +39,7 @@ 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")
 
@@ -65,105 +54,22 @@ def _status(response: object) -> int | None:
     return value
 
 
-def _read_bounded(
-    stream,
-    maximum: int,
-    length: int | None = None,
-    *,
-    deadline: float | None = None,
-    set_timeout=None,
-) -> bytes:
+def _read_bounded(stream, maximum: int, length: int | None = None) -> bytes:
     if length is not None and not 0 <= length <= maximum:
         raise PolicyError("SCM request exceeds the safe size limit")
-    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)
+    value = stream.read(maximum + 1 if length is None else length)
     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)
-    deadline = time.monotonic() + INBOUND_BODY_TIMEOUT
-    value = json.loads(
-        _read_bounded(
-            handler.rfile,
-            MAX_CONTROL_BODY,
-            length,
-            deadline=deadline,
-            set_timeout=handler.connection.settimeout,
-        )
-    )
+    value = json.loads(_read_bounded(handler.rfile, MAX_CONTROL_BODY, length))
     if not isinstance(value, dict):
         raise PolicyError("broker control request must be an object")
     return value
@@ -227,20 +133,8 @@ def _reject_credential_bytes(value: bytes, token: str, context: str) -> None:
         raise PolicyError(f"{context} contains runtime credential material")
 
 
-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:
+def _validate_receive_pack(body: bytes, 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
@@ -281,24 +175,16 @@ def _upstream_git_request(
     target: str,
     *,
     method: str,
-    body: bytes | BinaryIO | None,
+    body: bytes | None,
     content_type: str | None,
     expected_type: str,
     token: str,
-    body_length: int | None = None,
-    stream_result: bool = False,
     opener=UPSTREAM_OPENER.open,
-) -> bytes | tuple[BinaryIO, int]:
+) -> bytes:
     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,
@@ -310,13 +196,9 @@ 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, result_length = _spool_response(response, MAX_GIT_RESPONSE, token)
-    if stream_result:
-        return result, result_length
-    try:
-        return result.read()
-    finally:
-        result.close()
+        result = _read_bounded(response, MAX_GIT_RESPONSE)
+    _reject_credential_bytes(result, token, "Git upstream response")
+    return result
 
 
 class BrokerHandler(BaseHTTPRequestHandler):
@@ -325,42 +207,6 @@ 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
 
@@ -377,7 +223,6 @@ class BrokerHandler(BaseHTTPRequestHandler):
 
     def do_GET(self) -> None:
         try:
-            self._validate_headers()
             if self.path == "/healthz":
                 self._json(200, b'{"status":"ok"}')
                 return
@@ -387,32 +232,29 @@ class BrokerHandler(BaseHTTPRequestHandler):
             token = read_token()
             _reject_forbidden(repo, "repository", (token,))
             expected = f"application/x-{service}-advertisement"
-            streamed = _upstream_git_request(
+            body = _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,
             )
-            body, length = streamed
-            try:
-                self._stream(200, expected, body, length)
-            finally:
-                body.close()
-        except (OSError, PolicyError, socket.timeout, urllib.error.URLError, ValueError):
+            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):
             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, socket.timeout, urllib.error.URLError, ValueError, json.JSONDecodeError):
+        except (OSError, PolicyError, urllib.error.URLError, ValueError, json.JSONDecodeError):
             self._reject()
 
     def _control(self) -> None:
@@ -439,38 +281,27 @@ 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,))
-        body, body_length = _spool_bounded(
-            self.rfile,
-            MAX_GIT_REQUEST,
-            length,
+        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,
             token=token,
-            context="Git request",
-            deadline=time.monotonic() + INBOUND_BODY_TIMEOUT,
-            set_timeout=self.connection.settimeout,
         )
-        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()
+        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)
 
 
 def main() -> int:
@@ -478,7 +309,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()
-    BoundedThreadingHTTPServer((args.listen, args.port), BrokerHandler).serve_forever()
+    ThreadingHTTPServer((args.listen, args.port), BrokerHandler).serve_forever()
     return 0
 
 

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

@@ -1,54 +0,0 @@
-#!/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

@@ -1,44 +0,0 @@
-#!/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

@@ -1,138 +0,0 @@
-#!/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,30 +7,16 @@ 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
@@ -54,9 +40,23 @@ 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]:
-    snapshot = read_regular(path, maximum)
-    return snapshot.value, path.stat(follow_symlinks=False)
+    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
 
 
 def _records(value: bytes, fields: int, name: str) -> list[list[str]]:
@@ -111,6 +111,49 @@ 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 = (
@@ -119,24 +162,18 @@ def _reconcile_databases() -> None:
         ("shadow", 9, 0),
         ("gshadow", 4, 0),
     )
-    planned: list[tuple[Path, FileSnapshot, bytes]] = []
+    planned = []
     account_present = False
-    parsed: dict[str, list[list[str]]] = {}
     for name, fields, identity_index in definitions:
         path = HOST_ETC / name
-        snapshot = read_regular(path, MAX_ACCOUNT_FILE)
-        records = _records(snapshot.value, fields, name)
-        parsed[name] = records
+        value, metadata = _read_regular(path)
+        records = _records(value, fields, name)
         updated = _reconcile_record(
             records, expected[name], identity_index=identity_index
         )
         if name == "passwd":
             account_present = expected[name] in records
-        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
-    )
+        planned.append((path, value, metadata, _encode(updated)))
     try:
         (HOST_HOME / ACCOUNT).lstat()
     except FileNotFoundError:
@@ -144,85 +181,52 @@ def _reconcile_databases() -> None:
     else:
         if not account_present:
             raise HardeningError("dedicated Hermes account home already exists")
-    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]] = []
+    for path, value, metadata, _updated in planned:
+        _backup_once(path, value, metadata)
     try:
-        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:
+        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:
             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, 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)
+        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,
+            )
         raise
 
 
-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]]:
+def _validated_public_key(path: Path) -> 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 (
-        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]
-    ):
+    if len(fields) not in {2, 3} or fields[0] not in {
+        b"ssh-ed25519",
+        b"ecdsa-sha2-nistp256",
+        b"ssh-rsa",
+    }:
         raise HardeningError("Hermes public key format is unsupported")
-    return line, identity
+    try:
+        base64.b64decode(fields[1], validate=True)
+    except ValueError as exc:
+        raise HardeningError("Hermes public key payload is invalid") from exc
+    return line
 
 
 def _directory(path: Path, *, mode: int, uid: int, gid: int) -> None:
@@ -239,78 +243,43 @@ def _directory(path: Path, *, mode: int, uid: int, gid: int) -> None:
     path.chmod(mode)
 
 
-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 _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 _move_key(public_key: Path) -> None:
-    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.
+    key = _validated_public_key(public_key)
     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 = 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=(),
+        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,
         )
-    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")
 
@@ -389,11 +358,11 @@ def _validate_acl_backup(value: bytes) -> None:
         raise HardeningError("sensitive directory ACL backup is malformed")
 
 
-def _acl_backup_once(path: Path, value: bytes, backup_name: str | None = None) -> None:
+def _acl_backup_once(path: Path, value: bytes) -> 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"{backup_name or path.name}.acl"
+    backup = root / f"{path.name}.acl"
     encoded = b"A" + value if value else b"N"
     try:
         existing, _metadata = _read_regular(backup, 64 * 1024)
@@ -433,14 +402,14 @@ def _acl_backup_once(path: Path, value: bytes, backup_name: str | None = None) -
         os.close(directory)
 
 
-def _deny_sensitive_root(path: Path, backup_name: str | None = None) -> None:
+def _deny_sensitive_root(path: Path) -> 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, backup_name)
+    _acl_backup_once(path, current)
     updated = _acl_with_deny(current, stat.S_IMODE(metadata.st_mode))
     try:
         os.setxattr(path, ACL_XATTR, updated, follow_symlinks=False)
@@ -463,21 +432,15 @@ def _deny_sensitive_root(path: Path, backup_name: str | None = None) -> None:
 
 
 def _deny_sensitive_roots() -> None:
-    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)
+    for path in (HOST_K3S, HOST_KUBELET, HOST_RUN_K3S, HOST_RUN_CONTAINERD):
+        _deny_sensitive_root(path)
 
 
 def reconcile(public_key: Path) -> None:
     """Create the locked account and move only the Hermes authorization key."""
-    with account_lock(HOST_ETC, expected_uid=HOST_ROOT_UID):
-        _reconcile_databases()
-        _deny_sensitive_roots()
-        _move_key(public_key)
+    _reconcile_databases()
+    _deny_sensitive_roots()
+    _move_key(public_key)
 
 
 def main() -> int:

services/hermes/scripts/node_account_io.py

@@ -1,205 +0,0 @@
-#!/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,21 +13,6 @@ 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():
@@ -73,8 +58,8 @@ def _fixture(tmp_path: Path, monkeypatch):
     }
     for name, value in originals.items():
         (host_etc / name).write_text(value, encoding="utf-8")
-    key = _key(b"synthetic-hermes-key")
-    other = _key(b"human-operator-key")
+    key = "ssh-ed25519 " + base64.b64encode(b"synthetic-hermes-key").decode()
+    other = "ssh-ed25519 " + base64.b64encode(b"human-operator-key").decode()
     for user in ("atlas", "oceanus"):
         ssh = host_home / user / ".ssh"
         ssh.mkdir(parents=True)
@@ -87,8 +72,6 @@ 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)
@@ -223,12 +206,7 @@ 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 "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"
+    assert "supplementary" not in source.lower()
 
 
 def test_sensitive_roots_get_explicit_zero_permission_acl_for_hermes(
@@ -292,13 +270,9 @@ 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(
-        _key(b"synthetic-key") + "\n",
+        "ssh-ed25519 " + base64.b64encode(b"synthetic-key").decode() + "\n",
         encoding="utf-8",
     )
     moved = False
@@ -329,267 +303,3 @@ 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,7 +3,6 @@
 from __future__ import annotations
 
 import importlib.util
-import io
 import json
 import sys
 from email.message import Message
@@ -27,19 +26,9 @@ 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
@@ -51,7 +40,7 @@ class Response:
         return False
 
     def read(self, limit=-1):
-        return self.stream.read(limit)
+        return self.body if limit < 0 else self.body[:limit]
 
 
 def _receive_command(old: bytes, new: bytes, ref: bytes) -> bytes:
@@ -227,163 +216,6 @@ 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"])
@@ -459,13 +291,6 @@ 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(
@@ -525,67 +350,3 @@ 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"