fix: close ungated CLI merge bypasses in review_pr.py and merge_pr.py (#16)
Reviewer found the MCP merge surface is gated but two local CLI scripts remain ungated merge paths that LLM automations in this project have been using. Close them (Option B — minimal safe fix; full gated CLI merge left to a follow-up): - review_pr.py: `--merge` now fails closed BEFORE any API call with a clear message directing callers to the gated `gitea_merge_pr` MCP workflow. The review-only path is unchanged. The merge execution block was removed. - merge_pr.py: main() is now a fail-closed no-op — reads no credentials and makes no merge API call; prints that merge is only available via the gated workflow. - README: the `review_pr.py` row and Quick Examples no longer advertise a CLI `--merge` path; added an audit-logging clarification that #16 returns structured gate/merge results but does not add durable audit logging, which is tracked by #18. Tests updated/added: - test_review_pr.py: `--merge` fails closed with no API call; message points to the gated workflow. - test_merge_pr.py: merge fails closed with no API call, even with --force/--do/--title/--message; message points to the gated workflow. - test_mcp_server.py: README no longer advertises the ungated CLI merge example. The gated MCP `gitea_merge_pr` is unchanged and still gated; `gitea_review_pr` still fails closed on merge=True; `gitea_submit_pr_review` still cannot merge. No secrets, auth headers, raw env, or credential paths are exposed. No Jenkins/Ops/GlitchTip/Release/deploy/CI behavior added. #17/#18 not started. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -180,6 +180,9 @@ Notes:
|
|||||||
can inspect which runtime it is talking to before deciding to act.
|
can inspect which runtime it is talking to before deciding to act.
|
||||||
- See [`docs/gitea-execution-profiles.md`](docs/gitea-execution-profiles.md) for
|
- See [`docs/gitea-execution-profiles.md`](docs/gitea-execution-profiles.md) for
|
||||||
the full profile model.
|
the full profile model.
|
||||||
|
- **Audit logging:** #16 returns structured gate/merge results but does not add
|
||||||
|
durable audit logging. Durable audit logging for Gitea MCP mutating actions is
|
||||||
|
tracked by #18.
|
||||||
</details>
|
</details>
|
||||||
|
|
||||||
<details>
|
<details>
|
||||||
@@ -204,7 +207,7 @@ The MCP tools can also be used as standalone CLI scripts:
|
|||||||
| `create_issue.py` | Create an issue (`--remote`, `--title`, `--body`, `--body-file`) |
|
| `create_issue.py` | Create an issue (`--remote`, `--title`, `--body`, `--body-file`) |
|
||||||
| `create_pr.py` | Open a Pull Request (`--remote`, `--title`, `--head`, `--base`) |
|
| `create_pr.py` | Open a Pull Request (`--remote`, `--title`, `--head`, `--base`) |
|
||||||
| `edit_pr.py` | Edit a Pull Request (`--title`, `--body`, `--body-file`, etc.) |
|
| `edit_pr.py` | Edit a Pull Request (`--title`, `--body`, `--body-file`, etc.) |
|
||||||
| `review_pr.py` | Review and sign-off on a pull request (with optional merge) |
|
| `review_pr.py` | Review/sign-off on a pull request (`--merge` is disabled — fails closed; merge only via gated `gitea_merge_pr`) |
|
||||||
| `close_issue.py` | Close a specific issue |
|
| `close_issue.py` | Close a specific issue |
|
||||||
| `mark_issue.py` | Claim/release an issue via `status:in-progress` label |
|
| `mark_issue.py` | Claim/release an issue via `status:in-progress` label |
|
||||||
| `manage_labels.py` | Create label set and apply label mappings (`--dry` to preview) |
|
| `manage_labels.py` | Create label set and apply label mappings (`--dry` to preview) |
|
||||||
@@ -225,8 +228,9 @@ The MCP tools can also be used as standalone CLI scripts:
|
|||||||
# Edit a PR's description or title
|
# Edit a PR's description or title
|
||||||
./edit_pr.py 155 --body "Updated description wording"
|
./edit_pr.py 155 --body "Updated description wording"
|
||||||
|
|
||||||
# Review and approve a PR, then automatically merge it
|
# Review and approve a PR (review only — CLI merge is disabled; use the
|
||||||
./review_pr.py --pr-number 12 --event APPROVE --body "Approved" --merge
|
# gated gitea_merge_pr MCP workflow to merge)
|
||||||
|
./review_pr.py --pr-number 12 --event APPROVE --body "Approved"
|
||||||
|
|
||||||
# Close issue #5
|
# Close issue #5
|
||||||
./close_issue.py 5
|
./close_issue.py 5
|
||||||
|
|||||||
+22
-25
@@ -1,7 +1,13 @@
|
|||||||
#!/usr/bin/env python3
|
#!/usr/bin/env python3
|
||||||
"""Merge a Gitea pull request.
|
"""Merge a Gitea pull request — DISABLED (#16).
|
||||||
|
|
||||||
Usage:
|
Direct CLI merge is a fail-closed no-op. LLM automations in this project were
|
||||||
|
using this script as an ungated merge bypass, so it no longer performs a merge.
|
||||||
|
Merge is only available through the gated `gitea_merge_pr` MCP workflow (#16),
|
||||||
|
which enforces identity/profile/eligibility, explicit confirmation, expected
|
||||||
|
head SHA checking, and self-merge protection.
|
||||||
|
|
||||||
|
Usage (fails closed):
|
||||||
merge_pr.py --pr-number 84 --do squash --remote dadeschools
|
merge_pr.py --pr-number 84 --do squash --remote dadeschools
|
||||||
"""
|
"""
|
||||||
import os
|
import os
|
||||||
@@ -29,29 +35,20 @@ def main(argv=None):
|
|||||||
parser.add_argument("--force", action="store_true", help="Force merge, ignoring status checks.")
|
parser.add_argument("--force", action="store_true", help="Force merge, ignoring status checks.")
|
||||||
args = parser.parse_args(argv)
|
args = parser.parse_args(argv)
|
||||||
|
|
||||||
host, org, repo = resolve_remote(args)
|
# Fail closed: direct CLI merge is disabled (#16). No credentials are read
|
||||||
auth = get_auth_header(host)
|
# and no merge API call is made. Merge is only available through the gated
|
||||||
if not auth:
|
# `gitea_merge_pr` MCP workflow, which enforces identity/profile/eligibility,
|
||||||
print(f"Could not get credentials for {host}.", file=sys.stderr)
|
# explicit confirmation, expected head SHA checking, and self-merge
|
||||||
return 1
|
# protection.
|
||||||
|
print(
|
||||||
url = f"{repo_api_url(host, org, repo)}/pulls/{args.pr_number}/merge"
|
f"Direct CLI merge is disabled (#16). PR #{args.pr_number} was NOT "
|
||||||
payload = {
|
"merged. Merge is only available through the gated workflow (MCP tool "
|
||||||
"Do": args.do,
|
"'gitea_merge_pr'), which enforces identity/profile/eligibility, "
|
||||||
"force_merge": args.force,
|
"explicit confirmation, expected head SHA checking, and self-merge "
|
||||||
}
|
"protection.",
|
||||||
if args.title:
|
file=sys.stderr,
|
||||||
payload["MergeTitleField"] = args.title
|
)
|
||||||
if args.message:
|
return 2
|
||||||
payload["MergeMessageField"] = args.message
|
|
||||||
|
|
||||||
try:
|
|
||||||
api_request("POST", url, auth, payload)
|
|
||||||
print(f"Successfully merged PR #{args.pr_number} using '{args.do}' method.")
|
|
||||||
return 0
|
|
||||||
except Exception as e:
|
|
||||||
print(f"Error merging PR #{args.pr_number}: {e}", file=sys.stderr)
|
|
||||||
return 1
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
|
|||||||
+18
-19
@@ -37,6 +37,22 @@ def main(argv=None):
|
|||||||
help="Merge method/style to use if merging (default: merge).")
|
help="Merge method/style to use if merging (default: merge).")
|
||||||
args = parser.parse_args(argv)
|
args = parser.parse_args(argv)
|
||||||
|
|
||||||
|
# Fail closed: direct CLI merge is disabled (#16). LLM automations were
|
||||||
|
# using this flag as an ungated merge bypass. Merge is only available via
|
||||||
|
# the gated `gitea_merge_pr` MCP workflow, which enforces
|
||||||
|
# identity/profile/eligibility, explicit confirmation, expected head SHA,
|
||||||
|
# and self-merge protection. No API call is made here.
|
||||||
|
if args.merge:
|
||||||
|
print(
|
||||||
|
"Direct CLI merge is disabled. Merge is only available through the "
|
||||||
|
"gated #16 workflow (MCP tool 'gitea_merge_pr'), which enforces "
|
||||||
|
"identity/profile/eligibility, explicit confirmation, expected head "
|
||||||
|
"SHA checking, and self-merge protection. Re-run without --merge to "
|
||||||
|
"submit a review only.",
|
||||||
|
file=sys.stderr,
|
||||||
|
)
|
||||||
|
return 2
|
||||||
|
|
||||||
host, org, repo = resolve_remote(args)
|
host, org, repo = resolve_remote(args)
|
||||||
|
|
||||||
body = args.body
|
body = args.body
|
||||||
@@ -80,25 +96,8 @@ def main(argv=None):
|
|||||||
print(f"Error submitting review: {e}", file=sys.stderr)
|
print(f"Error submitting review: {e}", file=sys.stderr)
|
||||||
return 1
|
return 1
|
||||||
|
|
||||||
# 3. Merge PR if --merge is requested and event is APPROVE
|
# Merge is intentionally not performed here — see the fail-closed guard
|
||||||
if args.merge:
|
# above. Use the gated `gitea_merge_pr` MCP workflow (#16) to merge.
|
||||||
if args.event != "APPROVE":
|
|
||||||
print("Warning: Skipping merge because review event is not 'APPROVE'.", file=sys.stderr)
|
|
||||||
return 0
|
|
||||||
|
|
||||||
merge_url = f"{repo_api_url(host, org, repo)}/pulls/{args.pr_number}/merge"
|
|
||||||
merge_payload = {
|
|
||||||
"Do": args.merge_method,
|
|
||||||
"force_merge": False
|
|
||||||
}
|
|
||||||
|
|
||||||
try:
|
|
||||||
api_request("POST", merge_url, auth, merge_payload)
|
|
||||||
print(f"Successfully merged PR #{args.pr_number} using '{args.merge_method}' method.")
|
|
||||||
except Exception as e:
|
|
||||||
print(f"Error merging PR: {e}", file=sys.stderr)
|
|
||||||
return 1
|
|
||||||
|
|
||||||
return 0
|
return 0
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -602,6 +602,17 @@ class TestNoUngatedMergePath(unittest.TestCase):
|
|||||||
merge_src = inspect.getsource(mcp_server.gitea_merge_pr)
|
merge_src = inspect.getsource(mcp_server.gitea_merge_pr)
|
||||||
self.assertIn("/merge\"", merge_src)
|
self.assertIn("/merge\"", merge_src)
|
||||||
|
|
||||||
|
def test_readme_no_longer_advertises_ungated_cli_merge(self):
|
||||||
|
import pathlib
|
||||||
|
readme = (pathlib.Path(__file__).resolve().parent.parent
|
||||||
|
/ "README.md").read_text(encoding="utf-8")
|
||||||
|
# The old ungated example command must be gone.
|
||||||
|
self.assertNotIn('--body "Approved" --merge', readme)
|
||||||
|
# And the gated messaging / audit deferral must be present.
|
||||||
|
self.assertIn("disabled", readme.lower())
|
||||||
|
self.assertIn("gitea_merge_pr", readme)
|
||||||
|
self.assertIn("#18", readme)
|
||||||
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
# Review PR
|
# Review PR
|
||||||
|
|||||||
+26
-27
@@ -15,49 +15,48 @@ FAKE_CREDS = "Basic dGVzdHVzZXI6dGVzdHBhc3M="
|
|||||||
|
|
||||||
class TestArgParsing(unittest.TestCase):
|
class TestArgParsing(unittest.TestCase):
|
||||||
|
|
||||||
@patch("merge_pr.api_request")
|
|
||||||
@patch("merge_pr.get_auth_header", return_value=FAKE_CREDS)
|
|
||||||
def test_minimal_required_args(self, _auth, mock_api):
|
|
||||||
rc = merge_pr.main(["--pr-number", "81"])
|
|
||||||
self.assertEqual(rc, 0)
|
|
||||||
mock_api.assert_called_once()
|
|
||||||
|
|
||||||
def test_missing_pr_number_exits(self):
|
def test_missing_pr_number_exits(self):
|
||||||
with self.assertRaises(SystemExit):
|
with self.assertRaises(SystemExit):
|
||||||
merge_pr.main([])
|
merge_pr.main([])
|
||||||
|
|
||||||
|
|
||||||
class TestAPIPayload(unittest.TestCase):
|
class TestMergeDisabled(unittest.TestCase):
|
||||||
|
"""Direct CLI merge is disabled (#16) — fails closed, no API call."""
|
||||||
|
|
||||||
@patch("merge_pr.api_request")
|
@patch("merge_pr.api_request")
|
||||||
@patch("merge_pr.get_auth_header", return_value=FAKE_CREDS)
|
@patch("merge_pr.get_auth_header", return_value=FAKE_CREDS)
|
||||||
def test_payload_fields(self, _auth, mock_api):
|
def test_merge_fails_closed_without_api_call(self, _auth, mock_api):
|
||||||
|
rc = merge_pr.main(["--pr-number", "81"])
|
||||||
|
self.assertEqual(rc, 2)
|
||||||
|
mock_api.assert_not_called()
|
||||||
|
|
||||||
|
@patch("merge_pr.api_request")
|
||||||
|
@patch("merge_pr.get_auth_header", return_value=FAKE_CREDS)
|
||||||
|
def test_no_merge_even_with_force_and_method(self, _auth, mock_api):
|
||||||
rc = merge_pr.main([
|
rc = merge_pr.main([
|
||||||
"--pr-number", "81",
|
"--pr-number", "81",
|
||||||
"--do", "squash",
|
"--do", "squash",
|
||||||
"--title", "Squash title",
|
"--title", "Squash title",
|
||||||
"--message", "Squash message",
|
"--message", "Squash message",
|
||||||
"--force",
|
"--force",
|
||||||
|
"--remote", "prgs",
|
||||||
])
|
])
|
||||||
self.assertEqual(rc, 0)
|
self.assertEqual(rc, 2)
|
||||||
mock_api.assert_called_once()
|
mock_api.assert_not_called()
|
||||||
method, url, auth, payload = mock_api.call_args[0]
|
|
||||||
self.assertEqual(method, "POST")
|
|
||||||
self.assertEqual(auth, FAKE_CREDS)
|
|
||||||
self.assertEqual(payload["Do"], "squash")
|
|
||||||
self.assertEqual(payload["MergeTitleField"], "Squash title")
|
|
||||||
self.assertEqual(payload["MergeMessageField"], "Squash message")
|
|
||||||
self.assertEqual(payload["force_merge"], True)
|
|
||||||
|
|
||||||
@patch("merge_pr.api_request")
|
def test_message_points_to_gated_workflow(self):
|
||||||
@patch("merge_pr.get_auth_header", return_value=FAKE_CREDS)
|
import io
|
||||||
def test_url_construction(self, _auth, mock_api):
|
import contextlib
|
||||||
merge_pr.main(["--pr-number", "81", "--remote", "prgs"])
|
with patch("merge_pr.get_auth_header", return_value=FAKE_CREDS), \
|
||||||
url = mock_api.call_args[0][1]
|
patch("merge_pr.api_request") as mock_api:
|
||||||
self.assertEqual(
|
buf = io.StringIO()
|
||||||
url,
|
with contextlib.redirect_stderr(buf):
|
||||||
"https://gitea.prgs.cc/api/v1/repos/Scaled-Tech-Consulting/Timesheet/pulls/81/merge"
|
rc = merge_pr.main(["--pr-number", "81"])
|
||||||
)
|
self.assertEqual(rc, 2)
|
||||||
|
mock_api.assert_not_called()
|
||||||
|
msg = buf.getvalue().lower()
|
||||||
|
self.assertIn("disabled", msg)
|
||||||
|
self.assertIn("gitea_merge_pr", msg)
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
|
|||||||
+21
-14
@@ -65,27 +65,34 @@ class TestAPIPayload(unittest.TestCase):
|
|||||||
|
|
||||||
@patch("review_pr.api_request")
|
@patch("review_pr.api_request")
|
||||||
@patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
|
@patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
|
||||||
def test_approve_and_merge_workflow(self, _auth, mock_api):
|
def test_merge_flag_fails_closed_without_api_call(self, _auth, mock_api):
|
||||||
# Setup mock api_request to return PR details, review response, and merge response
|
# --merge is an ungated bypass and is disabled (#16). It must fail
|
||||||
mock_api.side_effect = [FAKE_PR_DATA, {}, {}]
|
# closed BEFORE any API call — no review, no merge.
|
||||||
|
|
||||||
rc = review_pr.main([
|
rc = review_pr.main([
|
||||||
"--pr-number", "81",
|
"--pr-number", "81",
|
||||||
"--event", "APPROVE",
|
"--event", "APPROVE",
|
||||||
"--body", "Approved",
|
"--body", "Approved",
|
||||||
"--merge",
|
"--merge",
|
||||||
"--merge-method", "squash"
|
"--merge-method", "squash",
|
||||||
])
|
])
|
||||||
self.assertEqual(rc, 0)
|
self.assertEqual(rc, 2)
|
||||||
self.assertEqual(mock_api.call_count, 3)
|
self.assertEqual(mock_api.call_count, 0)
|
||||||
|
|
||||||
# Verify third call: POST merge
|
def test_merge_flag_message_points_to_gated_workflow(self):
|
||||||
third_call_args = mock_api.call_args_list[2]
|
import io
|
||||||
self.assertEqual(third_call_args[0][0], "POST")
|
import contextlib
|
||||||
self.assertEqual(third_call_args[0][1], "https://gitea.dadeschools.net/api/v1/repos/Contractor/Timesheet/pulls/81/merge")
|
with patch("review_pr.get_auth_header", return_value=FAKE_CREDS), \
|
||||||
payload = third_call_args[0][3]
|
patch("review_pr.api_request") as mock_api:
|
||||||
self.assertEqual(payload["Do"], "squash")
|
buf = io.StringIO()
|
||||||
self.assertEqual(payload["force_merge"], False)
|
with contextlib.redirect_stderr(buf):
|
||||||
|
rc = review_pr.main([
|
||||||
|
"--pr-number", "81", "--event", "APPROVE", "--merge",
|
||||||
|
])
|
||||||
|
self.assertEqual(rc, 2)
|
||||||
|
self.assertEqual(mock_api.call_count, 0)
|
||||||
|
msg = buf.getvalue().lower()
|
||||||
|
self.assertIn("disabled", msg)
|
||||||
|
self.assertIn("gitea_merge_pr", msg)
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
|
|||||||
Reference in New Issue
Block a user