From c8ccc8f2498fc566cc27e748d1aaeae6afcda9b4 Mon Sep 17 00:00:00 2001 From: jenkins Date: Sun, 16 Aug 2026 20:41:14 -0300 Subject: [PATCH] hermes: harden draft PR input boundary --- services/hermes/kustomization.yaml | 1 + services/hermes/scripts/gitea_api.py | 141 +++------- services/hermes/scripts/gitea_api_policy.py | 226 ++++++++++++++++ testing/tests/test_hermes_gitea_pr_client.py | 253 ++++++++++++++++-- .../tests/test_hermes_gitea_pr_integration.py | 9 + testing/tests/test_hermes_runtime_access.py | 2 + 6 files changed, 510 insertions(+), 122 deletions(-) create mode 100644 services/hermes/scripts/gitea_api_policy.py diff --git a/services/hermes/kustomization.yaml b/services/hermes/kustomization.yaml index 87984530..86d334b6 100644 --- a/services/hermes/kustomization.yaml +++ b/services/hermes/kustomization.yaml @@ -70,6 +70,7 @@ configMapGenerator: - claude_oauth_broker.py=scripts/claude_oauth_broker.py - worker_route_broker.py=scripts/worker_route_broker.py - gitea_api.py=scripts/gitea_api.py + - gitea_api_policy.py=scripts/gitea_api_policy.py - gitea_askpass.sh=scripts/gitea_askpass.sh - hermes_coordinator.py=scripts/hermes_coordinator.py - hermes_model_routing.py=scripts/hermes_model_routing.py diff --git a/services/hermes/scripts/gitea_api.py b/services/hermes/scripts/gitea_api.py index b4a98eca..3ea16047 100755 --- a/services/hermes/scripts/gitea_api.py +++ b/services/hermes/scripts/gitea_api.py @@ -7,7 +7,6 @@ import argparse import json import os import re -import subprocess import sys import urllib.error import urllib.parse @@ -15,28 +14,26 @@ import urllib.request from collections.abc import Callable from pathlib import Path +from gitea_api_policy import ( + PolicyError, + _draft_title, + _validate_body, + _validate_pr_number, + _validate_pr_number_segment, + _validate_query, + _validate_ref, + _validate_repo, + _validate_sha, +) + CANONICAL_BASE_URL = "https://scm.bstein.dev" ALLOWED_OWNER = "atlas" DEFAULT_TOKEN_FILE = Path("/runtime-access/gitea-token") -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") CREATE_PATH_RE = re.compile(r"/api/v1/repos/atlas/([^/]+)/pulls\Z") MAX_ERROR_BYTES = 8192 MAX_RESPONSE_BYTES = 2 * 1024 * 1024 -DRAFT_TITLE_PREFIX = "WIP: " -GIT_BIN = "/usr/bin/git" -SENSITIVE_BODY_PATTERNS = ( - re.compile(r"-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----"), - re.compile(r"(?i)\b(?:password|passwd|secret|token|api[_-]?key)\s*[:=]"), - re.compile(r"(?i)\bauthorization\s*:\s*(?:bearer|token|basic)\s+\S+"), - re.compile(r"\b(?:gh[pousr]_[A-Za-z0-9]{20,}|sk-[A-Za-z0-9]{20,})\b"), - re.compile(r"\b(?:AKIA[0-9A-Z]{16}|AIza[0-9A-Za-z_-]{35})\b"), - re.compile(r"\beyJ[A-Za-z0-9_-]{12,}\.[A-Za-z0-9_-]{12,}\.[A-Za-z0-9_-]{8,}\b"), -) - - -class PolicyError(ValueError): - """Raised when a requested Forgejo operation exceeds the safe boundary.""" +MAX_API_TARGET_LENGTH = 768 +MAX_API_PATH_LENGTH = 512 class RejectRedirectHandler(urllib.request.HTTPRedirectHandler): @@ -70,101 +67,21 @@ def configured_base_url() -> str: return configured -def _validate_repo(repo: str) -> str: - if not REPO_RE.fullmatch(repo) or repo in {".", ".."}: - raise PolicyError("repository name is outside the Atlas allowlist") - return repo - - -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("-"): - 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") - result = subprocess.run( - [GIT_BIN, "check-ref-format", f"refs/heads/{value}"], - stdin=subprocess.DEVNULL, - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - check=False, - ) - if result.returncode != 0: - raise PolicyError(f"{name} must be a safe same-repository branch name") - return value - - -def _validate_sha(value: object, name: str = "head SHA") -> str: - if not isinstance(value, str) or not SHA_RE.fullmatch(value): - raise PolicyError(f"{name} must be a full Git SHA-1") - return value.lower() - - -def _validate_text(value: object, name: str, maximum: 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: - raise PolicyError(f"{name} exceeds the safe request limit") - return value - - -def _validate_body(value: object, *, forbidden: tuple[str, ...] = ()) -> str: - body = _validate_text(value, "body", 16384, required=False) - if any(secret and secret in body for secret in forbidden): - raise PolicyError("pull-request body contains runtime credential material") - if any(pattern.search(body) for pattern in SENSITIVE_BODY_PATTERNS): - raise PolicyError("pull-request body resembles credential material") - return body - - -def _draft_title(value: object) -> str: - """Return a bounded title using Gitea's configured default draft prefix.""" - title = _validate_text(value, "title", 251, required=True).strip() - for prefix in ("WIP:", "[WIP]"): - if title.upper().startswith(prefix): - title = title[len(prefix) :].lstrip() - break - if not title: - raise PolicyError("title must contain text after the draft prefix") - return DRAFT_TITLE_PREFIX + title - - 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") target = urllib.parse.urlsplit(path) 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}/": raise PolicyError("encoded or non-canonical API paths are not allowed") return target -def _validate_query(target: urllib.parse.SplitResult, allowed: set[str]) -> None: - try: - pairs = urllib.parse.parse_qsl( - target.query, keep_blank_values=True, strict_parsing=True - ) - except ValueError as exc: - raise PolicyError("invalid API query") from exc - if len({key for key, _ in pairs}) != len(pairs): - raise PolicyError("duplicate API query parameters are not allowed") - if any(key not in allowed for key, _ in pairs): - raise PolicyError("API query parameter is outside the read allowlist") - values = dict(pairs) - for name in ("page", "limit"): - if name not in values: - continue - if not values[name].isdigit() or int(values[name]) < 1: - raise PolicyError(f"{name} must be a positive integer") - if "limit" in values and int(values["limit"]) > 50: - raise PolicyError("read limit cannot exceed 50") - if "state" in values and values["state"] not in {"open", "closed", "all"}: - raise PolicyError("pull-request state is invalid") - - def _authorize_read(target: urllib.parse.SplitResult) -> str: """Allow only repository, PR, branch, commit, and status metadata reads.""" prefix = "/api/v1/repos/atlas/" @@ -180,10 +97,18 @@ def _authorize_read(target: urllib.parse.SplitResult) -> str: if suffix == "pulls": _validate_query(target, {"page", "limit", "state"}) return "pull-list" - if re.fullmatch(r"pulls/[1-9][0-9]*", suffix): + content_match = re.fullmatch(r"pulls/([^/]+)\.(?:patch|diff)", suffix) + if content_match: + _validate_pr_number_segment(content_match.group(1)) + raise PolicyError("repository API route is outside the metadata read allowlist") + pull_match = re.fullmatch(r"pulls/([^/]+)", suffix) + if pull_match: + _validate_pr_number_segment(pull_match.group(1)) _validate_query(target, set()) return "pull" - if re.fullmatch(r"pulls/[1-9][0-9]*/(?:commits|files)", suffix): + evidence_match = re.fullmatch(r"pulls/([^/]+)/(?:commits|files)", suffix) + if evidence_match: + _validate_pr_number_segment(evidence_match.group(1)) _validate_query(target, {"page", "limit"}) return "pull-evidence" if suffix == "branches": @@ -255,6 +180,10 @@ def build_request( ) -> urllib.request.Request: """Build an authorized request without putting the token in its URL or body.""" authorize_request(method, path, data) + if method.upper() == "POST": + assert isinstance(data, dict) + _draft_title(data["title"], forbidden=(token,)) + _validate_body(data["body"], forbidden=(token,)) payload = None if data is not None: payload = json.dumps(data, separators=(",", ":")).encode("utf-8") @@ -332,9 +261,7 @@ def _require_create_response( raise PolicyError("Forgejo returned invalid pull-request metadata") from exc if not isinstance(document, dict): raise PolicyError("Forgejo returned invalid pull-request metadata") - number = document.get("number") - if not isinstance(number, int) or isinstance(number, bool) or number < 1: - raise PolicyError("Forgejo omitted a valid pull-request number") + number = _validate_pr_number(document.get("number")) full_name = f"{ALLOWED_OWNER}/{repo}" expected_html = f"{CANONICAL_BASE_URL}/{full_name}/pulls/{number}" checks = ( @@ -372,7 +299,7 @@ def create_draft( base = _validate_ref(base, "base") head = _validate_ref(head, "head") head_sha = _validate_sha(head_sha) - title = _draft_title(title) + title = _draft_title(title, forbidden=(token,)) body = _validate_body(body, forbidden=(token,)) data = {"base": base, "body": body, "head": head, "title": title} result = _request( diff --git a/services/hermes/scripts/gitea_api_policy.py b/services/hermes/scripts/gitea_api_policy.py new file mode 100644 index 00000000..6f6c755c --- /dev/null +++ b/services/hermes/scripts/gitea_api_policy.py @@ -0,0 +1,226 @@ +"""Pure validation policy for Hermes' least-authority Atlas Forgejo client.""" + +from __future__ import annotations + +import re +import subprocess +import urllib.parse + +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") +DRAFT_TITLE_PREFIX = "WIP: " +GIT_BIN = "/usr/bin/git" +MAX_QUERY_LENGTH = 128 +MAX_QUERY_ITEMS = 3 +MAX_QUERY_KEY_LENGTH = 16 +MAX_QUERY_VALUE_LENGTH = 16 +MAX_PAGE = 10_000 +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) + (?: + (? str: + if not REPO_RE.fullmatch(repo) or repo in {".", ".."}: + raise PolicyError("repository name is outside the Atlas allowlist") + return repo + + +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("-"): + 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") + result = subprocess.run( + [GIT_BIN, "check-ref-format", f"refs/heads/{value}"], + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + check=False, + ) + if result.returncode != 0: + raise PolicyError(f"{name} must be a safe same-repository branch name") + return value + + +def _validate_sha(value: object, name: str = "head SHA") -> str: + if not isinstance(value, str) or not SHA_RE.fullmatch(value): + raise PolicyError(f"{name} must be a full Git SHA-1") + return value.lower() + + +def _validate_pr_number_segment(value: object) -> int: + """Validate a canonical bounded pull-request number from an API path.""" + if ( + not isinstance(value, str) + or len(value) > MAX_PR_NUMBER_DIGITS + or not re.fullmatch(r"[1-9][0-9]*", value) + ): + raise PolicyError("pull-request number must use bounded canonical ASCII digits") + number = int(value) + if number > MAX_PR_NUMBER: + raise PolicyError("pull-request number is outside its allowed range") + return number + + +def _validate_pr_number(value: object) -> int: + """Validate a bounded pull-request number returned by Forgejo.""" + if ( + not isinstance(value, int) + or isinstance(value, bool) + or not 1 <= value <= MAX_PR_NUMBER + ): + raise PolicyError("Forgejo omitted a valid pull-request number") + return value + + +def _validate_text(value: object, name: str, maximum: 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: + raise PolicyError(f"{name} exceeds the safe request limit") + return value + + +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): + raise PolicyError(f"pull-request {name} resembles credential material") + + +def _validate_body(value: object, *, forbidden: tuple[str, ...] = ()) -> str: + body = _validate_text(value, "body", 16384, 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() + _reject_sensitive(title, "title", forbidden) + for prefix in ("WIP:", "[WIP]"): + if title.upper().startswith(prefix): + title = title[len(prefix) :].lstrip() + break + if not title: + raise PolicyError("title must contain text after the draft prefix") + return DRAFT_TITLE_PREFIX + title + + +def _validate_query(target: urllib.parse.SplitResult, allowed: set[str]) -> None: + """Accept only short canonical ASCII query strings with bounded pagination.""" + raw = target.query + if ( + len(raw) > MAX_QUERY_LENGTH + or not raw.isascii() + or not CANONICAL_QUERY_RE.fullmatch(raw) + or "%" in raw + ): + raise PolicyError("API query must use short canonical ASCII form") + try: + pairs = urllib.parse.parse_qsl(raw, keep_blank_values=True, strict_parsing=True) + except ValueError as exc: + raise PolicyError("invalid API query") from exc + if len(pairs) > MAX_QUERY_ITEMS: + raise PolicyError("too many API query parameters") + if len({key for key, _ in pairs}) != len(pairs): + raise PolicyError("duplicate API query parameters are not allowed") + if any( + not key + or len(key) > MAX_QUERY_KEY_LENGTH + or len(value) > MAX_QUERY_VALUE_LENGTH + for key, value in pairs + ): + raise PolicyError("API query key or value exceeds its safe limit") + if any(key not in allowed for key, _ in pairs): + raise PolicyError("API query parameter is outside the read allowlist") + values = dict(pairs) + for name in ("page", "limit"): + if name not in values: + continue + if not re.fullmatch(r"[0-9]+", values[name]): + raise PolicyError(f"{name} must use ASCII decimal digits") + number = int(values[name]) + if values[name] != str(number): + raise PolicyError(f"{name} must use canonical ASCII decimal form") + ceiling = MAX_PAGE if name == "page" else MAX_LIMIT + if not 1 <= number <= ceiling: + raise PolicyError(f"{name} is outside its allowed range") + if "state" in values and values["state"] not in {"open", "closed", "all"}: + raise PolicyError("pull-request state is invalid") diff --git a/testing/tests/test_hermes_gitea_pr_client.py b/testing/tests/test_hermes_gitea_pr_client.py index 447f3067..068065a1 100644 --- a/testing/tests/test_hermes_gitea_pr_client.py +++ b/testing/tests/test_hermes_gitea_pr_client.py @@ -14,6 +14,8 @@ import pytest ROOT = Path(__file__).parents[2] CLIENT_PATH = ROOT / "services/hermes/scripts/gitea_api.py" HEAD_SHA = "465cf9146b05c174a2a8d310aff6c64be58277b6" +if str(CLIENT_PATH.parent) not in sys.path: + sys.path.insert(0, str(CLIENT_PATH.parent)) def _load(): @@ -179,6 +181,109 @@ def test_read_query_is_bounded(path: str): client.authorize_request("GET", path, None) +@pytest.mark.parametrize( + "query", + [ + "page=0", + "page=10001", + "page=01", + "limit=0", + "limit=51", + "limit=01", + "page=" + "9" * 4000, + "page=０", + "page=%EF%BC%90", + "state=%6fpen", + "p%61ge=1", + "page=1&page=2", + "page=1&" + "x" * 17 + "=1", + "state=" + "x" * 17, + "page=1&limit=2&state=open&extra=3", + ], +) +def test_read_query_requires_canonical_bounded_ascii(query: str): + client = _load() + + with pytest.raises(client.PolicyError): + client.authorize_request( + "GET", f"/api/v1/repos/atlas/cassandra/pulls?{query}", None + ) + + +@pytest.mark.parametrize( + "suffix", + [ + "pulls/{number}", + "pulls/{number}.patch", + "pulls/{number}.diff", + "pulls/{number}/commits", + "pulls/{number}/files", + "commits/{number}/status", + "commits/{number}/statuses", + "statuses/{number}", + "branches/{number}", + ], +) +def test_oversized_numeric_or_captured_path_never_reaches_opener(suffix: str): + client = _load() + called = False + + def opener(*_args, **_kwargs): + nonlocal called + called = True + return Response(b"{}") + + path = "/api/v1/repos/atlas/cassandra/" + suffix.format(number="9" * 4000) + with pytest.raises(client.PolicyError): + client.read(path, token="runtime", opener=opener) + assert called is False + + +@pytest.mark.parametrize( + "number", + ["0", "01", "2147483648", "１２", "%31", "12345678901"], +) +@pytest.mark.parametrize("tail", ["", ".patch", ".diff", "/commits", "/files"]) +def test_noncanonical_or_out_of_range_pr_number_never_reaches_opener( + number: str, tail: str +): + client = _load() + called = False + + def opener(*_args, **_kwargs): + nonlocal called + called = True + return Response(b"{}") + + with pytest.raises(client.PolicyError): + client.read( + f"/api/v1/repos/atlas/cassandra/pulls/{number}{tail}", + token="runtime", + opener=opener, + ) + assert called is False + + +def test_maximum_bounded_pr_number_is_readable(): + client = _load() + called = False + + def opener(*_args, **_kwargs): + nonlocal called + called = True + return Response(b"{}") + + assert ( + client.read( + "/api/v1/repos/atlas/cassandra/pulls/2147483647", + token="runtime", + opener=opener, + ) + == b"{}" + ) + assert called is True + + @pytest.mark.parametrize( ("method", "path", "data"), [ @@ -230,7 +335,7 @@ def test_complete_git_ref_validation_rejects_invalid_names(ref: str): def test_git_ref_validation_uses_fixed_trusted_binary(): client = _load() - assert client.GIT_BIN == "/usr/bin/git" + assert client._validate_ref.__globals__["GIT_BIN"] == "/usr/bin/git" assert client._validate_ref("hermes/valid-fix", "head") == "hermes/valid-fix" @@ -272,42 +377,159 @@ def test_runtime_token_is_only_an_authorization_header(): @pytest.mark.parametrize( - "body", + "sensitive", [ - "pass" + "word=not-a-real-credential", + "client_" + "secret=not-a-real-value", + "ACCESS_" + "TOKEN = 'not-a-real-value'", + '{"refresh_' + 'token": "not-a-real-value"}', + "private_" + "key: not-a-real-value", + "AWS_SECRET_ACCESS_" + "KEY=not-a-real-value", + "aws_access_key_" + "id: not-a-real-value", + "AWS_SESSION_" + "TOKEN = 'not-a-real-value'", + '{"AccessKey' + 'Id":"not-a-real-value"}', + '{"SecretAccess' + 'Key":"not-a-real-value"}', + '{"Session' + 'Token":"not-a-real-value"}', + "aws-security-" + "token: not-a-real-value", + "Account" + "Key=not-a-real-value", + "SharedAccess" + "Signature: not-a-real-value", + "AZURE_STORAGE_CONNECTION_" + "STRING='not-a-real-value'", + "DefaultEndpointsProtocol=https;AccountName=fake;Account" + + "Key=not-a-real-value;EndpointSuffix=example", + "?sv=2024-11-04&ss=b&srt=sco&sp=rwdlac&se=2099-01-01&sig=" + "not-a-real-value", + "DOCKER_AUTH_" + "CONFIG='not-a-real-value'", + '{"auths":{"registry.example":{"auth":"bm90LXJlYWw="}}}', + '{"identity' + 'token":"not-a-real-value"}', + '{"type":"service_' + 'account","client_email":"fake@example.test"}', + '{"private_key_' + 'id":"not-a-real-value"}', + '{"client_' + 'email":"fake@example.test"}', + "GOOGLE_CREDENTIALS" + "=not-a-real-value", + "personal_access_" + "token: not-a-real-value", + "GITEA_" + "TOKEN=not-a-real-value", + "gitlab-token" + ": not-a-real-value", + "pat" + "=not-a-real-value", "Authorization: " + "Bearer not-a-real-credential-value", - "token: " + "ghp_" + "notarealcredentialvalue123456", + "authorization = " + '"Basic not-a-real-credential-value"', + "Bearer" + "=not-a-real-credential-value", + "Basic" + ": not-a-real-credential-value", + "pass" + "word=not-a-real-credential", + "ghp_" + "notarealcredentialvalue123456", + "github_pat_" + "notarealcredentialvalue123456", + "glpat-" + "notarealcredentialvalue123456", + "xoxb-" + "not-a-real-credential-value-123456", + "sk-ant-" + "notarealcredentialvalue123456", + "sk-proj-" + "notarealcredentialvalue123456", + "sk_live_" + "notarealcredentialvalue123456", + "ya29." + "notarealcredentialvalue123456", + "gta_" + "notarealcredentialvalue123456", + "whsec_" + "notarealcredentialvalue123456", + "npm_" + "notarealcredentialvalue123456", + "pypi-" + "notarealcredentialvalue123456789012345", + "hf_" + "notarealcredentialvalue123456", + "SG." + "notarealvalue1234" + ".notarealcredentialvalue123456", + "SK" + "a" * 32, + "https://hooks.slack.com/services/" + "T000/B000/notarealvalue123456", + "https://discord.com/api/webhooks/123456789/" + "notarealcredentialvalue123456", + "https://fake.webhook.office.com/" + "notarealcredentialvalue123456", + "webhook_" + "url=https://example.test/not-real", + "AKIA" + "A" * 16, + "AIza" + "a" * 35, "-----BEGIN OPENSSH " + "PRIVATE KEY-----", + "ssh-ed25519 " + "bm90YXJlYWxjcmVkZW50aWFsdmFsdWU=", "eyJnotarealheader." + "notarealpayloadvalue." + "notarealsignature", ], ) -def test_body_rejects_common_credential_shapes(body: str): - client = _load() - - with pytest.raises(client.PolicyError, match="credential material"): - client._validate_body(body) - - -def test_create_rejects_exact_runtime_token_before_network(): +@pytest.mark.parametrize("field", ["title", "body"]) +def test_create_rejects_expanded_credential_shapes_before_network( + sensitive: str, field: str +): client = _load() called = False + request_built = 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 called called = True return Response(_draft_response()) + client.build_request = build_request + values = {"title": "Focused fix", "body": "Review evidence"} + values[field] = sensitive + + with pytest.raises(client.PolicyError, match="credential material"): + client.create_draft( + "cassandra", + base="main", + head="hermes/fix", + head_sha=HEAD_SHA, + title=values["title"], + body=values["body"], + token="runtime-sentinel", + opener=opener, + ) + assert request_built is False + assert called is False + + +@pytest.mark.parametrize( + "safe_text", + [ + "AWS_SECRET_ACCESS_KEY is injected at runtime", + "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", + ], +) +def test_credential_policy_keeps_normal_engineering_prose_usable(safe_text: str): + client = _load() + + assert client._validate_body(safe_text) == safe_text + assert client._draft_title(safe_text) == f"WIP: {safe_text}" + + +@pytest.mark.parametrize("field", ["title", "body"]) +def test_create_rejects_exact_runtime_token_before_network(field: str): + client = _load() + called = False + request_built = 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 called + called = True + return Response(_draft_response()) + + client.build_request = build_request + runtime_token = "exact-random-runtime-sentinel-7b73ac61" + values = {"title": "Focused fix", "body": "Review evidence"} + values[field] = f"Accidental {runtime_token} value" with pytest.raises(client.PolicyError, match="runtime credential"): client.create_draft( "cassandra", base="main", head="hermes/fix", head_sha=HEAD_SHA, - title="Focused fix", - body="accidental runtime-sentinel value", - token="runtime-sentinel", + title=values["title"], + body=values["body"], + token=runtime_token, opener=opener, ) + assert request_built is False assert called is False @@ -345,6 +567,7 @@ def test_create_postcondition_rejects_every_material_mismatch(): client = _load() mutations = [ ("number", 0), + ("number", 2_147_483_648), ("state", "closed"), ("draft", False), ("merged", True), diff --git a/testing/tests/test_hermes_gitea_pr_integration.py b/testing/tests/test_hermes_gitea_pr_integration.py index d3b47eef..4a589f4c 100644 --- a/testing/tests/test_hermes_gitea_pr_integration.py +++ b/testing/tests/test_hermes_gitea_pr_integration.py @@ -15,6 +15,8 @@ import yaml ROOT = Path(__file__).parents[2] CLIENT_PATH = ROOT / "services/hermes/scripts/gitea_api.py" HEAD_SHA = "465cf9146b05c174a2a8d310aff6c64be58277b6" +if str(CLIENT_PATH.parent) not in sys.path: + sys.path.insert(0, str(CLIENT_PATH.parent)) def _load(): @@ -136,6 +138,13 @@ def test_flux_manifest_projects_runtime_vault_token_and_skill_only(): kustomization = yaml.safe_load( (ROOT / "services/hermes/kustomization.yaml").read_text(encoding="utf-8") ) + coordinator = next( + item + for item in kustomization["configMapGenerator"] + if item["name"] == "hermes-coordinator" + ) + assert "gitea_api.py=scripts/gitea_api.py" in coordinator["files"] + assert "gitea_api_policy.py=scripts/gitea_api_policy.py" in coordinator["files"] generator = next( item for item in kustomization["configMapGenerator"] diff --git a/testing/tests/test_hermes_runtime_access.py b/testing/tests/test_hermes_runtime_access.py index 7ad5087a..b27806bf 100644 --- a/testing/tests/test_hermes_runtime_access.py +++ b/testing/tests/test_hermes_runtime_access.py @@ -14,6 +14,8 @@ import yaml ROOT = Path(__file__).parents[2] HERMES = ROOT / "services" / "hermes" SCRIPTS = HERMES / "scripts" +if str(SCRIPTS) not in sys.path: + sys.path.insert(0, str(SCRIPTS)) def _load(name: str):