security: encrypt synchronized Later state (Closes #1114)

src/later_store.py

@@ -1,12 +1,12 @@
 """Durable, account-scoped Later deferrals."""
 
-import json
 import sqlite3
 import time
 from datetime import datetime
 from pathlib import Path
 
 from src.private_state import connect_private_sqlite
+from src.state_encryption import PrivateStateCipher, PrivateStateEncryptionError, private_state_encryption_key
 
 
 class LaterStore:
@@ -17,6 +17,7 @@ class LaterStore:
         timeout: float = 1.0,
         operation_limit: int = 4096,
         operation_retention_seconds: float = 30 * 24 * 60 * 60,
+        encryption_key: bytes | None = None,
         clock=time.time,
     ):
         self.path = Path(path)
@@ -24,6 +25,10 @@ class LaterStore:
         self.operation_limit = operation_limit
         self.operation_retention_seconds = operation_retention_seconds
         self.clock = clock
+        self._cipher = PrivateStateCipher(
+            encryption_key if encryption_key is not None else private_state_encryption_key(),
+            store="later",
+        )
         self._initialize()
 
     def _initialize(self) -> None:
@@ -99,11 +104,55 @@ class LaterStore:
             raise ValueError("login is required")
         return normalized
 
-    @staticmethod
-    def _snapshot(row) -> dict:
+    def _snapshot(self, row, login: str) -> tuple[dict, bool]:
         if row is None:
-            return {"revision": 0, "records": {}}
-        return {"revision": int(row[0]), "records": json.loads(row[1])}
+            return {"revision": 0, "records": {}}, False
+        records, legacy = self._cipher.open(row[1], binding=f"plan:{login}")
+        if not isinstance(records, dict):
+            raise PrivateStateEncryptionError("private state could not be decrypted")
+        return {"revision": int(row[0]), "records": records}, legacy
+
+    def _sealed_records(self, login: str, records: dict) -> str:
+        return self._cipher.seal(records, binding=f"plan:{login}")
+
+    def _item_revisions(
+        self, connection: sqlite3.Connection, login: str
+    ) -> tuple[dict[str, int], dict[str, str]]:
+        revisions: dict[str, int] = {}
+        stored_ids: dict[str, str] = {}
+        for stored_item_id, revision in connection.execute(
+            "SELECT item_id, revision FROM later_item_revisions WHERE login = ?",
+            (login,),
+        ):
+            if stored_item_id.startswith("v1:"):
+                item_id, _legacy = self._cipher.open(
+                    stored_item_id, binding=f"item-revision:{login}"
+                )
+            else:
+                item_id = stored_item_id
+            if not isinstance(item_id, str) or not item_id:
+                raise PrivateStateEncryptionError("private state could not be decrypted")
+            revisions[item_id] = int(revision)
+            stored_ids[item_id] = stored_item_id
+        return revisions, stored_ids
+
+    def _migrate_item_ids(
+        self, connection: sqlite3.Connection, login: str, stored_ids: dict[str, str]
+    ) -> dict[str, str]:
+        migrated = dict(stored_ids)
+        for item_id, stored_item_id in stored_ids.items():
+            if stored_item_id.startswith("v1:"):
+                continue
+            sealed_item_id = self._cipher.seal(
+                item_id, binding=f"item-revision:{login}"
+            )
+            connection.execute(
+                "UPDATE later_item_revisions SET item_id = ? "
+                "WHERE login = ? AND item_id = ?",
+                (sealed_item_id, login, stored_item_id),
+            )
+            migrated[item_id] = sealed_item_id
+        return migrated
 
     @staticmethod
     def _validate_wake_at(wake_at: str | None) -> str:
@@ -116,12 +165,21 @@ class LaterStore:
         return wake_at
 
     def get(self, login: str) -> dict:
+        login = self._normalize_login(login)
         with self._connect() as connection:
             row = connection.execute(
                 "SELECT revision, records FROM later_plans WHERE login = ?",
-                (self._normalize_login(login),),
+                (login,),
             ).fetchone()
