lukas.pupkalipinski/BanGUI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
c110352e9e7b2ccf9e9ddd739ccbf2e8fb2183a4
BanGUI/backend/app/services/config_file_service.py
Lukas c110352e9e Config page tasks 1-4: dropdowns, key props, inactive jail full GUI, banaction fix 
Task 1: Backend/LogEncoding/DatePattern dropdowns in JailConfigDetail
- Added BACKENDS, LOG_ENCODINGS, DATE_PATTERN_PRESETS constants
- Backend and Log Encoding: <Input readOnly> → <Select> (editable, auto-saves)
- Date Pattern: <Input> → <Combobox freeform> with presets
- Extended JailConfigUpdate model (backend, log_encoding) and service
- Added readOnly prop to JailConfigDetail (all fields, toggles, buttons)
- Extended RegexList with readOnly prop

Task 2: Fix raw action/filter config always blank
- Added key={selectedAction.name} to ActionDetail in ActionsTab
- Added key={selectedFilter.name} to FilterDetail in FiltersTab

Task 3: Inactive jail full GUI same as active jails
- Extended InactiveJail Pydantic model with all config fields
- Added _parse_time_to_seconds helper to config_file_service
- Updated _build_inactive_jail to populate all extended fields
- Extended InactiveJail TypeScript type to match
- Rewrote InactiveJailDetail to reuse JailConfigDetail (readOnly=true)

Task 4: Fix banaction interpolation error when activating jails
- _write_local_override_sync now includes banaction=iptables-multiport
  and banaction_allports=iptables-allports in every .local file
2026-03-14 09:28:30 +01:00

