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): return { "id": view_id, "name": name, "query": query, "kind": scope.get("kind", "issue"), "state": scope.get("state", "open"), "repository": scope.get("repository", "stackchain/stackchain-dashboard"), } 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") created = store.replace(" Timmy ", 0, [view(), view("reviews", "My reviews", "review", kind="pull")]) assert created == {"revision": 1, "views": [view(), view("reviews", "My reviews", "review", kind="pull")]} assert store.get("timmy") == created assert store.get("alexander") == {"revision": 0, "views": []} with pytest.raises(SavedSearchConflict) as conflict: store.replace("timmy", 0, [view(name="Stale overwrite")]) assert conflict.value.snapshot == created def test_saved_searches_validate_and_bound_the_synced_collection(tmp_path): store = SavedSearchStore(tmp_path / "saved-searches.sqlite3", limit=2) with pytest.raises(ValueError, match="limited to 2"): store.replace("timmy", 0, [view("one"), view("two"), view("three")]) with pytest.raises(ValueError, match="unique"): store.replace("timmy", 0, [view("same"), view("same")]) invalid = [ ({**view(), "name": " "}, "name is required"), ({**view(), "query": "x"}, "query must be between"), ({**view(), "kind": "commit"}, "kind is invalid"), ({**view(), "state": "merged"}, "state is invalid"), ({**view(), "repository": "not-a-repository"}, "repository is invalid"), ] for candidate, message in invalid: with pytest.raises(ValueError, match=message): store.replace("timmy", 0, [candidate]) @pytest.mark.anyio async def test_saved_search_api_is_authenticated_csrf_protected_no_store_and_conflict_safe(monkeypatch, tmp_path): monkeypatch.setenv("STACKCHAIN_DASHBOARD_AUTH_MODE", "operator") monkeypatch.setenv("STACKCHAIN_DASHBOARD_ACCESS_TOKEN", "correct horse battery staple") monkeypatch.setenv( "STACKCHAIN_DASHBOARD_SESSION_SECRET", "a-separate-session-signing-secret-with-enough-entropy", ) monkeypatch.setenv("STACKCHAIN_SESSION_DB", str(tmp_path / "sessions.sqlite3")) monkeypatch.setenv("STACKCHAIN_LOGIN_ATTEMPT_DB", str(tmp_path / "login.sqlite3")) monkeypatch.setenv("STACKCHAIN_SAVED_SEARCH_DB", str(tmp_path / "searches.sqlite3")) async def user(): return {"id": 1, "login": "Timmy"} monkeypatch.setattr(main, "current_user", user) transport = httpx.ASGITransport(app=main.app) async with httpx.AsyncClient(transport=transport, base_url="https://test") as client: await client.post("/api/v1/session", json={"access_token": "correct horse battery staple"}) forbidden = await client.put( "/api/v1/saved-searches", json={"revision": 0, "views": [view()]} ) headers = {"Origin": "https://test", "X-CSRF-Token": client.cookies["stackchain_csrf"]} saved = await client.put( "/api/v1/saved-searches", json={"revision": 0, "views": [view()]}, headers=headers ) stale = await client.put( "/api/v1/saved-searches", json={"revision": 0, "views": [view(name="Overwrite")]}, headers=headers, ) fetched = await client.get("/api/v1/saved-searches") assert forbidden.status_code == 403 assert saved.status_code == 200 assert saved.json() == {"revision": 1, "views": [view()]} assert stale.status_code == 409 assert stale.json()["detail"] == { "message": "Saved searches changed on another device.", "snapshot": saved.json(), } assert fetched.json() == saved.json() assert fetched.headers["cache-control"] == "no-store"