Support multiple JWT issuers in jwt_authenticate

Accept a tuple of issuer strings for multi-tenant setups (e.g.
Microsoft Entra with multiple tenants). A token matching any
issuer is accepted. First issuer used for OIDC discovery when
jwks_uri is not provided.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: rustyconover

tests/test_oauth.py

@@ -85,12 +85,13 @@ def _mint_jwt(
 def _make_local_auth(
     public_key: dict[str, object],
     *,
-    issuer: str = "https://auth.example.com",
+    issuer: str | tuple[str, ...] = "https://auth.example.com",
     audience: str | tuple[str, ...] = "https://api.example.com/vgi",
     principal_claim: str = "sub",
     domain: str = "jwt",
 ) -> Callable[[falcon.Request], AuthContext]:
     """Create a local JWT authenticate callback (no JWKS endpoint needed)."""
+    issuers = (issuer,) if isinstance(issuer, str) else issuer
     audiences = (audience,) if isinstance(audience, str) else audience
 
     def authenticate(req: falcon.Request) -> AuthContext:
@@ -103,7 +104,7 @@ def authenticate(req: falcon.Request) -> AuthContext:
                 raw_token,
                 public_key,
                 claims_options={
-                    "iss": {"essential": True, "value": issuer},
+                    "iss": {"essential": True, "values": list(issuers)},
                     "aud": {"essential": True, "values": list(audiences)},
                 },
             )
@@ -816,6 +817,37 @@ def test_wrong_issuer_raises_value_error(self) -> None:
         with pytest.raises(ValueError):
             auth_fn(req)
 
+    def test_multiple_issuers_first_matches(self) -> None:
+        """A JWT matching the first of multiple issuers is accepted."""
+        priv, pub = _make_rsa_key()
+        iss1 = "https://auth.example.com"
+        iss2 = "https://auth2.example.com"
+        token = _mint_jwt(priv, iss=iss1)
+        auth_fn = _make_local_auth(pub, issuer=(iss1, iss2))
+        req = falcon.testing.helpers.create_req(headers={"Authorization": f"Bearer {token}"})
+        auth = auth_fn(req)
+        assert auth.authenticated is True
+
+    def test_multiple_issuers_second_matches(self) -> None:
+        """A JWT matching the second of multiple issuers is accepted."""
+        priv, pub = _make_rsa_key()
+        iss1 = "https://auth.example.com"
+        iss2 = "https://auth2.example.com"
+        token = _mint_jwt(priv, iss=iss2)
+        auth_fn = _make_local_auth(pub, issuer=(iss1, iss2))
+        req = falcon.testing.helpers.create_req(headers={"Authorization": f"Bearer {token}"})
+        auth = auth_fn(req)
+        assert auth.authenticated is True
+
+    def test_multiple_issuers_none_match(self) -> None:
+        """A JWT with an unrecognized issuer is rejected."""
+        priv, pub = _make_rsa_key()
+        token = _mint_jwt(priv, iss="https://wrong.example.com")
+        auth_fn = _make_local_auth(pub, issuer=("https://a.example.com", "https://b.example.com"))
+        req = falcon.testing.helpers.create_req(headers={"Authorization": f"Bearer {token}"})
+        with pytest.raises(ValueError):
+            auth_fn(req)
+
     def test_missing_bearer_raises_value_error(self) -> None:
         """Missing Authorization header raises ValueError."""
         _priv, pub = _make_rsa_key()
@@ -909,3 +941,30 @@ def test_empty_audience_tuple_raises(self) -> None:
                 audience=(),
                 jwks_uri="https://auth.example.com/.well-known/jwks.json",
             )
