Compare commits

...

12 commits

Author SHA1 Message Date
Jeremiah Lowin
c587bdd854
Docs: add v3.4.6 changelog entries (#4761) 2026-08-05 10:43:23 -04:00
Jeremiah Lowin
fcdf422e54
Backport #4412 to 3.x: support trusted SSRF proxies (#4755) 2026-08-05 10:29:38 -04:00
Jeremiah Lowin
8de0c94cbb
Docs: add v3.4.5 changelog entries (#4673)
* Docs: add v3.4.5 changelog entries

* Condense the 3.4.5 entries to patch-release length
2026-07-27 15:12:57 -04:00
Jeremiah Lowin
5daa91bafe
Backport #4492 to release/3.x: don't mutate the caller's schema in compress_schema (#4663)
* Don't mutate the caller's schema in compress_schema (#4492)

(cherry picked from commit 6202008cf3)

* Copy schemas iteratively so deep nesting still compresses

(cherry picked from commit 2662e675ca)

* Keep $defs when the reference scan hits its depth limit

(cherry picked from commit 054a6cf3c2)

---------

Co-authored-by: WinkleMad <winklemad@outlook.com>
2026-07-27 15:02:41 -04:00
Jeremiah Lowin
49c5ed5581
Make transformed tool required order deterministic (#4564) (#4665)
(cherry picked from commit 7934124fb5)

Co-authored-by: Marcelo Trylesinski <marcelotryle@gmail.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
2026-07-27 13:11:04 -04:00
Jeremiah Lowin
87ba18c898
Serialize deep object query parameters (#4523) (#4664)
(cherry picked from commit b623c23183)
2026-07-27 13:10:57 -04:00
Jeremiah Lowin
aa8ab1ee66
Fix Azure scope fallback (#4469) (#4662)
(cherry picked from commit 1d932cc778)

Co-authored-by: nate nowack <thrast36@gmail.com>
2026-07-27 13:10:27 -04:00
Kaki Wang
93696a59f4
Skip unsupported JWKS keys instead of failing the whole key set (#4515) (#4517) (#4631)
(cherry picked from commit 1fca15abe6)

Co-authored-by: Burt Matthews <80060660+earfman@users.noreply.github.com>
2026-07-26 14:02:20 -04:00
Jeremiah Lowin
9138d40e88
Docs: add v3.4.4 changelog entries (#4473) 2026-07-08 20:30:40 -04:00
shaun smith
d929882f77
Hugging Face Auth Integration (#4385)
Co-authored-by: Jeremiah Lowin <153965+jlowin@users.noreply.github.com>
2026-07-08 20:27:44 -04:00
Jeremiah Lowin
5fe4fae535
Restore HTTP host guard compatibility (#4472) 2026-07-08 20:16:44 -04:00
Jeremiah Lowin
400db61b8b
Relax host origin guard defaults (#4439) 2026-07-06 20:59:28 -04:00
30 changed files with 2241 additions and 67 deletions

View file

@ -5,6 +5,61 @@ rss: true
tag: NEW
---
<Update label="v3.4.6" description="2026-08-05">
**[v3.4.6: Trust, but Proxy](https://github.com/PrefectHQ/fastmcp/releases/tag/v3.4.6)**
FastMCP 3.4.6 backports trusted-proxy support for SSRF-protected OAuth metadata and JWKS fetches. Deployments can now route these requests through a mandated corporate proxy while preserving custom CA certificates; FastMCP refuses the fetch when no proxy is configured instead of risking an unprotected direct request.
### Fixes 🐞
* Backport #4412 to 3.x: support trusted SSRF proxies by [@jlowin](https://github.com/jlowin) in [#4755](https://github.com/PrefectHQ/fastmcp/pull/4755)
### Docs 📚
* Docs: add v3.4.6 changelog entries by [@jlowin](https://github.com/jlowin) in [#4761](https://github.com/PrefectHQ/fastmcp/pull/4761)
**Full Changelog**: [v3.4.5...v3.4.6](https://github.com/PrefectHQ/fastmcp/compare/v3.4.5...v3.4.6)
</Update>
<Update label="v3.4.5" description="2026-07-27">
**[v3.4.5: Key Change](https://github.com/PrefectHQ/fastmcp/releases/tag/v3.4.5)**
FastMCP 3.4.5 collects five fixes for the 3.x line, led by `JWTVerifier` no longer rejecting every token when an authorization server publishes an unrecognized key type such as Ed25519.
### Fixes 🐞
* Backport #4517 to release/3.x: skip unsupported JWKS keys (#4515) by [@kakiii](https://github.com/kakiii) in [#4631](https://github.com/PrefectHQ/fastmcp/pull/4631)
* Backport #4469 to release/3.x: fix Azure scope fallback by [@jlowin](https://github.com/jlowin) in [#4662](https://github.com/PrefectHQ/fastmcp/pull/4662)
* Backport #4523 to release/3.x: serialize deep object query parameters by [@jlowin](https://github.com/jlowin) in [#4664](https://github.com/PrefectHQ/fastmcp/pull/4664)
* Backport #4564 to release/3.x: make transformed tool required order deterministic by [@jlowin](https://github.com/jlowin) in [#4665](https://github.com/PrefectHQ/fastmcp/pull/4665)
* Backport #4492 to release/3.x: don't mutate the caller's schema in compress_schema by [@jlowin](https://github.com/jlowin) in [#4663](https://github.com/PrefectHQ/fastmcp/pull/4663)
## New Contributors
* @kakiii made their first contribution in [#4631](https://github.com/PrefectHQ/fastmcp/pull/4631)
**Full Changelog**: [v3.4.4...v3.4.5](https://github.com/PrefectHQ/fastmcp/compare/v3.4.4...v3.4.5)
</Update>
<Update label="v3.4.4" description="2026-07-08">
**[v3.4.4: Host in Translation](https://github.com/PrefectHQ/fastmcp/releases/tag/v3.4.4)**
FastMCP 3.4.4 restores HTTP deployment compatibility after the 3.4.3 Host/Origin guard changed default behavior for existing ASGI, serverless, and reverse-proxy deployments. The guard implementation remains available for deployments that opt in with explicit trusted hosts and origins, while 3.x returns to accepting traffic that worked before the patch. This release also adds Hugging Face OAuth provider support, with docs and examples for public and private apps, PKCE, Dynamic Client Registration, and CIMD.
### Enhancements ✨
* Hugging Face Auth Integration by [@evalstate](https://github.com/evalstate) in [#4385](https://github.com/PrefectHQ/fastmcp/pull/4385)
### Fixes 🐞
* Relax host origin guard defaults by [@jlowin](https://github.com/jlowin) in [#4439](https://github.com/PrefectHQ/fastmcp/pull/4439)
* Restore HTTP host guard compatibility by [@jlowin](https://github.com/jlowin) in [#4472](https://github.com/PrefectHQ/fastmcp/pull/4472)
## New Contributors
* @evalstate made their first contribution in [#4385](https://github.com/PrefectHQ/fastmcp/pull/4385)
**Full Changelog**: [v3.4.3...v3.4.4](https://github.com/PrefectHQ/fastmcp/compare/v3.4.3...v3.4.4)
</Update>
<Update label="v3.4.3" description="2026-07-05">
**[v3.4.3: The Fast and the Secure-ious](https://github.com/PrefectHQ/fastmcp/releases/tag/v3.4.3)**
@ -3737,4 +3792,4 @@ This release is highlighted by the ability to handle complex JSON objects as MCP
The very first release of FastMCP! 🎉
**Full Changelog**: [Initial commits](https://github.com/PrefectHQ/fastmcp/commits/v0.1.0)
</Update>
</Update>

View file

@ -103,11 +103,11 @@ If you're mounting an authenticated server under a path prefix, see [Mounting Au
### Host and Origin Protection
FastMCP validates `Host` and browser `Origin` headers for Streamable HTTP requests by default. This protects localhost-bound servers from DNS rebinding attacks and rejects browser requests from origins you have not trusted.
FastMCP can validate `Host` and browser `Origin` headers for Streamable HTTP requests before they reach MCP session handling. This request guard protects localhost-bound servers from DNS rebinding attacks, and it remains opt-in in FastMCP 3.x to preserve compatibility with existing ASGI, serverless, and reverse-proxy deployments.
Think of this as a request guard rather than CORS middleware. It decides whether a request can reach MCP session handling. CORS remains a separate browser response-header policy; configure CORS middleware separately when browser JavaScript must read cross-origin responses.
When you deploy behind a public hostname, add the hostname clients use to reach your MCP endpoint. If a browser-based MCP client runs on a separate origin, add that origin as well:
Enable strict validation with `host_origin_protection=True`. When you deploy behind a public hostname, add the hostname clients use to reach your MCP endpoint. If a browser-based MCP client runs on a separate origin, add that origin as well:
```python
from fastmcp import FastMCP
@ -115,6 +115,7 @@ from fastmcp import FastMCP
mcp = FastMCP("My Server")
app = mcp.http_app(
host_origin_protection=True,
allowed_hosts=["mcp.example.com"],
allowed_origins=["https://app.example.com"],
)
@ -132,6 +133,7 @@ if __name__ == "__main__":
transport="http",
host="0.0.0.0",
port=8000,
host_origin_protection=True,
allowed_hosts=["mcp.example.com"],
allowed_origins=["https://app.example.com"],
)
@ -140,11 +142,12 @@ if __name__ == "__main__":
You can also configure these values with environment variables:
```bash
export FASTMCP_HTTP_HOST_ORIGIN_PROTECTION=true
export FASTMCP_HTTP_ALLOWED_HOSTS='["mcp.example.com"]'
export FASTMCP_HTTP_ALLOWED_ORIGINS='["https://app.example.com"]'
```
Use `host_origin_protection=False` only for trusted internal deployments that provide equivalent validation at another layer, such as an ingress proxy.
Use `host_origin_protection="auto"` to protect localhost-bound direct servers while allowing ASGI, serverless, and reverse-proxy deployments to keep their existing Host handling unless they configure explicit trust rules. Use `host_origin_protection=False` to keep the request guard disabled.
### Health Checks
@ -201,7 +204,7 @@ Most MCP clients, including those that you access through a browser like ChatGPT
CORS (Cross-Origin Resource Sharing) is needed when JavaScript running in a web browser connects directly to your MCP server. This is different from using an LLM through a browser—in that case, the browser connects to the LLM service, and the LLM service connects to your MCP server (no CORS needed).
Host and Origin protection runs before CORS. Add browser client origins to `allowed_origins` so trusted browser requests reach the CORS middleware, then configure CORS to let browser JavaScript read the MCP response headers it needs. Setting `allowed_origins` trusts the request; it does not emit `Access-Control-Allow-Origin` or other CORS response headers.
Host and Origin protection runs before CORS when it is active for a request. Add browser client origins to `allowed_origins` so trusted browser requests reach the CORS middleware, then configure CORS to let browser JavaScript read the MCP response headers it needs. Setting `allowed_origins` trusts the request; it does not emit `Access-Control-Allow-Origin` or other CORS response headers.
Browser-based MCP clients that need CORS include:

View file

@ -284,6 +284,7 @@
"integrations/eunomia-authorization",
"integrations/github",
"integrations/google",
"integrations/huggingface",
"integrations/keycloak",
"integrations/oci",
"integrations/permit",

View file

@ -0,0 +1,304 @@
---
title: Hugging Face OAuth 🤝 FastMCP
sidebarTitle: Hugging Face
description: Secure your FastMCP server with Hugging Face OAuth
icon: hugging-face
iconType: brands
---
import { VersionBadge } from "/snippets/version-badge.mdx"
<VersionBadge version="3.4.3" />
This guide shows you how to secure your FastMCP server using **Hugging Face OAuth**.
The `HuggingFaceProvider` uses FastMCP's [OAuth Proxy](/servers/auth/oauth-proxy)
pattern with Hugging Face's OAuth and OpenID Connect endpoints. It works with
manually created confidential apps, public PKCE apps, and Client ID Metadata
Documents (CIMD).
When deploying your MCP server to Hugging Face Spaces, Spaces can create and
manage the OAuth app for you.
## Configuration
### Prerequisites
Before you begin, you will need:
1. A **[Hugging Face account](https://huggingface.co/join)** with access to create OAuth apps
2. Your FastMCP server's URL (can be localhost for development, e.g., `http://localhost:8000`)
### Step 1: Create a Hugging Face OAuth app
Create an OAuth app from your [Hugging Face application settings](https://huggingface.co/settings/applications/new).
For details, see Hugging Face's [OAuth documentation](https://huggingface.co/docs/hub/oauth).
<Steps>
<Step title="Create the OAuth app">
Go to your [Hugging Face application settings](https://huggingface.co/settings/applications/new)
and create a new OAuth application.
Choose a name users will recognize, then configure the redirect URL for
your FastMCP server:
- Development: `http://localhost:8000/auth/callback`
- Production: `https://your-domain.com/auth/callback`
<Warning>
The redirect URL must match exactly. The default path is `/auth/callback`,
but you can customize it using the `redirect_path` parameter. For
production, use HTTPS.
</Warning>
</Step>
<Step title="Save your credentials">
After creating the app, save:
- **Client ID**: The public identifier for your Hugging Face OAuth app
- **Client Secret**: The app secret, if you created a confidential app
<Tip>
Store the client secret securely. Never commit it to version control. Use
environment variables or a secrets manager in production.
</Tip>
</Step>
</Steps>
### Step 2: Configure FastMCP
```python server.py
from fastmcp import FastMCP
from fastmcp.server.auth.providers.huggingface import HuggingFaceProvider
# The HuggingFaceProvider handles Hugging Face's opaque OAuth access tokens
# and stores user data in token claims.
auth_provider = HuggingFaceProvider(
client_id="your-huggingface-client-id", # Your Hugging Face OAuth app client ID
client_secret="your-huggingface-client-secret", # Your Hugging Face OAuth app client secret
base_url="http://localhost:8000", # Must match your OAuth configuration
required_scopes=["openid", "profile"], # Default value
# redirect_path="/auth/callback" # Default value, customize if needed
)
mcp = FastMCP(name="Hugging Face Secured App", auth=auth_provider)
# Add a protected tool to test authentication
@mcp.tool
async def get_user_info() -> dict:
"""Returns information about the authenticated Hugging Face user."""
from fastmcp.server.dependencies import get_access_token
token = get_access_token()
return {
"subject": token.claims.get("sub"),
"username": token.claims.get("preferred_username"),
"profile": token.claims.get("profile"),
}
```
## Public OAuth apps, DCR, and CIMD
Hugging Face supports public OAuth apps (no client secret). For public apps,
omit `client_secret` and provide a `jwt_signing_key` so FastMCP can sign its
own proxy tokens:
```python
auth_provider = HuggingFaceProvider(
client_id="your-public-huggingface-client-id",
base_url="http://localhost:8000",
jwt_signing_key="replace-with-a-secure-secret",
)
```
MCP clients can use Dynamic Client Registration with your FastMCP server. The
`HuggingFaceProvider` inherits FastMCP's OAuth Proxy behavior, which handles
client registration locally and forwards authorization to Hugging Face using
your configured Hugging Face OAuth app. In other words, MCP clients register
with FastMCP, while FastMCP uses your Hugging Face `client_id` and optional
`client_secret` for the upstream OAuth flow.
You can also use a Client ID Metadata Document URL as the `client_id` when your
client metadata is hosted at a stable HTTPS URL:
```python
auth_provider = HuggingFaceProvider(
client_id="https://your-client.example/.well-known/oauth-cimd",
base_url="http://localhost:8000",
jwt_signing_key="replace-with-a-secure-secret",
)
```
## Testing
### Running the Server
Start your server with HTTP transport:
```bash
fastmcp run server.py --transport http --port 8000
```
Your server is now running and protected by Hugging Face OAuth authentication.
### Testing with a Client
Create a test client that authenticates with your Hugging Face-protected server:
```python test_client.py
import asyncio
from fastmcp import Client
async def main():
async with Client("http://localhost:8000/mcp", auth="oauth") as client:
result = await client.call_tool("get_user_info")
print(result)
if __name__ == "__main__":
asyncio.run(main())
```
When you run the client for the first time:
1. Your browser will open to Hugging Face's authorization page
2. Sign in with your Hugging Face account and grant the requested permissions
3. After authorization, you'll be redirected back
4. The client receives the token and can make authenticated requests
<Info>
The client caches tokens locally, so you won't need to re-authenticate for
subsequent runs unless the token expires or you explicitly clear the cache.
</Info>
## Hugging Face Spaces
When deploying to [Hugging Face Spaces](https://huggingface.co/docs/hub/spaces-oauth),
Spaces can create and manage the OAuth app for you. Add OAuth metadata to your
Space README:
```yaml
---
title: FastMCP Hugging Face OAuth
sdk: docker
hf_oauth: true
hf_oauth_expiration_minutes: 480
hf_oauth_scopes:
- email
- inference-api
---
```
Spaces provide `OAUTH_CLIENT_ID`, `OAUTH_CLIENT_SECRET`, `OAUTH_SCOPES`,
`OPENID_PROVIDER_URL`, and `SPACE_HOST` environment variables:
```python
import os
from fastmcp import FastMCP
from fastmcp.server.auth.providers.huggingface import HuggingFaceProvider
from fastmcp.utilities.auth import parse_scopes
base_url = f"https://{os.environ['SPACE_HOST']}"
auth_provider = HuggingFaceProvider(
client_id=os.environ["OAUTH_CLIENT_ID"],
client_secret=os.environ["OAUTH_CLIENT_SECRET"],
base_url=base_url,
jwt_signing_key=os.environ["JWT_SIGNING_KEY"],
required_scopes=parse_scopes(os.environ.get("OAUTH_SCOPES")) or ["openid", "profile"],
)
mcp = FastMCP(name="Hugging Face Space App", auth=auth_provider)
```
Set `JWT_SIGNING_KEY` as a Space secret.
## Hugging Face scopes
The default scopes are `openid` and `profile`. Add more scopes when your tools
need Hub capabilities:
| Scope | Description |
|-------|-------------|
| `email` | Access the user's email address |
| `read-billing` | Know whether the user has a payment method set up |
| `read-repos` | Read the user's personal repositories |
| `gated-repos` | Read public gated repositories the user can access |
| `contribute-repos` | Create repositories and access app-created repositories |
| `write-repos` | Read and write the user's personal repositories |
| `manage-repos` | Full repository access, including creation and deletion |
| `read-collections` | Read the user's personal collections |
| `write-collections` | Read and write the user's personal collections, including collection creation and deletion |
| `inference-api` | Use Hugging Face Inference Providers as the user |
| `jobs` | Run Hugging Face Jobs |
| `webhooks` | Manage webhooks |
| `write-discussions` | Open discussions and pull requests, and interact with discussions |
```python
auth_provider = HuggingFaceProvider(
client_id="your-huggingface-client-id",
client_secret="your-huggingface-client-secret",
base_url="https://your-domain.com",
required_scopes=["openid", "profile", "inference-api", "jobs"],
)
```
For organization resources, use Hugging Face's normal OAuth organization grant
flow. If you need a specific organization, pass Hugging Face's `orgIds`
authorization parameter. The value is the organization ID from the
`organizations.sub` field in the Hugging Face userinfo response:
```python
auth_provider = HuggingFaceProvider(
client_id="your-huggingface-client-id",
client_secret="your-huggingface-client-secret",
base_url="https://your-domain.com",
extra_authorize_params={"orgIds": "your-org-id"},
)
```
## Production Configuration
For production deployments with persistent token management across server
restarts, configure `jwt_signing_key` and `client_storage`:
```python server.py
import os
from cryptography.fernet import Fernet
from fastmcp import FastMCP
from fastmcp.server.auth.providers.huggingface import HuggingFaceProvider
from key_value.aio.stores.redis import RedisStore
from key_value.aio.wrappers.encryption import FernetEncryptionWrapper
# Production setup with encrypted persistent token storage
auth_provider = HuggingFaceProvider(
client_id="your-huggingface-client-id",
client_secret=os.environ["HUGGINGFACE_CLIENT_SECRET"],
base_url="https://your-production-domain.com",
required_scopes=["openid", "profile", "email"],
# Production token management
jwt_signing_key=os.environ["JWT_SIGNING_KEY"],
client_storage=FernetEncryptionWrapper(
key_value=RedisStore(
host=os.environ["REDIS_HOST"],
port=int(os.environ["REDIS_PORT"])
),
fernet=Fernet(os.environ["STORAGE_ENCRYPTION_KEY"])
)
)
mcp = FastMCP(name="Production Hugging Face App", auth=auth_provider)
```
<Note>
Parameters (`jwt_signing_key` and `client_storage`) work together to ensure
tokens and client registrations survive server restarts. **Wrap your storage in
`FernetEncryptionWrapper` to encrypt sensitive OAuth tokens at rest** - without
it, tokens are stored in plaintext. Store secrets in environment variables and
use a persistent storage backend like Redis for distributed deployments.
For complete details on these parameters, see the [OAuth Proxy documentation](/servers/auth/oauth-proxy#configuration-parameters).
</Note>

View file

@ -42,9 +42,9 @@ These control how the server listens when running with an HTTP transport.
| `FASTMCP_STREAMABLE_HTTP_PATH` | `str` | `/mcp` | Path for Streamable HTTP endpoint. |
| `FASTMCP_STATELESS_HTTP` | `bool` | `false` | Enable stateless HTTP mode (new transport per request). Useful for multi-worker deployments. |
| `FASTMCP_JSON_RESPONSE` | `bool` | `false` | Use JSON responses instead of SSE for Streamable HTTP. |
| `FASTMCP_HTTP_HOST_ORIGIN_PROTECTION` | `bool` | `true` | Validate `Host` and browser `Origin` headers for Streamable HTTP requests. |
| `FASTMCP_HTTP_ALLOWED_HOSTS` | `list[str] \| null` | `null` | Additional trusted hostnames for Streamable HTTP requests. Use a JSON array, such as `["mcp.example.com"]`. |
| `FASTMCP_HTTP_ALLOWED_ORIGINS` | `list[str] \| null` | `null` | Browser origins trusted by the Streamable HTTP request guard. Configure CORS separately for cross-origin browser reads. Use a JSON array, such as `["https://app.example.com"]`. |
| `FASTMCP_HTTP_HOST_ORIGIN_PROTECTION` | `bool \| "auto"` | `false` | Validate `Host` and browser `Origin` headers for Streamable HTTP requests. `auto` protects localhost-bound servers and explicit host/origin allowlists. |
| `FASTMCP_HTTP_ALLOWED_HOSTS` | `list[str] \| null` | `null` | Additional trusted hostnames when Host and Origin protection is enabled. Use a JSON array, such as `["mcp.example.com"]`. |
| `FASTMCP_HTTP_ALLOWED_ORIGINS` | `list[str] \| null` | `null` | Browser origins trusted when Host and Origin protection is enabled. Configure CORS separately for cross-origin browser reads. Use a JSON array, such as `["https://app.example.com"]`. |
| `FASTMCP_DEBUG` | `bool` | `false` | Enable debug mode. |
## Error Handling
@ -88,6 +88,26 @@ When setting Docket values in a `.env` file, use a **double** underscore: `FASTM
| `FASTMCP_DOCKET_RECONNECTION_DELAY` | `timedelta` | `5s` | Delay between reconnection attempts when the worker loses its backend connection. |
| `FASTMCP_DOCKET_MINIMUM_CHECK_INTERVAL` | `timedelta` | `50ms` | How frequently the worker polls for new tasks. Lower values reduce latency at the cost of more CPU usage. |
## Security
These control FastMCP's SSRF protection for the outbound fetches it makes during authentication (OAuth client metadata and JWKS).
| Environment Variable | Type | Default | Description |
|---|---|---|---|
| `FASTMCP_SSRF_TRUST_PROXY` | `bool` | `false` | Trust an outbound HTTP proxy for SSRF-protected fetches. When `false`, FastMCP resolves the target hostname itself and refuses to connect if it maps to a private, loopback, link-local, or reserved IP. When `true`, FastMCP routes auth metadata and JWKS fetches through the configured `HTTPS_PROXY`/`ALL_PROXY` and does not honor `NO_PROXY`; if no proxy is configured the fetch is refused. |
By default, FastMCP protects its OAuth and JWKS fetches against [SSRF](https://owasp.org/www-community/attacks/Server_Side_Request_Forgery) by resolving the target hostname, rejecting any address that maps to a private, loopback, link-local, or reserved IP, and then pinning the connection to that validated IP.
This breaks when a corporate `CONNECT` proxy is the only egress path: the container often cannot resolve external DNS at all (only the proxy can), and even when it can, pinning to the IP makes TLS verification fail because public certificates list hostnames, not IP addresses.
Set `FASTMCP_SSRF_TRUST_PROXY=true` when a trusted proxy is your mandated egress. FastMCP then skips DNS resolution and the IP blocklist entirely and makes a single request to the hostname URL, explicitly routed through the proxy named by the standard `HTTPS_PROXY` / `ALL_PROXY` environment variables (checked in that order). The HTTPS-only and hostname checks still apply.
<Warning>
This is a deliberate trust shift: the IP blocklist cannot be enforced through a proxy (the proxy does its own DNS, so an address FastMCP resolved is not the one the proxy dials). Only enable it when the proxy itself is trusted to mediate egress.
FastMCP reads the proxy URL from the environment and passes it to the HTTP client explicitly, so environment proxy selection and `NO_PROXY` do not participate — the request either goes through that exact proxy or fails outright. TLS trust environment variables remain enabled, so `SSL_CERT_FILE` and `SSL_CERT_DIR` continue to work for deployments that install a corporate CA. A host that `NO_PROXY` would otherwise exclude is still routed through the configured proxy rather than fetched direct with the IP blocklist disabled — the safer of the two options, since the blocklist cannot apply to a direct fetch here anyway. If you set `FASTMCP_SSRF_TRUST_PROXY=true` but neither `HTTPS_PROXY` nor `ALL_PROXY` is present in the server process's environment (an `HTTP_PROXY` alone never routes these HTTPS-only fetches), the request would otherwise go out **direct with the IP blocklist disabled** — no SSRF protection at all. Rather than send it, FastMCP refuses the fetch and raises `SSRFError` with an actionable message. The contract is crisp: proxy-trust mode delegates SSRF protection to the proxy, and with no proxy configured the fetch cannot proceed. Enable this setting only together with an active proxy that routes your auth endpoints.
</Warning>
## Advanced
| Environment Variable | Type | Default | Description |

View file

@ -5,6 +5,42 @@ icon: "sparkles"
tag: NEW
---
<Update label="FastMCP 3.4.6" description="August 5, 2026" tags={["Releases"]}>
<Card
title="FastMCP v3.4.6: Trust, but Proxy"
href="https://github.com/PrefectHQ/fastmcp/releases/tag/v3.4.6"
cta="Read the release notes"
>
FastMCP 3.4.6 adds trusted-proxy support for SSRF-protected OAuth metadata and JWKS fetches on the 3.x line. Deployments can route these requests through a mandated corporate proxy while preserving custom CA certificates, and FastMCP refuses the fetch when no proxy is configured instead of risking an unprotected direct request.
</Card>
</Update>
<Update label="FastMCP 3.4.5" description="July 27, 2026" tags={["Releases"]}>
<Card
title="FastMCP v3.4.5: Key Change"
href="https://github.com/PrefectHQ/fastmcp/releases/tag/v3.4.5"
cta="Read the release notes"
>
A maintenance release for the 3.x line. A single unrecognized JWKS key — Ed25519, which Rauthy and Ory Hydra publish by default — no longer poisons the entire key cache, alongside fixes for Azure scope fallback, OpenAPI `deepObject` query serialization, schema compression, and transformed tool `required` ordering.
</Card>
</Update>
<Update label="FastMCP 3.4.4" description="July 8, 2026" tags={["Releases"]}>
<Card
title="FastMCP v3.4.4: Host in Translation"
href="https://github.com/PrefectHQ/fastmcp/releases/tag/v3.4.4"
cta="Read the release notes"
>
A compatibility patch for HTTP deployments affected by the 3.4.3 Host/Origin guard defaults. FastMCP 3.x now keeps strict Host and Origin validation available for explicit opt-in deployments without rejecting existing ASGI, serverless, and reverse-proxy traffic by default.
🌐 **HTTP compatibility restored** — existing hosted deployments keep accepting their public Host headers unless strict host/origin protection is configured.
🔐 **Guard remains available** — deployments that know their public host and browser origins can still enable strict validation with `host_origin_protection=True`, `allowed_hosts`, and `allowed_origins`.
🤗 **Hugging Face auth** — new OAuth provider support covers public and private Hugging Face apps, with docs and examples for PKCE, Dynamic Client Registration, and CIMD.
</Card>
</Update>
<Update label="FastMCP 3.4.3" description="July 5, 2026" tags={["Releases"]}>
<Card
title="FastMCP v3.4.3: The Fast and the Secure-ious"
@ -725,4 +761,3 @@ Because life's too short for boilerplate.
This is where it all started. FastMCPs launch post introduced a clean, Pythonic way to build MCP servers without the protocol overhead. Just write functions; FastMCP handles the rest. What began as a weekend project quickly became the foundation of a growing ecosystem.
</Card>
</Update>

View file

@ -0,0 +1,31 @@
# Hugging FAce OAuth Example
Demonstrates FastMCP server protection with Hugging Face OAuth.
## Setup
1. Create a Hugging Face OAuth App:
- Go to Hugging Face Settings > Connected Apps > Create App (`https://huggingface.co/settings/applications/new`)
- Set Authorization callback URL to: `http://localhost:8000/auth/callback`
- Copy the Client ID and Client Secret
2. Set environment variables:
```bash
export FASTMCP_SERVER_AUTH_HF_CLIENT_ID="your-client-id"
export FASTMCP_SERVER_AUTH_HF_CLIENT_SECRET="your-client-secret"
```
3. Run the server:
```bash
python server.py
```
4. In another terminal, run the client:
```bash
python client.py
```
The client will open your browser for Hugging Face authentication.

View file

@ -0,0 +1,32 @@
"""OAuth client example for connecting to FastMCP servers.
This example demonstrates how to connect to an OAuth-protected FastMCP server.
To run:
python client.py
"""
import asyncio
from fastmcp.client import Client
SERVER_URL = "http://localhost:8000/mcp"
async def main():
try:
async with Client(SERVER_URL, auth="oauth") as client:
assert await client.ping()
print("✅ Successfully authenticated!")
tools = await client.list_tools()
print(f"🔧 Available tools ({len(tools)}):")
for tool in tools:
print(f" - {tool.name}: {tool.description}")
except Exception as e:
print(f"❌ Authentication failed: {e}")
raise
if __name__ == "__main__":
asyncio.run(main())

View file

@ -0,0 +1,35 @@
import os
from fastmcp import FastMCP
from fastmcp.server.auth.providers.huggingface import HuggingFaceProvider
auth_provider = HuggingFaceProvider(
# Your Hugging Face OAuth app client ID
client_id=os.getenv("FASTMCP_SERVER_AUTH_HF_CLIENT_ID") or "",
# Your Hugging Face OAuth app client secret
client_secret=os.getenv("FASTMCP_SERVER_AUTH_HF_CLIENT_SECRET") or "",
# Must match your OAuth configuration
base_url="http://localhost:8000",
# Supply jwt_signing_key instead of client_secret for public applications
# jwt_signing_key="replace-with-a-secure-secret"
)
mcp = FastMCP(name="Hugging Face Secured App", auth=auth_provider)
# Add a tool to test authentication
@mcp.tool
async def get_user_info() -> dict:
"""Returns information about the authenticated Hugging Face user."""
from fastmcp.server.dependencies import get_access_token
token = get_access_token()
return {
"subject": token.claims.get("sub"),
"username": token.claims.get("preferred_username"),
"profile": token.claims.get("profile"),
}
if __name__ == "__main__":
mcp.run(transport="http", port=8000)

View file

@ -501,8 +501,12 @@ class AzureProvider(OAuthProxy):
Returns:
List of scopes for Azure token endpoint
"""
# Prefix scopes for this API
prefixed_scopes = self._prefix_scopes_for_azure(scopes or [])
# Prefix scopes for this API. Some clients omit the scope parameter on
# the MCP authorization request; use the provider's configured scopes
# just like the authorize URL path does.
prefixed_scopes = self._prefix_scopes_for_azure(
scopes or self.required_scopes or []
)
# Add OIDC scopes only (not other API scopes) to avoid AADSTS28000
if self.additional_authorize_scopes:
@ -528,9 +532,13 @@ class AzureProvider(OAuthProxy):
"""
logger.debug("Base scopes from storage: %s", scopes)
# Some clients omit the scope parameter on the MCP authorization request;
# use the provider's configured scopes just like the authorize URL path does.
requested_scopes = scopes or self.required_scopes or []
# Filter out any additional_authorize_scopes that may have been stored
additional_scopes_set = set(self.additional_authorize_scopes or [])
base_scopes = [s for s in scopes if s not in additional_scopes_set]
base_scopes = [s for s in requested_scopes if s not in additional_scopes_set]
# Prefix base scopes with identifier_uri for Azure
prefixed_scopes = self._prefix_scopes_for_azure(base_scopes)

View file

@ -0,0 +1,279 @@
"""Hugging Face OAuth provider for FastMCP."""
from __future__ import annotations
import contextlib
from collections.abc import Mapping
from typing import Any, Literal
import httpx
from key_value.aio.protocols import AsyncKeyValue
from pydantic import AnyHttpUrl
from fastmcp.server.auth import TokenVerifier
from fastmcp.server.auth.auth import AccessToken
from fastmcp.server.auth.oauth_proxy import OAuthProxy
from fastmcp.utilities.auth import parse_scopes
from fastmcp.utilities.logging import get_logger
logger = get_logger(__name__)
HUGGINGFACE_AUTHORIZATION_ENDPOINT = "https://huggingface.co/oauth/authorize"
HUGGINGFACE_TOKEN_ENDPOINT = "https://huggingface.co/oauth/token"
HUGGINGFACE_USERINFO_ENDPOINT = "https://huggingface.co/oauth/userinfo"
HUGGINGFACE_WHOAMI_ENDPOINT = "https://huggingface.co/api/whoami-v2"
DEFAULT_HUGGINGFACE_SCOPES = ["openid", "profile"]
def _extract_scopes(data: Mapping[str, Any]) -> list[str]:
scope_value = data.get("scope") or data.get("scopes")
if isinstance(scope_value, str):
return parse_scopes(scope_value) or []
if isinstance(scope_value, list):
return [str(scope).strip() for scope in scope_value if str(scope).strip()]
auth = data.get("auth")
if not isinstance(auth, Mapping):
return []
access_token = auth.get("accessToken")
if not isinstance(access_token, Mapping):
return []
nested_scopes = access_token.get("scopes") or access_token.get("scope")
if isinstance(nested_scopes, str):
return parse_scopes(nested_scopes) or []
if isinstance(nested_scopes, list):
return [
str(scope.get("name") if isinstance(scope, Mapping) else scope).strip()
for scope in nested_scopes
if str(scope.get("name") if isinstance(scope, Mapping) else scope).strip()
]
return []
class HuggingFaceTokenVerifier(TokenVerifier):
"""Token verifier for Hugging Face OAuth access tokens.
Hugging Face OAuth access tokens are opaque, so validation is performed by
calling Hugging Face's userinfo endpoint.
"""
def __init__(
self,
*,
required_scopes: list[str] | None = None,
timeout_seconds: int = 10,
http_client: httpx.AsyncClient | None = None,
):
super().__init__(required_scopes=required_scopes)
self.timeout_seconds = timeout_seconds
self._http_client = http_client
async def verify_token(self, token: str) -> AccessToken | None:
"""Verify a Hugging Face OAuth token using the userinfo endpoint."""
try:
async with (
contextlib.nullcontext(self._http_client)
if self._http_client is not None
else httpx.AsyncClient(timeout=self.timeout_seconds)
) as client:
userinfo_response = await client.get(
HUGGINGFACE_USERINFO_ENDPOINT,
headers={
"Authorization": f"Bearer {token}",
"User-Agent": "FastMCP-HuggingFace-OAuth",
},
)
if userinfo_response.status_code != 200:
logger.debug(
"Hugging Face token verification failed: %d",
userinfo_response.status_code,
)
return None
userinfo = userinfo_response.json()
sub = userinfo.get("sub")
if not sub:
logger.debug("Hugging Face userinfo missing 'sub' claim")
return None
token_scopes = _extract_scopes(userinfo)
whoami: dict[str, Any] | None = None
if not token_scopes or (
self.required_scopes
and not set(self.required_scopes).issubset(set(token_scopes))
):
whoami = await self._fetch_whoami(client, token)
if whoami:
token_scopes = list(
dict.fromkeys([*token_scopes, *_extract_scopes(whoami)])
)
if not token_scopes:
token_scopes = list(DEFAULT_HUGGINGFACE_SCOPES)
if self.required_scopes and not set(self.required_scopes).issubset(
set(token_scopes)
):
logger.debug(
"Hugging Face token missing required scopes. Has %d, needs %d",
len(token_scopes),
len(self.required_scopes),
)
return None
username = (
userinfo.get("preferred_username")
or userinfo.get("nickname")
or userinfo.get("name")
)
return AccessToken(
token=token,
client_id=str(sub),
scopes=token_scopes,
expires_at=None,
claims={
"sub": str(sub),
"name": userinfo.get("name"),
"preferred_username": username,
"email": userinfo.get("email"),
"email_verified": userinfo.get("email_verified"),
"profile": userinfo.get("profile"),
"picture": userinfo.get("picture"),
"organizations": userinfo.get("organizations"),
"huggingface_userinfo": userinfo,
"huggingface_whoami": whoami,
},
)
except httpx.RequestError as e:
logger.debug("Failed to verify Hugging Face token: %s", e)
return None
except Exception as e:
logger.debug("Hugging Face token verification error: %s", e)
return None
async def _fetch_whoami(
self, client: httpx.AsyncClient, token: str
) -> dict[str, Any] | None:
response = await client.get(
HUGGINGFACE_WHOAMI_ENDPOINT,
headers={
"Authorization": f"Bearer {token}",
"User-Agent": "FastMCP-HuggingFace-OAuth",
},
)
if response.status_code != 200:
logger.debug("Hugging Face whoami lookup failed: %d", response.status_code)
return None
return response.json()
class HuggingFaceProvider(OAuthProxy):
"""Complete Hugging Face OAuth provider for FastMCP."""
def __init__(
self,
*,
client_id: str,
client_secret: str | None = None,
base_url: AnyHttpUrl | str,
resource_base_url: AnyHttpUrl | str | None = None,
issuer_url: AnyHttpUrl | str | None = None,
redirect_path: str | None = None,
required_scopes: list[str] | None = None,
valid_scopes: list[str] | None = None,
timeout_seconds: int = 10,
allowed_client_redirect_uris: list[str] | None = None,
client_storage: AsyncKeyValue | None = None,
jwt_signing_key: str | bytes | None = None,
require_authorization_consent: bool | Literal["remember", "external"] = True,
consent_csp_policy: str | None = None,
forward_resource: bool = True,
fallback_refresh_token_expiry_seconds: int | None = None,
fastmcp_access_token_expiry_seconds: int | None = None,
token_expiry_threshold_seconds: int = 0,
extra_authorize_params: dict[str, str] | None = None,
extra_token_params: dict[str, str] | None = None,
http_client: httpx.AsyncClient | None = None,
enable_cimd: bool = True,
):
"""Initialize Hugging Face OAuth provider.
Args:
client_id: Hugging Face OAuth app client ID. Public apps and CIMD
client IDs are supported.
client_secret: Hugging Face OAuth app client secret. Optional for
public PKCE apps; when omitted, ``jwt_signing_key`` is required.
base_url: Public URL where OAuth endpoints will be accessible.
required_scopes: Required Hugging Face scopes. Defaults to
``["openid", "profile"]``.
valid_scopes: Scopes clients may request. Defaults to required scopes.
extra_authorize_params: Extra authorization parameters, such as
``{"orgIds": "your-org-id"}`` for organization grants.
"""
required_scopes_final = (
parse_scopes(required_scopes)
if required_scopes is not None
else list(DEFAULT_HUGGINGFACE_SCOPES)
) or []
valid_scopes_final = parse_scopes(valid_scopes)
# Do not pass provider-level required_scopes into the verifier here.
# Hugging Face's userinfo endpoint validates opaque access tokens and
# returns identity claims, but granted scopes are carried reliably in
# the upstream token response. OAuthProxy stores those scopes, enforces
# provider.required_scopes against FastMCP-issued tokens, and
# _uses_alternate_verification() patches the stored upstream scopes
# onto the returned AccessToken.
token_verifier = HuggingFaceTokenVerifier(
timeout_seconds=timeout_seconds,
http_client=http_client,
)
super().__init__(
upstream_authorization_endpoint=HUGGINGFACE_AUTHORIZATION_ENDPOINT,
upstream_token_endpoint=HUGGINGFACE_TOKEN_ENDPOINT,
upstream_client_id=client_id,
upstream_client_secret=client_secret,
token_verifier=token_verifier,
base_url=base_url,
resource_base_url=resource_base_url,
redirect_path=redirect_path,
issuer_url=issuer_url or base_url,
allowed_client_redirect_uris=allowed_client_redirect_uris,
client_storage=client_storage,
jwt_signing_key=jwt_signing_key,
require_authorization_consent=require_authorization_consent,
consent_csp_policy=consent_csp_policy,
forward_resource=forward_resource,
fallback_refresh_token_expiry_seconds=fallback_refresh_token_expiry_seconds,
fastmcp_access_token_expiry_seconds=fastmcp_access_token_expiry_seconds,
token_expiry_threshold_seconds=token_expiry_threshold_seconds,
extra_authorize_params=extra_authorize_params,
extra_token_params=extra_token_params,
token_endpoint_auth_method="client_secret_basic"
if client_secret
else "none",
valid_scopes=valid_scopes_final,
enable_cimd=enable_cimd,
)
logger.debug(
"Initialized Hugging Face OAuth provider for client %s with scopes: %s",
client_id,
required_scopes_final,
)
self.required_scopes = required_scopes_final
self.update_default_scopes(valid_scopes_final or required_scopes_final)
def _uses_alternate_verification(self) -> bool:
"""Patch returned token scopes from the upstream token response.
Hugging Face OAuth access tokens are opaque. The userinfo endpoint
validates the token and returns identity claims, but scope information is
carried by the token response stored in OAuthProxy's upstream token set.
"""
return True

View file

@ -347,11 +347,26 @@ class JWTVerifier(TokenVerifier):
try:
jwks_data = await self._fetch_jwks()
# Cache all keys
# Cache all usable keys. A key that cannot be converted (e.g. an
# unsupported kty like OKP/Ed25519) is skipped rather than failing
# the whole set — per RFC 7517 §5, clients should ignore JWKs they
# don't understand. Otherwise one exotic key published by the
# authorization server would reject every token, including ones
# signed by supported keys in the same set (#4515).
self._jwks_cache = {}
skipped_kids: set[str] = set()
for key_data in jwks_data.get("keys", []):
if not isinstance(key_data, dict):
self.logger.debug("Skipping non-object JWKS entry: %r", key_data)
continue
key_kid = key_data.get("kid")
public_key = _jwk_to_pem(key_data)
try:
public_key = _jwk_to_pem(key_data)
except (JoseError, TypeError, KeyError, ValueError) as e:
self.logger.debug("Skipping unusable JWKS key %r: %s", key_kid, e)
if key_kid:
skipped_kids.add(key_kid)
continue
if key_kid:
self._jwks_cache[key_kid] = public_key
@ -364,6 +379,16 @@ class JWTVerifier(TokenVerifier):
# Select the appropriate key
if kid:
if kid not in self._jwks_cache:
if kid in skipped_kids:
self.logger.debug(
"JWKS key lookup failed: key ID '%s' is present "
"but its key type is unsupported",
kid,
)
raise ValueError(
f"Key ID '{kid}' found in JWKS but its key type "
"is unsupported"
)
self.logger.debug(
"JWKS key lookup failed: key ID '%s' not found", kid
)

View file

@ -4,12 +4,22 @@ This module provides SSRF-protected HTTP fetching with:
- DNS resolution and IP validation before requests
- DNS pinning to prevent rebinding TOCTOU attacks
- Support for both CIMD and JWKS fetches
When ``FASTMCP_SSRF_TRUST_PROXY`` is set, DNS resolution and the IP blocklist are
skipped and a single request is made to the hostname URL through the configured
HTTPS_PROXY/ALL_PROXY, delegating DNS and egress to that trusted proxy (the scheme
and hostname checks still apply). The proxy URL is read from the environment and
passed to httpx explicitly, so environment proxy selection and NO_PROXY are not
evaluated. ``trust_env`` remains enabled so SSL_CERT_FILE and SSL_CERT_DIR continue
to provide corporate CA trust. If no proxy is configured, the fetch is refused
rather than sent direct with the blocklist disabled.
"""
from __future__ import annotations
import asyncio
import ipaddress
import os
import socket
import time
from collections.abc import Mapping
@ -18,6 +28,7 @@ from urllib.parse import urlparse
import httpx
import fastmcp
from fastmcp.utilities.logging import get_logger
logger = get_logger(__name__)
@ -62,6 +73,26 @@ def format_ip_for_url(ip_str: str) -> str:
return ip_str
def _configured_proxy_url() -> str | None:
"""Return the proxy URL to route proxy-trust fetches through, if any is set.
Reads ``HTTPS_PROXY``/``https_proxy`` first, falling back to ``ALL_PROXY``/
``all_proxy``. This is a simple presence check: no host matching, no ``NO_PROXY``
evaluation. The caller passes the returned URL to httpx explicitly, which takes
precedence over environment proxy selection and NO_PROXY while preserving
environment-provided CA trust see the module docstring and :func:`validate_url`
for why that matters.
Returns:
The configured proxy URL, or None if none of the supported variables are set.
"""
for name in ("HTTPS_PROXY", "https_proxy", "ALL_PROXY", "all_proxy"):
value = os.environ.get(name)
if value:
return value
return None
class SSRFError(Exception):
"""Raised when an SSRF protection check fails."""
@ -179,6 +210,7 @@ class ValidatedURL:
port: int
path: str
resolved_ips: list[str]
proxy_url: str | None = None
@dataclass
@ -190,6 +222,55 @@ class SSRFFetchResponse:
headers: dict[str, str]
@dataclass
class _FetchTarget:
"""A single connection attempt for an SSRF-safe fetch.
In pinned (default) mode there is one target per resolved IP: the request goes to
an IP-literal URL with Host and SNI pinned to the validated hostname. In proxy
mode (FASTMCP_SSRF_TRUST_PROXY) there is a single target: the original hostname
URL with no pinning and an explicit ``proxy_url``, so the request is dialed
through the trusted proxy and the proxy (not httpx's environment-proxy routing)
owns DNS and TLS.
"""
url: str
host_header: str | None
sni_hostname: str | None
proxy_url: str | None = None
def _build_fetch_targets(validated: ValidatedURL) -> list[_FetchTarget]:
"""Build the ordered connection attempts for a validated URL.
An empty ``resolved_ips`` means proxy mode (see :func:`validate_url`): a single
unpinned request to the original hostname URL, explicitly routed through
``validated.proxy_url``. Otherwise, one pinned IP-literal request per resolved
IP, tried in order with fallback on connection error.
"""
if not validated.resolved_ips:
# Proxy mode: dial the original hostname URL verbatim and let the proxy parse
# and resolve it. validated.hostname is informational here — it does not
# constrain what gets dialed — so do not pin Host or SNI from it.
return [
_FetchTarget(
url=validated.original_url,
host_header=None,
sni_hostname=None,
proxy_url=validated.proxy_url,
)
]
return [
_FetchTarget(
url=f"https://{format_ip_for_url(ip)}:{validated.port}{validated.path}",
host_header=validated.hostname,
sni_hostname=validated.hostname,
)
for ip in validated.resolved_ips
]
async def validate_url(url: str, require_path: bool = False) -> ValidatedURL:
"""Validate URL for SSRF and resolve to IPs.
@ -201,7 +282,8 @@ async def validate_url(url: str, require_path: bool = False) -> ValidatedURL:
ValidatedURL with resolved IPs
Raises:
SSRFError: If URL is invalid or resolves to blocked IPs
SSRFError: If the URL is invalid, resolves to blocked IPs, or proxy-trust
mode is enabled but no configured proxy will route the request.
"""
try:
parsed = urlparse(url)
@ -219,8 +301,55 @@ async def validate_url(url: str, require_path: bool = False) -> ValidatedURL:
hostname = parsed.hostname or parsed.netloc
port = parsed.port or 443
path = parsed.path + ("?" + parsed.query if parsed.query else "")
# Resolve and validate IPs
# Proxy mode (FASTMCP_SSRF_TRUST_PROXY): a trusted outbound proxy owns DNS and
# egress, so resolving the hostname here is pointless — the IP we'd pin is not
# the one the proxy dials, making the blocklist unenforceable theater. Skip
# resolution and the blocklist entirely and signal proxy mode downstream with an
# empty resolved_ips list. The scheme (HTTPS) and host checks above still run.
if fastmcp.settings.ssrf_trust_proxy:
# Skipping the blocklist is only safe if the request is *actually* routed
# through a trusted proxy, so this does not try to predict whether it will
# be — it controls it. Earlier revisions predicted the HTTP client's routing
# decision, first by approximating NO_PROXY handling with urllib's proxy
# bypass helper,
# then by replicating the client's own environment-proxy matching
# internally. Both were still predictions of a library with
# open-ended NO_PROXY semantics, and each was found wrong for a different
# NO_PROXY form (port-qualified, IPv6, scheme-qualified entries each broke a
# different revision) — always in the dangerous direction of assuming
# "proxied" for a request that actually went out direct.
#
# Instead, read the proxy URL directly from the environment and hand it to
# httpx explicitly below. An explicit `proxy=` fixes httpx's proxy map without
# consulting NO_PROXY, even with `trust_env=True`; keeping trust_env enabled
# preserves SSL_CERT_FILE and SSL_CERT_DIR for corporate CA trust. The request
# therefore goes through that proxy or the connection fails. A NO_PROXY'd
# host is routed through the proxy rather than fetched direct with the
# blocklist already disabled, which is strictly safer than the alternative
# (see the module docstring). If no proxy is configured, there is nothing to
# route through, so refuse rather than fetch unprotected.
proxy_url = _configured_proxy_url()
if proxy_url is None:
raise SSRFError(
f"FASTMCP_SSRF_TRUST_PROXY is enabled but no HTTPS_PROXY/ALL_PROXY is "
f"configured, so the request to {hostname} would go direct with SSRF "
f"protection disabled. Set HTTPS_PROXY (or ALL_PROXY) to the trusted "
f"proxy, or unset FASTMCP_SSRF_TRUST_PROXY to restore DNS/IP "
f"validation."
)
return ValidatedURL(
original_url=url,
hostname=hostname,
port=port,
path=path,
resolved_ips=[],
proxy_url=proxy_url,
)
# Resolve and validate IPs (resolve_hostname raises rather than returning [], so a
# successful return here always yields a non-empty list — see ssrf_safe_fetch_response).
resolved_ips = await resolve_hostname(hostname, port)
blocked = [ip for ip in resolved_ips if not is_ip_allowed(ip)]
@ -234,7 +363,7 @@ async def validate_url(url: str, require_path: bool = False) -> ValidatedURL:
original_url=url,
hostname=hostname,
port=port,
path=parsed.path + ("?" + parsed.query if parsed.query else ""),
path=path,
resolved_ips=resolved_ips,
)
@ -305,31 +434,34 @@ async def ssrf_safe_fetch_response(
last_error: Exception | None = None
expected_statuses = allowed_status_codes or {200}
for pinned_ip in validated.resolved_ips:
# One target per pinned IP in default mode; a single unpinned target in proxy mode.
targets = _build_fetch_targets(validated)
for target in targets:
elapsed = time.monotonic() - start_time
if elapsed > overall_timeout:
raise SSRFFetchError(f"Overall timeout exceeded: {url}")
remaining = max(1.0, overall_timeout - elapsed)
pinned_url = (
f"https://{format_ip_for_url(pinned_ip)}:{validated.port}{validated.path}"
)
logger.debug("SSRF-safe fetch: %s -> %s", url, target.url)
logger.debug(
"SSRF-safe fetch: %s -> %s (pinned to %s)",
url,
pinned_url,
pinned_ip,
)
headers = {"Host": validated.hostname}
# In pinned mode Host is forced to the validated hostname; in proxy mode httpx
# derives it from the hostname URL. Either way, never let a caller override it.
headers: dict[str, str] = {}
if target.host_header is not None:
headers["Host"] = target.host_header
if request_headers:
for key, value in request_headers.items():
# Host must remain pinned to the validated hostname.
if key.lower() == "host":
continue
headers[key] = value
# Pin SNI to the hostname when connecting to an IP literal; in proxy mode httpx
# derives SNI from the URL, so no override is sent.
extensions: dict[str, str] = {}
if target.sni_hostname is not None:
extensions["sni_hostname"] = target.sni_hostname
try:
# Use httpx with streaming to enforce size limit during download
async with (
@ -342,12 +474,17 @@ async def ssrf_safe_fetch_response(
),
follow_redirects=False,
verify=True,
# An explicit proxy_url controls routing without consulting
# environment proxy selection or NO_PROXY. Keep trust_env enabled
# in both modes so SSL_CERT_FILE and SSL_CERT_DIR remain effective.
proxy=target.proxy_url,
trust_env=True,
) as client,
client.stream(
"GET",
pinned_url,
target.url,
headers=headers,
extensions={"sni_hostname": validated.hostname},
extensions=extensions,
) as response,
):
if time.monotonic() - start_time > overall_timeout:
@ -399,4 +536,4 @@ async def ssrf_safe_fetch_response(
raise SSRFFetchError(f"Timeout fetching {url}") from last_error
raise SSRFFetchError(f"Error fetching {url}: {last_error}") from last_error
raise SSRFFetchError(f"Error fetching {url}: no resolved IPs succeeded")
raise SSRFFetchError(f"Error fetching {url}: no fetch targets succeeded")

View file

@ -5,7 +5,7 @@ from contextlib import asynccontextmanager, contextmanager
from contextvars import ContextVar
from fnmatch import fnmatchcase
from ipaddress import ip_address
from typing import TYPE_CHECKING, Any
from typing import TYPE_CHECKING, Any, Literal
from urllib.parse import urlsplit
from uuid import uuid4
@ -36,6 +36,8 @@ if TYPE_CHECKING:
logger = get_logger(__name__)
DEFAULT_HOSTS = ("127.0.0.1", "localhost", "::1")
HostOriginProtection = bool | Literal["auto"]
HostOriginProtectionMode = Literal["auto", "strict"]
class FastMCPStreamableHTTPSessionManager(StreamableHTTPSessionManager):
@ -228,34 +230,81 @@ class HostOriginGuardMiddleware:
app: ASGIApp,
allowed_hosts: Sequence[str] | None = None,
allowed_origins: Sequence[str] | None = None,
mode: HostOriginProtectionMode = "auto",
) -> None:
self.app = app
self.allowed_hosts = tuple(allowed_hosts or ())
self.allowed_origins = tuple(allowed_origins or ())
self.mode = mode
self.has_explicit_allowed_hosts = allowed_hosts is not None
self.has_explicit_allowed_origins = allowed_origins is not None
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
if scope["type"] != "http":
await self.app(scope, receive, send)
return
allowed_hosts = self._allowed_hosts_for_scope(scope)
headers = Headers(scope=scope)
host = headers.get("host", "")
if not _host_matches(host, allowed_hosts):
if self._should_validate_host(scope) and not _host_matches(
host,
self._allowed_hosts_for_scope(scope),
):
response = Response("Misdirected Request", status_code=421)
await response(scope, receive, send)
return
origin = headers.get("origin")
request_origin = _request_origin(scope, host)
if origin and not self._origin_allowed(origin, request_origin, host):
if (
origin
and self._should_validate_origin(scope, host)
and not self._origin_allowed(
origin,
request_origin,
host,
allow_same_origin_fallback=self._allow_same_origin_fallback(
scope,
host,
),
)
):
response = Response("Forbidden Origin", status_code=403)
await response(scope, receive, send)
return
await self.app(scope, receive, send)
def _should_validate_host(self, scope: Scope) -> bool:
if self.mode == "strict" or self.has_explicit_allowed_hosts:
return True
server = scope.get("server")
return bool(server and _is_loopback_host(server[0]))
def _should_validate_origin(self, scope: Scope, host: str) -> bool:
if (
self.mode == "strict"
or self.has_explicit_allowed_hosts
or self.has_explicit_allowed_origins
or _is_loopback_host(host)
):
return True
server = scope.get("server")
return bool(server and _is_loopback_host(server[0]))
def _allow_same_origin_fallback(self, scope: Scope, host: str) -> bool:
if not self.has_explicit_allowed_origins:
return True
if self.mode == "strict" or self.has_explicit_allowed_hosts:
return True
server = scope.get("server")
return _is_loopback_host(host) or bool(server and _is_loopback_host(server[0]))
def _allowed_hosts_for_scope(self, scope: Scope) -> tuple[str, ...]:
allowed_hosts = list(DEFAULT_HOSTS)
allowed_hosts.extend(self.allowed_hosts)
@ -268,10 +317,19 @@ class HostOriginGuardMiddleware:
return tuple(allowed_hosts)
def _origin_allowed(self, origin: str, request_origin: str, host: str) -> bool:
def _origin_allowed(
self,
origin: str,
request_origin: str,
host: str,
allow_same_origin_fallback: bool,
) -> bool:
if _origin_matches(origin, self.allowed_origins):
return True
if not allow_same_origin_fallback:
return False
origin_host = _origin_host(origin)
if _is_loopback_host(origin_host) and _is_loopback_host(host):
return True
@ -491,7 +549,7 @@ def create_streamable_http_app(
debug: bool = False,
routes: list[BaseRoute] | None = None,
middleware: list[Middleware] | None = None,
host_origin_protection: bool = True,
host_origin_protection: HostOriginProtection = False,
allowed_hosts: Sequence[str] | None = None,
allowed_origins: Sequence[str] | None = None,
) -> StarletteWithLifespan:
@ -511,7 +569,9 @@ def create_streamable_http_app(
routes: Optional list of custom routes
middleware: Optional list of middleware
host_origin_protection: Whether to validate Host and Origin headers
before requests reach the MCP endpoint.
before requests reach the MCP endpoint. Defaults to False for
compatibility. "auto" protects localhost-bound servers and explicit
host/origin allowlists.
allowed_hosts: Additional hostnames that may appear in the Host header.
allowed_origins: Additional browser origins trusted by the request guard.
Configure CORS separately when browser JavaScript must read
@ -576,13 +636,17 @@ def create_streamable_http_app(
server_routes.extend(server._get_additional_http_routes())
# Add middleware
if host_origin_protection:
if host_origin_protection not in (True, False, "auto"):
raise ValueError("host_origin_protection must be True, False, or 'auto'.")
if host_origin_protection is not False:
server_middleware.insert(
0,
Middleware(
HostOriginGuardMiddleware,
allowed_hosts=allowed_hosts,
allowed_origins=allowed_origins,
mode="strict" if host_origin_protection is True else "auto",
),
)
if middleware:

View file

@ -19,7 +19,9 @@ from starlette.routing import BaseRoute, Route
import fastmcp
from fastmcp.server.event_store import EventStore
from fastmcp.server.http import (
HostOriginProtection,
StarletteWithLifespan,
_is_loopback_host,
create_sse_app,
create_streamable_http_app,
)
@ -48,6 +50,22 @@ def _format_host_for_url(host: str) -> str:
return host
def _resolve_allowed_hosts_for_run(
*,
host: str,
host_origin_protection: HostOriginProtection,
allowed_hosts: list[str] | None,
configured_allowed_hosts: list[str] | None,
) -> list[str] | None:
if allowed_hosts is not None:
return allowed_hosts
if host_origin_protection == "auto" and _is_loopback_host(host):
return [*(configured_allowed_hosts or []), host]
return configured_allowed_hosts
class TransportMixin:
"""Mixin providing transport-related methods for FastMCP.
@ -250,7 +268,7 @@ class TransportMixin:
json_response: bool | None = None,
stateless_http: bool | None = None,
stateless: bool | None = None,
host_origin_protection: bool | None = None,
host_origin_protection: HostOriginProtection | None = None,
allowed_hosts: list[str] | None = None,
allowed_origins: list[str] | None = None,
sockets: list[socket.socket] | None = None,
@ -269,7 +287,9 @@ class TransportMixin:
stateless_http: Whether to use stateless HTTP (defaults to settings.stateless_http)
stateless: Alias for stateless_http for CLI consistency
host_origin_protection: Whether to validate Host and Origin headers
before requests reach the MCP endpoint.
before requests reach the MCP endpoint. Defaults to
settings.http_host_origin_protection. "auto" protects
localhost-bound servers and explicit host/origin allowlists.
allowed_hosts: Additional hostnames that may appear in the Host header.
allowed_origins: Additional browser origins trusted by the request guard.
Configure CORS separately when browser JavaScript must read
@ -290,6 +310,17 @@ class TransportMixin:
host = host if host is not None else fastmcp.settings.host
port = port if port is not None else fastmcp.settings.port
resolved_host_origin_protection = (
host_origin_protection
if host_origin_protection is not None
else fastmcp.settings.http_host_origin_protection
)
resolved_allowed_hosts = _resolve_allowed_hosts_for_run(
host=host,
host_origin_protection=resolved_host_origin_protection,
allowed_hosts=allowed_hosts,
configured_allowed_hosts=fastmcp.settings.http_allowed_hosts,
)
default_log_level_to_use = (
log_level if log_level is not None else fastmcp.settings.log_level
).lower()
@ -300,8 +331,8 @@ class TransportMixin:
middleware=middleware,
json_response=json_response,
stateless_http=stateless_http,
host_origin_protection=host_origin_protection,
allowed_hosts=allowed_hosts,
host_origin_protection=resolved_host_origin_protection,
allowed_hosts=resolved_allowed_hosts,
allowed_origins=allowed_origins,
)
@ -345,7 +376,7 @@ class TransportMixin:
transport: Literal["http", "streamable-http", "sse"] = "http",
event_store: EventStore | None = None,
retry_interval: int | None = None,
host_origin_protection: bool | None = None,
host_origin_protection: HostOriginProtection | None = None,
allowed_hosts: list[str] | None = None,
allowed_origins: list[str] | None = None,
) -> StarletteWithLifespan:
@ -365,7 +396,9 @@ class TransportMixin:
disconnections. Requires event_store to be set. Only used with
streamable-http transport.
host_origin_protection: Whether to validate Host and Origin headers
before requests reach the MCP endpoint.
before requests reach the MCP endpoint. Defaults to
settings.http_host_origin_protection. "auto" protects
localhost-bound servers and explicit host/origin allowlists.
allowed_hosts: Additional hostnames that may appear in the Host header.
allowed_origins: Additional browser origins trusted by the request guard.
Configure CORS separately when browser JavaScript must read

View file

@ -310,6 +310,26 @@ class Settings(BaseSettings):
),
] = False
ssrf_trust_proxy: Annotated[
bool,
Field(
description=inspect.cleandoc(
"""
Trust an outbound HTTP proxy for SSRF-protected fetches (OAuth client
metadata and JWKS). When False (default), FastMCP resolves the target
hostname itself and refuses to connect if it maps to a private,
loopback, link-local, or otherwise reserved IP. When True, FastMCP
routes auth metadata and JWKS fetches through the configured
HTTPS_PROXY/ALL_PROXY and does not honor NO_PROXY; if no proxy is
configured the fetch is refused (raising SSRFError) rather than sent
direct with the blocklist disabled. Only enable this when a trusted
corporate proxy is the mandated egress path: it shifts SSRF trust to
that proxy. Scheme (HTTPS-only) and hostname checks still apply.
"""
),
),
] = False
server_dependencies: list[str] = Field(
default_factory=list,
description="List of dependencies to install in the server environment",
@ -320,7 +340,7 @@ class Settings(BaseSettings):
stateless_http: bool = (
False # If True, uses true stateless mode (new transport per request)
)
http_host_origin_protection: bool = True
http_host_origin_protection: bool | Literal["auto"] = False
http_allowed_hosts: list[str] | None = None
http_allowed_origins: list[str] | None = None

View file

@ -718,7 +718,8 @@ class TransformedTool(Tool):
schema = {
"type": "object",
"properties": new_props,
"required": list(new_required),
# Iterate props (not the set) for deterministic ordering
"required": [p for p in new_props if p in new_required],
"additionalProperties": False,
}
@ -910,7 +911,11 @@ class TransformedTool(Tool):
result = {
"type": "object",
"properties": merged_props,
"required": list(final_required),
# Iterate props (not the set) for deterministic ordering; keep any
# required names not present in properties (sorted) rather than
# silently dropping them.
"required": [p for p in merged_props if p in final_required]
+ sorted(final_required - set(merged_props)),
"additionalProperties": False,
}

View file

@ -1,12 +1,45 @@
from __future__ import annotations
import copy
from collections import defaultdict
from typing import Any
from jsonref import JsonRefError, replace_refs
def _copy_schema(schema: dict[str, Any]) -> dict[str, Any]:
"""Return a deep copy of a JSON schema without recursing.
`copy.deepcopy` consumes stack frames in proportion to nesting depth, so a
deeply nested schema raises `RecursionError` before the traversals in this
module can apply their own depth guards turning a schema that used to
compress into one that fails outright. Schemas are plain JSON, so an
explicit stack copies the containers at any depth and shares the immutable
scalars at the leaves.
"""
root: dict[str, Any] = {}
stack: list[tuple[Any, Any]] = [(schema, root)]
while stack:
source, target = stack.pop()
if isinstance(source, dict):
pairs: list[tuple[Any, Any]] = list(source.items())
else:
pairs = list(enumerate(source))
for key, value in pairs:
if isinstance(value, dict):
child: Any = {}
stack.append((value, child))
elif isinstance(value, list):
child = [None] * len(value)
stack.append((value, child))
else:
child = value
target[key] = child
return root
def _defs_have_cycles(defs: dict[str, Any]) -> bool:
"""Check whether any definitions in ``$defs`` form a reference cycle.
@ -348,7 +381,7 @@ def _prune_param(schema: dict[str, Any], param: str) -> dict[str, Any]:
"""Return a new schema with *param* removed from `properties`, `required`,
and (if no longer referenced) `$defs`.
"""
schema = copy.deepcopy(schema)
schema = _copy_schema(schema)
# ── 1. drop from properties/required ──────────────────────────────
props = schema.get("properties", {})
@ -498,6 +531,11 @@ def _single_pass_optimize(
if not (prune_defs or prune_titles or prune_additional_properties):
return schema # Nothing to do
# Work on a copy so the caller's schema is never mutated (see docstring). The
# pruning phases below pop keys/$defs in place, which would otherwise corrupt a
# shared dict such as a live Tool.input_schema passed straight to compress_schema.
schema = _copy_schema(schema)
# Phase 1: Collect references and apply simple cleanups
# Track which $defs are referenced from the main schema and from other $defs
root_refs: set[str] = set() # $defs referenced directly from main schema
@ -506,6 +544,11 @@ def _single_pass_optimize(
) # def A references def B
defs = schema.get("$defs")
# Set when the traversal below gives up at its depth limit. Once that
# happens the reference scan is incomplete, so we can no longer tell which
# definitions are genuinely unused.
reference_scan_truncated = False
def traverse_and_clean(
node: object,
current_def_name: str | None = None,
@ -523,7 +566,10 @@ def _single_pass_optimize(
about) but we skip all cleanups so we don't mutate user data that
happens to look metadata-shaped.
"""
nonlocal reference_scan_truncated
if depth > 50: # Prevent infinite recursion
reference_scan_truncated = True
return
if isinstance(node, dict):
@ -647,6 +693,13 @@ def _single_pass_optimize(
for def_name, def_schema in defs.items():
traverse_and_clean(def_schema, current_def_name=def_name, in_schema=True)
# An incomplete scan has not seen every $ref, so a definition that looks
# unused may simply be referenced below the cutoff. Keeping an unused
# definition is harmless; dropping a referenced one leaves a dangling
# $ref and an invalid schema.
if reference_scan_truncated:
return schema
# Phase 4: Remove unused definitions
def is_def_used(def_name: str, visiting: set[str] | None = None) -> bool:
"""Check if a definition is used, handling circular references."""

View file

@ -313,13 +313,13 @@ class RequestDirector:
if not value:
continue
if explode:
# form,explode=true on objects: each property becomes
# a separate query parameter.
# e.g. {"R": 100, "G": 200} → R=100&G=200
for k, v in value.items():
serialized[_query_scalar_to_str(k)] = _query_scalar_to_str(
v
)
# deepObject keeps the parent parameter name;
# form style emits each property as a bare key.
property_name = _query_scalar_to_str(k)
if param_info.style == "deepObject":
property_name = f"{key}[{property_name}]"
serialized[property_name] = _query_scalar_to_str(v)
else:
style = param_info.style or "form"
delimiter = self._STYLE_DELIMITERS.get(style, ",")

View file

@ -3,3 +3,4 @@ server:
- server-sse-polling
- resources-subscribe
- resources-unsubscribe
- dns-rebinding-protection

View file

@ -445,6 +445,25 @@ class TestAzureProvider:
assert "Mail.Read" in upstream_url
assert "User.Read" in upstream_url
def test_prepare_scopes_for_token_exchange_falls_back_to_required_scopes(
self, memory_storage: MemoryStore
):
"""Clients may omit scope; Azure still needs an API scope with offline_access."""
provider = AzureProvider(
client_id="test_client",
client_secret="test_secret",
tenant_id="test-tenant",
base_url="https://myserver.com",
identifier_uri="api://my-api",
required_scopes=["read"],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
result = provider._prepare_scopes_for_token_exchange([])
assert result == ["api://my-api/read", "offline_access"]
def test_base_authority_defaults_to_public_cloud(self, memory_storage: MemoryStore):
"""Test that base_authority defaults to login.microsoftonline.com."""
provider = AzureProvider(
@ -724,10 +743,10 @@ class TestAzureProvider:
"https://graph.microsoft.com/.default" in result
) # Not prefixed (contains ://)
def test_prepare_scopes_for_upstream_refresh_empty_scopes(
def test_prepare_scopes_for_upstream_refresh_empty_scopes_falls_back_to_required(
self, memory_storage: MemoryStore
):
"""Test behavior with empty scopes list."""
"""Clients may omit scope; refresh should still request the configured API scope."""
provider = AzureProvider(
client_id="test_client",
client_secret="test_secret",
@ -740,13 +759,13 @@ class TestAzureProvider:
client_storage=memory_storage,
)
# Empty scopes should still add OIDC scopes (not User.Read)
result = provider._prepare_scopes_for_upstream_refresh([])
assert "api://my-api/read" in result
assert "User.Read" not in result # Not OIDC
assert "openid" in result
assert "offline_access" in result # Auto-included
assert len(result) == 2 # Only OIDC scopes: openid + offline_access
assert len(result) == 3
def test_prepare_scopes_for_upstream_refresh_no_additional_scopes(
self, memory_storage: MemoryStore

View file

@ -0,0 +1,236 @@
"""Tests for Hugging Face OAuth provider."""
import re
import pytest
from key_value.aio.stores.memory import MemoryStore
from pytest_httpx import HTTPXMock
from fastmcp.server.auth.providers.huggingface import (
DEFAULT_HUGGINGFACE_SCOPES,
HUGGINGFACE_AUTHORIZATION_ENDPOINT,
HUGGINGFACE_TOKEN_ENDPOINT,
HUGGINGFACE_USERINFO_ENDPOINT,
HUGGINGFACE_WHOAMI_ENDPOINT,
HuggingFaceProvider,
HuggingFaceTokenVerifier,
)
@pytest.fixture
def memory_storage() -> MemoryStore:
"""Provide a MemoryStore for tests to avoid SQLite initialization on Windows."""
return MemoryStore()
_USERINFO_RE = re.compile(re.escape(HUGGINGFACE_USERINFO_ENDPOINT))
_WHOAMI_RE = re.compile(re.escape(HUGGINGFACE_WHOAMI_ENDPOINT))
class TestHuggingFaceProvider:
"""Test HuggingFaceProvider functionality."""
def test_init_with_explicit_params(self, memory_storage: MemoryStore):
provider = HuggingFaceProvider(
client_id="hf-client-id",
client_secret="hf-client-secret",
base_url="https://myserver.com",
required_scopes=["openid", "profile", "inference-api"],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
assert provider._upstream_client_id == "hf-client-id"
assert provider._upstream_client_secret is not None
assert provider._upstream_client_secret.get_secret_value() == "hf-client-secret"
assert str(provider.base_url) == "https://myserver.com/"
def test_init_defaults(self, memory_storage: MemoryStore):
provider = HuggingFaceProvider(
client_id="hf-client-id",
client_secret="hf-client-secret",
base_url="https://myserver.com",
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
assert provider._redirect_path == "/auth/callback"
assert provider.required_scopes == DEFAULT_HUGGINGFACE_SCOPES
assert provider._token_validator.required_scopes == []
def test_oauth_endpoints_configured_correctly(self, memory_storage: MemoryStore):
provider = HuggingFaceProvider(
client_id="hf-client-id",
client_secret="hf-client-secret",
base_url="https://myserver.com",
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
assert (
provider._upstream_authorization_endpoint
== HUGGINGFACE_AUTHORIZATION_ENDPOINT
)
assert provider._upstream_token_endpoint == HUGGINGFACE_TOKEN_ENDPOINT
assert provider._upstream_revocation_endpoint is None
def test_public_pkce_app_uses_none_token_auth(self, memory_storage: MemoryStore):
provider = HuggingFaceProvider(
client_id="https://client.example.com/.well-known/oauth-cimd",
base_url="https://myserver.com",
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
assert provider._upstream_client_secret is None
assert provider._token_endpoint_auth_method == "none"
def test_uses_upstream_token_response_scopes(self, memory_storage: MemoryStore):
provider = HuggingFaceProvider(
client_id="hf-client-id",
client_secret="hf-client-secret",
base_url="https://myserver.com",
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
assert provider._uses_alternate_verification() is True
def test_valid_scopes_passed_through(self, memory_storage: MemoryStore):
provider = HuggingFaceProvider(
client_id="hf-client-id",
client_secret="hf-client-secret",
base_url="https://myserver.com",
required_scopes=["openid", "profile"],
valid_scopes=["openid", "profile", "inference-api", "jobs"],
jwt_signing_key="test-secret",
client_storage=memory_storage,
)
reg_options = provider.client_registration_options
assert reg_options is not None
assert reg_options.valid_scopes == [
"openid",
"profile",
"inference-api",
"jobs",
]
class TestHuggingFaceTokenVerifier:
"""Test HuggingFaceTokenVerifier.verify_token()."""
async def test_valid_token_with_userinfo_scopes(self, httpx_mock: HTTPXMock):
httpx_mock.add_response(
url=_USERINFO_RE,
json={
"sub": "user-123",
"preferred_username": "alice",
"name": "Alice",
"email": "alice@example.com",
"email_verified": True,
"picture": "https://huggingface.co/alice.png",
"scope": "openid profile email",
},
)
verifier = HuggingFaceTokenVerifier(required_scopes=["openid", "email"])
result = await verifier.verify_token("hf_oauth_token")
assert result is not None
assert result.client_id == "user-123"
assert result.scopes == ["openid", "profile", "email"]
assert result.claims["sub"] == "user-123"
assert result.claims["preferred_username"] == "alice"
assert result.claims["email"] == "alice@example.com"
async def test_valid_token_with_whoami_scopes(self, httpx_mock: HTTPXMock):
httpx_mock.add_response(
url=_USERINFO_RE,
json={
"sub": "user-123",
"preferred_username": "alice",
"scope": "openid profile",
},
)
httpx_mock.add_response(
url=_WHOAMI_RE,
json={
"name": "alice",
"auth": {
"accessToken": {"scopes": ["openid", "profile", "inference-api"]}
},
},
)
verifier = HuggingFaceTokenVerifier(required_scopes=["openid", "inference-api"])
result = await verifier.verify_token("hf_oauth_token")
assert result is not None
assert result.scopes == ["openid", "profile", "inference-api"]
assert result.claims["huggingface_whoami"] is not None
async def test_defaults_scopes_when_userinfo_has_no_scope(
self, httpx_mock: HTTPXMock
):
httpx_mock.add_response(
url=_USERINFO_RE,
json={"sub": "user-123", "preferred_username": "alice"},
)
httpx_mock.add_response(url=_WHOAMI_RE, status_code=404)
verifier = HuggingFaceTokenVerifier()
result = await verifier.verify_token("hf_oauth_token")
assert result is not None
assert result.scopes == DEFAULT_HUGGINGFACE_SCOPES
async def test_missing_required_scope_returns_none(self, httpx_mock: HTTPXMock):
httpx_mock.add_response(
url=_USERINFO_RE,
json={
"sub": "user-123",
"scope": "openid profile",
},
)
httpx_mock.add_response(url=_WHOAMI_RE, json={"name": "alice"})
verifier = HuggingFaceTokenVerifier(required_scopes=["inference-api"])
result = await verifier.verify_token("hf_oauth_token")
assert result is None
async def test_invalid_token_returns_none(self, httpx_mock: HTTPXMock):
httpx_mock.add_response(
url=_USERINFO_RE,
status_code=401,
json={"error": "invalid_token"},
)
verifier = HuggingFaceTokenVerifier()
result = await verifier.verify_token("invalid")
assert result is None
async def test_missing_sub_returns_none(self, httpx_mock: HTTPXMock):
httpx_mock.add_response(
url=_USERINFO_RE,
json={"preferred_username": "alice"},
)
verifier = HuggingFaceTokenVerifier()
result = await verifier.verify_token("token-without-sub")
assert result is None
async def test_sends_bearer_token_to_userinfo(self, httpx_mock: HTTPXMock):
httpx_mock.add_response(
url=_USERINFO_RE,
json={"sub": "user-123", "scope": "openid profile"},
)
verifier = HuggingFaceTokenVerifier()
await verifier.verify_token("hf_oauth_token")
request = httpx_mock.get_requests()[0]
assert request.headers["Authorization"] == "Bearer hf_oauth_token"

View file

@ -1,6 +1,6 @@
import time
from collections.abc import AsyncGenerator
from typing import Any
from typing import Any, cast
from unittest.mock import patch
import pytest
@ -559,6 +559,85 @@ class TestBearerTokenJWKS:
assert access_token.claims.get("iss") == issuer
assert access_token.claims.get("aud") == audience
async def test_jwks_skips_unsupported_key_types(
self,
rsa_key_pair: RSAKeyPair,
jwks_provider: JWTVerifier,
mock_jwks_data: JWKSData,
httpx_mock: HTTPXMock,
mock_dns,
):
"""An unsupported key type in the JWKS (e.g. OKP/Ed25519) must be
skipped, not poison the whole key set - #4515.
Some authorization servers (e.g. Rauthy, Ory Hydra) publish an
Ed25519 key alongside RSA keys; tokens signed by the RSA keys must
still verify.
"""
okp_key = cast(
"JWKData",
{
"kty": "OKP",
"crv": "Ed25519",
"kid": "ed25519-key",
"alg": "EdDSA",
"use": "sig",
"x": "11qYAYKxCrfVS_7TyWQHOg7hcvPapiMlrwIaaPcHURo",
},
)
mock_jwks_data["keys"][0]["kid"] = "test-key-1"
# Unsupported key FIRST, so an unguarded conversion loop would
# abort before reaching the RSA key the token needs
mock_jwks_data["keys"].insert(0, okp_key)
httpx_mock.add_response(json=mock_jwks_data)
token = rsa_key_pair.create_token(
subject="test-user",
issuer="https://test.example.com",
audience="https://api.example.com",
kid="test-key-1",
)
access_token = await jwks_provider.load_access_token(token)
assert access_token is not None
assert access_token.client_id == "test-user"
async def test_jwks_with_only_unsupported_keys_rejects_cleanly(
self,
rsa_key_pair: RSAKeyPair,
jwks_provider: JWTVerifier,
httpx_mock: HTTPXMock,
mock_dns,
):
"""If every key in the JWKS is unsupported, verification fails
cleanly (returns None) rather than crashing - #4515."""
okp_only = {
"keys": [
cast(
"JWKData",
{
"kty": "OKP",
"crv": "Ed25519",
"kid": "ed25519-key",
"alg": "EdDSA",
"use": "sig",
"x": "11qYAYKxCrfVS_7TyWQHOg7hcvPapiMlrwIaaPcHURo",
},
)
]
}
httpx_mock.add_response(json=okp_only)
token = rsa_key_pair.create_token(
subject="test-user",
issuer="https://test.example.com",
audience="https://api.example.com",
kid="ed25519-key",
)
access_token = await jwks_provider.load_access_token(token)
assert access_token is None
async def test_jwks_token_validation_with_invalid_key(
self,
rsa_key_pair: RSAKeyPair,

View file

@ -3,18 +3,56 @@
This module tests the ssrf.py module which provides SSRF-protected HTTP fetching.
"""
import socket
from unittest.mock import AsyncMock, MagicMock, patch
import httpx
import pytest
import fastmcp
from fastmcp.server.auth.ssrf import (
SSRFError,
SSRFFetchError,
is_ip_allowed,
ssrf_safe_fetch,
ssrf_safe_fetch_response,
validate_url,
)
from fastmcp.utilities.tests import temporary_settings
def _mock_httpx_client(
*,
status_code: int = 200,
headers: dict[str, str] | None = None,
body_chunks: list[bytes] | None = None,
) -> AsyncMock:
"""Build a mock httpx.AsyncClient whose stream() yields a canned response.
The returned client's ``.stream.call_args`` exposes the request that was made.
"""
if headers is None:
headers = {"content-length": "2"}
if body_chunks is None:
body_chunks = [b"ok"]
mock_stream = MagicMock()
mock_stream.status_code = status_code
mock_stream.headers = headers
mock_stream.__aenter__ = AsyncMock(return_value=mock_stream)
mock_stream.__aexit__ = AsyncMock(return_value=None)
async def aiter_bytes():
for chunk in body_chunks:
yield chunk
mock_stream.aiter_bytes = aiter_bytes
mock_client = AsyncMock()
mock_client.stream = MagicMock(return_value=mock_stream)
mock_client.__aenter__.return_value = mock_client
mock_client.__aexit__ = AsyncMock(return_value=None)
return mock_client
class TestIsIPAllowed:
@ -506,3 +544,317 @@ class TestStreamingResponseSizeLimit:
with pytest.raises(SSRFFetchError, match="too large"):
await ssrf_safe_fetch("https://example.com/api", max_size=5120)
class TestProxyMode:
"""Tests for FASTMCP_SSRF_TRUST_PROXY (proxy trust) mode.
In proxy mode FastMCP skips its own DNS resolution and IP blocklist. Rather than
predicting whether the HTTP client would route a request through a proxy -- a strategy
that broke three times chasing different NO_PROXY forms (port-qualified, IPv6,
scheme-qualified) -- it reads the proxy URL directly from the environment and
hands it to httpx explicitly, so the request is provably routed through that
proxy rather than predicted to be. NO_PROXY is therefore not evaluated, while
trust_env remains enabled for environment-provided CA trust. The scheme (HTTPS)
and host checks still apply.
"""
@pytest.fixture(autouse=True)
def _clear_proxy_env(self, monkeypatch):
"""Start every test from a clean slate for both spellings of every proxy
variable, so a proxy inherited from the host/CI environment (or left behind
by another test) can't leak in and make behavior non-deterministic."""
for name in (
"HTTP_PROXY",
"http_proxy",
"HTTPS_PROXY",
"https_proxy",
"ALL_PROXY",
"all_proxy",
"NO_PROXY",
"no_proxy",
):
monkeypatch.delenv(name, raising=False)
def test_flag_defaults_to_false(self):
"""The trust-proxy flag must be off by default (no silent weakening)."""
assert fastmcp.settings.ssrf_trust_proxy is False
async def test_validate_url_skips_resolution_and_blocklist(self, monkeypatch):
"""Proxy mode returns resolved_ips=[] without resolving or blocklisting, and
carries the configured proxy URL for the fetch to use."""
monkeypatch.setenv("HTTPS_PROXY", "http://proxy.internal:3128")
with (
temporary_settings(ssrf_trust_proxy=True),
patch("fastmcp.server.auth.ssrf.resolve_hostname") as mock_resolve,
patch("fastmcp.server.auth.ssrf.is_ip_allowed") as mock_blocklist,
):
result = await validate_url("https://example.com/path")
assert result.resolved_ips == []
assert result.original_url == "https://example.com/path"
assert result.hostname == "example.com"
assert result.proxy_url == "http://proxy.internal:3128"
mock_resolve.assert_not_called()
mock_blocklist.assert_not_called()
async def test_validate_url_still_rejects_http(self, monkeypatch):
"""Proxy mode keeps the HTTPS-only scheme check."""
monkeypatch.setenv("HTTPS_PROXY", "http://proxy.internal:3128")
with temporary_settings(ssrf_trust_proxy=True):
with pytest.raises(SSRFError, match="must use HTTPS"):
await validate_url("http://example.com/path")
async def test_validate_url_still_rejects_missing_host(self, monkeypatch):
"""Proxy mode keeps the host check."""
monkeypatch.setenv("HTTPS_PROXY", "http://proxy.internal:3128")
with temporary_settings(ssrf_trust_proxy=True):
with pytest.raises(SSRFError, match="must have a host"):
await validate_url("https:///path")
async def test_validate_url_still_enforces_require_path(self, monkeypatch):
"""Proxy mode keeps the require_path check (CIMD)."""
monkeypatch.setenv("HTTPS_PROXY", "http://proxy.internal:3128")
with temporary_settings(ssrf_trust_proxy=True):
with pytest.raises(SSRFError, match="non-root path"):
await validate_url("https://example.com/", require_path=True)
async def test_raises_when_no_proxy_is_configured(self):
"""No proxy in the environment → refuse rather than fetch unprotected."""
with temporary_settings(ssrf_trust_proxy=True):
with pytest.raises(SSRFError, match="no HTTPS_PROXY/ALL_PROXY"):
await validate_url("https://example.com/path")
async def test_fetch_refuses_end_to_end_when_no_proxy_configured(self):
"""The refusal surfaces through ssrf_safe_fetch: no client is ever built.
The whole point of the hard failure is that the *fetch* cannot proceed, so
this drives it through the public entrypoint and asserts no httpx client is
ever constructed the request never leaves the process with the blocklist
disabled.
"""
with (
temporary_settings(ssrf_trust_proxy=True),
patch("httpx.AsyncClient") as mock_client_class,
):
with pytest.raises(SSRFError, match="no HTTPS_PROXY/ALL_PROXY"):
await ssrf_safe_fetch("https://example.com/api")
mock_client_class.assert_not_called()
async def test_https_proxy_used_explicitly(self, monkeypatch):
"""HTTPS_PROXY is passed to httpx explicitly while CA environment handling
remains enabled, and a single request goes to the original hostname URL not
an IP literal.
This is the property the whole redesign rests on: an explicit proxy= fixes
httpx's proxy map without consulting NO_PROXY, while trust_env=True preserves
SSL_CERT_FILE and SSL_CERT_DIR handling.
"""
monkeypatch.setenv("HTTPS_PROXY", "http://proxy.internal:3128")
monkeypatch.setenv("SSL_CERT_FILE", "/corporate-ca.pem")
mock_client = _mock_httpx_client()
with (
temporary_settings(ssrf_trust_proxy=True),
patch("fastmcp.server.auth.ssrf.resolve_hostname") as mock_resolve,
patch("httpx.AsyncClient", return_value=mock_client) as mock_client_class,
):
content = await ssrf_safe_fetch("https://example.com/api")
assert content == b"ok"
mock_resolve.assert_not_called()
client_kwargs = mock_client_class.call_args[1]
assert client_kwargs["proxy"] == "http://proxy.internal:3128"
assert client_kwargs["trust_env"] is True
# A single request to the original hostname URL — not an IP literal.
assert mock_client.stream.call_count == 1
url_called = mock_client.stream.call_args[0][1]
assert url_called == "https://example.com/api"
# No Host override and no SNI override — the client derives both from the URL.
call_kwargs = mock_client.stream.call_args[1]
assert "Host" not in call_kwargs["headers"]
assert call_kwargs["extensions"] == {}
# Redirects stay disabled and TLS verification stays on.
assert client_kwargs["follow_redirects"] is False
assert client_kwargs["verify"] is True
async def test_all_proxy_used_as_fallback(self, monkeypatch):
"""ALL_PROXY routes the fetch when HTTPS_PROXY is not set."""
monkeypatch.setenv("ALL_PROXY", "http://all-proxy.internal:3128")
mock_client = _mock_httpx_client()
with (
temporary_settings(ssrf_trust_proxy=True),
patch("fastmcp.server.auth.ssrf.resolve_hostname"),
patch("httpx.AsyncClient", return_value=mock_client) as mock_client_class,
):
content = await ssrf_safe_fetch("https://example.com/api")
assert content == b"ok"
client_kwargs = mock_client_class.call_args[1]
assert client_kwargs["proxy"] == "http://all-proxy.internal:3128"
assert client_kwargs["trust_env"] is True
async def test_https_proxy_preferred_over_all_proxy(self, monkeypatch):
"""When both are set, HTTPS_PROXY takes priority."""
monkeypatch.setenv("HTTPS_PROXY", "http://https-proxy.internal:3128")
monkeypatch.setenv("ALL_PROXY", "http://all-proxy.internal:3128")
mock_client = _mock_httpx_client()
with (
temporary_settings(ssrf_trust_proxy=True),
patch("fastmcp.server.auth.ssrf.resolve_hostname"),
patch("httpx.AsyncClient", return_value=mock_client) as mock_client_class,
):
await ssrf_safe_fetch("https://example.com/api")
proxy_used = mock_client_class.call_args[1]["proxy"]
assert proxy_used == "http://https-proxy.internal:3128"
async def test_no_proxy_is_not_honored(self, monkeypatch):
"""Documents the behavior change: a NO_PROXY entry that would previously have
matched the target host no longer excludes it. The fetch still proceeds
through the configured proxy rather than being refused, because routing a
NO_PROXY'd host through the proxy is strictly safer than the alternative —
fetching it direct with the IP blocklist already disabled.
"""
monkeypatch.setenv("HTTPS_PROXY", "http://proxy.internal:3128")
monkeypatch.setenv("NO_PROXY", "example.com")
mock_client = _mock_httpx_client()
with (
temporary_settings(ssrf_trust_proxy=True),
patch("fastmcp.server.auth.ssrf.resolve_hostname"),
patch("httpx.AsyncClient", return_value=mock_client) as mock_client_class,
):
content = await ssrf_safe_fetch("https://example.com/api")
assert content == b"ok"
client_kwargs = mock_client_class.call_args[1]
assert client_kwargs["proxy"] == "http://proxy.internal:3128"
assert client_kwargs["trust_env"] is True
async def test_fetch_preserves_request_headers_but_drops_host(self, monkeypatch):
"""Caller headers pass through, but a caller-supplied Host is dropped."""
monkeypatch.setenv("HTTPS_PROXY", "http://proxy.internal:3128")
mock_client = _mock_httpx_client()
with (
temporary_settings(ssrf_trust_proxy=True),
patch("fastmcp.server.auth.ssrf.resolve_hostname"),
patch("httpx.AsyncClient", return_value=mock_client),
):
await ssrf_safe_fetch_response(
"https://example.com/api",
request_headers={"If-None-Match": "etag", "Host": "evil.example"},
)
sent_headers = mock_client.stream.call_args[1]["headers"]
assert sent_headers["If-None-Match"] == "etag"
assert "Host" not in sent_headers
async def test_fetch_size_limit_preserved(self, monkeypatch):
"""Proxy mode still enforces the response size limit during streaming."""
monkeypatch.setenv("HTTPS_PROXY", "http://proxy.internal:3128")
big_chunks = [b"x" * 1024 for _ in range(10)]
mock_client = _mock_httpx_client(headers={}, body_chunks=big_chunks)
with (
temporary_settings(ssrf_trust_proxy=True),
patch("fastmcp.server.auth.ssrf.resolve_hostname"),
patch("httpx.AsyncClient", return_value=mock_client),
):
with pytest.raises(SSRFFetchError, match="too large"):
await ssrf_safe_fetch("https://example.com/api", max_size=5120)
async def test_fetch_status_check_preserved(self, monkeypatch):
"""Proxy mode still rejects non-allowed status codes."""
monkeypatch.setenv("HTTPS_PROXY", "http://proxy.internal:3128")
mock_client = _mock_httpx_client(status_code=404, body_chunks=[b"no"])
with (
temporary_settings(ssrf_trust_proxy=True),
patch("fastmcp.server.auth.ssrf.resolve_hostname"),
patch("httpx.AsyncClient", return_value=mock_client),
):
with pytest.raises(SSRFFetchError, match="HTTP 404"):
await ssrf_safe_fetch("https://example.com/api")
async def test_gaierror_repro_succeeds_through_proxy(self, monkeypatch):
"""Reproduces issue #4292: on a host with no external DNS at all (every
getaddrinfo() call raises gaierror), the OAuth/JWKS fetch still succeeds in
proxy-trust mode, because DNS resolution is never attempted only HTTPS_PROXY
is read and the proxy resolves the target. This is the reporter's exact
failure mode, and the strongest proof the redesign closes the issue: unlike
other tests in this class, resolve_hostname itself is *not* mocked, so if
proxy mode ever regressed into calling it, this test would fail with SSRFError
instead of succeeding.
"""
monkeypatch.setenv("HTTPS_PROXY", "http://proxy.internal:3128")
def _no_dns(*args, **kwargs):
raise socket.gaierror("Name or service not known")
monkeypatch.setattr(socket, "getaddrinfo", _no_dns)
mock_client = _mock_httpx_client()
with (
temporary_settings(ssrf_trust_proxy=True),
patch("httpx.AsyncClient", return_value=mock_client) as mock_client_class,
):
content = await ssrf_safe_fetch("https://example.com/api")
assert content == b"ok"
client_kwargs = mock_client_class.call_args[1]
assert client_kwargs["proxy"] == "http://proxy.internal:3128"
assert client_kwargs["trust_env"] is True
assert mock_client.stream.call_args[0][1] == "https://example.com/api"
async def test_default_mode_still_resolves_and_pins(self):
"""Regression: with the flag off, resolution + blocklist + IP pinning still
apply, and no explicit proxy is passed to the client."""
resolved_ip = "93.184.216.34"
mock_client = _mock_httpx_client()
with (
patch(
"fastmcp.server.auth.ssrf.resolve_hostname",
return_value=[resolved_ip],
) as mock_resolve,
patch("httpx.AsyncClient", return_value=mock_client) as mock_client_class,
):
assert fastmcp.settings.ssrf_trust_proxy is False
await ssrf_safe_fetch("https://example.com/api")
mock_resolve.assert_called_once()
# Connection is pinned to the resolved IP literal, with Host + SNI = hostname.
call_args = mock_client.stream.call_args
url_called = call_args[0][1]
assert resolved_ip in url_called
assert call_args[1]["headers"]["Host"] == "example.com"
assert call_args[1]["extensions"] == {"sni_hostname": "example.com"}
# No explicit proxy is passed, and trust_env keeps its normal default.
client_kwargs = mock_client_class.call_args[1]
assert client_kwargs["proxy"] is None
assert client_kwargs["trust_env"] is True
async def test_default_mode_ignores_proxy_env_vars(self, monkeypatch):
"""Regression: proxy env vars — including a hostile NO_PROXY that previously
caused non-deterministic failures must not affect the default (non-trust)
path at all, since it never reads them."""
monkeypatch.setenv("HTTPS_PROXY", "http://proxy.internal:3128")
monkeypatch.setenv("NO_PROXY", "localhost,127.0.0.1,::1")
resolved_ip = "93.184.216.34"
mock_client = _mock_httpx_client()
with (
patch(
"fastmcp.server.auth.ssrf.resolve_hostname",
return_value=[resolved_ip],
) as mock_resolve,
patch("httpx.AsyncClient", return_value=mock_client) as mock_client_class,
):
assert fastmcp.settings.ssrf_trust_proxy is False
await ssrf_safe_fetch("https://example.com/api")
mock_resolve.assert_called_once()
assert mock_client_class.call_args[1]["proxy"] is None
assert mock_client_class.call_args[1]["trust_env"] is True

View file

@ -1,11 +1,16 @@
from collections.abc import MutableMapping
from typing import Any, Literal
import pytest
from mcp.server.auth.middleware.bearer_auth import RequireAuthMiddleware
from starlette.responses import Response
from starlette.routing import Route
from starlette.testclient import TestClient
from starlette.types import Receive, Scope, Send
from fastmcp.server import FastMCP
from fastmcp.server.auth.providers.jwt import JWTVerifier, RSAKeyPair
from fastmcp.server.http import create_streamable_http_app
from fastmcp.server.http import HostOriginGuardMiddleware, create_streamable_http_app
INITIALIZE_REQUEST = {
"jsonrpc": "2.0",
@ -19,6 +24,59 @@ INITIALIZE_REQUEST = {
}
async def _ok_app(scope: Scope, receive: Receive, send: Send) -> None:
response = Response("OK")
await response(scope, receive, send)
async def _empty_receive() -> dict[str, Any]:
return {"type": "http.request", "body": b"", "more_body": False}
async def _guard_status(
*,
host: str,
origin: str | None = None,
server: tuple[str, int] | None = None,
mode: Literal["auto", "strict"] = "auto",
allowed_hosts: list[str] | None = None,
allowed_origins: list[str] | None = None,
) -> int:
app = HostOriginGuardMiddleware(
_ok_app,
allowed_hosts=allowed_hosts,
allowed_origins=allowed_origins,
mode=mode,
)
headers = [(b"host", host.encode())]
if origin is not None:
headers.append((b"origin", origin.encode()))
scope: Scope = {
"type": "http",
"asgi": {"version": "3.0"},
"http_version": "1.1",
"method": "POST",
"scheme": "https",
"path": "/mcp",
"raw_path": b"/mcp",
"query_string": b"",
"headers": headers,
"client": ("127.0.0.1", 12345),
"server": server,
}
sent_messages: list[MutableMapping[str, Any]] = []
async def send(message: MutableMapping[str, Any]) -> None:
sent_messages.append(message)
await app(scope, _empty_receive, send)
response_start = next(
message for message in sent_messages if message["type"] == "http.response.start"
)
return response_start["status"]
class TestStreamableHTTPAppResourceMetadataURL:
"""Test resource_metadata_url logic in create_streamable_http_app."""
@ -108,11 +166,84 @@ class TestStreamableHTTPAppResourceMetadataURL:
class TestStreamableHTTPHostOriginProtection:
"""Test host and origin validation for streamable HTTP apps."""
def test_rejects_untrusted_host_before_session_initialization(self):
def test_default_allows_untrusted_host_for_compatibility(self):
server = FastMCP(name="TestServer")
app = create_streamable_http_app(
server=server,
streamable_http_path="/mcp",
allowed_hosts=["apps.example.com"],
)
with TestClient(app, base_url="http://127.0.0.1") as client:
response = client.post(
"/mcp",
headers={
"accept": "application/json, text/event-stream",
"host": "internal-upstream",
"x-forwarded-host": "apps.example.com",
},
json=INITIALIZE_REQUEST,
)
assert response.status_code == 200
assert "mcp-session-id" in response.headers
async def test_auto_allows_public_host_when_server_scope_is_ambiguous(self):
status = await _guard_status(
host="mcp.example.com",
origin="https://app.example.com",
server=None,
)
assert status == 200
async def test_auto_rejects_untrusted_host_when_server_scope_is_loopback(self):
status = await _guard_status(
host="attacker.example",
origin="https://attacker.example",
server=("127.0.0.1", 8000),
)
assert status == 421
async def test_strict_rejects_public_host_when_server_scope_is_ambiguous(self):
status = await _guard_status(
host="mcp.example.com",
origin="https://app.example.com",
server=None,
mode="strict",
)
assert status == 421
async def test_auto_rejects_same_origin_fallback_without_trusted_host_boundary(
self,
):
status = await _guard_status(
host="attacker.example",
origin="https://attacker.example",
server=None,
allowed_origins=["https://app.example.com"],
)
assert status == 403
async def test_auto_allows_configured_origin_without_trusted_host_boundary(self):
status = await _guard_status(
host="mcp.example.com",
origin="https://app.example.com",
server=None,
allowed_origins=["https://app.example.com"],
)
assert status == 200
def test_auto_rejects_untrusted_host_before_session_initialization(self):
server = FastMCP(name="TestServer")
app = create_streamable_http_app(
server=server,
streamable_http_path="/mcp",
host_origin_protection="auto",
)
with TestClient(app, base_url="http://127.0.0.1") as client:
@ -128,11 +259,12 @@ class TestStreamableHTTPHostOriginProtection:
assert response.status_code == 421
assert "mcp-session-id" not in response.headers
def test_rejects_untrusted_origin_before_session_initialization(self):
def test_auto_rejects_untrusted_origin_before_session_initialization(self):
server = FastMCP(name="TestServer")
app = create_streamable_http_app(
server=server,
streamable_http_path="/mcp",
host_origin_protection="auto",
)
with TestClient(app, base_url="http://127.0.0.1") as client:
@ -153,6 +285,7 @@ class TestStreamableHTTPHostOriginProtection:
app = create_streamable_http_app(
server=server,
streamable_http_path="/mcp",
host_origin_protection="auto",
allowed_hosts=["mcp.example.com"],
allowed_origins=["https://app.example.com"],
)
@ -176,6 +309,7 @@ class TestStreamableHTTPHostOriginProtection:
app = create_streamable_http_app(
server=server,
streamable_http_path="/mcp",
host_origin_protection="auto",
allowed_hosts=["mcp.example.com"],
)
@ -197,6 +331,7 @@ class TestStreamableHTTPHostOriginProtection:
app = create_streamable_http_app(
server=server,
streamable_http_path="/mcp",
host_origin_protection="auto",
)
with TestClient(app, base_url="http://127.0.0.1") as client:
@ -217,6 +352,7 @@ class TestStreamableHTTPHostOriginProtection:
app = create_streamable_http_app(
server=server,
streamable_http_path="/mcp",
host_origin_protection="auto",
allowed_hosts=["mcp.example.com"],
)
@ -238,6 +374,7 @@ class TestStreamableHTTPHostOriginProtection:
app = create_streamable_http_app(
server=server,
streamable_http_path="/mcp",
host_origin_protection="auto",
allowed_hosts=["mcp.example.com"],
allowed_origins=["http://localhost:3000"],
)
@ -267,6 +404,7 @@ class TestStreamableHTTPHostOriginProtection:
app = create_streamable_http_app(
server=server,
streamable_http_path="/mcp",
host_origin_protection="auto",
allowed_hosts=["mcp.example.com"],
)

View file

@ -1,6 +1,9 @@
import pytest
from fastmcp.server.mixins.transport import _format_host_for_url
from fastmcp.server.mixins.transport import (
_format_host_for_url,
_resolve_allowed_hosts_for_run,
)
@pytest.mark.parametrize(
@ -18,3 +21,30 @@ from fastmcp.server.mixins.transport import _format_host_for_url
def test_format_host_for_url(host: str, expected: str):
"""IPv6 hosts are bracketed for use in a URL; everything else is unchanged."""
assert _format_host_for_url(host) == expected
def test_resolve_allowed_hosts_for_run_merges_configured_hosts_with_loopback_host():
assert _resolve_allowed_hosts_for_run(
host="127.0.0.1",
host_origin_protection="auto",
allowed_hosts=None,
configured_allowed_hosts=["mcp.example.com"],
) == ["mcp.example.com", "127.0.0.1"]
def test_resolve_allowed_hosts_for_run_preserves_configured_hosts_when_disabled():
assert _resolve_allowed_hosts_for_run(
host="127.0.0.1",
host_origin_protection=False,
allowed_hosts=None,
configured_allowed_hosts=["mcp.example.com"],
) == ["mcp.example.com"]
def test_resolve_allowed_hosts_for_run_preserves_explicit_hosts():
assert _resolve_allowed_hosts_for_run(
host="127.0.0.1",
host_origin_protection="auto",
allowed_hosts=["mcp.example.com"],
configured_allowed_hosts=["settings.example.com"],
) == ["mcp.example.com"]

View file

@ -37,3 +37,21 @@ def test_get_setting_raises_for_missing_nested_parent():
test_settings.get_setting("docket__missing__value")
assert str(exc_info.value) == "Setting missing does not exist."
def test_http_host_origin_protection_defaults_to_false():
assert Settings().http_host_origin_protection is False
@pytest.mark.parametrize(
("value", "expected"),
[
("auto", "auto"),
("true", True),
("false", False),
],
)
def test_http_host_origin_protection_env_var(value, expected, monkeypatch):
monkeypatch.setenv("FASTMCP_HTTP_HOST_ORIGIN_PROTECTION", value)
assert Settings().http_host_origin_protection == expected

View file

@ -41,6 +41,23 @@ def test_tool_from_tool_no_change(add_tool):
assert new_tool.description == add_tool.description
def test_transformed_tool_required_order_is_deterministic():
"""`required` must follow property order, not set iteration order.
Set iteration order varies with PYTHONHASHSEED, which broke snapshot
tests of tools/list output across processes.
"""
def fn(alpha: int, beta: str, gamma: float, delta: bool, epsilon: int) -> str:
return "x"
base = Tool.from_function(fn)
transformed = Tool.from_tool(base, transform_args={"alpha": ArgTransform(name="a")})
props = list(transformed.parameters["properties"])
assert transformed.parameters["required"] == props
assert props == ["a", "beta", "gamma", "delta", "epsilon"]
def test_from_tool_accepts_decorated_function():
@tool
def search(q: str, limit: int = 10) -> list[str]:

View file

@ -890,6 +890,45 @@ class TestQueryParameterSerialization:
assert "myAttribute=true" in url
assert "data=" not in url
def test_deep_object_explode_true_uses_bracket_notation(self, director):
route = HTTPRoute(
path="/items",
method="GET",
operation_id="list_items",
parameters=[
ParameterInfo(
name="filter",
location="query",
required=True,
schema={
"type": "object",
"properties": {
"eq": {"type": "string"},
"display name": {"type": "string"},
},
},
explode=True,
style="deepObject",
)
],
parameter_map={
"filter": {"location": "query", "openapi_name": "filter"},
},
)
request = director.build(
route,
{"filter": {"eq": "foo/bar", "display name": "active & ready"}},
"https://example.com",
)
assert request.url.params["filter[eq]"] == "foo/bar"
assert request.url.params["filter[display name]"] == "active & ready"
assert "filter%5Beq%5D=foo%2Fbar" in str(request.url)
assert "filter%5Bdisplay+name%5D=active+%26+ready" in str(request.url)
assert "eq" not in request.url.params
assert "display name" not in request.url.params
def test_explode_default_dict_expands_to_separate_params(self, director):
"""Default explode (None → true) on objects expands properties."""
route = HTTPRoute(

View file

@ -1,4 +1,6 @@
import copy
import sys
from typing import Any
from unittest.mock import patch
from jsonref import replace_refs
@ -12,6 +14,31 @@ from fastmcp.utilities.json_schema import (
)
def _measure_depth(schema: dict[str, Any]) -> int:
"""Return how many `items` levels deep an array-nested schema goes.
Walks iteratively so the assertion helpers cannot themselves hit the
recursion limit the tests are probing.
"""
depth = 0
node: Any = schema
while isinstance(node.get("items"), dict):
node = node["items"]
depth += 1
return depth
def _count_titles(schema: dict[str, Any]) -> int:
"""Count the `title` keys down an array-nested schema, iteratively."""
count = 0
node: Any = schema
while isinstance(node, dict):
if "title" in node:
count += 1
node = node.get("items")
return count
class TestPruneParam:
"""Tests for the _prune_param function."""
@ -358,6 +385,84 @@ class TestDereferenceRefs:
class TestCompressSchema:
"""Tests for the compress_schema function."""
def test_does_not_mutate_input(self):
"""compress_schema must return a new dict and leave the caller's schema
untouched, even when it prunes titles, additionalProperties and unused
$defs (a live Tool.input_schema is passed straight in at some call sites)."""
schema = {
"type": "object",
"title": "MySchema",
"additionalProperties": False,
"properties": {
"a": {"type": "string", "title": "A"},
"b": {
"type": "object",
"title": "B",
"properties": {"c": {"type": "integer", "title": "C"}},
},
},
"$defs": {"Unused": {"type": "string", "title": "Unused"}},
}
original = copy.deepcopy(schema)
result = compress_schema(
schema, prune_titles=True, prune_additional_properties=True
)
# The input is untouched...
assert schema == original
assert result is not schema
# ...and the returned copy really was optimized (so it is not a no-op).
assert "title" not in result
assert "title" not in result["properties"]["b"]["properties"]["c"]
assert "additionalProperties" not in result
assert "$defs" not in result
def test_compresses_schema_nested_far_beyond_the_recursion_limit(self):
"""Deeply nested schemas must compress rather than raise RecursionError.
Copying the schema is what sets the depth ceiling, so it must not
recurse: schemas this deep arrive from proxied or remote MCP servers,
and failing to compress them is worse than compressing them partially.
"""
depth = sys.getrecursionlimit() * 2
schema: dict[str, Any] = {"type": "string", "title": "Leaf"}
for _ in range(depth):
schema = {"type": "array", "title": "Level", "items": schema}
original_depth = _measure_depth(schema)
result = compress_schema(schema, prune_titles=True)
assert result is not schema
assert _measure_depth(result) == original_depth
# The caller's schema is still intact at every level...
assert _count_titles(schema) == original_depth + 1
# ...and the copy really was pruned as deep as the traversal reaches.
assert _count_titles(result) < _count_titles(schema)
def test_keeps_defs_referenced_below_the_traversal_cutoff(self):
"""A $ref deeper than the traversal walks must still pin its definition.
The reference scan stops at its depth guard, so past that point it
cannot prove a definition is unused. Dropping one anyway would leave a
dangling $ref an invalid schema is worse than an unpruned one.
"""
schema: dict[str, Any] = {"$ref": "#/$defs/Leaf"}
for _ in range(60):
schema = {"type": "array", "items": schema}
schema["$defs"] = {"Leaf": {"type": "string"}}
result = compress_schema(schema)
assert result["$defs"] == {"Leaf": {"type": "string"}}
node: Any = result
while isinstance(node.get("items"), dict):
node = node["items"]
assert node == {"$ref": "#/$defs/Leaf"}
def test_preserves_refs_by_default(self):
"""Test that compress_schema preserves $refs by default."""
schema = {