thermograph/backend/notifications/push.py

"""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"