hermes: harden draft PR input boundary

services/hermes/scripts/gitea_api.py

@@ -22,6 +22,7 @@ from gitea_api_policy import (
     _validate_pr_number_segment,
     _validate_query,
     _validate_ref,
+    _validate_ref_bounds,
     _validate_repo,
     _validate_sha,
 )
@@ -174,11 +175,13 @@ def authorize_request(method: str, path: str, data: object | None) -> str:
     _validate_repo(match.group(1))
     if not isinstance(data, dict) or set(data) != {"base", "body", "head", "title"}:
         raise PolicyError("draft creation accepts only base, body, head, and title")
-    _validate_ref(data["base"], "base")
-    _validate_ref(data["head"], "head")
     if data["title"] != _draft_title(data["title"]):
         raise PolicyError("new pull requests must use the Gitea draft-title prefix")
     _validate_body(data["body"])
+    _validate_ref_bounds(data["base"], "base")
+    _validate_ref_bounds(data["head"], "head")
+    _validate_ref(data["base"], "base")
+    _validate_ref(data["head"], "head")
     return "create-draft"
 
 
@@ -308,11 +311,13 @@ def create_draft(
 ) -> bytes:
     """Create and verify a same-repository draft pull request for human review."""
     repo = _validate_repo(repo)
-    base = _validate_ref(base, "base")
-    head = _validate_ref(head, "head")
-    head_sha = _validate_sha(head_sha)
     title = _draft_title(title, forbidden=(token,))
     body = _validate_body(body, forbidden=(token,))
