Files
BanGUI/backend/app/models/config.py
Lukas ab11ece001 Add fail2ban log viewer and service health to Config page
Task 2: adds a new Log tab to the Configuration page.

Backend:
- New Pydantic models: Fail2BanLogResponse, ServiceStatusResponse
  (backend/app/models/config.py)
- New service methods in config_service.py:
    read_fail2ban_log() — queries socket for log target/level, validates the
    resolved path against a safe-prefix allowlist (/var/log) to prevent
    path traversal, then reads the tail of the file via the existing
    _read_tail_lines() helper; optional substring filter applied server-side.
    get_service_status() — delegates to health_service.probe() and appends
    log level/target from the socket.
- New endpoints in routers/config.py:
    GET /api/config/fail2ban-log?lines=200&filter=...
    GET /api/config/service-status
  Both require authentication; log endpoint returns 400 for non-file log
  targets or path-traversal attempts, 502 when fail2ban is unreachable.

Frontend:
- New LogTab.tsx component:
    Service Health panel (Running/Offline badge, version, jail count, bans,
    failures, log level/target, offline warning banner).
    Log viewer with color-coded lines (error=red, warning=yellow,
    debug=grey), toolbar (filter input + debounce, lines selector, manual
    refresh, auto-refresh with interval selector), truncation notice, and
    auto-scroll to bottom on data updates.
  fetchData uses Promise.allSettled so a log-read failure never hides the
  service-health panel.
- Types: Fail2BanLogResponse, ServiceStatusResponse (types/config.ts)
- API functions: fetchFail2BanLog, fetchServiceStatus (api/config.ts)
- Endpoint constants (api/endpoints.ts)
- ConfigPage.tsx: Log tab added after existing tabs

Tests:
- Backend service tests: TestReadFail2BanLog (6), TestGetServiceStatus (2)
- Backend router tests: TestGetFail2BanLog (8), TestGetServiceStatus (3)
- Frontend: LogTab.test.tsx (8 tests)

Docs:
- Features.md: Log section added under Configuration View
- Architekture.md: config.py router and config_service.py descriptions updated
- Tasks.md: Task 2 marked done
2026-03-14 12:54:03 +01:00

894 lines
32 KiB
Python

