ipam_control/app/api/v1/auth.py

"""Вход в UI. Перебор паролей ограничен тремя областями (окно {WINDOW_MIN} минут → 429, изменения 012, 026, 027):
{MAX_PER_LOGIN_IP} неудач на пару логин+IP, {MAX_PER_IP} на IP (любые логины), {MAX_PER_LOGIN} на логин со всех IP, кроме «известных»
(IP, с которого этот пользователь успешно входил за последние {KNOWN_IP_DAYS} дн.) — иначе анонимный клиент, знающий логин,
мог бы держать чужую учётную запись заблокированной с любого IP (находка №2 ревью 2026-09-26)."""
from datetime import datetime, timedelta, timezone

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy import and_, delete, func, select
from sqlalchemy.dialects.postgresql import insert
from sqlalchemy.orm import Session

from app.db import get_db
from app.models import KnownLogin, LoginAttempt, User
from app.request_context import request_meta
from app.schemas import LoginIn, TokenOut, UserOut
from app.security import create_token, current_user, hash_password, verify_password
from app.services import ANONYMOUS, KNOWN_IP_DAYS, LOGIN_WINDOW, MAX_PER_IP, MAX_PER_LOGIN, MAX_PER_LOGIN_IP, audit

router = APIRouter(prefix="/auth", tags=["auth"])
# пороги перебора (LOGIN_WINDOW, MAX_PER_*, KNOWN_IP_DAYS) — в app.services (изменение 030, ревью 025-029 находка №2)
# пространства ключей двухаргументного advisory-lock (не пересекаются с одноаргументными: LOCK_KEY/USERS_ADMIN_LOCK)
LOGIN_LOCK_NS = 7031
IP_LOCK_NS = 7032
__doc__ = __doc__.format(MAX_PER_LOGIN_IP=MAX_PER_LOGIN_IP, MAX_PER_IP=MAX_PER_IP, MAX_PER_LOGIN=MAX_PER_LOGIN,
                          WINDOW_MIN=int(LOGIN_WINDOW.total_seconds() // 60), KNOWN_IP_DAYS=KNOWN_IP_DAYS)
_DUMMY_HASH = hash_password("dummy-password-for-timing")  # выравнивает время ответа для несуществующего логина


def _condition(login: str, ip: str | None, scope: str):
    if scope == "login_ip":
        return and_(LoginAttempt.username == login, LoginAttempt.client_ip == ip)
    if scope == "ip":
        return LoginAttempt.client_ip == ip
    return LoginAttempt.username == login  # "login" — по всем IP


def _is_known(db: Session, login: str, ip: str | None) -> bool:
    """IP считается известным для логина, если вход с него был успешен не более KNOWN_IP_DAYS дней назад."""
    if ip is None:
        return False
    cutoff = datetime.now(timezone.utc) - timedelta(days=KNOWN_IP_DAYS)
    return db.scalar(select(KnownLogin.username).where(KnownLogin.username == login, KnownLogin.client_ip == ip, KnownLogin.last_seen > cutoff)) is not None


def _retry_after(db: Session, login: str, ip: str | None, known: bool) -> tuple[int, str | None, dict[str, int]]:
    """(секунд до конца блокировки, область 'login_ip'|'ip'|'login', счётчики по каждой области, до лимита).
    Блокировка длится окно после последней неудачи. Область 'login' не блокирует, если IP запроса известен
    (изменение 026, находка №2); 'login_ip' и 'ip' действуют всегда. Счётчики считаются отдельно (изменение 024,
    находка №4): решение о записи в журнал принимается по конкретной области, а не по максимуму среди них."""
    now = datetime.now(timezone.utc)
    worst, why, counts = 0, None, {}
    for scope, limit in (("login_ip", MAX_PER_LOGIN_IP), ("ip", MAX_PER_IP), ("login", MAX_PER_LOGIN)):
        if scope in ("login_ip", "ip") and ip is None:
            continue
        rows = db.scalars(select(LoginAttempt.ts).where(_condition(login, ip, scope), LoginAttempt.ts > now - LOGIN_WINDOW).order_by(LoginAttempt.ts.desc()).limit(limit)).all()
        counts[scope] = len(rows)
        if len(rows) >= limit and not (scope == "login" and known):
            left = int((rows[0] + LOGIN_WINDOW - now).total_seconds()) + 1
            if left > worst:
                worst, why = left, scope
    return worst, why, counts


def _distinct_logins(db: Session, ip: str) -> int:
    """Число различных логинов, для которых была неудачная попытка с этого IP в окне (для diff записи session.locked)."""
    now = datetime.now(timezone.utc)
    return db.scalar(select(func.count(func.distinct(LoginAttempt.username))).where(LoginAttempt.client_ip == ip, LoginAttempt.ts > now - LOGIN_WINDOW)) or 0


def _remember_login(db: Session, login: str, ip: str | None) -> None:
    """Отмечает IP как известный для логина (upsert last_seen); вызывается при успешном входе."""
    if ip is None:
        return
    stmt = insert(KnownLogin).values(username=login, client_ip=ip, last_seen=func.now())
    db.execute(stmt.on_conflict_do_update(index_elements=[KnownLogin.username, KnownLogin.client_ip], set_={"last_seen": func.now()}))


def _locked(retry: int) -> HTTPException:
    return HTTPException(429, {"message": f"Слишком много неудачных попыток входа. Повторите через {max(1, -(-retry // 60))} мин.", "retry_after_seconds": retry},
                         headers={"Retry-After": str(retry)})


@router.post("/login", response_model=TokenOut)
def login(body: LoginIn, db: Session = Depends(get_db)):
    ctx = request_meta.get()
    ip = ctx["client_ip"] if ctx else None
    name = body.username.strip().lower()
    # сериализация попыток одного логина (изменение 024, находка №1), затем — этого IP (изменение 027, находка №7):
    # порядок всегда «логин, затем IP» исключает взаимную блокировку; без второй блокировки параллельный перебор
    # разных логинов с одного IP проходит проверку лимита по IP одновременно, и лимит превышается на степень
    # параллелизма — ценой служит то, что попытки с одного IP (в том числе за NAT) обрабатываются по одной,
    # и при массовом входе время ответа растёт на время проверки argon2.
    db.execute(select(func.pg_advisory_xact_lock(LOGIN_LOCK_NS, func.hashtext(name))))
    if ip is not None:
        db.execute(select(func.pg_advisory_xact_lock(IP_LOCK_NS, func.hashtext(ip))))
    known = _is_known(db, name, ip)
    retry, _, _ = _retry_after(db, name, ip, known)
    if retry:  # блокировка: пароль не проверяем, в журнал не пишем (запись о блокировке уже есть)
        raise _locked(retry)
    user = db.scalar(select(User).where(User.username == body.username, User.is_active))
    if user is None:
        verify_password(body.password, _DUMMY_HASH)
    if user is None or not verify_password(body.password, user.password_hash):
        _, _, before = _retry_after(db, name, ip, known)
        db.add(LoginAttempt(client_ip=ip, username=name))
        db.flush()
        retry, why, after = _retry_after(db, name, ip, known)
        if retry and before[why] < after[why]:  # именно этот запрос впервые пересёк лимит — запись пишем один раз
            attempts = after[why]
            diff = {"scope": why, "attempts": attempts, "retry_after_seconds": retry}
            if why == "ip":
                diff["distinct_logins"] = _distinct_logins(db, ip)
            scope_ru = {"login_ip": "по логину и IP", "ip": "с IP", "login": "по логину"}[why]
            audit(db, ANONYMOUS, "session", None, "locked", body.username[:100], diff,
                  message=f"Вход заблокирован на {max(1, -(-retry // 60))} мин.: {attempts} неудачных попыток ({scope_ru})")
        elif before["login"] == 0:  # в журнал — только первая неудача по этому логину в окне (не по IP: иначе перебор логинов с одного IP её не оставит)
            audit(db, ANONYMOUS, "session", None, "failed", body.username[:100], message="Неудачная попытка входа в UI")
        db.commit()
        if retry:
            raise _locked(retry)
        raise HTTPException(401, "Неверный логин или пароль")
    db.execute(delete(LoginAttempt).where(LoginAttempt.username == name))
    _remember_login(db, name, ip)
    user.last_login_at = func.now()  # изменение 035
    user.last_login_ip = ip  # изменение 035
    audit(db, user, "session", None, "login", user.username, message=f"Вход в UI: {user.username}")
    db.commit()
    return TokenOut(access_token=create_token(user))


@router.get("/me", response_model=UserOut)
def me(user: User = Depends(current_user)):
    return user