+    head_sha = _validate_sha(head_sha)
+    _validate_ref_bounds(base, "base")
+    _validate_ref_bounds(head, "head")
+    base = _validate_ref(base, "base")
+    head = _validate_ref(head, "head")
     data = {"base": base, "body": body, "head": head, "title": title}
     result = _request(
         "POST",
@@ -361,11 +366,15 @@ def _operation(args: argparse.Namespace) -> tuple[str, str, object | None]:
         return "GET", args.path, None
     repo = _validate_repo(args.repo)
     _validate_sha(args.head_sha)
+    body = _validate_body(args.body)
+    title = _draft_title(args.title)
+    _validate_ref_bounds(args.base, "base")
+    _validate_ref_bounds(args.head, "head")
     data = {
         "base": _validate_ref(args.base, "base"),
-        "body": _validate_body(args.body),
+        "body": body,
         "head": _validate_ref(args.head, "head"),
-        "title": _draft_title(args.title),
+        "title": title,
     }
     return "POST", f"/api/v1/repos/{ALLOWED_OWNER}/{repo}/pulls", data
 

services/hermes/scripts/gitea_api_policy.py

@@ -4,10 +4,15 @@ The screening catches structured credential assignments, known token formats,
 and long high-entropy values. It is a fail-closed accident barrier, not proof
 that text is secret-free: a short or multiword secret under an innocuous key is
 not reliably distinguishable from prose and must never be supplied by callers.
+JSON depth/nodes/documents and multiline gaps are deliberately bounded; large
+or unusual configuration blobs are not a supported pull-request field format.
 """
 
+# ruff: noqa: SIM905
+
 from __future__ import annotations
 
+import json
 import re
 import subprocess
 import urllib.parse
@@ -26,6 +31,13 @@ MAX_PAGE = 10_000
 MAX_LIMIT = 50
 MAX_PR_NUMBER = 2_147_483_647
 MAX_PR_NUMBER_DIGITS = 10
+MAX_REF_CHARACTERS = 200
+MAX_REF_UTF8_BYTES = 255
+MAX_TITLE_UTF8_BYTES = 512
+MAX_BODY_UTF8_BYTES = 32_768
+MAX_JSON_DOCUMENTS = 32
+MAX_JSON_NODES = 2_048
+MAX_JSON_DEPTH = 32
 CANONICAL_QUERY_RE = re.compile(r"[A-Za-z0-9_=&-]*\Z")
 ASSIGNMENT_RE = re.compile(
     r"""(?mx)
@@ -37,10 +49,21 @@ ASSIGNMENT_RE = re.compile(
     (?P<value>
       \"(?:\\.|[^\"\\\r\n])*\" |
       '(?:\\.|[^'\\\r\n])*' |
-      [^\r\n,;}&]*
+      [^\r\n,;}&]{0,2048}
     )
     """
 )
+JSON_KEY_RE = re.compile(r'"(?P<key>[^"]{1,512})"(?P<spacing>[\x00-\x20\x7f]{0,256}):')
+MULTILINE_ASSIGNMENT_RE = re.compile(
+    r"""(?mx)
+    (?<![A-Za-z0-9_.-])
+    (?P<key>\.?[A-Za-z][A-Za-z0-9_.-]{0,127})
+    [ \t]{0,64}(?:\r?\n[ \t]{0,64}){0,4}
+    (?P<separator>[:=])
+    [ \t]{0,64}(?:\r?\n[ \t]{0,64}){1,4}
+    (?P<value>[^\r\n,;}&]{1,2048})
+    """
+)
 CAMEL_BOUNDARY_RE = re.compile(r"(?<=[a-z0-9])(?=[A-Z])|(?<=[A-Z])(?=[A-Z][a-z])")
 CREDENTIAL_SCHEME_RE = re.compile(
     r"(?i)\b(?:bearer|basic|token)\s+"
@@ -76,107 +99,30 @@ STANDALONE_SECRET_PATTERNS = tuple(
     )
 )
 HIGH_ENTROPY_TOKEN_RE = re.compile(
-    r"(?<![A-Za-z0-9+/_=-])[A-Za-z0-9+/_=-]{48,}(?![A-Za-z0-9+/_=-])"
+    r"(?<![A-Za-z0-9+/_=-])[A-Za-z0-9+/_=-]{48,16384}(?![A-Za-z0-9+/_=-])"
 )
-PROSE_LEAD_WORDS = {
-    "add",
-    "allow",
-    "change",
-    "check",
-    "describe",
-    "document",
-    "ensure",
-    "explain",
-    "fix",
-    "handle",
-    "keep",
-    "preserve",
-    "prevent",
-    "reject",
-    "remove",
-    "review",
-    "support",
-    "test",
-    "validate",
-}
-SENSITIVE_KEY_WORDS = {
-    "auth",
-    "authentication",
-    "authorization",
-    "auths",
-    "basic",
-    "bearer",
-    "credential",
-    "credentials",
-    "passwd",
-    "password",
-    "pat",
-    "private",
-    "sas",
-    "secret",
-    "secrets",
-    "sig",
-    "signature",
-    "signing",
-    "token",
-    "tokens",
-    "webhook",
-}
-KEY_MODIFIER_WORDS = {
-    "access",
-    "account",
-    "api",
-    "auth",
-    "client",
-    "encryption",
-    "identity",
-    "private",
-    "registry",
-    "secret",
-    "service",
-    "session",
-    "signing",
-    "ssh",
-}
-SENSITIVE_COMPACT_KEYS = {
-    "accessid",
-    "accesskeyid",
-    "accountkey",
-    "clientemail",
-    "clientid",
-    "connectionstring",
-    "dockerconfigjson",
-    "privatekeyid",
-    "serviceaccountkey",
-    "sharedaccesssignature",
-}
-SENSITIVE_COMPACT_SUFFIXES = (
-    "accesskey",
-    "accesskeyid",
-    "accountkey",
-    "apikey",
-    "authkey",
-    "clientemail",
-    "clientid",
-    "clientkey",
-    "connectionstring",
-    "credential",
-    "credentials",
-    "encryptionkey",
-    "identitykey",
-    "password",
-    "passwd",
-    "privatekeyid",
-    "privatekey",
-    "secret",
-    "secretkey",
-    "servicekey",
-    "sessionkey",
-    "signature",
-    "signingkey",
-    "sshkey",
-    "token",
-    "webhook",
+PROSE_LEAD_WORDS = frozenset(
+    "add allow change check describe document ensure explain fix handle keep preserve "
+    "prevent reject remove review support test validate".split()
+)
+SENSITIVE_KEY_WORDS = frozenset(
+    "auth authentication authorization auths basic bearer credential credentials "
+    "passwd password pat private sas secret secrets sig signature signing token tokens "
+    "webhook".split()
+)
+KEY_MODIFIER_WORDS = frozenset(
+    "access account api auth client encryption identity private registry secret service "
+    "session signing ssh".split()
+)
+SENSITIVE_COMPACT_KEYS = frozenset(
+    "accessid accesskeyid accountkey clientemail clientid connectionstring "
+    "dockerconfigjson privatekeyid serviceaccountkey sharedaccesssignature".split()
+)
+SENSITIVE_COMPACT_SUFFIXES = tuple(
+    "accesskey accesskeyid accountkey apikey authkey clientemail clientid clientkey "
+    "connectionstring credential credentials encryptionkey identitykey password passwd "
+    "privatekeyid privatekey secret secretkey servicekey sessionkey signature signingkey "
+    "sshkey token webhook".split()
 )
 
 
@@ -190,9 +136,23 @@ def _validate_repo(repo: str) -> str:
     return repo
 
 
+def _validate_ref_bounds(value: object, name: str) -> str:
+    """Reject oversized or non-encodable refs before invoking Git."""
+    if not isinstance(value, str) or not value or len(value) > MAX_REF_CHARACTERS:
+        raise PolicyError(f"{name} exceeds the safe branch-name limit")
+    try:
+        encoded = value.encode("utf-8", errors="strict")
+    except UnicodeEncodeError as exc:
+        raise PolicyError(f"{name} is not valid UTF-8 text") from exc
+    if len(encoded) > MAX_REF_UTF8_BYTES:
+        raise PolicyError(f"{name} exceeds the safe UTF-8 branch-name limit")
+    return value
+
+
 def _validate_ref(value: object, name: str) -> str:
     """Validate the complete Git ref grammar using Git itself."""
-    if not isinstance(value, str) or not value or value.startswith("-"):
+    value = _validate_ref_bounds(value, name)
+    if value.startswith("-"):
         raise PolicyError(f"{name} must be a same-repository branch name")
     if any(ord(character) < 32 or ord(character) == 127 for character in value):
         raise PolicyError(f"{name} must be a safe same-repository branch name")
@@ -239,12 +199,23 @@ def _validate_pr_number(value: object) -> int:
     return value
 
 
-def _validate_text(value: object, name: str, maximum: int, *, required: bool) -> str:
+def _validate_text(
+    value: object,
+    name: str,
+    maximum: int,
+    maximum_bytes: int,
+    *,
+    required: bool,
+) -> str:
     if not isinstance(value, str):
         raise PolicyError(f"{name} must be text")
     if required and not value.strip():
         raise PolicyError(f"{name} must not be empty")
-    if len(value) > maximum or "\x00" in value:
+    try:
+        encoded = value.encode("utf-8", errors="strict")
+    except UnicodeEncodeError as exc:
+        raise PolicyError(f"{name} is not valid UTF-8 text") from exc
+    if len(value) > maximum or len(encoded) > maximum_bytes or "\x00" in value:
         raise PolicyError(f"{name} exceeds the safe request limit")
     return value
 
@@ -319,6 +290,89 @@ def _is_structural_credential_assignment(key: str, value: str) -> bool:
     ) or ("service" in words and "account" in words and bool(stripped))
 
 
+def _decode_json_key(raw: str) -> tuple[str, bool]:
+    """Decode one bounded JSON key without silently accepting bad escapes."""
+    try:
+        decoded = json.loads(f'"{raw}"')
+    except (json.JSONDecodeError, UnicodeError):
+        approximate = re.sub(r"[^A-Za-z0-9]+", "_", raw.replace("\\", ""))
+        return approximate, False
+    return decoded if isinstance(decoded, str) else "", True
+
+
+def _json_value_has_sensitive_assignment(
+    value: object, *, depth: int = 0, nodes: list[int] | None = None
+) -> bool:
+    """Walk one decoded JSON value with explicit depth and node ceilings."""
+    if nodes is None:
+        nodes = [0]
+    nodes[0] += 1
+    if nodes[0] > MAX_JSON_NODES or depth > MAX_JSON_DEPTH:
+        raise PolicyError("structured pull-request text exceeds the scan limit")
+    if isinstance(value, dict):
+        for key, assigned in value.items():
+            if not isinstance(key, str):
+                continue
+            if _is_sensitive_key(key):
+                return True
+            assigned_text = assigned if isinstance(assigned, str) else ""
+            if _is_structural_credential_assignment(key, assigned_text):
+                return True
+            if _json_value_has_sensitive_assignment(
+                assigned, depth=depth + 1, nodes=nodes
+            ):
+                return True
+    elif isinstance(value, list):
+        return any(
+            _json_value_has_sensitive_assignment(item, depth=depth + 1, nodes=nodes)
+            for item in value
+        )
+    return False
+
+
+def _decoded_json_documents(value: str):
+    """Yield bounded embedded JSON object or array fragments."""
+    decoder = json.JSONDecoder()
+    index = 0
+    attempts = 0
+    while attempts < MAX_JSON_DOCUMENTS:
+        positions = [
+            position for token in "{[" if (position := value.find(token, index)) >= 0
+        ]
+        if not positions:
+            return
+        start = min(positions)
+        attempts += 1
+        try:
+            document, end = decoder.raw_decode(value, start)
+        except (json.JSONDecodeError, RecursionError, ValueError):
+            index = start + 1
+            continue
+        yield document
+        index = max(end, start + 1)
+
+
+def _has_structured_sensitive_assignment(value: str) -> bool:
+    """Detect bounded JSON and multiline YAML/env credential assignments."""
+    for candidate in JSON_KEY_RE.finditer(value):
+        decoded, valid = _decode_json_key(candidate.group("key"))
+        if _is_sensitive_key(decoded):
+            return True
+        if not valid and _is_sensitive_key(candidate.group("key").replace("\\", "")):
+            return True
+    for candidate in MULTILINE_ASSIGNMENT_RE.finditer(value):
+        key = candidate.group("key")
+        assigned = candidate.group("value")
+        if _is_structural_credential_assignment(key, assigned) or (
+            _is_sensitive_key(key) and _looks_sensitive_assignment_value(assigned)
+        ):
+            return True
+    return any(
+        _json_value_has_sensitive_assignment(item)
+        for item in _decoded_json_documents(value)
+    )
+
+
 def _has_high_entropy_token(value: str) -> bool:
     for match in HIGH_ENTROPY_TOKEN_RE.finditer(value):
         token = match.group(0)
@@ -344,6 +398,8 @@ def _reject_sensitive(value: str, name: str, forbidden: tuple[str, ...]) -> None
         raise PolicyError(f"pull-request {name} resembles credential material")
     if any(pattern.search(value) for pattern in STANDALONE_SECRET_PATTERNS):
         raise PolicyError(f"pull-request {name} resembles credential material")
+    if _has_structured_sensitive_assignment(value):
+        raise PolicyError(f"pull-request {name} resembles credential material")
     for candidate in ASSIGNMENT_RE.finditer(value):
         key = candidate.group("key")
         assigned = candidate.group("value")
@@ -356,14 +412,16 @@ def _reject_sensitive(value: str, name: str, forbidden: tuple[str, ...]) -> None
 
 
 def _validate_body(value: object, *, forbidden: tuple[str, ...] = ()) -> str:
-    body = _validate_text(value, "body", 16384, required=False)
+    body = _validate_text(value, "body", 16384, MAX_BODY_UTF8_BYTES, required=False)
     _reject_sensitive(body, "body", forbidden)
     return body
 
 
 def _draft_title(value: object, *, forbidden: tuple[str, ...] = ()) -> str:
     """Return a bounded, secret-screened Gitea draft title."""
-    title = _validate_text(value, "title", 251, required=True).strip()
+    title = _validate_text(
+        value, "title", 251, MAX_TITLE_UTF8_BYTES, required=True
+    ).strip()
     _reject_sensitive(title, "title", forbidden)
     for prefix in ("WIP:", "[WIP]"):
         if title.upper().startswith(prefix):

services/hermes/skills/manage-atlas-pull-requests/SKILL.md

@@ -32,8 +32,10 @@ SHA explicitly. Keep the short body free of credentials and credential-like
 text. The client screens structured assignments, known token formats, and long
 high-entropy values as an accidental-disclosure barrier. It cannot prove text
 is secret-free: short or multiword secrets under innocuous names may resemble
-ordinary prose, so never place credential material in either field. Validate
-without reading a credential or using the network:
+ordinary prose, so never place credential material in either field. Structured
+inspection is deliberately bounded; large, deeply nested, or unusual config
+blobs do not belong in these fields and may be rejected. Validate without
+reading a credential or using the network:
 
 ```sh
 /opt/coordinator/gitea_api.py --dry-run create-draft REPO \

testing/tests/test_hermes_gitea_pr_client.py

@@ -409,6 +409,104 @@ def test_git_ref_validation_uses_fixed_trusted_binary():
     assert client._validate_ref("hermes/valid-fix", "head") == "hermes/valid-fix"
 
 
+@pytest.mark.parametrize("field", ["base", "head"])
+@pytest.mark.parametrize("oversized", ["r" * 100_000, "🧪" * 128])
+def test_oversized_ref_never_invokes_git_request_or_opener(
+    field: str, oversized: str, monkeypatch
+):
+    client = _load()
+    git_called = False
+    request_built = False
+    opener_called = False
+    original_build_request = client.build_request
+
+    def git_run(*_args, **_kwargs):
+        nonlocal git_called
+        git_called = True
+        raise AssertionError("Git must not receive an oversized ref")
+
+    def build_request(*args, **kwargs):
+        nonlocal request_built
+        request_built = True
+        return original_build_request(*args, **kwargs)
+
+    def opener(*_args, **_kwargs):
+        nonlocal opener_called
+        opener_called = True
+        return Response(_draft_response())
+
+    monkeypatch.setattr(client._validate_ref.__globals__["subprocess"], "run", git_run)
+    client.build_request = build_request
+    refs = {"base": "main", "head": "hermes/fix"}
+    refs[field] = oversized
+    with pytest.raises(client.PolicyError, match="branch-name limit"):
+        client.create_draft(
+            "cassandra",
+            base=refs["base"],
+            head=refs["head"],
+            head_sha=HEAD_SHA,
+            title="Focused fix",
+            body="Review evidence",
+            token="runtime",
+            opener=opener,
+        )
+    assert git_called is False
+    assert request_built is False
+    assert opener_called is False
+
+
+@pytest.mark.parametrize(
+    ("field", "oversized"),
+    [("repo", "r" * 101), ("title", "🧪" * 200), ("body", "🧪" * 9_000)],
+)
+def test_other_text_bounds_fail_before_git_request_or_opener(
+    field: str, oversized: str, monkeypatch
+):
+    client = _load()
+    git_called = False
+    request_built = False
+    opener_called = False
+    original_build_request = client.build_request
+
+    def git_run(*_args, **_kwargs):
+        nonlocal git_called
+        git_called = True
+        raise AssertionError("Git must not run before cheap input bounds")
+
+    def build_request(*args, **kwargs):
+        nonlocal request_built
+        request_built = True
+        return original_build_request(*args, **kwargs)
+
+    def opener(*_args, **_kwargs):
+        nonlocal opener_called
+        opener_called = True
+        return Response(_draft_response())
+
+    monkeypatch.setattr(client._validate_ref.__globals__["subprocess"], "run", git_run)
+    client.build_request = build_request
+    values = {
+        "repo": "cassandra",
+        "title": "Focused fix",
+        "body": "Review evidence",
+    }
+    values[field] = oversized
+    with pytest.raises(client.PolicyError):
+        client.create_draft(
+            values["repo"],
+            base="main",
+            head="hermes/fix",
+            head_sha=HEAD_SHA,
+            title=values["title"],
+            body=values["body"],
+            token="runtime",
+            opener=opener,
+        )
+    assert git_called is False
+    assert request_built is False
+    assert opener_called is False
+
+
 def test_create_forces_draft_title_and_same_repository_branch_names():
     client = _load()
     assert (
@@ -450,6 +548,13 @@ def test_runtime_token_is_only_an_authorization_header():
     "sensitive",
     [
         "client_" + "secret=not-a-real-value",
+        '{"client_' + 'secret":\n"synthetic-value"}',
+        '{"client_' + 'secret"\n:\n"synthetic-value"}',
+        '{"client\\u005f' + 'secret":"synthetic-value"}',
+        "client_" + "secret:\n  synthetic-value",
+        '{"client\\q' + 'secret":"synthetic-value"}',
+        '{"client\n' + 'secret":"synthetic-value"}',
+        '"client_' + 'secret"\x0b:\n"synthetic-value"',
         "ACCESS_" + "TOKEN = 'not-a-real-value'",
         '{"refresh_' + 'token": "not-a-real-value"}',
         "private_" + "key: not-a-real-value",