mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-09 15:19:10 +02:00
feat: add AzureB2CProvider for Azure AD B2C user flows (#3995)
This commit is contained in:
Carlos Rian
GitHub
parent
2d6143c6d8
commit
eebdc8c031
3 changed files with 533 additions and 1 deletions
|
|
@ -458,3 +458,85 @@ For advanced OBO scenarios, use `CurrentAccessToken()` to get the user's token,
|
|||
<Tip>
|
||||
For a complete working example of Azure OBO with FastMCP, see [Pamela Fox's blog post on OBO flow for Entra-based MCP servers](https://blog.pamelafox.org/2026/01/using-on-behalf-of-flow-for-entra-based.html).
|
||||
</Tip>
|
||||
|
||||
## Azure AD B2C
|
||||
|
||||
<VersionBadge version="3.3.0" />
|
||||
|
||||
Azure AD B2C (Business-to-Consumer) uses different endpoints, scope URIs, and
|
||||
token issuers than standard Microsoft Entra ID. The `AzureProvider.from_b2c()`
|
||||
factory handles all of these differences automatically.
|
||||
|
||||
<Warning>
|
||||
Azure AD B2C does **not** support the On-Behalf-Of (OBO) flow. If you need
|
||||
OBO for downstream API calls, use `AzureProvider` with standard Entra ID
|
||||
instead.
|
||||
</Warning>
|
||||
|
||||
### Quick Start
|
||||
|
||||
```python server.py
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.server.auth.providers.azure import AzureProvider
|
||||
|
||||
auth = AzureProvider.from_b2c(
|
||||
tenant_name="mytenant",
|
||||
policy_name="B2C_1_susi",
|
||||
client_id="00000000-0000-0000-0000-000000000000",
|
||||
client_secret="my-secret",
|
||||
required_scopes=["mcp-access"],
|
||||
base_url="https://myserver.com",
|
||||
)
|
||||
|
||||
mcp = FastMCP("My App", auth=auth)
|
||||
```
|
||||
|
||||
`from_b2c()` derives the following values automatically:
|
||||
|
||||
| Derived value | Formula |
|
||||
|---|---|
|
||||
| Authority host | `{tenant_name}.b2clogin.com` |
|
||||
| Authorization endpoint | `https://{tenant_name}.b2clogin.com/{tenant_name}.onmicrosoft.com/{policy_name}/oauth2/v2.0/authorize` |
|
||||
| Token endpoint | `https://{tenant_name}.b2clogin.com/{tenant_name}.onmicrosoft.com/{policy_name}/oauth2/v2.0/token` |
|
||||
| Scope identifier URI | `https://{tenant_name}.onmicrosoft.com/{client_id}` |
|
||||
|
||||
### Token Issuer Validation
|
||||
|
||||
B2C access tokens carry the **tenant GUID** (not the `.onmicrosoft.com` name)
|
||||
in the `iss` claim, and the exact format varies by policy and custom-domain
|
||||
configuration. `from_b2c()` therefore **disables issuer validation by
|
||||
default**; **audience validation still enforces that tokens target the correct
|
||||
application**.
|
||||
|
||||
Once you have confirmed a successful end-to-end login, read the actual `iss`
|
||||
value from the decoded claims and enable strict validation:
|
||||
|
||||
```python
|
||||
auth = AzureProvider.from_b2c(
|
||||
tenant_name="mytenant",
|
||||
policy_name="B2C_1_susi",
|
||||
client_id="00000000-0000-0000-0000-000000000000",
|
||||
client_secret="my-secret",
|
||||
required_scopes=["mcp-access"],
|
||||
base_url="https://myserver.com",
|
||||
token_issuer="https://mytenant.b2clogin.com/11111111-2222-3333-4444-555555555555/v2.0/",
|
||||
)
|
||||
```
|
||||
|
||||
### Custom Domains
|
||||
|
||||
If your B2C tenant uses a [custom domain](https://learn.microsoft.com/en-us/azure/active-directory-b2c/custom-domain)
|
||||
(e.g. `auth.mycompany.com` instead of `mytenant.b2clogin.com`), pass it via
|
||||
`custom_domain`:
|
||||
|
||||
```python
|
||||
auth = AzureProvider.from_b2c(
|
||||
tenant_name="mytenant",
|
||||
policy_name="B2C_1_susi",
|
||||
client_id="00000000-0000-0000-0000-000000000000",
|
||||
client_secret="my-secret",
|
||||
required_scopes=["mcp-access"],
|
||||
base_url="https://myserver.com",
|
||||
custom_domain="auth.mycompany.com",
|
||||
)
|
||||
```
|
||||
|
|
|
|||
|
|
@ -117,6 +117,7 @@ class AzureProvider(OAuthProxy):
|
|||
forward_resource: bool = True,
|
||||
fallback_refresh_token_expiry_seconds: int | None = None,
|
||||
base_authority: str = "login.microsoftonline.com",
|
||||
token_issuer: str | None = None,
|
||||
http_client: httpx.AsyncClient | None = None,
|
||||
enable_cimd: bool = True,
|
||||
) -> None:
|
||||
|
|
@ -141,6 +142,9 @@ class AzureProvider(OAuthProxy):
|
|||
redirect_path: Redirect path configured in Azure App registration (defaults to "/auth/callback")
|
||||
base_authority: Azure authority base URL (defaults to "login.microsoftonline.com").
|
||||
For Azure Government, use "login.microsoftonline.us".
|
||||
token_issuer: Override the expected `iss` claim value for JWT validation.
|
||||
Defaults to the standard Entra ID issuer derived from `base_authority`
|
||||
and `tenant_id`. Pass an explicit string to enforce a specific issuer.
|
||||
required_scopes: Custom API scope names WITHOUT prefix (e.g., ["read", "write"]).
|
||||
- Automatically prefixed with identifier_uri during initialization
|
||||
- Validated on all tokens
|
||||
|
|
@ -197,13 +201,14 @@ class AzureProvider(OAuthProxy):
|
|||
# to avoid redundant OBO exchanges for the same user + scopes.
|
||||
self._obo_credentials: OrderedDict[str, OnBehalfOfCredential] = OrderedDict()
|
||||
self._obo_max_credentials: int = 128
|
||||
self._obo_supported = True
|
||||
|
||||
# Apply defaults
|
||||
self.identifier_uri = identifier_uri or f"api://{client_id}"
|
||||
self.additional_authorize_scopes: list[str] = parsed_additional_scopes
|
||||
|
||||
# Always validate tokens against the app's API client ID using JWT
|
||||
issuer = f"https://{base_authority}/{tenant_id}/v2.0"
|
||||
issuer = token_issuer or f"https://{base_authority}/{tenant_id}/v2.0"
|
||||
jwks_uri = f"https://{base_authority}/{tenant_id}/discovery/v2.0/keys"
|
||||
|
||||
# Azure access tokens only include custom API scopes in the `scp` claim,
|
||||
|
|
@ -272,6 +277,84 @@ class AzureProvider(OAuthProxy):
|
|||
authority_info,
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def from_b2c(
|
||||
cls,
|
||||
*,
|
||||
tenant_name: str,
|
||||
policy_name: str,
|
||||
client_id: str,
|
||||
client_secret: str | None = None,
|
||||
required_scopes: list[str],
|
||||
base_url: str,
|
||||
custom_domain: str | None = None,
|
||||
identifier_uri: str | None = None,
|
||||
token_issuer: str | None = None,
|
||||
**kwargs: Any,
|
||||
) -> AzureProvider:
|
||||
"""Create an AzureProvider pre-configured for Azure AD B2C.
|
||||
|
||||
Derives authority host, tenant path, and identifier URI from
|
||||
`tenant_name` and `policy_name`, then delegates to the standard
|
||||
constructor. Returns a plain `AzureProvider` instance.
|
||||
|
||||
B2C issuer validation is disabled by default (`token_issuer=None`)
|
||||
because B2C issuers embed the tenant GUID. Pass an explicit
|
||||
`token_issuer` string once you know the real `iss` value.
|
||||
|
||||
Azure AD B2C does **not** support OBO.
|
||||
|
||||
Args:
|
||||
tenant_name: Short B2C tenant name without `.onmicrosoft.com`
|
||||
(e.g. `"mytenant"`).
|
||||
policy_name: User-flow or custom-policy name
|
||||
(e.g. `"B2C_1_susi"`).
|
||||
client_id: Application (client) ID from the B2C app registration.
|
||||
client_secret: Client secret from the B2C app registration.
|
||||
required_scopes: Custom API scope names without prefix
|
||||
(e.g. `["mcp-access"]`).
|
||||
base_url: Public base URL of this server.
|
||||
custom_domain: Custom domain for the B2C authority
|
||||
(e.g. `"auth.mycompany.com"`). Defaults to
|
||||
`{tenant_name}.b2clogin.com`.
|
||||
identifier_uri: Application ID URI. Defaults to
|
||||
`https://{tenant_name}.onmicrosoft.com/{client_id}`.
|
||||
token_issuer: Expected `iss` claim. `None` (default) disables
|
||||
issuer validation.
|
||||
**kwargs: Forwarded to `AzureProvider.__init__`.
|
||||
"""
|
||||
if ".onmicrosoft.com" in tenant_name:
|
||||
raise ValueError(
|
||||
f"tenant_name should be the short name without the "
|
||||
f".onmicrosoft.com suffix (e.g. 'mytenant'), got {tenant_name!r}"
|
||||
)
|
||||
|
||||
if custom_domain is not None:
|
||||
custom_domain = (
|
||||
custom_domain.removeprefix("https://")
|
||||
.removeprefix("http://")
|
||||
.rstrip("/")
|
||||
)
|
||||
|
||||
authority = custom_domain or f"{tenant_name}.b2clogin.com"
|
||||
tenant_path = f"{tenant_name}.onmicrosoft.com/{policy_name}"
|
||||
uri = identifier_uri or f"https://{tenant_name}.onmicrosoft.com/{client_id}"
|
||||
|
||||
provider = cls(
|
||||
client_id=client_id,
|
||||
client_secret=client_secret,
|
||||
tenant_id=tenant_path,
|
||||
required_scopes=required_scopes,
|
||||
base_url=base_url,
|
||||
base_authority=authority,
|
||||
identifier_uri=uri,
|
||||
token_issuer=token_issuer,
|
||||
**kwargs,
|
||||
)
|
||||
provider._token_validator.issuer = token_issuer # type: ignore[union-attr]
|
||||
provider._obo_supported = False
|
||||
return provider
|
||||
|
||||
async def authorize(
|
||||
self,
|
||||
client: OAuthClientInformationFull,
|
||||
|
|
@ -512,8 +595,14 @@ class AzureProvider(OAuthProxy):
|
|||
A configured OnBehalfOfCredential ready for get_token() calls.
|
||||
|
||||
Raises:
|
||||
NotImplementedError: If OBO is not supported (e.g. Azure AD B2C).
|
||||
ImportError: If azure-identity is not installed (requires fastmcp[azure]).
|
||||
"""
|
||||
if not self._obo_supported:
|
||||
raise NotImplementedError(
|
||||
"Azure AD B2C does not support the On-Behalf-Of (OBO) flow. "
|
||||
"Use AzureProvider with standard Entra ID for OBO scenarios."
|
||||
)
|
||||
_require_azure_identity("OBO token exchange")
|
||||
from azure.identity.aio import OnBehalfOfCredential
|
||||
|
||||
|
|
|
|||
|
|
@ -822,3 +822,364 @@ class TestAzureProvider:
|
|||
# Should have 3 items (read deduplicated, plus offline_access)
|
||||
assert len(result) == 3
|
||||
assert result.count("api://my-api/read") == 1
|
||||
|
||||
|
||||
class TestAzureProviderTokenIssuer:
|
||||
"""Tests for the token_issuer parameter on AzureProvider."""
|
||||
|
||||
def test_default_issuer_is_derived(self, memory_storage: MemoryStore):
|
||||
"""Without token_issuer, the issuer is derived from base_authority/tenant_id."""
|
||||
provider = AzureProvider(
|
||||
client_id="test_client",
|
||||
client_secret="test_secret",
|
||||
tenant_id="my-tenant",
|
||||
base_url="https://myserver.com",
|
||||
required_scopes=["read"],
|
||||
jwt_signing_key="test-secret",
|
||||
client_storage=memory_storage,
|
||||
)
|
||||
|
||||
assert isinstance(provider._token_validator, JWTVerifier)
|
||||
assert (
|
||||
provider._token_validator.issuer
|
||||
== "https://login.microsoftonline.com/my-tenant/v2.0"
|
||||
)
|
||||
|
||||
def test_explicit_token_issuer_is_used(self, memory_storage: MemoryStore):
|
||||
"""An explicit token_issuer string is passed to the verifier."""
|
||||
custom_issuer = "https://custom.issuer.com/v2.0"
|
||||
provider = AzureProvider(
|
||||
client_id="test_client",
|
||||
client_secret="test_secret",
|
||||
tenant_id="my-tenant",
|
||||
base_url="https://myserver.com",
|
||||
required_scopes=["read"],
|
||||
token_issuer=custom_issuer,
|
||||
jwt_signing_key="test-secret",
|
||||
client_storage=memory_storage,
|
||||
)
|
||||
|
||||
assert isinstance(provider._token_validator, JWTVerifier)
|
||||
assert provider._token_validator.issuer == custom_issuer
|
||||
|
||||
async def test_explicit_issuer_enforced(self, memory_storage: MemoryStore):
|
||||
"""With an explicit token_issuer, wrong issuers are rejected."""
|
||||
key_pair = RSAKeyPair.generate()
|
||||
expected = "https://expected.issuer.com/v2.0"
|
||||
provider = AzureProvider(
|
||||
client_id="test_client",
|
||||
client_secret="test_secret",
|
||||
tenant_id="my-tenant",
|
||||
base_url="https://myserver.com",
|
||||
required_scopes=["read"],
|
||||
token_issuer=expected,
|
||||
jwt_signing_key="test-secret",
|
||||
client_storage=memory_storage,
|
||||
)
|
||||
|
||||
assert isinstance(provider._token_validator, JWTVerifier)
|
||||
verifier = provider._token_validator
|
||||
verifier.public_key = key_pair.public_key
|
||||
verifier.jwks_uri = None
|
||||
|
||||
good_token = key_pair.create_token(
|
||||
subject="test-user",
|
||||
issuer=expected,
|
||||
audience="test_client",
|
||||
additional_claims={"scp": "read"},
|
||||
)
|
||||
assert await verifier.load_access_token(good_token) is not None
|
||||
|
||||
bad_token = key_pair.create_token(
|
||||
subject="test-user",
|
||||
issuer="https://wrong.issuer.com/v2.0",
|
||||
audience="test_client",
|
||||
additional_claims={"scp": "read"},
|
||||
)
|
||||
assert await verifier.load_access_token(bad_token) is None
|
||||
|
||||
|
||||
class TestAzureProviderFromB2C:
|
||||
"""Tests for the AzureProvider.from_b2c() classmethod factory."""
|
||||
|
||||
def test_b2c_endpoints_derived_correctly(self, memory_storage: MemoryStore):
|
||||
"""from_b2c() produces correct B2C authority and tenant path."""
|
||||
provider = AzureProvider.from_b2c(
|
||||
tenant_name="mytenant",
|
||||
policy_name="B2C_1_susi",
|
||||
client_id="client-id",
|
||||
client_secret="secret",
|
||||
required_scopes=["mcp-access"],
|
||||
base_url="https://myserver.com",
|
||||
jwt_signing_key="test-secret",
|
||||
client_storage=memory_storage,
|
||||
)
|
||||
|
||||
assert provider._upstream_authorization_endpoint == (
|
||||
"https://mytenant.b2clogin.com"
|
||||
"/mytenant.onmicrosoft.com/B2C_1_susi"
|
||||
"/oauth2/v2.0/authorize"
|
||||
)
|
||||
assert provider._upstream_token_endpoint == (
|
||||
"https://mytenant.b2clogin.com"
|
||||
"/mytenant.onmicrosoft.com/B2C_1_susi"
|
||||
"/oauth2/v2.0/token"
|
||||
)
|
||||
|
||||
def test_b2c_identifier_uri_uses_https(self, memory_storage: MemoryStore):
|
||||
"""from_b2c() sets identifier_uri with https:// scheme, not api://."""
|
||||
provider = AzureProvider.from_b2c(
|
||||
tenant_name="mytenant",
|
||||
policy_name="B2C_1_susi",
|
||||
client_id="00000000-0000-0000-0000-000000000001",
|
||||
client_secret="secret",
|
||||
required_scopes=["mcp-access"],
|
||||
base_url="https://myserver.com",
|
||||
jwt_signing_key="test-secret",
|
||||
client_storage=memory_storage,
|
||||
)
|
||||
|
||||
assert provider.identifier_uri == (
|
||||
"https://mytenant.onmicrosoft.com/00000000-0000-0000-0000-000000000001"
|
||||
)
|
||||
assert provider.identifier_uri.startswith("https://")
|
||||
assert not provider.identifier_uri.startswith("api://")
|
||||
|
||||
def test_b2c_issuer_disabled_by_default(self, memory_storage: MemoryStore):
|
||||
"""from_b2c() disables issuer validation by default (token_issuer=None)."""
|
||||
provider = AzureProvider.from_b2c(
|
||||
tenant_name="mytenant",
|
||||
policy_name="B2C_1_susi",
|
||||
client_id="client-id",
|
||||
client_secret="secret",
|
||||
required_scopes=["mcp-access"],
|
||||
base_url="https://myserver.com",
|
||||
jwt_signing_key="test-secret",
|
||||
client_storage=memory_storage,
|
||||
)
|
||||
|
||||
assert isinstance(provider._token_validator, JWTVerifier)
|
||||
assert provider._token_validator.issuer is None
|
||||
|
||||
def test_b2c_explicit_token_issuer(self, memory_storage: MemoryStore):
|
||||
"""from_b2c() forwards an explicit token_issuer to the verifier."""
|
||||
explicit_issuer = (
|
||||
"https://mytenant.b2clogin.com/11111111-2222-3333-4444-555555555555/v2.0/"
|
||||
)
|
||||
provider = AzureProvider.from_b2c(
|
||||
tenant_name="mytenant",
|
||||
policy_name="B2C_1_susi",
|
||||
client_id="client-id",
|
||||
client_secret="secret",
|
||||
required_scopes=["mcp-access"],
|
||||
base_url="https://myserver.com",
|
||||
token_issuer=explicit_issuer,
|
||||
jwt_signing_key="test-secret",
|
||||
client_storage=memory_storage,
|
||||
)
|
||||
|
||||
assert isinstance(provider._token_validator, JWTVerifier)
|
||||
assert provider._token_validator.issuer == explicit_issuer
|
||||
|
||||
def test_b2c_custom_domain(self, memory_storage: MemoryStore):
|
||||
"""from_b2c() uses custom_domain in place of {tenant}.b2clogin.com."""
|
||||
provider = AzureProvider.from_b2c(
|
||||
tenant_name="mytenant",
|
||||
policy_name="B2C_1_susi",
|
||||
client_id="client-id",
|
||||
client_secret="secret",
|
||||
required_scopes=["mcp-access"],
|
||||
base_url="https://myserver.com",
|
||||
custom_domain="auth.mycompany.com",
|
||||
jwt_signing_key="test-secret",
|
||||
client_storage=memory_storage,
|
||||
)
|
||||
|
||||
assert "auth.mycompany.com" in provider._upstream_authorization_endpoint
|
||||
assert "auth.mycompany.com" in provider._upstream_token_endpoint
|
||||
assert "mytenant.b2clogin.com" not in provider._upstream_authorization_endpoint
|
||||
|
||||
def test_b2c_custom_domain_with_scheme_normalized(
|
||||
self, memory_storage: MemoryStore
|
||||
):
|
||||
"""from_b2c() strips scheme and trailing slash from custom_domain."""
|
||||
provider = AzureProvider.from_b2c(
|
||||
tenant_name="mytenant",
|
||||
policy_name="B2C_1_susi",
|
||||
client_id="client-id",
|
||||
client_secret="secret",
|
||||
required_scopes=["mcp-access"],
|
||||
base_url="https://myserver.com",
|
||||
custom_domain="https://auth.mycompany.com/",
|
||||
jwt_signing_key="test-secret",
|
||||
client_storage=memory_storage,
|
||||
)
|
||||
|
||||
assert "auth.mycompany.com" in provider._upstream_authorization_endpoint
|
||||
assert "https://https://" not in provider._upstream_authorization_endpoint
|
||||
|
||||
def test_b2c_custom_identifier_uri(self, memory_storage: MemoryStore):
|
||||
"""from_b2c() respects an explicit identifier_uri override."""
|
||||
custom_uri = "https://mycompany.com/api/mcp"
|
||||
provider = AzureProvider.from_b2c(
|
||||
tenant_name="mytenant",
|
||||
policy_name="B2C_1_susi",
|
||||
client_id="client-id",
|
||||
client_secret="secret",
|
||||
required_scopes=["mcp-access"],
|
||||
base_url="https://myserver.com",
|
||||
identifier_uri=custom_uri,
|
||||
jwt_signing_key="test-secret",
|
||||
client_storage=memory_storage,
|
||||
)
|
||||
|
||||
assert provider.identifier_uri == custom_uri
|
||||
|
||||
def test_b2c_scope_prefix_uses_https(self, memory_storage: MemoryStore):
|
||||
"""from_b2c() scopes are prefixed with the https:// identifier URI."""
|
||||
provider = AzureProvider.from_b2c(
|
||||
tenant_name="mytenant",
|
||||
policy_name="B2C_1_susi",
|
||||
client_id="aabbccdd",
|
||||
client_secret="secret",
|
||||
required_scopes=["mcp-access"],
|
||||
base_url="https://myserver.com",
|
||||
jwt_signing_key="test-secret",
|
||||
client_storage=memory_storage,
|
||||
)
|
||||
|
||||
result = provider._prefix_scopes_for_azure(["mcp-access"])
|
||||
assert result == ["https://mytenant.onmicrosoft.com/aabbccdd/mcp-access"]
|
||||
|
||||
def test_b2c_returns_azure_provider_instance(self, memory_storage: MemoryStore):
|
||||
"""from_b2c() returns an AzureProvider, not a subclass."""
|
||||
provider = AzureProvider.from_b2c(
|
||||
tenant_name="mytenant",
|
||||
policy_name="B2C_1_susi",
|
||||
client_id="client-id",
|
||||
client_secret="secret",
|
||||
required_scopes=["mcp-access"],
|
||||
base_url="https://myserver.com",
|
||||
jwt_signing_key="test-secret",
|
||||
client_storage=memory_storage,
|
||||
)
|
||||
|
||||
assert type(provider) is AzureProvider
|
||||
|
||||
def test_b2c_custom_policy_name(self, memory_storage: MemoryStore):
|
||||
"""from_b2c() accepts custom policy names (B2C_1A_*)."""
|
||||
provider = AzureProvider.from_b2c(
|
||||
tenant_name="mytenant",
|
||||
policy_name="B2C_1A_SIGNUP_SIGNIN",
|
||||
client_id="client-id",
|
||||
client_secret="secret",
|
||||
required_scopes=["mcp-access"],
|
||||
base_url="https://myserver.com",
|
||||
jwt_signing_key="test-secret",
|
||||
client_storage=memory_storage,
|
||||
)
|
||||
|
||||
assert "B2C_1A_SIGNUP_SIGNIN" in provider._upstream_authorization_endpoint
|
||||
assert "B2C_1A_SIGNUP_SIGNIN" in provider._upstream_token_endpoint
|
||||
|
||||
async def test_b2c_token_accepted_with_any_issuer(
|
||||
self, memory_storage: MemoryStore
|
||||
):
|
||||
"""B2C provider (issuer=None) accepts tokens from any issuer."""
|
||||
key_pair = RSAKeyPair.generate()
|
||||
provider = AzureProvider.from_b2c(
|
||||
tenant_name="mytenant",
|
||||
policy_name="B2C_1_susi",
|
||||
client_id="my-client-id",
|
||||
client_secret="secret",
|
||||
required_scopes=["mcp-access"],
|
||||
base_url="https://myserver.com",
|
||||
jwt_signing_key="test-secret",
|
||||
client_storage=memory_storage,
|
||||
)
|
||||
|
||||
assert isinstance(provider._token_validator, JWTVerifier)
|
||||
verifier = provider._token_validator
|
||||
verifier.public_key = key_pair.public_key
|
||||
verifier.jwks_uri = None
|
||||
|
||||
token = key_pair.create_token(
|
||||
subject="test-user",
|
||||
issuer="https://mytenant.b2clogin.com/11111111-guid/v2.0/",
|
||||
audience="my-client-id",
|
||||
additional_claims={"scp": "mcp-access"},
|
||||
)
|
||||
result = await verifier.load_access_token(token)
|
||||
assert result is not None
|
||||
|
||||
async def test_b2c_token_rejected_with_wrong_audience(
|
||||
self, memory_storage: MemoryStore
|
||||
):
|
||||
"""B2C provider still rejects tokens with wrong audience."""
|
||||
key_pair = RSAKeyPair.generate()
|
||||
provider = AzureProvider.from_b2c(
|
||||
tenant_name="mytenant",
|
||||
policy_name="B2C_1_susi",
|
||||
client_id="my-client-id",
|
||||
client_secret="secret",
|
||||
required_scopes=["mcp-access"],
|
||||
base_url="https://myserver.com",
|
||||
jwt_signing_key="test-secret",
|
||||
client_storage=memory_storage,
|
||||
)
|
||||
|
||||
assert isinstance(provider._token_validator, JWTVerifier)
|
||||
verifier = provider._token_validator
|
||||
verifier.public_key = key_pair.public_key
|
||||
verifier.jwks_uri = None
|
||||
|
||||
token = key_pair.create_token(
|
||||
subject="test-user",
|
||||
issuer="https://mytenant.b2clogin.com/11111111-guid/v2.0/",
|
||||
audience="wrong-app-id",
|
||||
additional_claims={"scp": "mcp-access"},
|
||||
)
|
||||
result = await verifier.load_access_token(token)
|
||||
assert result is None
|
||||
|
||||
async def test_b2c_obo_raises_not_implemented(self, memory_storage: MemoryStore):
|
||||
"""from_b2c() providers must reject OBO calls with NotImplementedError."""
|
||||
provider = AzureProvider.from_b2c(
|
||||
tenant_name="mytenant",
|
||||
policy_name="B2C_1_susi",
|
||||
client_id="client-id",
|
||||
client_secret="secret",
|
||||
required_scopes=["mcp-access"],
|
||||
base_url="https://myserver.com",
|
||||
jwt_signing_key="test-secret",
|
||||
client_storage=memory_storage,
|
||||
)
|
||||
|
||||
with pytest.raises(NotImplementedError, match="OBO"):
|
||||
await provider.get_obo_credential(user_assertion="fake-token")
|
||||
|
||||
|
||||
class TestAzureProviderFromB2CInputValidation:
|
||||
"""Input validation for from_b2c() parameters."""
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"tenant_name",
|
||||
[
|
||||
"mytenant.onmicrosoft.com",
|
||||
"my.onmicrosoft.com.tenant",
|
||||
],
|
||||
)
|
||||
def test_tenant_name_with_onmicrosoft_suffix_rejected(
|
||||
self, memory_storage: MemoryStore, tenant_name: str
|
||||
):
|
||||
with pytest.raises(ValueError, match="onmicrosoft.com"):
|
||||
AzureProvider.from_b2c(
|
||||
tenant_name=tenant_name,
|
||||
policy_name="B2C_1_susi",
|
||||
client_id="client-id",
|
||||
client_secret="secret",
|
||||
required_scopes=["mcp-access"],
|
||||
base_url="https://myserver.com",
|
||||
jwt_signing_key="test-secret",
|
||||
client_storage=memory_storage,
|
||||
)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue