gendesign/backend/app/services/auth_session.py

"""Резолв сессионной куки общего реестра (БД `auth`) — сторона «Птицы».

Эпик «единый вход»: вместо браузерного popup'а Caddy basic_auth у продукта одна
нейтральная форма входа. Живёт она у «Меры» (`/trade-in/login`): та проверяет
пароль, пишет строку в `auth.sessions` и ставит куку host-only на gendsgn.ru с
`path="/"` — поэтому браузер шлёт её и на `/site-finder/**` тоже.

«Птица» эту куку ТОЛЬКО ЧИТАЕТ. Здесь нет и не должно появиться `create_session` /
`revoke_session`: выдача и отзыв — исключительная ответственность единственной
формы входа, второй эмитент сессий означал бы два места, где решается «кого
пускать», и расходящиеся правила блокировки.

Что модуль отдаёт вызывающему: `resolve_session_token(token)` → `SessionUser`
(username + состояние доступа) либо None. Что делать с username дальше — дело
guard'а: авторизация «Птицы» (какие пути кому видны) по-прежнему живёт в
`auth/roles.yaml` (`app.core.auth.get_role`), продуктовые роли реестра
(`auth.users.role` — admin/manager/employee, миграция data/sql/auth/004) сюда
намеренно НЕ протаскиваются: это другая ролевая модель, и её отображение на
roles.yaml — отдельное решение стадии 2, а не побочный эффект резолва сессии.

Токены опаковые (`secrets.token_urlsafe` на стороне «Меры») — не JWT, не подписаны:
валидность проверяется исключительно наличием строки в БД + `expires_at` +
состоянием доступа юзера. Никакого разделяемого секрета между стеками для этого
не нужно — только доступ к одной БД.

Имена таблиц (`users`, `sessions`) и колонок — литералы из data/sql/auth/001 и 004;
снаружи в SQL-строку не попадает ничего, значения идут bind-параметрами.

Зеркало по подходу: tradein-mvp/backend/app/services/auth_session.py («Мера»). Там
модуль дополнительно умеет две схемы (переходный `identity_store`) и выдачу сессий —
здесь этого нет за ненадобностью.
"""

from __future__ import annotations

import logging
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from enum import StrEnum

from sqlalchemy import text
from sqlalchemy.orm import Session

from app.core import auth_db
from app.core.config import settings

logger = logging.getLogger(__name__)

# Sliding-window refresh: last_seen_at/expires_at продлеваются НЕ чаще раза в 5
# минут — иначе каждый API-запрос авторизованного юзера бил бы в БД лишним UPDATE
# (guard резолвит сессию на КАЖДЫЙ non-public запрос). Значение и механика — те же,
# что у «Меры» (tradein-mvp/.../auth_session.py:51): сессия общая, и продлевать её
# два продукта обязаны одинаково.
_SLIDING_REFRESH_INTERVAL = timedelta(minutes=5)


class AccessState(StrEnum):
    """Состояние доступа аккаунта — значения дословно из `auth.users.access_state`.

    CHECK-констрейнт `users_access_state_ck`, миграция data/sql/auth/004; семантика
    оттуда же (решение владельца от 2026-07-31):
      active        — доступ есть;
      trial_expired — пароль верный, но пробный период истёк;
      disabled      — доступ закрыт владельцем.

    Для «Птицы» все три состояния делятся надвое (`can_sign_in`): отдельный экран
    «пробный доступ закончился» — сюжет формы входа, то есть «Меры»; сюда приходит
    уже вошедший человек, и всё, что не `active`, для него значит одно — сессии нет.
    """

    ACTIVE = "active"
    TRIAL_EXPIRED = "trial_expired"
    DISABLED = "disabled"

    @property
    def can_sign_in(self) -> bool:
        """True только для `active` — единственная проверка «пускать ли».

        Вынесена в свойство, чтобы вызывающий не писал `state == "active"`: добавится
        четвёртое состояние — оно по умолчанию окажется «не пускать», а не «пускать,
        потому что не disabled».
        """
        return self is AccessState.ACTIVE


