mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-09 07:09:11 +02:00
Merge remote-tracking branch 'origin/release/3.x' into codex/oauth-proxy-rfc9207-issuer
This commit is contained in:
Jeremiah Lowin
commit
9e34b1686c
18 changed files with 1267 additions and 28 deletions
|
|
@ -5,6 +5,25 @@ rss: true
|
|||
tag: NEW
|
||||
---
|
||||
|
||||
<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 +3756,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>
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
||||
|
|
|
|||
|
|
@ -284,6 +284,7 @@
|
|||
"integrations/eunomia-authorization",
|
||||
"integrations/github",
|
||||
"integrations/google",
|
||||
"integrations/huggingface",
|
||||
"integrations/keycloak",
|
||||
"integrations/oci",
|
||||
"integrations/permit",
|
||||
|
|
|
|||
304
docs/integrations/huggingface.mdx
Normal file
304
docs/integrations/huggingface.mdx
Normal 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>
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -5,6 +5,22 @@ icon: "sparkles"
|
|||
tag: NEW
|
||||
---
|
||||
|
||||
<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 +741,3 @@ Because life's too short for boilerplate.
|
|||
This is where it all started. FastMCP’s 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>
|
||||
|
||||
|
|
|
|||
31
examples/auth/huggingface_oauth/README.md
Normal file
31
examples/auth/huggingface_oauth/README.md
Normal 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.
|
||||
32
examples/auth/huggingface_oauth/client.py
Normal file
32
examples/auth/huggingface_oauth/client.py
Normal 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())
|
||||
35
examples/auth/huggingface_oauth/server.py
Normal file
35
examples/auth/huggingface_oauth/server.py
Normal 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)
|
||||
279
fastmcp_slim/fastmcp/server/auth/providers/huggingface.py
Normal file
279
fastmcp_slim/fastmcp/server/auth/providers/huggingface.py
Normal 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
|
||||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -320,7 +320,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
|
||||
|
||||
|
|
|
|||
|
|
@ -3,3 +3,4 @@ server:
|
|||
- server-sse-polling
|
||||
- resources-subscribe
|
||||
- resources-unsubscribe
|
||||
- dns-rebinding-protection
|
||||
|
|
|
|||
236
tests/server/auth/providers/test_huggingface.py
Normal file
236
tests/server/auth/providers/test_huggingface.py
Normal 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"
|
||||
|
|
@ -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"],
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -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"]
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue