199 lines
7.9 KiB
Python
199 lines
7.9 KiB
Python
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"
|