Emit scope step-up challenges for incremental authorization (SEP-2350) (#4623)

* Signal component-level scope shortfalls as insufficient_scope (SEP-2350)

* Fix ty type narrowing in scope step-up test

* Respect check short-circuit when reporting scope shortfall (P2)

* Report union of unmet scopes and document step-up contract

* Aggregate scope shortfall across the AuthMiddleware chain

* Stop chain scope aggregation at the first unevaluated gate
This commit is contained in:
Jeremiah Lowin 2026-07-26 14:13:26 -04:00 committed by GitHub
commit 4ebb3fd5e6
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 645 additions and 34 deletions

View file

@ -168,6 +168,7 @@ Sync and async checks can be freely combined in a list — each check is handled
Auth checks can raise exceptions for explicit denial with custom messages:
- **`AuthorizationError`**: Propagates with its custom message, useful for explaining why access was denied
- **`InsufficientScopeError`**: A subclass of `AuthorizationError` raised by `AuthMiddleware` when the denial is a missing scope; it [names the scopes the caller needs](#signaling-scope-shortfalls)
- **Other exceptions**: Masked for security (logged internally, treated as denial)
```python
@ -215,7 +216,7 @@ Component-level `auth` controls both visibility (list filtering) and access (dir
## Server-Level Authorization
For server-wide authorization enforcement, use `AuthMiddleware`. This middleware applies auth checks globally to all components—filtering list responses and blocking unauthorized execution with explicit `AuthorizationError` responses.
For server-wide authorization enforcement, use `AuthMiddleware`. This middleware applies auth checks globally to all components—filtering list responses and blocking unauthorized execution with explicit `AuthorizationError` responses. When the denial is specifically a missing scope, the error [names the scopes the caller needs](#signaling-scope-shortfalls).
```python
from fastmcp import FastMCP
@ -296,6 +297,46 @@ def read_record(id: str) -> str:
return f"Record {id}"
```
### Signaling Scope Shortfalls
A denial is more useful when it says what would fix it. When `AuthMiddleware` blocks a call because the token is missing scopes — rather than because some other policy rejected it — it raises `InsufficientScopeError`, which carries the specific scopes the caller needs in its `required_scopes` attribute. An agent that reads the error knows exactly which scopes to re-authorize for, instead of retrying blindly against an opaque refusal.
`InsufficientScopeError` subclasses `AuthorizationError`, so existing handlers that catch `AuthorizationError` keep catching it and nothing about your error handling has to change to adopt this.
Only the scopes the token *lacks* are named, so re-authorizing accumulates permissions rather than replacing them. A caller holding `read` that needs `read` and `write` is told to obtain `write` alone, and keeps `read` through the re-authorization. When several scope requirements fail at once, every unmet scope is reported together — a caller granted them all in one round succeeds on the retry, instead of discovering the next missing scope only after obtaining the first.
```python
from fastmcp import FastMCP
from fastmcp.exceptions import InsufficientScopeError
from fastmcp.server.auth import require_scopes
from fastmcp.server.middleware import AuthMiddleware
mcp = FastMCP(
"Step-Up Server",
middleware=[AuthMiddleware(auth=require_scopes("read", "write"))],
)
@mcp.tool
def update_record(id: str) -> str:
"""Requires both 'read' and 'write'."""
return f"Updated {id}"
# A token holding only "read" is denied with:
# InsufficientScopeError(required_scopes=["write"])
```
This holds across several `AuthMiddleware` instances too, not just several checks within one. In the [tag-based configuration](#tag-based-global-authorization) each middleware contributes its own requirement, and the first to find a shortfall reports the requirements of the others alongside its own — so one re-authorization covers the whole chain rather than one layer at a time.
A shortfall is reported only when the scope requirement is what actually caused the denial. If you [combine checks](#combining-checks) and a non-scope check rejects the request first — a tenant policy, say — the denial stays a plain `AuthorizationError` and names no scopes at all. Disclosing a scope requirement for a component the caller could not reach anyway would leak information about components they are not authorized to see.
That rule also bounds what gets aggregated. Combining requirements only reaches as far down the chain as the request itself would have gone: it stops at the first layer holding a custom check, since whether that layer would admit the caller is unknown until it runs, and running it early would trigger authorization logic the request had not reached yet. Requirements at or beyond that point sit behind an unverified gate and are left out.
So a custom check early in the chain makes the reported set partial, and a caller may need more than one round to satisfy everything. The reported set is complete when the layers ahead are scope-only and conservative otherwise: it may name fewer scopes than the full chain requires, but it never names scopes behind a policy that might reject the caller regardless.
<Note>
This names the missing scopes in the error rather than emitting an HTTP `403` challenge. A per-tool denial is a JSON-RPC error carried inside a `200` response, so there is no HTTP status at that layer to attach a `WWW-Authenticate` header to. Token-level scope failures — where the token does not satisfy the server's own `required_scopes` — are a separate concern handled by the transport middleware, which does return a spec-correct `403` with an `insufficient_scope` challenge.
</Note>
## Accessing Tokens in Tools
Tools can access the current authentication token using `get_access_token()` from `fastmcp.server.dependencies`. This enables tools to make decisions based on user identity or permissions beyond simple authorization checks.
@ -380,5 +421,10 @@ from fastmcp.server.auth import (
run_auth_checks, # Utility: run checks with AND logic
)
from fastmcp.exceptions import (
AuthorizationError, # Denial with a custom message
InsufficientScopeError, # Subclass of AuthorizationError; has .required_scopes
)
from fastmcp.server.middleware import AuthMiddleware
```

View file

@ -95,6 +95,30 @@ class AuthorizationError(FastMCPError):
"""Error when authorization check fails."""
class InsufficientScopeError(AuthorizationError):
"""Authorization failed because the token is missing required OAuth scopes.
Unlike a bare ``AuthorizationError``, this carries the specific scopes the
caller must obtain. A component-level scope shortfall can then be signalled
as a spec-correct ``insufficient_scope`` step-up (SEP-2350 / RFC 6750 §3),
naming exactly what to re-authorize for instead of an opaque denial. The
named scopes are only the *unmet* ones, so an existing grant is accumulated
rather than replaced when the caller re-authorizes.
"""
def __init__(
self,
required_scopes: list[str],
*,
message: str | None = None,
) -> None:
self.required_scopes = list(required_scopes)
if message is None:
named = ", ".join(self.required_scopes) or "(unknown)"
message = f"Insufficient scope. Required: {named}"
super().__init__(message)
def to_mcp_error(exc: Exception, *, default_code: int = INTERNAL_ERROR) -> MCPError:
"""Translate a FastMCP exception into a wire-format ``MCPError``.

View file

@ -6,6 +6,8 @@ from fastmcp.utilities.authorization import (
require_scopes,
restrict_tag,
run_auth_checks,
run_auth_checks_with_shortfall,
scope_requirements,
)
__all__ = [
@ -14,4 +16,6 @@ __all__ = [
"require_scopes",
"restrict_tag",
"run_auth_checks",
"run_auth_checks_with_shortfall",
"scope_requirements",
]

View file

@ -29,7 +29,7 @@ from typing import Any
import mcp_types as mt
from fastmcp.exceptions import AuthorizationError
from fastmcp.exceptions import AuthorizationError, InsufficientScopeError
from fastmcp.prompts.base import Prompt, PromptResult
from fastmcp.resources.base import Resource, ResourceResult
from fastmcp.resources.template import ResourceTemplate
@ -37,6 +37,8 @@ from fastmcp.server.auth.authorization import (
AuthCheck,
AuthContext,
run_auth_checks,
run_auth_checks_with_shortfall,
scope_requirements,
)
from fastmcp.server.dependencies import get_access_token
from fastmcp.server.middleware.middleware import (
@ -111,6 +113,60 @@ class AuthMiddleware(Middleware):
def __init__(self, auth: AuthCheck | list[AuthCheck]) -> None:
self.auth = auth
def _chain_shortfall(
self,
own_missing: list[str],
ctx: AuthContext,
server: Any,
) -> list[str]:
"""Widen this middleware's shortfall to cover the rest of the chain.
Scope requirements are commonly split across several `AuthMiddleware`
instances (the tag-based configuration does exactly this). Raising as
soon as the first one finds a shortfall means the middleware further in
never runs, so a caller told only the outer scope would obtain it, retry,
and be denied for the next the same non-convergent loop that reporting
a union within one middleware avoids.
Sibling requirements are read without evaluating them: `scope_requirements`
is a pure comparison against the token and component. Only the layers the
request would actually reach next may contribute, so the walk starts after
this middleware and stops at the first one whose requirements cannot be
read a layer holding an opaque check is an unverified gate, and anything
at or beyond it might be unreachable for reasons that have nothing to do
with scopes. Disclosing those requirements would leak what sits behind a
policy the request never passed.
Layers *before* this one are already known to have passed, so they neither
block the walk nor contribute anything. This middleware's own shortfall
always counts: the request demonstrably reached it.
The result is therefore complete when the reachable chain is scope-only,
and deliberately partial otherwise never a disclosure past an
unevaluated gate.
"""
middleware = getattr(server, "middleware", None)
if not isinstance(middleware, list):
return own_missing
position = next(
(i for i, mw in enumerate(middleware) if mw is self),
None,
)
if position is None:
return own_missing
missing = set(own_missing)
for mw in middleware[position + 1 :]:
if not isinstance(mw, AuthMiddleware):
continue
sibling = scope_requirements(mw.auth, ctx)
if sibling is None:
# An unverified gate. Nothing at or beyond it may contribute.
break
missing |= set(sibling)
return sorted(missing)
async def on_list_tools(
self,
context: MiddlewareContext[mt.ListToolsRequest],
@ -180,7 +236,17 @@ class AuthMiddleware(Middleware):
# Global auth check
token = get_access_token()
ctx = AuthContext(token=token, component=tool)
if not await run_auth_checks(self.auth, ctx):
authorized, missing = await run_auth_checks_with_shortfall(self.auth, ctx)
if not authorized:
if missing:
missing = self._chain_shortfall(missing, ctx, fastmcp.fastmcp)
raise InsufficientScopeError(
missing,
message=(
f"Authorization failed for tool '{tool_name}': "
f"insufficient scope (required: {', '.join(missing)})"
),
)
raise AuthorizationError(
f"Authorization failed for tool '{tool_name}': insufficient permissions"
)
@ -258,7 +324,17 @@ class AuthMiddleware(Middleware):
# Global auth check
token = get_access_token()
ctx = AuthContext(token=token, component=component)
if not await run_auth_checks(self.auth, ctx):
authorized, missing = await run_auth_checks_with_shortfall(self.auth, ctx)
if not authorized:
if missing:
missing = self._chain_shortfall(missing, ctx, fastmcp.fastmcp)
raise InsufficientScopeError(
missing,
message=(
f"Authorization failed for resource '{uri}': "
f"insufficient scope (required: {', '.join(missing)})"
),
)
raise AuthorizationError(
f"Authorization failed for resource '{uri}': insufficient permissions"
)
@ -360,7 +436,17 @@ class AuthMiddleware(Middleware):
# Global auth check
token = get_access_token()
ctx = AuthContext(token=token, component=prompt)
if not await run_auth_checks(self.auth, ctx):
authorized, missing = await run_auth_checks_with_shortfall(self.auth, ctx)
if not authorized:
if missing:
missing = self._chain_shortfall(missing, ctx, fastmcp.fastmcp)
raise InsufficientScopeError(
missing,
message=(
f"Authorization failed for prompt '{prompt_name}': "
f"insufficient scope (required: {', '.join(missing)})"
),
)
raise AuthorizationError(
f"Authorization failed for prompt '{prompt_name}': insufficient permissions"
)

View file

@ -47,55 +47,185 @@ class AuthContext:
AuthCheck = Callable[[AuthContext], bool] | Callable[[AuthContext], Awaitable[bool]]
def require_scopes(*scopes: str) -> AuthCheck:
"""Require all of the given OAuth scopes."""
required = set(scopes)
class _ScopeAwareCheck:
"""Base for auth checks that can name the scopes a token is missing.
def check(ctx: AuthContext) -> bool:
Ordinary auth checks are opaque booleans: on denial they reveal nothing
about *why*. Scope-based checks expose their unmet requirements through
`missing_scopes` so a shortfall can be surfaced as a spec-correct
``insufficient_scope`` step-up (SEP-2350 / RFC 6750 §3) naming exactly what
the caller must re-authorize for.
"""
def missing_scopes(self, ctx: AuthContext) -> set[str]:
"""Return the required scopes the token lacks (empty if satisfied).
An absent token yields an empty set: a missing token is an
authentication problem, not a scope shortfall, and must not be turned
into an ``insufficient_scope`` challenge (RFC 6750 §3.1).
"""
raise NotImplementedError
class _RequireScopes(_ScopeAwareCheck):
"""Callable auth check requiring all of a fixed set of OAuth scopes."""
def __init__(self, scopes: tuple[str, ...]) -> None:
self.required_scopes: frozenset[str] = frozenset(scopes)
def __call__(self, ctx: AuthContext) -> bool:
if ctx.token is None:
return False
return required.issubset(set(ctx.token.scopes))
return self.required_scopes.issubset(set(ctx.token.scopes))
return check
def missing_scopes(self, ctx: AuthContext) -> set[str]:
if ctx.token is None:
return set()
return set(self.required_scopes) - set(ctx.token.scopes)
class _RestrictTag(_ScopeAwareCheck):
"""Callable auth check requiring scopes only when a component has a tag."""
def __init__(self, tag: str, scopes: list[str]) -> None:
self.tag = tag
self.required_scopes: frozenset[str] = frozenset(scopes)
def __call__(self, ctx: AuthContext) -> bool:
if self.tag not in ctx.component.tags:
return True
if ctx.token is None:
return False
return self.required_scopes.issubset(set(ctx.token.scopes))
def missing_scopes(self, ctx: AuthContext) -> set[str]:
if self.tag not in ctx.component.tags or ctx.token is None:
return set()
return set(self.required_scopes) - set(ctx.token.scopes)
def require_scopes(*scopes: str) -> AuthCheck:
"""Require all of the given OAuth scopes."""
return _RequireScopes(scopes)
def restrict_tag(tag: str, *, scopes: list[str]) -> AuthCheck:
"""Require scopes when the accessed component has a specific tag."""
required = set(scopes)
return _RestrictTag(tag, scopes)
def check(ctx: AuthContext) -> bool:
if tag not in ctx.component.tags:
return True
if ctx.token is None:
return False
return required.issubset(set(ctx.token.scopes))
return check
def scope_requirements(
checks: AuthCheck | list[AuthCheck],
ctx: AuthContext,
) -> list[str] | None:
"""Scopes a check list requires but the token lacks, without running it.
Returns ``None`` when the list contains any opaque (non-scope) check. Such a
check might deny for a reason unrelated to scopes, and evaluating it here
would run authorization logic with whatever side effects it carries
outside its normal place in the chain. Since its verdict is unknown, its
siblings' scopes must not be disclosed either, so the whole list is withheld.
When every check is scope-aware, the result is their combined shortfall,
computed purely from the token and component (an empty list means the list is
already satisfied). This lets a shortfall be aggregated across authorization
layers without evaluating anything that would otherwise be skipped.
"""
check_list = [checks] if not isinstance(checks, list) else checks
check_list = cast(list[AuthCheck], check_list)
missing: set[str] = set()
for check in check_list:
if not isinstance(check, _ScopeAwareCheck):
return None
missing |= check.missing_scopes(ctx)
return sorted(missing)
async def _evaluate_check(check: AuthCheck, ctx: AuthContext) -> bool:
"""Evaluate a single auth check, masking unexpected failures as denial.
An ``AuthorizationError`` is the check's deliberate denial and propagates.
Any other exception is a bug in the check; it is logged and treated as a
denial so a broken check fails closed.
"""
try:
result = check(ctx)
if inspect.isawaitable(result):
result = await result
except AuthorizationError:
raise
except Exception:
logger.warning(
f"Auth check {getattr(check, '__name__', repr(check))} "
"raised an unexpected exception",
exc_info=True,
)
return False
return bool(result)
async def run_auth_checks_with_shortfall(
checks: AuthCheck | list[AuthCheck],
ctx: AuthContext,
) -> tuple[bool, list[str]]:
"""Run auth checks with AND logic, classifying the denial cause.
Returns ``(authorized, missing_scopes)``. ``missing_scopes`` names every
scope the caller must obtain to satisfy *all* scope requirements at once:
the union of the shortfalls across every scope-aware check, not just the
first one to fail. Reporting only the first would strand a caller in a
step-up loop it obtains that scope, retries, and is denied again for the
next so the union is what makes a single re-authorization converge.
The challenge is withheld entirely (an empty list, which the caller surfaces
as a plain ``AuthorizationError``) unless every non-scope check passes. A
custom policy denial a tenant check, say must never be reported as an
``insufficient_scope`` shortfall, and must never name the scopes of a
component the caller could not otherwise reach. To guarantee that, the
opaque checks are all evaluated before any scope is disclosed; a shortfall
is only reported once they have all passed.
An ``AuthorizationError`` raised by a check propagates unchanged.
"""
check_list = [checks] if not isinstance(checks, list) else checks
check_list = cast(list[AuthCheck], check_list)
scope_shortfall = False
for check in check_list:
if await _evaluate_check(check, ctx):
continue
if not isinstance(check, _ScopeAwareCheck):
# An opaque denial dominates: deny without disclosing any scope.
return False, []
# Keep going. The remaining opaque checks still have to pass before a
# scope shortfall may be disclosed, and the remaining scope checks
# contribute to the union.
scope_shortfall = True
if not scope_shortfall:
return True, []
# Every opaque check passed, so naming the shortfall is safe. `missing_scopes`
# is a pure comparison against the token and returns an empty set for checks
# that passed, so unioning across all of them yields exactly the unmet scopes.
missing: set[str] = set()
for check in check_list:
if isinstance(check, _ScopeAwareCheck):
missing |= check.missing_scopes(ctx)
return False, sorted(missing)
async def run_auth_checks(
checks: AuthCheck | list[AuthCheck],
ctx: AuthContext,
) -> bool:
"""Run auth checks with AND logic."""
"""Run auth checks with AND logic, stopping at the first failure."""
check_list = [checks] if not isinstance(checks, list) else checks
check_list = cast(list[AuthCheck], check_list)
for check in check_list:
try:
result = check(ctx)
if inspect.isawaitable(result):
result = await result
if not result:
return False
except AuthorizationError:
raise
except Exception:
logger.warning(
f"Auth check {getattr(check, '__name__', repr(check))} "
"raised an unexpected exception",
exc_info=True,
)
if not await _evaluate_check(check, ctx):
return False
return True

View file

@ -8,7 +8,7 @@ from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser
from fastmcp import FastMCP
from fastmcp.client import Client
from fastmcp.exceptions import AuthorizationError
from fastmcp.exceptions import AuthorizationError, InsufficientScopeError
from fastmcp.server.auth import (
AccessToken,
AuthContext,
@ -1008,3 +1008,324 @@ class TestComponentAuthDenialMessage:
assert "not found or not authorized" in message
finally:
auth_context_var.reset(tok)
# =============================================================================
# Tests for component-level scope step-up signalling (SEP-2350)
# =============================================================================
class TestInsufficientScopeSignal:
"""A scope shortfall on a globally-authorized component is surfaced as an
``InsufficientScopeError`` naming the unmet scopes, the component-level
analog of the transport-level ``insufficient_scope`` challenge. The named
scopes are only those the token lacks, so an existing grant accumulates
rather than being replaced when the caller re-authorizes.
"""
async def test_call_tool_missing_scope_names_required_scope(self):
mcp = FastMCP(middleware=[AuthMiddleware(auth=require_scopes("api"))])
@mcp.tool
def api_tool() -> str:
return "ok"
tok = set_token(make_token(scopes=["read"]))
try:
with pytest.raises(InsufficientScopeError) as exc_info:
await mcp.call_tool("api_tool", {})
finally:
auth_context_var.reset(tok)
assert exc_info.value.required_scopes == ["api"]
assert "api" in str(exc_info.value)
async def test_insufficient_scope_error_is_authorization_error(self):
# Existing `except AuthorizationError` sites must still catch it.
assert issubclass(InsufficientScopeError, AuthorizationError)
async def test_call_tool_sufficient_scope_passes(self):
mcp = FastMCP(middleware=[AuthMiddleware(auth=require_scopes("api"))])
@mcp.tool
def api_tool() -> str:
return "ok"
tok = set_token(make_token(scopes=["api", "read"]))
try:
result = await mcp.call_tool("api_tool", {})
finally:
auth_context_var.reset(tok)
assert result.content[0].text == "ok" # type: ignore[union-attr] # ty:ignore[unresolved-attribute]
async def test_shortfall_names_only_unmet_scopes(self):
# The token already carries "read"; only the missing "api" is named, so a
# re-authorization accumulates scopes rather than dropping "read".
mcp = FastMCP(middleware=[AuthMiddleware(auth=require_scopes("read", "api"))])
@mcp.tool
def api_tool() -> str:
return "ok"
tok = set_token(make_token(scopes=["read"]))
try:
with pytest.raises(InsufficientScopeError) as exc_info:
await mcp.call_tool("api_tool", {})
finally:
auth_context_var.reset(tok)
assert exc_info.value.required_scopes == ["api"]
async def test_missing_token_is_not_insufficient_scope(self):
# No token is an authentication failure (RFC 6750 §3.1), not a scope
# shortfall: it must not be turned into an insufficient_scope signal.
mcp = FastMCP(middleware=[AuthMiddleware(auth=require_scopes("api"))])
@mcp.tool
def api_tool() -> str:
return "ok"
with pytest.raises(AuthorizationError) as exc_info:
await mcp.call_tool("api_tool", {})
assert not isinstance(exc_info.value, InsufficientScopeError)
async def test_non_scope_denial_stays_opaque_and_names_no_scope(self):
# A non-scope check (e.g. a custom tenant policy) fails first and
# short-circuits before the scope check runs. The denial must stay a
# plain AuthorizationError and must NOT disclose or request the "admin"
# scope for a component the caller could not otherwise reach.
def deny_tenant(ctx: AuthContext) -> bool:
return False
mcp = FastMCP(
middleware=[
AuthMiddleware(auth=[deny_tenant, require_scopes("admin")]),
]
)
@mcp.tool
def api_tool() -> str:
return "ok"
tok = set_token(make_token(scopes=["read"]))
try:
with pytest.raises(AuthorizationError) as exc_info:
await mcp.call_tool("api_tool", {})
finally:
auth_context_var.reset(tok)
assert not isinstance(exc_info.value, InsufficientScopeError)
assert "admin" not in str(exc_info.value)
async def test_shortfall_unions_every_unmet_scope_check(self):
# Naming only the first failing check would strand the caller in a
# step-up loop: they obtain "read", retry, and are denied for "write".
mcp = FastMCP(
middleware=[
AuthMiddleware(auth=[require_scopes("read"), require_scopes("write")]),
]
)
@mcp.tool
def api_tool() -> str:
return "ok"
tok = set_token(make_token(scopes=[]))
try:
with pytest.raises(InsufficientScopeError) as exc_info:
await mcp.call_tool("api_tool", {})
finally:
auth_context_var.reset(tok)
assert exc_info.value.required_scopes == ["read", "write"]
async def test_shortfall_union_drops_already_granted_scope(self):
mcp = FastMCP(
middleware=[
AuthMiddleware(auth=[require_scopes("read"), require_scopes("write")]),
]
)
@mcp.tool
def api_tool() -> str:
return "ok"
tok = set_token(make_token(scopes=["read"]))
try:
with pytest.raises(InsufficientScopeError) as exc_info:
await mcp.call_tool("api_tool", {})
finally:
auth_context_var.reset(tok)
assert exc_info.value.required_scopes == ["write"]
async def test_shortfall_unions_across_middleware_chain(self):
# Scope requirements split across two AuthMiddleware instances. The
# outer one raises before the inner ever runs, so its shortfall has to
# account for the inner requirement or the caller loops.
mcp = FastMCP(
middleware=[
AuthMiddleware(auth=require_scopes("admin")),
AuthMiddleware(auth=require_scopes("write")),
]
)
@mcp.tool
def api_tool() -> str:
return "ok"
tok = set_token(make_token(scopes=[]))
try:
with pytest.raises(InsufficientScopeError) as exc_info:
await mcp.call_tool("api_tool", {})
finally:
auth_context_var.reset(tok)
assert exc_info.value.required_scopes == ["admin", "write"]
async def test_chain_shortfall_drops_already_granted_scope(self):
mcp = FastMCP(
middleware=[
AuthMiddleware(auth=require_scopes("admin")),
AuthMiddleware(auth=require_scopes("write")),
]
)
@mcp.tool
def api_tool() -> str:
return "ok"
tok = set_token(make_token(scopes=["admin"]))
try:
with pytest.raises(InsufficientScopeError) as exc_info:
await mcp.call_tool("api_tool", {})
finally:
auth_context_var.reset(tok)
assert exc_info.value.required_scopes == ["write"]
async def test_opaque_denial_in_chain_stays_opaque(self):
# An opaque check denies in the outer middleware; the inner middleware's
# scope requirement must not be disclosed.
def deny_tenant(ctx: AuthContext) -> bool:
return False
mcp = FastMCP(
middleware=[
AuthMiddleware(auth=deny_tenant),
AuthMiddleware(auth=require_scopes("write")),
]
)
@mcp.tool
def api_tool() -> str:
return "ok"
tok = set_token(make_token(scopes=[]))
try:
with pytest.raises(AuthorizationError) as exc_info:
await mcp.call_tool("api_tool", {})
finally:
auth_context_var.reset(tok)
assert not isinstance(exc_info.value, InsufficientScopeError)
assert "write" not in str(exc_info.value)
async def test_chain_shortfall_withholds_scopes_of_opaque_sibling(self):
# The outer middleware has a real scope shortfall, but the inner one
# pairs its scope requirement with an opaque check whose verdict is
# unknown. That layer's scope must not be disclosed.
def opaque_policy(ctx: AuthContext) -> bool:
return True
mcp = FastMCP(
middleware=[
AuthMiddleware(auth=require_scopes("admin")),
AuthMiddleware(auth=[opaque_policy, require_scopes("write")]),
]
)
@mcp.tool
def api_tool() -> str:
return "ok"
tok = set_token(make_token(scopes=[]))
try:
with pytest.raises(InsufficientScopeError) as exc_info:
await mcp.call_tool("api_tool", {})
finally:
auth_context_var.reset(tok)
assert exc_info.value.required_scopes == ["admin"]
async def test_chain_shortfall_stops_at_unevaluated_opaque_layer(self):
# The opaque middleware sits between two scope layers. The outer one
# raises before it ever runs, so whether it would admit the caller is
# unknown — and the scope behind it must not be disclosed. It returns
# True here to show the walk stops regardless of what the verdict
# would have been.
def tenant_check(ctx: AuthContext) -> bool:
return True
mcp = FastMCP(
middleware=[
AuthMiddleware(auth=require_scopes("admin")),
AuthMiddleware(auth=tenant_check),
AuthMiddleware(auth=require_scopes("write")),
]
)
@mcp.tool
def api_tool() -> str:
return "ok"
tok = set_token(make_token(scopes=[]))
try:
with pytest.raises(InsufficientScopeError) as exc_info:
await mcp.call_tool("api_tool", {})
finally:
auth_context_var.reset(tok)
assert exc_info.value.required_scopes == ["admin"]
async def test_chain_shortfall_spans_consecutive_scope_only_layers(self):
# The same chain without the opaque layer aggregates all of it, proving
# the reachability bound does not over-correct.
mcp = FastMCP(
middleware=[
AuthMiddleware(auth=require_scopes("admin")),
AuthMiddleware(auth=require_scopes("write")),
AuthMiddleware(auth=require_scopes("delete")),
]
)
@mcp.tool
def api_tool() -> str:
return "ok"
tok = set_token(make_token(scopes=[]))
try:
with pytest.raises(InsufficientScopeError) as exc_info:
await mcp.call_tool("api_tool", {})
finally:
auth_context_var.reset(tok)
assert exc_info.value.required_scopes == ["admin", "delete", "write"]
async def test_restrict_tag_shortfall_names_scope(self):
mcp = make_restricted_tag_server()
@mcp.tool(tags={"admin"})
def admin_tool() -> str:
return "ok"
tok = set_token(make_token(scopes=["read"]))
try:
with pytest.raises(InsufficientScopeError) as exc_info:
await mcp.call_tool("admin_tool", {})
finally:
auth_context_var.reset(tok)
assert exc_info.value.required_scopes == ["admin"]