diff --git a/services/hermes/agent-configmap.yaml b/services/hermes/agent-configmap.yaml index b53b2c92..28d520fb 100644 --- a/services/hermes/agent-configmap.yaml +++ b/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 diff --git a/services/hermes/scm-common/kustomization.yaml b/services/hermes/scm-common/kustomization.yaml index 17dc1da6..a4756ccd 100644 --- a/services/hermes/scm-common/kustomization.yaml +++ b/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 diff --git a/services/hermes/scm-common/scripts/git_pack_objects.py b/services/hermes/scm-common/scripts/git_pack_objects.py new file mode 100644 index 00000000..d9a3e715 --- /dev/null +++ b/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 diff --git a/services/hermes/scm-common/scripts/receive_pack_scan.py b/services/hermes/scm-common/scripts/receive_pack_scan.py new file mode 100644 index 00000000..76980880 --- /dev/null +++ b/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)(? 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 diff --git a/services/hermes/scm-common/scripts/scm_broker.py b/services/hermes/scm-common/scripts/scm_broker.py index a0c4043c..a9cb3685 100644 --- a/services/hermes/scm-common/scripts/scm_broker.py +++ b/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[A-Za-z0-9][A-Za-z0-9._-]{0,99})\.git/" r"(?Pinfo/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: diff --git a/testing/tests/test_hermes_deadline_stream.py b/testing/tests/test_hermes_deadline_stream.py new file mode 100644 index 00000000..d4f97efe --- /dev/null +++ b/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() diff --git a/testing/tests/test_hermes_git_pack_objects.py b/testing/tests/test_hermes_git_pack_objects.py new file mode 100644 index 00000000..f3a249ec --- /dev/null +++ b/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") \ No newline at end of file diff --git a/testing/tests/test_hermes_receive_pack_scan.py b/testing/tests/test_hermes_receive_pack_scan.py new file mode 100644 index 00000000..7d17fa6f --- /dev/null +++ b/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) \ No newline at end of file diff --git a/testing/tests/test_hermes_scm_broker_internal_coverage.py b/testing/tests/test_hermes_scm_broker_internal_coverage.py index abab8406..b38702ac 100644 --- a/testing/tests/test_hermes_scm_broker_internal_coverage.py +++ b/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") diff --git a/testing/tests/test_hermes_scm_broker_streaming.py b/testing/tests/test_hermes_scm_broker_streaming.py index 2fa9683f..78a63f27 100644 --- a/testing/tests/test_hermes_scm_broker_streaming.py +++ b/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 diff --git a/testing/tests/test_hermes_scm_broker_support.py b/testing/tests/test_hermes_scm_broker_support.py index 6e2184d9..e3682494 100644 --- a/testing/tests/test_hermes_scm_broker_support.py +++ b/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([]))