Deploying remote Model Context Protocol (MCP) servers over Server-Sent Events (SSE) opens two severe attack vectors if endpoints run without network isolation. An attacker website can execute DNS rebinding attacks to bypass browser Same-Origin Policy (SOP) and call local MCP tools directly. Simultaneously, an agent tricked by indirect prompt injection can use network-enabled MCP tools to perform Server-Side Request Forgery (SSRF) against internal VPC microservices or cloud metadata endpoints (169.254.169.254).

Here is how to enforce strict Host header filtering, validate IP resolution before socket creation, and lock down Linux egress rules with nftables.

---

1. Quick Solution: Starlette Host Header Guard

Block DNS rebinding attacks by validating the incoming HTTP Host header against an explicit allowlist in your FastMCP or Starlette ASGI stack:

# Save as host_guard.py
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response

ALLOWED_HOSTS = {"127.0.0.1:8000", "localhost:8000", "mcp.internal.company.com"}

class HostHeaderValidationMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        host = request.headers.get("host", "")
        if host not in ALLOWED_HOSTS:
            return Response(
                content="421 Misdirected Request: Invalid Host Header",
                status_code=421,
                media_type="text/plain",
            )
        return await call_next(request)

In a DNS rebinding exploit, an external domain points initially to an attacker IP, then switches DNS records to 127.0.0.1. The victim browser sends requests with the attacker domain in the Host header (e.g. Host: attacker-domain.xyz). Returning HTTP 421 immediately drops the connection before the MCP JSON-RPC router parses the tool payload.

---

2. Attack Surface Breakdown: Rebinding vs SSRF

Remote MCP servers bridging desktop coding agents (Cursor, Claude Code) and cloud environments face distinct inbound and outbound threats:

Attack Vector	Entry Point	Target Asset	Impact	Primary Mitigation
DNS Rebinding	Inbound HTTP/SSE	Local MCP JSON-RPC daemon	Arbitrary local tool execution	Strict Host header allowlisting
Cloud Metadata SSRF	Outbound Tool Call	169.254.169.254 (IMDSv1/v2)	IAM credential theft	Pre-dial DNS resolution & RFC 3927 drops
VPC Lateral Movement	Outbound Tool Call	RFC 1918 subnets (10.0.0.0/8)	Internal database/cache exfiltration	Socket-level IP validation & nftables
Origin Impersonation	Inbound CORS	/sse and /messages endpoints	Cross-site event stream interception	Restrict Access-Control-Allow-Origin

---

3. Preventing SSRF in Fetch and Web-Scraping Tools

Standard URL validation functions that rely solely on string checks (url.startswith("http")) fail against DNS resolution tricks, zero-padded IPs, and decimal encoding (http://2852039166 resolves to 169.254.169.254).

Implement an IP validation resolver that inspects resolved socket addresses before dispatching HTTP requests:

import ipaddress
import socket
from urllib.parse import urlparse
import httpx

BLOCKED_NETWORKS = [
    ipaddress.ip_network("127.0.0.0/8"),      # Loopback
    ipaddress.ip_network("10.0.0.0/8"),       # RFC 1918 Private
    ipaddress.ip_network("172.16.0.0/12"),    # RFC 1918 Private
    ipaddress.ip_network("192.168.0.0/16"),   # RFC 1918 Private
    ipaddress.ip_network("169.254.0.0/16"),   # Link-Local / Cloud Metadata
    ipaddress.ip_network("::1/128"),          # IPv6 Loopback
    ipaddress.ip_network("fe80::/10"),        # IPv6 Link-Local
]

def validate_safe_url(target_url: str) -> str:
    parsed = urlparse(target_url)
    if parsed.scheme not in ("http", "https"):
        raise ValueError(f"Unsupported protocol: {parsed.scheme}")

    hostname = parsed.hostname
    if not hostname:
        raise ValueError("Missing hostname")

    # Resolve all associated IP addresses
    addr_info = socket.getaddrinfo(hostname, None, socket.AF_UNSPEC, socket.SOCK_STREAM)
    for family, _, _, _, sockaddr in addr_info:
        ip = ipaddress.ip_address(sockaddr[0])
        for network in BLOCKED_NETWORKS:
            if ip in network:
                raise PermissionError(f"Access denied: {hostname} resolves to restricted IP {ip}")

    return target_url

# Example MCP Tool implementation
async def safe_fetch_html(target_url: str) -> str:
    clean_url = validate_safe_url(target_url)
    async with httpx.AsyncClient(timeout=10.0, follow_redirects=False) as client:
        res = await client.get(clean_url)
        return res.text

Setting follow_redirects=False is mandatory. Attackers frequently bypass initial checks by redirecting an allowed public URL (such as http://example.com/redirect) to http://169.254.169.254/latest/meta-data/.

---

4. Kernel-Level Egress Filtering with Linux nftables

Application-level checks can be circumvented if sub-dependencies spawn child processes (curl, wget, git clone). Apply kernel-level egress packet filtering on Linux using nftables:

sudo nft -f - << EOF
table inet mcp_security {
    chain egress_filter {
        type filter hook output priority 0; policy accept;

        # Drop outbound packets to AWS / GCP / Azure metadata services
        ip daddr 169.254.169.254 counter drop
        ip daddr 169.254.0.0/16 counter drop

        # Block direct access to internal corporate subnets from MCP worker UID (1001)
        meta skuid 1001 ip daddr { 10.0.0.0/8, 172.16.0.0/12, 192.168.0.0/16 } counter drop

        # Allow established and related traffic
        ct state established,related accept
    }
}
EOF

Verify that the table is active and packet counters increment on dropped probes:

sudo nft list table inet mcp_security

---

5. FastMCP Server Hardening Template

Here is the complete ASGI entrypoint combining Host header validation, strict CORS origin filtering, and token authentication:

import os
from fastmcp import FastMCP
from starlette.middleware.cors import CORSMiddleware
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response

mcp = FastMCP("hardened-remote-service")

EXPECTED_TOKEN = os.environ.get("MCP_AUTH_TOKEN", "prod_token_secret_9941")
ALLOWED_ORIGIN = "https://agentic-pulse.dev"

class SecurityGateMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # 1. Validate Host Header
        host = request.headers.get("host", "")
        if host not in ("127.0.0.1:8000", "localhost:8000"):
            return Response("421 Misdirected Request", status_code=421)

        # 2. Validate Bearer Token on incoming JSON-RPC calls
        if request.url.path in ("/messages", "/tools"):
            auth = request.headers.get("authorization", "")
            if auth != f"Bearer {EXPECTED_TOKEN}":
                return Response("401 Unauthorized", status_code=401)

        return await call_next(request)

app = mcp.get_asgi_app()
app.add_middleware(SecurityGateMiddleware)
app.add_middleware(
    CORSMiddleware,
    allow_origins=[ALLOWED_ORIGIN],
    allow_methods=["GET", "POST", "OPTIONS"],
    allow_headers=["Authorization", "Content-Type"],
)

Run this service behind an unprivileged system user (UID 1001) with read-only file root access to ensure full defense-in-depth across production agent swarms.