diff --git a/docs/gitea-execution-profiles.md b/docs/gitea-execution-profiles.md index 61324bd..e4ec36e 100644 --- a/docs/gitea-execution-profiles.md +++ b/docs/gitea-execution-profiles.md @@ -203,6 +203,29 @@ remote/org/repo arguments. Create operations are audit-logged redacted, and normal output contains no endpoint URLs (`GITEA_MCP_REVEAL_ENDPOINTS=1` is the local admin opt-in for web links). +## PR edits versus PR closure (#216) + +Editing a pull request and closing one are different capabilities: + +- **PR edits** (`gitea_edit_pr` with `title`/`body`/`base`, or reopening with + `state="open"`) stay on the ordinary edit path and need no dedicated + capability. +- **PR closure** (`gitea_edit_pr` with `state="closed"`) requires the + distinct `gitea.pr.close` operation. The resolver task is `close_pr` + (`gitea_resolve_task_capability(task="close_pr")`, author-side). Without + `gitea.pr.close` the close attempt fails closed — no API call, structured + `permission_report` — so the broad edit path can never be used as an + untracked close fallback. +- Closures are audited as a distinct `close_pr` action with + `required_permission: gitea.pr.close` in the request metadata, so final + reports can prove exactly which mutation capability was exercised (#191). + +`gitea.pr.close` has no legacy alias; spell it canonically. It is not part of +any default profile: the operator grants it deliberately (e.g. for an +explicit operator-directed closure of a contaminated PR). If `close_pr` ever +resolves as unknown, agents must fail closed rather than fall back to the +edit path. + ## Identity and fail-closed rules Before **any** mutating action, a workflow must know both: diff --git a/mcp_server.py b/mcp_server.py index 83c87fc..c6d5226 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -977,7 +977,7 @@ def gitea_get_pr_review_feedback( 'feedback_not_attempted' True, 'reasons', and 'permission_report' — deliberately distinct from a successful "no reviews yet" result. """ - reasons = _issue_comment_gate("gitea.read") + reasons = _profile_operation_gate("gitea.read") if reasons: return { "success": False, @@ -1220,11 +1220,20 @@ def gitea_edit_pr( ) -> dict: """Edit an existing pull request on a Gitea repository. + Closing a PR (``state='closed'``) is a distinct capability from other + edits (#216): it requires the ``gitea.pr.close`` operation (resolver + task ``close_pr``), fails closed with a structured permission report + when the active profile lacks it, and is audited as a distinct + ``close_pr`` action. Title/body/base edits and reopening stay on the + ordinary edit path, so the edit tool can never be used as an untracked + close fallback. + Args: pr_number: The pull request index/number (required). title: New PR title. body: New PR description. - state: New state — 'open' or 'closed'. + state: New state — 'open' or 'closed'. 'closed' requires the + ``gitea.pr.close`` capability. base: Target branch name. remote: Known instance — 'dadeschools' or 'prgs'. host: Override the Gitea host. @@ -1232,7 +1241,10 @@ def gitea_edit_pr( repo: Override the repository name. Returns: - dict with success status and details of the edited PR. + dict with success status and details of the edited PR. A close + attempt without ``gitea.pr.close`` returns 'success'/'performed' + False with 'reasons' and a structured 'permission_report' and makes + no API call. """ # Validate inputs BEFORE any auth/profile resolution or API setup: a # no-fields call is a pure validation error and must not depend on @@ -1243,6 +1255,9 @@ def gitea_edit_pr( if body is not None: payload["body"] = body if state is not None: + if state not in ("open", "closed"): + raise ValueError( + f"Invalid state {state!r}: must be 'open' or 'closed' (fail closed).") payload["state"] = state if base is not None: payload["base"] = base @@ -1250,12 +1265,33 @@ def gitea_edit_pr( if not payload: raise ValueError("At least one field to edit (title, body, state, base) must be provided.") + # PR closure is a first-class capability, distinct from retitling or + # rebasing edits (#216). Gate BEFORE auth/API setup so a blocked close + # never touches the network. + closing = payload.get("state") == "closed" + if closing: + gate_reasons = _profile_operation_gate("gitea.pr.close") + if gate_reasons: + return { + "success": False, + "performed": False, + "pr_number": pr_number, + "requested_state": "closed", + "required_permission": "gitea.pr.close", + "reasons": gate_reasons, + "permission_report": _permission_block_report("gitea.pr.close"), + } + h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) url = f"{repo_api_url(h, o, r)}/pulls/{pr_number}" - with _audited("edit_pr", host=h, remote=remote, org=o, repo=r, - pr_number=pr_number, request_metadata={"fields": sorted(payload)}): + request_metadata = {"fields": sorted(payload)} + if closing: + request_metadata["required_permission"] = "gitea.pr.close" + with _audited("close_pr" if closing else "edit_pr", + host=h, remote=remote, org=o, repo=r, + pr_number=pr_number, request_metadata=request_metadata): data = api_request("PATCH", url, auth, payload) cleanup_status = None @@ -2027,13 +2063,14 @@ def _permission_block_report(required_operation: str, return report -def _issue_comment_gate(op: str) -> list[str]: - """Profile permission check for issue-comment tools (#126). +def _profile_operation_gate(op: str) -> list[str]: + """Profile permission check for a single gated operation (#126, #216). Issue discussion comments are gated separately from the gitea.pr.* review/merge family: listing requires ``gitea.read``, creating requires - ``gitea.issue.comment``. Returns a list of block reasons (empty = allowed); - an unreadable profile fails closed. + ``gitea.issue.comment``. Closing a PR requires the distinct + ``gitea.pr.close`` (#216). Returns a list of block reasons (empty = + allowed); an unreadable profile fails closed. """ try: profile = get_profile() @@ -2085,7 +2122,7 @@ def gitea_list_issue_comments( 'success' False, 'reasons', and a structured 'permission_report' (#142) with no API call made. """ - reasons = _issue_comment_gate("gitea.read") + reasons = _profile_operation_gate("gitea.read") if reasons: return {"success": False, "issue_number": issue_number, "reasons": reasons, @@ -2147,7 +2184,7 @@ def gitea_create_issue_comment( (permission blocks also carry a structured 'permission_report', #142). """ - gate_reasons = _issue_comment_gate("gitea.issue.comment") + gate_reasons = _profile_operation_gate("gitea.issue.comment") reasons = list(gate_reasons) if not (body or "").strip(): reasons.append("comment body must be a non-empty string") @@ -3581,6 +3618,14 @@ def gitea_resolve_task_capability( "permission": "gitea.pr.comment", "role": "author", }, + # PR closure is a first-class capability (#216): distinct from + # comment_pr (gitea.pr.comment) and from ordinary PR edits, which + # need no dedicated capability. gitea_edit_pr(state='closed') is + # gated on the same operation, so no edit-path fallback exists. + "close_pr": { + "permission": "gitea.pr.close", + "role": "author", + }, "address_pr_change_requests": { "permission": "gitea.branch.push", "role": "author", diff --git a/tests/test_close_pr_capability.py b/tests/test_close_pr_capability.py new file mode 100644 index 0000000..8a2385e --- /dev/null +++ b/tests/test_close_pr_capability.py @@ -0,0 +1,154 @@ +"""Tests for the gated PR close path (Issue #216). + +``gitea_edit_pr(state='closed')`` requires the distinct ``gitea.pr.close`` +capability: without it the close fails closed (no auth lookup, no API call, +structured permission report). With it — the explicit operator-directed +contaminated-PR closure path — the close proceeds and is audited as a +distinct ``close_pr`` action carrying the required capability, so final +reports can prove exactly which mutation capability was exercised. + +Ordinary edits (title/body/base) and reopening never require the close +capability: PR comment, PR edit, and PR close remain distinct capabilities. +""" +import sys +import unittest +from unittest.mock import patch + +sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) + +import mcp_server +from mcp_server import gitea_edit_pr + +FAKE_AUTH = "token fake" + +AUTHOR_NO_CLOSE = { + "profile_name": "prgs-author", + "allowed_operations": [ + "gitea.read", "gitea.pr.create", "gitea.pr.comment", + "gitea.branch.push", "gitea.issue.comment", + ], + "forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"], + "audit_label": "prgs-author", +} + +AUTHOR_WITH_CLOSE = { + "profile_name": "prgs-author-closer", + "allowed_operations": AUTHOR_NO_CLOSE["allowed_operations"] + ["gitea.pr.close"], + "forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"], + "audit_label": "prgs-author-closer", +} + + +class TestEditPrCloseGate(unittest.TestCase): + + def setUp(self): + self.mock_api = patch("mcp_server.api_request").start() + self.mock_auth = patch("mcp_server.get_auth_header", return_value=FAKE_AUTH).start() + # Deterministic permission reports: no real operator config, no switching. + patch("gitea_config.load_config", return_value={}).start() + patch("gitea_config.is_runtime_switching_enabled", return_value=False).start() + patch("gitea_audit.audit_enabled", return_value=False).start() + mcp_server._IDENTITY_CACHE.clear() + + def tearDown(self): + patch.stopall() + mcp_server._IDENTITY_CACHE.clear() + + def _set_profile(self, profile): + patch("mcp_server.get_profile", return_value=profile).start() + + def test_close_blocked_without_close_capability(self): + # Author profile without gitea.pr.close: the broad edit path must not + # be usable as an untracked close fallback. + self._set_profile(AUTHOR_NO_CLOSE) + res = gitea_edit_pr(pr_number=205, state="closed", remote="prgs") + self.assertFalse(res["success"]) + self.assertFalse(res["performed"]) + self.assertEqual(res["requested_state"], "closed") + self.assertEqual(res["required_permission"], "gitea.pr.close") + self.assertTrue(res["reasons"]) + report = res["permission_report"] + self.assertEqual(report["required_permission"], "gitea.pr.close") + self.assertEqual(report["active_profile"], "prgs-author") + self.mock_api.assert_not_called() + self.mock_auth.assert_not_called() + + def test_close_blocked_when_close_forbidden(self): + profile = dict(AUTHOR_WITH_CLOSE) + profile["forbidden_operations"] = ["gitea.pr.close"] + self._set_profile(profile) + res = gitea_edit_pr(pr_number=205, state="closed", remote="prgs") + self.assertFalse(res["success"]) + self.assertIn("profile forbids 'gitea.pr.close'", res["reasons"]) + self.mock_api.assert_not_called() + + def test_close_blocked_when_profile_unresolvable(self): + patch("mcp_server.get_profile", side_effect=RuntimeError("no profile")).start() + res = gitea_edit_pr(pr_number=205, state="closed", remote="prgs") + self.assertFalse(res["success"]) + self.assertTrue(any("fail closed" in r for r in res["reasons"])) + self.mock_api.assert_not_called() + + def test_operator_directed_close_allowed_and_audited(self): + # Explicit operator-directed contaminated-PR closure: the operator + # granted gitea.pr.close, so the close proceeds and the audit trail + # records a distinct close_pr action with the capability proof. + self._set_profile(AUTHOR_WITH_CLOSE) + patch("gitea_audit.audit_enabled", return_value=True).start() + mock_write = patch("gitea_audit.write_event").start() + + def api_side_effect(method, url, auth, payload=None): + if method == "GET" and "/user" in url: + return {"login": "jcwalker3"} + if method == "PATCH" and "pulls/205" in url: + self.assertEqual(payload["state"], "closed") + return { + "number": 205, + "title": "Contaminated PR", + "state": "closed", + "html_url": "url", + "body": "No issue link", + "head": {"ref": "feat/invalid-provenance"}, + } + return {} + self.mock_api.side_effect = api_side_effect + + res = gitea_edit_pr(pr_number=205, state="closed", remote="prgs") + self.assertTrue(res["success"]) + self.assertEqual(res["state"], "closed") + mock_write.assert_called() + event = mock_write.call_args[0][0] + self.assertEqual(event["action"], "close_pr") + self.assertEqual( + event["request_metadata"]["required_permission"], "gitea.pr.close") + + def test_title_edit_needs_no_close_capability(self): + # PR edit and PR close are distinct capabilities: retitling stays on + # the ordinary edit path. + self._set_profile(AUTHOR_NO_CLOSE) + self.mock_api.return_value = { + "number": 7, "title": "Renamed", "state": "open", + "body": "", "html_url": "u"} + res = gitea_edit_pr(pr_number=7, title="Renamed", remote="prgs") + self.assertTrue(res["success"]) + + def test_reopen_needs_no_close_capability(self): + self._set_profile(AUTHOR_NO_CLOSE) + self.mock_api.return_value = { + "number": 7, "title": "T", "state": "open", + "body": "", "html_url": "u"} + res = gitea_edit_pr(pr_number=7, state="open", remote="prgs") + self.assertTrue(res["success"]) + + def test_invalid_state_fails_closed_before_auth(self): + # A case-variant state can neither bypass the close gate nor reach + # the API: it is rejected as pure validation, before auth. + self._set_profile(AUTHOR_WITH_CLOSE) + with self.assertRaises(ValueError): + gitea_edit_pr(pr_number=7, state="CLOSED", remote="prgs") + self.mock_api.assert_not_called() + self.mock_auth.assert_not_called() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index d22a311..d84a409 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -1623,7 +1623,8 @@ class TestTrackerHygieneCleanup(unittest.TestCase): self.mock_auth = patch("mcp_server.get_auth_header", return_value=FAKE_AUTH).start() patch("gitea_audit.audit_enabled", return_value=True).start() self.mock_audit = patch("gitea_audit.write_event").start() - patch("mcp_server.get_profile", return_value={"profile_name": "test", "allowed_operations": ["merge", "edit", "close"], "audit_label": "test", "forbidden_operations": []}).start() + # gitea.pr.close: closing a PR via gitea_edit_pr is capability-gated (#216). + patch("mcp_server.get_profile", return_value={"profile_name": "test", "allowed_operations": ["merge", "edit", "close", "gitea.pr.close"], "audit_label": "test", "forbidden_operations": []}).start() def tearDown(self): patch.stopall() diff --git a/tests/test_resolve_task_capability.py b/tests/test_resolve_task_capability.py index 358febe..991e141 100644 --- a/tests/test_resolve_task_capability.py +++ b/tests/test_resolve_task_capability.py @@ -203,5 +203,67 @@ class TestResolveTaskCapability(unittest.TestCase): reasons = res.get("reasons", []) self.assertTrue(any("author" in str(r).lower() for r in reasons) or len(reasons) > 0) + # Issue #216: close_pr is a first-class resolver task gated on gitea.pr.close. + @patch("mcp_server.api_request", return_value={"login": "author-user"}) + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_resolve_close_pr_author_profile_without_close_blocked(self, _auth, _api): + # Default author profile has PR create/comment but not close: the + # close capability must never be implied by broader author ops. + with patch.dict(os.environ, self._env("author-profile")): + res = mcp_server.gitea_resolve_task_capability(task="close_pr", remote="prgs") + self.assertEqual(res["requested_task"], "close_pr") + self.assertEqual(res["required_operation_permission"], "gitea.pr.close") + self.assertEqual(res["required_role_kind"], "author") + self.assertFalse(res["allowed_in_current_session"]) + self.assertTrue(res["stop_required"]) + self.assertEqual(res["matching_configured_profile"], []) + self.assertTrue(res["different_mcp_namespace_required"]) + + @patch("mcp_server.api_request", return_value={"login": "author-user"}) + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_resolve_close_pr_author_profile_with_close_allowed(self, _auth, _api): + # Operator-granted close capability resolves cleanly under an author profile. + config = json.loads(json.dumps(CONFIG_RESOLVER)) + config["profiles"]["author-closer"] = { + "enabled": True, + "context": "ctx", + "role": "author", + "username": "author-user", + "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"}, + "allowed_operations": ["gitea.read", "gitea.pr.close"], + "forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"], + "execution_profile": "author-closer", + } + self._write_config(config) + with patch.dict(os.environ, self._env("author-closer")): + res = mcp_server.gitea_resolve_task_capability(task="close_pr", remote="prgs") + self.assertTrue(res["allowed_in_current_session"]) + self.assertFalse(res["stop_required"]) + self.assertIn("author-closer", res["matching_configured_profile"]) + self.assertIn("ready for operations", res["exact_safe_next_action"]) + + @patch("mcp_server.api_request", return_value={"login": "reviewer-user"}) + @patch("mcp_server.get_auth_header", return_value="token reviewer-pass") + def test_resolve_close_pr_reviewer_profile_blocked(self, _auth, _api): + # close_pr is author-side: a reviewer profile must be told to stop. + with patch.dict(os.environ, self._env("reviewer-profile")): + res = mcp_server.gitea_resolve_task_capability(task="close_pr", remote="prgs") + self.assertFalse(res["allowed_in_current_session"]) + self.assertTrue(res["stop_required"]) + self.assertEqual(res["required_role_kind"], "author") + self.assertIn("author", res["exact_safe_next_action"].lower()) + + @patch("mcp_server.api_request", return_value={"login": "author-user"}) + @patch("mcp_server.get_auth_header", return_value="token author-pass") + def test_close_pr_known_and_lookalike_tasks_still_fail_closed(self, _auth, _api): + # close_pr resolves (no Unknown-task error); near-miss spellings keep + # failing closed so no untracked close fallback can be rationalized. + with patch.dict(os.environ, self._env("author-profile")): + res = mcp_server.gitea_resolve_task_capability(task="close_pr", remote="prgs") + self.assertEqual(res["required_operation_permission"], "gitea.pr.close") + for unknown in ("close_pull_request", "close", "pr_close"): + with self.assertRaises(ValueError): + mcp_server.gitea_resolve_task_capability(task=unknown, remote="prgs") + if __name__ == "__main__": unittest.main()