gitea_audit.py

@@ -163,12 +163,13 @@ def audit_enabled():
 def build_event(*, action, result, remote=None, server=None, repository=None,
                 issue_number=None, pr_number=None, profile_name=None,
                 audit_label=None, authenticated_username=None, target_branch=None,
-                head_sha=None, reason=None, request_metadata=None, now=None):
+                head_sha=None, reason=None, request_metadata=None, now=None,
+                mcp_namespace=None, task_role=None, operation=None):
     """Build a redacted, JSON-able audit record for a mutating action."""
     ts = now or datetime.datetime.now(datetime.timezone.utc)
     if isinstance(ts, datetime.datetime):
         ts = ts.isoformat()
-    return {
+    event = {
         "timestamp": ts,
         "action": action,
         "action_type": "mutating",
@@ -186,6 +187,13 @@ def build_event(*, action, result, remote=None, server=None, repository=None,
         "reason": _redact_str(reason) if reason else reason,
         "request_metadata": redact(request_metadata) if request_metadata is not None else None,
     }
+    if mcp_namespace is not None:
+        event["mcp_namespace"] = mcp_namespace
+    if task_role is not None:
+        event["task_role"] = task_role
+    if operation is not None:
+        event["operation"] = operation
+    return event
 
 
 def write_event(event, path=None):

gitea_mcp_server.py

@@ -257,6 +257,7 @@ import gitea_config  # noqa: E402
 import capability_stop_terminal  # noqa: E402
 import issue_duplicate_gate  # noqa: E402
 import role_session_router  # noqa: E402
+import role_namespace_gate  # noqa: E402
 import task_capability_map  # noqa: E402
 
 
@@ -415,7 +416,8 @@ def _authenticated_username(host: str):
 
 def _audit(action: str, *, host, remote, result, org=None, repo=None,
            reason=None, request_metadata=None, issue_number=None,
-           pr_number=None, target_branch=None, head_sha=None, username=_UNSET):
+           pr_number=None, target_branch=None, head_sha=None, username=_UNSET,
+           mutation_task: str | None = None):
     """Emit one audit record for a mutating action. No-op unless auditing is on.
 
     Never raises — auditing must not break the action it records.
@@ -426,6 +428,10 @@ def _audit(action: str, *, host, remote, result, org=None, repo=None,
         profile = get_profile()
         if username is _UNSET:
             username = _authenticated_username(host) if host else None
+        ns_ctx = {}
+        if mutation_task:
+            ns_ctx = role_namespace_gate.mutation_audit_context(
+                mutation_task, profile, remote=remote, repository=repo)
         event = gitea_audit.build_event(
             action=action,
             result=result,
@@ -441,6 +447,9 @@ def _audit(action: str, *, host, remote, result, org=None, repo=None,
             head_sha=head_sha,
             reason=reason,
             request_metadata=request_metadata,
+            mcp_namespace=ns_ctx.get("mcp_namespace"),
+            task_role=ns_ctx.get("task_role"),
+            operation=ns_ctx.get("operation"),
         )
         gitea_audit.write_event(event)
     except Exception:
@@ -556,6 +565,10 @@ def gitea_create_issue(
             "number": None,
             "reasons": block_reasons,
         }
+    blocked = _namespace_mutation_block(
+        "create_issue", number=None, remote=remote)
+    if blocked:
+        return blocked
     blocked = _profile_permission_block(
         task_capability_map.required_permission("create_issue"),
         number=None,
@@ -599,7 +612,7 @@ def gitea_create_issue(
         raise
     _audit("create_issue", host=h, remote=remote, org=o, repo=r,
            result=gitea_audit.SUCCEEDED, issue_number=data["number"],
-           request_metadata={"title": title})
+           request_metadata={"title": title}, mutation_task="create_issue")
     return _with_optional_url({"number": data["number"]}, data.get("html_url"))
 
 
@@ -707,6 +720,25 @@ def gitea_create_pr(
     Returns:
         dict with 'number' of the created PR ('url' only with the reveal opt-in).
     """
+    ok, block_reasons = role_session_router.check_author_mutation_after_reviewer_stop(
+        "create_pr")
+    if not ok:
+        return {
+            "success": False,
+            "performed": False,
+            "number": None,
+            "reasons": block_reasons,
+        }
+    blocked = _namespace_mutation_block(
+        "create_pr", number=None, remote=remote)
+    if blocked:
+        return blocked
+    blocked = _profile_permission_block(
+        task_capability_map.required_permission("create_pr"),
+        number=None,
+    )
+    if blocked:
+        return blocked
     verify_preflight_purity(remote)
     h, o, r = _resolve(remote, host, org, repo)
 
@@ -753,11 +785,13 @@ def gitea_create_pr(
     except Exception as exc:
         _audit("create_pr", host=h, remote=remote, org=o, repo=r,
                result=gitea_audit.FAILED, reason=_redact(str(exc)),
-               target_branch=head, request_metadata=meta)
+               target_branch=head, request_metadata=meta,
+               mutation_task="create_pr")
         raise
     _audit("create_pr", host=h, remote=remote, org=o, repo=r,
            result=gitea_audit.SUCCEEDED, pr_number=data["number"],
-           target_branch=head, request_metadata=meta)
+           target_branch=head, request_metadata=meta,
+           mutation_task="create_pr")
     return _with_optional_url({"number": data["number"]}, data.get("html_url"))
 
 
