"""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.") 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.""" 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: ````, ````, ````.", ) 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.", )