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

"""Best-effort integration with fail2ban for SSH login bans (OS level, not SAC DB)."""

from __future__ import annotations

import logging
import re
import subprocess
from dataclasses import dataclass

from app.config import get_settings

logger = logging.getLogger(__name__)

_BANNED_IP_RE = re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b")


@dataclass(frozen=True)
class Fail2banActionResult:
    ok: bool
    message: str


def _ssh_jail() -> str:
    settings = get_settings()
    jail = (getattr(settings, "sac_fail2ban_ssh_jail", None) or "sshd").strip()
    return jail or "sshd"


def _run_fail2ban(args: list[str], *, timeout: int = 15) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        ["fail2ban-client", *args],
        capture_output=True,
        text=True,
        timeout=timeout,
        check=False,
    )


def fail2ban_available() -> bool:
    try:
        proc = _run_fail2ban(["ping"], timeout=5)
        return proc.returncode == 0
    except (OSError, subprocess.SubprocessError):
        return False


def list_ssh_banned_ips() -> list[str]:
    if not fail2ban_available():
        return []
    jail = _ssh_jail()
    try:
        proc = _run_fail2ban(["status", jail])
    except (OSError, subprocess.SubprocessError) as exc:
        logger.warning("fail2ban status failed: %s", exc)
        return []
    if proc.returncode != 0:
        return []
    text = (proc.stdout or "") + "\n" + (proc.stderr or "")
    ips: list[str] = []
    capture = False
    for line in text.splitlines():
        lower = line.strip().lower()
        if "banned ip list" in lower:
            capture = True
            tail = line.split(":", 1)[-1]
            ips.extend(_BANNED_IP_RE.findall(tail))
            continue
        if capture:
            if not line.strip():
                break
            ips.extend(_BANNED_IP_RE.findall(line))
    # preserve order, unique
    seen: set[str] = set()
    out: list[str] = []
    for ip in ips:
        if ip not in seen:
            seen.add(ip)
            out.append(ip)
    return out


def unban_ssh_ip(ip_address: str) -> Fail2banActionResult:
    ip = (ip_address or "").strip()
    if not ip:
        return Fail2banActionResult(ok=False, message="empty ip")
    if not fail2ban_available():
        return Fail2banActionResult(ok=False, message="fail2ban-client недоступен на сервере SAC")
    jail = _ssh_jail()
    try:
        proc = _run_fail2ban(["set", jail, "unbanip", ip])
    except (OSError, subprocess.SubprocessError) as exc:
        return Fail2banActionResult(ok=False, message=str(exc))
    if proc.returncode != 0:
        err = (proc.stderr or proc.stdout or "unban failed").strip()
        return Fail2banActionResult(ok=False, message=err[:500])
    return Fail2banActionResult(ok=True, message=f"SSH: IP {ip} разблокирован (jail {jail})")


def sync_fail2ban_ignore_ips(ip_addresses: list[str]) -> Fail2banActionResult:
    settings = get_settings()
    if not getattr(settings, "sac_fail2ban_sync_enabled", False):
        return Fail2banActionResult(ok=True, message="sync отключён (SAC_FAIL2BAN_SYNC_ENABLED=false)")
    path = (getattr(settings, "sac_fail2ban_ignoreip_file", None) or "").strip()
    if not path:
        return Fail2banActionResult(ok=True, message="файл ignoreip не задан (SAC_FAIL2BAN_IGNOREIP_FILE)")

    jail = _ssh_jail()
    base_ignore = "127.0.0.1/8 ::1"
    extra = " ".join(ip for ip in ip_addresses if ip)
    ignore_line = f"{base_ignore} {extra}".strip()
    content = (
        f"# Managed by SAC — login IP whitelist for fail2ban\n"
        f"[{jail}]\n"
        f"ignoreip = {ignore_line}\n"
    )
    try:
        with open(path, "w", encoding="utf-8") as fh:
            fh.write(content)
    except OSError as exc:
        return Fail2banActionResult(ok=False, message=f"не удалось записать {path}: {exc}")

    if not fail2ban_available():
        return Fail2banActionResult(ok=True, message=f"файл записан ({path}); fail2ban reload пропущен")

    try:
        proc = _run_fail2ban(["reload"], timeout=30)
    except (OSError, subprocess.SubprocessError) as exc:
        return Fail2banActionResult(ok=False, message=f"файл записан, reload failed: {exc}")
    if proc.returncode != 0:
        err = (proc.stderr or proc.stdout or "reload failed").strip()
        return Fail2banActionResult(ok=False, message=f"файл записан, reload: {err[:300]}")
    return Fail2banActionResult(ok=True, message=f"fail2ban ignoreip обновлён ({path})")