autorevert: only honor disable killswitch when label applied by a write-access user (#8229)

## Summary

`check_autorevert_disabled()` treated **any** open issue carrying the
`ci: disable-autorevert` label as a global disable, with no authority
check. Because GitHub applies an issue template's labels on the author's
behalf at creation time **regardless of the author's permissions**, a
user with no write access could disable autorevert for the entire repo
just by opening the disable-autorevert issue.

This happened in production: pytorch/pytorch#188383 was opened by a
`NONE`-permission user from the `[DISABLE AUTOREVERT]` template, which
auto-applied the label and kept autorevert disabled for ~15h until a
maintainer stripped the label.

## Change

Gate the killswitch on **who applied the label**, not the issue author:
- Find the most recent `labeled` event for `ci: disable-autorevert` (via
`issue.get_events()`, latest wins so a remove-then-re-add reflects
whoever re-applied it).
- Require that actor to hold `write` / `maintain` / `admin` permission.
`triage` is intentionally excluded — triagers may manage labels but
shouldn't be able to globally disable autorevert.
- Checking the applier rather than the author means a maintainer
labeling *someone else's* issue is still correctly honored (the common
triage workflow).
- **Fail safe:** if the applier or their permission can't be resolved,
skip that issue (autorevert stays ON) and keep evaluating the rest —
consistent with the existing "on error, allow autorevert to continue"
behavior.

Adds `test_autorevert_circuit_breaker.py` covering: the exploit case
(unprivileged applier → not disabled), maintainer-labels-foreign-issue →
disabled, latest-applier-wins after relabel, `triage` rejected,
no-labeled-event anomaly, and the fail-safe / keep-evaluating paths (12
tests).

## Companion PR

The matching defense at the template layer (stop auto-applying the label
so only a write/triage user can ever add it) is in **pytorch/pytorch**:
see the `disable-autorevert` issue-template change. This lambda gate is
the durable guarantee independent of any one repo's template config.

Ref: pytorch/pytorch#188383

Signed-off-by: Ivan Zaitsev <izaitsevfb@meta.com>
Co-authored-by: Ivan Zaitsev <izaitsevfb@meta.com>

Author: izaitsevfb

aws/lambda/pytorch-auto-revert/pytorch_auto_revert/autorevert_circuit_breaker.py

@@ -6,10 +6,53 @@
 
 logger = logging.getLogger(__name__)
 
