airbyte_cdk.sources.streams.http.requests_native_auth.abstract_oauth

  1#
  2# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
  3#
  4
  5import logging
  6import re
  7import threading
  8from abc import abstractmethod
  9from datetime import datetime, timedelta
 10from json import JSONDecodeError
 11from typing import Any, List, Mapping, MutableMapping, Optional, Tuple, Union
 12
 13import backoff
 14import requests
 15from requests.auth import AuthBase
 16
 17from airbyte_cdk.models import FailureType, Level
 18from airbyte_cdk.sources.http_logger import format_http_message
 19from airbyte_cdk.sources.message import MessageRepository, NoopMessageRepository
 20from airbyte_cdk.utils import AirbyteTracedException
 21from airbyte_cdk.utils.airbyte_secrets_utils import add_to_secrets, filter_secrets
 22from airbyte_cdk.utils.datetime_helpers import AirbyteDateTime, ab_datetime_now, ab_datetime_parse
 23
 24from ..exceptions import DefaultBackoffException
 25
 26logger = logging.getLogger("airbyte")
 27_NOOP_MESSAGE_REPOSITORY = NoopMessageRepository()
 28
 29# Provider error codes that lead `error_description`, e.g. Microsoft Entra's `AADSTS50173`.
 30# Only the code is appended to the user-facing message: it is the entire grouping key, and unlike
 31# the surrounding prose it is identical on every attempt at the same failure.
 32_PROVIDER_ERROR_CODE_PATTERN = re.compile(r"^[A-Za-z]{3,}\d{4,}\b")
 33_REFRESH_TOKEN_REJECTED_GUIDANCE = (
 34    "Refresh token was rejected by the OAuth provider (invalid, expired, or already used). "
 35    "Re-authenticate this source's credentials in its connection settings."
 36)
 37# Upper bound on the provider-controlled text appended to the user-facing message. Both the
 38# RFC 6749 `error` token and the extracted code come from the provider, so both are still capped.
 39# The longest standard token (`unsupported_grant_type`, 22) plus a 7-digit provider code is 37
 40# characters, so this leaves roughly 2x headroom while still bounding a misbehaving provider.
 41_PROVIDER_ERROR_DETAIL_MAX_LENGTH = 64
 42# How much of the raw provider response is kept in the internal message, which is logged and is not
 43# shown to the user.
 44_PROVIDER_ERROR_RESPONSE_MAX_LENGTH = 1000
 45
 46
 47class ResponseKeysMaxRecurtionReached(AirbyteTracedException):
 48    """
 49    Raised when the max level of recursion is reached, when trying to
 50    find-and-get the target key, during the `_make_handled_request`
 51    """
 52
 53
 54class AbstractOauth2Authenticator(AuthBase):
 55    """
 56    Abstract class for an OAuth authenticators that implements the OAuth token refresh flow. The authenticator
 57    is designed to generically perform the refresh flow without regard to how config fields are get/set by
 58    delegating that behavior to the classes implementing the interface.
 59    """
 60
 61    _NO_STREAM_NAME = None
 62
 63    # Class-level lock to prevent concurrent token refresh across multiple authenticator instances.
 64    # This is necessary because multiple streams may share the same OAuth credentials (refresh token)
 65    # through the connector config. Without this lock, concurrent refresh attempts can cause race
 66    # conditions where one stream successfully refreshes the token while others fail because the
 67    # refresh token has been invalidated (especially for single-use refresh tokens).
 68    _token_refresh_lock = threading.RLock()
 69
 70    def __init__(
 71        self,
 72        refresh_token_error_status_codes: Tuple[int, ...] = (),
 73        refresh_token_error_key: str = "",
 74        refresh_token_error_values: Tuple[str, ...] = (),
 75        refresh_token_error_messages: Optional[Mapping[str, str]] = None,
 76    ) -> None:
 77        """
 78        If all of refresh_token_error_status_codes, refresh_token_error_key, and refresh_token_error_values are set,
 79        then http errors with such params will be wrapped in AirbyteTracedException.
 80        refresh_token_error_messages maps fixed provider error strings to fixed user-facing messages.
 81        """
 82        self._refresh_token_error_status_codes = refresh_token_error_status_codes
 83        self._refresh_token_error_key = refresh_token_error_key
 84        self._refresh_token_error_values = refresh_token_error_values
 85        self._refresh_token_error_messages: Mapping[str, str] = dict(
 86            refresh_token_error_messages or {}
 87        )
 88
 89    def __call__(self, request: requests.PreparedRequest) -> requests.PreparedRequest:
 90        """Attach the HTTP headers required to authenticate on the HTTP request"""
 91        request.headers.update(self.get_auth_header())
 92        return request
 93
 94    @property
 95    def _is_access_token_flow(self) -> bool:
 96        return self.get_token_refresh_endpoint() is None and self.access_token is not None
 97
 98    @property
 99    def token_expiry_is_time_of_expiration(self) -> bool:
