Compare commits

..
13 changed files with 348 additions and 521 deletions
-41
View File
@@ -589,47 +589,6 @@ When dynamic profile switching is enabled and a profile is activated via `gitea_
2. Call `gitea_whoami` with the target remote to prove and verify the fresh Gitea authenticated identity.
This guarantees the active profile operations align with the actual Gitea authenticated user credential.
### 4. Review-State Invalidation on Profile Switch (#690)
A cross-profile activation (e.g. reviewer → author → reviewer) is a session
boundary for formal review state. On any switch where the activated profile
differs from the previous one, `gitea_activate_profile` invalidates, in
memory **and** in durable session state (for both the old and new profile
identities):
- preflight identity/capability stamps (`gitea_whoami` / `gitea_resolve_task_capability` proof),
- review workflow-load proof (`gitea_load_review_workflow`),
- the review decision lock (including `final_review_decision_ready` markers),
- the reviewer PR session lease binding,
- live namespace-health assessments.
Before any formal verdict (`gitea_mark_final_review_decision` /
`gitea_submit_pr_review`) the full reviewer preflight must be re-established
under the new profile: `gitea_whoami`, `gitea_load_review_workflow`,
`gitea_resolve_task_capability(review_pr)`, live head re-pin, and lease
re-acquire/adopt. Switching back to the earlier profile does **not**
resurrect the prior run — durable state keyed by either profile identity is
cleared at switch time.
The supported pattern remains **separate session/namespace per role**
(dual-namespace, §2): file author-side follow-ups from an author session,
not by hopping profiles inside a formal review run. Runtime profile
switching is the operator-approved exception and always carries the
re-preflight cost above.
### 5. Namespace Provenance (#690)
The server cannot derive its own client-managed MCP namespace name, so a
launcher may declare it via the `GITEA_MCP_NAMESPACE` environment variable
(e.g. `gitea-reviewer`). `gitea_whoami`, `gitea_get_runtime_context`, and
`gitea_resolve_task_capability` report `namespace_provenance` — the
configured client namespace, the active execution profile, and, for tasks
with a required namespace (`review_pr` → `gitea-reviewer`, `merge_pr` →
`gitea-merger`), a mismatch verdict. A declared namespace that disagrees
with the requested task's required namespace **fails closed**. An
undeclared namespace is reported as `unknown` and is never treated as
proof either way.
## Gitea MCP Runtime Isolation and Worktree Safety
To ensure high availability and prevent broken feature worktrees from disabling essential security/identity controls, the Gitea MCP server implements runtime isolation:
+12
View File
@@ -153,7 +153,19 @@ not a tool argument: a session must never be able to authorize itself.
## Related
- #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.
- #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-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).
-21
View File
@@ -86,24 +86,3 @@ When a namespace returns EOF, follow
When blocked, repair the IDE namespace and re-record a healthy
`client_namespace` assessment before retrying the mutation.
## Namespace provenance (#690)
A server process cannot derive the name of the client-managed namespace it is
registered under, so the launcher may declare it with the
`GITEA_MCP_NAMESPACE` environment variable (e.g. `GITEA_MCP_NAMESPACE=gitea-reviewer`).
- `gitea_whoami`, `gitea_get_runtime_context`, and
`gitea_resolve_task_capability` report `namespace_provenance`: the declared
client namespace, the active execution profile, and — for tasks with a
required namespace (`review_pr`/`submit_review``gitea-reviewer`,
`merge_pr``gitea-merger`) — a `mismatch` verdict.
- A declared namespace that disagrees with the requested task's required
namespace **fails closed** (`allowed_in_current_session=false` with a STOP
guidance entry).
- An undeclared namespace is reported as `namespace_source="unknown"` and is
never treated as proof either way.
- A profile switch via `gitea_activate_profile` clears all recorded live
namespace-health assessments; re-probe through the client before further
review/merge mutations.
+50 -1
View File
@@ -1169,10 +1169,57 @@ def server_command():
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):
"""Return a thin MCP launcher entry for *profile_name*.
Contains only command/args and the two GITEA_MCP_* env vars — never a token
Contains command/args and the GITEA_MCP_* / GITEA_CLIENT_MANAGED env vars — never a token
or password. Suitable for Claude / Gemini / Codex ``mcpServers`` blocks.
"""
command, args = server_command()
@@ -1183,11 +1230,13 @@ def launcher_entry(profile_name, config_path=None):
"env": {
"GITEA_MCP_CONFIG": config_path or DEFAULT_CONFIG_PATH,
"GITEA_MCP_PROFILE": profile_name,
"GITEA_CLIENT_MANAGED": "1",
},
}
}
def keychain_set(item_id, token, account=None, runner=subprocess.run):
"""Store *token* in the macOS keychain under service *item_id*.
+121 -126
View File
@@ -839,75 +839,6 @@ def _invalidate_preflight_identity_state() -> None:
_clear_preflight_capability_state()
# #690: session-boundary invalidation record for the most recent cross-profile
# activation. Surfaced in runtime diagnostics so a formal review run can prove
# its state was reset by a profile switch and must be fully re-established.
_PROFILE_SWITCH_INVALIDATION: dict | None = None
def _invalidate_review_state_on_profile_switch(
before_profile: str,
after_profile: str,
) -> dict:
"""Invalidate review-critical session state on a profile switch (#690).
Workflow-load proof, reviewer lease binding, the review decision lock,
live namespace health, and preflight identity/capability stamps recorded
under the prior profile are contaminated for the new role. Durable state
keyed by *either* profile identity is cleared so a reviewer author
reviewer hop cannot resurrect a stale review run: the full reviewer
preflight (gitea_whoami, gitea_load_review_workflow,
gitea_resolve_task_capability(review_pr), live head re-pin, lease
re-acquire/adopt) must be re-established before any formal verdict.
"""
global _PROFILE_SWITCH_INVALIDATION
invalidated: list[str] = []
_invalidate_preflight_identity_state()
invalidated.append("preflight_identity_capability")
review_workflow_load.clear_review_workflow_load()
invalidated.append("review_workflow_load")
_save_review_decision_lock(None)
invalidated.append("review_decision_lock")
reviewer_pr_lease.clear_session_lease()
invalidated.append("reviewer_session_lease")
if _LIVE_NAMESPACE_HEALTH:
_LIVE_NAMESPACE_HEALTH.clear()
invalidated.append("live_namespace_health")
# Durable records keyed by either profile identity must not survive the
# switch, or activating author → reviewer → author could revive a stale
# review run without re-preflight.
for identity in {before_profile, after_profile}:
if not identity:
continue
try:
mcp_session_state.clear_state(
kind=mcp_session_state.KIND_DECISION_LOCK,
profile_identity=identity,
)
mcp_session_state.clear_state(
kind=mcp_session_state.KIND_WORKFLOW_LOAD,
profile_identity=identity,
)
except Exception:
pass # best-effort durable cleanup; in-memory state already reset
invalidated.append("durable_profile_state")
_PROFILE_SWITCH_INVALIDATION = {
"from_profile": before_profile,
"to_profile": after_profile,
"invalidated": invalidated,
"invalidated_at": datetime.now(timezone.utc).isoformat(),
"re_preflight_required": True,
}
return dict(_PROFILE_SWITCH_INVALIDATION)
def record_preflight_check(
type_name: str,
resolved_role: str | None = None,
@@ -14654,6 +14585,56 @@ def _session_context_mutation_block(
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:
"""Structured operation-gate denial for gated tools (#69, #142, #897).
@@ -14670,6 +14651,10 @@ def _profile_permission_block(required_operation: str, **extra_fields) -> dict |
# #714: evaluate active profile only — never auto-switch.
_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)
if reasons:
return _build_operation_gate_refusal(
@@ -14703,6 +14688,10 @@ def _namespace_mutation_block(mutation_task: str, **extra_fields) -> dict | None
# #714: evaluate active profile only — never auto-switch.
_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:
profile = get_profile()
except Exception as exc:
@@ -17279,11 +17268,6 @@ def gitea_whoami(
"session_context_audit": session_ctx.mutation_context_audit_fields(),
"identity_match": not id_match.get("block"),
"identity_match_reasons": id_match.get("reasons") or [],
# #690 AC4: report launcher-declared client namespace alongside the
# active execution profile so drift is visible in diagnostics.
"namespace_provenance": mcp_namespace_health.namespace_provenance(
active_profile=profile["profile_name"]
),
}
if id_match.get("block"):
_invalidate_preflight_identity_state()
@@ -18155,6 +18139,9 @@ def 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 = {
"active_profile": profile["profile_name"],
"authenticated_username": username,
@@ -18171,6 +18158,9 @@ def gitea_get_runtime_context(
"review_merge_blocked_reasons": blocked_reasons,
"suggested_fix": suggested_fix,
"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_block_reasons": preflight["preflight_block_reasons"],
"preflight_workspace": preflight.get("preflight_workspace"),
@@ -18182,13 +18172,15 @@ def gitea_get_runtime_context(
"shell_health": native_mcp_preference.shell_health_status(),
"workflow_load_proof": review_workflow_load.workflow_load_status(
PROJECT_ROOT),
# #690: namespace provenance + profile-switch invalidation evidence.
"namespace_provenance": mcp_namespace_health.namespace_provenance(
active_profile=profile["profile_name"]
),
"profile_switch_invalidation": _PROFILE_SWITCH_INVALIDATION,
}
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
# binding; recovery itself runs during capability resolution.
try:
@@ -18690,17 +18682,6 @@ def gitea_activate_profile(
source="gitea_activate_profile",
)
# 4.7 #690: a profile switch is a session-boundary event for review state.
# Any workflow-load proof, reviewer lease, decision lock, namespace
# health, or preflight stamp recorded under the prior profile is
# contaminated for the new role and must be re-established under the new
# profile before any formal review verdict.
switch_invalidation = None
if before_profile != after_profile:
switch_invalidation = _invalidate_review_state_on_profile_switch(
before_profile, after_profile
)
# 5. Audit the switch if auditing is on
_audit(
"activate_profile",
@@ -18711,12 +18692,11 @@ def gitea_activate_profile(
"before": before_profile,
"after": after_profile,
"session_context": session_ctx.mutation_context_audit_fields(),
"review_state_invalidated": bool(switch_invalidation),
},
username=after_identity,
)
result = {
return {
"success": True,
"message": f"Successfully activated profile '{profile_name}' (fresh identity verification complete).",
"before_profile": before_profile,
@@ -18726,18 +18706,6 @@ def gitea_activate_profile(
"session_context_audit": session_ctx.mutation_context_audit_fields(),
"auto_profile_substitution": False,
}
if switch_invalidation is not None:
result["review_state_invalidation"] = switch_invalidation
result["re_preflight_required"] = True
result["exact_next_action"] = (
"Profile switch invalidated workflow-load proof, reviewer lease, "
"decision lock, and preflight stamps (#690). Before any formal "
"review verdict, re-run the full reviewer preflight: "
"gitea_whoami, gitea_load_review_workflow, "
"gitea_resolve_task_capability(review_pr), live head re-pin, and "
"lease re-acquire/adopt."
)
return result
@mcp.tool()
@@ -20670,7 +20638,9 @@ def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) ->
self_pid = os.getpid()
self_stale = False
running_profiles = {}
all_profile_procs: dict[str, list[dict]] = {}
unsupported_env_found = set()
for line in proc.stdout.splitlines()[1:]:
line = line.strip()
if not line or "mcp_server.py" not in line:
@@ -20702,16 +20672,55 @@ def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) ->
if match:
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
if pid == self_pid and is_stale:
self_stale = True
if profile not in running_profiles or start_time > running_profiles[profile]["start_time"]:
running_profiles[profile] = {
"pid": pid,
"start_time": start_time,
"is_stale": is_stale
}
proc_info = {
"pid": pid,
"start_time": start_time,
"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:
# #685: report-only — no config utime, no thread, no os._exit.
@@ -20746,6 +20755,7 @@ def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) ->
return reasons
@mcp.tool()
def gitea_resolve_task_capability(
task: str,
@@ -20982,22 +20992,12 @@ def gitea_resolve_task_capability(
f"{required_role} task '{task}' even if nearby permissions are "
"present (fail closed)."
)
# #690 AC4: when the launcher declares a client namespace, a task with a
# required namespace must fail closed on mismatch (e.g. review_pr served
# from an author namespace).
ns_provenance = mcp_namespace_health.namespace_provenance(
task=task_key, active_profile=profile.get("profile_name")
)
ns_mismatch_reason = None
if ns_provenance.get("mismatch"):
ns_mismatch_reason = "; ".join(ns_provenance.get("reasons") or [])
cross_host_block = bool(remote_assess.get("block"))
identity_block = bool(id_assess.get("block"))
drift_block = bool(ctx_assess.get("block"))
allowed_in_current_session = (
permission_allowed_in_current_session
and role_matches_current_session
and not ns_provenance.get("mismatch")
and not cross_host_block
and not identity_block
and not drift_block
@@ -21048,8 +21048,6 @@ def gitea_resolve_task_capability(
)
if role_mismatch_reason:
deny_parts.append(role_mismatch_reason)
if ns_mismatch_reason:
deny_parts.append(ns_mismatch_reason)
if deny_parts:
reason_msg = "; ".join(deny_parts)
elif configured and switching:
@@ -21127,8 +21125,6 @@ def gitea_resolve_task_capability(
task_role_guidance = []
if role_mismatch_reason:
task_role_guidance.append(f"STOP: {role_mismatch_reason}")
if ns_mismatch_reason:
task_role_guidance.append(f"STOP: {ns_mismatch_reason}")
if required_role == "reviewer":
if allowed_in_current_session:
task_role_guidance.append(
@@ -21195,7 +21191,6 @@ def gitea_resolve_task_capability(
"session_context_audit": session_ctx.mutation_context_audit_fields(),
"profile_remote_compatible": not cross_host_block,
"identity_match": not identity_block,
"namespace_provenance": ns_provenance,
"auto_profile_substitution": False,
}
# #685: report typed reconnect blocker without mutating config or exiting.
+16 -49
View File
@@ -16,7 +16,6 @@ Probe sources
from __future__ import annotations
import os
from typing import Any
@@ -58,56 +57,8 @@ SAFE_ENV_KEYS = (
"GITEA_SERVICE",
"GITEA_EXECUTION_ROLE",
"GITEA_MCP_CONFIG",
"GITEA_MCP_NAMESPACE",
)
# Optional launcher-provided env declaring the client-managed MCP namespace
# this process is registered under (e.g. ``gitea-reviewer``). The server
# cannot derive its own IDE namespace name, so the launcher declares it; when
# declared, reviewers/mergers can fail closed on a namespace/task mismatch
# (#690 AC4). Absence means "unknown" — reported, never guessed.
NAMESPACE_ENV = "GITEA_MCP_NAMESPACE"
def configured_client_namespace(env: dict[str, str] | None = None) -> str | None:
"""Return the launcher-declared client namespace, or None when unknown."""
source = os.environ if env is None else env
value = (source.get(NAMESPACE_ENV) or "").strip()
return value or None
def namespace_provenance(
task: str | None = None,
*,
active_profile: str | None = None,
env: dict[str, str] | None = None,
) -> dict[str, Any]:
"""Report configured client namespace vs active execution profile (#690).
When *task* carries a required namespace (``TASK_REQUIRED_NAMESPACES``)
and the launcher declared a different one, ``mismatch`` is True and the
caller must fail closed for that task. An undeclared namespace is
reported as unknown — never treated as proof either way.
"""
configured = configured_client_namespace(env)
required = TASK_REQUIRED_NAMESPACES.get(task or "")
mismatch = bool(configured and required and configured != required)
reasons: list[str] = []
if mismatch:
reasons.append(
f"configured client namespace '{configured}' does not match "
f"required namespace '{required}' for task '{task}' (fail closed)"
)
return {
"configured_namespace": configured,
"namespace_source": NAMESPACE_ENV if configured else "unknown",
"active_profile": active_profile,
"requested_task": task,
"required_namespace": required,
"mismatch": mismatch,
"reasons": reasons,
}
def _as_list(value: Any) -> list[str] | None:
if value is None:
@@ -274,6 +225,16 @@ def classify_namespace_probe(
# on bad data without treating success as IDE proof).
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 {
"success": healthy,
"healthy": healthy,
@@ -289,6 +250,9 @@ def classify_namespace_probe(
"error_message": error_message or None,
"reasons": reasons,
"remediation": remediation,
"provenance": provenance,
"is_client_managed": is_client_managed,
"unconsumed_gitea_env": unconsumed_env,
"diagnostics": {
"namespace": ns,
"required_tool": tool,
@@ -297,6 +261,9 @@ def classify_namespace_probe(
"env": env_summary,
"config_path": config_path,
"probe_source": source,
"provenance": provenance,
"is_client_managed": is_client_managed,
"unconsumed_gitea_env": unconsumed_env,
},
"blocks_merge_workflow": blocks,
}
+2 -2
View File
@@ -41,10 +41,11 @@ def _reset_mutation_authority(monkeypatch):
"GITEA_REVIEWER_WORKTREE",
"GITEA_MERGER_WORKTREE",
"GITEA_RECONCILER_WORKTREE",
"GITEA_MCP_NAMESPACE",
]:
monkeypatch.delenv(env_key, raising=False)
monkeypatch.setenv("GITEA_CLIENT_MANAGED", "1")
# Isolate durable session-state files so tests never share host cache (#559).
import tempfile
@@ -116,7 +117,6 @@ def _reset_mutation_authority(monkeypatch):
monkeypatch.setattr(mcp_server, "_ACTOR_IDENTITY_CACHE", {})
monkeypatch.setattr(mcp_server, "_REVIEW_DECISION_LOCK", None)
monkeypatch.setattr(mcp_server, "_LIVE_NAMESPACE_HEALTH", {})
monkeypatch.setattr(mcp_server, "_PROFILE_SWITCH_INVALIDATION", None)
monkeypatch.setattr(mcp_server, "_preflight_whoami_called", False)
monkeypatch.setattr(mcp_server, "_preflight_capability_called", False)
monkeypatch.setattr(mcp_server, "_preflight_resolved_role", None)
+2 -2
View File
@@ -35,7 +35,7 @@ CONFIG = {
],
"forbidden_operations": [],
"execution_profile": "full-author",
"allowed_repositories": ["Example-Org/Example-Repo"],
"allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools", "Example-Org/Example-Repo"],
},
"reviewer-no-commit": {
"enabled": True,
@@ -50,7 +50,7 @@ CONFIG = {
"gitea.repo.commit", "gitea.pr.create", "gitea.branch.push"
],
"execution_profile": "reviewer-no-commit",
"allowed_repositories": ["Example-Org/Example-Repo"],
"allowed_repositories": ["Scaled-Tech-Consulting/Gitea-Tools", "Example-Org/Example-Repo"],
},
},
"rules": {"allow_runtime_switching": False},
+1 -1
View File
@@ -175,7 +175,7 @@ class TestLauncherSnippets(unittest.TestCase):
def test_only_safe_keys_no_secrets(self):
entry = gitea_config.launcher_entry("prgs", "/cfg/profiles.json")["gitea-tools"]
self.assertEqual(set(entry), {"command", "args", "env"})
self.assertEqual(set(entry["env"]), {"GITEA_MCP_CONFIG", "GITEA_MCP_PROFILE"})
self.assertEqual(set(entry["env"]), {"GITEA_MCP_CONFIG", "GITEA_MCP_PROFILE", "GITEA_CLIENT_MANAGED"})
self.assertEqual(entry["env"]["GITEA_MCP_PROFILE"], "prgs")
blob = json.dumps(entry).lower()
for word in ("token", "password", "secret"):
@@ -0,0 +1,139 @@
"""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()
@@ -1,274 +0,0 @@
"""Regression coverage for #690: cross-role profile activation invalidation.
A mid-run profile switch (e.g. reviewer → author → reviewer) must invalidate
workflow-load proof, reviewer lease binding, review decision lock, live
namespace health, and preflight identity/capability stamps, and must require
a full reviewer preflight before any formal verdict. Namespace provenance
must be reported and fail closed on task/namespace mismatch.
"""
import json
import os
import sys
import tempfile
import unittest
from unittest.mock import patch
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import gitea_config
import mcp_namespace_health
import mcp_server
import mcp_session_state
import review_workflow_load
import reviewer_pr_lease
from tests.test_runtime_clarity import CONFIG_SWITCHING_ENABLED
class TestProfileSwitchReviewGuard(unittest.TestCase):
def setUp(self):
self._remotes_patch = patch.dict(mcp_server.REMOTES, {
"dadeschools": {"host": "gitea.example.com", "org": "Example-Org", "repo": "Example-Repo"},
"prgs": {"host": "gitea.example.com", "org": "Example-Org", "repo": "Example-Repo"},
})
self._remotes_patch.start()
mcp_server._IDENTITY_CACHE.clear()
gitea_config._active_profile_override = None
self._dir = tempfile.TemporaryDirectory()
self.config_path = os.path.join(self._dir.name, "profiles.json")
with open(self.config_path, "w", encoding="utf-8") as fh:
fh.write(json.dumps(CONFIG_SWITCHING_ENABLED))
def tearDown(self):
self._remotes_patch.stop()
mcp_server._IDENTITY_CACHE.clear()
gitea_config._active_profile_override = None
self._dir.cleanup()
def _env(self, profile="reviewer-profile"):
return {
"GITEA_MCP_CONFIG": self.config_path,
"GITEA_MCP_PROFILE": profile,
"GITEA_TOKEN_AUTHOR": "author-pass",
"GITEA_TOKEN_REVIEWER": "reviewer-pass",
"GITEA_TOKEN_MERGER": "merger-pass",
}
def _seed_contaminated_review_state(self):
"""Simulate an in-flight reviewer run under reviewer-profile."""
mcp_server._preflight_whoami_called = True
mcp_server._preflight_capability_called = True
mcp_server._preflight_resolved_role = "reviewer"
mcp_server._preflight_resolved_task = "review_pr"
review_workflow_load._REVIEW_WORKFLOW_LOAD = {"loaded": True}
mcp_server._REVIEW_DECISION_LOCK = {
"session_profile": "reviewer-profile",
"final_review_decision_ready": True,
"ready_pr_number": 688,
}
reviewer_pr_lease.record_session_lease(
{"session_id": "lease-session-1", "pr_number": 688}
)
mcp_server._LIVE_NAMESPACE_HEALTH["gitea-reviewer"] = {
"namespace": "gitea-reviewer",
"healthy": True,
"ide_namespace_proven": True,
}
# Durable records keyed by the reviewer identity must also be cleared.
mcp_session_state.save_state(
kind=mcp_session_state.KIND_WORKFLOW_LOAD,
payload={"loaded": True},
profile_identity="reviewer-profile",
)
mcp_session_state.save_state(
kind=mcp_session_state.KIND_DECISION_LOCK,
payload={"final_review_decision_ready": True, "ready_pr_number": 688},
profile_identity="reviewer-profile",
)
def _activate(self, target, logins):
with patch.object(
mcp_server, "get_auth_header", side_effect=[f"token p" for _ in logins]
), patch.object(
mcp_server, "api_request", side_effect=[{"login": l} for l in logins]
), patch.object(
mcp_server,
"_workspace_repository_slug",
return_value="Example-Org/Example-Repo",
), patch.object(
mcp_server, "_canonical_repository_slug", return_value=(None, [])
):
return mcp_server.gitea_activate_profile(profile_name=target)
# -----------------------------------------------------------------
# AC1/AC2/AC3: switch invalidates review state; re-preflight required
# -----------------------------------------------------------------
def test_switch_invalidates_review_state_and_blocks_verdict(self):
with patch.dict(os.environ, self._env("reviewer-profile"), clear=True):
self._seed_contaminated_review_state()
res = self._activate("author-profile", ["reviewer-user", "author-user"])
self.assertTrue(res["success"])
self.assertTrue(res["re_preflight_required"])
inv = res["review_state_invalidation"]
self.assertEqual(inv["from_profile"], "reviewer-profile")
self.assertEqual(inv["to_profile"], "author-profile")
for item in (
"preflight_identity_capability",
"review_workflow_load",
"review_decision_lock",
"reviewer_session_lease",
"live_namespace_health",
):
self.assertIn(item, inv["invalidated"])
# In-memory state cleared.
self.assertFalse(mcp_server._preflight_whoami_called)
self.assertFalse(mcp_server._preflight_capability_called)
self.assertIsNone(mcp_server._preflight_resolved_task)
self.assertIsNone(review_workflow_load._REVIEW_WORKFLOW_LOAD)
self.assertIsNone(mcp_server._REVIEW_DECISION_LOCK)
self.assertIsNone(reviewer_pr_lease.get_session_lease())
self.assertEqual(mcp_server._LIVE_NAMESPACE_HEALTH, {})
self.assertIsNotNone(mcp_server._PROFILE_SWITCH_INVALIDATION)
# Durable records keyed by the reviewer identity are gone.
self.assertIsNone(
mcp_session_state.load_state(
kind=mcp_session_state.KIND_WORKFLOW_LOAD,
profile_identity="reviewer-profile",
)
)
self.assertIsNone(
mcp_session_state.load_state(
kind=mcp_session_state.KIND_DECISION_LOCK,
profile_identity="reviewer-profile",
)
)
# A formal verdict without re-preflight fails closed.
reasons = mcp_server.check_review_decision_gate(
688, "APPROVE", final_review_decision_ready=True
)
self.assertTrue(reasons)
def test_switch_back_cannot_resurrect_stale_review_run(self):
with patch.dict(os.environ, self._env("reviewer-profile"), clear=True):
self._seed_contaminated_review_state()
self._activate("author-profile", ["reviewer-user", "author-user"])
res = self._activate("reviewer-profile", ["author-user", "reviewer-user"])
self.assertTrue(res["success"])
# The pre-switch review run must not reappear.
self.assertIsNone(mcp_server._REVIEW_DECISION_LOCK)
self.assertIsNone(review_workflow_load._REVIEW_WORKFLOW_LOAD)
self.assertIsNone(reviewer_pr_lease.get_session_lease())
status = review_workflow_load.workflow_load_status()
self.assertFalse(status["workflow_load_valid"])
reasons = mcp_server.check_review_decision_gate(
688, "APPROVE", final_review_decision_ready=True
)
self.assertTrue(reasons)
def test_same_profile_reactivation_keeps_state(self):
with patch.dict(os.environ, self._env("reviewer-profile"), clear=True):
self._seed_contaminated_review_state()
res = self._activate("reviewer-profile", ["reviewer-user", "reviewer-user"])
self.assertTrue(res["success"], res)
self.assertNotIn("review_state_invalidation", res)
self.assertIsNotNone(mcp_server._REVIEW_DECISION_LOCK)
self.assertTrue(mcp_server._preflight_whoami_called)
def test_clean_repreflight_after_switch_allows_gate(self):
with patch.dict(os.environ, self._env("reviewer-profile"), clear=True):
self._seed_contaminated_review_state()
self._activate("author-profile", ["reviewer-user", "author-user"])
self._activate("reviewer-profile", ["author-user", "reviewer-user"])
# Re-establish the full reviewer preflight under the new profile.
mcp_server.record_preflight_check("whoami")
mcp_server.record_preflight_check(
"capability", resolved_role="reviewer", resolved_task="review_pr"
)
mcp_server.init_review_decision_lock("dadeschools", "review_pr")
lock = mcp_server._load_review_decision_lock()
self.assertIsNotNone(lock)
lock.update(
{
"final_review_decision_ready": True,
"ready_pr_number": 688,
"ready_action": "APPROVE",
"ready_remote": "dadeschools",
"ready_org": "Example-Org",
"ready_repo": "Example-Repo",
}
)
mcp_server._save_review_decision_lock(lock)
with patch.object(
mcp_server, "_review_workflow_load_gate_reasons", return_value=[]
):
reasons = mcp_server.check_review_decision_gate(
688,
"APPROVE",
final_review_decision_ready=True,
remote="dadeschools",
)
self.assertEqual(reasons, [])
# -----------------------------------------------------------------
# AC4: namespace provenance reporting + fail-closed mismatch
# -----------------------------------------------------------------
def test_namespace_provenance_mismatch_detection(self):
prov = mcp_namespace_health.namespace_provenance(
task="review_pr",
active_profile="reviewer-profile",
env={"GITEA_MCP_NAMESPACE": "gitea-author"},
)
self.assertTrue(prov["mismatch"])
self.assertEqual(prov["required_namespace"], "gitea-reviewer")
prov_ok = mcp_namespace_health.namespace_provenance(
task="review_pr",
active_profile="reviewer-profile",
env={"GITEA_MCP_NAMESPACE": "gitea-reviewer"},
)
self.assertFalse(prov_ok["mismatch"])
prov_unknown = mcp_namespace_health.namespace_provenance(
task="review_pr", active_profile="reviewer-profile", env={}
)
self.assertIsNone(prov_unknown["configured_namespace"])
self.assertFalse(prov_unknown["mismatch"])
self.assertEqual(prov_unknown["namespace_source"], "unknown")
@patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
def test_whoami_reports_namespace_provenance(self, _auth, _api):
env = self._env("reviewer-profile")
env["GITEA_MCP_NAMESPACE"] = "gitea-reviewer"
with patch.dict(os.environ, env, clear=True):
res = mcp_server.gitea_whoami(remote="dadeschools")
prov = res["namespace_provenance"]
self.assertEqual(prov["configured_namespace"], "gitea-reviewer")
self.assertEqual(prov["active_profile"], "reviewer-profile")
self.assertFalse(prov["mismatch"])
@patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
@patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
def test_resolve_fails_closed_on_namespace_mismatch(self, _auth, _api):
env = self._env("reviewer-profile")
env["GITEA_MCP_NAMESPACE"] = "gitea-author"
with patch.dict(os.environ, env, clear=True):
res = mcp_server.gitea_resolve_task_capability(
task="review_pr", kwargs="{}", remote="dadeschools"
)
self.assertFalse(res["allowed_in_current_session"])
self.assertTrue(res["namespace_provenance"]["mismatch"])
self.assertTrue(
any("namespace" in g for g in res["task_role_guidance"])
)
if __name__ == "__main__":
unittest.main()
+3 -3
View File
@@ -35,10 +35,10 @@ class TestMcpStaleRuntime(unittest.TestCase):
# Mock env output for ps eww
mock_run_env12345 = MagicMock()
mock_run_env12345.stdout = "GITEA_MCP_PROFILE=prgs-reconciler"
mock_run_env12345.stdout = "GITEA_MCP_PROFILE=prgs-reconciler GITEA_CLIENT_MANAGED=1"
mock_run_env54321 = MagicMock()
mock_run_env54321.stdout = "GITEA_MCP_PROFILE=prgs-author"
mock_run_env54321.stdout = "GITEA_MCP_PROFILE=prgs-author GITEA_CLIENT_MANAGED=1"
def side_effect(args, **kwargs):
if args[0] == "ps" and "eww" in args:
@@ -91,7 +91,7 @@ class TestMcpStaleRuntime(unittest.TestCase):
mock_run_ps.stdout = ps_output
mock_run_env = MagicMock()
mock_run_env.stdout = "GITEA_MCP_PROFILE=prgs-author"
mock_run_env.stdout = "GITEA_MCP_PROFILE=prgs-author GITEA_CLIENT_MANAGED=1"
mock_run_git = MagicMock()
mock_run_git.stdout = "FAKE2" # different SHA
+2 -1
View File
@@ -243,9 +243,10 @@ class TestRuntimeClarity(unittest.TestCase):
self.assertIn("switching is disabled", res["message"].lower())
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.get_auth_header")
def test_activate_profile_succeeds_when_enabled(self, mock_auth, mock_api):
def test_activate_profile_succeeds_when_enabled(self, mock_auth, mock_api, mock_trusted):
self._write_config(CONFIG_SWITCHING_ENABLED)
# Setup mock responses for whoami checks