-        return self._snapshot(row)
+            snapshot, legacy_plan = self._snapshot(row, login)
+            _revisions, stored_item_ids = self._item_revisions(connection, login)
+            if row is not None and legacy_plan:
+                connection.execute(
+                    "UPDATE later_plans SET records = ? WHERE login = ? AND records = ?",
+                    (self._sealed_records(login, snapshot["records"]), login, row[1]),
+                )
+            self._migrate_item_ids(connection, login, stored_item_ids)
+        return snapshot
 
     def apply(
         self,
@@ -152,7 +210,7 @@ class LaterStore:
             row = connection.execute(
                 "SELECT revision, records FROM later_plans WHERE login = ?", (login,)
             ).fetchone()
-            snapshot = self._snapshot(row)
+            snapshot, legacy_plan = self._snapshot(row, login)
             records = dict(snapshot["records"])
             revision = snapshot["revision"]
             accepted: list[str] = []
@@ -161,15 +219,19 @@ class LaterStore:
 
             # Existing databases predate per-item revisions. Conservatively mark
             # active deferrals as changed at the latest known plan revision.
+            item_revisions, stored_item_ids = self._item_revisions(connection, login)
+            stored_item_ids = self._migrate_item_ids(connection, login, stored_item_ids)
             for item_id in records:
-                connection.execute(
-                    "INSERT OR IGNORE INTO later_item_revisions(login, item_id, revision) VALUES (?, ?, ?)",
-                    (login, item_id, revision),
-                )
-            item_revisions = dict(connection.execute(
-                "SELECT item_id, revision FROM later_item_revisions WHERE login = ?",
-                (login,),
-            ).fetchall())
+                if item_id not in item_revisions:
+                    stored_item_id = self._cipher.seal(
+                        item_id, binding=f"item-revision:{login}"
+                    )
+                    connection.execute(
+                        "INSERT INTO later_item_revisions(login, item_id, revision) VALUES (?, ?, ?)",
+                        (login, stored_item_id, revision),
+                    )
+                    item_revisions[item_id] = revision
+                    stored_item_ids[item_id] = stored_item_id
             batch_start_item_revisions = dict(item_revisions)
 
             for operation in operations:
@@ -218,21 +280,32 @@ class LaterStore:
                 revision += 1 if changed else 0
                 if changed:
                     item_revisions[item_id] = revision
-                    connection.execute(
-                        "INSERT INTO later_item_revisions(login, item_id, revision) VALUES (?, ?, ?) "
-                        "ON CONFLICT(login, item_id) DO UPDATE SET revision = excluded.revision",
-                        (login, item_id, revision),
-                    )
+                    stored_item_id = stored_item_ids.get(item_id)
+                    if stored_item_id is None:
+                        stored_item_id = self._cipher.seal(
+                            item_id, binding=f"item-revision:{login}"
+                        )
+                        stored_item_ids[item_id] = stored_item_id
+                        connection.execute(
+                            "INSERT INTO later_item_revisions(login, item_id, revision) VALUES (?, ?, ?)",
+                            (login, stored_item_id, revision),
+                        )
+                    else:
+                        connection.execute(
+                            "UPDATE later_item_revisions SET revision = ? "
+                            "WHERE login = ? AND item_id = ?",
+                            (revision, login, stored_item_id),
+                        )
                 self._record_operation(connection, login, operation_id)
                 accepted.append(operation_id)
 
-            serialized = json.dumps(records, separators=(",", ":"), sort_keys=True)
+            serialized = self._sealed_records(login, records)
             if row is None:
                 connection.execute(
                     "INSERT INTO later_plans(login, revision, records) VALUES (?, ?, ?)",
                     (login, revision, serialized),
                 )
-            elif accepted:
+            elif accepted or legacy_plan:
                 connection.execute(
                     "UPDATE later_plans SET revision = ?, records = ? WHERE login = ?",
                     (revision, serialized, login),