sysadmin
553f745f23
Bind mutation/credential paths to a process-local native MCP runtime so env spoofing, direct imports, and offline helpers cannot reconstruct session gates after native transport failure. Quarantine contaminated formal reviews under controller authority and honor quarantine in review feedback, merge eligibility, merge mutation, and canonical handoff validation. Add regression coverage for the second (PR #694 / review 427) incident class.
93 lines
3.5 KiB
Python
93 lines
3.5 KiB
Python
"""Tests for sanctioned MCP daemon guards (#558 / #695)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
import mcp_daemon_guard
|
|
import gitea_auth
|
|
|
|
|
|
class TestMcpDaemonGuard(unittest.TestCase):
|
|
def tearDown(self) -> None:
|
|
mcp_daemon_guard.clear_native_runtime_for_tests()
|
|
os.environ.pop(mcp_daemon_guard.FORCE_PROVENANCE_FAIL_ENV, None)
|
|
|
|
def test_unsanctioned_blocks_mutation_runtime(self):
|
|
env = {k: v for k, v in os.environ.items() if k not in {
|
|
mcp_daemon_guard.SANCTIONED_DAEMON_ENV,
|
|
mcp_daemon_guard.ALLOW_DIRECT_IMPORT_ENV,
|
|
"PYTEST_CURRENT_TEST",
|
|
}}
|
|
env["GITEA_TEST_FORCE_UNSANCTIONED"] = "1"
|
|
with patch.dict(os.environ, env, clear=True):
|
|
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError):
|
|
mcp_daemon_guard.assert_sanctioned_mutation_runtime("test")
|
|
|
|
def test_pytest_allows_runtime(self):
|
|
# Running under pytest already sets PYTEST_CURRENT_TEST.
|
|
mcp_daemon_guard.assert_sanctioned_mutation_runtime("pytest")
|
|
|
|
def test_mark_sanctioned_bootstrap_allows(self):
|
|
mcp_daemon_guard.clear_native_runtime_for_tests()
|
|
mcp_daemon_guard.mark_sanctioned_daemon(allow_test_bootstrap=True)
|
|
mcp_daemon_guard.assert_sanctioned_mutation_runtime("daemon")
|
|
self.assertTrue(mcp_daemon_guard.is_native_mcp_transport())
|
|
|
|
def test_env_alone_insufficient_when_force_unsanctioned(self):
|
|
mcp_daemon_guard.clear_native_runtime_for_tests()
|
|
env = {
|
|
k: v
|
|
for k, v in os.environ.items()
|
|
if k not in {
|
|
mcp_daemon_guard.SANCTIONED_DAEMON_ENV,
|
|
mcp_daemon_guard.ALLOW_DIRECT_IMPORT_ENV,
|
|
"PYTEST_CURRENT_TEST",
|
|
}
|
|
}
|
|
env[mcp_daemon_guard.SANCTIONED_DAEMON_ENV] = "1"
|
|
env["GITEA_TEST_FORCE_UNSANCTIONED"] = "1"
|
|
with patch.dict(os.environ, env, clear=True):
|
|
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError) as ctx:
|
|
mcp_daemon_guard.assert_sanctioned_mutation_runtime("env-spoof")
|
|
self.assertIn("not sufficient", str(ctx.exception))
|
|
|
|
def test_keychain_blocked_without_sanction(self):
|
|
env = {
|
|
k: v
|
|
for k, v in os.environ.items()
|
|
if k
|
|
not in {
|
|
mcp_daemon_guard.SANCTIONED_DAEMON_ENV,
|
|
mcp_daemon_guard.ALLOW_DIRECT_IMPORT_ENV,
|
|
mcp_daemon_guard.ALLOW_KEYCHAIN_CLI_ENV,
|
|
"PYTEST_CURRENT_TEST",
|
|
}
|
|
}
|
|
env["GITEA_TEST_FORCE_UNSANCTIONED"] = "1"
|
|
with patch.dict(os.environ, env, clear=True):
|
|
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError):
|
|
mcp_daemon_guard.assert_keychain_access_allowed()
|
|
|
|
def test_get_auth_header_blocked_when_unsanctioned(self):
|
|
env = {
|
|
k: v
|
|
for k, v in os.environ.items()
|
|
if k
|
|
not in {
|
|
mcp_daemon_guard.SANCTIONED_DAEMON_ENV,
|
|
mcp_daemon_guard.ALLOW_DIRECT_IMPORT_ENV,
|
|
"PYTEST_CURRENT_TEST",
|
|
}
|
|
}
|
|
env["GITEA_TEST_FORCE_UNSANCTIONED"] = "1"
|
|
with patch.dict(os.environ, env, clear=True):
|
|
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError):
|
|
gitea_auth.get_auth_header("gitea.prgs.cc")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|