Merge branch 'main' into claude/issue-2245-20251024-2225

This commit is contained in:
William Easton 2025-10-24 18:16:34 -05:00 committed by GitHub
commit a83824c04d
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
47 changed files with 611 additions and 1003 deletions

View file

@ -24,7 +24,7 @@ uv run pytest # Run full test suite
| ------------------ | --------------------------------------------------- |
| `src/fastmcp/` | Library source code (Python ≥ 3.10) |
| `├─server/` | Server implementation, `FastMCP`, auth, networking |
| `│ ├─auth/` | Authentication providers (Bearer, JWT, WorkOS) |
| `│ ├─auth/` | Authentication providers (Google, GitHub, Azure, AWS, WorkOS, Auth0, JWT, and more) |
| `│ └─middleware/` | Error handling, logging, rate limiting |
| `├─client/` | High-level client SDK + transports |
| `│ └─auth/` | Client authentication (Bearer, OAuth) |
@ -262,4 +262,4 @@ uv sync # Installs all deps including dev tools
1. **Dependencies**: Always `uv sync` first
2. **Pre-commit fails**: Run `uv run pre-commit run --all-files` to see failures
3. **Type errors**: Use `uv run ty check` directly, check `pyproject.toml` config
4. **Test timeouts**: Default 3s - optimize or mark as integration tests
4. **Test timeouts**: Default 5s - optimize or mark as integration tests

View file

