https://pastein.ru/t/sJW
Загрузка данных
"""Санитизированная эталонная реализация автоматического отзыва сессий ChatGPT.
НАЗНАЧЕНИЕ
==========
В этом файле показан изолированный алгоритм обнаружения и отзыва неожиданной
сессии ChatGPT. Он намеренно не зависит от исходного приложения: здесь нет
моделей базы данных, идентификаторов арендаторов, меток учётных записей, адресов
электронной почты, рабочих URL, настроек прокси, ключей шифрования, уведомлений
или настоящих токенов.
Это эталонная выдержка, а не готовая замена полной службы. Два используемых ниже
эндпоинта ChatGPT являются внутренними веб-эндпоинтами и могут измениться без
предупреждения. Перед интеграцией этого кода необходимо проверить актуальное
поведение API, правила авторизации и применимые условия использования.
ЧТО ОЗНАЧАЕТ «ПРИНУДИТЕЛЬНО ЗАВЕРШИТЬ СЕССИЮ»
=============================================
Под принудительным завершением понимается отзыв одной конкретной сессии по её
идентификатору:
POST /backend-api/accounts/sessions/revoke
JSON: {"session_id": "..."}
Это НЕ означает выход из всех сессий, сброс пароля, удаление ключей доступа,
доступ к почтовому ящику или восстановление учётной записи. Такие операции
намеренно не входят в область ответственности этого файла.
РЕШЕНИЯ ПО БЕЗОПАСНОСТИ
=======================
1. Первый успешный запрос, давший достоверные данные, формирует базовый набор.
Существующие сессии не отзываются при первом включении защиты.
2. Сессия, уже присутствующая в базовом наборе, сохраняется.
3. Текущая сессия обработчика или браузера всегда сохраняется. Её отзыв может
лишить автоматизацию доступа к учётной записи.
4. Новая сессия сохраняется, если ровно одно событие успешной доставки TOTP-кода
соответствует времени входа и это событие соответствует ровно одной новой
сессии.
5. Сессия без пригодной метки времени входа направляется на ручную проверку.
6. Новая сессия без однозначного подтверждения TOTP-кодом отзывается немедленно.
7. За один цикл отзывается не более MAX_REVOKES_PER_CYCLE сессий.
8. Сессия после неудачного отзыва не добавляется в базовый набор, поэтому
следующая попытка выполняется в одном из последующих циклов.
Поле API ``is_trusted_device`` намеренно НЕ используется как правило авторизации.
Это метаданные устройства от удалённой службы, а не локальное подтверждение того,
что сессия была одобрена.
ТРЕБОВАНИЯ К ИНТЕГРАЦИИ
=======================
Вызывающая сторона должна предоставить постоянное хранилище и распределённую
блокировку. Перед каждым разрушающим отзывом снова вызывается
``lease_is_valid``. Реализация блокировки должна исключать одновременную
обработку одной учётной записи двумя обработчиками и прекращать владение
устаревшего владельца через ограниченный интервал.
События TOTP следует регистрировать только после успешной доставки кода
предполагаемому получателю. Они загружаются повторно после HTTP-запроса сессий,
что снижает вероятность состояния гонки, когда код доставляется во время
получения списка сессий.
``persist_state`` должен атомарно сохранять базовый набор и счётчики, пока
владение той же блокировкой сохраняется. Доставка уведомлений должна быть
идемпотентной и никогда не должна откатывать успешный удалённый отзыв.
ОБРАЩЕНИЕ С СЕКРЕТАМИ
=====================
- Никогда не задавайте учётные данные непосредственно в этом файле.
- Загружайте учётные данные из диспетчера секретов во время выполнения.
- Никогда не записывайте в журнал экземпляры Credentials, заголовки запросов,
файлы cookie, тела ответов, идентификаторы сессий и устройств, сведения о
местоположении или метки времени событий TOTP.
- В полях учётных данных используется ``repr=False``, чтобы снизить риск
случайного раскрытия, но это не заменяет надёжное хранение секретов и
фильтрацию журналов.
- Не передавайте этот файл кому-либо после добавления в него настоящих учётных
данных.
ЗАВИСИМОСТЬ
===========
HTTP-адаптер использует ``aiohttp``. Классификатор и модель состояния используют
только стандартную библиотеку Python.
"""
from __future__ import annotations
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"
SESSIONS_PATH = "/backend-api/accounts/sessions"
REVOKE_PATH = "/backend-api/accounts/sessions/revoke"
CODE_MATCH_BEFORE_SECONDS = 60
CODE_MATCH_AFTER_SECONDS = 60
MAX_REVOKES_PER_CYCLE = 5
FETCH_TIMEOUT_SECONDS = 25
REVOKE_TIMEOUT_SECONDS = 20
MAX_SESSION_ID_LENGTH = 220
DecisionName = Literal[
"baseline",
"current_device",
"authorized_by_code",
"manual_review",
"revoke",
]
def utc_now() -> datetime:
"""Вернуть дату и время с часовым поясом UTC; все сравнения используют UTC."""
return datetime.now(timezone.utc)
def parse_datetime(value: object) -> datetime | None:
"""Разобрать дату и время в формате ISO и привести их к UTC.
Значения без часового пояса считаются заданными в UTC. Для некорректных или
пустых входных данных возвращается None, поскольку предполагать метку времени
при принятии решения о разрушающем действии небезопасно.
"""
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, int(float(text)))
except ValueError:
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)
seconds = (retry_at.astimezone(timezone.utc) - (now or utc_now())).total_seconds()
return max(0, int(seconds))
@dataclass(slots=True)
class Credentials:
"""Учётные данные, используемые HTTP-адаптером только во время выполнения.
Необходимо передать хотя бы одно из значений access_token или session_token.
``account_id`` является значением заголовка учётной записи ChatGPT, а не
идентификатором приложения или базы данных. Настоящие значения должны
поступать из диспетчера секретов и никогда не должны сериализоваться вместе
с состоянием защиты.
"""
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 validate(self) -> None:
if not self.access_token and not self.session_token:
raise ValueError("Требуется токен доступа ChatGPT или токен сессии")
def access_token_is_expired(self, now: datetime | None = None) -> bool:
expires_at = parse_datetime(self.access_token_expires_at)
return bool(expires_at and expires_at <= (now or utc_now()))
@dataclass(frozen=True, slots=True)
class SessionDevice:
"""Минимальные метаданные сессии, необходимые для решения о защите.
``context_id`` содержит необязательные локальные метаданные. Он может
идентифицировать получателя или рабочий процесс, получивший TOTP-код, но не
должен содержать пароль или токен. ``is_trusted_device`` сохраняется только
для отображения и аудита и никогда не используется для авторизации новой
сессии.
"""
session_id: str
signed_in_at: datetime | None
is_current_device: bool
is_trusted_device: bool
display_name: str = "Устройство"
context_id: str = ""
@dataclass(frozen=True, slots=True)
class CodeEvent:
"""Подтверждение успешной доставки TOTP-кода.
Создавайте это событие только после успешной доставки. ``event_id`` должен
быть стабильным несекретным идентификатором для дедупликации. ``context_id``
может быть идентификатором заказа, рабочего процесса или получателя; сессии
с пустым контекстом может соответствовать любое событие.
"""
event_id: str
issued_at: datetime
context_id: str = ""
@dataclass(frozen=True, slots=True)
class SessionDecision:
"""Результат чистого классификатора для одной видимой сейчас сессии."""
session: SessionDevice
decision: DecisionName
reason: str
matched_event_id: str = ""
@dataclass(frozen=True, slots=True)
class Classification:
"""Все решения и идентификаторы, безопасные для следующего базового набора."""
decisions: tuple[SessionDecision, ...]
next_baseline_ids: frozenset[str]
def selected(self, decision: DecisionName) -> list[SessionDecision]:
return [item for item in self.decisions if item.decision == decision]
@dataclass(slots=True)
class ProtectionState:
"""Постоянное состояние одной физической учётной записи ChatGPT.
Рабочая реализация репозитория должна хранить этот объект по ключу в области
учётной записи и атомарно обновлять его при владении блокировкой.
Идентификаторы сессий после неудачного отзыва и сессий на ручной проверке
намеренно не входят в ``baseline_ids``.
"""
initialized: bool = False
baseline_ids: set[str] = field(default_factory=set)
fetch_attempts: int = 0
revoked_count: int = 0
failed_count: int = 0
last_checked_at: datetime | None = None
last_ok_at: datetime | None = None
last_action_at: datetime | None = None
last_action_status: str = ""
last_error: str = ""
next_check_at: datetime | None = None
@dataclass(frozen=True, slots=True)
class FetchResult:
"""Результат запроса списка сессий без конфиденциальных данных."""
ok: bool
devices: tuple[Mapping[str, Any], ...] = ()
status: str = ""
http_status: int = 0
retry_after_seconds: int = 0
error: str = ""
@dataclass(frozen=True, slots=True)
class RevokeResult:
"""Результат запроса отзыва одной сессии без конфиденциальных данных."""
ok: bool
http_status: int = 0
already_absent: bool = False
retry_after_seconds: int = 0
error: str = ""
@dataclass(frozen=True, slots=True)
class CycleResult:
"""Сводка для метрик и журналов без материалов учётных данных."""
status: str
checked: int = 0
baselined: int = 0
authorized: int = 0
manual_review: int = 0
revoke_candidates: int = 0
revoked: int = 0
failed: int = 0
next_delay_seconds: int = 0
lease_lost: bool = False
def normalize_session(raw: Mapping[str, Any]) -> SessionDevice | None:
"""Преобразовать объект удалённого устройства в минимальную внутреннюю модель.
Со временем API использовал несколько полей идентификатора, поэтому
предпочтение отдаётся session_id, а hashed_device_id и render_id используются
как резервные варианты. Объект без стабильного идентификатора отклоняется.
"""
session_id = str(
raw.get("session_id")
or raw.get("hashed_device_id")
or raw.get("render_id")
or ""
).strip()[:MAX_SESSION_ID_LENGTH]
if not session_id:
return None
signed_in_at = None
try:
timestamp = int(raw.get("last_signed_in_timestamp_second") or 0)
except (TypeError, ValueError, OverflowError):
timestamp = 0
if timestamp > 0:
try:
signed_in_at = datetime.fromtimestamp(timestamp, tz=timezone.utc)
except (OSError, OverflowError, ValueError):
signed_in_at = None
if signed_in_at is None:
signed_in_at = parse_datetime(raw.get("last_signed_in_at"))
display_name = str(
raw.get("display_name")
or raw.get("device_model")
or raw.get("platform")
or "Устройство"
).strip()[:120]
return SessionDevice(
session_id=session_id,
signed_in_at=signed_in_at,
is_current_device=bool(raw.get("is_current_device", False)),
is_trusted_device=bool(raw.get("is_trusted_device", False)),
display_name=display_name or "Устройство",
context_id=str(raw.get("context_id") or raw.get("order_id") or "").strip()[:80],
)
def code_event_matches(session: SessionDevice, event: CodeEvent) -> bool:
"""Вернуть True, если событие кода подходит по времени и контексту."""
if session.signed_in_at is None:
return False
if session.context_id and event.context_id != session.context_id:
return False
before = timedelta(seconds=CODE_MATCH_BEFORE_SECONDS)
after = timedelta(seconds=CODE_MATCH_AFTER_SECONDS)
return event.issued_at - before <= session.signed_in_at <= event.issued_at + after
def classify_sessions(
sessions: Sequence[SessionDevice],
baseline_ids: set[str] | frozenset[str],
code_events: Sequence[CodeEvent],
) -> Classification:
"""Классифицировать сессии без побочных эффектов в сети, БД или уведомлениях.
Авторизация намеренно требует взаимно-однозначного соответствия. Если одно
событие может объяснить две сессии или одну сессию можно объяснить двумя
событиями, неоднозначность не разрешается автоматически. Затронутая сессия
направляется на ручную проверку.
"""
baseline = {str(value).strip() for value in baseline_ids if str(value).strip()}
unique_sessions = {item.session_id: item for item in sessions if item.session_id}
ordered = sorted(
unique_sessions.values(),
key=lambda item: (item.signed_in_at or datetime.min.replace(tzinfo=timezone.utc), item.session_id),
)
matches_by_session: dict[str, list[CodeEvent]] = {}
session_ids_by_event: dict[str, set[str]] = {}
for session in ordered:
if session.session_id in baseline or session.is_current_device or session.signed_in_at is None:
continue
matches = [event for event in code_events if code_event_matches(session, event)]
matches_by_session[session.session_id] = matches
for event in matches:
session_ids_by_event.setdefault(event.event_id, set()).add(session.session_id)
decisions: list[SessionDecision] = []
next_baseline: set[str] = set()
for session in ordered:
if session.session_id in baseline:
next_baseline.add(session.session_id)
decisions.append(SessionDecision(
session,
"baseline",
"Сессия уже присутствовала в доверенном базовом наборе",
))
continue
if session.is_current_device:
next_baseline.add(session.session_id)
decisions.append(SessionDecision(
session,
"current_device",
"Текущая сессия обработчика или браузера никогда не отзывается",
))
continue
matches = matches_by_session.get(session.session_id, [])
if matches:
matches.sort(key=lambda event: abs((session.signed_in_at - event.issued_at).total_seconds()))
closest = matches[0]
one_event_for_session = len(matches) == 1
one_session_for_event = len(session_ids_by_event.get(closest.event_id, set())) == 1
if one_event_for_session and one_session_for_event:
next_baseline.add(session.session_id)
decisions.append(SessionDecision(
session,
"authorized_by_code",
"Ровно одно событие доставки TOTP соответствует ровно одной новой сессии",
matched_event_id=closest.event_id,
))
else:
decisions.append(SessionDecision(
session,
"manual_review",
"Данные TOTP неоднозначны; автоматический отзыв намеренно пропущен",
))
continue
if session.signed_in_at is None:
decisions.append(SessionDecision(
session,
"manual_review",
"Нет достоверной метки времени входа; автоматическое разрушающее действие небезопасно",
))
continue
decisions.append(SessionDecision(
session,
"revoke",
"У новой сессии нет однозначного подтверждения доставленным TOTP-кодом",
))
return Classification(tuple(decisions), frozenset(next_baseline))
class ChatGPTSessionClient:
"""Небольшой HTTP-адаптер для получения и отзыва сессий ChatGPT.
Тела ответов никогда не включаются в объекты ошибок, поскольку могут
содержать данные учётной записи или устройства. Вызывающая сторона должна
записывать в журнал только категории статусов и ошибок.
"""
def __init__(self, credentials: Credentials) -> None:
credentials.validate()
self._credentials = credentials
def _headers(self, target_path: str) -> dict[str, str]:
credentials = self._credentials
headers = {
"accept": "*/*",
"accept-language": "en-US,en;q=0.9",
"oai-language": "en-US",
"referer": CHATGPT_ORIGIN + "/",
"user-agent": "Mozilla/5.0 session-protection-reference/1.0",
"x-openai-target-path": target_path,
"x-openai-target-route": target_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_sessions(self) -> FetchResult:
"""Получить достоверный список сессий.
Пустой список или явно заданное ``show_session_manager=false`` считается
признаком недоступности, а не достоверным результатом. Это не позволяет
временной проблеме API стереть базовый набор и изменить решения о доверии.
"""
if self._credentials.access_token_is_expired():
return FetchResult(False, status="token_expired", error="Срок действия токена доступа истёк")
timeout = aiohttp.ClientTimeout(total=FETCH_TIMEOUT_SECONDS)
try:
async with aiohttp.ClientSession() as http:
async with http.get(
CHATGPT_ORIGIN + SESSIONS_PATH,
headers=self._headers(SESSIONS_PATH),
timeout=timeout,
) as response:
status = int(response.status)
retry_after = parse_retry_after(response.headers.get("Retry-After"))
try:
payload = await response.json(content_type=None)
except (aiohttp.ContentTypeError, ValueError):
payload = None
except (aiohttp.ClientError, TimeoutError):
return FetchResult(False, status="network_error", error="Запрос списка сессий завершился с ошибкой")
if not 200 <= status < 300:
return FetchResult(
False,
status="api_error",
http_status=status,
retry_after_seconds=retry_after,
error=f"Эндпоинт списка сессий вернул HTTP {status}",
)
if not isinstance(payload, dict):
return FetchResult(
False,
status="invalid_response",
http_status=status,
error="Ответ не является объектом",
)
devices = payload.get("devices")
manager_explicitly_disabled = (
"show_session_manager" in payload and payload.get("show_session_manager") is not True
)
if not isinstance(devices, list) or not devices or manager_explicitly_disabled:
return FetchResult(
False,
status="session_manager_unavailable",
http_status=status,
error="Диспетчер сессий пуст или недоступен",
)
safe_devices = tuple(item for item in devices if isinstance(item, Mapping))
if not safe_devices:
return FetchResult(
False,
status="invalid_response",
http_status=status,
error="В ответе нет корректных объектов устройств",
)
return FetchResult(True, devices=safe_devices, status="ok", http_status=status)
async def revoke_session(self, session_id: str) -> RevokeResult:
"""Отозвать ровно одну сессию; HTTP 404 означает, что она уже отсутствует."""
clean_id = str(session_id or "").strip()[:MAX_SESSION_ID_LENGTH]
if not clean_id:
return RevokeResult(False, error="Требуется идентификатор сессии")
if self._credentials.access_token_is_expired():
return RevokeResult(False, error="Срок действия токена доступа истёк")
headers = self._headers(REVOKE_PATH)
headers["content-type"] = "application/json"
timeout = aiohttp.ClientTimeout(total=REVOKE_TIMEOUT_SECONDS)
try:
async with aiohttp.ClientSession() as http:
async with http.post(
CHATGPT_ORIGIN + REVOKE_PATH,
json={"session_id": clean_id},
headers=headers,
timeout=timeout,
) as response:
status = int(response.status)
retry_after = parse_retry_after(response.headers.get("Retry-After"))
await response.read()
except (aiohttp.ClientError, TimeoutError):
return RevokeResult(False, error="Запрос на отзыв сессии завершился с ошибкой")
if 200 <= status < 300:
return RevokeResult(True, http_status=status)
if status == 404:
return RevokeResult(True, http_status=status, already_absent=True)
return RevokeResult(
False,
http_status=status,
retry_after_seconds=retry_after,
error=f"Эндпоинт отзыва сессии вернул HTTP {status}",
)
LoadCodeEvents = Callable[[], Awaitable[Sequence[CodeEvent]]]
PersistState = Callable[[ProtectionState], Awaitable[None]]
LeaseIsValid = Callable[[], Awaitable[bool]]
Heartbeat = Callable[[], Awaitable[None]]
Notify = Callable[[str, SessionDevice, str], Awaitable[None]]
async def _no_op() -> None:
return None
async def _no_notification(_status: str, _session: SessionDevice, _error: str) -> None:
return None
def fetch_retry_delay(
result: FetchResult,
attempt: int,
normal_interval_seconds: int,
) -> int:
"""Рассчитать ограниченную задержку повтора после сбоя запроса сессий."""
attempt = max(1, int(attempt))
normal = max(1, int(normal_interval_seconds))
if result.http_status == 429:
exponential = min(900, 30 * (2 ** min(attempt - 1, 5)))
return max(normal, result.retry_after_seconds, exponential)
if result.status == "network_error" or result.http_status >= 500:
exponential = min(300, 15 * (2 ** min(attempt - 1, 4)))
return max(normal, exponential)
return normal
async def run_protection_cycle(
*,
client: ChatGPTSessionClient,
state: ProtectionState,
load_code_events: LoadCodeEvents,
persist_state: PersistState,
lease_is_valid: LeaseIsValid,
slow_interval_seconds: int = 30,
fast_interval_seconds: int = 5,
heartbeat: Heartbeat = _no_op,
notify: Notify = _no_notification,
) -> CycleResult:
"""Выполнить полный цикл получения, классификации, отзыва и сохранения.
Вызывающая сторона уже должна владеть блокировкой. Состояние сохраняется при
обычном успехе и сбое получения данных, но никогда не сохраняется после
потери владения блокировкой. Уведомления доставляются по мере возможности и
не влияют на результат отзыва.
"""
if not await lease_is_valid():
return CycleResult(status="lease_lost", lease_lost=True)
checked_at = utc_now()
fetch = await client.fetch_sessions()
await heartbeat()
state.last_checked_at = checked_at
if not fetch.ok:
state.fetch_attempts += 1
delay = fetch_retry_delay(fetch, state.fetch_attempts, slow_interval_seconds)
state.last_error = fetch.error or fetch.status
state.next_check_at = checked_at + timedelta(seconds=delay)
if not await lease_is_valid():
return CycleResult(status="lease_lost", lease_lost=True)
await persist_state(state)
return CycleResult(status="fetch_failed", next_delay_seconds=delay)
normalized = [normalize_session(raw) for raw in fetch.devices]
sessions = [item for item in normalized if item is not None]
current_ids = {item.session_id for item in sessions}
if not state.initialized:
state.initialized = True
state.baseline_ids = current_ids
state.fetch_attempts = 0
state.last_ok_at = checked_at
state.last_action_at = checked_at
state.last_action_status = "baseline"
state.last_error = ""
state.next_check_at = checked_at + timedelta(seconds=max(1, slow_interval_seconds))
if not await lease_is_valid():
return CycleResult(status="lease_lost", lease_lost=True)
await persist_state(state)
return CycleResult(
status="baseline_created",
checked=len(sessions),
baselined=len(current_ids),
next_delay_seconds=max(1, slow_interval_seconds),
)
# Повторная загрузка после HTTP-запроса учитывает коды, доставленные во время запроса.
code_events = list(await load_code_events())
classification = classify_sessions(sessions, state.baseline_ids, code_events)
revoke_candidates = classification.selected("revoke")
manual_review = classification.selected("manual_review")
authorized = classification.selected("authorized_by_code")
revoked_ids: set[str] = set()
revoked = 0
failed = 0
attempted = 0
for decision in revoke_candidates[:MAX_REVOKES_PER_CYCLE]:
if not await lease_is_valid():
return CycleResult(status="lease_lost", lease_lost=True)
attempted += 1
result = await client.revoke_session(decision.session.session_id)
await heartbeat()
if result.ok:
revoked_ids.add(decision.session.session_id)
revoked += 1
try:
await notify("revoked", decision.session, "")
except Exception:
pass
else:
failed += 1
try:
await notify("revoke_failed", decision.session, result.error)
except Exception:
pass
for decision in manual_review:
try:
await notify("manual_review", decision.session, decision.reason)
except Exception:
pass
if not await lease_is_valid():
return CycleResult(status="lease_lost", lease_lost=True)
# Идентификаторы отозванных сессий, неудачных отзывов, сессий на ручной проверке
# и ещё не обработанных кандидатов не считаются доверенными. Здесь остаются
# только базовые, текущие и авторизованные кодом сессии.
state.baseline_ids = set(classification.next_baseline_ids & current_ids) - revoked_ids
state.fetch_attempts = 0
state.revoked_count += revoked
state.failed_count += failed
state.last_ok_at = checked_at
state.last_error = f"Количество сессий, которые не удалось отозвать: {failed}" if failed else ""
remaining_candidates = len(revoke_candidates) - attempted
needs_fast_retry = bool(failed or manual_review or remaining_candidates > 0)
delay = max(1, fast_interval_seconds if needs_fast_retry else slow_interval_seconds)
state.next_check_at = utc_now() + timedelta(seconds=delay)
if revoked or failed:
state.last_action_at = checked_at
state.last_action_status = "revoke_failed" if failed else "revoked"
elif manual_review:
state.last_action_at = checked_at
state.last_action_status = "manual_review"
await persist_state(state)
return CycleResult(
status="completed",
checked=len(sessions),
authorized=len(authorized),
manual_review=len(manual_review),
revoke_candidates=len(revoke_candidates),
revoked=revoked,
failed=failed,
next_delay_seconds=delay,
)
# Минимальная схема интеграции (намеренно не является исполняемым кодом):
#
# lease, state = await repository.claim_due_monitor(stale_after_seconds=180)
# if lease:
# credentials = await secret_manager.get_chatgpt_credentials(lease.account_ref)
# client = ChatGPTSessionClient(credentials)
# result = await run_protection_cycle(
# client=client,
# state=state,
# load_code_events=lambda: repository.load_delivered_code_events(lease),
# persist_state=lambda value: repository.save_if_owned(lease, value),
# lease_is_valid=lambda: repository.lease_is_owned(lease),
# heartbeat=lambda: repository.refresh_lease(lease),
# notify=lambda status, session, error: notifier.send_idempotently(
# status=status,
# session_fingerprint=hash_for_notification(session.session_id),
# error_category=error,
# ),
# )
