File size: 11,239 Bytes
7a278ee 0ba5932 7a278ee 4aac055 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 | import asyncio
import json
import urllib.parse
from datetime import datetime, timedelta, timezone
from pathlib import Path
import httpx
import pytest
from app import push as pushes
from app.answer_scans import scan_code
from app.auth import AuthUnavailable
from app.main import create_app
from scan_support import OTHER, OWNER, ScanAuth, answer_store, postgres_server # noqa: F401
KEY = pushes.new_private_key()
SUB = {"endpoint": "https://push.example/abc", "keys": {"p256dh": "p" * 87, "auth": "a" * 22}}
class PushAuth(ScanAuth):
"""ScanAuth with an in-memory math_push_subs behind the PostgREST calls Push makes."""
def __init__(self):
self.rows: dict[str, dict] = {}
def _admin(self, path, method="GET", body=None, prefer=""):
table, _, query = path.partition("?")
if table != "/rest/v1/math_push_subs":
return super()._admin(path) # no other tables here (a quiz answer looks for a final sitting)
q = dict(urllib.parse.parse_qsl(query))
match = lambda r: all(r[k] == v.removeprefix("eq.") for k, v in q.items() if k in ("endpoint", "user_id"))
if method == "POST":
self.rows[body["endpoint"]] = dict(body)
return None
if method == "DELETE":
for e in [e for e, r in self.rows.items() if match(r)]:
del self.rows[e]
return None
return [r for r in self.rows.values() if match(r)]
def scan(state="waiting", minutes=10, sid="7d9c2f0e-3c1a-4c55-9a1e-2b3c4d5e6f70"):
end = datetime.now(timezone.utc) + timedelta(minutes=minutes)
return {"attempt": {"id": "11111111-1111-1111-1111-111111111111"},
"scan": {"id": sid, "state": state, "expires_at": end.isoformat()}}
@pytest.fixture
def sent(monkeypatch):
calls = []
def webpush(sub, data, **kwargs):
calls.append({"endpoint": sub["endpoint"], "data": json.loads(data), **kwargs})
import pywebpush
monkeypatch.setattr(pywebpush, "webpush", webpush)
return calls
def test_payload_has_code_and_no_question_and_replaces_by_attempt():
result = scan()
msg = pushes.scan_payload(result)
assert scan_code(result["scan"]["id"]) in msg["body"]
assert msg["tag"] == "scan-11111111-1111-1111-1111-111111111111"
assert msg["url"] == "/?from=push-scan"
assert 590 <= pushes.seconds_left(result) <= 600
assert pushes.seconds_left(scan(minutes=-1)) == 0
assert pushes.seconds_left({"scan": {"expires_at": None}}) == 900
def test_keys_round_trip_and_off_reasons():
assert len(pushes.public_key(KEY)) == 87 # 65 bytes, base64url
assert pushes.Push(PushAuth(), "").why == "VAPID_PRIVATE_KEY is not set"
assert pushes.Push(PushAuth(), "not a key!").why == "VAPID_PRIVATE_KEY is not a valid key"
p = pushes.Push(PushAuth(), KEY, "mailto:a@test.invalid")
assert p.on and p.public == pushes.public_key(KEY)
def test_request_goes_once_to_the_accounts_phones_and_gone_ones_are_deleted(sent, monkeypatch):
auth = PushAuth()
p = pushes.Push(auth, KEY, "mailto:a@test.invalid")
p.save(OWNER, SUB)
p.save(OWNER, {**SUB, "endpoint": "https://push.example/gone"})
p.save(OTHER, {**SUB, "endpoint": "https://push.example/other"})
with pytest.raises(ValueError):
p.save(OWNER, {"endpoint": "http://insecure", "keys": SUB["keys"]})
import pywebpush
real = pywebpush.webpush
def webpush(sub, data, **kwargs):
if sub["endpoint"].endswith("/gone"):
raise pywebpush.WebPushException("gone", response=type("R", (), {"status_code": 410})())
return real(sub, data, **kwargs)
monkeypatch.setattr(pywebpush, "webpush", webpush)
assert p.scan_requested(OWNER, scan()) == 1
assert [c["endpoint"] for c in sent] == [SUB["endpoint"]]
assert sent[0]["headers"] == {"Urgency": "high"} and 0 < sent[0]["ttl"] <= 600
assert "https://push.example/gone" not in auth.rows and "https://push.example/other" in auth.rows
assert p.scan_requested(OWNER, scan()) == 0 # the same request again: no second ring
assert p.scan_requested(OWNER, scan("received", sid="22222222-2222-2222-2222-222222222222")) == 0
assert p.scan_requested(OWNER, scan(minutes=-1, sid="33333333-3333-3333-3333-333333333333")) == 0
assert len(sent) == 1
def test_missing_table_does_not_break_the_request(sent):
class NoTable(PushAuth):
def _admin(self, *a, **k):
raise AuthUnavailable("supabase answered 404: relation math_push_subs does not exist")
assert pushes.Push(NoTable(), KEY, "mailto:a@test.invalid").scan_requested(OWNER, scan()) == 0
def test_subscribe_api_moves_a_phone_between_accounts(monkeypatch):
monkeypatch.setenv("VAPID_PRIVATE_KEY", KEY)
auth = PushAuth()
app = create_app(auth=auth)
async def check():
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as c:
assert (await c.get("/api/push")).status_code == 401
owner, other = ({"Authorization": f"Bearer {u}"} for u in (OWNER, OTHER))
assert (await c.get("/api/push", headers=owner)).json()["key"] == pushes.public_key(KEY)
assert (await c.post("/api/push/subscribe", json={"endpoint": "x"}, headers=owner)).status_code == 400
assert (await c.post("/api/push/subscribe", json=SUB, headers=owner)).json() == {"ok": True}
assert auth.rows[SUB["endpoint"]]["user_id"] == OWNER
assert (await c.post("/api/push/subscribe", json=SUB, headers=other)).status_code == 200
assert auth.rows[SUB["endpoint"]]["user_id"] == OTHER # one row per phone: the account signed in now
await c.post("/api/push/unsubscribe", json={"endpoint": SUB["endpoint"]}, headers=owner)
assert SUB["endpoint"] in auth.rows # another account cannot remove it
await c.post("/api/push/unsubscribe", json={"endpoint": SUB["endpoint"]}, headers=other)
assert not auth.rows
sw = await c.get("/service-worker.js")
assert "showNotification" in sw.text and 'addEventListener("fetch"' not in sw.text and sw.headers["cache-control"] == "no-store"
asyncio.run(check())
def test_push_off_without_key(monkeypatch):
monkeypatch.delenv("VAPID_PRIVATE_KEY", raising=False)
app = create_app(auth=PushAuth())
async def check():
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test",
headers={"Authorization": f"Bearer {OWNER}"}) as c:
assert (await c.get("/api/push")).json() == {"key": ""}
assert (await c.post("/api/push/subscribe", json=SUB)).status_code == 503
asyncio.run(check())
def test_computer_request_pushes_to_the_phone(answer_store, sent, monkeypatch):
monkeypatch.setenv("VAPID_PRIVATE_KEY", KEY)
auth = PushAuth()
app = create_app(auth=auth)
app.state.answers = answer_store
app.state.push.save(OWNER, SUB)
aid = answer_store.action(OWNER, "resume", question_id="q", question_version="v1",
snapshot={"question": "Prove it", "type": "open-ended"})["attempt"]["id"]
async def check():
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test",
headers={"Authorization": f"Bearer {OWNER}"}) as c:
first = (await c.post(f"/api/math/answer-attempts/{aid}/scans", json={})).json()
again = (await c.post(f"/api/math/answer-attempts/{aid}/scans", json={})).json()
assert again["scan"]["id"] == first["scan"]["id"]
renewed = (await c.post(f"/api/math/answer-attempts/{aid}/scans", json={"renew": True})).json()
return first, renewed
first, renewed = asyncio.run(check())
assert len(sent) == 2 # the repeated request did not ring again; the renewed one did
assert first["scan"]["code"] in sent[0]["data"]["body"] and renewed["scan"]["code"] in sent[1]["data"]["body"]
assert sent[0]["data"]["tag"] == sent[1]["data"]["tag"] == f"scan-{aid}"
assert "Prove it" not in json.dumps(sent)
def test_migration_keeps_subscriptions_from_browsers(postgres_server):
import uuid
import psycopg
database = "push_" + uuid.uuid4().hex
with psycopg.connect(postgres_server.get_uri(), autocommit=True) as c:
c.execute("create database " + database)
try:
with psycopg.connect(postgres_server.get_uri(database=database), autocommit=True) as c:
c.execute("create schema auth; create table auth.users(id uuid primary key);"
"grant usage on schema public to anon, authenticated, service_role;")
c.execute(Path("supabase/migrations/20261103180000_math_push_subs.sql").read_text())
c.execute("insert into auth.users values(%s)", (OWNER,))
c.execute("set role service_role")
c.execute("insert into math_push_subs(endpoint,user_id,p256dh,auth) values(%s,%s,'p','a')", (SUB["endpoint"], OWNER))
c.execute("reset role; set role authenticated")
with pytest.raises(psycopg.errors.InsufficientPrivilege):
c.execute("select * from math_push_subs")
c.execute("reset role")
with pytest.raises(psycopg.errors.CheckViolation):
c.execute("insert into math_push_subs(endpoint,user_id,p256dh,auth) values('http://x',%s,'p','a')", (OWNER,))
finally:
with psycopg.connect(postgres_server.get_uri(), autocommit=True) as c:
c.execute("drop database " + database + " with (force)")
def test_a_devices_own_upload_rings_no_phone_and_is_offered_to_none(answer_store, sent, monkeypatch):
monkeypatch.setenv("VAPID_PRIVATE_KEY", KEY)
app = create_app(auth=PushAuth())
app.state.answers = answer_store
app.state.push.save(OWNER, SUB)
own = answer_store.action(OWNER, "resume", question_id="quiz:t:0", question_version="v1",
snapshot={"question": "Solve", "type": "open-ended"})["attempt"]["id"]
asked = answer_store.action(OWNER, "resume", question_id="quiz:t:1", question_version="v1",
snapshot={"question": "Prove", "type": "open-ended"})["attempt"]["id"]
async def check():
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test",
headers={"Authorization": f"Bearer {OWNER}"}) as c:
direct = (await c.post(f"/api/math/answer-attempts/{own}/scans", json={"renew": True, "direct": True})).json()
phone = (await c.post(f"/api/math/answer-attempts/{asked}/scans", json={"renew": True})).json()
listed = (await c.get("/api/math/answer-scans")).json()
return direct, phone, listed
direct, phone, listed = asyncio.run(check())
assert len(sent) == 1 and phone["scan"]["code"] in sent[0]["data"]["body"] # only the request for the phone rang
assert [r["scan"]["id"] for r in listed] == [phone["scan"]["id"]] # and only it waits for a phone
assert direct["scan"]["state"] == "waiting" # the device's own request works as before
|