Add explicit close_pr capability resolution and gated PR close path (Issue #216)
PR closure had no first-class capability: agents could close PRs through gitea_edit_pr(state=closed) with no close-specific capability proof, and gitea_resolve_task_capability(close_pr) failed as unknown, leaving the broad edit path as an untracked close fallback. Add close_pr to the resolver TASK_MAP (gitea.pr.close, author-side) and gate gitea_edit_pr(state=closed) on the same operation: without it the close fails closed before any auth or API call, with reasons and a structured permission_report; with it the close proceeds and is audited as a distinct close_pr action carrying the required capability. Reject invalid state values outright so case variants cannot bypass the gate. Generalize the profile gate helper (_profile_operation_gate) and document the PR comment / PR edit / PR close capability split. Co-Authored-By: Claude Fable 5 <[email protected]>
This commit is contained in:
@@ -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:
|
||||
|
||||
+56
-11
@@ -841,7 +841,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,
|
||||
@@ -1073,11 +1073,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.
|
||||
@@ -1085,7 +1094,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
|
||||
@@ -1096,6 +1108,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
|
||||
@@ -1103,12 +1118,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
|
||||
@@ -1869,13 +1905,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()
|
||||
@@ -1927,7 +1964,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,
|
||||
@@ -1989,7 +2026,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")
|
||||
@@ -3353,6 +3390,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",
|
||||
|
||||
@@ -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()
|
||||
@@ -1621,7 +1621,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()
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user