Exempt the trusted-internal dispatch from edge middleware

.env.example

@@ -64,7 +64,7 @@ CSRF_ROTATE_ON_LOGIN=true
 CSRF_TRUSTED_ORIGINS=["http://localhost:4444","http://localhost:8080"]
 
 # Paths exempt from CSRF protection (JSON array)
-CSRF_EXEMPT_PATHS=["/health","/auth/login","/auth/logout","/auth/refresh","/auth/email/login","/auth/email/register","/auth/email/forgot-password","/auth/email/reset-password","/admin","/oauth/fetch-tools","/docs","/redoc","/openapi.json","/metrics","/mcp/","/sse","/message"]
+CSRF_EXEMPT_PATHS=["/health","/auth/login","/auth/logout","/auth/refresh","/auth/email/login","/auth/email/register","/auth/email/forgot-password","/auth/email/reset-password","/admin","/oauth/fetch-tools","/docs","/redoc","/openapi.json","/metrics","/mcp/","/sse","/message","/_internal/mcp/"]
 
 # Bootstrap admin credentials (email auth)
 # PRODUCTION: Change these values

mcpgateway/auth_context.py

@@ -310,6 +310,94 @@ def has_valid_internal_mcp_runtime_auth_header(request: Request) -> bool:
     return hmac.compare_digest(provided, _expected_internal_mcp_runtime_auth_header())
 
 
+# Internal-dispatch trust gate, defined once and shared by all callers.
+INTERNAL_MCP_PATH_PREFIX = "/_internal/mcp"
+INTERNAL_A2A_PATH_PREFIX = "/_internal/a2a"
+INTERNAL_RUNTIME_MARKER_HEADER = "x-contextforge-mcp-runtime"
+INTERNAL_AUTH_CONTEXT_HEADER = "x-contextforge-auth-context"
+TRUSTED_INTERNAL_RUNTIME_MARKERS = frozenset({"rust", "affinity"})
+
+
+def _internal_path_requires_auth_context(path: str) -> bool:
+    """Whether an internal route requires an auth context.
+
+    Only ``*/authenticate`` is exempt, since it creates the context; every
+    other internal route must carry one.
+
+    Args:
+        path: The request path to classify.
+
+    Returns:
+        ``False`` for ``*/authenticate``, otherwise ``True``.
+    """
+    return not path.rstrip("/").endswith("/authenticate")
+
+
+def is_trusted_internal_runtime_request(
+    request: Request,
+    *,
+    allowed_prefixes: tuple[str, ...],
+    require_auth_context: bool,
+    path: Optional[str] = None,
+) -> bool:
+    """Return whether a request is a trusted in-process internal-runtime hop.
+
+    The trust boundary is the HMAC header and, when required, the encoded
+    ``x-contextforge-auth-context``. The loopback check is defense in depth,
+    not an independent gate: ProxyHeaders(trusted_hosts="*") lets a direct
+    external caller influence ``request.client.host`` (the replay is hardened
+    separately by stripping forwarded / client-IP headers).
+
+    Args:
+        request: Incoming request to inspect.
+        allowed_prefixes: Internal path prefixes this caller trusts.
+        require_auth_context: Require a non-empty ``x-contextforge-auth-context``.
+        path: Explicit path override for callers that strip a ``root_path``;
+            defaults to ``request.url.path``.
+
+    Returns:
+        ``True`` only when prefix, runtime marker, HMAC, optional auth context,
+        and loopback all hold; otherwise ``False``.
+    """
+    p = path if path is not None else (getattr(getattr(request, "url", None), "path", "") or "")
+    if not any(p == prefix or p.startswith(f"{prefix}/") for prefix in allowed_prefixes):
+        return False
+    if request.headers.get(INTERNAL_RUNTIME_MARKER_HEADER) not in TRUSTED_INTERNAL_RUNTIME_MARKERS:
+        return False
+    if not has_valid_internal_mcp_runtime_auth_header(request):
+        return False
+    if require_auth_context and not request.headers.get(INTERNAL_AUTH_CONTEXT_HEADER):
+        return False
+    client_host = getattr(getattr(request, "client", None), "host", None)
+    return client_host in ("127.0.0.1", "::1")
+
+
+def is_trusted_internal_mcp_request(request: Request, *, path: Optional[str] = None) -> bool:
+    """MCP + A2A internal trust gate with a path-aware auth-context requirement.
+
+    ``*/authenticate`` routes are exempt from the auth-context requirement;
+    every other internal route requires it. ``/_internal/a2a/*`` is trusted
+    only when the A2A feature is enabled.
+
+    Args:
+        request: Incoming request to inspect.
+        path: Explicit path override (e.g. a ``root_path``-stripped path);
+            defaults to ``request.url.path``.
+
+    Returns:
+        ``True`` when the request is a trusted internal MCP/A2A hop.
+    """
+    p = path if path is not None else (getattr(getattr(request, "url", None), "path", "") or "")
+    if p.startswith(f"{INTERNAL_A2A_PATH_PREFIX}/") and not settings.mcpgateway_a2a_enabled:
+        return False
+    return is_trusted_internal_runtime_request(
+        request,
+        allowed_prefixes=(INTERNAL_MCP_PATH_PREFIX, INTERNAL_A2A_PATH_PREFIX),
+        require_auth_context=_internal_path_requires_auth_context(p),
+        path=p,
+    )
+
+
 def get_token_teams_from_request(request: Request) -> Optional[List[str]]:
     """Extract and normalize teams from verified JWT token.
 

mcpgateway/main.py

@@ -81,8 +81,8 @@
     get_scoped_resource_access_context,
     get_token_teams_from_request,
     get_user_email,
-    has_valid_internal_mcp_runtime_auth_header,
     INTERNAL_MCP_SESSION_VALIDATED_HEADER,
+    is_trusted_internal_mcp_request,
 )
 from mcpgateway.cache import ResourceCache, SessionRegistry
 from mcpgateway.common.models import InitializeResult
@@ -296,17 +296,7 @@ def _is_trusted_internal_mcp_runtime_request(request: Request) -> bool:
         ``True`` when the request carries a trusted internal-runtime marker
         from loopback, otherwise ``False``.
     """
