|
| 1 | +import threading |
| 2 | +import time |
| 3 | +from typing import Callable, Dict, Iterable, List, Optional |
| 4 | + |
| 5 | +import requests |
| 6 | +from databricks.sql.auth.authenticators import CredentialsProvider |
| 7 | +from databricks.sql.auth.endpoint import get_oauth_endpoints |
| 8 | + |
| 9 | + |
| 10 | +class ServicePrincipalConfigurationError(ValueError): |
| 11 | + """Raised when the service principal configuration is incomplete.""" |
| 12 | + |
| 13 | + |
| 14 | +class ServicePrincipalAuthenticationError(RuntimeError): |
| 15 | + """Raised when fetching an OAuth token fails.""" |
| 16 | + |
| 17 | + |
| 18 | +def _normalize_hostname(hostname: str) -> str: |
| 19 | + maybe_scheme = "" if hostname.startswith("https://") else "https://" |
| 20 | + trimmed = ( |
| 21 | + hostname[len("https://") :] if hostname.startswith("https://") else hostname |
| 22 | + ) |
| 23 | + return f"{maybe_scheme}{trimmed}".rstrip("/") |
| 24 | + |
| 25 | + |
| 26 | +class ServicePrincipalCredentialsProvider(CredentialsProvider): |
| 27 | + """CredentialsProvider that performs the Databricks OAuth client credentials flow.""" |
| 28 | + |
| 29 | + DEFAULT_SCOPES = ("sql",) |
| 30 | + |
| 31 | + def __init__( |
| 32 | + self, |
| 33 | + server_hostname: str, |
| 34 | + client_id: str, |
| 35 | + client_secret: str, |
| 36 | + *, |
| 37 | + scopes: Optional[Iterable[str]] = None, |
| 38 | + refresh_margin: int = 60, |
| 39 | + request_timeout: int = 10, |
| 40 | + ): |
| 41 | + if not server_hostname: |
| 42 | + raise ServicePrincipalConfigurationError("server_hostname is required") |
| 43 | + if not client_id: |
| 44 | + raise ServicePrincipalConfigurationError("client_id is required") |
| 45 | + if not client_secret: |
| 46 | + raise ServicePrincipalConfigurationError("client_secret is required") |
| 47 | + |
| 48 | + self._hostname = _normalize_hostname(server_hostname) |
| 49 | + oauth_endpoints = get_oauth_endpoints(self._hostname, use_azure_auth=False) |
| 50 | + if not oauth_endpoints: |
| 51 | + raise ServicePrincipalConfigurationError( |
| 52 | + f"Unable to determine OAuth endpoints for host {server_hostname}" |
| 53 | + ) |
| 54 | + |
| 55 | + scope_tuple = tuple(scopes) if scopes else self.DEFAULT_SCOPES |
| 56 | + mapped_scopes = oauth_endpoints.get_scopes_mapping(list(scope_tuple)) |
| 57 | + |
| 58 | + self._client_id = client_id |
| 59 | + self._client_secret = client_secret |
| 60 | + self._scopes: List[str] = mapped_scopes |
| 61 | + self._refresh_margin = refresh_margin |
| 62 | + self._request_timeout = request_timeout |
| 63 | + self._access_token: Optional[str] = None |
| 64 | + self._expires_at: float = 0 |
| 65 | + self._lock = threading.Lock() |
| 66 | + self._token_endpoint = self._discover_token_endpoint(oauth_endpoints) |
| 67 | + |
| 68 | + def auth_type(self) -> str: |
| 69 | + return "databricks-service-principal" |
| 70 | + |
| 71 | + def __call__(self) -> Callable[[], Dict[str, str]]: |
| 72 | + def header_factory() -> Dict[str, str]: |
| 73 | + access_token = self._get_token() |
| 74 | + return {"Authorization": f"Bearer {access_token}"} |
| 75 | + |
| 76 | + return header_factory |
| 77 | + |
| 78 | + def _discover_token_endpoint(self, oauth_endpoints) -> str: |
| 79 | + openid_config_url = oauth_endpoints.get_openid_config_url(self._hostname) |
| 80 | + try: |
| 81 | + response = requests.get(openid_config_url, timeout=self._request_timeout) |
| 82 | + response.raise_for_status() |
| 83 | + config = response.json() |
| 84 | + except Exception as exc: |
| 85 | + raise ServicePrincipalAuthenticationError( |
| 86 | + "Failed to load Databricks OAuth configuration" |
| 87 | + ) from exc |
| 88 | + |
| 89 | + token_endpoint = config.get("token_endpoint") |
| 90 | + if not token_endpoint: |
| 91 | + raise ServicePrincipalAuthenticationError( |
| 92 | + "OAuth configuration did not include a token endpoint" |
| 93 | + ) |
| 94 | + return token_endpoint |
| 95 | + |
| 96 | + def _needs_refresh(self) -> bool: |
| 97 | + if not self._access_token: |
| 98 | + return True |
| 99 | + now = time.time() |
| 100 | + return now >= (self._expires_at - self._refresh_margin) |
| 101 | + |
| 102 | + def _get_token(self) -> str: |
| 103 | + with self._lock: |
| 104 | + if self._needs_refresh(): |
| 105 | + self._refresh_token() |
| 106 | + assert self._access_token |
| 107 | + return self._access_token |
| 108 | + |
| 109 | + def _refresh_token(self) -> None: |
| 110 | + |
| 111 | + payload = { |
| 112 | + "grant_type": "client_credentials", |
| 113 | + "client_id": self._client_id, |
| 114 | + "client_secret": self._client_secret, |
| 115 | + "scope": " ".join(self._scopes), |
| 116 | + } |
| 117 | + |
| 118 | + response = requests.post( |
| 119 | + self._token_endpoint, data=payload, timeout=self._request_timeout |
| 120 | + ) |
| 121 | + try: |
| 122 | + response.raise_for_status() |
| 123 | + except Exception as exc: |
| 124 | + raise ServicePrincipalAuthenticationError( |
| 125 | + "Failed to retrieve OAuth token for service principal" |
| 126 | + ) from exc |
| 127 | + |
| 128 | + try: |
| 129 | + parsed = response.json() |
| 130 | + access_token = parsed["access_token"] |
| 131 | + except Exception as exc: # pragma: no cover - defensive |
| 132 | + raise ServicePrincipalAuthenticationError( |
| 133 | + "OAuth response did not include an access token" |
| 134 | + ) from exc |
| 135 | + |
| 136 | + expires_in_raw = parsed.get("expires_in", 3600) |
| 137 | + try: |
| 138 | + expires_in = int(expires_in_raw) |
| 139 | + except (TypeError, ValueError): |
| 140 | + expires_in = 3600 |
| 141 | + |
| 142 | + self._access_token = access_token |
| 143 | + self._expires_at = time.time() + max(expires_in, self._refresh_margin + 1) |
0 commit comments