import asyncio from datetime import UTC, datetime, timedelta from typing import Any import httpx from .crypto import TokenCipher from .domain import Binding, OAuthCredentials from .oauth import BotOAuthRepository AUTH_ERRORS = {"expired_token", "invalid_token", "no_auth_found"} class BitrixClient: """REST-клиент Битрикса с OAuth-контекстом привязанного пользователя.""" def __init__( self, credentials: BotOAuthRepository, cipher: TokenCipher, client_id: str, client_secret: str, oauth_token_url: str, client: httpx.AsyncClient | None = None ) -> None: self.credentials = credentials self.cipher = cipher self.client_id = client_id self.client_secret = client_secret self.oauth_token_url = oauth_token_url # Передача клиента для упрощения тестирования. self._client = client or httpx.AsyncClient(timeout=15) self._owns_client = client is None async def call( self, binding: Binding, method: str, params: dict[str, Any] | None = None ) -> dict[str, Any]: credentials = await self.credentials.get(binding) if not credentials: raise RuntimeError("OAuth-данные пользователя не найдены") # Отправляем запрос и в случае истечения токена запрашиваем обновление. data = await self._request(credentials, method, params) if str(data.get("error") or "").lower() in AUTH_ERRORS: credentials = await self._refresh(credentials, binding) data = await self._request(credentials, method, params) if "error" in data: description = data.get("error_description", data["error"]) raise RuntimeError(f"Bitrix API error: {description}") return data async def _request( self, credentials: OAuthCredentials, method: str, params: dict[str, Any] | None ) -> dict[str, Any]: payload = dict(params or {}) payload["auth"] = self.cipher.decrypt(credentials.access_token) response = await self._client.post( f"https://{credentials.domain}/rest/{method}.json", json=payload ) try: data = response.json() except ValueError: response.raise_for_status() raise RuntimeError("Bitrix вернул некорректный ответ") from None # Битрикс присылает полезное описание ошибки и при HTTP 4xx. if response.is_error and "error" not in data: response.raise_for_status() return data async def _refresh( self, credentials: OAuthCredentials, binding: Binding ) -> OAuthCredentials: # Если другой процесс уже обновляет токен, ждем его завершения. if not await self.credentials.claim_refresh(credentials): return await self._wait_for_refresh(credentials, binding) try: # Битрикс возвращает новую пару, поэтому обновляем оба токена. try: response = await self._client.get( self.oauth_token_url, params={ "grant_type": "refresh_token", "client_id": self.client_id, "client_secret": self.client_secret, "refresh_token": self.cipher.decrypt( credentials.refresh_token) } ) response.raise_for_status() except httpx.HTTPError: # Не включаем URL с OAuth-секретами в traceback. raise RuntimeError("Не удалось обновить OAuth-токен") from None data = response.json() if "error" in data: raise RuntimeError( "Bitrix OAuth error: " + str(data.get("error_description") or data["error"]) ) # Проверяем, что обновленный токен принадлежит тому же порталу # и пользователю. if data.get("member_id") not in {None, credentials.member_id}: raise RuntimeError("Bitrix вернул токен другого портала") if int(data.get("user_id", credentials.bitrix_user_id)) != ( credentials.bitrix_user_id ): raise RuntimeError("Bitrix вернул токен другого пользователя") # Обновляем токены в базе и возвращаем новые данные. expires_at = datetime.now(UTC) + timedelta( seconds=int(data.get("expires_in", 3600)) ) saved = await self.credentials.finish_refresh( credentials, self.cipher.encrypt(str(data["access_token"])), self.cipher.encrypt(str(data["refresh_token"])), expires_at ) # Если другой процесс успел обновить токен, ждем его завершения. if not saved: return await self._wait_for_refresh(credentials, binding) updated = await self.credentials.get(binding) if not updated: raise RuntimeError("Обновленные OAuth-данные не найдены") return updated except Exception: await self.credentials.release_refresh(credentials) raise async def _wait_for_refresh( self, previous: OAuthCredentials, binding: Binding ) -> OAuthCredentials: for _ in range(80): await asyncio.sleep(0.2) current = await self.credentials.get(binding) if current and current.version > previous.version: return current raise RuntimeError("Не удалось дождаться обновления OAuth-токена") async def close(self) -> None: if self._owns_client: await self._client.aclose()