Sandbar/internal/sandbar/push.go

package sandbar

import (
	"bytes"
	"crypto/aes"
	"crypto/cipher"
	"crypto/ecdh"
	"crypto/ecdsa"
	"crypto/elliptic"
	"crypto/hkdf"
	"crypto/rand"
	"crypto/sha256"
	"database/sql"
	"encoding/base64"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log"
	"math/big"
	"net/http"
	"net/url"
	"strconv"
	"strings"
	"time"
	"unicode/utf8"
)

// Web Push lets the mobile companion (a PWA) get a notification while it is closed. Each
// device registers a push subscription and the topics it wants; Publish then sends one
// encrypted push per matching device. It is implemented here with the standard library only:
// VAPID (RFC 8292) identifies Sandbar to the push service, and the payload is encrypted with
// aes128gcm (RFC 8291) so the push service never sees the message.

const (
	vapidKind    = "vapid"
	pushRecord   = 4096
	pushMaxBody  = 1200 // characters of message body sent in a push; the app fetches the rest
	pushTTL      = 4 * 60 * 60
	maxPushSubs  = 20 // per person
	pushSendWait = 10 * time.Second
)

// PushSub is one device registered for push.
type PushSub struct {
	ID       string   `json:"id"`
	UserID   string   `json:"-"`
	Endpoint string   `json:"endpoint"`
	P256dh   string   `json:"-"`
	Auth     string   `json:"-"`
	Topics   []string `json:"topics"`
	Device   string   `json:"device"`
	Created  int64    `json:"created"`
}

// vapidKey returns the server's VAPID key pair, creating it on first use. The private key
// is stored sealed with the master key like channel secrets.
func (s *Server) vapidKey() (*ecdsa.PrivateKey, string, error) {
	s.pushMu.Lock()
	defer s.pushMu.Unlock()
	if s.vapid != nil {
		return s.vapid, s.vapidPub, nil
	}
	var doc struct {
		Priv string `json:"priv"`
	}
	var k *ecdsa.PrivateKey
	if err := s.st.get(vapidKind, "key", &doc); err == nil {
		if raw, err := base64.RawURLEncoding.DecodeString(s.st.open(doc.Priv)); err == nil {
			if ek, err := ecdh.P256().NewPrivateKey(raw); err == nil {
				k = ecdhToECDSA(ek)
			}
		}
	}
	if k == nil {
		ek, err := ecdh.P256().GenerateKey(rand.Reader)
		if err != nil {
			return nil, "", err
		}
		doc.Priv = s.st.seal(base64.RawURLEncoding.EncodeToString(ek.Bytes()))
		if err := s.st.put(vapidKind, "key", doc); err != nil {
			return nil, "", err
		}
		k = ecdhToECDSA(ek)
	}
	s.vapid = k
	pub, _ := k.PublicKey.ECDH()
	s.vapidPub = base64.RawURLEncoding.EncodeToString(pub.Bytes())
	return s.vapid, s.vapidPub, nil
}

func ecdhToECDSA(k *ecdh.PrivateKey) *ecdsa.PrivateKey {
	pub := k.PublicKey().Bytes()
	x, y := elliptic.Unmarshal(elliptic.P256(), pub)
	d := new(big.Int).SetBytes(k.Bytes())
	return &ecdsa.PrivateKey{PublicKey: ecdsa.PublicKey{Curve: elliptic.P256(), X: x, Y: y}, D: d}
}

func (s *Server) vapidSubject() string {
	if s.opts.VapidSubject != "" {
		return s.opts.VapidSubject
	}
	if strings.HasPrefix(s.opts.URL, "https://") {
		return s.opts.URL
	}
	return "mailto:sandbar@localhost"
}

