coder/coderd/webpush/webpush.go

package webpush

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"io"
	"net"
	"net/http"
	"net/netip"
	"slices"
	"sync"
	"syscall"
	"time"

	"github.com/SherClockHolmes/webpush-go"
	"github.com/google/uuid"
	"golang.org/x/sync/errgroup"
	"golang.org/x/xerrors"
	"tailscale.com/util/singleflight"

	"cdr.dev/slog/v3"
	"github.com/coder/coder/v2/coderd/database"
	"github.com/coder/coder/v2/coderd/database/dbauthz"
	"github.com/coder/coder/v2/codersdk"
	"github.com/coder/quartz"
)

const defaultSubscriptionCacheTTL = 3 * time.Minute

// isStaleSubscriptionStatus reports whether a status code from a push
// service indicates that the subscription is permanently invalid and
// should be removed from the database. Other 4xx and 5xx responses
// (rate limits, transient failures) leave the subscription in place
// so it can be retried on the next dispatch.
func isStaleSubscriptionStatus(statusCode int) bool {
	switch statusCode {
	case http.StatusBadRequest, // 400: malformed subscription per the push service.
		http.StatusForbidden, // 403: Apple BadJwtToken / VAPID rejected, key rotation.
		http.StatusNotFound,  // 404: FCM/Mozilla endpoint no longer valid.
		http.StatusGone:      // 410: standard "subscription expired" signal.
		return true
	}
	return false
}

// Dispatcher is an interface that can be used to dispatch
// web push notifications to clients such as browsers.
type Dispatcher interface {
	// Dispatch sends a web push notification to all subscriptions
	// for a user. Any notifications that fail to send are silently dropped.
	Dispatch(ctx context.Context, userID uuid.UUID, notification codersdk.WebpushMessage) error
	// Test sends a test web push notificatoin to a subscription to ensure it is valid.
	Test(ctx context.Context, req codersdk.WebpushSubscription) error
	// PublicKey returns the VAPID public key for the webpush dispatcher.
	PublicKey() string
}

// SubscriptionCacheInvalidator is an optional interface that lets local
// subscription mutation handlers invalidate cached subscriptions.
type SubscriptionCacheInvalidator interface {
	InvalidateUser(userID uuid.UUID)
}

type options struct {
	clock                quartz.Clock
	subscriptionCacheTTL time.Duration
	httpClient           *http.Client
}

// Option configures optional behavior for a Webpusher.
type Option func(*options)

// WithClock sets the clock used by the subscription cache. Defaults to a real
// clock when not provided.
func WithClock(clock quartz.Clock) Option {
	return func(o *options) {
		o.clock = clock
	}
}

// WithSubscriptionCacheTTL sets the in-memory subscription cache TTL. Defaults
// to three minutes when not provided or when given a non-positive duration.
func WithSubscriptionCacheTTL(ttl time.Duration) Option {
	return func(o *options) {
		o.subscriptionCacheTTL = ttl
	}
}

// WithHTTPClient overrides the default SSRF-safe HTTP client used to deliver
// push notifications. This is intended for tests that need to deliver to
// localhost test servers.
func WithHTTPClient(client *http.Client) Option {
	return func(o *options) {
		o.httpClient = client
	}
}

