fix: harden internal MCP trust boundaries

Signed-off-by: Mihai Criveti <crivetimihai@gmail.com>

Author: crivetimihai

mcpgateway/main.py

@@ -33,6 +33,7 @@
 from datetime import datetime, timezone
 from functools import lru_cache
 import hashlib
+import hmac
 import html
 import re
 import sys
@@ -309,6 +310,8 @@ def get_user_email(user):
 
 
 _INTERNAL_MCP_AUTH_CONTEXT_HEADER = "x-contextforge-auth-context"
+_INTERNAL_MCP_RUNTIME_AUTH_HEADER = "x-contextforge-mcp-runtime-auth"
+_INTERNAL_MCP_RUNTIME_AUTH_CONTEXT = "contextforge-internal-mcp-runtime-v1"
 _INTERNAL_MCP_SESSION_VALIDATED_HEADER = "x-contextforge-session-validated"
 
 
@@ -347,6 +350,33 @@ def _decode_internal_mcp_auth_context(header_value: str) -> Dict[str, Any]:
     return payload
 
 
+@lru_cache(maxsize=1)
+def _expected_internal_mcp_runtime_auth_header() -> str:
+    """Return the shared secret-derived trust header for Rust->Python MCP hops.
+
+    Returns:
+        Hex-encoded SHA-256 digest derived from the shared auth secret.
+    """
+    secret = settings.auth_encryption_secret.get_secret_value()
+    material = f"{secret}:{_INTERNAL_MCP_RUNTIME_AUTH_CONTEXT}".encode("utf-8")
+    return hashlib.sha256(material).hexdigest()
+
+
+def _has_valid_internal_mcp_runtime_auth_header(request: Request) -> bool:
+    """Validate the shared secret-derived trust header for internal MCP requests.
+
+    Args:
+        request: Incoming internal MCP request.
+
+    Returns:
+        ``True`` when the derived trust header matches the expected value.
+    """
+    provided = request.headers.get(_INTERNAL_MCP_RUNTIME_AUTH_HEADER)
+    if not provided:
+        return False
+    return hmac.compare_digest(provided, _expected_internal_mcp_runtime_auth_header())
+
+
 def _is_trusted_internal_mcp_runtime_request(request: Request) -> bool:
     """Return whether the request came from the local Rust runtime sidecar.
 
@@ -359,7 +389,7 @@ def _is_trusted_internal_mcp_runtime_request(request: Request) -> bool:
     """
     runtime_marker = request.headers.get("x-contextforge-mcp-runtime")
     client_host = getattr(getattr(request, "client", None), "host", None)
-    return runtime_marker == "rust" and client_host in ("127.0.0.1", "::1")
+    return runtime_marker == "rust" and _has_valid_internal_mcp_runtime_auth_header(request) and client_host in ("127.0.0.1", "::1")
 
 
 def _build_internal_mcp_forwarded_user(request: Request) -> Dict[str, Any]:
@@ -9290,9 +9320,7 @@ def _lowered_request_headers() -> Dict[str, str]:
             # Catch-all for other completion/* methods (currently unsupported)
             result = {}
         elif method == "logging/setLevel":
-            # MCP logging/setLevel is a standard MCP capability invoked by clients during
-            # initialization; servers.use (not admin.system_config) keeps the handshake working.
-            await _ensure_rpc_permission(user, db, "servers.use", method, request=request)
+            await _ensure_rpc_permission(user, db, "admin.system_config", method, request=request)
             level = LogLevel(params.get("level"))
             await logging_service.set_level(level)
             result = {}

mcpgateway/middleware/token_scoping.py

@@ -13,6 +13,8 @@
 # Standard
 from datetime import datetime, timedelta, timezone
 from functools import lru_cache
+import hashlib
+import hmac
 import ipaddress
 import re
 from typing import List, Optional, Pattern, Tuple
