152 lines
4.6 KiB
Python
152 lines
4.6 KiB
Python
"""Shared domain exception classes used across routers and services."""
|
|
|
|
from __future__ import annotations
|
|
|
|
|
|
class JailNotFoundError(Exception):
|
|
"""Raised when a requested jail name does not exist."""
|
|
|
|
def __init__(self, name: str) -> None:
|
|
self.name = name
|
|
super().__init__(f"Jail not found: {name!r}")
|
|
|
|
|
|
class JailOperationError(Exception):
|
|
"""Raised when a fail2ban jail operation fails."""
|
|
|
|
|
|
class ConfigValidationError(Exception):
|
|
"""Raised when config values fail validation before applying."""
|
|
|
|
|
|
class ConfigOperationError(Exception):
|
|
"""Raised when a config payload update or command fails."""
|
|
|
|
|
|
class ServerOperationError(Exception):
|
|
"""Raised when a server control command (e.g. refresh) fails."""
|
|
|
|
|
|
class Fail2BanConnectionError(Exception):
|
|
"""Raised when the fail2ban socket is unreachable or returns an error."""
|
|
|
|
def __init__(self, message: str, socket_path: str) -> None:
|
|
"""Initialize with a human-readable message and the socket path.
|
|
|
|
Args:
|
|
message: Description of the connection problem.
|
|
socket_path: The fail2ban socket path that was targeted.
|
|
"""
|
|
self.socket_path: str = socket_path
|
|
super().__init__(f"{message} (socket: {socket_path})")
|
|
|
|
|
|
class Fail2BanProtocolError(Exception):
|
|
"""Raised when the response from fail2ban cannot be parsed."""
|
|
|
|
|
|
class FilterInvalidRegexError(Exception):
|
|
"""Raised when a regex pattern fails to compile."""
|
|
|
|
def __init__(self, pattern: str, error: str) -> None:
|
|
"""Initialize with the invalid pattern and compile error."""
|
|
self.pattern = pattern
|
|
self.error = error
|
|
super().__init__(f"Invalid regex {pattern!r}: {error}")
|
|
|
|
|
|
class JailNotFoundInConfigError(Exception):
|
|
"""Raised when the requested jail name is not defined in any config file."""
|
|
|
|
def __init__(self, name: str) -> None:
|
|
self.name = name
|
|
super().__init__(f"Jail not found in config: {name!r}")
|
|
|
|
|
|
class ConfigWriteError(Exception):
|
|
"""Raised when writing a configuration file modification fails."""
|
|
|
|
def __init__(self, message: str) -> None:
|
|
self.message = message
|
|
super().__init__(message)
|
|
|
|
|
|
class JailNameError(Exception):
|
|
"""Raised when a jail name contains invalid characters."""
|
|
|
|
|
|
class JailAlreadyActiveError(Exception):
|
|
"""Raised when trying to activate a jail that is already active."""
|
|
|
|
def __init__(self, name: str) -> None:
|
|
self.name = 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:
|
|
self.name = name
|
|
super().__init__(f"Jail is already inactive: {name!r}")
|
|
|
|
|
|
class FilterNotFoundError(Exception):
|
|
"""Raised when the requested filter name is not found."""
|
|
|
|
def __init__(self, name: str) -> None:
|
|
self.name = name
|
|
super().__init__(f"Filter not found: {name!r}")
|
|
|
|
|
|
class FilterAlreadyExistsError(Exception):
|
|
"""Raised when trying to create a filter whose `.conf` or `.local` already exists."""
|
|
|
|
def __init__(self, name: str) -> None:
|
|
self.name = name
|
|
super().__init__(f"Filter already exists: {name!r}")
|
|
|
|
|
|
class FilterNameError(Exception):
|
|
"""Raised when a filter name contains invalid characters."""
|
|
|
|
|
|
class FilterReadonlyError(Exception):
|
|
"""Raised when trying to delete a shipped `.conf` filter with no `.local` override."""
|
|
|
|
def __init__(self, name: str) -> None:
|
|
self.name = name
|
|
super().__init__(
|
|
f"Filter {name!r} is a shipped default (.conf only); only user-created .local files can be deleted."
|
|
)
|
|
|
|
|
|
class ActionNotFoundError(Exception):
|
|
"""Raised when the requested action name is not found."""
|
|
|
|
def __init__(self, name: str) -> None:
|
|
self.name = 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:
|
|
self.name = name
|
|
super().__init__(f"Action already exists: {name!r}")
|
|
|
|
|
|
class ActionNameError(Exception):
|
|
"""Raised when an action name contains invalid characters."""
|
|
|
|
|
|
class ActionReadonlyError(Exception):
|
|
"""Raised when trying to delete a shipped `.conf` action with no `.local` override."""
|
|
|
|
def __init__(self, name: str) -> None:
|
|
self.name = name
|
|
super().__init__(
|
|
f"Action {name!r} is a shipped default (.conf only); only user-created .local files can be deleted."
|
|
)
|