security-alert-center/backend/app/services/login_rate_limit.py

"""Rate limit failed SAC UI logins + optional Telegram alert."""

from __future__ import annotations

import logging
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone

from fastapi import HTTPException, Request
from sqlalchemy import func, select
from sqlalchemy.orm import Session

from app.config import get_settings
from app.models.login_attempt import LoginAttempt
from app.services.client_ip import client_ip_from_request
from app.services.login_security_settings import (
    get_effective_login_security_config,
    is_ip_login_whitelisted,
)
from app.services.telegram_notify import send_telegram_text

logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class LoginRateLimitConfig:
    max_failures: int
    window_minutes: int
    alert_telegram: bool


def get_login_rate_limit_config() -> LoginRateLimitConfig:
    settings = get_settings()
    return LoginRateLimitConfig(
        max_failures=max(1, int(getattr(settings, "sac_login_max_failures", 3) or 3)),
        window_minutes=max(1, int(getattr(settings, "sac_login_failure_window_minutes", 15) or 15)),
        alert_telegram=bool(getattr(settings, "sac_login_alert_telegram", True)),
    )


def _failure_count(db: Session, ip_address: str, *, since: datetime) -> int:
    return int(
        db.scalar(
            select(func.count())
            .select_from(LoginAttempt)
            .where(
                LoginAttempt.ip_address == ip_address,
                LoginAttempt.success.is_(False),
                LoginAttempt.created_at >= since,
            )
        )
        or 0
    )


def _alert_already_sent(db: Session, ip_address: str, *, since: datetime) -> bool:
    """Avoid spamming Telegram on every subsequent 429."""
    row = db.scalar(
        select(LoginAttempt.username)
        .where(
            LoginAttempt.ip_address == ip_address,
            LoginAttempt.success.is_(False),
            LoginAttempt.created_at >= since,
            LoginAttempt.username == "__alert_sent__",
        )
        .limit(1)
    )
    return row is not None


def _send_login_bruteforce_alert(ip_address: str, username: str, failures: int) -> None:
    user_hint = username if username else "—"
    text = (
        f"⚠️ <b>SAC: подбор пароля UI</b>\n"
        f"IP: <code>{ip_address}</code>\n"
        f"Логин: <code>{user_hint}</code>\n"
        f"Неудачных попыток: {failures}\n"
        f"Дальнейший вход с этого IP временно заблокирован."
    )
    try:
        send_telegram_text(text, parse_mode="HTML")
    except Exception:
        logger.exception("failed to send SAC login brute-force Telegram alert")


def ensure_login_allowed(db: Session, request: Request) -> str:
    """Raise 429 if IP exceeded failed login threshold. Returns client IP."""
    cfg = get_login_rate_limit_config()
    ip_address = client_ip_from_request(request)
    sec = get_effective_login_security_config(db)
    if is_ip_login_whitelisted(ip_address, sec.ip_whitelist):
        return ip_address
    since = datetime.now(timezone.utc) - timedelta(minutes=cfg.window_minutes)
    failures = _failure_count(db, ip_address, since=since)
    if failures >= cfg.max_failures:
        raise HTTPException(
            status_code=429,
            detail=f"Слишком много неудачных попыток входа. Повторите через {cfg.window_minutes} мин.",
        )
    return ip_address


def record_login_success(db: Session, *, ip_address: str, username: str) -> None:
    db.add(
        LoginAttempt(
            ip_address=ip_address,
            username=username[:64],
            success=True,
        )
    )


def record_login_failure(
    db: Session,
    *,
    ip_address: str,
    username: str,
    request: Request | None = None,
) -> None:
    del request  # reserved
    sec = get_effective_login_security_config(db)
    if is_ip_login_whitelisted(ip_address, sec.ip_whitelist):
        return
    cfg = get_login_rate_limit_config()
    since = datetime.now(timezone.utc) - timedelta(minutes=cfg.window_minutes)
    prior_failures = _failure_count(db, ip_address, since=since)

    db.add(
        LoginAttempt(
            ip_address=ip_address,
            username=username[:64] if username else None,
            success=False,
        )
    )
    db.flush()

    new_failures = prior_failures + 1
    if new_failures >= cfg.max_failures and cfg.alert_telegram and not _alert_already_sent(db, ip_address, since=since):
        db.add(
            LoginAttempt(
                ip_address=ip_address,
                username="__alert_sent__",
                success=False,
            )
        )
        _send_login_bruteforce_alert(ip_address, username, new_failures)