@@ -62,6 +64,8 @@
 _INTERNAL_MCP_PATH_PREFIX = "/_internal/mcp"
 _INTERNAL_MCP_RUNTIME_HEADER = "x-contextforge-mcp-runtime"
 _INTERNAL_MCP_AUTH_CONTEXT_HEADER = "x-contextforge-auth-context"
+_INTERNAL_MCP_RUNTIME_AUTH_HEADER = "x-contextforge-mcp-runtime-auth"
+_INTERNAL_MCP_RUNTIME_AUTH_CONTEXT = "contextforge-internal-mcp-runtime-v1"
 
 # Permission map with precompiled patterns
 # Maps (HTTP method, path pattern) to required permission
@@ -1355,12 +1359,32 @@ def _is_trusted_internal_mcp_runtime_request(self, request: Request, normalized_
         if request.headers.get(_INTERNAL_MCP_RUNTIME_HEADER) != "rust":
             return False
 
+        provided_auth = request.headers.get(_INTERNAL_MCP_RUNTIME_AUTH_HEADER)
+        if not provided_auth:
+            return False
+
+        expected_auth = self._expected_internal_mcp_runtime_auth_header()
+        if not hmac.compare_digest(provided_auth, expected_auth):
+            return False
+
         if not request.headers.get(_INTERNAL_MCP_AUTH_CONTEXT_HEADER):
             return False
 
         client_host = getattr(getattr(request, "client", None), "host", None)
         return client_host in ("127.0.0.1", "::1")
 
+    @staticmethod
+    @lru_cache(maxsize=1)
+    def _expected_internal_mcp_runtime_auth_header() -> str:
+        """Return the expected shared internal-auth header for Rust MCP hops.
+
+        Returns:
+            Shared secret-derived digest expected on trusted internal Rust MCP calls.
+        """
+        secret = settings.auth_encryption_secret.get_secret_value()
+        material = f"{secret}:{_INTERNAL_MCP_RUNTIME_AUTH_CONTEXT}".encode("utf-8")
+        return hashlib.sha256(material).hexdigest()
+
 
 # Create middleware instance
 token_scoping_middleware = TokenScopingMiddleware()

mcpgateway/transports/streamablehttp_transport.py

@@ -2223,14 +2223,12 @@ async def set_logging_level(level: types.LoggingLevel) -> types.EmptyResult:
 
     if _should_enforce_streamable_rbac(user_context):
         # Layer 1: Token scope cap
-        # MCP logging/setLevel is a standard MCP capability invoked by clients during
-        # initialization; servers.use (not admin.system_config) keeps the handshake working.
-        if not _check_scoped_permission(user_context, "servers.use"):
+        if not _check_scoped_permission(user_context, "admin.system_config"):
             raise PermissionError(_ACCESS_DENIED_MSG)
         # Layer 2: RBAC check
         has_permission = await _check_streamable_permission(
             user_context=user_context,
-            permission="servers.use",
+            permission="admin.system_config",
             check_any_team=_check_any_team_for_server_scoped_rbac(user_context, server_id),
         )
         if not has_permission:

tests/unit/mcpgateway/middleware/test_token_scoping.py

@@ -12,6 +12,7 @@
 """
 
 # Standard
+import hashlib
 import json
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -21,10 +22,21 @@
 import pytest
 
 # First-Party
+from mcpgateway.config import settings
 from mcpgateway.db import Permissions
 from mcpgateway.middleware.token_scoping import _get_llm_permission_patterns, TokenScopingMiddleware
 
 
+def _trusted_internal_runtime_headers() -> dict[str, str]:
+    secret = settings.auth_encryption_secret.get_secret_value()
+    expected = hashlib.sha256(f"{secret}:contextforge-internal-mcp-runtime-v1".encode("utf-8")).hexdigest()
+    return {
+        "x-contextforge-mcp-runtime": "rust",
+        "x-contextforge-mcp-runtime-auth": expected,
+        "x-contextforge-auth-context": "trusted-payload",
+    }
+
+
 @pytest.fixture(autouse=True)
 def clear_llm_permission_pattern_cache():
     """Clear cached LLM permission regex patterns between tests."""
@@ -148,11 +160,7 @@ async def test_trusted_internal_mcp_runtime_request_bypasses_token_scoping(self,
         mock_request.url.path = "/_internal/mcp/rpc"
         mock_request.scope["path"] = "/_internal/mcp/rpc"
         mock_request.method = "POST"
-        mock_request.headers = {
-            "Authorization": "Bearer scoped-token",
-            "x-contextforge-mcp-runtime": "rust",
-            "x-contextforge-auth-context": "trusted-payload",
-        }
+        mock_request.headers = {"Authorization": "Bearer scoped-token", **_trusted_internal_runtime_headers()}
 
         call_next = AsyncMock(return_value="ok")
         with patch.object(middleware, "_extract_token_scopes", new=AsyncMock(side_effect=AssertionError("token scoping should be bypassed"))):
@@ -168,11 +176,7 @@ async def test_untrusted_internal_mcp_runtime_request_still_enforces_token_scopi
         mock_request.scope["path"] = "/_internal/mcp/rpc"
         mock_request.method = "POST"
         mock_request.client.host = "10.0.0.8"
-        mock_request.headers = {
-            "Authorization": "Bearer scoped-token",
-            "x-contextforge-mcp-runtime": "rust",
-            "x-contextforge-auth-context": "trusted-payload",
-        }
+        mock_request.headers = {"Authorization": "Bearer scoped-token", **_trusted_internal_runtime_headers()}
 
         payload = {"sub": "user@example.com", "scopes": {"permissions": ["tools.read"]}}
         with (
@@ -219,6 +223,7 @@ async def test_internal_mcp_request_without_auth_context_does_not_bypass(self, m
         mock_request.headers = {
             "Authorization": "Bearer scoped-token",
             "x-contextforge-mcp-runtime": "rust",
+            "x-contextforge-mcp-runtime-auth": _trusted_internal_runtime_headers()["x-contextforge-mcp-runtime-auth"],
         }
 
         with (

tests/unit/mcpgateway/test_main_extended.py

@@ -39,6 +39,7 @@
     InternalTrustedMCPTransportBridge,
     MCPRuntimeHeaderTransportWrapper,
     MCPPathRewriteMiddleware,
+    _expected_internal_mcp_runtime_auth_header,
     _build_internal_mcp_auth_scope,
     _build_internal_mcp_forwarded_user,
     _decode_internal_mcp_auth_context,
@@ -133,6 +134,17 @@ def _make_request(
     return request
 
 
+def _trusted_internal_mcp_headers(auth_context: dict[str, object], **extra_headers: str) -> dict[str, str]:
+    """Build trusted Rust->Python internal MCP headers for unit tests."""
+    headers = {
+        "x-contextforge-mcp-runtime": "rust",
+        "x-contextforge-mcp-runtime-auth": _expected_internal_mcp_runtime_auth_header(),
+        "x-contextforge-auth-context": base64.urlsafe_b64encode(orjson.dumps(auth_context)).decode().rstrip("="),
+    }
+    headers.update(extra_headers)
+    return headers
+
+
 def _import_fresh_main_module(
     monkeypatch: pytest.MonkeyPatch,
     *,
@@ -382,6 +394,7 @@ async def handle_streamable_http(self, scope, receive, send):
             "query_string": b"session_id=abc123",
             "headers": [
                 (b"x-contextforge-mcp-runtime", b"rust"),
+                (b"x-contextforge-mcp-runtime-auth", _expected_internal_mcp_runtime_auth_header().encode("ascii")),
                 (b"x-contextforge-auth-context", encoded_auth.encode("ascii")),
                 (b"x-contextforge-server-id", b"server-1"),
             ],
@@ -443,6 +456,7 @@ async def handle_streamable_http(self, _scope, _receive, send):
             "query_string": b"session_id=abc123",
             "headers": [
                 (b"x-contextforge-mcp-runtime", b"rust"),
+                (b"x-contextforge-mcp-runtime-auth", _expected_internal_mcp_runtime_auth_header().encode("ascii")),
                 (b"x-contextforge-auth-context", encoded_auth.encode("ascii")),
                 (b"x-contextforge-session-validated", b"rust"),
             ],
@@ -503,6 +517,7 @@ async def handle_streamable_http(self, scope, receive, send):
             "query_string": b"",
             "headers": [
                 (b"x-contextforge-mcp-runtime", b"rust"),
+                (b"x-contextforge-mcp-runtime-auth", _expected_internal_mcp_runtime_auth_header().encode("ascii")),
                 (b"x-contextforge-auth-context", encoded_auth.encode("ascii")),
                 (b"x-contextforge-server-id", b"server-1"),
             ],
@@ -532,7 +547,10 @@ async def test_bridge_rejects_missing_internal_auth_context(self):
             "method": "GET",
             "path": "/_internal/mcp/transport",
             "query_string": b"",
-            "headers": [(b"x-contextforge-mcp-runtime", b"rust")],
+            "headers": [
+                (b"x-contextforge-mcp-runtime", b"rust"),
+                (b"x-contextforge-mcp-runtime-auth", _expected_internal_mcp_runtime_auth_header().encode("ascii")),
+            ],
             "client": ("127.0.0.1", 5000),
         }
 
@@ -864,6 +882,7 @@ def test_build_internal_mcp_forwarded_user_rejects_invalid_auth_context(self):
         request = MagicMock(spec=Request)
         request.headers = {
             "x-contextforge-mcp-runtime": "rust",
+            "x-contextforge-mcp-runtime-auth": _expected_internal_mcp_runtime_auth_header(),
             "x-contextforge-auth-context": "not-base64",
         }
         request.client = SimpleNamespace(host="127.0.0.1")
@@ -875,28 +894,37 @@ def test_build_internal_mcp_forwarded_user_rejects_invalid_auth_context(self):
         assert excinfo.value.status_code == 400
         assert "Invalid trusted MCP auth context" in excinfo.value.detail
 
-    def test_build_internal_mcp_forwarded_user_sets_session_validated_and_token_teams(self):
-        """Trusted forwarded auth should copy teams and set the Rust session validation marker."""
+    def test_build_internal_mcp_forwarded_user_requires_internal_runtime_auth_header(self):
+        """Trusted Rust forwarding must include the shared internal-auth header."""
         request = MagicMock(spec=Request)
         request.headers = {
             "x-contextforge-mcp-runtime": "rust",
-            "x-contextforge-session-validated": "rust",
-            "x-contextforge-auth-context": base64.urlsafe_b64encode(
-                orjson.dumps(
-                    {
-                        "email": "user@example.com",
-                        "teams": ["team-a"],
-                        "is_authenticated": True,
-                        "is_admin": False,
-                        "permission_is_admin": True,
-                        "token_use": "session",
-                    }
-                )
-            )
-            .decode()
-            .rstrip("="),
+            "x-contextforge-auth-context": base64.urlsafe_b64encode(orjson.dumps({"email": "user@example.com"})).decode().rstrip("="),
         }
         request.client = SimpleNamespace(host="127.0.0.1")
+        request.state = MagicMock()
+
+        with pytest.raises(HTTPException) as excinfo:
+            _build_internal_mcp_forwarded_user(request)
+
+        assert excinfo.value.status_code == 403
+        assert "only available to the local Rust runtime" in excinfo.value.detail
+
+    def test_build_internal_mcp_forwarded_user_sets_session_validated_and_token_teams(self):
+        """Trusted forwarded auth should copy teams and set the Rust session validation marker."""
+        request = MagicMock(spec=Request)
+        request.headers = _trusted_internal_mcp_headers(
+            {
+                "email": "user@example.com",
+                "teams": ["team-a"],
+                "is_authenticated": True,
+                "is_admin": False,
+                "permission_is_admin": True,
+                "token_use": "session",
+            },
+            **{"x-contextforge-session-validated": "rust"},
+        )
+        request.client = SimpleNamespace(host="127.0.0.1")
         request.state = SimpleNamespace()
 
         forwarded = _build_internal_mcp_forwarded_user(request)
@@ -4814,6 +4842,19 @@ async def test_list_and_get_a2a_agents_branches(self, monkeypatch):
 class TestRpcHandling:
     """Cover RPC handler branches."""
 
+    @pytest.fixture(autouse=True)
+    def _trust_internal_rust_headers_for_handler_logic_tests(self, monkeypatch):
+        """Keep this suite focused on handler logic, not trust-boundary validation.
+
+        The trust boundary itself is covered separately by the dedicated helper
+        and middleware tests above.
+        """
+        monkeypatch.setattr(
+            "mcpgateway.main._is_trusted_internal_mcp_runtime_request",
+            lambda request: request.headers.get("x-contextforge-mcp-runtime") == "rust"
+            and getattr(getattr(request, "client", None), "host", None) in ("127.0.0.1", "::1"),
+        )
+
     @staticmethod
     def _make_request(payload: dict) -> MagicMock:
         request = MagicMock(spec=Request)
@@ -10784,18 +10825,18 @@ async def test_resources_subscribe_denied_with_servers_use_only(self):
         assert result["error"]["code"] == -32003
         assert "Access denied" in result["error"]["message"]
 
-    async def test_logging_set_level_allowed_with_servers_use(self):
-        """Token scoped to servers.use should be allowed logging/setLevel."""
+    async def test_logging_set_level_allowed_with_admin_system_config(self):
+        """Token scoped to admin.system_config should be allowed logging/setLevel."""
         payload = {"jsonrpc": "2.0", "id": 1, "method": "logging/setLevel", "params": {"level": "error"}}
-        request = self._make_request(payload, scoped_permissions=["servers.use"])
+        request = self._make_request(payload, scoped_permissions=["admin.system_config"])
 
         result = await handle_rpc(request, db=MagicMock(), user={"email": "user@example.com"})
         assert "error" not in result
 
-    async def test_logging_set_level_denied_without_servers_use(self):
-        """Token scoped to tools.read only (no servers.use) should be denied logging/setLevel."""
+    async def test_logging_set_level_denied_without_admin_system_config(self):
+        """Token scoped without admin.system_config should be denied logging/setLevel."""
         payload = {"jsonrpc": "2.0", "id": 1, "method": "logging/setLevel", "params": {"level": "error"}}
-        request = self._make_request(payload, scoped_permissions=["tools.read"])
+        request = self._make_request(payload, scoped_permissions=["servers.use"])
 
         result = await handle_rpc(request, db=MagicMock(), user={"email": "user@example.com"})
         assert result["error"]["code"] == -32003

tests/unit/mcpgateway/transports/test_streamablehttp_transport.py

@@ -3667,7 +3667,7 @@ async def test_set_logging_level_exception():
 
 @pytest.mark.asyncio
 async def test_set_logging_level_requires_servers_use(monkeypatch):
-    """logging/setLevel requires servers.use permission for authenticated users."""
+    """logging/setLevel requires admin.system_config permission for authenticated users."""
     # First-Party
     from mcpgateway.transports.streamablehttp_transport import set_logging_level
 
@@ -3691,15 +3691,15 @@ async def test_set_logging_level_requires_servers_use(monkeypatch):
     mock_logging_service.set_level = AsyncMock()
     monkeypatch.setattr("mcpgateway.transports.streamablehttp_transport.logging_service", mock_logging_service)
 
-    # Should raise PermissionError for non-admin user without servers.use
+    # Should raise PermissionError for non-admin user without admin.system_config
     with pytest.raises(PermissionError, match="Access denied"):
         await set_logging_level("info")
     mock_logging_service.set_level.assert_not_called()
 
 
 @pytest.mark.asyncio
 async def test_set_logging_level_admin_allowed(monkeypatch):
-    """logging/setLevel succeeds when the caller has servers.use permission."""
+    """logging/setLevel succeeds when the caller has admin.system_config permission."""
     # Third-Party
     from mcp import types as mcp_types