def to_access_state(value: object) -> AccessState:
    """Приводит значение колонки `users.access_state` к `AccessState`.

    Fail-closed: неизвестная строка, NULL и любой неожиданный тип → `disabled` +
    WARNING. Обратный выбор (пускать всё, что не `disabled`) означал бы, что новое
    состояние, добавленное миграцией раньше кода, молча раздаёт доступ — а миграции
    БД `auth` применяются деплоем «Птицы» (.forgejo/workflows/deploy.yml), то есть
    опередить код они могут запросто.
    """
    if isinstance(value, str):
        try:
            return AccessState(value)
        except ValueError:
            logger.warning(
                "auth_session: неизвестное состояние доступа %r → трактую как disabled", value
            )
            return AccessState.DISABLED
    logger.warning(
        "auth_session: состояние доступа %r неожиданного типа %s → трактую как disabled",
        value,
        type(value).__name__,
    )
    return AccessState.DISABLED


@dataclass(frozen=True, slots=True)
class SessionUser:
    """Кто стоит за валидной сессионной кукой.

    Attributes:
        username: логин из реестра. Именно он, а не значение куки, дальше едет в
            RBAC «Птицы» (`app.core.auth.get_role`).
        access_state: всегда `AccessState.ACTIVE` — не-active сюда не доходит
            (см. `get_session_user`). Поле оставлено явным, чтобы состояние доступа
            во всём коде называлось и выражалось одинаково, а не превращалось в
            неявное «раз объект вернулся, значит active».
    """

    username: str
    access_state: AccessState


def get_session_user(db: Session, token: str) -> SessionUser | None:
    """Резолвит сессионный токен в пользователя, или None если сессия невалидна.

    Невалидна = не найдена / истекла / состояние доступа юзера не `active`.

    Состояние доступа: пропускается ТОЛЬКО `AccessState.ACTIVE`. Любое другое
    (`disabled`, `trial_expired`, а также нераспознанное — `to_access_state`
    fail-closed'ит его в `disabled`) делает уже выданную сессию недействительной
    НЕМЕДЛЕННО, не дожидаясь `expires_at`. Иначе заблокированный человек продолжал
    бы работать до истечения TTL (до 30 дней), а sliding-refresh продлевал бы ему
    сессию бесконечно — то есть блокировка в реестре не блокировала бы ничего.

    Sliding refresh: если с последнего `last_seen_at` прошло >= 5 минут — продлевает
    `last_seen_at`/`expires_at` ОДНИМ UPDATE (ровно как «Мера»: тот же интервал, тот
    же одиночный UPDATE обеих колонок, тот же best-effort). Продлевать обе колонки
    обязательно: обновляй «Птица» только `last_seen_at`, человек, работающий весь
    день в ней одной, был бы разлогинен по `expires_at` несмотря на активность.
    Сбой refresh (напр. read-only реплика) логируется и НЕ мешает вернуть валидного
    юзера — это best-effort продление, а не часть решения «валидна ли сессия».

    Принимает уже открытую сессию БД `auth` (не открывает сам) — так модуль остаётся
    тривиально unit-тестируемым. Обычный вызывающий берёт `resolve_session_token`.

    ⚠️ `db` ОБЯЗАНА быть сессией БД `auth` (`app.core.auth_db.auth_session()`), а не
    `app.core.db.get_db`: в продуктовой БД gendesign таблиц `users`/`sessions` нет.

    Исключения БД наружу НЕ глушатся (кроме best-effort refresh): сбой реестра —
    часть auth-решения, и вызывающий обязан его увидеть, чтобы закрыться, а не
    трактовать как «сессии нет».
    """
    if not token:
        return None

    row = db.execute(
        text(
            """
            SELECT s.expires_at, s.last_seen_at, u.username, u.access_state
            FROM sessions s
            JOIN users u ON u.id = s.user_id
            WHERE s.token = :token
              AND s.expires_at > now()
            """
        ),
        {"token": token},
    ).fetchone()

    if row is None:
        return None

    now = datetime.now(UTC)
    # Второй пояс к `AND s.expires_at > now()` в SELECT'е выше. Первый пояс — часами
    # БД, и это принципиально: строку продлевает UPDATE ниже, где `expires_at =
    # now() + interval` считает СЕРВЕР. Реши мы срок годности только часами процесса
    # (`datetime.now(UTC)`), отставание этих часов давало бы не «сессия проживёт на
    # дельту дольше», а НЕОБРАТИМОЕ воскрешение: строку, которую БД уже считает
    # мёртвой, Python пропустил бы, тут же сработал бы sliding-refresh и отодвинул
    # expires_at на полный TTL от серверного now(). Секунда расхождения → +30 дней.
    # Обе стороны сравнения обязаны брать время из одного источника.
    #
    # Проверку на None оставляем первой: `expires_at` объявлен NOT NULL
    # (data/sql/auth/001), но если колонку когда-нибудь ослабят, это дешевле
    # разбирательства, почему сравнение с None упало TypeError'ом в auth-пути.
    if row.expires_at is None or row.expires_at <= now:
        return None
    access_state = to_access_state(row.access_state)
    if not access_state.can_sign_in:
        return None

    if row.last_seen_at is None or (now - row.last_seen_at) >= _SLIDING_REFRESH_INTERVAL:
        try:
            db.execute(
                text(
                    """
                    UPDATE sessions
                    SET last_seen_at = now(),
                        expires_at = now() + make_interval(hours => CAST(:ttl_hours AS integer))
                    WHERE token = :token
                    """
                ),
                {"ttl_hours": settings.session_ttl_hours, "token": token},
            )
            db.commit()
        except Exception:
            # Без username в сообщении: строка лога — не место для связки
            # «кто именно» + «в какой момент», а разбор всё равно идёт по времени.
            logger.warning("auth_session: sliding refresh failed", exc_info=True)
            try:
                db.rollback()
            except Exception:
                # Причина сбоя UPDATE'а может быть оборванным соединением — тогда и
                # rollback бросит. Без этого except «best-effort продление» переставало
                # бы быть best-effort: валидный юзер, чью сессию не удалось продлить,
                # получал бы не доступ, а исключение наружу (и в guard'е — деградацию
                # на легаси-заголовок, а в db_only — отказ).
                logger.warning("auth_session: rollback after failed refresh failed", exc_info=True)

    return SessionUser(username=row.username, access_state=access_state)


