fix: harden Rust runtime fail-closed handling

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

Author: crivetimihai

mcpgateway/transports/rust_mcp_runtime_proxy.py

@@ -96,6 +96,7 @@ async def handle_streamable_http(self, scope: Scope, receive: Receive, send: Sen
                 content=_stream_request_body(receive) if method == "POST" else b"",
                 headers=headers,
                 timeout=timeout,
+                follow_redirects=False,
             ) as response:
                 await send(
                     {
@@ -144,7 +145,7 @@ async def _get_runtime_client(self) -> httpx.AsyncClient:
                     transport=httpx.AsyncHTTPTransport(uds=uds_path),
                     limits=get_http_limits(),
                     timeout=httpx.Timeout(settings.experimental_rust_mcp_runtime_timeout_seconds),
-                    follow_redirects=True,
+                    follow_redirects=False,
                 )
             return self._uds_client
 

mcpgateway/transports/streamablehttp_transport.py

@@ -455,6 +455,7 @@ async def store_event(self, stream_id: StreamId, message: JSONRPCMessage | None)
                 "ttlSeconds": self.ttl,
             },
             timeout=httpx.Timeout(settings.experimental_rust_mcp_runtime_timeout_seconds),
+            follow_redirects=False,
         )
         response.raise_for_status()
         payload = response.json()
@@ -481,6 +482,7 @@ async def replay_events_after(self, last_event_id: EventId, send_callback: Event
                 "keyPrefix": self.key_prefix,
             },
             timeout=httpx.Timeout(settings.experimental_rust_mcp_runtime_timeout_seconds),
+            follow_redirects=False,
         )
         response.raise_for_status()
         payload = response.json()
@@ -515,7 +517,7 @@ async def _get_rust_event_store_client() -> httpx.AsyncClient:
                 transport=httpx.AsyncHTTPTransport(uds=uds_path),
                 limits=get_http_limits(),
                 timeout=httpx.Timeout(settings.experimental_rust_mcp_runtime_timeout_seconds),
-                follow_redirects=True,
+                follow_redirects=False,
             )
         return _rust_event_store_client
 

tests/unit/mcpgateway/transports/test_rust_mcp_runtime_proxy.py

@@ -66,11 +66,12 @@ async def __aexit__(self, exc_type, exc, tb):
             return False
 
     class FakeClient:
-        def stream(self, method, url, *, content, headers, timeout):  # noqa: ANN001
+        def stream(self, method, url, *, content, headers, timeout, follow_redirects):  # noqa: ANN001
             captured["method"] = method
             captured["url"] = url
             captured["headers"] = headers
             captured["timeout"] = timeout
+            captured["follow_redirects"] = follow_redirects
             return FakeStreamContext(content=content)
 
     monkeypatch.setattr("mcpgateway.transports.rust_mcp_runtime_proxy.settings.experimental_rust_mcp_runtime_url", "http://127.0.0.1:8787")
@@ -125,6 +126,7 @@ async def send(message):
     assert captured["method"] == "POST"
     assert captured["url"] == "http://127.0.0.1:8787/mcp/?session_id=abc123"
     assert captured["timeout"].connect == 17
+    assert captured["follow_redirects"] is False
 
     forwarded_headers = dict(captured["headers"])
     assert forwarded_headers["authorization"] == "Bearer test-token"
@@ -188,11 +190,12 @@ async def __aexit__(self, exc_type, exc, tb):
             return False
 
     class FakeClient:
-        def stream(self, method, url, *, content, headers, timeout):  # noqa: ANN001
+        def stream(self, method, url, *, content, headers, timeout, follow_redirects):  # noqa: ANN001
             captured["method"] = method
             captured["url"] = url
             captured["headers"] = headers
             captured["timeout"] = timeout
+            captured["follow_redirects"] = follow_redirects
             return FakeStreamContext(content=content)
 
     async def receive():
@@ -233,6 +236,7 @@ async def send(message):
     assert captured["method"] == "POST"
     assert captured["url"] == "http://127.0.0.1:8787/mcp/"
     assert captured["content"] == b'{"jsonrpc":"2.0","id":1,"method":"ping","params":{}}'
