services/hermes/agent-configmap.yaml

@@ -320,7 +320,9 @@ data:
     unauthenticated Gitea HTTP 404 as a missing private repository. Use
     brokered Git for clone, fetch, and creation of a new namespaced feature
     branch. Existing-ref updates, protected refs, deletion, and force-push are
-    rejected. For bounded
+    rejected. The broker inflates and scans every pushed object, so thin
+    packs are rejected; always push with `git push --no-thin` so the pack
+    is self-contained. For bounded
     repository/pull-request evidence or to create a review-ready draft PR, load
     `$manage-atlas-pull-requests` and use `/opt/scm/gitea_api.py`. The
     client carries no repository credential; a separate least-authority

services/hermes/scm-common/kustomization.yaml

@@ -4,8 +4,11 @@ kind: Kustomization
 configMapGenerator:
   - name: hermes-scm-boundary
     files:
+      - deadline_http.py=scripts/deadline_http.py
+      - git_pack_objects.py=scripts/git_pack_objects.py
       - gitea_api.py=scripts/gitea_api.py
       - gitea_api_policy.py=scripts/gitea_api_policy.py
+      - receive_pack_scan.py=scripts/receive_pack_scan.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

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

@@ -0,0 +1,308 @@
+#!/usr/bin/env python3
+"""Bounded Git pack inflation and delta resolution for quarantine scanning.
+
+Pushed objects arrive zlib-compressed inside a packfile, so any content
+screening must inflate the real payloads first.  This module unpacks one
+self-contained pack under explicit object, size, and checksum bounds and
+resolves every delta against bases inside the same pack.  Thin packs are
+rejected outright: a delta against a base the broker cannot see would leave
+part of the pushed content unscanned.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import tempfile
+import zlib
+from typing import BinaryIO
+
+from gitea_api_policy import PolicyError
+
+OBJECT_TYPE_NAMES = {1: b"commit", 2: b"tree", 3: b"blob", 4: b"tag"}
+OFS_DELTA = 6
+REF_DELTA = 7
+MAX_PACK_OBJECTS = 50_000
+MAX_OBJECT_BYTES = 32 * 1024 * 1024
+MAX_TOTAL_BYTES = 256 * 1024 * 1024
+MAX_RESOLVE_PASSES = 64
+SPOOL_MEMORY_LIMIT = 256 * 1024
+READ_CHUNK = 64 * 1024
+
+
+def _read_exact(stream: BinaryIO, size: int) -> bytes:
+    value = stream.read(size)
+    if len(value) != size:
+        raise PolicyError("Git pack data ended early")
+    return value
+
+
+def _object_header(stream: BinaryIO) -> tuple[int, int]:
+    """Parse one object's type and declared inflated size."""
+    byte = _read_exact(stream, 1)[0]
+    type_code = (byte >> 4) & 0x7
+    size = byte & 0x0F
+    shift = 4
+    while byte & 0x80:
+        if shift > 60:
+            raise PolicyError("Git object header is invalid")
+        byte = _read_exact(stream, 1)[0]
+        size |= (byte & 0x7F) << shift
+        shift += 7
+    return type_code, size
+
+
+def _delta_distance(stream: BinaryIO) -> int:
+    """Parse one ofs-delta base distance."""
+    byte = _read_exact(stream, 1)[0]
+    value = byte & 0x7F
+    while byte & 0x80:
+        byte = _read_exact(stream, 1)[0]
+        value = ((value + 1) << 7) | (byte & 0x7F)
+        if value > 1 << 48:
+            raise PolicyError("Git delta base distance is invalid")
+    return value
+
+
+def _inflate(stream: BinaryIO, declared: int) -> BinaryIO:
+    """Inflate one object into a quarantine spool, bounded by its header."""
+    spool = tempfile.SpooledTemporaryFile(  # noqa: SIM115 - caller owns returned spool
+        max_size=SPOOL_MEMORY_LIMIT, dir="/tmp"
+    )
+    decompressor = zlib.decompressobj()
+    produced = 0
+    try:
+        while not decompressor.eof:
+            data = stream.read(READ_CHUNK)
+            if not data:
+                raise PolicyError("Git pack object data ended early")
+            while True:
+                value = decompressor.decompress(data, READ_CHUNK)
+                produced += len(value)
+                if produced > declared:
+                    raise PolicyError("Git object size does not match its header")
+                spool.write(value)
+                data = decompressor.unconsumed_tail
+                if decompressor.eof or not data:
+                    break
+        if decompressor.unused_data:
+            stream.seek(-len(decompressor.unused_data), 1)
+        if produced != declared:
+            raise PolicyError("Git object size does not match its header")
+        spool.seek(0)
+        return spool
+    except zlib.error as exc:
+        spool.close()
+        raise PolicyError("Git pack object data is corrupt") from exc
+    except Exception:
+        spool.close()
+        raise
+
+
+def _delta_size(delta: bytes, position: int) -> tuple[int, int]:
+    size = 0
+    shift = 0
+    while True:
+        if position >= len(delta) or shift > 60:
+            raise PolicyError("Git delta header is invalid")
+        byte = delta[position]
+        position += 1
+        size |= (byte & 0x7F) << shift
+        shift += 7
+        if not byte & 0x80:
+            return size, position
+
+
+def _apply_delta(base: bytes, delta: bytes) -> bytes:
+    """Apply one Git delta and require exact declared sizes throughout."""
+    source_size, position = _delta_size(delta, 0)
+    target_size, position = _delta_size(delta, position)
+    if source_size != len(base) or target_size > MAX_OBJECT_BYTES:
+        raise PolicyError("Git delta sizes are invalid")
+    parts: list[bytes] = []
+    produced = 0
+    while position < len(delta):
+        opcode = delta[position]
+        position += 1
+        if opcode & 0x80:
+            operands = [bit for bit in (1, 2, 4, 8, 16, 32, 64) if opcode & bit]
+            if position + len(operands) > len(delta):
+                raise PolicyError("Git delta copy is truncated")
+            offset = 0
+            size = 0
+            for shift, bit in enumerate((1, 2, 4, 8)):
+                if opcode & bit:
+                    offset |= delta[position] << (8 * shift)
+                    position += 1
+            for shift, bit in enumerate((16, 32, 64)):
+                if opcode & bit:
+                    size |= delta[position] << (8 * shift)
+                    position += 1
+            size = size or 0x10000
+            if offset + size > len(base):
+                raise PolicyError("Git delta copy is out of bounds")
+            parts.append(base[offset : offset + size])
+            produced += size
+        elif opcode:
+            if position + opcode > len(delta):
+                raise PolicyError("Git delta insert is truncated")
+            parts.append(delta[position : position + opcode])
+            position += opcode
+            produced += opcode
+        else:
+            raise PolicyError("Git delta instruction is invalid")
+        if produced > target_size:
+            raise PolicyError("Git delta result exceeds its declared size")
+    if produced != target_size:
+        raise PolicyError("Git delta result size is invalid")
+    return b"".join(parts)
+
+
+def _payload_sha(type_code: int, payload: BinaryIO, size: int) -> bytes:
+    digest = hashlib.sha1(OBJECT_TYPE_NAMES[type_code] + b" %d\x00" % size)
+    payload.seek(0)
+    while True:
+        chunk = payload.read(READ_CHUNK)
+        if not chunk:
+            break
+        digest.update(chunk)
+    payload.seek(0)
+    return digest.digest()
+
+
+def _verify_checksum(stream: BinaryIO, start: int, end: int) -> None:
+    trailer = _read_exact(stream, 20)
+    stream.seek(start)
+    digest = hashlib.sha1()
+    remaining = end - start
+    while remaining:
+        chunk = stream.read(min(READ_CHUNK, remaining))
+        digest.update(chunk)
+        remaining -= len(chunk)
+    if digest.digest() != trailer:
+        raise PolicyError("Git pack checksum is invalid")
+    stream.seek(end + 20)
+
+
+class _Entry:
+    """One pack entry: either resolved content or an unresolved delta."""
+
+    def __init__(self, type_code: int, payload: BinaryIO, size: int, base):
+        self.type_code = type_code
+        self.payload = payload
+        self.size = size
+        self.base = base
+
+
+def _resolve(entries: dict[int, _Entry], total: list[int]) -> None:
+    """Resolve every delta against in-pack bases only, to a fixed point."""
+    by_sha: dict[bytes, int] = {}
+    for offset, entry in entries.items():
+        if entry.base is None:
+            by_sha.setdefault(
+                _payload_sha(entry.type_code, entry.payload, entry.size), offset
+            )
+    for _ in range(MAX_RESOLVE_PASSES):
+        pending = [item for item in entries.items() if item[1].base is not None]
+        if not pending:
+            return
+        progressed = False
+        for offset, entry in pending:
+            base_key = entry.base
+            base_offset = base_key if isinstance(base_key, int) else by_sha.get(base_key)
+            base = entries.get(base_offset) if base_offset is not None else None
+            if base is None or base.base is not None:
+                if isinstance(base_key, int) and base is None:
+                    raise PolicyError("Git delta base offset is invalid")
+                continue
+            base.payload.seek(0)
+            entry.payload.seek(0)
+            value = _apply_delta(base.payload.read(), entry.payload.read())
+            total[0] += len(value)
+            if total[0] > MAX_TOTAL_BYTES:
+                raise PolicyError("Git pack contents exceed the safe size limit")
+            entry.payload.close()
+            spool = tempfile.SpooledTemporaryFile(  # noqa: SIM115 - entry owns spool
+                max_size=SPOOL_MEMORY_LIMIT, dir="/tmp"
+            )
+            spool.write(value)
+            spool.seek(0)
+            entry.type_code = base.type_code
+            entry.payload = spool
+            entry.size = len(value)
+            entry.base = None
+            by_sha.setdefault(_payload_sha(entry.type_code, spool, entry.size), offset)
+            progressed = True
+        if not progressed:
+            raise PolicyError("Git delta base is outside the pack; push full packs")
+    raise PolicyError("Git delta chains exceed the safe resolution limit")
+
+
+def unpack_objects(stream: BinaryIO) -> list[tuple[int, BinaryIO]]:
+    """Inflate one self-contained pack and return every resolved payload."""
+    start = stream.tell()
+    header = _read_exact(stream, 12)
+    version = int.from_bytes(header[4:8], "big")
+    count = int.from_bytes(header[8:12], "big")
+    if header[:4] != b"PACK" or version not in {2, 3}:
+        raise PolicyError("Git pack header is invalid")
+    if count > MAX_PACK_OBJECTS:
+        raise PolicyError("Git pack contains too many objects")
+    entries: dict[int, _Entry] = {}
+    total = [0]
+    try:
+        for _ in range(count):
+            offset = stream.tell() - start
+            type_code, declared = _object_header(stream)
+            if declared > MAX_OBJECT_BYTES:
+                raise PolicyError("Git object exceeds the safe size limit")
+            total[0] += declared
+            if total[0] > MAX_TOTAL_BYTES:
+                raise PolicyError("Git pack contents exceed the safe size limit")
+            base: int | bytes | None = None
+            if type_code == OFS_DELTA:
+                distance = _delta_distance(stream)
+                base = offset - distance
+                if base < 0:
+                    raise PolicyError("Git delta base offset is invalid")
+            elif type_code == REF_DELTA:
+                base = _read_exact(stream, 20)
+            elif type_code not in OBJECT_TYPE_NAMES:
+                raise PolicyError("Git pack object type is invalid")
+            entries[offset] = _Entry(type_code, _inflate(stream, declared), declared, base)
+        _verify_checksum(stream, start, stream.tell())
+        _resolve(entries, total)
+    except Exception:
+        for entry in entries.values():
+            entry.payload.close()
+        raise
+    for entry in entries.values():
+        entry.payload.seek(0)
+    return [
+        (entries[offset].type_code, entries[offset].payload)
+        for offset in sorted(entries)
+    ]
+
+
+def parse_loose_object(value: bytes) -> tuple[int, bytes]:
+    """Inflate one quarantined loose object and return its typed payload."""
+    if len(value) > MAX_OBJECT_BYTES:
+        raise PolicyError("Git object exceeds the safe size limit")
+    decompressor = zlib.decompressobj()
+    try:
+        inflated = decompressor.decompress(value, MAX_OBJECT_BYTES + 1)
+    except zlib.error as exc:
+        raise PolicyError("Git loose object data is corrupt") from exc
+    if not decompressor.eof or decompressor.unused_data or len(inflated) > MAX_OBJECT_BYTES:
+        raise PolicyError("Git loose object framing is invalid")
+    kind, separator, remainder = inflated.partition(b"\x00")
+    name, space, size = kind.partition(b" ")
+    types = {label: code for code, label in OBJECT_TYPE_NAMES.items()}
+    if (
+        not separator
+        or not space
+        or name not in types
+        or not size.isdigit()
+        or int(size) != len(remainder)
+    ):
+        raise PolicyError("Git loose object header is invalid")
+    return types[name], remainder

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

