fix(guard): close allow_test_bootstrap and basename entrypoint bypasses (#695)
Remove the public allow_test_bootstrap production seam so caller-controlled flags cannot forge native mutation provenance. Bind provenance to the resolved canonical entrypoint path plus a live stdio transport bind; basename-only mcp_server.py stack frames and import-only launch no longer authorize mutations. Test-mode install_test_native_runtime is pytest-only and cannot reach production Gitea mutation endpoints. Add AC9 regressions for both reviewer-found bypasses and related spoof vectors. Refs: PR #696 REQUEST_CHANGES at 253269c; issue #695 comments 11002/11005.
This commit is contained in:
@@ -9,22 +9,36 @@ formal reviews then looked identical to native approvals.
|
|||||||
|
|
||||||
## Rule
|
## Rule
|
||||||
|
|
||||||
Mutation auth, keychain fill, and controller quarantine require a **native MCP
|
Mutation auth, keychain fill, and controller quarantine require a **production
|
||||||
transport runtime** established only by the official entrypoint.
|
native MCP transport runtime** established only by:
|
||||||
|
|
||||||
|
1. the **resolved absolute path** of the canonical entrypoint
|
||||||
|
(`mcp_server.py` / `gitea_mcp_server.py` next to `mcp_daemon_guard.py`), and
|
||||||
|
2. a live **transport bind** (`bind_native_mcp_transport(transport="stdio")`)
|
||||||
|
immediately before `mcp.run`.
|
||||||
|
|
||||||
|
Basename-only trust (a renamed file called `mcp_server.py`), caller-controlled
|
||||||
|
flags (there is **no** `allow_test_bootstrap`), environment variables, stack
|
||||||
|
frame spoofing, or import-only launch are insufficient.
|
||||||
|
|
||||||
| Context | Allowed |
|
| Context | Allowed |
|
||||||
|---------|---------|
|
|---------|---------|
|
||||||
| Official MCP entrypoint (`mcp_server.py` / `gitea_mcp_server` `__main__`) calls `mark_sanctioned_daemon()` and holds a process-local runtime token | yes |
|
| Official IDE-native MCP: resolved canonical entrypoint marks + binds stdio, holds process-local runtime token | yes |
|
||||||
| pytest (hermetic unit tests) | yes |
|
| pytest (hermetic unit tests) via `is_pytest_runtime()` | yes for unit gates |
|
||||||
|
| `install_test_native_runtime()` under pytest (test-mode record) | unit-test transport gates only — **never** production Gitea mutations |
|
||||||
|
| `allow_test_bootstrap=True` (removed; must not exist) | **no** |
|
||||||
|
| Renamed runner basename `mcp_server.py` outside package root | **no** |
|
||||||
|
| Import/launch of real entrypoint without transport bind | **no** |
|
||||||
| `GITEA_MCP_SANCTIONED_DAEMON=1` alone (no process-local native runtime) | **no** (#695) |
|
| `GITEA_MCP_SANCTIONED_DAEMON=1` alone (no process-local native runtime) | **no** (#695) |
|
||||||
| `GITEA_ALLOW_DIRECT_MCP_IMPORT=1` in LLM sessions | **no** — never set in agent sessions |
|
| `GITEA_ALLOW_DIRECT_MCP_IMPORT=1` in LLM sessions | **no** — never set in agent sessions |
|
||||||
| `GITEA_ALLOW_KEYCHAIN_CLI=1` in LLM sessions | **no** — human operator only |
|
| `GITEA_ALLOW_KEYCHAIN_CLI=1` in LLM sessions | **no** — human operator only |
|
||||||
| bare `python -c 'import gitea_mcp_server; …'` or offline runners | **no** |
|
| bare `python -c 'import gitea_mcp_server; …'` or offline runners | **no** |
|
||||||
| keychain fill outside native/pytest | **no** |
|
| keychain fill outside native/pytest | **no** |
|
||||||
|
|
||||||
Native runtime is **process-local**: a random token bound to the daemon PID.
|
Native runtime is **process-local**: a random token bound to the daemon PID and
|
||||||
It is never reconstructed from environment variables, session-state files, or
|
transport phase. It is never reconstructed from environment variables,
|
||||||
importing internals in a fresh Python process.
|
session-state files, caller-controlled flags, or importing internals in a
|
||||||
|
fresh Python process.
|
||||||
|
|
||||||
## Contaminated review quarantine (#695 AC8)
|
## Contaminated review quarantine (#695 AC8)
|
||||||
|
|
||||||
|
|||||||
+8
-5
@@ -12843,9 +12843,10 @@ def gitea_quarantine_contaminated_review(
|
|||||||
427 until an independent adversarial reviewer and controller deployment
|
427 until an independent adversarial reviewer and controller deployment
|
||||||
of this tooling have completed.
|
of this tooling have completed.
|
||||||
"""
|
"""
|
||||||
# Fail closed outside native MCP before any mutation assessment.
|
# Fail closed outside production native MCP before any mutation assessment.
|
||||||
|
# Test-mode bootstrap must never reach this production mutation endpoint (#695).
|
||||||
try:
|
try:
|
||||||
mcp_daemon_guard.assert_sanctioned_mutation_runtime(
|
mcp_daemon_guard.assert_production_mutation_runtime(
|
||||||
"gitea_quarantine_contaminated_review"
|
"gitea_quarantine_contaminated_review"
|
||||||
)
|
)
|
||||||
except mcp_daemon_guard.UnsanctionedRuntimeError as exc:
|
except mcp_daemon_guard.UnsanctionedRuntimeError as exc:
|
||||||
@@ -13051,10 +13052,12 @@ def gitea_quarantine_contaminated_review(
|
|||||||
# ── Entry point ───────────────────────────────────────────────────────────────
|
# ── Entry point ───────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
# #558 / #695: mark this process as the official native MCP daemon before
|
# #558 / #695: claim the resolved canonical entrypoint, then bind the live
|
||||||
# any tool dispatch. Env vars alone cannot reconstruct native transport;
|
# native MCP transport lifecycle before any tool dispatch. Env vars,
|
||||||
# offline imports / standalone scripts fail closed on mutations.
|
# basename-only stack frames, and import-only launch cannot reconstruct
|
||||||
|
# native transport; offline imports / standalone scripts fail closed.
|
||||||
mcp_daemon_guard.mark_sanctioned_daemon()
|
mcp_daemon_guard.mark_sanctioned_daemon()
|
||||||
|
mcp_daemon_guard.bind_native_mcp_transport(transport="stdio")
|
||||||
# Lock this session's launch profile into the environment so child CLI
|
# Lock this session's launch profile into the environment so child CLI
|
||||||
# processes (e.g. review_pr.py) can detect and refuse profile
|
# processes (e.g. review_pr.py) can detect and refuse profile
|
||||||
# side-channel overrides (#199).
|
# side-channel overrides (#199).
|
||||||
|
|||||||
+268
-34
@@ -6,12 +6,18 @@ Direct ``import gitea_mcp_server`` from a shell bypasses native MCP transport.
|
|||||||
- Environment variables alone cannot reconstruct a native session
|
- Environment variables alone cannot reconstruct a native session
|
||||||
(``GITEA_MCP_SANCTIONED_DAEMON=1`` / ``GITEA_ALLOW_DIRECT_MCP_IMPORT=1`` are
|
(``GITEA_MCP_SANCTIONED_DAEMON=1`` / ``GITEA_ALLOW_DIRECT_MCP_IMPORT=1`` are
|
||||||
insufficient for mutation gates).
|
insufficient for mutation gates).
|
||||||
- A process-local runtime record is created only by the official entrypoint
|
- A process-local runtime record is established only by the resolved canonical
|
||||||
(``mcp_server.py`` calling ``mark_sanctioned_daemon``), bound to PID and a
|
entrypoint path (not basename) **and** the actual native MCP transport
|
||||||
random secret that never leaves process memory.
|
lifecycle (``bind_native_mcp_transport`` before ``mcp.run``). Merely
|
||||||
|
importing or launching the entrypoint offline does not grant mutation
|
||||||
|
authority.
|
||||||
|
- Public caller-controlled flags (including any former
|
||||||
|
``allow_test_bootstrap``) never establish trusted mutation provenance.
|
||||||
- Offline scripts that import internals fail closed on mutations.
|
- Offline scripts that import internals fail closed on mutations.
|
||||||
- Pytest remains allowed (hermetic tests); optional force flags exist for
|
- Pytest remains allowed for hermetic unit tests via ``is_pytest_runtime()``.
|
||||||
provenance regression tests.
|
A separate test-only seam may establish a **test-mode** native record for
|
||||||
|
unit tests of transport gates; that record cannot authorize production
|
||||||
|
Gitea mutation endpoints.
|
||||||
|
|
||||||
Manual deletion of session-state files is never a recovery path.
|
Manual deletion of session-state files is never a recovery path.
|
||||||
"""
|
"""
|
||||||
@@ -23,6 +29,7 @@ import inspect
|
|||||||
import os
|
import os
|
||||||
import secrets
|
import secrets
|
||||||
import time
|
import time
|
||||||
|
from pathlib import Path
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
SANCTIONED_DAEMON_ENV = "GITEA_MCP_SANCTIONED_DAEMON"
|
SANCTIONED_DAEMON_ENV = "GITEA_MCP_SANCTIONED_DAEMON"
|
||||||
@@ -34,6 +41,13 @@ FORCE_PROVENANCE_FAIL_ENV = "GITEA_TEST_FORCE_UNSANCTIONED"
|
|||||||
# Process-local native runtime (never persisted, never read from env alone).
|
# Process-local native runtime (never persisted, never read from env alone).
|
||||||
_NATIVE_RUNTIME: dict[str, Any] | None = None
|
_NATIVE_RUNTIME: dict[str, Any] | None = None
|
||||||
|
|
||||||
|
# Production transport identifiers accepted by bind_native_mcp_transport.
|
||||||
|
_PRODUCTION_TRANSPORTS = frozenset({"stdio"})
|
||||||
|
_RUNTIME_MODE_PRODUCTION = "production"
|
||||||
|
_RUNTIME_MODE_TEST = "test"
|
||||||
|
_PHASE_ENTRYPOINT_CLAIMED = "entrypoint_claimed"
|
||||||
|
_PHASE_TRANSPORT_BOUND = "transport_bound"
|
||||||
|
|
||||||
|
|
||||||
class UnsanctionedRuntimeError(RuntimeError):
|
class UnsanctionedRuntimeError(RuntimeError):
|
||||||
"""Raised when mutation/credential code runs outside a native MCP daemon."""
|
"""Raised when mutation/credential code runs outside a native MCP daemon."""
|
||||||
@@ -53,44 +67,197 @@ def is_pytest_runtime() -> bool:
|
|||||||
return bool((os.environ.get("PYTEST_CURRENT_TEST") or "").strip())
|
return bool((os.environ.get("PYTEST_CURRENT_TEST") or "").strip())
|
||||||
|
|
||||||
|
|
||||||
|
def _package_root() -> Path:
|
||||||
|
"""Directory that contains the canonical MCP entrypoint modules."""
|
||||||
|
return Path(__file__).resolve().parent
|
||||||
|
|
||||||
|
|
||||||
|
def canonical_entrypoint_paths() -> frozenset[str]:
|
||||||
|
"""Resolved absolute paths of official entrypoints (not basenames)."""
|
||||||
|
root = _package_root()
|
||||||
|
return frozenset(
|
||||||
|
{
|
||||||
|
str((root / "mcp_server.py").resolve()),
|
||||||
|
str((root / "gitea_mcp_server.py").resolve()),
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _resolve_path(path: str | None) -> str | None:
|
||||||
|
if not path:
|
||||||
|
return None
|
||||||
|
try:
|
||||||
|
return str(Path(path).resolve())
|
||||||
|
except (OSError, RuntimeError, ValueError):
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def _caller_official_entrypoint_path() -> str | None:
|
||||||
|
"""Return the resolved canonical entrypoint path in the call stack, or None.
|
||||||
|
|
||||||
|
Basename-only matches (e.g. an attacker file named ``mcp_server.py``
|
||||||
|
elsewhere) are rejected. The path must equal one of
|
||||||
|
:func:`canonical_entrypoint_paths`.
|
||||||
|
"""
|
||||||
|
canonical = canonical_entrypoint_paths()
|
||||||
|
for frame in inspect.stack()[1:20]:
|
||||||
|
resolved = _resolve_path(frame.filename)
|
||||||
|
if resolved and resolved in canonical:
|
||||||
|
return resolved
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
def _caller_is_official_entrypoint() -> bool:
|
def _caller_is_official_entrypoint() -> bool:
|
||||||
"""True when mark_sanctioned_daemon is invoked from mcp_server.py."""
|
"""True when invoked from a resolved canonical entrypoint path (#695)."""
|
||||||
for frame in inspect.stack()[1:12]:
|
return _caller_official_entrypoint_path() is not None
|
||||||
path = (frame.filename or "").replace("\\", "/")
|
|
||||||
base = path.rsplit("/", 1)[-1]
|
|
||||||
if base == "mcp_server.py":
|
|
||||||
return True
|
|
||||||
return False
|
|
||||||
|
|
||||||
|
|
||||||
def mark_sanctioned_daemon(*, allow_test_bootstrap: bool = False) -> dict[str, Any]:
|
def _new_runtime_token() -> tuple[str, str]:
|
||||||
"""Mark this process as the official native MCP daemon (#695).
|
token = secrets.token_hex(32)
|
||||||
|
fingerprint = hashlib.sha256(token.encode()).hexdigest()[:16]
|
||||||
|
return token, fingerprint
|
||||||
|
|
||||||
Only the official ``mcp_server.py`` entrypoint (or pytest test bootstrap)
|
|
||||||
may establish native transport. Setting env vars alone is insufficient.
|
def mark_sanctioned_daemon() -> dict[str, Any]:
|
||||||
|
"""Claim the official entrypoint for this process (#695).
|
||||||
|
|
||||||
|
This alone does **not** authorize mutations. Callers must subsequently
|
||||||
|
bind the native MCP transport via :func:`bind_native_mcp_transport`.
|
||||||
|
|
||||||
|
Only a stack frame whose **resolved absolute path** is the canonical
|
||||||
|
``mcp_server.py`` or ``gitea_mcp_server.py`` next to this module may
|
||||||
|
claim the entrypoint. Basename spoofing is rejected.
|
||||||
|
|
||||||
|
There is no public ``allow_test_bootstrap`` argument: caller-controlled
|
||||||
|
flags must never establish trusted mutation provenance. Hermetic tests
|
||||||
|
use :func:`install_test_native_runtime` (pytest-only, test mode).
|
||||||
"""
|
"""
|
||||||
global _NATIVE_RUNTIME
|
global _NATIVE_RUNTIME
|
||||||
if not is_pytest_runtime() and not allow_test_bootstrap:
|
if is_pytest_runtime():
|
||||||
if not _caller_is_official_entrypoint():
|
# Under pytest, production mark is a no-op for transport authority.
|
||||||
raise UnsanctionedRuntimeError(
|
# Tests that need a native-transport record use install_test_native_runtime.
|
||||||
"mark_sanctioned_daemon rejected: not called from official "
|
return native_runtime_status()
|
||||||
"mcp_server.py entrypoint (#695). Offline import / standalone "
|
|
||||||
"scripts cannot reconstruct native transport. Stop after native "
|
entrypoint_path = _caller_official_entrypoint_path()
|
||||||
"MCP failure; do not run offline mutation helpers."
|
if entrypoint_path is None:
|
||||||
)
|
raise UnsanctionedRuntimeError(
|
||||||
token = secrets.token_hex(32)
|
"mark_sanctioned_daemon rejected: not called from the resolved "
|
||||||
|
"canonical MCP entrypoint path (#695). Basename-only names "
|
||||||
|
"(e.g. a renamed runner called mcp_server.py) are insufficient. "
|
||||||
|
"Offline import / standalone scripts cannot reconstruct native "
|
||||||
|
"transport. Stop after native MCP failure; do not run offline "
|
||||||
|
"mutation helpers."
|
||||||
|
)
|
||||||
|
|
||||||
|
token, fingerprint = _new_runtime_token()
|
||||||
_NATIVE_RUNTIME = {
|
_NATIVE_RUNTIME = {
|
||||||
"token": token,
|
"token": token,
|
||||||
"token_fingerprint": hashlib.sha256(token.encode()).hexdigest()[:16],
|
"token_fingerprint": fingerprint,
|
||||||
"pid": os.getpid(),
|
"pid": os.getpid(),
|
||||||
"started_at": time.time(),
|
"started_at": time.time(),
|
||||||
"entrypoint": "mcp_server",
|
"entrypoint": "mcp_server",
|
||||||
|
"entrypoint_path": entrypoint_path,
|
||||||
|
"phase": _PHASE_ENTRYPOINT_CLAIMED,
|
||||||
|
"transport": None,
|
||||||
|
"mode": _RUNTIME_MODE_PRODUCTION,
|
||||||
}
|
}
|
||||||
# Legacy signal for older probes; alone does not authorize mutations.
|
# Legacy signal for older probes; alone does not authorize mutations.
|
||||||
os.environ[SANCTIONED_DAEMON_ENV] = "1"
|
os.environ[SANCTIONED_DAEMON_ENV] = "1"
|
||||||
return native_runtime_status()
|
return native_runtime_status()
|
||||||
|
|
||||||
|
|
||||||
|
def bind_native_mcp_transport(*, transport: str) -> dict[str, Any]:
|
||||||
|
"""Bind the live native MCP transport lifecycle (#695).
|
||||||
|
|
||||||
|
Must be called from the resolved canonical entrypoint immediately before
|
||||||
|
the real MCP server transport loop (e.g. ``mcp.run(transport=\"stdio\")``).
|
||||||
|
Requires a prior successful :func:`mark_sanctioned_daemon` claim in this
|
||||||
|
process. Import-only or offline launch without this bind leaves
|
||||||
|
:func:`is_native_mcp_transport` false.
|
||||||
|
"""
|
||||||
|
global _NATIVE_RUNTIME
|
||||||
|
transport_name = (transport or "").strip().lower()
|
||||||
|
if transport_name not in _PRODUCTION_TRANSPORTS:
|
||||||
|
raise UnsanctionedRuntimeError(
|
||||||
|
f"bind_native_mcp_transport rejected: transport {transport!r} is "
|
||||||
|
f"not a production MCP transport (#695). Allowed: "
|
||||||
|
f"{sorted(_PRODUCTION_TRANSPORTS)}."
|
||||||
|
)
|
||||||
|
|
||||||
|
entrypoint_path = _caller_official_entrypoint_path()
|
||||||
|
if entrypoint_path is None:
|
||||||
|
raise UnsanctionedRuntimeError(
|
||||||
|
"bind_native_mcp_transport rejected: not called from the resolved "
|
||||||
|
"canonical MCP entrypoint path (#695)."
|
||||||
|
)
|
||||||
|
|
||||||
|
if _NATIVE_RUNTIME is None or int(_NATIVE_RUNTIME.get("pid") or -1) != os.getpid():
|
||||||
|
raise UnsanctionedRuntimeError(
|
||||||
|
"bind_native_mcp_transport rejected: no entrypoint claim in this "
|
||||||
|
"process (#695). Call mark_sanctioned_daemon() from the official "
|
||||||
|
"entrypoint first."
|
||||||
|
)
|
||||||
|
|
||||||
|
if _NATIVE_RUNTIME.get("mode") != _RUNTIME_MODE_PRODUCTION:
|
||||||
|
raise UnsanctionedRuntimeError(
|
||||||
|
"bind_native_mcp_transport rejected: runtime mode is not "
|
||||||
|
"production (#695)."
|
||||||
|
)
|
||||||
|
|
||||||
|
claimed = (_NATIVE_RUNTIME.get("entrypoint_path") or "").strip()
|
||||||
|
if claimed and claimed != entrypoint_path:
|
||||||
|
raise UnsanctionedRuntimeError(
|
||||||
|
"bind_native_mcp_transport rejected: entrypoint path mismatch "
|
||||||
|
"between mark and bind (#695)."
|
||||||
|
)
|
||||||
|
|
||||||
|
_NATIVE_RUNTIME["phase"] = _PHASE_TRANSPORT_BOUND
|
||||||
|
_NATIVE_RUNTIME["transport"] = transport_name
|
||||||
|
_NATIVE_RUNTIME["entrypoint_path"] = entrypoint_path
|
||||||
|
_NATIVE_RUNTIME["bound_at"] = time.time()
|
||||||
|
os.environ[SANCTIONED_DAEMON_ENV] = "1"
|
||||||
|
return native_runtime_status()
|
||||||
|
|
||||||
|
|
||||||
|
def install_test_native_runtime() -> dict[str, Any]:
|
||||||
|
"""Pytest-only seam for hermetic native-transport unit tests (#695).
|
||||||
|
|
||||||
|
Establishes a **test-mode** process-local record so unit tests can exercise
|
||||||
|
gates that require ``is_native_mcp_transport()``. This record:
|
||||||
|
|
||||||
|
- is rejected outside pytest (including a fresh offline interpreter);
|
||||||
|
- never uses production mode;
|
||||||
|
- cannot authorize production Gitea mutation endpoints
|
||||||
|
(:func:`assert_production_mutation_runtime` / production path of
|
||||||
|
:func:`assert_sanctioned_mutation_runtime` when not under pytest).
|
||||||
|
|
||||||
|
There is no public caller-controlled flag that forges production native
|
||||||
|
transport.
|
||||||
|
"""
|
||||||
|
global _NATIVE_RUNTIME
|
||||||
|
if not is_pytest_runtime():
|
||||||
|
raise UnsanctionedRuntimeError(
|
||||||
|
"install_test_native_runtime rejected: test-mode native runtime "
|
||||||
|
"is only available under pytest (#695). allow_test_bootstrap and "
|
||||||
|
"similar caller-controlled flags do not exist and cannot authorize "
|
||||||
|
"a fresh offline interpreter."
|
||||||
|
)
|
||||||
|
token, fingerprint = _new_runtime_token()
|
||||||
|
_NATIVE_RUNTIME = {
|
||||||
|
"token": token,
|
||||||
|
"token_fingerprint": fingerprint,
|
||||||
|
"pid": os.getpid(),
|
||||||
|
"started_at": time.time(),
|
||||||
|
"entrypoint": "test_bootstrap",
|
||||||
|
"entrypoint_path": None,
|
||||||
|
"phase": _PHASE_TRANSPORT_BOUND,
|
||||||
|
"transport": "test",
|
||||||
|
"mode": _RUNTIME_MODE_TEST,
|
||||||
|
"bound_at": time.time(),
|
||||||
|
}
|
||||||
|
return native_runtime_status()
|
||||||
|
|
||||||
|
|
||||||
def clear_native_runtime_for_tests() -> None:
|
def clear_native_runtime_for_tests() -> None:
|
||||||
"""Test helper: drop native runtime (does not clear env)."""
|
"""Test helper: drop native runtime (does not clear env)."""
|
||||||
global _NATIVE_RUNTIME
|
global _NATIVE_RUNTIME
|
||||||
@@ -98,7 +265,7 @@ def clear_native_runtime_for_tests() -> None:
|
|||||||
|
|
||||||
|
|
||||||
def is_native_mcp_transport() -> bool:
|
def is_native_mcp_transport() -> bool:
|
||||||
"""True when this process holds a live native MCP runtime record (#695)."""
|
"""True when this process holds a transport-bound native runtime (#695)."""
|
||||||
if (os.environ.get(FORCE_PROVENANCE_FAIL_ENV) or "").strip() in {
|
if (os.environ.get(FORCE_PROVENANCE_FAIL_ENV) or "").strip() in {
|
||||||
"1",
|
"1",
|
||||||
"true",
|
"true",
|
||||||
@@ -111,12 +278,23 @@ def is_native_mcp_transport() -> bool:
|
|||||||
return False
|
return False
|
||||||
if not (_NATIVE_RUNTIME.get("token") or "").strip():
|
if not (_NATIVE_RUNTIME.get("token") or "").strip():
|
||||||
return False
|
return False
|
||||||
|
if _NATIVE_RUNTIME.get("phase") != _PHASE_TRANSPORT_BOUND:
|
||||||
|
return False
|
||||||
|
if not (_NATIVE_RUNTIME.get("transport") or "").strip():
|
||||||
|
return False
|
||||||
return True
|
return True
|
||||||
|
|
||||||
|
|
||||||
|
def is_production_native_mcp_transport() -> bool:
|
||||||
|
"""True only for production-mode, transport-bound native runtime."""
|
||||||
|
if not is_native_mcp_transport():
|
||||||
|
return False
|
||||||
|
return (_NATIVE_RUNTIME or {}).get("mode") == _RUNTIME_MODE_PRODUCTION
|
||||||
|
|
||||||
|
|
||||||
def is_sanctioned_mcp_daemon() -> bool:
|
def is_sanctioned_mcp_daemon() -> bool:
|
||||||
"""Backward-compatible name; #695 requires native transport, not env alone."""
|
"""Backward-compatible name; #695 requires native transport, not env alone."""
|
||||||
if is_native_mcp_transport():
|
if is_production_native_mcp_transport():
|
||||||
return True
|
return True
|
||||||
if is_pytest_runtime():
|
if is_pytest_runtime():
|
||||||
return True
|
return True
|
||||||
@@ -126,10 +304,42 @@ def is_sanctioned_mcp_daemon() -> bool:
|
|||||||
return False
|
return False
|
||||||
|
|
||||||
|
|
||||||
def assert_sanctioned_mutation_runtime(context: str = "mutation") -> None:
|
def assert_production_mutation_runtime(context: str = "mutation") -> None:
|
||||||
"""Fail closed when mutation code runs outside native MCP transport (#695)."""
|
"""Fail closed unless production native MCP transport is bound (#695).
|
||||||
if is_sanctioned_mcp_daemon():
|
|
||||||
|
Test-mode bootstrap records and pytest-only hermetic allowances do **not**
|
||||||
|
satisfy this gate. Use for production Gitea mutation endpoints that must
|
||||||
|
never be reachable via test bootstrap.
|
||||||
|
"""
|
||||||
|
if is_production_native_mcp_transport():
|
||||||
return
|
return
|
||||||
|
mode = (_NATIVE_RUNTIME or {}).get("mode")
|
||||||
|
if mode == _RUNTIME_MODE_TEST:
|
||||||
|
raise UnsanctionedRuntimeError(
|
||||||
|
f"Test-mode native runtime cannot authorize production {context} "
|
||||||
|
"(#695). install_test_native_runtime / former allow_test_bootstrap "
|
||||||
|
"must never reach real Gitea mutation endpoints."
|
||||||
|
)
|
||||||
|
assert_sanctioned_mutation_runtime(context)
|
||||||
|
|
||||||
|
|
||||||
|
def assert_sanctioned_mutation_runtime(context: str = "mutation") -> None:
|
||||||
|
"""Fail closed when mutation code runs outside native MCP transport (#695).
|
||||||
|
|
||||||
|
Under pytest, hermetic unit tests are allowed (profile/permission tests).
|
||||||
|
Outside pytest, requires production-mode transport-bound native runtime.
|
||||||
|
Test-mode records do not authorize non-pytest production mutations.
|
||||||
|
"""
|
||||||
|
if is_pytest_runtime():
|
||||||
|
return
|
||||||
|
if is_production_native_mcp_transport():
|
||||||
|
return
|
||||||
|
mode = (_NATIVE_RUNTIME or {}).get("mode")
|
||||||
|
if mode == _RUNTIME_MODE_TEST:
|
||||||
|
raise UnsanctionedRuntimeError(
|
||||||
|
f"Test-mode native runtime cannot authorize production {context} "
|
||||||
|
"(#695). Test bootstrap cannot reach production mutation endpoints."
|
||||||
|
)
|
||||||
env_spoof = (os.environ.get(SANCTIONED_DAEMON_ENV) or "").strip() in {
|
env_spoof = (os.environ.get(SANCTIONED_DAEMON_ENV) or "").strip() in {
|
||||||
"1",
|
"1",
|
||||||
"true",
|
"true",
|
||||||
@@ -139,13 +349,24 @@ def assert_sanctioned_mutation_runtime(context: str = "mutation") -> None:
|
|||||||
if env_spoof:
|
if env_spoof:
|
||||||
extra = (
|
extra = (
|
||||||
f" Note: {SANCTIONED_DAEMON_ENV} alone is not sufficient (#695); "
|
f" Note: {SANCTIONED_DAEMON_ENV} alone is not sufficient (#695); "
|
||||||
"native transport requires the official MCP entrypoint."
|
"native transport requires the official MCP entrypoint and a live "
|
||||||
|
"transport bind."
|
||||||
|
)
|
||||||
|
phase = (_NATIVE_RUNTIME or {}).get("phase")
|
||||||
|
if phase == _PHASE_ENTRYPOINT_CLAIMED:
|
||||||
|
extra = (
|
||||||
|
(extra + " ") if extra else " "
|
||||||
|
) + (
|
||||||
|
"Entrypoint was claimed but native MCP transport was never bound "
|
||||||
|
"(#695); offline launch/import of the real entrypoint does not "
|
||||||
|
"grant mutation authority."
|
||||||
)
|
)
|
||||||
raise UnsanctionedRuntimeError(
|
raise UnsanctionedRuntimeError(
|
||||||
f"Unsanctioned / non-native runtime blocked {context} (#695). "
|
f"Unsanctioned / non-native runtime blocked {context} (#695). "
|
||||||
"Do not import gitea_mcp_server or call mutation helpers from a raw "
|
"Do not import gitea_mcp_server or call mutation helpers from a raw "
|
||||||
"shell, offline runner, or ad-hoc script after native MCP failure. "
|
"shell, offline runner, or ad-hoc script after native MCP failure. "
|
||||||
"Stop and reconnect the official MCP daemon (mcp_server.py). "
|
"Stop and reconnect the official MCP daemon (mcp_server.py) over "
|
||||||
|
"native transport. "
|
||||||
f"Do not set {ALLOW_DIRECT_IMPORT_ENV} or raw token env vars in LLM "
|
f"Do not set {ALLOW_DIRECT_IMPORT_ENV} or raw token env vars in LLM "
|
||||||
f"sessions.{extra}"
|
f"sessions.{extra}"
|
||||||
)
|
)
|
||||||
@@ -181,11 +402,16 @@ def native_runtime_status() -> dict[str, Any]:
|
|||||||
rt = _NATIVE_RUNTIME or {}
|
rt = _NATIVE_RUNTIME or {}
|
||||||
return {
|
return {
|
||||||
"native_mcp_transport": is_native_mcp_transport(),
|
"native_mcp_transport": is_native_mcp_transport(),
|
||||||
|
"production_native_mcp_transport": is_production_native_mcp_transport(),
|
||||||
"pytest": is_pytest_runtime(),
|
"pytest": is_pytest_runtime(),
|
||||||
"pid": rt.get("pid"),
|
"pid": rt.get("pid"),
|
||||||
"token_fingerprint": rt.get("token_fingerprint"),
|
"token_fingerprint": rt.get("token_fingerprint"),
|
||||||
"started_at": rt.get("started_at"),
|
"started_at": rt.get("started_at"),
|
||||||
"entrypoint": rt.get("entrypoint"),
|
"entrypoint": rt.get("entrypoint"),
|
||||||
|
"entrypoint_path": rt.get("entrypoint_path"),
|
||||||
|
"phase": rt.get("phase"),
|
||||||
|
"transport": rt.get("transport"),
|
||||||
|
"mode": rt.get("mode"),
|
||||||
"env_sanctioned_alone_insufficient": True,
|
"env_sanctioned_alone_insufficient": True,
|
||||||
"sanctioned_env": SANCTIONED_DAEMON_ENV,
|
"sanctioned_env": SANCTIONED_DAEMON_ENV,
|
||||||
"allow_direct_import_env": ALLOW_DIRECT_IMPORT_ENV,
|
"allow_direct_import_env": ALLOW_DIRECT_IMPORT_ENV,
|
||||||
@@ -203,10 +429,18 @@ def runtime_status() -> dict[str, Any]:
|
|||||||
def mutation_provenance_fields() -> dict[str, Any]:
|
def mutation_provenance_fields() -> dict[str, Any]:
|
||||||
"""Fields to attach to live mutation / review audit records (#695 AC6)."""
|
"""Fields to attach to live mutation / review audit records (#695 AC6)."""
|
||||||
st = native_runtime_status()
|
st = native_runtime_status()
|
||||||
|
transport = "native_mcp" if st["native_mcp_transport"] else "untrusted"
|
||||||
|
if st.get("mode") == _RUNTIME_MODE_TEST and st["native_mcp_transport"]:
|
||||||
|
transport = "test_native_mcp"
|
||||||
return {
|
return {
|
||||||
"transport": "native_mcp" if st["native_mcp_transport"] else "untrusted",
|
"transport": transport,
|
||||||
"native_mcp_transport": bool(st["native_mcp_transport"]),
|
"native_mcp_transport": bool(st["native_mcp_transport"]),
|
||||||
|
"production_native_mcp_transport": bool(
|
||||||
|
st.get("production_native_mcp_transport")
|
||||||
|
),
|
||||||
"native_runtime_pid": st.get("pid"),
|
"native_runtime_pid": st.get("pid"),
|
||||||
"native_token_fingerprint": st.get("token_fingerprint"),
|
"native_token_fingerprint": st.get("token_fingerprint"),
|
||||||
"entrypoint": st.get("entrypoint"),
|
"entrypoint": st.get("entrypoint"),
|
||||||
|
"phase": st.get("phase"),
|
||||||
|
"mode": st.get("mode"),
|
||||||
}
|
}
|
||||||
|
|||||||
+5
-3
@@ -41,15 +41,17 @@ def check_conflict_markers():
|
|||||||
|
|
||||||
check_conflict_markers()
|
check_conflict_markers()
|
||||||
|
|
||||||
# #558: official entrypoint marks the process as the sanctioned MCP daemon
|
# #558 / #695: claim the official entrypoint before loading mutation modules.
|
||||||
# before loading mutation modules (blocks raw shell import bypasses).
|
# This alone does NOT authorize mutations — gitea_mcp_server binds the live
|
||||||
|
# native MCP transport (stdio) immediately before mcp.run. Import-only or
|
||||||
|
# offline launch without that bind fails closed on mutations.
|
||||||
try:
|
try:
|
||||||
import mcp_daemon_guard
|
import mcp_daemon_guard
|
||||||
|
|
||||||
mcp_daemon_guard.mark_sanctioned_daemon()
|
mcp_daemon_guard.mark_sanctioned_daemon()
|
||||||
except Exception:
|
except Exception:
|
||||||
# Guard import failures must not hide conflict-marker infra_stop above;
|
# Guard import failures must not hide conflict-marker infra_stop above;
|
||||||
# gitea_mcp_server main also marks sanctioned when run as __main__.
|
# gitea_mcp_server main also marks + binds when run as __main__.
|
||||||
pass
|
pass
|
||||||
|
|
||||||
# Execute the actual server logic via exec in this namespace.
|
# Execute the actual server logic via exec in this namespace.
|
||||||
|
|||||||
@@ -1,14 +1,18 @@
|
|||||||
"""Regression tests for Issue #695 — second incident (PR #694 / review 427).
|
"""Regression tests for Issue #695 — second incident (PR #694 / review 427).
|
||||||
|
|
||||||
Reproduces offline import, env-only runtime spoof, exposed-token invocation,
|
Reproduces offline import, env-only runtime spoof, exposed-token invocation,
|
||||||
direct imports, locally generated runtime keys, standalone quarantine attempts,
|
direct imports, basename entrypoint spoof, allow_test_bootstrap forgery,
|
||||||
and false “official workflow” canonical claims. Gates must fail closed.
|
standalone quarantine attempts, and false “official workflow” canonical claims.
|
||||||
|
Gates must fail closed.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import os
|
import os
|
||||||
|
import subprocess
|
||||||
|
import sys
|
||||||
import tempfile
|
import tempfile
|
||||||
|
import textwrap
|
||||||
import unittest
|
import unittest
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from unittest.mock import patch
|
from unittest.mock import patch
|
||||||
@@ -20,6 +24,26 @@ import canonical_comment_validator as ccv
|
|||||||
|
|
||||||
HEAD_694 = "1844e298809373be19a526fd39b7d8b0669eb5bd"
|
HEAD_694 = "1844e298809373be19a526fd39b7d8b0669eb5bd"
|
||||||
HEAD_OTHER = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
|
HEAD_OTHER = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
|
||||||
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||||
|
|
||||||
|
|
||||||
|
def _run_offline_snippet(snippet: str, *, env_extra: dict[str, str] | None = None) -> subprocess.CompletedProcess:
|
||||||
|
"""Execute snippet in a fresh interpreter (no pytest modules)."""
|
||||||
|
env = os.environ.copy()
|
||||||
|
env.pop("PYTEST_CURRENT_TEST", None)
|
||||||
|
env.pop(mcp_daemon_guard.FORCE_PROVENANCE_FAIL_ENV, None)
|
||||||
|
env["PYTHONPATH"] = str(REPO_ROOT) + os.pathsep + env.get("PYTHONPATH", "")
|
||||||
|
if env_extra:
|
||||||
|
env.update(env_extra)
|
||||||
|
return subprocess.run(
|
||||||
|
[sys.executable, "-c", snippet],
|
||||||
|
capture_output=True,
|
||||||
|
text=True,
|
||||||
|
cwd=str(REPO_ROOT),
|
||||||
|
env=env,
|
||||||
|
timeout=30,
|
||||||
|
check=False,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
class TestNativeTransportBinding(unittest.TestCase):
|
class TestNativeTransportBinding(unittest.TestCase):
|
||||||
@@ -46,28 +70,37 @@ class TestNativeTransportBinding(unittest.TestCase):
|
|||||||
os.environ[mcp_daemon_guard.FORCE_PROVENANCE_FAIL_ENV] = "1"
|
os.environ[mcp_daemon_guard.FORCE_PROVENANCE_FAIL_ENV] = "1"
|
||||||
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError) as ctx:
|
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError) as ctx:
|
||||||
mcp_daemon_guard.mark_sanctioned_daemon()
|
mcp_daemon_guard.mark_sanctioned_daemon()
|
||||||
self.assertIn("mcp_server.py", str(ctx.exception))
|
self.assertIn("canonical", str(ctx.exception).lower())
|
||||||
self.assertFalse(mcp_daemon_guard.is_native_mcp_transport())
|
self.assertFalse(mcp_daemon_guard.is_native_mcp_transport())
|
||||||
|
|
||||||
def test_locally_generated_runtime_key_without_entrypoint_rejected(self):
|
def test_locally_generated_runtime_key_without_entrypoint_rejected(self):
|
||||||
"""Spoofing process-local fields via mark outside entrypoint fails."""
|
"""Spoofing process-local fields via mark outside entrypoint fails."""
|
||||||
mcp_daemon_guard.clear_native_runtime_for_tests()
|
mcp_daemon_guard.clear_native_runtime_for_tests()
|
||||||
os.environ[mcp_daemon_guard.FORCE_PROVENANCE_FAIL_ENV] = "1"
|
os.environ[mcp_daemon_guard.FORCE_PROVENANCE_FAIL_ENV] = "1"
|
||||||
# Even if a caller tries allow_test_bootstrap under force-unsanctioned
|
|
||||||
# pytest path is also forced off — only real entrypoint may mark.
|
|
||||||
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError):
|
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError):
|
||||||
mcp_daemon_guard.mark_sanctioned_daemon(allow_test_bootstrap=False)
|
mcp_daemon_guard.mark_sanctioned_daemon()
|
||||||
|
|
||||||
def test_test_bootstrap_establishes_native_for_hermetic_tests(self):
|
def test_test_native_runtime_for_hermetic_tests_not_production(self):
|
||||||
mcp_daemon_guard.clear_native_runtime_for_tests()
|
mcp_daemon_guard.clear_native_runtime_for_tests()
|
||||||
st = mcp_daemon_guard.mark_sanctioned_daemon(allow_test_bootstrap=True)
|
st = mcp_daemon_guard.install_test_native_runtime()
|
||||||
self.assertTrue(st["native_mcp_transport"])
|
self.assertTrue(st["native_mcp_transport"])
|
||||||
self.assertTrue(mcp_daemon_guard.is_native_mcp_transport())
|
self.assertTrue(mcp_daemon_guard.is_native_mcp_transport())
|
||||||
|
self.assertFalse(mcp_daemon_guard.is_production_native_mcp_transport())
|
||||||
mcp_daemon_guard.assert_sanctioned_mutation_runtime("test-bootstrap")
|
mcp_daemon_guard.assert_sanctioned_mutation_runtime("test-bootstrap")
|
||||||
fields = mcp_daemon_guard.mutation_provenance_fields()
|
fields = mcp_daemon_guard.mutation_provenance_fields()
|
||||||
self.assertEqual(fields["transport"], "native_mcp")
|
self.assertEqual(fields["transport"], "test_native_mcp")
|
||||||
self.assertTrue(fields["native_mcp_transport"])
|
self.assertTrue(fields["native_mcp_transport"])
|
||||||
|
self.assertFalse(fields["production_native_mcp_transport"])
|
||||||
self.assertIsNotNone(fields["native_token_fingerprint"])
|
self.assertIsNotNone(fields["native_token_fingerprint"])
|
||||||
|
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError) as ctx:
|
||||||
|
mcp_daemon_guard.assert_production_mutation_runtime("prod-endpoint")
|
||||||
|
self.assertIn("Test-mode", str(ctx.exception))
|
||||||
|
|
||||||
|
def test_no_allow_test_bootstrap_parameter_on_mark(self):
|
||||||
|
import inspect
|
||||||
|
|
||||||
|
sig = inspect.signature(mcp_daemon_guard.mark_sanctioned_daemon)
|
||||||
|
self.assertNotIn("allow_test_bootstrap", sig.parameters)
|
||||||
|
|
||||||
def test_exposed_token_env_never_grants_native(self):
|
def test_exposed_token_env_never_grants_native(self):
|
||||||
"""Raw / exposed token env vars must never reconstruct native transport."""
|
"""Raw / exposed token env vars must never reconstruct native transport."""
|
||||||
@@ -96,6 +129,249 @@ class TestNativeTransportBinding(unittest.TestCase):
|
|||||||
os.environ.pop(key, None)
|
os.environ.pop(key, None)
|
||||||
|
|
||||||
|
|
||||||
|
class TestAC9BypassRegressions(unittest.TestCase):
|
||||||
|
"""AC9: empirically reproduced offline bypasses must fail closed (#695)."""
|
||||||
|
|
||||||
|
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_allow_test_bootstrap_cannot_authorize_fresh_offline_interpreter(self):
|
||||||
|
"""Finding 1: former allow_test_bootstrap forge is gone and rejected offline."""
|
||||||
|
snippet = textwrap.dedent(
|
||||||
|
"""
|
||||||
|
import inspect
|
||||||
|
import mcp_daemon_guard as g
|
||||||
|
sig = inspect.signature(g.mark_sanctioned_daemon)
|
||||||
|
assert "allow_test_bootstrap" not in sig.parameters, "bootstrap flag must not exist"
|
||||||
|
try:
|
||||||
|
g.mark_sanctioned_daemon(allow_test_bootstrap=True)
|
||||||
|
except TypeError:
|
||||||
|
pass
|
||||||
|
else:
|
||||||
|
raise SystemExit("mark_sanctioned_daemon accepted allow_test_bootstrap")
|
||||||
|
try:
|
||||||
|
g.install_test_native_runtime()
|
||||||
|
except g.UnsanctionedRuntimeError as exc:
|
||||||
|
assert "pytest" in str(exc).lower() or "#695" in str(exc)
|
||||||
|
else:
|
||||||
|
raise SystemExit("install_test_native_runtime authorized offline interpreter")
|
||||||
|
assert g.is_native_mcp_transport() is False
|
||||||
|
try:
|
||||||
|
g.assert_sanctioned_mutation_runtime("offline-bootstrap")
|
||||||
|
except g.UnsanctionedRuntimeError:
|
||||||
|
pass
|
||||||
|
else:
|
||||||
|
raise SystemExit("mutation runtime authorized after offline bootstrap attempt")
|
||||||
|
print("OK")
|
||||||
|
"""
|
||||||
|
)
|
||||||
|
proc = _run_offline_snippet(snippet)
|
||||||
|
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
|
||||||
|
self.assertIn("OK", proc.stdout)
|
||||||
|
|
||||||
|
def test_renamed_runner_named_mcp_server_py_rejected(self):
|
||||||
|
"""Finding 2: basename-only entrypoint trust is insufficient."""
|
||||||
|
with tempfile.TemporaryDirectory() as tmp:
|
||||||
|
attacker = Path(tmp) / "mcp_server.py"
|
||||||
|
attacker.write_text(
|
||||||
|
textwrap.dedent(
|
||||||
|
"""
|
||||||
|
import mcp_daemon_guard as g
|
||||||
|
try:
|
||||||
|
g.mark_sanctioned_daemon()
|
||||||
|
except g.UnsanctionedRuntimeError as exc:
|
||||||
|
print("REJECTED:" + str(exc))
|
||||||
|
raise SystemExit(0)
|
||||||
|
print("AUTHORIZED")
|
||||||
|
raise SystemExit(1)
|
||||||
|
"""
|
||||||
|
),
|
||||||
|
encoding="utf-8",
|
||||||
|
)
|
||||||
|
env = os.environ.copy()
|
||||||
|
env.pop("PYTEST_CURRENT_TEST", None)
|
||||||
|
env["PYTHONPATH"] = str(REPO_ROOT) + os.pathsep + env.get("PYTHONPATH", "")
|
||||||
|
proc = subprocess.run(
|
||||||
|
[sys.executable, str(attacker)],
|
||||||
|
capture_output=True,
|
||||||
|
text=True,
|
||||||
|
cwd=tmp,
|
||||||
|
env=env,
|
||||||
|
timeout=30,
|
||||||
|
check=False,
|
||||||
|
)
|
||||||
|
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
|
||||||
|
self.assertIn("REJECTED:", proc.stdout)
|
||||||
|
self.assertIn("canonical", proc.stdout.lower())
|
||||||
|
self.assertNotIn("AUTHORIZED", proc.stdout)
|
||||||
|
|
||||||
|
def test_direct_launch_import_canonical_entrypoint_without_transport_rejected(self):
|
||||||
|
"""Merely importing/launching real entrypoint offline must not authorize."""
|
||||||
|
snippet = textwrap.dedent(
|
||||||
|
f"""
|
||||||
|
import importlib.util
|
||||||
|
import mcp_daemon_guard as g
|
||||||
|
# Simulate claim-only phase (no transport bind).
|
||||||
|
g.clear_native_runtime_for_tests()
|
||||||
|
# Direct mark from non-entrypoint must fail.
|
||||||
|
try:
|
||||||
|
g.mark_sanctioned_daemon()
|
||||||
|
except g.UnsanctionedRuntimeError:
|
||||||
|
pass
|
||||||
|
assert g.is_native_mcp_transport() is False
|
||||||
|
# Even if someone forges entrypoint_claimed without transport bind:
|
||||||
|
g._NATIVE_RUNTIME = {{
|
||||||
|
"token": "x" * 64,
|
||||||
|
"token_fingerprint": "deadbeefdeadbeef",
|
||||||
|
"pid": __import__("os").getpid(),
|
||||||
|
"started_at": 0,
|
||||||
|
"entrypoint": "mcp_server",
|
||||||
|
"entrypoint_path": {str(REPO_ROOT / "mcp_server.py")!r},
|
||||||
|
"phase": "entrypoint_claimed",
|
||||||
|
"transport": None,
|
||||||
|
"mode": "production",
|
||||||
|
}}
|
||||||
|
assert g.is_native_mcp_transport() is False
|
||||||
|
try:
|
||||||
|
g.assert_sanctioned_mutation_runtime("import-only")
|
||||||
|
except g.UnsanctionedRuntimeError as exc:
|
||||||
|
assert "transport" in str(exc).lower() or "entrypoint" in str(exc).lower() or "#695" in str(exc)
|
||||||
|
else:
|
||||||
|
raise SystemExit("import-only claim authorized mutation")
|
||||||
|
print("OK")
|
||||||
|
"""
|
||||||
|
)
|
||||||
|
proc = _run_offline_snippet(snippet)
|
||||||
|
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
|
||||||
|
self.assertIn("OK", proc.stdout)
|
||||||
|
|
||||||
|
def test_spoofed_pytest_env_stack_path_rejected(self):
|
||||||
|
"""Spoofed pytest/env/call-stack/path evidence must not authorize offline."""
|
||||||
|
snippet = textwrap.dedent(
|
||||||
|
f"""
|
||||||
|
import os
|
||||||
|
import mcp_daemon_guard as g
|
||||||
|
os.environ["PYTEST_CURRENT_TEST"] = "spoofed::test"
|
||||||
|
os.environ[g.SANCTIONED_DAEMON_ENV] = "1"
|
||||||
|
os.environ[g.ALLOW_DIRECT_IMPORT_ENV] = "1"
|
||||||
|
# Fresh interpreter has no pytest module; PYTEST_CURRENT_TEST alone
|
||||||
|
# might still trip is_pytest_runtime — force-unsanctioned is not set.
|
||||||
|
# But install_test_native_runtime requires real pytest path; if
|
||||||
|
# PYTEST_CURRENT_TEST alone grants is_pytest_runtime, production
|
||||||
|
# mutation still requires production transport outside true pytest.
|
||||||
|
if g.is_pytest_runtime():
|
||||||
|
# Env-only pytest spoof: test install may succeed, but production
|
||||||
|
# mutation gate must still reject test mode.
|
||||||
|
g.install_test_native_runtime()
|
||||||
|
assert g.is_production_native_mcp_transport() is False
|
||||||
|
try:
|
||||||
|
g.assert_production_mutation_runtime("spoofed-pytest")
|
||||||
|
except g.UnsanctionedRuntimeError:
|
||||||
|
pass
|
||||||
|
else:
|
||||||
|
raise SystemExit("production mutation accepted test-mode under spoofed pytest env")
|
||||||
|
else:
|
||||||
|
try:
|
||||||
|
g.install_test_native_runtime()
|
||||||
|
except g.UnsanctionedRuntimeError:
|
||||||
|
pass
|
||||||
|
else:
|
||||||
|
raise SystemExit("test install without pytest evidence")
|
||||||
|
# Basename path spoof via inspect is covered elsewhere; env alone:
|
||||||
|
g.clear_native_runtime_for_tests()
|
||||||
|
os.environ.pop("PYTEST_CURRENT_TEST", None)
|
||||||
|
assert g.is_native_mcp_transport() is False
|
||||||
|
try:
|
||||||
|
g.assert_sanctioned_mutation_runtime("env-path-spoof")
|
||||||
|
except g.UnsanctionedRuntimeError:
|
||||||
|
pass
|
||||||
|
else:
|
||||||
|
raise SystemExit("env/path spoof authorized mutation")
|
||||||
|
print("OK")
|
||||||
|
"""
|
||||||
|
)
|
||||||
|
proc = _run_offline_snippet(snippet)
|
||||||
|
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
|
||||||
|
self.assertIn("OK", proc.stdout)
|
||||||
|
|
||||||
|
def test_test_bootstrap_cannot_reach_production_mutation_endpoints(self):
|
||||||
|
"""Under pytest, test-mode runtime cannot satisfy production mutation gate."""
|
||||||
|
mcp_daemon_guard.clear_native_runtime_for_tests()
|
||||||
|
mcp_daemon_guard.install_test_native_runtime()
|
||||||
|
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError) as ctx:
|
||||||
|
mcp_daemon_guard.assert_production_mutation_runtime(
|
||||||
|
"gitea_quarantine_contaminated_review"
|
||||||
|
)
|
||||||
|
msg = str(ctx.exception)
|
||||||
|
self.assertIn("Test-mode", msg)
|
||||||
|
self.assertIn("#695", msg)
|
||||||
|
|
||||||
|
# Offline: install_test_native_runtime must not authorize production mutations.
|
||||||
|
snippet = textwrap.dedent(
|
||||||
|
"""
|
||||||
|
import mcp_daemon_guard as g
|
||||||
|
try:
|
||||||
|
g.install_test_native_runtime()
|
||||||
|
except g.UnsanctionedRuntimeError:
|
||||||
|
pass
|
||||||
|
assert g.is_production_native_mcp_transport() is False
|
||||||
|
try:
|
||||||
|
g.assert_production_mutation_runtime("gitea_quarantine_contaminated_review")
|
||||||
|
except g.UnsanctionedRuntimeError:
|
||||||
|
pass
|
||||||
|
else:
|
||||||
|
raise SystemExit("production mutation authorized offline")
|
||||||
|
try:
|
||||||
|
g.assert_sanctioned_mutation_runtime("gitea_mutation")
|
||||||
|
except g.UnsanctionedRuntimeError:
|
||||||
|
pass
|
||||||
|
else:
|
||||||
|
raise SystemExit("sanctioned mutation authorized offline")
|
||||||
|
print("OK")
|
||||||
|
"""
|
||||||
|
)
|
||||||
|
proc = _run_offline_snippet(snippet)
|
||||||
|
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
|
||||||
|
self.assertIn("OK", proc.stdout)
|
||||||
|
|
||||||
|
def test_legitimate_native_transport_bind_succeeds(self):
|
||||||
|
"""Canonical entrypoint path + stdio bind establishes production native."""
|
||||||
|
mcp_daemon_guard.clear_native_runtime_for_tests()
|
||||||
|
# Simulate production path under force-unsanctioned (no pytest allowance)
|
||||||
|
# by calling internal claim/bind with patched caller path.
|
||||||
|
canonical = str((REPO_ROOT / "mcp_server.py").resolve())
|
||||||
|
|
||||||
|
def _fake_caller():
|
||||||
|
return canonical
|
||||||
|
|
||||||
|
with patch.object(
|
||||||
|
mcp_daemon_guard, "_caller_official_entrypoint_path", side_effect=_fake_caller
|
||||||
|
):
|
||||||
|
# Force non-pytest path for mark/bind logic.
|
||||||
|
with patch.object(mcp_daemon_guard, "is_pytest_runtime", return_value=False):
|
||||||
|
st1 = mcp_daemon_guard.mark_sanctioned_daemon()
|
||||||
|
self.assertFalse(st1["native_mcp_transport"])
|
||||||
|
self.assertEqual(st1["phase"], "entrypoint_claimed")
|
||||||
|
st2 = mcp_daemon_guard.bind_native_mcp_transport(transport="stdio")
|
||||||
|
self.assertTrue(st2["native_mcp_transport"])
|
||||||
|
self.assertTrue(st2["production_native_mcp_transport"])
|
||||||
|
self.assertEqual(st2["transport"], "stdio")
|
||||||
|
self.assertEqual(st2["mode"], "production")
|
||||||
|
mcp_daemon_guard.assert_sanctioned_mutation_runtime("native-ide")
|
||||||
|
mcp_daemon_guard.assert_production_mutation_runtime("native-ide")
|
||||||
|
fields = mcp_daemon_guard.mutation_provenance_fields()
|
||||||
|
self.assertEqual(fields["transport"], "native_mcp")
|
||||||
|
self.assertTrue(fields["production_native_mcp_transport"])
|
||||||
|
|
||||||
|
def test_canonical_entrypoint_paths_are_resolved_absolute(self):
|
||||||
|
paths = mcp_daemon_guard.canonical_entrypoint_paths()
|
||||||
|
self.assertTrue(any(p.endswith("mcp_server.py") for p in paths))
|
||||||
|
for p in paths:
|
||||||
|
self.assertTrue(os.path.isabs(p), p)
|
||||||
|
self.assertEqual(p, str(Path(p).resolve()))
|
||||||
|
|
||||||
|
|
||||||
class TestQuarantineWriteNativeOnly(unittest.TestCase):
|
class TestQuarantineWriteNativeOnly(unittest.TestCase):
|
||||||
def setUp(self) -> None:
|
def setUp(self) -> None:
|
||||||
self._tmp = tempfile.TemporaryDirectory()
|
self._tmp = tempfile.TemporaryDirectory()
|
||||||
@@ -133,7 +409,12 @@ class TestQuarantineWriteNativeOnly(unittest.TestCase):
|
|||||||
forensic_comment_ids=[10883, 10886],
|
forensic_comment_ids=[10883, 10886],
|
||||||
)
|
)
|
||||||
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError):
|
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError):
|
||||||
review_quarantine.write_quarantine_record(record)
|
# Force non-pytest and non-native for write path.
|
||||||
|
with patch.object(mcp_daemon_guard, "is_pytest_runtime", return_value=False):
|
||||||
|
with patch.object(
|
||||||
|
mcp_daemon_guard, "is_native_mcp_transport", return_value=False
|
||||||
|
):
|
||||||
|
review_quarantine.write_quarantine_record(record)
|
||||||
|
|
||||||
def test_confirmation_must_match_exactly(self):
|
def test_confirmation_must_match_exactly(self):
|
||||||
assessment = review_quarantine.assess_quarantine_write(
|
assessment = review_quarantine.assess_quarantine_write(
|
||||||
@@ -170,7 +451,7 @@ class TestQuarantineWriteNativeOnly(unittest.TestCase):
|
|||||||
"review_quarantine.mcp_session_state.default_state_dir",
|
"review_quarantine.mcp_session_state.default_state_dir",
|
||||||
return_value=self._tmp.name,
|
return_value=self._tmp.name,
|
||||||
):
|
):
|
||||||
mcp_daemon_guard.mark_sanctioned_daemon(allow_test_bootstrap=True)
|
mcp_daemon_guard.install_test_native_runtime()
|
||||||
record = review_quarantine.build_quarantine_record(
|
record = review_quarantine.build_quarantine_record(
|
||||||
remote="prgs",
|
remote="prgs",
|
||||||
org="org",
|
org="org",
|
||||||
@@ -305,7 +586,7 @@ class TestFeedbackQuarantineIntegration(unittest.TestCase):
|
|||||||
self._tmp = tempfile.TemporaryDirectory()
|
self._tmp = tempfile.TemporaryDirectory()
|
||||||
self.addCleanup(self._tmp.cleanup)
|
self.addCleanup(self._tmp.cleanup)
|
||||||
mcp_daemon_guard.clear_native_runtime_for_tests()
|
mcp_daemon_guard.clear_native_runtime_for_tests()
|
||||||
mcp_daemon_guard.mark_sanctioned_daemon(allow_test_bootstrap=True)
|
mcp_daemon_guard.install_test_native_runtime()
|
||||||
|
|
||||||
def tearDown(self) -> None:
|
def tearDown(self) -> None:
|
||||||
mcp_daemon_guard.clear_native_runtime_for_tests()
|
mcp_daemon_guard.clear_native_runtime_for_tests()
|
||||||
|
|||||||
@@ -30,11 +30,17 @@ class TestMcpDaemonGuard(unittest.TestCase):
|
|||||||
# Running under pytest already sets PYTEST_CURRENT_TEST.
|
# Running under pytest already sets PYTEST_CURRENT_TEST.
|
||||||
mcp_daemon_guard.assert_sanctioned_mutation_runtime("pytest")
|
mcp_daemon_guard.assert_sanctioned_mutation_runtime("pytest")
|
||||||
|
|
||||||
def test_mark_sanctioned_bootstrap_allows(self):
|
def test_test_native_runtime_install_under_pytest(self):
|
||||||
mcp_daemon_guard.clear_native_runtime_for_tests()
|
mcp_daemon_guard.clear_native_runtime_for_tests()
|
||||||
mcp_daemon_guard.mark_sanctioned_daemon(allow_test_bootstrap=True)
|
mcp_daemon_guard.install_test_native_runtime()
|
||||||
mcp_daemon_guard.assert_sanctioned_mutation_runtime("daemon")
|
|
||||||
self.assertTrue(mcp_daemon_guard.is_native_mcp_transport())
|
self.assertTrue(mcp_daemon_guard.is_native_mcp_transport())
|
||||||
|
self.assertFalse(mcp_daemon_guard.is_production_native_mcp_transport())
|
||||||
|
# Hermetic unit path still passes under pytest.
|
||||||
|
mcp_daemon_guard.assert_sanctioned_mutation_runtime("daemon")
|
||||||
|
# Production mutation gate rejects test-mode records.
|
||||||
|
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError) as ctx:
|
||||||
|
mcp_daemon_guard.assert_production_mutation_runtime("gitea_mutation")
|
||||||
|
self.assertIn("Test-mode", str(ctx.exception))
|
||||||
|
|
||||||
def test_env_alone_insufficient_when_force_unsanctioned(self):
|
def test_env_alone_insufficient_when_force_unsanctioned(self):
|
||||||
mcp_daemon_guard.clear_native_runtime_for_tests()
|
mcp_daemon_guard.clear_native_runtime_for_tests()
|
||||||
@@ -87,6 +93,11 @@ class TestMcpDaemonGuard(unittest.TestCase):
|
|||||||
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError):
|
with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError):
|
||||||
gitea_auth.get_auth_header("gitea.prgs.cc")
|
gitea_auth.get_auth_header("gitea.prgs.cc")
|
||||||
|
|
||||||
|
def test_no_allow_test_bootstrap_public_parameter(self):
|
||||||
|
"""Production mark must not accept allow_test_bootstrap (#695)."""
|
||||||
|
sig = __import__("inspect").signature(mcp_daemon_guard.mark_sanctioned_daemon)
|
||||||
|
self.assertNotIn("allow_test_bootstrap", sig.parameters)
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
unittest.main()
|
unittest.main()
|
||||||
|
|||||||
Reference in New Issue
Block a user