335 lines
12 KiB
Python
335 lines
12 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
|
|
|
|
from src.private_state import connect_private_sqlite
|
|
from src.state_encryption import (
|
|
PrivateStateCipher,
|
|
PrivateStateEncryptionError,
|
|
private_state_encryption_config,
|
|
)
|
|
|
|
|
|
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,
|
|
encryption_key: bytes | None = None,
|
|
) -> 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
|
|
try:
|
|
self._cipher = PrivateStateCipher(
|
|
encryption_key if encryption_key is not None else private_state_encryption_config(),
|
|
store="security-events",
|
|
)
|
|
except PrivateStateEncryptionError as exc:
|
|
raise SecurityEventStoreError(
|
|
"Security activity is temporarily unavailable"
|
|
) from exc
|
|
|
|
@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:
|
|
connection = connect_private_sqlite(self.path, timeout=self.lock_timeout_seconds)
|
|
connection.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS security_events (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
payload 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"
|
|
)
|
|
plaintext_columns = {"kind", "method", "device_label", "target"}
|
|
if "payload" not in columns or plaintext_columns.intersection(columns):
|
|
self._migrate_plaintext(
|
|
connection, has_payload="payload" in columns
|
|
)
|
|
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 _migrate_plaintext(
|
|
self, connection: sqlite3.Connection, *, has_payload: bool
|
|
) -> None:
|
|
connection.execute("PRAGMA secure_delete = ON")
|
|
payload_column = "payload," if has_payload else "NULL AS payload,"
|
|
rows = connection.execute(
|
|
f"SELECT id, {payload_column} kind, method, device_label, target, "
|
|
"created_at, status, operation_id FROM security_events ORDER BY id"
|
|
).fetchall()
|
|
connection.execute(
|
|
"""
|
|
CREATE TABLE security_events_encrypted (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
payload TEXT,
|
|
created_at INTEGER NOT NULL,
|
|
status TEXT NOT NULL DEFAULT 'completed',
|
|
operation_id TEXT
|
|
)
|
|
"""
|
|
)
|
|
for (
|
|
event_id,
|
|
payload,
|
|
kind,
|
|
method,
|
|
device_label,
|
|
target,
|
|
created_at,
|
|
status,
|
|
operation_id,
|
|
) in rows:
|
|
connection.execute(
|
|
"INSERT INTO security_events_encrypted "
|
|
"(id, payload, created_at, status, operation_id) VALUES (?, ?, ?, ?, ?)",
|
|
(
|
|
event_id,
|
|
payload
|
|
or self._seal_event(
|
|
event_id, kind, method, device_label, target
|
|
),
|
|
created_at,
|
|
status,
|
|
operation_id,
|
|
),
|
|
)
|
|
connection.execute("DROP TABLE security_events")
|
|
connection.execute(
|
|
"ALTER TABLE security_events_encrypted RENAME TO security_events"
|
|
)
|
|
|
|
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 _seal_event(
|
|
self,
|
|
event_id: int,
|
|
kind: str,
|
|
method: str | None,
|
|
device_label: str | None,
|
|
target: str | None,
|
|
) -> str:
|
|
return self._cipher.seal(
|
|
{
|
|
"kind": self._bounded(kind, 48) or "security_event",
|
|
"method": self._bounded(method, 32),
|
|
"device_label": self._bounded(device_label, 64),
|
|
"target": self._bounded(target, 255),
|
|
},
|
|
binding=f"event:{event_id}",
|
|
)
|
|
|
|
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:
|
|
cursor = connection.execute(
|
|
"INSERT INTO security_events(created_at, status) VALUES (?, 'completed')",
|
|
(now,),
|
|
)
|
|
event_id = cursor.lastrowid
|
|
payload = self._seal_event(
|
|
event_id, kind, method, device_label, target
|
|
)
|
|
connection.execute(
|
|
"UPDATE security_events SET payload = ? WHERE id = ?",
|
|
(payload, event_id),
|
|
)
|
|
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:
|
|
cursor = connection.execute(
|
|
"INSERT INTO security_events(created_at, status, operation_id) "
|
|
"VALUES (?, 'pending', ?)",
|
|
(now, operation_id),
|
|
)
|
|
event_id = cursor.lastrowid
|
|
connection.execute(
|
|
"UPDATE security_events SET payload = ? WHERE id = ?",
|
|
(
|
|
self._seal_event(
|
|
event_id, kind, method, device_label, target
|
|
),
|
|
event_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, payload, 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]
|
|
events = []
|
|
try:
|
|
for event_id, payload, created_at, status in visible:
|
|
value, _legacy = self._cipher.open(
|
|
payload, binding=f"event:{event_id}"
|
|
)
|
|
if not isinstance(value, dict) or not isinstance(value.get("kind"), str):
|
|
raise PrivateStateEncryptionError(
|
|
"private state could not be decrypted"
|
|
)
|
|
events.append(
|
|
SecurityEvent(
|
|
event_id,
|
|
value["kind"],
|
|
value.get("method"),
|
|
value.get("device_label"),
|
|
value.get("target"),
|
|
created_at,
|
|
status,
|
|
)
|
|
)
|
|
except PrivateStateEncryptionError as exc:
|
|
raise SecurityEventStoreError(
|
|
"Security activity is temporarily unavailable"
|
|
) from exc
|
|
return SecurityEventPage(
|
|
events=events,
|
|
next_cursor=visible[-1][0] if has_more else None,
|
|
)
|