Merge pull request 'fix(guardrail): detect and reject manually launched duplicate MCP role servers (#686)' (#925) from fix/issue-686-detect-reject-manual-mcp into master

This commit was merged in pull request #925.
This commit is contained in:
2026-07-25 18:32:23 -05:00
10 changed files with 347 additions and 15 deletions
+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).
+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*.
+117 -4
View File
@@ -14585,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).
@@ -14601,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(
@@ -14634,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:
@@ -18081,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,
@@ -18097,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"),
@@ -18110,6 +18174,13 @@ def gitea_get_runtime_context(
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
# binding; recovery itself runs during capability resolution.
try:
@@ -20567,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:
@@ -20599,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] = {
proc_info = {
"pid": pid,
"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:
# #685: report-only — no config utime, no thread, no os._exit.
@@ -20643,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,
+16
View File
@@ -225,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,
@@ -240,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,
@@ -248,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
View File
@@ -44,6 +44,8 @@ def _reset_mutation_authority(monkeypatch):
]:
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
+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()
+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