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