// vapidAuth builds the Authorization header for a push endpoint.
func (s *Server) vapidAuth(endpoint string) (string, error) {
	k, pub, err := s.vapidKey()
	if err != nil {
		return "", err
	}
	u, err := url.Parse(endpoint)
	if err != nil {
		return "", err
	}
	hdr := base64.RawURLEncoding.EncodeToString([]byte(`{"typ":"JWT","alg":"ES256"}`))
	claims, _ := json.Marshal(map[string]any{"aud": u.Scheme + "://" + u.Host, "exp": s.now().Add(12 * time.Hour).Unix(), "sub": s.vapidSubject()})
	signing := hdr + "." + base64.RawURLEncoding.EncodeToString(claims)
	h := sha256.Sum256([]byte(signing))
	r, sg, err := ecdsa.Sign(rand.Reader, k, h[:])
	if err != nil {
		return "", err
	}
	sig := make([]byte, 64)
	r.FillBytes(sig[:32])
	sg.FillBytes(sig[32:])
	return "vapid t=" + signing + "." + base64.RawURLEncoding.EncodeToString(sig) + ", k=" + pub, nil
}

// encryptPush encrypts payload for a subscription (RFC 8291, aes128gcm, one record).
func encryptPush(payload []byte, p256dh, authSecret string) ([]byte, error) {
	uaRaw, err := base64.RawURLEncoding.DecodeString(strings.TrimRight(p256dh, "="))
	if err != nil {
		return nil, errors.New("bad p256dh key")
	}
	auth, err := base64.RawURLEncoding.DecodeString(strings.TrimRight(authSecret, "="))
	if err != nil || len(auth) != 16 {
		return nil, errors.New("bad auth secret")
	}
	uaPub, err := ecdh.P256().NewPublicKey(uaRaw)
	if err != nil {
		return nil, errors.New("bad p256dh key")
	}
	if len(payload) > pushRecord-17-1 {
		return nil, errors.New("payload too large")
	}
	asKey, err := ecdh.P256().GenerateKey(rand.Reader)
	if err != nil {
		return nil, err
	}
	secret, err := asKey.ECDH(uaPub)
	if err != nil {
		return nil, err
	}
	asPub := asKey.PublicKey().Bytes()
	salt := make([]byte, 16)
	if _, err := rand.Read(salt); err != nil {
		return nil, err
	}
	prkKey, err := hkdf.Extract(sha256.New, secret, auth)
	if err != nil {
		return nil, err
	}
	keyInfo := append(append([]byte("WebPush: info\x00"), uaRaw...), asPub...)
	ikm, err := hkdf.Expand(sha256.New, prkKey, string(keyInfo), 32)
	if err != nil {
		return nil, err
	}
	prk, err := hkdf.Extract(sha256.New, ikm, salt)
	if err != nil {
		return nil, err
	}
	cek, err := hkdf.Expand(sha256.New, prk, "Content-Encoding: aes128gcm\x00", 16)
	if err != nil {
		return nil, err
	}
	nonce, err := hkdf.Expand(sha256.New, prk, "Content-Encoding: nonce\x00", 12)
	if err != nil {
		return nil, err
	}
	blk, err := aes.NewCipher([]byte(cek))
	if err != nil {
		return nil, err
	}
	gcm, err := cipher.NewGCM(blk)
	if err != nil {
		return nil, err
	}
	ct := gcm.Seal(nil, []byte(nonce), append(append([]byte{}, payload...), 2), nil)
	var out bytes.Buffer
	out.Write(salt)
	out.Write([]byte{0, 0, byte(pushRecord >> 8), byte(pushRecord & 0xff), byte(len(asPub))})
	out.Write(asPub)
	out.Write(ct)
	return out.Bytes(), nil
}

// pushPayload is what the service worker receives.
type pushPayload struct {
	ID       string   `json:"id"`
	Topic    string   `json:"topic"`
	Title    string   `json:"title"`
	Message  string   `json:"message"`
	Priority int      `json:"priority"`
	Tags     []string `json:"tags,omitempty"`
	Time     int64    `json:"time"`
	Alert    string   `json:"alert,omitempty"` // set on a firing on-call alert so the app can offer Acknowledge
	Status   string   `json:"status,omitempty"`
}

func truncate(s string, n int) string {
	if utf8.RuneCountInString(s) <= n {
		return s
	}
	r := []rune(s)
	return string(r[:n]) + "..."
}