+    assert captured["follow_redirects"] is False
     assert events[-1] == {"type": "http.response.body", "body": b"", "more_body": False}
 
 
@@ -257,11 +261,12 @@ async def __aexit__(self, exc_type, exc, tb):
             return False
 
     class FakeClient:
-        def stream(self, method, url, *, content, headers, timeout):  # noqa: ANN001
+        def stream(self, method, url, *, content, headers, timeout, follow_redirects):  # noqa: ANN001
             captured["method"] = method
             captured["url"] = url
             captured["headers"] = dict(headers)
             captured["timeout"] = timeout
+            captured["follow_redirects"] = follow_redirects
             return FakeStreamContext()
 
     monkeypatch.setattr("mcpgateway.transports.rust_mcp_runtime_proxy.settings.experimental_rust_mcp_runtime_url", "http://127.0.0.1:8787")
@@ -298,6 +303,7 @@ async def send(message):
 
     assert captured["method"] == "GET"
     assert captured["url"] == "http://127.0.0.1:8787/mcp/?session_id=session-1"
+    assert captured["follow_redirects"] is False
     assert captured["headers"]["x-contextforge-affinity-forwarded"] == "rust"
     assert "x-forwarded-internally" not in captured["headers"]
     assert "x-original-worker" not in captured["headers"]
@@ -332,12 +338,13 @@ async def __aexit__(self, exc_type, exc, tb):
             return False
 
     class FakeClient:
-        def stream(self, method, url, *, content, headers, timeout):  # noqa: ANN001
+        def stream(self, method, url, *, content, headers, timeout, follow_redirects):  # noqa: ANN001
             captured["method"] = method
             captured["url"] = url
             captured["content"] = content
             captured["headers"] = headers
             captured["timeout"] = timeout
+            captured["follow_redirects"] = follow_redirects
             return FakeStreamContext()
 
     monkeypatch.setattr("mcpgateway.transports.rust_mcp_runtime_proxy.settings.experimental_rust_mcp_runtime_url", "http://127.0.0.1:8787")
@@ -371,6 +378,7 @@ async def send(message):
     assert captured["method"] == "GET"
     assert captured["url"] == "http://127.0.0.1:8787/mcp/?session_id=abc123"
     assert captured["content"] == b""
+    assert captured["follow_redirects"] is False
     assert dict(captured["headers"])["x-contextforge-server-id"] == "123e4567-e89b-12d3-a456-426614174000"
     assert events[0]["status"] == 200
     assert (b"content-type", b"text/event-stream") in events[0]["headers"]
@@ -402,11 +410,12 @@ class FakeAsyncClient:
         def __init__(self, **kwargs):
             constructed["kwargs"] = kwargs
 
-        def stream(self, method, url, *, content, headers, timeout):  # noqa: ANN001
+        def stream(self, method, url, *, content, headers, timeout, follow_redirects):  # noqa: ANN001
             constructed["method"] = method
             constructed["url"] = url
             constructed["headers"] = headers
             constructed["timeout"] = timeout
+            constructed["follow_redirects"] = follow_redirects
             return FakeStreamContext()
 
     get_http_client_mock = AsyncMock()
@@ -441,6 +450,57 @@ async def send(message):
     assert constructed["method"] == "POST"
     assert constructed["url"] == "http://localhost/mcp/"
     assert constructed["kwargs"]["transport"]._pool._uds == "/tmp/contextforge-mcp-rust.sock"  # pylint: disable=protected-access
