"""Web Push (VAPID) — key management and the low-level send. This is the transport under the notification engine: notify.py decides *what* to say and *to whom*; this module signs and delivers it to a browser push service. Keys are resolved once, in this order (mirrors the self-signed-cert flow in run.sh — generated on first use, reused thereafter): 1. env — ``THERMOGRAPH_VAPID_PRIVATE_KEY`` + ``THERMOGRAPH_VAPID_PUBLIC_KEY``, both base64url-raw (the private key is the 32-byte scalar, the public key the 65-byte uncompressed point — the standard VAPID interchange format). Set both in production so every worker/deploy signs with the same identity. 2. ``data/vapid.json`` — a gitignored keypair written on first run. 3. freshly generated, then written to (2). The private key signs the VAPID JWT; the public key is the ``applicationServerKey`` the browser needs at subscribe time (served via GET /push/vapid-key). Nothing here touches the database — ``send()`` returns a status string and the caller (which owns the session) prunes rows the push service says are gone. """ import base64 import json import logging import os import threading from core import audit import paths from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric import ec from pywebpush import WebPushException, webpush log = logging.getLogger("thermograph.push") _DATA_DIR = paths.DATA_DIR _VAPID_PATH = os.environ.get("THERMOGRAPH_VAPID_FILE") or os.path.join(_DATA_DIR, "vapid.json") # The VAPID "sub" claim — a contact the push service can reach about our traffic. _CONTACT = os.environ.get("THERMOGRAPH_VAPID_CONTACT", "mailto:admin@thermograph.org") # pywebpush 2.0.0 forwards `timeout` straight to requests.post with no default of # its own — omit the kwarg here and the send blocks with NO timeout at all (the # "10s default" people expect only applies if requests itself is called bare). # One hung push endpoint must never wedge the notifier: the singleton flock # (core/singleton.py) is held for the process's whole life, so no other worker # can take over while this thread is stuck, and /healthz keeps reporting green # on a notifier that's actually frozen. _SEND_TIMEOUT = 15 _lock = threading.Lock() _keys = None # cached {"private_key": str, "public_key": str} (base64url-raw) def _b64url(raw: bytes) -> str: return base64.urlsafe_b64encode(raw).rstrip(b"=").decode() def _generate() -> dict: """A fresh P-256 keypair as base64url-raw: 32-byte private scalar + 65-byte uncompressed public point (the format pywebpush's from_string and the browser applicationServerKey both accept).""" priv = ec.generate_private_key(ec.SECP256R1()) scalar = priv.private_numbers().private_value.to_bytes(32, "big") raw_pub = priv.public_key().public_bytes( serialization.Encoding.X962, serialization.PublicFormat.UncompressedPoint, ) return {"private_key": _b64url(scalar), "public_key": _b64url(raw_pub)} def _claim_file(data: dict) -> dict: """Persist a freshly generated keypair as `_VAPID_PATH`'s content, atomic first-writer-wins. `_lock` only keeps this process's own threads from racing each other — on a cold /state volume every uvicorn *worker* (a separate process) hits the missing-file branch at boot together, and without this each would generate its OWN keypair and cache it in-process, silently diverging from the file and from each other (a subscription signed against one worker's public key fails to verify under another's private key — the exact incident class deploy/entrypoint.sh's comments name). Write the full keypair to a private temp file first, then claim the real path with a hard link: `os.link` is atomic and the loser gets EEXIST immediately, with no window where `_VAPID_PATH` exists but is only half-written (unlike O_CREAT|O_EXCL directly on the destination followed by a separate write). A loser discards its own generation and reads back the winner's file instead, so every process ends up caching the SAME keys the file actually holds.""" tmp_path = f"{_VAPID_PATH}.{os.getpid()}.tmp" try: os.makedirs(_DATA_DIR, exist_ok=True) with open(tmp_path, "w", encoding="utf-8") as f: json.dump(data, f) os.chmod(tmp_path, 0o600) except OSError: log.warning("could not persist VAPID keys to %s; using an in-memory pair", _VAPID_PATH) return data try: try: os.link(tmp_path, _VAPID_PATH) return data except FileExistsError: # Lost the race — someone else's file is now the truth. Read IT back # rather than keep our own now-orphaned generation. try: with open(_VAPID_PATH, encoding="utf-8") as f: winner = json.load(f) if winner.get("private_key") and winner.get("public_key"): return winner except (OSError, ValueError): pass # Winner's file was unreadable/corrupt (very rare) — fall back to our # own in-memory generation rather than crash; the next _load() call # (a new process, or after the file heals) retries the file. return data except OSError: log.warning("could not persist VAPID keys to %s; using an in-memory pair", _VAPID_PATH) return data finally: try: os.unlink(tmp_path) except OSError: pass def _load() -> dict: """Resolve the keypair once (env → file → generate) and cache it.""" global _keys if _keys is not None: return _keys with _lock: if _keys is not None: return _keys env_priv = os.environ.get("THERMOGRAPH_VAPID_PRIVATE_KEY") env_pub = os.environ.get("THERMOGRAPH_VAPID_PUBLIC_KEY") if env_priv and env_pub: _keys = {"private_key": env_priv.strip(), "public_key": env_pub.strip()} return _keys try: with open(_VAPID_PATH, encoding="utf-8") as f: data = json.load(f) if data.get("private_key") and data.get("public_key"): _keys = {"private_key": data["private_key"], "public_key": data["public_key"]} return _keys except (OSError, ValueError): pass # Cache whatever _claim_file settles on — our own generation if we won # the race to create the file, or the winner's keys read back from disk # if we lost it. Either way this process's cache matches the file. _keys = _claim_file(_generate()) return _keys def public_key() -> str: """The applicationServerKey (base64url raw point) the browser subscribes with.""" return _load()["public_key"] def send(subscription_info: dict, payload: dict) -> str: """Deliver one push. Returns 'ok', 'gone' (prune the row), or 'error'. `subscription_info` is the browser shape: {endpoint, keys:{p256dh, auth}}. Never raises — a bad send must not break the caller's transaction. """ keys = _load() try: webpush( subscription_info=subscription_info, data=json.dumps(payload), vapid_private_key=keys["private_key"], vapid_claims={"sub": _CONTACT}, ttl=86400, timeout=_SEND_TIMEOUT, ) return "ok" except WebPushException as e: status = getattr(getattr(e, "response", None), "status_code", None) if status in (401, 403, 404, 410): # 404/410 = endpoint retired; 401/403 = VAPID mismatch (e.g. after a key # rotation) — the row was minted under a key we can no longer sign for and # will never authenticate again. All are permanently dead: caller deletes. return "gone" # Anything else (rate limits, 5xx, malformed payload) may be transient — keep # the row and record it where it's visible (the errors JSONL / dashboard). log.warning("web push failed (status=%s): %s", status, e) audit.log_event("error", {"phase": "push", "status": status, "endpoint": (subscription_info.get("endpoint") or "")[:120]}) return "error" except Exception as e: # noqa: BLE001 - network/encoding errors must not propagate log.warning("web push send raised", exc_info=True) audit.log_event("error", {"phase": "push", "error": repr(e), "endpoint": (subscription_info.get("endpoint") or "")[:120]}) return "error"