@@ -2728,6 +2762,43 @@ def _profile_permission_block(required_operation: str, **extra_fields) -> dict |
     return blocked
 
 
+def _namespace_mutation_block(mutation_task: str, **extra_fields) -> dict | None:
+    """Reviewer/author namespace alignment gate (#209)."""
+    try:
+        profile = get_profile()
+    except Exception as exc:
+        return {
+            "success": False,
+            "performed": False,
+            "reasons": [
+                f"profile could not be resolved (fail closed): {_redact(str(exc))}"],
+            **extra_fields,
+        }
+    ok, reasons = role_namespace_gate.check_author_mutation_namespace(
+        mutation_task, profile)
+    if ok:
+        return None
+    blocked = {
+        "success": False,
+        "performed": False,
+        "reasons": reasons,
+        "namespace_block": True,
+        "mcp_namespace": role_namespace_gate.infer_mcp_namespace(
+            profile.get("profile_name")),
+        **extra_fields,
+    }
+    _audit(
+        mutation_task,
+        host=None,
+        remote=extra_fields.get("remote"),
+        result=gitea_audit.BLOCKED,
+        repo=extra_fields.get("repository"),
+        reason="; ".join(reasons),
+        mutation_task=mutation_task,
+    )
+    return blocked
+
+
 @mcp.tool()
 def gitea_list_issue_comments(
     issue_number: int,

review_proofs.py

@@ -1139,6 +1139,41 @@ def assess_role_route_handoff(report_text, route_result=None):
     }
 
 
