feat(oauth): add refresh_token_error_messages to OAuthAuthenticator

airbyte_cdk/sources/declarative/auth/oauth.py

@@ -51,6 +51,7 @@ class DeclarativeOauth2Authenticator(AbstractOauth2Authenticator, DeclarativeAut
         refresh_token_error_status_codes (Tuple[int, ...]): Status codes to identify refresh token errors in response
         refresh_token_error_key (str): Key to identify refresh token error in response
         refresh_token_error_values (Tuple[str, ...]): List of values to check for exception during token refresh process
+        refresh_token_error_messages (Optional[Mapping[str, str]]): Fixed user-facing messages for refresh token errors
     """
 
     config: Mapping[str, Any]
@@ -81,12 +82,14 @@ class DeclarativeOauth2Authenticator(AbstractOauth2Authenticator, DeclarativeAut
     refresh_token_error_status_codes: Tuple[int, ...] = ()
     refresh_token_error_key: str = ""
     refresh_token_error_values: Tuple[str, ...] = ()
+    refresh_token_error_messages: Optional[Mapping[str, str]] = None
 
     def __post_init__(self, parameters: Mapping[str, Any]) -> None:
         super().__init__(
             refresh_token_error_status_codes=self.refresh_token_error_status_codes,
             refresh_token_error_key=self.refresh_token_error_key,
             refresh_token_error_values=self.refresh_token_error_values,
+            refresh_token_error_messages=self.refresh_token_error_messages,
         )
         if self.token_refresh_endpoint is not None:
             self._token_refresh_endpoint: Optional[InterpolatedString] = InterpolatedString.create(

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -1736,6 +1736,15 @@ definitions:
           type: string
         examples:
           - ["invalid_grant", "invalid_permissions"]
+      refresh_token_error_messages:
+        title: Refresh Token Error Messages
+        description: >-
+          Fixed user-facing messages for refresh token errors. Keys are matched in order against the provider error code leading error_description (a token such as ABC12345 followed by a word boundary), the whitespace-normalized full error_description, then the OAuth error value; the first match replaces the default re-authentication guidance, and the provider error detail is still appended. Only applies to responses already flagged as config errors by Refresh Token Error Status Codes, Refresh Token Error Key and Refresh Token Error Values (or their defaults); it does not flag any additional responses. Values are used verbatim, without interpolation, so the message is identical on every attempt.
+        type: object
+        additionalProperties:
+          type: string
+        examples:
+          - invalid_grant: "The refresh token expired or was revoked. Re-authenticate this source to restore syncs."
       refresh_token_updater:
         title: Refresh Token Updater
         description: When the refresh token updater is defined, new refresh tokens, access tokens and the access token expiry date are written back from the authentication response to the config object. This is important if the refresh token can only used once.

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -2221,6 +2221,16 @@ class OAuthAuthenticator(BaseModel):
         examples=[["invalid_grant", "invalid_permissions"]],
         title="Refresh Token Error Values",
     )
+    refresh_token_error_messages: Optional[Dict[str, str]] = Field(
+        None,
+        description="Fixed user-facing messages for refresh token errors. Keys are matched in order against the provider error code leading error_description (a token such as ABC12345 followed by a word boundary), the whitespace-normalized full error_description, then the OAuth error value; the first match replaces the default re-authentication guidance, and the provider error detail is still appended. Only applies to responses already flagged as config errors by Refresh Token Error Status Codes, Refresh Token Error Key and Refresh Token Error Values (or their defaults); it does not flag any additional responses. Values are used verbatim, without interpolation, so the message is identical on every attempt.",
+        examples=[
+            {
+                "invalid_grant": "The refresh token expired or was revoked. Re-authenticate this source to restore syncs."
+            }
+        ],
+        title="Refresh Token Error Messages",
+    )
     refresh_token_updater: Optional[RefreshTokenUpdater] = Field(
         None,
         description="When the refresh token updater is defined, new refresh tokens, access tokens and the access token expiry date are written back from the authentication response to the config object. This is important if the refresh token can only used once.",

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -3194,6 +3194,7 @@ def create_oauth_authenticator(
                 refresh_token_error_status_codes=refresh_token_error_status_codes,
                 refresh_token_error_key=refresh_token_error_key,
                 refresh_token_error_values=refresh_token_error_values,
+                refresh_token_error_messages=self._get_refresh_token_error_messages(model),
             )
         # ignore type error because fixing it would have a lot of dependencies, revisit later
         return DeclarativeOauth2Authenticator(  # type: ignore
@@ -3224,6 +3225,7 @@ def create_oauth_authenticator(
             refresh_token_error_status_codes=refresh_token_error_status_codes,
             refresh_token_error_key=refresh_token_error_key,
             refresh_token_error_values=refresh_token_error_values,
+            refresh_token_error_messages=self._get_refresh_token_error_messages(model),
         )
 
     @staticmethod
@@ -3274,6 +3276,11 @@ def _get_refresh_token_error_information(
         # returning default values we think cover most cases
         return (400,), "error", ("invalid_grant", "invalid_permissions")
 
+    @staticmethod
+    def _get_refresh_token_error_messages(model: OAuthAuthenticatorModel) -> Dict[str, str]:
+        """Resolve messages separately so they never disable the default gate."""
+        return dict(model.refresh_token_error_messages or {})
+
     @staticmethod
     def _reject_union_combined_extractor_for_offset_increment(
         extractor_model: Optional[BaseModel],

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

@@ -30,9 +30,13 @@
 # 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")
+_REFRESH_TOKEN_REJECTED_GUIDANCE = (
+    "Refresh token was rejected by the OAuth provider (invalid, expired, or already used). "
+    "Re-authenticate this source's credentials in its connection settings."
+)
 # 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
+# The longest standard token (`unsupported_grant_type`, 22) plus a 7-digit provider 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
@@ -68,14 +72,19 @@ def __init__(
         refresh_token_error_status_codes: Tuple[int, ...] = (),
         refresh_token_error_key: str = "",
         refresh_token_error_values: Tuple[str, ...] = (),
+        refresh_token_error_messages: Optional[Mapping[str, str]] = None,
     ) -> None:
         """
         If all of refresh_token_error_status_codes, refresh_token_error_key, and refresh_token_error_values are set,
         then http errors with such params will be wrapped in AirbyteTracedException.
+        refresh_token_error_messages maps fixed provider error strings to fixed user-facing messages.
         """
         self._refresh_token_error_status_codes = refresh_token_error_status_codes
         self._refresh_token_error_key = refresh_token_error_key
         self._refresh_token_error_values = refresh_token_error_values
+        self._refresh_token_error_messages: Mapping[str, str] = dict(
+            refresh_token_error_messages or {}
+        )
 
     def __call__(self, request: requests.PreparedRequest) -> requests.PreparedRequest:
         """Attach the HTTP headers required to authenticate on the HTTP request"""
@@ -315,6 +324,34 @@ def _redact_credentials(self, value: str) -> str:
     def _truncate(value: str, max_length: int) -> str:
         return value if len(value) <= max_length else value[:max_length] + "..."
 
+    @staticmethod
+    def _extract_provider_error_code(
+        response_content: Optional[Mapping[str, Any]],
+    ) -> Optional[str]:
+        if not response_content:
+            return None
+        description = response_content.get("error_description")
+        if isinstance(description, str):
+            code_match = _PROVIDER_ERROR_CODE_PATTERN.match(description.strip())
+            if code_match:
+                return code_match.group()
+        return None
+
+    def _get_refresh_token_error_message(
+        self, response_content: Optional[Mapping[str, Any]]
+    ) -> Optional[str]:
+        if not self._refresh_token_error_messages or not response_content:
+            return None
+        candidates = [self._extract_provider_error_code(response_content)]
+        for key in ("error_description", "error"):
+            value = response_content.get(key)
+            if isinstance(value, str):
+                candidates.append(" ".join(value.split()))
+        for candidate in candidates:
+            if candidate and candidate in self._refresh_token_error_messages:
+                return self._refresh_token_error_messages[candidate]
+        return None
+
     def _build_provider_error_detail(
         self, response_content: Optional[Mapping[str, Any]]
     ) -> Optional[str]:
@@ -324,26 +361,20 @@ def _build_provider_error_detail(
         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.
+        into a single failure summary. The description prose is deliberately excluded because it
+        can contain per-request values, which would make the grouping key unbounded. 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())
+        provider_error_code = self._extract_provider_error_code(response_content)
+        if provider_error_code:
+            parts.append(provider_error_code)
         if not parts:
             return None
         return self._truncate(
@@ -465,10 +496,10 @@ def _make_handled_request(self) -> Any:
                     )
             error_content = self._parse_error_response_content(e.response)
             if self._wrap_refresh_token_exception(e, response_content=error_content):
+                # Configured messages are fixed connector strings for failure-summary grouping.
                 message = (
-                    "Refresh token was rejected by the OAuth provider (invalid, expired, or "
-                    "already used). Re-authenticate this source's credentials in its connection "
-                    "settings."
+                    self._get_refresh_token_error_message(error_content)
+                    or _REFRESH_TOKEN_REJECTED_GUIDANCE
                 )
                 provider_error_detail = self._build_provider_error_detail(error_content)
                 if provider_error_detail:

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

@@ -28,6 +28,7 @@ class Oauth2Authenticator(AbstractOauth2Authenticator):
     Generates OAuth2.0 access tokens from an OAuth2.0 refresh token and client credentials.
     The generated access token is attached to each request via the Authorization header.
     If a connector_config is provided any mutation of it's value in the scope of this class will emit AirbyteControlConnectorConfigMessage.
+    `refresh_token_error_messages` provides fixed user-facing messages for configured refresh token errors.
     """
 
     def __init__(
@@ -53,6 +54,7 @@ def __init__(
         refresh_token_error_status_codes: Tuple[int, ...] = (),
         refresh_token_error_key: str = "",
         refresh_token_error_values: Tuple[str, ...] = (),
+        refresh_token_error_messages: Optional[Mapping[str, str]] = None,
     ) -> None:
         self._token_refresh_endpoint = token_refresh_endpoint
         self._client_secret_name = client_secret_name
@@ -75,7 +77,10 @@ def __init__(
         self._token_expiry_is_time_of_expiration = token_expiry_is_time_of_expiration
         self._access_token = None
         super().__init__(
-            refresh_token_error_status_codes, refresh_token_error_key, refresh_token_error_values
+            refresh_token_error_status_codes,
+            refresh_token_error_key,
+            refresh_token_error_values,
+            refresh_token_error_messages=refresh_token_error_messages,
         )
 
     def get_token_refresh_endpoint(self) -> str:
@@ -158,6 +163,7 @@ class SingleUseRefreshTokenOauth2Authenticator(Oauth2Authenticator):
     By default, this authenticator expects a connector config with a "credentials" field with the following nested fields: client_id,
     client_secret, refresh_token. This behavior can be changed by defining custom config path (using dpath paths) in client_id_config_path,
     client_secret_config_path, refresh_token_config_path constructor arguments.
+    `refresh_token_error_messages` provides fixed user-facing messages for configured refresh token errors.
     """
 
     def __init__(
@@ -186,6 +192,7 @@ def __init__(
         refresh_token_error_status_codes: Tuple[int, ...] = (),
         refresh_token_error_key: str = "",
         refresh_token_error_values: Tuple[str, ...] = (),
+        refresh_token_error_messages: Optional[Mapping[str, str]] = None,
     ) -> None:
         """
         Args:
@@ -207,6 +214,7 @@ def __init__(
             token_expiry_date_format (Optional[str]): Date format of the token expiry date field (set by expires_in_name). If specified the token expiry date is interpreted as time of expiration, as with token_expiry_is_time_of_expiration, otherwise as number of seconds until expiration.
             token_expiry_is_time_of_expiration bool: set True it if expires_in is returned as time of expiration instead of the number seconds until expiration
             message_repository (MessageRepository): the message repository used to emit logs on HTTP requests and control message on config update
+            refresh_token_error_messages (Optional[Mapping[str, str]]): Fixed user-facing messages for refresh token errors
         """
         self._connector_config = connector_config
         self._client_id: str = self._get_config_value_by_path(
@@ -248,6 +256,7 @@ def __init__(
             refresh_token_error_status_codes=refresh_token_error_status_codes,
             refresh_token_error_key=refresh_token_error_key,
             refresh_token_error_values=refresh_token_error_values,
+            refresh_token_error_messages=refresh_token_error_messages,
         )
 
     @property

unit_tests/sources/declarative/auth/test_oauth.py

@@ -788,7 +788,9 @@ def test_refresh_access_token_rejected_surfaces_provider_error_and_redacts(self,
             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 exc_info.value.message.startswith(
+            "Refresh token was rejected by the OAuth provider (invalid, expired, or already used)"
+        )
         # 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")
@@ -802,6 +804,36 @@ def test_refresh_access_token_rejected_surfaces_provider_error_and_redacts(self,
             assert parameters["refresh_token"] not in message
             assert config["client_secret"] not in message
 
+    def test_refresh_access_token_rejected_uses_configured_message(self, requests_mock):
+        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,
+            refresh_token_error_status_codes=(400,),
+            refresh_token_error_key="error",
+            refresh_token_error_values=("invalid_grant", "invalid_permissions"),
+            refresh_token_error_messages={"AADSTS50173": "Configured guidance."},
+        )
+        requests_mock.post(
+            config["refresh_endpoint"],
+            status_code=400,
+            json={
+                "error": "invalid_grant",
+                "error_description": "AADSTS50173: Grant revoked.\r\nTrace ID: x",
+            },
+        )
+
+        with pytest.raises(AirbyteTracedException) as exc_info:
+            oauth.refresh_access_token()
+
+        assert exc_info.value.message == (
+            "Configured guidance. Provider error: invalid_grant: AADSTS50173"
+        )
+        assert exc_info.value.failure_type == FailureType.config_error
+
 
 def mock_request(method, url, data, headers, **kwargs):
     if url == "https://refresh_endpoint.com":