fix(oauth): auto-learn IdP audience and persist as resource for token validation

.secrets.baseline

@@ -3,7 +3,7 @@
     "files": "(?x)( package-lock\\.json$ |Cargo\\.lock$ |uv\\.lock$ |go\\.sum$ |mcpgateway/sri_hashes\\.json$ )|^.secrets.baseline$",
     "lines": null
   },
-  "generated_at": "2026-04-27T12:40:08Z",
+  "generated_at": "2026-04-27T14:14:15Z",
   "plugins_used": [
     {
       "name": "AWSKeyDetector"
@@ -5000,7 +5000,7 @@
         "hashed_secret": "d3ecb0d890368d7659ee54010045b835dacb8efe",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 625,
+        "line_number": 643,
         "type": "Secret Keyword",
         "verified_result": null
       }
@@ -7290,7 +7290,7 @@
         "hashed_secret": "72cb70dbbafe97e5ea13ad88acd65d08389439b0",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 137,
+        "line_number": 229,
         "type": "Secret Keyword",
         "verified_result": null
       }

mcpgateway/routers/oauth_router.py

@@ -73,6 +73,42 @@ def _normalize_resource_url(url: str | None, *, preserve_query: bool = False) ->
     return normalized
 
 
+async def _persist_learned_audience(gateway: Gateway, oauth_result: Dict[str, Any], db: Session) -> None:
+    """Learn the IdP's audience identifier from the token and persist it.
+
+    Many IdPs (ServiceNow, Authentik, etc.) do not honor RFC 8707 and set the
+    ``aud`` claim to an abstract identifier (often the ``client_id``) rather than
+    the ``resource`` URL sent in the authorization request.  By persisting the
+    actual ``aud`` value as ``resource`` in the gateway's ``oauth_config``, we
+    ensure that subsequent token validation in ``_validate_audience`` succeeds
+    and that future OAuth requests use the IdP's preferred audience identifier.
+
+    This is a best-effort operation: opaque tokens and missing aud claims are
+    silently ignored.
+
+    Args:
+        gateway: The gateway ORM object (will be mutated and flushed).
+        oauth_result: The result dict from ``complete_authorization_code_flow``,
+            expected to contain ``token_aud``.
+        db: Active database session.
+    """
+    token_aud = oauth_result.get("token_aud")
+    if token_aud is None:
+        return
+
+    # Store aud as-is (string or list) -- RFC 7519 allows both forms.
+    current_resource = (gateway.oauth_config or {}).get("resource")
+    if current_resource == token_aud:
+        return  # Already correct
+
+    # Persist the learned audience as resource
+    updated_config = dict(gateway.oauth_config) if gateway.oauth_config else {}
+    updated_config["resource"] = token_aud
+    gateway.oauth_config = updated_config
+    db.flush()
+    logger.debug("Learned OAuth audience from IdP token for gateway %s; persisted as resource", gateway.name)
+
+
 oauth_router = APIRouter(prefix="/oauth", tags=["oauth"])
 
 
