hermes: harden draft PR input boundary

services/hermes/scripts/gitea_api.py

@@ -34,6 +34,7 @@ MAX_ERROR_BYTES = 8192
 MAX_RESPONSE_BYTES = 2 * 1024 * 1024
 MAX_API_TARGET_LENGTH = 768
 MAX_API_PATH_LENGTH = 512
+RAW_API_TARGET_RE = re.compile(r"[A-Za-z0-9/_.?&=-]+\Z")
 
 
 class RejectRedirectHandler(urllib.request.HTTPRedirectHandler):
@@ -61,7 +62,7 @@ def read_token(path: Path = DEFAULT_TOKEN_FILE) -> str:
 
 def configured_base_url() -> str:
     """Reject attempts to redirect credentials away from the canonical origin."""
-    configured = os.environ.get("GITEA_BASE_URL", CANONICAL_BASE_URL).rstrip("/")
+    configured = os.environ.get("GITEA_BASE_URL", CANONICAL_BASE_URL)
     if configured != CANONICAL_BASE_URL:
         raise PolicyError("Forgejo origin is fixed to the private Atlas SCM service")
     return configured
@@ -70,14 +71,25 @@ def configured_base_url() -> str:
 def _split_api_path(path: str) -> urllib.parse.SplitResult:
     if not isinstance(path, str) or len(path) > MAX_API_TARGET_LENGTH:
         raise PolicyError("API target exceeds the safe size limit")
+    if (
+        not path.isascii()
+        or any(ord(character) < 32 or ord(character) == 127 for character in path)
+        or "\\" in path
+        or not RAW_API_TARGET_RE.fullmatch(path)
+    ):
+        raise PolicyError("API target contains non-canonical raw characters")
     target = urllib.parse.urlsplit(path)
+    canonical = urllib.parse.urlunsplit(("", "", target.path, target.query, ""))
+    if canonical != path:
+        raise PolicyError("API target is not in exact canonical form")
     if target.scheme or target.netloc or target.fragment:
         raise PolicyError("API path must be relative to the Atlas SCM origin")
     if not target.path.startswith("/api/v1/"):
         raise PolicyError("API path must start with /api/v1/")
     if len(target.path) > MAX_API_PATH_LENGTH:
         raise PolicyError("API path exceeds the safe size limit")
-    if "%" in target.path or "//" in target.path or "/../" in f"{target.path}/":
+    segments = target.path.split("/")
+    if "//" in target.path or any(segment in {".", ".."} for segment in segments):
         raise PolicyError("encoded or non-canonical API paths are not allowed")
     return target
 
@@ -136,7 +148,7 @@ def _authorize_read(target: urllib.parse.SplitResult) -> str:
 
 def api_url(base_url: str, path: str) -> str:
     """Return a canonical same-origin URL without forwarding credentials."""
-    if base_url.rstrip("/") != CANONICAL_BASE_URL:
+    if base_url != CANONICAL_BASE_URL:
         raise PolicyError("Forgejo origin is fixed to the private Atlas SCM service")
     target = _split_api_path(path)
     return urllib.parse.urlunsplit(

services/hermes/scripts/gitea_api_policy.py

@@ -1,10 +1,18 @@
-"""Pure validation policy for Hermes' least-authority Atlas Forgejo client."""
+"""Validation policy for Hermes' least-authority Atlas Forgejo client.
+
+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.
+"""
 
 from __future__ import annotations
 
 import re
 import subprocess
 import urllib.parse
+from collections import Counter
+from math import log2
 
 REPO_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,99}\Z")
 SHA_RE = re.compile(r"[0-9a-fA-F]{40}\Z")
@@ -19,73 +27,157 @@ MAX_LIMIT = 50
 MAX_PR_NUMBER = 2_147_483_647
 MAX_PR_NUMBER_DIGITS = 10
 CANONICAL_QUERY_RE = re.compile(r"[A-Za-z0-9_=&-]*\Z")
