| 1 | package main |
| 2 | |
| 3 | import ( |
| 4 | "encoding/json" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "io" |
| 8 | "log" |
| 9 | "net" |
| 10 | "net/http" |
| 11 | "net/url" |
| 12 | "os" |
| 13 | "sync" |
| 14 | "sync/atomic" |
| 15 | "time" |
| 16 | |
| 17 | webpush "github.com/SherClockHolmes/webpush-go" |
| 18 | ) |
| 19 | |
| 20 | // pushClient sends Web Push POSTs. The endpoint is attacker-controlled data (any |
| 21 | // member registers one at subscribe time), so — like ogClient — its dialer |
| 22 | // validates the resolved IP of every connection to stop the server being pointed |
| 23 | // at its own network (SSRF), and re-validates each redirect hop (defeating DNS |
| 24 | // rebinding). webpush-go otherwise uses http.DefaultClient, which has no guard. |
| 25 | var pushClient = &http.Client{ |
| 26 | Timeout: 30 * time.Second, |
| 27 | Transport: &http.Transport{ |
| 28 | DialContext: (&net.Dialer{ |
| 29 | Timeout: 10 * time.Second, |
| 30 | Control: guardDialControl, |
| 31 | }).DialContext, |
| 32 | TLSHandshakeTimeout: 10 * time.Second, |
| 33 | ResponseHeaderTimeout: 20 * time.Second, |
| 34 | DisableKeepAlives: true, |
| 35 | }, |
| 36 | CheckRedirect: func(req *http.Request, via []*http.Request) error { |
| 37 | if len(via) >= 5 { |
| 38 | return errors.New("too many redirects") |
| 39 | } |
| 40 | if req.URL.Scheme != "https" { |
| 41 | return errors.New("non-https redirect") |
| 42 | } |
| 43 | return nil |
| 44 | }, |
| 45 | } |
| 46 | |
| 47 | // validatePushEndpoint is the first line of the SSRF guard: it rejects anything |
| 48 | // that isn't a plain https URL with a host before we ever store it. Real Web Push |
| 49 | // services (FCM, Mozilla, Apple, Windows) are always https. The resolved-IP check |
| 50 | // that actually blocks internal targets lives in pushClient's dialer above, so a |
| 51 | // stale stored endpoint is still guarded when we send. |
| 52 | func validatePushEndpoint(endpoint string) error { |
| 53 | if len(endpoint) > 2048 { |
| 54 | return errors.New("endpoint too long") |
| 55 | } |
| 56 | u, err := url.Parse(endpoint) |
| 57 | if err != nil { |
| 58 | return err |
| 59 | } |
| 60 | if u.Scheme != "https" { |
| 61 | return errors.New("endpoint must be https") |
| 62 | } |
| 63 | if u.Hostname() == "" { |
| 64 | return errors.New("endpoint has no host") |
| 65 | } |
| 66 | return nil |
| 67 | } |
| 68 | |
| 69 | // PushSub is a stored Web Push subscription, tied to the session token that made |
| 70 | // it. Only a hash of that token is persisted (it's a live credential); Token is |
| 71 | // the raw value on the way in, TokenHash is what comes back out and is matched on. |
| 72 | type PushSub struct { |
| 73 | Endpoint string |
| 74 | Token string // raw session token — input to addSubscription only, never stored |
| 75 | TokenHash string // hashToken(Token); the persisted, matched-on value |
| 76 | P256dh string |
| 77 | Auth string |
| 78 | UserID string // who this belongs to (so removing a user drops their subs) |
| 79 | } |
| 80 | |
| 81 | func (db *DB) addSubscription(s PushSub) error { |
| 82 | var uid any |
| 83 | if s.UserID != "" { |
| 84 | uid = s.UserID |
| 85 | } |
| 86 | // Persist only the token hash; the plaintext `token` column stays empty. |
| 87 | _, err := db.sql.Exec( |
| 88 | `INSERT INTO push_subscriptions(endpoint, token, token_hash, p256dh, auth, user_id, created_at) |
| 89 | VALUES(?, '', ?, ?, ?, ?, ?) |
| 90 | ON CONFLICT(endpoint) DO UPDATE SET |
| 91 | token='', token_hash=excluded.token_hash, p256dh=excluded.p256dh, auth=excluded.auth, user_id=excluded.user_id`, |
| 92 | s.Endpoint, hashToken(s.Token), s.P256dh, s.Auth, uid, time.Now().Unix(), |
| 93 | ) |
| 94 | return err |
| 95 | } |
| 96 | |
| 97 | func (db *DB) listSubscriptions() ([]PushSub, error) { |
| 98 | rows, err := db.sql.Query("SELECT endpoint, COALESCE(token_hash, ''), p256dh, auth FROM push_subscriptions") |
| 99 | if err != nil { |
| 100 | return nil, err |
| 101 | } |
| 102 | defer rows.Close() |
| 103 | out := []PushSub{} |
| 104 | for rows.Next() { |
| 105 | var s PushSub |
| 106 | if err := rows.Scan(&s.Endpoint, &s.TokenHash, &s.P256dh, &s.Auth); err != nil { |
| 107 | return nil, err |
| 108 | } |
| 109 | out = append(out, s) |
| 110 | } |
| 111 | return out, rows.Err() |
| 112 | } |
| 113 | |
| 114 | func (db *DB) deleteSubscription(endpoint string) error { |
| 115 | _, err := db.sql.Exec("DELETE FROM push_subscriptions WHERE endpoint = ?", endpoint) |
| 116 | return err |
| 117 | } |
| 118 | |
| 119 | // deleteOwnSubscription removes a subscription only if it belongs to the given |
| 120 | // token, so knowing someone else's endpoint isn't enough to silence their device. |
| 121 | func (db *DB) deleteOwnSubscription(endpoint, token string) error { |
| 122 | _, err := db.sql.Exec( |
| 123 | "DELETE FROM push_subscriptions WHERE endpoint = ? AND token_hash = ?", |
| 124 | endpoint, hashToken(token), |
| 125 | ) |
| 126 | return err |
| 127 | } |
| 128 | |
| 129 | func (db *DB) deleteSubscriptionsForToken(token string) error { |
| 130 | _, err := db.sql.Exec("DELETE FROM push_subscriptions WHERE token_hash = ?", hashToken(token)) |
| 131 | return err |
| 132 | } |
| 133 | |
| 134 | // ensureVAPID returns the server's VAPID keypair, generating and persisting one on |
| 135 | // first use. VAPID lets the server authenticate itself to the browser push services |
| 136 | // with no third-party account — it is unrelated to message encryption. |
| 137 | // |
| 138 | // The pair lives in a 0600 sidecar file next to the DB (<db>.vapid, like the |
| 139 | // <db>-blobs dir), deliberately NOT in the config table: hibernation replicates |
| 140 | // lchat.db to S3, and a bucket leak must not hand out private keys of any kind |
| 141 | // (docs/no-plaintext.md § 2). The pair used to live in config — those legacy |
| 142 | // rows are moved into the file and deleted on first boot, so existing push |
| 143 | // subscriptions keep working across the change. |
| 144 | func (db *DB) ensureVAPID() (pub, priv string, err error) { |
| 145 | // The file is authoritative once it exists. A restored/older DB snapshot may |
| 146 | // still carry the legacy config rows — clear them so the DB (and its S3 |
| 147 | // replica) never re-grows a private key. |
| 148 | b, readErr := os.ReadFile(db.vapidPath) |
| 149 | if readErr == nil { |
| 150 | var k struct{ Public, Private string } |
| 151 | if err := json.Unmarshal(b, &k); err != nil || k.Public == "" || k.Private == "" { |
| 152 | // Fail loud rather than silently minting a new pair: regenerating |
| 153 | // breaks every existing push subscription, so a corrupt file is an |
| 154 | // operator problem (restore or delete it), not something to paper over. |
| 155 | return "", "", fmt.Errorf("vapid keyfile %s is malformed; restore it or delete it to re-mint (breaks existing push subscriptions)", db.vapidPath) |
| 156 | } |
| 157 | db.deleteVAPIDConfig() |
| 158 | return k.Public, k.Private, nil |
| 159 | } |
| 160 | if !os.IsNotExist(readErr) { |
| 161 | // EACCES/EIO etc. must NOT fall through to the mint path — that would |
| 162 | // rename a fresh pair over a file that exists but couldn't be read |
| 163 | // (e.g. an operator's root-owned copy after a migration) and silently |
| 164 | // break every subscription. Only a genuinely absent file mints. |
| 165 | return "", "", fmt.Errorf("vapid keyfile %s: %w", db.vapidPath, readErr) |
| 166 | } |
| 167 | // Legacy: the pair is still in the DB — move it into the file. Fresh |
| 168 | // instance: mint one. Either way the file is written before the config |
| 169 | // rows are deleted, so a crash in between leaves both copies (file wins |
| 170 | // next boot), never neither. |
| 171 | pub, okPub, errPub := db.getConfig("vapid_public") |
| 172 | priv, okPriv, errPriv := db.getConfig("vapid_private") |
| 173 | if errPub != nil || errPriv != nil { |
| 174 | // A transient DB error here must not mint a fresh pair (the delete |
| 175 | // below would then destroy the intact legacy rows) — retry next boot. |
| 176 | return "", "", fmt.Errorf("vapid config read: %v / %v", errPub, errPriv) |
| 177 | } |
| 178 | if !okPub || !okPriv { |
| 179 | priv, pub, err = webpush.GenerateVAPIDKeys() |
| 180 | if err != nil { |
| 181 | return "", "", err |
| 182 | } |
| 183 | // A fresh mint with existing subscriptions means the old pair is gone |
| 184 | // (a DB restored onto a new box) — say so, since those subscriptions |
| 185 | // are now dead until members revisit (docs/secrets.md). |
| 186 | if subs, _ := db.listSubscriptions(); len(subs) > 0 { |
| 187 | log.Printf("vapid: minted a NEW keypair with %d existing push subscriptions — they break until members re-subscribe on next visit", len(subs)) |
| 188 | } |
| 189 | } |
| 190 | blob, err := json.Marshal(struct{ Public, Private string }{pub, priv}) |
| 191 | if err != nil { |
| 192 | return "", "", err |
| 193 | } |
| 194 | // Write-temp-then-rename so a crash mid-write can't leave a torn keyfile. |
| 195 | tmp := db.vapidPath + ".tmp" |
| 196 | if err = os.WriteFile(tmp, blob, 0o600); err != nil { |
| 197 | return "", "", err |
| 198 | } |
| 199 | if err = os.Rename(tmp, db.vapidPath); err != nil { |
| 200 | return "", "", err |
| 201 | } |
| 202 | db.deleteVAPIDConfig() |
| 203 | return pub, priv, nil |
| 204 | } |
| 205 | |
| 206 | // deleteVAPIDConfig scrubs the legacy config-table copy of the VAPID pair. |
| 207 | // Best-effort: a failed delete just means the next boot retries. |
| 208 | func (db *DB) deleteVAPIDConfig() { |
| 209 | db.sql.Exec("DELETE FROM config WHERE key IN ('vapid_public', 'vapid_private')") |
| 210 | } |
| 211 | |
| 212 | // pushNewMessage delivers the ciphertext blob to every subscription. The server |
| 213 | // never sees plaintext: the payload IS the same ciphertext stored in the DB, and |
| 214 | // the service worker decrypts it locally with the group key. Expired subscriptions |
| 215 | // (404/410) are pruned. Best-effort; runs in its own goroutine. |
| 216 | // pushNewMessage notifies everyone except the sender — you don't want a push on |
| 217 | // any of your own devices for a message you just sent. The sender's device token |
| 218 | // (from their WebSocket) identifies both that device and, via its owning user, |
| 219 | // all their other devices, which selectRecipients excludes. |
| 220 | func (s *Server) pushNewMessage(blob, threadID, senderToken string, id, at int64) { |
| 221 | // The space thread (spacethread.go) never pushes: its traffic is app |
| 222 | // machinery (install registry + space-app events) that must not buzz every |
| 223 | // phone in the space, and clients that don't know the thread (native, old) |
| 224 | // couldn't render the notification. Live WS delivery still applies. |
| 225 | if s.db.threadKind(threadID) == "space" { |
| 226 | return |
| 227 | } |
| 228 | members, err := s.db.memberUserIDs(threadID) |
| 229 | if err != nil { |
| 230 | log.Println("push: thread members:", err) |
| 231 | return |
| 232 | } |
| 233 | set := make(map[string]bool, len(members)) |
| 234 | for _, u := range members { |
| 235 | set[u] = true |
| 236 | } |
| 237 | s.dropMutedRecipients(threadID, set) |
| 238 | // And members who BLOCK the sender (issue #578): a blocked member's group |
| 239 | // messages must never buzz the blocker (the client folds them away too). |
| 240 | if senderUser, ok := s.db.deviceTokenUser(senderToken); ok { |
| 241 | s.dropBlockingRecipients(senderUser, set) |
| 242 | } |
| 243 | // id/at ride along so the service worker can file the row straight into |
| 244 | // the page's local message store (#457 phase 2) — opening from a |
| 245 | // notification then paints with the message already in place. Additive: |
| 246 | // an older SW just ignores them. Both are values the recipient could read |
| 247 | // from /api/messages anyway; nothing new is disclosed. |
| 248 | payload, _ := json.Marshal(map[string]any{"blob": blob, "thread": threadID, "id": id, "at": at}) |
| 249 | // skipActive=true: don't notify people currently reading on another device. |
| 250 | // recipients=set: only this thread's members, never non-members. |
| 251 | s.sendToRecipients(payload, "msg", senderToken, true, set) |
| 252 | // Same message, same rules, over APNs to the native iOS apps (no-op unless |
| 253 | // LCHAT_APNS_* is configured). |
| 254 | s.apnsFanOut(threadID, blob, id, at, senderToken, true, set) |
| 255 | } |
| 256 | |
| 257 | // dropMutedRecipients removes members who muted the thread from a push |
| 258 | // recipient set — a mute means no buzz on any device, over either transport. |
| 259 | // The messages themselves still flow; only notifications stop. |
| 260 | func (s *Server) dropMutedRecipients(threadID string, set map[string]bool) { |
| 261 | muted, err := s.db.mutedUserIDs(threadID) |
| 262 | if err != nil { |
| 263 | log.Println("push: muted members:", err) |
| 264 | return // best-effort: on error, prefer a spurious buzz over a lost one |
| 265 | } |
| 266 | for u := range muted { |
| 267 | delete(set, u) |
| 268 | } |
| 269 | } |
| 270 | |
| 271 | // dropBlockingRecipients removes members who have blocked the author from a |
| 272 | // push recipient set — block implies never hearing from them again. |
| 273 | func (s *Server) dropBlockingRecipients(authorID string, set map[string]bool) { |
| 274 | blockers, err := s.db.blockersOf(authorID) |
| 275 | if err != nil { |
| 276 | log.Println("push: blockers:", err) |
| 277 | return |
| 278 | } |
| 279 | for u := range blockers { |
| 280 | delete(set, u) |
| 281 | } |
| 282 | } |
| 283 | |
| 284 | // sendTestPush delivers a generic "it works" notification to every subscriber, |
| 285 | // bypassing the message flow (and ignoring active state, so it always lands). |
| 286 | // It covers BOTH transports: Web Push (browsers) and APNs (native iOS, via the |
| 287 | // relay when configured). The APNs side carries a dummy thread/blob the NSE |
| 288 | // can't decrypt, so the phone shows the placeholder "New message" banner (see |
| 289 | // ios/LChatNSE) — which is exactly the "push reached this device" signal a test |
| 290 | // wants. No exclusions (exceptToken="", recipients=nil) and skipActive=false, so |
| 291 | // it always lands, on every registered device including the admin's own. |
| 292 | func (s *Server) sendTestPush() int { |
| 293 | payload, _ := json.Marshal(map[string]bool{"test": true}) |
| 294 | sent := s.sendToRecipients(payload, "test", "", false, nil) |
| 295 | sent += s.apnsFanOut("test", "test", 0, 0, "", false, nil) // no-op unless APNs/relay configured |
| 296 | return sent |
| 297 | } |
| 298 | |
| 299 | // maxPushConcurrency bounds simultaneous sends so a large group doesn't open |
| 300 | // thousands of sockets at once. Each webpush send is a blocking HTTPS round-trip, |
| 301 | // so fanning out turns O(N) sequential latency into ~O(N/concurrency). |
| 302 | const maxPushConcurrency = 32 |
| 303 | |
| 304 | // selectRecipients decides which subscriptions should receive a payload. It |
| 305 | // excludes the sender — by exact device token AND by owning user, so none of the |
| 306 | // sender's *other* devices buzz for a message they sent (each device carries its |
| 307 | // own token now, so an exact-token match alone would miss the sibling devices). |
| 308 | // When skipActive it also drops ALL of a recipient's devices if they're reading on |
| 309 | // any one of them (per-user, mirroring the sender exclusion — a per-device check |
| 310 | // would still buzz your phone while you read on your laptop), and when recipients is |
| 311 | // non-nil it keeps only devices whose owner is in that set (thread membership). |
| 312 | func (s *Server) selectRecipients(subs []PushSub, exceptToken string, skipActive bool, recipients map[string]bool) []PushSub { |
| 313 | senderUser := "" |
| 314 | exceptHash := "" |
| 315 | if exceptToken != "" { |
| 316 | exceptHash = hashToken(exceptToken) |
| 317 | senderUser, _ = s.db.deviceTokenUser(exceptToken) |
| 318 | } |
| 319 | out := make([]PushSub, 0, len(subs)) |
| 320 | for _, sub := range subs { |
| 321 | if exceptHash != "" && sub.TokenHash == exceptHash { |
| 322 | continue // the exact device that sent it |
| 323 | } |
| 324 | // Resolve the owner once — needed for sender-user exclusion, the per-user |
| 325 | // active check, and membership. Resolved from the stored token hash. |
| 326 | var uid string |
| 327 | var known bool |
| 328 | if senderUser != "" || recipients != nil || skipActive { |
| 329 | uid, known = s.db.deviceTokenUserByHash(sub.TokenHash) |
| 330 | } |
| 331 | if senderUser != "" && known && uid == senderUser { |
| 332 | continue // another of the sender's own devices — don't notify yourself |
| 333 | } |
| 334 | if skipActive && s.isUserActive(uid) { |
| 335 | continue // this person is reading on one of their devices right now |
| 336 | } |
| 337 | if recipients != nil && (!known || !recipients[uid]) { |
| 338 | continue // not a member of this thread |
| 339 | } |
| 340 | out = append(out, sub) |
| 341 | } |
| 342 | return out |
| 343 | } |
| 344 | |
| 345 | // sendToRecipients delivers a payload to the subscriptions selectRecipients |
| 346 | // picks. Prunes dead subs (404/410). Sends run concurrently, bounded by |
| 347 | // maxPushConcurrency. Returns the number of successful sends. |
| 348 | func (s *Server) sendToRecipients(payload []byte, kind, exceptToken string, skipActive bool, recipients map[string]bool) int { |
| 349 | subs, err := s.db.listSubscriptions() |
| 350 | if err != nil { |
| 351 | log.Println("push: list subs:", err) |
| 352 | return 0 |
| 353 | } |
| 354 | targets := s.selectRecipients(subs, exceptToken, skipActive, recipients) |
| 355 | log.Printf("push[%s]: %d subscription(s), %d after exclusions", kind, len(subs), len(targets)) |
| 356 | |
| 357 | sem := make(chan struct{}, maxPushConcurrency) |
| 358 | var wg sync.WaitGroup |
| 359 | var sent atomic.Int64 |
| 360 | for _, sub := range targets { |
| 361 | wg.Add(1) |
| 362 | sem <- struct{}{} |
| 363 | go func(sub PushSub) { |
| 364 | defer wg.Done() |
| 365 | defer func() { <-sem }() |
| 366 | if s.sendOne(payload, kind, sub) { |
| 367 | sent.Add(1) |
| 368 | } |
| 369 | }(sub) |
| 370 | } |
| 371 | wg.Wait() |
| 372 | return int(sent.Load()) |
| 373 | } |
| 374 | |
| 375 | // sendOne delivers a payload to a single subscription, pruning it if the push |
| 376 | // service reports it gone (404/410). Returns true on a 2xx. |
| 377 | func (s *Server) sendOne(payload []byte, kind string, sub PushSub) bool { |
| 378 | resp, err := webpush.SendNotification(payload, &webpush.Subscription{ |
| 379 | Endpoint: sub.Endpoint, |
| 380 | Keys: webpush.Keys{P256dh: sub.P256dh, Auth: sub.Auth}, |
| 381 | }, &webpush.Options{ |
| 382 | // Route through the SSRF-guarded client, not webpush-go's default |
| 383 | // http.DefaultClient — the endpoint is attacker-controlled. |
| 384 | HTTPClient: pushClient, |
| 385 | // Pass the bare e-mail: webpush-go prepends "mailto:" itself. Passing |
| 386 | // "mailto:…" yields "mailto:mailto:…", which Apple rejects (BadJwtToken). |
| 387 | Subscriber: "admin@elevenmessenger.com", |
| 388 | VAPIDPublicKey: s.vapidPub, |
| 389 | VAPIDPrivateKey: s.vapidPriv, |
| 390 | TTL: 60, |
| 391 | Urgency: webpush.UrgencyHigh, |
| 392 | }) |
| 393 | if err != nil { |
| 394 | log.Printf("push[%s]: send error: %v", kind, err) |
| 395 | return false |
| 396 | } |
| 397 | defer resp.Body.Close() |
| 398 | if resp.StatusCode >= 200 && resp.StatusCode < 300 { |
| 399 | log.Printf("push[%s]: status %d (%.45s…)", kind, resp.StatusCode, sub.Endpoint) |
| 400 | return true |
| 401 | } |
| 402 | body, _ := io.ReadAll(io.LimitReader(resp.Body, 400)) |
| 403 | log.Printf("push[%s]: status %d body=%q (%.45s…)", kind, resp.StatusCode, string(body), sub.Endpoint) |
| 404 | if resp.StatusCode == http.StatusNotFound || resp.StatusCode == http.StatusGone { |
| 405 | _ = s.db.deleteSubscription(sub.Endpoint) |
| 406 | } |
| 407 | return false |
| 408 | } |
| 409 | |
amadan