fix(review): harden #211 decision lock and CLI gate per review blockers

- Bind review decision state to in-process session (pid + profile lock); drop /tmp file path.
- Fail closed on review_pr.py CLI; route live reviews through gated MCP tools only.
- Require operator_authorized on review correction; allow re-mark after correction.
- Validate remote/org/repo on mark and submit; wire review mutation proof into build_final_report.
- Add security regression tests for spoofed locks, correction flow, and CLI bypass.

Refs #211
This commit is contained in:
2026-07-05 20:16:04 -04:00
parent 7489107768
commit 16f977fde0
9 changed files with 272 additions and 172 deletions
+25 -54
View File
@@ -34,8 +34,7 @@ class TestArgParsing(unittest.TestCase):
self.exists_patcher.start()
self.addCleanup(self.exists_patcher.stop)
@patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
def test_missing_pr_number_exits(self, _auth):
def test_missing_pr_number_exits(self):
with self.assertRaises(SystemExit):
review_pr.main([])
@@ -47,37 +46,21 @@ class TestAPIPayload(unittest.TestCase):
self.exists_patcher.start()
self.addCleanup(self.exists_patcher.stop)
@patch("review_pr.api_request")
@patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
def test_payload_fields_and_workflow(self, _auth, mock_api):
# Setup mock api_request to return PR details, then review response
mock_api.side_effect = [FAKE_PR_DATA, {}]
def test_review_submission_fails_closed_without_api_call(self):
# #211: live review submission must use gated MCP tools, not CLI POST.
import io
buf = io.StringIO()
with patch.object(sys, "stderr", buf):
rc = review_pr.main([
"--pr-number", "81",
"--event", "APPROVE",
"--body", "Approved and ready to merge",
])
self.assertEqual(rc, 2)
self.assertIn("disabled", buf.getvalue().lower())
self.assertIn("gitea_submit_pr_review", buf.getvalue())
rc = review_pr.main([
"--pr-number", "81",
"--event", "APPROVE",
"--body", "Approved and ready to merge",
])
self.assertEqual(rc, 0)
self.assertEqual(mock_api.call_count, 2)
# Verify first call: GET PR
first_call_args = mock_api.call_args_list[0]
self.assertEqual(first_call_args[0][0], "GET")
self.assertEqual(first_call_args[0][1], "https://gitea.dadeschools.net/api/v1/repos/Contractor/Timesheet/pulls/81")
# Verify second call: POST review
second_call_args = mock_api.call_args_list[1]
self.assertEqual(second_call_args[0][0], "POST")
self.assertEqual(second_call_args[0][1], "https://gitea.dadeschools.net/api/v1/repos/Contractor/Timesheet/pulls/81/reviews")
payload = second_call_args[0][3]
self.assertEqual(payload["event"], "APPROVE")
self.assertEqual(payload["body"], "Approved and ready to merge")
self.assertEqual(payload["commit_id"], "abcdef1234567890")
@patch("review_pr.api_request")
@patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
def test_merge_flag_fails_closed_without_api_call(self, _auth, mock_api):
def test_merge_flag_fails_closed_without_api_call(self):
# --merge is an ungated bypass and is disabled (#16). It must fail
# closed BEFORE any API call — no review, no merge.
rc = review_pr.main([
@@ -88,24 +71,15 @@ class TestAPIPayload(unittest.TestCase):
"--merge-method", "squash",
])
self.assertEqual(rc, 2)
self.assertEqual(mock_api.call_count, 0)
def test_merge_flag_message_points_to_gated_workflow(self):
from _pytest.monkeypatch import MonkeyPatch
import io
with patch("review_pr.get_auth_header", return_value=FAKE_CREDS), \
patch("review_pr.api_request") as mock_api:
buf = io.StringIO()
monkeypatch = MonkeyPatch()
monkeypatch.setattr(sys, "stderr", buf)
try:
rc = review_pr.main([
"--pr-number", "81", "--event", "APPROVE", "--merge",
])
finally:
monkeypatch.undo()
buf = io.StringIO()
with patch.object(sys, "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)
@@ -126,21 +100,18 @@ class TestMutationAuthorityLock(unittest.TestCase):
])
self.assertEqual(rc, 2)
msg = buf.getvalue().lower()
self.assertIn("disabled within mcp sessions", msg)
self.assertIn("disabled", msg)
self.assertIn("gitea_submit_pr_review", msg)
@patch("review_pr.api_request")
@patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
def test_cli_allowed_without_session_lock(self, _auth, mock_api):
# No lock in the environment = direct operator CLI use; the wall
# does not apply and the normal flow proceeds.
mock_api.side_effect = [FAKE_PR_DATA, {}]
def test_cli_disabled_even_without_session_lock(self):
# #211: CLI review submission is fail-closed regardless of session lock.
env = {k: v for k, v in os.environ.items()
if k != "GITEA_SESSION_PROFILE_LOCK"}
with patch.dict(os.environ, env, clear=True):
rc = review_pr.main([
"--pr-number", "81", "--event", "APPROVE",
])
self.assertEqual(rc, 0)
self.assertEqual(rc, 2)
if __name__ == "__main__":