feat(reviewer-workflow): add hard wall against reviewer mutations through alternate profile or CLI side-channel

mcp_server.py

@@ -16,10 +16,69 @@ Configuration (mcp_config.json):
 import os
 import re
 import sys
+import json
 import functools
 import contextlib
 import subprocess
 
+
+LOCK_FILE = "/tmp/gitea_mutation_authority.lock"
+
+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 = {
+        "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,
+    }
+    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)")
+
+    if data.get("remote") != remote:
+        raise RuntimeError(
+            f"Mutation remote '{remote}' does not match locked remote '{data.get('remote')}' (fail closed)"
+        )
+
+    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
+
+    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)"
+        )
+
+    # 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():
+        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
@@ -991,6 +1050,12 @@ 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(
@@ -1311,6 +1376,12 @@ 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(
@@ -2987,6 +3058,25 @@ 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
+
     # 5. Audit the switch if auditing is on
     _audit(
         "activate_profile",
@@ -3476,6 +3566,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,