Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
2066623986 |
+155
-44
@@ -386,6 +386,68 @@ 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,
|
||||||
@@ -397,7 +459,13 @@ 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()
|
||||||
@@ -407,34 +475,50 @@ 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 {
|
return _author_bootstrap_assessment(
|
||||||
"not_applicable": True,
|
not_applicable=True,
|
||||||
"allowed": False,
|
allowed=False,
|
||||||
"block": False,
|
block=False,
|
||||||
"proven": False,
|
reasons=["task is not author_issue_bootstrap"],
|
||||||
"reasons": ["task is not author_issue_bootstrap"],
|
workspace=workspace,
|
||||||
}
|
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 {
|
return _author_bootstrap_assessment(
|
||||||
"not_applicable": False,
|
not_applicable=True,
|
||||||
"allowed": True,
|
allowed=False,
|
||||||
"block": False,
|
block=False,
|
||||||
"proven": True,
|
reasons=["workspace is under branches/; ordinary #274 path applies"],
|
||||||
"bootstrap_path": "existing_branches_worktree",
|
workspace=workspace,
|
||||||
"reasons": [
|
root=root,
|
||||||
"workspace is already a registered worktree under branches/"
|
branch=branch or None,
|
||||||
],
|
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 workspace != root:
|
if not root or 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 branch not in author_mutation_worktree.BASE_BRANCHES:
|
if not branch:
|
||||||
|
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))})"
|
||||||
@@ -444,37 +528,64 @@ 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])}"
|
||||||
)
|
)
|
||||||
|
|
||||||
if remote_master_sha_error:
|
# Fail closed on missing tip proof (same bar as create_issue bootstrap #757).
|
||||||
|
if not local_tip:
|
||||||
reasons.append(
|
reasons.append(
|
||||||
f"could not verify live master tip: {remote_master_sha_error}"
|
"control checkout HEAD SHA is unknown; base equivalence to live "
|
||||||
|
"master cannot be proven (fail closed)"
|
||||||
)
|
)
|
||||||
elif remote_master_sha and head_sha:
|
resolver_error = (remote_master_sha_error or "").strip() or None
|
||||||
h = head_sha.strip().lower()
|
if resolver_error:
|
||||||
rm = remote_master_sha.strip().lower()
|
|
||||||
if h != rm:
|
|
||||||
reasons.append(
|
reasons.append(
|
||||||
f"control checkout HEAD ({h[:12]}) != live master tip ({rm[:12]})"
|
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]})"
|
||||||
)
|
)
|
||||||
|
|
||||||
if reasons:
|
if reasons:
|
||||||
return {
|
return _author_bootstrap_assessment(
|
||||||
"not_applicable": False,
|
not_applicable=False,
|
||||||
"allowed": False,
|
allowed=False,
|
||||||
"block": True,
|
block=True,
|
||||||
"proven": False,
|
reasons=reasons,
|
||||||
"reasons": reasons,
|
workspace=workspace,
|
||||||
}
|
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,
|
||||||
|
)
|
||||||
|
|
||||||
return {
|
# Allowed: empty reasons so bootstrap_permits_control_checkout can pass.
|
||||||
"not_applicable": False,
|
return _author_bootstrap_assessment(
|
||||||
"allowed": True,
|
not_applicable=False,
|
||||||
"block": False,
|
allowed=True,
|
||||||
"proven": True,
|
block=False,
|
||||||
"bootstrap_path": "clean_canonical_control_checkout",
|
reasons=[],
|
||||||
"reasons": [
|
workspace=workspace,
|
||||||
"control checkout is clean on accepted base branch matching live master"
|
root=root,
|
||||||
],
|
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
|
||||||
|
|||||||
@@ -241,9 +241,14 @@ 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` from inspected repository state. It is
|
:func:`assess_create_issue_bootstrap` or
|
||||||
never accepted from an MCP tool argument, so no caller can assert
|
:func:`author_issue_bootstrap.assess_author_issue_bootstrap` from inspected
|
||||||
eligibility it has not proven.
|
repository state. It is never accepted from an MCP tool argument, so no
|
||||||
|
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
|
||||||
@@ -264,9 +269,16 @@ def bootstrap_permits_control_checkout(
|
|||||||
if assessment.get("reasons"):
|
if assessment.get("reasons"):
|
||||||
return False
|
return False
|
||||||
|
|
||||||
# Scope proof: only the create_issue bootstrap, only via the clean
|
# Scope proof: create_issue (#749) or author issue bootstrap (#850/#892),
|
||||||
# canonical control checkout path.
|
# only via the clean canonical control checkout path.
|
||||||
if assessment.get("task_scope") != "create_issue_only":
|
task_scope = assessment.get("task_scope")
|
||||||
|
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
|
||||||
|
|||||||
+11
-28
@@ -3325,16 +3325,10 @@ def _effective_remote(remote: str) -> str:
|
|||||||
return remote
|
return remote
|
||||||
|
|
||||||
|
|
||||||
def _resolve(
|
def _resolve(remote: str, host: str | None, org: str | None, repo: str | None):
|
||||||
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 / #707: when the caller omits org and/or repo, prefer the
|
#714 / #530: 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
|
||||||
@@ -3420,7 +3414,6 @@ def _resolve(
|
|||||||
# 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)
|
||||||
|
|
||||||
@@ -3482,9 +3475,8 @@ 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/#707).
|
"""Fail closed on a remote/repo mismatch vs. the local git remote (#530).
|
||||||
|
|
||||||
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)
|
||||||
@@ -3496,12 +3488,6 @@ 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,
|
||||||
@@ -3509,9 +3495,6 @@ 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))
|
||||||
@@ -4080,7 +4063,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, for_mutation=True)
|
h, o, r = _resolve(remote, host, org, repo)
|
||||||
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
|
||||||
)
|
)
|
||||||
@@ -5256,7 +5239,7 @@ def gitea_create_pr(
|
|||||||
org=org,
|
org=org,
|
||||||
repo=repo,
|
repo=repo,
|
||||||
)
|
)
|
||||||
h, o, r = _resolve(remote, host, org, repo, for_mutation=True)
|
h, o, r = _resolve(remote, host, org, repo)
|
||||||
|
|
||||||
# ── 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)
|
||||||
@@ -9577,7 +9560,7 @@ def gitea_edit_pr(
|
|||||||
required_permission="gitea.pr.close",
|
required_permission="gitea.pr.close",
|
||||||
)
|
)
|
||||||
|
|
||||||
h, o, r = _resolve(remote, host, org, repo, for_mutation=True)
|
h, o, r = _resolve(remote, host, org, repo)
|
||||||
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}"
|
||||||
|
|
||||||
@@ -9861,7 +9844,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, for_mutation=True)
|
h, o, r = _resolve(remote, host, org, repo)
|
||||||
auth = _auth(h)
|
auth = _auth(h)
|
||||||
url = f"{repo_api_url(h, o, r)}/contents"
|
url = f"{repo_api_url(h, o, r)}/contents"
|
||||||
|
|
||||||
@@ -9997,7 +9980,7 @@ def gitea_publish_unpublished_issue_branch(
|
|||||||
repo=repo,
|
repo=repo,
|
||||||
)
|
)
|
||||||
|
|
||||||
h, o, r = _resolve(remote, host, org, repo, for_mutation=True)
|
h, o, r = _resolve(remote, host, org, repo)
|
||||||
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(
|
||||||
@@ -10200,7 +10183,7 @@ def gitea_bootstrap_author_issue_worktree(
|
|||||||
repo=repo,
|
repo=repo,
|
||||||
)
|
)
|
||||||
|
|
||||||
h, o, r = _resolve(remote, host, org, repo, for_mutation=True)
|
h, o, r = _resolve(remote, host, org, repo)
|
||||||
canonical_root = _canonical_local_git_root()
|
canonical_root = _canonical_local_git_root()
|
||||||
|
|
||||||
import author_issue_bootstrap
|
import author_issue_bootstrap
|
||||||
@@ -12148,7 +12131,7 @@ def gitea_reconcile_merged_cleanups(
|
|||||||
"blocker_kind": "invalid_pr_number",
|
"blocker_kind": "invalid_pr_number",
|
||||||
}
|
}
|
||||||
|
|
||||||
h, o, r = _resolve(remote, host, org, repo, for_mutation=True)
|
h, o, r = _resolve(remote, host, org, repo)
|
||||||
auth = _auth(h)
|
auth = _auth(h)
|
||||||
base = repo_api_url(h, o, r)
|
base = repo_api_url(h, o, r)
|
||||||
|
|
||||||
@@ -12538,7 +12521,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, for_mutation=True)
|
h, o, r = _resolve(remote, host, org, repo)
|
||||||
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)
|
||||||
|
|
||||||
|
|||||||
+7
-60
@@ -54,62 +54,20 @@ 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 enforces two protection levels:
|
The guard is intentionally conservative:
|
||||||
|
|
||||||
1. Cross-Project Mutation Boundary (#707):
|
* When the caller passed both ``org`` and ``repo`` explicitly, their intent is
|
||||||
When ``for_mutation`` is True (codebase mutation operation: branch, commit, PR,
|
authoritative and the guard never blocks.
|
||||||
merge, branch deletion), the target repository (``resolved_org/resolved_repo``)
|
* When the local git remote URL is unavailable (``None``/empty), corroboration
|
||||||
MUST match the primary authorized project context (``primary_org/primary_repo``
|
is impossible, so the guard does not block (best-effort only).
|
||||||
or parsed from ``local_remote_url``). Any attempt to mutate a different project
|
* Otherwise, the resolved ``org/repo`` slug must appear in the local remote URL
|
||||||
fails closed, even if explicit org/repo were passed. Metadata operations (such
|
(case-insensitive); if it does not, the guard blocks.
|
||||||
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)
|
||||||
|
|
||||||
@@ -130,16 +88,6 @@ 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"]
|
||||||
)
|
)
|
||||||
@@ -172,4 +120,3 @@ def _assessment(
|
|||||||
"local_remote_url": local_remote_url,
|
"local_remote_url": local_remote_url,
|
||||||
"remediation": REMEDIATION,
|
"remediation": REMEDIATION,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -1,94 +0,0 @@
|
|||||||
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()
|
|
||||||
@@ -0,0 +1,215 @@
|
|||||||
|
"""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()
|
||||||
Reference in New Issue
Block a user