-SENSITIVE_TEXT_RE = re.compile(
-    r"""(?ix)
-    (?:
-      (?<![A-Za-z0-9])
-      (?:client[_-]?secret|access[_-]?token|refresh[_-]?token|private[_-]?key|
-         private[_-]?key[_-]?id|client[_-]?email|api[_-]?key|password|passwd|
-         secret|token|authorization|bearer|basic|personal[_-]?access[_-]?token|
-         pat|github[_-]?token|gitlab[_-]?token|gitea[_-]?token|forgejo[_-]?token|
-         aws[_-]?(?:access[_-]?key(?:[_-]?id)?|secret[_-]?(?:access[_-]?)?key|
-           session[_-]?token|security[_-]?token)|
-         access[_-]?key[_-]?id|secret[_-]?access[_-]?key|session[_-]?token|
-         account[_-]?key|shared[_-]?access[_-]?signature|sas[_-]?token|
-         azure[_-]?(?:storage[_-]?)?connection[_-]?string|connection[_-]?string|
-         docker[_-]?(?:auth[_-]?config|config[_-]?json)|dockerconfigjson|
-         registry[_-]?(?:auth|password|token)|identity[_-]?token|
-         google[_-]?(?:credentials|oauth[_-]?token)|service[_-]?account[_-]?key|
-         webhook[_-]?(?:url|token|secret)|slack[_-]?(?:token|webhook)|
-         twilio[_-]?auth[_-]?token|sendgrid[_-]?api[_-]?key)
-      [\"']?\s*[:=]\s*[\"']?
-      |
-      \b(?:bearer|basic)\s+
-        (?=[A-Za-z0-9+/_=.:-]{16,}\b)(?=[A-Za-z0-9+/_=.:-]*[0-9+/_=.-])
-        [A-Za-z0-9+/_=.:-]{16,}
-      |
-      [\"'](?:auth|identitytoken|registrytoken)[\"']\s*:\s*
-        [\"'][A-Za-z0-9+/_=.-]{8,}[\"']
-      |
-      [\"']auths[\"']\s*:\s*\{
-      |
-      [\"']type[\"']\s*:\s*[\"']service_account[\"']
-      |
-      \bDefaultEndpointsProtocol\s*=\s*https?;[^\r\n]{0,1024}
-        AccountKey\s*=
-      |
-      (?:[?;&]|\b)sv=20[0-9]{2}-[0-9]{2}-[0-9]{2}&[^\s]{0,1024}&sig=
-      |
-      -----BEGIN\s+[A-Z0-9 ][A-Z0-9 -]{1,62}-----
-      |
-      \bssh-(?:rsa|ed25519|dss|ecdsa-[A-Za-z0-9-]+)\s+[A-Za-z0-9+/]{16,}={0,3}
-      |
-      \b(?:gh[pousr]_[A-Za-z0-9]{20,}|github_pat_[A-Za-z0-9_]{20,}|
-          glpat-[A-Za-z0-9_-]{20,}|xox[baprs]-[A-Za-z0-9-]{20,}|
-          sk-ant-[A-Za-z0-9_-]{20,}|sk-(?:proj-|live_)?[A-Za-z0-9_-]{20,}|
-          [sr]k_live_[A-Za-z0-9]{20,}|
-          whsec_[A-Za-z0-9]{20,}|npm_[A-Za-z0-9]{20,}|
-          pypi-[A-Za-z0-9_-]{30,}|hf_[A-Za-z0-9]{20,}|
-          (?:gta|gto|gitea|forgejo)_[A-Za-z0-9_-]{20,}|
-          ya29\.[A-Za-z0-9_-]{20,})\b
-      |
-      \bSG\.[A-Za-z0-9_-]{16,}\.[A-Za-z0-9_-]{20,}\b
-      |
-      \bSK[0-9a-fA-F]{32}\b
-      |
-      https://hooks\.slack\.com/services/[A-Za-z0-9/_-]{20,}
-      |
-      https://(?:discord(?:app)?\.com)/api/webhooks/[0-9]+/[A-Za-z0-9._-]{20,}
-      |
-      https://[^\s/]*webhook\.office\.com/[^\s]{20,}
-      |
-      \b(?:AKIA|ASIA)[0-9A-Z]{16}\b
-      |
-      \bAIza[0-9A-Za-z_-]{35}\b
-      |
-      \beyJ[A-Za-z0-9_-]{12,}\.[A-Za-z0-9_-]{12,}\.[A-Za-z0-9_-]{8,}\b
+ASSIGNMENT_RE = re.compile(
+    r"""(?mx)
+    (?<![A-Za-z0-9_.-])
+    (?P<key_quote>[\"']?)
+    (?P<key>\.?[A-Za-z][A-Za-z0-9_.-]{0,127})
+    (?P=key_quote)
+    [ \t]*(?P<separator>[:=])[ \t]*
+    (?P<value>
+      \"(?:\\.|[^\"\\\r\n])*\" |
+      '(?:\\.|[^'\\\r\n])*' |
+      [^\r\n,;}&]*
     )
     """
 )
+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+"
+    r"(?=[A-Za-z0-9+/_=.:-]{8,}\b)(?=[A-Za-z0-9+/_=.:-]*[0-9+/_=.-])"
+    r"[A-Za-z0-9+/_=.:-]{8,}"
+)
+STANDALONE_SECRET_PATTERNS = tuple(
+    re.compile(pattern, re.IGNORECASE)
+    for pattern in (
+        r"\bgh[pousr]_[A-Za-z0-9]{20,}\b",
+        r"\bgithub_pat_[A-Za-z0-9_]{20,}\b",
+        r"\bglpat-[A-Za-z0-9_-]{20,}\b",
+        r"\bxox[baprs]-[A-Za-z0-9-]{20,}\b",
+        r"\bsk-(?:ant-|proj-)?[A-Za-z0-9_-]{20,}\b",
+        r"\b[rs]k_(?:test|live)_[A-Za-z0-9]{20,}\b",
+        r"\bwhsec_[A-Za-z0-9]{20,}\b",
+        r"\bnpm_[A-Za-z0-9]{20,}\b",
+        r"\bpypi-[A-Za-z0-9_-]{30,}\b",
+        r"\bhf_[A-Za-z0-9]{20,}\b",
+        r"\b(?:gta|gto|gitea|forgejo)_[A-Za-z0-9_-]{20,}\b",
+        r"\bya29\.[A-Za-z0-9_-]{20,}\b",
+        r"\boy2[A-Za-z0-9]{40,}\b",
+        r"\bSG\.[A-Za-z0-9_-]{16,}\.[A-Za-z0-9_-]{20,}\b",
+        r"\bSK[0-9a-f]{32}\b",
+        r"\b(?:AKIA|ASIA)[0-9A-Z]{16}\b",
+        r"\bAIza[0-9A-Za-z_-]{35}\b",
+        r"\beyJ[A-Za-z0-9_-]{12,}\.[A-Za-z0-9_-]{12,}\.[A-Za-z0-9_-]{8,}\b",
+        r"-----BEGIN\s+[A-Z0-9 ][A-Z0-9 -]{1,62}-----",
+        r"\bssh-(?:rsa|ed25519|dss|ecdsa-[A-Za-z0-9-]+)\s+[A-Za-z0-9+/]{16,}={0,3}",
+        r"https://hooks\.slack\.com/services/[A-Za-z0-9/_-]{20,}",
+        r"https://(?:discord(?:app)?\.com)/api/webhooks/[0-9]+/[A-Za-z0-9._-]{20,}",
+        r"https://[^\s/]*webhook\.office\.com/[^\s]{20,}",
+    )
+)
+HIGH_ENTROPY_TOKEN_RE = re.compile(
+    r"(?<![A-Za-z0-9+/_=-])[A-Za-z0-9+/_=-]{48,}(?![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",
+)
 
 
 class PolicyError(ValueError):
@@ -157,10 +249,109 @@ def _validate_text(value: object, name: str, maximum: int, *, required: bool) ->
     return value
 
 
+def _normalized_key(value: str) -> tuple[tuple[str, ...], str]:
+    """Split camelCase and separator-based assignment keys into semantics."""
+    separated = CAMEL_BOUNDARY_RE.sub(" ", value.strip(".\"'"))
+    words = tuple(re.findall(r"[A-Za-z0-9]+", separated.lower()))
+    return words, "".join(words)
+
+
+def _is_sensitive_key(value: str) -> bool:
+    words, compact = _normalized_key(value)
+    word_set = set(words)
+    if word_set & SENSITIVE_KEY_WORDS:
+        return True
+    if compact in SENSITIVE_COMPACT_KEYS:
+        return True
+    if compact.endswith(SENSITIVE_COMPACT_SUFFIXES):
+        return True
+    if "key" in word_set and word_set & KEY_MODIFIER_WORDS:
+        return True
+    if {"client", "email"} <= word_set or {"client", "id"} <= word_set:
+        return True
+    if {"access", "id"} <= word_set or {"connection", "string"} <= word_set:
+        return True
+    return compact.endswith("configjson")
+
+
+def _strip_assignment_value(value: str) -> tuple[str, bool]:
+    stripped = value.strip()
+    quoted = (
+        len(stripped) >= 2 and stripped[0] in {'"', "'"} and stripped[-1] == stripped[0]
+    )
+    if quoted:
+        stripped = stripped[1:-1].strip()
+    return stripped, quoted
+
+
+def _looks_like_prose(value: str) -> bool:
+    words = re.findall(r"[A-Za-z]+(?:[-'][A-Za-z]+)?", value)
+    if len(words) < 2:
+        return False
+    if re.search(r"[$`{}\[\]\\@/:=+_]", value):
+        return False
+    return words[0].lower() in PROSE_LEAD_WORDS or len(words) >= 4
+
+
+def _looks_sensitive_assignment_value(value: str) -> bool:
+    stripped, quoted = _strip_assignment_value(value)
+    if not stripped:
+        return False
+    if quoted or stripped.startswith(("{", "[", "|", ">", "!!", "&", "*")):
+        return True
+    if CREDENTIAL_SCHEME_RE.search(stripped):
+        return True
+    if any(pattern.search(stripped) for pattern in STANDALONE_SECRET_PATTERNS):
+        return True
+    if re.search(r"(?:https?://|[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+)", stripped):
+        return True
+    if re.search(r"(?:\$\{|\$\(|\\[nrt]|[A-Za-z0-9+/]{16,}={0,3}\Z)", stripped):
+        return True
+    return not _looks_like_prose(stripped)
+
+
+def _is_structural_credential_assignment(key: str, value: str) -> bool:
+    words, compact_key = _normalized_key(key)
+    stripped, _quoted = _strip_assignment_value(value)
+    _value_words, compact_value = _normalized_key(stripped)
+    return (
+        compact_key == "type" and compact_value in {"serviceaccount", "credential"}
+    ) or ("service" in words and "account" in words and bool(stripped))
+
+
+def _has_high_entropy_token(value: str) -> bool:
+    for match in HIGH_ENTROPY_TOKEN_RE.finditer(value):
+        token = match.group(0)
+        if len(token) > 256:
+            return True
+        groups = sum(
+            bool(re.search(pattern, token))
+            for pattern in (r"[a-z]", r"[A-Z]", r"[0-9]", r"[+/_=-]")
+        )
+        counts = Counter(token)
+        entropy = -sum(
+            (count / len(token)) * log2(count / len(token)) for count in counts.values()
+        )
+        if groups >= 3 and entropy >= 4.0:
+            return True
+    return False
+
+
 def _reject_sensitive(value: str, name: str, forbidden: tuple[str, ...]) -> None:
     if any(secret and secret in value for secret in forbidden):
         raise PolicyError(f"pull-request {name} contains runtime credential material")
-    if SENSITIVE_TEXT_RE.search(value):
+    if CREDENTIAL_SCHEME_RE.search(value):
+        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")
+    for candidate in 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)
+        ):
+            raise PolicyError(f"pull-request {name} resembles credential material")
+    if _has_high_entropy_token(value):
         raise PolicyError(f"pull-request {name} resembles credential material")
 
 

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

@@ -29,7 +29,11 @@ evidence, not authorization to change it.
 Before opening a PR, verify the remote, branch, clean worktree, diff, tests, and
 exact pushed commit. Never force-push. Pass that full 40-character pushed head
 SHA explicitly. Keep the short body free of credentials and credential-like
-text. Validate without reading a credential or using the network:
+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:
 
 ```sh
 /opt/coordinator/gitea_api.py --dry-run create-draft REPO \

testing/tests/test_hermes_gitea_pr_client.py

@@ -103,6 +103,10 @@ def test_redirect_handler_rejects_cross_origin_with_sentinel_authorization():
         ("https://evil.example", "/api/v1/repos/atlas/cassandra"),
         ("http://scm.bstein.dev", "/api/v1/repos/atlas/cassandra"),
         ("https://scm.bstein.dev:443", "/api/v1/repos/atlas/cassandra"),
+        ("https://scm.bstein.dev/", "/api/v1/repos/atlas/cassandra"),
+        ("HTTPS://scm.bstein.dev", "/api/v1/repos/atlas/cassandra"),
+        ("https://SCM.bstein.dev", "/api/v1/repos/atlas/cassandra"),
+        ("https://user@scm.bstein.dev", "/api/v1/repos/atlas/cassandra"),
         ("https://scm.bstein.dev", "https://evil.example/api/v1/repos/atlas/cassandra"),
         ("https://scm.bstein.dev", "/api/v1/repos/evil/cassandra"),
         ("https://scm.bstein.dev", "/api/v1/repos/%61tlas/cassandra"),
@@ -116,6 +120,72 @@ def test_host_owner_and_path_escape_attempts_are_rejected(base_url: str, path: s
         client.build_request("GET", path, base_url=base_url, token="secret")
 
 
+@pytest.mark.parametrize(
+    "path",
+    [
+        "/api/v1/repos/atlas/cassandra/pulls/1\nHost: evil.example",
+        "/api/v1/repos/atlas/cassandra/pulls/1\r\nX-Test: value",
+        "/api/v1/repos/atlas/cassandra/pulls/1\tignored",
+        "/api/v1/repos/atlas/cassandra/pulls/1\x00ignored",
+        "/api/v1/repos/atlas/cassandra/pulls/1\x1fignored",
+        "/api/v1/repos/atlas/cassandra/pulls/1\x7fignored",
+        "/api/v1/repos/atlas/cassandra\\pulls\\1",
+        "/api/v1/repos/atlas/cassandra/pulls/%31",
+        "/api/v1/repos/atlas/cassandra/pulls/１",
+        " https://scm.bstein.dev/api/v1/repos/atlas/cassandra",
+        "https://scm.bstein.dev/api/v1/repos/atlas/cassandra",
+    ],
+)
+def test_raw_noncanonical_target_is_rejected_before_urlsplit_and_opener(
+    path: str, monkeypatch
+):
+    client = _load()
+    split_called = False
+    opener_called = False
+    original_urlsplit = client.urllib.parse.urlsplit
+
+    def urlsplit(*args, **kwargs):
+        nonlocal split_called
+        split_called = True
+        return original_urlsplit(*args, **kwargs)
+
+    def opener(*_args, **_kwargs):
+        nonlocal opener_called
+        opener_called = True
+        return Response(b"{}")
+
+    monkeypatch.setattr(client.urllib.parse, "urlsplit", urlsplit)
+    with pytest.raises(client.PolicyError):
+        client.read(path, token="runtime", opener=opener)
+    assert split_called is False
+    assert opener_called is False
+
+
+@pytest.mark.parametrize(
+    "path",
+    [
+        "/api/v1/repos/atlas/cassandra/pulls/1?",
+        "//scm.bstein.dev/api/v1/repos/atlas/cassandra",
+        "/api/v1/repos/atlas/cassandra/./pulls/1",
+        "/api/v1/repos/atlas/cassandra/../admin",
+        "/api/v1/repos/atlas/cassandra//pulls/1",
+        "/api/v1/repos/atlas/cassandra/pulls/1?limit=01",
+    ],
+)
+def test_noncanonical_round_trip_or_segments_never_reach_opener(path: str):
+    client = _load()
+    opener_called = False
+
+    def opener(*_args, **_kwargs):
+        nonlocal opener_called
+        opener_called = True
+        return Response(b"{}")
+
+    with pytest.raises(client.PolicyError):
+        client.read(path, token="runtime", opener=opener)
+    assert opener_called is False
+
+
 @pytest.mark.parametrize(
     "path",
     [
@@ -431,6 +501,21 @@ def test_runtime_token_is_only_an_authorization_header():
         "https://discord.com/api/webhooks/123456789/" + "notarealcredentialvalue123456",
         "https://fake.webhook.office.com/" + "notarealcredentialvalue123456",
         "webhook_" + "url=https://example.test/not-real",
+        "FutureCloudSigning" + "Credential=not-a-real-value",
+        "future-client-signing-" + "key: not-a-real-value",
+        "future_client_signing_" + "key='not-a-real-value'",
+        '{"serviceAccountPrivate' + 'Key":"not-a-real-value"}',
+        "CONTAINER_REGISTRY_" + "CREDENTIAL=not-a-real-value",
+        "someWebhookSigning" + "Secret: not-a-real-value",
+        "client" + "Key=not-a-real-value",
+        "session" + "Key: not-a-real-value",
+        "access" + "Id=not-a-real-value",
+        "credentials" + ": {user: fake}",
+        "private" + "Key: |",
+        "nuget_api_" + "key=not-a-real-value",
+        "oy2" + "a" * 44,
+        "sk_test_" + "notarealcredentialvalue123456",
+        "A1b2C3d4E5f6G7h8I9j0K_l-M+n/O=pQ2rS3tU4vW5xY6zZ7aB8cC9d",
         "AKIA" + "A" * 16,
         "AIza" + "a" * 35,
         "-----BEGIN OPENSSH " + "PRIVATE KEY-----",
@@ -477,16 +562,54 @@ def test_create_rejects_expanded_credential_shapes_before_network(
     assert called is False
 
 
+def test_very_long_compact_body_is_rejected_before_request_or_network():
+    client = _load()
+    request_built = False
+    opener_called = False
+    original_build_request = client.build_request
+
+    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())
+
+    client.build_request = build_request
+    with pytest.raises(client.PolicyError, match="credential material"):
+        client.create_draft(
+            "cassandra",
+            base="main",
+            head="hermes/fix",
+            head_sha=HEAD_SHA,
+            title="Focused fix",
+            body="a" * 300,
+            token="runtime-sentinel",
+            opener=opener,
+        )
+    assert request_built is False
+    assert opener_called is False
+
+
 @pytest.mark.parametrize(
     "safe_text",
     [
         "AWS_SECRET_ACCESS_KEY is injected at runtime",
+        "Token: reject empty values",
+        "Authorization = preserve header behavior",
+        "Password: add regression",
         "Review the Authorization header behavior",
         "Bearer authentication is required for this route",
         "Document Docker auths payload rejection",
         "The client_email field belongs to service accounts",
         "AccountKey assignments must be blocked",
         "This patch changes token validation without including a value",
+        "FutureCloudSigningCredential handling needs a regression test",
+        "The clientKey name is documented without an assigned value",
+        "A SHA-256 digest 0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef is evidence",
     ],
 )
 def test_credential_policy_keeps_normal_engineering_prose_usable(safe_text: str):