security: encrypt worker-shared snapshots (Closes #1106)

tests/conftest.py

@@ -0,0 +1,9 @@
+"""Test-only secret injection for encrypted private snapshot stores."""
+
+import os
+
+
+os.environ.setdefault(
+    "STACKCHAIN_PRIVATE_STATE_ENCRYPTION_KEY",
+    "c3Nzc3Nzc3Nzc3Nzc3Nzc3Nzc3Nzc3Nzc3Nzc3Nzc3M=",
+)

tests/test_available_issue_snapshot_store.py

@@ -1,7 +1,78 @@
 import threading
 import os
+import sqlite3
+
+import pytest
 
 from src.available_issue_snapshot_store import AvailableIssueSnapshotStore
+from src.state_encryption import PrivateStateEncryptionError
+
+
+PRIVATE_KEY = b"a" * 32
+
+
+def test_published_find_work_catalog_is_encrypted_at_rest_and_survives_restart(tmp_path):
+    path = tmp_path / "available.sqlite3"
+    store = AvailableIssueSnapshotStore(path, clock=lambda: 100.0, encryption_key=PRIVATE_KEY)
+    owner = store.try_acquire_refresh(lease_seconds=5)
+    canary = "private-find-work-body-canary"
+
+    published = store.publish(
+        owner,
+        items=[{"repository": "stackchain/api", "number": 7, "body": canary}],
+    )
+
+    with sqlite3.connect(path) as connection:
+        payload = connection.execute(
+            "SELECT items_json FROM available_issue_snapshot WHERE singleton = 1"
+        ).fetchone()[0]
+    assert payload.startswith("v1:")
+    assert canary not in payload
+    assert AvailableIssueSnapshotStore(
+        path, clock=lambda: 100.0, encryption_key=PRIVATE_KEY
+    ).load().items == published.items
+
+
+def test_find_work_catalog_lazily_migrates_plaintext_without_changing_freshness(tmp_path):
+    path = tmp_path / "available.sqlite3"
+    store = AvailableIssueSnapshotStore(path, clock=lambda: 100.0, encryption_key=PRIVATE_KEY)
+    with sqlite3.connect(path) as connection:
+        connection.execute(
+            "UPDATE available_issue_snapshot SET items_json = ?, created_at = ?, retry_at = ?",
+            ('[{"repository":"stackchain/api","number":7}]', 91.0, 105.0),
+        )
+
+    state = store.load()
+
+    with sqlite3.connect(path) as connection:
+        migrated = connection.execute(
+            "SELECT items_json FROM available_issue_snapshot WHERE singleton = 1"
+        ).fetchone()[0]
+    assert state.items == [{"repository": "stackchain/api", "number": 7}]
+    assert (state.created_at, state.retry_at) == (91.0, 105.0)
+    assert migrated.startswith("v1:")
+
+
+def test_find_work_ciphertext_cannot_be_substituted_into_live_snapshot(tmp_path):
+    available_path = tmp_path / "available.sqlite3"
+    available = AvailableIssueSnapshotStore(
+        available_path, clock=lambda: 100.0, encryption_key=PRIVATE_KEY
+    )
+    owner = available.try_acquire_refresh(lease_seconds=5)
+    available.publish(owner, items=[{"body": "secret"}])
+    with sqlite3.connect(available_path) as connection:
+        payload = connection.execute(
+            "SELECT items_json FROM available_issue_snapshot WHERE singleton = 1"
+        ).fetchone()[0]
+
+    live_path = tmp_path / "live.sqlite3"
+    from src.live_snapshot_store import LiveSnapshotStore
+    live = LiveSnapshotStore(live_path, encryption_key=PRIVATE_KEY)
+    with sqlite3.connect(live_path) as connection:
+        connection.execute("UPDATE live_snapshot SET value_json = ?", (payload,))
+
+    with pytest.raises(PrivateStateEncryptionError, match="private state could not be decrypted"):
+        live.load()
 
 
 def test_independent_workers_allow_only_one_catalog_refresh(tmp_path):

tests/test_live_snapshot_store.py

@@ -6,6 +6,74 @@ import pytest
 
 from src import live_snapshot_store
 from src.live_snapshot_store import LiveSnapshotStore, RefreshLeaseLost
+from src.state_encryption import PrivateStateEncryptionError
+
+
+PRIVATE_KEY = b"l" * 32
+
+
+def test_published_live_snapshot_is_encrypted_at_rest_and_survives_restart(tmp_path):
+    path = tmp_path / "live.sqlite3"
+    store = LiveSnapshotStore(path, clock=lambda: 100.0, encryption_key=PRIVATE_KEY)
+    lease = store.try_acquire_refresh({"context"}, lease_seconds=5)
+    canary = "private-live-title-canary"
+
+    published = store.publish_refresh(
+        lease,
+        value={"context": {"title": canary}, "events": [], "notifications": []},
+        created_at={section: 100.0 for section in live_snapshot_store.SECTIONS},
+        failure_count={section: 0 for section in live_snapshot_store.SECTIONS},
+        retry_at={section: None for section in live_snapshot_store.SECTIONS},
+        changed_sections={"context"},
+    )
+
+    with sqlite3.connect(path) as connection:
+        payload = connection.execute(
+            "SELECT value_json FROM live_snapshot WHERE singleton = 1"
+        ).fetchone()[0]
+    assert payload.startswith("v1:")
+    assert canary not in payload
+    assert LiveSnapshotStore(path, clock=lambda: 100.0, encryption_key=PRIVATE_KEY).load().value == published.value
+
+
+def test_live_snapshot_lazily_migrates_plaintext_without_changing_metadata(tmp_path):
+    path = tmp_path / "live.sqlite3"
+    store = LiveSnapshotStore(path, clock=lambda: 100.0, encryption_key=PRIVATE_KEY)
+    legacy = {"context": {"title": "legacy"}, "events": [], "notifications": []}
+    with sqlite3.connect(path) as connection:
+        connection.execute(
+            "UPDATE live_snapshot SET value_json = ?, revisions_json = ? WHERE singleton = 1",
+            ('{"context":{"title":"legacy"},"events":[],"notifications":[]}',
+             '{"context":4,"events":2,"notifications":1}'),
+        )
+
+    state = store.load()
+
+    with sqlite3.connect(path) as connection:
+        migrated = connection.execute(
+            "SELECT value_json FROM live_snapshot WHERE singleton = 1"
+        ).fetchone()[0]
+    assert state.value == legacy
+    assert state.revisions == {"context": 4, "events": 2, "notifications": 1}
+    assert migrated.startswith("v1:")
+
+
+def test_live_snapshot_authentication_failure_returns_no_private_content(tmp_path):
+    path = tmp_path / "live.sqlite3"
+    store = LiveSnapshotStore(path, clock=lambda: 100.0, encryption_key=PRIVATE_KEY)
+    lease = store.try_acquire_refresh({"context"}, lease_seconds=5)
+    metadata = {section: None for section in live_snapshot_store.SECTIONS}
+    store.publish_refresh(
+        lease,
+        value={"context": {"title": "secret"}},
+        created_at=metadata,
+        failure_count={section: 0 for section in metadata},
+        retry_at=metadata,
+        changed_sections={"context"},
+    )
+
+    with pytest.raises(PrivateStateEncryptionError, match="private state could not be decrypted"):
+        LiveSnapshotStore(path, encryption_key=b"x" * 32).load()
 
 
 def test_metadata_load_does_not_retrieve_or_decode_snapshot_value(tmp_path):
@@ -24,7 +92,7 @@ def test_metadata_load_does_not_retrieve_or_decode_snapshot_value(tmp_path):
     assert metadata.revisions == {
         section: 0 for section in ("context", "events", "notifications")
     }
-    with pytest.raises(ValueError):
+    with pytest.raises(PrivateStateEncryptionError):
         store.load()