From 6832df6177ad973ad0e1b4f0a49d7a6da06c6e04 Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Fri, 9 Oct 2026 12:08:31 +0200 Subject: 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 --- packages/meshbay-hub/src/meshbay_hub/api/push.py | 277 +++++++++++++++++++++++ 1 file changed, 277 insertions(+) create mode 100644 packages/meshbay-hub/src/meshbay_hub/api/push.py (limited to 'packages/meshbay-hub/src/meshbay_hub/api/push.py') diff --git a/packages/meshbay-hub/src/meshbay_hub/api/push.py b/packages/meshbay-hub/src/meshbay_hub/api/push.py new file mode 100644 index 0000000..0c4a411 --- /dev/null +++ b/packages/meshbay-hub/src/meshbay_hub/api/push.py @@ -0,0 +1,277 @@ +""" +Notifications on a phone — /v1/push/*: pushed when it can be, fetched when not. + +A phone registers once and gets a row here. **With a UnifiedPush distributor** +it gives an endpoint and a P-256 key, and every notification +`create_notification` lets through is sent there, encrypted to the phone +(`webpush.py`). **Without one** — nothing to install is the default — the row has +no endpoint, and the phone fetches what is new with `POST /v1/push/poll` every +quarter of an hour or so. Both are the same rows and the same payload, so a phone +can move between them (a distributor installed, removed, refusing) without the +hub caring which. + +**Nothing reaches a phone that was not created**: a muted group and an account +with every notification turned off stop at `create_notification`, before this +module is reached, so the two switches the person sees are the only two there are. + +The poll is authenticated by a secret issued with the row, not by a session. It +reads notification lines and nothing else, so a phone running in the background +holds no token that could do anything more — and a sign-out, which deletes the +row, ends it. + +Who pays (§13.5b): a member's chat costs every other member's phones a push. +That fan-out is already bounded where it starts — `chat_notify` is budgeted per +node — and here a conversation reaches each phone at most once per +`CHAT_COALESCE` seconds: the phone shows one line per group, so the pushes in +between would only have replaced it. An account holds `MAX_SUBSCRIPTIONS` rows +at most, because each is one outbound request per notification; a row is polled +at most once per `POLL_MIN_INTERVAL`. +""" + +import asyncio +import hashlib +import hmac +import logging +import math +import secrets +import time +from datetime import UTC, datetime + +from fastapi import APIRouter, Depends, HTTPException +from pydantic import BaseModel, Field +from sqlalchemy import func, select, update +from sqlalchemy.ext.asyncio import AsyncSession + +from meshbay_hub import webpush +from meshbay_hub.api.deps import require_user_scope +from meshbay_hub.db.engine import get_db +from meshbay_hub.db.models import Notification, PushSubscription, User + +log = logging.getLogger(__name__) + +router = APIRouter(prefix="/v1/push", tags=["push"]) + +MAX_SUBSCRIPTIONS = 10 +CHAT_COALESCE = 30.0 +_COALESCE_ENTRIES = 10_000 +POLL_MIN_INTERVAL = 60.0 +POLL_LIMIT = 20 + +# Strong references: asyncio holds a task weakly, and a collected one is a push +# that silently never went (CLAUDE.md, "a background task nobody holds"). +_tasks: set[asyncio.Task] = set() +_last_chat: dict[tuple[str, str], float] = {} +_last_poll: dict[str, float] = {} +# Replaced by the tests; the real one never raises. +_send = webpush.send + + +class SubscriptionIn(BaseModel): + # The row this phone already has, to update rather than add one: a phone + # with no endpoint has nothing else to be recognised by. + id: str | None = Field(default=None, max_length=36) + endpoint: str | None = Field(default=None, max_length=webpush.MAX_ENDPOINT) + # Lengths bounded before decoding: base64 decoding skips characters outside + # its alphabet, so an unbounded string could still decode to 65 bytes. + p256dh: str | None = Field(default=None, max_length=128) + auth: str | None = Field(default=None, max_length=32) + + +def _hash(secret: str) -> str: + return hashlib.sha256(secret.encode()).hexdigest() + + +def _payload(notif: Notification) -> dict: + """What a phone is told, pushed or fetched: the hub's own line, never a message.""" + return { + "v": 1, + "id": notif.id, + "kind": notif.kind, + "title": notif.title, + "link": notif.link, + "group_id": notif.group_id, + "created_at": _iso(notif.created_at), + } + + +def _iso(at: datetime) -> str: + # SQLite hands back naive datetimes; every one stored here is UTC. + return (at if at.tzinfo else at.replace(tzinfo=UTC)).isoformat() + + +@router.post("/subscriptions") +async def subscribe( + body: SubscriptionIn, + current_user: User = Depends(require_user_scope), + db: AsyncSession = Depends(get_db), +): + """ + Register this phone, or update its row: with an endpoint and keys when it + has a push distributor, without them when it will fetch instead. + + Answers the row's id, a fresh secret for `POST /v1/push/poll` (the previous + one stops working) and the hub's time, from which the phone counts what is + new — what was there before it registered is not news. + """ + if body.endpoint is not None: + if body.p256dh is None or body.auth is None: + raise HTTPException(status_code=422, detail="an endpoint needs its keys") + try: + webpush.check_endpoint(body.endpoint) + webpush.check_keys(body.p256dh, body.auth) + except ValueError as e: + raise HTTPException(status_code=422, detail=str(e)) from e + + sub = None + if body.id is not None: + sub = await db.get(PushSubscription, body.id) + if sub is not None and sub.user_id != current_user.id: + sub = None + if body.endpoint is not None: + same = (await db.execute( + select(PushSubscription).where( + PushSubscription.user_id == current_user.id, + PushSubscription.endpoint == body.endpoint))).scalar_one_or_none() + if sub is None: + sub = same + elif same is not None and same.id != sub.id: + # The endpoint moved to this row; the old one would only repeat it. + await db.delete(same) + await db.flush() + if sub is None: + held = (await db.execute( + select(func.count()).select_from(PushSubscription) + .where(PushSubscription.user_id == current_user.id))).scalar() or 0 + if held >= MAX_SUBSCRIPTIONS: + raise HTTPException(status_code=429, detail="Too many push subscriptions") + sub = PushSubscription(user_id=current_user.id) + db.add(sub) + secret = secrets.token_urlsafe(32) + sub.endpoint, sub.p256dh, sub.auth = body.endpoint, body.p256dh, body.auth + sub.poll_hash = _hash(secret) + await db.commit() + return {"id": sub.id, "poll_secret": secret, "now": datetime.now(UTC).isoformat()} + + +class PollIn(BaseModel): + id: str = Field(max_length=36) + secret: str = Field(max_length=64) + since: datetime + + +@router.post("/poll") +async def poll(body: PollIn, db: AsyncSession = Depends(get_db)): + """ + What is new for this phone since `since`: the same payloads a push carries, + oldest first, at most twenty. + + Authenticated by the row's secret rather than a session, so what a phone + keeps for running in the background reads notification lines and nothing + else. A wrong secret and an unknown row answer the same 404. + """ + sub = await db.get(PushSubscription, body.id) + if (sub is None or sub.poll_hash is None + or not hmac.compare_digest(sub.poll_hash, _hash(body.secret))): + raise HTTPException(status_code=404, detail="Subscription not found") + now = time.monotonic() + last = _last_poll.get(sub.id) + if last is not None and now - last < POLL_MIN_INTERVAL: + raise HTTPException(status_code=429, detail="Polled too often", + headers={"Retry-After": str(int(POLL_MIN_INTERVAL - (now - last)) + 1)}) + if len(_last_poll) >= _COALESCE_ENTRIES: + for k in [k for k, at in _last_poll.items() if now - at >= POLL_MIN_INTERVAL]: + del _last_poll[k] + _last_poll[sub.id] = now + + since = body.since if body.since.tzinfo else body.since.replace(tzinfo=UTC) + rows = (await db.execute( + select(Notification).where( + Notification.user_id == sub.user_id, + Notification.created_at > since.astimezone(UTC), + ).order_by(Notification.created_at.desc()).limit(POLL_LIMIT) + )).scalars().all() + return {"notifications": [_payload(n) for n in reversed(rows)]} + + +@router.delete("/subscriptions/{subscription_id}") +async def unsubscribe( + subscription_id: str, + current_user: User = Depends(require_user_scope), + db: AsyncSession = Depends(get_db), +): + """Stop telling one phone anything: turned off there, or signed out of.""" + sub = await db.get(PushSubscription, subscription_id) + if sub is None or sub.user_id != current_user.id: + raise HTTPException(status_code=404, detail="Subscription not found") + await db.delete(sub) + await db.commit() + return {"status": "ok"} + + +def _coalesced(sub_id: str, group_id: str, now: float) -> bool: + key = (sub_id, group_id) + if now - _last_chat.get(key, -math.inf) < CHAT_COALESCE: + return True + if len(_last_chat) >= _COALESCE_ENTRIES: + for k in [k for k, at in _last_chat.items() if now - at >= CHAT_COALESCE]: + del _last_chat[k] + _last_chat[key] = now + return False + + +async def push_notification(db: AsyncSession, notif: Notification) -> None: + """ + Send `notif` to the person's phones, off the caller's path. + + Called by `create_notification` once the row exists, inside the caller's + transaction: the subscriptions are read there, the requests leave in a task + of their own, so a slow push server delays nobody's request. + """ + subs = (await db.execute( + select(PushSubscription).where(PushSubscription.user_id == notif.user_id, + PushSubscription.endpoint.is_not(None)) + )).scalars().all() + if not subs: + return + payload = _payload(notif) + now = time.monotonic() + targets = [ + (s.id, webpush.Target(s.endpoint, s.p256dh, s.auth)) for s in subs + if not (notif.kind == "chat_message" and notif.group_id + and _coalesced(s.id, notif.group_id, now)) + ] + if not targets: + return + ttl = webpush.TTL_CHAT if notif.kind == "chat_message" else webpush.TTL_OTHER + task = asyncio.get_running_loop().create_task(_deliver(targets, payload, ttl)) + _tasks.add(task) + task.add_done_callback(_tasks.discard) + + +async def _deliver(targets: list[tuple[str, webpush.Target]], payload: dict, + ttl: int) -> None: + results = await asyncio.gather(*(_send(t, payload, ttl=ttl) for _, t in targets), + return_exceptions=True) + gone = [sid for (sid, _), r in zip(targets, results, strict=True) if r == webpush.GONE] + if not gone: + return + # The distributor dropped the registration: pushing there would cost a + # request per notification for ever and reach nothing. The row stays, without + # its endpoint — the phone still fetches, and registers again when it can. + try: + from meshbay_hub.db.engine import get_session_factory + async with get_session_factory()() as db: + await db.execute( + update(PushSubscription).where(PushSubscription.id.in_(gone)) + .values(endpoint=None, p256dh=None, auth=None)) + await db.commit() + except Exception as e: + log.warning("Could not drop %d gone push subscription(s): %s", len(gone), e) + + +async def drain() -> None: + """Wait for every push in flight — for the tests, and for a clean shutdown.""" + # Done tasks leave the set from a callback the loop has not run yet, and + # awaiting a finished gather never yields to it: wait on the unfinished only. + while pending := [t for t in _tasks if not t.done()]: + await asyncio.gather(*pending, return_exceptions=True) -- cgit v1.2.3