sysadminandClaude Opus 4.8
b33844d74a
Share canonical workspace/repo-root resolution between gitea_get_runtime_context and verify_preflight_purity so valid branches/ worktrees are not rejected when the MCP process root differs from the stable control checkout. Fix relative git-common-dir resolution and add regression tests. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
232 lines
8.0 KiB
Python
232 lines
8.0 KiB
Python
"""Branches-only author mutation worktree guard (#274).
|
|
|
|
Author/coder mutations must run from a session-owned worktree under the
|
|
project's ``branches/`` directory, never from the stable control checkout.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import subprocess
|
|
|
|
BASE_BRANCHES = frozenset({"master", "main", "dev"})
|
|
ACTIVE_WORKTREE_ENV = "GITEA_ACTIVE_WORKTREE"
|
|
AUTHOR_WORKTREE_ENV = "GITEA_AUTHOR_WORKTREE"
|
|
|
|
|
|
def _normalize_path(path: str) -> str:
|
|
return (path or "").replace("\\", "/").rstrip("/")
|
|
|
|
|
|
def is_path_under_branches(path: str, project_root: str | None = None) -> bool:
|
|
"""True when *path* resolves inside ``<project_root>/branches/``."""
|
|
normalized = _normalize_path(path)
|
|
if not normalized:
|
|
return False
|
|
if "/branches/" in f"{normalized}/":
|
|
return True
|
|
if normalized.endswith("/branches"):
|
|
return True
|
|
if project_root:
|
|
root = _normalize_path(os.path.realpath(project_root))
|
|
real = _normalize_path(os.path.realpath(path))
|
|
if real.startswith(f"{root}/"):
|
|
rel = real[len(root) + 1 :]
|
|
return rel == "branches" or rel.startswith("branches/")
|
|
return False
|
|
|
|
|
|
def resolve_mutation_workspace(
|
|
worktree_path: str | None,
|
|
project_root: str,
|
|
*,
|
|
active_worktree_env: str | None = None,
|
|
author_worktree_env: str | None = None,
|
|
) -> str:
|
|
"""Resolve the workspace path inspected before author mutations."""
|
|
for candidate in (worktree_path, active_worktree_env, author_worktree_env):
|
|
text = (candidate or "").strip()
|
|
if text:
|
|
return os.path.realpath(os.path.abspath(text))
|
|
return os.path.realpath(project_root)
|
|
|
|
|
|
def _realpath_git_common_dir(workspace_path: str, common_dir: str) -> str:
|
|
"""Resolve ``git rev-parse --git-common-dir`` relative to *workspace_path*."""
|
|
raw = (common_dir or "").strip()
|
|
if not raw:
|
|
return raw
|
|
if os.path.isabs(raw):
|
|
return os.path.realpath(raw)
|
|
return os.path.realpath(os.path.join(workspace_path, raw))
|
|
|
|
|
|
def resolve_canonical_repo_root(workspace_path: str, fallback_project_root: str) -> str:
|
|
"""Return the stable repository root for *workspace_path* via git metadata (#460)."""
|
|
path = (workspace_path or "").strip()
|
|
fallback = os.path.realpath(fallback_project_root)
|
|
if not path:
|
|
return fallback
|
|
try:
|
|
res = subprocess.run(
|
|
["git", "-C", path, "rev-parse", "--git-common-dir"],
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
)
|
|
common = _realpath_git_common_dir(path, res.stdout)
|
|
except Exception:
|
|
return fallback
|
|
if common.endswith(f"{os.sep}.git"):
|
|
return os.path.dirname(common)
|
|
if os.path.basename(common) == ".git":
|
|
return os.path.dirname(common)
|
|
return fallback
|
|
|
|
|
|
def resolve_author_mutation_context(
|
|
worktree_path: str | None,
|
|
process_project_root: str,
|
|
*,
|
|
active_worktree_env: str | None = None,
|
|
author_worktree_env: str | None = None,
|
|
) -> dict:
|
|
"""Shared workspace resolution for runtime_context and mutation guards (#460)."""
|
|
workspace = resolve_mutation_workspace(
|
|
worktree_path,
|
|
process_project_root,
|
|
active_worktree_env=active_worktree_env,
|
|
author_worktree_env=author_worktree_env,
|
|
)
|
|
process_root = os.path.realpath(process_project_root)
|
|
# Canonical repository identity comes from the MCP process checkout (#460),
|
|
# not from the declared task workspace being validated.
|
|
canonical_root = resolve_canonical_repo_root(process_root, process_root)
|
|
return {
|
|
"workspace_path": workspace,
|
|
"process_project_root": process_root,
|
|
"canonical_repo_root": canonical_root,
|
|
"roots_aligned": canonical_root == process_root,
|
|
}
|
|
|
|
|
|
def assess_workspace_repo_membership(
|
|
*,
|
|
workspace_path: str,
|
|
canonical_repo_root: str,
|
|
) -> dict:
|
|
"""Fail closed when *workspace_path* is not a git worktree of *canonical_repo_root*."""
|
|
workspace = os.path.realpath(workspace_path)
|
|
root = os.path.realpath(canonical_repo_root)
|
|
reasons: list[str] = []
|
|
|
|
if not os.path.exists(workspace):
|
|
reasons.append(f"worktree path '{workspace}' does not exist")
|
|
return _membership_assessment(False, reasons, workspace, root, None)
|
|
|
|
if not os.path.isdir(workspace):
|
|
reasons.append(f"worktree path '{workspace}' is not a directory")
|
|
return _membership_assessment(False, reasons, workspace, root, None)
|
|
|
|
try:
|
|
res = subprocess.run(
|
|
["git", "-C", workspace, "rev-parse", "--git-common-dir"],
|
|
capture_output=True,
|
|
text=True,
|
|
check=True,
|
|
)
|
|
common_dir = _realpath_git_common_dir(workspace, res.stdout)
|
|
except Exception:
|
|
reasons.append(f"worktree '{workspace}' is not a valid git repository")
|
|
return _membership_assessment(False, reasons, workspace, root, None)
|
|
|
|
expected_dir = os.path.realpath(os.path.join(root, ".git"))
|
|
if common_dir != expected_dir:
|
|
reasons.append(
|
|
f"worktree '{workspace}' does not belong to the target repository '{root}'"
|
|
)
|
|
return _membership_assessment(not reasons, reasons, workspace, root, common_dir)
|
|
|
|
|
|
def _membership_assessment(
|
|
proven: bool,
|
|
reasons: list[str],
|
|
workspace: str,
|
|
root: str,
|
|
common_dir: str | None,
|
|
) -> dict:
|
|
return {
|
|
"proven": proven,
|
|
"block": not proven,
|
|
"reasons": reasons,
|
|
"workspace_path": workspace,
|
|
"canonical_repo_root": root,
|
|
"git_common_dir": common_dir,
|
|
}
|
|
|
|
|
|
def format_workspace_repo_membership_error(assessment: dict) -> str:
|
|
workspace = assessment.get("workspace_path") or "(unknown)"
|
|
root = assessment.get("canonical_repo_root") or "(unknown)"
|
|
reasons = "; ".join(assessment.get("reasons") or ["unknown repository membership violation"])
|
|
return (
|
|
f"Branches-only mutation guard (#274): {reasons} (fail closed). "
|
|
f"canonical repository root: {root}; workspace: {workspace}."
|
|
)
|
|
|
|
|
|
def assess_author_mutation_worktree(
|
|
*,
|
|
workspace_path: str,
|
|
project_root: str,
|
|
current_branch: str | None = None,
|
|
base_branches: frozenset[str] | None = None,
|
|
) -> dict:
|
|
"""Fail closed when author mutations are not rooted under ``branches/``."""
|
|
bases = base_branches or BASE_BRANCHES
|
|
reasons: list[str] = []
|
|
root = os.path.realpath(project_root)
|
|
workspace = os.path.realpath(workspace_path)
|
|
branch = (current_branch or "").strip()
|
|
|
|
under_branches = is_path_under_branches(workspace, root)
|
|
if not under_branches:
|
|
if workspace == root:
|
|
reasons.append(
|
|
"author mutation blocked: workspace is the stable control checkout; "
|
|
"create or switch to a session-owned worktree under branches/"
|
|
)
|
|
else:
|
|
reasons.append(
|
|
f"author mutation blocked: workspace '{workspace}' is not under "
|
|
f"'{root}/branches/'; create a branches/<task> worktree first"
|
|
)
|
|
|
|
if not under_branches and workspace == root and branch and branch not in bases:
|
|
reasons.append(
|
|
f"control checkout drift: branch '{branch}' is not a stable base "
|
|
f"branch ({'/'.join(sorted(bases))})"
|
|
)
|
|
|
|
proven = not reasons
|
|
return {
|
|
"proven": proven,
|
|
"block": not proven,
|
|
"reasons": reasons,
|
|
"project_root": root,
|
|
"workspace_path": workspace,
|
|
"under_branches": is_path_under_branches(workspace, root),
|
|
"current_branch": branch or None,
|
|
}
|
|
|
|
|
|
def format_author_mutation_worktree_error(assessment: dict) -> str:
|
|
"""Single RuntimeError message for MCP preflight gates."""
|
|
workspace = assessment.get("workspace_path") or "(unknown)"
|
|
root = assessment.get("project_root") or "(unknown)"
|
|
reasons = "; ".join(assessment.get("reasons") or ["unknown branches-only violation"])
|
|
return (
|
|
f"Branches-only mutation guard (#274): {reasons}. "
|
|
f"project root: {root}; workspace: {workspace}. "
|
|
"Create a session-owned worktree under branches/ before mutating."
|
|
) |