"""Configuration view/edit Pydantic models.
Request, response, and domain models for the config router and service.
"""
from pydantic import BaseModel, ConfigDict, Field
# ---------------------------------------------------------------------------
# Ban-time escalation
# ---------------------------------------------------------------------------
class BantimeEscalation(BaseModel):
"""Incremental ban-time escalation configuration for a jail."""
model_config = ConfigDict(strict=True)
increment: bool = Field(
default=False,
description="Whether incremental banning is enabled.",
)
factor: float | None = Field(
default=None,
description="Multiplier applied to the base ban time on each repeat offence.",
)
formula: str | None = Field(
default=None,
description="Python expression evaluated to compute the escalated ban time.",
)
multipliers: str | None = Field(
default=None,
description="Space-separated integers used as per-offence multipliers.",
)
max_time: int | None = Field(
default=None,
description="Maximum ban duration in seconds when escalation is active.",
)
rnd_time: int | None = Field(
default=None,
description="Random jitter (seconds) added to each escalated ban time.",
)
overall_jails: bool = Field(
default=False,
description="Count repeat offences across all jails, not just the current one.",
)
class BantimeEscalationUpdate(BaseModel):
"""Partial update payload for ban-time escalation settings."""
model_config = ConfigDict(strict=True)
increment: bool | None = Field(default=None)
factor: float | None = Field(default=None)
formula: str | None = Field(default=None)
multipliers: str | None = Field(default=None)
max_time: int | None = Field(default=None)
rnd_time: int | None = Field(default=None)
overall_jails: bool | None = Field(default=None)
# ---------------------------------------------------------------------------
# Jail configuration models
# ---------------------------------------------------------------------------
class JailConfig(BaseModel):
"""Configuration snapshot of a single jail (editable fields)."""
model_config = ConfigDict(strict=True)
name: str = Field(..., description="Jail name as configured in fail2ban.")
ban_time: int = Field(..., description="Ban duration in seconds. -1 for permanent.")
max_retry: int = Field(..., ge=1, description="Number of failures before a ban is issued.")
find_time: int = Field(..., ge=1, description="Time window (seconds) for counting failures.")
fail_regex: list[str] = Field(default_factory=list, description="Failure detection regex patterns.")
ignore_regex: list[str] = Field(default_factory=list, description="Regex patterns that bypass the ban logic.")
log_paths: list[str] = Field(default_factory=list, description="Monitored log files.")
date_pattern: str | None = Field(default=None, description="Custom date pattern for log parsing.")
log_encoding: str = Field(default="UTF-8", description="Log file encoding.")
backend: str = Field(default="polling", description="Log monitoring backend.")
use_dns: str = Field(default="warn", description="DNS lookup mode: yes | warn | no | raw.")
prefregex: str = Field(default="", description="Prefix regex prepended to every failregex; empty means disabled.")
actions: list[str] = Field(default_factory=list, description="Names of actions attached to this jail.")
bantime_escalation: BantimeEscalation | None = Field(
default=None,
description="Incremental ban-time escalation settings, or None if not configured.",
)
class JailConfigResponse(BaseModel):
"""Response for ``GET /api/config/jails/{name}``."""
model_config = ConfigDict(strict=True)
jail: JailConfig
class JailConfigListResponse(BaseModel):
"""Response for ``GET /api/config/jails``."""
model_config = ConfigDict(strict=True)
jails: list[JailConfig] = Field(default_factory=list)
total: int = Field(..., ge=0)
class JailConfigUpdate(BaseModel):
"""Payload for ``PUT /api/config/jails/{name}``."""
model_config = ConfigDict(strict=True)
ban_time: int | None = Field(default=None, description="Ban duration in seconds. -1 for permanent.")
max_retry: int | None = Field(default=None, ge=1)
find_time: int | None = Field(default=None, ge=1)
fail_regex: list[str] | None = Field(default=None, description="Failure detection regex patterns.")
ignore_regex: list[str] | None = Field(default=None)
prefregex: str | None = Field(default=None, description="Prefix regex; None = skip, '' = clear, non-empty = set.")
date_pattern: str | None = Field(default=None)
dns_mode: str | None = Field(default=None, description="DNS lookup mode: yes | warn | no | raw.")
backend: str | None = Field(default=None, description="Log monitoring backend.")
log_encoding: str | None = Field(default=None, description="Log file encoding.")
enabled: bool | None = Field(default=None)
bantime_escalation: BantimeEscalationUpdate | None = Field(
default=None,
description="Incremental ban-time escalation settings to update.",
)
# ---------------------------------------------------------------------------
# Regex tester models
# ---------------------------------------------------------------------------
class RegexTestRequest(BaseModel):
"""Payload for ``POST /api/config/regex-test``."""
model_config = ConfigDict(strict=True)
log_line: str = Field(..., description="Sample log line to test against.")
fail_regex: str = Field(..., description="Regex pattern to match.")
class RegexTestResponse(BaseModel):
"""Result of a regex test."""
model_config = ConfigDict(strict=True)
matched: bool = Field(..., description="Whether the pattern matched the log line.")
groups: list[str] = Field(
default_factory=list,
description="Named groups captured by a successful match.",
)
error: str | None = Field(
default=None,
description="Compilation error message if the regex is invalid.",
)
# ---------------------------------------------------------------------------
# Global config models
# ---------------------------------------------------------------------------
class GlobalConfigResponse(BaseModel):
"""Response for ``GET /api/config/global``."""
model_config = ConfigDict(strict=True)
log_level: str
log_target: str
db_purge_age: int = Field(..., description="Seconds after which ban records are purged from the fail2ban DB.")
db_max_matches: int = Field(..., description="Maximum stored log-line matches per ban record.")
class GlobalConfigUpdate(BaseModel):
"""Payload for ``PUT /api/config/global``."""
model_config = ConfigDict(strict=True)
log_level: str | None = Field(
default=None,
description="Log level: CRITICAL, ERROR, WARNING, NOTICE, INFO, DEBUG.",
)
log_target: str | None = Field(
default=None,
description="Log target: STDOUT, STDERR, SYSLOG, SYSTEMD-JOURNAL, or a file path.",
)
db_purge_age: int | None = Field(default=None, ge=0)
db_max_matches: int | None = Field(default=None, ge=0)
# ---------------------------------------------------------------------------
# Log observation / preview models
# ---------------------------------------------------------------------------
class AddLogPathRequest(BaseModel):
"""Payload for ``POST /api/config/jails/{name}/logpath``."""
model_config = ConfigDict(strict=True)
log_path: str = Field(..., description="Absolute path to the log file to monitor.")
tail: bool = Field(
default=True,
description="If true, monitor from current end of file (tail). If false, read from the beginning.",
)
class LogPreviewRequest(BaseModel):
"""Payload for ``POST /api/config/preview-log``."""
model_config = ConfigDict(strict=True)
log_path: str = Field(..., description="Absolute path to the log file to preview.")
fail_regex: str = Field(..., description="Regex pattern to test against log lines.")
num_lines: int = Field(default=200, ge=1, le=5000, description="Number of lines to read from the end of the file.")
class LogPreviewLine(BaseModel):
"""A single log line with match information."""
model_config = ConfigDict(strict=True)
line: str
matched: bool
groups: list[str] = Field(default_factory=list)
class LogPreviewResponse(BaseModel):
"""Response for ``POST /api/config/preview-log``."""
model_config = ConfigDict(strict=True)
lines: list[LogPreviewLine] = Field(default_factory=list)
total_lines: int = Field(..., ge=0)
matched_count: int = Field(..., ge=0)
regex_error: str | None = Field(default=None, description="Set if the regex failed to compile.")
# ---------------------------------------------------------------------------
# Map color threshold models
# ---------------------------------------------------------------------------
class MapColorThresholdsResponse(BaseModel):
"""Response for ``GET /api/config/map-thresholds``."""
model_config = ConfigDict(strict=True)
threshold_high: int = Field(
..., description="Ban count for red coloring."
)
threshold_medium: int = Field(
..., description="Ban count for yellow coloring."
)
threshold_low: int = Field(
..., description="Ban count for green coloring."
)
class MapColorThresholdsUpdate(BaseModel):
"""Payload for ``PUT /api/config/map-thresholds``."""
model_config = ConfigDict(strict=True)
threshold_high: int = Field(..., gt=0, description="Ban count for red.")
threshold_medium: int = Field(
..., gt=0, description="Ban count for yellow."
)
threshold_low: int = Field(..., gt=0, description="Ban count for green.")
# ---------------------------------------------------------------------------
# Parsed filter file models
# ---------------------------------------------------------------------------
class FilterConfig(BaseModel):
"""Structured representation of a ``filter.d/*.conf`` file.
The ``active``, ``used_by_jails``, ``source_file``, and
``has_local_override`` fields are populated by
:func:`~app.services.config_file_service.list_filters` and
:func:`~app.services.config_file_service.get_filter`. When the model is
returned from the raw file-based endpoints (``/filters/{name}/parsed``),
these fields carry their default values.
"""
model_config = ConfigDict(strict=True)
name: str = Field(..., description="Filter base name, e.g. ``sshd``.")
filename: str = Field(..., description="Actual filename, e.g. ``sshd.conf``.")
# [INCLUDES]
before: str | None = Field(default=None, description="Included file read before this one.")
after: str | None = Field(default=None, description="Included file read after this one.")
# [DEFAULT] — free-form key=value pairs
variables: dict[str, str] = Field(
default_factory=dict,
description="Free-form ``[DEFAULT]`` section variables.",
)
# [Definition]
prefregex: str | None = Field(
default=None,
description="Prefix regex prepended to every failregex.",
)
failregex: list[str] = Field(
default_factory=list,
description="Failure detection regex patterns (one per list entry).",
)
ignoreregex: list[str] = Field(
default_factory=list,
description="Regex patterns that bypass ban logic.",
)
maxlines: int | None = Field(
default=None,
description="Maximum number of log lines accumulated for a single match attempt.",
)
datepattern: str | None = Field(
default=None,
description="Custom date-parsing pattern, or ``None`` for auto-detect.",
)
journalmatch: str | None = Field(
default=None,
description="Systemd journal match expression.",
)
# Active-status fields — populated by config_file_service.list_filters /
# get_filter; default to safe "inactive" values when not computed.
active: bool = Field(
default=False,
description=(
"``True`` when this filter is referenced by at least one currently "
"enabled (running) jail."
),
)
used_by_jails: list[str] = Field(
default_factory=list,
description=(
"Names of currently enabled jails that reference this filter. "
"Empty when ``active`` is ``False``."
),
)
source_file: str = Field(
default="",
description="Absolute path to the ``.conf`` source file for this filter.",
)
has_local_override: bool = Field(
default=False,
description=(
"``True`` when a ``.local`` override file exists alongside the "
"base ``.conf`` file."
),
)
class FilterConfigUpdate(BaseModel):
"""Partial update payload for a parsed filter file.
Only explicitly set (non-``None``) fields are written back.
"""
model_config = ConfigDict(strict=True)
before: str | None = Field(default=None)
after: str | None = Field(default=None)
variables: dict[str, str] | None = Field(default=None)
prefregex: str | None = Field(default=None)
failregex: list[str] | None = Field(default=None)
ignoreregex: list[str] | None = Field(default=None)
maxlines: int | None = Field(default=None)
datepattern: str | None = Field(default=None)
journalmatch: str | None = Field(default=None)
class FilterUpdateRequest(BaseModel):
"""Payload for ``PUT /api/config/filters/{name}``.
Accepts only the user-editable ``[Definition]`` fields. Fields left as
``None`` are not changed; the existing value from the merged conf/local is
preserved.
"""
model_config = ConfigDict(strict=True)
failregex: list[str] | None = Field(
default=None,
description="Updated failure-detection regex patterns. ``None`` = keep existing.",
)
ignoreregex: list[str] | None = Field(
default=None,
description="Updated bypass-ban regex patterns. ``None`` = keep existing.",
)
datepattern: str | None = Field(
default=None,
description="Custom date-parsing pattern. ``None`` = keep existing.",
)
journalmatch: str | None = Field(
default=None,
description="Systemd journal match expression. ``None`` = keep existing.",
)
class FilterCreateRequest(BaseModel):
"""Payload for ``POST /api/config/filters``.
Creates a new user-defined filter at ``filter.d/{name}.local``.
"""
model_config = ConfigDict(strict=True)
name: str = Field(
...,
description="Filter base name (e.g. ``my-custom-filter``). Must not already exist in ``filter.d/``.",
)
failregex: list[str] = Field(
default_factory=list,
description="Failure-detection regex patterns.",
)
ignoreregex: list[str] = Field(
default_factory=list,
description="Regex patterns that bypass ban logic.",
)
prefregex: str | None = Field(
default=None,
description="Prefix regex prepended to every failregex.",
)
datepattern: str | None = Field(
default=None,
description="Custom date-parsing pattern.",
)
journalmatch: str | None = Field(
default=None,
description="Systemd journal match expression.",
)
class AssignFilterRequest(BaseModel):
"""Payload for ``POST /api/config/jails/{jail_name}/filter``."""
model_config = ConfigDict(strict=True)
filter_name: str = Field(
...,
description="Filter base name to assign to the jail (e.g. ``sshd``).",
)
class FilterListResponse(BaseModel):
"""Response for ``GET /api/config/filters``."""
model_config = ConfigDict(strict=True)
filters: list[FilterConfig] = Field(
default_factory=list,
description=(
"All discovered filters, each annotated with active/inactive status "
"and the jails that reference them."
),
)
total: int = Field(..., ge=0, description="Total number of filters found.")
# ---------------------------------------------------------------------------
# Parsed action file models
# ---------------------------------------------------------------------------
class ActionConfig(BaseModel):
"""Structured representation of an ``action.d/*.conf`` file."""
model_config = ConfigDict(strict=True)
name: str = Field(..., description="Action base name, e.g. ``iptables``.")
filename: str = Field(..., description="Actual filename, e.g. ``iptables.conf``.")
# [INCLUDES]
before: str | None = Field(default=None)
after: str | None = Field(default=None)
# [Definition] — well-known lifecycle commands
actionstart: str | None = Field(
default=None,
description="Executed at jail start or first ban.",
)
actionstop: str | None = Field(
default=None,
description="Executed at jail stop.",
)
actioncheck: str | None = Field(
default=None,
description="Executed before each ban.",
)
actionban: str | None = Field(
default=None,
description="Executed to ban an IP. Tags: ``<ip>``, ``<name>``, ``<port>``.",
)
actionunban: str | None = Field(
default=None,
description="Executed to unban an IP.",
)
actionflush: str | None = Field(
default=None,
description="Executed to flush all bans on shutdown.",
)
# [Definition] — extra variables not covered by the well-known keys
definition_vars: dict[str, str] = Field(
default_factory=dict,
description="Additional ``[Definition]`` variables.",
)
# [Init] — runtime-configurable parameters
init_vars: dict[str, str] = Field(
default_factory=dict,
description="Runtime parameters that can be overridden per jail.",
)
# Active-status fields — populated by config_file_service.list_actions /
# get_action; default to safe "inactive" values when not computed.
active: bool = Field(
default=False,
description=(
"``True`` when this action is referenced by at least one currently "
"enabled (running) jail."
),
)
used_by_jails: list[str] = Field(
default_factory=list,
description=(
"Names of currently enabled jails that reference this action. "
"Empty when ``active`` is ``False``."
),
)
source_file: str = Field(
default="",
description="Absolute path to the ``.conf`` source file for this action.",
)
has_local_override: bool = Field(
default=False,
description=(
"``True`` when a ``.local`` override file exists alongside the "
"base ``.conf`` file."
),
)
class ActionConfigUpdate(BaseModel):
"""Partial update payload for a parsed action file."""
model_config = ConfigDict(strict=True)
before: str | None = Field(default=None)
after: str | None = Field(default=None)
actionstart: str | None = Field(default=None)
actionstop: str | None = Field(default=None)
actioncheck: str | None = Field(default=None)
actionban: str | None = Field(default=None)
actionunban: str | None = Field(default=None)
actionflush: str | None = Field(default=None)
definition_vars: dict[str, str] | None = Field(default=None)
init_vars: dict[str, str] | None = Field(default=None)
class ActionListResponse(BaseModel):
"""Response for ``GET /api/config/actions``."""
model_config = ConfigDict(strict=True)
actions: list[ActionConfig] = Field(
default_factory=list,
description=(
"All discovered actions, each annotated with active/inactive status "
"and the jails that reference them."
),
)
total: int = Field(..., ge=0, description="Total number of actions found.")
class ActionUpdateRequest(BaseModel):
"""Payload for ``PUT /api/config/actions/{name}``.
Accepts only the user-editable ``[Definition]`` lifecycle fields and
``[Init]`` parameters. Fields left as ``None`` are not changed.
"""
model_config = ConfigDict(strict=True)
actionstart: str | None = Field(
default=None,
description="Updated ``actionstart`` command. ``None`` = keep existing.",
)
actionstop: str | None = Field(
default=None,
description="Updated ``actionstop`` command. ``None`` = keep existing.",
)
actioncheck: str | None = Field(
default=None,
description="Updated ``actioncheck`` command. ``None`` = keep existing.",
)
actionban: str | None = Field(
default=None,
description="Updated ``actionban`` command. ``None`` = keep existing.",
)
actionunban: str | None = Field(
default=None,
description="Updated ``actionunban`` command. ``None`` = keep existing.",
)
actionflush: str | None = Field(
default=None,
description="Updated ``actionflush`` command. ``None`` = keep existing.",
)
definition_vars: dict[str, str] | None = Field(
default=None,
description="Additional ``[Definition]`` variables to set. ``None`` = keep existing.",
)
init_vars: dict[str, str] | None = Field(
default=None,
description="``[Init]`` parameters to set. ``None`` = keep existing.",
)
class ActionCreateRequest(BaseModel):
"""Payload for ``POST /api/config/actions``.
Creates a new user-defined action at ``action.d/{name}.local``.
"""
model_config = ConfigDict(strict=True)
name: str = Field(
...,
description="Action base name (e.g. ``my-custom-action``). Must not already exist.",
)
actionstart: str | None = Field(default=None, description="Command to execute at jail start.")
actionstop: str | None = Field(default=None, description="Command to execute at jail stop.")
actioncheck: str | None = Field(default=None, description="Command to execute before each ban.")
actionban: str | None = Field(default=None, description="Command to execute to ban an IP.")
actionunban: str | None = Field(default=None, description="Command to execute to unban an IP.")
actionflush: str | None = Field(default=None, description="Command to flush all bans on shutdown.")
definition_vars: dict[str, str] = Field(
default_factory=dict,
description="Additional ``[Definition]`` variables.",
)
init_vars: dict[str, str] = Field(
default_factory=dict,
description="``[Init]`` runtime parameters.",
)
class AssignActionRequest(BaseModel):
"""Payload for ``POST /api/config/jails/{jail_name}/action``."""
model_config = ConfigDict(strict=True)
action_name: str = Field(
...,
description="Action base name to add to the jail (e.g. ``iptables-multiport``).",
)
params: dict[str, str] = Field(
default_factory=dict,
description=(
"Optional per-jail action parameters written as "
"``action_name[key=value, ...]`` in the jail config."
),
)
# ---------------------------------------------------------------------------
# Jail file config models (Task 6.1)
# ---------------------------------------------------------------------------
class JailSectionConfig(BaseModel):
"""Settings within a single [jailname] section of a jail.d file."""
model_config = ConfigDict(strict=True)
enabled: bool | None = Field(default=None, description="Whether this jail is enabled.")
port: str | None = Field(default=None, description="Port(s) to monitor (e.g. 'ssh' or '22,2222').")
filter: str | None = Field(default=None, description="Filter name to use (e.g. 'sshd').")
logpath: list[str] = Field(default_factory=list, description="Log file paths to monitor.")
maxretry: int | None = Field(default=None, ge=1, description="Failures before banning.")
findtime: int | None = Field(default=None, ge=1, description="Time window in seconds for counting failures.")
bantime: int | None = Field(default=None, description="Ban duration in seconds. -1 for permanent.")
action: list[str] = Field(default_factory=list, description="Action references.")
backend: str | None = Field(default=None, description="Log monitoring backend.")
extra: dict[str, str] = Field(default_factory=dict, description="Additional settings not captured by named fields.")
class JailFileConfig(BaseModel):
"""Structured representation of a jail.d/*.conf file."""
model_config = ConfigDict(strict=True)
filename: str = Field(..., description="Filename including extension (e.g. 'sshd.conf').")
jails: dict[str, JailSectionConfig] = Field(
default_factory=dict,
description="Mapping of jail name → settings for each [section] in the file.",
)
class JailFileConfigUpdate(BaseModel):
"""Partial update payload for a jail.d file."""
model_config = ConfigDict(strict=True)
jails: dict[str, JailSectionConfig] | None = Field(
default=None,
description="Jail section updates. Only jails present in this dict are updated.",
)
# ---------------------------------------------------------------------------
# Inactive jail models (Stage 1)
# ---------------------------------------------------------------------------
class InactiveJail(BaseModel):
"""A jail defined in fail2ban config files that is not currently active.
A jail is considered inactive when its ``enabled`` key is ``false`` (or
absent from the config, since fail2ban defaults to disabled) **or** when it
is explicitly enabled in config but fail2ban is not reporting it as
running.
"""
model_config = ConfigDict(strict=True)
name: str = Field(..., description="Jail name from the config section header.")
filter: str = Field(
...,
description=(
"Filter name used by this jail. May include fail2ban mode suffix, "
"e.g. ``sshd[mode=normal]``."
),
)
actions: list[str] = Field(
default_factory=list,
description="Action references listed in the config (raw strings).",
)
port: str | None = Field(
default=None,
description="Port(s) to monitor, e.g. ``ssh`` or ``22,2222``.",
)
logpath: list[str] = Field(
default_factory=list,
description="Log file paths to monitor.",
)
bantime: str | None = Field(
default=None,
description="Ban duration as a raw config string, e.g. ``10m`` or ``-1``.",
)
findtime: str | None = Field(
default=None,
description="Failure-counting window as a raw config string, e.g. ``10m``.",
)
maxretry: int | None = Field(
default=None,
description="Number of failures before a ban is issued.",
)
# ---- Extended fields for full GUI display ----
ban_time_seconds: int = Field(
default=600,
description="Ban duration in seconds, parsed from bantime string.",
)
find_time_seconds: int = Field(
default=600,
description="Failure-counting window in seconds, parsed from findtime string.",
)
log_encoding: str = Field(
default="auto",
description="Log encoding, e.g. ``utf-8`` or ``auto``.",
)
backend: str = Field(
default="auto",
description="Log-monitoring backend, e.g. ``auto``, ``pyinotify``, ``polling``.",
)
date_pattern: str | None = Field(
default=None,
description="Date pattern for log parsing, or None for auto-detect.",
)
use_dns: str = Field(
default="warn",
description="DNS resolution mode: ``yes``, ``warn``, ``no``, or ``raw``.",
)
prefregex: str = Field(
default="",
description="Prefix regex prepended to every failregex.",
)
fail_regex: list[str] = Field(
default_factory=list,
description="List of failure regex patterns.",
)
ignore_regex: list[str] = Field(
default_factory=list,
description="List of ignore regex patterns.",
)
bantime_escalation: BantimeEscalation | None = Field(
default=None,
description="Ban-time escalation configuration, if enabled.",
)
source_file: str = Field(
...,
description="Absolute path to the config file where this jail is defined.",
)
enabled: bool = Field(
...,
description=(
"Effective ``enabled`` value from the merged config. ``False`` for "
"inactive jails that appear in this list."
),
)
class InactiveJailListResponse(BaseModel):
"""Response for ``GET /api/config/jails/inactive``."""
model_config = ConfigDict(strict=True)
jails: list[InactiveJail] = Field(default_factory=list)
total: int = Field(..., ge=0)
class ActivateJailRequest(BaseModel):
"""Optional override values when activating an inactive jail.
All fields are optional. Omitted fields are not written to the
``.local`` override file so that fail2ban falls back to its default
values.
"""
model_config = ConfigDict(strict=True)
bantime: str | None = Field(
default=None,
description="Override ban duration, e.g. ``1h`` or ``3600``.",
)
findtime: str | None = Field(
default=None,
description="Override failure-counting window, e.g. ``10m``.",
)
maxretry: int | None = Field(
default=None,
ge=1,
description="Override maximum failures before a ban.",
)
port: str | None = Field(
default=None,
description="Override port(s) to monitor.",
)
logpath: list[str] | None = Field(
default=None,
description="Override log file paths.",
)
class JailActivationResponse(BaseModel):
"""Response for jail activation and deactivation endpoints."""
model_config = ConfigDict(strict=True)
name: str = Field(..., description="Name of the affected jail.")
active: bool = Field(
...,
description="New activation state: ``True`` after activate, ``False`` after deactivate.",
)
message: str = Field(..., description="Human-readable result message.")
# ---------------------------------------------------------------------------
# fail2ban log viewer models
# ---------------------------------------------------------------------------
class Fail2BanLogResponse(BaseModel):
"""Response for ``GET /api/config/fail2ban-log``."""
model_config = ConfigDict(strict=True)
log_path: str = Field(..., description="Resolved absolute path of the log file being read.")
lines: list[str] = Field(default_factory=list, description="Log lines returned (tail, optionally filtered).")
total_lines: int = Field(..., ge=0, description="Total number of lines in the file before filtering.")
log_level: str = Field(..., description="Current fail2ban log level.")
log_target: str = Field(..., description="Current fail2ban log target (file path or special value).")
class ServiceStatusResponse(BaseModel):
"""Response for ``GET /api/config/service-status``."""
model_config = ConfigDict(strict=True)
online: bool = Field(..., description="Whether fail2ban is reachable via its socket.")
version: str | None = Field(default=None, description="fail2ban version string, or None when offline.")
jail_count: int = Field(default=0, ge=0, description="Number of currently active jails.")
total_bans: int = Field(default=0, ge=0, description="Aggregated current ban count across all jails.")
total_failures: int = Field(default=0, ge=0, description="Aggregated current failure count across all jails.")
log_level: str = Field(default="UNKNOWN", description="Current fail2ban log level.")
log_target: str = Field(default="UNKNOWN", description="Current fail2ban log target.")