Merge pull request 'feat(reviewer-workflow): add hard wall against reviewer mutations through alternate profile or CLI side-channel' (#203) from feat/issue-194-reviewer-mutation-boundary into master

mcp_server.py

@@ -16,10 +16,135 @@ Configuration (mcp_config.json):
 import os
 import re
 import sys
+import json
 import functools
 import contextlib
 import subprocess
 
+
+# Mutation-authority record (#199, refs #194). Deliberately in-process, NOT a
+# file: a /tmp lock is host-global, writable (spoofable) by any local process,
+# goes silently stale across sessions, and races between concurrent agent
+# sessions. This record lives and dies with the MCP server process, so it can
+# never be forged from outside or leak between sessions. The CLI side-channel
+# (a subprocess overriding GITEA_MCP_PROFILE to escalate roles) is covered by
+# SESSION_PROFILE_LOCK_ENV below: the server exports its launch profile into
+# the environment, children inherit it, and reviewer CLIs (review_pr.py)
+# refuse to run under a different resolved profile.
+_MUTATION_AUTHORITY: dict | None = None
+
+SESSION_PROFILE_LOCK_ENV = "GITEA_SESSION_PROFILE_LOCK"
+
+
+def _export_session_profile_lock():
+    """Export this process's launch profile for child CLI processes.
+
+    setdefault: an already-locked environment (outer session) wins, so a
+    nested launch cannot relabel the session.
+    """
+    try:
+        name = (get_profile().get("profile_name") or "").strip()
+        if name:
+            os.environ.setdefault(SESSION_PROFILE_LOCK_ENV, name)
+    except Exception:
+        # Profile resolution problems surface loudly on the first real call;
+        # the lock export must not mask them here at import time.
+        pass
+
+
+def record_mutation_authority(profile_name: str | None, identity: str | None,
+                              remote: str | None, task: str | None):
+    """Record the resolved capability context for this process (fail-closed
+    consumers in verify_mutation_authority)."""
+    global _MUTATION_AUTHORITY
+    _MUTATION_AUTHORITY = {
+        "initial_profile": profile_name,
+        "initial_identity": identity,
+        "current_profile": profile_name,
+        "current_identity": identity,
+        "remote": remote,
+        "task": task,
+        "role_pivot_authorized": False,
+        "role_pivot_record": None,
+        "pid": os.getpid(),
+    }
+
+
+def verify_mutation_authority(remote: str | None, host: str | None = None,
+                              required_role: str = "reviewer",
+                              active_identity: str | None = None):
+    """Verify the current mutation matches this process's recorded authority.
+
+    Fail-closed rules:
+    - No recorded authority (or one from another process after a fork) is
+      seeded from the live, config-resolved context — the approved preflight
+      path (whoami → eligibility → mutation) therefore works without an
+      explicit resolve call — but an unresolvable profile still fails closed.
+    - GITEA_SESSION_PROFILE_LOCK (set by the launching session) must match
+      the active profile: a mid-session GITEA_MCP_PROFILE override flips the
+      active profile away from the lock and is refused.
+    - Remote, profile, and identity must match the recorded authority.
+    - An author→reviewer pivot requires an authorized pivot record
+      (gitea_activate_profile in dynamic mode); it can never be improvised.
+    """
+    global _MUTATION_AUTHORITY
+
+    profile = get_profile()
+    active_profile = profile.get("profile_name")
+    if active_identity is None:
+        # Callers that already proved the identity (eligibility gate) pass it
+        # in; otherwise resolve it here (cached, read-only).
+        h = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None)
+        active_identity = _authenticated_username(h) if h else None
+
+    if not active_profile:
+        raise RuntimeError(
+            "Mutation authority unavailable: active profile unresolved (fail closed)"
+        )
+
+    session_lock = (os.environ.get(SESSION_PROFILE_LOCK_ENV) or "").strip()
+    if session_lock and session_lock != active_profile:
+        raise RuntimeError(
+            f"Active profile '{active_profile}' does not match the session "
+            f"profile lock '{session_lock}' — profile side-channel override "
+            "rejected (fail closed)"
+        )
+
+    data = _MUTATION_AUTHORITY
+    if data is None or data.get("pid") != os.getpid():
+        # First mutation gate in this process (approved preflight path):
+        # seed the authority from the live context, then verify against it.
+        record_mutation_authority(
+            active_profile, active_identity, remote, "seeded-at-mutation-gate"
+        )
+        data = _MUTATION_AUTHORITY
+
+    if data.get("remote") != remote:
+        raise RuntimeError(
+            f"Mutation remote '{remote}' does not match locked remote "
+            f"'{data.get('remote')}' (fail closed)"
+        )
+
+    locked_profile = data.get("current_profile")
+    locked_identity = data.get("current_identity")
+    if active_profile != locked_profile or active_identity != locked_identity:
+        raise RuntimeError(
+            f"Mutation profile '{active_profile}' or identity '{active_identity}' "
+            f"does not match locked authority (profile: '{locked_profile}', "
+            f"identity: '{locked_identity}') (fail closed)"
+        )
+
+    # Reviewer/author role pivot boundary: only an authorized pivot
+    # (recorded by gitea_activate_profile) may cross author → reviewer.
+    if (required_role == "reviewer"
+            and "author" in str(data.get("initial_profile")).lower()
+            and "reviewer" in str(active_profile).lower()):
+        if not data.get("role_pivot_authorized"):
+            raise RuntimeError(
+                "Attempted reviewer mutation from author session without "
+                "authorized role pivot (fail closed)"
+            )
+
 # Resolve the project root. MCP clients must launch this script directly with
 # the venv interpreter (venv/bin/python3) — see the config example above. We do
 # NOT os.execv() to re-point the interpreter: replacing the process after the