// New creates a new Dispatcher to dispatch web push notifications.
//
// This is *not* integrated into the enqueue system unfortunately.
// That's because the notifications system has a enqueue system,
// and push notifications at time of implementation are being used
// for updates inside of a workspace, which we want to be immediate.
//
// See: https://github.com/coder/internal/issues/528
func New(ctx context.Context, log *slog.Logger, db database.Store, vapidSub string, opts ...Option) (Dispatcher, error) {
	cfg := options{
		clock:                quartz.NewReal(),
		subscriptionCacheTTL: defaultSubscriptionCacheTTL,
	}
	for _, opt := range opts {
		opt(&cfg)
	}
	if cfg.clock == nil {
		cfg.clock = quartz.NewReal()
	}
	if cfg.subscriptionCacheTTL <= 0 {
		cfg.subscriptionCacheTTL = defaultSubscriptionCacheTTL
	}
	if cfg.httpClient == nil {
		cfg.httpClient = newSSRFSafeHTTPClient()
	}

	keys, err := db.GetWebpushVAPIDKeys(ctx)
	if err != nil {
		if !errors.Is(err, sql.ErrNoRows) {
			return nil, xerrors.Errorf("get notification vapid keys: %w", err)
		}
	}

	if keys.VapidPublicKey == "" || keys.VapidPrivateKey == "" {
		// Generate new VAPID keys. This also deletes all existing push
		// subscriptions as part of the transaction, as they are no longer
		// valid.
		newPrivateKey, newPublicKey, err := RegenerateVAPIDKeys(ctx, db)
		if err != nil {
			return nil, xerrors.Errorf("regenerate vapid keys: %w", err)
		}

		keys.VapidPublicKey = newPublicKey
		keys.VapidPrivateKey = newPrivateKey
	}

	return &Webpusher{
		vapidSub:                vapidSub,
		store:                   db,
		log:                     log,
		VAPIDPublicKey:          keys.VapidPublicKey,
		VAPIDPrivateKey:         keys.VapidPrivateKey,
		clock:                   cfg.clock,
		subscriptionCacheTTL:    cfg.subscriptionCacheTTL,
		subscriptionCache:       make(map[uuid.UUID]cachedSubscriptions),
		subscriptionGenerations: make(map[uuid.UUID]uint64),
		httpClient:              cfg.httpClient,
	}, nil
}

type cachedSubscriptions struct {
	subscriptions []database.WebpushSubscription
	expiresAt     time.Time
}

type Webpusher struct {
	store database.Store
	log   *slog.Logger
	// VAPID allows us to identify the sender of the message.
	// This must be a https:// URL or an email address.
	// Some push services (such as Apple's) require this to be set.
	vapidSub string

	// public and private keys for VAPID. These are used to sign and encrypt
	// the message payload.
	VAPIDPublicKey  string
	VAPIDPrivateKey string

	// httpClient is an SSRF-safe HTTP client that rejects connections to
	// private, loopback, and link-local IP addresses at dial time. This
	// closes the DNS rebinding TOCTOU gap where a hostname passes URL
	// validation but resolves to a private IP when the connection is made.
	httpClient *http.Client

	clock quartz.Clock

	cacheMu                 sync.RWMutex
	subscriptionCache       map[uuid.UUID]cachedSubscriptions
	subscriptionGenerations map[uuid.UUID]uint64
	subscriptionCacheTTL    time.Duration
	subscriptionFetches     singleflight.Group[string, []database.WebpushSubscription]
}