@@ -298,22 +334,10 @@ async def initiate_oauth_flow(
 
         oauth_config = gateway.oauth_config.copy()  # Work with a copy to avoid mutating the original
 
-        # RFC 8707: Set resource parameter for JWT access tokens
-        # Respect pre-configured resource (e.g., for providers requiring pre-registered resources)
-        # Only derive from gateway.url if not explicitly configured
-        if oauth_config.get("resource"):
-            # Normalize existing resource - preserve query for explicit config (RFC 8707 allows when necessary)
-            existing = oauth_config["resource"]
-            if isinstance(existing, list):
-                original_count = len(existing)
-                normalized = [_normalize_resource_url(r, preserve_query=True) for r in existing]
-                oauth_config["resource"] = [r for r in normalized if r]
-                if not oauth_config["resource"] and original_count > 0:
-                    logger.warning(f"All {original_count} configured resource values were invalid and removed")
-            else:
-                oauth_config["resource"] = _normalize_resource_url(existing, preserve_query=True)
-        else:
-            # Default to gateway.url as the resource (strip query per RFC 8707 SHOULD NOT)
+        # RFC 8707: Set resource parameter for JWT access tokens.
+        # If resource was previously learned from the IdP's token aud claim, use it as-is.
+        # Otherwise derive from gateway.url for the first authorization request.
+        if not oauth_config.get("resource"):
             oauth_config["resource"] = _normalize_resource_url(gateway.url)
 
         # Phase 1.4: Auto-trigger DCR if credentials are missing
@@ -541,30 +565,24 @@ def _invalid_state_response() -> HTMLResponse:
 
         # Complete OAuth flow
 
-        # RFC 8707: Add resource parameter for JWT access tokens
-        # Must be set here in callback, not just in /authorize, because complete_authorization_code_flow
-        # needs it for the token exchange request
-        # Respect pre-configured resource; only derive from gateway.url if not explicitly configured
+        # RFC 8707: Set resource parameter for the token exchange request.
+        # If resource was previously learned from the IdP's token aud claim, use it as-is.
+        # Otherwise derive from gateway.url for the first authorization request.
         oauth_config_with_resource = gateway.oauth_config.copy()
-        if oauth_config_with_resource.get("resource"):
-            # Preserve query for explicit config (RFC 8707 allows when necessary)
-            existing = oauth_config_with_resource["resource"]
-            if isinstance(existing, list):
-                original_count = len(existing)
-                normalized = [_normalize_resource_url(r, preserve_query=True) for r in existing]
-                oauth_config_with_resource["resource"] = [r for r in normalized if r]
-                if not oauth_config_with_resource["resource"] and original_count > 0:
-                    logger.warning(f"All {original_count} configured resource values were invalid and removed")
-            else:
-                oauth_config_with_resource["resource"] = _normalize_resource_url(existing, preserve_query=True)
-        else:
-            # Strip query for auto-derived (RFC 8707 SHOULD NOT)
+        if not oauth_config_with_resource.get("resource"):
             oauth_config_with_resource["resource"] = _normalize_resource_url(gateway.url)
 
         result = await oauth_manager.complete_authorization_code_flow(
             gateway_id, code, state, oauth_config_with_resource, ca_certificate=gateway.ca_certificate, client_cert=gateway.client_cert, client_key=gateway.client_key
         )
 
+        # Learn the IdP's audience mapping from the token and persist as resource.
+        # RFC 8707 Section 2: "The authorization server may use the exact resource value
+        # as the audience or it may map from that value to a more general URI or abstract
+        # identifier for the given resource."  We persist whatever the IdP chose so that
+        # subsequent token validation matches.
+        await _persist_learned_audience(gateway, result, db)
+
         logger.info(f"Completed OAuth flow for gateway {SecurityValidator.sanitize_log_message(gateway_id)}, user {SecurityValidator.sanitize_log_message(str(result.get('user_id')))}")
 
         # Return success page with option to return to admin

mcpgateway/services/oauth_manager.py

@@ -777,6 +777,10 @@ async def complete_authorization_code_flow(
         # Extract user information from token response
         user_id = self._extract_user_id(token_response, credentials)
 
+        # Extract audience from token (best-effort) for caller to persist as resource.
+        # This enables audience learning for IdPs that map resource to a different aud.
+        token_aud = self._extract_token_audience(token_response.get("access_token", ""))
+
         # Store tokens if storage service is available
         if self.token_storage:
             token_record = await self.token_storage.store_tokens(
@@ -789,8 +793,8 @@ async def complete_authorization_code_flow(
                 scopes=token_response.get("scope", "").split(),
             )
 
-            return {"success": True, "user_id": user_id, "expires_at": token_record.expires_at.isoformat() if token_record.expires_at else None}
-        return {"success": True, "user_id": user_id, "expires_at": None}
+            return {"success": True, "user_id": user_id, "expires_at": token_record.expires_at.isoformat() if token_record.expires_at else None, "token_aud": token_aud}
+        return {"success": True, "user_id": user_id, "expires_at": None, "token_aud": token_aud}
 
     async def get_access_token_for_user(self, gateway_id: str, app_user_email: str) -> Optional[str]:
         """Get valid access token for a specific user.
@@ -1591,6 +1595,34 @@ def _extract_user_id(self, token_response: Dict[str, Any], credentials: Dict[str
         # Final fallback
         return "unknown_user"
 
+    @staticmethod
+    def _extract_token_audience(access_token: str) -> Any:
+        """Extract the ``aud`` claim from a JWT access token (best-effort).
+
+        Returns the raw ``aud`` value (string or list) or ``None`` for opaque
+        tokens or decode failures.  No signature verification is performed.
+
+        Args:
+            access_token: The raw access token string.
+
+        Returns:
+            The ``aud`` claim value, or None.
+        """
+        if not access_token:
+            return None
+        try:
+            # Third-Party
+            import jwt as pyjwt  # pylint: disable=import-outside-toplevel
+
+            claims = pyjwt.decode(
+                access_token,
+                options={"verify_signature": False, "verify_aud": False, "verify_iss": False, "verify_exp": False},
+                algorithms=["RS256", "RS384", "RS512", "ES256", "ES384", "ES512", "PS256", "PS384", "PS512", "HS256", "HS384", "HS512", "EdDSA"],
+            )
+            return claims.get("aud")
+        except Exception:  # noqa: BLE001
+            return None
+
 
 class OAuthError(Exception):
     """OAuth-related errors.

mcpgateway/services/token_validation_service.py

@@ -73,9 +73,9 @@ def blocking_errors(self) -> List[str]:
             >>> r.blocking_errors
             []
             >>> r.audience_match = False
-            >>> r.warnings.append("Token audience mismatch: token aud=[api://wrong], expected 'api://correct'")
+            >>> r.warnings.append("Token audience mismatch: token aud does not match expected resource or gateway URL")
             >>> r.blocking_errors
-            ["Token audience mismatch: token aud=[api://wrong], expected 'api://correct'"]
+            ['Token audience mismatch: token aud does not match expected resource or gateway URL']
         """
         if not self.warnings:
             return []
@@ -152,23 +152,24 @@ def _validate_audience(claims: Dict[str, Any], oauth_config: Dict[str, Any], gat
         gateway_name: Gateway name for log messages.
         result: Validation result to update in-place.
     """
-    expected_audience = oauth_config.get("resource") or gateway_url
-    if not expected_audience:
+    expected = oauth_config.get("resource") or gateway_url
+    if not expected:
         return
 
     token_aud = claims.get("aud")
     if token_aud is None:
         logger.debug("OAuth token for gateway %s has no 'aud' claim", gateway_name)
         return
 
+    # Normalize both sides to lists for a simple membership check.
+    # Per RFC 7519 Section 4.1.3, aud can be a string or array.
+    expected_list = expected if isinstance(expected, list) else [expected]
     aud_list = token_aud if isinstance(token_aud, list) else [token_aud]
-    if expected_audience in aud_list:
+    if any(a in expected_list for a in aud_list):
         result.audience_match = True
     else:
         result.audience_match = False
-        safe_aud = ", ".join(str(a)[:80] for a in aud_list[:3])
-        safe_expected = str(expected_audience)[:80]
-        result.warnings.append(f"Token audience mismatch: token aud=[{safe_aud}], expected '{safe_expected}'")
+        result.warnings.append("Token audience mismatch: token aud does not match expected resource or gateway URL")
 
 
 def _validate_scopes(claims: Dict[str, Any], oauth_config: Dict[str, Any], gateway_name: str, result: TokenValidationResult) -> None: