diff --git a/gitea_audit.py b/gitea_audit.py index f41a2f6..ceb73e9 100644 --- a/gitea_audit.py +++ b/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): diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index b1961fe..d6934da 100644 --- a/gitea_mcp_server.py +++ b/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, diff --git a/review_proofs.py b/review_proofs.py index f044b3b..7412cc0 100644 --- a/review_proofs.py +++ b/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, *, diff --git a/role_namespace_gate.py b/role_namespace_gate.py new file mode 100644 index 0000000..7ed1073 --- /dev/null +++ b/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, + } \ No newline at end of file diff --git a/tests/test_role_namespace_gate.py b/tests/test_role_namespace_gate.py new file mode 100644 index 0000000..564e39c --- /dev/null +++ b/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() \ No newline at end of file