| 1 | // The Eleven push relay: a small, content-blind APNs signing service. Every |
| 2 | // messenger instance pushes to the SAME Apple app via the SAME team's .p8, so |
| 3 | // rather than copy that signing key onto every instance (a wide blast radius, |
| 4 | // and a blocker for third-party self-hosting), the key lives ONLY here. An |
| 5 | // instance authenticates with a per-instance bearer token and POSTs the same |
| 6 | // {device_token, sandbox, thread, blob} it would send to Apple; the relay signs |
| 7 | // the provider JWT and forwards it. Instances never hold signing capability. |
| 8 | // |
| 9 | // It preserves end-to-end encryption: the relay sees a device token, an opaque |
| 10 | // thread id, and the ciphertext blob — a strict subset of what Apple already |
| 11 | // sees per notification — never plaintext and never a message-decryption key. |
| 12 | // The .p8 is Apple *auth*, not a decryption key. See docs/push-relay.md. |
| 13 | package main |
| 14 | |
| 15 | import ( |
| 16 | "crypto/subtle" |
| 17 | "encoding/json" |
| 18 | "errors" |
| 19 | "flag" |
| 20 | "fmt" |
| 21 | "log" |
| 22 | "net" |
| 23 | "net/http" |
| 24 | "os" |
| 25 | "strings" |
| 26 | "sync" |
| 27 | "time" |
| 28 | |
| 29 | "lchat/internal/apns" |
| 30 | ) |
| 31 | |
| 32 | // maxPushBytes caps a push request body. Apple rejects alert payloads over |
| 33 | // ~4 KB; this leaves generous headroom for the ciphertext blob + envelope and |
| 34 | // fails oversized requests fast at the relay instead of after a wasted Apple |
| 35 | // round-trip. |
| 36 | const maxPushBytes = 8 << 10 |
| 37 | |
| 38 | type Server struct { |
| 39 | db *DB |
| 40 | apns *apns.Client |
| 41 | |
| 42 | adminHash string // sha256 of LCHAT_ADMIN_TOKEN; compared in constant time |
| 43 | |
| 44 | // Rate limits. perInstance keys by instance label so one instance can't |
| 45 | // exhaust another's budget (never by IP — instances egress from shared |
| 46 | // platform IPs; never by device token — unbounded bucket growth). global is |
| 47 | // a fleet-wide backstop against a single compromised token flooding Apple. |
| 48 | perInstance *rateLimiter |
| 49 | global *rateLimiter |
| 50 | } |
| 51 | |
| 52 | func main() { |
| 53 | addr := flag.String("addr", envOr("ADDR", ":1749"), "listen address (env: ADDR)") |
| 54 | dbPath := flag.String("db", envOr("LCHAT_PUSH_DB", "push.db"), "SQLite path (env: LCHAT_PUSH_DB)") |
| 55 | flag.Parse() |
| 56 | |
| 57 | // Fail loud: a misconfigured relay is a fleet-wide iOS-push outage, so a |
| 58 | // missing/bad key or admin token must stop the boot, not silently disable. |
| 59 | admin := os.Getenv("LCHAT_ADMIN_TOKEN") |
| 60 | if admin == "" { |
| 61 | log.Fatal("push: LCHAT_ADMIN_TOKEN is required") |
| 62 | } |
| 63 | client, err := apns.New( |
| 64 | os.Getenv("LCHAT_APNS_KEY"), |
| 65 | os.Getenv("LCHAT_APNS_KEY_ID"), |
| 66 | os.Getenv("LCHAT_APNS_TEAM_ID"), |
| 67 | os.Getenv("LCHAT_APNS_TOPIC"), |
| 68 | ) |
| 69 | if err != nil { |
| 70 | log.Fatalf("push: APNs config: %v", err) |
| 71 | } |
| 72 | |
| 73 | db, err := openDB(*dbPath) |
| 74 | if err != nil { |
| 75 | log.Fatalf("open db: %v", err) |
| 76 | } |
| 77 | s := &Server{ |
| 78 | db: db, |
| 79 | apns: client, |
| 80 | adminHash: hashToken(admin), |
| 81 | // Generous: a legitimate large-group fan-out is a burst of hundreds. The |
| 82 | // limit is a runaway/quota backstop, not a tight boundary (a token can |
| 83 | // only reach device tokens its instance already holds). |
| 84 | perInstance: newRateLimiter(1000, 500), |
| 85 | global: newRateLimiter(4000, 2000), |
| 86 | } |
| 87 | log.Printf("push relay listening on %s (db: %s, topic: %s)", *addr, *dbPath, client.Topic()) |
| 88 | if err := serve(*addr, s.handler()); err != nil { |
| 89 | log.Fatal(err) |
| 90 | } |
| 91 | } |
| 92 | |
| 93 | // serve listens on addr and serves h. A path-like addr (starts with "/") is a |
| 94 | // unix socket — how the instance router fronts this service, reverse-proxying to |
| 95 | // a socket rather than a TCP port; anything else is a TCP address (the default, |
| 96 | // e.g. on a container platform that health-checks over HTTP). On the socket path |
| 97 | // it clobbers only a stale socket (nothing listening) and locks it to 0600, |
| 98 | // mirroring the main binary's listen(). |
| 99 | func serve(addr string, h http.Handler) error { |
| 100 | // ReadHeaderTimeout caps how long a client may dribble request headers |
| 101 | // (Slowloris): the TCP branch binds straight to public TCP on a container |
| 102 | // platform. Same 10s the messenger's own listeners use (main.go). |
| 103 | srv := &http.Server{Handler: h, ReadHeaderTimeout: 10 * time.Second} |
| 104 | if !strings.HasPrefix(addr, "/") { |
| 105 | srv.Addr = addr |
| 106 | return srv.ListenAndServe() |
| 107 | } |
| 108 | if c, err := net.DialTimeout("unix", addr, 200*time.Millisecond); err == nil { |
| 109 | c.Close() |
| 110 | return fmt.Errorf("socket %s is already in use by another process", addr) |
| 111 | } |
| 112 | if err := os.Remove(addr); err != nil && !os.IsNotExist(err) { |
| 113 | return fmt.Errorf("remove stale socket %s: %w", addr, err) |
| 114 | } |
| 115 | ln, err := net.Listen("unix", addr) |
| 116 | if err != nil { |
| 117 | return err |
| 118 | } |
| 119 | defer ln.Close() |
| 120 | if err := os.Chmod(addr, 0o600); err != nil { |
| 121 | return fmt.Errorf("chmod socket %s: %w", addr, err) |
| 122 | } |
| 123 | return srv.Serve(ln) |
| 124 | } |
| 125 | |
| 126 | // handler wires the routes. A method so tests exercise the real router via |
| 127 | // httptest. |
| 128 | func (s *Server) handler() http.Handler { |
| 129 | mux := http.NewServeMux() |
| 130 | mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, r *http.Request) { w.Write([]byte("ok")) }) |
| 131 | mux.HandleFunc("GET /", s.root) |
| 132 | |
| 133 | // The one data endpoint: an instance forwards a push. Bearer instance token. |
| 134 | mux.HandleFunc("POST /api/push", s.pushHandler) |
| 135 | |
| 136 | // Admin surface: mint / list / revoke per-instance tokens at runtime |
| 137 | // (rotation needs no redeploy). Gated by LCHAT_ADMIN_TOKEN. |
| 138 | mux.HandleFunc("POST /api/admin/tokens", s.admin(s.mintHandler)) |
| 139 | mux.HandleFunc("GET /api/admin/tokens", s.admin(s.listHandler)) |
| 140 | mux.HandleFunc("DELETE /api/admin/tokens/{id}", s.admin(s.revokeHandler)) |
| 141 | |
| 142 | return securityHeaders(mux) |
| 143 | } |
| 144 | |
| 145 | func (s *Server) root(w http.ResponseWriter, r *http.Request) { |
| 146 | if r.URL.Path != "/" { |
| 147 | http.NotFound(w, r) |
| 148 | return |
| 149 | } |
| 150 | w.Header().Set("Content-Type", "text/plain; charset=utf-8") |
| 151 | w.Write([]byte("Eleven push relay. Content-blind APNs signer. See docs/push-relay.md.\n")) |
| 152 | } |
| 153 | |
| 154 | // ---- push ------------------------------------------------------------------ |
| 155 | |
| 156 | func (s *Server) pushHandler(w http.ResponseWriter, r *http.Request) { |
| 157 | // Order matters: auth → per-instance limit → global limit → size → shape. |
| 158 | label, ok := s.db.authInstance(bearerToken(r)) |
| 159 | if !ok { |
| 160 | http.Error(w, "unauthorized", http.StatusUnauthorized) |
| 161 | return |
| 162 | } |
| 163 | if !s.perInstance.allow(label) || !s.global.allow("*") { |
| 164 | http.Error(w, "rate limited", http.StatusTooManyRequests) |
| 165 | return |
| 166 | } |
| 167 | var body struct { |
| 168 | DeviceToken string `json:"device_token"` |
| 169 | Sandbox bool `json:"sandbox"` |
| 170 | Thread string `json:"thread"` |
| 171 | Blob string `json:"blob"` |
| 172 | // Message id + created_at, forwarded verbatim into the aps envelope so |
| 173 | // the device's NSE can ingest the message into its local store. Still |
| 174 | // content-blind: integers the instance already exposes to Apple's |
| 175 | // delivery timing. Optional — older instances omit them. |
| 176 | ID int64 `json:"id"` |
| 177 | At int64 `json:"at"` |
| 178 | // A changelog announcement (announce.go in the messenger): a PLAINTEXT |
| 179 | // banner instead of a ciphertext blob. The one non-blind payload the |
| 180 | // relay carries, and deliberately so — release notes are branding-class |
| 181 | // text baked into every binary, never user content. Mutually exclusive |
| 182 | // with thread/blob. |
| 183 | Changelog bool `json:"changelog"` |
| 184 | Title string `json:"title"` |
| 185 | Body string `json:"body"` |
| 186 | } |
| 187 | r.Body = http.MaxBytesReader(w, r.Body, maxPushBytes) |
| 188 | if err := json.NewDecoder(r.Body).Decode(&body); err != nil { |
| 189 | var tooLarge *http.MaxBytesError |
| 190 | if errors.As(err, &tooLarge) { |
| 191 | http.Error(w, "payload too large", http.StatusRequestEntityTooLarge) |
| 192 | return |
| 193 | } |
| 194 | http.Error(w, "bad request", http.StatusBadRequest) |
| 195 | return |
| 196 | } |
| 197 | if !apns.ValidDeviceToken(body.DeviceToken) { |
| 198 | http.Error(w, "bad device token", http.StatusBadRequest) |
| 199 | return |
| 200 | } |
| 201 | |
| 202 | start := time.Now() |
| 203 | var err error |
| 204 | if body.Changelog { |
| 205 | err = s.apns.SendAlert(body.DeviceToken, body.Sandbox, body.Title, body.Body) |
| 206 | } else { |
| 207 | err = s.apns.Send(body.DeviceToken, body.Sandbox, body.Thread, body.Blob, body.ID, body.At) |
| 208 | } |
| 209 | latency := time.Since(start).Milliseconds() |
| 210 | switch { |
| 211 | case err == nil: |
| 212 | // Content-blind logging: instance label + outcome + latency ONLY — never |
| 213 | // the device token, thread, blob, or any key material. |
| 214 | log.Printf("push[%s]: ok (%dms)", label, latency) |
| 215 | w.WriteHeader(http.StatusOK) |
| 216 | case errors.Is(err, apns.ErrGone): |
| 217 | // Normalise "token dead" to a bare 410 so the instance prunes it. The |
| 218 | // body must NOT contain "BadDeviceToken" — the instance also sniffs that |
| 219 | // string, so echoing it from a non-gone path would trigger a false prune. |
| 220 | log.Printf("push[%s]: gone (%dms)", label, latency) |
| 221 | http.Error(w, "gone", http.StatusGone) |
| 222 | default: |
| 223 | // Any other failure: log for us (the apns error carries Apple's status |
| 224 | // but no device token), return a generic 502 that is neither 410 nor |
| 225 | // contains "BadDeviceToken", so the instance does not prune a live token. |
| 226 | log.Printf("push[%s]: upstream error (%dms): %v", label, latency, err) |
| 227 | http.Error(w, "upstream error", http.StatusBadGateway) |
| 228 | } |
| 229 | } |
| 230 | |
| 231 | // ---- admin ----------------------------------------------------------------- |
| 232 | |
| 233 | // admin gates a handler behind the relay admin token (constant-time hash |
| 234 | // compare, matching the messenger's checkAdmin). |
| 235 | func (s *Server) admin(next http.HandlerFunc) http.HandlerFunc { |
| 236 | return func(w http.ResponseWriter, r *http.Request) { |
| 237 | if subtle.ConstantTimeCompare([]byte(hashToken(bearerToken(r))), []byte(s.adminHash)) != 1 { |
| 238 | http.Error(w, "unauthorized", http.StatusUnauthorized) |
| 239 | return |
| 240 | } |
| 241 | next(w, r) |
| 242 | } |
| 243 | } |
| 244 | |
| 245 | func (s *Server) mintHandler(w http.ResponseWriter, r *http.Request) { |
| 246 | var body struct { |
| 247 | Label string `json:"label"` |
| 248 | } |
| 249 | if json.NewDecoder(http.MaxBytesReader(w, r.Body, 1<<10)).Decode(&body) != nil || body.Label == "" { |
| 250 | http.Error(w, "label required", http.StatusBadRequest) |
| 251 | return |
| 252 | } |
| 253 | id, token, err := s.db.mintToken(body.Label) |
| 254 | if err != nil { |
| 255 | http.Error(w, "db error", http.StatusInternalServerError) |
| 256 | return |
| 257 | } |
| 258 | // The plaintext token is returned exactly once; it is unrecoverable after. |
| 259 | writeJSON(w, http.StatusOK, map[string]any{"id": id, "label": body.Label, "token": token}) |
| 260 | } |
| 261 | |
| 262 | func (s *Server) listHandler(w http.ResponseWriter, r *http.Request) { |
| 263 | tokens, err := s.db.listTokens() |
| 264 | if err != nil { |
| 265 | http.Error(w, "db error", http.StatusInternalServerError) |
| 266 | return |
| 267 | } |
| 268 | writeJSON(w, http.StatusOK, map[string]any{"tokens": tokens}) |
| 269 | } |
| 270 | |
| 271 | func (s *Server) revokeHandler(w http.ResponseWriter, r *http.Request) { |
| 272 | deleted, err := s.db.revokeToken(r.PathValue("id")) |
| 273 | if err != nil { |
| 274 | http.Error(w, "db error", http.StatusInternalServerError) |
| 275 | return |
| 276 | } |
| 277 | if !deleted { |
| 278 | http.Error(w, "no such token", http.StatusNotFound) |
| 279 | return |
| 280 | } |
| 281 | writeJSON(w, http.StatusOK, map[string]any{"ok": true}) |
| 282 | } |
| 283 | |
| 284 | // ---- middleware & helpers -------------------------------------------------- |
| 285 | |
| 286 | // securityHeaders mirrors the messenger's / home's defense-in-depth headers. |
| 287 | func securityHeaders(next http.Handler) http.Handler { |
| 288 | return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| 289 | h := w.Header() |
| 290 | h.Set("X-Content-Type-Options", "nosniff") |
| 291 | h.Set("X-Frame-Options", "DENY") |
| 292 | h.Set("Referrer-Policy", "no-referrer") |
| 293 | h.Set("Strict-Transport-Security", "max-age=31536000") |
| 294 | next.ServeHTTP(w, r) |
| 295 | }) |
| 296 | } |
| 297 | |
| 298 | // bearerToken pulls the token out of "Authorization: Bearer <t>". |
| 299 | func bearerToken(r *http.Request) string { |
| 300 | const p = "Bearer " |
| 301 | h := r.Header.Get("Authorization") |
| 302 | if len(h) > len(p) && h[:len(p)] == p { |
| 303 | return h[len(p):] |
| 304 | } |
| 305 | return "" |
| 306 | } |
| 307 | |
| 308 | func writeJSON(w http.ResponseWriter, status int, v any) { |
| 309 | w.Header().Set("Content-Type", "application/json") |
| 310 | w.WriteHeader(status) |
| 311 | if err := json.NewEncoder(w).Encode(v); err != nil { |
| 312 | log.Printf("write json: %v", err) |
| 313 | } |
| 314 | } |
| 315 | |
| 316 | func envOr(key, def string) string { |
| 317 | if v := os.Getenv(key); v != "" { |
| 318 | return v |
| 319 | } |
| 320 | return def |
| 321 | } |
| 322 | |
| 323 | // ---- rate limiting (per-key token bucket, copied from home/main.go) -------- |
| 324 | |
| 325 | type rateLimiter struct { |
| 326 | mu sync.Mutex |
| 327 | buckets map[string]*bucket |
| 328 | burst float64 |
| 329 | perSec float64 |
| 330 | } |
| 331 | |
| 332 | type bucket struct { |
| 333 | tokens float64 |
| 334 | last time.Time |
| 335 | } |
| 336 | |
| 337 | func newRateLimiter(burst, perSec float64) *rateLimiter { |
| 338 | return &rateLimiter{buckets: map[string]*bucket{}, burst: burst, perSec: perSec} |
| 339 | } |
| 340 | |
| 341 | func (rl *rateLimiter) allow(key string) bool { |
| 342 | rl.mu.Lock() |
| 343 | defer rl.mu.Unlock() |
| 344 | now := time.Now() |
| 345 | if len(rl.buckets) > 10000 { |
| 346 | for k, b := range rl.buckets { |
| 347 | if b.tokens >= rl.burst || now.Sub(b.last) > time.Hour { |
| 348 | delete(rl.buckets, k) |
| 349 | } |
| 350 | } |
| 351 | } |
| 352 | b, ok := rl.buckets[key] |
| 353 | if !ok { |
| 354 | b = &bucket{tokens: rl.burst, last: now} |
| 355 | rl.buckets[key] = b |
| 356 | } |
| 357 | b.tokens += now.Sub(b.last).Seconds() * rl.perSec |
| 358 | if b.tokens > rl.burst { |
| 359 | b.tokens = rl.burst |
| 360 | } |
| 361 | b.last = now |
| 362 | if b.tokens < 1 { |
| 363 | return false |
| 364 | } |
| 365 | b.tokens-- |
| 366 | return true |
| 367 | } |
| 368 | |
amadan