feat: prepare conflict-safe release rollback pulls (Closes #1370)

src/gitea_proxy.py

@@ -1,4 +1,5 @@
 import asyncio
+import base64
 import os
 import re
 import shlex
@@ -12,6 +13,9 @@ GITEA_URL = os.getenv("GITEA_URL", "http://127.0.0.1:3000").rstrip("/")
 GITEA_TOKEN = os.getenv("GITEA_TOKEN", "")
 REVIEW_DIFF_MAX_BYTES = 64 * 1024
 REVIEW_DIFF_MAX_LINES = 400
+ROLLBACK_MAX_FILES = 20
+ROLLBACK_MAX_FILE_BYTES = 256 * 1024
+ROLLBACK_MAX_TOTAL_BYTES = 1024 * 1024
 AVAILABLE_ISSUE_PAGE_CONCURRENCY = 3
 _client: "GiteaTransport | None" = None
 
@@ -181,6 +185,14 @@ class PullCreateConflictError(ValueError):
     """Raised when the selected source branch changed before pull creation."""
 
 
+class ReleaseRollbackConflictError(ValueError):
+    """Raised when current target content no longer permits an exact rollback."""
+
+
+class ReleaseRollbackUnsupportedError(ValueError):
+    """Raised when a merge cannot be represented as one bounded text rollback."""
+
+
 class SourceBranchChangedError(ValueError):
     """Raised before deletion when a merged source branch has advanced."""
 
@@ -3113,6 +3125,193 @@ async def can_recover_merged_release(
     )
 
 
+async def prepare_release_rollback(
+    repository: str, number: int, commit_sha: str
+) -> dict:
+    """Create one bounded reverse commit and draft pull for an exact merged pull."""
+    login, pull = await _current_login_and_target(
+        f"repos/{repository}/pulls/{number}"
+    )
+    author = pull.get("user") if isinstance(pull.get("user"), dict) else {}
+    if not (
+        pull.get("state") == "closed"
+        and pull.get("merged") is True
+        and pull.get("merge_commit_sha") == commit_sha
+        and (
+            author.get("login", "").casefold() == login.casefold()
+            or _login_in_users(login, pull.get("assignees"))
+        )
+    ):
+        raise IssueNotAvailableError("merged pull request not found")
+
+    access = await repository_access(repository)
+    permissions = access.get("permissions", {}) if isinstance(access, dict) else {}
+    default_branch = access.get("default_branch") if isinstance(access, dict) else None
+    if permissions.get("push") is not True or not isinstance(default_branch, str):
+        raise IssueNotAvailableError("writable repository not found")
+
+    commit_response = await _get_client().get(
+        f"/api/v1/repos/{repository}/git/commits/{commit_sha}", headers=_auth()
+    )
+    commit_response.raise_for_status()
+    merge_commit = commit_response.json()
+    parents = merge_commit.get("parents") if isinstance(merge_commit, dict) else None
+    files = merge_commit.get("files") if isinstance(merge_commit, dict) else None
+    if (
+        merge_commit.get("sha") != commit_sha
+        or not isinstance(parents, list)
+        or len(parents) < 2
+        or not isinstance(parents[0], dict)
+        or not isinstance(parents[0].get("sha"), str)
+        or not isinstance(files, list)
+        or not 1 <= len(files) <= ROLLBACK_MAX_FILES
+    ):
+        raise ReleaseRollbackUnsupportedError("merge is not a bounded merge commit")
+    parent_sha = parents[0]["sha"]
+    branch = f"{login}/rollback-{number}-{commit_sha[:12]}"
+    branch_path = quote(branch, safe="")
+    branch_response = await _get_client().get(
+        f"/api/v1/repos/{repository}/branches/{branch_path}", headers=_auth()
+    )
+
+    if branch_response.status_code == 404:
+        async def content(path: str, ref: str) -> tuple[bytes, str] | None:
+            response = await _get_client().get(
+                f"/api/v1/repos/{repository}/contents/{quote(path, safe='/')}",
+                headers=_auth(),
+                params={"ref": ref},
+            )
+            if response.status_code == 404:
+                return None
+            response.raise_for_status()
+            value = response.json()
+            if (
+                not isinstance(value, dict)
+                or value.get("type") != "file"
+                or value.get("encoding") != "base64"
+                or not isinstance(value.get("content"), str)
+                or not isinstance(value.get("sha"), str)
+            ):
+                raise ReleaseRollbackUnsupportedError("rollback contains a non-file entry")
+            try:
+                raw = base64.b64decode(value["content"], validate=True)
+                raw.decode("utf-8")
+            except (ValueError, UnicodeDecodeError) as exc:
+                raise ReleaseRollbackUnsupportedError(
+                    "rollback contains binary or invalid content"
+                ) from exc
+            if len(raw) > ROLLBACK_MAX_FILE_BYTES:
+                raise ReleaseRollbackUnsupportedError("rollback file is too large")
+            return raw, value["sha"]
+
+        operations = []
+        total_bytes = 0
+        for changed in files:
+            filename = changed.get("filename") if isinstance(changed, dict) else None
+            status = changed.get("status") if isinstance(changed, dict) else None
+            if (
+                not isinstance(filename, str)
+                or not filename
+                or filename.startswith("/")
+                or ".." in filename.split("/")
+                or status not in {"added", "modified", "removed"}
+            ):
+                raise ReleaseRollbackUnsupportedError("rollback contains an unsupported change")
+            merged = await content(filename, commit_sha)
+            parent = await content(filename, parent_sha)
+            current = await content(filename, default_branch)
+            if (
+                (current is None) != (merged is None)
+                or (
+                    current is not None
+                    and merged is not None
+                    and current[0] != merged[0]
+                )
+            ):
+                raise ReleaseRollbackConflictError(
+                    f"{filename} changed after the failed release"
+                )
+            if status == "added" and merged is not None and parent is None:
+                operations.append({
+                    "operation": "delete", "path": filename, "sha": current[1],
+                })
+            elif status == "removed" and merged is None and parent is not None:
+                operations.append({
+                    "operation": "create", "path": filename,
+                    "content": base64.b64encode(parent[0]).decode(),
+                })
+                total_bytes += len(parent[0])
+            elif status == "modified" and merged is not None and parent is not None:
+                operations.append({
+                    "operation": "update", "path": filename, "sha": current[1],
+                    "content": base64.b64encode(parent[0]).decode(),
+                })
+                total_bytes += len(parent[0])
+            else:
+                raise ReleaseRollbackUnsupportedError("change metadata did not match content")
+            if total_bytes > ROLLBACK_MAX_TOTAL_BYTES:
+                raise ReleaseRollbackUnsupportedError("rollback content is too large")
+
+        mutation = await _get_client().post(
+            f"/api/v1/repos/{repository}/contents",
+            headers=_auth(),
+            json={
+                "branch": default_branch,
+                "new_branch": branch,
+                "message": f"Revert {commit_sha} from pull #{number}",
+                "files": operations,
+            },
+        )
+        if mutation.status_code in {409, 422}:
+            reconciled = await _get_client().get(
+                f"/api/v1/repos/{repository}/branches/{branch_path}", headers=_auth()
+            )
+            reconciled.raise_for_status()
+            branch_payload = reconciled.json()
+            branch_commit = (
+                branch_payload.get("commit") if isinstance(branch_payload, dict) else None
+            )
+            rollback_sha = (
+                branch_commit.get("id") if isinstance(branch_commit, dict) else None
+            )
+            if not isinstance(rollback_sha, str):
+                raise ValueError("Gitea did not confirm the concurrent rollback branch")
+        else:
+            mutation.raise_for_status()
+            payload = mutation.json()
+            created_commit = payload.get("commit") if isinstance(payload, dict) else None
+            rollback_sha = created_commit.get("sha") if isinstance(created_commit, dict) else None
+            if not isinstance(rollback_sha, str):
+                raise ValueError("Gitea did not confirm rollback commit creation")
+    else:
+        branch_response.raise_for_status()
+        branch_payload = branch_response.json()
+        branch_commit = branch_payload.get("commit") if isinstance(branch_payload, dict) else None
+        rollback_sha = branch_commit.get("id") if isinstance(branch_commit, dict) else None
+        if not isinstance(rollback_sha, str):
+            raise ValueError("Gitea did not confirm the rollback branch")
+
+    title = f"Rollback #{number}: {pull.get('title', 'failed release')}"
+    body = (
+        f"Rollback of #{number} at `{commit_sha}` after failed release checks.\n\n"
+        "This draft reverses only files that still matched the failed merge."
+    )
+    result = await create_pull(
+        repository,
+        head=branch,
+        base=default_branch,
+        title=title,
+        body=body,
+        draft=True,
+        expected_head_sha=rollback_sha,
+    )
+    return {
+        **result,
+        "rollback_of": commit_sha,
+        "files_changed": len(files),
+    }
+
+
 async def pull_workspace_capabilities(repository: str, number: int) -> dict[str, bool]:
     login, pull = await _current_login_and_target(
         f"repos/{repository}/pulls/{number}"

src/main.py

@@ -498,6 +498,7 @@ class FollowingNotificationPayload(BaseModel):
 StepUpAction = Literal[
     "merge_pull",
     "delete_source_branch",
+    "prepare_release_rollback",
     "submit_pull_review",
     "close_issue",
     "delete_comment",
@@ -7588,6 +7589,103 @@ async def retry_release_action_job(
     )
 
 
+@app.post(
+    "/api/v1/repos/{owner}/{repo}/pulls/{number}/release-receipt/{commit_sha}/rollback"
+)
+async def prepare_release_rollback(
+    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]+$"
+    ),
+    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}"
+    await _require_step_up(
+        request, step_up_grant, action="prepare_release_rollback", target=target
+    )
+    try:
+        status = await asyncio.wait_for(
+            gitea_proxy.release_receipt_status(repository, commit_sha),
+            timeout=REVIEW_DETAIL_TIMEOUT_SECONDS,
+        )
+    except Exception:
+        return JSONResponse(
+            {"error": "The failed release could not be revalidated. No rollback was created."},
+            status_code=503,
+            headers={"Cache-Control": "no-store", "Retry-After": "1"},
+        )
+    if (
+        status.get("commit_sha") != commit_sha
+        or status.get("ci_state") != "failure"
+        or status.get("release") is not None
+    ):
+        return JSONResponse(
+            {"error": "Only an unreleased commit with failed checks can be rolled back."},
+            status_code=409,
+            headers={"Cache-Control": "no-store"},
+        )
+
+    journal = _security_event_store()
+    try:
+        operation_id = await asyncio.to_thread(
+            journal.reserve, "release_rollback_prepared", target=target
+        )
+    except SecurityEventStoreError:
+        return JSONResponse(
+            {"error": "Security activity is temporarily unavailable. No rollback was created."},
+            status_code=503,
+            headers={"Cache-Control": "no-store", "Retry-After": "1"},
+        )
+    try:
+        result = await asyncio.wait_for(
+            gitea_proxy.prepare_release_rollback(repository, number, commit_sha),
+            timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
+        )
+        try:
+            await asyncio.to_thread(journal.finalize, operation_id)
+        except SecurityEventStoreError:
+            pass
+        return JSONResponse(
+            result,
+            status_code=200 if result.get("existing") else 201,
+            headers={"Cache-Control": "no-store"},
+        )
+    except gitea_proxy.ReleaseRollbackConflictError as exc:
+        try:
+            await asyncio.to_thread(journal.discard, operation_id)
+        except SecurityEventStoreError:
+            pass
+        return JSONResponse(
+            {"error": str(exc)}, status_code=409, headers={"Cache-Control": "no-store"}
+        )
+    except gitea_proxy.ReleaseRollbackUnsupportedError as exc:
+        try:
+            await asyncio.to_thread(journal.discard, operation_id)
+        except SecurityEventStoreError:
+            pass
+        return JSONResponse(
+            {"error": str(exc)}, status_code=422, headers={"Cache-Control": "no-store"}
+        )
+    except gitea_proxy.IssueNotAvailableError:
+        try:
+            await asyncio.to_thread(journal.discard, operation_id)
+        except SecurityEventStoreError:
+            pass
+        raise HTTPException(status_code=404, detail="Merged pull request not found")
+    except Exception:
+        return JSONResponse(
+            {"error": "Rollback preparation could not be confirmed. Retry to reconcile it."},
+            status_code=503,
+            headers={"Cache-Control": "no-store", "Retry-After": "1"},
+        )
+
+
 @app.post("/api/v1/repos/{owner}/{repo}/pulls/{number}/review", status_code=201)
 async def submit_review(
     submission: PullReviewSubmission,