mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-22 21:44:18 +02:00
Add consent_csp_policy parameter for CSP customization (#2484)
* Add consent_csp_policy parameter to OAuthProxy Allows customization or disabling of CSP directives on the consent page. Fixes #2476. * Add consent_csp_policy to OIDCProxy and update docs * Fix HTML injection vulnerability in CSP policy HTML-escape the CSP policy value before inserting into meta tag to prevent HTML injection when CSP policies contain quotes.
This commit is contained in:
parent
d770a76c79
commit
ba69fba305
6 changed files with 262 additions and 17 deletions
|
|
@ -299,6 +299,24 @@ auth = OAuthProxy(..., client_storage=MemoryStore())
|
|||
Disabling consent removes an important security layer. Only disable for local development or testing environments where you fully control all connecting clients.
|
||||
</Warning>
|
||||
</ParamField>
|
||||
|
||||
<ParamField body="consent_csp_policy" type="str | None" default="None">
|
||||
Content Security Policy for the consent page.
|
||||
|
||||
- `None` (default): Uses the built-in CSP policy with appropriate directives for form submission
|
||||
- Empty string `""`: Disables CSP entirely (no meta tag rendered)
|
||||
- Custom string: Uses the provided value as the CSP policy
|
||||
|
||||
This is useful for organizations that have their own CSP policies and need to override or disable FastMCP's built-in CSP directives.
|
||||
|
||||
```python
|
||||
# Disable CSP entirely (let org CSP policies apply)
|
||||
auth = OAuthProxy(..., consent_csp_policy="")
|
||||
|
||||
# Use custom CSP policy
|
||||
auth = OAuthProxy(..., consent_csp_policy="default-src 'self'; style-src 'unsafe-inline'")
|
||||
```
|
||||
</ParamField>
|
||||
</Card>
|
||||
|
||||
### Using Built-in Providers
|
||||
|
|
|
|||
|
|
@ -191,6 +191,20 @@ auth = OIDCProxy(
|
|||
```
|
||||
|
||||
</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. See [OAuthProxy documentation](/servers/auth/oauth-proxy#confused-deputy-attacks) for details on confused deputy attack protection.
|
||||
</ParamField>
|
||||
|
||||
<ParamField body="consent_csp_policy" type="str | None" default="None">
|
||||
Content Security Policy for the consent page.
|
||||
|
||||
- `None` (default): Uses the built-in CSP policy with appropriate directives for form submission
|
||||
- Empty string `""`: Disables CSP entirely (no meta tag rendered)
|
||||
- Custom string: Uses the provided value as the CSP policy
|
||||
|
||||
This is useful for organizations that have their own CSP policies and need to override or disable FastMCP's built-in CSP directives.
|
||||
</ParamField>
|
||||
</Card>
|
||||
|
||||
### Using Built-in Providers
|
||||
|
|
|
|||
|
|
@ -246,8 +246,16 @@ def create_consent_html(
|
|||
server_icon_url: str | None = None,
|
||||
server_website_url: str | None = None,
|
||||
client_website_url: str | None = None,
|
||||
csp_policy: str | None = None,
|
||||
) -> str:
|
||||
"""Create a styled HTML consent page for OAuth authorization requests."""
|
||||
"""Create a styled HTML consent page for OAuth authorization requests.
|
||||
|
||||
Args:
|
||||
csp_policy: Content Security Policy override.
|
||||
If None, uses the built-in CSP policy with appropriate directives.
|
||||
If empty string "", disables CSP entirely (no meta tag is rendered).
|
||||
If a non-empty string, uses that as the CSP policy value.
|
||||
"""
|
||||
import html as html_module
|
||||
|
||||
client_display = html_module.escape(client_name or client_id)
|
||||
|
|
@ -368,20 +376,25 @@ def create_consent_html(
|
|||
+ TOOLTIP_STYLES
|
||||
)
|
||||
|
||||
# Need to allow form-action for form submission
|
||||
# Chrome requires explicit scheme declarations in CSP form-action when redirect chains
|
||||
# end in custom protocol schemes (e.g., cursor://). Parse redirect_uri to include its scheme.
|
||||
parsed_redirect = urlparse(redirect_uri)
|
||||
redirect_scheme = parsed_redirect.scheme.lower()
|
||||
# Determine CSP policy to use
|
||||
# If csp_policy is None, build the default CSP policy
|
||||
# If csp_policy is empty string, CSP will be disabled entirely in create_page
|
||||
# If csp_policy is a non-empty string, use it as-is
|
||||
if csp_policy is None:
|
||||
# Need to allow form-action for form submission
|
||||
# Chrome requires explicit scheme declarations in CSP form-action when redirect chains
|
||||
# end in custom protocol schemes (e.g., cursor://). Parse redirect_uri to include its scheme.
|
||||
parsed_redirect = urlparse(redirect_uri)
|
||||
redirect_scheme = parsed_redirect.scheme.lower()
|
||||
|
||||
# Build form-action directive with standard schemes plus custom protocol if present
|
||||
form_action_schemes = ["https:", "http:"]
|
||||
if redirect_scheme and redirect_scheme not in ("http", "https"):
|
||||
# Custom protocol scheme (e.g., cursor:, vscode:, etc.)
|
||||
form_action_schemes.append(f"{redirect_scheme}:")
|
||||
# Build form-action directive with standard schemes plus custom protocol if present
|
||||
form_action_schemes = ["https:", "http:"]
|
||||
if redirect_scheme and redirect_scheme not in ("http", "https"):
|
||||
# Custom protocol scheme (e.g., cursor:, vscode:, etc.)
|
||||
form_action_schemes.append(f"{redirect_scheme}:")
|
||||
|
||||
form_action_directive = " ".join(form_action_schemes)
|
||||
csp_policy = f"default-src 'none'; style-src 'unsafe-inline'; img-src https: data:; base-uri 'none'; form-action {form_action_directive}"
|
||||
form_action_directive = " ".join(form_action_schemes)
|
||||
csp_policy = f"default-src 'none'; style-src 'unsafe-inline'; img-src https: data:; base-uri 'none'; form-action {form_action_directive}"
|
||||
|
||||
return create_page(
|
||||
content=content,
|
||||
|
|
@ -672,6 +685,7 @@ class OAuthProxy(OAuthProvider):
|
|||
jwt_signing_key: str | bytes | None = None,
|
||||
# Consent screen configuration
|
||||
require_authorization_consent: bool = True,
|
||||
consent_csp_policy: str | None = None,
|
||||
):
|
||||
"""Initialize the OAuth proxy provider.
|
||||
|
||||
|
|
@ -715,6 +729,12 @@ class OAuthProxy(OAuthProvider):
|
|||
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.
|
||||
consent_csp_policy: Content Security Policy for the consent page.
|
||||
If None (default), uses the built-in CSP policy with appropriate directives.
|
||||
If empty string "", disables CSP entirely (no meta tag is rendered).
|
||||
If a non-empty string, uses that as the CSP policy value.
|
||||
This allows organizations with their own CSP policies to override or disable
|
||||
the built-in CSP directives.
|
||||
"""
|
||||
|
||||
# Always enable DCR since we implement it locally for MCP clients
|
||||
|
|
@ -775,6 +795,7 @@ class OAuthProxy(OAuthProvider):
|
|||
|
||||
# Consent screen configuration
|
||||
self._require_authorization_consent: bool = require_authorization_consent
|
||||
self._consent_csp_policy: str | None = consent_csp_policy
|
||||
if not require_authorization_consent:
|
||||
logger.warning(
|
||||
"Authorization consent screen disabled - only use for local development or testing. "
|
||||
|
|
@ -2106,6 +2127,7 @@ class OAuthProxy(OAuthProvider):
|
|||
server_name=server_name,
|
||||
server_icon_url=server_icon_url,
|
||||
server_website_url=server_website_url,
|
||||
csp_policy=self._consent_csp_policy,
|
||||
)
|
||||
response = create_secure_html_response(html)
|
||||
# Store CSRF in cookie with short lifetime
|
||||
|
|
|
|||
|
|
@ -222,6 +222,7 @@ class OIDCProxy(OAuthProxy):
|
|||
token_endpoint_auth_method: str | None = None,
|
||||
# Consent screen configuration
|
||||
require_authorization_consent: bool = True,
|
||||
consent_csp_policy: str | None = None,
|
||||
# Extra parameters
|
||||
extra_authorize_params: dict[str, str] | None = None,
|
||||
extra_token_params: dict[str, str] | None = None,
|
||||
|
|
@ -262,6 +263,10 @@ class OIDCProxy(OAuthProxy):
|
|||
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.
|
||||
consent_csp_policy: Content Security Policy for the consent page.
|
||||
If None (default), uses the built-in CSP policy with appropriate directives.
|
||||
If empty string "", disables CSP entirely (no meta tag is rendered).
|
||||
If a non-empty string, uses that as the CSP policy value.
|
||||
extra_authorize_params: Additional parameters to forward to the upstream authorization endpoint.
|
||||
Useful for provider-specific parameters like prompt=consent or access_type=offline.
|
||||
Example: {"prompt": "consent", "access_type": "offline"}
|
||||
|
|
@ -338,6 +343,7 @@ class OIDCProxy(OAuthProxy):
|
|||
"jwt_signing_key": jwt_signing_key,
|
||||
"token_endpoint_auth_method": token_endpoint_auth_method,
|
||||
"require_authorization_consent": require_authorization_consent,
|
||||
"consent_csp_policy": consent_csp_policy,
|
||||
}
|
||||
|
||||
if redirect_path:
|
||||
|
|
|
|||
|
|
@ -463,12 +463,21 @@ def create_page(
|
|||
content: HTML content to place inside the page
|
||||
title: Page title
|
||||
additional_styles: Extra CSS to include
|
||||
csp_policy: Content Security Policy header value
|
||||
csp_policy: Content Security Policy header value.
|
||||
If empty string "", the CSP meta tag is omitted entirely.
|
||||
|
||||
Returns:
|
||||
Complete HTML page as string
|
||||
"""
|
||||
title = html.escape(title)
|
||||
|
||||
# Only include CSP meta tag if policy is non-empty
|
||||
csp_meta = (
|
||||
f'<meta http-equiv="Content-Security-Policy" content="{html.escape(csp_policy, quote=True)}" />'
|
||||
if csp_policy
|
||||
else ""
|
||||
)
|
||||
|
||||
return f"""
|
||||
<!DOCTYPE html>
|
||||
<html lang="en">
|
||||
|
|
@ -480,7 +489,7 @@ def create_page(
|
|||
{BASE_STYLES}
|
||||
{additional_styles}
|
||||
</style>
|
||||
<meta http-equiv="Content-Security-Policy" content="{csp_policy}" />
|
||||
{csp_meta}
|
||||
</head>
|
||||
<body>
|
||||
{content}
|
||||
|
|
|
|||
|
|
@ -884,3 +884,179 @@ class TestConsentPageServerIcon:
|
|||
'alt="<script>alert("xss")</script>Server"'
|
||||
in response.text
|
||||
)
|
||||
|
||||
|
||||
class TestConsentCSPPolicy:
|
||||
"""Tests for Content Security Policy customization on consent page."""
|
||||
|
||||
async def test_default_csp_includes_form_action(self):
|
||||
"""Test that default CSP includes form-action directive."""
|
||||
from unittest.mock import Mock
|
||||
|
||||
from fastmcp import FastMCP
|
||||
|
||||
verifier = Mock(spec=TokenVerifier)
|
||||
verifier.required_scopes = ["read"]
|
||||
verifier.verify_token = Mock(return_value=None)
|
||||
|
||||
# Create OAuthProxy with default CSP (no custom CSP)
|
||||
proxy = OAuthProxy(
|
||||
upstream_authorization_endpoint="https://oauth.example.com/authorize",
|
||||
upstream_token_endpoint="https://oauth.example.com/token",
|
||||
upstream_client_id="upstream-client",
|
||||
upstream_client_secret="upstream-secret",
|
||||
token_verifier=verifier,
|
||||
base_url="https://proxy.example.com",
|
||||
client_storage=MemoryStore(),
|
||||
jwt_signing_key="test-secret",
|
||||
)
|
||||
|
||||
server = FastMCP(name="Test Server", auth=proxy)
|
||||
app = server.http_app()
|
||||
|
||||
client_info = OAuthClientInformationFull(
|
||||
client_id="test-client",
|
||||
client_secret="test-secret",
|
||||
redirect_uris=[AnyUrl("http://localhost:12345/callback")],
|
||||
)
|
||||
await proxy.register_client(client_info)
|
||||
|
||||
from fastmcp.server.auth.oauth_proxy import OAuthTransaction
|
||||
|
||||
txn_id = "test-txn-id"
|
||||
transaction = OAuthTransaction(
|
||||
txn_id=txn_id,
|
||||
client_id="test-client",
|
||||
client_redirect_uri="http://localhost:12345/callback",
|
||||
client_state="client-state",
|
||||
code_challenge="challenge",
|
||||
code_challenge_method="S256",
|
||||
scopes=["read"],
|
||||
created_at=time.time(),
|
||||
)
|
||||
await proxy._transaction_store.put(key=txn_id, value=transaction)
|
||||
|
||||
with TestClient(app) as client:
|
||||
response = client.get(f"/consent?txn_id={txn_id}")
|
||||
|
||||
assert response.status_code == 200
|
||||
# Default CSP should be present with form-action
|
||||
assert 'http-equiv="Content-Security-Policy"' in response.text
|
||||
assert "form-action" in response.text
|
||||
|
||||
async def test_empty_csp_disables_csp_meta_tag(self):
|
||||
"""Test that empty string CSP disables CSP meta tag entirely."""
|
||||
from unittest.mock import Mock
|
||||
|
||||
from fastmcp import FastMCP
|
||||
|
||||
verifier = Mock(spec=TokenVerifier)
|
||||
verifier.required_scopes = ["read"]
|
||||
verifier.verify_token = Mock(return_value=None)
|
||||
|
||||
# Create OAuthProxy with empty CSP to disable it
|
||||
proxy = OAuthProxy(
|
||||
upstream_authorization_endpoint="https://oauth.example.com/authorize",
|
||||
upstream_token_endpoint="https://oauth.example.com/token",
|
||||
upstream_client_id="upstream-client",
|
||||
upstream_client_secret="upstream-secret",
|
||||
token_verifier=verifier,
|
||||
base_url="https://proxy.example.com",
|
||||
client_storage=MemoryStore(),
|
||||
jwt_signing_key="test-secret",
|
||||
consent_csp_policy="", # Empty string disables CSP
|
||||
)
|
||||
|
||||
server = FastMCP(name="Test Server", auth=proxy)
|
||||
app = server.http_app()
|
||||
|
||||
client_info = OAuthClientInformationFull(
|
||||
client_id="test-client",
|
||||
client_secret="test-secret",
|
||||
redirect_uris=[AnyUrl("http://localhost:12345/callback")],
|
||||
)
|
||||
await proxy.register_client(client_info)
|
||||
|
||||
from fastmcp.server.auth.oauth_proxy import OAuthTransaction
|
||||
|
||||
txn_id = "test-txn-id"
|
||||
transaction = OAuthTransaction(
|
||||
txn_id=txn_id,
|
||||
client_id="test-client",
|
||||
client_redirect_uri="http://localhost:12345/callback",
|
||||
client_state="client-state",
|
||||
code_challenge="challenge",
|
||||
code_challenge_method="S256",
|
||||
scopes=["read"],
|
||||
created_at=time.time(),
|
||||
)
|
||||
await proxy._transaction_store.put(key=txn_id, value=transaction)
|
||||
|
||||
with TestClient(app) as client:
|
||||
response = client.get(f"/consent?txn_id={txn_id}")
|
||||
|
||||
assert response.status_code == 200
|
||||
# CSP meta tag should NOT be present
|
||||
assert 'http-equiv="Content-Security-Policy"' not in response.text
|
||||
|
||||
async def test_custom_csp_policy_is_used(self):
|
||||
"""Test that custom CSP policy is applied to consent page."""
|
||||
from unittest.mock import Mock
|
||||
|
||||
from fastmcp import FastMCP
|
||||
|
||||
verifier = Mock(spec=TokenVerifier)
|
||||
verifier.required_scopes = ["read"]
|
||||
verifier.verify_token = Mock(return_value=None)
|
||||
|
||||
# Create OAuthProxy with custom CSP policy
|
||||
custom_csp = "default-src 'self'; script-src 'none'"
|
||||
proxy = OAuthProxy(
|
||||
upstream_authorization_endpoint="https://oauth.example.com/authorize",
|
||||
upstream_token_endpoint="https://oauth.example.com/token",
|
||||
upstream_client_id="upstream-client",
|
||||
upstream_client_secret="upstream-secret",
|
||||
token_verifier=verifier,
|
||||
base_url="https://proxy.example.com",
|
||||
client_storage=MemoryStore(),
|
||||
jwt_signing_key="test-secret",
|
||||
consent_csp_policy=custom_csp,
|
||||
)
|
||||
|
||||
server = FastMCP(name="Test Server", auth=proxy)
|
||||
app = server.http_app()
|
||||
|
||||
client_info = OAuthClientInformationFull(
|
||||
client_id="test-client",
|
||||
client_secret="test-secret",
|
||||
redirect_uris=[AnyUrl("http://localhost:12345/callback")],
|
||||
)
|
||||
await proxy.register_client(client_info)
|
||||
|
||||
from fastmcp.server.auth.oauth_proxy import OAuthTransaction
|
||||
|
||||
txn_id = "test-txn-id"
|
||||
transaction = OAuthTransaction(
|
||||
txn_id=txn_id,
|
||||
client_id="test-client",
|
||||
client_redirect_uri="http://localhost:12345/callback",
|
||||
client_state="client-state",
|
||||
code_challenge="challenge",
|
||||
code_challenge_method="S256",
|
||||
scopes=["read"],
|
||||
created_at=time.time(),
|
||||
)
|
||||
await proxy._transaction_store.put(key=txn_id, value=transaction)
|
||||
|
||||
with TestClient(app) as client:
|
||||
response = client.get(f"/consent?txn_id={txn_id}")
|
||||
|
||||
assert response.status_code == 200
|
||||
# Custom CSP should be present (HTML-escaped)
|
||||
assert 'http-equiv="Content-Security-Policy"' in response.text
|
||||
# Check for the HTML-escaped version (single quotes become ')
|
||||
import html
|
||||
|
||||
assert html.escape(custom_csp, quote=True) in response.text
|
||||
# Default form-action should NOT be present (we're using custom)
|
||||
assert "form-action" not in response.text
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue