security: remove plaintext journal metadata columns

This commit is contained in:
timmy committed 2026-08-19 07:37:35 +00:00
1 parent 7f2681c1c6
commit be9ce8ede6
2 files changed
+43 -24

No files matched your search

+27 -10
View File
@@ -8,6 +8,26 @@ from src.security_event_store import SecurityEventStore, SecurityEventStoreError
PRIVATE_KEY = b"e" * 32
def test_security_event_schema_exposes_no_plaintext_metadata_columns(tmp_path):
store = SecurityEventStore(
tmp_path / "security.sqlite3", clock=lambda: 1_000, encryption_key=PRIVATE_KEY
)
store.record("issue_closed", method="passkey", target="private/repo#42")
with sqlite3.connect(store.path) as connection:
columns = {
row[1]
for row in connection.execute("PRAGMA table_info(security_events)")
}
assert columns == {
"id",
"payload",
"created_at",
"status",
"operation_id",
}
def test_security_event_payload_is_encrypted_at_rest_and_survives_restart(tmp_path):
path = tmp_path / "security.sqlite3"
canaries = {
@@ -21,11 +41,10 @@ def test_security_event_payload_is_encrypted_at_rest_and_survives_restart(tmp_pa
store.record(**canaries)
with sqlite3.connect(path) as connection:
row = connection.execute(
"SELECT kind, method, device_label, target, payload FROM security_events"
).fetchone()
assert row[:4] == (None, None, None, None)
assert row[4].startswith("v1:")
payload = connection.execute(
"SELECT payload FROM security_events"
).fetchone()[0]
assert payload.startswith("v1:")
database_bytes = path.read_bytes()
assert all(value.encode() not in database_bytes for value in canaries.values())
reopened = SecurityEventStore(path, clock=lambda: 1_001, encryption_key=PRIVATE_KEY)
@@ -74,11 +93,10 @@ def test_legacy_security_events_migrate_without_changing_journal_semantics(tmp_p
assert store.list(limit=1).events[0].status == "completed"
with sqlite3.connect(path) as connection:
rows = connection.execute(
"SELECT id, kind, method, device_label, target, payload FROM security_events ORDER BY id"
"SELECT id, payload FROM security_events ORDER BY id"
).fetchall()
assert [row[0] for row in rows] == [7, 9]
assert all(row[1:5] == (None, None, None, None) for row in rows)
assert all(row[5].startswith("v1:") for row in rows)
assert all(row[1].startswith("v1:") for row in rows)
database_bytes = path.read_bytes()
assert b"legacy_sign_in_canary" not in database_bytes
assert b"Legacy Tablet Canary" not in database_bytes
@@ -144,8 +162,7 @@ def test_security_events_are_private_bounded_and_reverse_chronological(tmp_path)
row[1] for row in sqlite3.connect(store.path).execute("PRAGMA table_info(security_events)")
}
assert columns == {
"id", "kind", "method", "device_label", "target", "payload", "created_at",
"status", "operation_id",
"id", "payload", "created_at", "status", "operation_id",
}