fix: authorize secure source branch deletion (Closes #1368)

tests/test_dashboard_auth.py

@@ -43,6 +43,84 @@ async def fresh_grant(client, action: str, target: str) -> str:
     return response.json()["grant"]
 
 
+@pytest.mark.anyio
+async def test_source_branch_deletion_accepts_exact_one_time_fresh_authorization(
+    access_control, monkeypatch
+):
+    lifecycle = []
+
+    class Journal:
+        def record(self, *_args, **_kwargs):
+            pass
+
+        def reserve(self, kind, *, target):
+            lifecycle.append(("reserve", kind, target))
+            return "cleanup-operation"
+
+        def finalize(self, operation_id):
+            lifecycle.append(("finalize", operation_id))
+
+        def discard(self, operation_id):
+            lifecycle.append(("discard", operation_id))
+
+    async def cleanup(repository, number, source_branch, expected_head_sha):
+        lifecycle.append(
+            ("delete", repository, number, source_branch, expected_head_sha)
+        )
+        return {"number": number, "deleted": True}
+
+    monkeypatch.setattr(main, "_security_event_store", lambda: Journal())
+    monkeypatch.setattr(main.gitea_proxy, "delete_merged_source_branch", cleanup)
+    transport = httpx.ASGITransport(app=main.app)
+    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
+        signed_in = await client.post(
+            "/api/v1/session",
+            json={"access_token": "correct horse battery staple", "device_label": "Phone"},
+        )
+        assert signed_in.status_code == 200
+        headers = {
+            "Origin": "https://test",
+            "X-CSRF-Token": client.cookies["stackchain_csrf"],
+        }
+        request = {
+            "source_branch": "timmy/feature",
+            "expected_head_sha": "abc1234",
+        }
+        path = "/api/v1/repos/stackchain/api/pulls/7/source-branch"
+
+        challenged = await client.request("DELETE", path, json=request, headers=headers)
+        grant = await fresh_grant(
+            client, "delete_source_branch", "stackchain/api#7@abc1234"
+        )
+        deleted = await client.request(
+            "DELETE",
+            path,
+            json=request,
+            headers={**headers, "X-Step-Up-Grant": grant},
+        )
+        replayed = await client.request(
+            "DELETE",
+            path,
+            json=request,
+            headers={**headers, "X-Step-Up-Grant": grant},
+        )
+
+    assert challenged.status_code == 428
+    assert challenged.json()["detail"] == {
+        "detail": "Fresh authorization required",
+        "code": "step_up_required",
+        "action": "delete_source_branch",
+        "target": "stackchain/api#7@abc1234",
+    }
+    assert deleted.status_code == 200
+    assert replayed.status_code == 428
+    assert lifecycle == [
+        ("reserve", "source_branch_deleted", "stackchain/api#7@abc1234"),
+        ("delete", "stackchain/api", 7, "timmy/feature", "abc1234"),
+        ("finalize", "cleanup-operation"),
+    ]
+
+
 @pytest.mark.anyio
 async def test_passkey_enrollment_options_require_fresh_authorization_and_are_one_time(
     access_control,

tests/test_request_boundary.py

@@ -123,6 +123,44 @@ def test_request_limits_are_route_specific_and_cover_api_mutations():
     assert main.request_body_limit("POST", "/unrelated") is None
 
 
+@pytest.mark.anyio
+async def test_oversized_source_branch_delete_is_rejected_before_endpoint_work():
+    downstream_called = False
+    sent = []
+
+    async def downstream(_scope, _receive, _send):
+        nonlocal downstream_called
+        downstream_called = True
+
+    messages = iter(
+        [
+            {"type": "http.request", "body": b"x" * 40_000, "more_body": True},
+            {"type": "http.request", "body": b"y" * 40_000, "more_body": False},
+        ]
+    )
+
+    async def receive():
+        return next(messages)
+
+    async def send(message):
+        sent.append(message)
+
+    middleware = RequestBodyLimitMiddleware(downstream)
+    await middleware(
+        {
+            "type": "http",
+            "method": "DELETE",
+            "path": "/api/v1/repos/stackchain/api/pulls/7/source-branch",
+            "headers": [],
+        },
+        receive,
+        send,
+    )
+
+    assert sent[0]["status"] == 413
+    assert downstream_called is False
+
+
 def test_pull_screenshot_upload_uses_binary_attachment_boundary():
     assert (
         main.request_body_limit(