def resolve_session_token(token: str | None) -> SessionUser | None:
    """Резолвит токен сессионной куки, сам открывая соединение с БД `auth`.

    Точка входа для `rbac_guard` (`app/main.py`), который зовёт её в threadpool —
    внутри синхронный psycopg-I/O, а guard живёт на event loop'е. Возвращает None,
    если сессии нет или она недействительна.

    Режим `legacy` (`AUTH_MODE=legacy`, ДЕФОЛТ) → None СРАЗУ, без единого
    обращения к БД: инвариант «выключенный флаг = ни одного коннекта к реестру»
    держится этим модулем, а не соглашением с вызывающим. Тихий None здесь безопасен,
    потому что направлен в сторону fail-closed — он означает ровно «session-auth не
    используется», то есть сегодняшнее поведение (Caddy basic_auth + trusted-header),
    и никому ничего не открывает.

    Исключения НЕ глушатся — ни `AuthDatabaseNotConfiguredError` (флаг включён, DSN
    пуст/битый), ни ошибки соединения. Решение «что делать со сломанным реестром»
    принимает guard, и оно неочевидно: молча откатиться на trusted-header значит
    раздавать права из roles.yaml в обход реестра, включая заблокированные аккаунты.
    Прятать такое внутри резолвера нельзя.

    Raises:
        AuthDatabaseNotConfiguredError: флаг включён, а DSN БД `auth` пуст или не
            разобрался (см. `app.core.auth_db`).
    """
    if not settings.auth_session_enabled:
        return None
    if not token:
        return None
    with auth_db.auth_session() as db:
        return get_session_user(db, token)