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]>
89 lines
3.1 KiB
Python
89 lines
3.1 KiB
Python
"""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,
|
|
} |