|
|
@@ -219,6 +219,15 @@ def run(ctx):
|
|
|
check(block_hit["headers"].get("ttl") == "60"
|
|
|
and block_hit["headers"].get("content-type") == "application/octet-stream",
|
|
|
"the push POST carries TTL and the encrypted content type")
|
|
|
+ # RFC 8291 section 4: the aes128gcm body MUST announce its encoding
|
|
|
+ # — push services answer HTTP 400 without it (the test-notification
|
|
|
+ # 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"
|
|
|
+ and block_hit["headers"].get("authorization", "")
|
|
|
+ .startswith("vapid t="),
|
|
|
+ "the push POST carries 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
|
|
|
@@ -449,12 +458,17 @@ def run(ctx):
|
|
|
and len(push2.hits("/b")) > b_before, timeout=15,
|
|
|
what="test notification POSTs") is not None,
|
|
|
"push_test fires a REAL push to both of the user's endpoints")
|
|
|
- _, test_payload = decrypt_webpush(push2.hits("/a")[-1], browser_a,
|
|
|
+ test_hit = push2.hits("/a")[-1]
|
|
|
+ _, test_payload = decrypt_webpush(test_hit, browser_a,
|
|
|
browser_a_public, browser_a_auth)
|
|
|
check(test_payload["title"] == "Test notification"
|
|
|
and test_payload["url"] == "/",
|
|
|
"the test payload decrypts with title + url (goes to the dashboard)"
|
|
|
" (%s)" % test_payload)
|
|
|
+ check(test_hit["headers"].get("content-encoding") == "aes128gcm"
|
|
|
+ and test_hit["headers"].get("ttl") == "60",
|
|
|
+ "the test-notification POST carries Content-Encoding: aes128gcm"
|
|
|
+ " + TTL (the HTTP 400 fix)")
|
|
|
check("2 browser" in message,
|
|
|
"the action response reports both deliveries (%s)" % message)
|
|
|
set_types = [d.get("type") for d in result if d.get("type") == "set"]
|