2551 lines
84 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Fail2ban jail configuration file parser and activator.
Parses the full set of fail2ban jail configuration files
(``jail.conf``, ``jail.local``, ``jail.d/*.conf``, ``jail.d/*.local``)
to discover all defined jails — both active and inactive — and provides
functions to activate or deactivate them by writing ``.local`` override
files.
Merge order (fail2ban convention):
1. ``jail.conf``
2. ``jail.local``
3. ``jail.d/*.conf`` (alphabetical)
4. ``jail.d/*.local`` (alphabetical)
Security note: the ``activate_jail`` and ``deactivate_jail`` callers must
supply a validated jail name. This module validates the name against an
allowlist pattern before constructing any filesystem paths to prevent
directory traversal.
"""
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 Any
import structlog
from app.models.config import (
ActionConfig,
ActionConfigUpdate,
ActionCreateRequest,
ActionListResponse,
ActionUpdateRequest,
ActivateJailRequest,
AssignActionRequest,
AssignFilterRequest,
BantimeEscalation,
FilterConfig,
FilterConfigUpdate,
FilterCreateRequest,
FilterListResponse,
FilterUpdateRequest,
InactiveJail,
InactiveJailListResponse,
JailActivationResponse,
)
from app.services import conffile_parser, jail_service
from app.utils.fail2ban_client import Fail2BanClient, Fail2BanConnectionError
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"})
# ---------------------------------------------------------------------------
# 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."""
class FilterNameError(Exception):
"""Raised when a filter name contains invalid characters."""
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}")
# ---------------------------------------------------------------------------
# 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 _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 _ordered_config_files(config_dir: Path) -> list[Path]:
"""Return all jail config files in fail2ban merge order.
Args:
config_dir: The fail2ban configuration root directory.
Returns:
List of paths in ascending priority order (later entries override
earlier ones).
"""
files: list[Path] = []
jail_conf = config_dir / "jail.conf"
if jail_conf.is_file():
files.append(jail_conf)
jail_local = config_dir / "jail.local"
if jail_local.is_file():
files.append(jail_local)
jail_d = config_dir / "jail.d"
if jail_d.is_dir():
files.extend(sorted(jail_d.glob("*.conf")))
files.extend(sorted(jail_d.glob("*.local")))
return files
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_int_safe(value: str) -> int | None:
"""Parse *value* as int, returning ``None`` on failure.
Args:
value: Raw string to parse.
Returns:
Integer value, or ``None``.
"""
try:
return int(value.strip())
except (ValueError, AttributeError):
return None
def _parse_time_to_seconds(value: str | None, default: int) -> int:
"""Convert a fail2ban time string (e.g. ``1h``, ``10m``, ``3600``) to seconds.
Supports the suffixes ``s`` (seconds), ``m`` (minutes), ``h`` (hours),
``d`` (days), ``w`` (weeks), and plain integers (already seconds).
``-1`` is treated as a permanent ban and returned as-is.
Args:
value: Raw time string from config, or ``None``.
default: Value to return when ``value`` is absent or unparseable.
Returns:
Duration in seconds, or ``-1`` for permanent, or ``default`` on failure.
"""
if not value:
return default
stripped = value.strip()
if stripped == "-1":
return -1
multipliers: dict[str, int] = {
"w": 604800,
"d": 86400,
"h": 3600,
"m": 60,
"s": 1,
}
for suffix, factor in multipliers.items():
if stripped.endswith(suffix) and len(stripped) > 1:
try:
return int(stripped[:-1]) * factor
except ValueError:
return default
try:
return int(stripped)
except ValueError:
return default
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
def _parse_jails_sync(
config_dir: Path,
) -> tuple[dict[str, dict[str, str]], dict[str, str]]:
"""Synchronously parse all jail configs and return merged definitions.
This is a CPU-bound / IO-bound sync function; callers must dispatch to
an executor for async use.
Args:
config_dir: The fail2ban configuration root directory.
Returns:
A two-tuple ``(jails, source_files)`` where:
- ``jails``: ``{jail_name: {key: value}}`` – merged settings for each
jail with DEFAULT values already applied.
- ``source_files``: ``{jail_name: str(path)}`` – path of the file that
last defined each jail section (for display in the UI).
"""
parser = _build_parser()
files = _ordered_config_files(config_dir)
# Track which file each section came from (last write wins).
source_files: dict[str, str] = {}
for path in files:
try:
single = _build_parser()
single.read(str(path), encoding="utf-8")
for section in single.sections():
if section not in _META_SECTIONS:
source_files[section] = str(path)
except (configparser.Error, OSError) as exc:
log.warning("jail_config_read_error", path=str(path), error=str(exc))
# Full merged parse: configparser applies DEFAULT values to every section.
try:
parser.read([str(p) for p in files], encoding="utf-8")
except configparser.Error as exc:
log.warning("jail_config_parse_error", error=str(exc))
jails: dict[str, dict[str, str]] = {}
for section in parser.sections():
if section in _META_SECTIONS:
continue
try:
# items() merges DEFAULT values automatically.
jails[section] = dict(parser.items(section))
except configparser.Error as exc:
log.warning(
"jail_section_parse_error", section=section, error=str(exc)
)
log.debug("jails_parsed", count=len(jails), config_dir=str(config_dir))
return jails, source_files
def _build_inactive_jail(
name: str,
settings: dict[str, str],
source_file: str,
) -> InactiveJail:
"""Construct an :class:`~app.models.config.InactiveJail` from raw settings.
Args:
name: Jail section name.
settings: Merged key→value dict (DEFAULT values already applied).
source_file: Path of the file that last defined this section.
Returns:
Populated :class:`~app.models.config.InactiveJail`.
"""
raw_filter = settings.get("filter", "")
mode = settings.get("mode", "normal")
filter_name = _resolve_filter(raw_filter, name, mode) if raw_filter else name
raw_action = settings.get("action", "")
actions = _parse_multiline(raw_action) if raw_action else []
raw_logpath = settings.get("logpath", "")
logpath = _parse_multiline(raw_logpath) if raw_logpath else []
enabled_raw = settings.get("enabled", "false")
enabled = _is_truthy(enabled_raw)
maxretry_raw = settings.get("maxretry", "")
maxretry = _parse_int_safe(maxretry_raw)
# Extended fields for full GUI display
ban_time_seconds = _parse_time_to_seconds(settings.get("bantime"), 600)
find_time_seconds = _parse_time_to_seconds(settings.get("findtime"), 600)
log_encoding = settings.get("logencoding") or "auto"
backend = settings.get("backend") or "auto"
date_pattern = settings.get("datepattern") or None
use_dns = settings.get("usedns") or "warn"
prefregex = settings.get("prefregex") or ""
fail_regex = _parse_multiline(settings.get("failregex", ""))
ignore_regex = _parse_multiline(settings.get("ignoreregex", ""))
# Ban-time escalation
esc_increment = _is_truthy(settings.get("bantime.increment", "false"))
esc_factor_raw = settings.get("bantime.factor")
esc_factor = float(esc_factor_raw) if esc_factor_raw else None
esc_formula = settings.get("bantime.formula") or None
esc_multipliers = settings.get("bantime.multipliers") or None
esc_max_raw = settings.get("bantime.maxtime")
esc_max_time = _parse_time_to_seconds(esc_max_raw, 0) if esc_max_raw else None
esc_rnd_raw = settings.get("bantime.rndtime")
esc_rnd_time = _parse_time_to_seconds(esc_rnd_raw, 0) if esc_rnd_raw else None
esc_overall = _is_truthy(settings.get("bantime.overalljails", "false"))
bantime_escalation = (
BantimeEscalation(
increment=esc_increment,
factor=esc_factor,
formula=esc_formula,
multipliers=esc_multipliers,
max_time=esc_max_time,
rnd_time=esc_rnd_time,
overall_jails=esc_overall,
)
if esc_increment
else None
)
return InactiveJail(
name=name,
filter=filter_name,
actions=actions,
port=settings.get("port") or None,
logpath=logpath,
bantime=settings.get("bantime") or None,
findtime=settings.get("findtime") or None,
maxretry=maxretry,
ban_time_seconds=ban_time_seconds,
find_time_seconds=find_time_seconds,
log_encoding=log_encoding,
backend=backend,
date_pattern=date_pattern,
use_dns=use_dns,
prefregex=prefregex,
fail_regex=fail_regex,
ignore_regex=ignore_regex,
bantime_escalation=bantime_escalation,
source_file=source_file,
enabled=enabled,
)
async def _get_active_jail_names(socket_path: str) -> set[str]:
"""Fetch the set of currently running jail names from fail2ban.
Returns an empty set gracefully if fail2ban is unreachable.
Args:
socket_path: Path to the fail2ban Unix domain socket.
Returns:
Set of active jail names, or empty set on connection failure.
"""
try:
client = Fail2BanClient(socket_path=socket_path, timeout=_SOCKET_TIMEOUT)
def _to_dict_inner(pairs: Any) -> dict[str, Any]:
if not isinstance(pairs, (list, tuple)):
return {}
result: dict[str, Any] = {}
for item in pairs:
try:
k, v = item
result[str(k)] = v
except (TypeError, ValueError):
pass
return result
def _ok(response: Any) -> Any:
code, data = response
if code != 0:
raise ValueError(f"fail2ban error {code}: {data!r}")
return data
status_raw = _ok(await client.send(["status"]))
status_dict = _to_dict_inner(status_raw)
jail_list_raw: str = str(status_dict.get("Jail list", "") or "").strip()
if not jail_list_raw:
return set()
return {j.strip() for j in jail_list_raw.split(",") if j.strip()}
except Fail2BanConnectionError:
log.warning("fail2ban_unreachable_during_inactive_list")
return set()
except Exception as exc: # noqa: BLE001
log.warning(
"fail2ban_status_error_during_inactive_list", error=str(exc)
)
return set()
def _write_local_override_sync(
config_dir: Path,
jail_name: str,
enabled: bool,
overrides: dict[str, Any],
) -> 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] = 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 _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))
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),
)
# ---------------------------------------------------------------------------
# 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))
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.
Writes ``enabled = true`` (plus any override values from *req*) to
``jail.d/{name}.local`` and then triggers a full fail2ban reload so the
jail starts 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 activate. Must exist in the parsed config.
req: Optional override values to write alongside ``enabled = true``.
Returns:
:class:`~app.models.config.JailActivationResponse`.
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)
overrides: dict[str, Any] = {
"bantime": req.bantime,
"findtime": req.findtime,
"maxretry": req.maxretry,
"port": req.port,
"logpath": req.logpath,
}
await loop.run_in_executor(
None,
_write_local_override_sync,
Path(config_dir),
name,
True,
overrides,
)
try:
await jail_service.reload_all(socket_path)
except Exception as exc: # noqa: BLE001
log.warning("reload_after_activate_failed", jail=name, error=str(exc))
log.info("jail_activated", jail=name)
return JailActivationResponse(
name=name,
active=True,
message=f"Jail {name!r} activated successfully.",
)
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)
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.",
)
# ---------------------------------------------------------------------------
# Filter discovery helpers (Task 2.1)
# ---------------------------------------------------------------------------
# Allowlist pattern for filter names used in path construction.
_SAFE_FILTER_NAME_RE: re.Pattern[str] = re.compile(
r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$"
)
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}")
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
# ---------------------------------------------------------------------------
# Public API — filter discovery (Task 2.1)
# ---------------------------------------------------------------------------
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 (Task 2.2)
# ---------------------------------------------------------------------------
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.
JailNotFoundInConfigError: 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,
)
# ---------------------------------------------------------------------------
# Action discovery helpers (Task 3.1)
# ---------------------------------------------------------------------------
# Allowlist pattern for action names used in path construction.
_SAFE_ACTION_NAME_RE: re.Pattern[str] = re.compile(
r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$"
)
class ActionNotFoundError(Exception):
"""Raised when the requested action name is not found in ``action.d/``."""
def __init__(self, name: str) -> None:
"""Initialise with the action name that was not found.
Args:
name: The action name that could not be located.
"""
self.name: str = name
super().__init__(f"Action not found: {name!r}")
class ActionAlreadyExistsError(Exception):
"""Raised when trying to create an action whose ``.conf`` or ``.local`` already exists."""
def __init__(self, name: str) -> None:
"""Initialise with the action name that already exists.
Args:
name: The action name that already exists.
"""
self.name: str = name
super().__init__(f"Action already exists: {name!r}")
class ActionReadonlyError(Exception):
"""Raised when trying to delete a shipped ``.conf`` action with no ``.local`` override."""
def __init__(self, name: str) -> None:
"""Initialise with the action name that cannot be deleted.
Args:
name: The action name that is read-only (shipped ``.conf`` only).
"""
self.name: str = name
super().__init__(
f"Action {name!r} is a shipped default (.conf only); "
"only user-created .local files can be deleted."
)
class ActionNameError(Exception):
"""Raised when an action name contains invalid characters."""
def _safe_action_name(name: str) -> str:
"""Validate *name* and return it unchanged or raise :class:`ActionNameError`.
Args:
name: Proposed action name (without extension).
Returns:
The name unchanged if valid.
Raises:
ActionNameError: If *name* contains unsafe characters.
"""
if not _SAFE_ACTION_NAME_RE.match(name):
raise ActionNameError(
f"Action 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_action_to_jails_map(
all_jails: dict[str, dict[str, str]],
active_names: set[str],
) -> dict[str, list[str]]:
"""Return a mapping of action base name → list of active jail names.
Iterates over every jail whose name is in *active_names*, resolves each
entry in its ``action`` config key to an action base name (stripping
``[…]`` parameter blocks), and records the jail against each base name.
Args:
all_jails: Merged jail config dict — ``{jail_name: {key: value}}``.
active_names: Set of jail names currently running in fail2ban.
Returns:
``{action_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_action = settings.get("action", "")
if not raw_action:
continue
for line in raw_action.splitlines():
stripped = line.strip()
if not stripped or stripped.startswith("#"):
continue
# Strip optional [key=value] parameter block to get the base name.
bracket = stripped.find("[")
base = stripped[:bracket].strip() if bracket != -1 else stripped
if base:
mapping.setdefault(base, []).append(jail_name)
return mapping
def _parse_actions_sync(
action_d: Path,
) -> list[tuple[str, str, str, bool, str]]:
"""Synchronously scan ``action.d/`` and return per-action tuples.
Each tuple contains:
- ``name`` — action base name (``"iptables"``).
- ``filename`` — actual filename (``"iptables.conf"``).
- ``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) actions.
Also discovers ``.local``-only files (user-created actions with no
corresponding ``.conf``).
Args:
action_d: Path to the ``action.d`` directory.
Returns:
List of ``(name, filename, content, has_local, source_path)`` tuples,
sorted by name.
"""
if not action_d.is_dir():
log.warning("action_d_not_found", path=str(action_d))
return []
conf_names: set[str] = set()
results: list[tuple[str, str, str, bool, str]] = []
# ---- .conf-based actions (with optional .local override) ----------------
for conf_path in sorted(action_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(
"action_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")
content = content + "\n" + local_content
except OSError as exc:
log.warning(
"action_local_read_error",
name=name,
path=str(local_path),
error=str(exc),
)
results.append((name, filename, content, has_local, str(conf_path)))
# ---- .local-only actions (user-created, no corresponding .conf) ----------
for local_path in sorted(action_d.glob("*.local")):
if not local_path.is_file():
continue
name = local_path.stem
if name in conf_names:
continue
try:
content = local_path.read_text(encoding="utf-8")
except OSError as exc:
log.warning(
"action_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("actions_scanned", count=len(results), action_d=str(action_d))
return results
def _append_jail_action_sync(
config_dir: Path,
jail_name: str,
action_entry: str,
) -> None:
"""Append an action entry to the ``action`` key in ``jail.d/{jail_name}.local``.
If the ``.local`` file already contains an ``action`` key under the jail
section, the new entry is appended as an additional line (multi-line
configparser format) unless it is already present. If no ``action`` key
exists, one is created.
Args:
config_dir: The fail2ban configuration root directory.
jail_name: Validated jail name.
action_entry: Full action string including any ``[…]`` parameters.
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)
existing_raw = parser.get(jail_name, "action") if parser.has_option(jail_name, "action") else ""
existing_lines = [
line.strip()
for line in existing_raw.splitlines()
if line.strip() and not line.strip().startswith("#")
]
# Extract base names from existing entries for duplicate checking.
def _base(entry: str) -> str:
bracket = entry.find("[")
return entry[:bracket].strip() if bracket != -1 else entry.strip()
new_base = _base(action_entry)
if not any(_base(e) == new_base for e in existing_lines):
existing_lines.append(action_entry)
if existing_lines:
# configparser multi-line: continuation lines start with whitespace.
new_value = existing_lines[0] + "".join(
f"\n {line}" for line in existing_lines[1:]
)
parser.set(jail_name, "action", new_value)
else:
parser.set(jail_name, "action", action_entry)
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_action_appended",
jail=jail_name,
action=action_entry,
path=str(local_path),
)
def _remove_jail_action_sync(
config_dir: Path,
jail_name: str,
action_name: str,
) -> None:
"""Remove an action entry from the ``action`` key in ``jail.d/{jail_name}.local``.
Reads the ``.local`` file, removes any ``action`` entries whose base name
matches *action_name*, and writes the result back atomically. If no
``.local`` file exists, this is a no-op.
Args:
config_dir: The fail2ban configuration root directory.
jail_name: Validated jail name.
action_name: Base name of the action to remove (without ``[…]``).
Raises:
ConfigWriteError: If writing fails.
"""
jail_d = config_dir / "jail.d"
local_path = jail_d / f"{jail_name}.local"
if not local_path.is_file():
return
parser = _build_parser()
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),
)
return
if not parser.has_section(jail_name) or not parser.has_option(jail_name, "action"):
return
existing_raw = parser.get(jail_name, "action")
existing_lines = [
line.strip()
for line in existing_raw.splitlines()
if line.strip() and not line.strip().startswith("#")
]
def _base(entry: str) -> str:
bracket = entry.find("[")
return entry[:bracket].strip() if bracket != -1 else entry.strip()
filtered = [e for e in existing_lines if _base(e) != action_name]
if len(filtered) == len(existing_lines):
# Action was not found — silently return (idempotent).
return
if filtered:
new_value = filtered[0] + "".join(
f"\n {line}" for line in filtered[1:]
)
parser.set(jail_name, "action", new_value)
else:
parser.remove_option(jail_name, "action")
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_action_removed",
jail=jail_name,
action=action_name,
path=str(local_path),
)
def _write_action_local_sync(action_d: Path, name: str, content: str) -> None:
"""Write *content* to ``action.d/{name}.local`` atomically.
The write is atomic: content is written to a temp file first, then
renamed into place. The ``action.d/`` directory is created if absent.
Args:
action_d: Path to the ``action.d`` directory.
name: Validated action base name (used as filename stem).
content: Full serialized action content to write.
Raises:
ConfigWriteError: If writing fails.
"""
try:
action_d.mkdir(parents=True, exist_ok=True)
except OSError as exc:
raise ConfigWriteError(
f"Cannot create action.d directory: {exc}"
) from exc
local_path = action_d / f"{name}.local"
try:
with tempfile.NamedTemporaryFile(
mode="w",
encoding="utf-8",
dir=action_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("action_local_written", action=name, path=str(local_path))
# ---------------------------------------------------------------------------
# Public API — action discovery (Task 3.1)
# ---------------------------------------------------------------------------
async def list_actions(
config_dir: str,
socket_path: str,
) -> ActionListResponse:
"""Return all available actions from ``action.d/`` with active/inactive status.
Scans ``{config_dir}/action.d/`` for ``.conf`` files, merges any
corresponding ``.local`` overrides, parses each file into an
:class:`~app.models.config.ActionConfig`, and cross-references with the
currently running jails to determine which actions are active.
An action is considered *active* when its base name appears in the
``action`` 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.ActionListResponse` with all actions
sorted alphabetically, active ones carrying non-empty
``used_by_jails`` lists.
"""
action_d = Path(config_dir) / "action.d"
loop = asyncio.get_event_loop()
raw_actions: list[tuple[str, str, str, bool, str]] = await loop.run_in_executor(
None, _parse_actions_sync, action_d
)
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
action_to_jails = _build_action_to_jails_map(all_jails, active_names)
actions: list[ActionConfig] = []
for name, filename, content, has_local, source_path in raw_actions:
cfg = conffile_parser.parse_action_file(
content, name=name, filename=filename
)
used_by = sorted(action_to_jails.get(name, []))
actions.append(
ActionConfig(
name=cfg.name,
filename=cfg.filename,
before=cfg.before,
after=cfg.after,
actionstart=cfg.actionstart,
actionstop=cfg.actionstop,
actioncheck=cfg.actioncheck,
actionban=cfg.actionban,
actionunban=cfg.actionunban,
actionflush=cfg.actionflush,
definition_vars=cfg.definition_vars,
init_vars=cfg.init_vars,
active=len(used_by) > 0,
used_by_jails=used_by,
source_file=source_path,
has_local_override=has_local,
)
)
log.info("actions_listed", total=len(actions), active=sum(1 for a in actions if a.active))
return ActionListResponse(actions=actions, total=len(actions))
async def get_action(
config_dir: str,
socket_path: str,
name: str,
) -> ActionConfig:
"""Return a single action from ``action.d/`` with active/inactive status.
Reads ``{config_dir}/action.d/{name}.conf``, merges any ``.local``
override, and enriches the parsed :class:`~app.models.config.ActionConfig`
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: Action base name (e.g. ``"iptables"`` or ``"iptables.conf"``).
Returns:
:class:`~app.models.config.ActionConfig` with status fields populated.
Raises:
ActionNotFoundError: If no ``{name}.conf`` or ``{name}.local`` file
exists in ``action.d/``.
"""
if name.endswith(".conf"):
base_name = name[:-5]
elif name.endswith(".local"):
base_name = name[:-6]
else:
base_name = name
action_d = Path(config_dir) / "action.d"
conf_path = action_d / f"{base_name}.conf"
local_path = action_d / f"{base_name}.local"
loop = asyncio.get_event_loop()
def _read() -> tuple[str, bool, str]:
"""Read action 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(
"action_local_read_error",
name=base_name,
path=str(local_path),
error=str(exc),
)
return content, has_local, str(conf_path)
elif has_local:
content = local_path.read_text(encoding="utf-8")
return content, False, str(local_path)
else:
raise ActionNotFoundError(base_name)
content, has_local, source_path = await loop.run_in_executor(None, _read)
cfg = conffile_parser.parse_action_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
action_to_jails = _build_action_to_jails_map(all_jails, active_names)
used_by = sorted(action_to_jails.get(base_name, []))
log.info("action_fetched", name=base_name, active=len(used_by) > 0)
return ActionConfig(
name=cfg.name,
filename=cfg.filename,
before=cfg.before,
after=cfg.after,
actionstart=cfg.actionstart,
actionstop=cfg.actionstop,
actioncheck=cfg.actioncheck,
actionban=cfg.actionban,
actionunban=cfg.actionunban,
actionflush=cfg.actionflush,
definition_vars=cfg.definition_vars,
init_vars=cfg.init_vars,
active=len(used_by) > 0,
used_by_jails=used_by,
source_file=source_path,
has_local_override=has_local,
)
# ---------------------------------------------------------------------------
# Public API — action write operations (Task 3.2)
# ---------------------------------------------------------------------------
async def update_action(
config_dir: str,
socket_path: str,
name: str,
req: ActionUpdateRequest,
do_reload: bool = False,
) -> ActionConfig:
"""Update an action's ``.local`` override with new lifecycle command 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 ``action.d/{name}.local``. The
original ``.conf`` file is never modified.
Args:
config_dir: Absolute path to the fail2ban configuration directory.
socket_path: Path to the fail2ban Unix domain socket.
name: Action base name (e.g. ``"iptables"`` or ``"iptables.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.ActionConfig` reflecting the updated state.
Raises:
ActionNameError: If *name* contains invalid characters.
ActionNotFoundError: If no ``{name}.conf`` or ``{name}.local`` exists.
ConfigWriteError: If writing the ``.local`` file fails.
"""
base_name = name[:-5] if name.endswith((".conf", ".local")) else name
_safe_action_name(base_name)
current = await get_action(config_dir, socket_path, base_name)
update = ActionConfigUpdate(
actionstart=req.actionstart,
actionstop=req.actionstop,
actioncheck=req.actioncheck,
actionban=req.actionban,
actionunban=req.actionunban,
actionflush=req.actionflush,
definition_vars=req.definition_vars,
init_vars=req.init_vars,
)
merged = conffile_parser.merge_action_update(current, update)
content = conffile_parser.serialize_action_config(merged)
action_d = Path(config_dir) / "action.d"
loop = asyncio.get_event_loop()
await loop.run_in_executor(None, _write_action_local_sync, action_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_action_update_failed",
action=base_name,
error=str(exc),
)
log.info("action_updated", action=base_name, reload=do_reload)
return await get_action(config_dir, socket_path, base_name)
async def create_action(
config_dir: str,
socket_path: str,
req: ActionCreateRequest,
do_reload: bool = False,
) -> ActionConfig:
"""Create a brand-new user-defined action in ``action.d/{name}.local``.
No ``.conf`` is written; fail2ban loads ``.local`` files directly. If a
``.conf`` or ``.local`` file already exists for the requested name, an
:class:`ActionAlreadyExistsError` is raised.
Args:
config_dir: Absolute path to the fail2ban configuration directory.
socket_path: Path to the fail2ban Unix domain socket.
req: Action name and definition fields.
do_reload: When ``True``, trigger a full fail2ban reload after writing.
Returns:
:class:`~app.models.config.ActionConfig` for the newly created action.
Raises:
ActionNameError: If ``req.name`` contains invalid characters.
ActionAlreadyExistsError: If a ``.conf`` or ``.local`` already exists.
ConfigWriteError: If writing fails.
"""
_safe_action_name(req.name)
action_d = Path(config_dir) / "action.d"
conf_path = action_d / f"{req.name}.conf"
local_path = action_d / f"{req.name}.local"
def _check_not_exists() -> None:
if conf_path.is_file() or local_path.is_file():
raise ActionAlreadyExistsError(req.name)
loop = asyncio.get_event_loop()
await loop.run_in_executor(None, _check_not_exists)
cfg = ActionConfig(
name=req.name,
filename=f"{req.name}.local",
actionstart=req.actionstart,
actionstop=req.actionstop,
actioncheck=req.actioncheck,
actionban=req.actionban,
actionunban=req.actionunban,
actionflush=req.actionflush,
definition_vars=req.definition_vars,
init_vars=req.init_vars,
)
content = conffile_parser.serialize_action_config(cfg)
await loop.run_in_executor(None, _write_action_local_sync, action_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_action_create_failed",
action=req.name,
error=str(exc),
)
log.info("action_created", action=req.name, reload=do_reload)
return await get_action(config_dir, socket_path, req.name)
async def delete_action(
config_dir: str,
name: str,
) -> None:
"""Delete a user-created action's ``.local`` file.
Deletion rules:
- If only a ``.conf`` file exists (shipped default, no user override) →
:class:`ActionReadonlyError`.
- If a ``.local`` file exists (whether or not a ``.conf`` also exists) →
only the ``.local`` file is deleted.
- If neither file exists → :class:`ActionNotFoundError`.
Args:
config_dir: Absolute path to the fail2ban configuration directory.
name: Action base name (e.g. ``"iptables"``).
Raises:
ActionNameError: If *name* contains invalid characters.
ActionNotFoundError: If no action file is found for *name*.
ActionReadonlyError: If only a shipped ``.conf`` exists (no ``.local``).
ConfigWriteError: If deletion of the ``.local`` file fails.
"""
base_name = name[:-5] if name.endswith((".conf", ".local")) else name
_safe_action_name(base_name)
action_d = Path(config_dir) / "action.d"
conf_path = action_d / f"{base_name}.conf"
local_path = action_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 ActionNotFoundError(base_name)
if has_conf and not has_local:
raise ActionReadonlyError(base_name)
try:
local_path.unlink()
except OSError as exc:
raise ConfigWriteError(
f"Failed to delete {local_path}: {exc}"
) from exc
log.info("action_local_deleted", action=base_name, path=str(local_path))
await loop.run_in_executor(None, _delete)
async def assign_action_to_jail(
config_dir: str,
socket_path: str,
jail_name: str,
req: AssignActionRequest,
do_reload: bool = False,
) -> None:
"""Add an action to a jail by updating the jail's ``.local`` file.
Appends ``{req.action_name}[{params}]`` (or just ``{req.action_name}`` when
no params are given) to the ``action`` key in the ``[{jail_name}]`` section
of ``jail.d/{jail_name}.local``. If the action is already listed it is not
duplicated. If the ``.local`` file does not exist it is created.
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 action name and optional parameters.
do_reload: When ``True``, trigger a full fail2ban reload after writing.
Raises:
JailNameError: If *jail_name* contains invalid characters.
ActionNameError: If ``req.action_name`` contains invalid characters.
JailNotFoundInConfigError: If *jail_name* is not defined in any config
file.
ActionNotFoundError: If ``req.action_name`` does not exist in
``action.d/``.
ConfigWriteError: If writing fails.
"""
_safe_jail_name(jail_name)
_safe_action_name(req.action_name)
loop = asyncio.get_event_loop()
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)
action_d = Path(config_dir) / "action.d"
def _check_action() -> None:
if (
not (action_d / f"{req.action_name}.conf").is_file()
and not (action_d / f"{req.action_name}.local").is_file()
):
raise ActionNotFoundError(req.action_name)
await loop.run_in_executor(None, _check_action)
# Build the action string with optional parameters.
if req.params:
param_str = ", ".join(f"{k}={v}" for k, v in sorted(req.params.items()))
action_entry = f"{req.action_name}[{param_str}]"
else:
action_entry = req.action_name
await loop.run_in_executor(
None,
_append_jail_action_sync,
Path(config_dir),
jail_name,
action_entry,
)
if do_reload:
try:
await jail_service.reload_all(socket_path)
except Exception as exc: # noqa: BLE001
log.warning(
"reload_after_assign_action_failed",
jail=jail_name,
action=req.action_name,
error=str(exc),
)
log.info(
"action_assigned_to_jail",
jail=jail_name,
action=req.action_name,
reload=do_reload,
)
async def remove_action_from_jail(
config_dir: str,
socket_path: str,
jail_name: str,
action_name: str,
do_reload: bool = False,
) -> None:
"""Remove an action from a jail's ``.local`` config.
Reads ``jail.d/{jail_name}.local``, removes the line(s) that reference
``{action_name}`` from the ``action`` key (including any ``[…]`` parameter
blocks), and writes the file back atomically.
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.
action_name: Base name of the action to remove.
do_reload: When ``True``, trigger a full fail2ban reload after writing.
Raises:
JailNameError: If *jail_name* contains invalid characters.
ActionNameError: If *action_name* contains invalid characters.
JailNotFoundInConfigError: If *jail_name* is not defined in any config.
ConfigWriteError: If writing fails.
"""
_safe_jail_name(jail_name)
_safe_action_name(action_name)
loop = asyncio.get_event_loop()
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)
await loop.run_in_executor(
None,
_remove_jail_action_sync,
Path(config_dir),
jail_name,
action_name,
)
if do_reload:
try:
await jail_service.reload_all(socket_path)
except Exception as exc: # noqa: BLE001
log.warning(
"reload_after_remove_action_failed",
jail=jail_name,
action=action_name,
error=str(exc),
)
log.info(
"action_removed_from_jail",
jail=jail_name,
action=action_name,
reload=do_reload,
)
Reference in New Issue View Git Blame Copy Permalink