115 lines
3.6 KiB
Python
115 lines
3.6 KiB
Python
from enum import Enum
|
|
from typing import Annotated, List, Optional
|
|
|
|
from pydantic import BaseModel, Field, field_validator
|
|
from pydantic.types import StringConstraints
|
|
|
|
|
|
class FirewallTarget(str, Enum):
|
|
"""Firewall zone target types"""
|
|
|
|
ACCEPT = "ACCEPT"
|
|
REJECT = "REJECT"
|
|
DROP = "DROP"
|
|
DEFAULT = "default"
|
|
|
|
|
|
class Protocol(str, Enum):
|
|
"""Network protocols supported by firewalld"""
|
|
|
|
TCP = "tcp"
|
|
UDP = "udp"
|
|
|
|
|
|
class FirewallZone(BaseModel):
|
|
"""Represents a firewalld zone configuration"""
|
|
|
|
name: Annotated[str, StringConstraints(min_length=1)] = Field(
|
|
..., description="Name of the zone"
|
|
)
|
|
target: Optional[FirewallTarget] = Field(
|
|
default=FirewallTarget.DEFAULT, description="Zone target type"
|
|
)
|
|
services: List[str] = Field(
|
|
default_factory=list, description="List of enabled services"
|
|
)
|
|
ports: List[str] = Field(default_factory=list, description="List of enabled ports")
|
|
sources: List[str] = Field(
|
|
default_factory=list, description="List of source addresses"
|
|
)
|
|
interfaces: List[str] = Field(
|
|
default_factory=list, description="List of network interfaces"
|
|
)
|
|
|
|
|
|
class FirewallService(BaseModel):
|
|
"""Represents a firewalld service configuration"""
|
|
|
|
name: Annotated[str, StringConstraints(min_length=1)] = Field(
|
|
..., description="Name of the service"
|
|
)
|
|
ports: List[str] = Field(
|
|
default_factory=list, description="List of ports used by the service"
|
|
)
|
|
protocols: List[str] = Field(
|
|
default_factory=list, description="List of protocols used by the service"
|
|
)
|
|
modules: List[str] = Field(
|
|
default_factory=list,
|
|
description="List of kernel modules required by the service",
|
|
)
|
|
timeout: Optional[int] = Field(
|
|
default=0, ge=0, description="Timeout in seconds (0 for permanent)"
|
|
)
|
|
|
|
|
|
class FirewallPort(BaseModel):
|
|
"""Represents a firewalld port configuration"""
|
|
|
|
port: int = Field(..., ge=1, le=65535, description="Port number (1-65535)")
|
|
protocol: Protocol = Field(..., description="Network protocol (tcp/udp)")
|
|
timeout: Optional[int] = Field(
|
|
default=0, ge=0, description="Timeout in seconds (0 for permanent)"
|
|
)
|
|
|
|
def __str__(self) -> str:
|
|
return f"{self.port}/{self.protocol.value}"
|
|
|
|
|
|
class FirewallSource(BaseModel):
|
|
"""Represents a firewalld source configuration"""
|
|
|
|
address: Annotated[
|
|
str,
|
|
StringConstraints(pattern=r"^(?:[0-9]{1,3}\.){3}[0-9]{1,3}(?:/[0-9]{1,2})?$"),
|
|
] = Field(..., description="IP address or network in CIDR notation")
|
|
zone: Optional[str] = Field(default=None, description="Zone name for the source")
|
|
|
|
@field_validator("address")
|
|
def validate_ip_address(cls, v):
|
|
# Validate IP address octets
|
|
ip = v.split("/")[0]
|
|
octets = ip.split(".")
|
|
if not all(0 <= int(octet) <= 255 for octet in octets):
|
|
raise ValueError("Invalid IP address octets")
|
|
|
|
# Validate CIDR if present
|
|
if "/" in v:
|
|
cidr = int(v.split("/")[1])
|
|
if not 0 <= cidr <= 32:
|
|
raise ValueError("CIDR must be between 0 and 32")
|
|
return v
|
|
|
|
|
|
class FirewallRule(BaseModel):
|
|
"""Represents a complete firewalld rule"""
|
|
|
|
zone: str = Field(..., description="Zone name")
|
|
source: Optional[FirewallSource] = Field(
|
|
default=None, description="Source configuration"
|
|
)
|
|
port: Optional[FirewallPort] = Field(default=None, description="Port configuration")
|
|
service: Optional[str] = Field(default=None, description="Service name")
|
|
timeout: Optional[int] = Field(
|
|
default=0, ge=0, description="Rule timeout in seconds"
|
|
)
|