// sendPush delivers one payload to one subscription. A gone subscription is removed.
func (s *Server) sendPush(sub PushSub, payload []byte, urgency string) error {
	body, err := encryptPush(payload, sub.P256dh, sub.Auth)
	if err != nil {
		return err
	}
	authz, err := s.vapidAuth(sub.Endpoint)
	if err != nil {
		return err
	}
	req, err := http.NewRequest("POST", sub.Endpoint, bytes.NewReader(body))
	if err != nil {
		return err
	}
	req.Header.Set("Content-Encoding", "aes128gcm")
	req.Header.Set("Content-Type", "application/octet-stream")
	req.Header.Set("TTL", strconv.Itoa(pushTTL))
	req.Header.Set("Urgency", urgency)
	req.Header.Set("Authorization", authz)
	c := *s.httpClient(s.opts.PushAllowPrivate)
	c.Timeout = pushSendWait
	resp, err := c.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	io.Copy(io.Discard, io.LimitReader(resp.Body, 4096))
	if resp.StatusCode == 404 || resp.StatusCode == 410 {
		s.st.db.Exec(`DELETE FROM push_subs WHERE id=?`, sub.ID)
		return fmt.Errorf("subscription gone (%d), removed", resp.StatusCode)
	}
	if resp.StatusCode >= 300 {
		return fmt.Errorf("push service answered %d", resp.StatusCode)
	}
	return nil
}

func urgencyFor(prio int) string {
	switch {
	case prio >= 5:
		return "high"
	case prio == 4:
		return "normal"
	case prio == 2:
		return "low"
	case prio <= 1:
		return "very-low"
	}
	return "normal"
}

// pushMessage sends a published message to every device subscribed to its topic.
func (s *Server) pushMessage(m Message, status string) {
	subs := s.pushSubsForTopic(m.Topic)
	if len(subs) == 0 {
		return
	}
	p, _ := json.Marshal(pushPayload{ID: m.ID, Topic: m.Topic, Title: m.Title, Message: truncate(m.Message, pushMaxBody), Priority: m.Priority,
		Tags: m.Tags, Time: m.Time, Alert: m.Alert, Status: status})
	for _, sub := range subs {
		sub := sub
		go func() {
			if err := s.sendPush(sub, p, urgencyFor(m.Priority)); err != nil {
				log.Printf("push to %s: %v", sub.ID, err)
			}
		}()
	}
}

const pushCols = `id,user_id,endpoint,p256dh,auth,topics,device,created`

func scanSub(r scanner) (PushSub, error) {
	var p PushSub
	var tj string
	err := r.Scan(&p.ID, &p.UserID, &p.Endpoint, &p.P256dh, &p.Auth, &tj, &p.Device, &p.Created)
	json.Unmarshal([]byte(tj), &p.Topics)
	if p.Topics == nil {
		p.Topics = []string{}
	}
	return p, err
}

func (s *Server) pushSubs(where string, args ...any) []PushSub {
	out := []PushSub{}
	rows, err := s.st.db.Query(`SELECT `+pushCols+` FROM push_subs `+where, args...)
	if err != nil {
		return out
	}
	defer rows.Close()
	for rows.Next() {
		if p, err := scanSub(rows); err == nil {
			out = append(out, p)
		}
	}
	return out
}

func (s *Server) pushSubsForTopic(topic string) []PushSub {
	var out []PushSub
	for _, p := range s.pushSubs(`ORDER BY created`) { // topics live in a JSON column; there are few devices
		for _, t := range p.Topics {
			if t == topic {
				out = append(out, p)
				break
			}
		}
	}
	return out
}

// ---------- HTTP ----------

func (s *Server) pushKeyHTTP(w http.ResponseWriter, r *http.Request) {
	_, pub, err := s.vapidKey()
	if err != nil {
		fail(w, err)
		return
	}
	writeJSON(w, 200, map[string]string{"publicKey": pub})
}

func (s *Server) listPushSubs(w http.ResponseWriter, r *http.Request) {
	writeJSON(w, 200, s.pushSubs(`WHERE user_id=? ORDER BY created`, userFrom(r.Context()).ID))
}

