133 lines
4.5 KiB
Python
133 lines
4.5 KiB
Python
"""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})")
|