feat: notifications on Android while closed, with nothing to install

The phone fetches what is new every fifteen minutes with a poll secret
(POST /v1/push/poll) that reads notification lines and nothing else. When a
UnifiedPush distributor is already installed, the hub also pushes at once,
encrypted to the phone (RFC 8291); losing the distributor falls back to
fetching.

The hub now honours "disable all notifications" itself: create_notification
creates nothing for that account, as it already did for a muted group, so
neither switch lets anything reach a phone. The interface used to be the only
reader of the account-wide switch.

Push endpoints are member-supplied URLs: a send refuses non-public
addresses, connects to the address it checked, and follows no redirect.

Android build untested here (no SDK on this machine).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

1 files changed, 455 insertions, 0 deletions

diff --git a/packages/meshbay-hub/tests/test_push.py b/packages/meshbay-hub/tests/test_push.py
new file mode 100644
index 0000000..c3bca4f
--- /dev/null
+++ b/packages/meshbay-hub/tests/test_push.py
@@ -0,0 +1,455 @@
+"""
+Push to a phone: the encryption, the endpoint a member supplies, and — the
+point of the whole feature — that the two switches a person sees are honoured.
+
+Every flow test is two accounts, the one causing a notification and the one
+receiving it, because a one-member test proves a one-member property.
+"""
+
+import base64
+import hashlib
+import json
+
+import httpx
+import pytest
+from cryptography.hazmat.primitives import hashes, hmac, serialization
+from cryptography.hazmat.primitives.asymmetric import ec
+from cryptography.hazmat.primitives.ciphers.aead import AESGCM
+from membership import add_member
+from meshbay_hub import webpush
+from meshbay_hub.api import push
+from meshbay_hub.db.models import Notification, PushSubscription, User
+from sqlalchemy import select
+
+
+def _b64(s: str) -> bytes:
+    s = "".join(s.split())
+    return base64.urlsafe_b64decode(s + "=" * (-len(s) % 4))
+
+
+def _b64e(b: bytes) -> str:
+    return base64.urlsafe_b64encode(b).rstrip(b"=").decode()
+
+
+def _auth_key(password: str, username: str) -> str:
+    salt = hashlib.sha256(f"meshbay:auth:v1:{username}".encode()).digest()
+    return base64.b64encode(
+        hashlib.pbkdf2_hmac("sha512", password.encode(), salt, 600_000, 32)).decode()
+
+
+async def _user(client, username, password="a-long-enough-passphrase"):
+    await client.post("/v1/users/register", json={
+        "username": username, "email": f"{username}@example.com",
+        "auth_key": _auth_key(password, username)})
+    r = await client.post("/v1/users/login", json={
+        "username": username, "auth_key": _auth_key(password, username)})
+    return r.json()["access_token"]
+
+
+def _phone():
+    """A user agent's keys, as a distributor's connector would generate them."""
+    sk = ec.generate_private_key(ec.SECP256R1())
+    pk = sk.public_key().public_bytes(
+        serialization.Encoding.X962, serialization.PublicFormat.UncompressedPoint)
+    auth = b"0123456789abcdef"
+    return sk, _b64e(pk), _b64e(auth)
+
+
+def _hm(key, data):
+    h = hmac.HMAC(key, hashes.SHA256())
+    h.update(data)
+    return h.finalize()
+
+
+def _decrypt(body: bytes, ua_private, auth_secret: bytes) -> bytes:
+    """RFC 8291 from the receiving side, written from the RFC and not from webpush.py."""
+    salt, idlen = body[:16], body[20]
+    as_public = body[21:21 + idlen]
+    ua_public = ua_private.public_key().public_bytes(
+        serialization.Encoding.X962, serialization.PublicFormat.UncompressedPoint)
+    ecdh = ua_private.exchange(
+        ec.ECDH(), ec.EllipticCurvePublicKey.from_encoded_point(ec.SECP256R1(), as_public))
+    ikm = _hm(_hm(auth_secret, ecdh), b"WebPush: info\x00" + ua_public + as_public + b"\x01")
+    prk = _hm(salt, ikm)
+    cek = _hm(prk, b"Content-Encoding: aes128gcm\x00\x01")[:16]
+    nonce = _hm(prk, b"Content-Encoding: nonce\x00\x01")[:12]
+    plain = AESGCM(cek).decrypt(nonce, body[21 + idlen:], None)
+    assert plain.endswith(b"\x02")
+    return plain[:-1]
+
+
+@pytest.fixture
+def sent(monkeypatch):
+    """Every push the hub would have sent, instead of sending it."""
+    calls = []
+
+    async def fake_send(target, payload, *, ttl, client=None):
+        calls.append((target, payload, ttl))
+        return webpush.DELIVERED
+
+    monkeypatch.setattr(push, "_send", fake_send)
+    push._last_chat.clear()
+    return calls
+
+
+async def _subscribe(client, token, endpoint="https://push.example.net/up/abc"):
+    _, p256dh, auth = _phone()
+    r = await client.post("/v1/push/subscriptions",
+                          json={"endpoint": endpoint, "p256dh": p256dh, "auth": auth},
+                          headers={"Authorization": f"Bearer {token}"})
+    return r
+
+
+async def _two_in_a_group(client, db_session, name):
+    member = await _user(client, f"{name}_member")
+    owner = await _user(client, f"{name}_owner")
+    g = await client.post("/v1/groups", json={"name": name},
+                          headers={"Authorization": f"Bearer {owner}"})
+    gid = g.json()["group_id"]
+    await add_member(client, gid, f"{name}_member", {"Authorization": f"Bearer {owner}"})
+    uid = (await db_session.execute(
+        select(User.id).where(User.username == f"{name}_member"))).scalar_one()
+    return member, uid, gid
+
+
+# ── Encryption ───────────────────────────────────────────────────────────────
+
+def test_encryption_reproduces_the_rfc_8291_example():
+    as_private = ec.derive_private_key(
+        int.from_bytes(_b64("yfWPiYE-n46HLnH0KqZOF1fJJU3MYrct3AELtAQ-oRw"), "big"),
+        ec.SECP256R1())
+    out = webpush.encrypt(
+        _b64("V2hlbiBJIGdyb3cgdXAsIEkgd2FudCB0byBiZSBhIHdhdGVybWVsb24"),
+        _b64("BCVxsr7N_eNgVRqvHtD0zTZsEc6-VV-JvLexhqUzORcx"
+             "aOzi6-AYWXvTBHm4bjyPjs7Vd8pZGH6SRpkNtoIAiw4"),
+        _b64("BTBZMqHH6r4Tts7J_aSIgg"),
+        as_private=as_private, salt=_b64("DGv6ra1nlYgDCS1FRnbzlw"))
+    header = _b64("DGv6ra1nlYgDCS1FRnbzlwAAEABBBP4z9KsN6nGRTbVYI_c7VJSPQTBtkgcy27ml"
+                  "mlMoZIIgDll6e3vCYLocInmYWAmS6TlzAC8wEqKK6PBru3jl7A8")
+    ciphertext = _b64("8pfeW0KbunFT06SuDKoJH9Ql87S1QUrdirN6GcG7sFz1y1sqLgVi1VhjVkHsUoEs"
+                      "bI_0LpXMuGvnzQ")
+    assert out == header + ciphertext
+
+
+def test_a_fresh_encryption_opens_on_the_phone():
+    sk, p256dh, auth = _phone()
+    body = webpush.encrypt(b'{"kind":"x"}', _b64(p256dh), _b64(auth))
+    assert _decrypt(body, sk, _b64(auth)) == b'{"kind":"x"}'
+
+
+# ── The endpoint is a URL a member chose, and the hub fetches it ─────────────
+
+@pytest.mark.parametrize("endpoint", [
+    "http://push.example.net/up/abc",
+    "https://127.0.0.1/up",
+    "https://10.1.2.3/up",
+    "https://[::1]/up",
+    "https://169.254.169.254/latest",
+    "https://user:pw@push.example.net/up",
+    "ftp://push.example.net/up",
+])
+@pytest.mark.asyncio
+async def test_an_endpoint_the_hub_should_not_fetch_is_refused(client, endpoint):
+    token = await _user(client, "ssrf_shape")
+    r = await _subscribe(client, token, endpoint)
+    assert r.status_code == 422, (endpoint, r.text)
+
+
+@pytest.mark.asyncio
+async def test_a_name_resolving_to_a_private_address_is_never_sent_to():
+    _, p256dh, auth = _phone()
+    seen = []
+    mock = httpx.AsyncClient(transport=httpx.MockTransport(
+        lambda req: seen.append(req) or httpx.Response(201)))
+    result = await webpush.send(webpush.Target("https://localhost/up", p256dh, auth),
+                                {"v": 1}, ttl=60, client=mock)
+    assert result == webpush.FAILED and seen == []
+
+
+@pytest.mark.asyncio
+async def test_the_request_goes_to_the_address_that_was_checked(monkeypatch):
+    """The hostname is the TLS name and the Host header only: a second lookup
+    returning something else would never be made."""
+    sk, p256dh, auth = _phone()
+
+    async def resolved(host, port):
+        assert host == "push.example.net"
+        return "93.184.215.14"
+
+    monkeypatch.setattr(webpush, "_resolve_public", resolved)
+    seen = []
+    mock = httpx.AsyncClient(transport=httpx.MockTransport(
+        lambda req: seen.append(req) or httpx.Response(201)))
+    result = await webpush.send(
+        webpush.Target("https://push.example.net/up/abc?up=1", p256dh, auth),
+        {"title": "hello"}, ttl=60, client=mock)
+    assert result == webpush.DELIVERED
+    (req,) = seen
+    assert req.url.host == "93.184.215.14" and req.url.path == "/up/abc"
+    assert req.headers["host"] == "push.example.net"
+    assert req.extensions["sni_hostname"] == "push.example.net"
+    assert req.headers["content-encoding"] == "aes128gcm" and req.headers["ttl"] == "60"
+    assert json.loads(_decrypt(req.content, sk, _b64(auth))) == {"title": "hello"}
+
+
+@pytest.mark.asyncio
+async def test_bad_keys_are_refused(client):
+    token = await _user(client, "bad_keys_user")
+    r = await client.post("/v1/push/subscriptions", json={
+        "endpoint": "https://push.example.net/up/abc",
+        "p256dh": _b64e(b"\x04" + b"\x01" * 64), "auth": _b64e(b"0" * 16)},
+        headers={"Authorization": f"Bearer {token}"})
+    assert r.status_code == 422
+
+
+# ── Rows ─────────────────────────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_registering_again_updates_the_same_row(client):
+    token = await _user(client, "registers_again")
+    first = (await _subscribe(client, token)).json()["id"]
+    second = (await _subscribe(client, token)).json()["id"]
+    assert first == second
+
+
+@pytest.mark.asyncio
+async def test_an_account_holds_a_bounded_number_of_subscriptions(client):
+    token = await _user(client, "many_phones")
+    for i in range(push.MAX_SUBSCRIPTIONS):
+        r = await _subscribe(client, token, f"https://push.example.net/up/{i}")
+        assert r.status_code == 200, r.text
+    r = await _subscribe(client, token, "https://push.example.net/up/one-more")
+    assert r.status_code == 429
+
+
+@pytest.mark.asyncio
+async def test_only_the_owner_can_remove_a_subscription(client):
+    mine = await _user(client, "sub_owner")
+    other = await _user(client, "sub_other")
+    sid = (await _subscribe(client, mine)).json()["id"]
+    r = await client.delete(f"/v1/push/subscriptions/{sid}",
+                            headers={"Authorization": f"Bearer {other}"})
+    assert r.status_code == 404
+    r = await client.delete(f"/v1/push/subscriptions/{sid}",
+                            headers={"Authorization": f"Bearer {mine}"})
+    assert r.status_code == 200
+
+
+# ── What is pushed, and what is not ──────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_a_notification_reaches_the_phone(client, db_session, sent):
+    from meshbay_hub.api.notifications import create_notification
+
+    member, uid, gid = await _two_in_a_group(client, db_session, "pushed")
+    assert (await _subscribe(client, member)).status_code == 200
+    await create_notification(db_session, uid, "chat_message", "owner posted in pushed",
+                              link=f"#/group/{gid}", group_id=gid, aggregate=True)
+    await db_session.commit()
+    await push.drain()
+    (target, payload, ttl) = sent[0]
+    assert payload["kind"] == "chat_message" and payload["group_id"] == gid
+    assert payload["title"] == "owner posted in pushed" and ttl == webpush.TTL_CHAT
+
+
+@pytest.mark.asyncio
+async def test_a_muted_group_pushes_nothing(client, db_session, sent):
+    from meshbay_hub.api.notifications import create_notification
+
+    member, uid, gid = await _two_in_a_group(client, db_session, "hushed")
+    await _subscribe(client, member)
+    r = await client.post(f"/v1/groups/{gid}/mute", json={"muted": True},
+                          headers={"Authorization": f"Bearer {member}"})
+    assert r.status_code == 200
+    await create_notification(db_session, uid, "chat_message", "owner posted in hushed",
+                              group_id=gid, aggregate=True)
+    await db_session.commit()
+    await push.drain()
+    assert sent == []
+
+
+@pytest.mark.asyncio
+async def test_every_notification_turned_off_pushes_nothing_and_stores_nothing(
+        client, db_session, sent):
+    """The account-wide switch was read by the interface only: rows went on
+    being created and hidden. With a phone told about each row, that would have
+    been a switch that did nothing."""
+    from meshbay_hub.api.notifications import create_notification
+
+    member, uid, gid = await _two_in_a_group(client, db_session, "silenced")
+    await _subscribe(client, member)
+    r = await client.put("/v1/users/me/preferences/notifications_disabled",
+                         json={"value": "true"},
+                         headers={"Authorization": f"Bearer {member}"})
+    assert r.status_code == 200
+    for kind, group in (("chat_message", gid), ("group_invite", None)):
+        made = await create_notification(db_session, uid, kind, "something",
+                                         group_id=group, aggregate=group is not None)
+        assert made is None, kind
+    await db_session.commit()
+    await push.drain()
+    assert sent == []
+    rows = (await db_session.execute(
+        select(Notification).where(Notification.user_id == uid,
+                                   Notification.title == "something"))).scalars().all()
+    assert rows == []
+
+    await client.put("/v1/users/me/preferences/notifications_disabled",
+                     json={"value": "false"},
+                     headers={"Authorization": f"Bearer {member}"})
+    await create_notification(db_session, uid, "group_invite", "back on")
+    await db_session.commit()
+    await push.drain()
+    assert [p["title"] for _, p, _ in sent] == ["back on"]
+
+
+@pytest.mark.asyncio
+async def test_a_busy_conversation_reaches_a_phone_once_per_window(client, db_session, sent):
+    from meshbay_hub.api.notifications import create_notification
+
+    member, uid, gid = await _two_in_a_group(client, db_session, "chatty")
+    await _subscribe(client, member)
+    for i in range(5):
+        await create_notification(db_session, uid, "chat_message", f"line {i}",
+                                  group_id=gid, aggregate=True)
+    await create_notification(db_session, uid, "group_invite", "not chat")
+    await db_session.commit()
+    await push.drain()
+    assert [p["title"] for _, p, _ in sent] == ["line 0", "not chat"]
+
+
+@pytest.mark.asyncio
+async def test_a_registration_the_push_server_dropped_falls_back_to_fetching(
+        client, db_session, monkeypatch):
+    """The row loses its endpoint, not its existence: the phone keeps fetching."""
+    from meshbay_hub.api.notifications import create_notification
+
+    async def gone(target, payload, *, ttl, client=None):
+        return webpush.GONE
+
+    monkeypatch.setattr(push, "_send", gone)
+    push._last_chat.clear()
+    member, uid, gid = await _two_in_a_group(client, db_session, "dropped")
+    await _subscribe(client, member)
+    await create_notification(db_session, uid, "group_invite", "anyone there")
+    await db_session.commit()
+    await push.drain()
+    db_session.expire_all()
+    (row,) = (await db_session.execute(
+        select(PushSubscription).where(PushSubscription.user_id == uid))).scalars().all()
+    assert row.endpoint is None and row.p256dh is None and row.poll_hash is not None
+
+
+# ── Fetching, for a phone with no distributor ───────────────────────────────
+
+async def _poll(client, reg, since=None):
+    return await client.post("/v1/push/poll", json={
+        "id": reg["id"], "secret": reg["poll_secret"], "since": since or reg["now"]})
+
+
+@pytest.fixture
+def unthrottled(monkeypatch):
+    monkeypatch.setattr(push, "POLL_MIN_INTERVAL", 0.0)
+    push._last_poll.clear()
+
+
+@pytest.mark.asyncio
+async def test_a_phone_without_a_distributor_fetches_what_is_new(
+        client, db_session, sent, unthrottled):
+    from meshbay_hub.api.notifications import create_notification
+
+    member, uid, gid = await _two_in_a_group(client, db_session, "fetched")
+    r = await client.post("/v1/push/subscriptions", json={},
+                          headers={"Authorization": f"Bearer {member}"})
+    assert r.status_code == 200, r.text
+    reg = r.json()
+    # The invitation that made them a member predates the registration: not news.
+    assert (await _poll(client, reg)).json()["notifications"] == []
+
+    await create_notification(db_session, uid, "chat_message", "owner posted in fetched",
+                              link=f"#/group/{gid}", group_id=gid, aggregate=True)
+    await db_session.commit()
+    await push.drain()
+    assert sent == [], "a row with no endpoint is never pushed to"
+    (got,) = (await _poll(client, reg)).json()["notifications"]
+    assert got["kind"] == "chat_message" and got["group_id"] == gid
+    assert got["title"] == "owner posted in fetched"
+    assert (await _poll(client, reg, got["created_at"])).json()["notifications"] == []
+
+
+@pytest.mark.asyncio
+async def test_fetching_honours_both_switches(client, db_session, unthrottled):
+    from meshbay_hub.api.notifications import create_notification
+
+    member, uid, gid = await _two_in_a_group(client, db_session, "fetchmute")
+    reg = (await client.post("/v1/push/subscriptions", json={},
+                             headers={"Authorization": f"Bearer {member}"})).json()
+    await client.post(f"/v1/groups/{gid}/mute", json={"muted": True},
+                      headers={"Authorization": f"Bearer {member}"})
+    await create_notification(db_session, uid, "chat_message", "muted line",
+                              group_id=gid, aggregate=True)
+    await client.put("/v1/users/me/preferences/notifications_disabled",
+                     json={"value": "true"}, headers={"Authorization": f"Bearer {member}"})
+    await create_notification(db_session, uid, "group_invite", "all off")
+    await db_session.commit()
+    assert (await _poll(client, reg)).json()["notifications"] == []
+
+
+@pytest.mark.asyncio
+async def test_the_poll_secret_is_the_only_way_in(client, unthrottled):
+    mine = await _user(client, "poll_owner")
+    reg = (await client.post("/v1/push/subscriptions", json={},
+                             headers={"Authorization": f"Bearer {mine}"})).json()
+    assert (await _poll(client, {**reg, "poll_secret": "x" * 43})).status_code == 404
+    assert (await _poll(client, {**reg, "id": "0" * 36})).status_code == 404
+    again = (await client.post("/v1/push/subscriptions", json={"id": reg["id"]},
+                               headers={"Authorization": f"Bearer {mine}"})).json()
+    assert again["id"] == reg["id"]
+    assert (await _poll(client, reg)).status_code == 404, "a new secret retires the old"
+    assert (await _poll(client, again)).status_code == 200
+    await client.delete(f"/v1/push/subscriptions/{reg['id']}",
+                        headers={"Authorization": f"Bearer {mine}"})
+    assert (await _poll(client, again)).status_code == 404, "signed out: the row is gone"
+
+
+@pytest.mark.asyncio
+async def test_another_account_cannot_take_over_a_row(client):
+    mine = await _user(client, "row_owner")
+    other = await _user(client, "row_taker")
+    reg = (await client.post("/v1/push/subscriptions", json={},
+                             headers={"Authorization": f"Bearer {mine}"})).json()
+    theirs = (await client.post("/v1/push/subscriptions", json={"id": reg["id"]},
+                                headers={"Authorization": f"Bearer {other}"})).json()
+    assert theirs["id"] != reg["id"]
+
+
+@pytest.mark.asyncio
+async def test_a_phone_cannot_poll_faster_than_the_floor(client):
+    push._last_poll.clear()
+    mine = await _user(client, "eager_poller")
+    reg = (await client.post("/v1/push/subscriptions", json={},
+                             headers={"Authorization": f"Bearer {mine}"})).json()
+    assert (await _poll(client, reg)).status_code == 200
+    r = await _poll(client, reg)
+    assert r.status_code == 429 and int(r.headers["retry-after"]) > 0
+
+
+@pytest.mark.asyncio
+async def test_a_phone_gaining_a_distributor_keeps_its_row(client):
+    mine = await _user(client, "upgrading_phone")
+    reg = (await client.post("/v1/push/subscriptions", json={},
+                             headers={"Authorization": f"Bearer {mine}"})).json()
+    _, p256dh, auth = _phone()
+    r = await client.post("/v1/push/subscriptions", json={
+        "id": reg["id"], "endpoint": "https://push.example.net/up/new",
+        "p256dh": p256dh, "auth": auth}, headers={"Authorization": f"Bearer {mine}"})
+    assert r.json()["id"] == reg["id"]
+
+
+@pytest.mark.asyncio
+async def test_an_endpoint_without_its_keys_is_refused(client):
+    mine = await _user(client, "keyless_phone")
+    r = await client.post("/v1/push/subscriptions",
+                          json={"endpoint": "https://push.example.net/up/k"},
+                          headers={"Authorization": f"Bearer {mine}"})
+    assert r.status_code == 422