Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
ef34d938bf |
@@ -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).
|
|
||||||
|
|
||||||
|
|||||||
@@ -86,26 +86,3 @@ When a namespace returns EOF, follow
|
|||||||
|
|
||||||
When blocked, repair the IDE namespace and re-record a healthy
|
When blocked, repair the IDE namespace and re-record a healthy
|
||||||
`client_namespace` assessment before retrying the mutation.
|
`client_namespace` assessment before retrying the mutation.
|
||||||
|
|
||||||
## Connected vs Attached Tool Surface (#708)
|
|
||||||
|
|
||||||
MCP servers can report **Connected** at the CLI / host inventory layer while the **active LLM session exposes none of their tool namespaces**.
|
|
||||||
|
|
||||||
### Core principle
|
|
||||||
|
|
||||||
* **Connected status at host layer ≠ attached tools in active session.**
|
|
||||||
* Required preflight proof is **live tool visibility + `gitea_whoami` call** through the target namespace, not host `Connected` status alone.
|
|
||||||
* When servers report Connected but namespaces are absent from attached tools, classify as `mcp_connected_namespaces_missing`.
|
|
||||||
|
|
||||||
### Forbidden unsafe fallbacks
|
|
||||||
|
|
||||||
When `mcp_connected_namespaces_missing` is detected, workflows must **fail closed** and must **never** encourage or perform:
|
|
||||||
|
|
||||||
* direct imports of MCP server Python modules
|
|
||||||
* CLI or raw Gitea API mutations as a substitute for native tools
|
|
||||||
* profile hopping to another MCP profile/namespace to bypass the empty session
|
|
||||||
* session-state overrides or hand-edited session/ledger files
|
|
||||||
* process kills (`pkill`), config mtime touches, or `.env` edits
|
|
||||||
|
|
||||||
Only sanctioned recovery: **client reconnect path**, followed by full preflight (`whoami` → capability resolve → task).
|
|
||||||
|
|
||||||
|
|||||||
+62
-4
@@ -22,8 +22,62 @@ import gitea_config
|
|||||||
|
|
||||||
PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))
|
PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))
|
||||||
|
|
||||||
# Load standard .env if present
|
# Reserved runtime-control / workspace-binding environment variables (#704).
|
||||||
load_dotenv(os.path.join(PROJECT_ROOT, ".env"))
|
# Repository .env files MUST NOT populate or override any of these keys.
|
||||||
|
RESERVED_WORKTREE_ENV_KEYS: frozenset[str] = frozenset({
|
||||||
|
"GITEA_ACTIVE_WORKTREE",
|
||||||
|
"GITEA_AUTHOR_WORKTREE",
|
||||||
|
"GITEA_REVIEWER_WORKTREE",
|
||||||
|
"GITEA_MERGER_WORKTREE",
|
||||||
|
"GITEA_RECONCILER_WORKTREE",
|
||||||
|
})
|
||||||
|
|
||||||
|
|
||||||
|
def is_reserved_worktree_env_key(key: str | None) -> bool:
|
||||||
|
"""Return True if *key* is a reserved runtime workspace binding variable (#704)."""
|
||||||
|
if not key:
|
||||||
|
return False
|
||||||
|
k = str(key).upper().strip()
|
||||||
|
return k in RESERVED_WORKTREE_ENV_KEYS or (k.startswith("GITEA_") and k.endswith("_WORKTREE"))
|
||||||
|
|
||||||
|
|
||||||
|
def load_env_file_sanitized(
|
||||||
|
env_path: str,
|
||||||
|
*,
|
||||||
|
target_env: dict | os._Environ | None = None,
|
||||||
|
) -> list[str]:
|
||||||
|
"""Load a .env file without populating or overriding reserved workspace keys (#704).
|
||||||
|
|
||||||
|
Pre-existing process environment values retain their precedence. Reserved
|
||||||
|
runtime-control keys found in repository files are ignored (without logging
|
||||||
|
their values). Returns a list of sanitized rejection reasons.
|
||||||
|
"""
|
||||||
|
if target_env is None:
|
||||||
|
target_env = os.environ
|
||||||
|
if not os.path.exists(env_path) or os.path.isdir(env_path):
|
||||||
|
return []
|
||||||
|
|
||||||
|
rejection_reasons: list[str] = []
|
||||||
|
try:
|
||||||
|
file_vals = dotenv_values(env_path)
|
||||||
|
for key, val in file_vals.items():
|
||||||
|
if not key or val is None:
|
||||||
|
continue
|
||||||
|
if is_reserved_worktree_env_key(key):
|
||||||
|
filename = os.path.basename(env_path)
|
||||||
|
rejection_reasons.append(
|
||||||
|
f"Ignored reserved runtime workspace key '{key}' from repository {filename}"
|
||||||
|
)
|
||||||
|
continue
|
||||||
|
if key not in target_env:
|
||||||
|
target_env[key] = val
|
||||||
|
except Exception:
|
||||||
|
pass
|
||||||
|
return rejection_reasons
|
||||||
|
|
||||||
|
|
||||||
|
# Load standard .env if present (sanitized to prevent repo workspace binding contamination #704)
|
||||||
|
load_env_file_sanitized(os.path.join(PROJECT_ROOT, ".env"))
|
||||||
|
|
||||||
# Dictionary to store configurations parsed dynamically from .env.* files
|
# Dictionary to store configurations parsed dynamically from .env.* files
|
||||||
DYNAMIC_CONFIGS = {}
|
DYNAMIC_CONFIGS = {}
|
||||||
@@ -37,9 +91,13 @@ for env_path in glob.glob(os.path.join(PROJECT_ROOT, ".env*")):
|
|||||||
continue
|
continue
|
||||||
try:
|
try:
|
||||||
config_vals = dotenv_values(env_path)
|
config_vals = dotenv_values(env_path)
|
||||||
site = config_vals.get("GITEA_SITE") or config_vals.get("GITEA_HOST")
|
# Filter out reserved workspace keys from dynamic configs (#704)
|
||||||
|
sanitized_config = {
|
||||||
|
k: v for k, v in config_vals.items() if not is_reserved_worktree_env_key(k)
|
||||||
|
}
|
||||||
|
site = sanitized_config.get("GITEA_SITE") or sanitized_config.get("GITEA_HOST")
|
||||||
if site:
|
if site:
|
||||||
DYNAMIC_CONFIGS[site.lower().strip()] = config_vals
|
DYNAMIC_CONFIGS[site.lower().strip()] = sanitized_config
|
||||||
except Exception:
|
except Exception:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
|||||||
+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,
|
||||||
|
|||||||
@@ -51,14 +51,6 @@ EOF_PATTERNS = (
|
|||||||
"eof",
|
"eof",
|
||||||
)
|
)
|
||||||
|
|
||||||
ERROR_CONNECTED_NAMESPACES_MISSING = "mcp_connected_namespaces_missing"
|
|
||||||
|
|
||||||
UNSAFE_FALLBACK_WARNING = (
|
|
||||||
"Workflow Safety Hard Stop (#708): Connected-but-namespaces-missing recovery must "
|
|
||||||
"NEVER use direct imports, Gitea API mutations, profile hopping, session-state "
|
|
||||||
"overrides, PID kills, or config mtime touches. Use client reconnect only."
|
|
||||||
)
|
|
||||||
|
|
||||||
SAFE_ENV_KEYS = (
|
SAFE_ENV_KEYS = (
|
||||||
"GITEA_MCP_PROFILE",
|
"GITEA_MCP_PROFILE",
|
||||||
"GITEA_PROFILE_NAME",
|
"GITEA_PROFILE_NAME",
|
||||||
@@ -68,83 +60,6 @@ SAFE_ENV_KEYS = (
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def assess_connected_namespace_attachment(
|
|
||||||
*,
|
|
||||||
connected_servers: list[str] | tuple[str, ...] | set[str] | None = None,
|
|
||||||
attached_session_namespaces: list[str] | tuple[str, ...] | set[str] | None = None,
|
|
||||||
required_namespaces: list[str] | tuple[str, ...] | set[str] | None = None,
|
|
||||||
) -> dict[str, Any]:
|
|
||||||
"""Assess whether host-connected MCP servers have attached tool namespaces in the active session (#708).
|
|
||||||
|
|
||||||
Addresses the Connected-but-namespaces-missing defect: CLI/host status may report Connected
|
|
||||||
while the active LLM session tool surface exposes 0 attached tool namespaces.
|
|
||||||
|
|
||||||
Returns structured telemetry and detection details.
|
|
||||||
"""
|
|
||||||
connected = [str(s).strip() for s in (connected_servers or []) if str(s).strip()]
|
|
||||||
attached = set(str(ns).strip() for ns in (attached_session_namespaces or []) if str(ns).strip())
|
|
||||||
req = [str(r).strip() for r in (required_namespaces or DEFAULT_NAMESPACES) if str(r).strip()]
|
|
||||||
|
|
||||||
proof: dict[str, dict[str, bool]] = {}
|
|
||||||
missing: list[str] = []
|
|
||||||
|
|
||||||
for s in connected:
|
|
||||||
is_attached = s in attached
|
|
||||||
proof[s] = {"connected": True, "attached": is_attached}
|
|
||||||
if not is_attached and s in req:
|
|
||||||
missing.append(s)
|
|
||||||
|
|
||||||
for r in req:
|
|
||||||
if r not in proof:
|
|
||||||
proof[r] = {"connected": r in connected, "attached": r in attached}
|
|
||||||
if r in connected and r not in attached and r not in missing:
|
|
||||||
missing.append(r)
|
|
||||||
|
|
||||||
attachment_healthy = len(missing) == 0 and len(connected) > 0
|
|
||||||
discovery_status = (
|
|
||||||
"namespaces_attached" if attachment_healthy
|
|
||||||
else ("connected_but_namespaces_missing" if len(connected) > 0 else "disconnected")
|
|
||||||
)
|
|
||||||
|
|
||||||
reasons: list[str] = []
|
|
||||||
remediation: list[str] = []
|
|
||||||
if missing:
|
|
||||||
reasons.append(
|
|
||||||
f"MCP server(s) {missing} report Connected at host/CLI layer but tool namespaces "
|
|
||||||
f"are missing from active session attached tools (Connected ≠ attached tools, #708)."
|
|
||||||
)
|
|
||||||
remediation.append(
|
|
||||||
"Reconnect the IDE/client MCP session to attach namespaces to the active session. "
|
|
||||||
"Do not use direct imports, CLI API mutations, profile hopping, or session file overrides."
|
|
||||||
)
|
|
||||||
elif not connected:
|
|
||||||
reasons.append("No MCP servers reported Connected.")
|
|
||||||
remediation.append("Start or reconnect Gitea MCP servers in client config.")
|
|
||||||
else:
|
|
||||||
reasons.append("All connected MCP server namespaces are attached to the active session.")
|
|
||||||
|
|
||||||
return {
|
|
||||||
"success": attachment_healthy,
|
|
||||||
"attachment_healthy": attachment_healthy,
|
|
||||||
"discovery_status": discovery_status,
|
|
||||||
"connected_servers": connected,
|
|
||||||
"attached_session_namespaces": list(attached),
|
|
||||||
"missing_namespaces": missing,
|
|
||||||
"proof_of_connected_vs_attached": proof,
|
|
||||||
"error_type": None if attachment_healthy else ERROR_CONNECTED_NAMESPACES_MISSING,
|
|
||||||
"reasons": reasons,
|
|
||||||
"remediation": remediation,
|
|
||||||
"exact_next_action": (
|
|
||||||
"Reconnect the IDE/client MCP session so tool namespaces attach to the active session. "
|
|
||||||
"Do not use direct imports, CLI API mutations, profile hopping, or session-state overrides."
|
|
||||||
if not attachment_healthy
|
|
||||||
else "None; session tool namespaces attached."
|
|
||||||
),
|
|
||||||
"unsafe_fallback_policy": UNSAFE_FALLBACK_WARNING,
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
def _as_list(value: Any) -> list[str] | None:
|
def _as_list(value: Any) -> list[str] | None:
|
||||||
if value is None:
|
if value is None:
|
||||||
return None
|
return None
|
||||||
@@ -310,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,
|
||||||
@@ -335,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,
|
||||||
@@ -346,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,
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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,166 @@
|
|||||||
|
"""Tests for Issue #704: Preventing repository .env files from injecting workspace bindings.
|
||||||
|
|
||||||
|
Acceptance Criteria (#704):
|
||||||
|
1. Repository .env loading cannot populate or override GITEA_ACTIVE_WORKTREE or any role-specific GITEA_*_WORKTREE runtime-binding variable.
|
||||||
|
2. Runtime workspace bindings are accepted only from sanctioned managed-launch/session mechanisms.
|
||||||
|
3. Pre-existing sanctioned process environment values retain their intended precedence.
|
||||||
|
4. Importing gitea_auth or related modules does not mutate workspace-binding state from repository files.
|
||||||
|
5. Reserved runtime-control keys found in .env are ignored or rejected with a sanitized actionable reason; their values are never logged.
|
||||||
|
6. The protection applies consistently to author, reviewer, merger, and reconciler namespaces.
|
||||||
|
7. Comprehensive test coverage for stale worktree, missing worktree, task-specific, role-specific, launcher binding, repeated imports, namespace isolation, precedence, and absence of secret leakage.
|
||||||
|
8. Dirty-state and workspace-preflight gates cannot be bypassed by an injected missing-path binding.
|
||||||
|
9. No environment, dotenv, offline-import, or caller-controlled path can forge native transport or mutation provenance.
|
||||||
|
10. Cross-linked with #702, PR #703, #510.
|
||||||
|
11. Required immediate follow-up to Issue #702 / PR #703.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import os
|
||||||
|
import sys
|
||||||
|
import tempfile
|
||||||
|
import importlib
|
||||||
|
import unittest
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||||
|
|
||||||
|
import gitea_auth
|
||||||
|
from gitea_auth import is_reserved_worktree_env_key, load_env_file_sanitized
|
||||||
|
|
||||||
|
|
||||||
|
class TestIssue704PreventEnvWorkspaceBindings(unittest.TestCase):
|
||||||
|
"""Test suite verifying .env workspace-binding injection prevention (#704)."""
|
||||||
|
|
||||||
|
def setUp(self):
|
||||||
|
self.tmpdir = tempfile.TemporaryDirectory()
|
||||||
|
self.addCleanup(self.tmpdir.cleanup)
|
||||||
|
self.env_dir = Path(self.tmpdir.name)
|
||||||
|
|
||||||
|
def test_is_reserved_worktree_env_key(self):
|
||||||
|
"""Verify key classification for all role namespaces (#704 AC6)."""
|
||||||
|
reserved_keys = [
|
||||||
|
"GITEA_ACTIVE_WORKTREE",
|
||||||
|
"GITEA_AUTHOR_WORKTREE",
|
||||||
|
"GITEA_REVIEWER_WORKTREE",
|
||||||
|
"GITEA_MERGER_WORKTREE",
|
||||||
|
"GITEA_RECONCILER_WORKTREE",
|
||||||
|
"gitea_active_worktree",
|
||||||
|
"GITEA_CUSTOM_ROLE_WORKTREE",
|
||||||
|
]
|
||||||
|
for key in reserved_keys:
|
||||||
|
self.assertTrue(
|
||||||
|
is_reserved_worktree_env_key(key),
|
||||||
|
f"Expected {key} to be recognized as a reserved worktree key",
|
||||||
|
)
|
||||||
|
|
||||||
|
unreserved_keys = [
|
||||||
|
"GITEA_USER",
|
||||||
|
"GITEA_PASS",
|
||||||
|
"GITEA_TOKEN",
|
||||||
|
"GITEA_HOST",
|
||||||
|
"PATH",
|
||||||
|
]
|
||||||
|
for key in unreserved_keys:
|
||||||
|
self.assertFalse(
|
||||||
|
is_reserved_worktree_env_key(key),
|
||||||
|
f"Expected {key} to NOT be recognized as a reserved worktree key",
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_load_env_file_sanitized_ignores_reserved_keys(self):
|
||||||
|
"""Verify .env loading ignores GITEA_ACTIVE_WORKTREE and role-specific keys (#704 AC1)."""
|
||||||
|
env_file = self.env_dir / ".env"
|
||||||
|
stale_path = "/tmp/stale-worktree-path-1234"
|
||||||
|
env_file.write_text(
|
||||||
|
f"GITEA_USER=testuser\n"
|
||||||
|
f"GITEA_ACTIVE_WORKTREE={stale_path}\n"
|
||||||
|
f"GITEA_AUTHOR_WORKTREE={stale_path}\n"
|
||||||
|
f"GITEA_REVIEWER_WORKTREE={stale_path}\n"
|
||||||
|
f"GITEA_MERGER_WORKTREE={stale_path}\n"
|
||||||
|
f"GITEA_RECONCILER_WORKTREE={stale_path}\n"
|
||||||
|
)
|
||||||
|
|
||||||
|
test_env = {}
|
||||||
|
reasons = load_env_file_sanitized(str(env_file), target_env=test_env)
|
||||||
|
|
||||||
|
# Unreserved key loaded
|
||||||
|
self.assertEqual(test_env.get("GITEA_USER"), "testuser")
|
||||||
|
|
||||||
|
# Reserved keys ignored
|
||||||
|
self.assertNotIn("GITEA_ACTIVE_WORKTREE", test_env)
|
||||||
|
self.assertNotIn("GITEA_AUTHOR_WORKTREE", test_env)
|
||||||
|
self.assertNotIn("GITEA_REVIEWER_WORKTREE", test_env)
|
||||||
|
self.assertNotIn("GITEA_MERGER_WORKTREE", test_env)
|
||||||
|
self.assertNotIn("GITEA_RECONCILER_WORKTREE", test_env)
|
||||||
|
|
||||||
|
# Rejection reasons populated without leaking the secret value (#704 AC5)
|
||||||
|
self.assertTrue(len(reasons) >= 5)
|
||||||
|
for r in reasons:
|
||||||
|
self.assertNotIn(stale_path, r, "Secret/path value must not leak into rejection reason")
|
||||||
|
|
||||||
|
def test_preexisting_sanctioned_launcher_env_retained(self):
|
||||||
|
"""Sanctioned launcher values in process env are retained (#704 AC2, AC3)."""
|
||||||
|
sanctioned_path = "/tmp/sanctioned-launcher-worktree"
|
||||||
|
test_env = {"GITEA_ACTIVE_WORKTREE": sanctioned_path}
|
||||||
|
|
||||||
|
env_file = self.env_dir / ".env"
|
||||||
|
env_file.write_text("GITEA_ACTIVE_WORKTREE=/tmp/injected-repo-worktree\n")
|
||||||
|
|
||||||
|
load_env_file_sanitized(str(env_file), target_env=test_env)
|
||||||
|
|
||||||
|
# Pre-existing value retained, not overwritten by .env
|
||||||
|
self.assertEqual(test_env.get("GITEA_ACTIVE_WORKTREE"), sanctioned_path)
|
||||||
|
|
||||||
|
def test_stale_or_missing_worktree_in_env_ignored(self):
|
||||||
|
"""Stale or non-existent worktree path in .env file is ignored (#704 AC7)."""
|
||||||
|
nonexistent_path = "/nonexistent/branches/stale-issue-999"
|
||||||
|
env_file = self.env_dir / ".env"
|
||||||
|
env_file.write_text(f"GITEA_ACTIVE_WORKTREE={nonexistent_path}\n")
|
||||||
|
|
||||||
|
test_env = {}
|
||||||
|
load_env_file_sanitized(str(env_file), target_env=test_env)
|
||||||
|
|
||||||
|
self.assertNotIn("GITEA_ACTIVE_WORKTREE", test_env)
|
||||||
|
|
||||||
|
def test_repeated_module_import_does_not_mutate_workspace_env(self):
|
||||||
|
"""Repeated imports of gitea_auth leave os.environ un-contaminated (#704 AC4, AC7)."""
|
||||||
|
# Ensure no active worktree env exists initially
|
||||||
|
original_val = os.environ.pop("GITEA_ACTIVE_WORKTREE", None)
|
||||||
|
try:
|
||||||
|
importlib.reload(gitea_auth)
|
||||||
|
self.assertNotIn("GITEA_ACTIVE_WORKTREE", os.environ)
|
||||||
|
|
||||||
|
importlib.reload(gitea_auth)
|
||||||
|
self.assertNotIn("GITEA_ACTIVE_WORKTREE", os.environ)
|
||||||
|
finally:
|
||||||
|
if original_val is not None:
|
||||||
|
os.environ["GITEA_ACTIVE_WORKTREE"] = original_val
|
||||||
|
|
||||||
|
def test_namespace_isolation_all_roles_protected(self):
|
||||||
|
"""Verify protection across author, reviewer, merger, reconciler (#704 AC6)."""
|
||||||
|
env_file = self.env_dir / ".env"
|
||||||
|
env_file.write_text(
|
||||||
|
"GITEA_AUTHOR_WORKTREE=/bad/author\n"
|
||||||
|
"GITEA_REVIEWER_WORKTREE=/bad/reviewer\n"
|
||||||
|
"GITEA_MERGER_WORKTREE=/bad/merger\n"
|
||||||
|
"GITEA_RECONCILER_WORKTREE=/bad/reconciler\n"
|
||||||
|
)
|
||||||
|
test_env = {}
|
||||||
|
load_env_file_sanitized(str(env_file), target_env=test_env)
|
||||||
|
|
||||||
|
self.assertEqual(test_env, {})
|
||||||
|
|
||||||
|
def test_absence_of_secret_leakage(self):
|
||||||
|
"""Rejection reasons contain key names but never secret path values (#704 AC5)."""
|
||||||
|
sensitive_path = "/Users/secret/path/private_repo"
|
||||||
|
env_file = self.env_dir / ".env"
|
||||||
|
env_file.write_text(f"GITEA_ACTIVE_WORKTREE={sensitive_path}\n")
|
||||||
|
|
||||||
|
test_env = {}
|
||||||
|
reasons = load_env_file_sanitized(str(env_file), target_env=test_env)
|
||||||
|
for reason in reasons:
|
||||||
|
self.assertNotIn(sensitive_path, reason)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
unittest.main()
|
||||||
@@ -1,69 +0,0 @@
|
|||||||
"""Unit regression tests for Issue #708: Connected-but-namespaces-missing detection and attachment safety."""
|
|
||||||
|
|
||||||
import mcp_namespace_health
|
|
||||||
|
|
||||||
|
|
||||||
def test_assess_connected_namespace_attachment_success():
|
|
||||||
connected = ["gitea-author", "gitea-reviewer", "gitea-merger"]
|
|
||||||
attached = ["gitea-author", "gitea-reviewer", "gitea-merger"]
|
|
||||||
res = mcp_namespace_health.assess_connected_namespace_attachment(
|
|
||||||
connected_servers=connected,
|
|
||||||
attached_session_namespaces=attached,
|
|
||||||
)
|
|
||||||
assert res["success"] is True
|
|
||||||
assert res["attachment_healthy"] is True
|
|
||||||
assert res["discovery_status"] == "namespaces_attached"
|
|
||||||
assert res["error_type"] is None
|
|
||||||
assert res["missing_namespaces"] == []
|
|
||||||
assert res["exact_next_action"] == "None; session tool namespaces attached."
|
|
||||||
|
|
||||||
|
|
||||||
def test_assess_connected_namespace_attachment_missing():
|
|
||||||
connected = ["gitea-author", "gitea-reviewer", "gitea-merger"]
|
|
||||||
attached = ["gitea-author"]
|
|
||||||
res = mcp_namespace_health.assess_connected_namespace_attachment(
|
|
||||||
connected_servers=connected,
|
|
||||||
attached_session_namespaces=attached,
|
|
||||||
)
|
|
||||||
assert res["success"] is False
|
|
||||||
assert res["attachment_healthy"] is False
|
|
||||||
assert res["discovery_status"] == "connected_but_namespaces_missing"
|
|
||||||
assert res["error_type"] == "mcp_connected_namespaces_missing"
|
|
||||||
assert "gitea-reviewer" in res["missing_namespaces"]
|
|
||||||
assert "gitea-merger" in res["missing_namespaces"]
|
|
||||||
assert "Reconnect the IDE/client MCP session" in res["exact_next_action"]
|
|
||||||
|
|
||||||
|
|
||||||
def test_assess_connected_namespace_attachment_disconnected():
|
|
||||||
res = mcp_namespace_health.assess_connected_namespace_attachment(
|
|
||||||
connected_servers=[],
|
|
||||||
attached_session_namespaces=[],
|
|
||||||
)
|
|
||||||
assert res["success"] is False
|
|
||||||
assert res["attachment_healthy"] is False
|
|
||||||
assert res["discovery_status"] == "disconnected"
|
|
||||||
|
|
||||||
|
|
||||||
def test_proof_of_connected_vs_attached_mapping():
|
|
||||||
connected = ["gitea-author", "gitea-reviewer"]
|
|
||||||
attached = ["gitea-author"]
|
|
||||||
res = mcp_namespace_health.assess_connected_namespace_attachment(
|
|
||||||
connected_servers=connected,
|
|
||||||
attached_session_namespaces=attached,
|
|
||||||
)
|
|
||||||
proof = res["proof_of_connected_vs_attached"]
|
|
||||||
assert proof["gitea-author"] == {"connected": True, "attached": True}
|
|
||||||
assert proof["gitea-reviewer"] == {"connected": True, "attached": False}
|
|
||||||
|
|
||||||
|
|
||||||
def test_unsafe_fallback_policy_enforcement():
|
|
||||||
res = mcp_namespace_health.assess_connected_namespace_attachment(
|
|
||||||
connected_servers=["gitea-author"],
|
|
||||||
attached_session_namespaces=[],
|
|
||||||
)
|
|
||||||
policy = res["unsafe_fallback_policy"]
|
|
||||||
assert "Workflow Safety Hard Stop (#708)" in policy
|
|
||||||
assert "direct imports" in policy
|
|
||||||
assert "API mutations" in policy
|
|
||||||
assert "profile hopping" in policy
|
|
||||||
assert "session-state overrides" in policy
|
|
||||||
@@ -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