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
|
||||||
|
|||||||
@@ -153,19 +153,7 @@ not a tool argument: a session must never be able to authorize itself.
|
|||||||
## Related
|
## Related
|
||||||
|
|
||||||
- #630 — manual daemon killing as contaminated recovery (this contrast, enforced).
|
- #630 — manual daemon killing as contaminated recovery (this contrast, enforced).
|
||||||
- #657 — restart-path inventory and daemon classification.
|
|
||||||
- #686 — manual server launch detection & fail-closed provenance gate.
|
|
||||||
- #531 / #544 — stale-runtime detection (`ps`-based); sibling failure mode.
|
- #531 / #544 — stale-runtime detection (`ps`-based); sibling failure mode.
|
||||||
- #558 / `docs/mcp-daemon-import-guard.md` — why shell imports are not a repair.
|
- #558 / `docs/mcp-daemon-import-guard.md` — why shell imports are not a repair.
|
||||||
- `docs/mcp-client-registration.md` — per-server registration contract.
|
- `docs/mcp-client-registration.md` — per-server registration contract.
|
||||||
- `docs/mcp-namespace-health.md` — probe sources and mutation enforcement.
|
- `docs/mcp-namespace-health.md` — probe sources and mutation enforcement.
|
||||||
|
|
||||||
## Sanctioned reconnect vs forbidden manual launch (#686)
|
|
||||||
|
|
||||||
In addition to manual process killing (#630), manually launching a duplicate role server from an ad hoc shell (`python3 mcp_server.py`) is forbidden and fail-closed:
|
|
||||||
|
|
||||||
- **Why manual launches are unsupported:** A terminal-launched `mcp_server.py` holds its own stdio transport; it can never bind to the IDE client's stdio pipes. It cannot restore a dropped IDE namespace, and a manual duplicate process masks stale client-managed runtimes for that profile, defeating stale-runtime gates.
|
|
||||||
- **Sanctioned path:** Supported recovery is IDE/client-managed reconnect only (`/mcp reconnect`, IDE restart, or sanctioned reconnect exposure).
|
|
||||||
- **Fail-closed enforcement (#686):** Mutating tools on a server lacking client-managed launch provenance (`GITEA_CLIENT_MANAGED=1`) refuse execution fail-closed with typed blocker `unsupported_manual_launch` and an exact next action. Unsupported `GITEA_*` env overrides (e.g. `GITEA_DUMMY`) are surfaced in diagnostics rather than silently ignored.
|
|
||||||
- **Inventory & staleness:** Staleness diagnostics ignore non-client-managed duplicates when evaluating runtime freshness and inventory duplicate processes per profile (#657, #686).
|
|
||||||
|
|
||||||
|
|||||||
+1
-50
@@ -1169,57 +1169,10 @@ def server_command():
|
|||||||
return python, [os.path.join(root, "mcp_server.py")]
|
return python, [os.path.join(root, "mcp_server.py")]
|
||||||
|
|
||||||
|
|
||||||
RECOGNIZED_GITEA_ENV_KEYS = frozenset({
|
|
||||||
"GITEA_MCP_CONFIG",
|
|
||||||
"GITEA_MCP_PROFILE",
|
|
||||||
"GITEA_PROFILE_NAME",
|
|
||||||
"GITEA_SERVICE",
|
|
||||||
"GITEA_EXECUTION_ROLE",
|
|
||||||
"GITEA_CLIENT_MANAGED",
|
|
||||||
"GITEA_MCP_CLIENT_MANAGED",
|
|
||||||
"GITEA_SERVER_PROVENANCE",
|
|
||||||
"GITEA_AUTHOR_WORKTREE",
|
|
||||||
"GITEA_ACTIVE_WORKTREE",
|
|
||||||
"GITEA_DISABLE_KEYCHAIN",
|
|
||||||
"GITEA_CONTROL_PLANE_DB",
|
|
||||||
"GITEA_DB_PATH",
|
|
||||||
"GITEA_LOG_LEVEL",
|
|
||||||
"GITEA_DEBUG",
|
|
||||||
"GITEA_HMAC_SECRET",
|
|
||||||
"GITEA_IRRECOVERABLE_HMAC_SECRET",
|
|
||||||
"GITEA_FORCE_MCP_RUNTIME_CHECK",
|
|
||||||
"GITEA_FORCE_CLIENT_MANAGED",
|
|
||||||
})
|
|
||||||
|
|
||||||
RECOGNIZED_GITEA_ENV_PREFIXES = (
|
|
||||||
"GITEA_TOKEN_",
|
|
||||||
"GITEA_PASS_",
|
|
||||||
"GITEA_USER_",
|
|
||||||
"GITEA_URL_",
|
|
||||||
"GITEA_HOST_",
|
|
||||||
"GITEA_REMOTE_",
|
|
||||||
"GITEA_HTTP_HEADER_",
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
def get_unconsumed_gitea_env_overrides(env=None) -> dict[str, str]:
|
|
||||||
"""Find unsupported GITEA_* env vars present in *env* (defaults to os.environ)."""
|
|
||||||
target = os.environ if env is None else env
|
|
||||||
unconsumed = {}
|
|
||||||
for key, value in target.items():
|
|
||||||
if key.startswith("GITEA_"):
|
|
||||||
if key in RECOGNIZED_GITEA_ENV_KEYS:
|
|
||||||
continue
|
|
||||||
if any(key.startswith(p) for p in RECOGNIZED_GITEA_ENV_PREFIXES):
|
|
||||||
continue
|
|
||||||
unconsumed[key] = str(value)
|
|
||||||
return unconsumed
|
|
||||||
|
|
||||||
|
|
||||||
def launcher_entry(profile_name, config_path=None):
|
def launcher_entry(profile_name, config_path=None):
|
||||||
"""Return a thin MCP launcher entry for *profile_name*.
|
"""Return a thin MCP launcher entry for *profile_name*.
|
||||||
|
|
||||||
Contains command/args and the GITEA_MCP_* / GITEA_CLIENT_MANAGED env vars — never a token
|
Contains only command/args and the two GITEA_MCP_* env vars — never a token
|
||||||
or password. Suitable for Claude / Gemini / Codex ``mcpServers`` blocks.
|
or password. Suitable for Claude / Gemini / Codex ``mcpServers`` blocks.
|
||||||
"""
|
"""
|
||||||
command, args = server_command()
|
command, args = server_command()
|
||||||
@@ -1230,13 +1183,11 @@ def launcher_entry(profile_name, config_path=None):
|
|||||||
"env": {
|
"env": {
|
||||||
"GITEA_MCP_CONFIG": config_path or DEFAULT_CONFIG_PATH,
|
"GITEA_MCP_CONFIG": config_path or DEFAULT_CONFIG_PATH,
|
||||||
"GITEA_MCP_PROFILE": profile_name,
|
"GITEA_MCP_PROFILE": profile_name,
|
||||||
"GITEA_CLIENT_MANAGED": "1",
|
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
def keychain_set(item_id, token, account=None, runner=subprocess.run):
|
def keychain_set(item_id, token, account=None, runner=subprocess.run):
|
||||||
"""Store *token* in the macOS keychain under service *item_id*.
|
"""Store *token* in the macOS keychain under service *item_id*.
|
||||||
|
|
||||||
|
|||||||
+4
-117
@@ -14585,56 +14585,6 @@ def _session_context_mutation_block(
|
|||||||
return blocked
|
return blocked
|
||||||
|
|
||||||
|
|
||||||
def _is_client_managed_process() -> bool:
|
|
||||||
"""Check whether the current MCP server process has client-managed launch provenance (#686)."""
|
|
||||||
val = (
|
|
||||||
os.environ.get("GITEA_CLIENT_MANAGED")
|
|
||||||
or os.environ.get("GITEA_MCP_CLIENT_MANAGED")
|
|
||||||
or os.environ.get("GITEA_SERVER_PROVENANCE")
|
|
||||||
or os.environ.get("GITEA_FORCE_CLIENT_MANAGED")
|
|
||||||
or ""
|
|
||||||
).strip().lower()
|
|
||||||
|
|
||||||
if val in ("0", "false", "no", "manual", "manual_launch"):
|
|
||||||
return False
|
|
||||||
|
|
||||||
if val in ("1", "true", "yes", "client_managed"):
|
|
||||||
return True
|
|
||||||
|
|
||||||
# A terminal launch has an active TTY on stdin
|
|
||||||
try:
|
|
||||||
if sys.stdin and sys.stdin.isatty():
|
|
||||||
return False
|
|
||||||
except Exception:
|
|
||||||
pass
|
|
||||||
|
|
||||||
# Standard client launch or test runner with stdio pipe and profile env
|
|
||||||
if "GITEA_MCP_CONFIG" in os.environ or "GITEA_MCP_PROFILE" in os.environ or "GITEA_PROFILE_NAME" in os.environ:
|
|
||||||
return True
|
|
||||||
|
|
||||||
return False
|
|
||||||
|
|
||||||
|
|
||||||
def _provenance_mutation_block(**extra_fields) -> dict | None:
|
|
||||||
"""Refuse mutating tool calls on processes lacking client-managed launch provenance (#686)."""
|
|
||||||
if _is_client_managed_process():
|
|
||||||
return None
|
|
||||||
unconsumed = gitea_config.get_unconsumed_gitea_env_overrides()
|
|
||||||
blocked = {
|
|
||||||
"success": False,
|
|
||||||
"performed": False,
|
|
||||||
"blocker_kind": "unsupported_manual_launch",
|
|
||||||
"reasons": [
|
|
||||||
"mutation denied: server process was launched manually from a terminal without client-managed provenance (fail closed). Manually launched mcp_server.py processes cannot receive IDE stdio or serve workflow mutations."
|
|
||||||
],
|
|
||||||
"exact_next_action": "BLOCKED + RECONNECT: Reconnect the IDE/client-managed MCP server namespace instead of an ad hoc terminal launch. Hand-launched processes and mcp_config.json hand-edits are classified as workflow contamination.",
|
|
||||||
"provenance": "manual_launch",
|
|
||||||
"unconsumed_gitea_env": unconsumed,
|
|
||||||
}
|
|
||||||
blocked.update(extra_fields)
|
|
||||||
return blocked
|
|
||||||
|
|
||||||
|
|
||||||
def _profile_permission_block(required_operation: str, **extra_fields) -> dict | None:
|
def _profile_permission_block(required_operation: str, **extra_fields) -> dict | None:
|
||||||
"""Structured operation-gate denial for gated tools (#69, #142, #897).
|
"""Structured operation-gate denial for gated tools (#69, #142, #897).
|
||||||
|
|
||||||
@@ -14651,10 +14601,6 @@ def _profile_permission_block(required_operation: str, **extra_fields) -> dict |
|
|||||||
# #714: evaluate active profile only — never auto-switch.
|
# #714: evaluate active profile only — never auto-switch.
|
||||||
_ensure_matching_profile(required_operation, req_role, extra_fields.get("remote"))
|
_ensure_matching_profile(required_operation, req_role, extra_fields.get("remote"))
|
||||||
|
|
||||||
prov_block = _provenance_mutation_block(**extra_fields)
|
|
||||||
if prov_block is not None:
|
|
||||||
return prov_block
|
|
||||||
|
|
||||||
reasons = _profile_operation_gate(required_operation)
|
reasons = _profile_operation_gate(required_operation)
|
||||||
if reasons:
|
if reasons:
|
||||||
return _build_operation_gate_refusal(
|
return _build_operation_gate_refusal(
|
||||||
@@ -14688,10 +14634,6 @@ def _namespace_mutation_block(mutation_task: str, **extra_fields) -> dict | None
|
|||||||
# #714: evaluate active profile only — never auto-switch.
|
# #714: evaluate active profile only — never auto-switch.
|
||||||
_ensure_matching_profile(required_permission, required_role, extra_fields.get("remote"))
|
_ensure_matching_profile(required_permission, required_role, extra_fields.get("remote"))
|
||||||
|
|
||||||
prov_block = _provenance_mutation_block(**extra_fields)
|
|
||||||
if prov_block is not None:
|
|
||||||
return prov_block
|
|
||||||
|
|
||||||
try:
|
try:
|
||||||
profile = get_profile()
|
profile = get_profile()
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
@@ -18139,9 +18081,6 @@ def gitea_get_runtime_context(
|
|||||||
source="gitea_get_runtime_context",
|
source="gitea_get_runtime_context",
|
||||||
)
|
)
|
||||||
|
|
||||||
is_client_managed = _is_client_managed_process()
|
|
||||||
unconsumed_env = gitea_config.get_unconsumed_gitea_env_overrides()
|
|
||||||
|
|
||||||
result = {
|
result = {
|
||||||
"active_profile": profile["profile_name"],
|
"active_profile": profile["profile_name"],
|
||||||
"authenticated_username": username,
|
"authenticated_username": username,
|
||||||
@@ -18158,9 +18097,6 @@ def gitea_get_runtime_context(
|
|||||||
"review_merge_blocked_reasons": blocked_reasons,
|
"review_merge_blocked_reasons": blocked_reasons,
|
||||||
"suggested_fix": suggested_fix,
|
"suggested_fix": suggested_fix,
|
||||||
"safe_next_action": safe_next_action,
|
"safe_next_action": safe_next_action,
|
||||||
"server_provenance": "client_managed" if is_client_managed else "manual_launch",
|
|
||||||
"is_client_managed": is_client_managed,
|
|
||||||
"unconsumed_gitea_env": unconsumed_env,
|
|
||||||
"preflight_ready": preflight["preflight_ready"],
|
"preflight_ready": preflight["preflight_ready"],
|
||||||
"preflight_block_reasons": preflight["preflight_block_reasons"],
|
"preflight_block_reasons": preflight["preflight_block_reasons"],
|
||||||
"preflight_workspace": preflight.get("preflight_workspace"),
|
"preflight_workspace": preflight.get("preflight_workspace"),
|
||||||
@@ -18174,13 +18110,6 @@ def gitea_get_runtime_context(
|
|||||||
PROJECT_ROOT),
|
PROJECT_ROOT),
|
||||||
}
|
}
|
||||||
|
|
||||||
if not is_client_managed:
|
|
||||||
result["safe_next_action"] = (
|
|
||||||
"BLOCKED + RECONNECT: Serving process lacks client-managed launch provenance (manual launch). "
|
|
||||||
"Reconnect the IDE/client-managed MCP server namespace instead of an ad hoc terminal launch."
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
# #702: read-only visibility into the inherited GITEA_ACTIVE_WORKTREE
|
# #702: read-only visibility into the inherited GITEA_ACTIVE_WORKTREE
|
||||||
# binding; recovery itself runs during capability resolution.
|
# binding; recovery itself runs during capability resolution.
|
||||||
try:
|
try:
|
||||||
@@ -20638,9 +20567,7 @@ def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) ->
|
|||||||
self_pid = os.getpid()
|
self_pid = os.getpid()
|
||||||
self_stale = False
|
self_stale = False
|
||||||
|
|
||||||
all_profile_procs: dict[str, list[dict]] = {}
|
running_profiles = {}
|
||||||
unsupported_env_found = set()
|
|
||||||
|
|
||||||
for line in proc.stdout.splitlines()[1:]:
|
for line in proc.stdout.splitlines()[1:]:
|
||||||
line = line.strip()
|
line = line.strip()
|
||||||
if not line or "mcp_server.py" not in line:
|
if not line or "mcp_server.py" not in line:
|
||||||
@@ -20672,55 +20599,16 @@ def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) ->
|
|||||||
if match:
|
if match:
|
||||||
profile = match.group(1)
|
profile = match.group(1)
|
||||||
|
|
||||||
is_client_managed = bool(
|
|
||||||
re.search(r'\bGITEA_CLIENT_MANAGED=(1|true|yes|client_managed)\b', env_out, re.IGNORECASE)
|
|
||||||
or re.search(r'\bGITEA_MCP_CLIENT_MANAGED=(1|true|yes|client_managed)\b', env_out, re.IGNORECASE)
|
|
||||||
or re.search(r'\bGITEA_SERVER_PROVENANCE=client_managed\b', env_out, re.IGNORECASE)
|
|
||||||
)
|
|
||||||
|
|
||||||
for env_match in re.finditer(r'\b(GITEA_[A-Z0-9_]+)=([^\s]+)', env_out):
|
|
||||||
k, v = env_match.group(1), env_match.group(2)
|
|
||||||
if k not in gitea_config.RECOGNIZED_GITEA_ENV_KEYS and not any(k.startswith(p) for p in gitea_config.RECOGNIZED_GITEA_ENV_PREFIXES):
|
|
||||||
unsupported_env_found.add(f"{k}={v}")
|
|
||||||
|
|
||||||
is_stale = (start_time < code_mtime) or git_stale
|
is_stale = (start_time < code_mtime) or git_stale
|
||||||
if pid == self_pid and is_stale:
|
if pid == self_pid and is_stale:
|
||||||
self_stale = True
|
self_stale = True
|
||||||
|
|
||||||
proc_info = {
|
if profile not in running_profiles or start_time > running_profiles[profile]["start_time"]:
|
||||||
|
running_profiles[profile] = {
|
||||||
"pid": pid,
|
"pid": pid,
|
||||||
"start_time": start_time,
|
"start_time": start_time,
|
||||||
"is_stale": is_stale,
|
"is_stale": is_stale
|
||||||
"is_client_managed": is_client_managed,
|
|
||||||
}
|
}
|
||||||
if profile not in all_profile_procs:
|
|
||||||
all_profile_procs[profile] = []
|
|
||||||
all_profile_procs[profile].append(proc_info)
|
|
||||||
|
|
||||||
running_profiles = {}
|
|
||||||
for profile, procs in all_profile_procs.items():
|
|
||||||
if len(procs) > 1:
|
|
||||||
pids_str = ", ".join(str(p["pid"]) for p in procs)
|
|
||||||
reasons.append(
|
|
||||||
f"stale-runtime: Duplicate MCP server process(es) detected for profile '{profile}' (PIDs: {pids_str}). "
|
|
||||||
"Manual or duplicate launches defeat staleness detection and cannot receive client stdio."
|
|
||||||
)
|
|
||||||
client_procs = [p for p in procs if p["is_client_managed"]]
|
|
||||||
if client_procs:
|
|
||||||
client_procs.sort(key=lambda p: p["start_time"], reverse=True)
|
|
||||||
running_profiles[profile] = client_procs[0]
|
|
||||||
else:
|
|
||||||
pids_str = ", ".join(str(p["pid"]) for p in procs)
|
|
||||||
reasons.append(
|
|
||||||
f"stale-runtime: Manually launched MCP process(es) detected without client-managed provenance for profile '{profile}' (PIDs: {pids_str}). "
|
|
||||||
"Manual launches cannot serve client stdio and are ignored for runtime freshness."
|
|
||||||
)
|
|
||||||
|
|
||||||
if unsupported_env_found:
|
|
||||||
reasons.append(
|
|
||||||
f"unsupported-env: Unsupported GITEA_* environment variable override(s) detected: {', '.join(sorted(unsupported_env_found))}. "
|
|
||||||
"Unknown env overrides are unsupported."
|
|
||||||
)
|
|
||||||
|
|
||||||
if self_stale:
|
if self_stale:
|
||||||
# #685: report-only — no config utime, no thread, no os._exit.
|
# #685: report-only — no config utime, no thread, no os._exit.
|
||||||
@@ -20755,7 +20643,6 @@ def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) ->
|
|||||||
return reasons
|
return reasons
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
@mcp.tool()
|
@mcp.tool()
|
||||||
def gitea_resolve_task_capability(
|
def gitea_resolve_task_capability(
|
||||||
task: str,
|
task: str,
|
||||||
|
|||||||
@@ -225,16 +225,6 @@ def classify_namespace_probe(
|
|||||||
# on bad data without treating success as IDE proof).
|
# on bad data without treating success as IDE proof).
|
||||||
blocks = namespace_health_blocks_task("merge_pr", healthy)
|
blocks = namespace_health_blocks_task("merge_pr", healthy)
|
||||||
|
|
||||||
import gitea_config
|
|
||||||
raw_env = process.get("env") if isinstance(process, dict) else None
|
|
||||||
unconsumed_env = gitea_config.get_unconsumed_gitea_env_overrides(raw_env)
|
|
||||||
is_client_managed = bool(
|
|
||||||
env_summary.get("GITEA_CLIENT_MANAGED") in ("1", "true", "yes", "client_managed")
|
|
||||||
or env_summary.get("GITEA_MCP_CLIENT_MANAGED") in ("1", "true", "yes", "client_managed")
|
|
||||||
or env_summary.get("GITEA_SERVER_PROVENANCE") == "client_managed"
|
|
||||||
)
|
|
||||||
provenance = "client_managed" if is_client_managed else "manual_launch"
|
|
||||||
|
|
||||||
return {
|
return {
|
||||||
"success": healthy,
|
"success": healthy,
|
||||||
"healthy": healthy,
|
"healthy": healthy,
|
||||||
@@ -250,9 +240,6 @@ def classify_namespace_probe(
|
|||||||
"error_message": error_message or None,
|
"error_message": error_message or None,
|
||||||
"reasons": reasons,
|
"reasons": reasons,
|
||||||
"remediation": remediation,
|
"remediation": remediation,
|
||||||
"provenance": provenance,
|
|
||||||
"is_client_managed": is_client_managed,
|
|
||||||
"unconsumed_gitea_env": unconsumed_env,
|
|
||||||
"diagnostics": {
|
"diagnostics": {
|
||||||
"namespace": ns,
|
"namespace": ns,
|
||||||
"required_tool": tool,
|
"required_tool": tool,
|
||||||
@@ -261,9 +248,6 @@ def classify_namespace_probe(
|
|||||||
"env": env_summary,
|
"env": env_summary,
|
||||||
"config_path": config_path,
|
"config_path": config_path,
|
||||||
"probe_source": source,
|
"probe_source": source,
|
||||||
"provenance": provenance,
|
|
||||||
"is_client_managed": is_client_managed,
|
|
||||||
"unconsumed_gitea_env": unconsumed_env,
|
|
||||||
},
|
},
|
||||||
"blocks_merge_workflow": blocks,
|
"blocks_merge_workflow": blocks,
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,3 +0,0 @@
|
|||||||
[pytest]
|
|
||||||
testpaths = tests
|
|
||||||
norecursedirs = branches .git venv __pycache__ graphify-out
|
|
||||||
@@ -44,8 +44,6 @@ def _reset_mutation_authority(monkeypatch):
|
|||||||
]:
|
]:
|
||||||
monkeypatch.delenv(env_key, raising=False)
|
monkeypatch.delenv(env_key, raising=False)
|
||||||
|
|
||||||
monkeypatch.setenv("GITEA_CLIENT_MANAGED", "1")
|
|
||||||
|
|
||||||
# Isolate durable session-state files so tests never share host cache (#559).
|
# Isolate durable session-state files so tests never share host cache (#559).
|
||||||
import tempfile
|
import tempfile
|
||||||
|
|
||||||
|
|||||||
@@ -35,7 +35,7 @@ CONFIG = {
|
|||||||
],
|
],
|
||||||
"forbidden_operations": [],
|
"forbidden_operations": [],
|
||||||
"execution_profile": "full-author",
|
"execution_profile": "full-author",
|
||||||
"allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools", "Example-Org/Example-Repo"],
|
"allowed_repositories": ["Example-Org/Example-Repo"],
|
||||||
},
|
},
|
||||||
"reviewer-no-commit": {
|
"reviewer-no-commit": {
|
||||||
"enabled": True,
|
"enabled": True,
|
||||||
@@ -50,7 +50,7 @@ CONFIG = {
|
|||||||
"gitea.repo.commit", "gitea.pr.create", "gitea.branch.push"
|
"gitea.repo.commit", "gitea.pr.create", "gitea.branch.push"
|
||||||
],
|
],
|
||||||
"execution_profile": "reviewer-no-commit",
|
"execution_profile": "reviewer-no-commit",
|
||||||
"allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools", "Example-Org/Example-Repo"],
|
"allowed_repositories": ["Example-Org/Example-Repo"],
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
"rules": {"allow_runtime_switching": False},
|
"rules": {"allow_runtime_switching": False},
|
||||||
|
|||||||
@@ -175,7 +175,7 @@ class TestLauncherSnippets(unittest.TestCase):
|
|||||||
def test_only_safe_keys_no_secrets(self):
|
def test_only_safe_keys_no_secrets(self):
|
||||||
entry = gitea_config.launcher_entry("prgs", "/cfg/profiles.json")["gitea-tools"]
|
entry = gitea_config.launcher_entry("prgs", "/cfg/profiles.json")["gitea-tools"]
|
||||||
self.assertEqual(set(entry), {"command", "args", "env"})
|
self.assertEqual(set(entry), {"command", "args", "env"})
|
||||||
self.assertEqual(set(entry["env"]), {"GITEA_MCP_CONFIG", "GITEA_MCP_PROFILE", "GITEA_CLIENT_MANAGED"})
|
self.assertEqual(set(entry["env"]), {"GITEA_MCP_CONFIG", "GITEA_MCP_PROFILE"})
|
||||||
self.assertEqual(entry["env"]["GITEA_MCP_PROFILE"], "prgs")
|
self.assertEqual(entry["env"]["GITEA_MCP_PROFILE"], "prgs")
|
||||||
blob = json.dumps(entry).lower()
|
blob = json.dumps(entry).lower()
|
||||||
for word in ("token", "password", "secret"):
|
for word in ("token", "password", "secret"):
|
||||||
|
|||||||
@@ -1,139 +0,0 @@
|
|||||||
"""Tests for Issue #686: Detect and reject manually launched duplicate MCP role servers."""
|
|
||||||
import os
|
|
||||||
import unittest
|
|
||||||
from unittest.mock import patch, MagicMock
|
|
||||||
from datetime import datetime
|
|
||||||
|
|
||||||
import gitea_config
|
|
||||||
import gitea_mcp_server
|
|
||||||
import mcp_namespace_health
|
|
||||||
|
|
||||||
|
|
||||||
class TestIssue686ManualMcpProvenance(unittest.TestCase):
|
|
||||||
|
|
||||||
def test_client_managed_process_detection(self):
|
|
||||||
"""Test _is_client_managed_process correctly detects provenance markers."""
|
|
||||||
with patch.dict(os.environ, {"GITEA_CLIENT_MANAGED": "1"}, clear=True):
|
|
||||||
self.assertTrue(gitea_mcp_server._is_client_managed_process())
|
|
||||||
|
|
||||||
with patch.dict(os.environ, {"GITEA_MCP_CLIENT_MANAGED": "true"}, clear=True):
|
|
||||||
self.assertTrue(gitea_mcp_server._is_client_managed_process())
|
|
||||||
|
|
||||||
with patch.dict(os.environ, {"GITEA_SERVER_PROVENANCE": "client_managed"}, clear=True):
|
|
||||||
self.assertTrue(gitea_mcp_server._is_client_managed_process())
|
|
||||||
|
|
||||||
with patch.dict(os.environ, {"GITEA_CLIENT_MANAGED": "0"}, clear=True):
|
|
||||||
self.assertFalse(gitea_mcp_server._is_client_managed_process())
|
|
||||||
|
|
||||||
def test_unconsumed_gitea_env_overrides(self):
|
|
||||||
"""Test surfacing of unsupported GITEA_* env overrides (e.g. GITEA_DUMMY)."""
|
|
||||||
env = {
|
|
||||||
"GITEA_MCP_PROFILE": "prgs-author",
|
|
||||||
"GITEA_CLIENT_MANAGED": "1",
|
|
||||||
"GITEA_DUMMY": "2",
|
|
||||||
"GITEA_UNKNOWN_FLAG": "abc",
|
|
||||||
}
|
|
||||||
unconsumed = gitea_config.get_unconsumed_gitea_env_overrides(env)
|
|
||||||
self.assertIn("GITEA_DUMMY", unconsumed)
|
|
||||||
self.assertEqual(unconsumed["GITEA_DUMMY"], "2")
|
|
||||||
self.assertIn("GITEA_UNKNOWN_FLAG", unconsumed)
|
|
||||||
self.assertNotIn("GITEA_MCP_PROFILE", unconsumed)
|
|
||||||
self.assertNotIn("GITEA_CLIENT_MANAGED", unconsumed)
|
|
||||||
|
|
||||||
def test_manual_server_mutation_fail_closed(self):
|
|
||||||
"""AC 2: Mutating tools on a server without client-managed provenance fail closed with a typed blocker."""
|
|
||||||
with patch.dict(os.environ, {"GITEA_CLIENT_MANAGED": "0"}, clear=True):
|
|
||||||
block = gitea_mcp_server._provenance_mutation_block(task="create_issue")
|
|
||||||
self.assertIsNotNone(block)
|
|
||||||
self.assertFalse(block["success"])
|
|
||||||
self.assertFalse(block["performed"])
|
|
||||||
self.assertEqual(block["blocker_kind"], "unsupported_manual_launch")
|
|
||||||
self.assertEqual(block["provenance"], "manual_launch")
|
|
||||||
self.assertTrue(any("mutation denied: server process was launched manually" in r for r in block["reasons"]))
|
|
||||||
self.assertIn("BLOCKED + RECONNECT", block["exact_next_action"])
|
|
||||||
|
|
||||||
def test_client_managed_server_mutation_passes_provenance_gate(self):
|
|
||||||
"""AC 3: Clean client-managed baseline passes the provenance gate."""
|
|
||||||
with patch.dict(os.environ, {"GITEA_CLIENT_MANAGED": "1"}, clear=True):
|
|
||||||
block = gitea_mcp_server._provenance_mutation_block(task="create_issue")
|
|
||||||
self.assertIsNone(block)
|
|
||||||
|
|
||||||
@patch("subprocess.run")
|
|
||||||
@patch("os.path.getmtime")
|
|
||||||
@patch("os.path.exists")
|
|
||||||
@patch("os.getpid")
|
|
||||||
def test_manual_duplicate_does_not_mask_stale_runtime(
|
|
||||||
self, mock_getpid, mock_exists, mock_getmtime, mock_run
|
|
||||||
):
|
|
||||||
"""AC 1 & AC 3: Staleness detection ignores manual duplicates and reports stale supported runtimes."""
|
|
||||||
mock_getpid.return_value = 12345
|
|
||||||
mock_exists.return_value = True
|
|
||||||
|
|
||||||
code_time = datetime(2026, 7, 8, 14, 0, 0)
|
|
||||||
mock_getmtime.return_value = code_time.timestamp()
|
|
||||||
|
|
||||||
# PID 12345: stale client-managed process (started at 13:00)
|
|
||||||
# PID 99999: fresh manual duplicate process (started at 15:00, no GITEA_CLIENT_MANAGED)
|
|
||||||
ps_output = (
|
|
||||||
" PID LSTART COMMAND\n"
|
|
||||||
"12345 Wed Jul 8 13:00:00 2026 /path/to/python mcp_server.py\n"
|
|
||||||
"99999 Wed Jul 8 15:00:00 2026 /path/to/python mcp_server.py\n"
|
|
||||||
)
|
|
||||||
|
|
||||||
mock_run_ps = MagicMock()
|
|
||||||
mock_run_ps.stdout = ps_output
|
|
||||||
|
|
||||||
mock_env_12345 = MagicMock()
|
|
||||||
mock_env_12345.stdout = "GITEA_MCP_PROFILE=prgs-author GITEA_CLIENT_MANAGED=1"
|
|
||||||
|
|
||||||
mock_env_99999 = MagicMock()
|
|
||||||
mock_env_99999.stdout = "GITEA_MCP_PROFILE=prgs-author GITEA_DUMMY=2"
|
|
||||||
|
|
||||||
def side_effect(args, **kwargs):
|
|
||||||
if args[0] == "ps" and "eww" in args:
|
|
||||||
pid = args[2]
|
|
||||||
if pid == "12345":
|
|
||||||
return mock_env_12345
|
|
||||||
elif pid == "99999":
|
|
||||||
return mock_env_99999
|
|
||||||
elif args[0] == "ps":
|
|
||||||
return mock_run_ps
|
|
||||||
raise ValueError(f"Unexpected args: {args}")
|
|
||||||
|
|
||||||
mock_run.side_effect = side_effect
|
|
||||||
|
|
||||||
reasons = gitea_mcp_server._check_mcp_runtimes_diagnostics("create_issue", ["prgs-author"])
|
|
||||||
|
|
||||||
# Manual duplicate process must be flagged
|
|
||||||
self.assertTrue(any("Duplicate MCP server process(es) detected" in r for r in reasons))
|
|
||||||
# Unsupported env override (GITEA_DUMMY=2) must be flagged
|
|
||||||
self.assertTrue(any("unsupported-env: Unsupported GITEA_* environment variable override(s) detected: GITEA_DUMMY=2" in r for r in reasons))
|
|
||||||
# Stale runtime must NOT be masked by fresh manual process 99999!
|
|
||||||
self.assertTrue(any("All matching profiles for task 'create_issue' (['prgs-author']) are running but stale" in r for r in reasons))
|
|
||||||
|
|
||||||
def test_namespace_health_classification_includes_provenance(self):
|
|
||||||
"""AC 1 & 4: mcp_namespace_health diagnostics include provenance and unconsumed_gitea_env."""
|
|
||||||
process = {
|
|
||||||
"pid": 5555,
|
|
||||||
"profile": "prgs-author",
|
|
||||||
"env": {
|
|
||||||
"GITEA_MCP_PROFILE": "prgs-author",
|
|
||||||
"GITEA_DUMMY": "99",
|
|
||||||
},
|
|
||||||
}
|
|
||||||
res = mcp_namespace_health.classify_namespace_probe(
|
|
||||||
"gitea-author",
|
|
||||||
configured=True,
|
|
||||||
registered_tools=["gitea_whoami"],
|
|
||||||
probe_result={"success": True},
|
|
||||||
process=process,
|
|
||||||
probe_source="client_namespace",
|
|
||||||
)
|
|
||||||
self.assertEqual(res["provenance"], "manual_launch")
|
|
||||||
self.assertFalse(res["is_client_managed"])
|
|
||||||
self.assertEqual(res["unconsumed_gitea_env"], {"GITEA_DUMMY": "99"})
|
|
||||||
self.assertEqual(res["diagnostics"]["provenance"], "manual_launch")
|
|
||||||
|
|
||||||
|
|
||||||
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()
|
||||||
@@ -35,10 +35,10 @@ class TestMcpStaleRuntime(unittest.TestCase):
|
|||||||
|
|
||||||
# Mock env output for ps eww
|
# Mock env output for ps eww
|
||||||
mock_run_env12345 = MagicMock()
|
mock_run_env12345 = MagicMock()
|
||||||
mock_run_env12345.stdout = "GITEA_MCP_PROFILE=prgs-reconciler GITEA_CLIENT_MANAGED=1"
|
mock_run_env12345.stdout = "GITEA_MCP_PROFILE=prgs-reconciler"
|
||||||
|
|
||||||
mock_run_env54321 = MagicMock()
|
mock_run_env54321 = MagicMock()
|
||||||
mock_run_env54321.stdout = "GITEA_MCP_PROFILE=prgs-author GITEA_CLIENT_MANAGED=1"
|
mock_run_env54321.stdout = "GITEA_MCP_PROFILE=prgs-author"
|
||||||
|
|
||||||
def side_effect(args, **kwargs):
|
def side_effect(args, **kwargs):
|
||||||
if args[0] == "ps" and "eww" in args:
|
if args[0] == "ps" and "eww" in args:
|
||||||
@@ -91,7 +91,7 @@ class TestMcpStaleRuntime(unittest.TestCase):
|
|||||||
mock_run_ps.stdout = ps_output
|
mock_run_ps.stdout = ps_output
|
||||||
|
|
||||||
mock_run_env = MagicMock()
|
mock_run_env = MagicMock()
|
||||||
mock_run_env.stdout = "GITEA_MCP_PROFILE=prgs-author GITEA_CLIENT_MANAGED=1"
|
mock_run_env.stdout = "GITEA_MCP_PROFILE=prgs-author"
|
||||||
|
|
||||||
mock_run_git = MagicMock()
|
mock_run_git = MagicMock()
|
||||||
mock_run_git.stdout = "FAKE2" # different SHA
|
mock_run_git.stdout = "FAKE2" # different SHA
|
||||||
|
|||||||
@@ -243,10 +243,9 @@ class TestRuntimeClarity(unittest.TestCase):
|
|||||||
self.assertIn("switching is disabled", res["message"].lower())
|
self.assertIn("switching is disabled", res["message"].lower())
|
||||||
self.assertIsNone(gitea_config._active_profile_override)
|
self.assertIsNone(gitea_config._active_profile_override)
|
||||||
|
|
||||||
@patch("mcp_server._trusted_session_repository", return_value={"repository": "Example-Org/Example-Repo", "org": "Example-Org", "repo": "Example-Repo", "reasons": []})
|
|
||||||
@patch("mcp_server.api_request")
|
@patch("mcp_server.api_request")
|
||||||
@patch("mcp_server.get_auth_header")
|
@patch("mcp_server.get_auth_header")
|
||||||
def test_activate_profile_succeeds_when_enabled(self, mock_auth, mock_api, mock_trusted):
|
def test_activate_profile_succeeds_when_enabled(self, mock_auth, mock_api):
|
||||||
self._write_config(CONFIG_SWITCHING_ENABLED)
|
self._write_config(CONFIG_SWITCHING_ENABLED)
|
||||||
|
|
||||||
# Setup mock responses for whoami checks
|
# Setup mock responses for whoami checks
|
||||||
|
|||||||
Reference in New Issue
Block a user