|
| 1 | +# Copyright (c) 2024 Airbyte, Inc., all rights reserved. |
| 2 | +"""Cloud credentials configuration for Airbyte Cloud API authentication. |
| 3 | +
|
| 4 | +This module provides the CloudCredentials class for managing authentication |
| 5 | +credentials when connecting to Airbyte Cloud, OSS, or Enterprise instances. |
| 6 | +
|
| 7 | +Two authentication methods are supported (mutually exclusive): |
| 8 | +1. OAuth2 client credentials (client_id + client_secret) |
| 9 | +2. Bearer token authentication |
| 10 | +
|
| 11 | +Example usage with client credentials: |
| 12 | + ```python |
| 13 | + from airbyte.cloud.credentials import CloudCredentials |
| 14 | +
|
| 15 | + credentials = CloudCredentials( |
| 16 | + client_id="your-client-id", |
| 17 | + client_secret="your-client-secret", |
| 18 | + ) |
| 19 | + ``` |
| 20 | +
|
| 21 | +Example usage with bearer token: |
| 22 | + ```python |
| 23 | + from airbyte.cloud.credentials import CloudCredentials |
| 24 | +
|
| 25 | + credentials = CloudCredentials( |
| 26 | + bearer_token="your-bearer-token", |
| 27 | + ) |
| 28 | + ``` |
| 29 | +
|
| 30 | +Example using environment variables: |
| 31 | + ```python |
| 32 | + from airbyte.cloud.credentials import CloudCredentials |
| 33 | +
|
| 34 | + # Resolves from AIRBYTE_CLOUD_CLIENT_ID, AIRBYTE_CLOUD_CLIENT_SECRET, |
| 35 | + # AIRBYTE_CLOUD_BEARER_TOKEN, and AIRBYTE_CLOUD_API_URL environment variables |
| 36 | + credentials = CloudCredentials.from_env() |
| 37 | + ``` |
| 38 | +""" |
| 39 | + |
| 40 | +from __future__ import annotations |
| 41 | + |
| 42 | +from dataclasses import dataclass |
| 43 | + |
| 44 | +from airbyte._util import api_util |
| 45 | +from airbyte.cloud.auth import ( |
| 46 | + resolve_cloud_api_url, |
| 47 | + resolve_cloud_bearer_token, |
| 48 | + resolve_cloud_client_id, |
| 49 | + resolve_cloud_client_secret, |
| 50 | +) |
| 51 | +from airbyte.exceptions import PyAirbyteInputError |
| 52 | +from airbyte.secrets.base import SecretString |
| 53 | + |
| 54 | + |
| 55 | +@dataclass |
| 56 | +class CloudCredentials: |
| 57 | + """Authentication credentials for Airbyte Cloud API. |
| 58 | +
|
| 59 | + This class encapsulates the authentication configuration needed to connect |
| 60 | + to Airbyte Cloud, OSS, or Enterprise instances. It supports two mutually |
| 61 | + exclusive authentication methods: |
| 62 | +
|
| 63 | + 1. OAuth2 client credentials flow (client_id + client_secret) |
| 64 | + 2. Bearer token authentication |
| 65 | +
|
| 66 | + Exactly one authentication method must be provided. Providing both or neither |
| 67 | + will raise a validation error. |
| 68 | +
|
| 69 | + Attributes: |
| 70 | + client_id: OAuth2 client ID for client credentials flow. |
| 71 | + client_secret: OAuth2 client secret for client credentials flow. |
| 72 | + bearer_token: Pre-generated bearer token for direct authentication. |
| 73 | + api_root: The API root URL. Defaults to Airbyte Cloud API. |
| 74 | + """ |
| 75 | + |
| 76 | + client_id: SecretString | None = None |
| 77 | + """OAuth2 client ID for client credentials authentication.""" |
| 78 | + |
| 79 | + client_secret: SecretString | None = None |
| 80 | + """OAuth2 client secret for client credentials authentication.""" |
| 81 | + |
| 82 | + bearer_token: SecretString | None = None |
| 83 | + """Bearer token for direct authentication (alternative to client credentials).""" |
| 84 | + |
| 85 | + api_root: str = api_util.CLOUD_API_ROOT |
| 86 | + """The API root URL. Defaults to Airbyte Cloud API.""" |
| 87 | + |
| 88 | + def __post_init__(self) -> None: |
| 89 | + """Validate credentials and ensure secrets are properly wrapped.""" |
| 90 | + # Wrap secrets in SecretString if they aren't already |
| 91 | + if self.client_id is not None: |
| 92 | + self.client_id = SecretString(self.client_id) |
| 93 | + if self.client_secret is not None: |
| 94 | + self.client_secret = SecretString(self.client_secret) |
| 95 | + if self.bearer_token is not None: |
| 96 | + self.bearer_token = SecretString(self.bearer_token) |
| 97 | + |
| 98 | + # Validate mutual exclusivity |
| 99 | + has_client_credentials = self.client_id is not None or self.client_secret is not None |
| 100 | + has_bearer_token = self.bearer_token is not None |
| 101 | + |
| 102 | + if has_client_credentials and has_bearer_token: |
| 103 | + raise PyAirbyteInputError( |
| 104 | + message="Cannot use both client credentials and bearer token authentication.", |
| 105 | + guidance=( |
| 106 | + "Provide either client_id and client_secret together, " |
| 107 | + "or bearer_token alone, but not both." |
| 108 | + ), |
| 109 | + ) |
| 110 | + |
| 111 | + if has_client_credentials and (self.client_id is None or self.client_secret is None): |
| 112 | + # If using client credentials, both must be provided |
| 113 | + raise PyAirbyteInputError( |
| 114 | + message="Incomplete client credentials.", |
| 115 | + guidance=( |
| 116 | + "When using client credentials authentication, " |
| 117 | + "both client_id and client_secret must be provided." |
| 118 | + ), |
| 119 | + ) |
| 120 | + |
| 121 | + if not has_client_credentials and not has_bearer_token: |
| 122 | + raise PyAirbyteInputError( |
| 123 | + message="No authentication credentials provided.", |
| 124 | + guidance=( |
| 125 | + "Provide either client_id and client_secret together for OAuth2 " |
| 126 | + "client credentials flow, or bearer_token for direct authentication." |
| 127 | + ), |
| 128 | + ) |
| 129 | + |
| 130 | + @property |
| 131 | + def uses_bearer_token(self) -> bool: |
| 132 | + """Return True if using bearer token authentication.""" |
| 133 | + return self.bearer_token is not None |
| 134 | + |
| 135 | + @property |
| 136 | + def uses_client_credentials(self) -> bool: |
| 137 | + """Return True if using client credentials authentication.""" |
| 138 | + return self.client_id is not None and self.client_secret is not None |
| 139 | + |
| 140 | + @classmethod |
| 141 | + def from_env( |
| 142 | + cls, |
| 143 | + *, |
| 144 | + api_root: str | None = None, |
| 145 | + ) -> CloudCredentials: |
| 146 | + """Create CloudCredentials from environment variables. |
| 147 | +
|
| 148 | + This factory method resolves credentials from environment variables, |
| 149 | + providing a convenient way to create credentials without explicitly |
| 150 | + passing secrets. |
| 151 | +
|
| 152 | + Environment variables used: |
| 153 | + - `AIRBYTE_CLOUD_CLIENT_ID`: OAuth client ID (for client credentials flow). |
| 154 | + - `AIRBYTE_CLOUD_CLIENT_SECRET`: OAuth client secret (for client credentials flow). |
| 155 | + - `AIRBYTE_CLOUD_BEARER_TOKEN`: Bearer token (alternative to client credentials). |
| 156 | + - `AIRBYTE_CLOUD_API_URL`: Optional. The API root URL (defaults to Airbyte Cloud). |
| 157 | +
|
| 158 | + The method will first check for a bearer token. If not found, it will |
| 159 | + attempt to use client credentials. |
| 160 | +
|
| 161 | + Args: |
| 162 | + api_root: The API root URL. If not provided, will be resolved from |
| 163 | + the `AIRBYTE_CLOUD_API_URL` environment variable, or default to |
| 164 | + the Airbyte Cloud API. |
| 165 | +
|
| 166 | + Returns: |
| 167 | + A CloudCredentials instance configured with credentials from the environment. |
| 168 | +
|
| 169 | + Raises: |
| 170 | + PyAirbyteSecretNotFoundError: If required credentials are not found in |
| 171 | + the environment. |
| 172 | + """ |
| 173 | + resolved_api_root = resolve_cloud_api_url(api_root) |
| 174 | + |
| 175 | + # Try bearer token first |
| 176 | + bearer_token = resolve_cloud_bearer_token() |
| 177 | + if bearer_token: |
| 178 | + return cls( |
| 179 | + bearer_token=bearer_token, |
| 180 | + api_root=resolved_api_root, |
| 181 | + ) |
| 182 | + |
| 183 | + # Fall back to client credentials |
| 184 | + return cls( |
| 185 | + client_id=resolve_cloud_client_id(), |
| 186 | + client_secret=resolve_cloud_client_secret(), |
| 187 | + api_root=resolved_api_root, |
| 188 | + ) |
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?v=4&size=40){kind=avatar}
0 commit comments