stroblmeandClaude Opus 5
d01a8dad37
Follows the portal: the noun is "instance" everywhere the app says it — UI strings, CLI output, error details, docs and comments. The wire keys (`instance_id`, `instance_token`) and the hub route this calls move with it. An existing cloud.json is adopted rather than refused: without the key alias the dataclass fails to parse, which the caller swallows and reads as "never enrolled" instead of "reconnect". `instance_key` on a node type becomes `target_key`. It means the outside thing a node points at, which is a different sense of the word, and keeping both would put two meanings of "instance" in one codebase. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_015YrQnKV3bnQd4K342y8tKj
143 lines
4.4 KiB
Python
143 lines
4.4 KiB
Python
"""Web push: the crypto is the part that cannot be checked by reading it."""
|
|
|
|
import asyncio
|
|
import os
|
|
|
|
import http_ece
|
|
import httpx
|
|
import jwt
|
|
import pytest
|
|
from cryptography.hazmat.primitives import serialization
|
|
from cryptography.hazmat.primitives.asymmetric import ec
|
|
|
|
from fluksio.core.config import settings
|
|
from fluksio.flow import webpush
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def store(tmp_path, monkeypatch):
|
|
"""A store per test, so nothing writes into the real instance."""
|
|
monkeypatch.setattr(settings, "WEBPUSH_FILE", tmp_path / "webpush.json")
|
|
return tmp_path / "webpush.json"
|
|
|
|
|
|
def browser(
|
|
name: str = "one",
|
|
) -> tuple[ec.EllipticCurvePrivateKey, bytes, webpush.Subscription]:
|
|
"""What a browser hands over when it subscribes."""
|
|
key = ec.generate_private_key(ec.SECP256R1())
|
|
auth = os.urandom(16)
|
|
point = key.public_key().public_bytes(
|
|
serialization.Encoding.X962, serialization.PublicFormat.UncompressedPoint
|
|
)
|
|
return (
|
|
key,
|
|
auth,
|
|
webpush.Subscription(
|
|
endpoint=f"https://push.example.com/{name}",
|
|
keys={"p256dh": webpush._b64url(point), "auth": webpush._b64url(auth)},
|
|
),
|
|
)
|
|
|
|
|
|
def test_only_the_subscribed_browser_can_read_the_payload():
|
|
"""RFC 8291, and the reason the push service is not trusted with content."""
|
|
key, auth, subscription = browser()
|
|
payload = b'{"title":"A run failed"}'
|
|
|
|
sealed = webpush._encrypt(payload, subscription)
|
|
|
|
assert sealed != payload
|
|
assert (
|
|
http_ece.decrypt(sealed, private_key=key, auth_secret=auth, version="aes128gcm")
|
|
== payload
|
|
)
|
|
|
|
|
|
def test_the_request_is_signed_for_the_push_service_that_gets_it():
|
|
"""RFC 8292: the audience is the endpoint's origin, never its path."""
|
|
key = ec.generate_private_key(ec.SECP256R1())
|
|
|
|
headers = webpush._vapid_headers(key, "https://push.example.com/quite/a/long/path")
|
|
|
|
scheme, rest = headers["Authorization"].split(" ", 1)
|
|
parts = dict(part.split("=", 1) for part in rest.split(","))
|
|
assert scheme == "vapid"
|
|
claims = jwt.decode(
|
|
parts["t"],
|
|
key.public_key(),
|
|
algorithms=["ES256"],
|
|
audience="https://push.example.com",
|
|
)
|
|
assert claims["sub"].startswith("mailto:")
|
|
# The service looks this key up against the one the browser subscribed with.
|
|
assert webpush._unb64url(parts["k"]) == webpush._public_bytes(key)
|
|
|
|
|
|
def test_the_keypair_survives_the_next_ask():
|
|
first = webpush.public_key()
|
|
assert webpush.public_key() == first
|
|
|
|
|
|
def push_to(handler) -> int:
|
|
"""Send one alert to two browsers, the push services answering `handler`."""
|
|
webpush.add_subscription(browser("gone")[2])
|
|
webpush.add_subscription(browser("here")[2])
|
|
|
|
transport = httpx.MockTransport(handler)
|
|
original = httpx.AsyncClient
|
|
|
|
def client(**kwargs):
|
|
kwargs["transport"] = transport
|
|
return original(**kwargs)
|
|
|
|
webpush.httpx.AsyncClient = client # type: ignore[misc]
|
|
try:
|
|
return asyncio.run(webpush.send_to_all({"title": "t", "body": "b"}))
|
|
finally:
|
|
webpush.httpx.AsyncClient = original # type: ignore[misc]
|
|
|
|
|
|
def test_every_subscribed_browser_is_pushed_to():
|
|
seen: list[str] = []
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
seen.append(str(request.url))
|
|
assert request.headers["content-encoding"] == "aes128gcm"
|
|
assert request.headers["authorization"].startswith("vapid t=")
|
|
return httpx.Response(201)
|
|
|
|
assert push_to(handler) == 2
|
|
assert len(seen) == 2
|
|
|
|
|
|
def test_a_retired_subscription_is_dropped():
|
|
"""410 is the only word we get that a browser is never coming back."""
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(410 if request.url.path.endswith("gone") else 201)
|
|
|
|
delivered = push_to(handler)
|
|
|
|
assert delivered == 1
|
|
assert [s.endpoint for s in webpush.subscriptions()] == [
|
|
"https://push.example.com/here"
|
|
]
|
|
|
|
|
|
def test_a_channel_with_nobody_subscribed_says_so():
|
|
"""What the test button on the alerts screen reports."""
|
|
from fluksio.flow.alerts import Alert, AlertManager, Channel
|
|
from fluksio.flow.events import EventBus
|
|
|
|
alerts = AlertManager(EventBus())
|
|
|
|
with pytest.raises(ValueError, match="No browsers"):
|
|
asyncio.run(
|
|
alerts.send(
|
|
Channel(name="phones", kind="webpush"),
|
|
Alert(title="t", body="b"),
|
|
raise_on_error=True,
|
|
)
|
|
)
|