+    assert constructed["kwargs"]["follow_redirects"] is False
+    assert constructed["follow_redirects"] is False
+    assert events[-1] == {"type": "http.response.body", "body": b"", "more_body": False}
+
+
+@pytest.mark.asyncio
+async def test_runtime_proxy_surfaces_redirect_without_following(monkeypatch):
+    """Internal runtime redirects should be surfaced directly and never auto-followed."""
+    requests_seen = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        requests_seen.append(str(request.url))
+        if request.url.path == "/mcp/":
+            return httpx.Response(
+                307,
+                headers={"location": "http://127.0.0.1:8787/final"},
+                request=request,
+            )
+        return httpx.Response(200, json={"jsonrpc": "2.0", "id": 1, "result": {"unexpected": True}}, request=request)
+
+    client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
+    monkeypatch.setattr("mcpgateway.transports.rust_mcp_runtime_proxy.settings.experimental_rust_mcp_runtime_url", "http://127.0.0.1:8787")
+    monkeypatch.setattr("mcpgateway.transports.rust_mcp_runtime_proxy.get_http_client", AsyncMock(return_value=client))
+
+    fallback = AsyncMock()
+    proxy = RustMCPRuntimeProxy(fallback)
+    events = []
+
+    async def send(message):
+        events.append(message)
+
+    try:
+        await proxy.handle_streamable_http(
+            {
+                "type": "http",
+                "method": "GET",
+                "path": "/",
+                "modified_path": "/mcp",
+                "query_string": b"session_id=abc123",
+                "headers": [],
+            },
+            _make_receive(b""),
+            send,
+        )
+    finally:
+        await client.aclose()
+
+    fallback.assert_not_awaited()
+    assert requests_seen == ["http://127.0.0.1:8787/mcp/?session_id=abc123"]
+    assert events[0]["status"] == 307
+    assert (b"location", b"http://127.0.0.1:8787/final") in events[0]["headers"]
     assert events[-1] == {"type": "http.response.body", "body": b"", "more_body": False}
 
 

tests/unit/mcpgateway/transports/test_streamablehttp_transport.py

@@ -168,8 +168,8 @@ def json(self):
             return self._payload
 
     class FakeClient:
-        async def post(self, url, json=None, timeout=None):  # noqa: A002
-            captured_requests.append((url, json, timeout.read))
+        async def post(self, url, json=None, timeout=None, follow_redirects=None):  # noqa: A002
+            captured_requests.append((url, json, timeout.read, follow_redirects))
             if url.endswith("/store"):
                 return FakeResponse({"eventId": "event-123"})
             return FakeResponse(
@@ -208,11 +208,43 @@ async def collector(msg):
         "ttlSeconds": 321,
     }
     assert captured_requests[0][2] == 17
+    assert captured_requests[0][3] is False
     assert captured_requests[1][0] == "http://127.0.0.1:8787/_internal/event-store/replay"
     assert captured_requests[1][1] == {
         "lastEventId": "event-123",
         "keyPrefix": "mcpgw:eventstore:test",
     }
+    assert captured_requests[1][3] is False
+
+
+@pytest.mark.asyncio
+async def test_rust_event_store_replay_rejects_redirects_without_following(monkeypatch):
+    """Replay requests should fail closed on redirects from the Rust sidecar."""
+    requests_seen = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        requests_seen.append(str(request.url))
+        if request.url.path.endswith("/replay"):
+            return httpx.Response(
+                307,
+                headers={"location": "http://127.0.0.1:8787/final"},
+                request=request,
+            )
+        return httpx.Response(200, json={"streamId": "unexpected", "events": []}, request=request)
+
+    client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
+    monkeypatch.setattr(tr, "_get_rust_event_store_client", AsyncMock(return_value=client))
+    monkeypatch.setattr(tr.settings, "experimental_rust_mcp_runtime_url", "http://127.0.0.1:8787")
+
+    store = tr.RustEventStore()
+
+    try:
+        with pytest.raises(httpx.HTTPStatusError, match="307 Temporary Redirect"):
+            await store.replay_events_after("event-123", AsyncMock())
+    finally:
+        await client.aclose()
+
+    assert requests_seen == ["http://127.0.0.1:8787/_internal/event-store/replay"]
 
 
 @pytest.mark.asyncio
