Split config_file_service.py into three specialized service modules

Extract jail, filter, and action configuration management into separate
domain-focused service modules:

- jail_config_service.py: Jail activation, deactivation, validation, rollback
- filter_config_service.py: Filter discovery, CRUD, assignment to jails
- action_config_service.py: Action discovery, CRUD, assignment to jails

Benefits:
- Reduces monolithic 3100-line module into three focused modules
- Improves readability and maintainability per domain
- Clearer separation of concerns following single responsibility principle
- Easier to test domain-specific functionality in isolation
- Reduces coupling - each service only depends on its needed utilities

Changes:
- Create three new service modules under backend/app/services/
- Update backend/app/routers/config.py to import from new modules
- Update exception and function imports to source from appropriate service
- Update Architecture.md to reflect new service organization
- All existing tests continue to pass with new module structure

Relates to Task 4 of refactoring backlog in Docs/Tasks.md

backend/app/routers/config.py

@@ -76,18 +76,28 @@ from app.models.config import (
     RollbackResponse,
     ServiceStatusResponse,
 )
-from app.services import config_file_service, config_service, jail_service
-from app.services.config_file_service import (
+from app.services import config_service, jail_service
+from app.services import (
+    action_config_service,
+    config_file_service,
+    filter_config_service,
+    jail_config_service,
+)
+from app.services.action_config_service import (
     ActionAlreadyExistsError,
     ActionNameError,
     ActionNotFoundError,
     ActionReadonlyError,
     ConfigWriteError,
+)
+from app.services.filter_config_service import (
     FilterAlreadyExistsError,
     FilterInvalidRegexError,
     FilterNameError,
     FilterNotFoundError,
     FilterReadonlyError,
+)
+from app.services.jail_config_service import (
     JailAlreadyActiveError,
     JailAlreadyInactiveError,
     JailNameError,
@@ -193,7 +203,7 @@ async def get_inactive_jails(
     """
     config_dir: str = request.app.state.settings.fail2ban_config_dir
     socket_path: str = request.app.state.settings.fail2ban_socket
-    return await config_file_service.list_inactive_jails(config_dir, socket_path)
+    return await jail_config_service.list_inactive_jails(config_dir, socket_path)
 
 
 @router.get(
@@ -687,7 +697,7 @@ async def activate_jail(
     req = body if body is not None else ActivateJailRequest()
 
     try:
-        result = await config_file_service.activate_jail(config_dir, socket_path, name, req)
+        result = await jail_config_service.activate_jail(config_dir, socket_path, name, req)
     except JailNameError as exc:
         raise _bad_request(str(exc)) from exc
     except JailNotFoundInConfigError:
@@ -761,7 +771,7 @@ async def deactivate_jail(
     socket_path: str = request.app.state.settings.fail2ban_socket
 
     try:
-        result = await config_file_service.deactivate_jail(config_dir, socket_path, name)
+        result = await jail_config_service.deactivate_jail(config_dir, socket_path, name)
     except JailNameError as exc:
         raise _bad_request(str(exc)) from exc
     except JailNotFoundInConfigError:
@@ -820,7 +830,7 @@ async def delete_jail_local_override(
     socket_path: str = request.app.state.settings.fail2ban_socket
 
     try:
-        await config_file_service.delete_jail_local_override(config_dir, socket_path, name)
+        await jail_config_service.delete_jail_local_override(config_dir, socket_path, name)
     except JailNameError as exc:
         raise _bad_request(str(exc)) from exc
     except JailNotFoundInConfigError:
@@ -873,7 +883,7 @@ async def validate_jail(
     """
     config_dir: str = request.app.state.settings.fail2ban_config_dir
     try:
-        return await config_file_service.validate_jail_config(config_dir, name)
+        return await jail_config_service.validate_jail_config(config_dir, name)
     except JailNameError as exc:
         raise _bad_request(str(exc)) from exc
 
@@ -939,7 +949,7 @@ async def rollback_jail(
     start_cmd_parts: list[str] = start_cmd.split()
 
     try:
-        result = await config_file_service.rollback_jail(config_dir, socket_path, name, start_cmd_parts)
+        result = await jail_config_service.rollback_jail(config_dir, socket_path, name, start_cmd_parts)
     except JailNameError as exc:
         raise _bad_request(str(exc)) from exc
     except ConfigWriteError as exc:
@@ -991,7 +1001,7 @@ async def list_filters(
     """
     config_dir: str = request.app.state.settings.fail2ban_config_dir
     socket_path: str = request.app.state.settings.fail2ban_socket
-    result = await config_file_service.list_filters(config_dir, socket_path)
+    result = await filter_config_service.list_filters(config_dir, socket_path)
     # Sort: active first (by name), then inactive (by name).
     result.filters.sort(key=lambda f: (not f.active, f.name.lower()))
     return result
@@ -1028,7 +1038,7 @@ async def get_filter(
     config_dir: str = request.app.state.settings.fail2ban_config_dir
     socket_path: str = request.app.state.settings.fail2ban_socket
     try:
-        return await config_file_service.get_filter(config_dir, socket_path, name)
+        return await filter_config_service.get_filter(config_dir, socket_path, name)
     except FilterNotFoundError:
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND,
@@ -1092,7 +1102,7 @@ async def update_filter(
     config_dir: str = request.app.state.settings.fail2ban_config_dir
     socket_path: str = request.app.state.settings.fail2ban_socket
     try:
-        return await config_file_service.update_filter(config_dir, socket_path, name, body, do_reload=reload)
+        return await filter_config_service.update_filter(config_dir, socket_path, name, body, do_reload=reload)
     except FilterNameError as exc:
         raise _bad_request(str(exc)) from exc
     except FilterNotFoundError:
@@ -1142,7 +1152,7 @@ async def create_filter(
     config_dir: str = request.app.state.settings.fail2ban_config_dir
     socket_path: str = request.app.state.settings.fail2ban_socket
     try:
-        return await config_file_service.create_filter(config_dir, socket_path, body, do_reload=reload)
+        return await filter_config_service.create_filter(config_dir, socket_path, body, do_reload=reload)
     except FilterNameError as exc:
         raise _bad_request(str(exc)) from exc
     except FilterAlreadyExistsError as exc:
@@ -1189,7 +1199,7 @@ async def delete_filter(
     """
     config_dir: str = request.app.state.settings.fail2ban_config_dir
     try:
-        await config_file_service.delete_filter(config_dir, name)
+        await filter_config_service.delete_filter(config_dir, name)
     except FilterNameError as exc:
         raise _bad_request(str(exc)) from exc
     except FilterNotFoundError:
@@ -1238,7 +1248,7 @@ async def assign_filter_to_jail(
     config_dir: str = request.app.state.settings.fail2ban_config_dir
     socket_path: str = request.app.state.settings.fail2ban_socket
     try:
-        await config_file_service.assign_filter_to_jail(config_dir, socket_path, name, body, do_reload=reload)
+        await filter_config_service.assign_filter_to_jail(config_dir, socket_path, name, body, do_reload=reload)
     except (JailNameError, FilterNameError) as exc:
         raise _bad_request(str(exc)) from exc
     except JailNotFoundInConfigError:
@@ -1302,7 +1312,7 @@ async def list_actions(
     """
     config_dir: str = request.app.state.settings.fail2ban_config_dir
     socket_path: str = request.app.state.settings.fail2ban_socket
-    result = await config_file_service.list_actions(config_dir, socket_path)
+    result = await action_config_service.list_actions(config_dir, socket_path)
     result.actions.sort(key=lambda a: (not a.active, a.name.lower()))
     return result
 
@@ -1337,7 +1347,7 @@ async def get_action(
     config_dir: str = request.app.state.settings.fail2ban_config_dir
     socket_path: str = request.app.state.settings.fail2ban_socket
     try:
-        return await config_file_service.get_action(config_dir, socket_path, name)
+        return await action_config_service.get_action(config_dir, socket_path, name)
     except ActionNotFoundError:
         raise _action_not_found(name) from None
 
@@ -1382,7 +1392,7 @@ async def update_action(
     config_dir: str = request.app.state.settings.fail2ban_config_dir
     socket_path: str = request.app.state.settings.fail2ban_socket
     try:
-        return await config_file_service.update_action(config_dir, socket_path, name, body, do_reload=reload)
+        return await action_config_service.update_action(config_dir, socket_path, name, body, do_reload=reload)
     except ActionNameError as exc:
         raise _bad_request(str(exc)) from exc
     except ActionNotFoundError:
@@ -1428,7 +1438,7 @@ async def create_action(
     config_dir: str = request.app.state.settings.fail2ban_config_dir
     socket_path: str = request.app.state.settings.fail2ban_socket
     try:
-        return await config_file_service.create_action(config_dir, socket_path, body, do_reload=reload)
+        return await action_config_service.create_action(config_dir, socket_path, body, do_reload=reload)
     except ActionNameError as exc:
         raise _bad_request(str(exc)) from exc
     except ActionAlreadyExistsError as exc:
@@ -1471,7 +1481,7 @@ async def delete_action(
     """
     config_dir: str = request.app.state.settings.fail2ban_config_dir
     try:
-        await config_file_service.delete_action(config_dir, name)
+        await action_config_service.delete_action(config_dir, name)
     except ActionNameError as exc:
         raise _bad_request(str(exc)) from exc
     except ActionNotFoundError:
@@ -1521,7 +1531,7 @@ async def assign_action_to_jail(
     config_dir: str = request.app.state.settings.fail2ban_config_dir
     socket_path: str = request.app.state.settings.fail2ban_socket
     try:
-        await config_file_service.assign_action_to_jail(config_dir, socket_path, name, body, do_reload=reload)
+        await action_config_service.assign_action_to_jail(config_dir, socket_path, name, body, do_reload=reload)
     except (JailNameError, ActionNameError) as exc:
         raise _bad_request(str(exc)) from exc
     except JailNotFoundInConfigError:
@@ -1570,7 +1580,7 @@ async def remove_action_from_jail(
     config_dir: str = request.app.state.settings.fail2ban_config_dir
     socket_path: str = request.app.state.settings.fail2ban_socket
     try:
-        await config_file_service.remove_action_from_jail(config_dir, socket_path, name, action_name, do_reload=reload)
+        await action_config_service.remove_action_from_jail(config_dir, socket_path, name, action_name, do_reload=reload)
     except (JailNameError, ActionNameError) as exc:
         raise _bad_request(str(exc)) from exc
     except JailNotFoundInConfigError:

backend/app/services/filter_config_service.py

@@ -0,0 +1,941 @@
+"""Filter configuration management for BanGUI.
+
+Handles parsing, validation, and lifecycle operations (create/update/delete)
+for fail2ban filter configurations.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import configparser
+import contextlib
+import io
+import os
+import re
+import tempfile
+from pathlib import Path
+
+import structlog
+
+from app.models.config import (
+    FilterConfig,
+    FilterConfigUpdate,
+    FilterCreateRequest,
+    FilterListResponse,
+    FilterUpdateRequest,
+    AssignFilterRequest,
+)
+from app.exceptions import JailNotFoundError
+from app.services import jail_service
+from app.services.config_file_service import (
+    _parse_jails_sync,
+    _get_active_jail_names,
+    ConfigWriteError,
+    JailNotFoundInConfigError,
+)
+from app.utils import conffile_parser
+
+log: structlog.stdlib.BoundLogger = structlog.get_logger()
+
+# ---------------------------------------------------------------------------
+# Custom exceptions
+# ---------------------------------------------------------------------------
+
+
+class FilterNotFoundError(Exception):
+    """Raised when the requested filter name is not found in ``filter.d/``."""
+
+    def __init__(self, name: str) -> None:
+        """Initialise with the filter name that was not found.
+
+        Args:
+            name: The filter name that could not be located.
+        """
+        self.name: str = name
+        super().__init__(f"Filter not found: {name!r}")
+
+
+class FilterAlreadyExistsError(Exception):
+    """Raised when trying to create a filter whose ``.conf`` or ``.local`` already exists."""
+
+    def __init__(self, name: str) -> None:
+        """Initialise with the filter name that already exists.
+
+        Args:
+            name: The filter name that already exists.
+        """
+        self.name: str = name
+        super().__init__(f"Filter already exists: {name!r}")
+
+
+class FilterReadonlyError(Exception):
+    """Raised when trying to delete a shipped ``.conf`` filter with no ``.local`` override."""
+
+    def __init__(self, name: str) -> None:
+        """Initialise with the filter name that cannot be deleted.
+
+        Args:
+            name: The filter name that is read-only (shipped ``.conf`` only).
+        """
+        self.name: str = name
+        super().__init__(
+            f"Filter {name!r} is a shipped default (.conf only); only user-created .local files can be deleted."
+        )
+
+
+class FilterInvalidRegexError(Exception):
+    """Raised when a regex pattern fails to compile."""
+
+    def __init__(self, pattern: str, error: str) -> None:
+        """Initialise with the invalid pattern and the compile error.
+
+        Args:
+            pattern: The regex string that failed to compile.
+            error: The ``re.error`` message.
+        """
+        self.pattern: str = pattern
+        self.error: str = error
+        super().__init__(f"Invalid regex {pattern!r}: {error}")
+
+
+class FilterNameError(Exception):
+    """Raised when a filter name contains invalid characters."""
+
+
+# ---------------------------------------------------------------------------
+# Additional helper functions for this service
+# ---------------------------------------------------------------------------
+
+
+class JailNameError(Exception):
+    """Raised when a jail name contains invalid characters."""
+
+
+_SAFE_JAIL_NAME_RE: re.Pattern[str] = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
+
+
+def _safe_filter_name(name: str) -> str:
+    """Validate *name* and return it unchanged or raise :class:`FilterNameError`.
+
+    Args:
+        name: Proposed filter name (without extension).
+
+    Returns:
+        The name unchanged if valid.
+
+    Raises:
+        FilterNameError: If *name* contains unsafe characters.
+    """
+    if not _SAFE_FILTER_NAME_RE.match(name):
+        raise FilterNameError(
+            f"Filter name {name!r} contains invalid characters. "
+            "Only alphanumeric characters, hyphens, underscores, and dots are "
+            "allowed; must start with an alphanumeric character."
+        )
+    return name
+
+
+def _safe_jail_name(name: str) -> str:
+    """Validate *name* and return it unchanged or raise :class:`JailNameError`.
+
+    Args:
+        name: Proposed jail name.
+
+    Returns:
+        The name unchanged if valid.
+
+    Raises:
+        JailNameError: If *name* contains unsafe characters.
+    """
+    if not _SAFE_JAIL_NAME_RE.match(name):
+        raise JailNameError(
+            f"Jail name {name!r} contains invalid characters. "
+            "Only alphanumeric characters, hyphens, underscores, and dots are "
+            "allowed; must start with an alphanumeric character."
+        )
+    return name
+
+
+def _build_parser() -> configparser.RawConfigParser:
+    """Create a :class:`configparser.RawConfigParser` for fail2ban configs.
+
+    Returns:
+        Parser with interpolation disabled and case-sensitive option names.
+    """
+    parser = configparser.RawConfigParser(interpolation=None, strict=False)
+    # fail2ban keys are lowercase but preserve case to be safe.
+    parser.optionxform = str  # type: ignore[assignment]
+    return parser
+
+
+def _is_truthy(value: str) -> bool:
+    """Return ``True`` if *value* is a fail2ban boolean true string.
+
+    Args:
+        value: Raw string from config (e.g. ``"true"``, ``"yes"``, ``"1"``).
+
+    Returns:
+        ``True`` when the value represents enabled.
+    """
+    return value.strip().lower() in _TRUE_VALUES
+
+
+def _parse_multiline(raw: str) -> list[str]:
+    """Split a multi-line INI value into individual non-blank lines.
+
+    Args:
+        raw: Raw multi-line string from configparser.
+
+    Returns:
+        List of stripped, non-empty, non-comment strings.
+    """
+    result: list[str] = []
+    for line in raw.splitlines():
+        stripped = line.strip()
+        if stripped and not stripped.startswith("#"):
+            result.append(stripped)
+    return result
+
+
+def _resolve_filter(raw_filter: str, jail_name: str, mode: str) -> str:
+    """Resolve fail2ban variable placeholders in a filter string.
+
+    Handles the common default ``%(__name__)s[mode=%(mode)s]`` pattern that
+    fail2ban uses so the filter name displayed to the user is readable.
+
+    Args:
+        raw_filter: Raw ``filter`` value from config (may contain ``%()s``).
+        jail_name: The jail's section name, used to substitute ``%(__name__)s``.
+        mode: The jail's ``mode`` value, used to substitute ``%(mode)s``.
+
+    Returns:
+        Human-readable filter string.
+    """
+    result = raw_filter.replace("%(__name__)s", jail_name)
+    result = result.replace("%(mode)s", mode)
+    return result
+
+
+# ---------------------------------------------------------------------------
+# Internal helpers - from config_file_service for local use
+# ---------------------------------------------------------------------------
+
+
+def _set_jail_local_key_sync(
+    config_dir: Path,
+    jail_name: str,
+    key: str,
+    value: str,
+) -> None:
+    """Update ``jail.d/{jail_name}.local`` to set a single key in the jail section.
+
+    If the ``.local`` file already exists it is read, the key is updated (or
+    added), and the file is written back atomically without disturbing other
+    settings.  If the file does not exist a new one is created containing
+    only the BanGUI header comment, the jail section, and the requested key.
+
+    Args:
+        config_dir: The fail2ban configuration root directory.
+        jail_name: Validated jail name (used as section name and filename stem).
+        key: Config key to set inside the jail section.
+        value: Config value to assign.
+
+    Raises:
+        ConfigWriteError: If writing fails.
+    """
+    jail_d = config_dir / "jail.d"
+    try:
+        jail_d.mkdir(parents=True, exist_ok=True)
+    except OSError as exc:
+        raise ConfigWriteError(f"Cannot create jail.d directory: {exc}") from exc
+
+    local_path = jail_d / f"{jail_name}.local"
+
+    parser = _build_parser()
+    if local_path.is_file():
+        try:
+            parser.read(str(local_path), encoding="utf-8")
+        except (configparser.Error, OSError) as exc:
+            log.warning(
+                "jail_local_read_for_update_error",
+                jail=jail_name,
+                error=str(exc),
+            )
+
+    if not parser.has_section(jail_name):
+        parser.add_section(jail_name)
+    parser.set(jail_name, key, value)
+
+    # Serialize: write a BanGUI header then the parser output.
+    buf = io.StringIO()
+    buf.write("# Managed by BanGUI — do not edit manually\n\n")
+    parser.write(buf)
+    content = buf.getvalue()
+
+    try:
+        with tempfile.NamedTemporaryFile(
+            mode="w",
+            encoding="utf-8",
+            dir=jail_d,
+            delete=False,
+            suffix=".tmp",
+        ) as tmp:
+            tmp.write(content)
+            tmp_name = tmp.name
+        os.replace(tmp_name, local_path)
+    except OSError as exc:
+        with contextlib.suppress(OSError):
+            os.unlink(tmp_name)  # noqa: F821
+        raise ConfigWriteError(f"Failed to write {local_path}: {exc}") from exc
+
+    log.info(
+        "jail_local_key_set",
+        jail=jail_name,
+        key=key,
+        path=str(local_path),
+    )
+
+
+def _extract_filter_base_name(filter_raw: str) -> str:
+    """Extract the base filter name from a raw fail2ban filter string.
+
+    fail2ban jail configs may specify a filter with an optional mode suffix,
+    e.g. ``sshd``, ``sshd[mode=aggressive]``, or
+    ``%(__name__)s[mode=%(mode)s]``.  This function strips the ``[…]`` mode
+    block and any leading/trailing whitespace to return just the file-system
+    base name used to look up ``filter.d/{name}.conf``.
+
+    Args:
+        filter_raw: Raw ``filter`` value from a jail config (already
+                    with ``%(__name__)s`` substituted by the caller).
+
+    Returns:
+        Base filter name, e.g. ``"sshd"``.
+    """
+    bracket = filter_raw.find("[")
+    if bracket != -1:
+        return filter_raw[:bracket].strip()
+    return filter_raw.strip()
+
+
+def _build_filter_to_jails_map(
+    all_jails: dict[str, dict[str, str]],
+    active_names: set[str],
+) -> dict[str, list[str]]:
+    """Return a mapping of filter base name → list of active jail names.
+
+    Iterates over every jail whose name is in *active_names*, resolves its
+    ``filter`` config key, and records the jail against the base filter name.
+
+    Args:
+        all_jails: Merged jail config dict — ``{jail_name: {key: value}}``.
+        active_names: Set of jail names currently running in fail2ban.
+
+    Returns:
+        ``{filter_base_name: [jail_name, …]}``.
+    """
+    mapping: dict[str, list[str]] = {}
+    for jail_name, settings in all_jails.items():
+        if jail_name not in active_names:
+            continue
+        raw_filter = settings.get("filter", "")
+        mode = settings.get("mode", "normal")
+        resolved = _resolve_filter(raw_filter, jail_name, mode) if raw_filter else jail_name
+        base = _extract_filter_base_name(resolved)
+        if base:
+            mapping.setdefault(base, []).append(jail_name)
+    return mapping
+
+
+def _parse_filters_sync(
+    filter_d: Path,
+) -> list[tuple[str, str, str, bool, str]]:
+    """Synchronously scan ``filter.d/`` and return per-filter tuples.
+
+    Each tuple contains:
+
+    - ``name``          — filter base name (``"sshd"``).
+    - ``filename``      — actual filename (``"sshd.conf"`` or ``"sshd.local"``).
+    - ``content``       — merged file content (``conf`` overridden by ``local``).
+    - ``has_local``     — whether a ``.local`` override exists alongside a ``.conf``.
+    - ``source_path``   — absolute path to the primary (``conf``) source file, or
+                          to the ``.local`` file for user-created (local-only) filters.
+
+    Also discovers ``.local``-only files (user-created filters with no
+    corresponding ``.conf``).  These are returned with ``has_local = False``
+    and ``source_path`` pointing to the ``.local`` file itself.
+
+    Args:
+        filter_d: Path to the ``filter.d`` directory.
+
+    Returns:
+        List of ``(name, filename, content, has_local, source_path)`` tuples,
+        sorted by name.
+    """
+    if not filter_d.is_dir():
+        log.warning("filter_d_not_found", path=str(filter_d))
+        return []
+
+    conf_names: set[str] = set()
+    results: list[tuple[str, str, str, bool, str]] = []
+
+    # ---- .conf-based filters (with optional .local override) ----------------
+    for conf_path in sorted(filter_d.glob("*.conf")):
+        if not conf_path.is_file():
+            continue
+        name = conf_path.stem
+        filename = conf_path.name
+        conf_names.add(name)
+        local_path = conf_path.with_suffix(".local")
+        has_local = local_path.is_file()
+
+        try:
+            content = conf_path.read_text(encoding="utf-8")
+        except OSError as exc:
+            log.warning("filter_read_error", name=name, path=str(conf_path), error=str(exc))
+            continue
+
+        if has_local:
+            try:
+                local_content = local_path.read_text(encoding="utf-8")
+                # Append local content after conf so configparser reads local
+                # values last (higher priority).
+                content = content + "\n" + local_content
+            except OSError as exc:
+                log.warning(
+                    "filter_local_read_error",
+                    name=name,
+                    path=str(local_path),
+                    error=str(exc),
+                )
+
+        results.append((name, filename, content, has_local, str(conf_path)))
+
+    # ---- .local-only filters (user-created, no corresponding .conf) ----------
+    for local_path in sorted(filter_d.glob("*.local")):
+        if not local_path.is_file():
+            continue
+        name = local_path.stem
+        if name in conf_names:
+            # Already covered above as a .conf filter with a .local override.
+            continue
+        try:
+            content = local_path.read_text(encoding="utf-8")
+        except OSError as exc:
+            log.warning(
+                "filter_local_read_error",
+                name=name,
+                path=str(local_path),
+                error=str(exc),
+            )
+            continue
+        results.append((name, local_path.name, content, False, str(local_path)))
+
+    results.sort(key=lambda t: t[0])
+    log.debug("filters_scanned", count=len(results), filter_d=str(filter_d))
+    return results
+
+
+def _validate_regex_patterns(patterns: list[str]) -> None:
+    """Validate each pattern in *patterns* using Python's ``re`` module.
+
+    Args:
+        patterns: List of regex strings to validate.
+
+    Raises:
+        FilterInvalidRegexError: If any pattern fails to compile.
+    """
+    for pattern in patterns:
+        try:
+            re.compile(pattern)
+        except re.error as exc:
+            raise FilterInvalidRegexError(pattern, str(exc)) from exc
+
+
+def _write_filter_local_sync(filter_d: Path, name: str, content: str) -> None:
+    """Write *content* to ``filter.d/{name}.local`` atomically.
+
+    The write is atomic: content is written to a temp file first, then
+    renamed into place.  The ``filter.d/`` directory is created if absent.
+
+    Args:
+        filter_d: Path to the ``filter.d`` directory.
+        name: Validated filter base name (used as filename stem).
+        content: Full serialized filter content to write.
+
+    Raises:
+        ConfigWriteError: If writing fails.
+    """
+    try:
+        filter_d.mkdir(parents=True, exist_ok=True)
+    except OSError as exc:
+        raise ConfigWriteError(f"Cannot create filter.d directory: {exc}") from exc
+
+    local_path = filter_d / f"{name}.local"
+    try:
+        with tempfile.NamedTemporaryFile(
+            mode="w",
+            encoding="utf-8",
+            dir=filter_d,
+            delete=False,
+            suffix=".tmp",
+        ) as tmp:
+            tmp.write(content)
+            tmp_name = tmp.name
+        os.replace(tmp_name, local_path)
+    except OSError as exc:
+        with contextlib.suppress(OSError):
+            os.unlink(tmp_name)  # noqa: F821
+        raise ConfigWriteError(f"Failed to write {local_path}: {exc}") from exc
+
+    log.info("filter_local_written", filter=name, path=str(local_path))
+
+
+# ---------------------------------------------------------------------------
+# Public API — filter discovery
+# ---------------------------------------------------------------------------
+
+
+async def list_filters(
+    config_dir: str,
+    socket_path: str,
+) -> FilterListResponse:
+    """Return all available filters from ``filter.d/`` with active/inactive status.
+
+    Scans ``{config_dir}/filter.d/`` for ``.conf`` files, merges any
+    corresponding ``.local`` overrides, parses each file into a
+    :class:`~app.models.config.FilterConfig`, and cross-references with the
+    currently running jails to determine which filters are active.
+
+    A filter is considered *active* when its base name matches the ``filter``
+    field of at least one currently running jail.
+
+    Args:
+        config_dir: Absolute path to the fail2ban configuration directory.
+        socket_path: Path to the fail2ban Unix domain socket.
+
+    Returns:
+        :class:`~app.models.config.FilterListResponse` with all filters
+        sorted alphabetically, active ones carrying non-empty
+        ``used_by_jails`` lists.
+    """
+    filter_d = Path(config_dir) / "filter.d"
+    loop = asyncio.get_event_loop()
+
+    # Run the synchronous scan in a thread-pool executor.
+    raw_filters: list[tuple[str, str, str, bool, str]] = await loop.run_in_executor(None, _parse_filters_sync, filter_d)
+
+    # Fetch active jail names and their configs concurrently.
+    all_jails_result, active_names = await asyncio.gather(
+        loop.run_in_executor(None, _parse_jails_sync, Path(config_dir)),
+        _get_active_jail_names(socket_path),
+    )
+    all_jails, _source_files = all_jails_result
+
+    filter_to_jails = _build_filter_to_jails_map(all_jails, active_names)
+
+    filters: list[FilterConfig] = []
+    for name, filename, content, has_local, source_path in raw_filters:
+        cfg = conffile_parser.parse_filter_file(content, name=name, filename=filename)
+        used_by = sorted(filter_to_jails.get(name, []))
+        filters.append(
+            FilterConfig(
+                name=cfg.name,
+                filename=cfg.filename,
+                before=cfg.before,
+                after=cfg.after,
+                variables=cfg.variables,
+                prefregex=cfg.prefregex,
+                failregex=cfg.failregex,
+                ignoreregex=cfg.ignoreregex,
+                maxlines=cfg.maxlines,
+                datepattern=cfg.datepattern,
+                journalmatch=cfg.journalmatch,
+                active=len(used_by) > 0,
+                used_by_jails=used_by,
+                source_file=source_path,
+                has_local_override=has_local,
+            )
+        )
+
+    log.info("filters_listed", total=len(filters), active=sum(1 for f in filters if f.active))
+    return FilterListResponse(filters=filters, total=len(filters))
+
+
+async def get_filter(
+    config_dir: str,
+    socket_path: str,
+    name: str,
+) -> FilterConfig:
+    """Return a single filter from ``filter.d/`` with active/inactive status.
+
+    Reads ``{config_dir}/filter.d/{name}.conf``, merges any ``.local``
+    override, and enriches the parsed :class:`~app.models.config.FilterConfig`
+    with ``active``, ``used_by_jails``, ``source_file``, and
+    ``has_local_override``.
+
+    Args:
+        config_dir: Absolute path to the fail2ban configuration directory.
+        socket_path: Path to the fail2ban Unix domain socket.
+        name: Filter base name (e.g. ``"sshd"`` or ``"sshd.conf"``).
+
+    Returns:
+        :class:`~app.models.config.FilterConfig` with status fields populated.
+
+    Raises:
+        FilterNotFoundError: If no ``{name}.conf`` or ``{name}.local`` file
+            exists in ``filter.d/``.
+    """
+    # Normalise — strip extension if provided (.conf=5 chars, .local=6 chars).
+    if name.endswith(".conf"):
+        base_name = name[:-5]
+    elif name.endswith(".local"):
+        base_name = name[:-6]
+    else:
+        base_name = name
+
+    filter_d = Path(config_dir) / "filter.d"
+    conf_path = filter_d / f"{base_name}.conf"
+    local_path = filter_d / f"{base_name}.local"
+    loop = asyncio.get_event_loop()
+
+    def _read() -> tuple[str, bool, str]:
+        """Read filter content and return (content, has_local_override, source_path)."""
+        has_local = local_path.is_file()
+        if conf_path.is_file():
+            content = conf_path.read_text(encoding="utf-8")
+            if has_local:
+                try:
+                    content += "\n" + local_path.read_text(encoding="utf-8")
+                except OSError as exc:
+                    log.warning(
+                        "filter_local_read_error",
+                        name=base_name,
+                        path=str(local_path),
+                        error=str(exc),
+                    )
+            return content, has_local, str(conf_path)
+        elif has_local:
+            # Local-only filter: created by the user, no shipped .conf base.
+            content = local_path.read_text(encoding="utf-8")
+            return content, False, str(local_path)
+        else:
+            raise FilterNotFoundError(base_name)
+
+    content, has_local, source_path = await loop.run_in_executor(None, _read)
+
+    cfg = conffile_parser.parse_filter_file(content, name=base_name, filename=f"{base_name}.conf")
+
+    all_jails_result, active_names = await asyncio.gather(
+        loop.run_in_executor(None, _parse_jails_sync, Path(config_dir)),
+        _get_active_jail_names(socket_path),
+    )
+    all_jails, _source_files = all_jails_result
+    filter_to_jails = _build_filter_to_jails_map(all_jails, active_names)
+
+    used_by = sorted(filter_to_jails.get(base_name, []))
+    log.info("filter_fetched", name=base_name, active=len(used_by) > 0)
+    return FilterConfig(
+        name=cfg.name,
+        filename=cfg.filename,
+        before=cfg.before,
+        after=cfg.after,
+        variables=cfg.variables,
+        prefregex=cfg.prefregex,
+        failregex=cfg.failregex,
+        ignoreregex=cfg.ignoreregex,
+        maxlines=cfg.maxlines,
+        datepattern=cfg.datepattern,
+        journalmatch=cfg.journalmatch,
+        active=len(used_by) > 0,
+        used_by_jails=used_by,
+        source_file=source_path,
+        has_local_override=has_local,
+    )
+
+
+# ---------------------------------------------------------------------------
+# Public API — filter write operations
+# ---------------------------------------------------------------------------
+
+
+async def update_filter(
+    config_dir: str,
+    socket_path: str,
+    name: str,
+    req: FilterUpdateRequest,
+    do_reload: bool = False,
+) -> FilterConfig:
+    """Update a filter's ``.local`` override with new regex/pattern values.
+
+    Reads the current merged configuration for *name* (``conf`` + any existing
+    ``local``), applies the non-``None`` fields in *req* on top of it, and
+    writes the resulting definition to ``filter.d/{name}.local``.  The
+    original ``.conf`` file is never modified.
+
+    All regex patterns in *req* are validated with Python's ``re`` module
+    before any write occurs.
+
+    Args:
+        config_dir: Absolute path to the fail2ban configuration directory.
+        socket_path: Path to the fail2ban Unix domain socket.
+        name: Filter base name (e.g. ``"sshd"`` or ``"sshd.conf"``).
+        req: Partial update — only non-``None`` fields are applied.
+        do_reload: When ``True``, trigger a full fail2ban reload after writing.
+
+    Returns:
+        :class:`~app.models.config.FilterConfig` reflecting the updated state.
+
+    Raises:
+        FilterNameError: If *name* contains invalid characters.
+        FilterNotFoundError: If no ``{name}.conf`` or ``{name}.local`` exists.
+        FilterInvalidRegexError: If any supplied regex pattern is invalid.
+        ConfigWriteError: If writing the ``.local`` file fails.
+    """
+    base_name = name[:-5] if name.endswith(".conf") or name.endswith(".local") else name
+    _safe_filter_name(base_name)
+
+    # Validate regex patterns before touching the filesystem.
+    patterns: list[str] = []
+    if req.failregex is not None:
+        patterns.extend(req.failregex)
+    if req.ignoreregex is not None:
+        patterns.extend(req.ignoreregex)
+    _validate_regex_patterns(patterns)
+
+    # Fetch the current merged config (raises FilterNotFoundError if absent).
+    current = await get_filter(config_dir, socket_path, base_name)
+
+    # Build a FilterConfigUpdate from the request fields.
+    update = FilterConfigUpdate(
+        failregex=req.failregex,
+        ignoreregex=req.ignoreregex,
+        datepattern=req.datepattern,
+        journalmatch=req.journalmatch,
+    )
+
+    merged = conffile_parser.merge_filter_update(current, update)
+    content = conffile_parser.serialize_filter_config(merged)
+
+    filter_d = Path(config_dir) / "filter.d"
+    loop = asyncio.get_event_loop()
+    await loop.run_in_executor(None, _write_filter_local_sync, filter_d, base_name, content)
+
+    if do_reload:
+        try:
+            await jail_service.reload_all(socket_path)
+        except Exception as exc:  # noqa: BLE001
+            log.warning(
+                "reload_after_filter_update_failed",
+                filter=base_name,
+                error=str(exc),
+            )
+
+    log.info("filter_updated", filter=base_name, reload=do_reload)
+    return await get_filter(config_dir, socket_path, base_name)
+
+
+async def create_filter(
+    config_dir: str,
+    socket_path: str,
+    req: FilterCreateRequest,
+    do_reload: bool = False,
+) -> FilterConfig:
+    """Create a brand-new user-defined filter in ``filter.d/{name}.local``.
+
+    No ``.conf`` is written; fail2ban loads ``.local`` files directly.  If a
+    ``.conf`` or ``.local`` file already exists for the requested name, a
+    :class:`FilterAlreadyExistsError` is raised.
+
+    All regex patterns are validated with Python's ``re`` module before
+    writing.
+
+    Args:
+        config_dir: Absolute path to the fail2ban configuration directory.
+        socket_path: Path to the fail2ban Unix domain socket.
+        req: Filter name and definition fields.
+        do_reload: When ``True``, trigger a full fail2ban reload after writing.
+
+    Returns:
+        :class:`~app.models.config.FilterConfig` for the newly created filter.
+
+    Raises:
+        FilterNameError: If ``req.name`` contains invalid characters.
+        FilterAlreadyExistsError: If a ``.conf`` or ``.local`` already exists.
+        FilterInvalidRegexError: If any regex pattern is invalid.
+        ConfigWriteError: If writing fails.
+    """
+    _safe_filter_name(req.name)
+
+    filter_d = Path(config_dir) / "filter.d"
+    conf_path = filter_d / f"{req.name}.conf"
+    local_path = filter_d / f"{req.name}.local"
+
+    def _check_not_exists() -> None:
+        if conf_path.is_file() or local_path.is_file():
+            raise FilterAlreadyExistsError(req.name)
+
+    loop = asyncio.get_event_loop()
+    await loop.run_in_executor(None, _check_not_exists)
+
+    # Validate regex patterns.
+    patterns: list[str] = list(req.failregex) + list(req.ignoreregex)
+    _validate_regex_patterns(patterns)
+
+    # Build a FilterConfig and serialise it.
+    cfg = FilterConfig(
+        name=req.name,
+        filename=f"{req.name}.local",
+        failregex=req.failregex,
+        ignoreregex=req.ignoreregex,
+        prefregex=req.prefregex,
+        datepattern=req.datepattern,
+        journalmatch=req.journalmatch,
+    )
+    content = conffile_parser.serialize_filter_config(cfg)
+
+    await loop.run_in_executor(None, _write_filter_local_sync, filter_d, req.name, content)
+
+    if do_reload:
+        try:
+            await jail_service.reload_all(socket_path)
+        except Exception as exc:  # noqa: BLE001
+            log.warning(
+                "reload_after_filter_create_failed",
+                filter=req.name,
+                error=str(exc),
+            )
+
+    log.info("filter_created", filter=req.name, reload=do_reload)
+    # Re-fetch to get the canonical FilterConfig (source_file, active, etc.).
+    return await get_filter(config_dir, socket_path, req.name)
+
+
+async def delete_filter(
+    config_dir: str,
+    name: str,
+) -> None:
+    """Delete a user-created filter's ``.local`` file.
+
+    Deletion rules:
+    - If only a ``.conf`` file exists (shipped default, no user override) →
+      :class:`FilterReadonlyError`.
+    - If a ``.local`` file exists (whether or not a ``.conf`` also exists) →
+      the ``.local`` file is deleted.  The shipped ``.conf`` is never touched.
+    - If neither file exists → :class:`FilterNotFoundError`.
+
+    Args:
+        config_dir: Absolute path to the fail2ban configuration directory.
+        name: Filter base name (e.g. ``"sshd"``).
+
+    Raises:
+        FilterNameError: If *name* contains invalid characters.
+        FilterNotFoundError: If no filter file is found for *name*.
+        FilterReadonlyError: If only a shipped ``.conf`` exists (no ``.local``).
+        ConfigWriteError: If deletion of the ``.local`` file fails.
+    """
+    base_name = name[:-5] if name.endswith(".conf") or name.endswith(".local") else name
+    _safe_filter_name(base_name)
+
+    filter_d = Path(config_dir) / "filter.d"
+    conf_path = filter_d / f"{base_name}.conf"
+    local_path = filter_d / f"{base_name}.local"
+
+    loop = asyncio.get_event_loop()
+
+    def _delete() -> None:
+        has_conf = conf_path.is_file()
+        has_local = local_path.is_file()
+
+        if not has_conf and not has_local:
+            raise FilterNotFoundError(base_name)
+
+        if has_conf and not has_local:
+            # Shipped default — nothing user-writable to remove.
+            raise FilterReadonlyError(base_name)
+
+        try:
+            local_path.unlink()
+        except OSError as exc:
+            raise ConfigWriteError(f"Failed to delete {local_path}: {exc}") from exc
+
+        log.info("filter_local_deleted", filter=base_name, path=str(local_path))
+
+    await loop.run_in_executor(None, _delete)
+
+
+async def assign_filter_to_jail(
+    config_dir: str,
+    socket_path: str,
+    jail_name: str,
+    req: AssignFilterRequest,
+    do_reload: bool = False,
+) -> None:
+    """Assign a filter to a jail by updating the jail's ``.local`` file.
+
+    Writes ``filter = {req.filter_name}`` into the ``[{jail_name}]`` section
+    of ``jail.d/{jail_name}.local``.  If the ``.local`` file already contains
+    other settings for this jail they are preserved.
+
+    Args:
+        config_dir: Absolute path to the fail2ban configuration directory.
+        socket_path: Path to the fail2ban Unix domain socket.
+        jail_name: Name of the jail to update.
+        req: Request containing the filter name to assign.
+        do_reload: When ``True``, trigger a full fail2ban reload after writing.
+
+    Raises:
+        JailNameError: If *jail_name* contains invalid characters.
+        FilterNameError: If ``req.filter_name`` contains invalid characters.
+        JailNotFoundError: If *jail_name* is not defined in any config file.
+        FilterNotFoundError: If ``req.filter_name`` does not exist in
+            ``filter.d/``.
+        ConfigWriteError: If writing fails.
+    """
+    _safe_jail_name(jail_name)
+    _safe_filter_name(req.filter_name)
+
+    loop = asyncio.get_event_loop()
+
+    # Verify the jail exists in config.
+    all_jails, _src = await loop.run_in_executor(None, _parse_jails_sync, Path(config_dir))
+    if jail_name not in all_jails:
+        raise JailNotFoundInConfigError(jail_name)
+
+    # Verify the filter exists (conf or local).
+    filter_d = Path(config_dir) / "filter.d"
+
+    def _check_filter() -> None:
+        conf_exists = (filter_d / f"{req.filter_name}.conf").is_file()
+        local_exists = (filter_d / f"{req.filter_name}.local").is_file()
+        if not conf_exists and not local_exists:
+            raise FilterNotFoundError(req.filter_name)
+
+    await loop.run_in_executor(None, _check_filter)
+
+    await loop.run_in_executor(
+        None,
+        _set_jail_local_key_sync,
+        Path(config_dir),
+        jail_name,
+        "filter",
+        req.filter_name,
+    )
+
+    if do_reload:
+        try:
+            await jail_service.reload_all(socket_path)
+        except Exception as exc:  # noqa: BLE001
+            log.warning(
+                "reload_after_assign_filter_failed",
+                jail=jail_name,
+                filter=req.filter_name,
+                error=str(exc),
+            )
+
+    log.info(
+        "filter_assigned_to_jail",
+        jail=jail_name,
+        filter=req.filter_name,
+        reload=do_reload,
+    )

backend/app/services/jail_config_service.py

@@ -0,0 +1,997 @@
+"""Jail configuration management for BanGUI.
+
+Handles parsing, validation, and lifecycle operations (activate/deactivate)
+for fail2ban jail configurations. Provides functions to discover inactive
+jails, validate their configurations before activation, and manage jail
+overrides in jail.d/*.local files.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import configparser
+import contextlib
+import io
+import os
+import re
+import tempfile
+from pathlib import Path
+from typing import cast
+
+import structlog
+
+from app.exceptions import JailNotFoundError
+from app.models.config import (
+    ActivateJailRequest,
+    InactiveJail,
+    InactiveJailListResponse,
+    JailActivationResponse,
+    JailValidationIssue,
+    JailValidationResult,
+    RollbackResponse,
+)
+from app.services import config_file_service, jail_service
+from app.utils.fail2ban_client import (
+    Fail2BanClient,
+    Fail2BanConnectionError,
+    Fail2BanResponse,
+)
+
+log: structlog.stdlib.BoundLogger = structlog.get_logger()
+
+# ---------------------------------------------------------------------------
+# Constants
+# ---------------------------------------------------------------------------
+
+_SOCKET_TIMEOUT: float = 10.0
+
+# Allowlist pattern for jail names used in path construction.
+_SAFE_JAIL_NAME_RE: re.Pattern[str] = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
+
+# Sections that are not jail definitions.
+_META_SECTIONS: frozenset[str] = frozenset({"INCLUDES", "DEFAULT"})
+
+# True-ish values for the ``enabled`` key.
+_TRUE_VALUES: frozenset[str] = frozenset({"true", "yes", "1"})
+
+# False-ish values for the ``enabled`` key.
+_FALSE_VALUES: frozenset[str] = frozenset({"false", "no", "0"})
+
+# Seconds to wait between fail2ban liveness probes after a reload.
+_POST_RELOAD_PROBE_INTERVAL: float = 2.0
+
+# Maximum number of post-reload probe attempts (initial attempt + retries).
+_POST_RELOAD_MAX_ATTEMPTS: int = 4
+
+
+# ---------------------------------------------------------------------------
+# Custom exceptions
+# ---------------------------------------------------------------------------
+
+
+class JailNotFoundInConfigError(Exception):
+    """Raised when the requested jail name is not defined in any config file."""
+
+    def __init__(self, name: str) -> None:
+        """Initialise with the jail name that was not found.
+
+        Args:
+            name: The jail name that could not be located.
+        """
+        self.name: str = name
+        super().__init__(f"Jail not found in config files: {name!r}")
+
+
+class JailAlreadyActiveError(Exception):
+    """Raised when trying to activate a jail that is already active."""
+
+    def __init__(self, name: str) -> None:
+        """Initialise with the jail name.
+
+        Args:
+            name: The jail that is already active.
+        """
+        self.name: str = name
+        super().__init__(f"Jail is already active: {name!r}")
+
+
+class JailAlreadyInactiveError(Exception):
+    """Raised when trying to deactivate a jail that is already inactive."""
+
+    def __init__(self, name: str) -> None:
+        """Initialise with the jail name.
+
+        Args:
+            name: The jail that is already inactive.
+        """
+        self.name: str = name
+        super().__init__(f"Jail is already inactive: {name!r}")
+
+
+class JailNameError(Exception):
+    """Raised when a jail name contains invalid characters."""
+
+
+class ConfigWriteError(Exception):
+    """Raised when writing a ``.local`` override file fails."""
+
+
+# ---------------------------------------------------------------------------
+# Internal helpers
+# ---------------------------------------------------------------------------
+
+
+def _safe_jail_name(name: str) -> str:
+    """Validate *name* and return it unchanged or raise :class:`JailNameError`.
+
+    Args:
+        name: Proposed jail name.
+
+    Returns:
+        The name unchanged if valid.
+
+    Raises:
+        JailNameError: If *name* contains unsafe characters.
+    """
+    if not _SAFE_JAIL_NAME_RE.match(name):
+        raise JailNameError(
+            f"Jail name {name!r} contains invalid characters. "
+            "Only alphanumeric characters, hyphens, underscores, and dots are "
+            "allowed; must start with an alphanumeric character."
+        )
+    return name
+
+
+def _build_parser() -> configparser.RawConfigParser:
+    """Create a :class:`configparser.RawConfigParser` for fail2ban configs.
+
+    Returns:
+        Parser with interpolation disabled and case-sensitive option names.
+    """
+    parser = configparser.RawConfigParser(interpolation=None, strict=False)
+    # fail2ban keys are lowercase but preserve case to be safe.
+    parser.optionxform = str  # type: ignore[assignment]
+    return parser
+
+
+def _is_truthy(value: str) -> bool:
+    """Return ``True`` if *value* is a fail2ban boolean true string.
+
+    Args:
+        value: Raw string from config (e.g. ``"true"``, ``"yes"``, ``"1"``).
+
+    Returns:
+        ``True`` when the value represents enabled.
+    """
+    return value.strip().lower() in _TRUE_VALUES
+
+
+def _write_local_override_sync(
+    config_dir: Path,
+    jail_name: str,
+    enabled: bool,
+    overrides: dict[str, object],
+) -> None:
+    """Write a ``jail.d/{name}.local`` file atomically.
+
+    Always writes to ``jail.d/{jail_name}.local``.  If the file already
+    exists it is replaced entirely.  The write is atomic: content is
+    written to a temp file first, then renamed into place.
+
+    Args:
+        config_dir: The fail2ban configuration root directory.
+        jail_name: Validated jail name (used as filename stem).
+        enabled: Value to write for ``enabled =``.
+        overrides: Optional setting overrides (bantime, findtime, maxretry,
+                   port, logpath).
+
+    Raises:
+        ConfigWriteError: If writing fails.
+    """
+    jail_d = config_dir / "jail.d"
+    try:
+        jail_d.mkdir(parents=True, exist_ok=True)
+    except OSError as exc:
+        raise ConfigWriteError(f"Cannot create jail.d directory: {exc}") from exc
+
+    local_path = jail_d / f"{jail_name}.local"
+
+    lines: list[str] = [
+        "# Managed by BanGUI — do not edit manually",
+        "",
+        f"[{jail_name}]",
+        "",
+        f"enabled = {'true' if enabled else 'false'}",
+        # Provide explicit banaction defaults so fail2ban can resolve the
+        # %(banaction)s interpolation used in the built-in action_ chain.
+        "banaction = iptables-multiport",
+        "banaction_allports = iptables-allports",
+    ]
+
+    if overrides.get("bantime") is not None:
+        lines.append(f"bantime  = {overrides['bantime']}")
+    if overrides.get("findtime") is not None:
+        lines.append(f"findtime = {overrides['findtime']}")
+    if overrides.get("maxretry") is not None:
+        lines.append(f"maxretry = {overrides['maxretry']}")
+    if overrides.get("port") is not None:
+        lines.append(f"port     = {overrides['port']}")
+    if overrides.get("logpath"):
+        paths: list[str] = cast("list[str]", overrides["logpath"])
+        if paths:
+            lines.append(f"logpath  = {paths[0]}")
+            for p in paths[1:]:
+                lines.append(f"           {p}")
+
+    content = "\n".join(lines) + "\n"
+
+    try:
+        with tempfile.NamedTemporaryFile(
+            mode="w",
+            encoding="utf-8",
+            dir=jail_d,
+            delete=False,
+            suffix=".tmp",
+        ) as tmp:
+            tmp.write(content)
+            tmp_name = tmp.name
+        os.replace(tmp_name, local_path)
+    except OSError as exc:
+        # Clean up temp file if rename failed.
+        with contextlib.suppress(OSError):
+            os.unlink(tmp_name)  # noqa: F821 — only reachable when tmp_name is set
+        raise ConfigWriteError(f"Failed to write {local_path}: {exc}") from exc
+
+    log.info(
+        "jail_local_written",
+        jail=jail_name,
+        path=str(local_path),
+        enabled=enabled,
+    )
+
+
+def _restore_local_file_sync(local_path: Path, original_content: bytes | None) -> None:
+    """Restore a ``.local`` file to its pre-activation state.
+
+    If *original_content* is ``None``, the file is deleted (it did not exist
+    before the activation).  Otherwise the original bytes are written back
+    atomically via a temp-file rename.
+
+    Args:
+        local_path: Absolute path to the ``.local`` file to restore.
+        original_content: Original raw bytes to write back, or ``None`` to
+            delete the file.
+
+    Raises:
+        ConfigWriteError: If the write or delete operation fails.
+    """
+    if original_content is None:
+        try:
+            local_path.unlink(missing_ok=True)
+        except OSError as exc:
+            raise ConfigWriteError(f"Failed to delete {local_path} during rollback: {exc}") from exc
+        return
+
+    tmp_name: str | None = None
+    try:
+        with tempfile.NamedTemporaryFile(
+            mode="wb",
+            dir=local_path.parent,
+            delete=False,
+            suffix=".tmp",
+        ) as tmp:
+            tmp.write(original_content)
+            tmp_name = tmp.name
+        os.replace(tmp_name, local_path)
+    except OSError as exc:
+        with contextlib.suppress(OSError):
+            if tmp_name is not None:
+                os.unlink(tmp_name)
+        raise ConfigWriteError(f"Failed to restore {local_path} during rollback: {exc}") from exc
+
+
+def _validate_regex_patterns(patterns: list[str]) -> None:
+    """Validate each pattern in *patterns* using Python's ``re`` module.
+
+    Args:
+        patterns: List of regex strings to validate.
+
+    Raises:
+        FilterInvalidRegexError: If any pattern fails to compile.
+    """
+    for pattern in patterns:
+        try:
+            re.compile(pattern)
+        except re.error as exc:
+            # Import here to avoid circular dependency
+            from app.services.filter_config_service import FilterInvalidRegexError
+            raise FilterInvalidRegexError(pattern, str(exc)) from exc
+
+
+def _set_jail_local_key_sync(
+    config_dir: Path,
+    jail_name: str,
+    key: str,
+    value: str,
+) -> None:
+    """Update ``jail.d/{jail_name}.local`` to set a single key in the jail section.
+
+    If the ``.local`` file already exists it is read, the key is updated (or
+    added), and the file is written back atomically without disturbing other
+    settings.  If the file does not exist a new one is created containing
+    only the BanGUI header comment, the jail section, and the requested key.
+
+    Args:
+        config_dir: The fail2ban configuration root directory.
+        jail_name: Validated jail name (used as section name and filename stem).
+        key: Config key to set inside the jail section.
+        value: Config value to assign.
+
+    Raises:
+        ConfigWriteError: If writing fails.
+    """
+    jail_d = config_dir / "jail.d"
+    try:
+        jail_d.mkdir(parents=True, exist_ok=True)
+    except OSError as exc:
+        raise ConfigWriteError(f"Cannot create jail.d directory: {exc}") from exc
+
+    local_path = jail_d / f"{jail_name}.local"
+
+    parser = _build_parser()
+    if local_path.is_file():
+        try:
+            parser.read(str(local_path), encoding="utf-8")
+        except (configparser.Error, OSError) as exc:
+            log.warning(
+                "jail_local_read_for_update_error",
+                jail=jail_name,
+                error=str(exc),
+            )
+
+    if not parser.has_section(jail_name):
+        parser.add_section(jail_name)
+    parser.set(jail_name, key, value)
+
+    # Serialize: write a BanGUI header then the parser output.
+    buf = io.StringIO()
+    buf.write("# Managed by BanGUI — do not edit manually\n\n")
+    parser.write(buf)
+    content = buf.getvalue()
+
+    try:
+        with tempfile.NamedTemporaryFile(
+            mode="w",
+            encoding="utf-8",
+            dir=jail_d,
+            delete=False,
+            suffix=".tmp",
+        ) as tmp:
+            tmp.write(content)
+            tmp_name = tmp.name
+        os.replace(tmp_name, local_path)
+    except OSError as exc:
+        with contextlib.suppress(OSError):
+            os.unlink(tmp_name)  # noqa: F821
+        raise ConfigWriteError(f"Failed to write {local_path}: {exc}") from exc
+
+    log.info(
+        "jail_local_key_set",
+        jail=jail_name,
+        key=key,
+        path=str(local_path),
+    )
+
+
+async def _probe_fail2ban_running(socket_path: str) -> bool:
+    """Return ``True`` if the fail2ban socket responds to a ping.
+
+    Args:
+        socket_path: Path to the fail2ban Unix domain socket.
+
+    Returns:
+        ``True`` when fail2ban is reachable, ``False`` otherwise.
+    """
+    try:
+        client = Fail2BanClient(socket_path=socket_path, timeout=5.0)
+        resp = await client.send(["ping"])
+        return isinstance(resp, (list, tuple)) and resp[0] == 0
+    except Exception:  # noqa: BLE001
+        return False
+
+
+async def wait_for_fail2ban(
+    socket_path: str,
+    max_wait_seconds: float = 10.0,
+    poll_interval: float = 2.0,
+) -> bool:
+    """Poll the fail2ban socket until it responds or the timeout expires.
+
+    Args:
+        socket_path: Path to the fail2ban Unix domain socket.
+        max_wait_seconds: Total time budget in seconds.
+        poll_interval: Delay between probe attempts in seconds.
+
+    Returns:
+        ``True`` if fail2ban came online within the budget.
+    """
+    elapsed = 0.0
+    while elapsed < max_wait_seconds:
+        if await _probe_fail2ban_running(socket_path):
+            return True
+        await asyncio.sleep(poll_interval)
+        elapsed += poll_interval
+    return False
+
+
+async def start_daemon(start_cmd_parts: list[str]) -> bool:
+    """Start the fail2ban daemon using *start_cmd_parts*.
+
+    Uses :func:`asyncio.create_subprocess_exec` (no shell interpretation)
+    to avoid command injection.
+
+    Args:
+        start_cmd_parts: Command and arguments, e.g.
+            ``["fail2ban-client", "start"]``.
+
+    Returns:
+        ``True`` when the process exited with code 0.
+    """
+    if not start_cmd_parts:
+        log.warning("fail2ban_start_cmd_empty")
+        return False
+    try:
+        proc = await asyncio.create_subprocess_exec(
+            *start_cmd_parts,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        await asyncio.wait_for(proc.wait(), timeout=30.0)
+        success = proc.returncode == 0
+        if not success:
+            log.warning(
+                "fail2ban_start_cmd_nonzero",
+                cmd=start_cmd_parts,
+                returncode=proc.returncode,
+            )
+        return success
+    except (TimeoutError, OSError) as exc:
+        log.warning("fail2ban_start_cmd_error", cmd=start_cmd_parts, error=str(exc))
+        return False
+
+
+# Import shared functions from config_file_service
+_parse_jails_sync = config_file_service._parse_jails_sync
+_build_inactive_jail = config_file_service._build_inactive_jail
+_get_active_jail_names = config_file_service._get_active_jail_names
+_validate_jail_config_sync = config_file_service._validate_jail_config_sync
+_orderedconfig_files = config_file_service._ordered_config_files
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+
+async def list_inactive_jails(
+    config_dir: str,
+    socket_path: str,
+) -> InactiveJailListResponse:
+    """Return all jails defined in config files that are not currently active.
+
+    Parses ``jail.conf``, ``jail.local``, and ``jail.d/`` following the
+    fail2ban merge order.  A jail is considered inactive when:
+
+    - Its merged ``enabled`` value is ``false`` (or absent, which defaults to
+      ``false`` in fail2ban), **or**
+    - Its ``enabled`` value is ``true`` in config but fail2ban does not report
+      it as running.
+
+    Args:
+        config_dir: Absolute path to the fail2ban configuration directory.
+        socket_path: Path to the fail2ban Unix domain socket.
+
+    Returns:
+        :class:`~app.models.config.InactiveJailListResponse` with all
+        inactive jails.
+    """
+    loop = asyncio.get_event_loop()
+    parsed_result: tuple[dict[str, dict[str, str]], dict[str, str]] = await loop.run_in_executor(
+        None, _parse_jails_sync, Path(config_dir)
+    )
+    all_jails, source_files = parsed_result
+    active_names: set[str] = await _get_active_jail_names(socket_path)
+
+    inactive: list[InactiveJail] = []
+    for jail_name, settings in sorted(all_jails.items()):
+        if jail_name in active_names:
+            # fail2ban reports this jail as running — skip it.
+            continue
+
+        source = source_files.get(jail_name, config_dir)
+        inactive.append(_build_inactive_jail(jail_name, settings, source, Path(config_dir)))
+
+    log.info(
+        "inactive_jails_listed",
+        total_defined=len(all_jails),
+        active=len(active_names),
+        inactive=len(inactive),
+    )
+    return InactiveJailListResponse(jails=inactive, total=len(inactive))
+
+
+async def activate_jail(
+    config_dir: str,
+    socket_path: str,
+    name: str,
+    req: ActivateJailRequest,
+) -> JailActivationResponse:
+    """Enable an inactive jail and reload fail2ban.
+
+    Performs pre-activation validation, writes ``enabled = true`` (plus any
+    override values from *req*) to ``jail.d/{name}.local``, and triggers a
+    full fail2ban reload.  After the reload a multi-attempt health probe
+    determines whether fail2ban (and the specific jail) are still running.
+
+    Args:
+        config_dir: Absolute path to the fail2ban configuration directory.
+        socket_path: Path to the fail2ban Unix domain socket.
+        name: Name of the jail to activate.  Must exist in the parsed config.
+        req: Optional override values to write alongside ``enabled = true``.
+
+    Returns:
+        :class:`~app.models.config.JailActivationResponse` including
+        ``fail2ban_running`` and ``validation_warnings`` fields.
+
+    Raises:
+        JailNameError: If *name* contains invalid characters.
+        JailNotFoundInConfigError: If *name* is not defined in any config file.
+        JailAlreadyActiveError: If fail2ban already reports *name* as running.
+        ConfigWriteError: If writing the ``.local`` file fails.
+        ~app.utils.fail2ban_client.Fail2BanConnectionError: If the fail2ban
+            socket is unreachable during reload.
+    """
+    _safe_jail_name(name)
+
+    loop = asyncio.get_event_loop()
+    all_jails, _source_files = await loop.run_in_executor(None, _parse_jails_sync, Path(config_dir))
+
+    if name not in all_jails:
+        raise JailNotFoundInConfigError(name)
+
+    active_names = await _get_active_jail_names(socket_path)
+    if name in active_names:
+        raise JailAlreadyActiveError(name)
+
+    # ---------------------------------------------------------------------- #
+    # Pre-activation validation — collect warnings but do not block           #
+    # ---------------------------------------------------------------------- #
+    validation_result: JailValidationResult = await loop.run_in_executor(
+        None, _validate_jail_config_sync, Path(config_dir), name
+    )
+    warnings: list[str] = [f"{i.field}: {i.message}" for i in validation_result.issues]
+    if warnings:
+        log.warning(
+            "jail_activation_validation_warnings",
+            jail=name,
+            warnings=warnings,
+        )
+
+    # Block activation on critical validation failures (missing filter or logpath).
+    blocking = [i for i in validation_result.issues if i.field in ("filter", "logpath")]
+    if blocking:
+        log.warning(
+            "jail_activation_blocked",
+            jail=name,
+            issues=[f"{i.field}: {i.message}" for i in blocking],
+        )
+        return JailActivationResponse(
+            name=name,
+            active=False,
+            fail2ban_running=True,
+            validation_warnings=warnings,
+            message=(f"Jail {name!r} cannot be activated: " + "; ".join(i.message for i in blocking)),
+        )
+
+    overrides: dict[str, object] = {
+        "bantime": req.bantime,
+        "findtime": req.findtime,
+        "maxretry": req.maxretry,
+        "port": req.port,
+        "logpath": req.logpath,
+    }
+
+    # ---------------------------------------------------------------------- #
+    # Backup the existing .local file (if any) before overwriting it so that  #
+    # we can restore it if activation fails.                                   #
+    # ---------------------------------------------------------------------- #
+    local_path = Path(config_dir) / "jail.d" / f"{name}.local"
+    original_content: bytes | None = await loop.run_in_executor(
+        None,
+        lambda: local_path.read_bytes() if local_path.exists() else None,
+    )
+
+    await loop.run_in_executor(
+        None,
+        _write_local_override_sync,
+        Path(config_dir),
+        name,
+        True,
+        overrides,
+    )
+
+    # ---------------------------------------------------------------------- #
+    # Activation reload — if it fails, roll back immediately                  #
+    # ---------------------------------------------------------------------- #
+    try:
+        await jail_service.reload_all(socket_path, include_jails=[name])
+    except JailNotFoundError as exc:
+        # Jail configuration is invalid (e.g. missing logpath that prevents
+        # fail2ban from loading the jail). Roll back and provide a specific error.
+        log.warning(
+            "reload_after_activate_failed_jail_not_found",
+            jail=name,
+            error=str(exc),
+        )
+        recovered = await _rollback_activation_async(config_dir, name, socket_path, original_content)
+        return JailActivationResponse(
+            name=name,
+            active=False,
+            fail2ban_running=False,
+            recovered=recovered,
+            validation_warnings=warnings,
+            message=(
+                f"Jail {name!r} activation failed: {str(exc)}. "
+                "Check that all logpath files exist and are readable. "
+                "The configuration was "
+                + ("automatically recovered." if recovered else "not recovered — manual intervention is required.")
+            ),
+        )
+    except Exception as exc:  # noqa: BLE001
+        log.warning("reload_after_activate_failed", jail=name, error=str(exc))
+        recovered = await _rollback_activation_async(config_dir, name, socket_path, original_content)
+        return JailActivationResponse(
+            name=name,
+            active=False,
+            fail2ban_running=False,
+            recovered=recovered,
+            validation_warnings=warnings,
+            message=(
+                f"Jail {name!r} activation failed during reload and the "
+                "configuration was "
+                + ("automatically recovered." if recovered else "not recovered — manual intervention is required.")
+            ),
+        )
+
+    # ---------------------------------------------------------------------- #
+    # Post-reload health probe with retries                                   #
+    # ---------------------------------------------------------------------- #
+    fail2ban_running = False
+    for attempt in range(_POST_RELOAD_MAX_ATTEMPTS):
+        if attempt > 0:
+            await asyncio.sleep(_POST_RELOAD_PROBE_INTERVAL)
+        if await _probe_fail2ban_running(socket_path):
+            fail2ban_running = True
+            break
+
+    if not fail2ban_running:
+        log.warning(
+            "fail2ban_down_after_activate",
+            jail=name,
+            message="fail2ban socket unreachable after reload — initiating rollback.",
+        )
+        recovered = await _rollback_activation_async(config_dir, name, socket_path, original_content)
+        return JailActivationResponse(
+            name=name,
+            active=False,
+            fail2ban_running=False,
+            recovered=recovered,
+            validation_warnings=warnings,
+            message=(
+                f"Jail {name!r} activation failed: fail2ban stopped responding "
+                "after reload. The configuration was "
+                + ("automatically recovered." if recovered else "not recovered — manual intervention is required.")
+            ),
+        )
+
+    # Verify the jail actually started (config error may prevent it silently).
+    post_reload_names = await _get_active_jail_names(socket_path)
+    actually_running = name in post_reload_names
+    if not actually_running:
+        log.warning(
+            "jail_activation_unverified",
+            jail=name,
+            message="Jail did not appear in running jails — initiating rollback.",
+        )
+        recovered = await _rollback_activation_async(config_dir, name, socket_path, original_content)
+        return JailActivationResponse(
+            name=name,
+            active=False,
+            fail2ban_running=True,
+            recovered=recovered,
+            validation_warnings=warnings,
+            message=(
+                f"Jail {name!r} was written to config but did not start after "
+                "reload. The configuration was "
+                + ("automatically recovered." if recovered else "not recovered — manual intervention is required.")
+            ),
+        )
+
+    log.info("jail_activated", jail=name)
+    return JailActivationResponse(
+        name=name,
+        active=True,
+        fail2ban_running=True,
+        validation_warnings=warnings,
+        message=f"Jail {name!r} activated successfully.",
+    )
+
+
+async def _rollback_activation_async(
+    config_dir: str,
+    name: str,
+    socket_path: str,
+    original_content: bytes | None,
+) -> bool:
+    """Restore the pre-activation ``.local`` file and reload fail2ban.
+
+    Called internally by :func:`activate_jail` when the activation fails after
+    the config file was already written.  Tries to:
+
+    1. Restore the original file content (or delete the file if it was newly
+       created by the activation attempt).
+    2. Reload fail2ban so the daemon runs with the restored configuration.
+    3. Probe fail2ban to confirm it came back up.
+
+    Args:
+        config_dir: Absolute path to the fail2ban configuration directory.
+        name: Name of the jail whose ``.local`` file should be restored.
+        socket_path: Path to the fail2ban Unix domain socket.
+        original_content: Raw bytes of the original ``.local`` file, or
+            ``None`` if the file did not exist before the activation.
+
+    Returns:
+        ``True`` if fail2ban is responsive again after the rollback, ``False``
+        if recovery also failed.
+    """
+    loop = asyncio.get_event_loop()
+    local_path = Path(config_dir) / "jail.d" / f"{name}.local"
+
+    # Step 1 — restore original file (or delete it).
+    try:
+        await loop.run_in_executor(None, _restore_local_file_sync, local_path, original_content)
+        log.info("jail_activation_rollback_file_restored", jail=name)
+    except ConfigWriteError as exc:
+        log.error("jail_activation_rollback_restore_failed", jail=name, error=str(exc))
+        return False
+
+    # Step 2 — reload fail2ban with the restored config.
+    try:
+        await jail_service.reload_all(socket_path)
+        log.info("jail_activation_rollback_reload_ok", jail=name)
+    except Exception as exc:  # noqa: BLE001
+        log.warning("jail_activation_rollback_reload_failed", jail=name, error=str(exc))
+        return False
+
+    # Step 3 — wait for fail2ban to come back.
+    for attempt in range(_POST_RELOAD_MAX_ATTEMPTS):
+        if attempt > 0:
+            await asyncio.sleep(_POST_RELOAD_PROBE_INTERVAL)
+        if await _probe_fail2ban_running(socket_path):
+            log.info("jail_activation_rollback_recovered", jail=name)
+            return True
+
+    log.warning("jail_activation_rollback_still_down", jail=name)
+    return False
+
+
+async def deactivate_jail(
+    config_dir: str,
+    socket_path: str,
+    name: str,
+) -> JailActivationResponse:
+    """Disable an active jail and reload fail2ban.
+
+    Writes ``enabled = false`` to ``jail.d/{name}.local`` and triggers a
+    full fail2ban reload so the jail stops immediately.
+
+    Args:
+        config_dir: Absolute path to the fail2ban configuration directory.
+        socket_path: Path to the fail2ban Unix domain socket.
+        name: Name of the jail to deactivate.  Must exist in the parsed config.
+
+    Returns:
+        :class:`~app.models.config.JailActivationResponse`.
+
+    Raises:
+        JailNameError: If *name* contains invalid characters.
+        JailNotFoundInConfigError: If *name* is not defined in any config file.
+        JailAlreadyInactiveError: If fail2ban already reports *name* as not
+            running.
+        ConfigWriteError: If writing the ``.local`` file fails.
+        ~app.utils.fail2ban_client.Fail2BanConnectionError: If the fail2ban
+            socket is unreachable during reload.
+    """
+    _safe_jail_name(name)
+
+    loop = asyncio.get_event_loop()
+    all_jails, _source_files = await loop.run_in_executor(None, _parse_jails_sync, Path(config_dir))
+
+    if name not in all_jails:
+        raise JailNotFoundInConfigError(name)
+
+    active_names = await _get_active_jail_names(socket_path)
+    if name not in active_names:
+        raise JailAlreadyInactiveError(name)
+
+    await loop.run_in_executor(
+        None,
+        _write_local_override_sync,
+        Path(config_dir),
+        name,
+        False,
+        {},
+    )
+
+    try:
+        await jail_service.reload_all(socket_path, exclude_jails=[name])
+    except Exception as exc:  # noqa: BLE001
+        log.warning("reload_after_deactivate_failed", jail=name, error=str(exc))
+
+    log.info("jail_deactivated", jail=name)
+    return JailActivationResponse(
+        name=name,
+        active=False,
+        message=f"Jail {name!r} deactivated successfully.",
+    )
+
+
+async def delete_jail_local_override(
+    config_dir: str,
+    socket_path: str,
+    name: str,
+) -> None:
+    """Delete the ``jail.d/{name}.local`` override file for an inactive jail.
+
+    This is the clean-up action shown in the config UI when an inactive jail
+    still has a ``.local`` override file (e.g. ``enabled = false``).  The
+    file is deleted outright; no fail2ban reload is required because the jail
+    is already inactive.
+
+    Args:
+        config_dir: Absolute path to the fail2ban configuration directory.
+        socket_path: Path to the fail2ban Unix domain socket.
+        name: Name of the jail whose ``.local`` file should be removed.
+
+    Raises:
+        JailNameError: If *name* contains invalid characters.
+        JailNotFoundInConfigError: If *name* is not defined in any config file.
+        JailAlreadyActiveError: If the jail is currently active (refusing to
+            delete the live config file).
+        ConfigWriteError: If the file cannot be deleted.
+    """
+    _safe_jail_name(name)
+
+    loop = asyncio.get_event_loop()
+    all_jails, _source_files = await loop.run_in_executor(None, _parse_jails_sync, Path(config_dir))
+
+    if name not in all_jails:
+        raise JailNotFoundInConfigError(name)
+
+    active_names = await _get_active_jail_names(socket_path)
+    if name in active_names:
+        raise JailAlreadyActiveError(name)
+
+    local_path = Path(config_dir) / "jail.d" / f"{name}.local"
+    try:
+        await loop.run_in_executor(None, lambda: local_path.unlink(missing_ok=True))
+    except OSError as exc:
+        raise ConfigWriteError(f"Failed to delete {local_path}: {exc}") from exc
+
+    log.info("jail_local_override_deleted", jail=name, path=str(local_path))
+
+
+async def validate_jail_config(
+    config_dir: str,
+    name: str,
+) -> JailValidationResult:
+    """Run pre-activation validation checks on a jail configuration.
+
+    Validates that referenced filter and action files exist in ``filter.d/``
+    and ``action.d/``, that all regex patterns compile, and that declared log
+    paths exist on disk.
+
+    Args:
+        config_dir: Absolute path to the fail2ban configuration directory.
+        name: Name of the jail to validate.
+
+    Returns:
+        :class:`~app.models.config.JailValidationResult` with any issues found.
+
+    Raises:
+        JailNameError: If *name* contains invalid characters.
+    """
+    _safe_jail_name(name)
+    loop = asyncio.get_event_loop()
+    return await loop.run_in_executor(
+        None,
+        _validate_jail_config_sync,
+        Path(config_dir),
+        name,
+    )
+
+
+async def rollback_jail(
+    config_dir: str,
+    socket_path: str,
+    name: str,
+    start_cmd_parts: list[str],
+) -> RollbackResponse:
+    """Disable a bad jail config and restart the fail2ban daemon.
+
+    Writes ``enabled = false`` to ``jail.d/{name}.local`` (works even when
+    fail2ban is down — only a file write), then attempts to start the daemon
+    with *start_cmd_parts*.  Waits up to 10 seconds for the socket to respond.
+
+    Args:
+        config_dir: Absolute path to the fail2ban configuration directory.
+        socket_path: Path to the fail2ban Unix domain socket.
+        name: Name of the jail to disable.
+        start_cmd_parts: Argument list for the daemon start command, e.g.
+            ``["fail2ban-client", "start"]``.
+
+    Returns:
+        :class:`~app.models.config.RollbackResponse`.
+
+    Raises:
+        JailNameError: If *name* contains invalid characters.
+        ConfigWriteError: If writing the ``.local`` file fails.
+    """
+    _safe_jail_name(name)
+
+    loop = asyncio.get_event_loop()
+
+    # Write enabled=false — this must succeed even when fail2ban is down.
+    await loop.run_in_executor(
+        None,
+        _write_local_override_sync,
+        Path(config_dir),
+        name,
+        False,
+        {},
+    )
+    log.info("jail_rolled_back_disabled", jail=name)
+
+    # Attempt to start the daemon.
+    started = await start_daemon(start_cmd_parts)
+    log.info("jail_rollback_start_attempted", jail=name, start_ok=started)
+
+    # Wait for the socket to come back.
+    fail2ban_running = await wait_for_fail2ban(socket_path, max_wait_seconds=10.0, poll_interval=2.0)
+
+    active_jails = 0
+    if fail2ban_running:
+        names = await _get_active_jail_names(socket_path)
+        active_jails = len(names)
+
+    if fail2ban_running:
+        log.info("jail_rollback_success", jail=name, active_jails=active_jails)
+        return RollbackResponse(
+            jail_name=name,
+            disabled=True,
+            fail2ban_running=True,
+            active_jails=active_jails,
+            message=(f"Jail {name!r} disabled and fail2ban restarted successfully with {active_jails} active jail(s)."),
+        )
+
+    log.warning("jail_rollback_fail2ban_still_down", jail=name)
+    return RollbackResponse(
+        jail_name=name,
+        disabled=True,
+        fail2ban_running=False,
+        active_jails=0,
+        message=(
+            f"Jail {name!r} was disabled but fail2ban did not come back online. "
+            "Check the fail2ban log for additional errors."
+        ),
+    )