264 lines
9.7 KiB
Python
264 lines
9.7 KiB
Python
|
from typing import List, Optional, Annotated, Union
|
|||
|
|
from enum import Enum
|
|||
|
|
from pydantic import BaseModel, Field, validator
|
|||
|
|
import ipaddress
|
|||
|
|
|
|||
|
|
|
|||
|
|
class Protocol(str, Enum):
|
|||
|
|
"""
|
|||
|
|
Enum для сетевых протоколов UFW.
|
|||
|
|
"""
|
|||
|
|
TCP = "tcp"
|
|||
|
|
UDP = "udp"
|
|||
|
|
BOTH = "both"
|
|||
|
|
|
|||
|
|
@property
|
|||
|
|
def protocols(self) -> List[str]:
|
|||
|
|
"""Возвращает список протоколов для использования в команде UFW."""
|
|||
|
|
return ["tcp", "udp"] if self == Protocol.BOTH else [self.value]
|
|||
|
|
|
|||
|
|
|
|||
|
|
class Action(str, Enum):
|
|||
|
|
"""
|
|||
|
|
Enum для действий UFW.
|
|||
|
|
|
|||
|
|
Attributes:
|
|||
|
|
ALLOW: Разрешить трафик
|
|||
|
|
DENY: Запретить трафик без уведомления
|
|||
|
|
REJECT: Запретить трафик с уведомлением отправителю
|
|||
|
|
"""
|
|||
|
|
ALLOW = "allow"
|
|||
|
|
DENY = "deny"
|
|||
|
|
REJECT = "reject"
|
|||
|
|
|
|||
|
|
|
|||
|
|
class Direction(str, Enum):
|
|||
|
|
"""
|
|||
|
|
Enum для направления трафика UFW.
|
|||
|
|
|
|||
|
|
Attributes:
|
|||
|
|
IN: Входящий трафик
|
|||
|
|
OUT: Исходящий трафик
|
|||
|
|
"""
|
|||
|
|
IN = "in"
|
|||
|
|
OUT = "out"
|
|||
|
|
|
|||
|
|
|
|||
|
|
class UFWRule(BaseModel):
|
|||
|
|
"""Модель правила UFW"""
|
|||
|
|
action: Action
|
|||
|
|
direction: Direction
|
|||
|
|
from_ip: str
|
|||
|
|
to_ip: str
|
|||
|
|
port: Union[int, str]
|
|||
|
|
protocol: Protocol
|
|||
|
|
comment: Optional[str] = None
|
|||
|
|
|
|||
|
|
@validator("port")
|
|||
|
|
def validate_port(cls, v):
|
|||
|
|
"""
|
|||
|
|
Проверяет корректность порта или диапазона портов.
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
v: Порт (число) или диапазон портов (строка вида "3000:3999")
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
Union[int, str]: Валидированное значение порта
|
|||
|
|
|
|||
|
|
Raises:
|
|||
|
|
ValueError: Если порт или диапазон портов некорректен
|
|||
|
|
"""
|
|||
|
|
if isinstance(v, int):
|
|||
|
|
if not (1 <= v <= 65535):
|
|||
|
|
raise ValueError(f"Порт должен быть в диапазоне 1-65535, получено: {v}")
|
|||
|
|
return v
|
|||
|
|
elif isinstance(v, str):
|
|||
|
|
try:
|
|||
|
|
start, end = map(int, v.split(':'))
|
|||
|
|
if not (1 <= start <= end <= 65535):
|
|||
|
|
raise ValueError
|
|||
|
|
return v
|
|||
|
|
except (ValueError, TypeError):
|
|||
|
|
raise ValueError(f"Некорректный диапазон портов: {v}. Формат: start:end, где 1 <= start <= end <= 65535")
|
|||
|
|
else:
|
|||
|
|
raise ValueError(f"Порт должен быть числом или строкой с диапазоном, получено: {type(v)}")
|
|||
|
|
|
|||
|
|
@validator("from_ip", "to_ip")
|
|||
|
|
def validate_ip(cls, v):
|
|||
|
|
"""Проверяет корректность IP адреса"""
|
|||
|
|
try:
|
|||
|
|
# Пробуем создать объект IP-сети, что автоматически проверит корректность адреса
|
|||
|
|
ipaddress.ip_network(v, strict=False)
|
|||
|
|
return v
|
|||
|
|
except ValueError:
|
|||
|
|
raise ValueError(f"Некорректный IP адрес или подсеть: {v}")
|
|||
|
|
|
|||
|
|
@validator("to_ip")
|
|||
|
|
def validate_ip_version_compatibility(cls, v, values):
|
|||
|
|
"""Проверяет совместимость версий IP адресов"""
|
|||
|
|
if "from_ip" not in values:
|
|||
|
|
return v
|
|||
|
|
|
|||
|
|
# Определяем версии IP
|
|||
|
|
from_ip = ipaddress.ip_network(values["from_ip"], strict=False)
|
|||
|
|
to_ip = ipaddress.ip_network(v, strict=False)
|
|||
|
|
|
|||
|
|
if isinstance(from_ip, ipaddress.IPv6Network) != isinstance(to_ip, ipaddress.IPv6Network):
|
|||
|
|
raise ValueError("Несовместимые версии IP: нельзя смешивать IPv4 и IPv6 в одном правиле")
|
|||
|
|
|
|||
|
|
return v
|
|||
|
|
|
|||
|
|
@validator("action", pre=True)
|
|||
|
|
def validate_action(cls, v):
|
|||
|
|
"""Приводит значение action к нижнему регистру перед валидацией"""
|
|||
|
|
if isinstance(v, str):
|
|||
|
|
return v.lower()
|
|||
|
|
return v
|
|||
|
|
|
|||
|
|
@validator("protocol", pre=True)
|
|||
|
|
def validate_protocol(cls, v):
|
|||
|
|
"""Приводит значение protocol к нижнему регистру перед валидацией"""
|
|||
|
|
if isinstance(v, str):
|
|||
|
|
return v.lower()
|
|||
|
|
return v
|
|||
|
|
|
|||
|
|
def to_ufw_command(self) -> str:
|
|||
|
|
"""Преобразует правило в команду UFW."""
|
|||
|
|
cmd_parts = ["ufw"]
|
|||
|
|
|
|||
|
|
# Добавляем действие
|
|||
|
|
cmd_parts.append(self.action.value)
|
|||
|
|
|
|||
|
|
# Добавляем направление
|
|||
|
|
cmd_parts.append(self.direction.value)
|
|||
|
|
|
|||
|
|
# Добавляем from с IP
|
|||
|
|
cmd_parts.extend(["from", self.from_ip])
|
|||
|
|
|
|||
|
|
# Добавляем to с IP
|
|||
|
|
cmd_parts.extend(["to", self.to_ip])
|
|||
|
|
|
|||
|
|
# Добавляем порт и протокол
|
|||
|
|
port_str = str(self.port) # Работает как с числами, так и с диапазонами
|
|||
|
|
cmd_parts.extend(["port", port_str])
|
|||
|
|
|
|||
|
|
# Добавляем протокол
|
|||
|
|
proto = self.protocol.value
|
|||
|
|
if proto == "both":
|
|||
|
|
cmd_parts.append("proto tcp/udp")
|
|||
|
|
else:
|
|||
|
|
cmd_parts.extend(["proto", proto])
|
|||
|
|
|
|||
|
|
# Добавляем комментарий если есть
|
|||
|
|
if self.comment:
|
|||
|
|
cmd_parts.extend(["comment", f"'{self.comment}'"])
|
|||
|
|
|
|||
|
|
return " ".join(cmd_parts)
|
|||
|
|
|
|||
|
|
@classmethod
|
|||
|
|
def from_ufw_output(cls, line: str) -> "UFWRule":
|
|||
|
|
"""
|
|||
|
|
Создает экземпляр UFWRule из строки вывода команды 'ufw status numbered'.
|
|||
|
|
|
|||
|
|
Args:
|
|||
|
|
line: Строка вывода UFW (например, "[1] 80/tcp ALLOW IN 192.168.1.0/24 # Web Server")
|
|||
|
|
|
|||
|
|
Returns:
|
|||
|
|
UFWRule: Новый экземпляр правила
|
|||
|
|
"""
|
|||
|
|
import re
|
|||
|
|
|
|||
|
|
# Регулярное выражение для разбора строки правила UFW
|
|||
|
|
pattern = r"""^\s*
|
|||
|
|
\[\s*(?P<number>\d+)\s*\]\s*
|
|||
|
|
(?P<port>\d+(?::\d+)?)/(?P<protocol>tcp(?:/udp)?|udp)\s+
|
|||
|
|
(?P<action>ALLOW|DENY|REJECT)\s+(?P<direction>IN|OUT)\s+
|
|||
|
|
(?P<from_ip>[\w\d\./]+|\bAnywhere\b)
|
|||
|
|
(?:\s+\(v6\))?
|
|||
|
|
(?:\s+\((?:out|in)\))?
|
|||
|
|
\s*
|
|||
|
|
(?:\#\s*(?P<comment>.*?))?
|
|||
|
|
\s*$
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
match = re.match(pattern, line, re.VERBOSE | re.IGNORECASE)
|
|||
|
|
if not match:
|
|||
|
|
raise ValueError(f"Невозможно разобрать строку правила UFW: {line}")
|
|||
|
|
|
|||
|
|
groups = match.groupdict()
|
|||
|
|
|
|||
|
|
# Преобразуем протокол в enum
|
|||
|
|
proto_str = groups['protocol'].lower()
|
|||
|
|
protocol = Protocol.BOTH if proto_str == "tcp/udp" else \
|
|||
|
|
Protocol.UDP if proto_str == "udp" else Protocol.TCP
|
|||
|
|
|
|||
|
|
# Преобразуем действие в enum (игнорируем регистр)
|
|||
|
|
action_str = groups['action'].lower()
|
|||
|
|
action = Action.ALLOW if action_str == "allow" else \
|
|||
|
|
Action.DENY if action_str == "deny" else Action.REJECT
|
|||
|
|
|
|||
|
|
# Преобразуем направление в enum
|
|||
|
|
direction_str = groups['direction'].lower()
|
|||
|
|
direction = Direction.IN if direction_str == "in" else Direction.OUT
|
|||
|
|
|
|||
|
|
# Обрабатываем IP адреса
|
|||
|
|
from_ip = groups['from_ip'].strip()
|
|||
|
|
if from_ip.lower() == "anywhere":
|
|||
|
|
from_ip = "0.0.0.0/0"
|
|||
|
|
|
|||
|
|
# Для исходящих правил меняем местами from и to
|
|||
|
|
to_ip = "0.0.0.0/0" # По умолчанию anywhere
|
|||
|
|
|
|||
|
|
# Обрабатываем порт
|
|||
|
|
port = groups['port']
|
|||
|
|
if ':' in port:
|
|||
|
|
# Оставляем как строку для диапазона
|
|||
|
|
port = port.strip()
|
|||
|
|
else:
|
|||
|
|
# Преобразуем в число для одиночного порта
|
|||
|
|
port = int(port)
|
|||
|
|
|
|||
|
|
# Обрабатываем комментарий
|
|||
|
|
comment = groups.get('comment', '').strip() if groups.get('comment') else None
|
|||
|
|
|
|||
|
|
return cls(
|
|||
|
|
action=action,
|
|||
|
|
direction=direction,
|
|||
|
|
from_ip=from_ip,
|
|||
|
|
to_ip=to_ip,
|
|||
|
|
port=port,
|
|||
|
|
protocol=protocol,
|
|||
|
|
comment=comment
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
|
|||
|
|
class UFWRulesResponse(BaseModel):
|
|||
|
|
"""
|
|||
|
|
Модель для представления списка правил UFW.
|
|||
|
|
|
|||
|
|
Attributes:
|
|||
|
|
enabled: Включен ли UFW
|
|||
|
|
rules: Список правил UFW
|
|||
|
|
total_rules: Общее количество правил
|
|||
|
|
"""
|
|||
|
|
enabled: bool = Field(..., description="Включен ли UFW")
|
|||
|
|
rules: List[UFWRule] = Field(default_factory=list, description="Список правил UFW")
|
|||
|
|
|
|||
|
|
@property
|
|||
|
|
def total_rules(self) -> int:
|
|||
|
|
"""Возвращает общее количество правил."""
|
|||
|
|
return len(self.rules)
|
|||
|
|
|
|||
|
|
|
|||
|
|
class UFWStatus(BaseModel):
|
|||
|
|
"""
|
|||
|
|
Модель для представления статуса UFW.
|
|||
|
|
|
|||
|
|
Attributes:
|
|||
|
|
enabled: Включен ли UFW
|
|||
|
|
message: Сообщение о статусе
|
|||
|
|
last_checked: Время последней проверки статуса
|
|||
|
|
"""
|
|||
|
|
enabled: bool = Field(..., description="Включен ли UFW")
|
|||
|
|
message: str = Field(..., description="Сообщение о статусе")
|