Merge pull request 'Require fresh authorization and audit CI job reruns' (#1482)

frontend/security-center.js

@@ -233,6 +233,7 @@
         issue_closed: 'Issue closed', pull_merged: 'Pull request merged',
         source_branch_deleted: 'Source branch deleted',
         release_rollback_prepared: 'Release rollback prepared',
+        ci_job_retried: 'CI job retried',
         comment_deleted: 'Comment deleted',
         pull_review_approved: 'Pull request approved',
         pull_review_changes_requested: 'Changes requested', gitea_time_logged: 'Gitea time logged',

src/main.py

@@ -606,6 +606,7 @@ StepUpAction = Literal[
     "merge_pull",
     "delete_source_branch",
     "prepare_release_rollback",
+    "retry_ci_job",
     "submit_pull_review",
     "close_issue",
     "delete_comment",
@@ -7863,13 +7864,38 @@ async def pull_action_failure(
 )
 async def retry_pull_action_job(
     retry: PullReadyRequest,
+    request: Request,
     owner: str,
     repo: str,
     number: int = PathParam(gt=0),
     run_id: int = PathParam(gt=0),
     job_index: int = PathParam(ge=0),
+    step_up_grant: str | None = Header(
+        default=None, alias="X-Step-Up-Grant", max_length=128
+    ),
 ) -> JSONResponse:
     repository = f"{owner}/{repo}"
+    target = (
+        f"{repository}#{number}@{retry.expected_head_sha}:"
+        f"actions/{run_id}/jobs/{job_index}"
+    )
+    await _require_step_up(
+        request, step_up_grant, action="retry_ci_job", target=target
+    )
+    journal = _security_event_store()
+    try:
+        operation_id = await asyncio.to_thread(
+            journal.reserve,
+            "ci_job_retried",
+            principal_id=await _security_principal_id(request),
+            target=target,
+        )
+    except SecurityEventStoreError:
+        return JSONResponse(
+            {"error": "Security activity is temporarily unavailable. No job was retried."},
+            status_code=503,
+            headers={"Cache-Control": "no-store", "Retry-After": "1"},
+        )
 
     async def retry_job():
         if not _has_pull_workspace_access(
@@ -7885,14 +7911,26 @@ async def retry_pull_action_job(
             retry_job(), timeout=ISSUE_ACTION_TIMEOUT_SECONDS
         )
     except HTTPException:
+        try:
+            await asyncio.to_thread(journal.discard, operation_id)
+        except SecurityEventStoreError:
+            pass
         raise
     except gitea_proxy.StalePullError:
+        try:
+            await asyncio.to_thread(journal.discard, operation_id)
+        except SecurityEventStoreError:
+            pass
         return JSONResponse(
             {"error": "New commits arrived. Reload checks before retrying this job."},
             status_code=409,
             headers={"Cache-Control": "no-store"},
         )
     except ValueError:
+        try:
+            await asyncio.to_thread(journal.discard, operation_id)
+        except SecurityEventStoreError:
+            pass
         return JSONResponse(
             {"error": "This check is no longer failed or cannot be retried."},
             status_code=409,
@@ -7904,6 +7942,10 @@ async def retry_pull_action_job(
             status_code=503,
             headers={"Cache-Control": "no-store", "Retry-After": "1"},
         )
+    try:
+        await asyncio.to_thread(journal.finalize, operation_id)
+    except SecurityEventStoreError:
+        pass
     return JSONResponse(
         result, status_code=202, headers={"Cache-Control": "no-store"}
     )
@@ -8359,14 +8401,39 @@ async def release_action_failure(
     "/checks/{run_id}/jobs/{job_index}/retry"
 )
 async def retry_release_action_job(
+    request: Request,
     owner: str,
     repo: str,
     number: int = PathParam(gt=0),
     commit_sha: str = PathParam(min_length=7, max_length=64, pattern=r"^[A-Fa-f0-9]+$"),
     run_id: int = PathParam(gt=0),
     job_index: int = PathParam(ge=0),
+    step_up_grant: str | None = Header(
+        default=None, alias="X-Step-Up-Grant", max_length=128
+    ),
 ) -> JSONResponse:
     repository = f"{owner}/{repo}"
+    target = (
+        f"{repository}#{number}@{commit_sha}:"
+        f"actions/{run_id}/jobs/{job_index}"
+    )
+    await _require_step_up(
+        request, step_up_grant, action="retry_ci_job", target=target
+    )
+    journal = _security_event_store()
+    try:
+        operation_id = await asyncio.to_thread(
+            journal.reserve,
+            "ci_job_retried",
+            principal_id=await _security_principal_id(request),
+            target=target,
+        )
+    except SecurityEventStoreError:
+        return JSONResponse(
+            {"error": "Security activity is temporarily unavailable. No job was retried."},
+            status_code=503,
+            headers={"Cache-Control": "no-store", "Retry-After": "1"},
+        )
 
     async def retry_job():
         if not await gitea_proxy.can_recover_merged_release(
@@ -8382,8 +8449,16 @@ async def retry_release_action_job(
             retry_job(), timeout=ISSUE_ACTION_TIMEOUT_SECONDS
         )
     except HTTPException:
+        try:
+            await asyncio.to_thread(journal.discard, operation_id)
+        except SecurityEventStoreError:
+            pass
         raise
     except ValueError:
+        try:
+            await asyncio.to_thread(journal.discard, operation_id)
+        except SecurityEventStoreError:
+            pass
         return JSONResponse(
             {"error": "This release check is no longer failed or cannot be retried."},
             status_code=409,
@@ -8395,6 +8470,10 @@ async def retry_release_action_job(
             status_code=503,
             headers={"Cache-Control": "no-store", "Retry-After": "1"},
         )
+    try:
+        await asyncio.to_thread(journal.finalize, operation_id)
+    except SecurityEventStoreError:
+        pass
     return JSONResponse(
         result, status_code=202, headers={"Cache-Control": "no-store"}
     )

tests/test_dashboard_auth.py

@@ -18,6 +18,10 @@ def test_release_rollback_is_a_supported_step_up_action():
     assert "prepare_release_rollback" in get_args(main.StepUpAction)
 
 
+def test_ci_job_retry_is_a_supported_step_up_action():
+    assert "retry_ci_job" in get_args(main.StepUpAction)
+
+
 @pytest.fixture
 def access_control(monkeypatch, tmp_path):
     monkeypatch.setenv("STACKCHAIN_DASHBOARD_AUTH_MODE", "operator")
@@ -125,6 +129,158 @@ async def fresh_grant(client, action: str, target: str) -> str:
     return response.json()["grant"]
 
 
+@pytest.mark.anyio
+async def test_ci_job_reruns_require_exact_one_time_authorization_and_audit_both_flows(
+    access_control, monkeypatch
+):
+    lifecycle = []
+
+    class Journal:
+        def record(self, *_args, **_kwargs):
+            pass
+
+        def reserve(self, kind, *, principal_id, target):
+            assert principal_id == 42
+            operation_id = f"operation-{len(lifecycle)}"
+            lifecycle.append(("reserve", kind, target, operation_id))
+            return operation_id
+
+        def finalize(self, operation_id):
+            lifecycle.append(("finalize", operation_id))
+
+        def discard(self, operation_id):
+            lifecycle.append(("discard", operation_id))
+
+    async def capabilities(repository, number):
+        return {"authored": True, "assigned": False}
+
+    async def retry_pull(repository, number, head_sha, run_id, job_index):
+        lifecycle.append(("retry-pull", repository, number, head_sha, run_id, job_index))
+        return {"status": "queued"}
+
+    async def release_access(repository, number, commit_sha):
+        return True
+
+    async def retry_release(repository, commit_sha, run_id, job_index):
+        lifecycle.append(("retry-release", repository, commit_sha, run_id, job_index))
+        return {"status": "queued"}
+
+    monkeypatch.setattr(main, "_security_event_store", lambda: Journal())
+    monkeypatch.setattr(main, "_pull_workspace_capabilities", capabilities)
+    monkeypatch.setattr(main.gitea_proxy, "retry_action_job", retry_pull)
+    monkeypatch.setattr(main.gitea_proxy, "can_recover_merged_release", release_access)
+    monkeypatch.setattr(main.gitea_proxy, "retry_release_action_job", retry_release)
+    transport = httpx.ASGITransport(app=main.app)
+    pull_path = "/api/v1/repos/stackchain/api/pulls/7/checks/91/jobs/3/retry"
+    release_path = (
+        "/api/v1/repos/stackchain/api/pulls/7/release-receipt/abc1234"
+        "/checks/91/jobs/3/retry"
+    )
+    pull_target = "stackchain/api#7@abc1234:actions/91/jobs/3"
+    release_target = "stackchain/api#7@abc1234:actions/91/jobs/3"
+    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"},
+        )
+        assert signed_in.status_code == 200
+        headers = {
+            "Origin": "https://test",
+            "X-CSRF-Token": client.cookies["stackchain_csrf"],
+        }
+
+        missing = await client.post(
+            pull_path, json={"expected_head_sha": "abc1234"}, headers=headers
+        )
+        wrong_grant = await fresh_grant(
+            client, "retry_ci_job", "stackchain/api#7@different:actions/91/jobs/3"
+        )
+        mismatched = await client.post(
+            pull_path,
+            json={"expected_head_sha": "abc1234"},
+            headers={**headers, "X-Step-Up-Grant": wrong_grant},
+        )
+        pull_grant = await fresh_grant(client, "retry_ci_job", pull_target)
+        retried_pull = await client.post(
+            pull_path,
+            json={"expected_head_sha": "abc1234"},
+            headers={**headers, "X-Step-Up-Grant": pull_grant},
+        )
+        replayed = await client.post(
+            pull_path,
+            json={"expected_head_sha": "abc1234"},
+            headers={**headers, "X-Step-Up-Grant": pull_grant},
+        )
+        release_grant = await fresh_grant(client, "retry_ci_job", release_target)
+        retried_release = await client.post(
+            release_path,
+            headers={**headers, "X-Step-Up-Grant": release_grant},
+        )
+
+    assert missing.status_code == 428
+    assert missing.json()["detail"] == {
+        "detail": "Fresh authorization required",
+        "code": "step_up_required",
+        "action": "retry_ci_job",
+        "target": pull_target,
+    }
+    assert mismatched.status_code == 428
+    assert retried_pull.status_code == 202
+    assert replayed.status_code == 428
+    assert retried_release.status_code == 202
+    assert lifecycle == [
+        ("reserve", "ci_job_retried", pull_target, "operation-0"),
+        ("retry-pull", "stackchain/api", 7, "abc1234", 91, 3),
+        ("finalize", "operation-0"),
+        ("reserve", "ci_job_retried", release_target, "operation-3"),
+        ("retry-release", "stackchain/api", "abc1234", 91, 3),
+        ("finalize", "operation-3"),
+    ]
+
+
+@pytest.mark.anyio
+async def test_ci_job_retry_fails_closed_when_security_activity_is_unavailable(
+    access_control, monkeypatch
+):
+    retry_calls = []
+
+    class UnavailableJournal:
+        def reserve(self, *_args, **_kwargs):
+            from src.security_event_store import SecurityEventStoreError
+
+            raise SecurityEventStoreError("unavailable")
+
+    async def retry(*args):
+        retry_calls.append(args)
+        return {"status": "queued"}
+
+    transport = httpx.ASGITransport(app=main.app)
+    path = "/api/v1/repos/stackchain/api/pulls/7/checks/91/jobs/3/retry"
+    target = "stackchain/api#7@abc1234:actions/91/jobs/3"
+    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
+        await client.post(
+            "/api/v1/session", json={"access_token": "correct horse battery staple"}
+        )
+        grant = await fresh_grant(client, "retry_ci_job", target)
+        monkeypatch.setattr(main, "_security_event_store", lambda: UnavailableJournal())
+        monkeypatch.setattr(main.gitea_proxy, "retry_action_job", retry)
+        response = await client.post(
+            path,
+            json={"expected_head_sha": "abc1234"},
+            headers={
+                "Origin": "https://test",
+                "X-CSRF-Token": client.cookies["stackchain_csrf"],
+                "X-Step-Up-Grant": grant,
+            },
+        )
+
+    assert response.status_code == 503
+    assert response.json() == {
+        "error": "Security activity is temporarily unavailable. No job was retried."
+    }
+    assert retry_calls == []
+
+
 @pytest.mark.anyio
 async def test_source_branch_deletion_accepts_exact_one_time_fresh_authorization(
     access_control, monkeypatch

tests/test_security_center.py

@@ -18,6 +18,12 @@ def test_release_rollback_has_a_specific_security_activity_label():
     assert "release_rollback_prepared: 'Release rollback prepared'" in source
 
 
+def test_ci_job_retry_has_a_specific_security_activity_label():
+    source = SECURITY_CENTER.read_text()
+
+    assert "ci_job_retried: 'CI job retried'" in source
+
+
 def test_open_security_center_loads_all_sections_concurrently_and_is_awaitable():
     harness = f"""
 const attachSecurityCenter=require({json.dumps(str(SECURITY_CENTER))});