Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 77 Pull Requests 10 Actions Packages Projects Releases Wiki Activity

feat: isolate MCP workspace binding across role namespaces (Closes #510)

Browse Source 
Namespace-scoped workspace resolution prevents GITEA_AUTHOR_WORKTREE from
poisoning reviewer, merger, and reconciler purity checks. Each role uses
its own worktree env var with actionable binding-source errors and safe
reconnect guidance. Preserves #274/#475 author and control-checkout guards.
This commit is contained in:
sysadmin
2026-07-08 03:57:38 -04:00
parent f89949cea9
commit f1ad665242
7 changed files with 772 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gitea_mcp_server.py
+169 -43
View File
@@ -174,12 +174,76 @@ _preflight_reviewer_violation_files: list[str] = []
ACTIVE_WORKTREE_ENV = "GITEA_ACTIVE_WORKTREE"
AUTHOR_WORKTREE_ENV = "GITEA_AUTHOR_WORKTREE"
REVIEWER_WORKTREE_ENV = "GITEA_REVIEWER_WORKTREE"
MERGER_WORKTREE_ENV = "GITEA_MERGER_WORKTREE"
RECONCILER_WORKTREE_ENV = "GITEA_RECONCILER_WORKTREE"
import namespace_workspace_binding as nwb # noqa: E402
def _preflight_in_test_mode() -> bool:
return "pytest" in sys.modules or "unittest" in sys.modules
def _reviewer_session_worktree() -> str | None:
import reviewer_pr_lease as _reviewer_pr_lease
session = _reviewer_pr_lease.get_session_lease()
if not session:
return None
worktree = (session.get("worktree") or "").strip()
return worktree or None
def _effective_workspace_role() -> str:
"""Resolve the namespace key used for workspace binding (#510)."""
profile = get_profile()
role = _preflight_resolved_role
if not role:
role = _role_kind(
profile.get("allowed_operations") or [],
profile.get("forbidden_operations") or [],
)
return nwb.normalize_role_kind(
role,
profile_name=profile.get("profile_name"),
)
def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str:
"""Resolve the namespace-scoped workspace root inspected by pre-flight guards."""
role = _effective_workspace_role()
workspace, _source = nwb.resolve_namespace_workspace(
role_kind=role,
worktree_path=worktree_path,
process_project_root=PROJECT_ROOT,
session_lease_worktree=(
_reviewer_session_worktree() if role in {"reviewer", "merger"} else None
),
profile_name=get_profile().get("profile_name"),
)
return workspace
def _resolve_namespace_mutation_context(worktree_path: str | None = None) -> dict:
"""Canonical namespace workspace + repository root for guards (#460/#510)."""
role = _effective_workspace_role()
return nwb.resolve_namespace_mutation_context(
role_kind=role,
worktree_path=worktree_path,
process_project_root=PROJECT_ROOT,
session_lease_worktree=(
_reviewer_session_worktree() if role in {"reviewer", "merger"} else None
),
profile_name=get_profile().get("profile_name"),
)
def _resolve_author_mutation_context(worktree_path: str | None = None) -> dict:
"""Backward-compatible alias for namespace workspace context."""
return _resolve_namespace_mutation_context(worktree_path)
def _ensure_process_start_porcelain() -> str:
"""Capture the shared-worktree baseline once per MCP process (#252)."""
global _process_start_porcelain
@@ -188,26 +252,6 @@ def _ensure_process_start_porcelain() -> str:
return _process_start_porcelain
def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str:
"""Resolve the workspace root inspected by pre-flight guards."""
return author_mutation_worktree.resolve_mutation_workspace(
worktree_path,
PROJECT_ROOT,
active_worktree_env=os.environ.get(ACTIVE_WORKTREE_ENV),
author_worktree_env=os.environ.get(AUTHOR_WORKTREE_ENV),
)
def _resolve_author_mutation_context(worktree_path: str | None = None) -> dict:
"""Canonical workspace + repository root for runtime_context and guards (#460)."""
return author_mutation_worktree.resolve_author_mutation_context(
worktree_path,
PROJECT_ROOT,
active_worktree_env=os.environ.get(ACTIVE_WORKTREE_ENV),
author_worktree_env=os.environ.get(AUTHOR_WORKTREE_ENV),
)
def _get_git_root(path: str) -> str | None:
try:
res = subprocess.run(
@@ -275,7 +319,7 @@ def _format_preflight_files(files: list[str]) -> str:
def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[str]) -> dict:
ctx = _resolve_author_mutation_context(worktree_path)
ctx = _resolve_namespace_mutation_context(worktree_path)
workspace = ctx["workspace_path"]
inspected_root = _get_git_root(workspace)
process_root = ctx["process_project_root"]
@@ -293,6 +337,9 @@ def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[st
"dirty_files": list(dirty_files),
"dirty_scope": dirty_scope,
"workspace_roots_aligned": ctx["roots_aligned"],
"workspace_role_kind": ctx.get("workspace_role_kind"),
"workspace_binding_source": ctx.get("workspace_binding_source"),
"ignored_bindings": list(ctx.get("ignored_bindings") or []),
}
if not ctx["roots_aligned"]:
details["workspace_root_mismatch"] = (
@@ -303,13 +350,19 @@ def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[st
def _format_preflight_workspace_details(details: dict) -> str:
return (
f"MCP server process root: {details.get('mcp_server_process_root')}; "
f"active task workspace root: {details.get('active_task_workspace_root')}; "
f"inspected git root: {details.get('inspected_git_root')}; "
f"dirty files: {_format_preflight_files(details.get('dirty_files') or [])}; "
f"dirty scope: {details.get('dirty_scope')}"
)
parts = [
f"MCP server process root: {details.get('mcp_server_process_root')}",
f"active task workspace root: {details.get('active_task_workspace_root')}",
f"workspace role: {details.get('workspace_role_kind')}",
f"binding source: {details.get('workspace_binding_source')}",
f"inspected git root: {details.get('inspected_git_root')}",
f"dirty files: {_format_preflight_files(details.get('dirty_files') or [])}",
f"dirty scope: {details.get('dirty_scope')}",
]
ignored = details.get("ignored_bindings") or []
if ignored:
parts.append(f"ignored foreign bindings: {'; '.join(ignored)}")
return "; ".join(parts)
def assess_preflight_status(worktree_path: str | None = None) -> dict:
@@ -332,6 +385,25 @@ def assess_preflight_status(worktree_path: str | None = None) -> dict:
"Active task workspace has tracked file edits before mutation "
f"({_format_preflight_workspace_details(workspace_details)})"
)
role = _effective_workspace_role()
if role in nwb.NON_AUTHOR_ROLES:
binding = nwb.assess_metadata_only_worktree_binding(
role_kind=role,
declared_worktree_path=worktree_path,
mutation_workspace=_resolve_preflight_workspace_path(worktree_path),
process_project_root=PROJECT_ROOT,
profile_name=get_profile().get("profile_name"),
)
if binding.get("block"):
reasons.append(binding["reasons"][0])
reasons.append(
nwb.format_namespace_workspace_binding_error(
role_kind=role,
workspace_path=binding["mutation_workspace"],
binding_source="MCP server process root (default)",
reasons=binding.get("reasons"),
)
)
return {
"preflight_ready": not reasons,
"preflight_block_reasons": reasons,
@@ -419,13 +491,14 @@ def record_preflight_check(type_name: str, resolved_role: str | None = None):
def _enforce_branches_only_author_mutation(worktree_path: str | None = None) -> None:
"""#274: author file/branch mutations must run from a branches/ worktree.
Reviewer and reconciler roles are exempt: reconciler ``close_pr`` is a
Gitea metadata mutation and must not require ``GITEA_AUTHOR_WORKTREE``
(#468).
Reviewer, merger, and reconciler roles are exempt: reconciler ``close_pr``
is a Gitea metadata mutation and must not require ``GITEA_AUTHOR_WORKTREE``
(#468). Non-author namespaces use dedicated workspace env vars (#510).
"""
if _preflight_resolved_role in ("reviewer", "reconciler"):
role = _effective_workspace_role()
if role in nwb.NON_AUTHOR_ROLES:
return
ctx = _resolve_author_mutation_context(worktree_path)
ctx = _resolve_namespace_mutation_context(worktree_path)
workspace = ctx["workspace_path"]
git_state = issue_lock_worktree.read_worktree_git_state(workspace)
assessment = author_mutation_worktree.assess_author_mutation_worktree(
@@ -459,11 +532,12 @@ def verify_preflight_purity(remote: str | None = None, worktree_path: str | None
"Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)"
)
ctx = _resolve_author_mutation_context(worktree_path)
ctx = _resolve_namespace_mutation_context(worktree_path)
workspace = ctx["workspace_path"]
canonical_root = ctx["canonical_repo_root"]
process_root = ctx["process_project_root"]
real_workspace = os.path.realpath(workspace)
role = ctx.get("workspace_role_kind") or _effective_workspace_role()
if real_workspace != process_root:
if not _preflight_in_test_mode():
@@ -480,11 +554,15 @@ def verify_preflight_purity(remote: str | None = None, worktree_path: str | None
dirty_files = sorted(_parse_porcelain_entries(_get_workspace_porcelain(workspace)))
if dirty_files:
details = _preflight_workspace_details(workspace, dirty_files)
raise RuntimeError(
"Pre-flight order violation: Active task workspace has tracked "
"file edits before mutation (fail closed). "
f"{_format_preflight_workspace_details(details)}"
nwb.format_namespace_workspace_binding_error(
role_kind=role,
workspace_path=workspace,
binding_source=ctx.get("workspace_binding_source")
or "unknown binding source",
dirty_files=dirty_files,
ignored_bindings=ctx.get("ignored_bindings"),
)
)
else:
if _preflight_whoami_violation:
@@ -500,20 +578,58 @@ def verify_preflight_purity(remote: str | None = None, worktree_path: str | None
f"{_format_preflight_files(_preflight_capability_violation_files)}"
)
if _preflight_resolved_role == "reviewer":
if role in {"reviewer", "merger"}:
current = _get_workspace_porcelain()
baseline = _preflight_capability_baseline_porcelain or ""
reviewer_delta = _new_tracked_changes_since(baseline, current)
_preflight_reviewer_violation_files = reviewer_delta
if reviewer_delta:
raise RuntimeError(
"Reviewer role violation: Reviewer profile is forbidden from modifying "
f"{role.title()} role violation: profile is forbidden from modifying "
"tracked workspace files (fail closed). Offending files: "
f"{_format_preflight_files(reviewer_delta)}"
)
_enforce_branches_only_author_mutation(worktree_path)
def _verify_role_mutation_workspace(
remote: str | None = None,
*,
worktree_path: str | None = None,
worktree: str | None = None,
) -> str:
"""Bind reviewer/merger mutations to the active namespace workspace (#510)."""
role = _effective_workspace_role()
git_state = issue_lock_worktree.read_worktree_git_state(
_resolve_preflight_workspace_path(worktree_path)
)
assessment = nwb.assess_namespace_mutation_workspace(
role_kind=role,
worktree_path=worktree_path,
worktree=worktree,
process_project_root=PROJECT_ROOT,
session_lease_worktree=(
_reviewer_session_worktree() if role in {"reviewer", "merger"} else None
),
profile_name=get_profile().get("profile_name"),
current_branch=git_state.get("current_branch"),
)
if assessment["block"]:
raise RuntimeError(
nwb.format_namespace_workspace_binding_error(
role_kind=role,
workspace_path=assessment["mutation_workspace"],
binding_source=assessment.get("workspace_binding_source")
or "unknown binding source",
reasons=assessment.get("reasons"),
ignored_bindings=assessment.get("ignored_bindings"),
)
)
resolved = assessment["mutation_workspace"]
verify_preflight_purity(remote, worktree_path=resolved)
return resolved
from mcp.server.fastmcp import FastMCP # noqa: E402
from gitea_auth import ( # noqa: E402
@@ -2649,9 +2765,10 @@ def _evaluate_pr_review_submission(
*,
live: bool,
final_review_decision_ready: bool = False,
worktree_path: str | None = None,
) -> dict:
"""Shared gate chain for live submit and dry-run review tools."""
verify_preflight_purity(remote)
_verify_role_mutation_workspace(remote, worktree_path=worktree_path)
action = (action or "").strip().lower()
result = {
"requested_action": action,
@@ -3051,6 +3168,7 @@ def gitea_dry_run_pr_review(
host: str | None = None,
org: str | None = None,
repo: str | None = None,
worktree_path: str | None = None,
) -> dict:
"""Validate review submission mechanics without a live PR mutation."""
return _evaluate_pr_review_submission(
@@ -3063,6 +3181,7 @@ def gitea_dry_run_pr_review(
org=org,
repo=repo,
live=False,
worktree_path=worktree_path,
)
@@ -3078,6 +3197,7 @@ def gitea_submit_pr_review(
org: str | None = None,
repo: str | None = None,
final_review_decision_ready: bool = False,
worktree_path: str | None = None,
) -> dict:
"""Gated PR review mutation: comment findings, request changes, or approve.
@@ -3096,6 +3216,7 @@ def gitea_submit_pr_review(
repo=repo,
live=True,
final_review_decision_ready=final_review_decision_ready,
worktree_path=worktree_path,
)
@@ -3456,6 +3577,7 @@ def gitea_merge_pr(
host: str | None = None,
org: str | None = None,
repo: str | None = None,
worktree_path: str | None = None,
) -> dict:
"""Gated merge of a Gitea pull request (#16).
@@ -3502,6 +3624,10 @@ def gitea_merge_pr(
host: Override the Gitea host.
org: Override the owner/organization.
repo: Override the repository name.
worktree_path: Merger workspace path under ``branches/`` or clean
control checkout; defaults to ``GITEA_MERGER_WORKTREE``,
``GITEA_ACTIVE_WORKTREE``, or the MCP server process root. Ignores
foreign ``GITEA_AUTHOR_WORKTREE`` bindings (#510).
Returns:
dict describing the attempt: performed, authenticated user, profile
@@ -3509,7 +3635,7 @@ def gitea_merge_pr(
reasons/gates passed or blocked, and merge result / merge commit if
available. Never secrets.
"""
verify_preflight_purity(remote)
_verify_role_mutation_workspace(remote, worktree_path=worktree_path)
do = (do or "").strip().lower()
result = {
"performed": False,
@@ -4992,7 +5118,7 @@ def gitea_acquire_reviewer_pr_lease(
"permission_report": _permission_block_report("gitea.pr.comment"),
}
verify_preflight_purity(remote)
_verify_role_mutation_workspace(remote, worktree=worktree)
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
profile = get_profile()