fix(oauth): preserve provider error detail on refresh token failure

airbyte_cdk/sources/streams/http/requests_native_auth/abstract_oauth.py

@@ -3,6 +3,7 @@
 #
 
 import logging
+import re
 import threading
 from abc import abstractmethod
 from datetime import timedelta
@@ -17,14 +18,27 @@
 from airbyte_cdk.sources.http_logger import format_http_message
 from airbyte_cdk.sources.message import MessageRepository, NoopMessageRepository
 from airbyte_cdk.utils import AirbyteTracedException
-from airbyte_cdk.utils.airbyte_secrets_utils import add_to_secrets
+from airbyte_cdk.utils.airbyte_secrets_utils import add_to_secrets, filter_secrets
 from airbyte_cdk.utils.datetime_helpers import AirbyteDateTime, ab_datetime_now, ab_datetime_parse
 
 from ..exceptions import DefaultBackoffException
 
 logger = logging.getLogger("airbyte")
 _NOOP_MESSAGE_REPOSITORY = NoopMessageRepository()
 
+# Provider error codes that lead `error_description`, e.g. Microsoft Entra's `AADSTS50173`.
+# Only the code is appended to the user-facing message: it is the entire grouping key, and unlike
+# the surrounding prose it is identical on every attempt at the same failure.
+_PROVIDER_ERROR_CODE_PATTERN = re.compile(r"^[A-Za-z]{3,}\d{4,}\b")
+# Upper bound on the provider-controlled text appended to the user-facing message. Both the
+# RFC 6749 `error` token and the extracted code come from the provider, so both are still capped.
+# The longest standard token (`unsupported_grant_type`, 22) plus a 7-digit Entra code is 37
+# characters, so this leaves roughly 2x headroom while still bounding a misbehaving provider.
+_PROVIDER_ERROR_DETAIL_MAX_LENGTH = 64
+# How much of the raw provider response is kept in the internal message, which is logged and is not
+# shown to the user.
+_PROVIDER_ERROR_RESPONSE_MAX_LENGTH = 1000
+
 
 class ResponseKeysMaxRecurtionReached(AirbyteTracedException):
     """
@@ -238,8 +252,100 @@ def _default_token_expiry_date(self) -> AirbyteDateTime:
         default_token_expiry_duration_hours = 1  # 1 hour
         return ab_datetime_now() + timedelta(hours=default_token_expiry_duration_hours)
 
+    @staticmethod
+    def _parse_error_response_content(
+        response: Optional[requests.Response],
+    ) -> Optional[Mapping[str, Any]]:
+        """
+        Best-effort parse of an error response body as a JSON object.
+
+        Returns `None` when the response is missing, empty, not valid JSON, or not a JSON object,
+        so that callers can degrade gracefully instead of raising a new exception while they are
+        already handling an error.
+        """
+        if response is None:
+            return None
+        try:
+            content = response.json()
+        except (JSONDecodeError, ValueError):
+            return None
+        return content if isinstance(content, Mapping) else None
+
+    def _redact_credentials(self, value: str) -> str:
+        """
+        Redact credential material from a string before it is logged or surfaced to the user.
+
+        Only response bodies are passed here, so request headers (including `Authorization`) are
+        never echoed. On top of the config secrets already tracked by the CDK, the authenticator's
+        own refresh token and client secret are redacted explicitly, in case a provider echoes the
+        submitted credentials back in its error payload.
+        """
+        redacted = filter_secrets(value)
+        for get_credential in (self.get_refresh_token, self.get_client_secret):
+            try:
+                credential = get_credential()
+            except Exception:
+                # Never let redaction itself fail the error path we are already in.
+                continue
+            if credential and isinstance(credential, str):
+                redacted = redacted.replace(credential, "****")
+        return redacted
+
+    @staticmethod
+    def _truncate(value: str, max_length: int) -> str:
+        return value if len(value) <= max_length else value[:max_length] + "..."
+
+    def _build_provider_error_detail(
+        self, response_content: Optional[Mapping[str, Any]]
+    ) -> Optional[str]:
+        """
+        Build a short, deterministic provider error detail for the user-facing message.
+
+        Only the standard OAuth 2.0 `error` field (RFC 6749 section 5.2) and the provider error
+        code leading `error_description` are used. Both are stable for a given failure, so the same
+        failure produces a byte-identical message on every attempt and the platform groups them
+        into a single failure summary. The description prose is deliberately excluded: it is
+        free-form, and providers embed per-request values in it -- Microsoft Entra's
+        `AADSTS700082` carries the token issue timestamp in its first sentence -- which would make
+        the grouping key unbounded. The code alone is what distinguishes a revoked grant
+        (`AADSTS50173`) from a misconfigured client (`AADSTS7000218`) or a Conditional Access
+        requirement (`AADSTS50076`). The full response body is preserved in the internal message,
+        which is logged. Which provider errors reach this path at all is set by the
+        authenticator's `refresh_token_error_*` configuration.
+        """
+        if not response_content:
+            return None
+        parts = []
+        error = response_content.get("error")
+        if isinstance(error, str) and error.strip():
+            parts.append(" ".join(error.split()))
+        description = response_content.get("error_description")
+        if isinstance(description, str):
+            code_match = _PROVIDER_ERROR_CODE_PATTERN.match(description.strip())
+            if code_match:
+                parts.append(code_match.group())
+        if not parts:
+            return None
+        return self._truncate(
+            self._redact_credentials(": ".join(parts)), _PROVIDER_ERROR_DETAIL_MAX_LENGTH
+        )
+
+    def _build_provider_response_info(self, exception: requests.exceptions.RequestException) -> str:
+        """
+        Build the full provider response detail for the internal message, which goes to the logs.
+        """
+        if exception.response is None:
+            return self._redact_credentials(str(exception))
+        body = self._truncate(
+            self._redact_credentials(exception.response.text),
+            _PROVIDER_ERROR_RESPONSE_MAX_LENGTH,
+        )
+        return f"HTTP {exception.response.status_code}: {body}"
+
     def _wrap_refresh_token_exception(
-        self, exception: requests.exceptions.RequestException
+        self,
+        exception: requests.exceptions.RequestException,
+        response_content: Optional[Mapping[str, Any]] = None,
     ) -> bool:
         """
         Wraps and handles exceptions that occur during the refresh token process.
