"""Shared domain exception classes used across routers and services. Exception Taxonomy ================== All domain exceptions inherit from one of these base categories: - **NotFoundError** (404): Domain entity not found - **BadRequestError** (400): Invalid input, validation failure, invalid identifiers - **ConflictError** (409): State conflict, resource already exists, invalid state transition - **OperationError** (500): Operation failure, write errors - **ServiceUnavailableError** (503): Infrastructure/external service issues Service exceptions inherit from the appropriate category, allowing routers to handle categories rather than individual exception types. Exception handlers in main.py register only base category types. Example: def get_jail(name: str) -> Jail: # Raises JailNotFoundError (subclass of NotFoundError) ... @app.exception_handler(NotFoundError) async def handle_not_found(request, exc): return JSONResponse(status_code=404, content={"detail": str(exc)}) See Backend-Development.md for the complete exception contract. """ from __future__ import annotations # --------------------------------------------------------------------------- # Exception Base Classes (Categories) # --------------------------------------------------------------------------- class DomainError(Exception): """Base class for all domain exceptions.""" pass class NotFoundError(DomainError): """Raised when a requested domain entity is not found. HTTP 404.""" pass class BadRequestError(DomainError): """Raised for invalid input, validation failures, or invalid identifiers. HTTP 400.""" pass class ConflictError(DomainError): """Raised for state conflicts or resource constraints. HTTP 409.""" pass class OperationError(DomainError): """Raised when a domain operation fails (write, update, etc.). HTTP 500.""" pass class ServiceUnavailableError(DomainError): """Raised for infrastructure or external service issues. HTTP 503.""" pass # --------------------------------------------------------------------------- # Jail-Specific Exceptions # --------------------------------------------------------------------------- class JailNotFoundError(NotFoundError): """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(ConflictError): """Raised when a jail state operation fails (e.g. start/stop already in progress).""" class ConfigValidationError(BadRequestError): """Raised when config values fail validation before applying.""" class ConfigOperationError(BadRequestError): """Raised when a config payload update or command fails.""" class ConfigDirError(ServiceUnavailableError): """Raised when the fail2ban config directory is missing or inaccessible.""" class ConfigFileNotFoundError(NotFoundError): """Raised when a requested config file does not exist.""" def __init__(self, filename: str) -> None: """Initialize with the filename that was not found. Args: filename: The filename that could not be located. """ self.filename = filename super().__init__(f"Config file not found: {filename!r}") class ConfigFileExistsError(ConflictError): """Raised when trying to create a file that already exists.""" def __init__(self, filename: str) -> None: """Initialize with the filename that already exists. Args: filename: The filename that conflicts. """ self.filename = filename super().__init__(f"Config file already exists: {filename!r}") class ConfigFileWriteError(OperationError): """Raised when a file cannot be written (permissions, disk full, etc.).""" class ConfigFileNameError(BadRequestError): """Raised when a supplied filename is invalid or unsafe.""" class ServerOperationError(BadRequestError): """Raised when a server control command (e.g. refresh) fails.""" class Fail2BanConnectionError(ServiceUnavailableError): """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(ServiceUnavailableError): """Raised when the response from fail2ban cannot be parsed.""" class FilterInvalidRegexError(BadRequestError): """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(NotFoundError): """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(OperationError): """Raised when writing a configuration file modification fails.""" def __init__(self, message: str) -> None: self.message = message super().__init__(message) class JailNameError(BadRequestError): """Raised when a jail name contains invalid characters.""" class JailAlreadyActiveError(ConflictError): """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(ConflictError): """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(NotFoundError): """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(ConflictError): """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(BadRequestError): """Raised when a filter name contains invalid characters.""" class FilterReadonlyError(ConflictError): """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(NotFoundError): """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(ConflictError): """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(BadRequestError): """Raised when an action name contains invalid characters.""" class ActionReadonlyError(ConflictError): """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." )