"""Jail management service. Provides methods to list, inspect, and control fail2ban jails via the Unix domain socket. All socket I/O is performed through the async :class:`~app.utils.fail2ban_client.Fail2BanClient` wrapper. Architecture note: this module is a pure service — it contains **no** HTTP/FastAPI concerns. All results are returned as Pydantic models so routers can serialise them directly. """ from __future__ import annotations import asyncio import contextlib import ipaddress from typing import TYPE_CHECKING, TypedDict, cast import structlog from app.exceptions import JailNotFoundError, JailOperationError from app.models.ban import ActiveBan, JailBannedIpsResponse from app.models.config import BantimeEscalation from app.models.geo import GeoDetail from app.models.jail import ( Jail, JailDetailResponse, JailListResponse, JailStatus, JailSummary, ) from app.services import geo_service from app.utils.config_file_utils import start_daemon, wait_for_fail2ban from app.utils.fail2ban_client import ( Fail2BanClient, Fail2BanCommand, Fail2BanConnectionError, Fail2BanResponse, Fail2BanToken, ) from app.utils.fail2ban_response import ( ensure_list, is_not_found_error, ok, to_dict, ) if TYPE_CHECKING: from collections.abc import Awaitable import aiohttp import aiosqlite from app.models.geo import GeoBatchLookup, GeoEnricher, GeoInfo log: structlog.stdlib.BoundLogger = structlog.get_logger() class IpLookupResult(TypedDict): """Result returned by :func:`lookup_ip`. This is intentionally a :class:`TypedDict` to provide precise typing for callers (e.g. routers) while keeping the implementation flexible. """ ip: str currently_banned_in: list[str] geo: GeoDetail | None # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- _SOCKET_TIMEOUT: float = 10.0 # Guard against concurrent reload_all calls. Overlapping ``reload --all`` # commands sent to fail2ban's socket produce undefined behaviour and may cause # jails to be permanently removed from the daemon. Serialising them here # ensures only one reload stream is in-flight at a time. _reload_all_lock: asyncio.Lock | None = None # Capability detection for optional fail2ban transmitter commands (backend, idle). # These commands are not supported in all fail2ban versions. Caching the result # avoids sending unsupported commands every polling cycle and spamming the # fail2ban log with "Invalid command" errors. _backend_cmd_supported: bool | None = None _backend_cmd_lock: asyncio.Lock | None = None def _get_reload_all_lock() -> asyncio.Lock: """Return the shared reload-all lock, initialising it lazily. Asyncio primitives must be created inside an active event loop in test environments that create new loops per test. Lazily initialising the lock avoids binding it to the import-time loop. """ global _reload_all_lock if _reload_all_lock is None: _reload_all_lock = asyncio.Lock() return _reload_all_lock def _get_backend_cmd_lock() -> asyncio.Lock: """Return the shared backend capability probe lock, initialising it lazily. The caller must already be running inside the event loop when the lock is created, which is true for all service entry points in this module. """ global _backend_cmd_lock if _backend_cmd_lock is None: _backend_cmd_lock = asyncio.Lock() return _backend_cmd_lock # --------------------------------------------------------------------------- # Custom exceptions # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # Internal helpers # --------------------------------------------------------------------------- async def _resolve_geo_info( ip: str, *, http_session: aiohttp.ClientSession | None = None, geo_enricher: GeoEnricher | None = None, ) -> GeoInfo | None: """Resolve geolocation using either a custom enricher or HTTP session.""" if geo_enricher is not None: return await geo_enricher(ip) if http_session is not None: return await geo_service.lookup(ip, http_session) return None async def _safe_get( client: Fail2BanClient, command: Fail2BanCommand, default: object | None = None, ) -> object | None: """Send a ``get`` command and return ``default`` on error. Errors during optional detail queries (logpath, regex, etc.) should not abort the whole request — this helper swallows them gracefully. Args: client: The :class:`~app.utils.fail2ban_client.Fail2BanClient` to use. command: The command list to send. default: Value to return when the command fails. Returns: The response payload, or *default* on any error. """ try: response = await client.send(command) return ok(cast("Fail2BanResponse", response)) except (ValueError, TypeError, Exception): return default async def _check_backend_cmd_supported( client: Fail2BanClient, jail_name: str, ) -> bool: """Detect whether the fail2ban daemon supports optional ``get ... backend`` command. Some fail2ban versions (e.g. LinuxServer.io container) do not implement the optional ``get backend`` and ``get idle`` transmitter sub-commands. This helper probes the daemon once and caches the result to avoid repeated "Invalid command" errors in the fail2ban log. Uses double-check locking to minimize lock contention in concurrent polls. Args: client: The :class:`~app.utils.fail2ban_client.Fail2BanClient` to use. jail_name: Name of any jail to use for the probe command. Returns: ``True`` if the command is supported, ``False`` otherwise. Once determined, the result is cached and reused for all jails. """ global _backend_cmd_supported # Fast path: return cached result if already determined. if _backend_cmd_supported is not None: return _backend_cmd_supported # Slow path: acquire lock and probe the command once. async with _get_backend_cmd_lock(): # Double-check idiom: another coroutine may have probed while we waited. if _backend_cmd_supported is not None: return _backend_cmd_supported # Probe: send the command and catch any exception. try: ok(await client.send(["get", jail_name, "backend"])) _backend_cmd_supported = True log.debug("backend_cmd_supported_detected") except Exception: _backend_cmd_supported = False log.debug("backend_cmd_unsupported_detected") return _backend_cmd_supported async def _reset_backend_capability_cache() -> None: """Reset the cached backend/idle capability detection state. This helper is intended for test isolation and for any scenario where the cached probe result must be invalidated before the next detection attempt. """ global _backend_cmd_supported async with _get_backend_cmd_lock(): _backend_cmd_supported = None # --------------------------------------------------------------------------- # Public API — Jail listing & detail # --------------------------------------------------------------------------- async def list_jails(socket_path: str) -> JailListResponse: """Return a summary list of all active fail2ban jails. Queries the daemon for the global jail list and then fetches status and key configuration for each jail in parallel. Args: socket_path: Path to the fail2ban Unix domain socket. Returns: :class:`~app.models.jail.JailListResponse` with all active jails. Raises: ~app.utils.fail2ban_client.Fail2BanConnectionError: If the socket cannot be reached. """ client = Fail2BanClient(socket_path=socket_path, timeout=_SOCKET_TIMEOUT) # 1. Fetch global status to get jail names. global_status = to_dict(ok(await client.send(["status"]))) jail_list_raw: str = str(global_status.get("Jail list", "") or "").strip() jail_names: list[str] = ( [j.strip() for j in jail_list_raw.split(",") if j.strip()] if jail_list_raw else [] ) log.info("jail_list_fetched", count=len(jail_names)) if not jail_names: return JailListResponse(jails=[], total=0) # 2. Fetch summary data for every jail in parallel. summaries: list[JailSummary] = await asyncio.gather( *[_fetch_jail_summary(client, name) for name in jail_names], return_exceptions=False, ) return JailListResponse(jails=list(summaries), total=len(summaries)) async def _fetch_jail_summary( client: Fail2BanClient, name: str, ) -> JailSummary: """Fetch and build a :class:`~app.models.jail.JailSummary` for one jail. Sends the ``status``, ``get ... bantime``, ``findtime``, ``maxretry``, ``backend``, and ``idle`` commands in parallel (if supported). The ``backend`` and ``idle`` commands are optional and not supported in all fail2ban versions. If not supported, this function will not send them to avoid spamming the fail2ban log with "Invalid command" errors. Args: client: Shared :class:`~app.utils.fail2ban_client.Fail2BanClient`. name: Jail name. Returns: A :class:`~app.models.jail.JailSummary` populated from the responses. """ # Check whether optional backend/idle commands are supported. # This probe happens once per session and is cached to avoid repeated # "Invalid command" errors in the fail2ban log. backend_cmd_is_supported = await _check_backend_cmd_supported(client, name) # Build the gather list based on command support. gather_list: list[Awaitable[object]] = [ client.send(["status", name, "short"]), client.send(["get", name, "bantime"]), client.send(["get", name, "findtime"]), client.send(["get", name, "maxretry"]), ] if backend_cmd_is_supported: # Commands are supported; send them for real values. gather_list.extend([ client.send(["get", name, "backend"]), client.send(["get", name, "idle"]), ]) else: # Commands not supported; return default values without sending. async def _return_default(value: object | None) -> Fail2BanResponse: return (0, value) gather_list.extend([ _return_default("polling"), # backend default _return_default(False), # idle default ]) _r = await asyncio.gather(*gather_list, return_exceptions=True) status_raw: object | Exception = _r[0] bantime_raw: object | Exception = _r[1] findtime_raw: object | Exception = _r[2] maxretry_raw: object | Exception = _r[3] backend_raw: object | Exception = _r[4] idle_raw: object | Exception = _r[5] # Parse jail status (filter + actions). jail_status: JailStatus | None = None if not isinstance(status_raw, Exception): try: raw = to_dict(ok(status_raw)) filter_stats = to_dict(raw.get("Filter") or []) action_stats = to_dict(raw.get("Actions") or []) jail_status = JailStatus( currently_banned=int(str(action_stats.get("Currently banned", 0) or 0)), total_banned=int(str(action_stats.get("Total banned", 0) or 0)), currently_failed=int(str(filter_stats.get("Currently failed", 0) or 0)), total_failed=int(str(filter_stats.get("Total failed", 0) or 0)), ) except (ValueError, TypeError) as exc: log.warning("jail_status_parse_error", jail=name, error=str(exc)) def _safe_int(raw: object | Exception, fallback: int) -> int: if isinstance(raw, Exception): return fallback try: return int(str(ok(cast("Fail2BanResponse", raw)))) except (ValueError, TypeError): return fallback def _safe_str(raw: object | Exception, fallback: str) -> str: if isinstance(raw, Exception): return fallback try: return str(ok(cast("Fail2BanResponse", raw))) except (ValueError, TypeError): return fallback def _safe_bool(raw: object | Exception, fallback: bool = False) -> bool: if isinstance(raw, Exception): return fallback try: return bool(ok(cast("Fail2BanResponse", raw))) except (ValueError, TypeError): return fallback return JailSummary( name=name, enabled=True, running=True, idle=_safe_bool(idle_raw), backend=_safe_str(backend_raw, "polling"), find_time=_safe_int(findtime_raw, 600), ban_time=_safe_int(bantime_raw, 600), max_retry=_safe_int(maxretry_raw, 5), status=jail_status, ) async def get_jail(socket_path: str, name: str) -> JailDetailResponse: """Return full detail for a single fail2ban jail. Sends multiple ``get`` and ``status`` commands in parallel to build the complete jail snapshot. Args: socket_path: Path to the fail2ban Unix domain socket. name: Jail name. Returns: :class:`~app.models.jail.JailDetailResponse` with the full jail. Raises: JailNotFoundError: If *name* is not a known jail. ~app.utils.fail2ban_client.Fail2BanConnectionError: If the socket cannot be reached. """ client = Fail2BanClient(socket_path=socket_path, timeout=_SOCKET_TIMEOUT) # Verify the jail exists by sending a status command first. try: status_raw = ok(await client.send(["status", name, "short"])) except ValueError as exc: if is_not_found_error(exc): raise JailNotFoundError(name) from exc raise raw = to_dict(status_raw) filter_stats = to_dict(raw.get("Filter") or []) action_stats = to_dict(raw.get("Actions") or []) jail_status = JailStatus( currently_banned=int(str(action_stats.get("Currently banned", 0) or 0)), total_banned=int(str(action_stats.get("Total banned", 0) or 0)), currently_failed=int(str(filter_stats.get("Currently failed", 0) or 0)), total_failed=int(str(filter_stats.get("Total failed", 0) or 0)), ) # Fetch all detail fields in parallel. ( logpath_raw, failregex_raw, ignoreregex_raw, ignoreip_raw, datepattern_raw, logencoding_raw, bantime_raw, findtime_raw, maxretry_raw, backend_raw, idle_raw, actions_raw, bt_increment_raw, bt_factor_raw, bt_formula_raw, bt_multipliers_raw, bt_maxtime_raw, bt_rndtime_raw, bt_overalljails_raw, ) = await asyncio.gather( _safe_get(client, ["get", name, "logpath"], []), _safe_get(client, ["get", name, "failregex"], []), _safe_get(client, ["get", name, "ignoreregex"], []), _safe_get(client, ["get", name, "ignoreip"], []), _safe_get(client, ["get", name, "datepattern"], None), _safe_get(client, ["get", name, "logencoding"], "UTF-8"), _safe_get(client, ["get", name, "bantime"], 600), _safe_get(client, ["get", name, "findtime"], 600), _safe_get(client, ["get", name, "maxretry"], 5), _safe_get(client, ["get", name, "backend"], "polling"), _safe_get(client, ["get", name, "idle"], False), _safe_get(client, ["get", name, "actions"], []), _safe_get(client, ["get", name, "bantime.increment"], False), _safe_get(client, ["get", name, "bantime.factor"], None), _safe_get(client, ["get", name, "bantime.formula"], None), _safe_get(client, ["get", name, "bantime.multipliers"], None), _safe_get(client, ["get", name, "bantime.maxtime"], None), _safe_get(client, ["get", name, "bantime.rndtime"], None), _safe_get(client, ["get", name, "bantime.overalljails"], False), ) bt_increment: bool = bool(bt_increment_raw) bantime_escalation = BantimeEscalation( increment=bt_increment, factor=float(str(bt_factor_raw)) if bt_factor_raw is not None else None, formula=str(bt_formula_raw) if bt_formula_raw else None, multipliers=str(bt_multipliers_raw) if bt_multipliers_raw else None, max_time=int(str(bt_maxtime_raw)) if bt_maxtime_raw is not None else None, rnd_time=int(str(bt_rndtime_raw)) if bt_rndtime_raw is not None else None, overall_jails=bool(bt_overalljails_raw), ) jail = Jail( name=name, enabled=True, running=True, idle=bool(idle_raw), backend=str(backend_raw or "polling"), log_paths=ensure_list(logpath_raw), fail_regex=ensure_list(failregex_raw), ignore_regex=ensure_list(ignoreregex_raw), ignore_ips=ensure_list(ignoreip_raw), date_pattern=str(datepattern_raw) if datepattern_raw else None, log_encoding=str(logencoding_raw or "UTF-8"), find_time=int(str(findtime_raw or 600)), ban_time=int(str(bantime_raw or 600)), max_retry=int(str(maxretry_raw or 5)), bantime_escalation=bantime_escalation, status=jail_status, actions=ensure_list(actions_raw), ) log.info("jail_detail_fetched", jail=name) return JailDetailResponse(jail=jail) # --------------------------------------------------------------------------- # Public API — Jail control # --------------------------------------------------------------------------- async def start_jail(socket_path: str, name: str) -> None: """Start a stopped fail2ban jail. Args: socket_path: Path to the fail2ban Unix domain socket. name: Jail name to start. Raises: JailNotFoundError: If *name* is not a known jail. JailOperationError: If fail2ban reports the operation failed. ~app.utils.fail2ban_client.Fail2BanConnectionError: If the socket cannot be reached. """ client = Fail2BanClient(socket_path=socket_path, timeout=_SOCKET_TIMEOUT) try: ok(await client.send(["start", name])) log.info("jail_started", jail=name) except ValueError as exc: if is_not_found_error(exc): raise JailNotFoundError(name) from exc raise JailOperationError(str(exc)) from exc async def stop_jail(socket_path: str, name: str) -> None: """Stop a running fail2ban jail. If the jail is not currently active (already stopped), this is a no-op and no error is raised — the operation is idempotent. Args: socket_path: Path to the fail2ban Unix domain socket. name: Jail name to stop. Raises: JailOperationError: If fail2ban reports the operation failed. ~app.utils.fail2ban_client.Fail2BanConnectionError: If the socket cannot be reached. """ client = Fail2BanClient(socket_path=socket_path, timeout=_SOCKET_TIMEOUT) try: ok(await client.send(["stop", name])) log.info("jail_stopped", jail=name) except ValueError as exc: if is_not_found_error(exc): # Jail is already stopped or was never running — treat as a no-op. log.info("jail_stop_noop", jail=name) return raise JailOperationError(str(exc)) from exc async def set_idle(socket_path: str, name: str, *, on: bool) -> None: """Toggle the idle mode of a fail2ban jail. When idle mode is on the jail pauses monitoring without stopping completely; existing bans remain active. Args: socket_path: Path to the fail2ban Unix domain socket. name: Jail name. on: Pass ``True`` to enable idle, ``False`` to disable it. Raises: JailNotFoundError: If *name* is not a known jail. JailOperationError: If fail2ban reports the operation failed. ~app.utils.fail2ban_client.Fail2BanConnectionError: If the socket cannot be reached. """ state = "on" if on else "off" client = Fail2BanClient(socket_path=socket_path, timeout=_SOCKET_TIMEOUT) try: ok(await client.send(["set", name, "idle", state])) log.info("jail_idle_toggled", jail=name, idle=on) except ValueError as exc: if is_not_found_error(exc): raise JailNotFoundError(name) from exc raise JailOperationError(str(exc)) from exc async def reload_jail(socket_path: str, name: str) -> None: """Reload a single fail2ban jail to pick up configuration changes. The reload protocol requires a non-empty configuration stream. Without one, fail2ban's end-of-reload phase removes every jail that received no configuration commands — permanently deleting the jail from the running daemon. Sending ``['start', name]`` as the minimal stream is sufficient: ``startJail`` removes the jail from ``reload_state``, which causes the end phase to *commit* the reload instead of deleting the jail. Args: socket_path: Path to the fail2ban Unix domain socket. name: Jail name to reload. Raises: JailNotFoundError: If *name* is not a known jail. JailOperationError: If fail2ban reports the operation failed. ~app.utils.fail2ban_client.Fail2BanConnectionError: If the socket cannot be reached. """ client = Fail2BanClient(socket_path=socket_path, timeout=_SOCKET_TIMEOUT) try: ok(await client.send(["reload", name, [], [["start", name]]])) log.info("jail_reloaded", jail=name) except ValueError as exc: if is_not_found_error(exc): raise JailNotFoundError(name) from exc raise JailOperationError(str(exc)) from exc async def reload_all( socket_path: str, *, include_jails: list[str] | None = None, exclude_jails: list[str] | None = None, ) -> None: """Reload all fail2ban jails at once. Fetches the current jail list first so that a ``['start', name]`` entry can be included in the config stream for every active jail. Without a non-empty stream the end-of-reload phase deletes every jail that received no configuration commands. *include_jails* are added to the stream (e.g. a newly activated jail that is not yet running). *exclude_jails* are removed from the stream (e.g. a jail that was just deactivated and should not be restarted). Args: socket_path: Path to the fail2ban Unix domain socket. include_jails: Extra jail names to add to the start stream. exclude_jails: Jail names to remove from the start stream. Raises: JailNotFoundError: If a jail in *include_jails* does not exist or its configuration is invalid (e.g. missing logpath). JailOperationError: If fail2ban reports the operation failed for a different reason. ~app.utils.fail2ban_client.Fail2BanConnectionError: If the socket cannot be reached. """ client = Fail2BanClient(socket_path=socket_path, timeout=_SOCKET_TIMEOUT) async with _get_reload_all_lock(): try: # Resolve jail names so we can build the minimal config stream. status_raw = ok(await client.send(["status"])) status_dict = to_dict(status_raw) jail_list_raw: str = str(status_dict.get("Jail list", "")) jail_names = [n.strip() for n in jail_list_raw.split(",") if n.strip()] # Merge include/exclude sets so the stream matches the desired state. names_set: set[str] = set(jail_names) if include_jails: names_set.update(include_jails) if exclude_jails: names_set -= set(exclude_jails) stream: list[list[object]] = [["start", n] for n in sorted(names_set)] ok(await client.send(["reload", "--all", [], cast("Fail2BanToken", stream)])) log.info("all_jails_reloaded") except ValueError as exc: # Detect UnknownJailException (missing or invalid jail configuration) # and re-raise as JailNotFoundError for better error specificity. if is_not_found_error(exc): # Extract the jail name from include_jails if available. jail_name = include_jails[0] if include_jails else "unknown" raise JailNotFoundError(jail_name) from exc raise JailOperationError(str(exc)) from exc async def restart(socket_path: str) -> None: """Stop the fail2ban daemon via the Unix socket. Sends ``["stop"]`` to the fail2ban daemon, which calls ``server.quit()`` on the daemon side and tears down all jails. The caller is responsible for starting the daemon again (e.g. via ``fail2ban-client start``). Note: ``["restart"]`` is a *client-side* orchestration command that is not handled by the fail2ban server transmitter — sending it to the socket raises ``"Invalid command"`` in the daemon. Args: socket_path: Path to the fail2ban Unix domain socket. Raises: JailOperationError: If fail2ban reports the stop command failed. ~app.utils.fail2ban_client.Fail2BanConnectionError: If the socket cannot be reached. """ client = Fail2BanClient(socket_path=socket_path, timeout=_SOCKET_TIMEOUT) try: ok(await client.send(["stop"])) log.info("fail2ban_stopped_for_restart") except ValueError as exc: raise JailOperationError(str(exc)) from exc async def restart_daemon( socket_path: str, start_cmd_parts: list[str], max_wait_seconds: float = _SOCKET_TIMEOUT, ) -> bool: """Restart the fail2ban daemon and verify it comes back online. This function stops the daemon through the socket, starts it with the configured command, and probes the socket until fail2ban accepts status requests again. Args: socket_path: Path to the fail2ban Unix domain socket. start_cmd_parts: The configured fail2ban start command split into executable and arguments. max_wait_seconds: The maximum number of seconds to wait for the daemon to become responsive after starting. Returns: ``True`` when the daemon is started and responsive. ``False`` when the command failed or fail2ban never became responsive. Raises: JailOperationError: If the stop command failed. ~app.utils.fail2ban_client.Fail2BanConnectionError: If the socket cannot be reached while stopping fail2ban. """ await restart(socket_path) if not await start_daemon(start_cmd_parts): log.warning( "fail2ban_start_command_failed", command=" ".join(start_cmd_parts), ) return False return await wait_for_fail2ban( socket_path, max_wait_seconds=max_wait_seconds, ) def _parse_ban_entry(entry: str, jail: str) -> ActiveBan | None: """Parse a ban entry from ``get banip --with-time`` output. Expected format:: "1.2.3.4 \t2025-01-01 12:00:00 + 3600 = 2025-01-01 13:00:00" Args: entry: Raw ban entry string. jail: Jail name for the resulting record. Returns: An :class:`~app.models.ban.ActiveBan` or ``None`` if parsing fails. """ from datetime import UTC, datetime try: parts = entry.split("\t", 1) ip = parts[0].strip() # Validate IP ipaddress.ip_address(ip) if len(parts) < 2: # Entry has no time info — return with unknown times. return ActiveBan( ip=ip, jail=jail, banned_at=None, expires_at=None, ban_count=1, country=None, ) time_part = parts[1].strip() # Format: "2025-01-01 12:00:00 + 3600 = 2025-01-01 13:00:00" # Split at " + " to get banned_at and remainder. plus_idx = time_part.find(" + ") if plus_idx == -1: banned_at_str = time_part.strip() expires_at_str: str | None = None else: banned_at_str = time_part[:plus_idx].strip() remainder = time_part[plus_idx + 3 :] # skip " + " eq_idx = remainder.find(" = ") expires_at_str = remainder[eq_idx + 3 :].strip() if eq_idx != -1 else None _date_fmt = "%Y-%m-%d %H:%M:%S" def _to_iso(ts: str) -> str: dt = datetime.strptime(ts, _date_fmt).replace(tzinfo=UTC) return dt.isoformat() banned_at_iso: str | None = None expires_at_iso: str | None = None with contextlib.suppress(ValueError): banned_at_iso = _to_iso(banned_at_str) with contextlib.suppress(ValueError): if expires_at_str: expires_at_iso = _to_iso(expires_at_str) return ActiveBan( ip=ip, jail=jail, banned_at=banned_at_iso, expires_at=expires_at_iso, ban_count=1, country=None, ) except (ValueError, IndexError, AttributeError) as exc: log.debug("ban_entry_parse_error", entry=entry, jail=jail, error=str(exc)) return None # --------------------------------------------------------------------------- # Public API — Jail-specific paginated bans # --------------------------------------------------------------------------- #: Maximum allowed page size for :func:`get_jail_banned_ips`. _MAX_PAGE_SIZE: int = 100 async def get_jail_banned_ips( socket_path: str, jail_name: str, page: int = 1, page_size: int = 25, search: str | None = None, geo_batch_lookup: GeoBatchLookup | None = None, http_session: aiohttp.ClientSession | None = None, app_db: aiosqlite.Connection | None = None, ) -> JailBannedIpsResponse: """Return a paginated list of currently banned IPs for a single jail. Fetches the full ban list from the fail2ban socket, applies an optional substring search filter on the IP, paginates server-side, and geo-enriches **only** the current page slice to stay within rate limits. Args: socket_path: Path to the fail2ban Unix domain socket. jail_name: Name of the jail to query. page: 1-based page number (default 1). page_size: Items per page; clamped to :data:`_MAX_PAGE_SIZE` (default 25). search: Optional case-insensitive substring filter applied to IP addresses. http_session: Optional shared :class:`aiohttp.ClientSession` for geo enrichment via :func:`~app.services.geo_service.lookup_batch`. app_db: Optional BanGUI application database for persistent geo cache. Returns: :class:`~app.models.ban.JailBannedIpsResponse` with the paginated bans. Raises: JailNotFoundError: If *jail_name* is not a known active jail. ~app.utils.fail2ban_client.Fail2BanConnectionError: If the socket is unreachable. """ # Clamp page_size to the allowed maximum. page_size = min(page_size, _MAX_PAGE_SIZE) client = Fail2BanClient(socket_path=socket_path, timeout=_SOCKET_TIMEOUT) # Verify the jail exists. try: ok(await client.send(["status", jail_name, "short"])) except ValueError as exc: if is_not_found_error(exc): raise JailNotFoundError(jail_name) from exc raise # Fetch the full ban list for this jail. try: raw_result = ok(await client.send(["get", jail_name, "banip", "--with-time"])) except (ValueError, TypeError): raw_result = [] ban_list: list[str] = cast("list[str]", raw_result) or [] # Parse all entries. all_bans: list[ActiveBan] = [] for entry in ban_list: ban = _parse_ban_entry(str(entry), jail_name) if ban is not None: all_bans.append(ban) # Apply optional substring search filter (case-insensitive). if search: search_lower = search.lower() all_bans = [b for b in all_bans if search_lower in b.ip.lower()] total = len(all_bans) # Slice the requested page. start = (page - 1) * page_size page_bans = all_bans[start : start + page_size] # Geo-enrich only the page slice. if http_session is not None and page_bans and geo_batch_lookup is not None: page_ips = [b.ip for b in page_bans] try: geo_map = await geo_batch_lookup(page_ips, http_session, db=app_db) except Exception: # noqa: BLE001 log.warning("jail_banned_ips_geo_failed", jail=jail_name) geo_map = {} enriched_page: list[ActiveBan] = [] for ban in page_bans: geo = geo_map.get(ban.ip) if geo is not None: enriched_page.append(ban.model_copy(update={"country": geo.country_code})) else: enriched_page.append(ban) page_bans = enriched_page log.info( "jail_banned_ips_fetched", jail=jail_name, total=total, page=page, page_size=page_size, ) return JailBannedIpsResponse( items=page_bans, total=total, page=page, page_size=page_size, ) async def _enrich_bans( bans: list[ActiveBan], geo_enricher: GeoEnricher, ) -> list[ActiveBan]: """Enrich ban records with geo data asynchronously. Args: bans: The list of :class:`~app.models.ban.ActiveBan` records to enrich. geo_enricher: Async callable ``(ip) → GeoInfo | None``. Returns: The same list with ``country`` fields populated where lookup succeeded. """ geo_results: list[object | Exception] = await asyncio.gather( *[cast("Awaitable[object]", geo_enricher(ban.ip)) for ban in bans], return_exceptions=True, ) enriched: list[ActiveBan] = [] for ban, geo in zip(bans, geo_results, strict=False): if geo is not None and not isinstance(geo, Exception): geo_info = cast("GeoInfo", geo) enriched.append(ban.model_copy(update={"country": geo_info.country_code})) else: enriched.append(ban) return enriched # --------------------------------------------------------------------------- # Public API — Ignore list (IP whitelist) # --------------------------------------------------------------------------- async def get_ignore_list(socket_path: str, name: str) -> list[str]: """Return the ignore list for a fail2ban jail. Args: socket_path: Path to the fail2ban Unix domain socket. name: Jail name. Returns: List of IP addresses and CIDR networks on the jail's ignore list. Raises: JailNotFoundError: If *name* is not a known jail. ~app.utils.fail2ban_client.Fail2BanConnectionError: If the socket cannot be reached. """ client = Fail2BanClient(socket_path=socket_path, timeout=_SOCKET_TIMEOUT) try: raw = ok(await client.send(["get", name, "ignoreip"])) return ensure_list(raw) except ValueError as exc: if is_not_found_error(exc): raise JailNotFoundError(name) from exc raise async def add_ignore_ip(socket_path: str, name: str, ip: str) -> None: """Add an IP address or CIDR network to a jail's ignore list. Args: socket_path: Path to the fail2ban Unix domain socket. name: Jail name. ip: IP address or CIDR network to add. Raises: JailNotFoundError: If *name* is not a known jail. JailOperationError: If fail2ban reports the operation failed. ~app.utils.fail2ban_client.Fail2BanConnectionError: If the socket cannot be reached. """ # Basic format validation. try: ipaddress.ip_network(ip, strict=False) except ValueError as exc: raise ValueError(f"Invalid IP address or network: {ip!r}") from exc client = Fail2BanClient(socket_path=socket_path, timeout=_SOCKET_TIMEOUT) try: ok(await client.send(["set", name, "addignoreip", ip])) log.info("ignore_ip_added", jail=name, ip=ip) except ValueError as exc: if is_not_found_error(exc): raise JailNotFoundError(name) from exc raise JailOperationError(str(exc)) from exc async def del_ignore_ip(socket_path: str, name: str, ip: str) -> None: """Remove an IP address or CIDR network from a jail's ignore list. Args: socket_path: Path to the fail2ban Unix domain socket. name: Jail name. ip: IP address or CIDR network to remove. Raises: JailNotFoundError: If *name* is not a known jail. JailOperationError: If fail2ban reports the operation failed. ~app.utils.fail2ban_client.Fail2BanConnectionError: If the socket cannot be reached. """ client = Fail2BanClient(socket_path=socket_path, timeout=_SOCKET_TIMEOUT) try: ok(await client.send(["set", name, "delignoreip", ip])) log.info("ignore_ip_removed", jail=name, ip=ip) except ValueError as exc: if is_not_found_error(exc): raise JailNotFoundError(name) from exc raise JailOperationError(str(exc)) from exc async def get_ignore_self(socket_path: str, name: str) -> bool: """Return whether a jail ignores the server's own IP addresses. Args: socket_path: Path to the fail2ban Unix domain socket. name: Jail name. Returns: ``True`` when ``ignoreself`` is enabled for the jail. Raises: JailNotFoundError: If *name* is not a known jail. ~app.utils.fail2ban_client.Fail2BanConnectionError: If the socket cannot be reached. """ client = Fail2BanClient(socket_path=socket_path, timeout=_SOCKET_TIMEOUT) try: raw = ok(await client.send(["get", name, "ignoreself"])) return bool(raw) except ValueError as exc: if is_not_found_error(exc): raise JailNotFoundError(name) from exc raise async def set_ignore_self(socket_path: str, name: str, *, on: bool) -> None: """Toggle the ``ignoreself`` option for a fail2ban jail. Args: socket_path: Path to the fail2ban Unix domain socket. name: Jail name. on: ``True`` to enable ignoreself, ``False`` to disable. Raises: JailNotFoundError: If *name* is not a known jail. JailOperationError: If fail2ban reports the operation failed. ~app.utils.fail2ban_client.Fail2BanConnectionError: If the socket cannot be reached. """ value = "true" if on else "false" client = Fail2BanClient(socket_path=socket_path, timeout=_SOCKET_TIMEOUT) try: ok(await client.send(["set", name, "ignoreself", value])) log.info("ignore_self_toggled", jail=name, on=on) except ValueError as exc: if is_not_found_error(exc): raise JailNotFoundError(name) from exc raise JailOperationError(str(exc)) from exc # --------------------------------------------------------------------------- # Public API — IP lookup # --------------------------------------------------------------------------- async def lookup_ip( socket_path: str, ip: str, geo_enricher: GeoEnricher | None = None, http_session: aiohttp.ClientSession | None = None, ) -> IpLookupResult: """Return ban status and history for a single IP address. Checks every running jail for whether the IP is currently banned. Also queries the fail2ban database for historical ban records. Args: socket_path: Path to the fail2ban Unix domain socket. ip: IP address to look up. geo_enricher: Optional async callable ``(ip) → GeoInfo | None``. Returns: A dict with keys: * ``ip`` — the queried IP address * ``currently_banned_in`` — list of jails where the IP is active * ``geo`` — ``GeoInfo`` dataclass or ``None`` Raises: ValueError: If *ip* is not a valid IP address. ~app.utils.fail2ban_client.Fail2BanConnectionError: If the socket cannot be reached. """ try: ipaddress.ip_address(ip) except ValueError as exc: raise ValueError(f"Invalid IP address: {ip!r}") from exc client = Fail2BanClient(socket_path=socket_path, timeout=_SOCKET_TIMEOUT) with contextlib.suppress(ValueError, Fail2BanConnectionError): # Use fail2ban's "banned " command which checks all jails. ok(await client.send(["get", "--all", "banned", ip])) # Fetch jail names from status. global_status = to_dict(ok(await client.send(["status"]))) jail_list_raw: str = str(global_status.get("Jail list", "") or "").strip() jail_names: list[str] = ( [j.strip() for j in jail_list_raw.split(",") if j.strip()] if jail_list_raw else [] ) # Check ban status per jail in parallel. ban_results: list[object | Exception] = await asyncio.gather( *[client.send(["get", jn, "banip"]) for jn in jail_names], return_exceptions=True, ) currently_banned_in: list[str] = [] for jail_name, result in zip(jail_names, ban_results, strict=False): if isinstance(result, Exception): continue try: ban_list: list[str] = cast("list[str]", ok(result)) or [] if ip in ban_list: currently_banned_in.append(jail_name) except (ValueError, TypeError): pass geo: GeoDetail | None = None if geo_enricher is not None or http_session is not None: with contextlib.suppress(Exception): # noqa: BLE001 raw_geo = await _resolve_geo_info( ip, http_session=http_session, geo_enricher=geo_enricher, ) if raw_geo is not None: geo = GeoDetail( country_code=raw_geo.country_code, country_name=raw_geo.country_name, asn=raw_geo.asn, org=raw_geo.org, ) log.info("ip_lookup_completed", ip=ip, banned_in_jails=currently_banned_in) return { "ip": ip, "currently_banned_in": currently_banned_in, "geo": geo, } async def unban_all_ips(socket_path: str) -> int: """Unban every currently banned IP across all fail2ban jails. Uses fail2ban's global ``unban --all`` command, which atomically removes every active ban from every jail in a single socket round-trip. Args: socket_path: Path to the fail2ban Unix domain socket. Returns: The number of IP addresses that were unbanned. Raises: ~app.utils.fail2ban_client.Fail2BanConnectionError: If the socket cannot be reached. """ client = Fail2BanClient(socket_path=socket_path, timeout=_SOCKET_TIMEOUT) count: int = int(str(ok(await client.send(["unban", "--all"])) or 0)) log.info("all_ips_unbanned", count=count) return count