func (n *Webpusher) Dispatch(ctx context.Context, userID uuid.UUID, msg codersdk.WebpushMessage) error {
	subscriptions, err := n.subscriptionsForUser(ctx, userID)
	if err != nil {
		return xerrors.Errorf("get web push subscriptions by user ID: %w", err)
	}
	if len(subscriptions) == 0 {
		return nil
	}

	msgJSON, err := json.Marshal(msg)
	if err != nil {
		return xerrors.Errorf("marshal webpush notification: %w", err)
	}

	cleanupSubscriptions := make([]uuid.UUID, 0)
	var mu sync.Mutex
	var eg errgroup.Group
	for _, subscription := range subscriptions {
		eg.Go(func() error {
			// TODO: Implement some retry logic here. For now, this is just a
			// best-effort attempt.
			statusCode, body, err := n.webpushSend(ctx, msgJSON, subscription.Endpoint, webpush.Keys{
				Auth:   subscription.EndpointAuthKey,
				P256dh: subscription.EndpointP256dhKey,
			})
			if err != nil {
				return xerrors.Errorf("send webpush notification: %w", err)
			}

			if isStaleSubscriptionStatus(statusCode) {
				// Remove subscriptions that the push service has marked as
				// permanently invalid (Apple returns 403 BadJwtToken and 404
				// for invalidated subscriptions, FCM returns 404 for
				// expired endpoints, all push services return 410 for
				// permanently gone subscriptions, and 400 indicates a
				// malformed subscription that cannot be retried). Without
				// this, stale rows accumulate after PWA reinstalls and the
				// in-memory cache keeps trying to deliver to dead
				// subscriptions.
				mu.Lock()
				cleanupSubscriptions = append(cleanupSubscriptions, subscription.ID)
				mu.Unlock()
			}

			if statusCode == http.StatusGone {
				// 410 Gone is informational, not a delivery error.
				return nil
			}

			// 200, 201, and 202 are common for successful delivery.
			if statusCode > http.StatusAccepted {
				// It's likely the subscription failed to deliver for some reason.
				return xerrors.Errorf("web push dispatch failed with status code %d: %s", statusCode, string(body))
			}

			return nil
		})
	}

	dispatchErr := eg.Wait()

	// Always remove subscriptions that the push service rejected as
	// permanently invalid, even when sibling deliveries returned a
	// non-stale error. The cleanup must run before the error return so a
	// transient delivery failure on one subscription cannot block the
	// deletion of a 410/404/403/400 sibling. Without this ordering,
	// stale rows accumulate after PWA reinstalls and silently mask the
	// new subscription on every subsequent dispatch.
	n.cleanupStaleSubscriptions(ctx, userID, cleanupSubscriptions)

	if dispatchErr != nil {
		return xerrors.Errorf("send webpush notifications: %w", dispatchErr)
	}

	return nil
}

// cleanupStaleSubscriptions deletes the rows the push service flagged as
// permanently invalid (see isStaleSubscriptionStatus) and clears the cached
// entries for the affected user. Failures are logged at error level rather
// than returned: the caller is in the middle of returning a delivery error
// and shouldn't have its error shadowed by a cleanup failure. The cache
// prune is gated on a successful database delete so a partial state cannot
// leak into the cache.
func (n *Webpusher) cleanupStaleSubscriptions(ctx context.Context, userID uuid.UUID, ids []uuid.UUID) {
	if len(ids) == 0 {
		return
	}
	// nolint:gocritic // These are known to be invalid subscriptions.
	if err := n.store.DeleteWebpushSubscriptions(dbauthz.AsNotifier(ctx), ids); err != nil {
		n.log.Error(ctx, "failed to delete stale push subscriptions", slog.Error(err))
		return
	}
	n.pruneSubscriptions(userID, ids)
}

func (n *Webpusher) subscriptionsForUser(ctx context.Context, userID uuid.UUID) ([]database.WebpushSubscription, error) {
	if subscriptions, ok := n.cachedSubscriptions(userID); ok {
		return subscriptions, nil
	}

	subscriptions, err, _ := n.subscriptionFetches.Do(userID.String(), func() ([]database.WebpushSubscription, error) {
		if cached, ok := n.cachedSubscriptions(userID); ok {
			return cached, nil
		}

		generation := n.subscriptionGeneration(userID)
		fetched, err := n.store.GetWebpushSubscriptionsByUserID(ctx, userID)
		if err != nil {
			return nil, err
		}
		n.storeSubscriptions(userID, generation, fetched)
		return slices.Clone(fetched), nil
	})
	if err != nil {
		return nil, err
	}

	return slices.Clone(subscriptions), nil
}

func (n *Webpusher) cachedSubscriptions(userID uuid.UUID) ([]database.WebpushSubscription, bool) {
	n.cacheMu.RLock()
	entry, ok := n.subscriptionCache[userID]
	n.cacheMu.RUnlock()
	if !ok {
		return nil, false
	}
	if n.clock.Now().Before(entry.expiresAt) {
		return slices.Clone(entry.subscriptions), true
	}

	n.cacheMu.Lock()
	if current, ok := n.subscriptionCache[userID]; ok && !n.clock.Now().Before(current.expiresAt) {
		delete(n.subscriptionCache, userID)
	}
	n.cacheMu.Unlock()

	return nil, false
}