+# The label whose presence on an open issue disables autorevert.
+DISABLE_AUTOREVERT_LABEL = "ci: disable-autorevert"
+
+# Repository permission levels that authorize disabling autorevert. Adding a
+# label to an issue requires triage/write access, but GitHub applies an issue
+# template's labels on the author's behalf at creation time regardless of the
+# author's permissions, so label presence alone is NOT proof of authority. We
+# require the user who APPLIED the label to hold write access. "triage" is
+# intentionally excluded: triagers may manage labels but should not be able to
+# globally disable autorevert.
+_AUTHORIZED_PERMISSIONS = frozenset({"admin", "write", "maintain"})
+
+
+def _label_applier_login(issue, label_name: str):
+    """Return the login of the user who most recently applied ``label_name`` to
+    ``issue``, or ``None`` if it cannot be determined.
+
+    The label *applier* — not the issue author — is the actor whose authority
+    gates the killswitch. For a template-auto-applied label, GitHub records the
+    "labeled" event with the issue author as the actor, so an unprivileged
+    author who tripped the label via the template is correctly rejected here;
+    while a maintainer who labels someone else's issue is correctly honored.
+    """
+    applier = None
+    for ev in issue.get_events():
+        if (
+            ev.event == "labeled"
+            and ev.label is not None
+            and ev.label.name == label_name
+        ):
+            # get_events() is chronological; keep the latest applier so a
+            # remove-then-re-add reflects whoever re-applied the label.
+            applier = ev.actor.login if ev.actor is not None else None
+    return applier
+
 
 def check_autorevert_disabled(repo_full_name: str = "pytorch/pytorch") -> bool:
     """
-    Check if autorevert is disabled by looking for open issues with 'ci: disable-autorevert' label.
+    Check if autorevert is disabled by looking for open issues with the
+    'ci: disable-autorevert' label that was applied by a user with write access.
+
+    The label alone is not sufficient authority: a user without write
+    permissions can still get the label onto an issue (e.g. via the issue
+    template, whose labels GitHub applies on the author's behalf at creation
+    time). To prevent an unprivileged user from disabling autorevert for the
+    whole repo, we additionally require the user who applied the label to have
+    write/maintain/admin permission on the repository.
 
     Args:
         repo_full_name: Repository name in format 'owner/repo'
@@ -23,26 +66,58 @@ def check_autorevert_disabled(repo_full_name: str = "pytorch/pytorch") -> bool:
                 gh_client = GHClientFactory().client
                 repo = gh_client.get_repo(repo_full_name)
 
-                should_disable = False
-
                 # Search for open issues with the specific label
                 disable_issues = repo.get_issues(
-                    state="open", labels=["ci: disable-autorevert"]
+                    state="open", labels=[DISABLE_AUTOREVERT_LABEL]
                 )
 
                 for issue in disable_issues:
+                    try:
+                        applier = _label_applier_login(issue, DISABLE_AUTOREVERT_LABEL)
+                        permission = (
+                            repo.get_collaborator_permission(applier)
+                            if applier is not None
+                            else None
+                        )
+                    except Exception as e:
+                        # Fail safe: if we cannot positively confirm the label
+                        # was applied by a write-access user, do NOT disable
+                        # autorevert. Skip this issue, keep evaluating the rest.
+                        logger.warning(
+                            f"Could not resolve the '{DISABLE_AUTOREVERT_LABEL}' "
+                            f"applier permission on issue #{issue.number}: {e}. "
+                            f"Ignoring this issue for the autorevert circuit breaker."
+                        )
+                        continue
+
+                    if applier is None:
+                        logger.warning(
+                            f"Could not determine who applied "
+                            f"'{DISABLE_AUTOREVERT_LABEL}' to issue #{issue.number}; "
+                            f"ignoring it for the autorevert circuit breaker."
+                        )
+                        continue
+
+                    if permission not in _AUTHORIZED_PERMISSIONS:
+                        logger.warning(
+                            f"Ignoring open issue #{issue.number}: "
+                            f"'{DISABLE_AUTOREVERT_LABEL}' was applied by {applier} "
+                            f"with '{permission}' permission (write access required "
+                            f"to disable autorevert)."
+                        )
+                        continue
+
                     logger.info(
-                        f"Found open issue #{issue.number} with 'ci: disable-autorevert' label "
-                        f"created by user {issue.user.login}. "
+                        f"Found open issue #{issue.number} with "
+                        f"'{DISABLE_AUTOREVERT_LABEL}' applied by {applier} "
+                        f"('{permission}' permission). "
                         f"Autorevert circuit breaker is ACTIVE."
                     )
-                    should_disable = True
-
-                if should_disable:
                     return True
 
                 logger.debug(
-                    "No open issues with 'ci: disable-autorevert' label found."
+                    f"No open issues with '{DISABLE_AUTOREVERT_LABEL}' applied by a "
+                    f"write-access user found."
                 )
                 return False
 

aws/lambda/pytorch-auto-revert/pytorch_auto_revert/tests/test_autorevert_circuit_breaker.py

@@ -0,0 +1,179 @@
+import sys
+import unittest
+from unittest.mock import MagicMock, patch
+
+
+# Ensure package import when running from repo root
+sys.path.insert(0, "aws/lambda/pytorch-auto-revert")
+
+from pytorch_auto_revert import autorevert_circuit_breaker  # noqa: E402
+from pytorch_auto_revert.autorevert_circuit_breaker import (  # noqa: E402
+    check_autorevert_disabled,
+    DISABLE_AUTOREVERT_LABEL,
+)
+
+
+def _event(event_type, label_name=None, actor_login=None):
+    ev = MagicMock()
+    ev.event = event_type
+    if label_name is None:
+        ev.label = None
+    else:
+        ev.label = MagicMock()
+        ev.label.name = label_name
+    if actor_login is None:
+        ev.actor = None
+    else:
+        ev.actor = MagicMock()
+        ev.actor.login = actor_login
+    return ev
+
+
+def _make_issue(number, events):
+    """events: list of (event_type, label_name|None, actor_login|None)."""
+    issue = MagicMock()
+    issue.number = number
+    issue.user = MagicMock()
+    issue.user.login = "issue-author"
+    issue.get_events.return_value = [_event(*e) for e in events]
+    return issue
+
+
+def _labeled_issue(number, applier_login):
+    """An issue whose disable-autorevert label was applied by applier_login."""
+    return _make_issue(number, [("labeled", DISABLE_AUTOREVERT_LABEL, applier_login)])
+
+
+class TestCheckAutorevertDisabled(unittest.TestCase):
+    def _run_with(self, issues, permission_map, perm_side_effect=None):
+        """Run check_autorevert_disabled against a mocked GitHub repo.
+
+        issues: list of mock issues returned by repo.get_issues
+        permission_map: dict applier-login -> permission string
+        perm_side_effect: optional callable(login) used instead of the map
+        """
+        repo = MagicMock()
+        repo.get_issues.return_value = issues
+
+        if perm_side_effect is not None:
+            repo.get_collaborator_permission.side_effect = perm_side_effect
+        else:
+            repo.get_collaborator_permission.side_effect = lambda login: permission_map[
+                login
+            ]
+
+        client = MagicMock()
+        client.get_repo.return_value = repo
+
+        factory = MagicMock()
+        factory.client = client
+
+        with patch.object(
+            autorevert_circuit_breaker, "GHClientFactory", return_value=factory
+        ):
+            result = check_autorevert_disabled("pytorch/pytorch")
+        return result, repo
+
+    def test_no_issues_returns_false(self):
+        result, repo = self._run_with([], {})
+        self.assertFalse(result)
+        repo.get_collaborator_permission.assert_not_called()
+
+    def test_write_access_applier_disables(self):
+        result, _ = self._run_with(
+            [_labeled_issue(1, "maintainer")], {"maintainer": "write"}
+        )
+        self.assertTrue(result)
+
+    def test_admin_and_maintain_appliers_disable(self):
+        for perm in ("admin", "maintain"):
+            with self.subTest(perm=perm):
+                result, _ = self._run_with([_labeled_issue(1, "boss")], {"boss": perm})
+                self.assertTrue(result)
+
+    def test_unprivileged_applier_does_not_disable(self):
+        # The exploit case: a NONE/read user trips the label via the template,
+        # so the "labeled" event's actor is that unprivileged user.
+        for perm in ("none", "read"):
+            with self.subTest(perm=perm):
+                result, _ = self._run_with(
+                    [_labeled_issue(188383, "shameelvk9-png")],
+                    {"shameelvk9-png": perm},
+                )
+                self.assertFalse(result)
+
+    def test_triage_applier_does_not_disable(self):
+        # Triagers can manage labels but must not be able to disable autorevert.
+        result, _ = self._run_with(
+            [_labeled_issue(1, "triager")], {"triager": "triage"}
+        )
+        self.assertFalse(result)
+
+    def test_maintainer_labeling_unprivileged_authored_issue_disables(self):
+        # Author is unprivileged; a maintainer applied the label. This must be
+        # honored (the author-permission approach would wrongly ignore it).
+        issue = _make_issue(1, [("labeled", DISABLE_AUTOREVERT_LABEL, "maintainer")])
+        issue.user.login = "random-contributor"
+        result, _ = self._run_with([issue], {"maintainer": "write"})
+        self.assertTrue(result)
+
+    def test_latest_applier_wins_after_relabel(self):
+        # Removed then re-applied by an unprivileged user -> latest applier (none) governs.
+        issue = _make_issue(
+            1,
+            [
+                ("labeled", DISABLE_AUTOREVERT_LABEL, "maintainer"),
+                ("unlabeled", DISABLE_AUTOREVERT_LABEL, "maintainer"),
+                ("labeled", DISABLE_AUTOREVERT_LABEL, "rando"),
+            ],
+        )
+        result, _ = self._run_with([issue], {"maintainer": "write", "rando": "none"})
+        self.assertFalse(result)
+
+    def test_no_labeled_event_is_ignored(self):
+        # Label present but no "labeled" event for it (anomalous) -> fail safe.
+        issue = _make_issue(
+            1, [("closed", None, "someone"), ("assigned", None, "someone")]
+        )
+        result, repo = self._run_with([issue], {})
+        self.assertFalse(result)
+        repo.get_collaborator_permission.assert_not_called()
+
+    def test_unprivileged_then_privileged_keeps_evaluating(self):
+        result, _ = self._run_with(
+            [_labeled_issue(1, "rando"), _labeled_issue(2, "maintainer")],
+            {"rando": "none", "maintainer": "write"},
+        )
+        self.assertTrue(result)
+
+    def test_get_events_error_is_skipped_fail_safe(self):
+        issue = _labeled_issue(1, "maintainer")
+        issue.get_events.side_effect = RuntimeError("github api hiccup")
+        result, _ = self._run_with([issue], {"maintainer": "write"})
+        self.assertFalse(result)
+
+    def test_permission_lookup_error_is_skipped_fail_safe(self):
+        def boom(login):
+            raise RuntimeError("github api hiccup")
+
+        result, _ = self._run_with(
+            [_labeled_issue(1, "rando")], {}, perm_side_effect=boom
+        )
+        self.assertFalse(result)
+
+    def test_permission_error_on_one_issue_does_not_block_authorized_one(self):
+        def perm(login):
+            if login == "rando":
+                raise RuntimeError("github api hiccup")
+            return "admin"
+
+        result, _ = self._run_with(
+            [_labeled_issue(1, "rando"), _labeled_issue(2, "maintainer")],
+            {},
+            perm_side_effect=perm,
+        )
+        self.assertTrue(result)
+
+
+if __name__ == "__main__":
+    unittest.main()