diff --git a/mcp_server.py b/mcp_server.py index ba90eb3..35e3eb2 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -22,11 +22,42 @@ import contextlib import subprocess -LOCK_FILE = "/tmp/gitea_mutation_authority.lock" +# 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 -def record_mutation_authority(profile_name: str | None, identity: str | None, remote: str | None, task: str | None): - """Record the resolved capability context to fail-closed lock file.""" - data = { +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, @@ -35,48 +66,83 @@ def record_mutation_authority(profile_name: str | None, identity: str | None, re "task": task, "role_pivot_authorized": False, "role_pivot_record": None, + "pid": os.getpid(), } - try: - with open(LOCK_FILE, "w", encoding="utf-8") as f: - json.dump(data, f) - except Exception: - pass -def verify_mutation_authority(remote: str | None, host: str | None = None, required_role: str = "reviewer"): - """Verify that the current mutation matches the locked capability context.""" - if not os.path.exists(LOCK_FILE): - raise RuntimeError("Mutation authority lock is missing (fail closed)") - try: - with open(LOCK_FILE, "r", encoding="utf-8") as f: - data = json.load(f) - except Exception as e: - raise RuntimeError(f"Could not read mutation authority lock: {e} (fail closed)") +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. - if data.get("remote") != remote: - raise RuntimeError( - f"Mutation remote '{remote}' does not match locked remote '{data.get('remote')}' (fail closed)" - ) + 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") - h = host or (REMOTES.get(remote, {}).get("host") if remote in REMOTES else None) - active_identity = _authenticated_username(h) if h else None + 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}', identity: '{locked_identity}') (fail closed)" + f"does not match locked authority (profile: '{locked_profile}', " + f"identity: '{locked_identity}') (fail closed)" ) - # Check reviewer/author role pivot boundaries - if required_role == "reviewer" and "author" in str(data.get("initial_profile")).lower() and "reviewer" in str(active_profile).lower(): + # 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)" + "Attempted reviewer mutation from author session without " + "authorized role pivot (fail closed)" ) # Resolve the project root. MCP clients must launch this script directly with @@ -1050,12 +1116,6 @@ def gitea_submit_pr_review( } reasons = result["reasons"] - try: - verify_mutation_authority(remote, host, required_role="reviewer") - except RuntimeError as e: - reasons.append(str(e)) - return result - # Gate 1 — valid review action (no mutation on unknown action). if action not in _REVIEW_ACTIONS: reasons.append( @@ -1108,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: @@ -1376,12 +1447,6 @@ def gitea_merge_pr( } reasons = result["reasons"] - try: - verify_mutation_authority(remote, host, required_role="reviewer") - except RuntimeError as e: - reasons.append(str(e)) - return result - # Gate 1 — valid merge method (no API call on a bad method). if do not in _MERGE_METHODS: reasons.append( @@ -1460,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) @@ -3058,24 +3134,21 @@ def gitea_activate_profile( after_profile = get_profile()["profile_name"] after_identity = _authenticated_username(h) if h else None - # 4.5 Record pivot in mutation authority lock - if os.path.exists(LOCK_FILE): - try: - with open(LOCK_FILE, "r", encoding="utf-8") as f: - lock_data = json.load(f) - lock_data["current_profile"] = after_profile - lock_data["current_identity"] = after_identity - lock_data["role_pivot_authorized"] = True - lock_data["role_pivot_record"] = { - "from_profile": before_profile, - "to_profile": after_profile, - "from_identity": before_identity, - "to_identity": after_identity - } - with open(LOCK_FILE, "w", encoding="utf-8") as f: - json.dump(lock_data, f) - except Exception: - pass + # 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( @@ -3588,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") diff --git a/review_pr.py b/review_pr.py index 1614720..b4fe4dc 100755 --- a/review_pr.py +++ b/review_pr.py @@ -60,28 +60,31 @@ def main(argv=None): host, org, repo = resolve_remote(args) - # ── Mutation Authority context wall check (Issue #194) ── - LOCK_FILE = "/tmp/gitea_mutation_authority.lock" - - if os.path.exists(LOCK_FILE): + # ── 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: - with open(LOCK_FILE, "r", encoding="utf-8") as f: - lock_data = json.load(f) - - # Resolve current CLI profile - cli_profile = get_profile().get("profile_name") - locked_profile = lock_data.get("current_profile") - - if cli_profile != locked_profile: - print( - f"Mismatched active profile vs mutation profile (CLI override rejected): " - f"CLI profile '{cli_profile}' does not match locked active profile '{locked_profile}' (fail closed)", - file=sys.stderr - ) - return 3 + 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: diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..e08e2df --- /dev/null +++ b/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 diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index f1c1cce..d22a311 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -38,9 +38,6 @@ from gitea_auth import get_profile # noqa: E402 import gitea_config # noqa: E402 import mcp_server -# Globally disable verification check for existing isolated unit tests -_real_verify = mcp_server.verify_mutation_authority -mcp_server.verify_mutation_authority = lambda *args, **kwargs: None FAKE_AUTH = "Basic dGVzdDp0ZXN0" @@ -2304,87 +2301,119 @@ class TestIssueCommentPermissionSeparation(unittest.TestCase): class TestVerifyMutationAuthority(unittest.TestCase): - """Test verification lock logic under various configurations.""" + """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() - - # Restore real function for these tests - self._old_verify = mcp_server.verify_mutation_authority - mcp_server.verify_mutation_authority = _real_verify + 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() - mcp_server.verify_mutation_authority = self._old_verify - # Clean up lock file - if os.path.exists("/tmp/gitea_mutation_authority.lock"): - os.remove("/tmp/gitea_mutation_authority.lock") - def test_missing_lock_fails_closed(self): - if os.path.exists("/tmp/gitea_mutation_authority.lock"): - os.remove("/tmp/gitea_mutation_authority.lock") + 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: - _real_verify("prgs") - self.assertIn("lock is missing", str(ctx.exception)) + mcp_server.verify_mutation_authority("prgs") + self.assertIn("profile unresolved", str(ctx.exception)) def test_mismatched_remote_fails(self): - with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f: - json.dump({ - "remote": "dadeschools", - "current_profile": "prgs-reviewer", - "current_identity": "sysadmin" - }, f) + self._authority(remote="dadeschools") with self.assertRaises(RuntimeError) as ctx: - _real_verify("prgs") + mcp_server.verify_mutation_authority("prgs") self.assertIn("does not match locked remote", str(ctx.exception)) - def test_mismatched_profile_fails(self): - with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f: - json.dump({ - "remote": "prgs", - "current_profile": "prgs-author", - "current_identity": "jcwalker3" - }, f) - self.mock_profile.return_value = {"profile_name": "prgs-reviewer"} - self.mock_username.return_value = "sysadmin" - + 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: - _real_verify("prgs") + mcp_server.verify_mutation_authority("prgs") self.assertIn("does not match locked authority", str(ctx.exception)) - def test_author_to_reviewer_pivot_blocked_without_authorization(self): - with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f: - json.dump({ - "remote": "prgs", - "initial_profile": "prgs-author", - "initial_identity": "jcwalker3", - "current_profile": "prgs-reviewer", - "current_identity": "sysadmin", - "role_pivot_authorized": False - }, f) - self.mock_profile.return_value = {"profile_name": "prgs-reviewer"} - self.mock_username.return_value = "sysadmin" + 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: - _real_verify("prgs", required_role="reviewer") + mcp_server.verify_mutation_authority("prgs", required_role="reviewer") self.assertIn("without authorized role pivot", str(ctx.exception)) - def test_allowed_when_match(self): - with open("/tmp/gitea_mutation_authority.lock", "w", encoding="utf-8") as f: - json.dump({ - "remote": "prgs", - "initial_profile": "prgs-reviewer", - "initial_identity": "sysadmin", - "current_profile": "prgs-reviewer", - "current_identity": "sysadmin", - "role_pivot_authorized": False - }, f) - self.mock_profile.return_value = {"profile_name": "prgs-reviewer"} - self.mock_username.return_value = "sysadmin" + 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") - # Should pass without exception - _real_verify("prgs") + 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") diff --git a/tests/test_review_pr.py b/tests/test_review_pr.py index 2575b8f..1344745 100644 --- a/tests/test_review_pr.py +++ b/tests/test_review_pr.py @@ -112,39 +112,23 @@ class TestAPIPayload(unittest.TestCase): class TestMutationAuthorityLock(unittest.TestCase): - """Issue #194: verify that the CLI tool rejects profile overrides when mismatched with lock.""" + """#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_profile_mismatch(self, mock_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"} - - original_exists = os.path.exists - def conditional_exists(path): - if "gitea_mutation_authority.lock" in str(path): - return True - return original_exists(path) - - original_open = open - def conditional_open(file, *args, **kwargs): - if "gitea_mutation_authority.lock" in str(file): - return io.StringIO('{"current_profile": "prgs-author"}') - return original_open(file, *args, **kwargs) - - from _pytest.monkeypatch import MonkeyPatch import io buf = io.StringIO() - monkeypatch = MonkeyPatch() - monkeypatch.setattr(sys, "stderr", buf) - - with patch("os.path.exists", side_effect=conditional_exists), \ - patch("builtins.open", side_effect=conditional_open): - try: - rc = review_pr.main([ - "--pr-number", "81", "--event", "APPROVE", - ]) - finally: - monkeypatch.undo() - + 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) @@ -155,21 +139,22 @@ class TestMutationAuthorityLock(unittest.TestCase): 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) - original_exists = os.path.exists - def conditional_exists(path): - if "gitea_mutation_authority.lock" in str(path): - return True - return original_exists(path) - - original_open = open - def conditional_open(file, *args, **kwargs): - if "gitea_mutation_authority.lock" in str(file): - return io.StringIO('{"current_profile": "prgs-reviewer"}') - return original_open(file, *args, **kwargs) - - with patch("os.path.exists", side_effect=conditional_exists), \ - patch("builtins.open", side_effect=conditional_open): + @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", ])