https://pastein.ru/t/CJW
Загрузка данных
"""Санитизированный функциональный эталон мониторинга 2FA/MFA в ChatGPT.
НАЗНАЧЕНИЕ
==========
Файл показывает изолированный цикл чтения состояния MFA, создания исходного
набора ключей доступа и обнаружения последующих изменений. Это самостоятельная
справочная реализация: при установленном ``aiohttp`` HTTP-клиент и функции
мониторинга исполняемы, но подключение постоянного хранилища, диспетчера секретов,
распределённой блокировки и обработчика инцидентов оставлено вызывающей стороне.
МОДЕЛЬ УГРОЗ
============
Монитор ищет три наблюдаемых признака: MFA была выключена; действующая
авторизация несколько раз подряд получила подтверждённый структурированным
ответом отказ; после исходной фиксации появился новый passkey. Новый passkey
сравнивается с локальным списком заранее доверенных идентификаторов. Недоверенный
ключ запрашивает обработку инцидента, а любой новый ключ создаёт уведомление
``PASSKEY_DISCOVERED``.
Ответ HTTP 401 сам по себе не доказывает потерю авторизации. Причиной могут быть
промежуточный узел, временная ошибка или изменившийся формат API. Поэтому статус
``auth_failed`` разрешён только для JSON-объекта со строго распознанным кодом или
маркером. Обычная строка, HTML и пустое тело не увеличивают соответствующую
последовательность отказов.
ТОЧНОЕ ПОВЕДЕНИЕ
================
1. Первый корректный ответ создаёт исходный набор только при
``mfa_enabled is True``. Все видимые тогда passkey считаются известными.
2. ``mfa_enabled is False`` немедленно создаёт запрос инцидента
``mfa_disabled`` и не позволяет впервые активировать монитор.
3. Ровно третий последовательный ``auth_failed`` по умолчанию создаёт запрос
``session_auth_failed``. Успех или любой иной сбой обнуляет эту
последовательность.
4. Разность текущих ключей и монотонного исходного набора считается новой.
Исходный набор пополняется объединением и никогда не уменьшается.
5. Для каждого нового ключа создаётся уведомление. Только новый ключ, которого
нет в доверенном наборе, создаёт запрос ``unexpected_security_key``.
6. Незавершённые события остаются в состоянии и повторяются с тем же отпечатком.
Обработчики должны использовать отпечаток как ключ идемпотентности.
7. Владение блокировкой проверяется до запроса, после него, перед каждым
функцией обработки инцидента и непосредственно перед сохранением состояния.
8. Ускорение после успешно использованного кода сохраняет более раннее
расписание даже тогда, когда код был зарегистрирован во время HTTP-запроса.
ЧТО НЕ ВХОДИТ В ФАЙЛ
====================
Здесь нет удаления passkey, завершения всех входов, просмотра или отзыва
отдельных сессий, сброса пароля, восстановления доступа и реализации изоляции
учётной записи. Функция ``trigger_incident`` лишь передаёт безопасное событие
внешней политике. Монитор ничего не удаляет и не выполняет разрушающих действий.
КОНФИДЕНЦИАЛЬНОСТЬ
==================
- Настоящие токены, cookie и идентификаторы нельзя записывать в этот файл.
- Учётные данные должны загружаться из диспетчера секретов непосредственно перед
запросом. Их поля имеют ``repr=False``, но это только дополнительная защита.
- Нельзя журналировать ``Credentials``, заголовки, тело ответа, значения cookie,
идентификаторы факторов или полные объекты событий.
- HTTP-клиент возвращает только нормализованные поля и безопасные категории
ошибок. Тело и заголовки ответа не возвращаются и не включаются в ошибки.
- Отпечатки не содержат идентификаторов приложения. Хранилище обязано отделять
состояние разных учётных записей собственным внешним ключом области.
КОНТРАКТ ИНТЕГРАЦИИ
===================
Репозиторий должен атомарно захватывать один монитор, выдавать ограниченную по
времени блокировку и сохранять ``MonitorState`` только текущему владельцу.
``persist_state`` сохраняет исходный набор, расписание и незавершённые события в
одной транзакции. ``trigger_incident`` и ``notify`` должны быть идемпотентными по
``MonitorEvent.fingerprint``; желательно, чтобы они надёжно ставили событие во
внешнюю доставку до возврата из функции. Ошибка уведомления не отменяет
обнаружение, запрос инцидента или рост исходного набора.
Если регистрация успешного кода и цикл выполняются в разных процессах,
репозиторий должен после сетевого запроса слить более новую ревизию состояния
либо применить условное сохранение. В этой выдержке та же гонка наглядно
моделируется полями ``revision`` и ``schedule_updated_at``.
Эндпоинт ``/backend-api/accounts/mfa_info`` является закрытым веб-интерфейсом,
не публичным стабильным API. Его путь, схема, требования авторизации и семантика
могут измениться без предупреждения. Перед практической интеграцией необходимо
повторно проверить контракт и применимые правила использования.
"""
from __future__ import annotations
import hashlib
import json
import math
import re
from collections.abc import Awaitable, Callable, Mapping, Sequence
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from email.utils import parsedate_to_datetime
from typing import Any, Literal
import aiohttp
CHATGPT_ORIGIN = "https://chatgpt.com"
MFA_INFO_PATH = "/backend-api/accounts/mfa_info"
DEFAULT_FETCH_TIMEOUT_SECONDS = 20
MIN_FETCH_TIMEOUT_SECONDS = 3
MAX_FETCH_TIMEOUT_SECONDS = 25
MAX_RESPONSE_BYTES = 100_000
MAX_PASSKEYS = 100
MAX_FACTOR_ID_LENGTH = 200
AUTH_FAILURE_THRESHOLD = 3
DEFAULT_FAST_INTERVAL_SECONDS = 5
MIN_FAST_INTERVAL_SECONDS = 5
MAX_FAST_INTERVAL_SECONDS = 30
DEFAULT_SLOW_INTERVAL_SECONDS = 15
MIN_SLOW_INTERVAL_SECONDS = 10
MAX_SLOW_INTERVAL_SECONDS = 300
DEFAULT_FAST_WINDOW_SECONDS = 600
MIN_FAST_WINDOW_SECONDS = 60
MAX_FAST_WINDOW_SECONDS = 3600
FACTOR_ID_PATTERN = re.compile(r"[A-Za-z0-9._:-]{1,200}\Z")
AUTH_FAILURE_MARKERS = frozenset({
"unauthorized",
"invalid_token",
"token_expired",
"token_revoked",
"authentication_required",
"not_authenticated",
"session_expired",
})
FETCH_STATUSES = frozenset({
"ok",
"credentials_missing",
"token_expired",
"network_error",
"invalid_response",
"unsupported",
"api_error",
"auth_failed",
"integration_error",
})
INCIDENT_REASONS = frozenset({
"mfa_disabled",
"session_auth_failed",
"unexpected_security_key",
})
FetchStatus = Literal[
"ok",
"credentials_missing",
"token_expired",
"network_error",
"invalid_response",
"unsupported",
"api_error",
"auth_failed",
"integration_error",
]
EventKind = Literal["PASSKEY_DISCOVERED", "INCIDENT_REQUESTED"]
IncidentReason = Literal[
"mfa_disabled",
"session_auth_failed",
"unexpected_security_key",
]
def utc_now() -> datetime:
"""Вернуть текущую дату с часовым поясом UTC."""
return datetime.now(timezone.utc)
def parse_datetime(value: object) -> datetime | None:
"""Безопасно разобрать ISO-дату и привести её к UTC."""
if isinstance(value, datetime):
parsed = value
else:
text = str(value or "").strip()
if not text:
return None
try:
parsed = datetime.fromisoformat(text.replace("Z", "+00:00"))
except (TypeError, ValueError):
return None
if parsed.tzinfo is None:
parsed = parsed.replace(tzinfo=timezone.utc)
return parsed.astimezone(timezone.utc)
def parse_retry_after(value: str | None, now: datetime | None = None) -> int:
"""Разобрать ``Retry-After`` как секунды или дату HTTP."""
text = str(value or "").strip()
if not text:
return 0
try:
return max(0, math.ceil(float(text)))
except (TypeError, ValueError, OverflowError):
pass
try:
retry_at = parsedate_to_datetime(text)
except (TypeError, ValueError, OverflowError):
return 0
if retry_at.tzinfo is None:
retry_at = retry_at.replace(tzinfo=timezone.utc)
delay = (retry_at.astimezone(timezone.utc) - (now or utc_now())).total_seconds()
return max(0, math.ceil(delay))
def valid_factor_id(value: object) -> str:
"""Вернуть строгий идентификатор фактора или вызвать ``ValueError``."""
factor_id = str(value or "").strip()
if not FACTOR_ID_PATTERN.fullmatch(factor_id):
raise ValueError("Некорректный идентификатор ключа безопасности")
return factor_id
def event_fingerprint(*parts: object) -> str:
"""Создать стабильный отпечаток события без проектных идентификаторов."""
material = "\0".join(str(part or "") for part in parts)
return hashlib.sha256(material.encode("utf-8")).hexdigest()
@dataclass(slots=True)
class Credentials:
"""Учётные данные, получаемые во время выполнения из защищённого источника."""
account_id: str = field(default="", repr=False)
access_token: str = field(default="", repr=False)
session_token: str = field(default="", repr=False)
access_token_expires_at: datetime | None = field(default=None, repr=False)
def __post_init__(self) -> None:
if not isinstance(self.account_id, str):
raise TypeError("Идентификатор учётной записи должен быть строкой")
if not isinstance(self.access_token, str) or not isinstance(self.session_token, str):
raise TypeError("Токены должны быть строками")
if not self.access_token.strip() and not self.session_token.strip():
raise ValueError("Требуется токен доступа или токен сессии")
if self.access_token_expires_at is not None and parse_datetime(self.access_token_expires_at) is None:
raise ValueError("Некорректен срок действия токена доступа")
def access_token_is_expired(self, now: datetime | None = None) -> bool:
"""Проверить локальный срок только для присутствующего токена доступа."""
expires_at = parse_datetime(self.access_token_expires_at)
return bool(self.access_token and expires_at and expires_at <= (now or utc_now()))
@dataclass(frozen=True, slots=True)
class Passkey:
"""Строго нормализованный passkey из ответа удалённого интерфейса."""
id: str
factor_type: str = "passkey"
factor_name: str = ""
authenticator_name: str = "Ключ безопасности"
is_security_key: bool = False
is_non_device_bound: bool = False
enrollment_type: str = ""
created_at_seconds: int = 0
updated_at_seconds: int = 0
def __post_init__(self) -> None:
valid_factor_id(self.id)
if self.factor_type != "passkey":
raise ValueError("Тип фактора должен быть passkey")
if not all(isinstance(value, str) for value in (
self.factor_name,
self.authenticator_name,
self.enrollment_type,
)):
raise TypeError("Названия и тип регистрации passkey должны быть строками")
if type(self.is_security_key) is not bool or type(self.is_non_device_bound) is not bool:
raise TypeError("Признаки passkey должны иметь логический тип")
if type(self.created_at_seconds) is not int or type(self.updated_at_seconds) is not int:
raise TypeError("Метки времени passkey должны быть целыми числами")
if self.created_at_seconds < 0 or self.updated_at_seconds < 0:
raise ValueError("Метки времени passkey не могут быть отрицательными")
@dataclass(frozen=True, slots=True)
class MfaSnapshot:
"""Достоверный нормализованный снимок состояния MFA."""
mfa_enabled: bool
mfa_enabled_v2: bool | None
native_default_factor_id: str
totp_count: int
sms_count: int
passkeys: tuple[Passkey, ...]
def __post_init__(self) -> None:
if type(self.mfa_enabled) is not bool:
raise TypeError("mfa_enabled должен иметь точный логический тип")
if self.mfa_enabled_v2 is not None and type(self.mfa_enabled_v2) is not bool:
raise TypeError("mfa_enabled_v2 должен иметь логический тип или None")
if not isinstance(self.native_default_factor_id, str):
raise TypeError("native_default_factor_id должен быть строкой")
if type(self.totp_count) is not int or type(self.sms_count) is not int:
raise TypeError("Счётчики факторов должны быть целыми числами")
if self.totp_count < 0 or self.sms_count < 0:
raise ValueError("Счётчики факторов не могут быть отрицательными")
if not isinstance(self.passkeys, tuple) or not all(isinstance(item, Passkey) for item in self.passkeys):
raise TypeError("passkeys должен быть кортежем нормализованных объектов Passkey")
if len(self.passkeys) > MAX_PASSKEYS:
raise ValueError("Превышено допустимое число passkey")
if len({item.id for item in self.passkeys}) != len(self.passkeys):
raise ValueError("Снимок MFA содержит повторные идентификаторы passkey")
@dataclass(frozen=True, slots=True)
class MonitorConfig:
"""Ограниченные настройки частоты запросов и порога авторизации."""
fast_interval_seconds: int = DEFAULT_FAST_INTERVAL_SECONDS
slow_interval_seconds: int = DEFAULT_SLOW_INTERVAL_SECONDS
fast_window_seconds: int = DEFAULT_FAST_WINDOW_SECONDS
auth_failure_threshold: int = AUTH_FAILURE_THRESHOLD
fetch_timeout_seconds: int = DEFAULT_FETCH_TIMEOUT_SECONDS
def __post_init__(self) -> None:
fast = max(MIN_FAST_INTERVAL_SECONDS, min(int(self.fast_interval_seconds), MAX_FAST_INTERVAL_SECONDS))
slow = max(MIN_SLOW_INTERVAL_SECONDS, min(int(self.slow_interval_seconds), MAX_SLOW_INTERVAL_SECONDS))
window = max(MIN_FAST_WINDOW_SECONDS, min(int(self.fast_window_seconds), MAX_FAST_WINDOW_SECONDS))
timeout = max(MIN_FETCH_TIMEOUT_SECONDS, min(int(self.fetch_timeout_seconds), MAX_FETCH_TIMEOUT_SECONDS))
object.__setattr__(self, "fast_interval_seconds", fast)
object.__setattr__(self, "slow_interval_seconds", max(fast, slow))
object.__setattr__(self, "fast_window_seconds", window)
object.__setattr__(self, "auth_failure_threshold", max(1, min(int(self.auth_failure_threshold), 10)))
object.__setattr__(self, "fetch_timeout_seconds", timeout)
@dataclass(frozen=True, slots=True)
class FetchResult:
"""Безопасный результат HTTP-запроса без тела и заголовков ответа."""
ok: bool
status: FetchStatus
snapshot: MfaSnapshot | None = None
http_status: int = 0
retry_after_seconds: int = 0
error_code: str = ""
error: str = ""
def __post_init__(self) -> None:
if type(self.ok) is not bool:
raise TypeError("Признак результата должен иметь логический тип")
if self.status not in FETCH_STATUSES:
raise ValueError("Неизвестный статус запроса MFA")
if self.ok != (self.snapshot is not None):
raise ValueError("Успешный результат обязан содержать ровно один снимок")
if self.ok and self.status != "ok":
raise ValueError("Успешный результат должен иметь статус ok")
if type(self.http_status) is not int or type(self.retry_after_seconds) is not int:
raise TypeError("HTTP-статус и задержка должны быть целыми числами")
if self.http_status < 0 or self.retry_after_seconds < 0:
raise ValueError("Коды и задержки не могут быть отрицательными")
@dataclass(frozen=True, slots=True)
class MonitorEvent:
"""Идемпотентное событие обнаружения или запроса обработки инцидента."""
kind: EventKind
fingerprint: str
reason: str
message: str
passkey: Passkey | None = None
detected_at: datetime = field(default_factory=utc_now)
def __post_init__(self) -> None:
if self.kind not in {"PASSKEY_DISCOVERED", "INCIDENT_REQUESTED"}:
raise ValueError("Неизвестный вид события мониторинга")
if self.kind == "PASSKEY_DISCOVERED" and (self.reason != "PASSKEY_DISCOVERED" or self.passkey is None):
raise ValueError("Событие обнаружения обязано содержать нормализованный passkey")
if self.kind == "INCIDENT_REQUESTED" and self.reason not in INCIDENT_REASONS:
raise ValueError("Неизвестна причина запроса инцидента")
if self.reason == "unexpected_security_key" and self.passkey is None:
raise ValueError("Инцидент нового ключа обязан содержать нормализованный passkey")
if not re.fullmatch(r"[a-f0-9]{64}", self.fingerprint):
raise ValueError("Некорректный отпечаток события")
if parse_datetime(self.detected_at) is None:
raise ValueError("Некорректно время обнаружения события")
@dataclass(slots=True)
class MonitorState:
"""Постоянное состояние монитора одной внешней учётной записи."""
initialized: bool = False
baseline_factor_ids: set[str] = field(default_factory=set)
detected_factor_ids: set[str] = field(default_factory=set)
auth_failure_streak: int = 0
fetch_attempts: int = 0
last_mfa_enabled: bool | None = None
mfa_disabled_active: bool = False
mfa_disabled_epoch: int = 0
auth_failure_epoch: int = 0
event_fingerprints: set[str] = field(default_factory=set)
pending_events: dict[str, MonitorEvent] = field(default_factory=dict)
fast_until: datetime | None = None
next_check_at: datetime | None = None
schedule_updated_at: datetime | None = None
last_checked_at: datetime | None = None
last_ok_at: datetime | None = None
last_error: str = ""
revision: int = 0
def __post_init__(self) -> None:
if type(self.initialized) is not bool:
raise TypeError("Признак инициализации должен иметь логический тип")
for factor_id in self.baseline_factor_ids | self.detected_factor_ids:
valid_factor_id(factor_id)
if self.auth_failure_streak < 0 or self.fetch_attempts < 0 or self.revision < 0:
raise ValueError("Счётчики состояния не могут быть отрицательными")
if any(key != event.fingerprint for key, event in self.pending_events.items()):
raise ValueError("Ключ незавершённого события не совпадает с отпечатком")
@dataclass(frozen=True, slots=True)
class MonitoringDecision:
"""Результат чистого перехода состояния без внешних побочных эффектов."""
state: MonitorState
status: str
new_events: tuple[MonitorEvent, ...] = ()
new_passkeys: int = 0
baseline_created: bool = False
@dataclass(frozen=True, slots=True)
class CycleResult:
"""Сводка цикла, пригодная для безопасных метрик."""
status: str
new_passkeys: int = 0
emitted_events: int = 0
incidents_requested: int = 0
notification_errors: int = 0
callback_errors: int = 0
next_delay_seconds: int = 0
lease_lost: bool = False
def _safe_name(value: object, default: str, maximum: int) -> str:
text = str(value or "").strip()[:maximum]
return text or default
def _non_negative_timestamp(value: object) -> int:
try:
return max(0, int(value or 0))
except (TypeError, ValueError, OverflowError):
raise ValueError("Некорректна метка времени passkey") from None
def normalize_mfa_info(data: object) -> MfaSnapshot:
"""Строго проверить верхний уровень и нормализовать допустимые факторы.
Некорректная общая схема отклоняет весь ответ. Некорректный, повторный или
неполный объект отдельного passkey пропускается, чтобы один повреждённый
элемент не останавливал мониторинг остальных факторов.
"""
if not isinstance(data, dict) or not isinstance(data.get("factors"), dict):
raise ValueError("Ответ MFA не содержит объект factors")
if type(data.get("mfa_enabled")) is not bool:
raise ValueError("Ответ MFA не содержит точное логическое поле mfa_enabled")
if data.get("mfa_enabled_v2") is not None and type(data.get("mfa_enabled_v2")) is not bool:
raise ValueError("Поле mfa_enabled_v2 имеет некорректный тип")
factors = data["factors"]
raw_passkeys = factors.get("passkeys")
if raw_passkeys is not None and not isinstance(raw_passkeys, list):
raise ValueError("Поле factors.passkeys имеет некорректный тип")
passkeys: list[Passkey] = []
seen: set[str] = set()
for raw in raw_passkeys or []:
if not isinstance(raw, dict):
continue
try:
factor_id = valid_factor_id(raw.get("id"))
if factor_id in seen:
continue
metadata = raw.get("metadata") if isinstance(raw.get("metadata"), dict) else {}
passkey = Passkey(
id=factor_id,
factor_name=_safe_name(metadata.get("factor_name"), "", 160),
authenticator_name=_safe_name(metadata.get("authenticator_name"), "Ключ безопасности", 160),
is_security_key=bool(metadata.get("is_security_key", False)),
is_non_device_bound=bool(metadata.get("is_non_device_bound", False)),
enrollment_type=_safe_name(metadata.get("passkey_enrollment_type"), "", 80),
created_at_seconds=_non_negative_timestamp(metadata.get("create_time_seconds")),
updated_at_seconds=_non_negative_timestamp(metadata.get("update_time_seconds")),
)
except (TypeError, ValueError):
continue
seen.add(factor_id)
passkeys.append(passkey)
if len(passkeys) >= MAX_PASSKEYS:
break
return MfaSnapshot(
mfa_enabled=data["mfa_enabled"],
mfa_enabled_v2=data.get("mfa_enabled_v2"),
native_default_factor_id=str(data.get("native_default_factor_id") or "")[:MAX_FACTOR_ID_LENGTH],
totp_count=len(factors.get("totp") or []) if isinstance(factors.get("totp"), list) else 0,
sms_count=len(factors.get("sms") or []) if isinstance(factors.get("sms"), list) else 0,
passkeys=tuple(passkeys),
)
def structured_auth_failure(error_body: object) -> tuple[bool, str]:
"""Подтвердить отказ авторизации только по структурированному JSON-объекту."""
if not isinstance(error_body, dict):
return False, ""
detail = error_body.get("detail")
error = error_body.get("error")
values = (
error_body.get("message"),
error_body.get("code"),
detail if isinstance(detail, str) else detail.get("message") if isinstance(detail, dict) else "",
detail.get("code") if isinstance(detail, dict) else "",
error if isinstance(error, str) else error.get("message") if isinstance(error, dict) else "",
error.get("code") if isinstance(error, dict) else "",
)
normalized = {
re.sub(r"[^a-z0-9]+", "_", str(value or "").casefold()).strip("_")
for value in values
if str(value or "").strip()
}
exact = normalized & AUTH_FAILURE_MARKERS
if exact:
marker = sorted(exact)[0]
return True, marker
text = " ".join(str(value or "") for value in values).casefold()
phrases = {
"unauthorized": "unauthorized",
"invalid token": "invalid_token",
"token expired": "token_expired",
"token revoked": "token_revoked",
"authentication required": "authentication_required",
"not authenticated": "not_authenticated",
"session expired": "session_expired",
}
for phrase, marker in phrases.items():
if phrase in text:
return True, marker
return False, ""
class ChatGPTMfaClient:
"""Минимальный асинхронный клиент только для чтения состояния MFA."""
def __init__(self, credentials: Credentials, timeout_seconds: int = DEFAULT_FETCH_TIMEOUT_SECONDS) -> None:
self._credentials = credentials
self._timeout_seconds = max(
MIN_FETCH_TIMEOUT_SECONDS,
min(int(timeout_seconds), MAX_FETCH_TIMEOUT_SECONDS),
)
def _build_headers(self) -> dict[str, str]:
"""Сформировать заголовки в памяти непосредственно перед запросом."""
credentials = self._credentials
headers = {
"accept": "*/*",
"accept-language": "ru-RU,ru;q=0.9,en;q=0.7",
"oai-language": "ru-RU",
"referer": CHATGPT_ORIGIN + "/",
"user-agent": "Mozilla/5.0 mfa-monitor-reference/1.0",
"x-openai-target-path": MFA_INFO_PATH,
"x-openai-target-route": MFA_INFO_PATH,
}
if credentials.access_token:
headers["authorization"] = f"Bearer {credentials.access_token}"
if credentials.account_id:
headers["chatgpt-account-id"] = credentials.account_id
if credentials.session_token:
cookie = f"__Secure-next-auth.session-token={credentials.session_token}"
if credentials.account_id:
cookie += f"; _account={credentials.account_id}"
headers["cookie"] = cookie
return headers
async def fetch_mfa_info(self) -> FetchResult:
"""Получить и нормализовать снимок, не раскрывая исходный ответ."""
credentials = self._credentials
if credentials.access_token_is_expired() and not credentials.session_token:
return FetchResult(
False,
"token_expired",
error="Локально установленный срок действия токена доступа истёк",
)
timeout = aiohttp.ClientTimeout(total=self._timeout_seconds)
try:
async with aiohttp.ClientSession() as session:
async with session.get(
CHATGPT_ORIGIN + MFA_INFO_PATH,
headers=self._build_headers(),
timeout=timeout,
allow_redirects=False,
) as response:
status = int(response.status)
retry_after = parse_retry_after(response.headers.get("Retry-After"))
if status == 404:
await response.read()
return FetchResult(
False,
"unsupported",
http_status=status,
retry_after_seconds=retry_after,
error="Интерфейс чтения MFA недоступен",
)
if not 200 <= status < 300 and status != 401:
await response.read()
return FetchResult(
False,
"api_error",
http_status=status,
retry_after_seconds=retry_after,
error=f"Интерфейс MFA вернул HTTP {status}",
)
raw_body = await response.content.read(MAX_RESPONSE_BYTES + 1)
if len(raw_body) > MAX_RESPONSE_BYTES:
return FetchResult(
False,
"invalid_response" if 200 <= status < 300 else "api_error",
http_status=status,
retry_after_seconds=retry_after,
error="Ответ интерфейса MFA превышает безопасный размер",
)
try:
payload = json.loads(raw_body) if raw_body else {}
except (TypeError, ValueError, UnicodeError):
payload = None
if status == 401:
confirmed, error_code = structured_auth_failure(payload)
return FetchResult(
False,
"auth_failed" if confirmed else "api_error",
http_status=status,
retry_after_seconds=retry_after,
error_code=error_code,
error=(
"Удалённый интерфейс подтвердил отказ авторизации"
if confirmed
else "Интерфейс MFA вернул неподтверждённый HTTP 401"
),
)
except (aiohttp.ClientError, TimeoutError):
return FetchResult(
False,
"network_error",
error="Сетевой запрос состояния MFA завершился с ошибкой",
)
try:
snapshot = normalize_mfa_info(payload)
except (TypeError, ValueError):
return FetchResult(
False,
"invalid_response",
http_status=status,
error="Интерфейс MFA вернул неполный или некорректный ответ",
)
return FetchResult(True, "ok", snapshot=snapshot, http_status=status)
def _copy_state(state: MonitorState) -> MonitorState:
return MonitorState(
initialized=state.initialized,
baseline_factor_ids=set(state.baseline_factor_ids),
detected_factor_ids=set(state.detected_factor_ids),
auth_failure_streak=state.auth_failure_streak,
fetch_attempts=state.fetch_attempts,
last_mfa_enabled=state.last_mfa_enabled,
mfa_disabled_active=state.mfa_disabled_active,
mfa_disabled_epoch=state.mfa_disabled_epoch,
auth_failure_epoch=state.auth_failure_epoch,
event_fingerprints=set(state.event_fingerprints),
pending_events=dict(state.pending_events),
fast_until=state.fast_until,
next_check_at=state.next_check_at,
schedule_updated_at=state.schedule_updated_at,
last_checked_at=state.last_checked_at,
last_ok_at=state.last_ok_at,
last_error=state.last_error,
revision=state.revision,
)
def _add_event(state: MonitorState, event: MonitorEvent, target: list[MonitorEvent]) -> None:
if event.fingerprint in state.event_fingerprints:
return
state.event_fingerprints.add(event.fingerprint)
state.pending_events[event.fingerprint] = event
target.append(event)
def apply_monitoring_result(
state: MonitorState,
fetch: FetchResult,
trusted_factor_ids: Sequence[str],
checked_at: datetime,
config: MonitorConfig = MonitorConfig(),
) -> MonitoringDecision:
"""Чисто вычислить следующее состояние и события одного результата запроса."""
checked = parse_datetime(checked_at)
if checked is None:
raise ValueError("Требуется корректное время проверки")
next_state = _copy_state(state)
next_state.last_checked_at = checked
new_events: list[MonitorEvent] = []
if not fetch.ok:
next_state.fetch_attempts += 1
if fetch.status == "auth_failed":
next_state.auth_failure_streak += 1
if next_state.auth_failure_streak == config.auth_failure_threshold:
next_state.auth_failure_epoch += 1
event = MonitorEvent(
kind="INCIDENT_REQUESTED",
fingerprint=event_fingerprint("session_auth_failed", next_state.auth_failure_epoch),
reason="session_auth_failed",
message="Достигнут порог подтверждённых отказов авторизации",
detected_at=checked,
)
_add_event(next_state, event, new_events)
else:
next_state.auth_failure_streak = 0
next_state.last_error = fetch.error or "Запрос состояния MFA завершился ошибкой"
return MonitoringDecision(next_state, "fetch_failed", tuple(new_events))
snapshot = fetch.snapshot
if snapshot is None:
raise ValueError("Успешный результат не содержит снимок MFA")
next_state.fetch_attempts = 0
next_state.auth_failure_streak = 0
next_state.last_ok_at = checked
next_state.last_error = ""
next_state.last_mfa_enabled = snapshot.mfa_enabled
if snapshot.mfa_enabled is False:
if not next_state.mfa_disabled_active:
next_state.mfa_disabled_active = True
next_state.mfa_disabled_epoch += 1
event = MonitorEvent(
kind="INCIDENT_REQUESTED",
fingerprint=event_fingerprint("mfa_disabled", next_state.mfa_disabled_epoch),
reason="mfa_disabled",
message="Удалённый снимок явно сообщает, что MFA выключена",
detected_at=checked,
)
_add_event(next_state, event, new_events)
else:
next_state.mfa_disabled_active = False
current_ids = {item.id for item in snapshot.passkeys}
if not next_state.initialized:
if snapshot.mfa_enabled is not True:
return MonitoringDecision(next_state, "mfa_disabled", tuple(new_events))
next_state.initialized = True
next_state.baseline_factor_ids = current_ids
return MonitoringDecision(
next_state,
"baseline_created",
tuple(new_events),
baseline_created=True,
)
trusted: set[str] = set()
for value in trusted_factor_ids:
try:
trusted.add(valid_factor_id(value))
except ValueError:
continue
new_ids = current_ids - next_state.baseline_factor_ids
new_items = [item for item in snapshot.passkeys if item.id in new_ids]
for item in new_items:
discovered = MonitorEvent(
kind="PASSKEY_DISCOVERED",
fingerprint=event_fingerprint("PASSKEY_DISCOVERED", item.id),
reason="PASSKEY_DISCOVERED",
message="После исходной фиксации обнаружен новый passkey",
passkey=item,
detected_at=checked,
)
_add_event(next_state, discovered, new_events)
if item.id not in trusted:
incident = MonitorEvent(
kind="INCIDENT_REQUESTED",
fingerprint=event_fingerprint("unexpected_security_key", item.id),
reason="unexpected_security_key",
message="Новый passkey отсутствует в локальном доверенном наборе",
passkey=item,
detected_at=checked,
)
_add_event(next_state, incident, new_events)
next_state.baseline_factor_ids |= current_ids
next_state.detected_factor_ids |= new_ids
return MonitoringDecision(
next_state,
"completed",
tuple(new_events),
new_passkeys=len(new_items),
)
def record_successful_code(
state: MonitorState,
issued_at: datetime,
config: MonitorConfig = MonitorConfig(),
now: datetime | None = None,
) -> None:
"""Включить быстрое окно и приблизить следующий запрос после успешного кода."""
issued = parse_datetime(issued_at)
changed_at = parse_datetime(now or utc_now())
if issued is None or changed_at is None:
raise ValueError("Требуется корректное время успешного кода")
fast_until = issued + timedelta(seconds=config.fast_window_seconds)
if state.fast_until is None or fast_until > state.fast_until:
state.fast_until = fast_until
due = issued + timedelta(seconds=config.fast_interval_seconds)
if state.next_check_at is None or due < state.next_check_at:
state.next_check_at = due
state.schedule_updated_at = changed_at
state.revision += 1
def monitor_interval(state: MonitorState, now: datetime, config: MonitorConfig) -> int:
"""Выбрать быстрый или обычный интервал по активному окну."""
current = parse_datetime(now)
fast_until = parse_datetime(state.fast_until)
if current is None:
raise ValueError("Требуется корректное текущее время")
return config.fast_interval_seconds if fast_until and current < fast_until else config.slow_interval_seconds
def fetch_retry_delay(result: FetchResult, attempt: int, base_interval_seconds: int) -> int:
"""Рассчитать ограниченную задержку повтора, эквивалентную рабочим правилам."""
attempt = max(1, int(attempt))
base = max(1, int(base_interval_seconds))
if result.http_status == 429:
sequence = min(900, 30 * (2 ** min(attempt - 1, 5)))
return max(base, result.retry_after_seconds, sequence)
if result.status == "network_error" or result.http_status >= 500:
sequence = min(300, 15 * (2 ** min(attempt - 1, 4)))
return max(base, sequence)
return base
FetchMfaInfo = Callable[[], Awaitable[FetchResult]]
LeaseIsValid = Callable[[], Awaitable[bool]]
LoadTrustedFactorIds = Callable[[], Awaitable[Sequence[str]]]
PersistState = Callable[[MonitorState], Awaitable[None]]
TriggerIncident = Callable[[IncidentReason, MonitorEvent], Awaitable[None]]
Notify = Callable[[MonitorEvent], Awaitable[None]]
Heartbeat = Callable[[], Awaitable[None]]
async def _no_op() -> None:
return None
async def _no_notification(_event: MonitorEvent) -> None:
return None
def _schedule_next(
state: MonitorState,
candidate: datetime,
start_revision: int,
preserve_inflight_acceleration: bool,
) -> None:
"""Не затереть более раннее расписание, созданное во время запроса."""
if (
preserve_inflight_acceleration
and state.revision != start_revision
and state.next_check_at is not None
and state.next_check_at < candidate
):
return
state.next_check_at = candidate
state.schedule_updated_at = utc_now()
async def run_monitor_cycle(
*,
fetch_mfa_info: FetchMfaInfo,
state: MonitorState,
lease_is_valid: LeaseIsValid,
load_trusted_factor_ids: LoadTrustedFactorIds,
persist_state: PersistState,
trigger_incident: TriggerIncident,
notify: Notify = _no_notification,
heartbeat: Heartbeat = _no_op,
config: MonitorConfig = MonitorConfig(),
) -> CycleResult:
"""Выполнить один запрос, переход состояния, обратные вызовы и сохранение.
Вызывающая сторона должна уже владеть блокировкой. Ошибки уведомлений
учитываются в результате, но не отменяют инциденты и не откатывают состояние.
Незавершённое событие сохраняется и будет повторено с прежним отпечатком.
"""
if not await lease_is_valid():
return CycleResult("lease_lost", lease_lost=True)
start_revision = state.revision
try:
fetch = await fetch_mfa_info()
except Exception:
fetch = FetchResult(
False,
"network_error",
error="Адаптер чтения состояния MFA завершился с ошибкой",
)
try:
await heartbeat()
except Exception:
pass
if not await lease_is_valid():
return CycleResult("lease_lost", lease_lost=True)
trusted: Sequence[str] = ()
effective_fetch = fetch
if fetch.ok:
try:
trusted = await load_trusted_factor_ids()
except Exception:
effective_fetch = FetchResult(
False,
"integration_error",
error="Не удалось загрузить доверенные идентификаторы факторов",
)
checked_at = utc_now()
decision = apply_monitoring_result(state, effective_fetch, trusted, checked_at, config)
next_state = decision.state
interval = monitor_interval(next_state, checked_at, config)
if effective_fetch.ok:
delay = interval
else:
delay = fetch_retry_delay(effective_fetch, next_state.fetch_attempts, interval)
_schedule_next(
next_state,
checked_at + timedelta(seconds=delay),
start_revision,
preserve_inflight_acceleration=effective_fetch.http_status != 429,
)
incidents_requested = 0
callback_errors = 0
notification_errors = 0
pending = list(next_state.pending_events.values())
for event in pending:
if event.kind != "INCIDENT_REQUESTED":
continue
if not await lease_is_valid():
return CycleResult("lease_lost", lease_lost=True)
try:
await trigger_incident(event.reason, event) # type: ignore[arg-type]
except Exception:
callback_errors += 1
else:
incidents_requested += 1
next_state.pending_events.pop(event.fingerprint, None)
for event in pending:
if event.kind != "PASSKEY_DISCOVERED":
continue
try:
await notify(event)
except Exception:
notification_errors += 1
else:
next_state.pending_events.pop(event.fingerprint, None)
if not await lease_is_valid():
return CycleResult("lease_lost", lease_lost=True)
next_state.revision += 1
await persist_state(next_state)
return CycleResult(
status=decision.status,
new_passkeys=decision.new_passkeys,
emitted_events=len(decision.new_events),
incidents_requested=incidents_requested,
notification_errors=notification_errors,
callback_errors=callback_errors,
next_delay_seconds=delay,
)
# Схема интеграции ниже намеренно не является исполняемым кодом.
#
# lease, state = await repository.claim_due_monitor()
# if lease:
# credentials = await secret_manager.load_credentials(lease.secret_reference)
# client = ChatGPTMfaClient(credentials)
# result = await run_monitor_cycle(
# fetch_mfa_info=client.fetch_mfa_info,
# state=state,
# lease_is_valid=lambda: repository.lease_is_valid(lease),
# load_trusted_factor_ids=lambda: repository.load_trusted_factor_ids(lease),
# persist_state=lambda value: repository.save_if_revision_owned(lease, value),
# trigger_incident=lambda reason, event: incident_handler.request_idempotently(
# reason=reason, fingerprint=event.fingerprint,
# ),
# notify=lambda event: notifier.deliver_idempotently(
# event_type=event.kind, fingerprint=event.fingerprint,
# ),
# heartbeat=lambda: repository.refresh_lease(lease),
# )
#
# При отдельном процессе регистрации кода репозиторий должен атомарно вызвать
# record_successful_code и увеличить revision. После HTTP-запроса цикл обязан
# перечитать или слить эту ревизию перед условным сохранением, чтобы не заменить
# ранний next_check_at более поздним обычным расписанием.