@@ -290,11 +322,12 @@ async def test_get_rust_event_store_client_uses_shared_http_client_without_uds(m
 @pytest.mark.asyncio
 async def test_get_rust_event_store_client_reuses_uds_client(monkeypatch):
     """UDS-backed Rust event-store client should be created once and then reused."""
-    constructed = {"count": 0}
+    constructed = {"count": 0, "kwargs": None}
 
     class FakeAsyncClient:
         def __init__(self, **_kwargs):
             constructed["count"] += 1
+            constructed["kwargs"] = _kwargs
 
     monkeypatch.setattr(tr, "_rust_event_store_client", None)
     monkeypatch.setattr(tr.settings, "experimental_rust_mcp_runtime_uds", "/tmp/contextforge-mcp-rust.sock")
@@ -305,6 +338,7 @@ def __init__(self, **_kwargs):
 
     assert first is second
     assert constructed["count"] == 1
+    assert constructed["kwargs"]["follow_redirects"] is False
 
 
 def test_get_streamable_http_auth_context_returns_empty_for_non_dict_context():

tools_rust/mcp_runtime/FOLLOWUPS.md

@@ -528,7 +528,36 @@ Recommended next step:
 - Add explicit shutdown cleanup on the Rust side and a `close()`/shutdown hook
   for the Python proxy's cached UDS client in a separate follow-up.
 
-#### 19. Session-auth reuse still trades freshness for fewer auth round-trips
+#### 19. Redis hot-path round-trip and cache single-flight polish
+
+Status:
+- Deferred performance polish
+
+Observed behavior:
+- Runtime-session refresh in Redis still uses `GET` followed by `EXPIRE`
+  instead of a single `GETEX`-style refresh.
+- Event-store replay still fetches replay payloads one entry at a time from
+  the Redis hash instead of batching those lookups.
+- A few in-process caches still use simple double-checked locking rather than
+  a stronger single-flight pattern, so duplicate initialization work is still
+  possible under contention.
+
+Why this matters:
+- These are performance and efficiency opportunities, not current correctness
+  regressions.
+- They are most visible under heavy load or when many workers race to populate
+  the same hot cache entries.
+
+Likely area:
+- `tools_rust/mcp_runtime/src/lib.rs`
+
+Recommended next step:
+- Revisit the Redis/runtime hot paths in a focused performance pass and assess:
+  - `GETEX` or equivalent atomic session-touch semantics
+  - `HMGET`/pipeline replay fetches for event batches
+  - `OnceCell` or another single-flight pattern for expensive cache fills
+
+#### 20. Session-auth reuse still trades freshness for fewer auth round-trips
 
 Status:
 - Deferred Rust-specific design follow-up
@@ -552,7 +581,7 @@ Recommended next step:
   should consume a revocation/invalidation signal from Python to drop cached
   auth state immediately.
 
-#### 20. Legacy migration suites are still red
+#### 21. Legacy migration suites are still red
 
 Status:
 - Deferred broader release/upgrade follow-up
@@ -581,7 +610,7 @@ Recommended next step:
   fix the legacy migration/data-retention issues independently of the Rust MCP
   transport PR.
 
-#### 21. Live PostgreSQL TLS validation is still unexecuted
+#### 22. Live PostgreSQL TLS validation is still unexecuted
 
 Status:
 - Deferred release-validation follow-up
@@ -609,7 +638,7 @@ Recommended next step:
   - Rust with `sslrootcert=/path/to/ca.pem`
   - explicit failure for unsupported `sslcert` / `sslkey`
 
-#### 22. Minikube clean reinstall flow still looks unhealthy
+#### 23. Minikube clean reinstall flow still looks unhealthy
 
 Status:
 - Deferred Helm/deployment follow-up
@@ -635,7 +664,7 @@ Recommended next step:
   issue is in Helm invocation, namespace lifecycle timing, or local Minikube
   state.
 
-#### 23. Optional `2025-11-25-report` surface is not release-clean
+#### 24. Optional `2025-11-25-report` surface is not release-clean
 
 Status:
 - Deferred protocol-surface follow-up

tools_rust/mcp_runtime/src/lib.rs

@@ -3446,8 +3446,19 @@ async fn replay_events_from_rust_event_store(
             continue;
         };
 
-        let message = serde_json::from_str::<Value>(&message_json).unwrap_or(Value::Null);
-        events.push(EventStoreReplayEvent { event_id, message });
+        match serde_json::from_str::<Value>(&message_json) {
+            Ok(message) => events.push(EventStoreReplayEvent { event_id, message }),
+            Err(err) => {
+                error!(
+                    "Rust event store replay decode failed for stream {} event {}: {err}",
+                    index_record.stream_id, event_id
+                );
+                return Err(json_response(
+                    StatusCode::BAD_GATEWAY,
+                    json!({"detail": "Rust event store replay decode failed"}),
+                ));
+            }
+        }
     }
 
     Ok(EventStoreReplayResponse {