100        """
101        Indicates that the Token Expiry returns the date until which the token will be valid, not the amount of time it will be valid.
102        """
103
104        return False
105
106    @property
107    def token_expiry_date_format(self) -> Optional[str]:
108        """
109        Format of the datetime; exists it if expires_in is returned as the expiration datetime instead of seconds until it expires
110        """
111
112        return None
113
114    def get_auth_header(self) -> Mapping[str, Any]:
115        """HTTP header to set on the requests"""
116        token = self.access_token if self._is_access_token_flow else self.get_access_token()
117        return {"Authorization": f"Bearer {token}"}
118
119    def get_access_token(self) -> str:
120        """
121        Returns the access token.
122
123        This method uses double-checked locking to ensure thread-safe token refresh.
124        When multiple threads (streams) detect an expired token simultaneously, only one
125        will perform the refresh while others wait. After acquiring the lock, the token
126        expiry is re-checked to avoid redundant refresh attempts.
127        """
128        if self.token_has_expired():
129            with self._token_refresh_lock:
130                # Double-check after acquiring lock - another thread may have already refreshed
131                if self.token_has_expired():
132                    self.refresh_and_set_access_token()
133
134        return self.access_token
135
136    def _current_access_token_or_none(self) -> Optional[str]:
137        """The current access token, or None when the implementation has none to report
138        (e.g., a declarative authenticator whose token has not been initialized yet)."""
139        try:
140            return self.access_token
141        except Exception:
142            return None
143
144    def refresh_and_set_access_token(self) -> None:
145        """Force refresh the access token and update internal state.
146
147        Refreshes regardless of expiry, serialized on the class-level refresh lock. If another
148        thread using this same authenticator instance replaced the access token while this one
149        waited for the lock, the refresh is skipped and the request is retried with that token.
150        `SingleUseRefreshTokenOauth2Authenticator` reads `access_token` from the connector config
151        shared by all stream instances, so the early return also covers separate instances there.
152        Only per-instance token authenticators (base and declarative) are limited to same-instance
153        detection; the Authorization-header check in `HttpClient._handle_error_resolution` covers
154        the rest by skipping the forced refresh when the rejected request's token was already
155        replaced.
156        Subclasses may override this to handle additional state updates (e.g., persisting new
157        refresh tokens).
158        """
159        token_before_waiting = self._current_access_token_or_none()
160        with self._token_refresh_lock:
161            if self._current_access_token_or_none() != token_before_waiting:
162                return
163            token, expires_in = self.refresh_access_token()
164            self.access_token = token
165            self.set_token_expiry_date(expires_in)
166
167    def token_has_expired(self) -> bool:
168        """Returns True if the token is expired"""
169        return ab_datetime_now() > self.get_token_expiry_date()
170
171    def _build_standard_refresh_args(self) -> MutableMapping[str, Any]:
172        """Build the standard OAuth refresh args (grant_type, refresh_token, client
173        credentials, scopes, plus any user-configured `refresh_request_body` extras).
174
175        Used by both `build_refresh_request_body()` and
176        `build_refresh_request_query_params()` so the same set of args can be emitted
177        in either the body or the URL query string depending on
178        `should_send_refresh_request_as_query_params()`.
179
180        Client credentials (client_id and client_secret) are excluded when
181        `refresh_request_headers` contains an `Authorization` header (e.g. Basic
182        auth). This is required by OAuth providers like Gong that expect credentials
183        ONLY in the Authorization header and reject requests that include them in
184        both places.
185        """
186        headers = self.get_refresh_request_headers()
187        credentials_in_header = headers and "Authorization" in headers
188        include_client_credentials = not credentials_in_header
189
190        payload: MutableMapping[str, Any] = {
191            self.get_grant_type_name(): self.get_grant_type(),
192        }
193
194        if include_client_credentials:
195            payload[self.get_client_id_name()] = self.get_client_id()
196            payload[self.get_client_secret_name()] = self.get_client_secret()
197
198        payload[self.get_refresh_token_name()] = self.get_refresh_token()
199
200        if self.get_scopes():
201            payload["scopes"] = self.get_scopes()
202
203        if self.get_refresh_request_body():
204            for key, val in self.get_refresh_request_body().items():
205                # Existing oauth args take precedence over custom configured fields.
206                if key not in payload:
207                    payload[key] = val
208
209        return payload
210
211    def build_refresh_request_body(self) -> Mapping[str, Any]:
212        """Returns the request body to set on the refresh request.
213
214        When `should_send_refresh_request_as_query_params()` is `True`, the standard
215        refresh args are emitted on the URL query string instead and this method
216        returns an empty body. This supports OAuth providers like Gong that document
217        their refresh endpoint as a `POST` with parameters on the URL query string
218        and an empty body.
219        """
220        if self.should_send_refresh_request_as_query_params():
221            return {}
222        return self._build_standard_refresh_args()
223
224    def build_refresh_request_headers(self) -> Mapping[str, Any] | None:
225        """
226        Returns the request headers to set on the refresh request
227
228        """
229        headers = self.get_refresh_request_headers()
230        return headers if headers else None
231
232    def build_refresh_request_query_params(self) -> Mapping[str, Any] | None:
233        """Returns the URL query string parameters to set on the refresh request.
234
235        When `should_send_refresh_request_as_query_params()` is `True`, the standard
236        refresh args (grant_type, refresh_token, client credentials, scopes, plus
237        any user-configured `refresh_request_body` extras) are returned here and
238        `build_refresh_request_body()` returns an empty body.
239
240        Returns `None` otherwise so existing authenticators retain their previous
241        behavior (no query params on the refresh URL).
242        """
243        if not self.should_send_refresh_request_as_query_params():
244            return None
245        return self._build_standard_refresh_args()
246
247    def refresh_access_token(self) -> Tuple[str, AirbyteDateTime]:
248        """
249        Returns the refresh token and its expiration datetime
250
251        :return: a tuple of (access_token, token_lifespan)
252        """
253        try:
254            response_json = self._make_handled_request()
255        except (
256            requests.exceptions.ConnectionError,
257            requests.exceptions.ConnectTimeout,
258            requests.exceptions.ReadTimeout,
259        ) as e:
260            raise AirbyteTracedException(
261                message="OAuth access token refresh request failed due to a network error.",
262                internal_message=f"Network error during OAuth token refresh after retries were exhausted: {e}",
263                failure_type=FailureType.transient_error,
264            ) from e
265        self._ensure_access_token_in_response(response_json)
266
267        return (
268            self._extract_access_token(response_json),
269            self._extract_token_expiry_date(response_json),
270        )
271
272    # ----------------
273    # PRIVATE METHODS
274    # ----------------
275
276    def _default_token_expiry_date(self) -> AirbyteDateTime:
277        """
278        Returns the default token expiry date
279        """
280        # 1 hour was chosen as a middle ground to avoid unnecessary frequent refreshes and token expiration
281        default_token_expiry_duration_hours = 1  # 1 hour
282        return ab_datetime_now() + timedelta(hours=default_token_expiry_duration_hours)
283
284    @staticmethod
285    def _parse_error_response_content(
286        response: Optional[requests.Response],
287    ) -> Optional[Mapping[str, Any]]:
288        """
289        Best-effort parse of an error response body as a JSON object.
290
291        Returns `None` when the response is missing, empty, not valid JSON, or not a JSON object,
292        so that callers can degrade gracefully instead of raising a new exception while they are
293        already handling an error.
294        """
295        if response is None:
296            return None
297        try:
298            content = response.json()
299        except (JSONDecodeError, ValueError):
300            return None
301        return content if isinstance(content, Mapping) else None
302
303    def _redact_credentials(self, value: str) -> str:
304        """
305        Redact credential material from a string before it is logged or surfaced to the user.
306
307        Only response bodies are passed here, so request headers (including `Authorization`) are
308        never echoed. On top of the config secrets already tracked by the CDK, the authenticator's
309        own refresh token and client secret are redacted explicitly, in case a provider echoes the
310        submitted credentials back in its error payload.
311        """
312        redacted = filter_secrets(value)
313        for get_credential in (self.get_refresh_token, self.get_client_secret):
314            try:
315                credential = get_credential()
316            except Exception:
317                # Never let redaction itself fail the error path we are already in.
318                continue
319            if credential and isinstance(credential, str):
320                redacted = redacted.replace(credential, "****")
321        return redacted
322
323    @staticmethod
324    def _truncate(value: str, max_length: int) -> str:
325        return value if len(value) <= max_length else value[:max_length] + "..."
326
327    @staticmethod
328    def _extract_provider_error_code(
329        response_content: Optional[Mapping[str, Any]],
330    ) -> Optional[str]:
331        if not response_content:
332            return None
333        description = response_content.get("error_description")
334        if isinstance(description, str):
335            code_match = _PROVIDER_ERROR_CODE_PATTERN.match(description.strip())
336            if code_match:
337                return code_match.group()
338        return None
339
340    def _get_refresh_token_error_message(
341        self, response_content: Optional[Mapping[str, Any]]
342    ) -> Optional[str]:
343        if not self._refresh_token_error_messages or not response_content:
344            return None
345        candidates = [self._extract_provider_error_code(response_content)]
346        for key in ("error_description", "error"):
347            value = response_content.get(key)
348            if isinstance(value, str):
349                candidates.append(" ".join(value.split()))
350        for candidate in candidates:
351            if candidate and candidate in self._refresh_token_error_messages:
352                return self._refresh_token_error_messages[candidate]
353        return None
354
355    def _build_provider_error_detail(
356        self, response_content: Optional[Mapping[str, Any]]
357    ) -> Optional[str]:
358        """
359        Build a short, deterministic provider error detail for the user-facing message.
360
361        Only the standard OAuth 2.0 `error` field (RFC 6749 section 5.2) and the provider error
362        code leading `error_description` are used. Both are stable for a given failure, so the same
363        failure produces a byte-identical message on every attempt and the platform groups them
364        into a single failure summary. The description prose is deliberately excluded because it
365        can contain per-request values, which would make the grouping key unbounded. The full
366        response body is preserved in the internal message, which is logged. Which provider errors
367        reach this path at all is set by the authenticator's `refresh_token_error_*` configuration.
368        """
369        if not response_content:
370            return None
371        parts = []
372        error = response_content.get("error")
373        if isinstance(error, str) and error.strip():
374            parts.append(" ".join(error.split()))
375        provider_error_code = self._extract_provider_error_code(response_content)
376        if provider_error_code:
377            parts.append(provider_error_code)
378        if not parts:
379            return None
380        return self._truncate(
381            self._redact_credentials(": ".join(parts)), _PROVIDER_ERROR_DETAIL_MAX_LENGTH
382        )
383
384    def _build_provider_response_info(self, exception: requests.exceptions.RequestException) -> str:
385        """
386        Build the full provider response detail for the internal message, which goes to the logs.
387        """
388        if exception.response is None:
389            return self._redact_credentials(str(exception))
390        body = self._truncate(
391            self._redact_credentials(exception.response.text),
392            _PROVIDER_ERROR_RESPONSE_MAX_LENGTH,
393        )
394        return f"HTTP {exception.response.status_code}: {body}"
395
396    def _wrap_refresh_token_exception(
397        self,
398        exception: requests.exceptions.RequestException,
399        response_content: Optional[Mapping[str, Any]] = None,
400    ) -> bool:
401        """
402        Wraps and handles exceptions that occur during the refresh token process.
403
404        This method checks if the provided exception is related to a refresh token error
405        by examining the response status code and specific error content.
406
407        Args:
408            exception (requests.exceptions.RequestException): The exception raised during the request.
409            response_content (Optional[Mapping[str, Any]]): The already-parsed response body, when
410                the caller has one, so the body is not parsed twice. Parsed on demand otherwise.
411
412        Returns:
413            bool: True if the exception is related to a refresh token error, False otherwise.
414        """
415        if exception.response is None:
416            return False
417        exception_content = (
418            response_content
419            if response_content is not None
420            else self._parse_error_response_content(exception.response)
421        )
422        if exception_content is None:
423            return False
424        return (
425            exception.response.status_code in self._refresh_token_error_status_codes
426            and exception_content.get(self._refresh_token_error_key)
427            in self._refresh_token_error_values
428        )
429
430    @backoff.on_exception(
431        backoff.expo,
432        (
433            DefaultBackoffException,
434            requests.exceptions.ConnectionError,
435            requests.exceptions.ConnectTimeout,
436            requests.exceptions.ReadTimeout,
437        ),
438        on_backoff=lambda details: logger.info(
439            f"Caught retryable error after {details['tries']} tries. Waiting {details['wait']} seconds then retrying..."
440        ),
441        max_time=300,
442    )
443    def _make_handled_request(self) -> Any:
444        """
445        Makes a handled HTTP request to refresh an OAuth token.
446
447        This method sends a POST request to the token refresh endpoint with the necessary
448        headers and body to obtain a new access token. It handles various exceptions that
449        may occur during the request and logs the response for troubleshooting purposes.
450
451        Returns:
452            Mapping[str, Any]: The JSON response from the token refresh endpoint.
453
454        Raises:
455            DefaultBackoffException: If the response status code is 429 (Too Many Requests)
456                                     or any 5xx server error.
457            AirbyteTracedException: If the refresh token is invalid or expired, prompting
458                                    re-authentication.
459            Exception: For any other exceptions that occur during the request.
460        """
461        try:
462            response = requests.request(
463                method="POST",
464                url=self.get_token_refresh_endpoint(),  # type: ignore # returns None, if not provided, but str | bytes is expected.
465                data=self.build_refresh_request_body(),
466                headers=self.build_refresh_request_headers(),
467                params=self.build_refresh_request_query_params(),
468            )
469
470            if not response.ok:
471                # log the response even if the request failed for troubleshooting purposes
472                self._log_response(response)
473                response.raise_for_status()
474
475            response_json = response.json()
476
477            try:
478                # extract the access token and add to secrets to avoid logging the raw value
479                access_key = self._extract_access_token(response_json)
480                if access_key:
481                    add_to_secrets(access_key)
482            except ResponseKeysMaxRecurtionReached as e:
483                # could not find the access token in the response, so do nothing
484                pass
485
486            self._log_response(response)
487
488            return response_json
489        except requests.exceptions.RequestException as e:
490            if e.response is not None:
491                if e.response.status_code == 429 or e.response.status_code >= 500:
492                    raise DefaultBackoffException(
493                        request=e.response.request,
494                        response=e.response,
495                        failure_type=FailureType.transient_error,
496                    )
497            error_content = self._parse_error_response_content(e.response)
498            if self._wrap_refresh_token_exception(e, response_content=error_content):
499                # Configured messages are fixed connector strings for failure-summary grouping.
500                message = (
501                    self._get_refresh_token_error_message(error_content)
502                    or _REFRESH_TOKEN_REJECTED_GUIDANCE
503                )
504                provider_error_detail = self._build_provider_error_detail(error_content)
505                if provider_error_detail:
506                    # The provider's own diagnostic is what tells apart otherwise identical-looking
507                    # failures (revoked grant vs. misconfigured client vs. Conditional Access), so
508                    # a short form of it is appended after the actionable guidance.
509                    message = f"{message} Provider error: {provider_error_detail}"
510                raise AirbyteTracedException(
511                    internal_message=(
512                        "Refresh token rejected by the OAuth token endpoint. "
513                        f"{self._build_provider_response_info(e)}"
514                    ),
515                    message=message,
516                    failure_type=FailureType.config_error,
517                ) from e
518            raise
519        except Exception as e:
520            raise AirbyteTracedException(
521                message="OAuth access token refresh request failed.",
522                internal_message=f"Unexpected error during OAuth token refresh: {e}",
523                failure_type=FailureType.system_error,
524            ) from e
525
526    def _ensure_access_token_in_response(self, response_data: Mapping[str, Any]) -> None:
527        """
528        Ensures that the access token is present in the response data.
529
530        This method attempts to extract the access token from the provided response data.
531        If the access token is not found, it raises an exception indicating that the token
532        refresh API response was missing the access token.
533
534        Args:
535            response_data (Mapping[str, Any]): The response data from which to extract the access token.
536
537        Raises:
538            Exception: If the access token is not found in the response data.
539            ResponseKeysMaxRecurtionReached: If the maximum recursion depth is reached while extracting the access token.
540        """
541        try:
542            access_key = self._extract_access_token(response_data)
543            if not access_key:
544                raise Exception(
545                    f"Token refresh API response was missing access token {self.get_access_token_name()}"
546                )
547        except ResponseKeysMaxRecurtionReached as e:
548            raise e
549
550    def _parse_token_expiration_date(self, value: Union[str, int]) -> AirbyteDateTime:
551        """
552        Parse a string or integer token expiration date into a datetime object. A datetime, which
553        `refresh_access_token` returns since 6.45.5, is passed through for setters written earlier.
554
555        :return: expiration datetime
556        """
557        if isinstance(value, datetime):
558            return AirbyteDateTime.from_datetime(value)
559        if self.token_expiry_is_time_of_expiration:
560            if not self.token_expiry_date_format:
561                raise ValueError(
562                    f"Invalid token expiry date format {self.token_expiry_date_format}; a string representing the format is required."
563                )
564            try:
565                return ab_datetime_parse(str(value))
566            except ValueError as e:
567                raise ValueError(f"Invalid token expiry date format: {e}")
568        else:
569            try:
570                # Only accept numeric values (as int/float/string) when no format specified
571                seconds = int(float(str(value)))
572                return ab_datetime_now() + timedelta(seconds=seconds)
573            except (ValueError, TypeError):
574                raise ValueError(
575                    f"Invalid expires_in value: {value}. Expected number of seconds when no format specified."
576                )
577
578    def _extract_access_token(self, response_data: Mapping[str, Any]) -> Any:
579        """
580        Extracts the access token from the given response data.
581
582        Args:
583            response_data (Mapping[str, Any]): The response data from which to extract the access token.
584
585        Returns:
586            str: The extracted access token.
587        """
588        return self._find_and_get_value_from_response(response_data, self.get_access_token_name())
589
590    def _extract_refresh_token(self, response_data: Mapping[str, Any]) -> Any:
591        """
592        Extracts the refresh token from the given response data.
593
594        Args:
595            response_data (Mapping[str, Any]): The response data from which to extract the refresh token.
596
597        Returns:
598            str: The extracted refresh token.
599        """
600        return self._find_and_get_value_from_response(response_data, self.get_refresh_token_name())
601
602    def _extract_token_expiry_date(self, response_data: Mapping[str, Any]) -> AirbyteDateTime:
603        """
604        Extracts the token_expiry_date, like `expires_in` or `expires_at`, etc from the given response data.
605
606        If the token_expiry_date is not found, it will return an existing token expiry date if set, or a default token expiry date.
607
608        Args:
609            response_data (Mapping[str, Any]): The response data from which to extract the token_expiry_date.
610
611        Returns:
612            The extracted token_expiry_date or None if not found.
613        """
614        expires_in = self._find_and_get_value_from_response(
615            response_data, self.get_expires_in_name()
616        )
617        if expires_in is not None:
618            return self._parse_token_expiration_date(expires_in)
619
620        # expires_in is None
621        existing_expiry_date = self.get_token_expiry_date()
622        if existing_expiry_date and not self.token_has_expired():
623            return existing_expiry_date
624
625        return self._default_token_expiry_date()
626
627    def _find_and_get_value_from_response(
628        self,
629        response_data: Mapping[str, Any],
630        key_name: str,
631        max_depth: int = 5,
632        current_depth: int = 0,
633    ) -> Any:
634        """
635        Recursively searches for a specified key in a nested dictionary or list and returns its value if found.
636
637        Args:
638            response_data (Mapping[str, Any]): The response data to search through, which can be a dictionary or a list.
639            key_name (str): The key to search for in the response data.
640            max_depth (int, optional): The maximum depth to search for the key to avoid infinite recursion. Defaults to 5.
641            current_depth (int, optional): The current depth of the recursion. Defaults to 0.
642
643        Returns:
644            Any: The value associated with the specified key if found, otherwise None.
645
646        Raises:
647            AirbyteTracedException: If the maximum recursion depth is reached without finding the key.
648        """
649        if current_depth > max_depth:
650            # this is needed to avoid an inf loop, possible with a very deep nesting observed.
651            message = f"The maximum level of recursion is reached. Couldn't find the specified `{key_name}` in the response."
652            raise ResponseKeysMaxRecurtionReached(
653                internal_message=message, message=message, failure_type=FailureType.config_error
654            )
655
656        if isinstance(response_data, dict):
657            # get from the root level
658            if key_name in response_data:
659                return response_data[key_name]
660
661            # get from the nested object
662            for _, value in response_data.items():
663                result = self._find_and_get_value_from_response(
664                    value, key_name, max_depth, current_depth + 1
665                )
666                if result is not None:
667                    return result
668
669        # get from the nested array object
670        elif isinstance(response_data, list):
671            for item in response_data:
672                result = self._find_and_get_value_from_response(
673                    item, key_name, max_depth, current_depth + 1
674                )
675                if result is not None:
676                    return result
677
678        return None
679
680    @property
681    def _message_repository(self) -> Optional[MessageRepository]:
682        """
683        The implementation can define a message_repository if it wants debugging logs for HTTP requests
684        """
685        return _NOOP_MESSAGE_REPOSITORY
686
687    def _log_response(self, response: requests.Response) -> None:
688        """
689        Logs the HTTP response using the message repository if it is available.
690
691        Args:
692            response (requests.Response): The HTTP response to log.
693        """
694        if self._message_repository:
695            self._message_repository.log_message(
696                Level.DEBUG,
697                lambda: format_http_message(
698                    response,
699                    "Refresh token",
700                    "Obtains access token",
701                    self._NO_STREAM_NAME,
702                    is_auxiliary=True,
703                    type="AUTH",
704                ),
705            )
706
707    # ----------------
708    # ABSTR METHODS
709    # ----------------
710
711    @abstractmethod
712    def get_token_refresh_endpoint(self) -> Optional[str]:
713        """Returns the endpoint to refresh the access token"""
714
715    @abstractmethod
716    def get_client_id_name(self) -> str:
717        """The client id name to authenticate"""
718
719    @abstractmethod
720    def get_client_id(self) -> str:
721        """The client id to authenticate"""
722
723    @abstractmethod
724    def get_client_secret_name(self) -> str:
725        """The client secret name to authenticate"""
726
727    @abstractmethod
728    def get_client_secret(self) -> str:
729        """The client secret to authenticate"""
730
731    @abstractmethod
732    def get_refresh_token_name(self) -> str:
733        """The refresh token name to authenticate"""
734
735    @abstractmethod
736    def get_refresh_token(self) -> Optional[str]:
737        """The token used to refresh the access token when it expires"""
738
739    @abstractmethod
740    def get_scopes(self) -> List[str]:
741        """List of requested scopes"""
742
743    @abstractmethod
744    def get_token_expiry_date(self) -> AirbyteDateTime:
745        """Expiration date of the access token"""
746
747    @abstractmethod
748    def set_token_expiry_date(self, value: AirbyteDateTime) -> None:
749        """Setter for access token expiration date"""
750
751    @abstractmethod
752    def get_access_token_name(self) -> str:
753        """Field to extract access token from in the response"""
754
755    @abstractmethod
756    def get_expires_in_name(self) -> str:
757        """Returns the expires_in field name"""
758
759    @abstractmethod
760    def get_refresh_request_body(self) -> Mapping[str, Any]:
761        """Returns the request body to set on the refresh request"""
762
763    @abstractmethod
764    def get_refresh_request_headers(self) -> Mapping[str, Any]:
765        """Returns the request headers to set on the refresh request"""
766
767    def should_send_refresh_request_as_query_params(self) -> bool:
768        """Returns `True` if the standard refresh args should be sent on the URL
769        query string instead of in the request body.
770
771        Defaults to `False` so existing authenticators retain their previous
772        behavior (params in body, no query params on the refresh URL). Subclasses
773        can override this to opt into the URL-query-string shape required by OAuth
774        providers like Gong.
775        """
776        return False
777
778    @abstractmethod
779    def get_grant_type(self) -> str:
780        """Returns grant_type specified for requesting access_token"""
781
782    @abstractmethod
783    def get_grant_type_name(self) -> str:
784        """Returns grant_type specified name for requesting access_token"""
785
786    @property
787    @abstractmethod
788    def access_token(self) -> str:
789        """Returns the access token"""
790
791    @access_token.setter
792    @abstractmethod
793    def access_token(self, value: str) -> str:
794        """Setter for the access token"""