This is an automated email from the ASF dual-hosted git repository. potiuk pushed a commit to branch fork-ci-6-reconciler-followups in repository https://gitbox.apache.org/repos/asf/airflow.git
commit bb9616cd47ee3bccfd675abefd7af8f86a5894e5 Author: Jarek Potiuk <[email protected]> AuthorDate: Sat Oct 17 03:01:39 2026 +0200 Attribute fork CI runs to the right pull request in the tampering check Runs from forks carry no pull request, so they were matched by fork branch alone. Legitimate project CI against a base branch without the gate, made before a retarget, or a run of another pull request from the same branch, could close a gated pull request for gate tampering. Generated-by: Claude Opus 5 --- .../doc/ci/08_fork_ci_for_external_contributors.md | 9 ++- .../src/airflow_breeze/utils/fork_ci_reconcile.py | 82 +++++++++++++++++-- dev/breeze/tests/test_fork_ci_reconcile.py | 93 +++++++++++++++++++++- 3 files changed, 174 insertions(+), 10 deletions(-) diff --git a/dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md b/dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md index 0820e007ebf..d5703324a12 100644 --- a/dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md +++ b/dev/breeze/doc/ci/08_fork_ci_for_external_contributors.md @@ -509,8 +509,13 @@ ignored, because they ran legitimately while the label was on. So are runs creat most recent reopen of the pull request by someone other than its author, which is how a maintainer brings back a pull request closed for tampering; reopening one whose gate is still missing triggers a new run, which is caught again. A reopen by the author does not count, so -closing and reopening one's own pull request cannot hide a tampered run. A run counts as a -bypass when both signals hold: +closing and reopening one's own pull request cannot hide a tampered run. Runs are listed per +fork branch, because GitHub records no pull request on runs from forks. A run made while the +pull request targeted a base branch without the gate, known from its retargets in the +timeline, is project CI by design and ignored; its changes are compared with that base. When +another pull request from the same branch was open within the window, a matching run could be +that one's, so it is reported to Slack as an attention item rather than closing the pull +request. A run counts as a bypass when both signals hold: 1. **The run did project CI work.** Either the default branch's version of the run's workflow is not triggered by the run's event, which covers an injected workflow and a trigger the diff --git a/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py b/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py index faef461b4a8..81f71b19cce 100644 --- a/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py +++ b/dev/breeze/src/airflow_breeze/utils/fork_ci_reconcile.py @@ -168,6 +168,8 @@ class PullRequestSnapshot: last_reopened_at: datetime | None = None last_reopened_by_anyone_at: datetime | None = None last_base_change_at: datetime | None = None + # (changed at, previous base, new base) for each retarget, oldest first. + base_changes: tuple[tuple[datetime, str, str], ...] = () current_status: tuple[str, str, str] | None = None status_created_at: datetime | None = None marker: Marker | None = None @@ -557,7 +559,7 @@ query($q: String!, $cursor: String) { ... on ConvertToDraftEvent { createdAt actor { login } } ... on UnlabeledEvent { createdAt label { name } } ... on ReopenedEvent { createdAt actor { login } } - ... on BaseRefChangedEvent { createdAt } + ... on BaseRefChangedEvent { createdAt previousRefName currentRefName } } } commits(last: 1) { nodes { commit { @@ -609,11 +611,15 @@ def parse_pull_request(node: dict[str, Any]) -> PullRequestSnapshot: last_reopened = None last_reopened_by_anyone = None last_base_change = None + base_changes = [] for item in (node.get("timelineItems") or {}).get("nodes") or []: if item["__typename"] == "ConvertToDraftEvent": last_draft_actor = _login(item.get("actor")) elif item["__typename"] == "BaseRefChangedEvent": last_base_change = parse_github_datetime(item["createdAt"]) + base_changes.append( + (last_base_change, item.get("previousRefName") or "", item.get("currentRefName") or "") + ) elif item["__typename"] == "ReopenedEvent": last_reopened_by_anyone = parse_github_datetime(item["createdAt"]) # Only a reopen by someone else counts for the gate tampering check: an author closing @@ -665,6 +671,7 @@ def parse_pull_request(node: dict[str, Any]) -> PullRequestSnapshot: last_reopened_at=last_reopened, last_reopened_by_anyone_at=last_reopened_by_anyone, last_base_change_at=last_base_change, + base_changes=tuple(base_changes), current_status=current_status, status_created_at=status_created_at, marker=marker, @@ -870,6 +877,15 @@ def load_workflow_triggers(workflows_dir: Path) -> dict[str, frozenset[str]]: return triggers +def base_at(pr: PullRequestSnapshot, when: datetime) -> str: + """The base branch the pull request had at ``when``, from its retargets.""" + base = pr.base_ref + for changed_at, previous, _ in reversed(pr.base_changes): + if when < changed_at: + base = previous + return base + + class BypassFinder: """ Finds project CI runs that bypassed the gate, from the runs apache/airflow recorded. @@ -885,12 +901,23 @@ class BypassFinder: """ def __init__( - self, client: GitHubClient, *, repo: str, now: datetime, base_triggers: dict[str, frozenset[str]] + self, + client: GitHubClient, + *, + repo: str, + now: datetime, + base_triggers: dict[str, frozenset[str]], + is_gated_base: Callable[[str], bool] = lambda base_ref: True, ): + """:param is_gated_base: whether a base branch has the gate; runs against one without it are project CI by design.""" self._client = client self._repo = repo self._now = now self._base_triggers = base_triggers + self._is_gated_base = is_gated_base + # Pull requests whose matching run could belong to another pull request from the same + # branch; reported instead of closed. + self.ambiguous: list[tuple[int, str]] = [] since = (now - TAMPER_WINDOW).strftime("%Y-%m-%dT%H:%M:%SZ") self._runs_by_branch: dict[tuple[str, str], list[WorkflowRun]] = {} incomplete = [] @@ -961,11 +988,26 @@ class BypassFinder: self._changes_workflows[key] = None if len(files) >= COMPARE_FILES_CAP else False return self._changes_workflows[key] + def _other_pull_requests(self, pr: PullRequestSnapshot) -> list[int]: + """Other pull requests from the same branch that were open within the tampering window.""" + owner = pr.head_repo.split("/", 1)[0] # type: ignore[union-attr] + pulls = self._client.rest_get( + f"repos/{self._repo}/pulls?head={quote(owner, safe='')}:{quote(pr.head_ref, safe='')}" + "&state=all&per_page=100", + paginate=True, + ) + window_start = self._now - TAMPER_WINDOW + return [ + pull["number"] + for pull in pulls + if pull["number"] != pr.number + and (pull.get("closed_at") is None or parse_github_datetime(pull["closed_at"]) >= window_start) + ] + def find(self, pr: PullRequestSnapshot) -> str | None: if not pr.head_repo: return None - # Matching by fork branch alone also picks up runs of another pull request from that branch - # or from before a retarget; tracked at https://github.com/apache/airflow/issues/74906 + # Runs are listed per fork branch: GitHub records no pull request on runs from forks. runs = self._runs_by_branch.get((pr.head_repo.casefold(), pr.head_ref), []) # Runs before `use project ci` was removed were legitimate. Runs before a maintainer # reopened the pull request were already acted on; reopening without restoring the gate @@ -976,11 +1018,26 @@ class BypassFinder: for run in sorted(runs, key=lambda r: r.created_at, reverse=True): if not_before is not None and run.created_at <= not_before: continue + base_ref = base_at(pr, run.created_at) + # Project CI against a base branch without the gate is what its gate job decides. + if not self._is_gated_base(base_ref): + continue if not self._did_run_project_ci(run): continue # A change set too large to list fails closed: padding a tampered commit past the # compare API's cap must not hide the workflow change. - if self._commit_changes_workflows(run, pr.base_ref) is not False: + if self._commit_changes_workflows(run, base_ref) is not False: + others = self._other_pull_requests(pr) + if others: + self.ambiguous.append( + ( + pr.number, + f"project CI run {run.html_url} may have bypassed the gate, but it could also " + f"belong to {', '.join(f'#{n}' for n in others)} from the same branch; not closed, " + "check it", + ) + ) + return None return run.html_url return None @@ -1144,8 +1201,20 @@ def run_reconcile( ) ] observations = fetch_fork_observations(client, prs) + + def is_gated_base(base_ref: str) -> bool: + # A base branch whose configuration cannot be read is treated as gated. + try: + return configs.get(base_ref) is not None + except (GitHubError, ValueError): + return True + bypass_finder = BypassFinder( - client, repo=repo, now=now, base_triggers=load_workflow_triggers(workflows_dir) + client, + repo=repo, + now=now, + base_triggers=load_workflow_triggers(workflows_dir), + is_gated_base=is_gated_base, ) shared_heads = find_heads_shared_across_bases(prs) @@ -1178,6 +1247,7 @@ def run_reconcile( console_print(f"[error]#{listed.number}: {e}[/]") unknown.append((listed.number, f"{type(e).__name__}: {e}")) + attention.extend(bypass_finder.ambiguous) notes = [bypass_finder.incomplete] if bypass_finder.incomplete else [] # The full list for the run log; Slack may show only part of a very long one. for number, reason in [*unknown, *attention]: diff --git a/dev/breeze/tests/test_fork_ci_reconcile.py b/dev/breeze/tests/test_fork_ci_reconcile.py index e5b28b13572..c01d28c4ab5 100644 --- a/dev/breeze/tests/test_fork_ci_reconcile.py +++ b/dev/breeze/tests/test_fork_ci_reconcile.py @@ -651,6 +651,8 @@ def _finder( jobs=(), compare_files=(".github/workflows/ci-amd.yml",), total_count=None, + other_pulls=(), + is_gated_base=lambda base_ref: True, ): def rest_get_pages(path): event = path.split("event=", 1)[1].split("&", 1)[0] @@ -663,11 +665,13 @@ def _finder( return list(jobs) if "/compare/" in path: return [{"filename": name} for name in compare_files] + if "/pulls?head=" in path: + return [{"number": 42, "closed_at": None}, *other_pulls] raise AssertionError(path) client.rest_get_pages.side_effect = rest_get_pages client.rest_get.side_effect = rest_get - return BypassFinder(client, repo=REPO, now=NOW, base_triggers=BASE_TRIGGERS) + return BypassFinder(client, repo=REPO, now=NOW, base_triggers=BASE_TRIGGERS, is_gated_base=is_gated_base) @pytest.fixture @@ -782,11 +786,44 @@ def test_finder_fails_closed_on_a_change_set_it_cannot_check(client): assert finder.find(_pr()) == "https://github.com/apache/airflow/actions/runs/7"; +def test_finder_ignores_runs_made_while_the_base_had_no_gate(client): + finder = _finder( + client, [_run(7)], jobs=TAMPERED_JOBS, is_gated_base=lambda base_ref: base_ref != "v2-10-test" + ) + pr = _pr(base_changes=((NOW - timedelta(minutes=30), "v2-10-test", "main"),)) + + assert finder.find(pr) is None + + [email protected]( + "other_pulls", + [ + pytest.param([{"number": 43, "closed_at": None}], id="open"), + pytest.param( + [{"number": 43, "closed_at": (NOW - timedelta(hours=2)).isoformat()}], id="closed-recently" + ), + ], +) +def test_finder_reports_a_run_another_pull_request_may_own(client, other_pulls): + finder = _finder(client, [_run(7)], jobs=TAMPERED_JOBS, other_pulls=other_pulls) + + assert finder.find(_pr()) is None + assert finder.ambiguous == [ + ( + 42, + "project CI run https://github.com/apache/airflow/actions/runs/7 may have bypassed the gate, " + "but it could also belong to #43 from the same branch; not closed, check it", + ) + ] + + def test_finder_catches_a_workflow_moved_away(client): finder = _finder(client, [_run(7)], jobs=TAMPERED_JOBS) client.rest_get.side_effect = lambda path, paginate=False, items_key=None: ( TAMPERED_JOBS if path.endswith("jobs?filter=latest&per_page=100") + else [{"number": 42, "closed_at": None}] + if "/pulls?head=" in path else [{"filename": "docs/x.yml", "previous_filename": ".github/workflows/x.yml"}] ) @@ -867,7 +904,12 @@ def test_parse_pull_request_reads_timeline_status_and_marker(): "createdAt": "2026-11-06T00:00:00Z", "actor": {"login": "someone"}, }, - {"__typename": "BaseRefChangedEvent", "createdAt": "2026-11-07T00:00:00Z"}, + { + "__typename": "BaseRefChangedEvent", + "createdAt": "2026-11-07T00:00:00Z", + "previousRefName": "v3-1-test", + "currentRefName": "main", + }, ] }, commits={ @@ -906,6 +948,7 @@ def test_parse_pull_request_reads_timeline_status_and_marker(): assert pr.last_reopened_at == datetime(2026, 11, 5, tzinfo=UTC) assert pr.last_reopened_by_anyone_at == datetime(2026, 11, 6, tzinfo=UTC) assert pr.last_base_change_at == datetime(2026, 11, 7, tzinfo=UTC) + assert pr.base_changes == ((datetime(2026, 11, 7, tzinfo=UTC), "v3-1-test", "main"),) assert pr.current_status == ("pending", "u", "d") assert pr.status_created_at == datetime(2026, 11, 8, tzinfo=UTC) assert pr.marker == marker @@ -1234,3 +1277,49 @@ def test_finder_ignores_an_added_workflow_whose_jobs_all_skipped(client): finder = _finder(client, [run], jobs=[_job("Fork only"), _job("Also fork only")]) assert finder.find(_pr()) is None + + [email protected]("airflow_breeze.utils.fork_ci_reconcile.BypassFinder", autospec=True) +def test_run_reconcile_reports_ambiguous_runs(mock_finder, client, tmp_path): + output = tmp_path / "slack-message.json" + client.graphql.side_effect = [_search_result(), _search_result(), _search_result()] + mock_finder.return_value.incomplete = None + mock_finder.return_value.ambiguous = [(42, "could also belong to #43")] + + has_alerts = run_reconcile( + client, CONFIG, now=NOW, run_url="https://run";, output_file=output, workflows_dir=tmp_path + ) + + assert has_alerts is True + assert "|#42>: could also belong to #43" in output.read_text() + + [email protected]( + ("base_ref", "rest_get", "expected"), + [ + pytest.param("main", None, True, id="default-branch"), + pytest.param("v2-10-test", _contents({}), False, id="no-gate"), + pytest.param("v3-1-test", GitHubError("gh api failed: HTTP 500"), True, id="unreadable"), + pytest.param( + "v3-1-test", _contents({(GATE_FILES[1], "v3-1-test"): "# gate\n"}), True, id="half-gate" + ), + ], +) [email protected]("airflow_breeze.utils.fork_ci_reconcile.BypassFinder", autospec=True) +def test_run_reconcile_tells_the_finder_which_bases_are_gated( + mock_finder, client, tmp_path, base_ref, rest_get, expected +): + client.graphql.side_effect = [_search_result(), _search_result(), _search_result()] + mock_finder.return_value.incomplete = None + mock_finder.return_value.ambiguous = [] + run_reconcile( + client, + CONFIG, + now=NOW, + run_url="https://run";, + output_file=tmp_path / "s.json", + workflows_dir=tmp_path, + ) + client.rest_get.side_effect = rest_get + + assert mock_finder.call_args.kwargs["is_gated_base"](base_ref) is expected
