diff --git a/src/selectel_ml_tui/auth.py b/src/selectel_ml_tui/auth.py new file mode 100644 index 0000000..bcf2bb9 --- /dev/null +++ b/src/selectel_ml_tui/auth.py @@ -0,0 +1,137 @@ +"""IAM-авторизация Selectel. + +Получение и кэширование IAM-токена проекта через +POST {identity_url}/auth/tokens (сервисный пользователь, scope project). +Время жизни токена — 24 часа; IamTokenManager автоматически обновляет токен +при приближении к истечению (запас 5 минут). +""" + +from __future__ import annotations + +import datetime as dt +from typing import cast + +import httpx +from pydantic import BaseModel, ConfigDict, Field + +from selectel_ml_tui.config import Credentials, Settings + + +class IamToken(BaseModel): + """IAM-токен с временем жизни.""" + + model_config = ConfigDict(populate_by_name=True) + + value: str = Field(alias="X-Subject-Token") + issued_at: dt.datetime = Field( + default_factory=lambda: dt.datetime.now(dt.timezone.utc) + ) + expires_at: dt.datetime = Field(alias="expires_at") + + +class IamAuthError(RuntimeError): + """Ошибка получения IAM-токена.""" + + +def _parse_expires_at(raw: str) -> dt.datetime: + """Распарсить expires_at из ответа IAM (ISO8601, возможно с 'Z').""" + text = raw.replace("Z", "+00:00") + return dt.datetime.fromisoformat(text) + + +class IamClient: + """Клиент получения IAM-токена проекта.""" + + def __init__( + self, + settings: Settings, + credentials: Credentials, + transport: httpx.AsyncBaseTransport | None = None, + ) -> None: + self._settings = settings + self._credentials = credentials + self._transport = transport + + async def get_token(self) -> IamToken: + """Получить IAM-токен проекта для аккаунта/домена.""" + if not self._credentials.service_user or not self._credentials.password: + raise IamAuthError( + "Не заданы логин/пароль сервисного пользователя. " + "Выполните: selectel-ml-tui config" + ) + body = { + "auth": { + "identity": { + "methods": ["password"], + "password": { + "user": { + "name": self._credentials.service_user, + "domain": {"name": self._credentials.domain_id}, + "password": self._credentials.password, + } + }, + }, + "scope": {"project": {"id": self._credentials.project_id}}, + } + } + url = f"{self._settings.identity_url}/auth/tokens" + try: + async with httpx.AsyncClient( + timeout=30, transport=self._transport + ) as client: + response = await client.post(url, json=body) + except httpx.HTTPError as exc: + raise IamAuthError(f"Ошибка сети при обращении к {url}: {exc}") from exc + if response.status_code not in (200, 201): + raise IamAuthError( + f"IAM: HTTP {response.status_code}: {response.text[:300]}" + ) + token = cast(str | None, response.headers.get("X-Subject-Token")) + if not token: + raise IamAuthError("IAM: отсутствует заголовок X-Subject-Token в ответе") + data = response.json() + expires_raw = ( + data.get("token", {}).get("expires_at") if isinstance(data, dict) else None + ) + if not isinstance(expires_raw, str): + raise IamAuthError("IAM: отсутствует поле token.expires_at в ответе") + return IamToken.model_validate( + { + "X-Subject-Token": token, + "expires_at": _parse_expires_at(expires_raw), + } + ) + + +class IamTokenManager: + """Кэширует IAM-токен и обновляет его по истечении TTL. + + Позволяет разделить получение токена и его использование клиентами. + """ + + SAFETY_MARGIN = dt.timedelta(minutes=5) + + def __init__(self, client: IamClient) -> None: + self._client = client + self._token: str | None = None + self._expires_at: dt.datetime | None = None + + def invalidate(self) -> None: + """Сбросить кэш (например, после HTTP 401).""" + self._token = None + self._expires_at = None + + def is_valid(self) -> bool: + if self._token is None or self._expires_at is None: + return False + now = dt.datetime.now(dt.timezone.utc) + return self._expires_at > now + self.SAFETY_MARGIN + + async def get_token(self) -> str: + """Вернуть валидный токен, при необходимости обновив его.""" + if self.is_valid(): + return cast(str, self._token) + iam_token = await self._client.get_token() + self._token = iam_token.value + self._expires_at = iam_token.expires_at + return iam_token.value