@@ -249,16 +355,20 @@ def _wrap_refresh_token_exception(
 
         Args:
             exception (requests.exceptions.RequestException): The exception raised during the request.
+            response_content (Optional[Mapping[str, Any]]): The already-parsed response body, when
+                the caller has one, so the body is not parsed twice. Parsed on demand otherwise.
 
         Returns:
             bool: True if the exception is related to a refresh token error, False otherwise.
         """
-        try:
-            if exception.response is not None:
-                exception_content = exception.response.json()
-            else:
-                return False
-        except JSONDecodeError:
+        if exception.response is None:
+            return False
+        exception_content = (
+            response_content
+            if response_content is not None
+            else self._parse_error_response_content(exception.response)
+        )
+        if exception_content is None:
             return False
         return (
             exception.response.status_code in self._refresh_token_error_status_codes
@@ -333,15 +443,25 @@ def _make_handled_request(self) -> Any:
                         response=e.response,
                         failure_type=FailureType.transient_error,
                     )
-            if self._wrap_refresh_token_exception(e):
-                response_info = (
-                    f"HTTP {e.response.status_code}: {e.response.text[:1000]}"
-                    if e.response is not None
-                    else str(e)
+            error_content = self._parse_error_response_content(e.response)
+            if self._wrap_refresh_token_exception(e, response_content=error_content):
+                message = (
+                    "Refresh token was rejected by the OAuth provider (invalid, expired, or "
+                    "already used). Re-authenticate this source's credentials in its connection "
+                    "settings."
                 )
+                provider_error_detail = self._build_provider_error_detail(error_content)
+                if provider_error_detail:
+                    # The provider's own diagnostic is what tells apart otherwise identical-looking
+                    # failures (revoked grant vs. misconfigured client vs. Conditional Access), so
+                    # a short form of it is appended after the actionable guidance.
+                    message = f"{message} Provider error: {provider_error_detail}"
                 raise AirbyteTracedException(
-                    internal_message=f"Refresh token rejected by the OAuth token endpoint. {response_info}",
-                    message="Refresh token was rejected by the OAuth provider (invalid, expired, or already used). Re-authenticate this source's credentials in its connection settings.",
+                    internal_message=(
+                        "Refresh token rejected by the OAuth token endpoint. "
+                        f"{self._build_provider_response_info(e)}"
+                    ),
+                    message=message,
                     failure_type=FailureType.config_error,
                 ) from e
             raise

unit_tests/sources/declarative/auth/test_oauth.py

@@ -19,7 +19,7 @@
 from airbyte_cdk.sources.declarative.auth.jwt import JwtAuthenticator
 from airbyte_cdk.test.mock_http import HttpMocker, HttpRequest, HttpResponse
 from airbyte_cdk.utils import AirbyteTracedException
-from airbyte_cdk.utils.airbyte_secrets_utils import filter_secrets
+from airbyte_cdk.utils.airbyte_secrets_utils import filter_secrets, update_secrets
 from airbyte_cdk.utils.datetime_helpers import AirbyteDateTime, ab_datetime_now, ab_datetime_parse
 
 LOGGER = logging.getLogger(__name__)
@@ -687,6 +687,53 @@ def test_error_handling(self, mocker):
             oauth.refresh_access_token()
             assert e.value.errno == 400
 
+    def test_refresh_access_token_rejected_surfaces_provider_error_and_redacts(self, requests_mock):
+        """The declarative authenticator (what manifest connectors use) carries the provider's own
+        error code into the user-facing message and never the interpolated credentials."""
+        update_secrets([])
+        oauth = DeclarativeOauth2Authenticator(
+            token_refresh_endpoint="{{ config['refresh_endpoint'] }}",
+            client_id="{{ config['client_id'] }}",
+            client_secret="{{ config['client_secret'] }}",
+            refresh_token="{{ parameters['refresh_token'] }}",
+            config=config,
+            parameters=parameters,
+            # The defaults the component factory applies when a manifest sets none of these.
+            refresh_token_error_status_codes=(400,),
+            refresh_token_error_key="error",
+            refresh_token_error_values=("invalid_grant", "invalid_permissions"),
+        )
+        requests_mock.post(
+            config["refresh_endpoint"],
+            status_code=400,
+            json={
+                "error": "invalid_grant",
+                "error_description": (
+                    f"AADSTS50173: Rejected {parameters['refresh_token']} for {config['client_secret']}. "
+                    "The provided grant has expired due to it being revoked, a fresh auth token is "
+                    "needed.\r\nTrace ID: 00000000-0000-0000-0000-000000000000"
+                ),
+            },
+        )
+
+        with pytest.raises(AirbyteTracedException) as exc_info:
+            oauth.refresh_access_token()
+
+        assert exc_info.value.failure_type == FailureType.config_error
+        assert exc_info.value.message.startswith("Refresh token was rejected by the OAuth provider")
+        # Only the provider code is surfaced, so the echoed credentials cannot reach the
+        # user-facing message at all -- they are absent by construction, not merely masked.
+        assert exc_info.value.message.endswith("Provider error: invalid_grant: AADSTS50173")
+        assert "****" not in exc_info.value.message
+        assert "Rejected" not in exc_info.value.message
+        assert "Trace ID" not in exc_info.value.message
+        # The internal message keeps the full body with both credentials redacted.
+        assert exc_info.value.internal_message.count("****") == 2
+        assert "Trace ID" in exc_info.value.internal_message
+        for message in (exc_info.value.message, exc_info.value.internal_message):
+            assert parameters["refresh_token"] not in message
+            assert config["client_secret"] not in message
+
 
 def mock_request(method, url, data, headers, **kwargs):
     if url == "https://refresh_endpoint.com":