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

"""SAC UI login security: IP whitelist, blocked list, unblock."""

from __future__ import annotations

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

from sqlalchemy import delete, func, select
from sqlalchemy.orm import Session

from app.config import get_settings
from app.models.login_attempt import LoginAttempt
from app.models.ui_settings import UI_SETTINGS_ROW_ID, UiSettings
from app.services.fail2ban_sync import (
    Fail2banActionResult,
    list_ssh_banned_ips,
    sync_fail2ban_ignore_ips,
    unban_ssh_ip,
)

_IP_RE = re.compile(
    r"^(?:(?:25[0-5]|2[0-4]\d|[01]?\d?\d)\.){3}(?:25[0-5]|2[0-4]\d|[01]?\d?\d)$"
)


@dataclass(frozen=True)
class LoginSecurityConfig:
    ip_whitelist: tuple[str, ...]
    sync_fail2ban: bool
    max_failures: int
    window_minutes: int
    source: str


@dataclass(frozen=True)
class LoginBlockEntry:
    ip_address: str
    scope: str
    failure_count: int | None
    usernames: tuple[str, ...]
    blocked_until: datetime | None
    reason: str


def normalize_ip_token(raw: str) -> str | None:
    text = (raw or "").strip()
    if not text or text.startswith("#"):
        return None
    if _IP_RE.match(text):
        return text
    return None


def parse_ip_whitelist_text(text: str) -> list[str]:
    if not text:
        return []
    tokens = re.split(r"[\s,;]+", text.replace("\n", " "))
    out: list[str] = []
    seen: set[str] = set()
    for token in tokens:
        ip = normalize_ip_token(token)
        if ip and ip not in seen:
            seen.add(ip)
            out.append(ip)
    return out


def _env_whitelist() -> list[str]:
    settings = get_settings()
    raw = (getattr(settings, "sac_login_ip_whitelist", None) or "").strip()
    return parse_ip_whitelist_text(raw.replace(",", " "))


def _db_whitelist(row: UiSettings | None) -> list[str]:
    if row is None:
        return []
    return parse_ip_whitelist_text(row.login_ip_whitelist or "")


def merge_whitelist(*parts: list[str]) -> list[str]:
    seen: set[str] = set()
    out: list[str] = []
    for part in parts:
        for ip in part:
            if ip not in seen:
                seen.add(ip)
                out.append(ip)
    return out


def get_effective_login_security_config(db: Session) -> LoginSecurityConfig:
    row = db.get(UiSettings, UI_SETTINGS_ROW_ID)
    env_ips = _env_whitelist()
    db_ips = _db_whitelist(row)
    merged = merge_whitelist(env_ips, db_ips)
    settings = get_settings()
    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))
    sources: list[str] = []
    if env_ips:
        sources.append("env")
    if db_ips:
        sources.append("db")
    source = "+".join(sources) if sources else "default"
    sync_fb = bool(row.login_sync_fail2ban) if row is not None else False
    return LoginSecurityConfig(
        ip_whitelist=tuple(merged),
        sync_fail2ban=sync_fb,
        max_failures=max_failures,
        window_minutes=window_minutes,
        source=source,
    )


def is_ip_login_whitelisted(ip_address: str, whitelist: tuple[str, ...] | list[str]) -> bool:
    ip = (ip_address or "").strip()
    if not ip:
        return False
    return ip in whitelist


def upsert_login_security_settings(
    db: Session,
    *,
    ip_whitelist_text: str,
    sync_fail2ban: bool,
) -> LoginSecurityConfig:
    row = db.get(UiSettings, UI_SETTINGS_ROW_ID)
    if row is None:
        row = UiSettings(id=UI_SETTINGS_ROW_ID, show_sidebar_system_stats=True)
        db.add(row)
    row.login_ip_whitelist = ip_whitelist_text.strip() or None
    row.login_sync_fail2ban = bool(sync_fail2ban)
    db.commit()
    db.refresh(row)
    cfg = get_effective_login_security_config(db)
    if cfg.sync_fail2ban:
        sync_fail2ban_ignore_ips(list(cfg.ip_whitelist))
    return cfg


