awg-dash/backend/awg_api/store/models.py

"""Модель данных хранилища awg-dash (issue #20).

Каждая сущность описана неизменяемым dataclass-ом; идентификатор ``None``
означает «ещё не сохранено». Строки времени — UTC ISO-8601, суффикс ``Z``.
"""

from __future__ import annotations

from dataclasses import dataclass
from enum import StrEnum
from typing import Any

RULE_KINDS: tuple[str, ...] = ("domain", "cidr", "ipset")

DEVICE_POLICIES: tuple[str, ...] = ("default", "tunnel", "direct", "block")

AUDIT_STATUS_OK = "ok"
AUDIT_STATUS_FAILED = "failed"
AUDIT_STATUS_STAGED = "staged"


class Mode(StrEnum):
    """Режим маршрутизации шлюза."""

    OFF = "off"
    SPLIT = "split"
    FULL = "full"

    @classmethod
    def values(cls) -> tuple[str, ...]:
        return tuple(member.value for member in cls)


class Entity(StrEnum):
    """Имена сущностей для audit_log и снимков состояния."""

    RULE = "rule"
    EXCEPTION = "exception"
    MODE = "mode"
    GEOIP_SCHEDULE = "geoip_schedule"
    CONFIG = "config"
    DEVICE = "device"
    APPLY = "apply"


@dataclass(frozen=True, slots=True)
class Rule:
    """Правило маршрутизации (domain / cidr / ipset -> target)."""

    name: str
    kind: str
    match: str
    target: str
    priority: int = 100
    enabled: bool = True
    id: int | None = None
    created_at: str | None = None
    updated_at: str | None = None

    def as_dict(self) -> dict[str, Any]:
        return {
            "id": self.id,
            "name": self.name,
            "kind": self.kind,
            "match": self.match,
            "target": self.target,
            "priority": self.priority,
            "enabled": self.enabled,
        }


@dataclass(frozen=True, slots=True)
class RuleException:
    """Исключение из правил (host/domain, который никогда не маршрутизируется в туннель)."""

    match: str
    reason: str = ""
    enabled: bool = True
    id: int | None = None
    created_at: str | None = None
    updated_at: str | None = None

    def as_dict(self) -> dict[str, Any]:
        return {
            "id": self.id,
            "match": self.match,
            "reason": self.reason,
            "enabled": self.enabled,
        }


@dataclass(frozen=True, slots=True)
class GeoIpSchedule:
    """Расписание обновления GeoIP-наборов."""

    enabled: bool = True
    cron: str = "0 4 * * *"
    provider: str = "ripe"
    updated_at: str | None = None

    def as_dict(self) -> dict[str, Any]:
        return {
            "enabled": self.enabled,
            "cron": self.cron,
            "provider": self.provider,
        }


@dataclass(frozen=True, slots=True)
class ConfigMeta:
    """Метаданные сгенерированного конфига (сам файл живёт в /config)."""

    name: str
    path: str
    version: int = 1
    checksum: str = ""
    size_bytes: int = 0
    applied: bool = False
    id: int | None = None
    created_at: str | None = None
    updated_at: str | None = None

    def as_dict(self) -> dict[str, Any]:
        return {
            "id": self.id,
            "name": self.name,
            "path": self.path,
            "version": self.version,
            "checksum": self.checksum,
            "size_bytes": self.size_bytes,
            "applied": self.applied,
        }


@dataclass(frozen=True, slots=True)
class Device:
    """Известный клиент шлюза."""

    mac: str
    name: str
    address: str = ""
    policy: str = "default"
    last_seen: str | None = None
    id: int | None = None
    created_at: str | None = None
    updated_at: str | None = None

    def as_dict(self) -> dict[str, Any]:
        return {
            "id": self.id,
            "mac": self.mac,
            "name": self.name,
            "address": self.address,
            "policy": self.policy,
            "last_seen": self.last_seen,
        }


@dataclass(frozen=True, slots=True)
class AuditEntry:
    """Запись журнала «кто/когда/что/old->new»."""

    id: int
    ts: str
    actor: str
    action: str
    entity: str
    entity_id: str | None
    old_value: Any
    new_value: Any
    status: str
    detail: str | None

    def as_dict(self) -> dict[str, Any]:
        return {
            "id": self.id,
            "ts": self.ts,
            "actor": self.actor,
            "action": self.action,
            "entity": self.entity,
            "entity_id": self.entity_id,
            "old_value": self.old_value,
            "new_value": self.new_value,
            "status": self.status,
            "detail": self.detail,
        }