@@ -1043,6 +1168,17 @@ def gitea_submit_pr_review(
         reasons.append("PR head SHA unavailable (fail closed)")
         return result
 
+    # Gate 5 — in-process mutation authority (#199): the last check before
+    # the mutating POST, using the identity the eligibility gate proved.
+    # A profile/identity flip or side-channel override between preflight
+    # and mutation fails closed here.
+    try:
+        verify_mutation_authority(remote, host, required_role="reviewer",
+                                  active_identity=auth_user)
+    except RuntimeError as e:
+        reasons.append(str(e))
+        return result
+
     # All gates passed — perform the single mutating call.
     h, o, r = _resolve(remote, host, org, repo)
     try:
@@ -1389,6 +1525,17 @@ def gitea_merge_pr(
         reasons.append("self-merge blocked (authenticated user is PR author)")
         return result
 
+    # Gate 7 — in-process mutation authority (#199): the last check before
+    # the merge mutation, using the identity the eligibility gate proved.
+    # A profile/identity flip or side-channel override between preflight
+    # and merge fails closed here.
+    try:
+        verify_mutation_authority(remote, host, required_role="reviewer",
+                                  active_identity=auth_user)
+    except RuntimeError as e:
+        reasons.append(str(e))
+        return result
+
     # All gates passed — perform the single merge mutation.
     try:
         auth = _auth(h)
@@ -2987,6 +3134,22 @@ def gitea_activate_profile(
     after_profile = get_profile()["profile_name"]
     after_identity = _authenticated_username(h) if h else None
 
+    # 4.5 Record the authorized pivot in the in-process mutation authority
+    # and keep the session profile lock in sync — this is the ONLY path that
+    # may authorize an author→reviewer role pivot.
+    if _MUTATION_AUTHORITY is not None:
+        _MUTATION_AUTHORITY["current_profile"] = after_profile
+        _MUTATION_AUTHORITY["current_identity"] = after_identity
+        _MUTATION_AUTHORITY["role_pivot_authorized"] = True
+        _MUTATION_AUTHORITY["role_pivot_record"] = {
+            "from_profile": before_profile,
+            "to_profile": after_profile,
+            "from_identity": before_identity,
+            "to_identity": after_identity,
+        }
+    if os.environ.get(SESSION_PROFILE_LOCK_ENV) and after_profile:
+        os.environ[SESSION_PROFILE_LOCK_ENV] = after_profile
+
     # 5. Audit the switch if auditing is on
     _audit(
         "activate_profile",
@@ -3476,6 +3639,8 @@ def gitea_resolve_task_capability(
             "STOP: the active profile cannot perform the requested task; "
             "follow exact_safe_next_action instead of improvising.")
 
+    record_mutation_authority(profile["profile_name"], username, remote if remote in REMOTES else None, task)
+
     return {
         "requested_task": task,
         "required_operation_permission": required_permission,
@@ -3496,4 +3661,8 @@ def gitea_resolve_task_capability(
 # ── Entry point ───────────────────────────────────────────────────────────────
 
 if __name__ == "__main__":
+    # Lock this session's launch profile into the environment so child CLI
+    # processes (e.g. review_pr.py) can detect and refuse profile
+    # side-channel overrides (#199).
+    _export_session_profile_lock()
     mcp.run(transport="stdio")

review_pr.py

@@ -24,7 +24,7 @@ venv_python = os.path.join(PROJECT_ROOT, "venv", "bin", "python3")
 if os.path.exists(venv_python) and sys.executable != venv_python:
     os.execv(venv_python, [venv_python] + sys.argv)
 
-from gitea_auth import get_auth_header, resolve_remote, add_remote_args, api_request, repo_api_url
+from gitea_auth import get_auth_header, resolve_remote, add_remote_args, api_request, repo_api_url, get_profile
 
 
 def main(argv=None):
@@ -60,6 +60,32 @@ def main(argv=None):
 
     host, org, repo = resolve_remote(args)
 
+    # ── Reviewer mutation side-channel wall (#199, refs #194) ──
+    # The launching MCP session exports GITEA_SESSION_PROFILE_LOCK with the
+    # profile it was started with; child processes inherit it. If this CLI
+    # resolves a different profile — e.g. an ad-hoc GITEA_MCP_PROFILE
+    # override escalating an author-bound session to reviewer — refuse
+    # before any API call. No lock in the environment means no session
+    # context (direct operator CLI use), which stays allowed. Unlike a /tmp
+    # lock file, the environment is per-process-tree: other sessions cannot
+    # spoof it and it cannot go stale across sessions.
+    session_lock = (os.environ.get("GITEA_SESSION_PROFILE_LOCK") or "").strip()
+    if session_lock:
+        try:
+            cli_profile = (get_profile().get("profile_name") or "").strip()
+        except Exception as e:
+            print(f"Mutation authority check failed: {e}", file=sys.stderr)
+            return 3
+        if cli_profile != session_lock:
+            print(
+                f"Mismatched active profile vs session profile lock "
+                f"(CLI override rejected): CLI profile '{cli_profile}' does "
+                f"not match locked session profile '{session_lock}' "
+                f"(fail closed)",
+                file=sys.stderr,
+            )
+            return 3
+
     body = args.body
     if args.body_file:
         if args.body_file == "-":

tests/conftest.py

@@ -0,0 +1,28 @@
+"""Shared pytest fixtures for the Gitea-Tools test suite."""
+import sys
+
+import pytest
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+
+@pytest.fixture(autouse=True)
+def _reset_mutation_authority(monkeypatch):
+    """Isolate the in-process mutation authority between tests (#199).
+
+    The mutation-authority gate stays LIVE in every test — this fixture only
+    clears the per-process record and the session profile lock so one test's
+    seeded authority (or an intentionally mismatched one) cannot leak into
+    the next test. It must never replace verify_mutation_authority with a
+    no-op: individual tests that need a specific authority state set it up
+    explicitly.
+    """
+    monkeypatch.delenv("GITEA_SESSION_PROFILE_LOCK", raising=False)
+    try:
+        import mcp_server
+    except Exception:
+        yield
+        return
+    monkeypatch.setattr(mcp_server, "_MUTATION_AUTHORITY", None)
+    monkeypatch.setattr(mcp_server, "_IDENTITY_CACHE", {})
+    yield

tests/test_mcp_server.py

@@ -37,6 +37,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"
 
 
@@ -2296,3 +2298,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")

tests/test_review_pr.py

@@ -2,6 +2,8 @@
 
 Mocks api_request and credentials.
 """
+import io
+import os
 import sys
 import unittest
 from unittest.mock import patch
@@ -27,6 +29,11 @@ FAKE_PR_DATA = {
 
 class TestArgParsing(unittest.TestCase):
 
+    def setUp(self):
+        self.exists_patcher = patch("os.path.exists", return_value=False)
+        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):
         with self.assertRaises(SystemExit):
@@ -35,6 +42,11 @@ class TestArgParsing(unittest.TestCase):
 
 class TestAPIPayload(unittest.TestCase):
 
+    def setUp(self):
+        self.exists_patcher = patch("os.path.exists", return_value=False)
+        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):
@@ -99,5 +111,55 @@ class TestAPIPayload(unittest.TestCase):
         self.assertIn("gitea_merge_pr", msg)
 
 
+class TestMutationAuthorityLock(unittest.TestCase):
+    """#199 (refs #194): the CLI refuses to run under a profile that differs
+    from the session profile lock exported by the launching MCP session."""
+
+    @patch("review_pr.get_profile")
+    def test_cli_blocked_on_session_lock_mismatch(self, mock_get_profile):
+        # An author-bound session exported the lock; the CLI resolves a
+        # reviewer profile (GITEA_MCP_PROFILE side-channel override) — reject
+        # before any API call.
+        mock_get_profile.return_value = {"profile_name": "prgs-reviewer"}
+        import io
+        buf = io.StringIO()
+        with patch.dict(os.environ,
+                        {"GITEA_SESSION_PROFILE_LOCK": "prgs-author"}), \
+             patch.object(sys, "stderr", buf):
+            rc = review_pr.main([
+                "--pr-number", "81", "--event", "APPROVE",
+            ])
+        self.assertEqual(rc, 3)
+        msg = buf.getvalue().lower()
+        self.assertIn("cli override rejected", msg)
+
+    @patch("review_pr.get_profile")
+    @patch("review_pr.api_request")
+    @patch("review_pr.get_auth_header", return_value=FAKE_CREDS)
+    def test_cli_allowed_on_profile_match(self, _auth, mock_api, mock_get_profile):
+        mock_get_profile.return_value = {"profile_name": "prgs-reviewer"}
+        mock_api.side_effect = [FAKE_PR_DATA, {}]
+        with patch.dict(os.environ,
+                        {"GITEA_SESSION_PROFILE_LOCK": "prgs-reviewer"}):
+            rc = review_pr.main([
+                "--pr-number", "81", "--event", "APPROVE",
+            ])
+        self.assertEqual(rc, 0)
+
+    @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, {}]
+        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)
+
+
 if __name__ == "__main__":
     unittest.main()