""" Authentication manager for Stockbit API. Handles main token and trading token lifecycle. """ import time import requests from typing import Optional from .exceptions import AuthenticationError, TokenExpiredError, InvalidPINError class AuthManager: EXODUS_BASE = "https://exodus.stockbit.com" CARINA_BASE = "https://carina.stockbit.com" def __init__(self, main_token: str, trading_pin: Optional[str] = None): self.main_token = main_token self.trading_pin = trading_pin self.trading_token: Optional[str] = None self.trading_refresh: Optional[str] = None self.trading_expires_at: float = 0.0 self.account_number: Optional[int] = None self._session = requests.Session() self._session.headers.update({ "Authorization": f"Bearer {main_token}", "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)", }) @property def _exodus_headers(self): return {"Authorization": f"Bearer {self.main_token}"} @property def _carina_headers(self): return {"Authorization": f"Bearer {self.trading_token}"} def get_sekuritas_token(self) -> str: resp = self._session.get( f"{self.EXODUS_BASE}/sekuritas/auth/token", headers=self._exodus_headers, ) resp.raise_for_status() body = resp.json() if "data" not in body: raise AuthenticationError(f"Failed to get sekuritas token: {body}") d = body["data"] if isinstance(d, dict) and "token" in d: return d["token"] return d def login_trading(self, sekuritas_token: Optional[str] = None, pin: Optional[str] = None) -> dict: sekuritas_token = sekuritas_token or self.get_sekuritas_token() pin = pin or self.trading_pin if not pin: raise InvalidPINError("Trading PIN is required") resp = self._session.post( f"{self.CARINA_BASE}/auth/v2/login", json={"login_token": sekuritas_token, "pin": pin}, ) resp.raise_for_status() payload = resp.json() if "data" not in payload: raise AuthenticationError(f"Trading login failed: {payload}") data = payload["data"] self.trading_token = data["access_token"] self.trading_refresh = data.get("refresh_token") self.account_number = data.get("account", {}).get("number") # JWT typically expires in 24h self.trading_expires_at = time.time() + 86400 return data def refresh_trading_token(self) -> str: if not self.trading_refresh: raise AuthenticationError("No refresh token available") resp = self._session.post( f"{self.CARINA_BASE}/auth/refresh", headers={"Authorization": f"Bearer {self.trading_token}"}, ) resp.raise_for_status() payload = resp.json() if "data" not in payload: raise AuthenticationError(f"Token refresh failed: {payload}") data = payload["data"] self.trading_token = data.get("access_token", self.trading_token) self.trading_refresh = data.get("refresh_token", self.trading_refresh) self.trading_expires_at = time.time() + 86400 return self.trading_token def ensure_trading_token(self) -> str: if not self.trading_token: if not self.trading_pin: raise AuthenticationError("Trading not logged in. Call login_trading() first.") self.login_trading() elif time.time() > self.trading_expires_at - 300: try: self.refresh_trading_token() except Exception: if self.trading_pin: self.login_trading() else: raise TokenExpiredError("Trading token expired and cannot refresh") return self.trading_token