def _window_since(cfg: LoginSecurityConfig) -> datetime:
    return datetime.now(timezone.utc) - timedelta(minutes=cfg.window_minutes)


def list_web_login_blocks(db: Session) -> list[LoginBlockEntry]:
    cfg = get_effective_login_security_config(db)
    since = _window_since(cfg)
    rows = db.execute(
        select(
            LoginAttempt.ip_address,
            func.count().label("cnt"),
            func.max(LoginAttempt.created_at).label("last_at"),
        )
        .where(
            LoginAttempt.success.is_(False),
            LoginAttempt.username != "__alert_sent__",
            LoginAttempt.created_at >= since,
        )
        .group_by(LoginAttempt.ip_address)
        .having(func.count() >= cfg.max_failures)
    ).all()

    blocks: list[LoginBlockEntry] = []
    for ip_address, cnt, last_at in rows:
        ip = str(ip_address)
        if is_ip_login_whitelisted(ip, cfg.ip_whitelist):
            continue
        user_rows = db.scalars(
            select(LoginAttempt.username)
            .where(
                LoginAttempt.ip_address == ip,
                LoginAttempt.success.is_(False),
                LoginAttempt.username.isnot(None),
                LoginAttempt.username != "__alert_sent__",
                LoginAttempt.created_at >= since,
            )
            .distinct()
            .limit(10)
        ).all()
        usernames = tuple(sorted({u for u in user_rows if u}))
        blocked_until = (last_at + timedelta(minutes=cfg.window_minutes)) if last_at else None
        blocks.append(
            LoginBlockEntry(
                ip_address=ip,
                scope="web",
                failure_count=int(cnt or 0),
                usernames=usernames,
                blocked_until=blocked_until,
                reason=f"≥{cfg.max_failures} неудачных входов в UI за {cfg.window_minutes} мин",
            )
        )
    return blocks


def list_ssh_login_blocks() -> list[LoginBlockEntry]:
    banned = list_ssh_banned_ips()
    return [
        LoginBlockEntry(
            ip_address=ip,
            scope="ssh",
            failure_count=None,
            usernames=(),
            blocked_until=None,
            reason="fail2ban (sshd)",
        )
        for ip in banned
    ]


def list_all_login_blocks(db: Session) -> list[LoginBlockEntry]:
    web = {b.ip_address: b for b in list_web_login_blocks(db)}
    ssh = list_ssh_login_blocks()
    out = list(web.values())
    for entry in ssh:
        if entry.ip_address in web:
            existing = web[entry.ip_address]
            out[out.index(existing)] = LoginBlockEntry(
                ip_address=entry.ip_address,
                scope="web+ssh",
                failure_count=existing.failure_count,
                usernames=existing.usernames,
                blocked_until=existing.blocked_until,
                reason=f"{existing.reason}; {entry.reason}",
            )
        else:
            out.append(entry)
    out.sort(key=lambda e: e.ip_address)
    return out


def clear_web_login_block(db: Session, ip_address: str) -> int:
    ip = (ip_address or "").strip()
    if not ip:
        return 0
    result = db.execute(delete(LoginAttempt).where(LoginAttempt.ip_address == ip))
    db.commit()
    return int(result.rowcount or 0)


def unblock_login_ip(db: Session, ip_address: str, *, scope: str = "both") -> list[str]:
    ip = (ip_address or "").strip()
    if not ip:
        return ["empty ip"]
    messages: list[str] = []
    scope_norm = (scope or "both").strip().lower()
    if scope_norm in ("web", "both"):
        deleted = clear_web_login_block(db, ip)
        messages.append(f"Web UI: удалено записей login_attempts: {deleted}")
    if scope_norm in ("ssh", "both"):
        res: Fail2banActionResult = unban_ssh_ip(ip)
        messages.append(res.message)
    return messages