diff --git a/docs/cli.md b/docs/cli.md index da2366f..7e1eb8e 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -165,6 +165,7 @@ Some settings have no CLI flag and use environment variables: | `G2G_LOG_LEVEL` | `WARNING` | Logging level; set `DEBUG` for verbose output | | `G2G_TOPIC_PREFIX` | `GH` | Prefix used when generating Gerrit topics | | `G2G_SKIP_GERRIT_COMMENTS` | `false` | Skip posting back-reference comments on Gerrit changes | +| `G2G_TRUSTED_ASSOCIATIONS` | `OWNER,MEMBER,COLLABORATOR` | Author associations trusted to issue `@github2gerrit` comment directives | | `G2G_ENABLE_DERIVATION` | `true` | Enable automatic derivation of Gerrit parameters from organization defaults | | `G2G_AUTO_SAVE_CONFIG` | context-dependent | Save derived parameters back to the configuration file | | `G2G_RESPECT_USER_SSH` | `false` (`true` in direct URL mode) | Use the local user's SSH configuration and keys instead of provided key material | diff --git a/docs/features.md b/docs/features.md index a660408..4193ed2 100644 --- a/docs/features.md +++ b/docs/features.md @@ -102,6 +102,57 @@ acts on it during the next workflow run. - When the same command appears in more than one comment, only the latest occurrence takes effect. - The tool logs unrecognized directives at debug level and ignores them. +- Only comments from trusted authors count. See below. + +### Who May Issue Commands + +These repositories are public, so the tool checks who wrote a comment before +acting on it. Without that check, any GitHub user able to leave a comment +could direct the tool. + +A comment counts only when GitHub reports its author's `author_association` +as one of: + +| Association | Meaning | +| -------------- | ------------------------------------------ | +| `OWNER` | Owns the repository | +| `MEMBER` | Member of the owning organization | +| `COLLABORATOR` | Invited collaborator on the repository | + +`CONTRIBUTOR` is **not** trusted. It means only that the author has had a +pull request merged at some point, which any outside contributor can achieve +and which carries no authority. + +The tool ignores directives from anyone else and logs a warning naming the +author and their association, so a refused command is visible rather than +silently dropped. + +#### For contributors adding commands + +Adding a `CommandDefinition` to the registry gives the new command the same +authorisation as the existing one, so long as consumers reach it through +`github2gerrit.pr_directives`, which fetches comments, discards untrusted +authors and parses in a single step. + +The parsing module performs no authorisation of its own. Calling +`parse_commands`, `find_command` or `has_command` with comments taken straight +from the GitHub API skips the check and recreates the original defect. + +Override the trusted set with `G2G_TRUSTED_ASSOCIATIONS`, a comma-separated +list. Setting it to `OWNER` alone is the strictest useful value: + +```yaml +env: + G2G_TRUSTED_ASSOCIATIONS: "OWNER,MEMBER" +``` + +An empty or blank value keeps the default rather than trusting nobody or +everyone. + +Note that `MEMBER` means organization member, which is not the same as write +access to a given repository. On a Gerrit mirror that remains the only +available signal, because the GitHub collaborator list holds infrastructure +accounts rather than the project's reviewers. ### Available Commands diff --git a/src/github2gerrit/config.py b/src/github2gerrit/config.py index f29ec62..727cde9 100644 --- a/src/github2gerrit/config.py +++ b/src/github2gerrit/config.py @@ -99,6 +99,7 @@ "G2G_NO_GERRIT", "G2G_DISABLED", "G2G_TOPIC_PREFIX", + "G2G_TRUSTED_ASSOCIATIONS", "G2G_LOG_LEVEL", "G2G_SHOW_PROGRESS", "G2G_RESPECT_USER_SSH", diff --git a/src/github2gerrit/core.py b/src/github2gerrit/core.py index 80815d3..4af69e6 100644 --- a/src/github2gerrit/core.py +++ b/src/github2gerrit/core.py @@ -1655,7 +1655,12 @@ def _should_create_missing( Returns ``True`` when either the ``--create-missing`` CLI flag is active **or** a ``@github2gerrit create missing change`` - comment is present on the PR. + comment is present on the PR *from a trusted author*. + + Comment authorship is checked because these mirrors are public: + without it, anyone able to comment could force creation of a + Gerrit change that the UPDATE path deliberately declined to + make. """ # 1. Explicit CLI / environment flag if inputs.create_missing: @@ -1670,16 +1675,13 @@ def _should_create_missing( try: from .pr_commands import CMD_CREATE_MISSING - from .pr_commands import find_command + from .pr_directives import find_pr_command client_gh = build_client() repo = get_repo_from_env(client_gh) pr_obj = get_pull(repo, int(gh.pr_number)) - issue = pr_obj.as_issue() - comment_bodies = [c.body or "" for c in issue.get_comments()] - - match = find_command(comment_bodies, CMD_CREATE_MISSING.name) + match = find_pr_command(pr_obj, CMD_CREATE_MISSING.name) if match is not None: log.info( "✅ Found '@github2gerrit %s' in PR #%s comment #%d; " @@ -1692,7 +1694,7 @@ def _should_create_missing( log.debug( "No @github2gerrit create-missing command found in " - "PR #%s comments", + "PR #%s comments from trusted authors", gh.pr_number, ) except Exception as exc: diff --git a/src/github2gerrit/github_api.py b/src/github2gerrit/github_api.py index 6fc3e53..d49edf9 100644 --- a/src/github2gerrit/github_api.py +++ b/src/github2gerrit/github_api.py @@ -19,6 +19,7 @@ import logging import os import re +from collections.abc import Callable from collections.abc import Iterable from importlib import import_module from typing import Any @@ -30,6 +31,7 @@ from .error_codes import is_github_api_permission_error from .external_api import ApiType from .external_api import external_api_call +from .trust import is_trusted_association # Error message constants to comply with TRY003 @@ -74,8 +76,14 @@ def __init__(self, *args: Any, **kwargs: Any) -> None: raise RuntimeError(_MSG_PYGITHUB_REQUIRED) +class GhUser(Protocol): + login: str + + class GhIssueComment(Protocol): body: str | None + author_association: str | None + user: GhUser | None class GhIssue(Protocol): @@ -127,6 +135,7 @@ def get_repo(self, full: str) -> GhRepository: "get_pull", "get_recent_change_ids_from_comments", "get_repo_from_env", + "get_trusted_comment_bodies", "iter_open_pulls", ] @@ -318,6 +327,57 @@ def get_recent_change_ids_from_comments( return found +@external_api_call(ApiType.GITHUB, "get_trusted_comment_bodies") +def get_trusted_comment_bodies( + pr: GhPullRequest, + *, + directive_detector: Callable[[str], bool] | None = None, +) -> tuple[list[str], list[str]]: + """Split PR comments into trusted bodies and ignored directives. + + Comments are only obeyed when their author's ``author_association`` + is trusted (see :mod:`github2gerrit.trust`). Without this, anyone + able to comment on a public mirror could direct the tool. + + Args: + pr: Pull request. + directive_detector: Predicate deciding whether an untrusted + comment was *attempting* to direct the tool, and so is worth + reporting. Supplied by the caller rather than matched here, so + that directive grammar stays owned by the command system and + cannot drift from it. When ``None``, nothing is reported. + + Returns: + A ``(trusted_bodies, ignored)`` pair, ordered oldest to newest. + ``ignored`` holds ``"login (ASSOCIATION)"`` descriptions for + untrusted comments that the detector accepted, so the caller can + explain the omission rather than failing silently. + """ + issue = _get_issue(pr) + + trusted_bodies: list[str] = [] + ignored: list[str] = [] + + for comment in issue.get_comments(): + body = getattr(comment, "body", "") or "" + if not body: + continue + + association = str(getattr(comment, "author_association", "") or "") + if is_trusted_association(association): + trusted_bodies.append(body) + continue + + if directive_detector is not None and directive_detector(body): + login = ( + getattr(getattr(comment, "user", None), "login", "") + or "unknown" + ) + ignored.append(f"{login} ({association or 'unknown'})") + + return trusted_bodies, ignored + + @external_api_call(ApiType.GITHUB, "create_pr_comment") def create_pr_comment(pr: GhPullRequest, body: str) -> None: """Create a new comment on the pull request.""" diff --git a/src/github2gerrit/pr_commands.py b/src/github2gerrit/pr_commands.py index db5ea86..79fb25f 100644 --- a/src/github2gerrit/pr_commands.py +++ b/src/github2gerrit/pr_commands.py @@ -12,8 +12,8 @@ Design principles ───────────────── - **Registry-based**: new commands are added by appending a - ``CommandDefinition`` to ``COMMAND_REGISTRY``; no other code changes - are required for recognition. + ``CommandDefinition`` to ``COMMAND_REGISTRY``; no change to this + module's parsing logic is required. - **Case-insensitive**: command matching ignores case so that ``@github2gerrit Create Missing Change`` works identically to ``@github2gerrit create missing change``. @@ -25,6 +25,21 @@ - **Minimal coupling**: the module depends only on the standard library and exposes typed dataclasses consumed by the orchestrator. +.. warning:: + + **This module performs no authorisation.** Every function here + assumes its input has already been filtered to comments from trusted + authors. Passing raw comments straight from the GitHub API lets any + user of a public mirror direct the tool — the defect fixed in issue + #382. + + Adding a command to ``COMMAND_REGISTRY`` requires no changes here, + but it does **not** grant that command a security gate on its own. + Consume commands through :mod:`github2gerrit.pr_directives`, which + fetches, authorises and parses in one step, rather than calling + :func:`parse_commands`, :func:`find_command` or :func:`has_command` + directly. + Supported commands ────────────────── ``create missing change`` (aliases ``create missing``, ``create-missing``) @@ -48,6 +63,7 @@ "CommandDefinition", "CommandMatch", "CommandParseResult", + "contains_directive", "find_command", "has_command", "list_commands", @@ -208,16 +224,52 @@ def _build_phrase_index() -> dict[str, str]: # ── Public API ────────────────────────────────────────────────────── -def parse_commands(comment_bodies: list[str]) -> CommandParseResult: - """Scan PR comment bodies for ``@github2gerrit`` commands. +def contains_directive(body: str) -> bool: + """Report whether *body* holds a syntactically valid directive. + + Applies the same grammar as :func:`parse_commands`, so a bare + ``@github2gerrit`` mention is *not* a directive — consistent with + that function, which reports it as neither a match nor an + unrecognised directive. + + A mention followed by whitespace alone is likewise not a directive. + The mention pattern's ``.`` matches spaces, so such text satisfies + the regex, but there is no command in it for anyone to act on. + + Callers use this to distinguish an ordinary comment that happens to + mention the tool from one that attempts to direct it. + + Args: + body: Comment body text. + + Returns: + ``True`` when the mention is followed by non-whitespace text. + """ + if not body: + return False + match = _MENTION_RE.search(body) + return match is not None and bool(match.group(1).strip()) + + +def parse_commands( + trusted_comment_bodies: list[str], +) -> CommandParseResult: + """Scan trusted PR comment bodies for ``@github2gerrit`` commands. Comments are processed oldest-first. When the same command appears in multiple comments the *latest* occurrence is kept (deduplication by canonical command name). + .. warning:: + + Performs no authorisation. *trusted_comment_bodies* must + already exclude comments from untrusted authors; prefer + :func:`github2gerrit.pr_directives.scan_pr_directives`, which + does that for you. + Args: - comment_bodies: Ordered list of comment body strings - (oldest → newest). + trusted_comment_bodies: Ordered list of comment body strings + (oldest → newest), already filtered by author trust. Returns: A ``CommandParseResult`` containing de-duplicated matches and @@ -228,7 +280,7 @@ def parse_commands(comment_bodies: list[str]) -> CommandParseResult: seen: dict[str, CommandMatch] = {} unrecognised: list[str] = [] - for idx, body in enumerate(comment_bodies): + for idx, body in enumerate(trusted_comment_bodies): if not body: continue for m in _MENTION_RE.finditer(body): @@ -278,37 +330,49 @@ def parse_commands(comment_bodies: list[str]) -> CommandParseResult: return result -def has_command(comment_bodies: list[str], command_name: str) -> bool: - """Check whether a specific command exists in the PR comments. +def has_command(trusted_comment_bodies: list[str], command_name: str) -> bool: + """Check whether a specific command exists in trusted PR comments. This is a convenience wrapper around ``parse_commands`` for the common case where only one command matters. + .. warning:: + + Performs no authorisation; see :func:`parse_commands`. + Args: - comment_bodies: Ordered list of comment body strings. + trusted_comment_bodies: Ordered list of comment body strings, + already filtered by author trust. command_name: Canonical command name to check for. Returns: ``True`` if the command was found in at least one comment. """ - result = parse_commands(comment_bodies) + result = parse_commands(trusted_comment_bodies) return result.has(command_name) def find_command( - comment_bodies: list[str], + trusted_comment_bodies: list[str], command_name: str, ) -> CommandMatch | None: - """Find a specific command match in the PR comments. + """Find a specific command match in trusted PR comments. + + .. warning:: + + Performs no authorisation. Prefer + :func:`github2gerrit.pr_directives.find_pr_command`, which + filters by author trust first. Args: - comment_bodies: Ordered list of comment body strings. + trusted_comment_bodies: Ordered list of comment body strings, + already filtered by author trust. command_name: Canonical command name to search for. Returns: The ``CommandMatch`` if found, otherwise ``None``. """ - result = parse_commands(comment_bodies) + result = parse_commands(trusted_comment_bodies) target = command_name.lower().strip() for m in result.matches: if m.command_name == target: diff --git a/src/github2gerrit/pr_directives.py b/src/github2gerrit/pr_directives.py new file mode 100644 index 0000000..82c18ae --- /dev/null +++ b/src/github2gerrit/pr_directives.py @@ -0,0 +1,115 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: 2026 The Linux Foundation +"""The entry point for acting on ``@github2gerrit`` PR directives. + +This module composes the three steps that must always happen together: + +1. fetch the pull request's comments (:mod:`github2gerrit.github_api`), +2. discard those whose author is not trusted + (:mod:`github2gerrit.trust`), and +3. parse the survivors for commands + (:mod:`github2gerrit.pr_commands`). + +**Consumers of the command registry must come through here.** Calling +:func:`github2gerrit.pr_commands.parse_commands` directly with comments +straight from the API skips step 2, which on a public mirror lets any +GitHub user direct the tool. That was the defect fixed in issue #382, +and a second command added later would reintroduce it just as easily as +the first one did. + +The split exists so that no single module holds both API access and +directive grammar: ``github_api`` fetches and partitions by trust +without knowing what a command looks like, ``pr_commands`` parses text +without reaching the network, and this module joins them. +""" + +from __future__ import annotations + +import logging +from typing import Any + +from .github_api import get_trusted_comment_bodies +from .pr_commands import MENTION_PREFIX +from .pr_commands import CommandMatch +from .pr_commands import CommandParseResult +from .pr_commands import contains_directive +from .pr_commands import parse_commands +from .trust import describe_trust_policy + + +__all__ = [ + "DirectiveScan", + "find_pr_command", + "scan_pr_directives", +] + +log = logging.getLogger("github2gerrit.pr_directives") + + +class DirectiveScan: + """Result of scanning a pull request for directives. + + Attributes: + result: Commands parsed from trusted comments. + ignored: ``"login (ASSOCIATION)"`` for each untrusted comment + that attempted to issue a directive. + """ + + __slots__ = ("ignored", "result") + + def __init__( + self, + result: CommandParseResult, + ignored: list[str], + ) -> None: + self.result = result + self.ignored = ignored + + +def scan_pr_directives(pr: Any) -> DirectiveScan: + """Fetch, authorise and parse a pull request's directives. + + Refused directives are logged here rather than left to each caller, + so a maintainer whose command is declined always learns why. + + Args: + pr: Pull request object. + + Returns: + A :class:`DirectiveScan`. + """ + bodies, ignored = get_trusted_comment_bodies( + pr, directive_detector=contains_directive + ) + + if ignored: + log.warning( + "🚫 Ignoring %s directive(s) from untrusted comment " + "author(s): %s. Trusted associations: %s", + MENTION_PREFIX, + ", ".join(ignored), + describe_trust_policy(), + ) + + return DirectiveScan(parse_commands(bodies), ignored) + + +def find_pr_command(pr: Any, command_name: str) -> CommandMatch | None: + """Return a trusted occurrence of *command_name*, or ``None``. + + The authorised counterpart of + :func:`github2gerrit.pr_commands.find_command`. + + Args: + pr: Pull request object. + command_name: Canonical command name to look for. + + Returns: + The :class:`~github2gerrit.pr_commands.CommandMatch` when a + trusted author issued the command, otherwise ``None``. + """ + target = command_name.lower().strip() + for match in scan_pr_directives(pr).result.matches: + if match.command_name == target: + return match + return None diff --git a/src/github2gerrit/trust.py b/src/github2gerrit/trust.py new file mode 100644 index 0000000..2ba4829 --- /dev/null +++ b/src/github2gerrit/trust.py @@ -0,0 +1,122 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: 2026 The Linux Foundation +"""Who is trusted to direct this tool. + +GitHub reports an ``author_association`` on every comment, review and +pull request, describing the author's standing in the repository. This +module turns that into a single trust decision, so every place that acts +on user input agrees on who may be obeyed. + +Why ``author_association`` rather than collaborator permission +────────────────────────────────────────────────────────────── +The repositories this tool serves are Gerrit mirrors. Their GitHub +collaborator lists hold infrastructure staff and bots, not the people +who review the code — on ``opendaylight/mdsal`` every collaborator is +LF releng, while project committers appear only as organisation +``MEMBER``\\ s. Requiring ``write`` or ``admin`` there would trust the +wrong people and exclude the right ones. + +``author_association`` also arrives on the payloads already being +fetched, so it costs no extra API calls and no extra token scope. + +Its limits are real and worth stating: ``MEMBER`` means organisation +member, which is not the same as write access to a given repository. +Callers that need a stronger guarantee should combine this with a +signal GitHub enforces structurally — for example, that a pull request +author cannot approve their own pull request. +""" + +from __future__ import annotations + +import logging +import os +from collections.abc import Iterable + + +__all__ = [ + "DEFAULT_TRUSTED_ASSOCIATIONS", + "TRUSTED_ASSOCIATIONS_ENV", + "describe_trust_policy", + "is_trusted_association", + "trusted_associations", +] + +log = logging.getLogger("github2gerrit.trust") + +TRUSTED_ASSOCIATIONS_ENV = "G2G_TRUSTED_ASSOCIATIONS" +"""Environment variable overriding the trusted set (comma-separated).""" + +DEFAULT_TRUSTED_ASSOCIATIONS: frozenset[str] = frozenset( + { + "OWNER", + "MEMBER", + "COLLABORATOR", + } +) +"""Associations trusted to direct the tool by default. + +``OWNER`` and ``COLLABORATOR`` carry repository standing; ``MEMBER`` +carries organisation standing, which is the only signal available on a +Gerrit mirror. + +``CONTRIBUTOR`` is deliberately excluded: it means only that the author +has had a pull request merged at some point, which any outside +contributor can achieve and which conveys no authority. +""" + + +def trusted_associations() -> frozenset[str]: + """Return the trusted association set, honouring the environment. + + ``G2G_TRUSTED_ASSOCIATIONS`` accepts a comma-separated list, for + example ``OWNER,MEMBER``. Values are upper-cased and stripped. An + unset, empty or entirely blank value keeps the default. + """ + raw = os.getenv(TRUSTED_ASSOCIATIONS_ENV, "").strip() + if not raw: + return DEFAULT_TRUSTED_ASSOCIATIONS + + parsed = {part.strip().upper() for part in raw.split(",") if part.strip()} + if not parsed: + return DEFAULT_TRUSTED_ASSOCIATIONS + + log.debug( + "Trusted associations overridden via %s: %s", + TRUSTED_ASSOCIATIONS_ENV, + sorted(parsed), + ) + return frozenset(parsed) + + +def is_trusted_association( + association: str | None, + *, + allowed: Iterable[str] | None = None, +) -> bool: + """Report whether *association* is trusted to direct the tool. + + Args: + association: A GitHub ``author_association`` value. ``None``, + empty, or an unrecognised value is untrusted. + allowed: Override the trusted set. Defaults to + :func:`trusted_associations`. + + Returns: + ``True`` only for an explicitly trusted association. Every + other input — including a missing or malformed one — is + untrusted, so an absent signal never grants authority. + """ + if not association: + return False + + permitted = ( + frozenset(a.strip().upper() for a in allowed) + if allowed is not None + else trusted_associations() + ) + return association.strip().upper() in permitted + + +def describe_trust_policy() -> str: + """Return a human-readable summary of the trusted set, for logs.""" + return ", ".join(sorted(trusted_associations())) diff --git a/tests/test_pr_command_authorisation.py b/tests/test_pr_command_authorisation.py new file mode 100644 index 0000000..012f176 --- /dev/null +++ b/tests/test_pr_command_authorisation.py @@ -0,0 +1,357 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: 2026 The Linux Foundation +"""Authorisation of ``@github2gerrit`` PR comment directives. + +These mirrors are public. Without an authorship check any GitHub user +able to leave a comment could direct the tool, so command recognition is +gated on the comment author's ``author_association``. +""" + +from __future__ import annotations + +import inspect +from typing import Any +from unittest.mock import MagicMock +from unittest.mock import patch + +import pytest + +from github2gerrit import pr_commands +from github2gerrit.github_api import get_trusted_comment_bodies +from github2gerrit.pr_commands import CMD_CREATE_MISSING +from github2gerrit.pr_commands import MENTION_PREFIX +from github2gerrit.pr_commands import contains_directive +from github2gerrit.pr_commands import find_command +from github2gerrit.pr_commands import has_command +from github2gerrit.pr_commands import parse_commands +from github2gerrit.pr_directives import find_pr_command +from github2gerrit.pr_directives import scan_pr_directives +from github2gerrit.trust import DEFAULT_TRUSTED_ASSOCIATIONS +from github2gerrit.trust import describe_trust_policy +from github2gerrit.trust import is_trusted_association +from github2gerrit.trust import trusted_associations + + +DIRECTIVE = f"{MENTION_PREFIX} {CMD_CREATE_MISSING.name}" + + +def _comment(body: str, association: str, login: str = "someone") -> Any: + comment = MagicMock() + comment.body = body + comment.author_association = association + comment.user = MagicMock() + comment.user.login = login + return comment + + +def _pr_with(comments: list[Any]) -> Any: + issue = MagicMock() + issue.get_comments.return_value = comments + pr = MagicMock() + pr.as_issue.return_value = issue + return pr + + +class TestIsTrustedAssociation: + """The trust rule itself.""" + + @pytest.mark.parametrize("association", ["OWNER", "MEMBER", "COLLABORATOR"]) + def test_trusted_values(self, association: str) -> None: + assert is_trusted_association(association) is True + + @pytest.mark.parametrize( + "association", + [ + "CONTRIBUTOR", + "FIRST_TIME_CONTRIBUTOR", + "FIRST_TIMER", + "MANNEQUIN", + "NONE", + ], + ) + def test_untrusted_values(self, association: str) -> None: + assert is_trusted_association(association) is False + + def test_contributor_is_not_trusted(self) -> None: + """CONTRIBUTOR only means a PR was merged once; not authority.""" + assert "CONTRIBUTOR" not in DEFAULT_TRUSTED_ASSOCIATIONS + + @pytest.mark.parametrize("association", [None, "", " ", "bogus"]) + def test_absent_or_unknown_is_untrusted( + self, association: str | None + ) -> None: + assert is_trusted_association(association) is False + + def test_case_and_whitespace_insensitive(self) -> None: + assert is_trusted_association(" member ") is True + + def test_explicit_allowed_set_overrides(self) -> None: + assert is_trusted_association("MEMBER", allowed=["OWNER"]) is False + assert is_trusted_association("OWNER", allowed=["OWNER"]) is True + + +class TestTrustedAssociationsEnv: + """Operators may narrow or widen the trusted set.""" + + def test_default_when_unset(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("G2G_TRUSTED_ASSOCIATIONS", raising=False) + assert trusted_associations() == DEFAULT_TRUSTED_ASSOCIATIONS + + def test_override_narrows(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("G2G_TRUSTED_ASSOCIATIONS", "OWNER") + assert trusted_associations() == frozenset({"OWNER"}) + assert is_trusted_association("MEMBER") is False + + def test_override_is_normalised( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("G2G_TRUSTED_ASSOCIATIONS", " owner , member ") + assert trusted_associations() == frozenset({"OWNER", "MEMBER"}) + + @pytest.mark.parametrize("raw", ["", " ", ",", " , , "]) + def test_blank_override_keeps_default( + self, raw: str, monkeypatch: pytest.MonkeyPatch + ) -> None: + """An empty override must not silently trust nobody or everybody.""" + monkeypatch.setenv("G2G_TRUSTED_ASSOCIATIONS", raw) + assert trusted_associations() == DEFAULT_TRUSTED_ASSOCIATIONS + + def test_policy_description_is_stable( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setenv("G2G_TRUSTED_ASSOCIATIONS", "MEMBER,OWNER") + assert describe_trust_policy() == "MEMBER, OWNER" + + +class TestGetTrustedCommentBodies: + """Comment partitioning at the API boundary.""" + + def test_untrusted_bodies_excluded(self) -> None: + pr = _pr_with( + [ + _comment(DIRECTIVE, "NONE", "outsider"), + _comment("looks good", "MEMBER", "maintainer"), + ] + ) + + bodies, ignored = get_trusted_comment_bodies( + pr, directive_detector=contains_directive + ) + + assert bodies == ["looks good"] + assert ignored == ["outsider (NONE)"] + + def test_trusted_directive_kept(self) -> None: + pr = _pr_with([_comment(DIRECTIVE, "MEMBER", "maintainer")]) + + bodies, ignored = get_trusted_comment_bodies( + pr, directive_detector=contains_directive + ) + + assert bodies == [DIRECTIVE] + assert ignored == [] + + def test_untrusted_chatter_is_not_reported(self) -> None: + """Only ignored *directives* are worth reporting.""" + pr = _pr_with([_comment("nice work", "NONE", "outsider")]) + + bodies, ignored = get_trusted_comment_bodies( + pr, directive_detector=contains_directive + ) + + assert bodies == [] + assert ignored == [] + + @pytest.mark.parametrize( + "body", + [ + MENTION_PREFIX, + f"{MENTION_PREFIX} ", + f"thanks {MENTION_PREFIX}", + ], + ) + def test_bare_mention_is_not_a_directive(self, body: str) -> None: + """A mention with no command must not warn on every scan. + + ``parse_commands`` treats a bare mention as neither a match nor + an unrecognised directive, so reporting it here would produce a + persistent warning about a comment nobody can act on. + """ + pr = _pr_with([_comment(body, "NONE", "outsider")]) + + bodies, ignored = get_trusted_comment_bodies( + pr, directive_detector=contains_directive + ) + + assert bodies == [] + assert ignored == [] + + def test_missing_association_is_untrusted(self) -> None: + comment = MagicMock() + comment.body = DIRECTIVE + comment.author_association = None + comment.user = MagicMock() + comment.user.login = "ghost" + + bodies, ignored = get_trusted_comment_bodies( + _pr_with([comment]), directive_detector=contains_directive + ) + + assert bodies == [] + assert ignored == ["ghost (unknown)"] + + def test_no_detector_reports_nothing(self) -> None: + """Reporting is opt-in; the grammar belongs to the caller.""" + pr = _pr_with([_comment(DIRECTIVE, "NONE", "outsider")]) + + bodies, ignored = get_trusted_comment_bodies(pr) + + assert bodies == [] + assert ignored == [] + + def test_ordering_preserved(self) -> None: + pr = _pr_with( + [ + _comment("first", "OWNER"), + _comment("second", "COLLABORATOR"), + ] + ) + + bodies, _ = get_trusted_comment_bodies(pr) + + assert bodies == ["first", "second"] + + def test_empty_bodies_skipped(self) -> None: + pr = _pr_with([_comment("", "OWNER"), _comment("kept", "OWNER")]) + + bodies, _ = get_trusted_comment_bodies(pr) + + assert bodies == ["kept"] + + +class TestPrDirectivesEntryPoint: + """The composed entry point is the supported way in. + + Consumers reaching for the registry must not have to remember to + filter by author trust themselves; forgetting is exactly how the + original defect arose. + """ + + def test_scan_authorises_before_parsing(self) -> None: + pr = _pr_with( + [ + _comment(DIRECTIVE, "NONE", "outsider"), + _comment("looks good", "MEMBER", "maintainer"), + ] + ) + + scan = scan_pr_directives(pr) + + assert scan.result.has(CMD_CREATE_MISSING.name) is False + assert scan.ignored == ["outsider (NONE)"] + + def test_scan_keeps_trusted_command(self) -> None: + pr = _pr_with([_comment(DIRECTIVE, "OWNER", "owner")]) + + scan = scan_pr_directives(pr) + + assert scan.result.has(CMD_CREATE_MISSING.name) is True + assert scan.ignored == [] + + def test_find_pr_command_matches_trusted_only(self) -> None: + untrusted = _pr_with([_comment(DIRECTIVE, "CONTRIBUTOR")]) + trusted = _pr_with([_comment(DIRECTIVE, "COLLABORATOR")]) + + assert find_pr_command(untrusted, CMD_CREATE_MISSING.name) is None + match = find_pr_command(trusted, CMD_CREATE_MISSING.name) + assert match is not None + assert match.command_name == CMD_CREATE_MISSING.name + + def test_unknown_command_returns_none(self) -> None: + pr = _pr_with([_comment(DIRECTIVE, "OWNER")]) + + assert find_pr_command(pr, "no such command") is None + + def test_refusal_logged_once_by_the_entry_point( + self, caplog: pytest.LogCaptureFixture + ) -> None: + pr = _pr_with([_comment(DIRECTIVE, "NONE", "outsider")]) + + with caplog.at_level("WARNING"): + find_pr_command(pr, CMD_CREATE_MISSING.name) + + assert caplog.text.count("outsider (NONE)") == 1 + + +class TestRegistryDocumentsTheGate: + """Issue #382 asks that new commands inherit the gate. + + The parser cannot enforce that itself without reaching the network, + so the contract is carried in the naming and the documentation. + A future contributor adding command number two should meet it. + """ + + def test_parser_parameters_name_the_contract(self) -> None: + for func in (parse_commands, has_command, find_command): + first = next(iter(inspect.signature(func).parameters)) + assert first == "trusted_comment_bodies", ( + f"{func.__name__} should name its input as trusted, so a " + "caller passing raw API comments notices" + ) + + def test_registry_module_warns_against_direct_use(self) -> None: + doc = pr_commands.__doc__ or "" + assert "no authorisation" in doc.lower() + assert "pr_directives" in doc + + +class TestShouldCreateMissingAuthorisation: + """End-to-end gating of the one registered command.""" + + def _run(self, comments: list[Any]) -> bool: + from github2gerrit.core import Orchestrator + + orch = Orchestrator.__new__(Orchestrator) + inputs = MagicMock() + inputs.create_missing = False + gh = MagicMock() + gh.pr_number = 29 + + pr = _pr_with(comments) + with ( + patch("github2gerrit.core.build_client"), + patch("github2gerrit.core.get_repo_from_env"), + patch("github2gerrit.core.get_pull", return_value=pr), + ): + return orch._should_create_missing(inputs, gh) + + def test_directive_from_outsider_ignored(self) -> None: + assert self._run([_comment(DIRECTIVE, "NONE", "outsider")]) is False + + def test_directive_from_contributor_ignored(self) -> None: + """A merged PR in the past confers no authority.""" + assert self._run([_comment(DIRECTIVE, "CONTRIBUTOR")]) is False + + @pytest.mark.parametrize("association", ["OWNER", "MEMBER", "COLLABORATOR"]) + def test_directive_from_trusted_author_honoured( + self, association: str + ) -> None: + assert self._run([_comment(DIRECTIVE, association)]) is True + + def test_outsider_cannot_ride_on_trusted_chatter(self) -> None: + """A trusted unrelated comment must not launder the directive.""" + comments = [ + _comment("looks good to me", "MEMBER", "maintainer"), + _comment(DIRECTIVE, "NONE", "outsider"), + ] + assert self._run(comments) is False + + def test_ignored_directive_is_logged( + self, caplog: pytest.LogCaptureFixture + ) -> None: + """Silent refusal would leave the author with no explanation.""" + with caplog.at_level("WARNING"): + self._run([_comment(DIRECTIVE, "NONE", "outsider")]) + + assert "outsider (NONE)" in caplog.text + assert "untrusted" in caplog.text.lower() diff --git a/tests/test_pr_commands.py b/tests/test_pr_commands.py index aa18b74..df8fd1a 100644 --- a/tests/test_pr_commands.py +++ b/tests/test_pr_commands.py @@ -515,9 +515,13 @@ def test_comment_directive_returns_true(self, tmp_path): mock_comment_1 = MagicMock() mock_comment_1.body = "CI is stuck, let me try this:" + mock_comment_1.author_association = "MEMBER" mock_comment_2 = MagicMock() mock_comment_2.body = "@github2gerrit create missing change" + # Directives are only obeyed from a trusted author; see + # tests/test_pr_command_authorisation.py. + mock_comment_2.author_association = "MEMBER" mock_issue = MagicMock() mock_issue.get_comments.return_value = [mock_comment_1, mock_comment_2] @@ -578,6 +582,9 @@ def test_alias_in_comment_returns_true(self, tmp_path): mock_comment = MagicMock() mock_comment.body = "@github2gerrit create-missing" + # Directives are only obeyed from a trusted author; see + # tests/test_pr_command_authorisation.py. + mock_comment.author_association = "MEMBER" mock_issue = MagicMock() mock_issue.get_comments.return_value = [mock_comment]