feat(backend): add conf-file parser and extend config models

- Add conffile_parser.py: reads, writes and manipulates fail2ban .conf
  files while preserving comments and section structure
- Extend config models with ActionConfig, FilterConfig, ConfFileContent,
  and related Pydantic schemas for jails, actions, and filters
This commit is contained in:
2026-03-13 13:47:09 +01:00
parent d6da81131f
commit 63b48849a7
2 changed files with 875 additions and 0 deletions

View File

@@ -267,3 +267,183 @@ class MapColorThresholdsUpdate(BaseModel):
..., 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."""
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.",
)
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)
# ---------------------------------------------------------------------------
# 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.",
)
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)
# ---------------------------------------------------------------------------
# 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.",
)