fix(mcp): detect and reject manually launched duplicate MCP role servers (Closes #686)
This commit is contained in:
@@ -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()
|
||||
Reference in New Issue
Block a user