sysadminandClaude Opus 4.8
cbce6cb2e7
Add role_namespace_gate to fail closed when reviewer-bound sessions attempt author mutations (create_pr always blocked; create_issue only when profile explicitly grants gitea.issue.create). Extend mutation audit records with mcp_namespace, task_role, and operation fields. Add handoff parity checks in review_proofs for author role vs reviewer namespace mismatches. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
215 lines
8.1 KiB
Python
215 lines
8.1 KiB
Python
"""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() |