Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 103 Pull Requests 11 Actions Packages Projects Releases Wiki Activity

fix(controller): production client_instance_id launch path and instance-aware mutation gate

Browse Source 
Remediate review 655 blockers on PR #979 (issue #978):

B1 — Production launcher (mcp_application_launcher) mints one trusted
client_instance_id per application launch and propagates it to all five
namespace workers via GITEA_MCP_CLIENT_INSTANCE. launcher_entry and
multi_namespace_launcher_entries use that path. Workers never invent a
trusted ID; missing/malformed/user-supplied values fail closed.

B2 — _check_mcp_runtimes_diagnostics is instance-aware: two legitimate
instances sharing a profile are allowed when each has a distinct trusted
client_instance_id; duplicate workers for the same (instance, profile)
still fail closed. Worker identity/generation exported for peer scans.

Tests cover shared ID across five namespaces, distinct launches, multi-
instance same profile, same-instance duplicates, untrusted attribution,
and the production launcher path.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
sysadmin
2026-07-30 10:12:16 -04:00
co-authored by Claude Opus 4.8
parent 4a1cc63e94
commit 0dbff6dcd5
7 changed files with 1055 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_config_menu.py
+14 -1
View File
@@ -175,8 +175,21 @@ 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", "GITEA_CLIENT_MANAGED"})
# #978 B1: production launcher also injects trusted client identity.
required = {
"GITEA_MCP_CONFIG",
"GITEA_MCP_PROFILE",
"GITEA_CLIENT_MANAGED",
"GITEA_MCP_CLIENT",
"GITEA_MCP_CLIENT_INSTANCE",
"GITEA_MCP_INSTANCE_PROVENANCE",
}
self.assertTrue(required.issubset(set(entry["env"])), entry["env"])
self.assertEqual(entry["env"]["GITEA_MCP_PROFILE"], "prgs")
self.assertTrue(
entry["env"]["GITEA_MCP_CLIENT_INSTANCE"].startswith("inst-"),
entry["env"]["GITEA_MCP_CLIENT_INSTANCE"],
)
blob = json.dumps(entry).lower()
for word in ("token", "password", "secret"):
self.assertNotIn(word, blob)
tests/test_issue_978_instance_fleet_snapshot.py
+380
View File
@@ -1049,6 +1049,386 @@ class ClientHintsTests(unittest.TestCase):
self.assertEqual(hints["client_instance_id"], inst)
self.assertEqual(hints["fleet_run_id"], "run-1")
def test_malformed_user_supplied_instance_not_trusted(self):
import gitea_mcp_server as server
with mock.patch.dict(
os.environ,
{
"GITEA_MCP_CLIENT": "codex",
"GITEA_MCP_CLIENT_INSTANCE": "user-spoofed-not-launcher",
},
clear=False,
):
hints = server._client_identity_hints()
self.assertFalse(hints["instance_identity_trusted"])
self.assertEqual(hints["client_instance_id"], "user-spoofed-not-launcher")
class ProductionLauncherB1Tests(unittest.TestCase):
"""#978 review 655 B1: production launcher mints + propagates instance ID."""
def test_one_launch_shares_id_across_five_namespaces(self):
import mcp_application_launcher as launcher
profiles = {ns: f"prgs-{ns}" for ns in NAMESPACES}
built = launcher.build_application_mcp_servers(
profiles,
client_type="codex",
config_path="/cfg/profiles.json",
launch_nonce="launch-one",
now=NOW,
command="/venv/bin/python3",
args=["mcp_server.py"],
)
servers = built["mcpServers"]
self.assertEqual(len(servers), 5)
ids = {
servers[f"gitea-{ns}"]["env"]["GITEA_MCP_CLIENT_INSTANCE"]
for ns in NAMESPACES
}
self.assertEqual(len(ids), 1, ids)
shared = next(iter(ids))
self.assertTrue(fleet.assess_instance_identity(shared)["trusted"])
self.assertEqual(built["client_instance_id"], shared)
# No legacy placeholder on any production worker env.
for ns in NAMESPACES:
env = servers[f"gitea-{ns}"]["env"]
self.assertEqual(env["GITEA_MCP_CLIENT_INSTANCE"], shared)
self.assertEqual(
env["GITEA_MCP_INSTANCE_PROVENANCE"],
fleet.INSTANCE_ID_PROVENANCE_TRUSTED,
)
self.assertEqual(env["GITEA_MCP_CLIENT"], "codex")
self.assertEqual(env["GITEA_CLIENT_MANAGED"], "1")
self.assertFalse(env["GITEA_MCP_CLIENT_INSTANCE"].startswith("legacy-"))
proof = launcher.collect_instance_ids_from_mcp_servers(servers)
self.assertTrue(proof["shared_single_trusted_id"], proof)
def test_two_launches_receive_different_ids(self):
import mcp_application_launcher as launcher
profiles = {ns: f"prgs-{ns}" for ns in NAMESPACES}
a = launcher.build_application_mcp_servers(
profiles,
client_type="codex",
launch_nonce="L1",
now=NOW,
command="python3",
args=["mcp_server.py"],
)
b = launcher.build_application_mcp_servers(
profiles,
client_type="codex",
launch_nonce="L2",
now=NOW,
command="python3",
args=["mcp_server.py"],
)
self.assertNotEqual(a["client_instance_id"], b["client_instance_id"])
def test_resume_reuses_supplied_trusted_id(self):
import mcp_application_launcher as launcher
profiles = {ns: f"prgs-{ns}" for ns in NAMESPACES}
first = launcher.build_application_mcp_servers(
profiles,
client_type="claude_code",
launch_nonce="resume-src",
now=NOW,
command="python3",
args=["mcp_server.py"],
)
resumed = launcher.build_application_mcp_servers(
profiles,
client_type="claude_code",
client_instance_id=first["client_instance_id"],
command="python3",
args=["mcp_server.py"],
)
self.assertEqual(first["client_instance_id"], resumed["client_instance_id"])
def test_refuses_untrusted_instance_id_on_production_path(self):
import mcp_application_launcher as launcher
profiles = {ns: f"prgs-{ns}" for ns in NAMESPACES}
with self.assertRaises(ValueError):
launcher.build_application_mcp_servers(
profiles,
client_type="codex",
client_instance_id="pid-999",
command="python3",
args=["mcp_server.py"],
)
with self.assertRaises(ValueError):
launcher.namespace_worker_env(
profile_name="prgs-author",
client_type="codex",
client_instance_id="user-supplied-garbage",
)
def test_extra_env_cannot_override_trusted_instance_keys(self):
import mcp_application_launcher as launcher
trusted = fleet.generate_client_instance_id(
"codex", launch_nonce="seal", now=NOW
)
env = launcher.namespace_worker_env(
profile_name="prgs-author",
client_type="codex",
client_instance_id=trusted,
extra_env={
"GITEA_MCP_CLIENT_INSTANCE": "inst-spoofed-20260730T000000Z-deadbeefdead",
"GITEA_MCP_INSTANCE_PROVENANCE": "trusted_launcher",
"GITEA_CLIENT_MANAGED": "0",
"GITEA_MCP_CLIENT": "gemini",
"GITEA_AUTHOR_WORKTREE": "/ok/extra",
},
)
self.assertEqual(env["GITEA_MCP_CLIENT_INSTANCE"], trusted)
self.assertEqual(env["GITEA_CLIENT_MANAGED"], "1")
self.assertEqual(env["GITEA_MCP_CLIENT"], "codex")
self.assertEqual(env["GITEA_AUTHOR_WORKTREE"], "/ok/extra")
def test_gitea_config_launcher_entry_uses_production_path(self):
"""Production launcher_entry — not a test helper — mints trusted IDs."""
import gitea_config
entry_a = gitea_config.launcher_entry(
"prgs-author",
"/cfg/profiles.json",
client_type="codex",
launch_nonce="cfg-a",
)["gitea-tools"]
entry_b = gitea_config.launcher_entry(
"prgs-author",
"/cfg/profiles.json",
client_type="codex",
launch_nonce="cfg-b",
)["gitea-tools"]
id_a = entry_a["env"]["GITEA_MCP_CLIENT_INSTANCE"]
id_b = entry_b["env"]["GITEA_MCP_CLIENT_INSTANCE"]
self.assertNotEqual(id_a, id_b)
self.assertTrue(fleet.assess_instance_identity(id_a)["trusted"])
self.assertTrue(fleet.assess_instance_identity(id_b)["trusted"])
def test_multi_namespace_launcher_entries_api(self):
import gitea_config
profiles = {ns: f"prgs-{ns}" for ns in NAMESPACES}
built = gitea_config.multi_namespace_launcher_entries(
profiles,
client_type="grok",
config_path="/cfg/profiles.json",
launch_nonce="multi-api",
)
self.assertTrue(built["shared_instance_id_across_namespaces"])
self.assertEqual(built["namespace_count"], 5)
proof = __import__(
"mcp_application_launcher", fromlist=["*"]
).collect_instance_ids_from_mcp_servers(built["mcpServers"])
self.assertTrue(proof["shared_single_trusted_id"], proof)
def test_malformed_and_missing_fail_safely(self):
self.assertFalse(fleet.assess_instance_identity("")["trusted"])
self.assertFalse(fleet.assess_instance_identity(None)["trusted"])
self.assertFalse(fleet.assess_instance_identity("legacy-pid-1")["trusted"])
self.assertFalse(fleet.assess_instance_identity("not-an-inst-id")["trusted"])
self.assertFalse(fleet.assess_instance_identity("inst-")["trusted"])
self.assertFalse(fleet.assess_instance_identity("user-spoofed")["trusted"])
# Reserved inst- prefix from the launcher is trusted.
self.assertTrue(fleet.assess_instance_identity("inst-codex-1")["trusted"])
class InstanceAwareMutationGateB2Tests(unittest.TestCase):
"""#978 review 655 B2: mutation gate is instance-aware, not profile-only."""
def _ps_lines(self, rows):
# rows: list of (pid, lstart, command) — command ignored beyond mcp_server.py
header = " PID LSTART COMMAND\n"
body = "".join(
f"{pid} {lstart} /path/to/python mcp_server.py\n"
for pid, lstart, _env in rows
)
return header + body
def _side_effect(self, rows, *, self_pid=1000):
ps_out = self._ps_lines(rows)
env_by_pid = {str(pid): env for pid, _ls, env in rows}
def side_effect(args, **kwargs):
mock_run = mock.MagicMock()
if args[0] == "ps" and "eww" in args:
mock_run.stdout = env_by_pid.get(str(args[2]), "")
return mock_run
if args[0] == "ps":
mock_run.stdout = ps_out
return mock_run
if args[0] == "git":
mock_run.stdout = "deadbeef"
return mock_run
raise ValueError(args)
return side_effect
def test_two_valid_instances_same_profile_allowed(self):
import gitea_mcp_server as server
inst_a = fleet.generate_client_instance_id(
"codex", launch_nonce="gate-a", now=NOW
)
inst_b = fleet.generate_client_instance_id(
"codex", launch_nonce="gate-b", now=NOW
)
lstart = "Wed Jul 8 15:00:00 2026"
rows = [
(
2001,
lstart,
(
f"GITEA_MCP_PROFILE=prgs-author GITEA_CLIENT_MANAGED=1 "
f"GITEA_MCP_CLIENT_INSTANCE={inst_a} "
f"GITEA_MCP_WORKER_IDENTITY=worker-a "
f"GITEA_MCP_GENERATION_ID=gen-a"
),
),
(
2002,
lstart,
(
f"GITEA_MCP_PROFILE=prgs-author GITEA_CLIENT_MANAGED=1 "
f"GITEA_MCP_CLIENT_INSTANCE={inst_b} "
f"GITEA_MCP_WORKER_IDENTITY=worker-b "
f"GITEA_MCP_GENERATION_ID=gen-b"
),
),
]
with mock.patch("subprocess.run") as mock_run, mock.patch(
"os.path.getmtime", return_value=datetime(2026, 7, 8, 11, 0, 0).timestamp()
), mock.patch("os.path.exists", return_value=True), mock.patch(
"os.getpid", return_value=9999
):
mock_run.side_effect = self._side_effect(rows)
reasons = server._check_mcp_runtimes_diagnostics(
"create_issue", ["prgs-author"]
)
self.assertFalse(
any("Duplicate MCP server process(es) detected" in r for r in reasons),
reasons,
)
def test_duplicate_same_instance_and_namespace_blocked(self):
import gitea_mcp_server as server
inst = fleet.generate_client_instance_id(
"codex", launch_nonce="dup", now=NOW
)
lstart = "Wed Jul 8 15:00:00 2026"
rows = [
(
3001,
lstart,
(
f"GITEA_MCP_PROFILE=prgs-author GITEA_CLIENT_MANAGED=1 "
f"GITEA_MCP_CLIENT_INSTANCE={inst} "
f"GITEA_MCP_WORKER_IDENTITY=worker-1 "
f"GITEA_MCP_GENERATION_ID=gen-1"
),
),
(
3002,
lstart,
(
f"GITEA_MCP_PROFILE=prgs-author GITEA_CLIENT_MANAGED=1 "
f"GITEA_MCP_CLIENT_INSTANCE={inst} "
f"GITEA_MCP_WORKER_IDENTITY=worker-2 "
f"GITEA_MCP_GENERATION_ID=gen-2"
),
),
]
with mock.patch("subprocess.run") as mock_run, mock.patch(
"os.path.getmtime", return_value=datetime(2026, 7, 8, 11, 0, 0).timestamp()
), mock.patch("os.path.exists", return_value=True), mock.patch(
"os.getpid", return_value=9999
):
mock_run.side_effect = self._side_effect(rows)
reasons = server._check_mcp_runtimes_diagnostics(
"create_issue", ["prgs-author"]
)
self.assertTrue(
any("Duplicate MCP server process(es) detected" in r for r in reasons),
reasons,
)
self.assertTrue(
any("duplicate namespace" in r.lower() or "client_instance_id" in r for r in reasons),
reasons,
)
def test_missing_instance_evidence_still_fail_closed(self):
"""Two profile-sharing processes without trusted IDs remain blocked."""
import gitea_mcp_server as server
lstart = "Wed Jul 8 15:00:00 2026"
rows = [
(
4001,
lstart,
"GITEA_MCP_PROFILE=prgs-author GITEA_CLIENT_MANAGED=1",
),
(
4002,
lstart,
"GITEA_MCP_PROFILE=prgs-author GITEA_CLIENT_MANAGED=1",
),
]
with mock.patch("subprocess.run") as mock_run, mock.patch(
"os.path.getmtime", return_value=datetime(2026, 7, 8, 11, 0, 0).timestamp()
), mock.patch("os.path.exists", return_value=True), mock.patch(
"os.getpid", return_value=9999
):
mock_run.side_effect = self._side_effect(rows)
reasons = server._check_mcp_runtimes_diagnostics(
"create_issue", ["prgs-author"]
)
self.assertTrue(
any("Duplicate MCP server process(es) detected" in r for r in reasons),
reasons,
)
def test_single_instance_still_ok(self):
import gitea_mcp_server as server
inst = fleet.generate_client_instance_id(
"codex", launch_nonce="single", now=NOW
)
lstart = "Wed Jul 8 15:00:00 2026"
rows = [
(
5001,
lstart,
(
f"GITEA_MCP_PROFILE=prgs-author GITEA_CLIENT_MANAGED=1 "
f"GITEA_MCP_CLIENT_INSTANCE={inst} "
f"GITEA_MCP_WORKER_IDENTITY=worker-only "
f"GITEA_MCP_GENERATION_ID=gen-only"
),
),
]
with mock.patch("subprocess.run") as mock_run, mock.patch(
"os.path.getmtime", return_value=datetime(2026, 7, 8, 11, 0, 0).timestamp()
), mock.patch("os.path.exists", return_value=True), mock.patch(
"os.getpid", return_value=9999
):
mock_run.side_effect = self._side_effect(rows)
reasons = server._check_mcp_runtimes_diagnostics(
"create_issue", ["prgs-author"]
)
self.assertFalse(
any("Duplicate MCP server process(es) detected" in r for r in reasons),
reasons,
)
if __name__ == "__main__":
unittest.main()