From 73618ca8f813eba374c6c49ed4b91f03ac131538 Mon Sep 17 00:00:00 2001 From: jenkins Date: Sun, 13 Sep 2026 20:32:12 -0500 Subject: [PATCH] hermes: accept verified Forgejo draft update acknowledgments --- .../scm-common/scripts/scm_task_drafts.py | 21 ++++- testing/tests/test_hermes_scm_task_grants.py | 79 +++++++++++++++++++ 2 files changed, 98 insertions(+), 2 deletions(-) diff --git a/services/hermes/scm-common/scripts/scm_task_drafts.py b/services/hermes/scm-common/scripts/scm_task_drafts.py index 7539de64..0903772a 100644 --- a/services/hermes/scm-common/scripts/scm_task_drafts.py +++ b/services/hermes/scm-common/scripts/scm_task_drafts.py @@ -10,6 +10,8 @@ import deadline_http from gitea_api import CANONICAL_BASE_URL from gitea_api_policy import PolicyError, _draft_title, _validate_body, _validate_pr_number +UPDATE_SUCCESS_STATUSES = {200, 201} + def request_fields(value: dict[str, Any], token: str) -> tuple[str, int, str, str]: if set(value) != {"grant", "pr_number", "title", "body"} or not isinstance(value["grant"], str): @@ -39,10 +41,25 @@ def update(token: str, repo: str, number: int, title: str, body: str) -> bytes: method="PATCH", headers={"Authorization": f"Basic {auth}", "Content-Type": "application/json", "Accept": "application/json"}, ) with deadline_http.open_bounded(request, maximum=2 * 1024 * 1024, timeout=30) as response: - if getattr(response, "status", None) != 200 or response.headers.get_content_type() != "application/json": + if getattr(response, "status", None) not in UPDATE_SUCCESS_STATUSES or response.headers.get_content_type() != "application/json": raise PolicyError("draft update upstream response is invalid") result = response.read(2 * 1024 * 1024 + 1) if len(result) > 2 * 1024 * 1024 or token.encode() in result: raise PolicyError("draft update upstream response is invalid") - json.loads(result) + try: + payload = json.loads(result) + except json.JSONDecodeError as error: + raise PolicyError("draft update upstream response is invalid") from error + try: + response_number = _validate_pr_number( + payload.get("number") if isinstance(payload, dict) else None + ) + except PolicyError as error: + raise PolicyError("draft update upstream response is invalid") from error + if ( + not isinstance(payload, dict) + or response_number != number + or payload.get("html_url") != f"{CANONICAL_BASE_URL}/titan/{repo}/pulls/{number}" + ): + raise PolicyError("draft update upstream response is invalid") return result diff --git a/testing/tests/test_hermes_scm_task_grants.py b/testing/tests/test_hermes_scm_task_grants.py index 42fedbe0..859b34d5 100644 --- a/testing/tests/test_hermes_scm_task_grants.py +++ b/testing/tests/test_hermes_scm_task_grants.py @@ -160,6 +160,85 @@ def test_draft_refresh_requires_exact_owned_pr_and_has_no_mutating_fields(): drafts.request_fields({"grant": "x", "pr_number": 55, "title": "x", "body": "x", "state": "closed"}, "token") +def _draft_update_response(status: int, payload: bytes, content_type: str = "application/json"): + class Response: + def __init__(self): + self.headers = Message() + + def __enter__(self): + self.status = status + self.headers["Content-Type"] = content_type + return self + + def __exit__(self, *_args): + return False + + @staticmethod + def read(_limit): + return payload + + return Response() + + +@pytest.mark.parametrize("status", [200, 201]) +def test_draft_update_accepts_only_known_success_statuses_with_matching_pr_shape( + monkeypatch, status +): + drafts = _load("scm_task_drafts") + body = json.dumps({ + "number": 55, + "html_url": "https://scm.bstein.dev/titan/atlas-iac/pulls/55", + }).encode() + seen = [] + + def open_bounded(request, maximum, timeout): + seen.append((request, maximum, timeout)) + return _draft_update_response(status, body) + + monkeypatch.setattr(drafts.deadline_http, "open_bounded", open_bounded) + + assert drafts.update("token", "atlas-iac", 55, "WIP: refreshed", "evidence") == body + request, maximum, timeout = seen[0] + assert request.get_method() == "PATCH" + assert request.full_url.endswith("/api/v1/repos/titan/atlas-iac/pulls/55") + assert json.loads(request.data) == {"title": "WIP: refreshed", "body": "evidence"} + assert maximum == 2 * 1024 * 1024 and timeout == 30 + + +def test_draft_update_rejects_unapproved_success_status(monkeypatch): + drafts = _load("scm_task_drafts") + body = b'{"number":55,"html_url":"https://scm.bstein.dev/titan/atlas-iac/pulls/55"}' + monkeypatch.setattr( + drafts.deadline_http, + "open_bounded", + lambda *_args, **_kwargs: _draft_update_response(202, body), + ) + + with pytest.raises(drafts.PolicyError, match="upstream response"): + drafts.update("token", "atlas-iac", 55, "WIP: refreshed", "evidence") + + +@pytest.mark.parametrize( + "payload", + [ + b"[]", + b'{"number":56,"html_url":"https://scm.bstein.dev/titan/atlas-iac/pulls/55"}', + b'{"number":55,"html_url":"https://evil.example/titan/atlas-iac/pulls/55"}', + b"{not-json", + ], +) +def test_draft_update_rejects_malformed_success_response_shape(monkeypatch, payload): + drafts = _load("scm_task_drafts") + monkeypatch.setattr( + drafts.deadline_http, + "open_bounded", + lambda *_args, **_kwargs: _draft_update_response(201, payload), + ) + + with pytest.raises(drafts.PolicyError, match="upstream response"): + drafts.update("token", "atlas-iac", 55, "WIP: refreshed", "evidence") + + def _control_handler(broker, body: dict): """Build the broker's real control handler without opening a TCP listener.""" encoded = json.dumps(body).encode()