feat: encrypt completed Filed review history (Closes #1252)

src/completed_filed_review_store.py

@@ -1,4 +1,4 @@
-"""Durable, account-scoped completed Filed review receipts."""
+"""Durable, encrypted, account-scoped completed Filed review receipts."""
 
 import re
 import sqlite3
@@ -6,33 +6,68 @@ 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_config
 
 
 _REPOSITORY = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$")
 
 
 class CompletedFiledReviewStore:
-    def __init__(self, path: str | Path, *, limit: int = 200, timeout: float = 1.0):
+    def __init__(
+        self,
+        path: str | Path,
+        *,
+        limit: int = 200,
+        timeout: float = 1.0,
+        encryption_key: bytes | None = None,
+    ):
         self.path = Path(path)
         self.limit = limit
         self.timeout = timeout
+        self._cipher = PrivateStateCipher(
+            encryption_key if encryption_key is not None else private_state_encryption_config(),
+            store="completed-filed-reviews",
+        )
         self._initialize()
 
     def _initialize(self) -> None:
         with self._connect() as connection:
             connection.execute("PRAGMA journal_mode=WAL")
+            connection.execute("PRAGMA secure_delete=ON")
             connection.execute(
                 """
-                CREATE TABLE IF NOT EXISTS completed_filed_reviews (
-                    login TEXT NOT NULL,
-                    repository TEXT NOT NULL,
-                    issue_number INTEGER NOT NULL,
-                    updated_at TEXT NOT NULL,
-                    touched_at INTEGER NOT NULL,
-                    PRIMARY KEY (login, repository, issue_number)
+                CREATE TABLE IF NOT EXISTS completed_filed_review_collections (
+                    login TEXT PRIMARY KEY,
+                    receipts TEXT NOT NULL
                 )
                 """
             )
+            legacy = connection.execute(
+                "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'completed_filed_reviews'"
+            ).fetchone()
+            if legacy:
+                connection.execute("BEGIN IMMEDIATE")
+                rows = connection.execute(
+                    "SELECT login, repository, issue_number, updated_at "
+                    "FROM completed_filed_reviews ORDER BY login, touched_at"
+                ).fetchall()
+                grouped: dict[str, list[dict]] = {}
+                for login, repository, number, updated_at in rows:
+                    grouped.setdefault(login, []).append({
+                        "repository": repository,
+                        "number": int(number),
+                        "updated_at": updated_at,
+                    })
+                for login, receipts in grouped.items():
+                    normalized = [self._receipt(item) for item in receipts][-self.limit:]
+                    connection.execute(
+                        "INSERT INTO completed_filed_review_collections(login, receipts) VALUES (?, ?) "
+                        "ON CONFLICT(login) DO UPDATE SET receipts=excluded.receipts",
+                        (login, self._seal(login, normalized)),
+                    )
+                connection.execute("DROP TABLE completed_filed_reviews")
+        with self._connect() as connection:
+            connection.execute("PRAGMA wal_checkpoint(TRUNCATE)")
 
     def _connect(self) -> sqlite3.Connection:
         return connect_private_sqlite(self.path, timeout=self.timeout)
@@ -45,7 +80,7 @@ class CompletedFiledReviewStore:
         return normalized
 
     @staticmethod
-    def _receipt(raw: dict) -> tuple[str, int, str]:
+    def _receipt(raw: dict) -> dict:
         if not isinstance(raw, dict):
             raise ValueError("receipt must be an object")
         repository = raw.get("repository")
@@ -63,23 +98,41 @@ class CompletedFiledReviewStore:
             raise ValueError("updated_at is invalid") from error
         if parsed.tzinfo is None:
             raise ValueError("updated_at is invalid")
-        return repository, number, updated_at
+        return {"repository": repository, "number": number, "updated_at": updated_at}
+
+    def _open(self, login: str, payload: str | None) -> tuple[list[dict], bool]:
+        if payload is None:
+            return [], False
+        value, stale = self._cipher.open(payload, binding=f"receipts:{login}")
+        if not isinstance(value, list):
+            raise PrivateStateEncryptionError("private state could not be decrypted")
+        try:
+            return [self._receipt(item) for item in value], stale
+        except ValueError as error:
+            raise PrivateStateEncryptionError("private state could not be decrypted") from error
+
+    def _seal(self, login: str, receipts: list[dict]) -> str:
+        return self._cipher.seal(receipts, binding=f"receipts:{login}")
 
     @staticmethod
-    def _snapshot(rows) -> dict:
-        return {"receipts": [
-            {"repository": row[0], "number": int(row[1]), "updated_at": row[2]}
-            for row in rows
-        ]}
+    def _snapshot(receipts: list[dict]) -> dict:
+        return {"receipts": receipts}
 
     def get(self, login: str) -> dict:
+        login = self._login(login)
         with self._connect() as connection:
-            rows = connection.execute(
-                "SELECT repository, issue_number, updated_at FROM completed_filed_reviews "
-                "WHERE login = ? ORDER BY touched_at",
-                (self._login(login),),
-            ).fetchall()
-        return self._snapshot(rows)
+            row = connection.execute(
+                "SELECT receipts FROM completed_filed_review_collections WHERE login = ?",
+                (login,),
+            ).fetchone()
+            receipts, stale = self._open(login, row[0] if row else None)
+            if row is not None and stale:
+                connection.execute(
+                    "UPDATE completed_filed_review_collections SET receipts = ? "
+                    "WHERE login = ? AND receipts = ?",
+                    (self._seal(login, receipts), login, row[0]),
+                )
+        return self._snapshot(receipts)
 
     def merge(self, login: str, receipts: list[dict]) -> dict:
         login = self._login(login)
@@ -88,35 +141,25 @@ class CompletedFiledReviewStore:
         normalized = [self._receipt(receipt) for receipt in receipts]
         with self._connect() as connection:
             connection.execute("BEGIN IMMEDIATE")
-            touched = int(connection.execute(
-                "SELECT COALESCE(MAX(touched_at), 0) FROM completed_filed_reviews WHERE login = ?",
+            row = connection.execute(
+                "SELECT receipts FROM completed_filed_review_collections WHERE login = ?",
                 (login,),
-            ).fetchone()[0])
-            for repository, number, updated_at in normalized:
-                current = connection.execute(
-                    "SELECT updated_at FROM completed_filed_reviews "
-                    "WHERE login = ? AND repository = ? AND issue_number = ?",
-                    (login, repository, number),
-                ).fetchone()
-                if current is not None and current[0] >= updated_at:
+            ).fetchone()
+            current, _stale = self._open(login, row[0] if row else None)
+            for incoming in normalized:
+                match = next((
+                    item for item in current
+                    if item["repository"] == incoming["repository"] and item["number"] == incoming["number"]
+                ), None)
+                if match is not None and match["updated_at"] >= incoming["updated_at"]:
                     continue
-                touched += 1
-                connection.execute(
-                    "INSERT INTO completed_filed_reviews "
-                    "(login, repository, issue_number, updated_at, touched_at) VALUES (?, ?, ?, ?, ?) "
-                    "ON CONFLICT(login, repository, issue_number) DO UPDATE SET "
-                    "updated_at=excluded.updated_at, touched_at=excluded.touched_at",
-                    (login, repository, number, updated_at, touched),
-                )
+                if match is not None:
+                    current.remove(match)
+                current.append(incoming)
+            current = current[-self.limit:]
             connection.execute(
-                "DELETE FROM completed_filed_reviews WHERE login = ? AND rowid NOT IN ("
-                "SELECT rowid FROM completed_filed_reviews WHERE login = ? "
-                "ORDER BY touched_at DESC LIMIT ?)",
-                (login, login, self.limit),
+                "INSERT INTO completed_filed_review_collections(login, receipts) VALUES (?, ?) "
+                "ON CONFLICT(login) DO UPDATE SET receipts=excluded.receipts",
+                (login, self._seal(login, current)),
             )
-            rows = connection.execute(
-                "SELECT repository, issue_number, updated_at FROM completed_filed_reviews "
-                "WHERE login = ? ORDER BY touched_at",
-                (login,),
-            ).fetchall()
-        return self._snapshot(rows)
+        return self._snapshot(current)