Files
PacketStation/tests/server/test_audit.py
T
spencermandClaude Fable 5.1 0b663bb2b2 Phase 0: repo skeleton, image, compose stacks, doctor, migration runner, frontend shell
New repo for the Packet Station rebuild (Python 3.12, FastAPI, PostgreSQL 16,
React, VisualKit, MFilesKit, Docker Compose on docker-1). Nothing is migrated
from the M: install.

- CLAUDE.md with the safety rules, README, CHANGELOG, docs (BRIEF, PLAN,
  SECURITY rewritten for the server with the requirement ID table, DEPLOY,
  stubs for FEED, POLICY, SCHEMA, GO-LIVE, PAPER-FALLBACK).
- packetstation CLI: app, worker (heartbeat stub), migrate, doctor, version,
  healthcheck. Settings from the environment with secret files. Stack guard
  (test never points at VISUAL production). Security headers and CSP.
  Trusted proxy resolved by name. JSON logs with redaction.
- Migration runner and 0001_base: settings, append-only audit_events with
  hash chain, ps_notify, grants for ps_app.
- Multi-stage Dockerfile (ODBC Driver 18, PESCO libraries from the Gitea
  index only), compose base, proxy overlay for npm_default, CI overlay,
  postgres initdb roles, env examples, NPM proxy host notes, Gitea Actions.
- Frontend: Vite, React, TypeScript, the PESCO dark theme ported with
  self-hosted fonts, placeholder sign-in and home.
- Verified on docker-1: image builds, CI stack comes up healthy, doctor
  database checks pass, 77 server tests pass against PostgreSQL.

Co-Authored-By: Claude Fable 5.1 <[email protected]>
2026-10-15 23:08:47 -06:00

68 lines
2.5 KiB
Python

from __future__ import annotations
import hashlib
import uuid
from datetime import UTC, datetime, timedelta, timezone
from pshelpers import needs_pg
from packetstation import audit, db
from packetstation.settings import Settings
AT = datetime(2026, 10, 15, 12, 0, 0, 123456, tzinfo=UTC)
def test_canonical_is_stable_and_normalised() -> None:
uid = uuid.uuid4()
a = audit.canonical({"at": AT, "action": "x", "user_id": uid, "detail": {"b": 1, "a": [1, 2]}, "ip": "10.0.0.1"})
denver = AT.astimezone(timezone(timedelta(hours=-6)))
b = audit.canonical(
{"detail": {"a": [1, 2], "b": 1}, "ip": "10.0.0.1/32", "user_id": str(uid), "action": "x", "at": denver}
)
assert a == b
assert b'"at":"2026-10-15T12:00:00.123456+00:00"' in a
assert b" " not in a
def test_row_hash_chains() -> None:
f1 = {"at": AT, "action": "one"}
h1 = audit.row_hash(None, f1)
assert h1 == hashlib.sha256(audit.canonical(f1)).digest()
h2 = audit.row_hash(h1, {"at": AT, "action": "two"})
assert h2 == hashlib.sha256(h1 + audit.canonical({"at": AT, "action": "two"})).digest()
assert audit.row_hash(None, {"at": AT, "action": "two"}) != h2
@needs_pg
def test_pg_chain_verifies_and_detects_tamper(pg_settings: Settings) -> None:
with db.connect_as("app", pg_settings, autocommit=True) as conn:
first = audit.append(
conn,
action="a.one",
user_id=uuid.uuid4(),
detail={"n": 1, "s": "x"},
ip="10.1.2.3",
packet_id=5,
subject_type="packet",
subject_id=5,
)
second = audit.append(conn, action="a.two", detail=None)
assert second.id > first.id
prev = conn.execute("SELECT prev_hash FROM audit_events WHERE id = %s", (second.id,)).fetchone()
assert bytes(prev[0]) == first.row_hash
ok, n, detail = audit.verify_chain(conn)
assert ok, detail
assert n == 2
# Tamper as the superuser with the trigger disabled, then verification must fail
import psycopg
from pshelpers import PG_URL
from psycopg.conninfo import make_conninfo
assert PG_URL
with psycopg.connect(make_conninfo(PG_URL, dbname=pg_settings.pg_database), autocommit=True) as su:
su.execute("ALTER TABLE audit_events DISABLE TRIGGER audit_events_append_only")
su.execute("UPDATE audit_events SET action = 'forged' WHERE id = %s", (first.id,))
with db.connect_as("app", pg_settings) as conn:
ok, _, detail = audit.verify_chain(conn)
assert not ok and "row_hash" in detail