tests/test_subrep.py (17029 bytes)
1 """Subrep Pro: the plan of the Subrep desktop app, sold with the accounts of this site. 2 3 yearly $9.99 a year (first on the page) 4 monthly $0.99 a month 5 6 A Stripe subscription makes the account Pro. The desktop app does not have the 7 cookie of the browser: the page gives a link command with the account id and a 8 refresh key, and the app sends the two to /api/subrep/refresh. The answer is a 9 licence token that the app checks offline with the public key in its build. 10 A Subrep plan is not a book plan: an account can have both, or one of them. 11 """ 12 13 from __future__ import annotations 14 15 import base64 16 import json 17 import time 18 from urllib.parse import parse_qs, urlparse 19 20 import pytest 21 import stripe 22 from cryptography.exceptions import InvalidSignature 23 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey 24 from hypothesis import given 25 from hypothesis import settings as hsettings 26 from hypothesis import strategies as st 27 28 from backend import accounts, subrep 29 from backend.db import Account, SessionLocal 30 from backend.settings import settings 31 32 from .conftest import FakeStripeObject, account_id, checkout_event, post_webhook 33 34 SEED = bytes(range(32)) # a fixed key for the tests, never the key of a server 35 PRIVATE_KEY = base64.urlsafe_b64encode(SEED).rstrip(b"=").decode() 36 PUBLIC_KEY = Ed25519PrivateKey.from_private_bytes(SEED).public_key() 37 DAY = 86400 38 39 40 @pytest.fixture(autouse=True) 41 def licence_key(monkeypatch): 42 monkeypatch.setattr(settings, "subrep_license_key", PRIVATE_KEY) 43 44 45 def b64d(text: str) -> bytes: 46 return base64.urlsafe_b64decode(text + "=" * (-len(text) % 4)) 47 48 49 def verify(token: str) -> dict: 50 """The check of the desktop app (verify_token in subrep/licensing.py).""" 51 payload, sig = token.split(".") 52 PUBLIC_KEY.verify(b64d(sig), b64d(payload)) 53 return json.loads(b64d(payload)) 54 55 56 def sub_object(account, plan_id, status="active", period_end=None, sub_id=None, 57 price_key=None) -> dict: 58 item: dict = {"current_period_end": period_end or int(time.time()) + 30 * DAY} 59 if price_key: 60 item["price"] = {"id": f"price_{price_key}", "lookup_key": price_key} 61 return { 62 "id": sub_id or f"sub_subrep_{account}", "object": "subscription", 63 "status": status, "customer": f"cus_{account}", 64 "metadata": {"account_id": account, "plan_id": plan_id}, 65 "items": {"data": [item]}, 66 } 67 68 69 def sub_event(kind: str, obj: dict) -> dict: 70 return {"id": f"evt_{kind}_{time.time_ns()}", "type": f"customer.subscription.{kind}", 71 "data": {"object": obj}} 72 73 74 def buy(client, monkeypatch, plan_id="subrep_year", email=None) -> str: 75 """Pay for Subrep Pro, as the checkout webhook tells it. Returns the account id.""" 76 acct = account_id(client) 77 sub = sub_object(acct, plan_id) 78 monkeypatch.setattr(stripe.Subscription, "retrieve", lambda sid: FakeStripeObject(sub)) 79 event = checkout_event(f"cs_{time.time_ns()}", acct, plan_id, 80 email=email or f"pro-{time.time_ns()}@example.com", 81 subscription=sub["id"]) 82 assert post_webhook(client, event).status_code == 200 83 return acct 84 85 86 def state(client) -> dict: 87 return client.get("/api/subrep").json() 88 89 90 def refresh(client, account: str, key: str): 91 return client.post("/api/subrep/refresh", json={"account": account, "refresh_key": key}) 92 93 94 def fake_checkout(monkeypatch) -> dict: 95 seen: dict = {} 96 monkeypatch.setattr( 97 stripe.checkout.Session, "create", 98 lambda **p: seen.update(p) or FakeStripeObject({"url": "https://stripe.test/pro"}), 99 ) 100 return seen 101 102 103 # --- plans and checkout ------------------------------------------------------ 104 105 def test_the_page_offers_the_yearly_plan_first_then_the_monthly_plan(client): 106 plans = state(client)["plans"] 107 assert [(p["id"], p["price_cents"], p["interval"]) for p in plans] == [ 108 ("subrep_year", 999, "year"), ("subrep_month", 99, "month")] 109 110 111 def test_the_book_page_does_not_show_subrep_pro(client): 112 ids = {p["id"] for p in client.get("/api/pricing").json()["plans"]} 113 assert not ids & {p.id for p in subrep.PLANS} 114 115 116 def test_a_plan_checks_out_as_a_subscription_of_its_interval(client, monkeypatch): 117 seen = fake_checkout(monkeypatch) 118 r = client.post("/api/subrep/checkout", json={"plan_id": "subrep_year"}) 119 assert r.status_code == 200 and r.json()["url"] == "https://stripe.test/pro" 120 assert seen["mode"] == "subscription" 121 price = seen["line_items"][0]["price_data"] 122 assert price["recurring"] == {"interval": "year"} and price["unit_amount"] == 999 123 assert seen["subscription_data"]["metadata"]["plan_id"] == "subrep_year" 124 125 126 def test_the_checkout_comes_back_to_the_subrep_page(client, monkeypatch): 127 seen = fake_checkout(monkeypatch) 128 client.post("/api/subrep/checkout", json={"plan_id": "subrep_month"}) 129 assert seen["cancel_url"] == "https://example.test/subrep.html?checkout=cancelled" 130 success = urlparse(seen["success_url"]) 131 assert success.path == "/api/billing/return" 132 assert parse_qs(success.query)["page"] == ["/subrep.html"] 133 134 135 def test_the_return_trip_goes_only_to_a_known_page(client, monkeypatch): 136 monkeypatch.setattr(stripe.checkout.Session, "retrieve", 137 lambda sid: FakeStripeObject({"id": sid, "payment_status": "unpaid"})) 138 r = client.get("/api/billing/return?session_id=cs_x&page=/subrep.html", 139 follow_redirects=False) 140 assert r.status_code == 303 and r.headers["location"] == "/subrep.html?checkout=pending" 141 r = client.get("/api/billing/return?session_id=cs_x&page=https://evil.test/", 142 follow_redirects=False) 143 assert r.headers["location"] == "/?checkout=pending" 144 145 146 def test_a_book_plan_does_not_stop_a_subrep_checkout(client, monkeypatch): 147 acct = account_id(client) 148 books = sub_object(acct, "month10", sub_id=f"sub_books_{acct}") 149 assert post_webhook(client, sub_event("created", books)).status_code == 200 150 assert client.get("/api/account").json()["subscribed"] is True 151 fake_checkout(monkeypatch) 152 assert client.post("/api/subrep/checkout", json={"plan_id": "subrep_month"}).status_code == 200 153 154 155 def test_a_second_subrep_plan_is_refused(client, monkeypatch): 156 buy(client, monkeypatch) 157 fake_checkout(monkeypatch) 158 r = client.post("/api/subrep/checkout", json={"plan_id": "subrep_month"}) 159 assert r.status_code == 400 and "Manage plan" in r.json()["detail"] 160 161 162 def test_only_a_subrep_plan_is_sold_here(client): 163 assert client.post("/api/subrep/checkout", json={"plan_id": "month10"}).status_code == 404 164 165 166 def test_checkout_needs_a_verified_email_when_the_server_can_send_one(client, monkeypatch): 167 # A $0.99 checkout draws card testing. With email sign-in, only a verified 168 # account may pay. Without it, the checks of Stripe are all there is. 169 monkeypatch.setattr(settings, "smtp_host", "smtp.example.test") 170 fake_checkout(monkeypatch) 171 assert state(client)["needs_sign_in"] is True 172 assert client.post("/api/subrep/checkout", json={"plan_id": "subrep_year"}).status_code == 403 173 with SessionLocal() as s: 174 row = s.get(Account, account_id(client)) 175 row.email, row.email_verified = f"v-{time.time_ns()}@example.com", 1 176 s.commit() 177 assert client.post("/api/subrep/checkout", json={"plan_id": "subrep_year"}).status_code == 200 178 179 180 def test_a_server_without_a_licence_key_sells_nothing(client, monkeypatch): 181 monkeypatch.setattr(settings, "subrep_license_key", "") 182 assert state(client)["available"] is False 183 assert client.post("/api/subrep/checkout", json={"plan_id": "subrep_year"}).status_code == 503 184 185 186 # --- Pro, the link command and the licence ------------------------------------- 187 188 def test_paying_makes_the_account_pro_and_gives_a_link_command(client, monkeypatch): 189 acct = buy(client, monkeypatch) 190 s = state(client) 191 assert s["pro"] is True and s["plan_id"] == "subrep_year" 192 link = s["link"] 193 assert link["account"] == acct and link["server"] == "https://example.test/api/subrep" 194 assert link["command"] == ( 195 f'subrep license --link "{acct}" "{link["refresh_key"]}" "{link["server"]}"') 196 197 198 def test_the_page_shows_the_same_link_command_on_each_visit(client, monkeypatch): 199 # Each computer of the customer runs the same command. 200 buy(client, monkeypatch) 201 assert state(client)["link"] == state(client)["link"] 202 203 204 def test_a_subrep_plan_does_not_touch_the_book_plan(client, monkeypatch): 205 buy(client, monkeypatch) 206 a = client.get("/api/account").json() 207 assert a["subscribed"] is False and a["plan_credits"] == 0 208 209 210 def test_the_desktop_app_gets_a_licence_that_the_public_key_verifies(client, monkeypatch): 211 email = f"desk-{time.time_ns()}@example.com" 212 acct = buy(client, monkeypatch, email=email) 213 r = refresh(client, acct, state(client)["link"]["refresh_key"]) 214 assert r.status_code == 200 215 claims = verify(r.json()["token"]) 216 assert claims["tier"] == "pro" and claims["sub"] == email 217 assert time.time() < claims["exp"] <= time.time() + 31 * DAY 218 219 220 def test_a_licence_ends_with_the_paid_period(client, monkeypatch): 221 acct = buy(client, monkeypatch, plan_id="subrep_month") 222 end = int(time.time()) + 3 * DAY 223 post_webhook(client, sub_event("updated", sub_object(acct, "subrep_month", period_end=end))) 224 token = refresh(client, acct, state(client)["link"]["refresh_key"]).json()["token"] 225 assert verify(token)["exp"] == end 226 227 228 def test_a_yearly_licence_ends_within_31_days(client, monkeypatch): 229 # The app asks again before then, so a refund ends Pro within a month. 230 acct = buy(client, monkeypatch) 231 post_webhook(client, sub_event("updated", sub_object( 232 acct, "subrep_year", period_end=int(time.time()) + 365 * DAY))) 233 token = refresh(client, acct, state(client)["link"]["refresh_key"]).json()["token"] 234 assert verify(token)["exp"] <= time.time() + 31 * DAY 235 236 237 def test_a_wrong_refresh_key_gets_no_licence(client, monkeypatch): 238 acct = buy(client, monkeypatch) 239 key = state(client)["link"]["refresh_key"] # the account has a key now 240 r = refresh(client, acct, key[:-1] + ("A" if key[-1] != "A" else "B")) 241 assert r.status_code == 403 and "error" in r.json() 242 assert refresh(client, acct, key).status_code == 200 243 244 245 def test_an_account_without_a_link_command_gets_no_licence(client, monkeypatch): 246 acct = buy(client, monkeypatch) # the page was never opened: no key yet 247 assert refresh(client, acct, "").status_code == 403 248 249 250 def test_the_control_panel_links_with_the_email_and_the_setup_key(client, monkeypatch): 251 # The Licence card of the desktop control panel asks for the email that 252 # paid and the setup key (the refresh key). The email names the account. 253 email = f"Panel-{time.time_ns()}@Example.com" 254 buy(client, monkeypatch, email=email) 255 link = state(client)["link"] 256 assert link["email"] == email.lower() 257 r = refresh(client, email, link["refresh_key"]) 258 assert r.status_code == 200 and verify(r.json()["token"])["sub"] == email.lower() 259 260 261 def test_another_email_gets_no_licence(client, monkeypatch): 262 buy(client, monkeypatch) 263 key = state(client)["link"]["refresh_key"] 264 assert refresh(client, f"other-{time.time_ns()}@example.com", key).status_code == 403 265 266 267 def test_a_licence_says_when_the_paid_period_ends(client, monkeypatch): 268 # The desktop app shows "paid until" from the "paid" claim. The token 269 # itself ends after TOKEN_DAYS at most. 270 acct = buy(client, monkeypatch) 271 end = int(time.time()) + 365 * DAY 272 post_webhook(client, sub_event("updated", sub_object(acct, "subrep_year", period_end=end))) 273 claims = verify(refresh(client, acct, state(client)["link"]["refresh_key"]).json()["token"]) 274 assert claims["paid"] == end and claims["exp"] < end 275 276 277 def test_a_cancelled_plan_gets_no_licence(client, monkeypatch): 278 acct = buy(client, monkeypatch, plan_id="subrep_month") 279 key = state(client)["link"]["refresh_key"] 280 ended = sub_object(acct, "subrep_month", status="canceled") 281 assert post_webhook(client, sub_event("deleted", ended)).status_code == 200 282 s = state(client) 283 assert s["pro"] is False and s["link"] is None 284 r = refresh(client, acct, key) 285 assert r.status_code == 402 and "error" in r.json() 286 287 288 def test_a_switch_in_the_portal_changes_the_plan(client, monkeypatch): 289 acct = buy(client, monkeypatch, plan_id="subrep_month") 290 post_webhook(client, sub_event("updated", sub_object( 291 acct, "subrep_month", price_key="subrep_year"))) 292 assert state(client)["plan_id"] == "subrep_year" 293 294 295 def test_the_end_of_an_old_subscription_does_not_end_its_replacement(client, monkeypatch): 296 acct = buy(client, monkeypatch) 297 old = sub_object(acct, "subrep_month", status="canceled", sub_id="sub_old") 298 post_webhook(client, sub_event("deleted", old)) 299 assert state(client)["pro"] is True 300 301 302 def test_merging_accounts_keeps_the_licence(client, monkeypatch): 303 acct = buy(client, monkeypatch) 304 with SessionLocal() as s: 305 owner = accounts.new_account(s) 306 accounts.merge(s, s.get(Account, acct), owner) 307 s.commit() 308 assert subrep.is_pro(subrep.licence(s, owner.id)) 309 assert subrep.licence(s, acct) is None 310 311 312 def test_the_link_command_of_a_merged_account_still_works(client, monkeypatch): 313 acct = buy(client, monkeypatch) 314 key = state(client)["link"]["refresh_key"] 315 with SessionLocal() as s: 316 owner = accounts.new_account(s) 317 accounts.merge(s, s.get(Account, acct), owner) 318 s.commit() 319 assert refresh(client, acct, key).status_code == 200 320 321 322 # --- the iOS waitlist ------------------------------------------------------------ 323 324 def test_the_ios_waitlist_keeps_one_entry_for_each_email(client): 325 email = f"Wait-{time.time_ns()}@Example.com" 326 for _ in range(2): 327 r = client.post("/api/subrep/waitlist", json={"email": email}) 328 assert r.status_code == 200 and r.json()["joined"] is True 329 with SessionLocal() as s: 330 rows = s.query(subrep.WaitlistEntry).filter_by(email=email.lower()).all() 331 assert [row.kind for row in rows] == ["ios"] 332 assert subrep.waitlist_count(s, "ios") >= 1 333 334 335 def test_the_waitlist_refuses_what_is_not_an_email(client): 336 assert client.post("/api/subrep/waitlist", json={"email": "nope"}).status_code == 400 337 338 339 def test_there_is_no_waitlist_for_an_unknown_app(client): 340 r = client.post("/api/subrep/waitlist", json={"email": "a@example.com", "kind": "tv"}) 341 assert r.status_code == 404 342 343 344 # --- the token format ------------------------------------------------------------ 345 346 claims_st = st.dictionaries( 347 st.text(min_size=1, max_size=12), 348 st.one_of(st.integers(), st.text(max_size=20), st.booleans()), 349 max_size=6, 350 ) 351 352 353 @hsettings(max_examples=60, deadline=None) 354 @given(claims=claims_st) 355 def test_a_token_carries_its_claims_and_its_signature_checks(claims): 356 assert verify(subrep.sign(claims)) == claims 357 358 359 @hsettings(max_examples=60, deadline=None) 360 @given(claims=claims_st, where=st.integers(min_value=0)) 361 def test_a_changed_token_fails_the_check(claims, where): 362 payload, sig = subrep.sign(claims).split(".") 363 raw = bytearray(b64d(payload)) 364 raw[where % len(raw)] ^= 1 365 changed = base64.urlsafe_b64encode(bytes(raw)).rstrip(b"=").decode() + "." + sig 366 with pytest.raises(InvalidSignature): 367 verify(changed) 368 369 370 def test_the_page_shows_the_public_key_for_the_desktop_build(client): 371 assert b64d(state(client)["public_key"]) == PUBLIC_KEY.public_bytes_raw() 372 373 374 def test_the_key_tool_makes_one_key_and_never_replaces_it(tmp_path, monkeypatch, capsys): 375 import importlib.util 376 from pathlib import Path 377 378 spec = importlib.util.spec_from_file_location( 379 "subrep_key", Path(__file__).resolve().parent.parent / "tools" / "subrep_key.py") 380 tool = importlib.util.module_from_spec(spec) 381 spec.loader.exec_module(tool) 382 env = tmp_path / ".env" 383 env.write_text("SUBPLZ_WEB_OTHER=1", encoding="utf-8") # no newline at the end 384 monkeypatch.setattr(tool, "ROOT", tmp_path) 385 monkeypatch.setattr(tool, "ENV", env) 386 387 assert tool.main() == 0 388 public = capsys.readouterr().out.strip() 389 text = env.read_text(encoding="utf-8") 390 assert text.startswith("SUBPLZ_WEB_OTHER=1\nSUBPLZ_WEB_SUBREP_LICENSE_KEY=") 391 saved = text.split("SUBPLZ_WEB_SUBREP_LICENSE_KEY=")[1].strip() 392 monkeypatch.setattr(settings, "subrep_license_key", saved) 393 assert subrep.public_key() == public # the printed key belongs to the saved key 394 395 assert tool.main() == 0 396 assert capsys.readouterr().out.strip() == public 397 assert env.read_text(encoding="utf-8") == text 398 399 400 def test_the_page_gives_a_buyer_the_windows_installer(client): 401 # A buyer lands on the Pro part of the page after paying, and needs the 402 # app there, with the setup key. 403 page = client.get("/subrep.html").text 404 pro = page[page.index('id="pro"'):page.index('id="ios"')] 405 assert 'href="https://honjimaku.com/subrep/SubrepSetup.exe"' in pro 406 assert 'id="link-key"' in pro