func (n *Webpusher) subscriptionGeneration(userID uuid.UUID) uint64 {
	n.cacheMu.RLock()
	generation := n.subscriptionGenerations[userID]
	n.cacheMu.RUnlock()
	return generation
}

func (n *Webpusher) storeSubscriptions(userID uuid.UUID, generation uint64, subscriptions []database.WebpushSubscription) {
	n.cacheMu.Lock()
	defer n.cacheMu.Unlock()

	if n.subscriptionGenerations[userID] != generation {
		return
	}

	n.subscriptionCache[userID] = cachedSubscriptions{
		subscriptions: slices.Clone(subscriptions),
		expiresAt:     n.clock.Now().Add(n.subscriptionCacheTTL),
	}
}

func (n *Webpusher) pruneSubscriptions(userID uuid.UUID, staleIDs []uuid.UUID) {
	if len(staleIDs) == 0 {
		return
	}

	stale := make(map[uuid.UUID]struct{}, len(staleIDs))
	for _, id := range staleIDs {
		stale[id] = struct{}{}
	}

	n.cacheMu.Lock()
	defer n.cacheMu.Unlock()

	entry, ok := n.subscriptionCache[userID]
	if !ok {
		return
	}
	if !n.clock.Now().Before(entry.expiresAt) {
		delete(n.subscriptionCache, userID)
		return
	}

	filtered := make([]database.WebpushSubscription, 0, len(entry.subscriptions))
	for _, subscription := range entry.subscriptions {
		if _, shouldDelete := stale[subscription.ID]; shouldDelete {
			continue
		}
		filtered = append(filtered, subscription)
	}
	if len(filtered) == 0 {
		delete(n.subscriptionCache, userID)
		return
	}

	entry.subscriptions = filtered
	n.subscriptionCache[userID] = entry
}

// InvalidateUser clears the cached subscriptions for a user and advances
// its invalidation generation. Local subscribe and unsubscribe handlers call
// this after mutating subscriptions in the same process.
func (n *Webpusher) InvalidateUser(userID uuid.UUID) {
	n.cacheMu.Lock()
	delete(n.subscriptionCache, userID)
	n.subscriptionGenerations[userID]++
	n.cacheMu.Unlock()
	n.subscriptionFetches.Forget(userID.String())
}

func (n *Webpusher) webpushSend(ctx context.Context, msg []byte, endpoint string, keys webpush.Keys) (int, []byte, error) {
	// Copy the message to avoid modifying the original.
	cpy := slices.Clone(msg)
	resp, err := webpush.SendNotificationWithContext(ctx, cpy, &webpush.Subscription{
		Endpoint: endpoint,
		Keys:     keys,
	}, &webpush.Options{
		HTTPClient:      n.httpClient,
		Subscriber:      n.vapidSub,
		VAPIDPublicKey:  n.VAPIDPublicKey,
		VAPIDPrivateKey: n.VAPIDPrivateKey,
	})
	if err != nil {
		n.log.Error(ctx, "failed to send webpush notification", slog.Error(err), slog.F("endpoint", endpoint))
		return -1, nil, xerrors.Errorf("send webpush notification: %w", err)
	}
	defer resp.Body.Close()
	body, err := io.ReadAll(resp.Body)
	if err != nil {
		return -1, nil, xerrors.Errorf("read response body: %w", err)
	}

	return resp.StatusCode, body, nil
}

