|
|
@@ -1,15 +1,17 @@
|
|
|
-"""The push bundle (moved verbatim from the pre-split web_stack_test.py;
|
|
|
-see test/harness.py for the shared machinery and the runner)."""
|
|
|
+"""The push bundle (goal 47: rewritten for Spry.Push semantics; see
|
|
|
+test/harness.py for the shared machinery and the runner)."""
|
|
|
from harness import * # noqa: F401,F403 — the shared wire helpers
|
|
|
|
|
|
|
|
|
# =============================================================================
|
|
|
-# Bundle: push — plan 11 task E: real Web Push over VAPID. VAPID
|
|
|
-# persistence, the service worker, subscription CRUD + validation,
|
|
|
-# notify-me prefs, the block/lifecycle dispatch matrix (global vs
|
|
|
-# opted-in), 410 pruning, retry-next-tick delivery and protocol-level
|
|
|
-# payload verification (RFC 8291 decryption + ES256 JWT) against the
|
|
|
-# fake push service.
|
|
|
+# Bundle: push — real Web Push over VAPID, Spry.Push underneath. The store
|
|
|
+# (spry_push_subscriptions), the RFC 8291/8292 crypto and the delivery POST
|
|
|
+# are Spry's (inline 2-attempt/250ms retry, 401/403/404/410 sweeping); the
|
|
|
+# payloads, audiences, notify-me prefs and the bell UX stay the app's.
|
|
|
+# Subscription CRUD + validation, the block/lifecycle dispatch matrix
|
|
|
+# (global vs opted-in), sweeping, inline retry, the send_test probe and
|
|
|
+# protocol-level payload verification (RFC 8291 decryption + ES256 JWT)
|
|
|
+# against the fake push service.
|
|
|
# =============================================================================
|
|
|
|
|
|
def run(ctx):
|
|
|
@@ -19,7 +21,7 @@ def run(ctx):
|
|
|
conn = db()
|
|
|
try:
|
|
|
return conn.execute(
|
|
|
- "select count(*) from push_subscriptions where endpoint = ?",
|
|
|
+ "select count(*) from spry_push_subscriptions where endpoint = ?",
|
|
|
(endpoint,)).fetchone()[0]
|
|
|
finally:
|
|
|
conn.close()
|
|
|
@@ -30,20 +32,24 @@ def run(ctx):
|
|
|
and "notificationclick" in sw and "clients.openWindow" in sw,
|
|
|
"the service worker serves at the origin root with push + click routing")
|
|
|
|
|
|
- print("== task E: VAPID keypair generation + persistence ==")
|
|
|
+ print("== goal 47: VAPID keys from the \"spry\" config section ==")
|
|
|
+ # Spry owns the keys: `spry keys` writes the "spry" section the app
|
|
|
+ # reads — the app no longer generates or persists key material.
|
|
|
config = json.load(open(os.path.join(WORK, "web-config.json")))
|
|
|
- vapid = config.get("vapid") or {}
|
|
|
- check(bool(vapid.get("public_key")) and bool(vapid.get("private_key")),
|
|
|
- "the VAPID keypair persists into web-config.json's vapid section")
|
|
|
+ spry = config.get("spry") or {}
|
|
|
+ check(bool(spry.get("vapid_public_key")) and bool(spry.get("vapid_private_key")),
|
|
|
+ "the VAPID keypair lives in web-config.json's spry section")
|
|
|
+ check("vapid" not in config,
|
|
|
+ "the dead boot-time vapid section is gone")
|
|
|
vapid_private_key = ec.derive_private_key(
|
|
|
- int.from_bytes(b64url_dec(vapid["private_key"]), "big"), ec.SECP256R1())
|
|
|
+ int.from_bytes(b64url_dec(spry["vapid_private_key"]), "big"), ec.SECP256R1())
|
|
|
derived_public = vapid_private_key.public_key().public_bytes(
|
|
|
Encoding.X962, PublicFormat.UncompressedPoint)
|
|
|
- check(b64url_dec(vapid["public_key"]) == derived_public,
|
|
|
- "the persisted public key matches the private scalar (a valid P-256 pair)")
|
|
|
+ check(b64url_dec(spry["vapid_public_key"]) == derived_public,
|
|
|
+ "the configured public key matches the private scalar (a valid P-256 pair)")
|
|
|
dash = slot_public(auth, "/", "dashboard")
|
|
|
- check(dash.get("push_vapid_key") == vapid["public_key"],
|
|
|
- "the dashboard exposes the same VAPID key to subscribers")
|
|
|
+ check(dash.get("push_vapid_key") == spry["vapid_public_key"],
|
|
|
+ "the dashboard exposes the spry section's VAPID key to subscribers")
|
|
|
check(dash.get("push_subscribed") is False and bool(dash.get("push_subscribe")),
|
|
|
"the bell starts unsubscribed and carries the subscribe action")
|
|
|
|
|
|
@@ -72,20 +78,24 @@ def run(ctx):
|
|
|
act(auth, dash["push_subscribe"],
|
|
|
{"endpoint": endpoint_main, **subscribe_fields})
|
|
|
conn = db()
|
|
|
- rows = conn.execute("select user_id, p256dh, auth from push_subscriptions"
|
|
|
+ rows = conn.execute("select user_id, p256dh, auth from spry_push_subscriptions"
|
|
|
" where endpoint = ?", (endpoint_main,)).fetchall()
|
|
|
admin_id = conn.execute("select id from spry_users where username = 'admin'"
|
|
|
).fetchone()[0]
|
|
|
+ stale = conn.execute("select count(*) from sqlite_master where name ="
|
|
|
+ " 'push_subscriptions'").fetchone()[0]
|
|
|
conn.close()
|
|
|
check(len(rows) == 1 and rows[0][0] == admin_id,
|
|
|
"a valid subscription stores against the signed-in user")
|
|
|
- # Re-subscribing the same endpoint replaces (never duplicates).
|
|
|
+ check(stale == 0, "the old push_subscriptions table is gone (fresh database)")
|
|
|
+ # Re-subscribing the same endpoint replaces (never duplicates) — the
|
|
|
+ # endpoint is Spry's subscription identity.
|
|
|
rotated_auth = os.urandom(16)
|
|
|
act(auth, dash["push_subscribe"],
|
|
|
{"endpoint": endpoint_main, "p256dh": subscribe_fields["p256dh"],
|
|
|
"auth": b64url(rotated_auth)})
|
|
|
conn = db()
|
|
|
- rows = conn.execute("select auth from push_subscriptions where endpoint = ?",
|
|
|
+ rows = conn.execute("select auth from spry_push_subscriptions where endpoint = ?",
|
|
|
(endpoint_main,)).fetchall()
|
|
|
conn.close()
|
|
|
check(len(rows) == 1 and rows[0][0] == b64url(rotated_auth),
|
|
|
@@ -125,8 +135,9 @@ def run(ctx):
|
|
|
other_auth = os.urandom(16)
|
|
|
endpoint_other = push.url("/other")
|
|
|
conn = db()
|
|
|
- conn.execute("insert into push_subscriptions (user_id, endpoint, p256dh, auth,"
|
|
|
- " created_at) values (999, ?, ?, ?, strftime('%s','now'))",
|
|
|
+ conn.execute("insert into spry_push_subscriptions (user_id, endpoint, p256dh,"
|
|
|
+ " auth, created) values (999, ?, ?, ?,"
|
|
|
+ " strftime('%s','now'))",
|
|
|
(endpoint_other, b64url(other_public), b64url(other_auth)))
|
|
|
conn.commit()
|
|
|
conn.close()
|
|
|
@@ -145,7 +156,6 @@ def run(ctx):
|
|
|
check(wait_for(lambda: fake.running_task == task1 or False, timeout=30,
|
|
|
what="task 1 assignment") is not None,
|
|
|
"task 1 dispatches to the fake daemon")
|
|
|
- task1_session = fake.session_id
|
|
|
check(wait_for(lambda: len(push.hits("/main")) > main_before, timeout=15,
|
|
|
what="first-start push to the opted-in admin") is not None,
|
|
|
"first start notifies the opted-in user's subscription")
|
|
|
@@ -166,14 +176,13 @@ def run(ctx):
|
|
|
check(len(push.hits("/other")) == other_before,
|
|
|
"completion stays with the opted-in user only")
|
|
|
check(wait_for(lambda: subscription_count(endpoint_gone) == 0, timeout=15,
|
|
|
- what="410 endpoint pruned") is not None,
|
|
|
- "a 410 Gone endpoint is pruned from push_subscriptions")
|
|
|
-
|
|
|
- # Block events go to EVERY push-subscribed user globally. The
|
|
|
- # poll-era /update raise is gone: a live (non-terminal) host task
|
|
|
- # with an open session on the fake carries the block through the
|
|
|
- # results push — task1 is done and terminal pushes are consumed
|
|
|
- # silently, so the host is its own fixture row.
|
|
|
+ what="410 endpoint swept") is not None,
|
|
|
+ "a 410 Gone endpoint is swept from spry_push_subscriptions")
|
|
|
+
|
|
|
+ # Block events go to EVERY push-subscribed user globally. A live
|
|
|
+ # (non-terminal) host task with an open session on the fake carries
|
|
|
+ # the block through the results push — task1 is done and terminal
|
|
|
+ # pushes are consumed silently, so the host is its own fixture row.
|
|
|
conn = db()
|
|
|
conn.execute("insert into tasks (project_id, title, status, priority,"
|
|
|
" created_at) values (1, 'Push block host', 'working', 5,"
|
|
|
@@ -205,13 +214,16 @@ def run(ctx):
|
|
|
and payload["url"] == "/tasks/%d" % push_host,
|
|
|
"the block payload carries task title + kind, question and click URL (%s)"
|
|
|
% payload)
|
|
|
- ciphertext_len = len(block_hit["body"]) - (21 + 65)
|
|
|
envelope = block_hit["body"]
|
|
|
record_size = int.from_bytes(envelope[16:20], "big")
|
|
|
- # rs = plaintext + delimiter (1) + tag (16) == ciphertext-with-tag len.
|
|
|
- check(record_size == ciphertext_len,
|
|
|
- "the RFC 8291 envelope record size covers payload + delimiter + tag")
|
|
|
- claims = verify_vapid(block_hit, vapid["public_key"], vapid["private_key"])
|
|
|
+ # Spry/Locksmith's aes128gcm envelope declares the fixed 4096 max
|
|
|
+ # record size (RFC 8291 §4.2) — the record itself is payload +
|
|
|
+ # delimiter + 16-byte tag, as long as the ciphertext.
|
|
|
+ ciphertext_len = len(envelope) - (21 + 65)
|
|
|
+ check(record_size == 4096 and ciphertext_len > 16,
|
|
|
+ "the RFC 8291 envelope declares Spry's fixed 4096 record size"
|
|
|
+ " over a well-formed record")
|
|
|
+ claims = verify_vapid(block_hit, spry["vapid_public_key"], spry["vapid_private_key"])
|
|
|
check(claims.get("aud") == "http://127.0.0.1:%d" % push.port
|
|
|
and claims.get("sub", "").startswith("mailto:")
|
|
|
and claims.get("exp", 0) > time.time(),
|
|
|
@@ -224,10 +236,10 @@ def run(ctx):
|
|
|
# 400 regression).
|
|
|
check(block_hit["headers"].get("content-encoding") == "aes128gcm",
|
|
|
"the push POST carries Content-Encoding: aes128gcm (RFC 8291)")
|
|
|
- check(block_hit["headers"].get("urgency") == "high"
|
|
|
+ check(block_hit["headers"].get("urgency") == "normal"
|
|
|
and block_hit["headers"].get("authorization", "")
|
|
|
.startswith("vapid t="),
|
|
|
- "the push POST carries Urgency and the VAPID Authorization")
|
|
|
+ "the push POST carries Spry's Urgency and the VAPID Authorization")
|
|
|
|
|
|
# Task 2: only the OTHER user opted in — start + failure reach them
|
|
|
# alone. Baseline again before the enqueue (first start fires at
|
|
|
@@ -251,7 +263,7 @@ def run(ctx):
|
|
|
or False, timeout=30, what="task 2 failed") is not None,
|
|
|
"task 2 fails")
|
|
|
check(wait_for(lambda: len(push.hits("/other")) >= other_before2 + 2,
|
|
|
- timeout=15, what="opted-in user's started+failed pushes"),
|
|
|
+ timeout=15, what="opted-in user's started+failed pushes") is not None,
|
|
|
"the opted-in user gets first start AND failure for the task")
|
|
|
started_hit = [hit for hit in push.hits("/other")[other_before2:]][0]
|
|
|
_, started_payload = decrypt_webpush(started_hit, other_browser,
|
|
|
@@ -266,7 +278,10 @@ def run(ctx):
|
|
|
check(len(push.hits("/main")) == main_before2,
|
|
|
"an agent failure never reaches users without the pref")
|
|
|
|
|
|
- print("== task E: retry next tick (best-effort delivery) ==")
|
|
|
+ print("== goal 47: inline retry (Spry's bounded delivery) ==")
|
|
|
+ # Spry's sender retries INLINE: two attempts, 250ms apart, then
|
|
|
+ # classifies. One transient 503 therefore lands on the second
|
|
|
+ # attempt moments later — no dispatcher tick, no queue row.
|
|
|
endpoint_flaky = push.url("/flaky")
|
|
|
act(auth, dash["push_subscribe"],
|
|
|
{"endpoint": endpoint_flaky, **subscribe_fields})
|
|
|
@@ -275,12 +290,44 @@ def run(ctx):
|
|
|
raise_global_block("Flaky delivery question?")
|
|
|
check(wait_for(lambda: len(push.hits("/flaky")) >= flaky_before + 2,
|
|
|
timeout=20, what="flaky endpoint retried") is not None,
|
|
|
- "a failed send retries on the next dispatcher tick and lands")
|
|
|
+ "a 503 send retries inline (Spry: 2 attempts) and lands")
|
|
|
+ flaky_hits = push.hits("/flaky")
|
|
|
+ if len(flaky_hits) >= flaky_before + 2:
|
|
|
+ first, second = flaky_hits[-2], flaky_hits[-1]
|
|
|
+ gap = second["at"] - first["at"]
|
|
|
+ check(0.05 <= gap <= 5.0,
|
|
|
+ "the retry follows the first attempt closely (Spry: 250ms) (%.2fs)"
|
|
|
+ % gap)
|
|
|
+ _, flaky_payload = decrypt_webpush(second, browser, browser_public,
|
|
|
+ browser_auth)
|
|
|
+ check(flaky_payload["body"] == "Flaky delivery question?",
|
|
|
+ "the retried delivery is a fresh, complete encryption (%s)"
|
|
|
+ % flaky_payload)
|
|
|
+ else:
|
|
|
+ check(False, "the inline retry fixture did not land")
|
|
|
+
|
|
|
+ print("== goal 47: a hard-failing endpoint is RETRYABLE, never swept ==")
|
|
|
+ # RFC 8030 §7 semantics (Spry): a persistent 503 exhausts the inline
|
|
|
+ # attempts and classifies RETRYABLE — the notification is NOT
|
|
|
+ # delivered, but the subscription row SURVIVES (only 401/403/404/410
|
|
|
+ # sweep). No give-up log line and no queue table anymore.
|
|
|
+ endpoint_dead = push.url("/dead")
|
|
|
+ push.status_for["/dead"] = 503
|
|
|
+ act(auth, dash["push_subscribe"],
|
|
|
+ {"endpoint": endpoint_dead, **subscribe_fields})
|
|
|
+ dead_before = len(push.hits("/dead"))
|
|
|
+ raise_global_block("Dead endpoint question?")
|
|
|
+ check(wait_for(lambda: len(push.hits("/dead")) >= dead_before + 2,
|
|
|
+ timeout=20, what="dead endpoint attempts") is not None,
|
|
|
+ "a 503 endpoint is POSTed twice (both inline attempts)")
|
|
|
+ check(len(push.hits("/dead")) <= dead_before + 2,
|
|
|
+ "exactly two attempts — Spry's ATTEMPTS bound, no dispatcher-tick retries")
|
|
|
+ check(subscription_count(endpoint_dead) == 1,
|
|
|
+ "a retryable endpoint is NOT swept (the row survives for the next send)")
|
|
|
|
|
|
print("== plan 12: goal notifications — blocks/success/failure always ==")
|
|
|
# Goal blocks, completion approval and failure push to EVERY
|
|
|
# subscribed user (the mirror on the goal-run task stays quiet).
|
|
|
- push_goal = None
|
|
|
act(auth, slot_public(auth, "/goals", "goals")["create"], {
|
|
|
"project_id": "1", "title": "Push goal", "brief": "Notifies everyone.",
|
|
|
"branch_name": "goal/push"})
|
|
|
@@ -401,9 +448,10 @@ def run(ctx):
|
|
|
return hashlib.sha256(endpoint.encode()).hexdigest()[:16]
|
|
|
|
|
|
print("== per-browser bell: fingerprint state shape ==")
|
|
|
- # The retry fixture left admin subscribed to /flaky; drop it so
|
|
|
- # this section owns exactly the subscriptions it asserts on.
|
|
|
- act(auth, dash["push_unsubscribe"], {"endpoint": endpoint_flaky})
|
|
|
+ # This section owns exactly the subscriptions it asserts on: drop
|
|
|
+ # the dispatch fixtures left on admin's account first.
|
|
|
+ for fixture_endpoint in (endpoint_flaky, endpoint_dead):
|
|
|
+ act(auth, dash["push_unsubscribe"], {"endpoint": fixture_endpoint})
|
|
|
# Fresh admin subscriptions on TWO fake endpoints = two browsers.
|
|
|
push2 = FakePushService()
|
|
|
push2.serve()
|
|
|
@@ -474,50 +522,56 @@ def run(ctx):
|
|
|
set_types = [d.get("type") for d in result if d.get("type") == "set"]
|
|
|
check(len(set_types) >= 1, "the action also refreshes the dashboard slot")
|
|
|
|
|
|
- print("== test notification: failure surfaces the real reason ==")
|
|
|
+ print("== test notification: a dead endpoint prunes + reports ==")
|
|
|
+ # Spry sweeps 401/403 exactly like 404/410 (RFC 8030: invalid
|
|
|
+ # credentials for THAT subscription can never heal) — the probe's
|
|
|
+ # DeliveryOutcome[] maps that to `pruned` (NOT failed: nothing
|
|
|
+ # retryable happened), the row goes away and the toast still
|
|
|
+ # separates the two outcomes.
|
|
|
push2.status_for["/b"] = 401
|
|
|
b_before = len(push2.hits("/b"))
|
|
|
result = act(auth, dash["push_test"], {})
|
|
|
message = action_notify(result) or ""
|
|
|
- check("reached 1 browser" in message and "HTTP 401" in message,
|
|
|
- "a refused endpoint reports reached/failed + the status (%s)" % message)
|
|
|
check(len(push2.hits("/b")) > b_before, "the refused endpoint was still POSTed")
|
|
|
+ check("1 browser" in message and "0 failed" in message,
|
|
|
+ "a refused endpoint reports the delivery + the sweep as not-failed (%s)"
|
|
|
+ % message)
|
|
|
+ check(subscription_count(endpoint_b) == 0,
|
|
|
+ "a 401 endpoint's row is swept (Spry: UNAUTHORIZED sweeps like GONE)")
|
|
|
+ check(wait_for(lambda: len(fingerprints_of(auth)) == 1, timeout=15,
|
|
|
+ what="bell state drops the swept endpoint") is not None,
|
|
|
+ "the refreshed bell state drops the swept endpoint")
|
|
|
|
|
|
- print("== truthful delivery: 4xx is refused, retried, never delivered ==")
|
|
|
- # A refused (401) endpoint must NOT count as delivered; the queue
|
|
|
- # retries on later ticks and — after MAX_ATTEMPTS — gives up WITH a
|
|
|
- # warning instead of silently claiming success. The subscription
|
|
|
- # row survives (a VAPID key rotation may heal it).
|
|
|
- row_before = subscription_count(endpoint_b)
|
|
|
- b_before = len(push2.hits("/b"))
|
|
|
- raise_global_block("Refused delivery question?")
|
|
|
- check(wait_for(lambda: len(push2.hits("/b")) > b_before, timeout=15,
|
|
|
- what="first attempt at the refused endpoint") is not None,
|
|
|
- "the event still reaches the refused endpoint")
|
|
|
- def give_up_in_log():
|
|
|
- try:
|
|
|
- with open(os.path.join(WORK, "app.log")) as handle:
|
|
|
- return "gave up after %d attempts" % 5 in handle.read()
|
|
|
- except IOError:
|
|
|
- return False
|
|
|
- check(wait_for(give_up_in_log, timeout=30, what="give-up warning in app.log")
|
|
|
- is not None,
|
|
|
- "after MAX_ATTEMPTS the give-up warning lands in the log (never silent)")
|
|
|
- check(subscription_count(endpoint_b) == row_before,
|
|
|
- "a refused endpoint is NOT pruned (it may recover)")
|
|
|
- check(len(fingerprints_of(auth)) == 2,
|
|
|
- "the bell state still lists both endpoints after give-up")
|
|
|
-
|
|
|
- print("== truthful delivery: 404/410 pruning regression ==")
|
|
|
- push2.status_for["/b"] = 410
|
|
|
- b_before = len(push2.hits("/b"))
|
|
|
+ print("== truthful delivery: retryable failures never count as delivered ==")
|
|
|
+ # A persistently 503ing endpoint stays RETRYABLE in Spry's model:
|
|
|
+ # both inline attempts fail, the row SURVIVES and the send_test
|
|
|
+ # probe surfaces the honest `failed` count in its toast.
|
|
|
+ endpoint_bad = push2.url("/bad")
|
|
|
+ push2.status_for["/bad"] = 503
|
|
|
+ act(auth, dash["push_subscribe"],
|
|
|
+ {"endpoint": endpoint_bad, "p256dh": b64url(browser_b_public),
|
|
|
+ "auth": b64url(browser_b_auth)})
|
|
|
+ bad_before = len(push2.hits("/bad"))
|
|
|
+ result = act(auth, dash["push_test"], {})
|
|
|
+ message = action_notify(result) or ""
|
|
|
+ check(len(push2.hits("/bad")) == bad_before + 2,
|
|
|
+ "the retryable endpoint got exactly two inline attempts (%d)"
|
|
|
+ % (len(push2.hits("/bad")) - bad_before))
|
|
|
+ check("1 browser" in message and "failed" in message,
|
|
|
+ "the probe reports the delivery that landed + the failure (%s)" % message)
|
|
|
+ check(subscription_count(endpoint_bad) == 1,
|
|
|
+ "a retryable endpoint is NOT swept — it may recover")
|
|
|
+
|
|
|
+ print("== truthful delivery: 410 pruning regression ==")
|
|
|
+ push2.status_for["/bad"] = 410
|
|
|
+ bad_before = len(push2.hits("/bad"))
|
|
|
raise_global_block("Prune after refusal question?")
|
|
|
- check(wait_for(lambda: len(push2.hits("/b")) > b_before, timeout=15,
|
|
|
+ check(wait_for(lambda: len(push2.hits("/bad")) > bad_before, timeout=15,
|
|
|
what="POST to the now-gone endpoint") is not None,
|
|
|
"the event reaches the endpoint before its 410")
|
|
|
- check(wait_for(lambda: subscription_count(endpoint_b) == 0, timeout=15,
|
|
|
- what="410 endpoint pruned after refusal episode") is not None,
|
|
|
- "a 410 endpoint is pruned even after an earlier refusal episode")
|
|
|
+ check(wait_for(lambda: subscription_count(endpoint_bad) == 0, timeout=15,
|
|
|
+ what="410 endpoint swept after retryable episode") is not None,
|
|
|
+ "a 410 endpoint is swept even after an earlier retryable episode")
|
|
|
check(wait_for(lambda: len(fingerprints_of(auth)) == 1, timeout=15,
|
|
|
what="bell state drops the pruned endpoint") is not None,
|
|
|
"the refreshed bell state drops the pruned endpoint")
|