Merge pull request 'Encrypt worker-shared dashboard snapshots at rest' (#1107) from timmy/1106-encrypt-worker-snapshots into main

README.md

@@ -191,10 +191,14 @@ cannot alter a newer crash-recovery claim. Device purge cancels an active drain
 private outbox storage. Results are coordinated through a bounded SQLite ledger. All private SQLite stores enforce a
 filesystem boundary independently of the service umask: the database directory is repaired to
 owner-only `0700`, database and SQLite sidecar files are owner-only `0600`, and symlinked database
-paths are rejected before access. Synchronized unfiled Draft collections add AES-256-GCM payload
-encryption with account and revision authentication; existing plaintext rows migrate on their first
-read without changing revision or order. Other private stores are not encrypted at the application
-layer, so secure host access, encrypted volumes, and private backups are still required.
+paths are rejected before access. Worker-shared live and Find Work snapshots add AES-256-GCM
+envelopes authenticated to their store identity (and live generation), so copied databases do not
+expose issue bodies, titles, notification metadata, or repository context. Existing plaintext
+snapshot rows migrate on their first read without changing freshness, revisions, ordering, or claim
+filters. Synchronized unfiled Draft collections use a separate AES-256-GCM key and authenticate the
+account and revision; existing plaintext rows likewise migrate on first read. Other private stores
+are not encrypted at the application layer, so secure host access, encrypted volumes, and private
+backups are still required.
 Set `STACKCHAIN_STATE_DIR` to a
 persistent, writable service directory (or set `STACKCHAIN_IDEMPOTENCY_DB` to an explicit
 SQLite path); the local default is `.stackchain-state/idempotency.sqlite3`. Ledger reads and
@@ -260,6 +264,11 @@ export STACKCHAIN_PASSKEY_MAX_CHALLENGES_PER_SOURCE=10
 export STACKCHAIN_PASSKEY_MAX_CHALLENGES=10000
 # Optional; defaults to STACKCHAIN_STATE_DIR/login-attempts.sqlite3.
 export STACKCHAIN_LOGIN_ATTEMPT_DB='/var/lib/stackchain-dashboard/login-attempts.sqlite3'
+# Required for worker-shared live and Find Work snapshots. Keep this key
+# independent from the Draft key and inject the base64 encoding of exactly 32
+# random bytes from a secret manager. Never commit it. Missing, malformed,
+# wrong-key, or modified snapshot state fails closed without returning content.
+export STACKCHAIN_PRIVATE_STATE_ENCRYPTION_KEY='<base64-encoded-32-byte-key>'
 # Required for cross-device unfiled Draft sync. The single-key setting remains
 # supported for the first deployment of keyring-capable code and writes v1 envelopes.
 # Inject the base64 encoding of exactly 32 random bytes from a secret manager.

src/available_issue_snapshot_store.py

@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import json
 import secrets
 import sqlite3
 import time
@@ -10,6 +9,11 @@ from dataclasses import dataclass
 from pathlib import Path
 
 from src.private_state import connect_private_sqlite
+from src.state_encryption import (
+    PrivateStateCipher,
+    PrivateStateEncryptionError,
+    private_state_encryption_key,
+)
 
 
 class RefreshLeaseLost(RuntimeError):
@@ -26,9 +30,13 @@ class AvailableIssueSnapshotState:
 
 
 class AvailableIssueSnapshotStore:
-    def __init__(self, path, *, clock=None):
+    def __init__(self, path, *, clock=None, encryption_key=None):
         self.path = Path(path)
         self.clock = clock or time.time
+        self._cipher = PrivateStateCipher(
+            encryption_key if encryption_key is not None else private_state_encryption_key(),
+            store="available-issue-snapshot",
+        )
         with self._connect() as connection:
             connection.executescript(
                     """
@@ -91,8 +99,21 @@ class AvailableIssueSnapshotStore:
                 "SELECT expires_at FROM available_issue_refresh_lease "
                 "WHERE singleton = 1 AND expires_at > ?", (now,)
             ).fetchone()
+        items = None
+        if row["items_json"]:
+            items, legacy = self._cipher.open(row["items_json"])
+            if not isinstance(items, list):
+                raise PrivateStateEncryptionError("private state could not be decrypted")
+            if legacy:
+                migrated = self._cipher.seal(items)
+                with self._connect() as connection:
+                    connection.execute(
+                        "UPDATE available_issue_snapshot SET items_json = ? "
+                        "WHERE singleton = 1 AND items_json = ?",
+                        (migrated, row["items_json"]),
+                    )
         return AvailableIssueSnapshotState(
-            items=json.loads(row["items_json"]) if row["items_json"] else None,
+            items=items,
             created_at=row["created_at"],
             retry_at=row["retry_at"],
             refreshing=lease is not None,
@@ -127,7 +148,7 @@ class AvailableIssueSnapshotStore:
             connection.execute(
                 "UPDATE available_issue_snapshot SET items_json = ?, created_at = ?, "
                 "retry_at = NULL WHERE singleton = 1",
-                (json.dumps(items, separators=(",", ":")), now),
+                (self._cipher.seal(items), now),
             )
             connection.execute("DELETE FROM available_issue_refresh_lease WHERE singleton = 1")
             connection.commit()
@@ -143,7 +164,7 @@ class AvailableIssueSnapshotStore:
             row = connection.execute(
                 "SELECT items_json FROM available_issue_snapshot WHERE singleton = 1"
             ).fetchone()
-            items = json.loads(row["items_json"]) if row["items_json"] else None
+            items = self._cipher.open(row["items_json"])[0] if row["items_json"] else None
             if items is not None:
                 items = [
                     item for item in items
@@ -151,7 +172,7 @@ class AvailableIssueSnapshotStore:
                 ]
                 connection.execute(
                     "UPDATE available_issue_snapshot SET items_json = ? WHERE singleton = 1",
-                    (json.dumps(items, separators=(",", ":")),),
+                    (self._cipher.seal(items),),
                 )
             connection.commit()
         return self.load()

src/live_snapshot_store.py

@@ -12,6 +12,11 @@ from pathlib import Path
 from typing import Callable, Iterable
 
 from src.private_state import connect_private_sqlite
+from src.state_encryption import (
+    PrivateStateCipher,
+    PrivateStateEncryptionError,
+    private_state_encryption_key,
+)
 
 SECTIONS = ("context", "events", "notifications")
 
@@ -53,9 +58,14 @@ class LiveSnapshotStore:
         path: str | os.PathLike[str],
         *,
         clock: Callable[[], float] | None = None,
+        encryption_key: bytes | None = None,
     ):
         self.path = Path(path)
         self.clock = clock or time.time
+        self._cipher = PrivateStateCipher(
+            encryption_key if encryption_key is not None else private_state_encryption_key(),
+            store="live-snapshot",
+        )
         self._initialize()
 
     def _connect(self) -> sqlite3.Connection:
@@ -138,8 +148,23 @@ class LiveSnapshotStore:
                 (now,),
             ).fetchone()
         assert row is not None
+        value = None
+        if row["value_json"] is not None:
+            value, legacy = self._cipher.open(
+                row["value_json"], binding=row["generation"]
+            )
+            if not isinstance(value, dict):
+                raise PrivateStateEncryptionError("private state could not be decrypted")
+            if legacy:
+                migrated = self._cipher.seal(value, binding=row["generation"])
+                with self._connect() as connection:
+                    connection.execute(
+                        "UPDATE live_snapshot SET value_json = ? "
+                        "WHERE singleton = 1 AND value_json = ?",
+                        (migrated, row["value_json"]),
+                    )
         return LiveSnapshotState(
-            value=json.loads(row["value_json"]) if row["value_json"] is not None else None,
+            value=value,
             created_at=json.loads(row["created_at_json"]),
             failure_count=json.loads(row["failure_count_json"]),
             retry_at=json.loads(row["retry_at_json"]),
@@ -198,7 +223,7 @@ class LiveSnapshotStore:
                 connection.rollback()
                 raise RefreshLeaseLost("live refresh lease expired or changed owner")
             row = connection.execute(
-                "SELECT revisions_json FROM live_snapshot WHERE singleton = 1"
+                "SELECT revisions_json, generation FROM live_snapshot WHERE singleton = 1"
             ).fetchone()
             revisions = json.loads(row["revisions_json"])
             notifications = value.get("notifications")
@@ -232,7 +257,7 @@ class LiveSnapshotStore:
                 failure_count_json = ?, retry_at_json = ?, revisions_json = ?
                 WHERE singleton = 1""",
                 (
-                    json.dumps(value, separators=(",", ":")),
+                    self._cipher.seal(value, binding=row["generation"]),
                     json.dumps(created_at, separators=(",", ":")),
                     json.dumps(failure_count, separators=(",", ":")),
                     json.dumps(retry_at, separators=(",", ":")),
@@ -257,9 +282,11 @@ class LiveSnapshotStore:
                 ((notification_id,) for notification_id in read_ids),
             )
             row = connection.execute(
-                "SELECT value_json, revisions_json FROM live_snapshot WHERE singleton = 1"
+                "SELECT value_json, revisions_json, generation FROM live_snapshot WHERE singleton = 1"
             ).fetchone()
-            value = json.loads(row["value_json"]) if row["value_json"] else None
+            value = self._cipher.open(
+                row["value_json"], binding=row["generation"]
+            )[0] if row["value_json"] else None
             revisions = json.loads(row["revisions_json"])
             if value is not None and isinstance(value.get("notifications"), list):
                 previous = value["notifications"]
@@ -274,7 +301,7 @@ class LiveSnapshotStore:
                     connection.execute(
                         "UPDATE live_snapshot SET value_json = ?, revisions_json = ? WHERE singleton = 1",
                         (
-                            json.dumps(value, separators=(",", ":")),
+                            self._cipher.seal(value, binding=row["generation"]),
                             json.dumps(revisions, separators=(",", ":")),
                         ),
                     )

src/state_encryption.py

@@ -0,0 +1,89 @@
+"""Authenticated envelopes for retained private dashboard state."""
+
+from __future__ import annotations
+
+import base64
+import binascii
+import json
+import os
+
+from cryptography.exceptions import InvalidTag
+from cryptography.hazmat.primitives.ciphers.aead import AESGCM
+
+
+class PrivateStateEncryptionError(RuntimeError):
+    """Private retained state could not be authenticated or decrypted."""
+
+
+def decode_private_state_encryption_key(encoded: str) -> bytes:
+    """Decode the independently injected 256-bit private-state key."""
+    try:
+        key = base64.b64decode(encoded, validate=True)
+    except (binascii.Error, ValueError) as error:
+        raise PrivateStateEncryptionError(
+            "private state encryption key is invalid"
+        ) from error
+    if len(key) != 32:
+        raise PrivateStateEncryptionError(
+            "private state encryption key must decode to exactly 32 bytes"
+        )
+    return key
+
+
+def private_state_encryption_key() -> bytes:
+    return decode_private_state_encryption_key(
+        os.getenv("STACKCHAIN_PRIVATE_STATE_ENCRYPTION_KEY", "")
+    )
+
+
+class PrivateStateCipher:
+    """Seal JSON values with store-specific authenticated context."""
+
+    def __init__(self, key: bytes, *, store: str):
+        if not isinstance(key, bytes) or len(key) != 32:
+            raise PrivateStateEncryptionError(
+                "private state encryption requires exactly 32 key bytes"
+            )
+        if not store or "\0" in store:
+            raise ValueError("private state store identity is invalid")
+        self._cipher = AESGCM(key)
+        self._store = store
+
+    def _aad(self, binding: str) -> bytes:
+        return f"stackchain:private-state:v1\0{self._store}\0{binding}".encode()
+
+    def seal(self, value: object, *, binding: str = "singleton") -> str:
+        plaintext = json.dumps(value, separators=(",", ":")).encode()
+        nonce = os.urandom(12)
+        sealed = nonce + self._cipher.encrypt(nonce, plaintext, self._aad(binding))
+        return "v1:" + base64.urlsafe_b64encode(sealed).decode()
+
+    def open(self, payload: str, *, binding: str = "singleton") -> tuple[object, bool]:
+        """Return the decoded value and whether plaintext migration is required."""
+        if not isinstance(payload, str):
+            raise PrivateStateEncryptionError("private state could not be decrypted")
+        if not payload.startswith("v1:"):
+            try:
+                return json.loads(payload), True
+            except (TypeError, json.JSONDecodeError) as error:
+                raise PrivateStateEncryptionError(
+                    "private state could not be decrypted"
+                ) from error
+        try:
+            sealed = base64.b64decode(payload[3:], altchars=b"-_", validate=True)
+            if len(sealed) < 28:
+                raise ValueError("encrypted payload is too short")
+            plaintext = self._cipher.decrypt(
+                sealed[:12], sealed[12:], self._aad(binding)
+            )
+            return json.loads(plaintext), False
+        except (
+            binascii.Error,
+            InvalidTag,
+            UnicodeDecodeError,
+            ValueError,
+            json.JSONDecodeError,
+        ) as error:
+            raise PrivateStateEncryptionError(
+                "private state could not be decrypted"
+            ) from error

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()