Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
caaec9ac7a |
@@ -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*.
|
||||||
|
|
||||||
|
|||||||
+35
-131
@@ -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)
|
||||||
|
|
||||||
@@ -14585,56 +14602,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 +14618,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 +14651,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 +18098,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 +18114,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 +18127,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 +20584,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 +20616,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"]:
|
||||||
"pid": pid,
|
running_profiles[profile] = {
|
||||||
"start_time": start_time,
|
"pid": pid,
|
||||||
"is_stale": is_stale,
|
"start_time": start_time,
|
||||||
"is_client_managed": is_client_managed,
|
"is_stale": is_stale
|
||||||
}
|
}
|
||||||
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 +20660,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,
|
||||||
}
|
}
|
||||||
|
|||||||
+60
-7
@@ -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,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -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"):
|
||||||
|
|||||||
@@ -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,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()
|
|
||||||
@@ -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