tinyrelaycommit

Commit a5b469fc4059

browse files at this commit | history

1diff --git a/internal/daemon/http.go b/internal/daemon/http.go
2index 0af5d8b..a4b1f0c 100644
3--- a/internal/daemon/http.go
4+++ b/internal/daemon/http.go
5@@ -64,6 +64,10 @@ func (t *Tenant) ServeHTTP(w http.ResponseWriter, r *http.Request) {
6 t.sessionHTTP(w, r)
7 return
8 }
9+ if strings.HasPrefix(r.URL.Path, "/push/") {
10+ t.pushHTTP(w, r)
11+ return
12+ }
13 if t.tryBrowseHTTP(w, r) {
14 return
15 }
16diff --git a/internal/daemon/notifications.go b/internal/daemon/notifications.go
17index b234655..31013b8 100644
18--- a/internal/daemon/notifications.go
19+++ b/internal/daemon/notifications.go
20@@ -22,6 +22,7 @@ import (
21 const (
22 notificationBroadcast = "notification-broadcast"
23 notificationDelivery = "notification-delivery"
24+ notificationPush = "notification-push"
25 )
26
27 type notificationPayload struct {
28diff --git a/internal/daemon/replication_services.go b/internal/daemon/replication_services.go
29index e68a07b..93260b7 100644
30--- a/internal/daemon/replication_services.go
31+++ b/internal/daemon/replication_services.go
32@@ -199,7 +199,7 @@ func (p tenantReplicationProvider) Restore(ctx context.Context, state replicatio
33 if _, err := p.tenant.config.ApplyWithOptions(ctx, parsed, configport.ApplyOptions{MigrationOwner: identity.Owner}); err != nil {
34 return fmt.Errorf("daemon: legacy config apply: %w", err)
35 }
36- fresh, err := records.New(ctx, records.Config{Store: p.tenant.store, Policy: p.tenant.Policy, RelayURL: p.tenant.RelayURL(), GroupID: p.tenant.meta.Name, OnGenerated: p.tenant.generatedRecord, DeliverNotification: p.tenant.deliverNotification, SetPolicy: func(next policy.Policy) error { return p.tenant.applyPolicy(context.Background(), next) }})
37+ fresh, err := records.New(ctx, records.Config{Store: p.tenant.store, Policy: p.tenant.Policy, RelayURL: p.tenant.RelayURL(), GroupID: p.tenant.meta.Name, OnGenerated: p.tenant.generatedRecord, DeliverNotification: p.tenant.deliverNotification, PushNotification: p.tenant.enqueuePush, SetPolicy: func(next policy.Policy) error { return p.tenant.applyPolicy(context.Background(), next) }})
38 if err != nil {
39 return fmt.Errorf("daemon: refresh records after legacy config: %w", err)
40 }
41@@ -231,7 +231,7 @@ func (p tenantReplicationProvider) Restore(ctx context.Context, state replicatio
42 p.tenant.policy = next
43 p.tenant.mu.Unlock()
44 }
45- fresh, err := records.New(ctx, records.Config{Store: p.tenant.store, Policy: p.tenant.Policy, RelayURL: p.tenant.RelayURL(), GroupID: p.tenant.meta.Name, OnGenerated: p.tenant.generatedRecord, DeliverNotification: p.tenant.deliverNotification, SetPolicy: func(next policy.Policy) error { return p.tenant.applyPolicy(context.Background(), next) }})
46+ fresh, err := records.New(ctx, records.Config{Store: p.tenant.store, Policy: p.tenant.Policy, RelayURL: p.tenant.RelayURL(), GroupID: p.tenant.meta.Name, OnGenerated: p.tenant.generatedRecord, DeliverNotification: p.tenant.deliverNotification, PushNotification: p.tenant.enqueuePush, SetPolicy: func(next policy.Policy) error { return p.tenant.applyPolicy(context.Background(), next) }})
47 if err != nil {
48 return fmt.Errorf("daemon: refresh records identity: %w", err)
49 }
50diff --git a/internal/daemon/services.go b/internal/daemon/services.go
51index a3aea8d..9c37113 100644
52--- a/internal/daemon/services.go
53+++ b/internal/daemon/services.go
54@@ -39,6 +39,9 @@ func (t *Tenant) initServices(ctx context.Context) error {
55 if err := t.initSessions(ctx); err != nil {
56 return err
57 }
58+ if err := t.initPush(ctx); err != nil {
59+ return err
60+ }
61 t.blobs, err = blob.New(ctx, blob.Config{
62 Root: t.meta.Paths.Root, PublicURL: t.publicURL, Store: t.store, Authorize: t.authorizeBlob,
63 Limits: func() blob.Limits {
64@@ -96,7 +99,7 @@ func (t *Tenant) initServices(ctx context.Context) error {
65 if err != nil {
66 return err
67 }
68- t.records, err = records.New(ctx, records.Config{Store: t.store, Policy: t.Policy, RelayURL: t.RelayURL(), GroupID: t.meta.Name, OnGenerated: t.generatedRecord, DeliverNotification: t.deliverNotification, SetPolicy: func(next policy.Policy) error { return t.applyPolicy(context.Background(), next) }})
69+ t.records, err = records.New(ctx, records.Config{Store: t.store, Policy: t.Policy, RelayURL: t.RelayURL(), GroupID: t.meta.Name, OnGenerated: t.generatedRecord, DeliverNotification: t.deliverNotification, PushNotification: t.enqueuePush, SetPolicy: func(next policy.Policy) error { return t.applyPolicy(context.Background(), next) }})
70 if err != nil {
71 return err
72 }
73@@ -492,6 +495,7 @@ func (t *Tenant) workHandlers() map[string]work.Handler {
74 for kind, handler := range t.notificationHandlers() {
75 handlers[kind] = handler
76 }
77+ handlers[notificationPush] = t.handleNotificationPush
78 return handlers
79 }
80
81diff --git a/internal/daemon/tenant.go b/internal/daemon/tenant.go
82index 3deb80a..d0160a8 100644
83--- a/internal/daemon/tenant.go
84+++ b/internal/daemon/tenant.go
85@@ -27,6 +27,7 @@ import (
86 "github.com/FelineStateMachine/tinyrelay/internal/sites"
87 "github.com/FelineStateMachine/tinyrelay/internal/storage"
88 "github.com/FelineStateMachine/tinyrelay/internal/syncprotocol"
89+ "github.com/FelineStateMachine/tinyrelay/internal/webpush"
90 "github.com/FelineStateMachine/tinyrelay/internal/webui"
91 )
92
93@@ -65,6 +66,9 @@ type Tenant struct {
94 policy policy.Policy
95 maintenance maintenanceGate
96 gitLegacy *replication.LegacyCache
97+ pushMu sync.Mutex
98+ pushVAPID *webpush.Keys
99+ pushClient *http.Client
100 }
101
102 func newTenant(ctx context.Context, cfg tenantConfig) (*Tenant, error) {
103diff --git a/internal/daemon/webpush.go b/internal/daemon/webpush.go
104new file mode 100644
105index 0000000..9381479
106--- /dev/null
107+++ b/internal/daemon/webpush.go
108@@ -0,0 +1,268 @@
109+package daemon
110+
111+// Web Push delivers relay notifications to installed browsers. A member
112+// registers a device subscription with a signed request; the relay keeps one
113+// VAPID key pair per tenant and sends each notification summary encrypted to
114+// every registered device. Push services see only ciphertext.
115+
116+import (
117+ "context"
118+ "database/sql"
119+ "encoding/json"
120+ "errors"
121+ "fmt"
122+ "net/http"
123+ "strings"
124+ "time"
125+
126+ "github.com/FelineStateMachine/tinyrelay/internal/replication"
127+ "github.com/FelineStateMachine/tinyrelay/internal/storage"
128+ "github.com/FelineStateMachine/tinyrelay/internal/webpush"
129+ "github.com/FelineStateMachine/tinyrelay/internal/work"
130+ "github.com/google/uuid"
131+)
132+
133+const (
134+ pushSettingKey = "webpush.vapid"
135+ pushMaxPerPubkey = 8
136+ pushMaxFailures = 5
137+ pushTTL = 24 * time.Hour
138+ pushSummaryMaxLength = 160
139+)
140+
141+const pushSchema = `CREATE TABLE IF NOT EXISTS web_push(endpoint TEXT PRIMARY KEY, pubkey TEXT NOT NULL, subscription TEXT NOT NULL, created_at INTEGER NOT NULL, failures INTEGER NOT NULL DEFAULT 0);
142+CREATE INDEX IF NOT EXISTS web_push_pubkey ON web_push(pubkey);`
143+
144+type pushPayload struct {
145+ Recipient string `json:"recipient"`
146+ Kind string `json:"kind"`
147+ Subject string `json:"subject"`
148+ Text string `json:"text"`
149+}
150+
151+// pushMessage is what the service worker receives after decryption.
152+type pushMessage struct {
153+ Title string `json:"title"`
154+ Body string `json:"body"`
155+ URL string `json:"url"`
156+ Tag string `json:"tag"`
157+}
158+
159+func (t *Tenant) initPush(ctx context.Context) error {
160+ _, err := t.store.DB().ExecContext(ctx, pushSchema)
161+ return err
162+}
163+
164+// pushVAPIDKeys returns the tenant's VAPID pair, creating it on first use.
165+func (t *Tenant) pushVAPIDKeys(ctx context.Context) (webpush.Keys, error) {
166+ t.pushMu.Lock()
167+ defer t.pushMu.Unlock()
168+ if t.pushVAPID != nil {
169+ return *t.pushVAPID, nil
170+ }
171+ var encoded string
172+ err := t.store.GetSetting(ctx, pushSettingKey, &encoded)
173+ if err != nil && !errors.Is(err, sql.ErrNoRows) {
174+ return webpush.Keys{}, err
175+ }
176+ var keys webpush.Keys
177+ if encoded != "" {
178+ keys, err = webpush.ParseKeys(encoded)
179+ } else {
180+ keys, err = webpush.GenerateKeys()
181+ if err == nil {
182+ encoded, err = keys.Marshal()
183+ }
184+ if err == nil {
185+ err = t.store.PutSetting(ctx, pushSettingKey, encoded)
186+ }
187+ }
188+ if err != nil {
189+ return webpush.Keys{}, err
190+ }
191+ t.pushVAPID = &keys
192+ return keys, nil
193+}
194+
195+// pushSubject identifies the relay operator to push services.
196+func (t *Tenant) pushSubject() string {
197+ if contact := strings.TrimSpace(t.Policy().Contact); contact != "" && strings.Contains(contact, "@") && !strings.Contains(contact, ":") {
198+ return "mailto:" + contact
199+ }
200+ return t.publicURL
201+}
202+
203+func (t *Tenant) pushHTTP(w http.ResponseWriter, r *http.Request) {
204+ switch {
205+ case r.URL.Path == "/push/key" && r.Method == http.MethodGet:
206+ keys, err := t.pushVAPIDKeys(r.Context())
207+ if err != nil {
208+ writeJSON(w, 500, map[string]string{"error": "push keys unavailable"})
209+ return
210+ }
211+ writeJSON(w, 200, map[string]string{"key": keys.PublicKey()})
212+ case (r.URL.Path == "/push/subscribe" || r.URL.Path == "/push/unsubscribe") && r.Method == http.MethodPost:
213+ // Registration is a signed mutation like every other change made
214+ // from the browser; the session cookie alone is not enough.
215+ actor, err := t.resolveUIActor(r)
216+ if err != nil || actor == "" {
217+ writeJSON(w, 401, map[string]string{"error": "auth-required: sign this request"})
218+ return
219+ }
220+ body, err := t.requestBody(w, r)
221+ if err != nil {
222+ writeJSON(w, 400, map[string]string{"error": err.Error()})
223+ return
224+ }
225+ var subscription webpush.Subscription
226+ if err := json.Unmarshal(body, &subscription); err != nil || subscription.Validate() != nil {
227+ writeJSON(w, 400, map[string]string{"error": "invalid: push subscription"})
228+ return
229+ }
230+ if r.URL.Path == "/push/unsubscribe" {
231+ if _, err := t.store.DB().ExecContext(r.Context(), "DELETE FROM web_push WHERE endpoint=? AND pubkey=?", subscription.Endpoint, actor); err != nil {
232+ writeJSON(w, 500, map[string]string{"error": "unsubscribe failed"})
233+ return
234+ }
235+ writeJSON(w, 200, map[string]any{"ok": true})
236+ return
237+ }
238+ if role, err := t.community.Role(r.Context(), actor); err != nil || (role == "" && actor != t.Policy().Owner) {
239+ writeJSON(w, 403, map[string]string{"error": "restricted: notifications are limited to relay members"})
240+ return
241+ }
242+ if err := t.savePushSubscription(r.Context(), actor, subscription); err != nil {
243+ writeJSON(w, statusForPush(err), map[string]string{"error": err.Error()})
244+ return
245+ }
246+ writeJSON(w, 200, map[string]any{"ok": true})
247+ default:
248+ http.NotFound(w, r)
249+ }
250+}
251+
252+func statusForPush(err error) int {
253+ if strings.HasPrefix(err.Error(), "invalid:") {
254+ return 400
255+ }
256+ return 500
257+}
258+
259+func (t *Tenant) savePushSubscription(ctx context.Context, actor string, subscription webpush.Subscription) error {
260+ raw, err := json.Marshal(subscription)
261+ if err != nil {
262+ return err
263+ }
264+ return t.store.WithTx(ctx, func(tx *sql.Tx) error {
265+ var count int
266+ if err := tx.QueryRowContext(ctx, "SELECT count(*) FROM web_push WHERE pubkey=? AND endpoint<>?", actor, subscription.Endpoint).Scan(&count); err != nil {
267+ return err
268+ }
269+ if count >= pushMaxPerPubkey {
270+ return fmt.Errorf("invalid: at most %d devices per key", pushMaxPerPubkey)
271+ }
272+ _, err := tx.ExecContext(ctx, "INSERT INTO web_push(endpoint,pubkey,subscription,created_at,failures) VALUES(?,?,?,?,0) ON CONFLICT(endpoint) DO UPDATE SET pubkey=excluded.pubkey, subscription=excluded.subscription, failures=0", subscription.Endpoint, actor, string(raw), time.Now().Unix())
273+ return err
274+ })
275+}
276+
277+// enqueuePush records one notification summary for device delivery. It is
278+// suitable for records.Config.PushNotification and does nothing for
279+// recipients without registered devices.
280+func (t *Tenant) enqueuePush(ctx context.Context, recipient, kind, subject, text string) error {
281+ var count int
282+ if err := t.store.DB().QueryRowContext(ctx, "SELECT count(*) FROM web_push WHERE pubkey=?", recipient).Scan(&count); err != nil || count == 0 {
283+ return err
284+ }
285+ if len(text) > pushSummaryMaxLength {
286+ text = text[:pushSummaryMaxLength]
287+ }
288+ payload, err := json.Marshal(pushPayload{Recipient: recipient, Kind: kind, Subject: subject, Text: text})
289+ if err != nil {
290+ return err
291+ }
292+ return t.store.WithTx(ctx, func(tx *sql.Tx) error {
293+ return storage.AddIntents(ctx, tx, []storage.Intent{{Kind: notificationPush, EventID: uuid.NewString(), Target: recipient, Payload: string(payload)}}, time.Now().Unix())
294+ })
295+}
296+
297+func (t *Tenant) handleNotificationPush(ctx context.Context, intent work.Intent) error {
298+ var payload pushPayload
299+ if err := json.Unmarshal([]byte(intent.Payload), &payload); err != nil || payload.Recipient != intent.Target {
300+ return errors.New("notification-push: invalid payload")
301+ }
302+ rows, err := t.store.DB().QueryContext(ctx, "SELECT endpoint, subscription, failures FROM web_push WHERE pubkey=?", payload.Recipient)
303+ if err != nil {
304+ return err
305+ }
306+ type device struct {
307+ endpoint string
308+ subscription webpush.Subscription
309+ failures int
310+ }
311+ var devices []device
312+ for rows.Next() {
313+ var item device
314+ var raw string
315+ if err := rows.Scan(&item.endpoint, &raw, &item.failures); err != nil {
316+ rows.Close()
317+ return err
318+ }
319+ if json.Unmarshal([]byte(raw), &item.subscription) == nil {
320+ devices = append(devices, item)
321+ }
322+ }
323+ rows.Close()
324+ if len(devices) == 0 {
325+ return nil
326+ }
327+ keys, err := t.pushVAPIDKeys(ctx)
328+ if err != nil {
329+ return err
330+ }
331+ message, err := json.Marshal(t.pushMessage(payload))
332+ if err != nil {
333+ return err
334+ }
335+ client := t.pushClient
336+ if client == nil {
337+ client = replication.NewPinnedClient(10 * time.Second)
338+ }
339+ var result error
340+ for _, item := range devices {
341+ sent, sendErr := webpush.Send(ctx, client, keys, t.pushSubject(), item.subscription, message, pushTTL)
342+ switch {
343+ case sendErr == nil:
344+ _, _ = t.store.DB().ExecContext(ctx, "UPDATE web_push SET failures=0 WHERE endpoint=?", item.endpoint)
345+ case sent.Gone || item.failures+1 >= pushMaxFailures:
346+ _, _ = t.store.DB().ExecContext(ctx, "DELETE FROM web_push WHERE endpoint=?", item.endpoint)
347+ default:
348+ _, _ = t.store.DB().ExecContext(ctx, "UPDATE web_push SET failures=failures+1 WHERE endpoint=?", item.endpoint)
349+ result = errors.Join(result, sendErr)
350+ }
351+ }
352+ return result
353+}
354+
355+// pushMessage turns a notification summary into what the device shows.
356+func (t *Tenant) pushMessage(payload pushPayload) pushMessage {
357+ title := t.Policy().Name
358+ if title == "" {
359+ title = t.meta.Name
360+ }
361+ body := payload.Text
362+ if body == "" {
363+ body = map[string]string{
364+ "reports": "A new report needs moderation.", "jobs": "A background job finished.",
365+ "succession": "Relay succession changed.", "digest": "Your relay digest is ready.",
366+ "test": "This is a test notification.", "succession-transfer": "Relay ownership was transferred.",
367+ }[payload.Kind]
368+ }
369+ if body == "" {
370+ body = "New relay notification."
371+ }
372+ if payload.Subject != "" && payload.Subject != "relay" && !strings.EqualFold(payload.Subject, payload.Kind) {
373+ title = title + ": " + payload.Subject
374+ }
375+ return pushMessage{Title: title, Body: body, URL: strings.TrimRight(t.publicURL, "/") + "/inbox", Tag: "tiny-" + payload.Kind}
376+}
377diff --git a/internal/daemon/webpush_test.go b/internal/daemon/webpush_test.go
378new file mode 100644
379index 0000000..af9b52f
380--- /dev/null
381+++ b/internal/daemon/webpush_test.go
382@@ -0,0 +1,113 @@
383+package daemon
384+
385+import (
386+ "context"
387+ "crypto/ecdh"
388+ "crypto/rand"
389+ "encoding/base64"
390+ "encoding/json"
391+ "net/http"
392+ "net/http/httptest"
393+ "strings"
394+ "testing"
395+
396+ "github.com/FelineStateMachine/tinyrelay/internal/webpush"
397+ "github.com/FelineStateMachine/tinyrelay/internal/work"
398+)
399+
400+func testSubscription(t *testing.T, endpoint string) webpush.Subscription {
401+ t.Helper()
402+ private, err := ecdh.P256().GenerateKey(rand.Reader)
403+ if err != nil {
404+ t.Fatal(err)
405+ }
406+ auth := make([]byte, 16)
407+ rand.Read(auth)
408+ var subscription webpush.Subscription
409+ subscription.Endpoint = endpoint
410+ subscription.Keys.P256dh = base64.RawURLEncoding.EncodeToString(private.PublicKey().Bytes())
411+ subscription.Keys.Auth = base64.RawURLEncoding.EncodeToString(auth)
412+ return subscription
413+}
414+
415+func TestPushSubscriptionRequiresSignatureAndDeliversNotifications(t *testing.T) {
416+ a, tenant := testTenant(t)
417+ received := make(chan http.Header, 4)
418+ status := http.StatusCreated
419+ service := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
420+ received <- r.Header.Clone()
421+ w.WriteHeader(status)
422+ }))
423+ defer service.Close()
424+ tenant.pushClient = service.Client()
425+ subscription := testSubscription(t, service.URL+"/send/device-1")
426+ body, _ := json.Marshal(subscription)
427+
428+ keyReq := httptest.NewRequest("GET", "http://relay.test/push/key", nil)
429+ keyRes := httptest.NewRecorder()
430+ a.ServeHTTP(keyRes, keyReq)
431+ var key map[string]string
432+ if keyRes.Code != 200 || json.NewDecoder(keyRes.Body).Decode(&key) != nil || len(key["key"]) < 80 {
433+ t.Fatalf("push key %d %s", keyRes.Code, keyRes.Body.String())
434+ }
435+
436+ unsigned := httptest.NewRequest("POST", "http://relay.test/push/subscribe", strings.NewReader(string(body)))
437+ unsignedRes := httptest.NewRecorder()
438+ a.ServeHTTP(unsignedRes, unsigned)
439+ if unsignedRes.Code != 401 {
440+ t.Fatalf("unsigned subscribe %d", unsignedRes.Code)
441+ }
442+
443+ signed := httptest.NewRequest("POST", "http://relay.test/push/subscribe", strings.NewReader(string(body)))
444+ signRequest(t, signed, string(body))
445+ signedRes := httptest.NewRecorder()
446+ a.ServeHTTP(signedRes, signed)
447+ if signedRes.Code != 200 {
448+ t.Fatalf("subscribe %d %s", signedRes.Code, signedRes.Body.String())
449+ }
450+
451+ ctx := context.Background()
452+ owner := tenant.Policy().Owner
453+ if err := tenant.enqueuePush(ctx, owner, "test", "relay", "hello from the relay"); err != nil {
454+ t.Fatal(err)
455+ }
456+ var payload string
457+ if err := tenant.store.DB().QueryRowContext(ctx, "SELECT payload FROM work_intents WHERE kind=? AND target=?", notificationPush, owner).Scan(&payload); err != nil {
458+ t.Fatalf("push intent not queued: %v", err)
459+ }
460+ if err := tenant.handleNotificationPush(ctx, work.Intent{Kind: notificationPush, Target: owner, Payload: payload}); err != nil {
461+ t.Fatal(err)
462+ }
463+ headers := <-received
464+ if headers.Get("Content-Encoding") != "aes128gcm" || !strings.HasPrefix(headers.Get("Authorization"), "vapid t=") || headers.Get("TTL") == "" {
465+ t.Fatalf("push headers %v", headers)
466+ }
467+
468+ // A subscription the push service reports gone is removed.
469+ status = http.StatusGone
470+ if err := tenant.handleNotificationPush(ctx, work.Intent{Kind: notificationPush, Target: owner, Payload: payload}); err != nil {
471+ t.Fatalf("gone device should not fail the intent: %v", err)
472+ }
473+ <-received
474+ var remaining int
475+ if err := tenant.store.DB().QueryRowContext(ctx, "SELECT count(*) FROM web_push").Scan(&remaining); err != nil || remaining != 0 {
476+ t.Fatalf("gone subscription kept: %d %v", remaining, err)
477+ }
478+ // Without devices, summaries are not queued at all.
479+ before := countIntents(t, tenant, notificationPush)
480+ if err := tenant.enqueuePush(ctx, owner, "test", "relay", ""); err != nil {
481+ t.Fatal(err)
482+ }
483+ if countIntents(t, tenant, notificationPush) != before {
484+ t.Fatal("push intent queued without a device")
485+ }
486+}
487+
488+func countIntents(t *testing.T, tenant *Tenant, kind string) int {
489+ t.Helper()
490+ var count int
491+ if err := tenant.store.DB().QueryRowContext(context.Background(), "SELECT count(*) FROM work_intents WHERE kind=?", kind).Scan(&count); err != nil {
492+ t.Fatal(err)
493+ }
494+ return count
495+}
496diff --git a/internal/records/records.go b/internal/records/records.go
497index 1aa9b87..d1350f0 100644
498--- a/internal/records/records.go
499+++ b/internal/records/records.go
500@@ -32,7 +32,10 @@ type Config struct {
501 GroupID string
502 OnGenerated func(context.Context, event.Event) error
503 DeliverNotification func(context.Context, event.Event, string) error
504- OnTransfer func(context.Context, string, string) error
505+ // PushNotification receives a plaintext summary of each notification for
506+ // device delivery. The gift wrap itself is opaque to the relay once sealed.
507+ PushNotification func(context.Context, string, string, string, string) error
508+ OnTransfer func(context.Context, string, string) error
509 }
510
511 type Service struct {
512@@ -44,6 +47,7 @@ type Service struct {
513 secret string
514 onGenerated func(context.Context, event.Event) error
515 deliverNotification func(context.Context, event.Event, string) error
516+ pushNotification func(context.Context, string, string, string, string) error
517 onTransfer func(context.Context, string, string) error
518 mu sync.Mutex
519 }
520@@ -86,7 +90,7 @@ func New(ctx context.Context, cfg Config) (*Service, error) {
521 if groupID == "" {
522 groupID = cfg.RelayURL
523 }
524- return &Service{store: cfg.Store, policy: cfg.Policy, setPolicy: cfg.SetPolicy, relayURL: cfg.RelayURL, groupID: groupID, secret: secret, onGenerated: cfg.OnGenerated, deliverNotification: cfg.DeliverNotification, onTransfer: cfg.OnTransfer}, nil
525+ return &Service{store: cfg.Store, policy: cfg.Policy, setPolicy: cfg.SetPolicy, relayURL: cfg.RelayURL, groupID: groupID, secret: secret, onGenerated: cfg.OnGenerated, deliverNotification: cfg.DeliverNotification, pushNotification: cfg.PushNotification, onTransfer: cfg.OnTransfer}, nil
526 }
527
528 func (s *Service) group() string { return s.groupID }
529@@ -1386,7 +1390,10 @@ func (s *Service) notify(ctx context.Context, typ, owner, target string, now int
530 return err
531 }
532 if s.deliverNotification != nil {
533- return s.deliverNotification(ctx, wrapped, target)
534+ if err := s.deliverNotification(ctx, wrapped, target); err != nil {
535+ return err
536+ }
537+ return s.push(ctx, target, typ, "relay", "")
538 }
539 if s.onGenerated != nil {
540 return s.onGenerated(ctx, wrapped)
541@@ -1394,6 +1401,13 @@ func (s *Service) notify(ctx context.Context, typ, owner, target string, now int
542 return err
543 }
544
545+func (s *Service) push(ctx context.Context, recipient, kind, subject, text string) error {
546+ if s.pushNotification == nil {
547+ return nil
548+ }
549+ return s.pushNotification(ctx, recipient, kind, subject, text)
550+}
551+
552 // Notify applies the source notification preferences and emits a durable
553 // gift-wrap through OnGenerated. The host callback is responsible for saving
554 // the event and enqueueing remote delivery in the same transaction.
555@@ -1423,7 +1437,10 @@ func (s *Service) notifyText(ctx context.Context, kind, text, subject, target st
556 return err
557 }
558 if s.deliverNotification != nil {
559- return s.deliverNotification(ctx, wrapped, target)
560+ if err := s.deliverNotification(ctx, wrapped, target); err != nil {
561+ return err
562+ }
563+ return s.push(ctx, target, kind, subject, text)
564 }
565 if s.onGenerated != nil {
566 return s.onGenerated(ctx, wrapped)
567diff --git a/internal/replication/callback.go b/internal/replication/callback.go
568index d04691a..011e186 100644
569--- a/internal/replication/callback.go
570+++ b/internal/replication/callback.go
571@@ -133,6 +133,10 @@ func (c CallbackClient) Post(ctx context.Context, registration PushRegistration,
572 return nil
573 }
574
575+// NewPinnedClient returns an HTTPS client that refuses private addresses and
576+// redirects, for calls to third-party services such as push endpoints.
577+func NewPinnedClient(timeout time.Duration) *http.Client { return pinnedClient(nil, timeout) }
578+
579 func pinnedClient(resolver Resolver, timeout time.Duration) *http.Client {
580 if timeout <= 0 {
581 timeout = 5 * time.Second
582diff --git a/internal/telemetry/telemetry.go b/internal/telemetry/telemetry.go
583index 0ce3936..806b068 100644
584--- a/internal/telemetry/telemetry.go
585+++ b/internal/telemetry/telemetry.go
586@@ -239,7 +239,7 @@ func latencyBuckets() []float64 {
587 var operationNames = map[string]struct{}{"http": {}, "websocket": {}, "publish": {}, "subscribe": {}, "query": {}, "count": {}, "fanout": {}, "delivery": {}, "commit": {}, "save": {}, "parse": {}, "authorize": {}, "upgrade": {}, "callback": {}, "delivery-discovery": {}, "notification-broadcast": {}, "browserepos": {}, "browserepo": {}, "browsefiles": {}, "browsefile": {}, "browsestatus": {}, "browseissues": {}, "browsepulls": {}, "browseissue": {}, "browsepull": {}, "browse-download": {}}
588 var outcomeNames = map[string]struct{}{"success": {}, "ok": {}, "invalid": {}, "unauthorized": {}, "timeout": {}, "busy": {}, "closed": {}, "error": {}}
589 var queueNames = map[string]struct{}{"inbox": {}, "outbox": {}, "write": {}, "delivery": {}}
590-var workNames = map[string]struct{}{"delivery": {}, "push": {}, "replication": {}, "webhook": {}, "job": {}, "records-projection": {}, "site-mirror": {}, "git-metadata": {}, "catalog-owner": {}, "callback": {}, "delivery-discovery": {}, "notification-delivery": {}, "notification-broadcast": {}, "notification": {}}
591+var workNames = map[string]struct{}{"delivery": {}, "push": {}, "replication": {}, "webhook": {}, "job": {}, "records-projection": {}, "site-mirror": {}, "git-metadata": {}, "catalog-owner": {}, "callback": {}, "delivery-discovery": {}, "notification-delivery": {}, "notification-broadcast": {}, "notification-push": {}, "notification": {}}
592
593 func bounded(value string, allowed map[string]struct{}) string {
594 if _, ok := allowed[value]; ok {
595diff --git a/internal/webpush/webpush.go b/internal/webpush/webpush.go
596new file mode 100644
597index 0000000..056bdd4
598--- /dev/null
599+++ b/internal/webpush/webpush.go
600@@ -0,0 +1,242 @@
601+// Package webpush sends Web Push messages: RFC 8291 payload encryption with
602+// the aes128gcm content encoding and RFC 8292 VAPID authorization. It uses
603+// only the standard library so the relay keeps its dependency footprint.
604+package webpush
605+
606+import (
607+ "bytes"
608+ "context"
609+ "crypto/aes"
610+ "crypto/cipher"
611+ "crypto/ecdh"
612+ "crypto/ecdsa"
613+ "crypto/elliptic"
614+ "crypto/hkdf"
615+ "crypto/rand"
616+ "crypto/sha256"
617+ "crypto/x509"
618+ "encoding/base64"
619+ "encoding/binary"
620+ "encoding/json"
621+ "errors"
622+ "fmt"
623+ "io"
624+ "net/http"
625+ "net/url"
626+ "strconv"
627+ "time"
628+)
629+
630+// Subscription is what the browser's PushManager hands to the page.
631+type Subscription struct {
632+ Endpoint string `json:"endpoint"`
633+ Keys struct {
634+ P256dh string `json:"p256dh"`
635+ Auth string `json:"auth"`
636+ } `json:"keys"`
637+}
638+
639+// Validate checks the shape of a subscription before it is stored.
640+func (s Subscription) Validate() error {
641+ parsed, err := url.Parse(s.Endpoint)
642+ if err != nil || parsed.Scheme != "https" || parsed.Hostname() == "" || len(s.Endpoint) > 2048 {
643+ return errors.New("invalid: push endpoint must be an HTTPS URL")
644+ }
645+ if key, err := decode(s.Keys.P256dh); err != nil || len(key) != 65 || key[0] != 4 {
646+ return errors.New("invalid: push p256dh key must be an uncompressed P-256 point")
647+ }
648+ if auth, err := decode(s.Keys.Auth); err != nil || len(auth) != 16 {
649+ return errors.New("invalid: push auth secret must be 16 bytes")
650+ }
651+ return nil
652+}
653+
654+// Keys is the VAPID key pair identifying this relay to push services.
655+type Keys struct{ private *ecdsa.PrivateKey }
656+
657+// GenerateKeys creates a fresh P-256 VAPID key pair.
658+func GenerateKeys() (Keys, error) {
659+ private, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
660+ if err != nil {
661+ return Keys{}, err
662+ }
663+ return Keys{private: private}, nil
664+}
665+
666+// Marshal serializes the private key for the settings table.
667+func (k Keys) Marshal() (string, error) {
668+ der, err := x509.MarshalPKCS8PrivateKey(k.private)
669+ if err != nil {
670+ return "", err
671+ }
672+ return base64.RawURLEncoding.EncodeToString(der), nil
673+}
674+
675+// ParseKeys restores a key pair from Marshal output.
676+func ParseKeys(encoded string) (Keys, error) {
677+ der, err := decode(encoded)
678+ if err != nil {
679+ return Keys{}, err
680+ }
681+ parsed, err := x509.ParsePKCS8PrivateKey(der)
682+ if err != nil {
683+ return Keys{}, err
684+ }
685+ private, ok := parsed.(*ecdsa.PrivateKey)
686+ if !ok || private.Curve != elliptic.P256() {
687+ return Keys{}, errors.New("webpush: VAPID key is not a P-256 key")
688+ }
689+ return Keys{private: private}, nil
690+}
691+
692+// PublicKey is the applicationServerKey the browser subscribes with.
693+func (k Keys) PublicKey() string {
694+ return base64.RawURLEncoding.EncodeToString(elliptic.Marshal(elliptic.P256(), k.private.PublicKey.X, k.private.PublicKey.Y))
695+}
696+
697+// Authorization builds the VAPID header for one push service origin. The
698+// token is an ES256 JWT with the endpoint origin as its audience.
699+func (k Keys) Authorization(endpoint, subject string, now time.Time) (string, error) {
700+ parsed, err := url.Parse(endpoint)
701+ if err != nil {
702+ return "", err
703+ }
704+ header := base64.RawURLEncoding.EncodeToString([]byte(`{"typ":"JWT","alg":"ES256"}`))
705+ claims, err := json.Marshal(map[string]any{"aud": parsed.Scheme + "://" + parsed.Host, "exp": now.Add(12 * time.Hour).Unix(), "sub": subject})
706+ if err != nil {
707+ return "", err
708+ }
709+ signing := header + "." + base64.RawURLEncoding.EncodeToString(claims)
710+ digest := sha256.Sum256([]byte(signing))
711+ r, s, err := ecdsa.Sign(rand.Reader, k.private, digest[:])
712+ if err != nil {
713+ return "", err
714+ }
715+ signature := make([]byte, 64)
716+ r.FillBytes(signature[:32])
717+ s.FillBytes(signature[32:])
718+ token := signing + "." + base64.RawURLEncoding.EncodeToString(signature)
719+ return "vapid t=" + token + ", k=" + k.PublicKey(), nil
720+}
721+
722+// Encrypt produces an aes128gcm body for the subscription. Each call uses a
723+// fresh ephemeral key and salt, so the same payload never encrypts twice
724+// the same way.
725+func Encrypt(subscription Subscription, plaintext []byte) ([]byte, error) {
726+ if len(plaintext) > 3993 {
727+ return nil, errors.New("webpush: payload exceeds one record")
728+ }
729+ receiverPublic, err := decode(subscription.Keys.P256dh)
730+ if err != nil {
731+ return nil, err
732+ }
733+ auth, err := decode(subscription.Keys.Auth)
734+ if err != nil {
735+ return nil, err
736+ }
737+ curve := ecdh.P256()
738+ receiver, err := curve.NewPublicKey(receiverPublic)
739+ if err != nil {
740+ return nil, fmt.Errorf("webpush: subscriber key: %w", err)
741+ }
742+ ephemeral, err := curve.GenerateKey(rand.Reader)
743+ if err != nil {
744+ return nil, err
745+ }
746+ shared, err := ephemeral.ECDH(receiver)
747+ if err != nil {
748+ return nil, err
749+ }
750+ senderPublic := ephemeral.PublicKey().Bytes()
751+ salt := make([]byte, 16)
752+ if _, err := io.ReadFull(rand.Reader, salt); err != nil {
753+ return nil, err
754+ }
755+ return encrypt(plaintext, shared, receiverPublic, senderPublic, auth, salt)
756+}
757+
758+func encrypt(plaintext, shared, receiverPublic, senderPublic, auth, salt []byte) ([]byte, error) {
759+ keyInfo := append(append([]byte("WebPush: info\x00"), receiverPublic...), senderPublic...)
760+ ikm, err := hkdf.Key(sha256.New, shared, auth, string(keyInfo), 32)
761+ if err != nil {
762+ return nil, err
763+ }
764+ contentKey, err := hkdf.Key(sha256.New, ikm, salt, "Content-Encoding: aes128gcm\x00", 16)
765+ if err != nil {
766+ return nil, err
767+ }
768+ nonce, err := hkdf.Key(sha256.New, ikm, salt, "Content-Encoding: nonce\x00", 12)
769+ if err != nil {
770+ return nil, err
771+ }
772+ block, err := aes.NewCipher(contentKey)
773+ if err != nil {
774+ return nil, err
775+ }
776+ gcm, err := cipher.NewGCM(block)
777+ if err != nil {
778+ return nil, err
779+ }
780+ record := gcm.Seal(nil, nonce, append(append([]byte{}, plaintext...), 2), nil)
781+ var body bytes.Buffer
782+ body.Write(salt)
783+ _ = binary.Write(&body, binary.BigEndian, uint32(4096))
784+ body.WriteByte(byte(len(senderPublic)))
785+ body.Write(senderPublic)
786+ body.Write(record)
787+ return body.Bytes(), nil
788+}
789+
790+// Result reports the push service's answer for one message.
791+type Result struct {
792+ Status int
793+ // Gone means the subscription no longer exists and should be removed.
794+ Gone bool
795+}
796+
797+// Send delivers one encrypted message. A nil client uses http.DefaultClient;
798+// callers that reach the public internet should pass a client that refuses
799+// private addresses.
800+func Send(ctx context.Context, client *http.Client, keys Keys, subject string, subscription Subscription, payload []byte, ttl time.Duration) (Result, error) {
801+ body, err := Encrypt(subscription, payload)
802+ if err != nil {
803+ return Result{}, err
804+ }
805+ authorization, err := keys.Authorization(subscription.Endpoint, subject, time.Now())
806+ if err != nil {
807+ return Result{}, err
808+ }
809+ request, err := http.NewRequestWithContext(ctx, http.MethodPost, subscription.Endpoint, bytes.NewReader(body))
810+ if err != nil {
811+ return Result{}, err
812+ }
813+ request.Header.Set("Content-Type", "application/octet-stream")
814+ request.Header.Set("Content-Encoding", "aes128gcm")
815+ request.Header.Set("TTL", strconv.Itoa(int(ttl.Seconds())))
816+ request.Header.Set("Urgency", "normal")
817+ request.Header.Set("Authorization", authorization)
818+ if client == nil {
819+ client = http.DefaultClient
820+ }
821+ response, err := client.Do(request)
822+ if err != nil {
823+ return Result{}, err
824+ }
825+ defer response.Body.Close()
826+ _, _ = io.Copy(io.Discard, io.LimitReader(response.Body, 4096))
827+ result := Result{Status: response.StatusCode, Gone: response.StatusCode == http.StatusNotFound || response.StatusCode == http.StatusGone}
828+ if response.StatusCode < 200 || response.StatusCode >= 300 {
829+ return result, fmt.Errorf("webpush: push service returned HTTP %d", response.StatusCode)
830+ }
831+ return result, nil
832+}
833+
834+func decode(value string) ([]byte, error) {
835+ if raw, err := base64.RawURLEncoding.DecodeString(value); err == nil {
836+ return raw, nil
837+ }
838+ if raw, err := base64.URLEncoding.DecodeString(value); err == nil {
839+ return raw, nil
840+ }
841+ return base64.StdEncoding.DecodeString(value)
842+}
843diff --git a/internal/webpush/webpush_test.go b/internal/webpush/webpush_test.go
844new file mode 100644
845index 0000000..aae7e94
846--- /dev/null
847+++ b/internal/webpush/webpush_test.go
848@@ -0,0 +1,161 @@
849+package webpush
850+
851+import (
852+ "bytes"
853+ "context"
854+ "crypto/aes"
855+ "crypto/cipher"
856+ "crypto/ecdh"
857+ "crypto/ecdsa"
858+ "crypto/elliptic"
859+ "crypto/hkdf"
860+ "crypto/rand"
861+ "crypto/sha256"
862+ "encoding/base64"
863+ "encoding/binary"
864+ "encoding/json"
865+ "math/big"
866+ "net/http"
867+ "net/http/httptest"
868+ "strings"
869+ "testing"
870+ "time"
871+)
872+
873+func subscriber(t *testing.T) (Subscription, *ecdh.PrivateKey, []byte) {
874+ t.Helper()
875+ private, err := ecdh.P256().GenerateKey(rand.Reader)
876+ if err != nil {
877+ t.Fatal(err)
878+ }
879+ auth := make([]byte, 16)
880+ rand.Read(auth)
881+ var subscription Subscription
882+ subscription.Endpoint = "https://push.example/send/abc"
883+ subscription.Keys.P256dh = base64.RawURLEncoding.EncodeToString(private.PublicKey().Bytes())
884+ subscription.Keys.Auth = base64.RawURLEncoding.EncodeToString(auth)
885+ return subscription, private, auth
886+}
887+
888+// decrypt mirrors the browser side of RFC 8291 so the test proves the body
889+// is readable by a real subscriber.
890+func decrypt(t *testing.T, body []byte, receiver *ecdh.PrivateKey, auth []byte) []byte {
891+ t.Helper()
892+ salt := body[:16]
893+ recordSize := binary.BigEndian.Uint32(body[16:20])
894+ idLength := int(body[20])
895+ senderPublic := body[21 : 21+idLength]
896+ record := body[21+idLength:]
897+ if recordSize != 4096 || idLength != 65 {
898+ t.Fatalf("header rs=%d idlen=%d", recordSize, idLength)
899+ }
900+ sender, err := ecdh.P256().NewPublicKey(senderPublic)
901+ if err != nil {
902+ t.Fatal(err)
903+ }
904+ shared, err := receiver.ECDH(sender)
905+ if err != nil {
906+ t.Fatal(err)
907+ }
908+ keyInfo := append(append([]byte("WebPush: info\x00"), receiver.PublicKey().Bytes()...), senderPublic...)
909+ ikm, _ := hkdf.Key(sha256.New, shared, auth, string(keyInfo), 32)
910+ key, _ := hkdf.Key(sha256.New, ikm, salt, "Content-Encoding: aes128gcm\x00", 16)
911+ nonce, _ := hkdf.Key(sha256.New, ikm, salt, "Content-Encoding: nonce\x00", 12)
912+ block, _ := aes.NewCipher(key)
913+ gcm, _ := cipher.NewGCM(block)
914+ plain, err := gcm.Open(nil, nonce, record, nil)
915+ if err != nil {
916+ t.Fatal(err)
917+ }
918+ if plain[len(plain)-1] != 2 {
919+ t.Fatalf("missing final record delimiter: %x", plain[len(plain)-1])
920+ }
921+ return plain[:len(plain)-1]
922+}
923+
924+func TestEncryptRoundTripsForASubscriber(t *testing.T) {
925+ subscription, private, auth := subscriber(t)
926+ if err := subscription.Validate(); err != nil {
927+ t.Fatal(err)
928+ }
929+ payload := []byte(`{"title":"tiny","body":"hello"}`)
930+ first, err := Encrypt(subscription, payload)
931+ if err != nil {
932+ t.Fatal(err)
933+ }
934+ second, _ := Encrypt(subscription, payload)
935+ if bytes.Equal(first, second) {
936+ t.Fatal("encryption reused its salt or ephemeral key")
937+ }
938+ if got := decrypt(t, first, private, auth); !bytes.Equal(got, payload) {
939+ t.Fatalf("decrypted %q", got)
940+ }
941+ if _, err := Encrypt(subscription, bytes.Repeat([]byte("x"), 4000)); err == nil {
942+ t.Fatal("oversized payload accepted")
943+ }
944+}
945+
946+func TestVAPIDAuthorizationVerifies(t *testing.T) {
947+ keys, err := GenerateKeys()
948+ if err != nil {
949+ t.Fatal(err)
950+ }
951+ encoded, err := keys.Marshal()
952+ if err != nil {
953+ t.Fatal(err)
954+ }
955+ restored, err := ParseKeys(encoded)
956+ if err != nil || restored.PublicKey() != keys.PublicKey() {
957+ t.Fatalf("restore keys: %v", err)
958+ }
959+ now := time.Unix(1_700_000_000, 0)
960+ header, err := keys.Authorization("https://push.example/send/abc?x=1", "mailto:owner@example", now)
961+ if err != nil {
962+ t.Fatal(err)
963+ }
964+ if !strings.HasPrefix(header, "vapid t=") || !strings.Contains(header, ", k="+keys.PublicKey()) {
965+ t.Fatalf("header %q", header)
966+ }
967+ token := strings.TrimPrefix(strings.SplitN(header, ", k=", 2)[0], "vapid t=")
968+ parts := strings.Split(token, ".")
969+ if len(parts) != 3 {
970+ t.Fatalf("token %q", token)
971+ }
972+ claimsRaw, _ := base64.RawURLEncoding.DecodeString(parts[1])
973+ var claims map[string]any
974+ if err := json.Unmarshal(claimsRaw, &claims); err != nil || claims["aud"] != "https://push.example" || claims["sub"] != "mailto:owner@example" || int64(claims["exp"].(float64)) != now.Add(12*time.Hour).Unix() {
975+ t.Fatalf("claims %s %v", claimsRaw, err)
976+ }
977+ signature, _ := base64.RawURLEncoding.DecodeString(parts[2])
978+ digest := sha256.Sum256([]byte(parts[0] + "." + parts[1]))
979+ public, _ := base64.RawURLEncoding.DecodeString(keys.PublicKey())
980+ x, y := elliptic.Unmarshal(elliptic.P256(), public)
981+ if !ecdsa.Verify(&ecdsa.PublicKey{Curve: elliptic.P256(), X: x, Y: y}, digest[:], new(big.Int).SetBytes(signature[:32]), new(big.Int).SetBytes(signature[32:])) {
982+ t.Fatal("JWT signature does not verify")
983+ }
984+}
985+
986+func TestSendReportsGoneSubscriptions(t *testing.T) {
987+ keys, _ := GenerateKeys()
988+ subscription, _, _ := subscriber(t)
989+ var headers http.Header
990+ status := 201
991+ server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
992+ headers = r.Header.Clone()
993+ w.WriteHeader(status)
994+ }))
995+ defer server.Close()
996+ subscription.Endpoint = server.URL + "/send/abc"
997+ result, err := Send(context.Background(), server.Client(), keys, "mailto:owner@example", subscription, []byte(`{"title":"t"}`), time.Hour)
998+ if err != nil || result.Status != 201 || result.Gone {
999+ t.Fatalf("send: %v %+v", err, result)
1000+ }
1001+ if headers.Get("Content-Encoding") != "aes128gcm" || headers.Get("TTL") != "3600" || !strings.HasPrefix(headers.Get("Authorization"), "vapid t=") {
1002+ t.Fatalf("headers %v", headers)
1003+ }
1004+ status = http.StatusGone
1005+ result, err = Send(context.Background(), server.Client(), keys, "mailto:owner@example", subscription, []byte(`{}`), time.Hour)
1006+ if err == nil || !result.Gone {
1007+ t.Fatalf("gone: %v %+v", err, result)
1008+ }
1009+}

About

ownerc01503e8035f
visibilitypublic
heada5b469fc4059

Maintainers

  • c01503e8035f owner
  • ba1b5beed5b5 agent hermes

Clone

  • https://012.run/npub1cq2s86qrtadf36eqq3epm38wmhdynrrcwxf84eldk3f94dkv7ttsrulrz5/tinyrelay.git

Refs