func (s *Server) savePushSub(w http.ResponseWriter, r *http.Request) {
	var in struct {
		Endpoint string `json:"endpoint"`
		Keys     struct {
			P256dh string `json:"p256dh"`
			Auth   string `json:"auth"`
		} `json:"keys"`
		Topics []string `json:"topics"`
		Device string   `json:"device"`
	}
	if !decode(r, &in) {
		bad(w, "invalid request")
		return
	}
	u := userFrom(r.Context())
	ep, err := url.Parse(in.Endpoint)
	if err != nil || (ep.Scheme != "https" && !(s.opts.PushAllowPrivate && ep.Scheme == "http")) || ep.Host == "" || len(in.Endpoint) > 1024 {
		bad(w, "the push endpoint must be an https address")
		return
	}
	// Prove the keys are usable now rather than when the first alert fails to reach the phone.
	if _, err := encryptPush([]byte("x"), in.Keys.P256dh, in.Keys.Auth); err != nil {
		bad(w, err.Error())
		return
	}
	topics := []string{userTopic(u.ID)} // the personal topic always comes along: it carries on-call pages
	seen := map[string]bool{topics[0]: true}
	for _, t := range in.Topics {
		if t = strings.TrimSpace(t); validTopic(t) && !seen[t] {
			seen[t] = true
			topics = append(topics, t)
		}
	}
	if len(topics) > 50 {
		bad(w, "subscribe to at most 50 topics")
		return
	}
	tj, _ := json.Marshal(topics)
	device := truncate(strings.TrimSpace(in.Device), 80)
	var id string
	err = s.st.db.QueryRow(`SELECT id FROM push_subs WHERE endpoint=?`, in.Endpoint).Scan(&id)
	switch {
	case err == nil:
		_, err = s.st.db.Exec(`UPDATE push_subs SET user_id=?,p256dh=?,auth=?,topics=?,device=? WHERE id=?`, u.ID, in.Keys.P256dh, in.Keys.Auth, string(tj), device, id)
	case errors.Is(err, sql.ErrNoRows):
		var n int
		s.st.db.QueryRow(`SELECT COUNT(*) FROM push_subs WHERE user_id=?`, u.ID).Scan(&n)
		if n >= maxPushSubs {
			bad(w, "this account has too many registered devices, remove one first")
			return
		}
		id = newID()
		_, err = s.st.db.Exec(`INSERT INTO push_subs(`+pushCols+`) VALUES(?,?,?,?,?,?,?,?)`, id, u.ID, in.Endpoint, in.Keys.P256dh, in.Keys.Auth, string(tj), device, s.now().Unix())
	}
	if err != nil {
		fail(w, err)
		return
	}
	p, err := scanSub(s.st.db.QueryRow(`SELECT `+pushCols+` FROM push_subs WHERE id=?`, id))
	if err != nil {
		fail(w, err)
		return
	}
	writeJSON(w, 200, p)
}

func (s *Server) deletePushSub(w http.ResponseWriter, r *http.Request) {
	_, err := s.st.db.Exec(`DELETE FROM push_subs WHERE id=? AND user_id=?`, r.PathValue("id"), userFrom(r.Context()).ID)
	fail2(w, err)
}

// testPush sends a test notification to the caller's own devices and says how each one went.
func (s *Server) testPush(w http.ResponseWriter, r *http.Request) {
	u := userFrom(r.Context())
	p, _ := json.Marshal(pushPayload{ID: newID(), Topic: userTopic(u.ID), Title: "Sandbar test", Message: "If you can read this, push notifications work on this device.",
		Priority: 3, Time: s.now().Unix(), Status: "test"})
	type result struct {
		ID     string `json:"id"`
		Device string `json:"device"`
		Error  string `json:"error,omitempty"`
	}
	out := []result{}
	for _, sub := range s.pushSubs(`WHERE user_id=?`, u.ID) {
		res := result{ID: sub.ID, Device: sub.Device}
		if err := s.sendPush(sub, p, "normal"); err != nil {
			res.Error = err.Error()
		}
		out = append(out, res)
	}
	writeJSON(w, 200, out)
}