"""Модель данных хранилища 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, }