from __future__ import annotations import base64 import json import logging import os import re import time from datetime import datetime, timezone from pathlib import Path from typing import Any from urllib.error import HTTPError, URLError from urllib.parse import urlencode from urllib.request import Request, urlopen from playwright.sync_api import BrowserContext, Page, TimeoutError as PlaywrightTimeoutError from ..browser.factory import BrowserFactory from ..core.config import OpenLaneConfig logger = logging.getLogger("openlane_scraper.openlane.auth") # Порог предупреждения об истечении refresh_token (дни). _TOKEN_REFRESH_THRESHOLD_DAYS = 3 _ACCESS_TOKEN_MIN_TTL_SECONDS = 300 class OpenLaneAuthError(RuntimeError): pass class OpenLaneAuthenticator: def __init__(self, config: OpenLaneConfig) -> None: self.config = config def validate_credentials(self) -> None: if not self.config.username or not self.config.password: raise OpenLaneAuthError( "OPENLANE_USERNAME and OPENLANE_PASSWORD must be set in the environment" ) def interactive_login_and_persist(self, context: BrowserContext) -> str: page = context.new_page() page.set_default_timeout(self.config.request_timeout_ms) logger.info("OpenLane interactive login: open %s and complete sign-in manually", self.config.sign_in_url) page.goto(self.config.sign_in_url, wait_until="domcontentloaded") self._wait_for_authenticated_session(page) self._persist_storage_state(context) logger.info("OpenLane interactive login successful") return str(Path(self.config.storage_state_file)) def _inject_refresh_token(self, context: BrowserContext) -> None: # Инъекция refresh_token cookie из env. token = self.config.refresh_token if not token: return # Проверка: токен уже истёк? days = self._token_days_remaining(token) if days is not None and days <= 0: logger.warning( "OPENLANE_REFRESH_TOKEN is EXPIRED (%.1f days ago). " "Will attempt login with username/password.", abs(days), ) # Очищаем токен — fallback на login() self.config.refresh_token = "" return if days is not None: logger.info("refresh_token: %.1f days remaining (expires %s)", days, datetime.fromtimestamp( self._decode_jwt_exp(token), tz=timezone.utc # type: ignore[arg-type] ).strftime("%Y-%m-%d %H:%M UTC")) if days < _TOKEN_REFRESH_THRESHOLD_DAYS: logger.warning( "refresh_token expires in %.1f days! " "Will try to re-login after auth to get a fresh one.", days, ) context.add_cookies([ { "name": "refresh_token", "value": token, "domain": ".openlane.com", "path": "/", "httpOnly": True, "secure": True, "sameSite": "None", } ]) logger.info("Injected OPENLANE_REFRESH_TOKEN cookie into browser context") # ------------------------------------------------------------------ # Инспекция и ротация токенов # ------------------------------------------------------------------ @staticmethod def _decode_jwt_exp(token: str) -> int | None: # Декодирование JWT exp (без проверки подписи). try: parts = token.split(".") if len(parts) < 2: return None payload_b64 = parts[1] # Дополнение base64 padding. payload_b64 += "=" * (-len(payload_b64) % 4) payload = json.loads(base64.urlsafe_b64decode(payload_b64)) exp = payload.get("exp") return int(exp) if exp is not None else None except Exception: return None @staticmethod def _decode_jwt_payload(token: str) -> dict[str, Any]: # Декодирование JWT payload (без проверки подписи). try: parts = token.split(".") if len(parts) < 2: return {} payload_b64 = parts[1] payload_b64 += "=" * (-len(payload_b64) % 4) payload = json.loads(base64.urlsafe_b64decode(payload_b64)) return payload if isinstance(payload, dict) else {} except Exception: return {} @staticmethod def _token_days_remaining(token: str) -> float | None: # Дней до истечения токена. exp = OpenLaneAuthenticator._decode_jwt_exp(token) if exp is None: return None now = datetime.now(timezone.utc).timestamp() return (exp - now) / 86400 @staticmethod def _token_seconds_remaining(token: str) -> int | None: # Секунд до истечения токена. exp = OpenLaneAuthenticator._decode_jwt_exp(token) if exp is None: return None return int(exp - time.time()) @staticmethod def _extract_cookie_value(context: BrowserContext, name: str) -> str | None: # Извлекаем значение cookie по имени. cookies = context.cookies("https://app.openlane.com") for cookie in cookies: if cookie.get("name") == name: value = str(cookie.get("value") or "").strip() if value: return value return None def _is_access_token_fresh(self, token: str) -> bool: # Проверяем что access_token ещё жив. remain = self._token_seconds_remaining(token) return remain is not None and remain > _ACCESS_TOKEN_MIN_TTL_SECONDS def _extract_refresh_token_from_cookies(self, context: BrowserContext) -> str | None: # Извлечение refresh_token из cookies. return self._extract_cookie_value(context, "refresh_token") def _extract_access_token_from_cookies(self, context: BrowserContext) -> str | None: # Извлечение access_token из cookies. return self._extract_cookie_value(context, "access_token") def _refresh_access_token_via_okta(self, refresh_token: str) -> tuple[str | None, str | None]: # Прямой refresh через Okta token endpoint. if not refresh_token: return None, None if not self.config.okta_client_id: logger.warning("OPENLANE_OKTA_CLIENT_ID is empty, direct refresh disabled") return None, None form_data = urlencode( { "grant_type": "refresh_token", "client_id": self.config.okta_client_id, "redirect_uri": self.config.okta_redirect_uri, "refresh_token": refresh_token, } ).encode("utf-8") req = Request( self.config.okta_token_endpoint, data=form_data, method="POST", headers={ "Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json", }, ) try: with urlopen(req, timeout=self.config.okta_timeout_seconds) as resp: payload = json.loads(resp.read().decode("utf-8")) access_token = str(payload.get("access_token") or "").strip() new_refresh_token = str(payload.get("refresh_token") or "").strip() or refresh_token if not access_token: logger.warning("Okta refresh response has no access_token") return None, None logger.info("Okta direct refresh succeeded") return access_token, new_refresh_token except HTTPError as exc: try: body = exc.read().decode("utf-8", errors="ignore")[:300] except Exception: body = "" logger.warning("Okta direct refresh failed: HTTP %s %s", exc.code, body) return None, None except URLError as exc: logger.warning("Okta direct refresh failed: %s", exc) return None, None except Exception as exc: logger.warning("Okta direct refresh failed: %s", exc) return None, None def _persist_access_token_to_storage_state(self, context: BrowserContext, access_token: str) -> None: # Сохраняем access_token cookie в контекст. expires = self._decode_jwt_exp(access_token) if expires is None: expires = int(time.time()) + 7200 context.add_cookies( [ { "name": "access_token", "value": access_token, "domain": ".openlane.com", "path": "/", "httpOnly": True, "secure": True, "sameSite": "None", "expires": int(expires), } ] ) def _maybe_rotate_token(self, context: BrowserContext) -> None: # Проверка и ротация refresh_token. new_token = self._extract_refresh_token_from_cookies(context) if not new_token: logger.debug("No refresh_token cookie found after auth — nothing to rotate") return old_token = self.config.refresh_token or "" # Проверяем изменился ли токен. if new_token != old_token: days = self._token_days_remaining(new_token) logger.info( "refresh_token CHANGED (new exp: %.1f days). Persisting to .env", days if days is not None else -1, ) self.config.refresh_token = new_token self._persist_token_to_dotenv(new_token) return # Токен не изменился — проверяем срок. days = self._token_days_remaining(new_token) if days is not None: logger.info("refresh_token unchanged, %.1f days remaining", days) if days < _TOKEN_REFRESH_THRESHOLD_DAYS: logger.warning( "⚠ refresh_token expires in %.1f days! " "Run `openlane-login` or set OPENLANE_USERNAME + OPENLANE_PASSWORD " "to auto-renew on next sync.", days, ) @staticmethod def _persist_token_to_dotenv(token: str) -> None: # Запись refresh_token в .env. env_path = Path(".env") if not env_path.exists(): # Создаём .env с токеном. env_path.write_text( f"OPENLANE_REFRESH_TOKEN={token}\n", encoding="utf-8", ) logger.info("Created .env with rotated OPENLANE_REFRESH_TOKEN") return content = env_path.read_text(encoding="utf-8") pattern = re.compile(r"^OPENLANE_REFRESH_TOKEN=.*$", re.MULTILINE) replacement = f"OPENLANE_REFRESH_TOKEN={token}" if pattern.search(content): new_content = pattern.sub(replacement, content) else: new_content = content.rstrip("\n") + f"\n{replacement}\n" env_path.write_text(new_content, encoding="utf-8") # Обновляем env процесса. os.environ["OPENLANE_REFRESH_TOKEN"] = token logger.info("Persisted rotated OPENLANE_REFRESH_TOKEN to .env") def bootstrap_authenticated_context(self, context: BrowserContext) -> Page: # Восстановление сессии: storage_state → login → ошибка. storage_state_path = Path(self.config.storage_state_file) if storage_state_path.exists(): logger.info("OpenLane: restoring session from %s", storage_state_path) page = context.new_page() page.set_default_timeout(self.config.request_timeout_ms) # Блокируем трекинг/картинки — нужен только fetch(). BrowserFactory.enable_resource_blocking(page) # Проверка cookies без сетевых запросов. cookies = context.cookies("https://app.openlane.com") cookie_names = {c["name"] for c in cookies} access_token = self._extract_access_token_from_cookies(context) if access_token and self._is_access_token_fresh(access_token): logger.info( "OpenLane session restored: access_token cookie present (%d cookies total)", len(cookies), ) # Лёгкий API endpoint для установки origin в браузере. page.goto( "https://app.openlane.com/api/_next/time", wait_until="domcontentloaded", ) if self._is_api_authenticated(page): return page logger.warning("Access token cookie exists but API returned unauthorized, will re-authenticate") # Пробуем прямой refresh access_token без навигации. refresh_token = self._extract_refresh_token_from_cookies(context) or self.config.refresh_token if refresh_token: new_access_token, new_refresh_token = self._refresh_access_token_via_okta(refresh_token) if new_access_token: self._persist_access_token_to_storage_state(context, new_access_token) if new_refresh_token and new_refresh_token != (self.config.refresh_token or ""): self.config.refresh_token = new_refresh_token self._persist_token_to_dotenv(new_refresh_token) self._persist_storage_state(context) page.goto( "https://app.openlane.com/api/_next/time", wait_until="domcontentloaded", ) if self._is_api_authenticated(page): logger.info("OpenLane session restored via direct Okta refresh") return page logger.warning("Direct Okta refresh produced token but API still unauthorized") logger.warning("Direct refresh failed, trying sign_in fallback") # Если refresh_token был только в env, добавляем cookie вручную. if "refresh_token" not in cookie_names and self.config.refresh_token: self._inject_refresh_token(context) cookies = context.cookies("https://app.openlane.com") cookie_names = {c["name"] for c in cookies} logger.warning("storage_state exists but no access_token cookie — trying navigation") # Fallback: SPA обменяет refresh_token на access_token. if "refresh_token" in cookie_names: try: page.goto(self.config.sign_in_url, wait_until="domcontentloaded") self._wait_for_authenticated_session(page) self._persist_storage_state(context) self._maybe_rotate_token(context) if self._is_api_authenticated(page): return page logger.warning("sign_in fallback completed but API is still unauthorized") except OpenLaneAuthError: logger.warning("SPA refresh_token exchange failed") page.close() # Нет storage_state — пробуем логин. if self.config.username and self.config.password: return self.login(context) raise OpenLaneAuthError( "No valid session found. Either:\n" " 1. Run `python scripts/explore_site.py` to login manually and save storage_state, or\n" " 2. Set OPENLANE_USERNAME and OPENLANE_PASSWORD in .env" ) def _is_api_authenticated(self, page: Page) -> bool: """Проверяет, что сессия реально авторизована для OpenLane API. Нужна для случаев, когда access_token cookie есть (и даже не истёк), но сервер уже считает сессию невалидной. """ try: status = page.evaluate( """async () => { try { const resp = await fetch('https://app.openlane.com/api/v4/search?page=1&per_page=1&source_tab=marketplace&sale_types=marketplace', { method: 'GET', credentials: 'include', headers: { 'Accept': 'application/json, application/vnd.backlotcars.v3' } }); return resp.status; } catch (e) { return 0; } }""" ) except Exception: logger.warning("Failed to validate OpenLane API auth state", exc_info=True) return False try: code = int(status) except (TypeError, ValueError): return False # 401/403 — точно неавторизованы. if code in (401, 403, 0): return False return True def login(self, context: BrowserContext) -> Page: self.validate_credentials() page = context.new_page() page.set_default_timeout(self.config.request_timeout_ms) logger.info("OpenLane login: opening sign-in page %s", self.config.sign_in_url) page.goto(self.config.sign_in_url, wait_until="domcontentloaded") self._fill_login_form(page) self._wait_for_authenticated_session(page) self._persist_storage_state(context) self._maybe_rotate_token(context) logger.info("OpenLane login successful") return page def _fill_login_form(self, page: Page) -> None: email_selectors = [ 'input[name="username"]', 'input[name="user[login]"]', 'input[name="email"]', 'input[name="user[email]"]', 'input[autocomplete="username"]', 'input[type="email"]', 'input[type="text"]', '#email', ] password_selectors = [ 'input[name="password"]', 'input[name="user[password]"]', 'input[type="password"]', '#password', ] submit_selectors = [ 'button[type="submit"]', 'input[type="submit"]', 'button:has-text("Sign in")', 'button:has-text("Log in")', ] email_filled = self._fill_first(page, email_selectors, self.config.username) password_filled = self._fill_first(page, password_selectors, self.config.password) if not email_filled or not password_filled: raise OpenLaneAuthError("OpenLane login form fields were not found") if not self._click_first(page, submit_selectors): raise OpenLaneAuthError("OpenLane login submit button was not found") def _wait_for_authenticated_session(self, page: Page) -> None: # Ждём пока SPA обменяет refresh_token → access_token. try: page.wait_for_url( lambda url: "/sign_in" not in url.lower(), timeout=min(self.config.request_timeout_ms, 30000), ) except PlaywrightTimeoutError: raise OpenLaneAuthError( "OpenLane authentication failed: page stayed on sign_in " "(refresh_token likely invalid or expired)" ) # Проверяем что не на странице ошибки. current_url = page.url.lower() logger.debug("After auth redirect, URL: %s", page.url) if "/sign_in" in current_url: raise OpenLaneAuthError("OpenLane authentication failed: still on sign_in page") # Ждём появления access_token cookie. for attempt in range(15): cookies = page.context.cookies("https://app.openlane.com") cookie_names = {c["name"] for c in cookies} if "access_token" in cookie_names: logger.info("OpenLane authenticated: access_token cookie present (attempt %d)", attempt) return page.wait_for_timeout(500) # Нет access_token cookie, но URL не sign_in — продолжаем. logger.warning( "access_token cookie not found after redirect, but page is at %s — continuing", page.url, ) def _persist_storage_state(self, context: BrowserContext) -> None: if not self.config.persist_storage_state: return storage_state_path = Path(self.config.storage_state_file) storage_state_path.parent.mkdir(parents=True, exist_ok=True) context.storage_state(path=str(storage_state_path)) logger.info("OpenLane storage state saved to %s", storage_state_path) @staticmethod def _fill_first(page: Page, selectors: list[str], value: str) -> bool: for selector in selectors: locator = page.locator(selector) if locator.count() == 0: continue try: locator.first.fill(value) return True except PlaywrightTimeoutError: continue return False @staticmethod def _click_first(page: Page, selectors: list[str]) -> bool: for selector in selectors: locator = page.locator(selector) if locator.count() == 0: continue try: locator.first.click() return True except PlaywrightTimeoutError: continue return False