Merge branch 'prgs/master' into feat/issue-211-single-terminal-review-decision

tests/test_mcp_server.py

@@ -42,6 +42,8 @@ from mcp_server import (  # noqa: E402
 from gitea_auth import get_profile  # noqa: E402
 import gitea_config  # noqa: E402
 
+import mcp_server
+
 FAKE_AUTH = "Basic dGVzdDp0ZXN0"
 
 
@@ -1776,7 +1778,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()
@@ -2451,3 +2454,122 @@ class TestIssueCommentPermissionSeparation(unittest.TestCase):
             "gitea.issue.comment", reviewer["allowed_operations"],
             reviewer.get("forbidden_operations", []))
         self.assertTrue(ok)
+
+
+class TestVerifyMutationAuthority(unittest.TestCase):
+    """In-process mutation authority (#199, refs #194).
+
+    The authority record lives in mcp_server._MUTATION_AUTHORITY (per
+    process, reset between tests by conftest); the CLI side-channel is
+    covered by the GITEA_SESSION_PROFILE_LOCK environment lock. There is no
+    lock file — nothing here touches /tmp.
+    """
+
+    def setUp(self):
+        self.patch_profile = patch("mcp_server.get_profile")
+        self.mock_profile = self.patch_profile.start()
+        self.patch_username = patch("mcp_server._authenticated_username")
+        self.mock_username = self.patch_username.start()
+        self.mock_profile.return_value = {"profile_name": "prgs-reviewer"}
+        self.mock_username.return_value = "sysadmin"
+
+    def tearDown(self):
+        self.patch_profile.stop()
+        self.patch_username.stop()
+
+    def _authority(self, **overrides):
+        data = {
+            "initial_profile": "prgs-reviewer",
+            "initial_identity": "sysadmin",
+            "current_profile": "prgs-reviewer",
+            "current_identity": "sysadmin",
+            "remote": "prgs",
+            "task": "review_pr",
+            "role_pivot_authorized": False,
+            "role_pivot_record": None,
+            "pid": os.getpid(),
+        }
+        data.update(overrides)
+        mcp_server._MUTATION_AUTHORITY = data
+
+    def test_missing_authority_seeds_from_live_context(self):
+        # Approved preflight path (whoami → eligibility → mutation): the
+        # first mutation gate seeds the authority instead of failing closed,
+        # so the standard reviewer workflow keeps working.
+        mcp_server._MUTATION_AUTHORITY = None
+        mcp_server.verify_mutation_authority("prgs")
+        seeded = mcp_server._MUTATION_AUTHORITY
+        self.assertIsNotNone(seeded)
+        self.assertEqual(seeded["current_profile"], "prgs-reviewer")
+        self.assertEqual(seeded["current_identity"], "sysadmin")
+        self.assertEqual(seeded["remote"], "prgs")
+
+    def test_unresolved_profile_fails_closed(self):
+        self.mock_profile.return_value = {}
+        mcp_server._MUTATION_AUTHORITY = None
+        with self.assertRaises(RuntimeError) as ctx:
+            mcp_server.verify_mutation_authority("prgs")
+        self.assertIn("profile unresolved", str(ctx.exception))
+
+    def test_mismatched_remote_fails(self):
+        self._authority(remote="dadeschools")
+        with self.assertRaises(RuntimeError) as ctx:
+            mcp_server.verify_mutation_authority("prgs")
+        self.assertIn("does not match locked remote", str(ctx.exception))
+
+    def test_profile_flip_after_record_fails(self):
+        # Authority was recorded as author; the active profile now resolves
+        # as reviewer (e.g. an env-var flip mid-session) — refuse.
+        self._authority(
+            initial_profile="prgs-author",
+            initial_identity="jcwalker3",
+            current_profile="prgs-author",
+            current_identity="jcwalker3",
+        )
+        with self.assertRaises(RuntimeError) as ctx:
+            mcp_server.verify_mutation_authority("prgs")
+        self.assertIn("does not match locked authority", str(ctx.exception))
+
+    def test_session_lock_env_mismatch_fails(self):
+        # The launching session locked the environment to the author
+        # profile; the active profile resolves as reviewer — side-channel
+        # override rejected even with a matching in-process authority.
+        self._authority()
+        with patch.dict(os.environ,
+                        {"GITEA_SESSION_PROFILE_LOCK": "prgs-author"}):
+            with self.assertRaises(RuntimeError) as ctx:
+                mcp_server.verify_mutation_authority("prgs")
+        self.assertIn("side-channel override rejected", str(ctx.exception))
+
+    def test_foreign_pid_authority_is_not_trusted(self):
+        # An authority record from another process (fork leftovers) is
+        # discarded and reseeded from the live context, never reused.
+        self._authority(current_profile="prgs-author", pid=os.getpid() + 1)
+        mcp_server.verify_mutation_authority("prgs")
+        self.assertEqual(
+            mcp_server._MUTATION_AUTHORITY["current_profile"], "prgs-reviewer"
+        )
+        self.assertEqual(mcp_server._MUTATION_AUTHORITY["pid"], os.getpid())
+
+    def test_author_to_reviewer_pivot_blocked_without_authorization(self):
+        self._authority(
+            initial_profile="prgs-author",
+            initial_identity="jcwalker3",
+        )
+        with self.assertRaises(RuntimeError) as ctx:
+            mcp_server.verify_mutation_authority("prgs", required_role="reviewer")
+        self.assertIn("without authorized role pivot", str(ctx.exception))
+
+    def test_authorized_pivot_is_allowed(self):
+        self._authority(
+            initial_profile="prgs-author",
+            initial_identity="jcwalker3",
+            role_pivot_authorized=True,
+        )
+        mcp_server.verify_mutation_authority("prgs", required_role="reviewer")
+
+    def test_allowed_when_match(self):
+        self._authority()
+        with patch.dict(os.environ,
+                        {"GITEA_SESSION_PROFILE_LOCK": "prgs-reviewer"}):
+            mcp_server.verify_mutation_authority("prgs")