"""Репозиторий сущностей состояния (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