@ -211,7 +211,6 @@ Access MCP session capabilities within your tools, resources, or prompts by addi
- **Logging:** Log messages to MCP clients with `ctx.info()`, `ctx.error()`, etc.
- **LLM Sampling:** Use `ctx.sample()` to request completions from the client's LLM.
- **HTTP Request:** Use `ctx.http_request()` to make HTTP requests to other servers.
- **Resource Access:** Use `ctx.read_resource()` to access resources on the server
- **Progress Reporting:** Use `ctx.report_progress()` to report progress to the client.
- and more...
@ -321,7 +320,7 @@ FastMCP provides comprehensive authentication support that sets it apart from ba
Protecting a server takes just two lines:
```python
from fastmcp.server.auth import GoogleProvider
from fastmcp.server.auth.providers.google import GoogleProvider
auth = GoogleProvider(client_id="...", client_secret="...", base_url="https://myserver.com")
mcp = FastMCP("Protected Server", auth=auth)

View file

@ -557,30 +557,33 @@ This automatic approach is convenient for development but not suitable for produ
**For Production:**
Production requires explicit key management to ensure tokens survive restarts and can be shared across multiple server instances. This requires three things working together:
Production requires explicit key management to ensure tokens survive restarts and can be shared across multiple server instances. This requires the following two things working together:
1. **Explicit JWT signing key** for signing tokens issued to clients
2. **Explicit token encryption key** for encrypting upstream OAuth tokens at rest
3. **Persistent network-accessible storage** for encrypted upstream tokens
The keys can be any secret strings (environment variables, secret manager, etc.) and should be different from each other. FastMCP derives proper cryptographic keys from whatever you provide using HKDF.
3. **Persistent network-accessible storage** for upstream tokens (wrapped in `FernetEncryptionWrapper` to encrypt sensitive data at rest)
**Configuration:**
Add three parameters to your auth provider:
Add two parameters to your auth provider:
```python {8-12}
from key_value.aio.stores.redis import RedisStore
from key_value.aio.wrappers.encryption import FernetEncryptionWrapper
from cryptography.fernet import Fernet
```python {4-7}
auth = GitHubProvider(
client_id=os.environ["GITHUB_CLIENT_ID"],
client_secret=os.environ["GITHUB_CLIENT_SECRET"],
jwt_signing_key=os.environ["JWT_SIGNING_KEY"],
token_encryption_key=os.environ["TOKEN_ENCRYPTION_KEY"],
client_storage=RedisStore(host="redis.example.com", ...),
client_storage=FernetEncryptionWrapper(
key_value=RedisStore(host="redis.example.com", port=6379),
fernet=Fernet(os.environ["STORAGE_ENCRYPTION_KEY"])
),
base_url="https://your-server.com" # use HTTPS
)
```
All three parameters are required for production. Without explicit keys, new keys are generated each time the server starts (on Mac/Windows from keyring, on Linux ephemeral). Without persistent storage, encrypted tokens are lost. Both cause token validation to fail after restart, requiring all clients to re-authenticate.
Both parameters are required for production. Without an explicit signing key, keys are signed using a key derived from the client_secret, which will cause invalidation upon rotation of the client secret. Without persistent storage, tokens are local to the server and won't be trusted across hosts. **Wrap your storage backend in `FernetEncryptionWrapper` to encrypt sensitive OAuth tokens at rest** - without encryption, tokens are stored in plaintext.
For more details on the token architecture and key management, see [OAuth Proxy Key and Storage Management](/servers/auth/oauth-proxy#key-and-storage-management).

View file

@ -4,6 +4,8 @@ description: "Testing patterns and requirements for FastMCP"
icon: vial
---
import { VersionBadge } from "/snippets/version-badge.mdx"
Good tests are the foundation of reliable software. In FastMCP, we treat tests as first-class documentation that demonstrates how features work while protecting against regressions. Every new capability needs comprehensive tests that demonstrate correctness.
## FastMCP Tests
@ -301,6 +303,8 @@ While in-memory testing covers most unit testing needs, you'll occasionally need
#### In-Process Network Testing (Preferred)
<VersionBadge version="2.13.0" />
For most network transport tests, use `run_server_async` with AnyIO task groups. This runs the server as a task in the same process, providing fast, deterministic tests with full debugger support:
```python

View file

@ -38,15 +38,12 @@ auth = GitHubProvider(
# Explicit keys (required for production)
jwt_signing_key=os.environ["JWT_SIGNING_KEY"],
token_encryption_key=os.environ["TOKEN_ENCRYPTION_KEY"],
# Persistent network storage (required for production)
client_storage=RedisStore(host="redis.example.com", port=6379)
)
```
All three are required for production. The keys accept any secret string and should be different from each other.
**More information:**
- [OAuth Token Security](/deployment/http#oauth-token-security) - Complete production setup guide
- [Key and Storage Management](/servers/auth/oauth-proxy#key-and-storage-management) - Detailed explanation of defaults and production requirements

View file

@ -236,7 +236,8 @@
"patterns/tool-transformation",
"patterns/decorating-methods",
"patterns/cli",
"patterns/contrib"
"patterns/contrib",
"patterns/testing"
]
},
{

View file

@ -153,15 +153,17 @@ When you run the client for the first time:
<VersionBadge version="2.13.0" />
For production deployments with persistent token management across server restarts, configure `jwt_signing_key`, `token_encryption_key`, and `client_storage`:
For production deployments with persistent token management across server restarts, configure `jwt_signing_key`, and `client_storage`:
```python server.py
import os
from fastmcp import FastMCP
from fastmcp.server.auth.providers.auth0 import Auth0Provider
from key_value.aio.stores.redis import RedisStore
from key_value.aio.wrappers.encryption import FernetEncryptionWrapper
from cryptography.fernet import Fernet
# Production setup with persistent token storage
# Production setup with encrypted persistent token storage
auth_provider = Auth0Provider(
config_url="https://.../.well-known/openid-configuration",
client_id="tv2ObNgaZAWWhhycr7Bz1LU2mxlnsmsB",
@ -170,11 +172,13 @@ auth_provider = Auth0Provider(
base_url="https://your-production-domain.com",
# Production token management
jwt_signing_key=os.environ["JWT_SIGNING_KEY"], # Secret for signing JWT tokens
token_encryption_key=os.environ["TOKEN_ENCRYPTION_KEY"], # Secret for encrypting tokens at rest
client_storage=RedisStore( # Persistent storage for client registrations
host=os.environ["REDIS_HOST"],
port=int(os.environ["REDIS_PORT"])
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"])
)
)
@ -182,7 +186,7 @@ mcp = FastMCP(name="Production Auth0 App", auth=auth_provider)
```
<Note>
All three parameters (`jwt_signing_key`, `token_encryption_key`, and `client_storage`) work together to ensure tokens and client registrations survive server restarts. Store secrets in environment variables and use a persistent storage backend like Redis or PostgreSQL for distributed deployments.
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

@ -200,15 +200,17 @@ The client caches tokens locally, so you won't need to re-authenticate for subse
<VersionBadge version="2.13.0" />
For production deployments with persistent token management across server restarts, configure `jwt_signing_key`, `token_encryption_key`, and `client_storage`:
For production deployments with persistent token management across server restarts, configure `jwt_signing_key`, and `client_storage`:
```python server.py
import os
from fastmcp import FastMCP
from fastmcp.server.auth.providers.aws import AWSCognitoProvider
from key_value.aio.stores.redis import RedisStore
from key_value.aio.wrappers.encryption import FernetEncryptionWrapper
from cryptography.fernet import Fernet
# Production setup with persistent token storage
# Production setup with encrypted persistent token storage
auth_provider = AWSCognitoProvider(
user_pool_id="eu-central-1_XXXXXXXXX",
aws_region="eu-central-1",
@ -217,11 +219,13 @@ auth_provider = AWSCognitoProvider(
base_url="https://your-production-domain.com",
# Production token management
jwt_signing_key=os.environ["JWT_SIGNING_KEY"], # Secret for signing JWT tokens
token_encryption_key=os.environ["TOKEN_ENCRYPTION_KEY"], # Secret for encrypting tokens at rest
client_storage=RedisStore( # Persistent storage for client registrations
host=os.environ["REDIS_HOST"],
port=int(os.environ["REDIS_PORT"])
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"])
)
)
@ -229,7 +233,7 @@ mcp = FastMCP(name="Production AWS Cognito App", auth=auth_provider)
```
<Note>
All three parameters (`jwt_signing_key`, `token_encryption_key`, and `client_storage`) work together to ensure tokens and client registrations survive server restarts. Store secrets in environment variables and use a persistent storage backend like Redis or PostgreSQL for distributed deployments.
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

@ -209,15 +209,17 @@ The client caches tokens locally, so you won't need to re-authenticate for subse
<VersionBadge version="2.13.0" />
For production deployments with persistent token management across server restarts, configure `jwt_signing_key`, `token_encryption_key`, and `client_storage`:
For production deployments with persistent token management across server restarts, configure `jwt_signing_key` and `client_storage`:
```python server.py
import os
from fastmcp import FastMCP
from fastmcp.server.auth.providers.azure import AzureProvider
from key_value.aio.stores.redis import RedisStore
from key_value.aio.wrappers.encryption import FernetEncryptionWrapper
from cryptography.fernet import Fernet
# Production setup with persistent token storage
# Production setup with encrypted persistent token storage
auth_provider = AzureProvider(
client_id="835f09b6-0f0f-40cc-85cb-f32c5829a149",
client_secret="your-client-secret",
@ -226,11 +228,13 @@ auth_provider = AzureProvider(
required_scopes=["your-scope"],
# Production token management
jwt_signing_key=os.environ["JWT_SIGNING_KEY"], # Secret for signing JWT tokens
token_encryption_key=os.environ["TOKEN_ENCRYPTION_KEY"], # Secret for encrypting tokens at rest
client_storage=RedisStore( # Persistent storage for client registrations
host=os.environ["REDIS_HOST"],
port=int(os.environ["REDIS_PORT"])
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"])
)
)
@ -238,7 +242,7 @@ mcp = FastMCP(name="Production Azure App", auth=auth_provider)
```
<Note>
All three parameters (`jwt_signing_key`, `token_encryption_key`, and `client_storage`) work together to ensure tokens and client registrations survive server restarts. Store secrets in environment variables and use a persistent storage backend like Redis or PostgreSQL for distributed deployments.
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

@ -139,26 +139,30 @@ The client caches tokens locally, so you won't need to re-authenticate for subse
<VersionBadge version="2.13.0" />
For production deployments with persistent token management across server restarts, configure `jwt_signing_key`, `token_encryption_key`, and `client_storage`:
For production deployments with persistent token management across server restarts, configure `jwt_signing_key` and `client_storage`:
```python server.py
import os
from fastmcp import FastMCP
from fastmcp.server.auth.providers.github import GitHubProvider
from key_value.aio.stores.redis import RedisStore
from key_value.aio.wrappers.encryption import FernetEncryptionWrapper
from cryptography.fernet import Fernet
# Production setup with persistent token storage
# Production setup with encrypted persistent token storage
auth_provider = GitHubProvider(
client_id="Ov23liAbcDefGhiJkLmN",
client_secret="github_pat_...",
base_url="https://your-production-domain.com",
# Production token management
jwt_signing_key=os.environ["JWT_SIGNING_KEY"], # Secret for signing JWT tokens
token_encryption_key=os.environ["TOKEN_ENCRYPTION_KEY"], # Secret for encrypting tokens at rest
client_storage=RedisStore( # Persistent storage for client registrations
host=os.environ["REDIS_HOST"],
port=int(os.environ["REDIS_PORT"])
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"])
)
)
@ -166,7 +170,7 @@ mcp = FastMCP(name="Production GitHub App", auth=auth_provider)
```
<Note>
All three parameters (`jwt_signing_key`, `token_encryption_key`, and `client_storage`) work together to ensure tokens and client registrations survive server restarts. Store secrets in environment variables and use a persistent storage backend like Redis or PostgreSQL for distributed deployments.
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

@ -152,15 +152,17 @@ The client caches tokens locally, so you won't need to re-authenticate for subse
<VersionBadge version="2.13.0" />
For production deployments with persistent token management across server restarts, configure `jwt_signing_key`, `token_encryption_key`, and `client_storage`:
For production deployments with persistent token management across server restarts, configure `jwt_signing_key` and `client_storage`:
```python server.py
import os
from fastmcp import FastMCP
from fastmcp.server.auth.providers.google import GoogleProvider
from key_value.aio.stores.redis import RedisStore
from key_value.aio.wrappers.encryption import FernetEncryptionWrapper
from cryptography.fernet import Fernet
# Production setup with persistent token storage
# Production setup with encrypted persistent token storage
auth_provider = GoogleProvider(
client_id="123456789.apps.googleusercontent.com",
client_secret="GOCSPX-abc123...",
@ -168,11 +170,13 @@ auth_provider = GoogleProvider(
required_scopes=["openid", "https://www.googleapis.com/auth/userinfo.email"],
# Production token management
jwt_signing_key=os.environ["JWT_SIGNING_KEY"], # Secret for signing JWT tokens
token_encryption_key=os.environ["TOKEN_ENCRYPTION_KEY"], # Secret for encrypting tokens at rest
client_storage=RedisStore( # Persistent storage for client registrations
host=os.environ["REDIS_HOST"],
port=int(os.environ["REDIS_PORT"])
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"])
)
)
@ -180,7 +184,7 @@ mcp = FastMCP(name="Production Google App", auth=auth_provider)
```
<Note>
All three parameters (`jwt_signing_key`, `token_encryption_key`, and `client_storage`) work together to ensure tokens and client registrations survive server restarts. Store secrets in environment variables and use a persistent storage backend like Redis or PostgreSQL for distributed deployments.
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

@ -130,15 +130,17 @@ The client caches tokens locally, so you won't need to re-authenticate for subse
<VersionBadge version="2.13.0" />
For production deployments with persistent token management across server restarts, configure `jwt_signing_key`, `token_encryption_key`, and `client_storage`:
For production deployments with persistent token management across server restarts, configure `jwt_signing_key`, and `client_storage`:
```python server.py
import os
from fastmcp import FastMCP
from fastmcp.server.auth.providers.workos import WorkOSProvider
from key_value.aio.stores.redis import RedisStore
from key_value.aio.wrappers.encryption import FernetEncryptionWrapper
from cryptography.fernet import Fernet
# Production setup with persistent token storage
# Production setup with encrypted persistent token storage
auth = WorkOSProvider(
client_id="client_YOUR_CLIENT_ID",
client_secret="YOUR_CLIENT_SECRET",
@ -147,11 +149,13 @@ auth = WorkOSProvider(
required_scopes=["openid", "profile", "email"],
# Production token management
jwt_signing_key=os.environ["JWT_SIGNING_KEY"], # Secret for signing JWT tokens
token_encryption_key=os.environ["TOKEN_ENCRYPTION_KEY"], # Secret for encrypting tokens at rest
client_storage=RedisStore( # Persistent storage for client registrations
host=os.environ["REDIS_HOST"],
port=int(os.environ["REDIS_PORT"])
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"])
)
)
@ -159,7 +163,7 @@ mcp = FastMCP(name="Production WorkOS App", auth=auth)
```
<Note>
All three parameters (`jwt_signing_key`, `token_encryption_key`, and `client_storage`) work together to ensure tokens and client registrations survive server restarts. Store secrets in environment variables and use a persistent storage backend like Redis or PostgreSQL for distributed deployments.
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>

85
docs/patterns/testing.mdx Normal file
View file

@ -0,0 +1,85 @@
---
title: Testing your FastMCP Server
sidebarTitle: Testing
description: How to test your FastMCP server.
icon: vial
---
The best way to ensure a reliable and maintainable FastMCP Server is to test it! The FastMCP Client combined with Pytest provides a simple and powerful way to test your FastMCP servers.
Using Pytest Fixtures, you can wrap your FastMCP Server in a Client instance that makes interacting with your server fast and easy. This is especially useful when building your own MCP Servers and enables a tight development loop by allowing you to avoid using a separate tool like MCP Inspector during development:
```python
import pytest
from fastmcp.client import Client
from fastmcp.client.transports import FastMCPTransport
from my_project.main import mcp
@pytest.fixture
async def main_mcp_client():
async with Client(transport=mcp) as mcp_client:
yield mcp_client
async def test_list_tools(main_mcp_client: Client[FastMCPTransport]):
list_tools = await main_mcp_client.list_tools()
assert len(list_tools) == 5
```
We recommend the [inline-snapshot library](https://github.com/15r10nk/inline-snapshot) for asserting complex data structures coming from your MCP Server. This library allows you to write tests that are easy to read and understand, and are also easy to update when the data structure changes.
```python
from inline_snapshot import snapshot
async def test_list_tools(main_mcp_client: Client[FastMCPTransport]):
list_tools = await main_mcp_client.list_tools()
assert list_tools == snapshot()
```
Simply run `pytest --inline-snapshot=fix,create` to fill in the `snapshot()` with actual data.
<Tip>
For values that change you can leverage the [dirty-equals](https://github.com/samuelcolvin/dirty-equals) library to perform flexible equality assertions on dynamic or non-deterministic values.
</Tip>
Using the pytest `parametrize` decorator, you can easily test your tools with a wide variety of inputs.
```python
import pytest
from my_project.main import mcp
from fastmcp.client import Client
from fastmcp.client.transports import FastMCPTransport
@pytest.fixture
async def main_mcp_client():
async with Client(mcp) as client:
yield client
@pytest.mark.parametrize(
"first_number, second_number, expected",
[
(1, 2, 3),
(2, 3, 5),
(3, 4, 7),
],
)
async def test_add(
first_number: int,
second_number: int,
expected: int,
main_mcp_client: Client[FastMCPTransport],
):
result = await main_mcp_client.call_tool(
name="add", arguments={"x": first_number, "y": second_number}
)
assert result.data is not None
assert isinstance(result.data, int)
assert result.data == expected
```
<Tip>
The [FastMCP Repository contains thousands of tests](https://github.com/jlowin/fastmcp/tree/main/tests) for the FastMCP Client and Server. Everything from connecting to remote MCP servers, to testing tools, resources, and prompts is covered, take a look for inspiration!
</Tip>

View file

@ -18,7 +18,7 @@ This maintains proper OAuth 2.0 token audience boundaries.
### `derive_jwt_key` <sup><a href="https://github.com/jlowin/fastmcp/blob/main/src/fastmcp/server/auth/jwt_issuer.py#L25" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>
```python
derive_jwt_key(upstream_secret: str, server_salt: str) -> bytes
derive_jwt_key(from_secret: str, server_salt: str) -> bytes
```
@ -28,7 +28,7 @@ Uses HKDF (RFC 5869) to derive a cryptographically secure signing key from
the upstream OAuth client secret combined with a server-specific salt.
**Args:**
- `upstream_secret`: The OAuth client secret from upstream provider
- `from_secret`: The OAuth client secret from upstream provider
- `server_salt`: Random salt unique to this server instance
**Returns:**
@ -38,7 +38,7 @@ the upstream OAuth client secret combined with a server-specific salt.
### `derive_encryption_key` <sup><a href="https://github.com/jlowin/fastmcp/blob/main/src/fastmcp/server/auth/jwt_issuer.py#L46" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>
```python
derive_encryption_key(upstream_secret: str) -> bytes
derive_encryption_key(from_secret: str) -> bytes
```
@ -48,7 +48,7 @@ Uses HKDF to derive a cryptographically secure encryption key for
encrypting upstream tokens at rest.
**Args:**
- `upstream_secret`: The OAuth client secret from upstream provider
- `from_secret`: The OAuth client secret from upstream provider
**Returns:**
- 32-byte Fernet key (base64url-encoded)

View file

@ -213,19 +213,36 @@ These parameters are included in all token requests to the upstream provider.
<ParamField body="client_storage" type="AsyncKeyValue | None">
<VersionBadge version="2.13.0" />
Storage backend for persisting OAuth client registrations and encrypted upstream tokens.
Storage backend for persisting OAuth client registrations and upstream tokens.
**Default behavior:**
- **Mac/Windows**: DiskStore in your platform's data directory (derived from `platformdirs`)
- **Linux**: MemoryStore (ephemeral - clients lost on restart)
By default on Mac/Windows, clients are automatically persisted to disk, allowing them to survive server restarts as long as the filesystem remains accessible. This means MCP clients only need to register once and can reconnect seamlessly. On Linux where keyring isn't available, ephemeral storage is used to match the ephemeral key strategy.
By default, clients are automatically persisted to an encrypted disk store, allowing them to survive server restarts as long as the filesystem remains accessible. This means MCP clients only need to register once and can reconnect seamlessly. The disk store is encrypted using a key derived from the JWT Signing Key (which is derived from the upstream client secret by default). For client registrations to survive upstream client secret rotation, you should provide a JWT Signing Key or your own client_storage.
For production deployments with multiple servers or cloud deployments, see [Storage Backends](/servers/storage-backends) for available options.
For production token persistence, use this with `jwt_signing_key` and `token_encryption_key` - all three work together to ensure tokens survive restarts. See [OAuth Token Security](/deployment/http#oauth-token-security).
<Warning>
**When providing custom storage**, wrap it in `FernetEncryptionWrapper` to encrypt sensitive OAuth tokens at rest:
Testing with in-memory storage:
```python
from key_value.aio.stores.redis import RedisStore
from key_value.aio.wrappers.encryption import FernetEncryptionWrapper
from cryptography.fernet import Fernet
import os
auth = OAuthProxy(
...,
jwt_signing_key=os.environ["JWT_SIGNING_KEY"],
client_storage=FernetEncryptionWrapper(
key_value=RedisStore(host="redis.example.com", port=6379),
fernet=Fernet(os.environ["STORAGE_ENCRYPTION_KEY"])
)
)
```
Without encryption, upstream OAuth tokens are stored in plaintext.
</Warning>
Testing with in-memory storage (unencrypted):
```python
from key_value.aio.stores.memory import MemoryStore
@ -234,20 +251,6 @@ from key_value.aio.stores.memory import MemoryStore
auth = OAuthProxy(..., client_storage=MemoryStore())
```
Production with Redis for distributed deployments:
```python
from key_value.aio.stores.redis import RedisStore
import os
auth = OAuthProxy(
...,
jwt_signing_key=os.environ["JWT_SIGNING_KEY"],
token_encryption_key=os.environ["TOKEN_ENCRYPTION_KEY"],
client_storage=RedisStore(host="redis.example.com", port=6379)
)
```
</ParamField>
<ParamField body="jwt_signing_key" type="str | bytes | None">
@ -256,19 +259,17 @@ auth = OAuthProxy(
Secret used to sign FastMCP JWT tokens issued to clients. Accepts any string or bytes - will be derived into a proper 32-byte cryptographic key using HKDF.
**Default behavior (`None`):**
- **Mac/Windows**: Auto-managed via system keyring. Keys are generated once and persisted, surviving server restarts with zero configuration. Keys are automatically derived from server attributes, so this approach, while convenient, is **only** suitable for development and local testing. For production, you must provide an explicit secret.
- **Linux**: Ephemeral (random salt at startup). Tokens become invalid on server restart, triggering client re-authentication.
Derives a 32-byte key using PBKDF2 from the upstream client secret.
**For production:**
Provide an explicit secret (e.g., from environment variable) to use a fixed key instead of the auto-generated one. Works with `token_encryption_key` and `client_storage` to ensure tokens survive restarts - all three parameters are required for production deployments. This allows you to manage keys securely in cloud environments and across multiple instances.
Provide an explicit secret (e.g., from environment variable) to use a fixed key instead of the key derived from the upstream client secret. This allows you to manage keys securely in cloud environments, allows keys to work across multiple instances, and allows you to rotate keys without losing client registrations.
```python
import os
auth = OAuthProxy(
...,
jwt_signing_key=os.environ["JWT_SIGNING_KEY"], # Any string!
token_encryption_key=os.environ["TOKEN_ENCRYPTION_KEY"],
jwt_signing_key=os.environ["JWT_SIGNING_KEY"], # Any sufficiently complex string!
client_storage=RedisStore(...) # Persistent storage
)
```
@ -276,26 +277,6 @@ auth = OAuthProxy(
See [HTTP Deployment - OAuth Token Security](/deployment/http#oauth-token-security) for complete production setup.
</ParamField>
<ParamField body="token_encryption_key" type="str | bytes | None">
<VersionBadge version="2.13.0" />
Secret used to encrypt upstream tokens at rest in `client_storage`. Accepts any string or bytes - will be derived into a proper 32-byte cryptographic key using HKDF.
**Default behavior (`None`):**
- **Mac/Windows**: FastMCP will generate a key and store it in the system's keyring. Keys are automatically derived from server attributes, so this approach, while convenient, is **only** suitable for development and local testing. For production, you must provide an explicit secret.
- **Linux**: Ephemeral (random salt at startup). Like `jwt_signing_key`, this is ephemeral, though without a valid JWT signing key, encrypted tokens are useless anyway (JWT validation fails first).
**For production:**
Provide an explicit secret distinct from `jwt_signing_key`. Works with `jwt_signing_key` and persistent `client_storage` - all three are required for production deployments.
```python
# Use different secrets for each key
jwt_signing_key="my-jwt-secret-v1"
token_encryption_key="my-encryption-secret-v1"
```
See [HTTP Deployment - OAuth Token Security](/deployment/http#oauth-token-security).
</ParamField>
<ParamField body="require_authorization_consent" type="bool" default="True">
Whether to require user consent before authorizing MCP clients. When enabled (default), users see a consent screen that displays which client is requesting access, preventing [confused deputy attacks](https://modelcontextprotocol.io/specification/2025-06-18/basic/security_best_practices#confused-deputy-problem) by ensuring users explicitly approve new clients.
@ -527,14 +508,14 @@ Check your server logs for "Client registered with redirect_uri" messages to ide
### Key and Storage Management
<VersionBadge version="2.13.0" />
The OAuth proxy requires cryptographic keys for JWT signing and token encryption, plus persistent storage to maintain valid tokens across server restarts.
The OAuth proxy requires cryptographic keys for JWT signing and storage encryption, plus persistent storage to maintain valid tokens across server restarts.
**Default behavior (appropriate for development only):**
- **Mac/Windows**: FastMCP automatically generates keys and stores them in your system keyring. Storage defaults to disk. Tokens survive server restarts. This is **only** suitable for development and local testing.
- **Linux**: Keys are ephemeral (random salt at startup). Storage defaults to memory. Tokens become invalid on server restart.
**For production:**
Configure three parameters together: provide a unique `jwt_signing_key` (for signing FastMCP JWTs), a unique `token_encryption_key` (for encrypting upstream tokens at rest), and a shared `client_storage` backend (for storing encrypted tokens). All three are required for production deployments. Use a network-accessible storage backend like Redis or DynamoDB rather than local disk storage. The keys accept any secret string and derive proper cryptographic keys using HKDF. See [OAuth Token Security](/deployment/http#oauth-token-security) and [Storage Backends](/servers/storage-backends) for complete production setup.
Configure the following parameters together: provide a unique `jwt_signing_key` (for signing FastMCP JWTs), and a shared `client_storage` backend (for storing tokens). Both are required for production deployments. Use a network-accessible storage backend like Redis or DynamoDB rather than local disk storage. **Wrap your storage in `FernetEncryptionWrapper` to encrypt sensitive OAuth tokens at rest** (see the `client_storage` parameter documentation above for examples). The keys accept any secret string and derive proper cryptographic keys using HKDF. See [OAuth Token Security](/deployment/http#oauth-token-security) and [Storage Backends](/servers/storage-backends) for complete production setup.
### Confused Deputy Attacks

View file

@ -139,36 +139,23 @@ Set this if your provider requires a specific authentication method and the defa
- **Linux**: Ephemeral (random salt at startup). Tokens become invalid on server restart, triggering client re-authentication.
**For production:**
Provide an explicit secret (e.g., from environment variable) to use a fixed key instead of the auto-generated one. Works with `token_encryption_key` and `client_storage` to ensure tokens survive restarts - all three parameters are required for production deployments.
</ParamField>
<ParamField body="token_encryption_key" type="str | bytes | None">
<VersionBadge version="2.13.0" />
Secret used to encrypt upstream tokens at rest in `client_storage`. Accepts any string or bytes - will be derived into a proper 32-byte cryptographic key using HKDF.
**Default behavior (`None`):**
- **Mac/Windows**: FastMCP will generate a key and store it in the system's keyring. Keys are automatically derived from server attributes, so this approach, while convenient, is **only** suitable for development and local testing. For production, you must provide an explicit secret.
- **Linux**: Ephemeral (random salt at startup). Like `jwt_signing_key`, this is ephemeral, though without a valid JWT signing key, encrypted tokens are useless anyway (JWT validation fails first).
**For production:**
Provide an explicit secret distinct from `jwt_signing_key`. Works with `jwt_signing_key` and persistent `client_storage` - all three are required for production deployments.
Provide an explicit secret (e.g., from environment variable) to use a fixed key instead of the auto-generated one.
</ParamField>
<ParamField body="client_storage" type="AsyncKeyValue | None">
<VersionBadge version="2.13.0" />
Storage backend for persisting OAuth client registrations and encrypted upstream tokens.
Storage backend for persisting OAuth client registrations and upstream tokens.
**Default behavior:**
- **Mac/Windows**: DiskStore in your platform's data directory (derived from `platformdirs`)
- **Mac/Windows**: Encrypted DiskStore in your platform's data directory (derived from `platformdirs`)
- **Linux**: MemoryStore (ephemeral - clients lost on restart)
By default on Mac/Windows, clients are automatically persisted to disk, allowing them to survive server restarts as long as the filesystem remains accessible. This means MCP clients only need to register once and can reconnect seamlessly. On Linux where keyring isn't available, ephemeral storage is used to match the ephemeral key strategy.
By default on Mac/Windows, clients are automatically persisted to encrypted disk storage, allowing them to survive server restarts as long as the filesystem remains accessible. This means MCP clients only need to register once and can reconnect seamlessly. On Linux where keyring isn't available, ephemeral storage is used to match the ephemeral key strategy.
For production deployments with multiple servers or cloud deployments, use a network-accessible storage backend rather than local disk storage. See [Storage Backends](/servers/storage-backends) for available options.
For production deployments with multiple servers or cloud deployments, use a network-accessible storage backend rather than local disk storage. **Wrap your storage in `FernetEncryptionWrapper` to encrypt sensitive OAuth tokens at rest.** See [Storage Backends](/servers/storage-backends) for available options.
Testing with in-memory storage:
Testing with in-memory storage (unencrypted):
```python
from key_value.aio.stores.memory import MemoryStore
@ -177,6 +164,24 @@ from key_value.aio.stores.memory import MemoryStore
auth = OIDCProxy(..., client_storage=MemoryStore())
```
Production with encrypted Redis storage:
```python
from key_value.aio.stores.redis import RedisStore
from key_value.aio.wrappers.encryption import FernetEncryptionWrapper
from cryptography.fernet import Fernet
import os
auth = OIDCProxy(
...,
jwt_signing_key=os.environ["JWT_SIGNING_KEY"],
client_storage=FernetEncryptionWrapper(
key_value=RedisStore(host="redis.example.com", port=6379),
fernet=Fernet(os.environ["STORAGE_ENCRYPTION_KEY"])
)
)
```
</ParamField>
</Card>

View file

@ -235,8 +235,8 @@ class UserAuthMiddleware(Middleware):
# Middleware stores user info in context state
context.fastmcp_context.set_state("user_id", "user_123")
context.fastmcp_context.set_state("permissions", ["read", "write"])
return await call_next()
return await call_next(context)
@mcp.tool
async def secure_operation(data: str, ctx: Context) -> str:

View file

@ -118,7 +118,6 @@ auth = GitHubProvider(
client_secret=os.environ["GITHUB_CLIENT_SECRET"],
base_url="https://your-server.com",
jwt_signing_key=os.environ["JWT_SIGNING_KEY"],
token_encryption_key=os.environ["TOKEN_ENCRYPTION_KEY"],
client_storage=RedisStore(host="redis.example.com", port=6379)
)
```
@ -148,7 +147,7 @@ For configuration details on these backends, consult the [py-key-value-aio docum
### Server-Side OAuth Token Storage
The [OAuth Proxy](/servers/auth/oauth-proxy) and OAuth auth providers use storage for persisting OAuth client registrations and encrypted upstream tokens.
The [OAuth Proxy](/servers/auth/oauth-proxy) and OAuth auth providers use storage for persisting OAuth client registrations and upstream tokens. **By default, storage is automatically encrypted using `FernetEncryptionWrapper`.** When providing custom storage, wrap it in `FernetEncryptionWrapper` to encrypt sensitive OAuth tokens at rest.
**Development (default behavior):**
@ -170,26 +169,30 @@ auth = GitHubProvider(
**Production:**
For production deployments, configure explicit keys and persistent network-accessible storage:
For production deployments, configure explicit keys and persistent network-accessible storage with encryption:
```python
import os
from fastmcp.server.auth.providers.github import GitHubProvider
from key_value.aio.stores.redis import RedisStore
from key_value.aio.wrappers.encryption import FernetEncryptionWrapper
from cryptography.fernet import Fernet
auth = GitHubProvider(
client_id=os.environ["GITHUB_CLIENT_ID"],
client_secret=os.environ["GITHUB_CLIENT_SECRET"],
base_url="https://your-server.com",
# Explicit token encryption and signing keys (required for production)
# Explicit JWT signing key (required for production)
jwt_signing_key=os.environ["JWT_SIGNING_KEY"],
token_encryption_key=os.environ["TOKEN_ENCRYPTION_KEY"],
# Persistent distributed storage (required for production)
client_storage=RedisStore(host="redis.example.com", port=6379)
# Encrypted persistent storage (required for production)
client_storage=FernetEncryptionWrapper(
key_value=RedisStore(host="redis.example.com", port=6379),
fernet=Fernet(os.environ["STORAGE_ENCRYPTION_KEY"])
)
)
```
All three parameters (both keys and storage) are required for production. See [OAuth Token Security](/deployment/http#oauth-token-security) and [Key and Storage Management](/servers/auth/oauth-proxy#key-and-storage-management) for complete setup details.
Both parameters are required for production. **Wrap your storage in `FernetEncryptionWrapper` to encrypt sensitive OAuth tokens at rest** - without it, tokens are stored in plaintext. See [OAuth Token Security](/deployment/http#oauth-token-security) and [Key and Storage Management](/servers/auth/oauth-proxy#key-and-storage-management) for complete setup details.
### Response Caching Middleware

View file

@ -16,9 +16,8 @@ dependencies = [
"pydantic[email]>=2.11.7",
"pyperclip>=1.9.0",
"openapi-core>=0.19.5",
"py-key-value-aio[disk,memory,keyring]>=0.2.6,<0.3.0",
"py-key-value-aio[disk,keyring,memory]>=0.2.6,<0.3.0",
"websockets>=15.0.1",
"pytest-asyncio>=1.2.0",
]
requires-python = ">=3.10"
@ -41,6 +40,7 @@ classifiers = [
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3.13",
"Typing :: Typed",
]
@ -61,6 +61,7 @@ dev = [
"pyinstrument>=5.0.2",
"pyperclip>=1.9.0",
"pytest>=8.3.3",
"pytest-asyncio>=1.2.0",
"pytest-cov>=6.1.1",
"pytest-env>=1.1.5",
"pytest-flakefinder",

View file

@ -9,82 +9,66 @@ from __future__ import annotations
import base64
import time
from typing import Any
from typing import Any, overload
from authlib.jose import JsonWebToken
from authlib.jose.errors import JoseError
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
from fastmcp.utilities.logging import get_logger
logger = get_logger(__name__)
def derive_jwt_key(upstream_secret: str, server_salt: str) -> bytes:
"""Derive JWT signing key from upstream client secret and server salt.
Uses HKDF (RFC 5869) to derive a cryptographically secure signing key from
the upstream OAuth client secret combined with a server-specific salt.
Args:
upstream_secret: The OAuth client secret from upstream provider
server_salt: Random salt unique to this server instance
Returns:
32-byte key suitable for HS256 JWT signing
"""
return HKDF(
algorithm=hashes.SHA256(),
length=32,
salt=f"fastmcp-jwt-signing-v1-{server_salt}".encode(),
info=b"HS256",
).derive(upstream_secret.encode())
KDF_ITERATIONS = 1000000
def derive_encryption_key(upstream_secret: str) -> bytes:
"""Derive Fernet encryption key from upstream client secret.
Uses HKDF to derive a cryptographically secure encryption key for
encrypting upstream tokens at rest.
Args:
upstream_secret: The OAuth client secret from upstream provider
Returns:
32-byte Fernet key (base64url-encoded)
"""
key_material = HKDF(
algorithm=hashes.SHA256(),
length=32,
salt=b"fastmcp-token-encryption-v1",
info=b"Fernet",
).derive(upstream_secret.encode())
return base64.urlsafe_b64encode(key_material)
@overload
def derive_jwt_key(*, high_entropy_material: str, salt: str) -> bytes:
"""Derive JWT signing key from a high-entropy key material and server salt."""
def derive_key_from_secret(secret: str | bytes, salt: str, info: bytes) -> bytes:
"""Derive 32-byte key from user-provided secret (string or bytes).
@overload
def derive_jwt_key(*, low_entropy_material: str, salt: str) -> bytes:
"""Derive JWT signing key from a low-entropy key material and server salt."""
Accepts any length input and derives a proper cryptographic key.
Uses HKDF to stretch weak inputs into strong keys.
Args:
secret: User-provided secret (any string or bytes)
salt: Application-specific salt string
info: Key purpose identifier
def derive_jwt_key(
*,
high_entropy_material: str | None = None,
low_entropy_material: str | None = None,
salt: str,
) -> bytes:
"""Derive JWT signing key from a high-entropy or low-entropy key material and server salt."""
if high_entropy_material is not None and low_entropy_material is not None:
raise ValueError(
"Either high_entropy_material or low_entropy_material must be provided, but not both"
)
Returns:
32-byte key suitable for HS256 JWT signing or Fernet encryption
"""
secret_bytes = secret.encode() if isinstance(secret, str) else secret
return HKDF(
algorithm=hashes.SHA256(),
length=32,
salt=salt.encode(),
info=info,
).derive(secret_bytes)
if high_entropy_material is not None:
derived_key = HKDF(
algorithm=hashes.SHA256(),
length=32,
salt=salt.encode(),
info=b"Fernet",
).derive(key_material=high_entropy_material.encode())
return base64.urlsafe_b64encode(derived_key)
if low_entropy_material is not None:
pbkdf2 = PBKDF2HMAC(
algorithm=hashes.SHA256(),
length=32,
salt=salt.encode(),
iterations=KDF_ITERATIONS,
).derive(key_material=low_entropy_material.encode())
return base64.urlsafe_b64encode(pbkdf2)
raise ValueError(
"Either high_entropy_material or low_entropy_material must be provided"
)
class JWTIssuer:
@ -250,40 +234,3 @@ class JWTIssuer:
except JoseError as e:
logger.debug("Token validation failed: %s", e)
raise
class TokenEncryption:
"""Handles encryption/decryption of upstream OAuth tokens at rest."""
def __init__(self, encryption_key: bytes):
"""Initialize token encryption.
Args:
encryption_key: Fernet encryption key (32 bytes, base64url-encoded)
"""
self._fernet = Fernet(encryption_key)
def encrypt(self, token: str) -> bytes:
"""Encrypt a token for storage.
Args:
token: Plain text token
Returns:
Encrypted token bytes
"""
return self._fernet.encrypt(token.encode())
def decrypt(self, encrypted_token: bytes) -> str:
"""Decrypt a token from storage.
Args:
encrypted_token: Encrypted token bytes
Returns:
Plain text token
Raises:
cryptography.fernet.InvalidToken: If token is corrupted or key is wrong
"""
return self._fernet.decrypt(encrypted_token).decode()

View file

@ -22,7 +22,6 @@ import base64
import hashlib
import hmac
import json
import platform
import secrets
import time
from base64 import urlsafe_b64encode
@ -32,10 +31,11 @@ from urllib.parse import urlencode, urlparse
import httpx
from authlib.common.security import generate_token
from authlib.integrations.httpx_client import AsyncOAuth2Client
from cryptography.fernet import Fernet
from key_value.aio.adapters.pydantic import PydanticAdapter
from key_value.aio.protocols import AsyncKeyValue
from key_value.aio.stores.disk import DiskStore
from key_value.aio.stores.memory import MemoryStore
from key_value.aio.wrappers.encryption import FernetEncryptionWrapper
from mcp.server.auth.handlers.token import TokenErrorResponse, TokenSuccessResponse
from mcp.server.auth.handlers.token import TokenHandler as _SDKTokenHandler
from mcp.server.auth.json_response import PydanticJSONResponse
@ -57,18 +57,18 @@ from pydantic import AnyHttpUrl, AnyUrl, BaseModel, Field, SecretStr
from starlette.requests import Request
from starlette.responses import HTMLResponse, RedirectResponse
from starlette.routing import Route
from typing_extensions import override
from fastmcp import settings
from fastmcp.server.auth.auth import OAuthProvider, TokenVerifier
from fastmcp.server.auth.handlers.authorize import AuthorizationHandler
from fastmcp.server.auth.jwt_issuer import (
JWTIssuer,
TokenEncryption,
derive_jwt_key,
)
from fastmcp.server.auth.redirect_validation import (
validate_redirect_uri,
)
from fastmcp.utilities.key_management import get_or_generate_keyring_key
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.ui import (
BUTTON_STYLES,
@ -148,12 +148,13 @@ class UpstreamTokenSet(BaseModel):
"""Stored upstream OAuth tokens from identity provider.
These tokens are obtained from the upstream provider (Google, GitHub, etc.)
and are stored encrypted at rest. They are never exposed to MCP clients.
and stored in plaintext within this model. Encryption is handled transparently
at the storage layer via FernetEncryptionWrapper. Tokens are never exposed to MCP clients.
"""
upstream_token_id: str # Unique ID for this token set
access_token: bytes # Encrypted upstream access token
refresh_token: bytes | None # Encrypted upstream refresh token
access_token: str # Upstream access token
refresh_token: str | None # Upstream refresh token
refresh_token_expires_at: (
float | None
) # Unix timestamp when refresh token expires (if known)
@ -563,10 +564,8 @@ class OAuthProxy(OAuthProvider):
extra_token_params: dict[str, str] | None = None,
# Client storage
client_storage: AsyncKeyValue | None = None,
# JWT signing key (optional, ephemeral if not provided)
# JWT signing key
jwt_signing_key: str | bytes | None = None,
# Token encryption key (optional, ephemeral if not provided)
token_encryption_key: str | bytes | None = None,
# Consent screen configuration
require_authorization_consent: bool = True,
):
@ -602,21 +601,18 @@ class OAuthProxy(OAuthProvider):
Example: {"audience": "https://api.example.com"}
extra_token_params: Additional parameters to forward to the upstream token endpoint.
Useful for provider-specific parameters during token exchange.
client_storage: Storage backend for OAuth state (client registrations, encrypted tokens).
Default (Mac/Windows): DiskStore at $FASTMCP_HOME/oauth-proxy (~/.fastmcp/oauth-proxy).
Default (Linux): MemoryStore (ephemeral keys make persistence pointless).
Custom: Pass DiskStore/RedisStore instance or override location via FASTMCP_HOME.
client_storage: Storage backend for OAuth state (client registrations, tokens).
If None, an encrypted DiskStore will be created in the data directory.
jwt_signing_key: Secret for signing FastMCP JWT tokens (any string or bytes).
None (default): Auto-managed via system keyring (Mac/Windows) or ephemeral (Linux).
Explicit value: For production deployments. Recommended to store in environment variable.
token_encryption_key: Secret for encrypting upstream tokens at rest (any string or bytes).
None (default): Auto-managed via system keyring (Mac/Windows) or ephemeral (Linux).
Explicit value: For production deployments. Recommended to store in environment variable.
If bytes are provided, they will be used as-is.
If a string is provided, it will be derived into a 32-byte key using PBKDF2 (1.2M iterations).
If not provided, it will be derived from the upstream client secret using HKDF.
require_authorization_consent: Whether to require user consent before authorizing clients (default True).
When True, users see a consent screen before being redirected to the upstream IdP.
When False, authorization proceeds directly without user confirmation.
SECURITY WARNING: Only disable for local development or testing environments.
"""
# Always enable DCR since we implement it locally for MCP clients
client_registration_options = ClientRegistrationOptions(
enabled=True,
@ -638,12 +634,14 @@ class OAuthProxy(OAuthProvider):
)
# Store upstream configuration
self._upstream_authorization_endpoint = upstream_authorization_endpoint
self._upstream_token_endpoint = upstream_token_endpoint
self._upstream_client_id = upstream_client_id
self._upstream_client_secret = SecretStr(upstream_client_secret)
self._upstream_revocation_endpoint = upstream_revocation_endpoint
self._default_scope_str = " ".join(self.required_scopes or [])
self._upstream_authorization_endpoint: str = upstream_authorization_endpoint
self._upstream_token_endpoint: str = upstream_token_endpoint
self._upstream_client_id: str = upstream_client_id
self._upstream_client_secret: SecretStr = SecretStr(
secret_value=upstream_client_secret
)
self._upstream_revocation_endpoint: str | None = upstream_revocation_endpoint
self._default_scope_str: str = " ".join(self.required_scopes or [])
# Store redirect configuration
if not redirect_path:
@ -659,73 +657,85 @@ class OAuthProxy(OAuthProvider):
):
logger.warning(
"allowed_client_redirect_uris is empty list; no redirect URIs will be accepted. "
"This will block all OAuth clients."
+ "This will block all OAuth clients."
)
self._allowed_client_redirect_uris = allowed_client_redirect_uris
self._allowed_client_redirect_uris: list[str] | None = (
allowed_client_redirect_uris
)
# PKCE configuration
self._forward_pkce = forward_pkce
self._forward_pkce: bool = forward_pkce
# Token endpoint authentication
self._token_endpoint_auth_method = token_endpoint_auth_method
self._token_endpoint_auth_method: str | None = token_endpoint_auth_method
# Consent screen configuration
self._require_authorization_consent = require_authorization_consent
self._require_authorization_consent: bool = require_authorization_consent
if not require_authorization_consent:
logger.warning(
"Authorization consent screen disabled - only use for local development or testing. "
"In production, this screen protects against confused deputy attacks."
+ "In production, this screen protects against confused deputy attacks."
)
# Extra parameters for authorization and token endpoints
self._extra_authorize_params = extra_authorize_params or {}
self._extra_token_params = extra_token_params or {}
self._extra_authorize_params: dict[str, str] = extra_authorize_params or {}
self._extra_token_params: dict[str, str] = extra_token_params or {}
# Default storage: match persistence to key availability
# On Mac/Windows: DiskStore + keyring keys = full persistence
# On Linux: MemoryStore + ephemeral keys = consistent (nothing persists)
if client_storage is None:
if platform.system() != "Linux":
# Keyring available: use persistent storage
default_storage_path = settings.home / "oauth-proxy"
default_storage_path.mkdir(parents=True, exist_ok=True)
self._client_storage = DiskStore(directory=str(default_storage_path))
logger.debug(
"Using disk storage for OAuth state: %s", default_storage_path
)
else:
# Keyring unavailable: use memory storage (ephemeral keys make disk pointless)
self._client_storage = MemoryStore()
logger.debug(
"Using in-memory storage on Linux (keyring unavailable). "
"For persistent tokens, provide explicit jwt_signing_key, "
"token_encryption_key, and client_storage."
)
self._auto_selected_storage = True
else:
self._client_storage = client_storage
self._auto_selected_storage = False
# Warn if explicitly using MemoryStore when keyring is available
if (
isinstance(self._client_storage, MemoryStore)
and not self._auto_selected_storage
and platform.system() != "Linux"
):
logger.warning(
"Using in-memory storage on a platform with keyring support. "
"OAuth state will be lost on restart. Consider using default storage "
"or providing explicit jwt_signing_key and token_encryption_key with persistent storage."
if jwt_signing_key is None:
jwt_signing_key = derive_jwt_key(
high_entropy_material=upstream_client_secret,
salt="fastmcp-jwt-signing-key",
)
if isinstance(jwt_signing_key, str):
if len(jwt_signing_key) < 12:
logger.warning(
"jwt_signing_key is less than 12 characters; it is recommended to use a longer. "
+ "string for the key derivation."
)
jwt_signing_key = derive_jwt_key(
low_entropy_material=jwt_signing_key,
salt="fastmcp-jwt-signing-key",
)
self._jwt_issuer: JWTIssuer = JWTIssuer(
issuer=str(self.base_url),
audience=f"{str(self.base_url).rstrip('/')}/mcp",
signing_key=jwt_signing_key,
)
# If the user does not provide a store, we will provide an encrypted disk store
if client_storage is None:
storage_encryption_key = derive_jwt_key(
high_entropy_material=jwt_signing_key.decode(),
salt="fastmcp-storage-encryption-key",
)
client_storage = FernetEncryptionWrapper(
key_value=DiskStore(directory=settings.home / "oauth-proxy"),
fernet=Fernet(key=storage_encryption_key),
)
self._client_storage: AsyncKeyValue = client_storage
# Cache HTTPS check to avoid repeated logging
self._is_https = str(self.base_url).startswith("https://")
self._is_https: bool = str(self.base_url).startswith("https://")
if not self._is_https:
logger.warning(
"Using non-secure cookies for development; deploy with HTTPS for production."
)
self._client_store = PydanticAdapter[ProxyDCRClient](
self._upstream_token_store: PydanticAdapter[UpstreamTokenSet] = PydanticAdapter[
UpstreamTokenSet
](
key_value=self._client_storage,
pydantic_model=UpstreamTokenSet,
default_collection="mcp-upstream-tokens",
raise_on_validation_error=True,
)
self._client_store: PydanticAdapter[ProxyDCRClient] = PydanticAdapter[
ProxyDCRClient
](
key_value=self._client_storage,
pydantic_model=ProxyDCRClient,
default_collection="mcp-oauth-proxy-clients",
@ -734,43 +744,32 @@ class OAuthProxy(OAuthProvider):
# OAuth transaction storage for IdP callback forwarding
# Reuse client_storage with different collections for state management
self._transaction_store = PydanticAdapter[OAuthTransaction](
self._transaction_store: PydanticAdapter[OAuthTransaction] = PydanticAdapter[
OAuthTransaction
](
key_value=self._client_storage,
pydantic_model=OAuthTransaction,
default_collection="mcp-oauth-transactions",
raise_on_validation_error=True,
)
self._code_store = PydanticAdapter[ClientCode](
self._code_store: PydanticAdapter[ClientCode] = PydanticAdapter[ClientCode](
key_value=self._client_storage,
pydantic_model=ClientCode,
default_collection="mcp-authorization-codes",
raise_on_validation_error=True,
)
# Storage for upstream tokens (encrypted at rest)
self._upstream_token_store = PydanticAdapter[UpstreamTokenSet](
key_value=self._client_storage,
pydantic_model=UpstreamTokenSet,
default_collection="mcp-upstream-tokens",
raise_on_validation_error=True,
)
# Storage for JTI mappings (FastMCP token -> upstream token)
self._jti_mapping_store = PydanticAdapter[JTIMapping](
self._jti_mapping_store: PydanticAdapter[JTIMapping] = PydanticAdapter[
JTIMapping
](
key_value=self._client_storage,
pydantic_model=JTIMapping,
default_collection="mcp-jti-mappings",
raise_on_validation_error=True,
)
# JWT issuer and encryption (initialized lazily on first use)
self._custom_jwt_key = jwt_signing_key
self._custom_encryption_key = token_encryption_key
self._jwt_issuer: JWTIssuer | None = None
self._token_encryption: TokenEncryption | None = None
self._jwt_initialized = False
# Local state for token bookkeeping only (no client caching)
self._access_tokens: dict[str, AccessToken] = {}
self._refresh_tokens: dict[str, RefreshToken] = {}
@ -780,7 +779,7 @@ class OAuthProxy(OAuthProvider):
self._refresh_to_access: dict[str, str] = {}
# Use the provided token validator
self._token_validator = token_verifier
self._token_validator: TokenVerifier = token_verifier
logger.debug(
"Initialized OAuth proxy provider with upstream server %s",
@ -806,110 +805,11 @@ class OAuthProxy(OAuthProvider):
return code_verifier, code_challenge
# -------------------------------------------------------------------------
# JWT Token Factory Initialization
# -------------------------------------------------------------------------
async def _ensure_jwt_initialized(self) -> None:
"""Initialize JWT issuer and token encryption (lazy initialization).
Key derivation strategy:
- Explicit key (production): User-provided via parameters
- Keyring key (local/dev): Auto-managed via system keyring
- Ephemeral key (fallback): Random salt at startup when keyring unavailable
"""
if self._jwt_initialized:
return
# Derive or use custom JWT signing key
from fastmcp.server.auth.jwt_issuer import derive_key_from_secret
if self._custom_jwt_key:
jwt_key = derive_key_from_secret(
secret=self._custom_jwt_key,
salt="fastmcp-jwt-signing-v1",
info=b"HS256",
)
logger.debug("Using user-provided JWT signing key")
else:
keyring_key = get_or_generate_keyring_key(
"jwt-signing", self._upstream_client_id
)
if keyring_key:
jwt_key = derive_key_from_secret(
secret=keyring_key,
salt="fastmcp-jwt-signing-v1",
info=b"HS256",
)
else:
server_salt = secrets.token_urlsafe(32)
upstream_secret = self._upstream_client_secret.get_secret_value()
jwt_key = derive_key_from_secret(
secret=upstream_secret,
salt=f"fastmcp-jwt-signing-v1-{server_salt}",
info=b"HS256",
)
if platform.system() == "Linux":
logger.warning(
"Keyring unavailable on Linux - using ephemeral keys. "
"Storage persists at %s but tokens will become unreadable after restart. "
"For persistent tokens, provide explicit jwt_signing_key and token_encryption_key.",
self._client_storage
if hasattr(self, "_client_storage")
else "disk",
)
else:
logger.warning(
"Keyring unavailable - using ephemeral keys. "
"For production, provide explicit jwt_signing_key and token_encryption_key."
)
# Initialize JWT issuer
issuer = str(self.base_url)
audience = f"{str(self.base_url).rstrip('/')}/mcp"
self._jwt_issuer = JWTIssuer(
issuer=issuer,
audience=audience,
signing_key=jwt_key,
)
if self._custom_encryption_key:
encryption_key = derive_key_from_secret(
secret=self._custom_encryption_key,
salt="fastmcp-token-encryption-v1",
info=b"Fernet",
)
encryption_key = base64.urlsafe_b64encode(encryption_key)
logger.debug("Using user-provided token encryption key")
else:
encryption_keyring_key = get_or_generate_keyring_key(
"token-encryption", self._upstream_client_id
)
if encryption_keyring_key:
key_material = derive_key_from_secret(
secret=encryption_keyring_key,
salt="fastmcp-token-encryption-v1",
info=b"Fernet",
)
encryption_key = base64.urlsafe_b64encode(key_material)
else:
server_salt = secrets.token_urlsafe(32)
upstream_secret = self._upstream_client_secret.get_secret_value()
key_material = derive_key_from_secret(
secret=upstream_secret,
salt=f"fastmcp-token-encryption-v1-{server_salt}",
info=b"Fernet",
)
encryption_key = base64.urlsafe_b64encode(key_material)
self._token_encryption = TokenEncryption(encryption_key)
self._jwt_initialized = True
# -------------------------------------------------------------------------
# Client Registration (Local Implementation)
# -------------------------------------------------------------------------
@override
async def get_client(self, client_id: str) -> OAuthClientInformationFull | None:
"""Get client information by ID. This is generally the random ID
provided to the DCR client during registration, not the upstream client ID.
@ -925,6 +825,7 @@ class OAuthProxy(OAuthProvider):
return client
@override
async def register_client(self, client_info: OAuthClientInformationFull) -> None:
"""Register a client locally
@ -971,6 +872,7 @@ class OAuthProxy(OAuthProvider):
# Authorization Flow (Proxy to Upstream)
# -------------------------------------------------------------------------
@override
async def authorize(
self,
client: OAuthClientInformationFull,
@ -1045,6 +947,7 @@ class OAuthProxy(OAuthProvider):
# Authorization Code Handling
# -------------------------------------------------------------------------
@override
async def load_authorization_code(
self,
client: OAuthClientInformationFull,
@ -1064,7 +967,7 @@ class OAuthProxy(OAuthProvider):
# Check if code expired
if time.time() > code_model.expires_at:
logger.debug("Authorization code expired: %s", authorization_code)
await self._code_store.delete(key=authorization_code)
_ = await self._code_store.delete(key=authorization_code)
return None
# Verify client ID matches
@ -1080,13 +983,14 @@ class OAuthProxy(OAuthProvider):
return AuthorizationCode(
code=authorization_code,
client_id=client.client_id,
redirect_uri=code_model.redirect_uri,
redirect_uri=AnyUrl(url=code_model.redirect_uri),
redirect_uri_provided_explicitly=True,
scopes=code_model.scopes,
expires_at=code_model.expires_at,
code_challenge=code_model.code_challenge or "",
)
@override
async def exchange_authorization_code(
self,
client: OAuthClientInformationFull,
@ -1103,11 +1007,6 @@ class OAuthProxy(OAuthProvider):
PKCE validation is handled by the MCP framework before this method is called.
"""
# Ensure JWT issuer is initialized
await self._ensure_jwt_initialized()
assert self._jwt_issuer is not None
assert self._token_encryption is not None
# Look up stored code data
code_model = await self._code_store.get(key=authorization_code.code)
if not code_model:
@ -1158,8 +1057,8 @@ class OAuthProxy(OAuthProvider):
# Encrypt and store upstream tokens
upstream_token_set = UpstreamTokenSet(
upstream_token_id=upstream_token_id,
access_token=self._token_encryption.encrypt(idp_tokens["access_token"]),
refresh_token=self._token_encryption.encrypt(idp_tokens["refresh_token"])
access_token=idp_tokens["access_token"],
refresh_token=idp_tokens["refresh_token"]
if idp_tokens.get("refresh_token")
else None,
refresh_token_expires_at=refresh_token_expires_at,
@ -1281,11 +1180,6 @@ class OAuthProxy(OAuthProvider):
5. Issue new FastMCP access token
6. Keep same FastMCP refresh token (unless upstream rotates)
"""
# Ensure JWT issuer is initialized
await self._ensure_jwt_initialized()
assert self._jwt_issuer is not None
assert self._token_encryption is not None
# Verify FastMCP refresh token
try:
refresh_payload = self._jwt_issuer.verify_token(refresh_token.token)
@ -1314,10 +1208,6 @@ class OAuthProxy(OAuthProvider):
logger.error("No upstream refresh token available")
raise TokenError("invalid_grant", "Refresh not supported for this token")
upstream_refresh_token = self._token_encryption.decrypt(
upstream_token_set.refresh_token
)
# Refresh upstream token using authlib
oauth_client = AsyncOAuth2Client(
client_id=self._upstream_client_id,
@ -1330,7 +1220,7 @@ class OAuthProxy(OAuthProvider):
logger.debug("Refreshing upstream token (jti=%s)", refresh_jti[:8])
token_response: dict[str, Any] = await oauth_client.refresh_token( # type: ignore[misc]
url=self._upstream_token_endpoint,
refresh_token=upstream_refresh_token,
refresh_token=upstream_token_set.refresh_token,
scope=" ".join(scopes) if scopes else None,
)
logger.debug("Successfully refreshed upstream token")
@ -1342,18 +1232,14 @@ class OAuthProxy(OAuthProvider):
new_expires_in = int(
token_response.get("expires_in", DEFAULT_ACCESS_TOKEN_EXPIRY_SECONDS)
)
upstream_token_set.access_token = self._token_encryption.encrypt(
token_response["access_token"]
)
upstream_token_set.access_token = token_response["access_token"]
upstream_token_set.expires_at = time.time() + new_expires_in
# Handle upstream refresh token rotation and expiry
new_refresh_expires_in = None
if new_upstream_refresh := token_response.get("refresh_token"):
if new_upstream_refresh != upstream_refresh_token:
upstream_token_set.refresh_token = self._token_encryption.encrypt(
new_upstream_refresh
)
if new_upstream_refresh != upstream_token_set.refresh_token:
upstream_token_set.refresh_token = new_upstream_refresh
logger.debug("Upstream refresh token rotated")
# Update refresh token expiry if provided
@ -1492,11 +1378,6 @@ class OAuthProxy(OAuthProvider):
The FastMCP JWT is a reference token - all authorization data comes
from validating the upstream token via the TokenVerifier.
"""
# Ensure JWT issuer and encryption are initialized
await self._ensure_jwt_initialized()
assert self._jwt_issuer is not None
assert self._token_encryption is not None
try:
# 1. Verify FastMCP JWT signature and claims
payload = self._jwt_issuer.verify_token(token)
@ -1517,15 +1398,12 @@ class OAuthProxy(OAuthProvider):
)
return None
# 3. Decrypt upstream token
upstream_token = self._token_encryption.decrypt(
# 3. Validate with upstream provider (delegated to TokenVerifier)
# This calls the real token validator (GitHub API, JWKS, etc.)
validated = await self._token_validator.verify_token(
upstream_token_set.access_token
)
# 4. Validate with upstream provider (delegated to TokenVerifier)
# This calls the real token validator (GitHub API, JWKS, etc.)
validated = await self._token_validator.verify_token(upstream_token)
if not validated:
logger.debug("Upstream token validation failed")
return None

View file

@ -217,7 +217,6 @@ class OIDCProxy(OAuthProxy):
client_storage: AsyncKeyValue | None = None,
# JWT and encryption keys
jwt_signing_key: str | bytes | None = None,
token_encryption_key: str | bytes | None = None,
# Token validation configuration
token_endpoint_auth_method: str | None = None,
# Consent screen configuration
@ -243,13 +242,12 @@ class OIDCProxy(OAuthProxy):
If None (default), only localhost redirect URIs are allowed.
If empty list, all redirect URIs are allowed (not recommended for production).
These are for MCP clients performing loopback redirects, NOT for the upstream OAuth app.
client_storage: An AsyncKeyValue-compatible store for client registrations, registrations are stored in memory if not provided
jwt_signing_key: Secret for signing FastMCP JWT tokens (any string or bytes).
None (default): Auto-managed via system keyring (Mac/Windows) or ephemeral (Linux).
Explicit value: For production deployments. Recommended to store in environment variable.
token_encryption_key: Secret for encrypting upstream tokens at rest (any string or bytes).
None (default): Auto-managed via system keyring (Mac/Windows) or ephemeral (Linux).
Explicit value: For production deployments. Recommended to store in environment variable.
client_storage: Storage backend for OAuth state (client registrations, encrypted tokens).
If None, a DiskStore will be created in the data directory (derived from `platformdirs`). The
disk store will be encrypted using a key derived from the JWT Signing Key.
jwt_signing_key: Secret for signing FastMCP JWT tokens (any string or bytes). If bytes are provided,
they will be used as is. If a string is provided, it will be derived into a 32-byte key. If not
provided, the upstream client secret will be used to derive a 32-byte key using PBKDF2.
token_endpoint_auth_method: Token endpoint authentication method for upstream server.
Common values: "client_secret_basic", "client_secret_post", "none".
If None, authlib will use its default (typically "client_secret_basic").
@ -311,7 +309,6 @@ class OIDCProxy(OAuthProxy):
"allowed_client_redirect_uris": allowed_client_redirect_uris,
"client_storage": client_storage,
"jwt_signing_key": jwt_signing_key,
"token_encryption_key": token_encryption_key,
"token_endpoint_auth_method": token_endpoint_auth_method,
"require_authorization_consent": require_authorization_consent,
}

View file

@ -52,6 +52,7 @@ class Auth0ProviderSettings(BaseSettings):
redirect_path: str | None = None
required_scopes: list[str] | None = None
allowed_client_redirect_uris: list[str] | None = None
jwt_signing_key: str | None = None
@field_validator("required_scopes", mode="before")
@classmethod
@ -96,8 +97,7 @@ class Auth0Provider(OIDCProxy):
redirect_path: str | NotSetT = NotSet,
allowed_client_redirect_uris: list[str] | NotSetT = NotSet,
client_storage: AsyncKeyValue | None = None,
jwt_signing_key: str | bytes | None = None,
token_encryption_key: str | bytes | None = None,
jwt_signing_key: str | bytes | NotSetT = NotSet,
require_authorization_consent: bool = True,
) -> None:
"""Initialize Auth0 OAuth provider.
@ -114,13 +114,12 @@ class Auth0Provider(OIDCProxy):
redirect_path: Redirect path configured in Auth0 application
allowed_client_redirect_uris: List of allowed redirect URI patterns for MCP clients.
If None (default), all URIs are allowed. If empty list, no URIs are allowed.
client_storage: An AsyncKeyValue-compatible store for client registrations, registrations are stored in memory if not provided
jwt_signing_key: Secret for signing FastMCP JWT tokens (any string or bytes).
None (default): Auto-managed via system keyring (Mac/Windows) or ephemeral (Linux).
Explicit value: For production deployments. Recommended to store in environment variable.
token_encryption_key: Secret for encrypting upstream tokens at rest (any string or bytes).
None (default): Auto-managed via system keyring (Mac/Windows) or ephemeral (Linux).
Explicit value: For production deployments. Recommended to store in environment variable.
client_storage: Storage backend for OAuth state (client registrations, encrypted tokens).
If None, a DiskStore will be created in the data directory (derived from `platformdirs`). The
disk store will be encrypted using a key derived from the JWT Signing Key.
jwt_signing_key: Secret for signing FastMCP JWT tokens (any string or bytes). If bytes are provided,
they will be used as is. If a string is provided, it will be derived into a 32-byte key. If not
provided, the upstream client secret will be used to derive a 32-byte key using PBKDF2.
require_authorization_consent: Whether to require user consent before authorizing clients (default True).
When True, users see a consent screen before being redirected to Auth0.
When False, authorization proceeds directly without user confirmation.
@ -139,6 +138,7 @@ class Auth0Provider(OIDCProxy):
"required_scopes": required_scopes,
"redirect_path": redirect_path,
"allowed_client_redirect_uris": allowed_client_redirect_uris,
"jwt_signing_key": jwt_signing_key,
}.items()
if v is not NotSet
}
@ -171,23 +171,20 @@ class Auth0Provider(OIDCProxy):
auth0_required_scopes = settings.required_scopes or ["openid"]
init_kwargs = {
"config_url": settings.config_url,
"client_id": settings.client_id,
"client_secret": settings.client_secret.get_secret_value(),
"audience": settings.audience,
"base_url": settings.base_url,
"issuer_url": settings.issuer_url,
"redirect_path": settings.redirect_path,
"required_scopes": auth0_required_scopes,
"allowed_client_redirect_uris": settings.allowed_client_redirect_uris,
"client_storage": client_storage,
"jwt_signing_key": jwt_signing_key,
"token_encryption_key": token_encryption_key,
"require_authorization_consent": require_authorization_consent,
}
super().__init__(**init_kwargs)
super().__init__(
config_url=settings.config_url,
client_id=settings.client_id,
client_secret=settings.client_secret.get_secret_value(),
audience=settings.audience,
base_url=settings.base_url,
issuer_url=settings.issuer_url,
redirect_path=settings.redirect_path,
required_scopes=auth0_required_scopes,
allowed_client_redirect_uris=settings.allowed_client_redirect_uris,
client_storage=client_storage,
jwt_signing_key=settings.jwt_signing_key,
require_authorization_consent=require_authorization_consent,
)
logger.debug(
"Initialized Auth0 OAuth provider for client %s with scopes: %s",

View file

@ -57,6 +57,7 @@ class AWSCognitoProviderSettings(BaseSettings):
redirect_path: str | None = None
required_scopes: list[str] | None = None
allowed_client_redirect_uris: list[str] | None = None
jwt_signing_key: str | None = None
@field_validator("required_scopes", mode="before")
@classmethod
@ -135,8 +136,7 @@ class AWSCognitoProvider(OIDCProxy):
required_scopes: list[str] | NotSetT = NotSet,
allowed_client_redirect_uris: list[str] | NotSetT = NotSet,
client_storage: AsyncKeyValue | None = None,
jwt_signing_key: str | bytes | None = None,
token_encryption_key: str | bytes | None = None,
jwt_signing_key: str | bytes | NotSetT = NotSet,
require_authorization_consent: bool = True,
):
"""Initialize AWS Cognito OAuth provider.
@ -153,13 +153,12 @@ class AWSCognitoProvider(OIDCProxy):
required_scopes: Required Cognito scopes (defaults to ["openid"])
allowed_client_redirect_uris: List of allowed redirect URI patterns for MCP clients.
If None (default), all URIs are allowed. If empty list, no URIs are allowed.
client_storage: An AsyncKeyValue-compatible store for client registrations, registrations are stored in memory if not provided
jwt_signing_key: Secret for signing FastMCP JWT tokens (any string or bytes).
None (default): Auto-managed via system keyring (Mac/Windows) or ephemeral (Linux).
Explicit value: For production deployments. Recommended to store in environment variable.
token_encryption_key: Secret for encrypting upstream tokens at rest (any string or bytes).
None (default): Auto-managed via system keyring (Mac/Windows) or ephemeral (Linux).
Explicit value: For production deployments. Recommended to store in environment variable.
client_storage: Storage backend for OAuth state (client registrations, encrypted tokens).
If None, a DiskStore will be created in the data directory (derived from `platformdirs`). The
disk store will be encrypted using a key derived from the JWT Signing Key.
jwt_signing_key: Secret for signing FastMCP JWT tokens (any string or bytes). If bytes are provided,
they will be used as is. If a string is provided, it will be derived into a 32-byte key. If not
provided, the upstream client secret will be used to derive a 32-byte key using PBKDF2.
require_authorization_consent: Whether to require user consent before authorizing clients (default True).
When True, users see a consent screen before being redirected to AWS Cognito.
When False, authorization proceeds directly without user confirmation.
@ -179,6 +178,7 @@ class AWSCognitoProvider(OIDCProxy):
"redirect_path": redirect_path,
"required_scopes": required_scopes,
"allowed_client_redirect_uris": allowed_client_redirect_uris,
"jwt_signing_key": jwt_signing_key,
}.items()
if v is not NotSet
}
@ -228,8 +228,7 @@ class AWSCognitoProvider(OIDCProxy):
redirect_path=redirect_path_final,
allowed_client_redirect_uris=allowed_client_redirect_uris_final,
client_storage=client_storage,
jwt_signing_key=jwt_signing_key,
token_encryption_key=token_encryption_key,
jwt_signing_key=settings.jwt_signing_key,
require_authorization_consent=require_authorization_consent,
)

View file

@ -45,6 +45,7 @@ class AzureProviderSettings(BaseSettings):
required_scopes: list[str] | None = None
additional_authorize_scopes: list[str] | None = None
allowed_client_redirect_uris: list[str] | None = None
jwt_signing_key: str | None = None
@field_validator("required_scopes", mode="before")
@classmethod
@ -120,8 +121,7 @@ class AzureProvider(OAuthProxy):
additional_authorize_scopes: list[str] | None | NotSetT = NotSet,
allowed_client_redirect_uris: list[str] | NotSetT = NotSet,
client_storage: AsyncKeyValue | None = None,
jwt_signing_key: str | bytes | None = None,
token_encryption_key: str | bytes | None = None,
jwt_signing_key: str | bytes | NotSetT = NotSet,
require_authorization_consent: bool = True,
) -> None:
"""Initialize Azure OAuth provider.
@ -154,13 +154,12 @@ class AzureProvider(OAuthProxy):
upstream Azure token, but MCP clients are unaware of them.
allowed_client_redirect_uris: List of allowed redirect URI patterns for MCP clients.
If None (default), all URIs are allowed. If empty list, no URIs are allowed.
client_storage: An AsyncKeyValue-compatible store for client registrations, registrations are stored in memory if not provided
jwt_signing_key: Secret for signing FastMCP JWT tokens (any string or bytes).
None (default): Auto-managed via system keyring (Mac/Windows) or ephemeral (Linux).
Explicit value: For production deployments. Recommended to store in environment variable.
token_encryption_key: Secret for encrypting upstream tokens at rest (any string or bytes).
None (default): Auto-managed via system keyring (Mac/Windows) or ephemeral (Linux).
Explicit value: For production deployments. Recommended to store in environment variable.
client_storage: Storage backend for OAuth state (client registrations, encrypted tokens).
If None, a DiskStore will be created in the data directory (derived from `platformdirs`). The
disk store will be encrypted using a key derived from the JWT Signing Key.
jwt_signing_key: Secret for signing FastMCP JWT tokens (any string or bytes). If bytes are provided,
they will be used as is. If a string is provided, it will be derived into a 32-byte key. If not
provided, the upstream client secret will be used to derive a 32-byte key using PBKDF2.
require_authorization_consent: Whether to require user consent before authorizing clients (default True).
When True, users see a consent screen before being redirected to Azure.
When False, authorization proceeds directly without user confirmation.
@ -180,6 +179,7 @@ class AzureProvider(OAuthProxy):
"required_scopes": required_scopes,
"additional_authorize_scopes": additional_authorize_scopes,
"allowed_client_redirect_uris": allowed_client_redirect_uris,
"jwt_signing_key": jwt_signing_key,
}.items()
if v is not NotSet
}
@ -256,8 +256,7 @@ class AzureProvider(OAuthProxy):
or settings.base_url, # Default to base_url if not specified
allowed_client_redirect_uris=settings.allowed_client_redirect_uris,
client_storage=client_storage,
jwt_signing_key=jwt_signing_key,
token_encryption_key=token_encryption_key,
jwt_signing_key=settings.jwt_signing_key,
require_authorization_consent=require_authorization_consent,
)

View file

@ -54,6 +54,7 @@ class GitHubProviderSettings(BaseSettings):
required_scopes: list[str] | None = None
timeout_seconds: int | None = None
allowed_client_redirect_uris: list[str] | None = None
jwt_signing_key: str | None = None
@field_validator("required_scopes", mode="before")
@classmethod
@ -206,8 +207,7 @@ class GitHubProvider(OAuthProxy):
timeout_seconds: int | NotSetT = NotSet,
allowed_client_redirect_uris: list[str] | NotSetT = NotSet,
client_storage: AsyncKeyValue | None = None,
jwt_signing_key: str | bytes | None = None,
token_encryption_key: str | bytes | None = None,
jwt_signing_key: str | bytes | NotSetT = NotSet,
require_authorization_consent: bool = True,
):
"""Initialize GitHub OAuth provider.
@ -223,13 +223,12 @@ class GitHubProvider(OAuthProxy):
timeout_seconds: HTTP request timeout for GitHub API calls
allowed_client_redirect_uris: List of allowed redirect URI patterns for MCP clients.
If None (default), all URIs are allowed. If empty list, no URIs are allowed.
client_storage: An AsyncKeyValue-compatible store for client registrations, registrations are stored in memory if not provided
jwt_signing_key: Secret for signing FastMCP JWT tokens (any string or bytes).
None (default): Auto-managed via system keyring (Mac/Windows) or ephemeral (Linux).
Explicit value: For production deployments. Recommended to store in environment variable.
token_encryption_key: Secret for encrypting upstream tokens at rest (any string or bytes).
None (default): Auto-managed via system keyring (Mac/Windows) or ephemeral (Linux).
Explicit value: For production deployments. Recommended to store in environment variable.
client_storage: Storage backend for OAuth state (client registrations, encrypted tokens).
If None, a DiskStore will be created in the data directory (derived from `platformdirs`). The
disk store will be encrypted using a key derived from the JWT Signing Key.
jwt_signing_key: Secret for signing FastMCP JWT tokens (any string or bytes). If bytes are provided,
they will be used as is. If a string is provided, it will be derived into a 32-byte key. If not
provided, the upstream client secret will be used to derive a 32-byte key using PBKDF2.
require_authorization_consent: Whether to require user consent before authorizing clients (default True).
When True, users see a consent screen before being redirected to GitHub.
When False, authorization proceeds directly without user confirmation.
@ -248,6 +247,7 @@ class GitHubProvider(OAuthProxy):
"required_scopes": required_scopes,
"timeout_seconds": timeout_seconds,
"allowed_client_redirect_uris": allowed_client_redirect_uris,
"jwt_signing_key": jwt_signing_key,
}.items()
if v is not NotSet
}
@ -293,8 +293,7 @@ class GitHubProvider(OAuthProxy):
or settings.base_url, # Default to base_url if not specified
allowed_client_redirect_uris=allowed_client_redirect_uris_final,
client_storage=client_storage,
jwt_signing_key=jwt_signing_key,
token_encryption_key=token_encryption_key,
jwt_signing_key=settings.jwt_signing_key,
require_authorization_consent=require_authorization_consent,
)

View file

@ -56,6 +56,7 @@ class GoogleProviderSettings(BaseSettings):
required_scopes: list[str] | None = None
timeout_seconds: int | None = None
allowed_client_redirect_uris: list[str] | None = None
jwt_signing_key: str | None = None
@field_validator("required_scopes", mode="before")
@classmethod
@ -222,8 +223,7 @@ class GoogleProvider(OAuthProxy):
timeout_seconds: int | NotSetT = NotSet,
allowed_client_redirect_uris: list[str] | NotSetT = NotSet,
client_storage: AsyncKeyValue | None = None,
jwt_signing_key: str | bytes | None = None,
token_encryption_key: str | bytes | None = None,
jwt_signing_key: str | bytes | NotSetT = NotSet,
require_authorization_consent: bool = True,
):
"""Initialize Google OAuth provider.
@ -242,13 +242,12 @@ class GoogleProvider(OAuthProxy):
timeout_seconds: HTTP request timeout for Google API calls
allowed_client_redirect_uris: List of allowed redirect URI patterns for MCP clients.
If None (default), all URIs are allowed. If empty list, no URIs are allowed.
client_storage: An AsyncKeyValue-compatible store for client registrations, registrations are stored in memory if not provided
jwt_signing_key: Secret for signing FastMCP JWT tokens (any string or bytes).
None (default): Auto-managed via system keyring (Mac/Windows) or ephemeral (Linux).
Explicit value: For production deployments. Recommended to store in environment variable.
token_encryption_key: Secret for encrypting upstream tokens at rest (any string or bytes).
None (default): Auto-managed via system keyring (Mac/Windows) or ephemeral (Linux).
Explicit value: For production deployments. Recommended to store in environment variable.
client_storage: Storage backend for OAuth state (client registrations, encrypted tokens).
If None, a DiskStore will be created in the data directory (derived from `platformdirs`). The
disk store will be encrypted using a key derived from the JWT Signing Key.
jwt_signing_key: Secret for signing FastMCP JWT tokens (any string or bytes). If bytes are provided,
they will be used as is. If a string is provided, it will be derived into a 32-byte key. If not
provided, the upstream client secret will be used to derive a 32-byte key using PBKDF2.
require_authorization_consent: Whether to require user consent before authorizing clients (default True).
When True, users see a consent screen before being redirected to Google.
When False, authorization proceeds directly without user confirmation.
@ -267,6 +266,7 @@ class GoogleProvider(OAuthProxy):
"required_scopes": required_scopes,
"timeout_seconds": timeout_seconds,
"allowed_client_redirect_uris": allowed_client_redirect_uris,
"jwt_signing_key": jwt_signing_key,
}.items()
if v is not NotSet
}
@ -312,8 +312,7 @@ class GoogleProvider(OAuthProxy):
or settings.base_url, # Default to base_url if not specified
allowed_client_redirect_uris=allowed_client_redirect_uris_final,
client_storage=client_storage,
jwt_signing_key=jwt_signing_key,
token_encryption_key=token_encryption_key,
jwt_signing_key=settings.jwt_signing_key,
require_authorization_consent=require_authorization_consent,
)

View file

@ -46,6 +46,7 @@ class WorkOSProviderSettings(BaseSettings):
required_scopes: list[str] | None = None
timeout_seconds: int | None = None
allowed_client_redirect_uris: list[str] | None = None
jwt_signing_key: str | None = None
@field_validator("required_scopes", mode="before")
@classmethod
@ -172,8 +173,7 @@ class WorkOSProvider(OAuthProxy):
timeout_seconds: int | NotSetT = NotSet,
allowed_client_redirect_uris: list[str] | NotSetT = NotSet,
client_storage: AsyncKeyValue | None = None,
jwt_signing_key: str | bytes | None = None,
token_encryption_key: str | bytes | None = None,
jwt_signing_key: str | bytes | NotSetT = NotSet,
require_authorization_consent: bool = True,
):
"""Initialize WorkOS OAuth provider.
@ -190,13 +190,12 @@ class WorkOSProvider(OAuthProxy):
timeout_seconds: HTTP request timeout for WorkOS API calls
allowed_client_redirect_uris: List of allowed redirect URI patterns for MCP clients.
If None (default), all URIs are allowed. If empty list, no URIs are allowed.
client_storage: An AsyncKeyValue-compatible store for client registrations, registrations are stored in memory if not provided
jwt_signing_key: Secret for signing FastMCP JWT tokens (any string or bytes).
None (default): Auto-managed via system keyring (Mac/Windows) or ephemeral (Linux).
Explicit value: For production deployments. Recommended to store in environment variable.
token_encryption_key: Secret for encrypting upstream tokens at rest (any string or bytes).
None (default): Auto-managed via system keyring (Mac/Windows) or ephemeral (Linux).
Explicit value: For production deployments. Recommended to store in environment variable.
client_storage: Storage backend for OAuth state (client registrations, encrypted tokens).
If None, a DiskStore will be created in the data directory (derived from `platformdirs`). The
disk store will be encrypted using a key derived from the JWT Signing Key.
jwt_signing_key: Secret for signing FastMCP JWT tokens (any string or bytes). If bytes are provided,
they will be used as is. If a string is provided, it will be derived into a 32-byte key. If not
provided, the upstream client secret will be used to derive a 32-byte key using PBKDF2.
require_authorization_consent: Whether to require user consent before authorizing clients (default True).
When True, users see a consent screen before being redirected to WorkOS.
When False, authorization proceeds directly without user confirmation.
@ -216,6 +215,7 @@ class WorkOSProvider(OAuthProxy):
"required_scopes": required_scopes,
"timeout_seconds": timeout_seconds,
"allowed_client_redirect_uris": allowed_client_redirect_uris,
"jwt_signing_key": jwt_signing_key,
}.items()
if v is not NotSet
}
@ -269,8 +269,7 @@ class WorkOSProvider(OAuthProxy):
or settings.base_url, # Default to base_url if not specified
allowed_client_redirect_uris=allowed_client_redirect_uris_final,
client_storage=client_storage,
jwt_signing_key=jwt_signing_key,
token_encryption_key=token_encryption_key,
jwt_signing_key=settings.jwt_signing_key,
require_authorization_consent=require_authorization_consent,
)

View file

@ -1,76 +0,0 @@
"""Key management utilities for FastMCP.
Provides automatic key generation and storage in system keyring for
Mac/Windows platforms, with graceful fallback for Linux/headless systems.
"""
from __future__ import annotations
import base64
import platform
import secrets
import keyring
from fastmcp.utilities.logging import get_logger
logger = get_logger(__name__)
def get_or_generate_keyring_key(key_type: str, namespace: str) -> str | None:
"""Get or generate a key from the system keyring.
Keys are namespaced to allow multiple isolated key sets.
Args:
key_type: Type of key (e.g., "jwt-signing", "token-encryption", "api-key")
namespace: Unique identifier for this key set (e.g., client ID, server name)
Returns:
Base64-encoded key string, or None if keyring unavailable
Example:
>>> key = get_or_generate_keyring_key("jwt-signing", "my-github-client-id")
>>> # Returns key from keyring or generates new one
"""
# Linux keyring support is unreliable (GUI sessions, unlock prompts, backend issues)
if platform.system() == "Linux":
return None
service_name = "fastmcp"
# Namespace keys for isolation
key_name = f"{key_type}-{namespace}"
try:
# Try to get existing key from keyring
existing_key = keyring.get_password(service_name, key_name)
if existing_key:
logger.debug(
"Retrieved %s for namespace=%s from system keyring",
key_type,
namespace,
)
return existing_key
# Generate new secure random key (32 bytes for Fernet/HMAC)
key_bytes = secrets.token_bytes(32)
key_b64 = base64.b64encode(key_bytes).decode()
# Store in keyring for future use
keyring.set_password(service_name, key_name, key_b64)
logger.info(
"Generated new %s for namespace=%s and stored in system keyring",
key_type,
namespace,
)
return key_b64
except Exception as e:
# Keyring backend may not be available (headless systems, permissions, etc.)
logger.warning(
"Failed to access system keyring for %s: %s. "
"Will use ephemeral key (tokens will not survive restart).",
key_type,
e,
)
return None

View file

@ -1,7 +1,6 @@
import socket
from collections.abc import Callable
from typing import Any
from unittest.mock import patch
import pytest
@ -22,19 +21,6 @@ def import_rich_rule():
yield
@pytest.fixture(autouse=True)
def mock_keyring():
"""Globally mock keyring to prevent OS keyring pollution during tests.
This prevents any test from accidentally writing to the system keyring.
Individual tests can override this mock if they need to test keyring behavior.
"""
with patch("fastmcp.utilities.key_management.keyring") as mock:
# Return None by default (keyring unavailable)
mock.get_password.return_value = None
yield mock
def get_fn_name(fn: Callable[..., Any]) -> str:
return fn.__name__ # ty: ignore[unresolved-attribute]

View file

@ -213,12 +213,6 @@ class TestPerformanceComparison:
f"Legacy should also be fast on small specs, got {legacy_avg:.4f}s"
)
# Performance should be comparable (within reasonable margin)
performance_ratio = max(new_avg, legacy_avg) / min(new_avg, legacy_avg)
assert performance_ratio < 3.0, (
f"Performance should be comparable, ratio: {performance_ratio:.2f}x"
)
def test_functionality_identical_after_optimization(self, comprehensive_spec):
"""Verify that performance optimization doesn't break functionality."""
client = httpx.AsyncClient(base_url="https://api.example.com")

View file

@ -46,6 +46,7 @@ def create_github_server(base_url: str) -> FastMCP:
client_id=FASTMCP_TEST_AUTH_GITHUB_CLIENT_ID,
client_secret=FASTMCP_TEST_AUTH_GITHUB_CLIENT_SECRET,
base_url=base_url,
jwt_signing_key="test-secret",
)
# Create FastMCP server with GitHub authentication
@ -74,6 +75,7 @@ def create_github_server_with_mock_callback(base_url: str) -> FastMCP:
client_id=FASTMCP_TEST_AUTH_GITHUB_CLIENT_ID,
client_secret=FASTMCP_TEST_AUTH_GITHUB_CLIENT_SECRET,
base_url=base_url,
jwt_signing_key="test-secret",
)
# Mock the authorize method to return a fake code instead of redirecting to GitHub

View file

@ -49,6 +49,7 @@ class TestAuth0ProviderSettings:
"FASTMCP_SERVER_AUTH_AUTH0_REQUIRED_SCOPES": ",".join(
TEST_REQUIRED_SCOPES
),
"FASTMCP_SERVER_AUTH_AUTH0_JWT_SIGNING_KEY": "test-secret",
},
):
settings = Auth0ProviderSettings()
@ -108,6 +109,7 @@ class TestAuth0Provider:
base_url=TEST_BASE_URL,
redirect_path=TEST_REDIRECT_PATH,
required_scopes=TEST_REQUIRED_SCOPES,
jwt_signing_key="test-secret",
)
mock_get.assert_called_once()
@ -147,6 +149,7 @@ class TestAuth0Provider:
"FASTMCP_SERVER_AUTH_AUTH0_AUDIENCE": TEST_AUDIENCE,
"FASTMCP_SERVER_AUTH_AUTH0_BASE_URL": TEST_BASE_URL,
"FASTMCP_SERVER_AUTH_AUTH0_REQUIRED_SCOPES": scopes_env,
"FASTMCP_SERVER_AUTH_AUTH0_JWT_SIGNING_KEY": "test-secret",
},
),
patch(
@ -188,6 +191,7 @@ class TestAuth0Provider:
"FASTMCP_SERVER_AUTH_AUTH0_CLIENT_SECRET": TEST_CLIENT_SECRET,
"FASTMCP_SERVER_AUTH_AUTH0_AUDIENCE": TEST_AUDIENCE,
"FASTMCP_SERVER_AUTH_AUTH0_BASE_URL": TEST_BASE_URL,
"FASTMCP_SERVER_AUTH_AUTH0_JWT_SIGNING_KEY": "test-secret",
},
),
patch(
@ -239,6 +243,7 @@ class TestAuth0Provider:
config_url=TEST_CONFIG_URL,
client_id=TEST_CLIENT_ID,
client_secret=TEST_CLIENT_SECRET,
jwt_signing_key="test-secret",
)
def test_init_missing_base_url_raises_error(self):
@ -251,6 +256,7 @@ class TestAuth0Provider:
client_id=TEST_CLIENT_ID,
client_secret=TEST_CLIENT_SECRET,
audience=TEST_AUDIENCE,
jwt_signing_key="test-secret",
)
def test_init_defaults(self, valid_oidc_configuration_dict):
@ -269,6 +275,7 @@ class TestAuth0Provider:
client_secret=TEST_CLIENT_SECRET,
audience=TEST_AUDIENCE,
base_url=TEST_BASE_URL,
jwt_signing_key="test-secret",
)
# Check defaults

View file

@ -106,6 +106,7 @@ class TestAWSCognitoProvider:
base_url="https://example.com",
redirect_path="/custom/callback",
required_scopes=["openid", "email"],
jwt_signing_key="test-secret",
)
# Check that the provider was initialized correctly
@ -143,6 +144,7 @@ class TestAWSCognitoProvider:
"FASTMCP_SERVER_AUTH_AWS_COGNITO_CLIENT_SECRET": "env_secret",
"FASTMCP_SERVER_AUTH_AWS_COGNITO_BASE_URL": "https://env-example.com",
"FASTMCP_SERVER_AUTH_AWS_COGNITO_REQUIRED_SCOPES": scopes_env,
"FASTMCP_SERVER_AUTH_AWS_COGNITO_JWT_SIGNING_KEY": "test-secret",
},
):
with mock_cognito_oidc_discovery():
@ -163,6 +165,7 @@ class TestAWSCognitoProvider:
"FASTMCP_SERVER_AUTH_AWS_COGNITO_USER_POOL_ID": "env_pool_id",
"FASTMCP_SERVER_AUTH_AWS_COGNITO_CLIENT_ID": "env_client_id",
"FASTMCP_SERVER_AUTH_AWS_COGNITO_CLIENT_SECRET": "env_secret",
"FASTMCP_SERVER_AUTH_AWS_COGNITO_JWT_SIGNING_KEY": "test-secret",
},
):
with mock_cognito_oidc_discovery():
@ -171,6 +174,7 @@ class TestAWSCognitoProvider:
client_id="explicit_client",
client_secret="explicit_secret",
base_url="https://example.com",
jwt_signing_key="test-secret",
)
assert provider._upstream_client_id == "explicit_client"
@ -216,6 +220,7 @@ class TestAWSCognitoProvider:
client_id="test_client",
client_secret="test_secret",
base_url="https://example.com",
jwt_signing_key="test-secret",
)
# Check defaults
@ -233,6 +238,7 @@ class TestAWSCognitoProvider:
client_id="test_client",
client_secret="test_secret",
base_url="https://example.com",
jwt_signing_key="test-secret",
)
# OIDC discovery should have configured the endpoints automatically

View file

@ -24,6 +24,7 @@ class TestAzureProvider:
tenant_id="87654321-4321-4321-4321-210987654321",
base_url="https://myserver.com",
required_scopes=["read", "write"],
jwt_signing_key="test-secret",
)
assert provider._upstream_client_id == "12345678-1234-1234-1234-123456789012"
@ -52,6 +53,7 @@ class TestAzureProvider:
"FASTMCP_SERVER_AUTH_AZURE_TENANT_ID": "env-tenant-id",
"FASTMCP_SERVER_AUTH_AZURE_BASE_URL": "https://envserver.com",
"FASTMCP_SERVER_AUTH_AZURE_REQUIRED_SCOPES": scopes_env,
"FASTMCP_SERVER_AUTH_AZURE_JWT_SIGNING_KEY": "test-secret",
},
):
provider = AzureProvider()
@ -101,6 +103,7 @@ class TestAzureProvider:
client_secret="test_secret",
tenant_id="test-tenant",
required_scopes=["read"],
jwt_signing_key="test-secret",
)
# Check defaults
@ -116,6 +119,7 @@ class TestAzureProvider:
tenant_id="my-tenant-id",
base_url="https://myserver.com",
required_scopes=["read"],
jwt_signing_key="test_secret",
)
# Check that endpoints use the correct Azure OAuth2 v2.0 endpoints with tenant
@ -139,6 +143,7 @@ class TestAzureProvider:
client_secret="test_secret",
tenant_id="organizations",
required_scopes=["read"],
jwt_signing_key="test-secret",
)
parsed = urlparse(provider1._upstream_authorization_endpoint)
assert "/organizations/" in parsed.path
@ -149,6 +154,7 @@ class TestAzureProvider:
client_secret="test_secret",
tenant_id="consumers",
required_scopes=["read"],
jwt_signing_key="test-secret",
)
parsed = urlparse(provider2._upstream_authorization_endpoint)
assert "/consumers/" in parsed.path
@ -165,6 +171,7 @@ class TestAzureProvider:
"write",
"admin",
],
jwt_signing_key="test-secret",
)
# Provider should initialize successfully with these scopes
@ -183,6 +190,7 @@ class TestAzureProvider:
client_secret="test_secret",
tenant_id="test-tenant",
required_scopes=["read"],
jwt_signing_key="test-secret",
)
assert provider is not None
@ -194,6 +202,7 @@ class TestAzureProvider:
tenant_id="my-tenant",
identifier_uri="api://my-api",
required_scopes=[".default"],
jwt_signing_key="test-secret",
)
assert provider._token_validator is not None
@ -208,7 +217,6 @@ class TestAzureProvider:
# Scopes should be prefixed with identifier_uri
assert verifier.required_scopes == ["api://my-api/.default"]
@pytest.mark.asyncio
async def test_authorize_filters_resource_and_accepts_prefixed_scopes(self):
"""authorize() should drop resource parameter and accept prefixed scopes from clients."""
provider = AzureProvider(
@ -218,6 +226,7 @@ class TestAzureProvider:
identifier_uri="api://my-api",
required_scopes=["read", "write"],
base_url="https://srv.example",
jwt_signing_key="test-secret",
)
await provider.register_client(
@ -262,7 +271,6 @@ class TestAzureProvider:
# Azure provider filters resource parameter (not stored in transaction)
assert transaction.resource is None
@pytest.mark.asyncio
async def test_authorize_appends_additional_scopes(self):
"""authorize() should append additional_authorize_scopes to the authorization request."""
provider = AzureProvider(
@ -273,6 +281,7 @@ class TestAzureProvider:
required_scopes=["read"],
base_url="https://srv.example",
additional_authorize_scopes=["Mail.Read", "User.Read"],
jwt_signing_key="test-secret",
)
await provider.register_client(

View file

@ -25,6 +25,7 @@ class TestGitHubProviderSettings:
"FASTMCP_SERVER_AUTH_GITHUB_BASE_URL": "https://example.com",
"FASTMCP_SERVER_AUTH_GITHUB_REDIRECT_PATH": "/custom/callback",
"FASTMCP_SERVER_AUTH_GITHUB_TIMEOUT_SECONDS": "30",
"FASTMCP_SERVER_AUTH_GITHUB_JWT_SIGNING_KEY": "test-secret",
},
):
settings = GitHubProviderSettings()
@ -51,6 +52,7 @@ class TestGitHubProviderSettings:
{
"client_id": "explicit_client_id",
"client_secret": "explicit_secret",
"jwt_signing_key": "test-secret",
}
)
@ -73,6 +75,7 @@ class TestGitHubProvider:
redirect_path="/custom/callback",
required_scopes=["user", "repo"],
timeout_seconds=30,
jwt_signing_key="test-secret",
)
# Check that the provider was initialized correctly
@ -99,6 +102,7 @@ class TestGitHubProvider:
"FASTMCP_SERVER_AUTH_GITHUB_CLIENT_SECRET": "env_secret",
"FASTMCP_SERVER_AUTH_GITHUB_BASE_URL": "https://env-example.com",
"FASTMCP_SERVER_AUTH_GITHUB_REQUIRED_SCOPES": scopes_env,
"FASTMCP_SERVER_AUTH_GITHUB_JWT_SIGNING_KEY": "test-secret",
},
):
provider = GitHubProvider()
@ -115,11 +119,13 @@ class TestGitHubProvider:
{
"FASTMCP_SERVER_AUTH_GITHUB_CLIENT_ID": "env_client_id",
"FASTMCP_SERVER_AUTH_GITHUB_CLIENT_SECRET": "env_secret",
"FASTMCP_SERVER_AUTH_GITHUB_JWT_SIGNING_KEY": "test-secret",
},
):
provider = GitHubProvider(
client_id="explicit_client",
client_secret="explicit_secret",
jwt_signing_key="test-secret",
)
assert provider._upstream_client_id == "explicit_client"
@ -146,6 +152,7 @@ class TestGitHubProvider:
provider = GitHubProvider(
client_id="test_client",
client_secret="test_secret",
jwt_signing_key="test-secret",
)
# Check defaults

View file

@ -18,6 +18,7 @@ class TestGoogleProvider:
client_secret="GOCSPX-test123",
base_url="https://myserver.com",
required_scopes=["openid", "email", "profile"],
jwt_signing_key="test-secret",
)
assert provider._upstream_client_id == "123456789.apps.googleusercontent.com"
@ -40,6 +41,7 @@ class TestGoogleProvider:
"FASTMCP_SERVER_AUTH_GOOGLE_CLIENT_SECRET": "GOCSPX-env456",
"FASTMCP_SERVER_AUTH_GOOGLE_BASE_URL": "https://envserver.com",
"FASTMCP_SERVER_AUTH_GOOGLE_REQUIRED_SCOPES": scopes_env,
"FASTMCP_SERVER_AUTH_GOOGLE_JWT_SIGNING_KEY": "test-secret",
},
):
provider = GoogleProvider()
@ -73,6 +75,7 @@ class TestGoogleProvider:
provider = GoogleProvider(
client_id="123456789.apps.googleusercontent.com",
client_secret="GOCSPX-test123",
jwt_signing_key="test-secret",
)
# Check defaults
@ -86,6 +89,7 @@ class TestGoogleProvider:
client_id="123456789.apps.googleusercontent.com",
client_secret="GOCSPX-test123",
base_url="https://myserver.com",
jwt_signing_key="test-secret",
)
# Check that endpoints use Google's OAuth2 endpoints
@ -110,6 +114,7 @@ class TestGoogleProvider:
"https://www.googleapis.com/auth/userinfo.email",
"https://www.googleapis.com/auth/userinfo.profile",
],
jwt_signing_key="test-secret",
)
# Provider should initialize successfully with these scopes

View file

@ -24,6 +24,7 @@ class TestWorkOSProvider:
authkit_domain="https://test.authkit.app",
base_url="https://myserver.com",
required_scopes=["openid", "profile"],
jwt_signing_key="test-secret",
)
assert provider._upstream_client_id == "client_test123"
@ -47,6 +48,7 @@ class TestWorkOSProvider:
"FASTMCP_SERVER_AUTH_WORKOS_AUTHKIT_DOMAIN": "https://env.authkit.app",
"FASTMCP_SERVER_AUTH_WORKOS_BASE_URL": "https://envserver.com",
"FASTMCP_SERVER_AUTH_WORKOS_REQUIRED_SCOPES": scopes_env,
"FASTMCP_SERVER_AUTH_WORKOS_JWT_SIGNING_KEY": "test-secret",
},
):
provider = WorkOSProvider()
@ -91,6 +93,7 @@ class TestWorkOSProvider:
client_secret="test_secret",
authkit_domain="test.authkit.app",
base_url="https://myserver.com",
jwt_signing_key="test-secret",
)
parsed = urlparse(provider1._upstream_authorization_endpoint)
assert parsed.scheme == "https"
@ -103,6 +106,7 @@ class TestWorkOSProvider:
client_secret="test_secret",
authkit_domain="https://test.authkit.app",
base_url="https://myserver.com",
jwt_signing_key="test-secret",
)
parsed = urlparse(provider2._upstream_authorization_endpoint)
assert parsed.scheme == "https"
@ -115,6 +119,7 @@ class TestWorkOSProvider:
client_secret="test_secret",
authkit_domain="http://localhost:8080",
base_url="https://myserver.com",
jwt_signing_key="test-secret",
)
parsed = urlparse(provider3._upstream_authorization_endpoint)
assert parsed.scheme == "http"
@ -127,6 +132,7 @@ class TestWorkOSProvider:
client_id="test_client",
client_secret="test_secret",
authkit_domain="https://test.authkit.app",
jwt_signing_key="test-secret",
)
# Check defaults
@ -141,6 +147,7 @@ class TestWorkOSProvider:
client_secret="test_secret",
authkit_domain="https://test.authkit.app",
base_url="https://myserver.com",
jwt_signing_key="test-secret",
)
# Check that endpoints use the authkit domain

View file

@ -41,6 +41,7 @@ class TestEnhancedAuthorizationHandler:
base_url="https://test.com",
),
base_url="https://myserver.com",
jwt_signing_key="test-secret",
)
def test_unregistered_client_returns_html_for_browser(self, oauth_proxy):
@ -301,6 +302,7 @@ class TestContentNegotiation:
base_url="https://test.com",
),
base_url="https://myserver.com",
jwt_signing_key="test-secret",
)
def test_html_preferred_when_both_accepted(self, oauth_proxy):

View file

@ -8,8 +8,6 @@ from authlib.jose.errors import JoseError
from fastmcp.server.auth.jwt_issuer import (
JWTIssuer,
TokenEncryption,
derive_encryption_key,
derive_jwt_key,
)
@ -19,59 +17,52 @@ class TestKeyDerivation:
def test_derive_jwt_key_produces_32_bytes(self):
"""Test that JWT key derivation produces 32-byte key."""
key = derive_jwt_key("test-secret", "test-salt")
assert len(key) == 32
key = derive_jwt_key(high_entropy_material="test-secret", salt="test-salt")
assert len(key) == 44
assert isinstance(key, bytes)
# base64 decode and make sure its 32 bytes
key_bytes = base64.urlsafe_b64decode(key)
assert len(key_bytes) == 32
key = derive_jwt_key(low_entropy_material="test-secret", salt="test-salt")
assert len(key) == 44
assert isinstance(key, bytes)
# base64 decode and make sure its 32 bytes
key_bytes = base64.urlsafe_b64decode(key)
assert len(key_bytes) == 32
def test_derive_jwt_key_with_different_secrets_produces_different_keys(self):
"""Test that different secrets produce different keys."""
key1 = derive_jwt_key("secret1", "salt")
key2 = derive_jwt_key("secret2", "salt")
key1 = derive_jwt_key(high_entropy_material="secret1", salt="salt")
key2 = derive_jwt_key(high_entropy_material="secret2", salt="salt")
assert key1 != key2
key1 = derive_jwt_key(low_entropy_material="secret1", salt="salt")
key2 = derive_jwt_key(low_entropy_material="secret2", salt="salt")
assert key1 != key2
def test_derive_jwt_key_with_different_salts_produces_different_keys(self):
"""Test that different salts produce different keys."""
key1 = derive_jwt_key("secret", "salt1")
key2 = derive_jwt_key("secret", "salt2")
key1 = derive_jwt_key(high_entropy_material="secret", salt="salt1")
key2 = derive_jwt_key(high_entropy_material="secret", salt="salt2")
assert key1 != key2
key1 = derive_jwt_key(low_entropy_material="secret", salt="salt1")
key2 = derive_jwt_key(low_entropy_material="secret", salt="salt2")
assert key1 != key2
def test_derive_jwt_key_is_deterministic(self):
"""Test that same inputs always produce same key."""
key1 = derive_jwt_key("secret", "salt")
key2 = derive_jwt_key("secret", "salt")
key1 = derive_jwt_key(high_entropy_material="secret", salt="salt")
key2 = derive_jwt_key(high_entropy_material="secret", salt="salt")
assert key1 == key2
def test_derive_encryption_key_produces_base64_key(self):
"""Test that encryption key is base64url-encoded."""
key = derive_encryption_key("test-secret")
assert len(key) == 44 # 32 bytes base64url-encoded = 44 chars
assert isinstance(key, bytes)
# Should be valid base64url (no padding issues)
import base64
decoded = base64.urlsafe_b64decode(key)
assert len(decoded) == 32
def test_derive_encryption_key_with_different_secrets_produces_different_keys(
self,
):
"""Test that different secrets produce different encryption keys."""
key1 = derive_encryption_key("secret1")
key2 = derive_encryption_key("secret2")
assert key1 != key2
def test_derive_encryption_key_is_deterministic(self):
"""Test that same input always produces same encryption key."""
key1 = derive_encryption_key("secret")
key2 = derive_encryption_key("secret")
key1 = derive_jwt_key(low_entropy_material="secret", salt="salt")
key2 = derive_jwt_key(low_entropy_material="secret", salt="salt")
assert key1 == key2
def test_jwt_and_encryption_keys_are_different(self):
"""Test that JWT and encryption keys derived from same secret are different."""
jwt_key = derive_jwt_key("secret", "salt")
enc_key_raw = base64.urlsafe_b64decode(derive_encryption_key("secret"))
assert jwt_key != enc_key_raw
class TestJWTIssuer:
"""Tests for JWT token issuance and verification."""
@ -79,7 +70,9 @@ class TestJWTIssuer:
@pytest.fixture
def issuer(self):
"""Create a JWT issuer for testing."""
signing_key = derive_jwt_key("test-secret", "test-salt")
signing_key = derive_jwt_key(
low_entropy_material="test-secret", salt="test-salt"
)
return JWTIssuer(
issuer="https://test-server.com",
audience="https://test-server.com/mcp",
@ -156,7 +149,9 @@ class TestJWTIssuer:
)
# Try to verify with different issuer (different key)
other_key = derive_jwt_key("different-secret", "different-salt")
other_key = derive_jwt_key(
low_entropy_material="different-secret", salt="different-salt"
)
other_issuer = JWTIssuer(
issuer="https://test-server.com",
audience="https://test-server.com/mcp",
@ -233,65 +228,3 @@ class TestJWTIssuer:
with pytest.raises(JoseError):
issuer.verify_token("header.payload") # Missing signature
class TestTokenEncryption:
"""Tests for token encryption/decryption."""
@pytest.fixture
def encryption(self):
"""Create token encryption instance for testing."""
key = derive_encryption_key("test-secret")
return TokenEncryption(key)
def test_encrypt_decrypt_roundtrip(self, encryption):
"""Test that encryption and decryption work correctly."""
plaintext = "sensitive-token-value"
encrypted = encryption.encrypt(plaintext)
decrypted = encryption.decrypt(encrypted)
assert decrypted == plaintext
def test_encrypt_produces_different_ciphertext_each_time(self, encryption):
"""Test that encrypting the same plaintext produces different ciphertext."""
plaintext = "token-value"
ciphertext1 = encryption.encrypt(plaintext)
ciphertext2 = encryption.encrypt(plaintext)
# Fernet includes timestamp and IV, so ciphertext differs each time
assert ciphertext1 != ciphertext2
# But both decrypt to same plaintext
assert encryption.decrypt(ciphertext1) == plaintext
assert encryption.decrypt(ciphertext2) == plaintext
def test_decrypt_with_wrong_key_fails(self, encryption):
"""Test that decryption with wrong key fails."""
plaintext = "token-value"
encrypted = encryption.encrypt(plaintext)
# Create different encryption instance with different key
other_key = derive_encryption_key("different-secret")
other_encryption = TokenEncryption(other_key)
from cryptography.fernet import InvalidToken
with pytest.raises(InvalidToken):
other_encryption.decrypt(encrypted)
def test_encrypt_handles_unicode(self, encryption):
"""Test that encryption handles unicode strings correctly."""
plaintext = "token-with-émojis-🔒"
encrypted = encryption.encrypt(plaintext)
decrypted = encryption.decrypt(encrypted)
assert decrypted == plaintext
def test_decrypt_rejects_tampered_ciphertext(self, encryption):
"""Test that tampered ciphertext is rejected."""
plaintext = "token-value"
encrypted = encryption.encrypt(plaintext)
# Tamper with ciphertext
tampered = encrypted[:-1] + b"X"
from cryptography.fernet import InvalidToken
with pytest.raises(InvalidToken):
encryption.decrypt(tampered)

View file

@ -78,6 +78,7 @@ def oauth_proxy_with_storage(storage):
base_url="https://myserver.com",
redirect_path="/auth/callback",
client_storage=storage, # Use our test storage
jwt_signing_key="test-secret",
)
@ -92,6 +93,7 @@ def oauth_proxy_https():
token_verifier=_Verifier(),
base_url="https://myserver.example",
client_storage=MemoryStore(),
jwt_signing_key="test-secret",
)
@ -681,6 +683,7 @@ class TestConsentPageServerIcon:
upstream_client_secret="upstream-secret",
token_verifier=verifier,
base_url="https://proxy.example.com",
jwt_signing_key="test-secret",
)
# Create FastMCP server with custom icon
@ -752,6 +755,7 @@ class TestConsentPageServerIcon:
upstream_client_secret="upstream-secret",
token_verifier=verifier,
base_url="https://proxy.example.com",
jwt_signing_key="test-secret",
)
# Create FastMCP server without icon
@ -818,6 +822,7 @@ class TestConsentPageServerIcon:
upstream_client_secret="upstream-secret",
token_verifier=verifier,
base_url="https://proxy.example.com",
jwt_signing_key="test-secret",
)
# Create FastMCP server with special characters in name

View file

@ -319,6 +319,7 @@ def oauth_proxy(jwt_verifier):
token_verifier=jwt_verifier,
base_url="https://myserver.com",
redirect_path="/auth/callback",
jwt_signing_key="test-secret",
)
@ -348,6 +349,7 @@ class TestOAuthProxyInitialization:
upstream_client_secret="secret-456",
token_verifier=jwt_verifier,
base_url="https://api.example.com",
jwt_signing_key="test-secret",
)
assert (
@ -376,6 +378,7 @@ class TestOAuthProxyInitialization:
valid_scopes=["custom", "scopes"],
forward_pkce=False,
token_endpoint_auth_method="client_secret_post",
jwt_signing_key="test-secret",
)
assert proxy._upstream_revocation_endpoint == "https://auth.example.com/revoke"
@ -395,6 +398,7 @@ class TestOAuthProxyInitialization:
token_verifier=jwt_verifier,
base_url="https://api.com",
redirect_path="auth/callback", # No leading slash
jwt_signing_key="test-secret",
)
assert proxy._redirect_path == "/auth/callback"
@ -446,6 +450,7 @@ class TestOAuthProxyAuthorization:
client_id="test-client",
client_secret="test-secret",
redirect_uris=[AnyUrl("http://localhost:54321/callback")],
jwt_signing_key="test-secret",
)
# Register client first (required for consent flow)
@ -493,6 +498,7 @@ class TestOAuthProxyPKCE:
token_verifier=jwt_verifier,
base_url="https://proxy.example.com",
forward_pkce=True,
jwt_signing_key="test-secret",
)
@pytest.fixture
@ -505,6 +511,7 @@ class TestOAuthProxyPKCE:
token_verifier=jwt_verifier,
base_url="https://proxy.example.com",
forward_pkce=False,
jwt_signing_key="test-secret",
)
async def test_pkce_forwarding_enabled(self, proxy_with_pkce):
@ -591,6 +598,7 @@ class TestOAuthProxyTokenEndpointAuth:
token_verifier=jwt_verifier,
base_url="https://proxy.example.com",
token_endpoint_auth_method="client_secret_post",
jwt_signing_key="test-secret",
)
assert proxy_post._token_endpoint_auth_method == "client_secret_post"
@ -603,6 +611,7 @@ class TestOAuthProxyTokenEndpointAuth:
token_verifier=jwt_verifier,
base_url="https://proxy.example.com",
token_endpoint_auth_method="client_secret_basic",
jwt_signing_key="test-secret",
)
assert proxy_basic._token_endpoint_auth_method == "client_secret_basic"
@ -614,6 +623,7 @@ class TestOAuthProxyTokenEndpointAuth:
upstream_client_secret="secret",
token_verifier=jwt_verifier,
base_url="https://proxy.example.com",
jwt_signing_key="test-secret",
)
assert proxy_default._token_endpoint_auth_method is None
@ -627,6 +637,7 @@ class TestOAuthProxyTokenEndpointAuth:
token_verifier=jwt_verifier,
base_url="https://proxy.example.com",
token_endpoint_auth_method="client_secret_post",
jwt_signing_key="test-secret",
)
# First, create a valid FastMCP token via full OAuth flow
@ -747,6 +758,7 @@ class TestOAuthProxyE2E:
upstream_client_secret="mock-secret",
token_verifier=MockTokenVerifier(),
base_url="http://localhost:8000",
jwt_signing_key="test-secret",
)
# Create FastMCP server with proxy
@ -800,6 +812,7 @@ class TestOAuthProxyE2E:
upstream_client_secret="mock-secret",
token_verifier=MockTokenVerifier(),
base_url="http://localhost:8000",
jwt_signing_key="test-secret",
)
client = OAuthClientInformationFull(
@ -915,6 +928,7 @@ class TestOAuthProxyE2E:
token_verifier=MockTokenVerifier(),
base_url="http://localhost:8000",
forward_pkce=True, # Enable PKCE forwarding
jwt_signing_key="test-secret",
)
client = OAuthClientInformationFull(
@ -970,6 +984,7 @@ class TestParameterForwarding:
base_url="https://proxy.example.com",
extra_authorize_params={"audience": "https://api.example.com"},
extra_token_params={"audience": "https://api.example.com"},
jwt_signing_key="test-secret",
)
@pytest.fixture
@ -982,6 +997,7 @@ class TestParameterForwarding:
upstream_client_secret="upstream-secret",
token_verifier=jwt_verifier,
base_url="https://proxy.example.com",
jwt_signing_key="test-secret",
)
async def test_resource_parameter_forwarding(self, proxy_without_extra_params):
@ -1133,6 +1149,7 @@ class TestParameterForwarding:
"prompt": "consent",
"max_age": "3600",
},
jwt_signing_key="test-secret",
)
client = OAuthClientInformationFull(
@ -1189,6 +1206,7 @@ class TestParameterForwarding:
upstream_client_secret="upstream-secret",
token_verifier=jwt_verifier,
base_url="https://proxy.example.com",
jwt_signing_key="test-secret",
)
# Create a test app with OAuth routes

View file

@ -110,6 +110,7 @@ class TestOAuthProxyRedirectValidation:
upstream_client_secret="test-secret",
token_verifier=MockTokenVerifier(),
base_url="http://localhost:8000",
jwt_signing_key="test-secret",
)
# The proxy should store None for default (allow all)
@ -127,6 +128,7 @@ class TestOAuthProxyRedirectValidation:
token_verifier=MockTokenVerifier(),
base_url="http://localhost:8000",
allowed_client_redirect_uris=custom_patterns,
jwt_signing_key="test-secret",
)
assert proxy._allowed_client_redirect_uris == custom_patterns
@ -141,6 +143,7 @@ class TestOAuthProxyRedirectValidation:
token_verifier=MockTokenVerifier(),
base_url="http://localhost:8000",
allowed_client_redirect_uris=[],
jwt_signing_key="test-secret",
)
assert proxy._allowed_client_redirect_uris == []
@ -157,6 +160,7 @@ class TestOAuthProxyRedirectValidation:
token_verifier=MockTokenVerifier(),
base_url="http://localhost:8000",
allowed_client_redirect_uris=custom_patterns,
jwt_signing_key="test-secret",
)
# Register a client
@ -189,6 +193,7 @@ class TestOAuthProxyRedirectValidation:
token_verifier=MockTokenVerifier(),
base_url="http://localhost:8000",
allowed_client_redirect_uris=custom_patterns,
jwt_signing_key="test-secret",
)
# Get an unregistered client

View file

@ -1,18 +1,19 @@
"""Tests for OAuth proxy with persistent storage."""
import platform
from collections.abc import AsyncGenerator
from pathlib import Path
from unittest.mock import AsyncMock, Mock, patch
from unittest.mock import AsyncMock, Mock
import pytest
from diskcache.core import tempfile
from inline_snapshot import snapshot
from key_value.aio.stores.disk import DiskStore, MultiDiskStore
from key_value.aio.protocols import AsyncKeyValue
from key_value.aio.stores.disk import MultiDiskStore
from key_value.aio.stores.memory import MemoryStore
from mcp.shared.auth import OAuthClientInformationFull
from pydantic import AnyUrl
from fastmcp.server.auth.auth import TokenVerifier
from fastmcp.server.auth.oauth_proxy import OAuthProxy
@ -40,7 +41,9 @@ class TestOAuthProxyStorage:
"""Create in-memory storage for testing."""
return MemoryStore()
def create_proxy(self, jwt_verifier, storage=None) -> OAuthProxy:
def create_proxy(
self, jwt_verifier: TokenVerifier, storage: AsyncKeyValue | None = None
) -> OAuthProxy:
"""Create an OAuth proxy with specified storage."""
return OAuthProxy(
upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
@ -51,18 +54,9 @@ class TestOAuthProxyStorage:
base_url="https://myserver.com",
redirect_path="/auth/callback",
client_storage=storage,
jwt_signing_key="test-secret",
)
async def test_default_storage_is_platform_appropriate(self, jwt_verifier):
"""Test that proxy defaults to appropriate storage for platform."""
proxy = self.create_proxy(jwt_verifier, storage=None)
if platform.system() == "Linux":
# Linux: no keyring support, use MemoryStore
assert isinstance(proxy._client_storage, MemoryStore)
else:
# Mac/Windows: keyring available, use DiskStore
assert isinstance(proxy._client_storage, DiskStore)
async def test_register_and_get_client(self, jwt_verifier, temp_storage):
"""Test registering and retrieving a client."""
proxy = self.create_proxy(jwt_verifier, storage=temp_storage)
@ -85,7 +79,7 @@ class TestOAuthProxyStorage:
assert client.scope == "read write"
async def test_client_persists_across_proxy_instances(
self, jwt_verifier, temp_storage
self, jwt_verifier: TokenVerifier, temp_storage: AsyncKeyValue
):
"""Test that clients persist when proxy is recreated."""
# First proxy registers client
@ -105,14 +99,16 @@ class TestOAuthProxyStorage:
assert client.client_secret == "persistent-secret"
assert client.scope == "openid profile"
async def test_nonexistent_client_returns_none(self, jwt_verifier, temp_storage):
async def test_nonexistent_client_returns_none(
self, jwt_verifier: TokenVerifier, temp_storage: AsyncKeyValue
):
"""Test that requesting non-existent client returns None."""
proxy = self.create_proxy(jwt_verifier, storage=temp_storage)
client = await proxy.get_client("does-not-exist")
assert client is None
async def test_proxy_dcr_client_redirect_validation(
self, jwt_verifier, temp_storage
self, jwt_verifier: TokenVerifier, temp_storage: AsyncKeyValue
):
"""Test that ProxyDCRClient is created with redirect URI patterns."""
proxy = OAuthProxy(
@ -124,6 +120,7 @@ class TestOAuthProxyStorage:
base_url="https://myserver.com",
allowed_client_redirect_uris=["http://localhost:*"],
client_storage=temp_storage,
jwt_signing_key="test-secret",
)
client_info = OAuthClientInformationFull(
@ -208,226 +205,3 @@ class TestOAuthProxyStorage:
"allowed_redirect_uri_patterns": None,
}
)
class TestOAuthProxyKeyring:
"""Tests for OAuth proxy keyring integration.
All tests mock keyring to prevent pollution of the OS keyring during testing.
"""
@pytest.fixture
def jwt_verifier(self):
"""Create a mock JWT verifier."""
verifier = Mock()
verifier.required_scopes = ["read", "write"]
verifier.verify_token = AsyncMock(return_value=None)
return verifier
@pytest.fixture
def memory_storage(self) -> MemoryStore:
"""Create in-memory storage for testing."""
return MemoryStore()
@patch("fastmcp.utilities.key_management.platform.system")
@patch("fastmcp.utilities.key_management.keyring")
async def test_keyring_used_on_mac_windows(
self, mock_keyring, mock_platform, jwt_verifier, memory_storage
):
"""Test that keyring is used on Mac/Windows platforms."""
# Simulate Mac platform
mock_platform.return_value = "Darwin"
# Mock keyring to return None (first time, no existing key)
mock_keyring.get_password.return_value = None
# Create proxy without explicit keys (should use keyring)
proxy = OAuthProxy(
upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
upstream_token_endpoint="https://github.com/login/oauth/access_token",
upstream_client_id="test-keyring-client",
upstream_client_secret="test-secret",
token_verifier=jwt_verifier,
base_url="https://myserver.com",
client_storage=memory_storage,
)
# Trigger JWT initialization to activate keyring calls
await proxy._ensure_jwt_initialized()
# Verify keyring was accessed for both JWT and encryption keys
assert mock_keyring.get_password.call_count == 2
assert mock_keyring.set_password.call_count == 2
# Verify service name and key names
jwt_calls = [
call
for call in mock_keyring.get_password.call_args_list
if "jwt-signing" in str(call)
]
encryption_calls = [
call
for call in mock_keyring.get_password.call_args_list
if "token-encryption" in str(call)
]
assert len(jwt_calls) == 1
assert len(encryption_calls) == 1
# Check that keys were stored with correct service name
set_calls = mock_keyring.set_password.call_args_list
for call in set_calls:
assert call[0][0] == "fastmcp" # service name
assert "test-keyring-client" in call[0][1] # namespace in key name
@patch("fastmcp.utilities.key_management.platform.system")
@patch("fastmcp.utilities.key_management.keyring")
async def test_keyring_skipped_on_linux(
self, mock_keyring, mock_platform, jwt_verifier, memory_storage
):
"""Test that keyring is skipped on Linux platforms."""
# Simulate Linux platform
mock_platform.return_value = "Linux"
# Create proxy without explicit keys
OAuthProxy(
upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
upstream_token_endpoint="https://github.com/login/oauth/access_token",
upstream_client_id="linux-client",
upstream_client_secret="test-secret",
token_verifier=jwt_verifier,
base_url="https://myserver.com",
client_storage=memory_storage,
)
# Keyring should never be accessed on Linux
mock_keyring.get_password.assert_not_called()
mock_keyring.set_password.assert_not_called()
@patch("fastmcp.utilities.key_management.platform.system")
@patch("fastmcp.utilities.key_management.keyring")
async def test_explicit_keys_bypass_keyring(
self, mock_keyring, mock_platform, jwt_verifier, memory_storage
):
"""Test that explicit keys bypass keyring entirely."""
mock_platform.return_value = "Darwin"
# Create proxy with explicit keys
OAuthProxy(
upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
upstream_token_endpoint="https://github.com/login/oauth/access_token",
upstream_client_id="explicit-keys-client",
upstream_client_secret="test-secret",
token_verifier=jwt_verifier,
base_url="https://myserver.com",
jwt_signing_key="my-custom-jwt-key",
token_encryption_key="my-custom-encryption-key",
client_storage=memory_storage,
)
# Keyring should never be accessed when explicit keys provided
mock_keyring.get_password.assert_not_called()
mock_keyring.set_password.assert_not_called()
@patch("fastmcp.utilities.key_management.platform.system")
@patch("fastmcp.utilities.key_management.keyring")
async def test_keyring_namespace_isolation(
self, mock_keyring, mock_platform, jwt_verifier, memory_storage
):
"""Test that different upstream client IDs create isolated keyring entries."""
mock_platform.return_value = "Darwin"
mock_keyring.get_password.return_value = None
# Create first proxy with client-A
OAuthProxy(
upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
upstream_token_endpoint="https://github.com/login/oauth/access_token",
upstream_client_id="client-A",
upstream_client_secret="secret-A",
token_verifier=jwt_verifier,
base_url="https://myserver.com",
client_storage=memory_storage,
)
# Reset mock to track second proxy separately
mock_keyring.reset_mock()
mock_keyring.get_password.return_value = None
# Create second proxy with client-B
OAuthProxy(
upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
upstream_token_endpoint="https://github.com/login/oauth/access_token",
upstream_client_id="client-B",
upstream_client_secret="secret-B",
token_verifier=jwt_verifier,
base_url="https://myserver.com",
client_storage=MemoryStore(), # Different storage instance
)
# Verify that client-B keys were stored with different namespace
set_calls = mock_keyring.set_password.call_args_list
for call in set_calls:
assert call[0][0] == "fastmcp"
assert "client-B" in call[0][1] # Namespace includes client-B
assert "client-A" not in call[0][1] # Not client-A
@patch("fastmcp.utilities.key_management.platform.system")
@patch("fastmcp.utilities.key_management.keyring")
async def test_keyring_retrieves_existing_keys(
self, mock_keyring, mock_platform, jwt_verifier, memory_storage
):
"""Test that existing keyring keys are retrieved and reused."""
mock_platform.return_value = "Darwin"
# Mock existing keys in keyring
def get_password_side_effect(service, key):
if "jwt-signing" in key:
return "existing-jwt-key-base64"
elif "token-encryption" in key:
return "existing-encryption-key-base64"
return None
mock_keyring.get_password.side_effect = get_password_side_effect
# Create proxy - should retrieve existing keys
proxy = OAuthProxy(
upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
upstream_token_endpoint="https://github.com/login/oauth/access_token",
upstream_client_id="existing-keys-client",
upstream_client_secret="test-secret",
token_verifier=jwt_verifier,
base_url="https://myserver.com",
client_storage=memory_storage,
)
# Trigger JWT initialization
await proxy._ensure_jwt_initialized()
# Should retrieve but not set new keys
assert mock_keyring.get_password.call_count == 2
mock_keyring.set_password.assert_not_called()
@patch("fastmcp.utilities.key_management.platform.system")
@patch("fastmcp.utilities.key_management.keyring")
async def test_keyring_failure_uses_ephemeral_keys(
self, mock_keyring, mock_platform, jwt_verifier, memory_storage
):
"""Test graceful fallback to ephemeral keys when keyring fails."""
mock_platform.return_value = "Darwin"
# Simulate keyring failure
mock_keyring.get_password.side_effect = Exception("Keyring backend unavailable")
# Should not raise - should fall back to ephemeral keys
proxy = OAuthProxy(
upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
upstream_token_endpoint="https://github.com/login/oauth/access_token",
upstream_client_id="fallback-client",
upstream_client_secret="test-secret",
token_verifier=jwt_verifier,
base_url="https://myserver.com",
client_storage=memory_storage,
)
# Proxy should be created successfully despite keyring failure
assert proxy is not None

View file

@ -458,6 +458,7 @@ class TestOIDCProxyInitialization:
client_id=TEST_CLIENT_ID,
client_secret=TEST_CLIENT_SECRET,
base_url=TEST_BASE_URL,
jwt_signing_key="test-secret",
)
validate_proxy(mock_get, proxy, oidc_config)
@ -478,6 +479,7 @@ class TestOIDCProxyInitialization:
client_secret=TEST_CLIENT_SECRET,
base_url=TEST_BASE_URL,
timeout_seconds=12,
jwt_signing_key="test-secret",
)
validate_proxy(mock_get, proxy, oidc_config)
@ -503,6 +505,7 @@ class TestOIDCProxyInitialization:
algorithm="RS256",
audience="oidc-proxy-test-audience",
required_scopes=["required", "scopes"],
jwt_signing_key="test-secret",
)
validate_proxy(mock_get, proxy, oidc_config)
@ -529,6 +532,7 @@ class TestOIDCProxyInitialization:
client_secret=TEST_CLIENT_SECRET,
base_url=TEST_BASE_URL,
audience="oidc-proxy-test-audience",
jwt_signing_key="test-secret",
)
validate_proxy(mock_get, proxy, oidc_config)
@ -556,6 +560,7 @@ class TestOIDCProxyInitialization:
redirect_path="/oidc/proxy",
allowed_client_redirect_uris=["http://localhost:*"],
token_endpoint_auth_method="client_secret_post",
jwt_signing_key="test-secret",
)
validate_proxy(mock_get, proxy, oidc_config)
@ -582,6 +587,7 @@ class TestOIDCProxyInitialization:
client_id=TEST_CLIENT_ID,
client_secret=TEST_CLIENT_SECRET,
base_url=TEST_BASE_URL,
jwt_signing_key="test-secret",
)
def test_no_client_id_initialization_raises_error(

View file

@ -1608,6 +1608,7 @@ class TestSettingsFromEnvironment:
os.environ["FASTMCP_SERVER_AUTH_AZURE_REDIRECT_PATH"] = "/auth/callback"
os.environ["FASTMCP_SERVER_AUTH_AZURE_BASE_URL"] = "http://localhost:8000"
os.environ["FASTMCP_SERVER_AUTH_AZURE_REQUIRED_SCOPES"] = "User.Read,email,profile"
os.environ["FASTMCP_SERVER_AUTH_AZURE_JWT_SIGNING_KEY"] = "test-secret"
import fastmcp

14
uv.lock generated
View file

@ -1,5 +1,5 @@
version = 1
revision = 3
revision = 2
requires-python = ">=3.10"
resolution-markers = [
"python_full_version >= '3.11'",
@ -1405,15 +1405,15 @@ wheels = [
[[package]]
name = "py-key-value-aio"
version = "0.2.6"
version = "0.2.8"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "beartype" },
{ name = "py-key-value-shared" },
]
sdist = { url = "https://files.pythonhosted.org/packages/72/a6/2ff5e11aa38f5c5433b30e65e4aa8e0b82f82041a735ae7e5036b160d260/py_key_value_aio-0.2.6.tar.gz", hash = "sha256:bb6cc2249c7d5f334365829487009d1ee97b1a8c16d201ab7dc94ad872fd52a7", size = 29975, upload-time = "2025-10-21T16:32:41.403Z" }
sdist = { url = "https://files.pythonhosted.org/packages/ca/35/65310a4818acec0f87a46e5565e341c5a96fc062a9a03495ad28828ff4d7/py_key_value_aio-0.2.8.tar.gz", hash = "sha256:c0cfbb0bd4e962a3fa1a9fa6db9ba9df812899bd9312fa6368aaea7b26008b36", size = 32853, upload-time = "2025-10-24T13:31:04.688Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/c9/0a/b5902e788a015d7a2fb55de59d4ef7b390d00929934c3c7224ec546a4199/py_key_value_aio-0.2.6-py3-none-any.whl", hash = "sha256:676e5be6f2818e6c51ad54f07b75aaded1aa5f6c3b2b94cfa623c19cbaa77d9e", size = 63359, upload-time = "2025-10-21T16:32:40.202Z" },
{ url = "https://files.pythonhosted.org/packages/cd/5a/e56747d87a97ad2aff0f3700d77f186f0704c90c2da03bfed9e113dae284/py_key_value_aio-0.2.8-py3-none-any.whl", hash = "sha256:561565547ce8162128fd2bd0b9d70ce04a5f4586da8500cce79a54dfac78c46a", size = 69200, upload-time = "2025-10-24T13:31:03.81Z" },
]
[package.optional-dependencies]
@ -1430,15 +1430,15 @@ memory = [
[[package]]
name = "py-key-value-shared"
version = "0.2.6"
version = "0.2.8"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "beartype" },
{ name = "typing-extensions" },
]
sdist = { url = "https://files.pythonhosted.org/packages/c6/05/0dca4cb7f1674133f13c37a9976d5834887ca922449bd56f58c1578e5bbe/py_key_value_shared-0.2.6.tar.gz", hash = "sha256:6e807af74a289bc5ee0372b1695dc2f3953ade35bdacf8dca59a7bbf80aac52e", size = 8080, upload-time = "2025-10-21T16:31:35.527Z" }
sdist = { url = "https://files.pythonhosted.org/packages/26/79/05a1f9280cfa0709479319cbfd2b1c5beb23d5034624f548c83fb65b0b61/py_key_value_shared-0.2.8.tar.gz", hash = "sha256:703b4d3c61af124f0d528ba85995c3c8d78f8bd3d2b217377bd3278598070cc1", size = 8216, upload-time = "2025-10-24T13:31:03.601Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/11/bc/6149a9d9180d28cf99e81d71cacdf8988effb84c6f8da8abf66a94a10748/py_key_value_shared-0.2.6-py3-none-any.whl", hash = "sha256:f115316f2612733b47f8da70c2b463af8d3c7d031619f452cac0dc95cda42f89", size = 14179, upload-time = "2025-10-21T16:31:34.405Z" },
{ url = "https://files.pythonhosted.org/packages/84/7a/1726ceaa3343874f322dd83c9ec376ad81f533df8422b8b1e1233a59f8ce/py_key_value_shared-0.2.8-py3-none-any.whl", hash = "sha256:aff1bbfd46d065b2d67897d298642e80e5349eae588c6d11b48452b46b8d46ba", size = 14586, upload-time = "2025-10-24T13:31:02.838Z" },
]
[[package]]