220 lines
7.8 KiB
Python
220 lines
7.8 KiB
Python
"""Bounded, privacy-preserving journal of operator security activity."""
|
|
|
|
import secrets
|
|
import sqlite3
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Callable
|
|
|
|
|
|
class SecurityEventStoreError(RuntimeError):
|
|
"""Raised when security activity cannot be persisted or read safely."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SecurityEvent:
|
|
id: int
|
|
kind: str
|
|
method: str | None
|
|
device_label: str | None
|
|
target: str | None
|
|
created_at: int
|
|
status: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SecurityEventPage:
|
|
events: list[SecurityEvent]
|
|
next_cursor: int | None
|
|
|
|
|
|
class SecurityEventStore:
|
|
def __init__(
|
|
self,
|
|
path: str | Path,
|
|
*,
|
|
clock: Callable[[], float],
|
|
max_events: int = 10_000,
|
|
retention_seconds: int = 90 * 24 * 60 * 60,
|
|
lock_timeout_seconds: float = 0.1,
|
|
) -> None:
|
|
self.path = Path(path)
|
|
self.clock = clock
|
|
self.max_events = max(1, max_events)
|
|
self.retention_seconds = max(1, retention_seconds)
|
|
self.lock_timeout_seconds = lock_timeout_seconds
|
|
|
|
@staticmethod
|
|
def _bounded(value: str | None, limit: int) -> str | None:
|
|
if value is None:
|
|
return None
|
|
normalized = " ".join(str(value).split())[:limit]
|
|
return normalized or None
|
|
|
|
def _connect(self) -> sqlite3.Connection:
|
|
try:
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
connection = sqlite3.connect(self.path, timeout=self.lock_timeout_seconds)
|
|
connection.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS security_events (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
kind TEXT NOT NULL,
|
|
method TEXT,
|
|
device_label TEXT,
|
|
target TEXT,
|
|
created_at INTEGER NOT NULL,
|
|
status TEXT NOT NULL DEFAULT 'completed',
|
|
operation_id TEXT
|
|
)
|
|
"""
|
|
)
|
|
columns = {
|
|
row[1] for row in connection.execute("PRAGMA table_info(security_events)")
|
|
}
|
|
if "status" not in columns:
|
|
connection.execute(
|
|
"ALTER TABLE security_events ADD COLUMN status TEXT NOT NULL DEFAULT 'completed'"
|
|
)
|
|
if "operation_id" not in columns:
|
|
connection.execute(
|
|
"ALTER TABLE security_events ADD COLUMN operation_id TEXT"
|
|
)
|
|
connection.execute(
|
|
"CREATE INDEX IF NOT EXISTS security_events_created "
|
|
"ON security_events(created_at DESC, id DESC)"
|
|
)
|
|
connection.execute(
|
|
"CREATE UNIQUE INDEX IF NOT EXISTS security_events_operation "
|
|
"ON security_events(operation_id) WHERE operation_id IS NOT NULL"
|
|
)
|
|
return connection
|
|
except (OSError, sqlite3.Error) as exc:
|
|
raise SecurityEventStoreError(
|
|
"Security activity is temporarily unavailable"
|
|
) from exc
|
|
|
|
def _prune(self, connection: sqlite3.Connection, now: int) -> None:
|
|
connection.execute(
|
|
"DELETE FROM security_events WHERE created_at < ?",
|
|
(now - self.retention_seconds,),
|
|
)
|
|
connection.execute(
|
|
"DELETE FROM security_events WHERE id NOT IN "
|
|
"(SELECT id FROM security_events ORDER BY id DESC LIMIT ?)",
|
|
(self.max_events,),
|
|
)
|
|
|
|
def record(
|
|
self,
|
|
kind: str,
|
|
*,
|
|
method: str | None = None,
|
|
device_label: str | None = None,
|
|
target: str | None = None,
|
|
) -> None:
|
|
now = int(self.clock())
|
|
try:
|
|
with self._connect() as connection:
|
|
connection.execute(
|
|
"INSERT INTO security_events(kind, method, device_label, target, created_at, status) "
|
|
"VALUES (?, ?, ?, ?, ?, 'completed')",
|
|
(
|
|
self._bounded(kind, 48) or "security_event",
|
|
self._bounded(method, 32),
|
|
self._bounded(device_label, 64),
|
|
self._bounded(target, 255),
|
|
now,
|
|
),
|
|
)
|
|
self._prune(connection, now)
|
|
except (OSError, sqlite3.Error) as exc:
|
|
raise SecurityEventStoreError(
|
|
"Security activity is temporarily unavailable"
|
|
) from exc
|
|
|
|
def reserve(
|
|
self,
|
|
kind: str,
|
|
*,
|
|
method: str | None = None,
|
|
device_label: str | None = None,
|
|
target: str | None = None,
|
|
) -> str:
|
|
now = int(self.clock())
|
|
operation_id = secrets.token_urlsafe(24)
|
|
try:
|
|
with self._connect() as connection:
|
|
connection.execute(
|
|
"INSERT INTO security_events(kind, method, device_label, target, created_at, status, operation_id) "
|
|
"VALUES (?, ?, ?, ?, ?, 'pending', ?)",
|
|
(
|
|
self._bounded(kind, 48) or "security_event",
|
|
self._bounded(method, 32),
|
|
self._bounded(device_label, 64),
|
|
self._bounded(target, 255),
|
|
now,
|
|
operation_id,
|
|
),
|
|
)
|
|
self._prune(connection, now)
|
|
except (OSError, sqlite3.Error) as exc:
|
|
raise SecurityEventStoreError(
|
|
"Security activity is temporarily unavailable"
|
|
) from exc
|
|
return operation_id
|
|
|
|
def finalize(self, operation_id: str) -> None:
|
|
try:
|
|
with self._connect() as connection:
|
|
cursor = connection.execute(
|
|
"UPDATE security_events SET status = 'completed' WHERE operation_id = ?",
|
|
(operation_id,),
|
|
)
|
|
if cursor.rowcount != 1:
|
|
raise SecurityEventStoreError(
|
|
"Security activity reservation was not found"
|
|
)
|
|
except (OSError, sqlite3.Error) as exc:
|
|
raise SecurityEventStoreError(
|
|
"Security activity is temporarily unavailable"
|
|
) from exc
|
|
|
|
def discard(self, operation_id: str) -> None:
|
|
try:
|
|
with self._connect() as connection:
|
|
connection.execute(
|
|
"DELETE FROM security_events WHERE operation_id = ? AND status = 'pending'",
|
|
(operation_id,),
|
|
)
|
|
except (OSError, sqlite3.Error) as exc:
|
|
raise SecurityEventStoreError(
|
|
"Security activity is temporarily unavailable"
|
|
) from exc
|
|
|
|
def list(self, *, limit: int = 50, cursor: int | None = None) -> SecurityEventPage:
|
|
bounded_limit = min(100, max(1, limit))
|
|
parameters: list[int] = []
|
|
where = ""
|
|
if cursor is not None:
|
|
where = "WHERE id < ?"
|
|
parameters.append(cursor)
|
|
parameters.append(bounded_limit + 1)
|
|
try:
|
|
with self._connect() as connection:
|
|
rows = connection.execute(
|
|
"SELECT id, kind, method, device_label, target, created_at, status "
|
|
f"FROM security_events {where} ORDER BY id DESC LIMIT ?",
|
|
parameters,
|
|
).fetchall()
|
|
except (OSError, sqlite3.Error) as exc:
|
|
raise SecurityEventStoreError(
|
|
"Security activity is temporarily unavailable"
|
|
) from exc
|
|
has_more = len(rows) > bounded_limit
|
|
visible = rows[:bounded_limit]
|
|
return SecurityEventPage(
|
|
events=[SecurityEvent(*row) for row in visible],
|
|
next_cursor=visible[-1][0] if has_more else None,
|
|
)
|