528 lines
22 KiB
Python
528 lines
22 KiB
Python
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
|