@@ -0,0 +1,143 @@
+#!/usr/bin/env python3
+"""Validate Git receive-pack requests and scan pushed object content.
+
+Ref commands may only create new namespaced feature branches.  Every pushed
+object is then inflated in quarantine and its decompressed payload scanned,
+so credentials cannot ride through the broker inside compressed pack data.
+The screening is a fail-closed accident barrier for credential-shaped
+content: exact runtime-token forms, private keys, SSH key material, and
+known provider token formats.  Entropy heuristics are deliberately excluded
+so SOPS ciphertext, pinned digests, and lock files remain pushable; a short
+secret disguised as prose is not reliably detectable and must never be
+supplied by callers.
+"""
+
+from __future__ import annotations
+
+import io
+import re
+from typing import BinaryIO
+
+import git_pack_objects
+from gitea_api_policy import (
+    STANDALONE_SECRET_PATTERNS,
+    PolicyError,
+    _validate_ref,
+)
+
+ZERO_SHA = b"0" * 40
+MAX_PUSH_COMMANDS = 16
+SCAN_CHUNK = 256 * 1024
+SCAN_CARRY = 8 * 1024
+FEATURE_REF_RE = re.compile(
+    r"refs/heads/(?:(?:feature|fix|hermes|handoff)/[A-Za-z0-9][A-Za-z0-9._/-]{0,190})\Z"
+)
+PKT_HEADER_RE = re.compile(rb"[0-9a-f]{4}\Z")
+CONTENT_SECRET_PATTERNS = tuple(
+    re.compile(pattern.pattern.encode("ascii"), re.IGNORECASE)
+    for pattern in STANDALONE_SECRET_PATTERNS
+) + (
+    re.compile(rb"(?i)(?<![A-Za-z0-9])AGE-SECRET-KEY-1[A-Z0-9]{20,}"),
+    re.compile(rb"(?i)PuTTY-User-Key-File"),
+    re.compile(
+        rb"(?im)^[ \t]*authorization[ \t]*:[ \t]*(?:basic|bearer|token)[ \t]+"
+        rb"(?=[A-Za-z0-9+/_=.:-]*[0-9+/=.-])[A-Za-z0-9+/_=.:-]{8,}"
+    ),
+)
+
+
+def _read_pkt_line(stream: BinaryIO) -> bytes | None:
+    """Read one pkt-line; None marks the flush packet before the pack."""
+    header = stream.read(4)
+    if not header:
+        raise PolicyError("Git receive-pack request omitted its command terminator")
+    if not PKT_HEADER_RE.fullmatch(header):
+        raise PolicyError("Git receive-pack command framing is invalid")
+    size = int(header, 16)
+    if size == 0:
+        return None
+    if size < 4:
+        raise PolicyError("Git receive-pack command framing is invalid")
+    value = stream.read(size - 4)
+    if len(value) != size - 4:
+        raise PolicyError("Git receive-pack command framing is invalid")
+    return value
+
+
+def _validate_command(line: bytes, token: str, forbidden: tuple[bytes, ...]) -> None:
+    """Permit only creation of one new, namespaced feature branch."""
+    if any(form in line for form in forbidden):
+        raise PolicyError("Git request contains runtime credential material")
+    command = line.rstrip(b"\n").split(b"\x00", 1)[0]
+    fields = command.split(b" ")
+    if len(fields) != 3 or not all(
+        re.fullmatch(rb"[0-9a-f]{40}", item) for item in fields[:2]
+    ):
+        raise PolicyError("Git receive-pack ref command is invalid")
+    old_sha, new_sha, raw_ref = fields
+    if old_sha != ZERO_SHA or new_sha == ZERO_SHA:
+        raise PolicyError("Git broker permits only new feature-branch creation")
+    try:
+        ref = raw_ref.decode("ascii")
+    except UnicodeDecodeError as exc:
+        raise PolicyError("Git ref must be canonical ASCII") from exc
+    if not FEATURE_REF_RE.fullmatch(ref):
+        raise PolicyError("Git push is limited to namespaced feature branches")
+    _validate_ref(ref.removeprefix("refs/heads/"), "head", forbidden=(token,))
+
+
+def _scan_payload(payload: BinaryIO, forbidden: tuple[bytes, ...]) -> None:
+    """Scan one decompressed object payload across chunk boundaries."""
+    carry = b""
+    while True:
+        chunk = payload.read(SCAN_CHUNK)
+        if not chunk:
+            return
+        window = carry + chunk
+        if any(form in window for form in forbidden):
+            raise PolicyError("Git push contains runtime credential material")
+        if any(pattern.search(window) for pattern in CONTENT_SECRET_PATTERNS):
+            raise PolicyError("Git push contains credential-shaped content")
+        carry = window[-SCAN_CARRY:]
+
+
+def _scan_pack(stream: BinaryIO, forbidden: tuple[bytes, ...]) -> None:
+    objects = git_pack_objects.unpack_objects(stream)
+    try:
+        if stream.read(1):
+            raise PolicyError("Git request has bytes after its pack")
+        for _type_code, payload in objects:
+            _scan_payload(payload, forbidden)
+    finally:
+        for _type_code, payload in objects:
+            payload.close()
+
+
+def validate_receive_pack(
+    body: bytes | BinaryIO, token: str, forbidden: tuple[bytes, ...]
+) -> None:
+    """Validate one whole receive-pack request: commands, pack, and content."""
+    stream = io.BytesIO(body) if isinstance(body, bytes) else body
+    stream.seek(0)
+    try:
+        commands = 0
+        while True:
+            line = _read_pkt_line(stream)
+            if line is None:
+                break
+            _validate_command(line, token, forbidden)
+            commands += 1
+            if commands > MAX_PUSH_COMMANDS:
+                raise PolicyError("Git push contains too many ref commands")
+        if commands == 0:
+            raise PolicyError("Git receive-pack request has no ref command")
+        _scan_pack(stream, forbidden)
+    finally:
+        stream.seek(0)
+
+
+def scan_loose_object(value: bytes, forbidden: tuple[bytes, ...]) -> int:
+    """Scan one quarantined loose object and return its parsed type code."""
+    type_code, payload = git_pack_objects.parse_loose_object(value)
+    _scan_payload(io.BytesIO(payload), forbidden)
+    return type_code

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

