From ef0422a8f08da22f8d103a3a406cc1f753b66280 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Matthew=20Meyer=20=F0=9F=90=89=E2=9A=94=EF=B8=8F?= Date: Mon, 3 Aug 2026 11:36:18 -0700 Subject: [PATCH 1/2] Implement JWT validation tests and enhance authorization configuration - TID Issuer cross check - Added a new utility module for generating signed RS256 JWTs for testing purposes. - Created comprehensive tests for the JwtTokenValidator, covering various scenarios including audience validation, signature verification, and issuer checks. - Enhanced the AgentAuthConfiguration to support issuer lists from environment variables and improved validation logic for issuer settings. - Ensured that the configuration defaults and behaviors are consistent with expected security practices, including handling of non-string audience claims. --- .../aiohttp/jwt_authorization_middleware.py | 33 +- .../core/authorization/_entra_issuers.py | 172 ++++ .../authorization/agent_auth_configuration.py | 93 +- .../authorization/jwt/_authorize_request.py | 7 +- .../authorization/jwt/jwt_token_validator.py | 176 +++- .../fastapi/jwt_authorization_middleware.py | 1 - tests/_common/jwt_test_utils.py | 67 ++ .../authorization/test_jwt_token_validator.py | 865 ++++++++++++++++++ tests/hosting_core/test_auth_configuration.py | 168 +++- 9 files changed, 1552 insertions(+), 30 deletions(-) create mode 100644 libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/_entra_issuers.py create mode 100644 tests/_common/jwt_test_utils.py create mode 100644 tests/hosting_core/authorization/test_jwt_token_validator.py diff --git a/libraries/microsoft-agents-hosting-aiohttp/microsoft_agents/hosting/aiohttp/jwt_authorization_middleware.py b/libraries/microsoft-agents-hosting-aiohttp/microsoft_agents/hosting/aiohttp/jwt_authorization_middleware.py index b2237c73..c2ba8901 100644 --- a/libraries/microsoft-agents-hosting-aiohttp/microsoft_agents/hosting/aiohttp/jwt_authorization_middleware.py +++ b/libraries/microsoft-agents-hosting-aiohttp/microsoft_agents/hosting/aiohttp/jwt_authorization_middleware.py @@ -2,7 +2,7 @@ # Licensed under the MIT License. import functools - +import logging from typing import cast from aiohttp.web import Request, middleware, json_response @@ -11,6 +11,25 @@ from microsoft_agents.hosting.core.authorization.jwt import _authorize_request from microsoft_agents.hosting.core.http import HttpResponse +logger = logging.getLogger(__name__) + +_GENERIC_AUTH_ERROR = {"error": "Invalid token or authentication failed."} + + +def _extract_bearer_token(auth_header: str) -> str | None: + """Extracts the bearer token from a raw Authorization header value. + + Surrounding whitespace on the token is ignored for backward compatibility. + Returns None for anything malformed so callers can respond with a + consistent 401 instead of raising. + """ + parts = auth_header.split(maxsplit=1) + if len(parts) != 2 or parts[0].lower() != "bearer": + return None + + token = parts[1].strip() + return token if token and not any(char.isspace() for char in token) else None + async def _jwt_authorization_middleware(request: Request, handler): """ @@ -21,6 +40,18 @@ async def _jwt_authorization_middleware(request: Request, handler): ) auth_header = request.headers.get("Authorization") + if auth_header is not None: + # aiohttp-specific tolerance: trailing whitespace (spaces/tabs) after + # the bearer token is ignored for backward compatibility, but internal + # or extra non-whitespace content is rejected. Normalizing here keeps + # the shared `_authorize_request` parsing (used identically by the + # FastAPI adapter) strict and unchanged. + token = _extract_bearer_token(auth_header) + if token is None: + logger.warning("Malformed authorization header.") + return json_response(_GENERIC_AUTH_ERROR, status=401) + auth_header = f"Bearer {token}" + res = await _authorize_request(auth_header, auth_config) if isinstance(res, HttpResponse): diff --git a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/_entra_issuers.py b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/_entra_issuers.py new file mode 100644 index 00000000..63bd9985 --- /dev/null +++ b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/_entra_issuers.py @@ -0,0 +1,172 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. + +"""Shared helpers for recognizing Microsoft Entra ID / Bot Framework token issuers. + +Centralizes the cloud (public vs. US Government) and tenant-GUID parsing rules +used by both :class:`AgentAuthConfiguration` (default ``ISSUERS``) and +:class:`JwtTokenValidator` (issuer allow-list validation, tid-to-issuer +binding, and JWKS routing) so the two stay consistent. +""" + +from __future__ import annotations + +import re +from typing import Any, NamedTuple +from urllib.parse import urlparse + +# Well-known Microsoft first-party token issuer tenant IDs that are always +# trusted, mirroring the default ``ValidIssuers`` set used by the .NET SDK. +# These identify Microsoft infrastructure tenants used by Azure Bot Service, +# Teams and skill/agent-to-agent flows, so enabling issuer validation does not +# reject legitimate first-party traffic. +WELL_KNOWN_PUBLIC_TENANT_IDS = ( + "d6d49420-f39b-4df7-a1dc-d59a935871db", + "f8cdef31-a31e-4b4a-93e4-5f571e91255a", + "69e9b82d-4842-4902-8d1e-abc5b98a55e8", +) +WELL_KNOWN_GOV_TENANT_ID = "cab8a31a-1906-4287-a0d8-4eef66b95f6e" + +BOTFRAMEWORK_PUBLIC_ISSUER = "https://api.botframework.com" +BOTFRAMEWORK_GOV_ISSUER = "https://api.botframework.us" + +BOTFRAMEWORK_JWKS_URIS = { + BOTFRAMEWORK_PUBLIC_ISSUER: "https://login.botframework.com/v1/.well-known/keys", + BOTFRAMEWORK_GOV_ISSUER: "https://login.botframework.azure.us/v1/.well-known/keys", +} + +_GOV_AUTHORITY_RE = re.compile(r"login\.microsoftonline\.us", re.IGNORECASE) +_ENTRA_TENANT_GUID_RE = re.compile( + r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$", re.IGNORECASE +) +_V1_ISSUER_RE = re.compile(r"^https://sts\.windows\.net/([^/]+)/$", re.IGNORECASE) +_V2_ISSUER_RE = re.compile( + r"^(?i:https://login\.microsoftonline\.(com|us)/)([^/]+)/v2\.0$" +) + + +class EntraIssuerInfo(NamedTuple): + """Cloud-affinity metadata for a recognized Entra issuer.""" + + tenant: str + """The lowercased tenant GUID embedded in the issuer.""" + + gov: bool | None + """``True``/``False`` for a cloud-specific v2 issuer (US Gov / public), or + ``None`` for the cloud-agnostic v1 ``sts.windows.net`` host, which is + shared across the public and US Government clouds.""" + + +def is_gov_authority(authority: str | None) -> bool: + """Returns whether the configured authority targets Azure US Government.""" + return bool(authority) and bool(_GOV_AUTHORITY_RE.search(authority)) + + +def effective_tenant(tenant_id: str | None, authority: str | None) -> str | None: + """Returns the effective tenant identifier for a connection. + + The tenant segment embedded in ``authority``'s path (e.g. + ``https://login.microsoftonline.com/common`` or + ``.../{tenant-guid}``) takes precedence over a separately configured + ``tenant_id`` when present, mirroring the JS reference's + ``getEffectiveTenant``/``resolveAuthority`` precedence: the authority is + the more specific/authoritative signal when both are configured. Falls + back to ``tenant_id`` when ``authority`` has no path segment (or is not + configured). + """ + if authority: + segments = [ + segment + for segment in urlparse(authority.rstrip("/")).path.split("/") + if segment + ] + if segments: + return segments[-1] + return tenant_id + + +def entra_issuer_info(iss: Any) -> EntraIssuerInfo | None: + """Parses a recognized public or US Government Entra issuer. + + Only GUID tenants are recognized: a token's ``tid`` claim is always the + tenant GUID, so an issuer whose tenant segment is a domain alias (e.g. + ``contoso.onmicrosoft.com``) cannot be compared to ``tid`` and is + intentionally left unrecognized. Non-Entra issuers such as the Azure Bot + Service ``api.botframework.*`` issuers carry no ``tid`` claim and are not + matched here. + + :return: The issuer's tenant and cloud affinity, or ``None`` when ``iss`` + is not a (non-empty) string, or is not a recognized Entra issuer with + a GUID tenant. A non-string ``iss`` (e.g. a malformed array/object + claim) is rejected up front rather than passed to the regexes, which + require string/buffer-like input. + """ + if not isinstance(iss, str) or not iss: + return None + + v1_match = _V1_ISSUER_RE.match(iss) + if v1_match: + tenant = v1_match.group(1) + if _ENTRA_TENANT_GUID_RE.match(tenant): + return EntraIssuerInfo(tenant.lower(), None) + return None + + v2_match = _V2_ISSUER_RE.match(iss) + if v2_match: + cloud, tenant = v2_match.group(1), v2_match.group(2) + if _ENTRA_TENANT_GUID_RE.match(tenant): + return EntraIssuerInfo(tenant.lower(), cloud.lower() == "us") + return None + + +def default_connection_issuers( + tenant_id: str | None, authority: str | None +) -> list[str]: + """Builds the default (tenant-scoped) issuer allow-list for a connection. + + Used when ``ISSUERS`` were not explicitly configured. The effective + tenant (authority-embedded segment, when present, otherwise + ``tenant_id``; see :func:`effective_tenant`) is used so an + authority-scoped concrete or ``common``/``organizations`` tenant is + reflected correctly instead of falling back to a stale/absent + ``tenant_id``. + """ + tenant = effective_tenant(tenant_id, authority) or "common" + gov = is_gov_authority(authority) + bf_issuer = BOTFRAMEWORK_GOV_ISSUER if gov else BOTFRAMEWORK_PUBLIC_ISSUER + login_host = ( + "https://login.microsoftonline.us" + if gov + else "https://login.microsoftonline.com" + ) + return [ + bf_issuer, + f"https://sts.windows.net/{tenant}/", + f"{login_host}/{tenant}/v2.0", + ] + + +def well_known_first_party_issuers(authority: str | None) -> list[str]: + """Returns the well-known Microsoft first-party issuers always trusted for + the cloud implied by ``authority`` (public by default, US Government when + the authority is a US Government endpoint).""" + if is_gov_authority(authority): + return [ + BOTFRAMEWORK_GOV_ISSUER, + f"https://sts.windows.net/{WELL_KNOWN_GOV_TENANT_ID}/", + f"https://login.microsoftonline.us/{WELL_KNOWN_GOV_TENANT_ID}/v2.0", + ] + issuers = [BOTFRAMEWORK_PUBLIC_ISSUER] + for tenant in WELL_KNOWN_PUBLIC_TENANT_IDS: + issuers.append(f"https://sts.windows.net/{tenant}/") + issuers.append(f"https://login.microsoftonline.com/{tenant}/v2.0") + return issuers + + +def jwks_login_host(authority: str | None) -> str: + """Returns the Entra discovery-keys login host for the configured cloud.""" + return ( + "https://login.microsoftonline.us" + if is_gov_authority(authority) + else "https://login.microsoftonline.com" + ) diff --git a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/agent_auth_configuration.py b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/agent_auth_configuration.py index 63a1f540..ff0cd0ae 100644 --- a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/agent_auth_configuration.py +++ b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/agent_auth_configuration.py @@ -8,6 +8,9 @@ from microsoft_agents.activity.config._coercion import coerce_bool from microsoft_agents.hosting.core.authorization.auth_types import AuthTypes +from microsoft_agents.hosting.core.authorization._entra_issuers import ( + default_connection_issuers, +) # Env-style configuration keys that ``__init__`` recognizes and binds into # first-class fields (via the ``kwargs.get("...")`` aliases below). These are @@ -31,10 +34,25 @@ "ALT_BLUEPRINT_NAME", "ALTERNATEBLUEPRINTCONNECTIONNAME", "ANONYMOUS_ALLOWED", + "ISSUERS", + "VALIDATE_ISSUER", } ) +def _normalize_issuers(value: Any) -> list[str] | None: + if value is None: + return None + if isinstance(value, str): + values = value.replace(",", " ").split() + elif isinstance(value, dict): + values = value.values() + else: + values = value + issuers = [str(item).strip() for item in values if item and str(item).strip()] + return issuers or None + + class AgentAuthConfiguration: """ Configuration for Agent authentication. @@ -54,6 +72,15 @@ class AgentAuthConfiguration: IDPM_RESOURCE: The resource URL for Identity Proxy Manager (IDPM) token acquisition. Only meaningful when AUTH_TYPE is AuthTypes.identity_proxy_manager. When not set, it defaults to "api://AzureAdTokenExchange/.default". + ISSUERS: An optional explicit list of accepted token issuers. When not provided, + a cloud/tenant-scoped default is computed from TENANT_ID and AUTHORITY. + VALIDATE_ISSUER: Explicit opt-in flag (default False) that enables issuer + allow-list validation in JwtTokenValidator. Preserved as an opt-in for + backward compatibility: existing deployments are unaffected unless + they explicitly enable it. Note: tid-to-issuer binding is always + enforced by JwtTokenValidator (per issue #626) whenever the verified + token's issuer is a recognized Entra issuer with a GUID tenant, + regardless of this flag. """ TENANT_ID: str | None @@ -69,6 +96,7 @@ class AgentAuthConfiguration: AZURE_REGION: str | None IDPM_RESOURCE: str | None ANONYMOUS_ALLOWED: bool = False + VALIDATE_ISSUER: bool = False # Provider-specific settings that aren't first-class fields (e.g. the Entra # sidecar's SERVICE_NAME, SIDECAR_BASE_URL). Preserved here as a single dict @@ -99,6 +127,8 @@ def __init__( azure_region: str | None = None, idpm_resource: str | None = None, anonymous_allowed: bool | None = None, + issuers: list[str] | None = None, + validate_issuer: bool | None = None, **kwargs: Any, ): @@ -149,6 +179,27 @@ def __init__( name="ANONYMOUS_ALLOWED", ) + # Explicit, optional issuer allow-list. When not provided, ISSUERS falls + # back to a cloud/tenant-scoped default (see the ISSUERS property below). + self._configured_issuers = _normalize_issuers( + issuers if issuers is not None else kwargs.get("ISSUERS", None) + ) + # Explicit opt-in for issuer allow-list validation in JwtTokenValidator. + # Off by default so existing deployments are unaffected unless they + # explicitly enable it. Note: tid-to-issuer binding is always enforced + # by JwtTokenValidator regardless of this flag (see VALIDATE_ISSUER + # docstring above). Same fail-safe string coercion as ANONYMOUS_ALLOWED + # applies here. + self.VALIDATE_ISSUER = coerce_bool( + ( + validate_issuer + if validate_issuer is not None + else kwargs.get("VALIDATE_ISSUER", False) + ), + default=False, + name="VALIDATE_ISSUER", + ) + # Preserve genuinely provider-specific settings that aren't first-class # fields (e.g. the Entra sidecar's SERVICE_NAME, SIDECAR_BASE_URL) so # custom providers can read them via ``provider_settings``. Recognized @@ -166,13 +217,16 @@ def __init__( @property def ISSUERS(self) -> list[str]: """ - Gets the list of issuers. + Gets the list of accepted issuers: the explicitly configured list when + provided, otherwise a cloud/tenant-scoped default derived from the + effective tenant (the tenant segment embedded in AUTHORITY's path when + present, e.g. ``https://login.microsoftonline.com/common`` or + ``.../{tenant-guid}``, otherwise TENANT_ID) and AUTHORITY (US + Government authorities yield US Government issuer defaults). """ - return [ - "https://api.botframework.com", - f"https://sts.windows.net/{self.TENANT_ID}/", - f"https://login.microsoftonline.com/{self.TENANT_ID}/v2.0", - ] + if self._configured_issuers: + return list(self._configured_issuers) + return default_connection_issuers(self.TENANT_ID, self.AUTHORITY) # .NET-aligned, read-only property aliases. These mirror the property names on # the .NET ``ConnectionSettingsBase`` so provider code and cross-language readers @@ -203,13 +257,36 @@ def alternate_blueprint_connection_name(self) -> str | None: """Alias for :attr:`ALT_BLUEPRINT_ID` (.NET ``AlternateBlueprintConnectionName``).""" return self.ALT_BLUEPRINT_ID - def _jwt_patch_is_valid_aud(self, aud: str) -> bool: + def _jwt_patch_is_valid_aud(self, aud: Any) -> bool: """ - JWT-patch: Checks if the given audience is valid for any of the connections. + JWT-patch: Checks if the given audience is valid for any of the + connections. A non-string ``aud`` (e.g. the JWT-spec-permitted array + form, or a malformed numeric/object claim) is never valid: only a + single string audience is accepted, so this returns ``False`` rather + than raising, letting the caller reject it as an invalid audience. """ + if not isinstance(aud, str): + return False for conn in self._connections.values(): if not conn.CLIENT_ID: continue if aud.lower() == conn.CLIENT_ID.lower(): return True return False + + def _jwt_patch_find_connection(self, aud: Any) -> "AgentAuthConfiguration | None": + """ + JWT-patch: Finds the configured connection whose CLIENT_ID matches the + given audience (case-insensitive), so JwtTokenValidator can route JWKS + lookup and issuer validation to the correct connection's tenant/authority + in multi-connection setups. Returns None when no connection matches, + or when ``aud`` is not a string (e.g. an unverified array-form or + malformed claim) -- callers fall back to default routing rather than + failing. + """ + if not aud or not isinstance(aud, str): + return None + for conn in self._connections.values(): + if conn.CLIENT_ID and aud.lower() == conn.CLIENT_ID.lower(): + return conn + return None diff --git a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/jwt/_authorize_request.py b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/jwt/_authorize_request.py index 9e0a692d..a402d836 100644 --- a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/jwt/_authorize_request.py +++ b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/jwt/_authorize_request.py @@ -22,7 +22,8 @@ async def _authorize_request( :param authorization_header: The value of the Authorization header from the request. :param auth_config: The AgentAuthConfiguration instance containing authentication settings. - :return: A ClaimsIdentity object if the token is valid, or an HttpResponse with an error message and status code if the token is invalid or missing. + :return: A ClaimsIdentity object if the token is valid, or an HttpResponse with an + error message and status code if the token is invalid or missing. """ if auth_config is None: @@ -52,7 +53,9 @@ async def _authorize_request( claims = await validator.validate_token(parts[1]) return claims except (PyJWTError, ValueError) as e: - logger.warning("JWT validation error: %s", e) + # Log only the exception type -- the message/claims are not surfaced + # to the caller (or the logs) to avoid leaking validation internals. + logger.warning("JWT validation error: %s", type(e).__name__) return HttpResponse( body={"error": "Invalid token or authentication failed."}, status_code=401, diff --git a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/jwt/jwt_token_validator.py b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/jwt/jwt_token_validator.py index a46d1318..90cd7e31 100644 --- a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/jwt/jwt_token_validator.py +++ b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/jwt/jwt_token_validator.py @@ -11,6 +11,14 @@ from ..agent_auth_configuration import AgentAuthConfiguration from ..claims_identity import ClaimsIdentity +from .._entra_issuers import ( + BOTFRAMEWORK_JWKS_URIS, + effective_tenant, + entra_issuer_info, + is_gov_authority, + jwks_login_host, + well_known_first_party_issuers, +) logger = logging.getLogger(__name__) @@ -31,7 +39,8 @@ def __init__(self): self._cache = {} def _get_jwk_client(self, jwks_uri: str) -> _JwkClientCacheEntry: - """Retrieves a PyJWKClient for the given JWKS URI, using a cache to avoid creating multiple clients for the same URI.""" + """Retrieves a PyJWKClient for the given JWKS URI, using a cache to + avoid creating multiple clients for the same URI.""" if jwks_uri not in self._cache: self._cache[jwks_uri] = _JwkClientCacheEntry( PyJWKClient(jwks_uri), threading.Lock() @@ -73,13 +82,36 @@ def __init__(self, configuration: AgentAuthConfiguration): async def validate_token(self, token: str) -> ClaimsIdentity: """Validates a JWT token. + The unverified token is used only to select the matching configured + connection (by audience) for JWKS/tenant routing. All acceptance + checks -- audience, tid-to-issuer binding (when the issuer is a + recognized Entra issuer with a GUID tenant), and, when the matched + connection opts in via ``VALIDATE_ISSUER``, the issuer allow-list -- + are evaluated against the signature-verified claims. + :param token: The JWT token to validate. :return: A ClaimsIdentity object containing the token's claims if validation is successful. - :raises ValueError: If the token is invalid or if the audience claim is not valid + :raises ValueError: If the token, audience, tenant binding, or (when opted in) issuer is not valid. """ logger.debug("Validating JWT token.") - key = await self._get_public_key_or_secret(token) + header = get_unverified_header(token) + unverified_payload: dict = decode(token, options={"verify_signature": False}) + + # Route by the unverified audience only where the selected connection's + # cloud affects JWKS lookup. Public-cloud routing retains the legacy + # root-configuration endpoint for backward-compatible network egress. + # This is routing only -- final acceptance is checked against the + # signature-verified claims below. + routing_config = ( + self.configuration._jwt_patch_find_connection(unverified_payload.get("aud")) + or self.configuration + ) + jwks_uri = _build_jwks_uri( + unverified_payload.get("iss"), self.configuration, routing_config + ) + key = await self._jwk_client_manager.get_signing_key(jwks_uri, header) + decoded_token = decode( token, key=key, @@ -87,11 +119,28 @@ async def validate_token(self, token: str) -> ClaimsIdentity: leeway=300.0, options={"verify_aud": False}, ) - if not self.configuration._jwt_patch_is_valid_aud(decoded_token["aud"]): - logger.error(f"Invalid audience: {decoded_token['aud']}", stack_info=True) + + aud = decoded_token.get("aud", "") + if not self.configuration._jwt_patch_is_valid_aud(aud): + logger.warning("JWT audience not accepted.") raise ValueError("Invalid audience.") - # This probably should return a ClaimsIdentity + matched_config = ( + self.configuration._jwt_patch_find_connection(aud) or routing_config + ) + + # Issuer allow-list validation is explicit opt-in + # (AgentAuthConfiguration.VALIDATE_ISSUER) to preserve backward + # compatibility for existing deployments. Tid-to-issuer binding, + # however, is always enforced per issue #626: it only engages when + # the (verified) issuer is itself a recognized Entra issuer carrying + # a GUID tenant, so Bot Framework/non-Entra and tenant-alias issuers + # are unaffected, and a missing ``tid`` claim skips the check rather + # than failing closed. + if matched_config.VALIDATE_ISSUER: + _validate_issuer(decoded_token.get("iss"), matched_config) + _validate_tenant_binding(decoded_token.get("iss"), decoded_token.get("tid")) + logger.debug("JWT token validated successfully.") return ClaimsIdentity(decoded_token, True, security_token=token) @@ -100,17 +149,112 @@ def get_anonymous_claims(self) -> ClaimsIdentity: logger.debug("Returning anonymous claims identity.") return ClaimsIdentity({}, False, authentication_type="Anonymous") - async def _get_public_key_or_secret(self, token: str) -> PyJWK: - """Retrieves the public key or secret for validating the JWT token.""" - header = get_unverified_header(token) - unverified_payload: dict = decode(token, options={"verify_signature": False}) - jwks_uri = ( - "https://login.botframework.com/v1/.well-known/keys" - if unverified_payload.get("iss") == "https://api.botframework.com" - else f"https://login.microsoftonline.com/{self.configuration.TENANT_ID}/discovery/v2.0/keys" +def _build_jwks_uri( + iss: Any, + root_config: AgentAuthConfiguration, + routing_config: AgentAuthConfiguration, +) -> str: + """Builds the JWKS URI for the (unverified, routing-only) issuer and the + root and audience-selected connection configurations. + + Recognizes the Bot Framework public/US Government issuers directly; + US Government Entra connections use the audience-selected connection's + effective tenant. Public-cloud Entra routing deliberately preserves the + pre-existing endpoint based on the root validator configuration's + ``TENANT_ID`` so existing multi-connection and network-egress deployments + do not begin contacting a different discovery URL. + + A non-string ``iss`` (e.g. a malformed array/object claim) is never a + recognized Bot Framework issuer and is never used as a dict key here -- + it falls through to the default (non-Bot-Framework) routing instead of + risking a dict-lookup/hash failure on an unhashable value. + """ + bf_uri = BOTFRAMEWORK_JWKS_URIS.get(iss) if isinstance(iss, str) else None + if bf_uri: + return bf_uri + + if is_gov_authority(routing_config.AUTHORITY): + host = jwks_login_host(routing_config.AUTHORITY) + tenant = ( + effective_tenant(routing_config.TENANT_ID, routing_config.AUTHORITY) + or "common" ) + return f"{host}/{tenant}/discovery/v2.0/keys" - key = await self._jwk_client_manager.get_signing_key(jwks_uri, header) + return ( + "https://login.microsoftonline.com/" + f"{root_config.TENANT_ID}/discovery/v2.0/keys" + ) - return key + +def _get_valid_issuers(config: AgentAuthConfiguration) -> set[str]: + """Case-insensitive union of the connection's configured/default issuers + (``AgentAuthConfiguration.ISSUERS``) and the always-trusted Microsoft + first-party issuers for the connection's cloud.""" + combined = list(config.ISSUERS) + well_known_first_party_issuers(config.AUTHORITY) + return {issuer.lower() for issuer in combined} + + +def _is_multi_tenant(config: AgentAuthConfiguration) -> bool: + """A connection configured for the Entra ``common``/``organizations`` + meta-tenant ("blueprint" agent) has no single known calling tenant at + configuration time, so :func:`_is_acceptable_tenant_issuer` is used instead + of the strict allow-list. The token's tenant is still bound to its ``tid`` + claim by :func:`_validate_tenant_binding` and anchored by the signature and + audience checks. + + The effective tenant (authority-embedded segment when present, otherwise + TENANT_ID; see :func:`effective_tenant`) is used so a connection whose + AUTHORITY embeds ``common``/``organizations`` (rather than TENANT_ID) is + still recognized as multi-tenant. + """ + tenant = (effective_tenant(config.TENANT_ID, config.AUTHORITY) or "").lower() + return tenant in ("common", "organizations") + + +def _is_acceptable_tenant_issuer(iss: str, config: AgentAuthConfiguration) -> bool: + """Whether ``iss`` is a canonical Entra issuer acceptable for a multi-tenant + connection: it must carry a tenant GUID and, for cloud-specific v2 issuers, + match the connection's cloud (public vs US Government). The cloud-agnostic + v1 ``sts.windows.net`` issuer is accepted for either cloud.""" + info = entra_issuer_info(iss) + if info is None: + return False + return info.gov is None or info.gov == is_gov_authority(config.AUTHORITY) + + +def _validate_issuer(iss: Any, config: AgentAuthConfiguration) -> None: + """Validates that the token's ``iss`` claim is accepted for the matched + connection: either present in the connection's issuer allow-list, or, for + a multi-tenant (``common``/``organizations``) connection, a canonical + Entra issuer for the connection's cloud.""" + if isinstance(iss, str): + if iss.lower() in _get_valid_issuers(config): + return + if _is_multi_tenant(config) and _is_acceptable_tenant_issuer(iss, config): + return + logger.warning("JWT issuer not accepted for this connection.") + raise ValueError("Invalid issuer.") + + +def _validate_tenant_binding(iss: Any, tid: Any) -> None: + """Validates that an Entra token's ``tid`` claim matches the tenant GUID + embedded in its ``iss`` claim, preventing a token whose issuer was + allow-listed (e.g. one of the always-trusted Microsoft first-party + tenants) from being accepted on behalf of a different tenant. + + The binding only applies when ``iss`` is a recognized Entra issuer + carrying a GUID tenant; Bot Framework/non-Entra issuers (no ``tid`` + claim) and tenant-alias issuers are skipped. A missing ``tid`` claim also + skips the binding rather than failing closed, so tokens/issuers that do + not carry one remain unaffected. + """ + if not isinstance(iss, str): + return + info = entra_issuer_info(iss) + if info is None or not isinstance(tid, str): + return + if tid.lower() != info.tenant: + logger.warning("JWT tenant binding mismatch.") + raise ValueError("Invalid issuer.") diff --git a/libraries/microsoft-agents-hosting-fastapi/microsoft_agents/hosting/fastapi/jwt_authorization_middleware.py b/libraries/microsoft-agents-hosting-fastapi/microsoft_agents/hosting/fastapi/jwt_authorization_middleware.py index ad220763..ad22537e 100644 --- a/libraries/microsoft-agents-hosting-fastapi/microsoft_agents/hosting/fastapi/jwt_authorization_middleware.py +++ b/libraries/microsoft-agents-hosting-fastapi/microsoft_agents/hosting/fastapi/jwt_authorization_middleware.py @@ -6,7 +6,6 @@ from fastapi import Request from fastapi.responses import JSONResponse - from starlette.types import ASGIApp, Receive, Scope, Send from microsoft_agents.hosting.core import AgentAuthConfiguration diff --git a/tests/_common/jwt_test_utils.py b/tests/_common/jwt_test_utils.py new file mode 100644 index 00000000..42a66468 --- /dev/null +++ b/tests/_common/jwt_test_utils.py @@ -0,0 +1,67 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. + +"""Shared helpers for building signed RS256 JWTs in tests. + +Generates an in-memory RSA keypair per call so tests can sign tokens with the +private key and have JwtTokenValidator "fetch" the matching public key via a +monkeypatched JWKS client, without any real network access. +""" + +from __future__ import annotations + +import json +import time +from typing import Any + +import jwt +from cryptography.hazmat.primitives.asymmetric import rsa +from cryptography.hazmat.primitives.asymmetric.rsa import RSAPrivateKey, RSAPublicKey + + +def generate_rsa_keypair() -> tuple[RSAPrivateKey, RSAPublicKey]: + """Generates a fresh RSA keypair for signing/verifying test tokens.""" + private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + return private_key, private_key.public_key() + + +def make_signed_jwt( + private_key: RSAPrivateKey, + claims: dict[str, Any], + kid: str = "test-kid", + expires_in: float = 3600.0, +) -> str: + """Encodes ``claims`` as an RS256 JWT signed with ``private_key``. + + ``exp`` is filled in from ``expires_in`` (seconds from now) unless the + caller already supplied one. + """ + payload = dict(claims) + payload.setdefault("exp", int(time.time() + expires_in)) + return jwt.encode(payload, private_key, algorithm="RS256", headers={"kid": kid}) + + +def make_signed_jwt_with_raw_claims( + private_key: RSAPrivateKey, + claims: dict[str, Any], + kid: str = "test-kid", + expires_in: float = 3600.0, +) -> str: + """Like :func:`make_signed_jwt`, but signs via the lower-level JWS API so + claim values are serialized as-is (no PyJWT claims-shape validation). + + PyJWT's ``jwt.encode`` (the higher-level JWT API) rejects a non-string + ``iss`` at encode time (``TypeError: Issuer (iss) must be a string.``), + which makes it impossible to build regression tokens for malformed + ``iss`` shapes (list/dict) via ``make_signed_jwt``. This helper drops + down to ``jwt.api_jws`` (JWS: a signed, opaque payload) to produce a + structurally valid, signed token carrying any JSON-serializable claims, + exactly mirroring what a non-conformant or malicious token producer + could hand to a real deployment. + """ + payload = dict(claims) + payload.setdefault("exp", int(time.time() + expires_in)) + payload_bytes = json.dumps(payload).encode("utf-8") + return jwt.api_jws.encode( + payload_bytes, private_key, algorithm="RS256", headers={"kid": kid} + ) diff --git a/tests/hosting_core/authorization/test_jwt_token_validator.py b/tests/hosting_core/authorization/test_jwt_token_validator.py new file mode 100644 index 00000000..39be5cbd --- /dev/null +++ b/tests/hosting_core/authorization/test_jwt_token_validator.py @@ -0,0 +1,865 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. + +import uuid + +import jwt as pyjwt +import pytest + +from microsoft_agents.hosting.core import AgentAuthConfiguration +from microsoft_agents.hosting.core.authorization.jwt.jwt_token_validator import ( + JwtTokenValidator, +) + +from tests._common.jwt_test_utils import ( + generate_rsa_keypair, + make_signed_jwt, + make_signed_jwt_with_raw_claims, +) + + +def _patch_signing_key(monkeypatch, validator, public_key, captured_uris=None): + async def fake_get_signing_key(jwks_uri, header): + if captured_uris is not None: + captured_uris.append(jwks_uri) + return public_key + + # Only mocked member: the JWKS client manager's network call. + monkeypatch.setattr( + validator._jwk_client_manager, "get_signing_key", fake_get_signing_key + ) + + +class TestJwtTokenValidatorAudienceAndSignature: + @pytest.mark.asyncio + async def test_validate_token_success_returns_authenticated_claims( + self, monkeypatch + ): + private_key, public_key = generate_rsa_keypair() + config = AgentAuthConfiguration(client_id="client-1", tenant_id="tenant-1") + validator = JwtTokenValidator(config) + _patch_signing_key(monkeypatch, validator, public_key) + + token = make_signed_jwt(private_key, {"aud": "client-1"}) + identity = await validator.validate_token(token) + + assert identity.is_authenticated is True + assert identity.claims["aud"] == "client-1" + + @pytest.mark.asyncio + async def test_validate_token_invalid_audience_rejected(self, monkeypatch): + private_key, public_key = generate_rsa_keypair() + config = AgentAuthConfiguration(client_id="client-1", tenant_id="tenant-1") + validator = JwtTokenValidator(config) + _patch_signing_key(monkeypatch, validator, public_key) + + token = make_signed_jwt(private_key, {"aud": "someone-else"}) + + with pytest.raises(ValueError, match="Invalid audience"): + await validator.validate_token(token) + + @pytest.mark.asyncio + async def test_validate_token_bad_signature_rejected(self, monkeypatch): + _, public_key = generate_rsa_keypair() + wrong_private_key, _ = generate_rsa_keypair() + config = AgentAuthConfiguration(client_id="client-1", tenant_id="tenant-1") + validator = JwtTokenValidator(config) + _patch_signing_key(monkeypatch, validator, public_key) + + # Signed with a different private key than the one whose public key + # the (mocked) JWKS lookup returns. + token = make_signed_jwt(wrong_private_key, {"aud": "client-1"}) + + with pytest.raises(pyjwt.PyJWTError): + await validator.validate_token(token) + + @pytest.mark.asyncio + async def test_validate_token_expired_rejected(self, monkeypatch): + private_key, public_key = generate_rsa_keypair() + config = AgentAuthConfiguration(client_id="client-1", tenant_id="tenant-1") + validator = JwtTokenValidator(config) + _patch_signing_key(monkeypatch, validator, public_key) + + # Leeway is 300s, so put the expiry well beyond that. + token = make_signed_jwt(private_key, {"aud": "client-1"}, expires_in=-3600.0) + + with pytest.raises(pyjwt.ExpiredSignatureError): + await validator.validate_token(token) + + +class TestJwtTokenValidatorMalformedClaimTypes: + """Regression tests for non-string ``aud``/``iss`` claims (e.g. the + JWT-spec-permitted array form, or malformed numeric/object claims), which + must never raise AttributeError/TypeError -- only the well-defined + ValueError rejections (or, where a check is skipped, successful + authentication) -- so middleware never leaks an unhandled 500. + """ + + @pytest.mark.asyncio + async def test_array_audience_rejected_as_invalid_audience(self, monkeypatch): + private_key, public_key = generate_rsa_keypair() + config = AgentAuthConfiguration(client_id="client-1", tenant_id="tenant-1") + validator = JwtTokenValidator(config) + _patch_signing_key(monkeypatch, validator, public_key) + + # RFC 7519 permits `aud` as an array of strings; this codebase only + # accepts a single string audience, so it must be cleanly rejected + # rather than crashing on `.lower()` (AttributeError on a list). + token = make_signed_jwt(private_key, {"aud": ["client-1", "someone-else"]}) + + with pytest.raises(ValueError, match="Invalid audience"): + await validator.validate_token(token) + + @pytest.mark.asyncio + async def test_numeric_audience_rejected_as_invalid_audience(self, monkeypatch): + private_key, public_key = generate_rsa_keypair() + config = AgentAuthConfiguration(client_id="client-1", tenant_id="tenant-1") + validator = JwtTokenValidator(config) + _patch_signing_key(monkeypatch, validator, public_key) + + token = make_signed_jwt(private_key, {"aud": 12345}) + + with pytest.raises(ValueError, match="Invalid audience"): + await validator.validate_token(token) + + @pytest.mark.asyncio + async def test_list_issuer_does_not_crash_routing_or_tenant_binding( + self, monkeypatch + ): + # A list `iss` is unhashable and must not be used as a dict key for + # the Bot Framework JWKS lookup (routing) nor crash tenant binding. + # With VALIDATE_ISSUER left at its default (False), tenant binding is + # skipped for a non-string issuer, so the token is otherwise accepted. + private_key, public_key = generate_rsa_keypair() + config = AgentAuthConfiguration(client_id="client-1", tenant_id="tenant-1") + validator = JwtTokenValidator(config) + captured_uris = [] + _patch_signing_key(monkeypatch, validator, public_key, captured_uris) + + token = make_signed_jwt_with_raw_claims( + private_key, + { + "aud": "client-1", + "iss": ["https://a.example.com", "https://b.example.com"], + }, + ) + + identity = await validator.validate_token(token) + assert identity.is_authenticated is True + # Falls through to default (non-Bot-Framework) routing. + assert captured_uris == [ + "https://login.microsoftonline.com/tenant-1/discovery/v2.0/keys" + ] + + @pytest.mark.asyncio + async def test_dict_issuer_does_not_crash_routing_or_tenant_binding( + self, monkeypatch + ): + # A dict `iss` is unhashable, same concern as the list case above. + private_key, public_key = generate_rsa_keypair() + config = AgentAuthConfiguration(client_id="client-1", tenant_id="tenant-1") + validator = JwtTokenValidator(config) + captured_uris = [] + _patch_signing_key(monkeypatch, validator, public_key, captured_uris) + + token = make_signed_jwt_with_raw_claims( + private_key, + {"aud": "client-1", "iss": {"unexpected": "object"}}, + ) + + identity = await validator.validate_token(token) + assert identity.is_authenticated is True + assert captured_uris == [ + "https://login.microsoftonline.com/tenant-1/discovery/v2.0/keys" + ] + + @pytest.mark.asyncio + async def test_list_issuer_rejected_as_invalid_issuer_when_validate_issuer_enabled( + self, monkeypatch + ): + # With VALIDATE_ISSUER opted in, a non-string issuer cannot match the + # (string) allow-list or the multi-tenant canonical-issuer check, and + # must be cleanly rejected rather than crash. + private_key, public_key = generate_rsa_keypair() + config = AgentAuthConfiguration( + client_id="client-1", tenant_id="tenant-1", validate_issuer=True + ) + validator = JwtTokenValidator(config) + _patch_signing_key(monkeypatch, validator, public_key) + + token = make_signed_jwt_with_raw_claims( + private_key, + {"aud": "client-1", "iss": ["https://a.example.com"]}, + ) + + with pytest.raises(ValueError, match="Invalid issuer"): + await validator.validate_token(token) + + @pytest.mark.asyncio + async def test_non_string_tid_skips_tenant_binding(self, monkeypatch): + private_key, public_key = generate_rsa_keypair() + tenant_id = str(uuid.uuid4()) + config = AgentAuthConfiguration(client_id="client-1", tenant_id=tenant_id) + validator = JwtTokenValidator(config) + _patch_signing_key(monkeypatch, validator, public_key) + + token = make_signed_jwt_with_raw_claims( + private_key, + { + "aud": "client-1", + "iss": f"https://login.microsoftonline.com/{tenant_id}/v2.0", + "tid": ["malformed"], + }, + ) + + identity = await validator.validate_token(token) + assert identity.is_authenticated is True + + +class TestJwtTokenValidatorIssuerOptIn: + @pytest.mark.asyncio + async def test_tenant_binding_enforced_even_when_issuer_validation_disabled( + self, monkeypatch + ): + # Issue #626: tid-to-issuer binding is always enforced (it is not + # gated by VALIDATE_ISSUER); only the issuer allow-list check is + # opt-in. A matching-audience token whose recognized Entra `iss` + # carries a GUID tenant that does not match its own `tid` must be + # rejected even with VALIDATE_ISSUER left at its default (False). + private_key, public_key = generate_rsa_keypair() + config = AgentAuthConfiguration(client_id="client-1", tenant_id="tenant-1") + assert config.VALIDATE_ISSUER is False + validator = JwtTokenValidator(config) + _patch_signing_key(monkeypatch, validator, public_key) + + other_tenant = str(uuid.uuid4()) + token = make_signed_jwt( + private_key, + { + "aud": "client-1", + "iss": f"https://login.microsoftonline.com/{other_tenant}/v2.0", + "tid": str(uuid.uuid4()), # deliberately mismatched + }, + ) + + with pytest.raises(ValueError, match="Invalid issuer"): + await validator.validate_token(token) + + @pytest.mark.asyncio + async def test_missing_tid_skips_binding_even_when_issuer_validation_disabled( + self, monkeypatch + ): + # Resolved choice: a missing tid must SKIP binding rather than reject, + # regardless of VALIDATE_ISSUER. + private_key, public_key = generate_rsa_keypair() + other_tenant = str(uuid.uuid4()) + config = AgentAuthConfiguration(client_id="client-1", tenant_id="tenant-1") + assert config.VALIDATE_ISSUER is False + validator = JwtTokenValidator(config) + _patch_signing_key(monkeypatch, validator, public_key) + + token = make_signed_jwt( + private_key, + { + "aud": "client-1", + "iss": f"https://login.microsoftonline.com/{other_tenant}/v2.0", + # no "tid" claim at all + }, + ) + + identity = await validator.validate_token(token) + assert identity.is_authenticated is True + + @pytest.mark.asyncio + async def test_noncanonical_entra_issuer_variants_skip_binding( + self, monkeypatch + ): + private_key, public_key = generate_rsa_keypair() + issuer_tenant = str(uuid.uuid4()) + mismatched_tid = str(uuid.uuid4()) + config = AgentAuthConfiguration(client_id="client-1", tenant_id="tenant-1") + validator = JwtTokenValidator(config) + _patch_signing_key(monkeypatch, validator, public_key) + + for issuer in ( + f"https://sts.windows.net/{issuer_tenant}", + f"https://login.microsoftonline.com/{issuer_tenant}/v2.0/", + f"https://login.microsoftonline.com/{issuer_tenant}/V2.0", + ): + token = make_signed_jwt( + private_key, + {"aud": "client-1", "iss": issuer, "tid": mismatched_tid}, + ) + identity = await validator.validate_token(token) + assert identity.is_authenticated is True + + @pytest.mark.asyncio + async def test_issuer_allow_list_not_enforced_when_disabled(self, monkeypatch): + # With VALIDATE_ISSUER left at its default (False), an unrecognized + # issuer (not in the allow-list, and self-consistent with its own + # tid so tenant binding does not reject it) must still be accepted: + # only the issuer allow-list check is opt-in. + private_key, public_key = generate_rsa_keypair() + other_tenant = str(uuid.uuid4()) + config = AgentAuthConfiguration(client_id="client-1", tenant_id="tenant-1") + assert config.VALIDATE_ISSUER is False + validator = JwtTokenValidator(config) + _patch_signing_key(monkeypatch, validator, public_key) + + token = make_signed_jwt( + private_key, + { + "aud": "client-1", + "iss": f"https://login.microsoftonline.com/{other_tenant}/v2.0", + "tid": other_tenant, # self-consistent: binding passes + }, + ) + + identity = await validator.validate_token(token) + assert identity.is_authenticated is True + + @pytest.mark.asyncio + async def test_validate_issuer_enabled_default_issuer_accepted(self, monkeypatch): + private_key, public_key = generate_rsa_keypair() + tenant_id = str(uuid.uuid4()) + config = AgentAuthConfiguration( + client_id="client-1", tenant_id=tenant_id, validate_issuer=True + ) + validator = JwtTokenValidator(config) + _patch_signing_key(monkeypatch, validator, public_key) + + token = make_signed_jwt( + private_key, + { + "aud": "client-1", + "iss": f"https://login.microsoftonline.com/{tenant_id}/v2.0", + "tid": tenant_id, + }, + ) + + identity = await validator.validate_token(token) + assert identity.is_authenticated is True + + @pytest.mark.asyncio + async def test_validate_issuer_enabled_unrecognized_issuer_rejected( + self, monkeypatch + ): + private_key, public_key = generate_rsa_keypair() + tenant_id = str(uuid.uuid4()) + other_tenant = str(uuid.uuid4()) + config = AgentAuthConfiguration( + client_id="client-1", tenant_id=tenant_id, validate_issuer=True + ) + validator = JwtTokenValidator(config) + _patch_signing_key(monkeypatch, validator, public_key) + + token = make_signed_jwt( + private_key, + { + "aud": "client-1", + "iss": f"https://login.microsoftonline.com/{other_tenant}/v2.0", + "tid": other_tenant, + }, + ) + + with pytest.raises(ValueError, match="Invalid issuer"): + await validator.validate_token(token) + + @pytest.mark.asyncio + async def test_validate_issuer_enabled_v1_issuer_recognized_and_bound( + self, monkeypatch + ): + private_key, public_key = generate_rsa_keypair() + tenant_id = str(uuid.uuid4()) + config = AgentAuthConfiguration( + client_id="client-1", tenant_id=tenant_id, validate_issuer=True + ) + validator = JwtTokenValidator(config) + _patch_signing_key(monkeypatch, validator, public_key) + + token = make_signed_jwt( + private_key, + { + "aud": "client-1", + "iss": f"https://sts.windows.net/{tenant_id}/", + "tid": tenant_id, + }, + ) + + identity = await validator.validate_token(token) + assert identity.is_authenticated is True + + @pytest.mark.asyncio + async def test_validate_issuer_enabled_tid_mismatch_rejected(self, monkeypatch): + private_key, public_key = generate_rsa_keypair() + tenant_id = str(uuid.uuid4()) + mismatched_tid = str(uuid.uuid4()) + config = AgentAuthConfiguration( + client_id="client-1", tenant_id=tenant_id, validate_issuer=True + ) + validator = JwtTokenValidator(config) + _patch_signing_key(monkeypatch, validator, public_key) + + # iss matches the configured tenant's default issuer, but tid claims a + # different tenant -- the binding must reject this. + token = make_signed_jwt( + private_key, + { + "aud": "client-1", + "iss": f"https://login.microsoftonline.com/{tenant_id}/v2.0", + "tid": mismatched_tid, + }, + ) + + with pytest.raises(ValueError, match="Invalid issuer"): + await validator.validate_token(token) + + @pytest.mark.asyncio + async def test_validate_issuer_enabled_missing_tid_skips_binding(self, monkeypatch): + # Resolved choice: a missing tid must SKIP binding rather than reject. + private_key, public_key = generate_rsa_keypair() + tenant_id = str(uuid.uuid4()) + config = AgentAuthConfiguration( + client_id="client-1", tenant_id=tenant_id, validate_issuer=True + ) + validator = JwtTokenValidator(config) + _patch_signing_key(monkeypatch, validator, public_key) + + token = make_signed_jwt( + private_key, + { + "aud": "client-1", + "iss": f"https://login.microsoftonline.com/{tenant_id}/v2.0", + # no "tid" claim at all + }, + ) + + identity = await validator.validate_token(token) + assert identity.is_authenticated is True + + @pytest.mark.asyncio + async def test_validate_issuer_enabled_bot_framework_issuer_skips_binding( + self, monkeypatch + ): + private_key, public_key = generate_rsa_keypair() + config = AgentAuthConfiguration( + client_id="client-1", tenant_id="tenant-1", validate_issuer=True + ) + validator = JwtTokenValidator(config) + captured_uris = [] + _patch_signing_key(monkeypatch, validator, public_key, captured_uris) + + token = make_signed_jwt( + private_key, + { + "aud": "client-1", + "iss": "https://api.botframework.com", + # Bot Framework tokens carry no tid; binding must be skipped. + }, + ) + + identity = await validator.validate_token(token) + assert identity.is_authenticated is True + assert captured_uris == ["https://login.botframework.com/v1/.well-known/keys"] + + @pytest.mark.asyncio + async def test_validate_issuer_enabled_alias_tenant_issuer_skips_binding( + self, monkeypatch + ): + # An operator-configured issuer using a tenant domain alias (rather + # than a GUID) cannot be compared to the GUID `tid` claim, so binding + # is skipped even though the issuer itself is explicitly allow-listed. + private_key, public_key = generate_rsa_keypair() + config = AgentAuthConfiguration( + client_id="client-1", + tenant_id="tenant-1", + validate_issuer=True, + issuers=["https://login.microsoftonline.com/contoso.onmicrosoft.com/v2.0"], + ) + validator = JwtTokenValidator(config) + _patch_signing_key(monkeypatch, validator, public_key) + + token = make_signed_jwt( + private_key, + { + "aud": "client-1", + "iss": "https://login.microsoftonline.com/contoso.onmicrosoft.com/v2.0", + "tid": str(uuid.uuid4()), + }, + ) + + identity = await validator.validate_token(token) + assert identity.is_authenticated is True + + @pytest.mark.asyncio + async def test_validate_issuer_enabled_well_known_first_party_issuer_accepted( + self, monkeypatch + ): + # Well-known Microsoft first-party tenants are always trusted even + # though they are not the connection's own configured tenant. + private_key, public_key = generate_rsa_keypair() + well_known_tenant = "d6d49420-f39b-4df7-a1dc-d59a935871db" + config = AgentAuthConfiguration( + client_id="client-1", + tenant_id=str(uuid.uuid4()), + validate_issuer=True, + ) + validator = JwtTokenValidator(config) + _patch_signing_key(monkeypatch, validator, public_key) + + token = make_signed_jwt( + private_key, + { + "aud": "client-1", + "iss": f"https://login.microsoftonline.com/{well_known_tenant}/v2.0", + "tid": well_known_tenant, + }, + ) + + identity = await validator.validate_token(token) + assert identity.is_authenticated is True + + @pytest.mark.asyncio + async def test_validate_issuer_enabled_explicit_issuers_used(self, monkeypatch): + private_key, public_key = generate_rsa_keypair() + config = AgentAuthConfiguration( + client_id="client-1", + tenant_id="tenant-1", + validate_issuer=True, + issuers=["https://custom-issuer.example.com/"], + ) + validator = JwtTokenValidator(config) + _patch_signing_key(monkeypatch, validator, public_key) + + token = make_signed_jwt( + private_key, + {"aud": "client-1", "iss": "https://custom-issuer.example.com/"}, + ) + + identity = await validator.validate_token(token) + assert identity.is_authenticated is True + + @pytest.mark.asyncio + async def test_validate_issuer_enabled_common_tenant_accepts_any_same_cloud_tenant( + self, monkeypatch + ): + private_key, public_key = generate_rsa_keypair() + caller_tenant = str(uuid.uuid4()) + config = AgentAuthConfiguration( + client_id="client-1", tenant_id="common", validate_issuer=True + ) + validator = JwtTokenValidator(config) + _patch_signing_key(monkeypatch, validator, public_key) + + token = make_signed_jwt( + private_key, + { + "aud": "client-1", + "iss": f"https://login.microsoftonline.com/{caller_tenant}/v2.0", + "tid": caller_tenant, + }, + ) + + identity = await validator.validate_token(token) + assert identity.is_authenticated is True + + @pytest.mark.asyncio + async def test_validate_issuer_enabled_organizations_tenant_accepts_any_same_cloud_tenant( + self, monkeypatch + ): + private_key, public_key = generate_rsa_keypair() + caller_tenant = str(uuid.uuid4()) + config = AgentAuthConfiguration( + client_id="client-1", tenant_id="organizations", validate_issuer=True + ) + validator = JwtTokenValidator(config) + _patch_signing_key(monkeypatch, validator, public_key) + + token = make_signed_jwt( + private_key, + { + "aud": "client-1", + "iss": f"https://sts.windows.net/{caller_tenant}/", + "tid": caller_tenant, + }, + ) + + identity = await validator.validate_token(token) + assert identity.is_authenticated is True + + @pytest.mark.asyncio + async def test_validate_issuer_enabled_gov_authority_routes_and_accepts_gov_issuer( + self, monkeypatch + ): + private_key, public_key = generate_rsa_keypair() + tenant_id = str(uuid.uuid4()) + config = AgentAuthConfiguration( + client_id="client-1", + tenant_id=tenant_id, + authority="https://login.microsoftonline.us", + validate_issuer=True, + ) + validator = JwtTokenValidator(config) + captured_uris = [] + _patch_signing_key(monkeypatch, validator, public_key, captured_uris) + + token = make_signed_jwt( + private_key, + { + "aud": "client-1", + "iss": f"https://login.microsoftonline.us/{tenant_id}/v2.0", + "tid": tenant_id, + }, + ) + + identity = await validator.validate_token(token) + assert identity.is_authenticated is True + assert captured_uris == [ + f"https://login.microsoftonline.us/{tenant_id}/discovery/v2.0/keys" + ] + + @pytest.mark.asyncio + async def test_validate_issuer_enabled_gov_authority_rejects_public_cloud_issuer( + self, monkeypatch + ): + # A v2 issuer from the *other* cloud must not be accepted even if the + # tenant GUID happens to match: cloud affinity is enforced too. + private_key, public_key = generate_rsa_keypair() + tenant_id = str(uuid.uuid4()) + config = AgentAuthConfiguration( + client_id="client-1", + tenant_id="organizations", + authority="https://login.microsoftonline.us", + validate_issuer=True, + ) + validator = JwtTokenValidator(config) + _patch_signing_key(monkeypatch, validator, public_key) + + token = make_signed_jwt( + private_key, + { + "aud": "client-1", + "iss": f"https://login.microsoftonline.com/{tenant_id}/v2.0", + "tid": tenant_id, + }, + ) + + with pytest.raises(ValueError, match="Invalid issuer"): + await validator.validate_token(token) + + +class TestJwtTokenValidatorMultiConnection: + @pytest.mark.asyncio + async def test_public_jwks_routing_preserves_root_connection_endpoint( + self, monkeypatch + ): + # Public-cloud routing intentionally retains the pre-existing root + # connection endpoint even when another connection matches the token's + # audience. This avoids changing network egress for existing agents. + private_key, public_key = generate_rsa_keypair() + tenant_a = str(uuid.uuid4()) + tenant_b = str(uuid.uuid4()) + config_a = AgentAuthConfiguration( + client_id="client-a", + tenant_id=tenant_a, + connection_name="SERVICE_CONNECTION", + ) + config_b = AgentAuthConfiguration( + client_id="client-b", tenant_id=tenant_b, connection_name="MCS" + ) + shared_connections = {"SERVICE_CONNECTION": config_a, "MCS": config_b} + config_a._connections = shared_connections + config_b._connections = shared_connections + + # Validator constructed against connection A's config, but the token + # is issued for connection B's audience/tenant. + validator = JwtTokenValidator(config_a) + captured_uris = [] + _patch_signing_key(monkeypatch, validator, public_key, captured_uris) + + token = make_signed_jwt(private_key, {"aud": "client-b"}) + identity = await validator.validate_token(token) + + assert identity.is_authenticated is True + assert captured_uris == [ + f"https://login.microsoftonline.com/{tenant_a}/discovery/v2.0/keys" + ] + + @pytest.mark.asyncio + async def test_gov_jwks_routing_uses_matching_connection_by_audience( + self, monkeypatch + ): + private_key, public_key = generate_rsa_keypair() + public_tenant = str(uuid.uuid4()) + gov_tenant = str(uuid.uuid4()) + config_a = AgentAuthConfiguration( + client_id="client-a", + tenant_id=public_tenant, + connection_name="SERVICE_CONNECTION", + ) + config_b = AgentAuthConfiguration( + client_id="client-b", + tenant_id=gov_tenant, + authority="https://login.microsoftonline.us", + connection_name="MCS", + ) + shared_connections = {"SERVICE_CONNECTION": config_a, "MCS": config_b} + config_a._connections = shared_connections + config_b._connections = shared_connections + + validator = JwtTokenValidator(config_a) + captured_uris = [] + _patch_signing_key(monkeypatch, validator, public_key, captured_uris) + + token = make_signed_jwt(private_key, {"aud": "client-b"}) + identity = await validator.validate_token(token) + + assert identity.is_authenticated is True + assert captured_uris == [ + f"https://login.microsoftonline.us/{gov_tenant}/discovery/v2.0/keys" + ] + + @pytest.mark.asyncio + async def test_validate_token_multi_connection_issuer_validation_uses_matched_tenant( + self, monkeypatch + ): + private_key, public_key = generate_rsa_keypair() + tenant_a = str(uuid.uuid4()) + tenant_b = str(uuid.uuid4()) + config_a = AgentAuthConfiguration( + client_id="client-a", + tenant_id=tenant_a, + connection_name="SERVICE_CONNECTION", + validate_issuer=True, + ) + config_b = AgentAuthConfiguration( + client_id="client-b", + tenant_id=tenant_b, + connection_name="MCS", + validate_issuer=True, + ) + shared_connections = {"SERVICE_CONNECTION": config_a, "MCS": config_b} + config_a._connections = shared_connections + config_b._connections = shared_connections + + validator = JwtTokenValidator(config_a) + _patch_signing_key(monkeypatch, validator, public_key) + + # Issuer/tid belong to tenant B, matching audience client-b: must be + # validated against connection B's tenant, not A's. + token = make_signed_jwt( + private_key, + { + "aud": "client-b", + "iss": f"https://login.microsoftonline.com/{tenant_b}/v2.0", + "tid": tenant_b, + }, + ) + + identity = await validator.validate_token(token) + assert identity.is_authenticated is True + + +class TestJwtTokenValidatorEffectiveTenant: + """Covers the authority-embedded tenant segment (e.g. + ``https://login.microsoftonline.com/common`` or + ``.../{tenant-guid}``) taking precedence over a separately configured + TENANT_ID for JWKS routing, multi-tenant detection, and default issuers. + """ + + @pytest.mark.asyncio + async def test_public_jwks_routing_ignores_authority_embedded_common_tenant( + self, monkeypatch + ): + # Issuer policy uses AUTHORITY's effective tenant, but public JWKS + # routing preserves the legacy root TENANT_ID endpoint. + private_key, public_key = generate_rsa_keypair() + caller_tenant = str(uuid.uuid4()) + config = AgentAuthConfiguration( + client_id="client-1", + tenant_id="concrete-tenant-id", + authority="https://login.microsoftonline.com/common", + validate_issuer=True, + ) + validator = JwtTokenValidator(config) + captured_uris = [] + _patch_signing_key(monkeypatch, validator, public_key, captured_uris) + + token = make_signed_jwt( + private_key, + { + "aud": "client-1", + "iss": f"https://login.microsoftonline.com/{caller_tenant}/v2.0", + "tid": caller_tenant, + }, + ) + + identity = await validator.validate_token(token) + assert identity.is_authenticated is True + assert captured_uris == [ + "https://login.microsoftonline.com/concrete-tenant-id/discovery/v2.0/keys" + ] + + @pytest.mark.asyncio + async def test_public_jwks_routing_ignores_authority_embedded_concrete_tenant( + self, monkeypatch + ): + # Issuer policy uses AUTHORITY's concrete tenant, but public JWKS + # routing preserves the legacy root TENANT_ID endpoint. + private_key, public_key = generate_rsa_keypair() + concrete_tenant = str(uuid.uuid4()) + config = AgentAuthConfiguration( + client_id="client-1", + tenant_id="common", + authority=f"https://login.microsoftonline.com/{concrete_tenant}", + validate_issuer=True, + ) + validator = JwtTokenValidator(config) + captured_uris = [] + _patch_signing_key(monkeypatch, validator, public_key, captured_uris) + + token = make_signed_jwt( + private_key, + { + "aud": "client-1", + "iss": f"https://login.microsoftonline.com/{concrete_tenant}/v2.0", + "tid": concrete_tenant, + }, + ) + + identity = await validator.validate_token(token) + assert identity.is_authenticated is True + assert captured_uris == [ + "https://login.microsoftonline.com/common/discovery/v2.0/keys" + ] + + @pytest.mark.asyncio + async def test_authority_embedded_concrete_tenant_rejects_other_tenant_issuer( + self, monkeypatch + ): + # Because AUTHORITY's embedded concrete tenant takes precedence over + # the "common" TENANT_ID, this connection is NOT treated as + # multi-tenant: an issuer for a different tenant must be rejected. + private_key, public_key = generate_rsa_keypair() + concrete_tenant = str(uuid.uuid4()) + other_tenant = str(uuid.uuid4()) + config = AgentAuthConfiguration( + client_id="client-1", + tenant_id="common", + authority=f"https://login.microsoftonline.com/{concrete_tenant}", + validate_issuer=True, + ) + validator = JwtTokenValidator(config) + _patch_signing_key(monkeypatch, validator, public_key) + + token = make_signed_jwt( + private_key, + { + "aud": "client-1", + "iss": f"https://login.microsoftonline.com/{other_tenant}/v2.0", + "tid": other_tenant, + }, + ) + + with pytest.raises(ValueError, match="Invalid issuer"): + await validator.validate_token(token) diff --git a/tests/hosting_core/test_auth_configuration.py b/tests/hosting_core/test_auth_configuration.py index 8abf6b98..2ef8a4c7 100644 --- a/tests/hosting_core/test_auth_configuration.py +++ b/tests/hosting_core/test_auth_configuration.py @@ -31,8 +31,8 @@ def test_auth_configuration_basic(self): assert auth_config.SCOPES == ["test-scope-1", "test-scope-2"] assert auth_config.ISSUERS == [ "https://api.botframework.com", - f"https://sts.windows.net/test-tenant-id/", - f"https://login.microsoftonline.com/test-tenant-id/v2.0", + "https://sts.windows.net/test-tenant-id/", + "https://login.microsoftonline.com/test-tenant-id/v2.0", ] def test_load_configuration_from_env(self): @@ -63,6 +63,32 @@ def test_load_configuration_from_env(self): f"https://login.microsoftonline.com/test-tenant-id-{name}/v2.0", ] + def test_issuer_list_from_env(self): + mock_config = load_configuration_from_env( + { + "CONNECTIONS__SERVICE_CONNECTION__SETTINGS__ISSUERS__0": "https://issuer-one.example/", + "CONNECTIONS__SERVICE_CONNECTION__SETTINGS__ISSUERS__1": "https://issuer-two.example/", + "CONNECTIONS__SERVICE_CONNECTION__SETTINGS__VALIDATE_ISSUER": "true", + } + ) + + auth_config = AgentAuthConfiguration( + **mock_config["CONNECTIONS"]["SERVICE_CONNECTION"]["SETTINGS"] + ) + + assert auth_config.ISSUERS == [ + "https://issuer-one.example/", + "https://issuer-two.example/", + ] + assert auth_config.VALIDATE_ISSUER is True + + def test_scalar_issuer_string_is_single_entry(self): + auth_config = AgentAuthConfiguration( + ISSUERS="https://issuer.example/", VALIDATE_ISSUER="true" + ) + + assert auth_config.ISSUERS == ["https://issuer.example/"] + def test_empty_settings(self): auth_config = AgentAuthConfiguration() assert auth_config.AUTH_TYPE == AuthTypes.client_secret @@ -183,3 +209,141 @@ def test_anonymous_allowed_kwarg_used_when_param_unset(self): # When the constructor arg is not provided, the kwarg is honored. auth_config = AgentAuthConfiguration(ANONYMOUS_ALLOWED="true") assert auth_config.ANONYMOUS_ALLOWED is True + + def test_validate_issuer_default_false(self): + assert AgentAuthConfiguration().VALIDATE_ISSUER is False + + def test_validate_issuer_true_bool_param(self): + auth_config = AgentAuthConfiguration(validate_issuer=True) + assert auth_config.VALIDATE_ISSUER is True + + def test_validate_issuer_false_string_kwarg_is_false(self): + # Same fail-safe coercion as ANONYMOUS_ALLOWED: bool("false") would be + # True and silently enable issuer validation when configured off. + auth_config = AgentAuthConfiguration(VALIDATE_ISSUER="false") + assert auth_config.VALIDATE_ISSUER is False + + def test_validate_issuer_true_string_kwarg_is_true(self): + auth_config = AgentAuthConfiguration(VALIDATE_ISSUER="true") + assert auth_config.VALIDATE_ISSUER is True + + def test_validate_issuer_explicit_false_overrides_kwarg(self): + auth_config = AgentAuthConfiguration( + validate_issuer=False, VALIDATE_ISSUER="true" + ) + assert auth_config.VALIDATE_ISSUER is False + + def test_issuers_default_when_not_configured(self): + auth_config = AgentAuthConfiguration(tenant_id="tenant-1") + assert auth_config.ISSUERS == [ + "https://api.botframework.com", + "https://sts.windows.net/tenant-1/", + "https://login.microsoftonline.com/tenant-1/v2.0", + ] + + def test_issuers_explicit_list_overrides_default(self): + auth_config = AgentAuthConfiguration( + tenant_id="tenant-1", + issuers=["https://custom-issuer.example.com/"], + ) + assert auth_config.ISSUERS == ["https://custom-issuer.example.com/"] + + def test_issuers_kwarg_alias(self): + auth_config = AgentAuthConfiguration( + tenant_id="tenant-1", ISSUERS=["https://custom-issuer.example.com/"] + ) + assert auth_config.ISSUERS == ["https://custom-issuer.example.com/"] + + def test_issuers_param_preferred_over_kwarg(self): + auth_config = AgentAuthConfiguration( + tenant_id="tenant-1", + issuers=["https://primary.example.com/"], + ISSUERS=["https://secondary.example.com/"], + ) + assert auth_config.ISSUERS == ["https://primary.example.com/"] + + def test_issuers_default_uses_gov_cloud_when_authority_is_gov(self): + auth_config = AgentAuthConfiguration( + tenant_id="tenant-1", authority="https://login.microsoftonline.us" + ) + assert auth_config.ISSUERS == [ + "https://api.botframework.us", + "https://sts.windows.net/tenant-1/", + "https://login.microsoftonline.us/tenant-1/v2.0", + ] + + def test_issuers_default_uses_authority_embedded_common_tenant(self): + # The authority-embedded tenant segment takes precedence over a + # separately configured (concrete) TENANT_ID, matching the JS + # reference's getEffectiveTenant/resolveAuthority precedence. + auth_config = AgentAuthConfiguration( + tenant_id="concrete-tenant-id", + authority="https://login.microsoftonline.com/common", + ) + assert auth_config.ISSUERS == [ + "https://api.botframework.com", + "https://sts.windows.net/common/", + "https://login.microsoftonline.com/common/v2.0", + ] + + def test_issuers_default_uses_authority_embedded_concrete_tenant(self): + # A concrete tenant embedded in AUTHORITY must be used instead of a + # "common"/absent TENANT_ID, avoiding an incorrect "/common" or + # "/None" default issuer. + auth_config = AgentAuthConfiguration( + tenant_id="common", + authority="https://login.microsoftonline.com/concrete-tenant-id", + ) + assert auth_config.ISSUERS == [ + "https://api.botframework.com", + "https://sts.windows.net/concrete-tenant-id/", + "https://login.microsoftonline.com/concrete-tenant-id/v2.0", + ] + + def test_issuers_default_falls_back_to_common_without_tenant_id_or_authority_path( + self, + ): + # Neither TENANT_ID nor an authority-embedded tenant segment is + # configured: the default must fall back to "common" rather than + # embedding a literal "None" in the issuer URLs. + auth_config = AgentAuthConfiguration( + authority="https://login.microsoftonline.com" + ) + assert auth_config.ISSUERS == [ + "https://api.botframework.com", + "https://sts.windows.net/common/", + "https://login.microsoftonline.com/common/v2.0", + ] + + def test_issuers_and_validate_issuer_not_in_provider_settings(self): + # Recognized keys (bound into first-class fields) must never be + # duplicated into the provider-specific settings bag. + auth_config = AgentAuthConfiguration( + ISSUERS=["https://custom-issuer.example.com/"], + VALIDATE_ISSUER="true", + SOME_PROVIDER_KEY="keep-me", + ) + assert "ISSUERS" not in auth_config.provider_settings + assert "VALIDATE_ISSUER" not in auth_config.provider_settings + assert auth_config.provider_settings == {"SOME_PROVIDER_KEY": "keep-me"} + + def test_jwt_patch_is_valid_aud_rejects_non_string_audience(self): + # A non-string `aud` (e.g. the JWT-spec-permitted array form, or a + # malformed numeric/object claim) must be reported as invalid rather + # than raising AttributeError from `.lower()` on a non-string value. + auth_config = AgentAuthConfiguration(client_id="client-1") + assert auth_config._jwt_patch_is_valid_aud(["client-1"]) is False + assert auth_config._jwt_patch_is_valid_aud(12345) is False + assert auth_config._jwt_patch_is_valid_aud({"aud": "client-1"}) is False + # Sanity check: normal string behavior is unaffected. + assert auth_config._jwt_patch_is_valid_aud("client-1") is True + + def test_jwt_patch_find_connection_treats_non_string_audience_as_no_match(self): + # Routing must not raise on a non-string `aud`; it should behave as + # "no matching connection" so callers fall back to default routing. + auth_config = AgentAuthConfiguration(client_id="client-1") + assert auth_config._jwt_patch_find_connection(["client-1"]) is None + assert auth_config._jwt_patch_find_connection(12345) is None + assert auth_config._jwt_patch_find_connection({"aud": "client-1"}) is None + # Sanity check: normal string behavior is unaffected. + assert auth_config._jwt_patch_find_connection("client-1") is auth_config From e7aabe58093d5d68f9f7e1bfffc7451e1c50c918 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Matthew=20Meyer=20=F0=9F=90=89=E2=9A=94=EF=B8=8F?= Date: Mon, 3 Aug 2026 12:21:47 -0700 Subject: [PATCH 2/2] Refactor token issuer handling and enhance JWT validation logic --- .../core/authorization/_entra_issuers.py | 69 +++++++++++++------ .../authorization/authentication_constants.py | 14 +++- .../authorization/jwt/jwt_token_validator.py | 26 +++++-- .../authorization/test_jwt_token_validator.py | 28 +++++++- 4 files changed, 103 insertions(+), 34 deletions(-) diff --git a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/_entra_issuers.py b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/_entra_issuers.py index 63bd9985..f6350f48 100644 --- a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/_entra_issuers.py +++ b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/_entra_issuers.py @@ -15,6 +15,8 @@ from typing import Any, NamedTuple from urllib.parse import urlparse +from .authentication_constants import AuthenticationConstants + # Well-known Microsoft first-party token issuer tenant IDs that are always # trusted, mirroring the default ``ValidIssuers`` set used by the .NET SDK. # These identify Microsoft infrastructure tenants used by Azure Bot Service, @@ -27,21 +29,32 @@ ) WELL_KNOWN_GOV_TENANT_ID = "cab8a31a-1906-4287-a0d8-4eef66b95f6e" -BOTFRAMEWORK_PUBLIC_ISSUER = "https://api.botframework.com" -BOTFRAMEWORK_GOV_ISSUER = "https://api.botframework.us" +BOTFRAMEWORK_PUBLIC_ISSUER = AuthenticationConstants.AGENTS_SDK_TOKEN_ISSUER +BOTFRAMEWORK_GOV_ISSUER = AuthenticationConstants.GOV_AGENTS_SDK_TOKEN_ISSUER BOTFRAMEWORK_JWKS_URIS = { - BOTFRAMEWORK_PUBLIC_ISSUER: "https://login.botframework.com/v1/.well-known/keys", - BOTFRAMEWORK_GOV_ISSUER: "https://login.botframework.azure.us/v1/.well-known/keys", + BOTFRAMEWORK_PUBLIC_ISSUER: AuthenticationConstants.PUBLIC_ABS_JWKS_URL, + BOTFRAMEWORK_GOV_ISSUER: AuthenticationConstants.GOV_ABS_JWKS_URL, } + +def _issuer_pattern(template: str) -> re.Pattern[str]: + prefix, suffix = template.split("{0}") + return re.compile(rf"^(?i:{re.escape(prefix)})([^/]+){re.escape(suffix)}$") + + _GOV_AUTHORITY_RE = re.compile(r"login\.microsoftonline\.us", re.IGNORECASE) _ENTRA_TENANT_GUID_RE = re.compile( r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$", re.IGNORECASE ) -_V1_ISSUER_RE = re.compile(r"^https://sts\.windows\.net/([^/]+)/$", re.IGNORECASE) -_V2_ISSUER_RE = re.compile( - r"^(?i:https://login\.microsoftonline\.(com|us)/)([^/]+)/v2\.0$" +_V1_ISSUER_RE = _issuer_pattern( + AuthenticationConstants.VALID_TOKEN_ISSUER_URL_TEMPLATE_V1 +) +_PUBLIC_V2_ISSUER_RE = _issuer_pattern( + AuthenticationConstants.VALID_TOKEN_ISSUER_URL_TEMPLATE_V2 +) +_GOV_V2_ISSUER_RE = _issuer_pattern( + AuthenticationConstants.VALID_GOV_TOKEN_ISSUER_URL_TEMPLATE_V2 ) @@ -111,11 +124,16 @@ def entra_issuer_info(iss: Any) -> EntraIssuerInfo | None: return EntraIssuerInfo(tenant.lower(), None) return None - v2_match = _V2_ISSUER_RE.match(iss) - if v2_match: - cloud, tenant = v2_match.group(1), v2_match.group(2) + for pattern, gov in ( + (_PUBLIC_V2_ISSUER_RE, False), + (_GOV_V2_ISSUER_RE, True), + ): + v2_match = pattern.match(iss) + if not v2_match: + continue + tenant = v2_match.group(1) if _ENTRA_TENANT_GUID_RE.match(tenant): - return EntraIssuerInfo(tenant.lower(), cloud.lower() == "us") + return EntraIssuerInfo(tenant.lower(), gov) return None @@ -134,15 +152,14 @@ def default_connection_issuers( tenant = effective_tenant(tenant_id, authority) or "common" gov = is_gov_authority(authority) bf_issuer = BOTFRAMEWORK_GOV_ISSUER if gov else BOTFRAMEWORK_PUBLIC_ISSUER - login_host = ( - "https://login.microsoftonline.us" - if gov - else "https://login.microsoftonline.com" - ) return [ bf_issuer, - f"https://sts.windows.net/{tenant}/", - f"{login_host}/{tenant}/v2.0", + AuthenticationConstants.VALID_TOKEN_ISSUER_URL_TEMPLATE_V1.format(tenant), + ( + AuthenticationConstants.VALID_GOV_TOKEN_ISSUER_URL_TEMPLATE_V2 + if gov + else AuthenticationConstants.VALID_TOKEN_ISSUER_URL_TEMPLATE_V2 + ).format(tenant), ] @@ -153,13 +170,21 @@ def well_known_first_party_issuers(authority: str | None) -> list[str]: if is_gov_authority(authority): return [ BOTFRAMEWORK_GOV_ISSUER, - f"https://sts.windows.net/{WELL_KNOWN_GOV_TENANT_ID}/", - f"https://login.microsoftonline.us/{WELL_KNOWN_GOV_TENANT_ID}/v2.0", + AuthenticationConstants.VALID_TOKEN_ISSUER_URL_TEMPLATE_V1.format( + WELL_KNOWN_GOV_TENANT_ID + ), + AuthenticationConstants.VALID_GOV_TOKEN_ISSUER_URL_TEMPLATE_V2.format( + WELL_KNOWN_GOV_TENANT_ID + ), ] issuers = [BOTFRAMEWORK_PUBLIC_ISSUER] for tenant in WELL_KNOWN_PUBLIC_TENANT_IDS: - issuers.append(f"https://sts.windows.net/{tenant}/") - issuers.append(f"https://login.microsoftonline.com/{tenant}/v2.0") + issuers.append( + AuthenticationConstants.VALID_TOKEN_ISSUER_URL_TEMPLATE_V1.format(tenant) + ) + issuers.append( + AuthenticationConstants.VALID_TOKEN_ISSUER_URL_TEMPLATE_V2.format(tenant) + ) return issuers diff --git a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/authentication_constants.py b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/authentication_constants.py index 296a8df2..d76068d9 100644 --- a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/authentication_constants.py +++ b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/authentication_constants.py @@ -10,6 +10,7 @@ class AuthenticationConstants(ABC): # Token issuer for ABS tokens. AGENTS_SDK_TOKEN_ISSUER = "https://api.botframework.com" + GOV_AGENTS_SDK_TOKEN_ISSUER = "https://api.botframework.us" # Default OAuth Url used to get a token from IUserTokenClient. AGENTS_SDK_OAUTH_URL = "https://api.botframework.com" @@ -25,15 +26,23 @@ class AuthenticationConstants(ABC): ) # Enterprise Channel OpenId Metadata URL format. - ENTERPRISE_CHANNEL_OPENID_METADATA_URL_FORMAT = "https://{0}.enterprisechannel.botframework.com/v1/.well-known/openidconfiguration" + ENTERPRISE_CHANNEL_OPENID_METADATA_URL_FORMAT = ( + "https://{0}.enterprisechannel.botframework.com/v1/.well-known/" + "openidconfiguration" + ) # Gov ABS OpenId Metadata URL. GOV_ABS_OPENID_METADATA_URL = ( "https://login.botframework.azure.us/v1/.well-known/openidconfiguration" ) + PUBLIC_ABS_JWKS_URL = "https://login.botframework.com/v1/.well-known/keys" + GOV_ABS_JWKS_URL = "https://login.botframework.azure.us/v1/.well-known/keys" # Gov OpenId Metadata URL. - GOV_OPENID_METADATA_URL = "https://login.microsoftonline.us/cab8a31a-1906-4287-a0d8-4eef66b95f6e/v2.0/.well-known/openid-configuration" + GOV_OPENID_METADATA_URL = ( + "https://login.microsoftonline.us/cab8a31a-1906-4287-a0d8-4eef66b95f6e/" + "v2.0/.well-known/openid-configuration" + ) # The V1 Azure AD token issuer URL template that will contain the tenant id where # the token was issued from. @@ -42,6 +51,7 @@ class AuthenticationConstants(ABC): # The V2 Azure AD token issuer URL template that will contain the tenant id where # the token was issued from. VALID_TOKEN_ISSUER_URL_TEMPLATE_V2 = "https://login.microsoftonline.com/{0}/v2.0" + VALID_GOV_TOKEN_ISSUER_URL_TEMPLATE_V2 = "https://login.microsoftonline.us/{0}/v2.0" # "azp" Claim. # Authorized party - the party to which the ID Token was issued. diff --git a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/jwt/jwt_token_validator.py b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/jwt/jwt_token_validator.py index 90cd7e31..1a9a14d6 100644 --- a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/jwt/jwt_token_validator.py +++ b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/jwt/jwt_token_validator.py @@ -10,6 +10,7 @@ from jwt import PyJWKClient, PyJWK, decode, get_unverified_header from ..agent_auth_configuration import AgentAuthConfiguration +from ..authentication_constants import AuthenticationConstants from ..claims_identity import ClaimsIdentity from .._entra_issuers import ( BOTFRAMEWORK_JWKS_URIS, @@ -61,7 +62,9 @@ async def get_signing_key(self, jwks_uri: str, header: dict[str, Any]) -> PyJWK: def _helper(): with jwk_cache_entry.lock: - return jwk_cache_entry.jwk_client.get_signing_key(header["kid"]) + return jwk_cache_entry.jwk_client.get_signing_key( + header[AuthenticationConstants.KEY_ID_HEADER] + ) key = await asyncio.to_thread(_helper) return key @@ -104,11 +107,15 @@ async def validate_token(self, token: str) -> ClaimsIdentity: # This is routing only -- final acceptance is checked against the # signature-verified claims below. routing_config = ( - self.configuration._jwt_patch_find_connection(unverified_payload.get("aud")) + self.configuration._jwt_patch_find_connection( + unverified_payload.get(AuthenticationConstants.AUDIENCE_CLAIM) + ) or self.configuration ) jwks_uri = _build_jwks_uri( - unverified_payload.get("iss"), self.configuration, routing_config + unverified_payload.get(AuthenticationConstants.ISSUER_CLAIM), + self.configuration, + routing_config, ) key = await self._jwk_client_manager.get_signing_key(jwks_uri, header) @@ -120,7 +127,7 @@ async def validate_token(self, token: str) -> ClaimsIdentity: options={"verify_aud": False}, ) - aud = decoded_token.get("aud", "") + aud = decoded_token.get(AuthenticationConstants.AUDIENCE_CLAIM, "") if not self.configuration._jwt_patch_is_valid_aud(aud): logger.warning("JWT audience not accepted.") raise ValueError("Invalid audience.") @@ -138,8 +145,13 @@ async def validate_token(self, token: str) -> ClaimsIdentity: # are unaffected, and a missing ``tid`` claim skips the check rather # than failing closed. if matched_config.VALIDATE_ISSUER: - _validate_issuer(decoded_token.get("iss"), matched_config) - _validate_tenant_binding(decoded_token.get("iss"), decoded_token.get("tid")) + _validate_issuer( + decoded_token.get(AuthenticationConstants.ISSUER_CLAIM), matched_config + ) + _validate_tenant_binding( + decoded_token.get(AuthenticationConstants.ISSUER_CLAIM), + decoded_token.get(AuthenticationConstants.TENANT_ID_CLAIM), + ) logger.debug("JWT token validated successfully.") return ClaimsIdentity(decoded_token, True, security_token=token) @@ -184,7 +196,7 @@ def _build_jwks_uri( return ( "https://login.microsoftonline.com/" - f"{root_config.TENANT_ID}/discovery/v2.0/keys" + f"{root_config.TENANT_ID or 'common'}/discovery/v2.0/keys" ) diff --git a/tests/hosting_core/authorization/test_jwt_token_validator.py b/tests/hosting_core/authorization/test_jwt_token_validator.py index 39be5cbd..dea055bb 100644 --- a/tests/hosting_core/authorization/test_jwt_token_validator.py +++ b/tests/hosting_core/authorization/test_jwt_token_validator.py @@ -271,9 +271,7 @@ async def test_missing_tid_skips_binding_even_when_issuer_validation_disabled( assert identity.is_authenticated is True @pytest.mark.asyncio - async def test_noncanonical_entra_issuer_variants_skip_binding( - self, monkeypatch - ): + async def test_noncanonical_entra_issuer_variants_skip_binding(self, monkeypatch): private_key, public_key = generate_rsa_keypair() issuer_tenant = str(uuid.uuid4()) mismatched_tid = str(uuid.uuid4()) @@ -767,6 +765,30 @@ class TestJwtTokenValidatorEffectiveTenant: TENANT_ID for JWKS routing, multi-tenant detection, and default issuers. """ + @pytest.mark.asyncio + async def test_public_jwks_routing_defaults_to_common_without_tenant( + self, monkeypatch + ): + private_key, public_key = generate_rsa_keypair() + config = AgentAuthConfiguration(client_id="client-1") + validator = JwtTokenValidator(config) + captured_uris = [] + _patch_signing_key(monkeypatch, validator, public_key, captured_uris) + + token = make_signed_jwt( + private_key, + { + "aud": "client-1", + "iss": "https://custom.example.com", + }, + ) + + identity = await validator.validate_token(token) + assert identity.is_authenticated is True + assert captured_uris == [ + "https://login.microsoftonline.com/common/discovery/v2.0/keys" + ] + @pytest.mark.asyncio async def test_public_jwks_routing_ignores_authority_embedded_common_tenant( self, monkeypatch