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

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

@@ -17,14 +17,24 @@
 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()
 
+# How much of the provider's error detail is appended to the user-facing message. Long enough to
+# keep a provider error code and the start of its description (e.g. Microsoft Entra puts the
+# `AADSTS<code>` at the front of `error_description`), short enough to keep the message readable
+# and to stop before the per-request values (issue dates, trace ids) that some providers embed
+# further into the description, so the same failure produces the same message on every attempt.
+_PROVIDER_ERROR_DETAIL_MAX_LENGTH = 120
+# 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 +248,94 @@ 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, single-line provider error detail suitable for the user-facing message.
+
+        Only the standard OAuth 2.0 `error` and `error_description` fields (RFC 6749 section 5.2)
+        are used, and only the first line of the description: providers put the actionable code
+        there (Microsoft Entra leads with `AADSTS<code>`, which tells a revoked or expired grant
+        such as `AADSTS50173` / `AADSTS700082` apart from a client misconfiguration such as
+        `AADSTS7000218`), while per-request trace ids and timestamps follow on later lines. 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 = [
+            response_content[key].strip()
+            for key in ("error", "error_description")
+            if isinstance(response_content.get(key), str) and response_content[key].strip()
+        ]
+        if not parts:
+            return None
+        # Keep the first line only and collapse its whitespace: the actionable code leads the
+        # description, while trace ids and timestamps that differ on every attempt follow on later
+        # lines and would make the same failure read differently each time.
+        detail = " ".join(": ".join(parts).splitlines()[0].split())
+        return self._truncate(self._redact_credentials(detail), _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 +345,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 +433,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,50 @@ 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")
+        assert (
+            "Provider error: invalid_grant: AADSTS50173: Rejected **** for ****."
+            in exc_info.value.message
+        )
+        assert "Trace ID" not in exc_info.value.message
+        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":