security: encrypt synchronized Saved Searches (Closes #1126)

tests/test_saved_search_store.py

@@ -1,8 +1,12 @@
+import json
+import sqlite3
+
 import httpx
 import pytest
 
 from src import main
 from src.saved_search_store import SavedSearchConflict, SavedSearchStore
+from src.state_encryption import PrivateStateEncryptionError
 
 
 def view(view_id="release", name="Release queue", query="mobile", **scope):
@@ -16,6 +20,105 @@ def view(view_id="release", name="Release queue", query="mobile", **scope):
     }
 
 
+def test_saved_searches_encrypt_private_views_and_survive_restart(tmp_path):
+    database = tmp_path / "saved-searches.sqlite3"
+    key = b"s" * 32
+    canary = "private-launch-query-canary"
+    store = SavedSearchStore(database, encryption_key=key)
+
+    created = store.replace(
+        "Timmy",
+        0,
+        [view(name="Private launch", query=canary, repository="private/launch")],
+    )
+
+    with sqlite3.connect(database) as connection:
+        payload = connection.execute(
+            "SELECT views FROM saved_searches WHERE login = 'timmy'"
+        ).fetchone()[0]
+    assert payload.startswith("v1:")
+    assert "Private launch" not in payload
+    assert canary not in payload
+    assert "private/launch" not in payload
+    assert SavedSearchStore(database, encryption_key=key).get("timmy") == created
+
+
+def test_saved_searches_lazily_migrate_plaintext_without_changing_revision(tmp_path):
+    database = tmp_path / "saved-searches.sqlite3"
+    key = b"s" * 32
+    legacy_views = [view(name="Legacy private view", query="legacy-canary")]
+    SavedSearchStore(database, encryption_key=key)
+    with sqlite3.connect(database) as connection:
+        connection.execute(
+            "INSERT INTO saved_searches(login, revision, views) VALUES (?, ?, ?)",
+            ("timmy", 7, json.dumps(legacy_views)),
+        )
+
+    snapshot = SavedSearchStore(database, encryption_key=key).get("Timmy")
+
+    with sqlite3.connect(database) as connection:
+        migrated = connection.execute(
+            "SELECT revision, views FROM saved_searches WHERE login = 'timmy'"
+        ).fetchone()
+    assert snapshot == {"revision": 7, "views": legacy_views}
+    assert migrated[0] == 7
+    assert migrated[1].startswith("v1:")
+    assert "legacy-canary" not in migrated[1]
+
+
+def test_saved_searches_fail_closed_with_wrong_key(tmp_path):
+    database = tmp_path / "saved-searches.sqlite3"
+    SavedSearchStore(database, encryption_key=b"s" * 32).replace(
+        "timmy", 0, [view(query="wrong-key-canary")]
+    )
+
+    with pytest.raises(
+        PrivateStateEncryptionError, match="private state could not be decrypted"
+    ):
+        SavedSearchStore(database, encryption_key=b"x" * 32).get("timmy")
+
+
+def test_saved_searches_reject_tampered_or_malformed_payloads(tmp_path):
+    database = tmp_path / "saved-searches.sqlite3"
+    store = SavedSearchStore(database, encryption_key=b"s" * 32)
+    store.replace("timmy", 0, [view(query="tamper-canary")])
+    with sqlite3.connect(database) as connection:
+        payload = connection.execute(
+            "SELECT views FROM saved_searches WHERE login = 'timmy'"
+        ).fetchone()[0]
+
+    for invalid in (payload[:-1] + ("A" if payload[-1] != "A" else "B"), "not-json"):
+        with sqlite3.connect(database) as connection:
+            connection.execute(
+                "UPDATE saved_searches SET views = ? WHERE login = 'timmy'",
+                (invalid,),
+            )
+        with pytest.raises(
+            PrivateStateEncryptionError, match="private state could not be decrypted"
+        ):
+            store.get("timmy")
+
+
+def test_saved_searches_reject_ciphertext_substituted_between_accounts(tmp_path):
+    database = tmp_path / "saved-searches.sqlite3"
+    store = SavedSearchStore(database, encryption_key=b"s" * 32)
+    store.replace("timmy", 0, [view(query="timmy-private")])
+    store.replace("alexander", 0, [view(query="alexander-private")])
+    with sqlite3.connect(database) as connection:
+        timmy_payload = connection.execute(
+            "SELECT views FROM saved_searches WHERE login = 'timmy'"
+        ).fetchone()[0]
+        connection.execute(
+            "UPDATE saved_searches SET views = ? WHERE login = 'alexander'",
+            (timmy_payload,),
+        )
+
+    with pytest.raises(
+        PrivateStateEncryptionError, match="private state could not be decrypted"
+    ):
+        store.get("alexander")
+
+
 def test_saved_searches_are_revisioned_ordered_and_account_scoped(tmp_path):
     store = SavedSearchStore(tmp_path / "saved-searches.sqlite3")