diff --git a/airbyte_cdk/sources/streams/http/requests_native_auth/abstract_oauth.py b/airbyte_cdk/sources/streams/http/requests_native_auth/abstract_oauth.py index eae5be13a1..ed737684dd 100644 --- a/airbyte_cdk/sources/streams/http/requests_native_auth/abstract_oauth.py +++ b/airbyte_cdk/sources/streams/http/requests_native_auth/abstract_oauth.py @@ -17,7 +17,7 @@ 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 @@ -25,6 +25,16 @@ 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` 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`, 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 diff --git a/unit_tests/sources/declarative/auth/test_oauth.py b/unit_tests/sources/declarative/auth/test_oauth.py index 906390c90f..9c3ffe73f2 100644 --- a/unit_tests/sources/declarative/auth/test_oauth.py +++ b/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": diff --git a/unit_tests/sources/streams/http/requests_native_auth/test_requests_native_auth.py b/unit_tests/sources/streams/http/requests_native_auth/test_requests_native_auth.py index 3a7cb8d3d6..b03a6180d5 100644 --- a/unit_tests/sources/streams/http/requests_native_auth/test_requests_native_auth.py +++ b/unit_tests/sources/streams/http/requests_native_auth/test_requests_native_auth.py @@ -25,9 +25,11 @@ TokenAuthenticator, ) from airbyte_cdk.sources.streams.http.requests_native_auth.abstract_oauth import ( + _PROVIDER_ERROR_DETAIL_MAX_LENGTH, ResponseKeysMaxRecurtionReached, ) from airbyte_cdk.utils import AirbyteTracedException +from airbyte_cdk.utils.airbyte_secrets_utils import update_secrets from airbyte_cdk.utils.datetime_helpers import AirbyteDateTime, ab_datetime_now, ab_datetime_parse LOGGER = logging.getLogger(__name__) @@ -648,9 +650,184 @@ def test_refresh_access_token_wrapped( ) assert f"HTTP {response_code}" in exc_info.value.internal_message assert response_value in exc_info.value.internal_message - assert exc_info.value.message == error_message + assert exc_info.value.message.startswith(error_message) assert exc_info.value.failure_type == FailureType.config_error + def _entra_style_authenticator(self): + return Oauth2Authenticator( + f"https://{TestOauth2Authenticator.refresh_endpoint}", + TestOauth2Authenticator.client_id, + TestOauth2Authenticator.client_secret, + TestOauth2Authenticator.refresh_token, + refresh_token_error_status_codes=(400,), + refresh_token_error_key="error", + refresh_token_error_values=("invalid_grant", "invalid_client"), + ) + + # Sits on a later line of each provider payload below, like the trace ids and timestamps + # Microsoft Entra appends, so it must reach the internal message and never the user-facing one. + trailing_marker = "Correlation ID: 11111111-2222-3333-4444-555555555555" + + @pytest.mark.parametrize( + "error, error_description, expected_code", + ( + ( + # Grant revoked, e.g. the user changed their password: re-authentication is the fix. + "invalid_grant", + "AADSTS50173: The provided grant has expired due to it being revoked, a fresh auth " + "token is needed. The user might have changed or reset their password.\r\n" + "Trace ID: 00000000-0000-0000-0000-000000000000\r\n" + trailing_marker, + "AADSTS50173", + ), + ( + # Client type / credential misconfiguration: re-authentication will not help. + "invalid_client", + "AADSTS7000218: The request body must contain the following parameter: " + "'client_assertion' or 'client_secret'.\r\n" + trailing_marker, + "AADSTS7000218", + ), + ( + # Conditional Access requires an interactive sign-in. + "invalid_grant", + "AADSTS50076: Due to a configuration change made by your administrator, or because " + "you moved to a new location, you must use multi-factor authentication to access " + "the resource.\r\n" + trailing_marker, + "AADSTS50076", + ), + ), + ) + def test_refresh_access_token_surfaces_provider_error_code( + self, requests_mock, error, error_description, expected_code + ): + """The provider's own diagnostic must reach the logs and the user-facing message.""" + oauth = self._entra_style_authenticator() + requests_mock.post( + f"https://{TestOauth2Authenticator.refresh_endpoint}", + status_code=400, + json={"error": error, "error_description": error_description}, + ) + + with pytest.raises(AirbyteTracedException) as exc_info: + oauth.refresh_access_token() + + guidance = ( + "Refresh token was rejected by the OAuth provider (invalid, expired, or already used). " + "Re-authenticate this source's credentials in its connection settings." + ) + # The full provider response reaches the internal message, which is logged. + assert expected_code in exc_info.value.internal_message + assert TestOauth2Authenticator.trailing_marker in exc_info.value.internal_message + # The actionable guidance still leads the user-facing message ... + assert exc_info.value.message.startswith(guidance) + # ... followed by a short, single-line provider detail carrying the error code. + assert f"Provider error: {error}: {expected_code}" in exc_info.value.message + assert "\n" not in exc_info.value.message and "\r" not in exc_info.value.message + # Later lines of the description (trace ids, timestamps) stay out of the user-facing message, + # so the same failure produces the same message on every attempt. + assert TestOauth2Authenticator.trailing_marker not in exc_info.value.message + assert exc_info.value.failure_type == FailureType.config_error + + def test_refresh_access_token_truncates_long_provider_error_detail(self, requests_mock): + oauth = self._entra_style_authenticator() + error_description = "AADSTS50173: " + ("x" * 5000) + requests_mock.post( + f"https://{TestOauth2Authenticator.refresh_endpoint}", + status_code=400, + json={"error": "invalid_grant", "error_description": error_description}, + ) + + with pytest.raises(AirbyteTracedException) as exc_info: + oauth.refresh_access_token() + + provider_detail = exc_info.value.message.split("Provider error: ", 1)[1] + assert provider_detail.startswith("invalid_grant: AADSTS50173: ") + assert provider_detail.endswith("...") + assert len(provider_detail) == _PROVIDER_ERROR_DETAIL_MAX_LENGTH + len("...") + assert len(exc_info.value.internal_message) < 1200 + + def test_refresh_access_token_redacts_credentials_from_provider_error(self, requests_mock): + """A provider that echoes the submitted credentials must not leak them into the error.""" + # Start from no config-level secrets: earlier tests register values such as "token" through + # add_to_secrets, which would otherwise mask the credentials on their own. + update_secrets([]) + refresh_token = "0.AXoA-rt-9f3ZqW7kP2" + client_secret = "s3cr3t~Xyz-1Qp" + oauth = Oauth2Authenticator( + f"https://{TestOauth2Authenticator.refresh_endpoint}", + TestOauth2Authenticator.client_id, + client_secret, + refresh_token, + refresh_token_error_status_codes=(400,), + refresh_token_error_key="error", + refresh_token_error_values=("invalid_grant",), + ) + requests_mock.post( + f"https://{TestOauth2Authenticator.refresh_endpoint}", + status_code=400, + json={ + "error": "invalid_grant", + "error_description": f"AADSTS50173: rejected rt={refresh_token} cs={client_secret}", + }, + ) + + with pytest.raises(AirbyteTracedException) as exc_info: + oauth.refresh_access_token() + + for message in (exc_info.value.message, exc_info.value.internal_message): + assert refresh_token not in message + assert client_secret not in message + assert message.count("****") == 2 + assert "AADSTS50173" in exc_info.value.message + + @pytest.mark.parametrize( + "response_kwargs", + ( + {"text": ""}, + {"text": "Bad Request"}, + {"json": ["invalid_grant"]}, + ), + ids=["empty_body", "html_body", "json_array_body"], + ) + def test_refresh_access_token_non_json_body_does_not_raise_new_exception( + self, requests_mock, response_kwargs + ): + """A body we cannot read as a JSON object falls back to the raw RequestException.""" + oauth = self._entra_style_authenticator() + requests_mock.post( + f"https://{TestOauth2Authenticator.refresh_endpoint}", + status_code=400, + **response_kwargs, + ) + + with pytest.raises(RequestException): + oauth.refresh_access_token() + + def test_refresh_access_token_without_error_fields_keeps_bare_guidance(self, requests_mock): + """No usable provider detail means the user-facing message is left exactly as before.""" + oauth = Oauth2Authenticator( + f"https://{TestOauth2Authenticator.refresh_endpoint}", + TestOauth2Authenticator.client_id, + TestOauth2Authenticator.client_secret, + TestOauth2Authenticator.refresh_token, + refresh_token_error_status_codes=(400,), + refresh_token_error_key="errorCode", + refresh_token_error_values=("invalid_grant",), + ) + requests_mock.post( + f"https://{TestOauth2Authenticator.refresh_endpoint}", + status_code=400, + json={"errorCode": "invalid_grant", "errorSummary": "the grant was revoked"}, + ) + + with pytest.raises(AirbyteTracedException) as exc_info: + oauth.refresh_access_token() + + assert exc_info.value.message == ( + "Refresh token was rejected by the OAuth provider (invalid, expired, or already used). " + "Re-authenticate this source's credentials in its connection settings." + ) + assert "the grant was revoked" in exc_info.value.internal_message + @freezegun.freeze_time("2022-12-31") class TestSingleUseRefreshTokenOauth2Authenticator: