diff --git a/backend/awg_api/store/repository.py b/backend/awg_api/store/repository.py new file mode 100644 index 0000000..7faaa13 --- /dev/null +++ b/backend/awg_api/store/repository.py @@ -0,0 +1,579 @@ +"""Репозиторий сущностей состояния (issue #20). + +Каждая мутация выполняется в одной транзакции, запись в ``audit_log`` +(«кто/когда/что/old→new») коммитится там же. Чтения идут по соединению WAL +и не блокируют писателя. +""" + +from __future__ import annotations + +import sqlite3 +from dataclasses import replace +from typing import cast + +from . import audit, validation +from .clock import utcnow +from .db import Database +from .errors import StorageError +from .models import ( + AuditEntry, + ConfigMeta, + Device, + Entity, + GeoIpSchedule, + Mode, + Rule, + RuleException, +) + + +def _rule_from_row(row: sqlite3.Row) -> Rule: + return Rule( + id=int(row["id"]), + name=str(row["name"]), + kind=str(row["kind"]), + match=str(row["match"]), + target=str(row["target"]), + priority=int(row["priority"]), + enabled=bool(row["enabled"]), + created_at=str(row["created_at"]), + updated_at=str(row["updated_at"]), + ) + + +def _exception_from_row(row: sqlite3.Row) -> RuleException: + return RuleException( + id=int(row["id"]), + match=str(row["match"]), + reason=str(row["reason"]), + enabled=bool(row["enabled"]), + created_at=str(row["created_at"]), + updated_at=str(row["updated_at"]), + ) + + +def _config_from_row(row: sqlite3.Row) -> ConfigMeta: + return ConfigMeta( + id=int(row["id"]), + name=str(row["name"]), + path=str(row["path"]), + version=int(row["version"]), + checksum=str(row["checksum"]), + size_bytes=int(row["size_bytes"]), + applied=bool(row["applied"]), + created_at=str(row["created_at"]), + updated_at=str(row["updated_at"]), + ) + + +def _device_from_row(row: sqlite3.Row) -> Device: + return Device( + id=int(row["id"]), + mac=str(row["mac"]), + name=str(row["name"]), + address=str(row["address"]), + policy=str(row["policy"]), + last_seen=None if row["last_seen"] is None else str(row["last_seen"]), + created_at=str(row["created_at"]), + updated_at=str(row["updated_at"]), + ) + + +class Store: + """Доступ к состоянию: чтение + аудируемые мутации.""" + + def __init__(self, db: Database) -> None: + self._db = db + + @property + def database(self) -> Database: + return self._db + + # -- чтение --------------------------------------------------------- + def get_mode(self) -> Mode: + row = self._db.connection.execute("SELECT mode FROM routing_mode WHERE id = 1").fetchone() + if row is None: + return Mode.OFF + return Mode(str(row["mode"])) + + def get_geoip_schedule(self) -> GeoIpSchedule: + row = self._db.connection.execute( + "SELECT enabled, cron, provider, updated_at FROM geoip_schedule WHERE id = 1" + ).fetchone() + if row is None: + return GeoIpSchedule() + return GeoIpSchedule( + enabled=bool(row["enabled"]), + cron=str(row["cron"]), + provider=str(row["provider"]), + updated_at=str(row["updated_at"]), + ) + + def list_rules(self, *, enabled_only: bool = False) -> list[Rule]: + sql = "SELECT * FROM routing_rules" + if enabled_only: + sql += " WHERE enabled = 1" + sql += " ORDER BY priority, id" + return [_rule_from_row(row) for row in self._db.connection.execute(sql).fetchall()] + + def get_rule(self, rule_id: int) -> Rule | None: + row = self._db.connection.execute( + "SELECT * FROM routing_rules WHERE id = ?", (rule_id,) + ).fetchone() + return None if row is None else _rule_from_row(row) + + def get_rule_by_name(self, name: str) -> Rule | None: + row = self._db.connection.execute( + "SELECT * FROM routing_rules WHERE name = ?", (name,) + ).fetchone() + return None if row is None else _rule_from_row(row) + + def list_exceptions(self) -> list[RuleException]: + rows = self._db.connection.execute("SELECT * FROM rule_exceptions ORDER BY id").fetchall() + return [_exception_from_row(row) for row in rows] + + def get_exception(self, match: str) -> RuleException | None: + row = self._db.connection.execute( + "SELECT * FROM rule_exceptions WHERE match = ?", (match,) + ).fetchone() + return None if row is None else _exception_from_row(row) + + def list_configs(self) -> list[ConfigMeta]: + rows = self._db.connection.execute("SELECT * FROM configs ORDER BY name").fetchall() + return [_config_from_row(row) for row in rows] + + def get_config(self, name: str) -> ConfigMeta | None: + row = self._db.connection.execute( + "SELECT * FROM configs WHERE name = ?", (name,) + ).fetchone() + return None if row is None else _config_from_row(row) + + def list_devices(self) -> list[Device]: + rows = self._db.connection.execute("SELECT * FROM devices ORDER BY name").fetchall() + return [_device_from_row(row) for row in rows] + + def get_device(self, mac: str) -> Device | None: + row = self._db.connection.execute( + "SELECT * FROM devices WHERE mac = ?", (validation.normalize_mac(mac),) + ).fetchone() + return None if row is None else _device_from_row(row) + + def list_audit( + self, *, entity: str | None = None, entity_id: str | int | None = None, limit: int = 100 + ) -> list[AuditEntry]: + return audit.list_entries( + self._db.connection, entity=entity, entity_id=entity_id, limit=limit + ) + + # -- мутации: правила ------------------------------------------------ + def upsert_rule(self, rule: Rule, *, actor: str) -> Rule: + """Создать или обновить правило (по ``id`` либо по уникальному ``name``).""" + validation.validate_name("name", rule.name) + kind = validation.validate_rule_kind(rule.kind) + match = validation.validate_match(kind, rule.match) + target = validation.validate_target(rule.target) + priority = validation.validate_priority(rule.priority) + now = utcnow() + with self._db.transaction() as conn: + existing = self._find_rule_row(conn, rule.id, rule.name) + core: tuple[str, str, str, str, int, int] = ( + rule.name, + kind, + match, + target, + priority, + 1 if rule.enabled else 0, + ) + if existing is None: + cursor = conn.execute( + "INSERT INTO routing_rules (name, kind, match, target, priority, enabled," + " created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)", + (*core, now, now), + ) + new_id = int(cursor.lastrowid or 0) + saved = replace( + rule, + id=new_id, + kind=kind, + match=match, + target=target, + priority=priority, + created_at=now, + updated_at=now, + ) + audit.record( + conn, + actor=actor, + action="create", + entity=Entity.RULE.value, + entity_id=new_id, + old_value=None, + new_value=saved.as_dict(), + ) + else: + rule_id = int(existing["id"]) + conn.execute( + "UPDATE routing_rules SET name = ?, kind = ?, match = ?, target = ?," + " priority = ?, enabled = ?, updated_at = ? WHERE id = ?", + (*core, now, rule_id), + ) + saved = replace( + rule, + id=rule_id, + kind=kind, + match=match, + target=target, + priority=priority, + created_at=str(existing["created_at"]), + updated_at=now, + ) + audit.record( + conn, + actor=actor, + action="update", + entity=Entity.RULE.value, + entity_id=rule_id, + old_value=_rule_from_row(existing).as_dict(), + new_value=saved.as_dict(), + ) + return saved + + def delete_rule(self, rule_id: int, *, actor: str) -> bool: + """Удалить правило; ``False`` — если правила не было.""" + with self._db.transaction() as conn: + row = self._find_rule_row(conn, rule_id, None) + if row is None: + return False + removed = _rule_from_row(row) + conn.execute("DELETE FROM routing_rules WHERE id = ?", (int(row["id"]),)) + audit.record( + conn, + actor=actor, + action="delete", + entity=Entity.RULE.value, + entity_id=int(row["id"]), + old_value=removed.as_dict(), + new_value=None, + ) + return True + + @staticmethod + def _find_rule_row( + conn: sqlite3.Connection, rule_id: int | None, name: str | None + ) -> sqlite3.Row | None: + if rule_id is not None: + row = conn.execute("SELECT * FROM routing_rules WHERE id = ?", (rule_id,)).fetchone() + return cast("sqlite3.Row | None", row) + if name is not None: + row = conn.execute("SELECT * FROM routing_rules WHERE name = ?", (name,)).fetchone() + return cast("sqlite3.Row | None", row) + return None + + # -- мутации: исключения --------------------------------------------- + def upsert_exception(self, exception: RuleException, *, actor: str) -> RuleException: + match = validation.validate_exception_match(exception.match) + now = utcnow() + with self._db.transaction() as conn: + existing = conn.execute( + "SELECT * FROM rule_exceptions WHERE match = ?", (match,) + ).fetchone() + if existing is None: + conn.execute( + "INSERT INTO rule_exceptions (match, reason, enabled, created_at, updated_at)" + " VALUES (?, ?, ?, ?, ?)", + (match, exception.reason, 1 if exception.enabled else 0, now, now), + ) + else: + conn.execute( + "UPDATE rule_exceptions SET reason = ?, enabled = ?, updated_at = ?" + " WHERE match = ?", + (exception.reason, 1 if exception.enabled else 0, now, match), + ) + row = conn.execute("SELECT * FROM rule_exceptions WHERE match = ?", (match,)).fetchone() + if row is None: # pragma: no cover - невозможен после успешной записи + raise StorageError("правило-исключение исчезло сразу после записи") + saved = _exception_from_row(row) + audit.record( + conn, + actor=actor, + action="create" if existing is None else "update", + entity=Entity.EXCEPTION.value, + entity_id=saved.id, + old_value=None if existing is None else _exception_from_row(existing).as_dict(), + new_value=saved.as_dict(), + ) + return saved + + def delete_exception(self, exception_id: int, *, actor: str) -> bool: + with self._db.transaction() as conn: + row = conn.execute( + "SELECT * FROM rule_exceptions WHERE id = ?", (exception_id,) + ).fetchone() + if row is None: + return False + removed = _exception_from_row(row) + conn.execute("DELETE FROM rule_exceptions WHERE id = ?", (exception_id,)) + audit.record( + conn, + actor=actor, + action="delete", + entity=Entity.EXCEPTION.value, + entity_id=exception_id, + old_value=removed.as_dict(), + new_value=None, + ) + return True + + # -- мутации: режим и GeoIP ------------------------------------------ + def set_mode(self, mode: str | Mode, *, actor: str) -> Mode: + """Установить режим маршрутизации; в журнал пишется old→new.""" + validated = validation.validate_mode(mode) + with self._db.transaction() as conn: + previous = conn.execute("SELECT mode FROM routing_mode WHERE id = 1").fetchone() + old = "off" if previous is None else str(previous["mode"]) + now = utcnow() + conn.execute( + "INSERT INTO routing_mode (id, mode, updated_at) VALUES (1, ?, ?)" + " ON CONFLICT(id) DO UPDATE SET mode = excluded.mode," + " updated_at = excluded.updated_at", + (validated.value, now), + ) + audit.record( + conn, + actor=actor, + action="set_mode", + entity=Entity.MODE.value, + entity_id=1, + old_value={"mode": old}, + new_value={"mode": validated.value}, + ) + return validated + + def set_geoip_schedule( + self, + *, + enabled: bool | None = None, + cron: str | None = None, + provider: str | None = None, + actor: str, + ) -> GeoIpSchedule: + """Обновить расписание GeoIP (частичное обновление) и записать аудит.""" + current = self.get_geoip_schedule() + updated = GeoIpSchedule( + enabled=current.enabled if enabled is None else enabled, + cron=validation.validate_cron(current.cron if cron is None else cron), + provider=validation.validate_provider( + current.provider if provider is None else provider + ), + ) + now = utcnow() + with self._db.transaction() as conn: + conn.execute( + "INSERT INTO geoip_schedule (id, enabled, cron, provider, updated_at)" + " VALUES (1, ?, ?, ?, ?)" + " ON CONFLICT(id) DO UPDATE SET enabled = excluded.enabled, cron = excluded.cron," + " provider = excluded.provider, updated_at = excluded.updated_at", + (1 if updated.enabled else 0, updated.cron, updated.provider, now), + ) + audit.record( + conn, + actor=actor, + action="set_geoip_schedule", + entity=Entity.GEOIP_SCHEDULE.value, + entity_id=1, + old_value=current.as_dict(), + new_value=updated.as_dict(), + ) + return replace(updated, updated_at=now) + + # -- мутации: конфиги ------------------------------------------------- + def upsert_config(self, config: ConfigMeta, *, actor: str) -> ConfigMeta: + name = validation.validate_name("name", config.name) + path = validation.validate_config_path(config.path) + now = utcnow() + with self._db.transaction() as conn: + existing = conn.execute("SELECT * FROM configs WHERE name = ?", (name,)).fetchone() + if existing is None: + cursor = conn.execute( + "INSERT INTO configs (name, version, path, checksum, size_bytes, applied," + " created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)", + ( + name, + config.version, + path, + config.checksum, + config.size_bytes, + 1 if config.applied else 0, + now, + now, + ), + ) + config_id = int(cursor.lastrowid or 0) + saved = replace( + config, id=config_id, name=name, path=path, created_at=now, updated_at=now + ) + audit.record( + conn, + actor=actor, + action="create", + entity=Entity.CONFIG.value, + entity_id=config_id, + old_value=None, + new_value=saved.as_dict(), + ) + else: + config_id = int(existing["id"]) + conn.execute( + "UPDATE configs SET version = ?, path = ?, checksum = ?, size_bytes = ?," + " applied = ?, updated_at = ? WHERE id = ?", + ( + config.version, + path, + config.checksum, + config.size_bytes, + 1 if config.applied else 0, + now, + config_id, + ), + ) + saved = replace( + config, + id=config_id, + name=name, + path=path, + created_at=str(existing["created_at"]), + updated_at=now, + ) + audit.record( + conn, + actor=actor, + action="update", + entity=Entity.CONFIG.value, + entity_id=config_id, + old_value=_config_from_row(existing).as_dict(), + new_value=saved.as_dict(), + ) + return saved + + def mark_config_applied(self, name: str, *, applied: bool, actor: str) -> ConfigMeta: + """Отметить конфиг применённым/снятым и записать аудит.""" + with self._db.transaction() as conn: + row = conn.execute("SELECT * FROM configs WHERE name = ?", (name,)).fetchone() + if row is None: + raise StorageError(f"конфиг {name!r} не найден") + now = utcnow() + conn.execute( + "UPDATE configs SET applied = ?, updated_at = ? WHERE id = ?", + (1 if applied else 0, now, int(row["id"])), + ) + updated = replace(_config_from_row(row), applied=applied, updated_at=now) + audit.record( + conn, + actor=actor, + action="set_applied", + entity=Entity.CONFIG.value, + entity_id=int(row["id"]), + old_value=_config_from_row(row).as_dict(), + new_value=updated.as_dict(), + ) + return updated + + # -- мутации: устройства ---------------------------------------------- + def register_device(self, device: Device, *, actor: str) -> Device: + mac = validation.normalize_mac(device.mac) + name = validation.validate_name("name", device.name) + policy = validation.validate_device_policy(device.policy) + now = utcnow() + with self._db.transaction() as conn: + existing = conn.execute("SELECT * FROM devices WHERE mac = ?", (mac,)).fetchone() + if existing is None: + cursor = conn.execute( + "INSERT INTO devices (mac, name, address, policy, last_seen, created_at," + " updated_at) VALUES (?, ?, ?, ?, ?, ?, ?)", + (mac, name, device.address, policy, device.last_seen, now, now), + ) + device_id = int(cursor.lastrowid or 0) + saved = replace( + device, + id=device_id, + mac=mac, + name=name, + policy=policy, + created_at=now, + updated_at=now, + ) + audit.record( + conn, + actor=actor, + action="create", + entity=Entity.DEVICE.value, + entity_id=device_id, + old_value=None, + new_value=saved.as_dict(), + ) + else: + device_id = int(existing["id"]) + conn.execute( + "UPDATE devices SET name = ?, address = ?, policy = ?, last_seen = ?," + " updated_at = ? WHERE id = ?", + (name, device.address, policy, device.last_seen, now, device_id), + ) + saved = replace( + device, + id=device_id, + mac=mac, + name=name, + policy=policy, + created_at=str(existing["created_at"]), + updated_at=now, + ) + audit.record( + conn, + actor=actor, + action="update", + entity=Entity.DEVICE.value, + entity_id=device_id, + old_value=_device_from_row(existing).as_dict(), + new_value=saved.as_dict(), + ) + return saved + + def delete_device(self, mac: str, *, actor: str) -> bool: + normalized = validation.normalize_mac(mac) + with self._db.transaction() as conn: + row = conn.execute("SELECT * FROM devices WHERE mac = ?", (normalized,)).fetchone() + if row is None: + return False + removed = _device_from_row(row) + conn.execute("DELETE FROM devices WHERE mac = ?", (normalized,)) + audit.record( + conn, + actor=actor, + action="delete", + entity=Entity.DEVICE.value, + entity_id=removed.id, + old_value=removed.as_dict(), + new_value=None, + ) + return True + + # -- служебное -------------------------------------------------------- + def counts(self) -> dict[str, int]: + """Количество строк в изменяемых таблицах (для отчётов и тестов).""" + tables = { + "rules": "routing_rules", + "exceptions": "rule_exceptions", + "configs": "configs", + "devices": "devices", + "audit": "audit_log", + "snapshots": "state_snapshots", + } + result: dict[str, int] = {} + for key, table in tables.items(): + row = self._db.connection.execute(f"SELECT COUNT(*) FROM {table}").fetchone() # noqa: S608 + result[key] = int(row[0]) if row is not None else 0 + return result + + def raw(self) -> sqlite3.Connection: + """Соединение для кода, которому нужен доступ ниже уровня репозитория.""" + return self._db.connection