-    runtime_marker = request.headers.get("x-contextforge-mcp-runtime")
-    client_host = getattr(getattr(request, "client", None), "host", None)
-    if runtime_marker not in ("rust", "affinity") or not has_valid_internal_mcp_runtime_auth_header(request) or client_host not in ("127.0.0.1", "::1"):
-        return False
-    # Defense-in-depth: /_internal/a2a/* endpoints must refuse requests when
-    # A2A support is disabled, even from an otherwise-trusted local sidecar.
-    # A legitimate sidecar should not be running when the feature is off.
-    path = getattr(getattr(request, "url", None), "path", "") or ""
-    if path.startswith("/_internal/a2a/") and not settings.mcpgateway_a2a_enabled:
-        return False
-    return True
+    return is_trusted_internal_mcp_request(request)
 
 
 def _is_jwt_token(token: str) -> bool:

mcpgateway/middleware/rate_limit_middleware.py

@@ -38,6 +38,7 @@
 
 # First-Party
 from mcpgateway import auth
+from mcpgateway.auth_context import is_trusted_internal_mcp_request
 from mcpgateway.config import settings
 from mcpgateway.services.security_logger import SecurityEventType, SecurityLogger, SecuritySeverity
 
@@ -199,6 +200,10 @@ async def dispatch(self, request: Request, call_next):
         if not self.enabled:
             return await call_next(request)
 
+        # Skip rate limiting for the trusted-internal dispatch; the edge request was already counted.
+        if is_trusted_internal_mcp_request(request):
+            return await call_next(request)
+
         tier = self.get_endpoint_tier(request.url.path)
         dimensions = self._get_client_dimensions(request)
 

mcpgateway/middleware/token_scoping.py

@@ -13,8 +13,6 @@
 # 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
@@ -64,11 +62,6 @@
     (re.compile(r"/gateways/?([a-f0-9\-]+)"), "gateway"),
 ]
 _AUTH_COOKIE_NAMES = ("jwt_token", "access_token")
-_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
@@ -1350,70 +1343,24 @@ async def __call__(self, request: Request, call_next):
             )
 
     def _is_trusted_internal_mcp_runtime_request(self, request: Request, normalized_path: str) -> bool:
-        """Return whether the request is a trusted loopback Rust MCP sidecar hop.
+        """Return whether the request is a trusted internal MCP/A2A runtime hop.
+
+        Delegates to the shared gate in ``auth_context`` so the trust decision is
+        defined in one place. Unlike the previous local check, this trusts both
+        the ``rust`` and ``affinity`` runtime markers — closing the gap where the
+        in-process session-affinity dispatch was still token-scoped.
 
         Args:
             request: Incoming HTTP request.
             normalized_path: Canonicalized request path used for route matching.
 
         Returns:
-            ``True`` when the request originated from the local Rust MCP runtime and
-            includes the expected trusted headers.
-        """
-        if normalized_path != _INTERNAL_MCP_PATH_PREFIX and not normalized_path.startswith(f"{_INTERNAL_MCP_PATH_PREFIX}/"):
-            return False
-
-        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
-    def _auth_encryption_secret_value() -> str:
-        """Return the configured auth-encryption secret as a plain string.
-
-        Returns:
-            The auth-encryption secret, normalized to a regular string.
-        """
-        secret = settings.auth_encryption_secret
-        if hasattr(secret, "get_secret_value"):
-            return secret.get_secret_value()
-        return str(secret)
-
-    @staticmethod
-    @lru_cache(maxsize=8)
-    def _expected_internal_mcp_runtime_auth_header_for_secret(secret: str) -> str:
-        """Return the expected shared internal-auth header for a specific secret.
-
-        Args:
-            secret: Auth-encryption secret to derive the trust header from.
-
-        Returns:
-            Hex-encoded SHA-256 digest derived from the provided auth secret.
+            ``True`` when the request is a trusted internal MCP/A2A hop.
         """
-        material = f"{secret}:{_INTERNAL_MCP_RUNTIME_AUTH_CONTEXT}".encode("utf-8")
-        return hashlib.sha256(material).hexdigest()
-
-    @staticmethod
-    def _expected_internal_mcp_runtime_auth_header() -> str:
-        """Return the expected shared internal-auth header for Rust MCP hops.
+        # First-Party
+        from mcpgateway.auth_context import is_trusted_internal_mcp_request  # pylint: disable=import-outside-toplevel
 