@@ -16,6 +16,7 @@ import urllib.request
 from http.server import BaseHTTPRequestHandler
 from typing import BinaryIO
 
+import deadline_http
 from gitea_api import (
     CANONICAL_BASE_URL,
     PolicyError,
@@ -23,7 +24,8 @@ from gitea_api import (
     read,
     read_token,
 )
-from gitea_api_policy import _reject_forbidden, _validate_ref, _validate_repo
+from gitea_api_policy import _reject_forbidden, _validate_repo
+from receive_pack_scan import validate_receive_pack
 from scm_broker_io import RejectRedirect, response_status as _status, spool_response
 from scm_broker_server import AbsoluteHeaderDeadlineMixin, BoundedThreadingHTTPServer
 
@@ -32,24 +34,17 @@ GIT_USER = "hermes-automation"
 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
+UPSTREAM_DEADLINE_SECONDS = 300.0
 GIT_PATH_RE = re.compile(
     r"/git/atlas/(?P<repo>[A-Za-z0-9][A-Za-z0-9._-]{0,99})\.git/"
     r"(?P<operation>info/refs|git-upload-pack|git-receive-pack)\Z"
 )
-FEATURE_REF_RE = re.compile(
-    r"refs/heads/(?:(?:feature|fix|hermes|handoff)/[A-Za-z0-9][A-Za-z0-9._/-]{0,190})\Z"
-)
-
-
-UPSTREAM_OPENER = urllib.request.build_opener(RejectRedirect())
 
 
 def _read_bounded(
@@ -126,14 +121,16 @@ def _spool_bounded(
         raise
 
 
-def _spool_response(response, maximum: int, token: str) -> tuple[BinaryIO, int]:
+def _spool_response(
+    response, maximum: int, token: str, deadline_seconds: float
+) -> 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,
+        deadline_seconds=deadline_seconds,
     )
 
 
@@ -211,61 +208,14 @@ def _credential_forms(token: str) -> tuple[bytes, ...]:
     return token.encode("utf-8"), basic, b"Basic " + basic
 
 
-def _reject_credential_bytes(value: bytes, token: str, context: str) -> None:
-    if any(form in value for form in _credential_forms(token)):
-        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:
-    """Permit only creation of new, namespaced feature branches."""
-    body = _receive_prefix(body)
-    _reject_credential_bytes(body, token, "Git request")
-    position = 0
-    commands = 0
-    while position + 4 <= len(body):
-        header = body[position : position + 4]
-        if not re.fullmatch(rb"[0-9a-f]{4}", header):
-            raise PolicyError("Git receive-pack command framing is invalid")
-        size = int(header, 16)
-        if size == 0:
-            if commands == 0:
-                raise PolicyError("Git receive-pack request has no ref command")
-            return
-        if size < 4 or position + size > len(body):
-            raise PolicyError("Git receive-pack command framing is invalid")
-        command = body[position + 4 : position + size].rstrip(b"\n")
-        command = command.split(b"\x00", 1)[0]
-        fields = command.split(b" ")
-        if len(fields) != 3 or not all(
-            re.fullmatch(rb"[0-9a-f]{40}", item) for item in fields[:2]
-        ):
-            raise PolicyError("Git receive-pack ref command is invalid")
-        old_sha, new_sha, raw_ref = fields
-        if old_sha != ZERO_SHA or new_sha == ZERO_SHA:
-            raise PolicyError("Git broker permits only new feature-branch creation")
-        try:
-            ref = raw_ref.decode("ascii")
-        except UnicodeDecodeError as exc:
-            raise PolicyError("Git ref must be canonical ASCII") from exc
-        if not FEATURE_REF_RE.fullmatch(ref):
-            raise PolicyError("Git push is limited to namespaced feature branches")
-        _validate_ref(ref.removeprefix("refs/heads/"), "head", forbidden=(token,))
-        commands += 1
-        if commands > MAX_PUSH_COMMANDS:
-            raise PolicyError("Git push contains too many ref commands")
-        position += size
-    raise PolicyError("Git receive-pack request omitted its command terminator")
+    """Permit only vetted new-feature-branch pushes with scanned contents."""
+    validate_receive_pack(body, token, _credential_forms(token))
+
+
+def _guarded_opener(guard: deadline_http.StreamDeadline):
+    """Build an opener whose connection obeys one absolute stream deadline."""
+    return urllib.request.build_opener(RejectRedirect(), *guard.handlers()).open
 
 
 def _upstream_git_request(
@@ -278,7 +228,7 @@ def _upstream_git_request(
     token: str,
     body_length: int | None = None,
     stream_result: bool = False,
-    opener=UPSTREAM_OPENER.open,
+    opener=None,
 ) -> bytes | tuple[BinaryIO, int]:
     credentials = base64.b64encode(f"{GIT_USER}:{token}".encode()).decode("ascii")
     headers = {"Accept": expected_type, "Authorization": f"Basic {credentials}"}
@@ -296,12 +246,21 @@ def _upstream_git_request(
         method=method,
         headers=headers,
     )
-    with opener(request, timeout=120) as response:
-        if _status(response) != 200:
-            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)
+    # One wall-clock budget covers connect, send, headers, and body read; the
+    # watchdog closes the upstream socket if any phase overstays it.
+    guard = deadline_http.StreamDeadline(UPSTREAM_DEADLINE_SECONDS)
+    try:
+        open_upstream = opener if opener is not None else _guarded_opener(guard)
+        with open_upstream(request, timeout=guard.remaining()) as response:
+            if _status(response) != 200:
+                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, guard.remaining()
+            )
+    finally:
+        guard.cancel()
     if stream_result:
         return result, result_length
     try:

testing/tests/test_hermes_deadline_stream.py

@@ -0,0 +1,217 @@
+"""Watchdog contracts for absolute deadlines on streaming Git exchanges."""
+
+from __future__ import annotations
+
+import threading
+import time
+import urllib.request
+from http.server import BaseHTTPRequestHandler, HTTPServer
+
+import pytest
+
+from testing.tests.test_hermes_scm_broker_support import Response, _load
+
+
+def test_stream_deadline_validates_bounds_and_counts_down():
+    module = _load("deadline_http")
+    for timeout in (0, -1, module.MAX_STREAM_SECONDS + 1):
+        with pytest.raises(module.PolicyError, match="bounds are invalid"):
+            module.StreamDeadline(timeout)
+    guard = module.StreamDeadline(30)
+    try:
+        first = guard.remaining()
+        assert 0 < first <= 30
+    finally:
+        guard.cancel()
+
+
+class _FakeTimer:
+    def __init__(self, timeout, callback):
+        self.timeout = timeout
+        self.callback = callback
+        self.daemon = False
+        self.started = False
+        self.cancelled = False
+
+    def start(self):
+        self.started = True
+
+    def cancel(self):
+        self.cancelled = True
+
+
+class _FakeSocket:
+    def __init__(self, fail: bool = False):
+        self.fail = fail
+        self.shut = False
+
+    def shutdown(self, _how):
+        if self.fail:
+            raise OSError("already gone")
+        self.shut = True
+
+
+class _FakeConnection:
+    def __init__(self, sock=None):
+        self.sock = sock
+        self.closed = False
+
+    def close(self):
+        self.closed = True
+
+
+def test_expiry_shuts_down_and_closes_every_tracked_connection():
+    module = _load("deadline_http")
+    guard = module.StreamDeadline(30, timer=_FakeTimer)
+    assert guard._timer.started and guard._timer.daemon
+    healthy = _FakeConnection(_FakeSocket())
+    broken = _FakeConnection(_FakeSocket(fail=True))
+    bare = _FakeConnection()
+    for connection in (healthy, broken, bare):
+        assert guard._track(connection) is connection
+    guard._expire()
+    assert guard.expired
+    assert healthy.sock.shut and healthy.closed
+    assert broken.closed and not broken.sock.shut
+    assert bare.closed
+    with pytest.raises(module.PolicyError, match="deadline exceeded"):
+        guard.remaining()
+    late = _FakeConnection()
+    with pytest.raises(module.PolicyError, match="deadline exceeded"):
+        guard._track(late)
+    assert late.closed
+    guard.cancel()
+    assert guard._timer.cancelled
+
+
+def test_handlers_build_guarded_connections_without_network(monkeypatch):
+    module = _load("deadline_http")
+    guard = module.StreamDeadline(30, timer=_FakeTimer)
+    http_handler, https_handler = guard.handlers()
+    assert isinstance(http_handler, urllib.request.HTTPHandler)
+    assert isinstance(https_handler, urllib.request.HTTPSHandler)
+    for handler, opener_name in (
+        (http_handler, "http_open"),
+        (https_handler, "https_open"),
+    ):
+        seen = {}
+        monkeypatch.setattr(
+            handler,
+            "do_open",
+            lambda factory, req, seen=seen: seen.update(factory=factory, req=req),
+        )
+        getattr(handler, opener_name)("request")
+        connection = seen["factory"]("127.0.0.1", timeout=1)
+        assert connection in guard._connections
+        connection.close()
+    guard.cancel()
+
+
+class _TrickleHandler(BaseHTTPRequestHandler):
+    def do_GET(self):
+        time.sleep(10)
+
+    def log_message(self, *_args):
+        return
+
+
+def test_watchdog_interrupts_a_hung_streaming_exchange():
+    module = _load("deadline_http")
+    server = HTTPServer(("127.0.0.1", 0), _TrickleHandler)
+    threading.Thread(target=server.serve_forever, daemon=True).start()
+    guard = module.StreamDeadline(1.0)
+    opener = urllib.request.build_opener(*guard.handlers())
+    started = time.monotonic()
+    try:
+        with pytest.raises(OSError):
+            opener.open(
+                f"http://127.0.0.1:{server.server_port}/slow", timeout=30
+            ).read()
+        assert time.monotonic() - started < 5
+        assert guard.expired
+    finally:
+        guard.cancel()
+        server.shutdown()
+
+
+def test_upstream_git_request_runs_under_one_wall_clock_budget(monkeypatch):
+    broker = _load("scm_broker")
+    seen = {}
+
+    class RecordingGuard:
+        def __init__(self, timeout, **_kwargs):
+            seen["timeout"] = timeout
+            self.cancelled = False
+            seen["guard"] = self
+
+        def remaining(self):
+            return 12.5
+
+        def cancel(self):
+            self.cancelled = True
+
+        def handlers(self):
+            return ()
+
+    monkeypatch.setattr(broker.deadline_http, "StreamDeadline", RecordingGuard)
+
+    def opener(request, timeout):
+        seen["opener_timeout"] = timeout
+        return Response(b"result", content_type="application/x-git-upload-pack-result")
+
+    result = broker._upstream_git_request(
+        "/atlas/cassandra.git/git-upload-pack",
+        method="POST",
+        body=b"request",
+        content_type="application/x-git-upload-pack-request",
+        expected_type="application/x-git-upload-pack-result",
+        token="sentinel",
+        opener=opener,
+    )
+    assert result == b"result"
+    assert seen["timeout"] == broker.UPSTREAM_DEADLINE_SECONDS
+    assert seen["opener_timeout"] == 12.5
+    assert seen["guard"].cancelled
+
+
+def test_upstream_git_request_cancels_guard_when_deadline_already_passed(monkeypatch):
+    broker = _load("scm_broker")
+    cancelled = []
+
+    class ExpiredGuard:
+        def __init__(self, _timeout, **_kwargs):
+            pass
+
+        def remaining(self):
+            raise broker.PolicyError("HTTP stream deadline exceeded")
+
+        def cancel(self):
+            cancelled.append(True)
+
+        def handlers(self):
+            return ()
+
+    monkeypatch.setattr(broker.deadline_http, "StreamDeadline", ExpiredGuard)
+    with pytest.raises(broker.PolicyError, match="deadline exceeded"):
+        broker._upstream_git_request(
+            "/atlas/cassandra.git/git-upload-pack",
+            method="POST",
+            body=b"request",
+            content_type="application/x-git-upload-pack-request",
+            expected_type="application/x-git-upload-pack-result",
+            token="sentinel",
+            opener=lambda *_a, **_k: None,
+        )
+    assert cancelled == [True]
+
+
+def test_guarded_opener_wires_redirect_rejection_and_deadline_handlers():
+    broker = _load("scm_broker")
+    module = _load("deadline_http")
+    guard = module.StreamDeadline(30, timer=_FakeTimer)
+    opener = broker._guarded_opener(guard)
+    assert callable(opener)
+    handlers = opener.__self__.handlers
+    assert any(isinstance(item, broker.RejectRedirect) for item in handlers)
+    assert any(isinstance(item, urllib.request.HTTPSHandler) for item in handlers)
+    guard.cancel()

testing/tests/test_hermes_git_pack_objects.py

@@ -0,0 +1,288 @@
+"""Bounded pack inflation and delta-resolution contracts for pushed objects."""
+
+from __future__ import annotations
+
+import hashlib
+import io
+import zlib
+
+import pytest
+
+from testing.tests.test_hermes_scm_broker_support import (
+    _load,
+    _object_entry,
+    _pack_of,
+)
+
+
+def _delta_header(type_code: int, size: int, suffix: bytes = b"") -> bytes:
+    byte = (type_code << 4) | (size & 0x0F)
+    size >>= 4
+    header = bytearray()
+    while size:
+        header.append(byte | 0x80)
+        byte = size & 0x7F
+        size >>= 7
+    header.append(byte)
+    return bytes(header) + suffix
+
+
+def _size_varint(value: int) -> bytes:
+    out = bytearray()
+    while True:
+        byte = value & 0x7F
+        value >>= 7
+        if value:
+            out.append(byte | 0x80)
+        else:
+            out.append(byte)
+            return bytes(out)
+
+
+def _ofs_distance(value: int) -> bytes:
+    out = [value & 0x7F]
+    value >>= 7
+    while value:
+        value -= 1
+        out.insert(0, 0x80 | (value & 0x7F))
+        value >>= 7
+    return bytes(out)
+
+
+def _delta(source: bytes, target_size: int, instructions: bytes) -> bytes:
+    return _size_varint(len(source)) + _size_varint(target_size) + instructions
+
+
+def _blob_sha(payload: bytes) -> bytes:
+    return hashlib.sha1(b"blob %d\x00" % len(payload) + payload).digest()
+
+
+def _read_all(objects):
+    try:
+        return [(code, payload.read()) for code, payload in objects]
+    finally:
+        for _code, payload in objects:
+            payload.close()
+
+
+def test_unpack_returns_typed_payloads_for_every_base_object_kind():
+    module = _load("git_pack_objects")
+    entries = [
+        _object_entry(1, b"tree 1\n"),
+        _object_entry(2, b"\x00tree-bytes"),
+        _object_entry(3, b"blob contents"),
+        _object_entry(4, b"tag contents"),
+    ]
+    result = _read_all(module.unpack_objects(io.BytesIO(_pack_of(entries))))
+    assert result == [
+        (1, b"tree 1\n"),
+        (2, b"\x00tree-bytes"),
+        (3, b"blob contents"),
+        (4, b"tag contents"),
+    ]
+
+
+def test_unpack_accepts_the_empty_pack():
+    module = _load("git_pack_objects")
+    assert module.unpack_objects(io.BytesIO(_pack_of([]))) == []
+
+
+def test_ofs_delta_resolves_against_its_in_pack_base():
+    module = _load("git_pack_objects")
+    base = b"base payload for copying"
+    base_entry = _object_entry(3, base)
+    # Copy the first 4 bytes, then insert new bytes.
+    instructions = bytes([0x80 | 0x10, 4]) + bytes([5]) + b"+tail"
+    delta = _delta(base, 9, instructions)
+    delta_entry = _delta_header(
+        6, len(delta), _ofs_distance(len(base_entry))
+    ) + zlib.compress(delta)
+    result = _read_all(module.unpack_objects(io.BytesIO(_pack_of([base_entry, delta_entry]))))
+    assert result == [(3, base), (3, b"base+tail")]
+
+
+def test_ref_delta_resolves_by_sha_and_supports_chains():
+    module = _load("git_pack_objects")
+    base = b"chain base"
+    first = _delta(base, 4, bytes([4]) + b"one!")
+    second = _delta(b"one!", 4, bytes([4]) + b"two!")
+    entries = [
+        _object_entry(3, base),
+        _delta_header(7, len(first), _blob_sha(base)) + zlib.compress(first),
+        _delta_header(7, len(second), _blob_sha(b"one!")) + zlib.compress(second),
+    ]
+    result = _read_all(module.unpack_objects(io.BytesIO(_pack_of(entries))))
+    assert [payload for _code, payload in result] == [base, b"one!", b"two!"]
+
+
+def test_thin_pack_ref_delta_with_external_base_is_rejected():
+    module = _load("git_pack_objects")
+    delta = _delta(b"absent", 1, bytes([1]) + b"x")
+    entry = _delta_header(7, len(delta), b"\xaa" * 20) + zlib.compress(delta)
+    with pytest.raises(module.PolicyError, match="push full packs"):
+        module.unpack_objects(io.BytesIO(_pack_of([entry])))
+
+
+def test_ofs_delta_bad_offsets_fail_closed():
+    module = _load("git_pack_objects")
+    base_entry = _object_entry(3, b"base")
+    delta = _delta(b"base", 1, bytes([1]) + b"x")
+    # Distance beyond the pack start.
+    early = _delta_header(6, len(delta), _ofs_distance(999)) + zlib.compress(delta)
+    with pytest.raises(module.PolicyError, match="base offset is invalid"):
+        module.unpack_objects(io.BytesIO(_pack_of([base_entry, early])))
+    # Distance landing between object boundaries.
+    misaligned = _delta_header(6, len(delta), _ofs_distance(3)) + zlib.compress(delta)
+    with pytest.raises(module.PolicyError, match="base offset is invalid"):
+        module.unpack_objects(io.BytesIO(_pack_of([base_entry, misaligned])))
+
+
+@pytest.mark.parametrize(
+    ("mutate", "match"),
+    [
+        (lambda pack: b"JUNK" + pack[4:], "header is invalid"),
+        (lambda pack: pack[:4] + (9).to_bytes(4, "big") + pack[8:], "header is invalid"),
+        (lambda pack: pack[:8] + (2).to_bytes(4, "big") + pack[12:], "Git"),
+        (lambda pack: pack[:-1], "ended early"),
+        (lambda pack: pack[:-20] + b"\x00" * 20, "checksum is invalid"),
+        (lambda pack: pack[:11], "ended early"),
+    ],
+)
+def test_malformed_pack_containers_fail_closed(mutate, match):
+    module = _load("git_pack_objects")
+    pack = _pack_of([_object_entry(3, b"payload")])
+    with pytest.raises(module.PolicyError, match=match):
+        module.unpack_objects(io.BytesIO(mutate(pack)))
+
+
+def test_object_count_and_type_bounds_fail_closed(monkeypatch):
+    module = _load("git_pack_objects")
+    body = b"PACK" + (2).to_bytes(4, "big") + (module.MAX_PACK_OBJECTS + 1).to_bytes(
+        4, "big"
+    )
+    with pytest.raises(module.PolicyError, match="too many objects"):
+        module.unpack_objects(io.BytesIO(body + hashlib.sha1(body).digest()))
+    reserved = _delta_header(5, 1) + zlib.compress(b"x")
+    with pytest.raises(module.PolicyError, match="object type is invalid"):
+        module.unpack_objects(io.BytesIO(_pack_of([reserved])))
+
+
+def test_declared_size_mismatches_and_zlib_bombs_fail_closed(monkeypatch):
+    module = _load("git_pack_objects")
+    longer = _delta_header(3, 2) + zlib.compress(b"longer-than-two")
+    with pytest.raises(module.PolicyError, match="does not match its header"):
+        module.unpack_objects(io.BytesIO(_pack_of([longer])))
+    shorter = _delta_header(3, 9) + zlib.compress(b"x")
+    with pytest.raises(module.PolicyError, match="does not match its header"):
+        module.unpack_objects(io.BytesIO(_pack_of([shorter])))
+    corrupt = _delta_header(3, 4) + b"\x00not-zlib\x00"
+    with pytest.raises(module.PolicyError, match="corrupt"):
+        module.unpack_objects(io.BytesIO(_pack_of([corrupt])))
+
+    monkeypatch.setattr(module, "MAX_OBJECT_BYTES", 4)
+    with pytest.raises(module.PolicyError, match="exceeds the safe size limit"):
+        module.unpack_objects(io.BytesIO(_pack_of([_object_entry(3, b"12345")])))
+    monkeypatch.setattr(module, "MAX_OBJECT_BYTES", 1024)
+    monkeypatch.setattr(module, "MAX_TOTAL_BYTES", 8)
+    with pytest.raises(module.PolicyError, match="contents exceed"):
+        module.unpack_objects(
+            io.BytesIO(_pack_of([_object_entry(3, b"12345"), _object_entry(3, b"12345")]))
+        )
+
+
+def test_object_and_distance_varints_are_bounded():
+    module = _load("git_pack_objects")
+    with pytest.raises(module.PolicyError, match="object header is invalid"):
+        module._object_header(io.BytesIO(b"\xff" * 12))
+    with pytest.raises(module.PolicyError, match="ended early"):
+        module._object_header(io.BytesIO(b"\x80"))
+    with pytest.raises(module.PolicyError, match="distance is invalid"):
+        module._delta_distance(io.BytesIO(b"\xff" * 12))
+
+
+@pytest.mark.parametrize(
+    ("base", "delta", "match"),
+    [
+        (b"base", b"", "delta header is invalid"),
+        (b"base", _size_varint(4), "delta header is invalid"),
+        (b"base", _size_varint(9) + _size_varint(1) + bytes([1]) + b"x", "sizes are invalid"),
+        (b"base", _size_varint(4) + _size_varint(2) + bytes([0]), "instruction is invalid"),
+        (b"base", _size_varint(4) + _size_varint(2) + bytes([5]) + b"ab", "insert is truncated"),
+        (b"base", _size_varint(4) + _size_varint(2) + bytes([0x80 | 0x10]), "copy is truncated"),
+        (
+            b"base",
+            _size_varint(4) + _size_varint(2) + bytes([0x80 | 0x01 | 0x10, 3, 9]),
+            "copy is out of bounds",
+        ),
+        (b"base", _size_varint(4) + _size_varint(1) + bytes([2]) + b"ab", "exceeds its declared size"),
+        (b"base", _size_varint(4) + _size_varint(3) + bytes([2]) + b"ab", "result size is invalid"),
+    ],
+)
+def test_delta_application_rejects_malformed_instructions(base, delta, match):
+    module = _load("git_pack_objects")
+    with pytest.raises(module.PolicyError, match=match):
+        module._apply_delta(base, delta)
+
+
+def test_delta_copy_with_zero_size_field_means_the_git_default():
+    module = _load("git_pack_objects")
+    base = b"z" * 0x10000
+    delta = _delta(base, 0x10000, bytes([0x80 | 0x10 | 0x20, 0, 0]))
+    assert module._apply_delta(base, delta) == base
+
+
+def test_delta_target_size_cap_applies(monkeypatch):
+    module = _load("git_pack_objects")
+    monkeypatch.setattr(module, "MAX_OBJECT_BYTES", 4)
+    delta = _delta(b"base", 5, bytes([5]) + b"abcde")
+    with pytest.raises(module.PolicyError, match="sizes are invalid"):
+        module._apply_delta(b"base", delta)
+
+
+def test_resolution_pass_and_total_caps_apply(monkeypatch):
+    module = _load("git_pack_objects")
+    base = b"pass base"
+    first = _delta(base, 4, bytes([4]) + b"one!")
+    second = _delta(b"one!", 4, bytes([4]) + b"two!")
+    entries = [
+        _delta_header(7, len(second), _blob_sha(b"one!")) + zlib.compress(second),
+        _object_entry(3, base),
+        _delta_header(7, len(first), _blob_sha(base)) + zlib.compress(first),
+    ]
+    pack = _pack_of(entries)
+    monkeypatch.setattr(module, "MAX_RESOLVE_PASSES", 1)
+    with pytest.raises(module.PolicyError, match="resolution limit"):
+        module.unpack_objects(io.BytesIO(pack))
+    monkeypatch.setattr(module, "MAX_RESOLVE_PASSES", 64)
+    result = _read_all(module.unpack_objects(io.BytesIO(pack)))
+    assert [payload for _code, payload in result] == [b"two!", base, b"one!"]
+
+    monkeypatch.setattr(module, "MAX_TOTAL_BYTES", len(base) + len(first) + len(second) + 1)
+    with pytest.raises(module.PolicyError, match="contents exceed"):
+        module.unpack_objects(io.BytesIO(pack))
+
+
+def test_parse_loose_object_round_trip_and_bounds(monkeypatch):
+    module = _load("git_pack_objects")
+    payload = b"loose blob"
+    value = zlib.compress(b"blob %d\x00" % len(payload) + payload)
+    assert module.parse_loose_object(value) == (3, payload)
+
+    with pytest.raises(module.PolicyError, match="corrupt"):
+        module.parse_loose_object(b"\x00junk")
+    with pytest.raises(module.PolicyError, match="framing is invalid"):
+        module.parse_loose_object(value + b"trailing")
+    with pytest.raises(module.PolicyError, match="framing is invalid"):
+        module.parse_loose_object(value[:-4])
+    for header in (
+        b"noodle 3\x00abc",
+        b"blob3\x00abc",
+        b"blob x\x00abc",
+        b"blob 4\x00abc",
+        b"blob 3",
+    ):
+        with pytest.raises(module.PolicyError, match="header is invalid"):
+            module.parse_loose_object(zlib.compress(header))
+    monkeypatch.setattr(module, "MAX_OBJECT_BYTES", 4)
+    with pytest.raises(module.PolicyError, match="exceeds the safe size limit"):
+        module.parse_loose_object(b"12345")

testing/tests/test_hermes_receive_pack_scan.py

@@ -0,0 +1,217 @@
+"""Quarantine content-scanning contracts for brokered Git pushes."""
+
+from __future__ import annotations
+
+import hashlib
+import io
+import zlib
+
+import pytest
+
+from testing.tests.test_hermes_scm_broker_support import (
+    _blob_pack,
+    _load,
+    _object_entry,
+    _pack_of,
+    _receive_command,
+)
+
+ZERO = b"0" * 40
+COMMIT = b"1" * 40
+TOKEN = "runtime-sentinel"
+FORMS = (b"runtime-sentinel",)
+
+
+def _request(pack: bytes, ref: bytes = b"refs/heads/hermes/scan") -> bytes:
+    return _receive_command(ZERO, COMMIT, ref, pack)
+
+
+def test_clean_push_with_real_content_passes():
+    scan = _load("receive_pack_scan")
+    pack = _blob_pack(
+        b"# services/example/deployment.yaml\napiVersion: apps/v1\n",
+        b"tree 5b3a\nparent none\n",
+    )
+    scan.validate_receive_pack(_request(pack), TOKEN, FORMS)
+
+
+@pytest.mark.parametrize(
+    "payload",
+    [
+        b"-----BEGIN OPENSSH PRIVATE KEY-----\nb3BlbnNzaA==\n",
+        b"-----BEGIN RSA PRIVATE KEY-----\nMIIE\n",
+        b"key = ghp_" + b"a1" * 12,
+        b"aws_access_key_id = AKIAABCDEFGHIJKLMNOP",
+        b"jwt eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiIxIn0.c2lnbmF0dXJlLXNlZ21lbnQ",
+        b"hook https://hooks.slack.com/services/T000/B000/tokentokentokentoken",
+        b"AGE-SECRET-KEY-1QQPCXR7QQPCXR7QQPCXR7QQPCXR7",
+        b"PuTTY-User-Key-File-3: ssh-ed25519",
+        b"Authorization: Bearer abc123-tokenvalue",
+        b"ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIF2c hermes@atlas",
+        b"gitea gta_" + b"z9" * 12,
+    ],
+)
+def test_credential_shaped_object_content_is_rejected(payload):
+    scan = _load("receive_pack_scan")
+    with pytest.raises(scan.PolicyError, match="credential-shaped"):
+        scan.validate_receive_pack(_request(_blob_pack(payload)), TOKEN, FORMS)
+
+
+def test_runtime_token_hidden_by_compression_is_still_caught():
+    scan = _load("receive_pack_scan")
+    pack = _blob_pack(b"config value: runtime-sentinel\n")
+    # The compressed wire bytes do not contain the token; only the
+    # decompressed payload does.
+    assert b"runtime-sentinel" not in pack
+    with pytest.raises(scan.PolicyError, match="runtime credential material"):
+        scan.validate_receive_pack(_request(pack), TOKEN, FORMS)
+
+
+def _size_varint(value: int) -> bytes:
+    out = bytearray()
+    while True:
+        byte = value & 0x7F
+        value >>= 7
+        if value:
+            out.append(byte | 0x80)
+        else:
+            out.append(byte)
+            return bytes(out)
+
+
+def _typed_header(type_code: int, size: int) -> bytes:
+    byte = (type_code << 4) | (size & 0x0F)
+    size >>= 4
+    header = bytearray()
+    while size:
+        header.append(byte | 0x80)
+        byte = size & 0x7F
+        size >>= 7
+    header.append(byte)
+    return bytes(header)
+
+
+def test_secret_assembled_only_by_delta_resolution_is_caught():
+    scan = _load("receive_pack_scan")
+    base = b"docs: AKIAABCD"
+    completion = b"EFGHIJKLMNOP"
+    delta = (
+        _size_varint(len(base))
+        + _size_varint(len(base) + len(completion))
+        + bytes([0x80 | 0x10, len(base)])
+        + bytes([len(completion)])
+        + completion
+    )
+    base_sha = hashlib.sha1(b"blob %d\x00" % len(base) + base).digest()
+    delta_entry = _typed_header(7, len(delta)) + base_sha + zlib.compress(delta)
+    pack = _pack_of([_object_entry(3, base), delta_entry])
+    # Neither the base nor the delta insert alone matches; the resolved
+    # payload is the only place the credential exists.
+    with pytest.raises(scan.PolicyError, match="runtime credential|credential-shaped"):
+        scan.validate_receive_pack(_request(pack), TOKEN, FORMS)
+
+
+def test_thin_pack_deltas_are_rejected_outright():
+    scan = _load("receive_pack_scan")
+    delta = _size_varint(4) + _size_varint(1) + bytes([1]) + b"x"
+    entry = _typed_header(7, len(delta)) + b"\xbb" * 20 + zlib.compress(delta)
+    with pytest.raises(scan.PolicyError, match="push full packs"):
+        scan.validate_receive_pack(_request(_pack_of([entry])), TOKEN, FORMS)
+
+
+def test_trailing_bytes_after_the_pack_are_rejected():
+    scan = _load("receive_pack_scan")
+    body = _request(_pack_of([])) + b"extra"
+    with pytest.raises(scan.PolicyError, match="bytes after its pack"):
+        scan.validate_receive_pack(body, TOKEN, FORMS)
+
+
+@pytest.mark.parametrize(
+    ("line", "match"),
+    [
+        (b"shallow " + COMMIT, "ref command is invalid"),
+        (COMMIT + b" " + COMMIT + b" refs/heads/hermes/x", "feature-branch creation"),
+        (ZERO + b" " + ZERO + b" refs/heads/hermes/x", "feature-branch creation"),
+        (ZERO + b" " + COMMIT + b" refs/heads/main", "namespaced feature branches"),
+        (ZERO + b" " + COMMIT + b" refs/tags/v1", "namespaced feature branches"),
+        (ZERO + b" " + COMMIT + b" refs/heads/hermes/\xff", "canonical ASCII"),
+        (b"zz" * 20 + b" " + COMMIT + b" refs/heads/hermes/x", "ref command is invalid"),
+    ],
+)
+def test_push_commands_outside_the_narrow_grant_are_rejected(line, match):
+    scan = _load("receive_pack_scan")
+    framed = f"{len(line) + 5:04x}".encode() + line + b"\n" + b"0000"
+    with pytest.raises(scan.PolicyError, match=match):
+        scan.validate_receive_pack(framed + _pack_of([]), TOKEN, FORMS)
+
+
+def test_command_stream_bounds_are_enforced():
+    scan = _load("receive_pack_scan")
+    with pytest.raises(scan.PolicyError, match="no ref command"):
+        scan.validate_receive_pack(b"0000" + _pack_of([]), TOKEN, FORMS)
+    with pytest.raises(scan.PolicyError, match="omitted its command terminator"):
+        scan.validate_receive_pack(b"", TOKEN, FORMS)
+    line = ZERO + b" " + COMMIT + b" refs/heads/hermes/x\n"
+    framed = f"{len(line) + 4:04x}".encode() + line
+    many = framed * (scan.MAX_PUSH_COMMANDS + 1) + b"0000" + _pack_of([])
+    with pytest.raises(scan.PolicyError, match="too many ref commands"):
+        scan.validate_receive_pack(many, TOKEN, FORMS)
+    with pytest.raises(scan.PolicyError, match="framing is invalid"):
+        scan.validate_receive_pack(b"zzzz", TOKEN, FORMS)
+    with pytest.raises(scan.PolicyError, match="framing is invalid"):
+        scan.validate_receive_pack(b"0003", TOKEN, FORMS)
+    with pytest.raises(scan.PolicyError, match="framing is invalid"):
+        scan.validate_receive_pack(b"0009abc", TOKEN, FORMS)
+
+
+def test_credential_forms_in_command_lines_are_rejected():
+    scan = _load("receive_pack_scan")
+    line = ZERO + b" " + COMMIT + b" refs/heads/hermes/x runtime-sentinel\n"
+    framed = f"{len(line) + 4:04x}".encode() + line + b"0000"
+    with pytest.raises(scan.PolicyError, match="runtime credential material"):
+        scan.validate_receive_pack(framed + _pack_of([]), TOKEN, FORMS)
+
+
+def test_scan_crosses_chunk_boundaries_with_carry(monkeypatch):
+    scan = _load("receive_pack_scan")
+    monkeypatch.setattr(scan, "SCAN_CHUNK", 8)
+    monkeypatch.setattr(scan, "SCAN_CARRY", 48)
+    payload = b"x" * 20 + b" AKIAABCDEFGHIJKLMNOP " + b"y" * 20
+    with pytest.raises(scan.PolicyError, match="credential-shaped"):
+        scan._scan_payload(io.BytesIO(payload), FORMS)
+    scan._scan_payload(io.BytesIO(b"clean " * 40), FORMS)
+    scan._scan_payload(io.BytesIO(b""), FORMS)
+
+
+def test_scan_loose_object_reports_type_and_scans_content():
+    scan = _load("receive_pack_scan")
+    clean = zlib.compress(b"blob 5\x00clean")
+    assert scan.scan_loose_object(clean, FORMS) == 3
+    secret = b"-----BEGIN EC PRIVATE KEY-----"
+    dirty = zlib.compress(b"blob %d\x00" % len(secret) + secret)
+    with pytest.raises(scan.PolicyError, match="credential-shaped"):
+        scan.scan_loose_object(dirty, FORMS)
+    leaked = b"the runtime-sentinel value"
+    with pytest.raises(scan.PolicyError, match="runtime credential material"):
+        scan.scan_loose_object(
+            zlib.compress(b"blob %d\x00" % len(leaked) + leaked), FORMS
+        )
+
+
+def test_content_patterns_extend_the_shared_standalone_corpus():
+    scan = _load("receive_pack_scan")
+    policy = _load("gitea_api_policy")
+    assert len(scan.CONTENT_SECRET_PATTERNS) == len(policy.STANDALONE_SECRET_PATTERNS) + 3
+    assert all(
+        isinstance(pattern.pattern, bytes) for pattern in scan.CONTENT_SECRET_PATTERNS
+    )
+
+
+def test_validation_rewinds_the_stream_for_upstream_forwarding():
+    scan = _load("receive_pack_scan")
+    body = _request(_blob_pack(b"safe content"))
+    stream = io.BytesIO(body)
+    scan.validate_receive_pack(stream, TOKEN, FORMS)
+    assert stream.tell() == 0
+    with pytest.raises(scan.PolicyError):
+        scan.validate_receive_pack(io.BytesIO(b"zzzz"), TOKEN, FORMS)

testing/tests/test_hermes_scm_broker_internal_coverage.py

@@ -7,7 +7,11 @@ from email.message import Message
 
 import pytest
 
-from testing.tests.test_hermes_scm_broker_support import Response, _load
+from testing.tests.test_hermes_scm_broker_support import (
+    Response,
+    _load,
+    _receive_command,
+)
 
 
 def test_broker_redirect_status_and_bounded_read_helpers(monkeypatch):
@@ -139,18 +143,22 @@ def test_content_length_and_credential_helpers_cover_safe_and_rejected_values():
     assert broker._content_length(headers, 1) == 0
     forms = broker._credential_forms("sentinel")
     assert len(forms) == 3 and forms[0] == b"sentinel"
-    broker._reject_credential_bytes(b"safe", "sentinel", "response")
+    zero = b"0" * 40
+    commit = b"1" * 40
+    line = zero + b" " + commit + b" refs/heads/hermes/x " + forms[2] + b"\n"
+    framed = f"{len(line) + 4:04x}".encode() + line + b"0000"
     with pytest.raises(broker.PolicyError, match="credential material"):
-        broker._reject_credential_bytes(b"prefix " + forms[2], "sentinel", "response")
+        broker._validate_receive_pack(framed, "sentinel")
 
 
-def test_receive_prefix_preserves_stream_position_and_accepts_bytes():
+def test_receive_pack_validation_rewinds_stream_and_accepts_bytes():
     broker = _load("scm_broker")
-    assert broker._receive_prefix(b"bytes") == b"bytes"
-    stream = io.BytesIO(b"prefix")
-    stream.seek(3)
-    assert broker._receive_prefix(stream) == b"prefix"
-    assert stream.tell() == 3
+    body = _receive_command(b"0" * 40, b"1" * 40, b"refs/heads/hermes/rewind")
+    stream = io.BytesIO(body)
+    stream.seek(9)
+    broker._validate_receive_pack(stream, "sentinel")
+    assert stream.tell() == 0
+    broker._validate_receive_pack(body, "sentinel")
 
 
 @pytest.mark.parametrize(
@@ -188,7 +196,8 @@ def test_receive_pack_rejects_nonascii_many_commands_and_missing_terminator(
     command = _packet(zero, commit, b"refs/heads/hermes/fix", terminator=False)
     with pytest.raises(broker.PolicyError, match="terminator"):
         broker._validate_receive_pack(command, "sentinel")
-    many = command * (broker.MAX_PUSH_COMMANDS + 1) + b"0000"
+    limit = _load("receive_pack_scan").MAX_PUSH_COMMANDS
+    many = command * (limit + 1) + b"0000"
     with pytest.raises(broker.PolicyError, match="too many"):
         broker._validate_receive_pack(many, "sentinel")
 

testing/tests/test_hermes_scm_broker_streaming.py

@@ -130,7 +130,7 @@ def test_upstream_stream_applies_absolute_socket_deadline():
     sock = DeadlineSocket()
     response.fp = type("FP", (), {"raw": type("Raw", (), {"_sock": sock})()})()
     spool, length = broker._spool_response(
-        response, broker.MAX_GIT_RESPONSE, "sentinel"
+        response, broker.MAX_GIT_RESPONSE, "sentinel", broker.INBOUND_BODY_TIMEOUT
     )
     try:
         assert length == 4

testing/tests/test_hermes_scm_broker_support.py

@@ -2,9 +2,11 @@
 
 from __future__ import annotations
 
+import hashlib
 import importlib.util
 import io
 import sys
+import zlib
 from email.message import Message
 from pathlib import Path
 
@@ -50,6 +52,33 @@ class Response:
         return self.stream.read(limit)
 
 
-def _receive_command(old: bytes, new: bytes, ref: bytes) -> bytes:
+def _object_entry(type_code: int, payload: bytes) -> bytes:
+    """Encode one non-delta pack entry with a real header and zlib body."""
+    size = len(payload)
+    byte = (type_code << 4) | (size & 0x0F)
+    size >>= 4
+    header = bytearray()
+    while size:
+        header.append(byte | 0x80)
+        byte = size & 0x7F
+        size >>= 7
+    header.append(byte)
+    return bytes(header) + zlib.compress(payload)
+
+
+def _pack_of(entries: list[bytes]) -> bytes:
+    """Assemble encoded entries into a checksummed version-2 pack."""
+    body = b"PACK" + (2).to_bytes(4, "big") + len(entries).to_bytes(4, "big")
+    for entry in entries:
+        body += entry
+    return body + hashlib.sha1(body).digest()
+
+
+def _blob_pack(*payloads: bytes) -> bytes:
+    return _pack_of([_object_entry(3, payload) for payload in payloads])
+
+
+def _receive_command(old: bytes, new: bytes, ref: bytes, pack: bytes = b"") -> bytes:
     command = old + b" " + new + b" " + ref + b"\x00report-status\n"
-    return f"{len(command) + 4:04x}".encode() + command + b"0000PACK"
+    framed = f"{len(command) + 4:04x}".encode() + command + b"0000"
+    return framed + (pack or _pack_of([]))