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

#!/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