diff --git a/mcp_server.py b/mcp_server.py index 5f26704..35e3eb2 100644 --- a/mcp_server.py +++ b/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") diff --git a/review_pr.py b/review_pr.py index 464d348..b4fe4dc 100755 --- a/review_pr.py +++ b/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 == "-": 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 d39ff87..d22a311 100644 --- a/tests/test_mcp_server.py +++ b/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") diff --git a/tests/test_review_pr.py b/tests/test_review_pr.py index d28ff01..1344745 100644 --- a/tests/test_review_pr.py +++ b/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()