Compare commits

..
6 changed files with 234 additions and 408 deletions
+46 -157
View File
@@ -386,68 +386,6 @@ def run_compensating_recovery(
return recovery_info return recovery_info
def _normalize_sha(value: str | None) -> str | None:
"""Normalize a Git object id for comparison, or ``None`` when unknown."""
normalized = (value or "").strip().lower()
return normalized or None
def _author_bootstrap_assessment(
*,
not_applicable: bool,
allowed: bool,
block: bool,
reasons: list[str],
workspace: str,
root: str,
branch: str | None,
dirty: list[str],
under_branches: bool,
bootstrap_path: str | None = None,
local_head_sha: str | None = None,
remote_master_sha: str | None = None,
exact_next_action: str | None = None,
) -> dict[str, Any]:
"""Structured author-bootstrap assessment consumable by bootstrap_permits (#892).
Field shape mirrors :func:`create_issue_bootstrap._result` so the shared
``bootstrap_permits_control_checkout`` predicate can prove control-checkout
eligibility for ``gitea_bootstrap_author_issue_worktree`` the same way it
does for ``create_issue``. Allowed control assessments must use empty
``reasons`` — narrative belongs in other fields, not the refusal list.
"""
local_tip = _normalize_sha(local_head_sha)
remote_tip = _normalize_sha(remote_master_sha)
base_tips_verified = bool(local_tip and remote_tip and local_tip == remote_tip)
return {
"not_applicable": not_applicable,
"allowed": allowed,
"block": block,
"proven": bool(allowed and not block and not not_applicable),
"reasons": list(reasons),
"workspace_path": workspace,
"canonical_repo_root": root,
"current_branch": branch,
"dirty_files": list(dirty),
"under_branches": under_branches,
"exact_next_action": exact_next_action,
"bootstrap_path": bootstrap_path,
"task_scope": "author_issue_bootstrap",
"local_head_sha": local_tip,
"remote_master_sha": remote_tip,
"base_tips_verified": base_tips_verified,
}
EXACT_NEXT_ACTION_AUTHOR_BOOTSTRAP = (
"Restore the canonical control checkout to a clean accepted base branch "
"(master/main/dev) that matches live master, with no tracked local edits. "
"Re-resolve bootstrap_author_issue_worktree, then re-run "
"gitea_bootstrap_author_issue_worktree from that clean control checkout. "
"Do not use shell git worktree add as the primary path once bootstrap is healthy."
)
def assess_author_issue_bootstrap( def assess_author_issue_bootstrap(
*, *,
workspace_path: str, workspace_path: str,
@@ -459,13 +397,7 @@ def assess_author_issue_bootstrap(
remote_master_sha_error: str | None = None, remote_master_sha_error: str | None = None,
task: str | None = None, task: str | None = None,
) -> dict[str, Any]: ) -> dict[str, Any]:
"""Assess whether author issue worktree bootstrap may proceed from control or worktree root. """Assess whether author issue worktree bootstrap may proceed from control or worktree root."""
#892: control-checkout successes emit the full field set required by
``create_issue_bootstrap.bootstrap_permits_control_checkout`` (empty reasons,
task_scope, base tip proof, binding paths) so the #274/#604 guards can
waive control-checkout for this one sanctioned bootstrap task.
"""
root = os.path.realpath(canonical_repo_root or "") root = os.path.realpath(canonical_repo_root or "")
workspace = os.path.realpath(workspace_path or root or ".") workspace = os.path.realpath(workspace_path or root or ".")
branch = (current_branch or "").strip() branch = (current_branch or "").strip()
@@ -475,50 +407,34 @@ def assess_author_issue_bootstrap(
if root if root
else False else False
) )
local_tip = _normalize_sha(head_sha)
remote_tip = _normalize_sha(remote_master_sha)
if not is_author_issue_bootstrap_task(task): if not is_author_issue_bootstrap_task(task):
return _author_bootstrap_assessment( return {
not_applicable=True, "not_applicable": True,
allowed=False, "allowed": False,
block=False, "block": False,
reasons=["task is not author_issue_bootstrap"], "proven": False,
workspace=workspace, "reasons": ["task is not author_issue_bootstrap"],
root=root, }
branch=branch or None,
dirty=dirty,
under_branches=under_branches,
)
# Already under branches/: ordinary #274 path applies; not a control waiver.
if under_branches: if under_branches:
return _author_bootstrap_assessment( return {
not_applicable=True, "not_applicable": False,
allowed=False, "allowed": True,
block=False, "block": False,
reasons=["workspace is under branches/; ordinary #274 path applies"], "proven": True,
workspace=workspace, "bootstrap_path": "existing_branches_worktree",
root=root, "reasons": [
branch=branch or None, "workspace is already a registered worktree under branches/"
dirty=dirty, ],
under_branches=True, }
bootstrap_path="existing_branches_worktree",
local_head_sha=local_tip,
remote_master_sha=remote_tip,
)
reasons: list[str] = [] reasons: list[str] = []
if not root or workspace != root: if workspace != root:
reasons.append( reasons.append(
"bootstrap requires workspace to be canonical control checkout or branches/ worktree" "bootstrap requires workspace to be canonical control checkout or branches/ worktree"
) )
if not branch: if branch not in author_mutation_worktree.BASE_BRANCHES:
reasons.append(
"control checkout is detached HEAD; expected an accepted base branch "
f"({', '.join(sorted(author_mutation_worktree.BASE_BRANCHES))})"
)
elif branch not in author_mutation_worktree.BASE_BRANCHES:
reasons.append( reasons.append(
f"control checkout branch '{branch}' is not an accepted base branch " f"control checkout branch '{branch}' is not an accepted base branch "
f"({', '.join(sorted(author_mutation_worktree.BASE_BRANCHES))})" f"({', '.join(sorted(author_mutation_worktree.BASE_BRANCHES))})"
@@ -528,64 +444,37 @@ def assess_author_issue_bootstrap(
f"control checkout has tracked local edits: {', '.join(dirty[:5])}" f"control checkout has tracked local edits: {', '.join(dirty[:5])}"
) )
# Fail closed on missing tip proof (same bar as create_issue bootstrap #757). if remote_master_sha_error:
if not local_tip:
reasons.append( reasons.append(
"control checkout HEAD SHA is unknown; base equivalence to live " f"could not verify live master tip: {remote_master_sha_error}"
"master cannot be proven (fail closed)"
)
resolver_error = (remote_master_sha_error or "").strip() or None
if resolver_error:
reasons.append(
f"live master tip could not be resolved ({resolver_error}); "
"base equivalence cannot be proven (fail closed)"
)
elif not remote_tip:
reasons.append(
"live master tip is unknown; base equivalence cannot be proven "
"(fail closed)"
)
elif local_tip and remote_tip and local_tip != remote_tip:
reasons.append(
f"control checkout HEAD ({local_tip[:12]}) != live master tip "
f"({remote_tip[:12]})"
) )
elif remote_master_sha and head_sha:
h = head_sha.strip().lower()
rm = remote_master_sha.strip().lower()
if h != rm:
reasons.append(
f"control checkout HEAD ({h[:12]}) != live master tip ({rm[:12]})"
)
if reasons: if reasons:
return _author_bootstrap_assessment( return {
not_applicable=False, "not_applicable": False,
allowed=False, "allowed": False,
block=True, "block": True,
reasons=reasons, "proven": False,
workspace=workspace, "reasons": reasons,
root=root, }
branch=branch or None,
dirty=dirty,
under_branches=False,
local_head_sha=local_tip,
remote_master_sha=remote_tip,
exact_next_action=EXACT_NEXT_ACTION_AUTHOR_BOOTSTRAP,
)
# Allowed: empty reasons so bootstrap_permits_control_checkout can pass. return {
return _author_bootstrap_assessment( "not_applicable": False,
not_applicable=False, "allowed": True,
allowed=True, "block": False,
block=False, "proven": True,
reasons=[], "bootstrap_path": "clean_canonical_control_checkout",
workspace=workspace, "reasons": [
root=root, "control checkout is clean on accepted base branch matching live master"
branch=branch or None, ],
dirty=dirty, }
under_branches=False,
bootstrap_path="clean_canonical_control_checkout",
local_head_sha=local_tip,
remote_master_sha=remote_tip,
exact_next_action=(
"Call gitea_bootstrap_author_issue_worktree with the allocated "
"issue/lease pins; it will create the branches/ worktree and lock."
),
)
import fcntl import fcntl
+6 -18
View File
@@ -241,14 +241,9 @@ def bootstrap_permits_control_checkout(
caller's ordinary block in force. caller's ordinary block in force.
``assessment`` is server-derived only: it is produced by ``assessment`` is server-derived only: it is produced by
:func:`assess_create_issue_bootstrap` or :func:`assess_create_issue_bootstrap` from inspected repository state. It is
:func:`author_issue_bootstrap.assess_author_issue_bootstrap` from inspected never accepted from an MCP tool argument, so no caller can assert
repository state. It is never accepted from an MCP tool argument, so no eligibility it has not proven.
caller can assert eligibility it has not proven.
#892: author issue worktree bootstrap uses the same predicate with
``task_scope='author_issue_bootstrap'`` so a clean control checkout can
create the first ``branches/`` worktree without the lock↔worktree cycle.
""" """
if not isinstance(assessment, dict): if not isinstance(assessment, dict):
return False return False
@@ -269,16 +264,9 @@ def bootstrap_permits_control_checkout(
if assessment.get("reasons"): if assessment.get("reasons"):
return False return False
# Scope proof: create_issue (#749) or author issue bootstrap (#850/#892), # Scope proof: only the create_issue bootstrap, only via the clean
# only via the clean canonical control checkout path. # canonical control checkout path.
task_scope = assessment.get("task_scope") if assessment.get("task_scope") != "create_issue_only":
if is_create_issue_task(task):
if task_scope != "create_issue_only":
return False
elif author_issue_bootstrap.is_author_issue_bootstrap_task(task):
if task_scope != "author_issue_bootstrap":
return False
else:
return False return False
if assessment.get("bootstrap_path") != "clean_canonical_control_checkout": if assessment.get("bootstrap_path") != "clean_canonical_control_checkout":
return False return False
+28 -11
View File
@@ -3325,10 +3325,16 @@ def _effective_remote(remote: str) -> str:
return remote return remote
def _resolve(remote: str, host: str | None, org: str | None, repo: str | None): def _resolve(
remote: str,
host: str | None,
org: str | None,
repo: str | None,
for_mutation: bool = False,
):
"""Resolve remote + overrides to (host, org, repo). """Resolve remote + overrides to (host, org, repo).
#714 / #530: when the caller omits org and/or repo, prefer the #714 / #530 / #707: when the caller omits org and/or repo, prefer the
workspace-aligned git remote over ``REMOTES`` defaults (e.g. bare workspace-aligned git remote over ``REMOTES`` defaults (e.g. bare
``remote=prgs`` must not force ``Timesheet`` when the checkout is ``remote=prgs`` must not force ``Timesheet`` when the checkout is
``Gitea-Tools``). Explicit caller org/repo always win and are validated ``Gitea-Tools``). Explicit caller org/repo always win and are validated
@@ -3414,6 +3420,7 @@ def _resolve(remote: str, host: str | None, org: str | None, repo: str | None):
# Workspace-filled sides are intentional alignment for #530. # Workspace-filled sides are intentional alignment for #530.
org_explicit=org_explicit or filled_org, org_explicit=org_explicit or filled_org,
repo_explicit=repo_explicit or filled_repo, repo_explicit=repo_explicit or filled_repo,
for_mutation=for_mutation,
) )
return (resolved_host, resolved_org, resolved_repo) return (resolved_host, resolved_org, resolved_repo)
@@ -3475,8 +3482,9 @@ def _enforce_remote_repo_guard(
*, *,
org_explicit: bool, org_explicit: bool,
repo_explicit: bool, repo_explicit: bool,
for_mutation: bool = False,
) -> None: ) -> None:
"""Fail closed on a remote/repo mismatch vs. the local git remote (#530). """Fail closed on a remote/repo mismatch vs. the local git remote (#530/#707).
Best-effort: bypassed under pytest unless ``GITEA_FORCE_REMOTE_REPO_CHECK`` is Best-effort: bypassed under pytest unless ``GITEA_FORCE_REMOTE_REPO_CHECK`` is
set, so the unit suite (which calls tools with bare remotes against mocked APIs) set, so the unit suite (which calls tools with bare remotes against mocked APIs)
@@ -3488,6 +3496,12 @@ def _enforce_remote_repo_guard(
): ):
return return
local_remote_url = _local_git_remote_url(remote) local_remote_url = _local_git_remote_url(remote)
primary_org = None
primary_repo = None
ctx = session_ctx.get_session_context()
if ctx:
primary_org = ctx.get("org")
primary_repo = ctx.get("repository")
assessment = remote_repo_guard.assess_remote_repo_match( assessment = remote_repo_guard.assess_remote_repo_match(
remote=remote, remote=remote,
resolved_org=resolved_org, resolved_org=resolved_org,
@@ -3495,6 +3509,9 @@ def _enforce_remote_repo_guard(
local_remote_url=local_remote_url, local_remote_url=local_remote_url,
org_explicit=org_explicit, org_explicit=org_explicit,
repo_explicit=repo_explicit, repo_explicit=repo_explicit,
for_mutation=for_mutation,
primary_org=primary_org,
primary_repo=primary_repo,
) )
if assessment["block"]: if assessment["block"]:
raise RuntimeError(remote_repo_guard.format_remote_repo_guard_error(assessment)) raise RuntimeError(remote_repo_guard.format_remote_repo_guard_error(assessment))
@@ -4063,7 +4080,7 @@ def gitea_lock_issue(
resolved_worktree = issue_lock_worktree.resolve_author_worktree_path( resolved_worktree = issue_lock_worktree.resolve_author_worktree_path(
worktree_path, _canonical_local_git_root() worktree_path, _canonical_local_git_root()
) )
h, o, r = _resolve(remote, host, org, repo) h, o, r = _resolve(remote, host, org, repo, for_mutation=True)
existing_issue_lock = _load_existing_issue_lock( existing_issue_lock = _load_existing_issue_lock(
remote=remote, org=o, repo=r, issue_number=issue_number remote=remote, org=o, repo=r, issue_number=issue_number
) )
@@ -5239,7 +5256,7 @@ def gitea_create_pr(
org=org, org=org,
repo=repo, repo=repo,
) )
h, o, r = _resolve(remote, host, org, repo) h, o, r = _resolve(remote, host, org, repo, for_mutation=True)
# ── Issue Lock Validation (Issue #194 / #196 / #443) ── # ── Issue Lock Validation (Issue #194 / #196 / #443) ──
lock_data = _resolve_issue_lock_for_pr(remote=remote, org=o, repo=r, head=head) lock_data = _resolve_issue_lock_for_pr(remote=remote, org=o, repo=r, head=head)
@@ -9560,7 +9577,7 @@ def gitea_edit_pr(
required_permission="gitea.pr.close", required_permission="gitea.pr.close",
) )
h, o, r = _resolve(remote, host, org, repo) h, o, r = _resolve(remote, host, org, repo, for_mutation=True)
auth = _auth(h) auth = _auth(h)
url = f"{repo_api_url(h, o, r)}/pulls/{pr_number}" url = f"{repo_api_url(h, o, r)}/pulls/{pr_number}"
@@ -9844,7 +9861,7 @@ def gitea_commit_files(
verify_preflight_purity(remote=remote, worktree_path=worktree_path, task="commit_files", org=org, repo=repo) verify_preflight_purity(remote=remote, worktree_path=worktree_path, task="commit_files", org=org, repo=repo)
processed_files, source_proofs = _prepare_commit_payload_files(files) processed_files, source_proofs = _prepare_commit_payload_files(files)
h, o, r = _resolve(remote, host, org, repo) h, o, r = _resolve(remote, host, org, repo, for_mutation=True)
auth = _auth(h) auth = _auth(h)
url = f"{repo_api_url(h, o, r)}/contents" url = f"{repo_api_url(h, o, r)}/contents"
@@ -9980,7 +9997,7 @@ def gitea_publish_unpublished_issue_branch(
repo=repo, repo=repo,
) )
h, o, r = _resolve(remote, host, org, repo) h, o, r = _resolve(remote, host, org, repo, for_mutation=True)
git_remote = (git_remote_name or remote or "").strip() git_remote = (git_remote_name or remote or "").strip()
existing_lock = issue_lock_store.load_issue_lock( existing_lock = issue_lock_store.load_issue_lock(
@@ -10183,7 +10200,7 @@ def gitea_bootstrap_author_issue_worktree(
repo=repo, repo=repo,
) )
h, o, r = _resolve(remote, host, org, repo) h, o, r = _resolve(remote, host, org, repo, for_mutation=True)
canonical_root = _canonical_local_git_root() canonical_root = _canonical_local_git_root()
import author_issue_bootstrap import author_issue_bootstrap
@@ -12131,7 +12148,7 @@ def gitea_reconcile_merged_cleanups(
"blocker_kind": "invalid_pr_number", "blocker_kind": "invalid_pr_number",
} }
h, o, r = _resolve(remote, host, org, repo) h, o, r = _resolve(remote, host, org, repo, for_mutation=True)
auth = _auth(h) auth = _auth(h)
base = repo_api_url(h, o, r) base = repo_api_url(h, o, r)
@@ -12521,7 +12538,7 @@ def gitea_assess_already_landed_reconciliation(
"permission_report": _permission_block_report("gitea.read"), "permission_report": _permission_block_report("gitea.read"),
} }
h, o, r = _resolve(remote, host, org, repo) h, o, r = _resolve(remote, host, org, repo, for_mutation=True)
auth = _auth(h) auth = _auth(h)
pr = api_request("GET", f"{repo_api_url(h, o, r)}/pulls/{pr_number}", auth) pr = api_request("GET", f"{repo_api_url(h, o, r)}/pulls/{pr_number}", auth)
+60 -7
View File
@@ -54,20 +54,62 @@ def assess_remote_repo_match(
local_remote_url: str | None, local_remote_url: str | None,
org_explicit: bool, org_explicit: bool,
repo_explicit: bool, repo_explicit: bool,
for_mutation: bool = False,
primary_org: str | None = None,
primary_repo: str | None = None,
) -> dict: ) -> dict:
"""Fail closed when the resolved org/repo disagrees with the local git remote. """Fail closed when the resolved org/repo disagrees with the local git remote.
The guard is intentionally conservative: The guard enforces two protection levels:
* When the caller passed both ``org`` and ``repo`` explicitly, their intent is 1. Cross-Project Mutation Boundary (#707):
authoritative and the guard never blocks. When ``for_mutation`` is True (codebase mutation operation: branch, commit, PR,
* When the local git remote URL is unavailable (``None``/empty), corroboration merge, branch deletion), the target repository (``resolved_org/resolved_repo``)
is impossible, so the guard does not block (best-effort only). MUST match the primary authorized project context (``primary_org/primary_repo``
* Otherwise, the resolved ``org/repo`` slug must appear in the local remote URL or parsed from ``local_remote_url``). Any attempt to mutate a different project
(case-insensitive); if it does not, the guard blocks. fails closed, even if explicit org/repo were passed. Metadata operations (such
as creating issues or commenting on issues) across projects remain allowed.
2. Workspace Mismatch Guard (#530):
When ``for_mutation`` is False (or targets match), bare remotes must resolve
to an org/repo slug present in the local git remote URL. When explicit org/repo
are supplied for non-mutation operations, the caller's intent is authoritative.
""" """
reasons: list[str] = [] reasons: list[str] = []
eff_primary_org = primary_org
eff_primary_repo = primary_repo
if not eff_primary_org or not eff_primary_repo:
parsed_primary = parse_org_repo_from_remote_url(local_remote_url)
if parsed_primary:
eff_primary_org = eff_primary_org or parsed_primary[0]
eff_primary_repo = eff_primary_repo or parsed_primary[1]
# #707: Enforce cross-project codebase mutation boundary if for_mutation is True
if for_mutation and eff_primary_org and eff_primary_repo:
if (
resolved_org.lower() != eff_primary_org.lower()
or resolved_repo.lower() != eff_primary_repo.lower()
):
reasons.append(
f"Cross-project mutation guard (#707): Attempted codebase mutation targeting "
f"'{resolved_org}/{resolved_repo}' outside of primary authorized project "
f"context '{eff_primary_org}/{eff_primary_repo}'"
)
return {
"proven": False,
"block": True,
"cross_project_mutation_block": True,
"reasons": reasons,
"remote": remote,
"resolved_org": resolved_org,
"resolved_repo": resolved_repo,
"primary_org": eff_primary_org,
"primary_repo": eff_primary_repo,
"local_remote_url": local_remote_url,
"remediation": f"Cross-project codebase work is forbidden. Create an issue in the target repository ('{resolved_org}/{resolved_repo}') instead.",
}
if org_explicit and repo_explicit: if org_explicit and repo_explicit:
return _assessment(True, reasons, remote, resolved_org, resolved_repo, local_remote_url) return _assessment(True, reasons, remote, resolved_org, resolved_repo, local_remote_url)
@@ -88,6 +130,16 @@ def assess_remote_repo_match(
def format_remote_repo_guard_error(assessment: dict) -> str: def format_remote_repo_guard_error(assessment: dict) -> str:
"""Single RuntimeError message for the MCP resolver gate.""" """Single RuntimeError message for the MCP resolver gate."""
if assessment.get("cross_project_mutation_block"):
resolved = f"{assessment.get('resolved_org')}/{assessment.get('resolved_repo')}"
primary = f"{assessment.get('primary_org')}/{assessment.get('primary_repo')}"
return (
f"Cross-project mutation guard (#707): Attempted codebase mutation targeting '{resolved}' "
f"outside of primary authorized project context '{primary}'. "
f"Cross-project codebase work (creating branches, committing files, creating PRs) is forbidden; "
f"create an issue in the target repository ('{resolved}') instead."
)
reasons = "; ".join( reasons = "; ".join(
assessment.get("reasons") or ["remote/repo resolution mismatch"] assessment.get("reasons") or ["remote/repo resolution mismatch"]
) )
@@ -120,3 +172,4 @@ def _assessment(
"local_remote_url": local_remote_url, "local_remote_url": local_remote_url,
"remediation": REMEDIATION, "remediation": REMEDIATION,
} }
@@ -0,0 +1,94 @@
import os
import sys
import unittest
from unittest import mock
import remote_repo_guard
import gitea_mcp_server as server
from session_context_binding import _reset_session_context_for_testing, bind_session_context
LOCAL_GITEA_TOOLS_URL = "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
class TestCrossProjectMutationBoundary(unittest.TestCase):
def setUp(self):
os.environ["PYTEST_CURRENT_TEST"] = "test"
_reset_session_context_for_testing()
def tearDown(self):
_reset_session_context_for_testing()
os.environ.pop("PYTEST_CURRENT_TEST", None)
def test_cross_project_codebase_mutation_blocked(self):
"""Codebase mutations targeting another project must fail closed (#707)."""
assessment = remote_repo_guard.assess_remote_repo_match(
remote="prgs",
resolved_org="Other-Org",
resolved_repo="Other-Repo",
local_remote_url=LOCAL_GITEA_TOOLS_URL,
org_explicit=True,
repo_explicit=True,
for_mutation=True,
)
self.assertTrue(assessment["block"])
self.assertTrue(assessment.get("cross_project_mutation_block"))
self.assertEqual(assessment["primary_org"], "Scaled-Tech-Consulting")
self.assertEqual(assessment["primary_repo"], "Gitea-Tools")
err_msg = remote_repo_guard.format_remote_repo_guard_error(assessment)
self.assertIn("Cross-project mutation guard (#707)", err_msg)
self.assertIn("Attempted codebase mutation targeting 'Other-Org/Other-Repo'", err_msg)
self.assertIn("outside of primary authorized project context 'Scaled-Tech-Consulting/Gitea-Tools'", err_msg)
self.assertIn("create an issue in the target repository ('Other-Org/Other-Repo') instead", err_msg)
def test_same_project_codebase_mutation_allowed(self):
"""Codebase mutations targeting the primary project context are allowed."""
assessment = remote_repo_guard.assess_remote_repo_match(
remote="prgs",
resolved_org="Scaled-Tech-Consulting",
resolved_repo="Gitea-Tools",
local_remote_url=LOCAL_GITEA_TOOLS_URL,
org_explicit=True,
repo_explicit=True,
for_mutation=True,
)
self.assertFalse(assessment["block"])
self.assertFalse(assessment.get("cross_project_mutation_block", False))
def test_cross_project_metadata_operation_allowed(self):
"""Metadata operations (issue creation, comments) across project boundaries are allowed."""
assessment = remote_repo_guard.assess_remote_repo_match(
remote="prgs",
resolved_org="Other-Org",
resolved_repo="Other-Repo",
local_remote_url=LOCAL_GITEA_TOOLS_URL,
org_explicit=True,
repo_explicit=True,
for_mutation=False,
)
self.assertTrue(assessment["proven"])
self.assertFalse(assessment["block"])
@mock.patch.dict(os.environ, {"GITEA_FORCE_REMOTE_REPO_CHECK": "1"})
@mock.patch("gitea_mcp_server._local_git_remote_url", return_value=LOCAL_GITEA_TOOLS_URL)
def test_mcp_server_resolve_cross_project_mutation_fails_closed(self, mock_remote):
"""_resolve with for_mutation=True blocks cross-project targets."""
with self.assertRaises(RuntimeError) as ctx:
server._resolve("prgs", None, "Other-Org", "Other-Repo", for_mutation=True)
err = str(ctx.exception)
self.assertIn("Cross-project mutation guard (#707)", err)
self.assertIn("create an issue in the target repository", err)
@mock.patch.dict(os.environ, {"GITEA_FORCE_REMOTE_REPO_CHECK": "1"})
@mock.patch("gitea_mcp_server._local_git_remote_url", return_value=LOCAL_GITEA_TOOLS_URL)
def test_mcp_server_resolve_cross_project_metadata_succeeds(self, mock_remote):
"""_resolve with for_mutation=False allows explicit cross-project target."""
host, org, repo = server._resolve("prgs", None, "Other-Org", "Other-Repo", for_mutation=False)
self.assertEqual(org, "Other-Org")
self.assertEqual(repo, "Other-Repo")
if __name__ == "__main__":
unittest.main()
@@ -1,215 +0,0 @@
"""Regression: author worktree bootstrap from clean control checkout (#892).
#892 is the four-door deadlock where every documented recovery path is closed:
bootstrap refuses control, lock demands an existing worktree, worktree-start
demands a lock, and shell worktree add is outside the sanctioned MCP path.
Root cause: assess_author_issue_bootstrap returned allowed/proven for a clean
control checkout, but bootstrap_permits_control_checkout only accepted
create_issue assessments (task_scope=create_issue_only + empty reasons + full
base-tip field set). Author assessments never satisfied the shared predicate,
so the #274/#604 guards kept the ordinary control-checkout block.
"""
from __future__ import annotations
import os
import tempfile
import unittest
from unittest import mock
import author_issue_bootstrap as aib
import create_issue_bootstrap as cib
CONTROL = "/repo/Gitea-Tools"
MASTER = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
OTHER = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
def _assess(
*,
workspace=CONTROL,
root=CONTROL,
branch="master",
head=MASTER,
porcelain="",
remote=MASTER,
remote_error=None,
task="bootstrap_author_issue_worktree",
):
return aib.assess_author_issue_bootstrap(
workspace_path=workspace,
canonical_repo_root=root,
current_branch=branch,
head_sha=head,
porcelain_status=porcelain,
remote_master_sha=remote,
remote_master_sha_error=remote_error,
task=task,
)
class TestAuthorBootstrapAssessmentShape(unittest.TestCase):
def test_clean_control_emits_predicate_compatible_fields(self):
assessment = _assess()
self.assertTrue(assessment["allowed"])
self.assertTrue(assessment["proven"])
self.assertFalse(assessment["block"])
self.assertFalse(assessment["not_applicable"])
self.assertEqual(assessment["reasons"], [])
self.assertEqual(assessment["task_scope"], "author_issue_bootstrap")
self.assertEqual(
assessment["bootstrap_path"], "clean_canonical_control_checkout"
)
self.assertEqual(assessment["dirty_files"], [])
self.assertIs(assessment["under_branches"], False)
self.assertTrue(assessment["base_tips_verified"])
self.assertEqual(assessment["local_head_sha"], MASTER)
self.assertEqual(assessment["remote_master_sha"], MASTER)
self.assertEqual(assessment["workspace_path"], os.path.realpath(CONTROL))
self.assertEqual(
assessment["canonical_repo_root"], os.path.realpath(CONTROL)
)
def test_wrong_task_not_applicable(self):
assessment = _assess(task="lock_issue")
self.assertTrue(assessment["not_applicable"])
self.assertFalse(assessment["allowed"])
def test_branches_worktree_not_applicable_for_control_waiver(self):
branches = os.path.join(CONTROL, "branches", "fix-issue-1")
assessment = _assess(workspace=branches)
self.assertTrue(assessment["not_applicable"])
self.assertFalse(assessment["allowed"])
self.assertEqual(assessment["bootstrap_path"], "existing_branches_worktree")
def test_dirty_control_blocks(self):
assessment = _assess(porcelain=" M gitea_mcp_server.py\n")
self.assertTrue(assessment["block"])
self.assertFalse(assessment["allowed"])
self.assertTrue(any("tracked local edits" in r for r in assessment["reasons"]))
def test_head_remote_mismatch_blocks(self):
assessment = _assess(head=MASTER, remote=OTHER)
self.assertTrue(assessment["block"])
self.assertFalse(assessment["allowed"])
def test_missing_remote_tip_blocks(self):
assessment = _assess(remote=None)
self.assertTrue(assessment["block"])
self.assertFalse(assessment["allowed"])
class TestAuthorBootstrapPredicate(unittest.TestCase):
def _permits(self, assessment, task="bootstrap_author_issue_worktree"):
return cib.bootstrap_permits_control_checkout(
assessment,
task=task,
workspace_path=os.path.realpath(CONTROL),
canonical_repo_root=os.path.realpath(CONTROL),
)
def test_clean_author_bootstrap_permits(self):
self.assertTrue(self._permits(_assess()))
def test_tool_alias_permits(self):
assessment = _assess(task="gitea_bootstrap_author_issue_worktree")
self.assertTrue(
self._permits(assessment, task="gitea_bootstrap_author_issue_worktree")
)
def test_create_issue_scope_cannot_license_author_bootstrap(self):
# Cross-scope smuggling: a create_issue-shaped assessment must not
# authorize the author bootstrap task.
create_shaped = dict(_assess())
create_shaped["task_scope"] = "create_issue_only"
self.assertFalse(self._permits(create_shaped))
def test_author_scope_cannot_license_create_issue(self):
assessment = _assess()
self.assertFalse(
cib.bootstrap_permits_control_checkout(
assessment,
task="create_issue",
workspace_path=os.path.realpath(CONTROL),
canonical_repo_root=os.path.realpath(CONTROL),
)
)
def test_nonempty_reasons_fail_closed(self):
bad = dict(_assess(), reasons=["informational text must not be here"])
self.assertFalse(self._permits(bad))
def test_dirty_fails_closed(self):
self.assertFalse(self._permits(_assess(porcelain=" M x.py\n")))
def test_mismatch_fails_closed(self):
self.assertFalse(self._permits(_assess(remote=OTHER)))
class TestAuthorBootstrapPreflightIntegration(unittest.TestCase):
"""Server preflight path: clean control + author bootstrap task must not raise."""
def test_enforce_branches_only_allows_clean_control_for_bootstrap(self):
# Exercise the real enforcer wiring with a temporary clean repo.
import gitea_mcp_server as srv
with tempfile.TemporaryDirectory() as tmp:
repo = os.path.join(tmp, "repo")
os.makedirs(os.path.join(repo, "branches"))
# Minimal git repo on master at a known tip.
import subprocess
subprocess.check_call(["git", "init", "-b", "master", repo])
subprocess.check_call(
["git", "-C", repo, "commit", "--allow-empty", "-m", "init"]
)
head = subprocess.check_output(
["git", "-C", repo, "rev-parse", "HEAD"], text=True
).strip()
assessment = aib.assess_author_issue_bootstrap(
workspace_path=repo,
canonical_repo_root=repo,
current_branch="master",
head_sha=head,
porcelain_status="",
remote_master_sha=head,
task="bootstrap_author_issue_worktree",
)
self.assertTrue(
cib.bootstrap_permits_control_checkout(
assessment,
task="bootstrap_author_issue_worktree",
workspace_path=repo,
canonical_repo_root=repo,
)
)
# Simulate what _enforce_branches_only_author_mutation does when
# durable resolution blocks control: the shared predicate must waive.
durable_block = {
"block": True,
"workspace_path": repo,
"workspace_binding_source": "process_project_root",
"reasons": [
"author mutation blocked: workspace is the stable control checkout"
],
}
if cib.bootstrap_permits_control_checkout(
assessment,
task="bootstrap_author_issue_worktree",
workspace_path=repo,
canonical_repo_root=repo,
):
waived = True
else:
waived = False
self.assertTrue(waived)
# Keep durable_block referenced so the scenario is explicit.
self.assertTrue(durable_block["block"])
if __name__ == "__main__":
unittest.main()