From 6729359ba2acb6d775b04ca9d2236bdf42f59355 Mon Sep 17 00:00:00 2001 From: timmy Date: Mon, 17 Aug 2026 15:28:00 +0000 Subject: [PATCH] fix: make pull merges durably auditable (Closes #1028) --- src/main.py | 59 ++++++++++++++---- tests/test_pull_api.py | 132 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 178 insertions(+), 13 deletions(-) diff --git a/src/main.py b/src/main.py index fc101e1..449b7e8 100644 --- a/src/main.py +++ b/src/main.py @@ -5918,31 +5918,60 @@ async def merge_assigned_pull( target=f"{repository}#{number}", ) - async def merge_pull(): - if not await gitea_proxy.is_assigned_pull(repository, number): - raise HTTPException(status_code=404, detail="Assigned pull request not found") - return await gitea_proxy.merge_assigned_pull( - repository, number, submission.expected_head_sha + try: + assigned = await asyncio.wait_for( + gitea_proxy.is_assigned_pull(repository, number), + timeout=ISSUE_ACTION_TIMEOUT_SECONDS, + ) + except Exception: + return JSONResponse( + {"error": "The pull request assignment could not be verified. Nothing was merged."}, + status_code=503, + headers={"Retry-After": "1"}, + ) + if not assigned: + raise HTTPException(status_code=404, detail="Assigned pull request not found") + + journal = _security_event_store() + try: + operation_id = await asyncio.to_thread( + journal.reserve, + "pull_merged", + target=f"{repository}#{number}", + ) + except SecurityEventStoreError: + return JSONResponse( + {"error": "Security activity is temporarily unavailable. Nothing was merged."}, + status_code=503, + headers={"Retry-After": "1"}, ) try: result = await asyncio.wait_for( - merge_pull(), timeout=ISSUE_ACTION_TIMEOUT_SECONDS - ) - await asyncio.to_thread( - _security_event_store().record, - "pull_merged", - target=f"{repository}#{number}", + gitea_proxy.merge_assigned_pull( + repository, number, submission.expected_head_sha + ), + timeout=ISSUE_ACTION_TIMEOUT_SECONDS, ) + try: + await asyncio.to_thread(journal.finalize, operation_id) + except SecurityEventStoreError: + pass return result - except HTTPException: - raise except gitea_proxy.StalePullError: + try: + await asyncio.to_thread(journal.discard, operation_id) + except SecurityEventStoreError: + pass return JSONResponse( {"error": "New commits were pushed. Refresh before merging."}, status_code=409, ) except gitea_proxy.PullNotMergeableError: + try: + await asyncio.to_thread(journal.discard, operation_id) + except SecurityEventStoreError: + pass return JSONResponse( {"error": "This pull request is not currently safe to merge. Refresh its status."}, status_code=409, @@ -5958,6 +5987,10 @@ async def merge_assigned_pull( except Exception: merged = False if merged: + try: + await asyncio.to_thread(journal.finalize, operation_id) + except SecurityEventStoreError: + pass return {"number": number, "merged": True, "state": "closed"} return JSONResponse( { diff --git a/tests/test_pull_api.py b/tests/test_pull_api.py index cec47d0..4f566b0 100644 --- a/tests/test_pull_api.py +++ b/tests/test_pull_api.py @@ -4,6 +4,7 @@ import httpx import pytest from src import gitea_proxy, main +from src.security_event_store import SecurityEventStoreError @pytest.mark.anyio @@ -312,11 +313,98 @@ async def test_assigned_pull_merge_requires_current_eligible_head(monkeypatch): assert calls == [("stackchain/api", 7, "abc123")] +@pytest.mark.anyio +async def test_assigned_pull_merge_fails_before_gitea_when_audit_reservation_fails( + monkeypatch, +): + merge_calls = [] + + class UnavailableJournal: + def reserve(self, *_args, **_kwargs): + raise SecurityEventStoreError("unavailable") + + async def assigned(*_args): + return True + + async def merge(*args): + merge_calls.append(args) + return {"number": 7, "merged": True, "state": "closed"} + + monkeypatch.setattr(main, "_security_event_store", lambda: UnavailableJournal()) + monkeypatch.setattr(main.gitea_proxy, "is_assigned_pull", assigned) + monkeypatch.setattr(main.gitea_proxy, "merge_assigned_pull", merge) + transport = httpx.ASGITransport(app=main.app) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + response = await client.post( + "/api/v1/repos/stackchain/api/pulls/7/merge", + json={"expected_head_sha": "abc123"}, + ) + + assert response.status_code == 503 + assert response.json() == { + "error": "Security activity is temporarily unavailable. Nothing was merged." + } + assert merge_calls == [] + + +@pytest.mark.anyio +async def test_assigned_pull_merge_reports_success_and_retains_pending_audit_when_finalize_fails( + monkeypatch, +): + lifecycle = [] + + class InterruptedJournal: + def reserve(self, kind, *, target): + lifecycle.append(("reserve", kind, target)) + return "merge-operation" + + def finalize(self, operation_id): + lifecycle.append(("finalize", operation_id)) + raise SecurityEventStoreError("unavailable") + + def discard(self, operation_id): + lifecycle.append(("discard", operation_id)) + + async def assigned(*_args): + return True + + async def merge(*_args): + return {"number": 7, "merged": True, "state": "closed"} + + monkeypatch.setattr(main, "_security_event_store", lambda: InterruptedJournal()) + monkeypatch.setattr(main.gitea_proxy, "is_assigned_pull", assigned) + monkeypatch.setattr(main.gitea_proxy, "merge_assigned_pull", merge) + transport = httpx.ASGITransport(app=main.app) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + response = await client.post( + "/api/v1/repos/stackchain/api/pulls/7/merge", + json={"expected_head_sha": "abc123"}, + ) + + assert response.status_code == 200 + assert response.json() == {"number": 7, "merged": True, "state": "closed"} + assert lifecycle == [ + ("reserve", "pull_merged", "stackchain/api#7"), + ("finalize", "merge-operation"), + ] + + @pytest.mark.anyio async def test_assigned_pull_merge_reconciles_acceptance_before_timeout(monkeypatch): calls = [] merged = False + class LifecycleJournal: + def reserve(self, kind, *, target): + calls.append(("reserve", kind, target)) + return "merge-operation" + + def finalize(self, operation_id): + calls.append(("finalize", operation_id)) + + def discard(self, operation_id): + calls.append(("discard", operation_id)) + async def assigned(repository, number): return True @@ -332,6 +420,7 @@ async def test_assigned_pull_merge_reconciles_acceptance_before_timeout(monkeypa return merged monkeypatch.setattr(main, "ISSUE_ACTION_TIMEOUT_SECONDS", 0.01) + monkeypatch.setattr(main, "_security_event_store", lambda: LifecycleJournal()) monkeypatch.setattr(main.gitea_proxy, "is_assigned_pull", assigned) monkeypatch.setattr(main.gitea_proxy, "merge_assigned_pull", merge) monkeypatch.setattr(main.gitea_proxy, "is_pull_merged_at_head", confirm, raising=False) @@ -345,8 +434,10 @@ async def test_assigned_pull_merge_reconciles_acceptance_before_timeout(monkeypa assert response.status_code == 200 assert response.json() == {"number": 7, "merged": True, "state": "closed"} assert calls == [ + ("reserve", "pull_merged", "stackchain/api#7"), ("merge", "stackchain/api", 7, "abc123"), ("confirm", "stackchain/api", 7, "abc123"), + ("finalize", "merge-operation"), ] @@ -435,6 +526,47 @@ async def test_assigned_pull_merge_returns_conflict_without_mutating_stale_head( assert "New commits" in response.json()["error"] +@pytest.mark.anyio +@pytest.mark.parametrize( + "rejection", + [gitea_proxy.StalePullError("changed"), gitea_proxy.PullNotMergeableError("unsafe")], +) +async def test_assigned_pull_merge_discards_audit_reservation_after_definite_rejection( + monkeypatch, rejection +): + lifecycle = [] + + class LifecycleJournal: + def reserve(self, kind, *, target): + lifecycle.append(("reserve", kind, target)) + return "merge-operation" + + def discard(self, operation_id): + lifecycle.append(("discard", operation_id)) + + async def assigned(*_args): + return True + + async def merge(*_args): + raise rejection + + monkeypatch.setattr(main, "_security_event_store", lambda: LifecycleJournal()) + monkeypatch.setattr(main.gitea_proxy, "is_assigned_pull", assigned) + monkeypatch.setattr(main.gitea_proxy, "merge_assigned_pull", merge) + transport = httpx.ASGITransport(app=main.app) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + response = await client.post( + "/api/v1/repos/stackchain/api/pulls/7/merge", + json={"expected_head_sha": "abc123"}, + ) + + assert response.status_code == 409 + assert lifecycle == [ + ("reserve", "pull_merged", "stackchain/api#7"), + ("discard", "merge-operation"), + ] + + @pytest.mark.anyio async def test_gitea_merge_rejects_failed_ci_without_upstream_mutation(): requests = [] -- 2.43.0