JWT authorization configuration refactor

changelog.md

@@ -1,3 +1,17 @@
+# Microsoft 365 Agents SDK for Python - Release Notes v1.10.0 (Unreleased)
+
+**Release Date:** Unreleased
+**Previous Version:** 1.9.0 (Released 2026-10-19)
+
+## Major Features & Enhancements
+
+- **Shared Configuration API**: Added a dictionary-backed `Configuration` API with nested path and alias lookup, typed getters, wildcard collection, configurable settings binding, and mapping/default resolution with indexed-list normalization. Retained the existing `load_configuration_from_env()` return contract (#636).
+- **JWT Authorization Configuration**: Added `ConnectionSettingsBase`, `JwtConnectionSettings`, and `TokenValidationConfiguration` to bind inbound JWT policies from application configuration, including explicit audiences, tenant and anonymous-access overrides, and per-connection issuer settings. Retained legacy `AgentAuthConfiguration` support and added application-scoped validator caching without changing `Connections` or `ConnectionManager` (#638).
+
+- **Explicit JWT Audience Lists**: Scalar and delimited audience strings now bind as explicit allow-lists instead of falling back to all configured connection client IDs. Missing or empty audiences continue to use connection defaults (#638).
+
+---
+
 # Microsoft 365 Agents SDK for Python - Release Notes v1.9.0
 
 **Release Date:** 2026-10-19

libraries/microsoft-agents-activity/microsoft_agents/activity/config/_load_configuration.py

@@ -4,6 +4,7 @@
 from typing import Any, Mapping
 
 from ._logging import _configure_logging
+from ._resolve import _resolve
 
 
 def load_configuration_from_env(env_vars: Mapping[str, Any]) -> dict:
@@ -30,8 +31,15 @@ def load_configuration_from_env(env_vars: Mapping[str, Any]) -> dict:
 
     _configure_logging(result.get("LOGGING", {}))
 
+    # temporary fix before next PR
+    token_validation_options = _resolve(
+        result.get("TOKENVALIDATION", {}),
+    )
+
     return {
-        "AGENTAPPLICATION": result.get("AGENTAPPLICATION", {}),
-        "CONNECTIONS": result.get("CONNECTIONS", {}),
-        "CONNECTIONSMAP": result.get("CONNECTIONSMAP", []),
+        "AGENTAPPLICATION": result.pop("AGENTAPPLICATION", {}),
+        "CONNECTIONS": result.pop("CONNECTIONS", {}),
+        "CONNECTIONSMAP": result.pop("CONNECTIONSMAP", []),
+        "LOGGING": result.pop("LOGGING", {}),
+        "TOKENVALIDATION": token_validation_options,
     }

libraries/microsoft-agents-activity/microsoft_agents/activity/config/configuration.py

@@ -352,7 +352,7 @@ def get_str_list(
             name = self._resolve_path(_PATH_SEPARATOR.join(path) or keys[0])
             before_split = _handle_str(val, None, name)
             if before_split is None:
-                return default
+                return default if default is not None else []
             if "," in before_split:
                 return [s.strip() for s in before_split.split(",")]
             return [s.strip() for s in before_split.strip().split()]

libraries/microsoft-agents-hosting-aiohttp/microsoft_agents/hosting/aiohttp/jwt_authorization_middleware.py

@@ -35,10 +35,6 @@ async def _jwt_authorization_middleware(request: Request, handler):
     """
     JWT Authorization Middleware for aiohttp endpoints.
     """
-    auth_config = cast(
-        AgentAuthConfiguration | None, request.app.get("agent_configuration", None)
-    )
-
     auth_header = request.headers.get("Authorization")
     if auth_header is not None:
         # aiohttp-specific tolerance: trailing whitespace (spaces/tabs) after
@@ -52,7 +48,11 @@ async def _jwt_authorization_middleware(request: Request, handler):
             return json_response(_GENERIC_AUTH_ERROR, status=401)
         auth_header = f"Bearer {token}"
 
-    res = await _authorize_request(auth_header, auth_config)
+    res = await _authorize_request(
+        auth_header,
+        request.app.get("agent_configuration"),
+        cache_key=request.app,
+    )
 
     if isinstance(res, HttpResponse):
         return json_response(res.body, status=res.status_code)

libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/connection_settings_base.py

@@ -4,7 +4,16 @@
 from __future__ import annotations
 
 from collections.abc import Mapping
+from dataclasses import dataclass, field, fields
 from typing import Any, TYPE_CHECKING
+from typing_extensions import Self
+
+from microsoft_agents.activity.config import (
+    Configuration,
+    _Configurable,
+    _resolve,
+    xpath,
+)
 
 if TYPE_CHECKING:
     from .agent_auth_configuration import AgentAuthConfiguration
@@ -109,3 +118,87 @@ def from_configuration(
     ) -> "_ConnectionSettingsBase":
         """Build base settings from an :class:`AgentAuthConfiguration`."""
         return cls(**cls.base_kwargs_from_configuration(configuration))
+
+
+_KNOWN_KWARGS_CACHE: dict[type[ConnectionSettingsBase], list[str]] = {}
+
+
+@dataclass(frozen=True)
+class ConnectionSettingsBase(_Configurable):
+    """Provider-independent settings bound from Configuration.
+
+    Frozen fields cannot be reassigned, but scopes and extras remain mutable.
+    Treat those containers as read-only after handing settings to a provider.
+    """
+
+    client_id: str | None = None
+    authority: str | None = None
+    tenant_id: str | None = None
+    alternate_blueprint_connection_name: str | None = None
+    scopes: list[str] = field(default_factory=list)
+    _extras: Configuration = field(default_factory=Configuration)
+
+    @property
+    def extras(self) -> Configuration:
+        """Return unrecognized provider settings, not an immutable copy."""
+        return self._extras
+
+    @classmethod
+    def _from_agent_auth_config(cls, agent_auth_config: AgentAuthConfiguration) -> Self:
+        attrs = dict(vars(agent_auth_config))
+        attrs.update(attrs.pop("provider_settings", {}))
+        attrs["ISSUERS"] = agent_auth_config.ISSUERS
+        if isinstance(attrs.get("SCOPES"), Mapping):
+            # TODO: Temporary compatibility for legacy indexed scopes; the next
+            # PR's configuration-loader migration will normalize these mappings.
+            attrs["SCOPES"] = _resolve({"SCOPES": attrs["SCOPES"]})["SCOPES"]
+        return cls.from_mapping(attrs)
+
+    @classmethod
+    def _get_init_kwargs(cls, config: Configuration) -> dict:
+        return {
+            "client_id": config.get_strn(cls._xpath("client_id")),
+            "authority": config.get_strn(
+                cls._xpath("authority"), cls._xpath("authorityendpoint")
+            ),
+            "tenant_id": config.get_strn(cls._xpath("tenant_id")),
+            "scopes": config.get_scopes(cls._xpath("scopes")),
+            "alternate_blueprint_connection_name": config.get_strn(
+                "alt_blueprint_name",
+                "ALT_BLUEPRINT_NAME",
+                "alt_blueprint_id",
+                "ALT_BLUEPRINT_ID",
+                *cls._xpath("alternate_blueprint_connection_name"),
+            ),
+        }
+
+    @classmethod
+    def _known_kwargs(cls) -> list[str]:
+        if cls not in _KNOWN_KWARGS_CACHE:
+            _KNOWN_KWARGS_CACHE[cls] = [
+                alias
+                for f in fields(cls)
+                if f.init and f.name != "_extras"
+                for alias in xpath(f.name)
+            ]
+        return _KNOWN_KWARGS_CACHE[cls]
+
+    @classmethod
+    def _get_extras(cls, config: Configuration) -> Configuration:
+        known_kwargs = cls._known_kwargs()
+        base = {k: v for k, v in config.items() if k not in known_kwargs}
+        return Configuration.from_mapping(
+            base, _base_path=config._base_path, **base.pop("extras", {})
+        )
+
+    @classmethod
+    def _from_configuration(cls, config: Configuration) -> Self:
+        try:
+            settings = config.get_section(xpath("settings"))
+        except KeyError:
+            settings = config
+        return cls(**cls._get_init_kwargs(settings), _extras=cls._get_extras(settings))
+
+    @staticmethod
+    def _xpath(name: str) -> list[str]:
+        return xpath(name, lower_snake=True)

libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/jwt/__init__.py

@@ -3,8 +3,12 @@
 
 from ._authorize_request import _authorize_request
 from .jwt_token_validator import JwtTokenValidator
+from .jwt_connection_settings import JwtConnectionSettings
+from .token_validation_configuration import TokenValidationConfiguration
 
 __all__ = [
     "JwtTokenValidator",
     "_authorize_request",
+    "JwtConnectionSettings",
+    "TokenValidationConfiguration",
 ]

libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/jwt/_authorize_request.py

@@ -4,7 +4,10 @@
 import logging
 
 from jwt import PyJWTError
+from typing import Any, Mapping
+from weakref import WeakKeyDictionary
 
+from microsoft_agents.activity.config import Configuration
 from microsoft_agents.hosting.core.http import HttpResponse
 
 from ..agent_auth_configuration import AgentAuthConfiguration
@@ -13,9 +16,33 @@
 
 logger = logging.getLogger(__name__)
 
+_TOKEN_VALIDATION_CACHE: WeakKeyDictionary[Any, JwtTokenValidator] = WeakKeyDictionary()
+
+
+def _resolve_token_validator(
+    auth_config: Configuration | AgentAuthConfiguration | None,
+    cache_key: Any = None,
+) -> JwtTokenValidator | None:
+
+    if cache_key is not None and cache_key in _TOKEN_VALIDATION_CACHE:
+        return _TOKEN_VALIDATION_CACHE[cache_key]
+
+    if auth_config is not None:
+        config: AgentAuthConfiguration | Configuration = auth_config
+        if not isinstance(auth_config, AgentAuthConfiguration):
+            config = Configuration(auth_config)
+        validator = JwtTokenValidator(config)
+        if cache_key is not None:
+            _TOKEN_VALIDATION_CACHE[cache_key] = validator
+        return validator
+    return None
+
 
 async def _authorize_request(
-    authorization_header: str | None, auth_config: AgentAuthConfiguration | None
+    authorization_header: str | None,
+    auth_config: Configuration | AgentAuthConfiguration | None,
+    *,
+    cache_key: Any = None
 ) -> ClaimsIdentity | HttpResponse:
     """
     Authorizes a request based on the provided JWT token in the Authorization header.
@@ -25,16 +52,16 @@ async def _authorize_request(
     :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.
     """
+    validator = _resolve_token_validator(auth_config, cache_key)
 
-    if auth_config is None:
+    if not validator:
         return HttpResponse(
             body={"error": "Agent Authentication configuration not found"},
             status_code=500,
         )
-    validator = JwtTokenValidator(auth_config)
 
     if not authorization_header:
-        if auth_config.ANONYMOUS_ALLOWED:
+        if validator.config.anonymous_allowed:
             claims_identity = validator.get_anonymous_claims()
             return claims_identity
         return HttpResponse(

libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/authorization/jwt/jwt_connection_settings.py

@@ -0,0 +1,40 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+from microsoft_agents.activity.config import Configuration, xpath
+
+from .._entra_issuers import default_connection_issuers
+from ..connection_settings_base import ConnectionSettingsBase
+
+
+@dataclass(frozen=True)
+class JwtConnectionSettings(ConnectionSettingsBase):
+    """Provider-independent connection settings for inbound JWT validation."""
+
+    validate_issuer: bool = False
+    issuers: list[str] = field(default_factory=list)
+    anonymous_allowed: bool = False
+
+    @classmethod
+    def _get_init_kwargs(cls, config: Configuration) -> dict:
+        base = super()._get_init_kwargs(config)
+        issuers = config.get_str_list(xpath("issuers"))
+        issuers = issuers or default_connection_issuers(
+            base.get("tenant_id"), base.get("authority")
+        )
+        base.update(
+            {
+                "validate_issuer": config.get_bool(
+                    xpath("validate_issuer"), default=False
+                ),
+                "issuers": issuers,
+                "anonymous_allowed": config.get_bool(
+                    xpath("anonymous_allowed"), default=False
+                ),
+            }
+        )
+        return base