func (n *Webpusher) Test(ctx context.Context, req codersdk.WebpushSubscription) error {
	msgJSON, err := json.Marshal(codersdk.WebpushMessage{
		Title: "It's working!",
		Body:  "You've subscribed to push notifications.",
	})
	if err != nil {
		return xerrors.Errorf("marshal webpush notification: %w", err)
	}
	statusCode, body, err := n.webpushSend(ctx, msgJSON, req.Endpoint, webpush.Keys{
		Auth:   req.AuthKey,
		P256dh: req.P256DHKey,
	})
	if err != nil {
		return xerrors.Errorf("send test webpush notification: %w", err)
	}

	// 200, 201, and 202 are common for successful delivery.
	if statusCode > http.StatusAccepted {
		// It's likely the subscription failed to deliver for some reason.
		return xerrors.Errorf("web push dispatch failed with status code %d: %s", statusCode, string(body))
	}

	return nil
}

// PublicKey returns the VAPID public key for the webpush dispatcher.
// Clients need this, so it's exposed via the BuildInfo endpoint.
func (n *Webpusher) PublicKey() string {
	return n.VAPIDPublicKey
}

// NoopWebpusher is a Dispatcher that always fails, returning Msg as
// the error. It is used as a fallback when VAPID key setup fails.
// The underlying error is not included to avoid leaking internal
// details (e.g. database errors) in API responses; it is logged at
// the call site instead.
type NoopWebpusher struct {
	Msg string
}

func (n *NoopWebpusher) Dispatch(context.Context, uuid.UUID, codersdk.WebpushMessage) error {
	return xerrors.New(n.Msg)
}

func (n *NoopWebpusher) Test(context.Context, codersdk.WebpushSubscription) error {
	return xerrors.New(n.Msg)
}

func (*NoopWebpusher) PublicKey() string {
	return ""
}

// newSSRFSafeHTTPClient returns an HTTP client that rejects connections to
// private, loopback, link-local, multicast, and unspecified IP addresses.
// This prevents DNS rebinding attacks where a hostname passes URL-level
// validation but resolves to an internal IP at dial time.
func newSSRFSafeHTTPClient() *http.Client {
	return &http.Client{
		Transport: &http.Transport{
			DialContext: (&net.Dialer{
				Control: func(_ string, address string, _ syscall.RawConn) error {
					host, _, err := net.SplitHostPort(address)
					if err != nil {
						return xerrors.Errorf("split host/port: %w", err)
					}
					ip, err := netip.ParseAddr(host)
					if err != nil {
						return xerrors.Errorf("parse resolved IP: %w", err)
					}
					if ip.IsPrivate() || ip.IsLoopback() || ip.IsLinkLocalUnicast() ||
						ip.IsLinkLocalMulticast() || ip.IsMulticast() ||
						ip.IsUnspecified() {
						return xerrors.Errorf(
							"webpush endpoint resolved to non-public address %s", ip.String(),
						)
					}
					return nil
				},
			}).DialContext,
		},
	}
}

// RegenerateVAPIDKeys regenerates the VAPID keys and deletes all existing
// push subscriptions as part of the transaction, as they are no longer valid.
func RegenerateVAPIDKeys(ctx context.Context, db database.Store) (newPrivateKey string, newPublicKey string, err error) {
	newPrivateKey, newPublicKey, err = webpush.GenerateVAPIDKeys()
	if err != nil {
		return "", "", xerrors.Errorf("generate new vapid keypair: %w", err)
	}

	if txErr := db.InTx(func(tx database.Store) error {
		if err := tx.DeleteAllWebpushSubscriptions(ctx); err != nil {
			return xerrors.Errorf("delete all webpush subscriptions: %w", err)
		}
		if err := tx.UpsertWebpushVAPIDKeys(ctx, database.UpsertWebpushVAPIDKeysParams{
			VapidPrivateKey: newPrivateKey,
			VapidPublicKey:  newPublicKey,
		}); err != nil {
			return xerrors.Errorf("upsert notification vapid key: %w", err)
		}
		return nil
	}, nil); txErr != nil {
		return "", "", xerrors.Errorf("regenerate vapid keypair: %w", txErr)
	}

	return newPrivateKey, newPublicKey, nil
}