-        Returns:
-            Shared secret-derived digest expected on trusted internal Rust MCP calls.
-        """
-        return TokenScopingMiddleware._expected_internal_mcp_runtime_auth_header_for_secret(TokenScopingMiddleware._auth_encryption_secret_value())
+        return is_trusted_internal_mcp_request(request, path=normalized_path)
 
 
 # Create middleware instance

mcpgateway/middleware/token_usage_middleware.py

@@ -30,6 +30,7 @@
 from starlette.types import ASGIApp, Receive, Scope, Send
 
 # First-Party
+from mcpgateway.auth_context import is_trusted_internal_mcp_request
 from mcpgateway.db import fresh_db_session
 from mcpgateway.middleware.path_filter import should_skip_auth_context
 from mcpgateway.services.token_catalog_service import TokenCatalogService
@@ -87,6 +88,11 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
             await self.app(scope, receive, send)
             return
 
+        # Skip usage logging for the trusted-internal dispatch; the edge request is already recorded.
+        if is_trusted_internal_mcp_request(Request(scope), path=path):
+            await self.app(scope, receive, send)
+            return
+
         # Record start time
         start_time = time.time()
 

mcpgateway/utils/passthrough_headers.py

@@ -573,6 +573,17 @@ async def set_global_passthrough_headers(db: Session) -> None:
         "x-contextforge-mcp-runtime",
         "x-contextforge-mcp-runtime-auth",
         "x-contextforge-session-validated",
+        # Forwarded / client-IP headers: strip so a caller cannot spoof the loopback
+        # client address that ProxyHeaders(trusted_hosts="*") would otherwise trust.
+        "forwarded",
+        "x-forwarded-for",
+        "x-forwarded-host",
+        "x-forwarded-proto",
+        "x-forwarded-port",
+        "x-forwarded-prefix",
+        "x-real-ip",
+        "cf-connecting-ip",
+        "true-client-ip",
     }
 )
 

tests/unit/mcpgateway/middleware/test_rate_limit_middleware.py

@@ -16,6 +16,30 @@
 import pytest
 
 
+def _trusted_internal_request(path, *, marker="rust", hmac_value=None, ctx="ctx", client="127.0.0.1"):
+    """Build a real loopback internal request for trust-gate exemption tests."""
+    from starlette.requests import Request
+
+    from mcpgateway.auth_context import _expected_internal_mcp_runtime_auth_header
+
+    headers = [
+        (b"x-contextforge-mcp-runtime", marker.encode()),
+        (b"x-contextforge-mcp-runtime-auth", (hmac_value or _expected_internal_mcp_runtime_auth_header()).encode()),
+    ]
+    if ctx is not None:
+        headers.append((b"x-contextforge-auth-context", ctx.encode()))
+    scope = {
+        "type": "http",
+        "method": "POST",
+        "path": path,
+        "raw_path": path.encode(),
+        "query_string": b"",
+        "headers": headers,
+        "client": (client, 12345),
+    }
+    return Request(scope)
+
+
 class TestRateLimitMiddleware:
     """Tests for RateLimitMiddleware."""
 
@@ -58,6 +82,29 @@ def mock_request(self):
         request.scope = {"client": ("192.168.1.100", 12345)}
         return request
 
+    @pytest.mark.asyncio
+    async def test_trusted_internal_dispatch_skips_rate_limiting(self, middleware):
+        """A trusted-internal hop is not rate-limited; the edge request already was."""
+        request = _trusted_internal_request("/_internal/mcp/rpc")
+        call_next = AsyncMock(return_value="passthrough")
+        # If the exemption fires, tier computation is never reached.
+        middleware.get_endpoint_tier = MagicMock(side_effect=AssertionError("trusted internal must not be rate-limited"))
+
+        result = await middleware.dispatch(request, call_next)
+
+        assert result == "passthrough"
+        call_next.assert_awaited_once_with(request)
+
+    @pytest.mark.asyncio
+    async def test_forged_internal_request_is_still_rate_limited(self, middleware):
+        """A forged HMAC fails the trust gate, so the rate-limit path runs."""
+        request = _trusted_internal_request("/_internal/mcp/rpc", hmac_value="forged")
+        call_next = AsyncMock(return_value="passthrough")
+        middleware.get_endpoint_tier = MagicMock(side_effect=RuntimeError("reached rate-limit path"))
+
+        with pytest.raises(RuntimeError, match="reached rate-limit path"):
+            await middleware.dispatch(request, call_next)
+
     def test_endpoint_tier_critical(self, middleware):
         """Test CRITICAL tier detection for auth endpoints."""
         tier = middleware.get_endpoint_tier("/auth/email/login")