+
+    def test_empty_issuer_tuple_raises(self) -> None:
+        """Passing an empty issuer tuple raises ValueError eagerly."""
+        with pytest.raises(ValueError, match="issuer must not be empty"):
+            jwt_authenticate(
+                issuer=(),
+                audience="https://api.example.com/vgi",
+                jwks_uri="https://auth.example.com/.well-known/jwks.json",
+            )
+
+    def test_jwt_authenticate_factory_multiple_issuers(self) -> None:
+        """jwt_authenticate() accepts a tuple of issuers."""
+        auth_fn = jwt_authenticate(
+            issuer=("https://auth.example.com", "https://auth2.example.com"),
+            audience="https://api.example.com/vgi",
+            jwks_uri="https://auth.example.com/.well-known/jwks.json",
+        )
+        assert callable(auth_fn)
+
+    def test_single_issuer_string_still_works(self) -> None:
+        """Passing issuer as a plain string still works (backwards compat)."""
+        priv, pub = _make_rsa_key()
+        token = _mint_jwt(priv)
+        auth_fn = _make_local_auth(pub, issuer="https://auth.example.com")
+        req = falcon.testing.helpers.create_req(headers={"Authorization": f"Bearer {token}"})
+        auth = auth_fn(req)
+        assert auth.authenticated is True

uv.lock

@@ -33,7 +33,7 @@
 
 def jwt_authenticate(
     *,
-    issuer: str,
+    issuer: str | tuple[str, ...],
     audience: str | tuple[str, ...],
     jwks_uri: str | None = None,
     claims_options: Mapping[str, Any] | None = None,
@@ -47,7 +47,11 @@ def jwt_authenticate(
     Keys are cached in-process with automatic refresh on unknown ``kid``.
 
     Args:
-        issuer: Expected ``iss`` claim in the JWT.
+        issuer: Expected ``iss`` claim(s) in the JWT.  A single string
+            or a tuple of strings.  When multiple issuers are given, a
+            token matching **any** of them is accepted.  When multiple
+            issuers are provided and ``jwks_uri`` is not set, the first
+            issuer is used for OIDC discovery.
         audience: Expected ``aud`` claim(s) in the JWT.  A single string
             or a tuple of strings.  When multiple audiences are given, a
             token matching **any** of them is accepted.
@@ -71,6 +75,12 @@ def jwt_authenticate(
     # Authlib's KeySet type is not exported in public stubs; alias to contain Any.
     type _KeySet = Any
 
+    issuers = (issuer,) if isinstance(issuer, str) else issuer
+    if not issuers:
+        raise ValueError("issuer must not be empty")
+    # Use the first issuer for OIDC discovery when jwks_uri is not provided.
+    discovery_issuer = issuers[0]
+
     resolved_jwks_uri = jwks_uri
     lock = threading.Lock()
     key_set: _KeySet = None
@@ -79,7 +89,7 @@ def _fetch_jwks() -> _KeySet:
         nonlocal resolved_jwks_uri, key_set
         if resolved_jwks_uri is None:
             with httpx.Client() as client:
-                oidc_resp = client.get(f"{issuer.rstrip('/')}/.well-known/openid-configuration")
+                oidc_resp = client.get(f"{discovery_issuer.rstrip('/')}/.well-known/openid-configuration")
                 oidc_resp.raise_for_status()
                 resolved_jwks_uri = oidc_resp.json()["jwks_uri"]
 
@@ -102,7 +112,7 @@ def _get_key_set(force_refresh: bool = False) -> _KeySet:
         raise ValueError("audience must not be empty")
 
     base_claims_options: dict[str, Any] = {
-        "iss": {"essential": True, "value": issuer},
+        "iss": {"essential": True, "values": list(issuers)},
         "aud": {"essential": True, "values": list(audiences)},
     }
     if claims_options:
@@ -117,12 +127,12 @@ def _log_claim_mismatch(token: str, keys: _KeySet, exc: JoseError) -> None:
             token_iss = raw_claims.get("iss")
             logger.warning(
                 "JWT validation failed: %s\n"
-                "  expected iss: %s\n"
+                "  expected iss (any of): %s\n"
                 "  token   iss: %s\n"
                 "  expected aud (any of): %s\n"
                 "  token   aud: %s",
                 exc,
-                issuer,
+                list(issuers),
                 token_iss,
                 list(audiences),
                 token_aud,