+def assess_mutation_namespace_handoff(
+    *,
+    declared_role: str | None = None,
+    namespaces_used: list[str] | None = None,
+    capability_profile: str | None = None,
+):
+    """Issue #209: handoff role and capability proof must match mutation namespace."""
+    reasons = []
+    role = (declared_role or "").strip().lower()
+    namespaces = list(namespaces_used or [])
+
+    reviewer_namespaces = [
+        ns for ns in namespaces if "reviewer" in (ns or "").lower()
+    ]
+    if role == "author" and reviewer_namespaces:
+        reasons.append(
+            "author handoff reported reviewer MCP namespace for mutations")
+
+    cap = (capability_profile or "").strip().lower()
+    if cap and "author" in cap and reviewer_namespaces:
+        reasons.append(
+            "capability proof profile does not match mutation namespace "
+            f"(profile={capability_profile}, namespaces={reviewer_namespaces})")
+
+    blocked = bool(reasons)
+    return {
+        "valid": not blocked,
+        "blocked": blocked,
+        "complete": not blocked,
+        "downgraded": blocked,
+        "reasons": reasons,
+        "missing_fields": [] if not blocked else ["namespace parity"],
+    }
+
+
 def assess_capability_stop_terminal_report(
     report_text,
     *,

role_namespace_gate.py

@@ -0,0 +1,89 @@
+"""Block author mutations from reviewer-bound MCP namespaces (#209).
+
+Every mutation must prove that the active profile role, inferred MCP
+namespace, and declared task role align. Reviewer sessions cannot silently
+perform author-side work (especially PR creation).
+"""
+
+from __future__ import annotations
+
+import gitea_config
+import role_session_router
+
+
+def infer_mcp_namespace(profile_name: str | None) -> str:
+    """Map a runtime profile name to its MCP namespace label."""
+    lower = (profile_name or "").strip().lower()
+    if "reviewer" in lower and "author" not in lower:
+        return "gitea-reviewer"
+    if "author" in lower:
+        return "gitea-author"
+    return profile_name or "gitea-default"
+
+
+def derive_role_kind(allowed, forbidden=()) -> str:
+    """Classify the active profile the same way as ``mcp_server._role_kind``."""
+
+    def can(op):
+        return gitea_config.check_operation(op, allowed, forbidden)[0]
+
+    review = can("gitea.pr.approve") or can("gitea.pr.merge")
+    author = can("gitea.pr.create") or can("gitea.branch.push")
+    if review and author:
+        return "mixed"
+    if review:
+        return "reviewer"
+    if author:
+        return "author"
+    return "limited"
+
+
+def check_author_mutation_namespace(
+    mutation_task: str,
+    profile: dict,
+) -> tuple[bool, list[str]]:
+    """Return (allowed, reasons). Fail closed on reviewer/author mismatch."""
+    required_role = role_session_router.required_role_for_task(mutation_task)
+    if required_role != "author":
+        return True, []
+
+    allowed = profile.get("allowed_operations") or []
+    forbidden = profile.get("forbidden_operations") or []
+    active_role = derive_role_kind(allowed, forbidden)
+    profile_name = profile.get("profile_name") or ""
+    namespace = infer_mcp_namespace(profile_name)
+
+    if mutation_task == "create_pr":
+        if active_role == "reviewer" or namespace == "gitea-reviewer":
+            return False, [
+                "author mutation 'create_pr' blocked in reviewer MCP namespace "
+                f"({namespace}); launch gitea-author",
+            ]
+
+    if active_role == "reviewer" or namespace == "gitea-reviewer":
+        if mutation_task == "create_issue":
+            ok, _ = gitea_config.check_operation(
+                "gitea.issue.create", allowed, forbidden)
+            if ok:
+                return True, []
+        return False, [
+            f"author mutation '{mutation_task}' blocked: active session is "
+            f"reviewer-bound ({profile_name} / {namespace})",
+        ]
+
+    return True, []
+
+
+def mutation_audit_context(mutation_task: str, profile: dict, *,
+                           remote=None, repository=None) -> dict:
+    """Structured mutation metadata for audit records (#209)."""
+    allowed = profile.get("allowed_operations") or []
+    forbidden = profile.get("forbidden_operations") or []
+    return {
+        "mcp_namespace": infer_mcp_namespace(profile.get("profile_name")),
+        "profile_name": profile.get("profile_name"),
+        "task_role": role_session_router.required_role_for_task(mutation_task),
+        "operation": mutation_task,
+        "remote": remote,
+        "repository": repository,
+    }

tests/test_role_namespace_gate.py

@@ -0,0 +1,215 @@
+"""Tests for reviewer/author namespace mismatch walls (#209)."""
+import json
+import os
+import sys
+import tempfile
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import gitea_audit
+import mcp_server
+import role_namespace_gate
+from review_proofs import assess_mutation_namespace_handoff
+
+CONFIG = {
+    "version": 2,
+    "contexts": {
+        "ctx": {
+            "enabled": True,
+            "gitea": {"enabled": True, "base_url": "https://gitea.example.com"},
+        }
+    },
+    "profiles": {
+        "prgs-author": {
+            "enabled": True,
+            "context": "ctx",
+            "role": "author",
+            "username": "jcwalker3",
+            "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
+            "allowed_operations": [
+                "gitea.read", "gitea.issue.create", "gitea.pr.create",
+                "gitea.branch.push", "gitea.issue.comment",
+            ],
+            "forbidden_operations": [
+                "gitea.pr.approve", "gitea.pr.merge", "gitea.pr.review",
+            ],
+            "execution_profile": "prgs-author",
+        },
+        "prgs-reviewer": {
+            "enabled": True,
+            "context": "ctx",
+            "role": "reviewer",
+            "username": "sysadmin",
+            "auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"},
+            "allowed_operations": [
+                "gitea.read", "gitea.pr.review", "gitea.pr.approve",
+                "gitea.pr.merge", "gitea.issue.comment",
+            ],
+            "forbidden_operations": [
+                "gitea.pr.create", "gitea.branch.push", "gitea.issue.create",
+            ],
+            "execution_profile": "prgs-reviewer",
+        },
+        "prgs-reviewer-issue-writer": {
+            "enabled": True,
+            "context": "ctx",
+            "role": "reviewer",
+            "username": "sysadmin",
+            "auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"},
+            "allowed_operations": [
+                "gitea.read", "gitea.pr.review", "gitea.issue.create",
+            ],
+            "forbidden_operations": ["gitea.pr.create"],
+            "execution_profile": "prgs-reviewer-issue-writer",
+        },
+    },
+    "rules": {"allow_runtime_switching": False},
+}
+
+ISSUE_WRITE_ENV = {"GITEA_ALLOWED_OPERATIONS": "gitea.issue.create"}
+
+
+class NamespaceGateBase(unittest.TestCase):
+    def setUp(self):
+        self._remotes = patch.dict(mcp_server.REMOTES, {
+            "prgs": {
+                "host": "gitea.example.com",
+                "org": "Scaled-Tech-Consulting",
+                "repo": "Gitea-Tools",
+            },
+        })
+        self._remotes.start()
+        mcp_server._IDENTITY_CACHE.clear()
+        self._dir = tempfile.TemporaryDirectory()
+        self.config_path = os.path.join(self._dir.name, "profiles.json")
+        with open(self.config_path, "w", encoding="utf-8") as fh:
+            fh.write(json.dumps(CONFIG))
+        self.audit_path = os.path.join(self._dir.name, "audit.log")
+
+    def tearDown(self):
+        self._remotes.stop()
+        mcp_server._IDENTITY_CACHE.clear()
+        self._dir.cleanup()
+
+    def _env(self, profile, *, audit=False):
+        env = {
+            "GITEA_MCP_CONFIG": self.config_path,
+            "GITEA_MCP_PROFILE": profile,
+            "GITEA_TOKEN_AUTHOR": "author-pass",
+            "GITEA_TOKEN_REVIEWER": "reviewer-pass",
+        }
+        if audit:
+            env["GITEA_AUDIT_LOG"] = self.audit_path
+        return env
+
+
+class TestNamespaceInference(unittest.TestCase):
+    def test_reviewer_and_author_namespaces(self):
+        self.assertEqual(
+            role_namespace_gate.infer_mcp_namespace("prgs-reviewer"),
+            "gitea-reviewer",
+        )
+        self.assertEqual(
+            role_namespace_gate.infer_mcp_namespace("prgs-author"),
+            "gitea-author",
+        )
+
+
+class TestReviewerAuthorMutationBlocks(NamespaceGateBase):
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
+    @patch("mcp_server.api_get_all", return_value=[])
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
+    def test_reviewer_namespace_create_issue_blocked(
+        self, _auth, mock_api, _get_all, _role
+    ):
+        with patch.dict(os.environ, self._env("prgs-reviewer"), clear=True):
+            result = mcp_server.gitea_create_issue(
+                title="should not post", remote="prgs")
+        self.assertFalse(result["success"])
+        self.assertFalse(result["performed"])
+        self.assertTrue(result.get("namespace_block"))
+        mock_api.assert_not_called()
+
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
+    @patch("mcp_server.api_get_all", return_value=[])
+    @patch("mcp_server.api_request")
+    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
+    def test_reviewer_namespace_create_pr_blocked(
+        self, _auth, mock_api, _get_all, _role
+    ):
+        with patch.dict(os.environ, self._env("prgs-reviewer"), clear=True):
+            result = mcp_server.gitea_create_pr(
+                title="feat: x Closes #1",
+                head="feat/issue-1-x",
+                body="Closes #1",
+                remote="prgs",
+            )
+        self.assertFalse(result.get("success", True))
+        self.assertFalse(result.get("performed", True))
+        self.assertTrue(result.get("namespace_block"))
+        mock_api.assert_not_called()
+
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
+    @patch("mcp_server.api_get_all", return_value=[])
+    @patch("mcp_server.api_request", return_value={"number": 12})
+    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
+    def test_reviewer_with_explicit_issue_create_allowed(
+        self, _auth, mock_api, _get_all, _role
+    ):
+        with patch.dict(os.environ, self._env("prgs-reviewer-issue-writer"),
+                        clear=True):
+            result = mcp_server.gitea_create_issue(
+                title="escalation issue", remote="prgs")
+        self.assertEqual(result.get("number"), 12)
+        mock_api.assert_called()
+
+
+class TestHandoffNamespaceParity(unittest.TestCase):
+    def test_author_handoff_with_reviewer_namespace_invalid(self):
+        result = assess_mutation_namespace_handoff(
+            declared_role="author",
+            namespaces_used=["gitea-reviewer"],
+        )
+        self.assertFalse(result["valid"])
+        self.assertTrue(result["blocked"])
+
+    def test_capability_author_profile_with_reviewer_namespace_blocked(self):
+        result = assess_mutation_namespace_handoff(
+            declared_role="author",
+            namespaces_used=["gitea-reviewer"],
+            capability_profile="prgs-author",
+        )
+        self.assertFalse(result["valid"])
+        blob = " ".join(result["reasons"])
+        self.assertIn("capability proof profile", blob)
+
+
+class TestMutationAuditFields(NamespaceGateBase):
+    @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+           return_value=(True, []))
+    @patch("mcp_server.api_get_all", return_value=[])
+    @patch("mcp_server.api_request", return_value={"number": 3})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_successful_create_issue_audit_includes_namespace_fields(
+        self, _auth, mock_api, _get_all, _role
+    ):
+        env = {**self._env("prgs-author", audit=True), **ISSUE_WRITE_ENV}
+        with patch.dict(os.environ, env, clear=True):
+            mcp_server.gitea_create_issue(title="audited", remote="prgs")
+        with open(self.audit_path, encoding="utf-8") as fh:
+            record = json.loads(fh.readline())
+        self.assertEqual(record["mcp_namespace"], "gitea-author")
+        self.assertEqual(record["task_role"], "author")
+        self.assertEqual(record["operation"], "create_issue")
+        self.assertEqual(record["profile_name"], "prgs-author")
+        self.assertEqual(record["remote"], "prgs")
+
+
+if __name__ == "__main__":
+    unittest.main()