| 1 | package aviso |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "io" |
| 8 | "net" |
| 9 | "net/http" |
| 10 | "strconv" |
| 11 | "strings" |
| 12 | "sync" |
| 13 | "time" |
| 14 | |
| 15 | webpush "github.com/SherClockHolmes/webpush-go" |
| 16 | ) |
| 17 | |
| 18 | // Options tune one batch. Zero values mean aviso's defaults, not the |
| 19 | // push service's: webpush-go always sends a TTL header, and a literal |
| 20 | // 0 there means "deliver now or drop". |
| 21 | type Options struct { |
| 22 | // TTL is how long the push service may hold the message; whole |
| 23 | // seconds, >= 0. 0 means 24 hours. |
| 24 | TTL time.Duration |
| 25 | // Urgency is "very-low", "low", "normal" or "high"; "" means normal. |
| 26 | Urgency string |
| 27 | // Topic collapses pending messages with the same topic; <= 32 |
| 28 | // URL-safe characters; "" means none. |
| 29 | Topic string |
| 30 | } |
| 31 | |
| 32 | // Result is one device's outcome. Status is the push service's |
| 33 | // acceptance, not delivery; a 2xx means the service took it. |
| 34 | type Result struct { |
| 35 | ID string |
| 36 | Status int // 0 when Err is transport-level or the row was never attempted |
| 37 | RetryAfter time.Duration // from a 429/503, else 0 |
| 38 | Err error |
| 39 | } |
| 40 | |
| 41 | // ErrPayloadTooLarge means the plaintext exceeds RFC 8291's one-record |
| 42 | // limit; a larger payload would be split, which no browser accepts. |
| 43 | var ErrPayloadTooLarge = errors.New("aviso: payload over 3993 bytes") |
| 44 | |
| 45 | // ErrBadOptions means Options failed validation. |
| 46 | var ErrBadOptions = errors.New("aviso: invalid Options") |
| 47 | |
| 48 | const ( |
| 49 | maxPayload = 3993 |
| 50 | defaultTTL = 24 * time.Hour |
| 51 | requestTimeout = 30 * time.Second |
| 52 | maxBodyRead = 4096 |
| 53 | ) |
| 54 | |
| 55 | func (o Options) validate() error { |
| 56 | if o.TTL < 0 || o.TTL%time.Second != 0 { |
| 57 | return fmt.Errorf("%w: TTL must be whole non-negative seconds", ErrBadOptions) |
| 58 | } |
| 59 | switch o.Urgency { |
| 60 | case "", "very-low", "low", "normal", "high": |
| 61 | default: |
| 62 | return fmt.Errorf("%w: Urgency %q", ErrBadOptions, o.Urgency) |
| 63 | } |
| 64 | if len(o.Topic) > 32 || strings.Trim(o.Topic, "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_") != "" { |
| 65 | return fmt.Errorf("%w: Topic must be <= 32 URL-safe characters", ErrBadOptions) |
| 66 | } |
| 67 | return nil |
| 68 | } |
| 69 | |
| 70 | // SendTo fans payload out to every device subject enrolled — the |
| 71 | // common case, so an app never touches Stored. The batch error covers |
| 72 | // what stops the batch (the query, validation, cancellation); each |
| 73 | // Result covers one device. |
| 74 | func (s *Service) SendTo(ctx context.Context, subject string, payload []byte, o Options) ([]Result, error) { |
| 75 | if err := checkBatch(ctx, payload, o); err != nil { |
| 76 | return nil, err |
| 77 | } |
| 78 | rows, err := s.List(ctx, subject) |
| 79 | if err != nil { |
| 80 | return nil, err |
| 81 | } |
| 82 | return s.Send(ctx, rows, payload, o) |
| 83 | } |
| 84 | |
| 85 | // Send delivers payload to each of to, bounded by Config.Concurrency |
| 86 | // across the Service. It never retries: RetryAfter is for the app's |
| 87 | // own scheduler. |
| 88 | // |
| 89 | // On cancellation, rows not yet started carry ctx.Err(), rows already |
| 90 | // in flight finish (their own request honours ctx), and Send returns |
| 91 | // only once every goroutine it started has stopped writing results — |
| 92 | // a caller that reads results after Send returns must never race a |
| 93 | // straggler. |
| 94 | func (s *Service) Send(ctx context.Context, to []Stored, payload []byte, o Options) ([]Result, error) { |
| 95 | if err := checkBatch(ctx, payload, o); err != nil { |
| 96 | return nil, err |
| 97 | } |
| 98 | results := make([]Result, len(to)) |
| 99 | var wg sync.WaitGroup |
| 100 | var batchErr error |
| 101 | for i := range to { |
| 102 | st := to[i] |
| 103 | results[i].ID = st.ID |
| 104 | if st.VAPIDKeyID != s.keyID { |
| 105 | results[i].Err = ErrKeyMismatch |
| 106 | continue |
| 107 | } |
| 108 | if err := validateEndpoint(st.Endpoint); err != nil { |
| 109 | results[i].Err = err |
| 110 | continue |
| 111 | } |
| 112 | if batchErr != nil { |
| 113 | results[i].Err = batchErr |
| 114 | continue |
| 115 | } |
| 116 | select { |
| 117 | case s.sem <- struct{}{}: |
| 118 | case <-ctx.Done(): |
| 119 | batchErr = ctx.Err() |
| 120 | results[i].Err = batchErr |
| 121 | continue |
| 122 | } |
| 123 | // A slot freed by a request the cancellation cut short can win |
| 124 | // the select above; nothing starts after cancellation. |
| 125 | if err := ctx.Err(); err != nil { |
| 126 | <-s.sem |
| 127 | batchErr = err |
| 128 | results[i].Err = err |
| 129 | continue |
| 130 | } |
| 131 | wg.Add(1) |
| 132 | go func(i int, st Stored) { |
| 133 | defer wg.Done() |
| 134 | defer func() { <-s.sem }() |
| 135 | results[i] = s.sendOne(ctx, st, payload, o) |
| 136 | }(i, st) |
| 137 | } |
| 138 | wg.Wait() |
| 139 | if batchErr == nil && ctx.Err() != nil { |
| 140 | batchErr = ctx.Err() |
| 141 | } |
| 142 | return results, batchErr |
| 143 | } |
| 144 | |
| 145 | func checkBatch(ctx context.Context, payload []byte, o Options) error { |
| 146 | if err := ctx.Err(); err != nil { |
| 147 | return err |
| 148 | } |
| 149 | if len(payload) > maxPayload { |
| 150 | return ErrPayloadTooLarge |
| 151 | } |
| 152 | return o.validate() |
| 153 | } |
| 154 | |
| 155 | func (s *Service) sendOne(ctx context.Context, st Stored, payload []byte, o Options) Result { |
| 156 | res := Result{ID: st.ID} |
| 157 | ctx, cancel := context.WithTimeout(ctx, requestTimeout) |
| 158 | defer cancel() |
| 159 | ttl := o.TTL |
| 160 | if ttl == 0 { |
| 161 | ttl = defaultTTL |
| 162 | } |
| 163 | urgency := webpush.Urgency(o.Urgency) |
| 164 | if urgency == "" { |
| 165 | urgency = webpush.UrgencyNormal |
| 166 | } |
| 167 | // webpush-go wraps the payload in a bytes.Buffer and appends the |
| 168 | // record delimiter and padding in place. With spare capacity, the |
| 169 | // concurrent sends of one batch would write into one backing |
| 170 | // array; capping the capacity forces each append to reallocate. |
| 171 | payload = payload[:len(payload):len(payload)] |
| 172 | resp, err := webpush.SendNotificationWithContext(ctx, payload, |
| 173 | &webpush.Subscription{Endpoint: st.Endpoint, Keys: webpush.Keys{P256dh: st.P256dh, Auth: st.Auth}}, |
| 174 | &webpush.Options{ |
| 175 | HTTPClient: s.client, |
| 176 | Subscriber: s.wireContact, |
| 177 | TTL: int(ttl / time.Second), |
| 178 | Urgency: urgency, |
| 179 | Topic: o.Topic, |
| 180 | VAPIDPublicKey: s.pub, |
| 181 | VAPIDPrivateKey: s.cfg.PrivateKey, |
| 182 | }) |
| 183 | if err != nil { |
| 184 | res.Err = redact(err) |
| 185 | return res |
| 186 | } |
| 187 | defer resp.Body.Close() |
| 188 | // Drain a bounded amount for keep-alive; the body is never read |
| 189 | // into anything a log could see. |
| 190 | _, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, maxBodyRead)) |
| 191 | return s.settle(ctx, st, resp.StatusCode, resp.Header.Get("Retry-After")) |
| 192 | } |
| 193 | |
| 194 | // settle turns the push service's answer into a Result and the store |
| 195 | // update it implies. Store updates run under a context the caller's |
| 196 | // cancellation cannot interrupt — a 410 the service already answered |
| 197 | // must prune whether or not the batch was cancelled a moment later — |
| 198 | // but with their own bound, because the writer is one connection and |
| 199 | // a transaction holding it must not pin a semaphore slot forever. |
| 200 | func (s *Service) settle(ctx context.Context, st Stored, status int, retryAfter string) Result { |
| 201 | res := Result{ID: st.ID, Status: status} |
| 202 | dbCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), s.dbTimeout) |
| 203 | defer cancel() |
| 204 | switch { |
| 205 | case status >= 200 && status < 300: |
| 206 | if err := s.confirm(dbCtx, st.ID, st.Revision); err != nil { |
| 207 | s.cfg.Logger.Warn("aviso: confirm failed", "id", st.ID, "err", err) |
| 208 | } |
| 209 | case status == http.StatusNotFound || status == http.StatusGone: |
| 210 | res.Err = fmt.Errorf("aviso: push service says subscription gone (%d)", status) |
| 211 | if err := s.prune(dbCtx, st.ID, st.Revision); err != nil { |
| 212 | s.cfg.Logger.Warn("aviso: prune failed", "id", st.ID, "err", err) |
| 213 | } |
| 214 | case status == http.StatusTooManyRequests || status == http.StatusServiceUnavailable: |
| 215 | res.RetryAfter = parseRetryAfter(retryAfter, s.now()) |
| 216 | res.Err = fmt.Errorf("aviso: push service throttled (%d)", status) |
| 217 | default: |
| 218 | res.Err = fmt.Errorf("aviso: push service refused (%d)", status) |
| 219 | } |
| 220 | return res |
| 221 | } |
| 222 | |
| 223 | // ErrTransport is the sanitised form of any transport failure that is |
| 224 | // not a cancellation, a timeout or the guard's own refusal. The |
| 225 | // original is discarded, not wrapped: net/http's errors quote the |
| 226 | // request URL and even a bad Location header verbatim, and the |
| 227 | // endpoint is the one secret in this package that must never reach a |
| 228 | // log. |
| 229 | var ErrTransport = errors.New("aviso: transport error") |
| 230 | |
| 231 | // ErrDialRefused is the guard's refusal, surfaced without the address. |
| 232 | var ErrDialRefused = errors.New("aviso: dial refused by the SSRF guard") |
| 233 | |
| 234 | func redact(err error) error { |
| 235 | switch { |
| 236 | case errors.Is(err, context.Canceled): |
| 237 | return fmt.Errorf("aviso: send: %w", context.Canceled) |
| 238 | case errors.Is(err, context.DeadlineExceeded): |
| 239 | return fmt.Errorf("aviso: send: %w", context.DeadlineExceeded) |
| 240 | case strings.Contains(err.Error(), "aviso: dial refused"): |
| 241 | return ErrDialRefused |
| 242 | } |
| 243 | var ne net.Error |
| 244 | if errors.As(err, &ne) && ne.Timeout() { |
| 245 | return fmt.Errorf("aviso: send: %w", context.DeadlineExceeded) |
| 246 | } |
| 247 | return ErrTransport |
| 248 | } |
| 249 | |
| 250 | // maxRetryAfter caps what a push service can ask for: a value past |
| 251 | // this is meaningless to a scheduler, and an unbounded one overflows |
| 252 | // time.Duration into a negative number. |
| 253 | const maxRetryAfter = 24 * time.Hour |
| 254 | |
| 255 | func parseRetryAfter(v string, now time.Time) time.Duration { |
| 256 | if v == "" { |
| 257 | return 0 |
| 258 | } |
| 259 | var d time.Duration |
| 260 | if strings.Trim(v, "0123456789") == "" { // delta-seconds, however many digits |
| 261 | secs, err := strconv.ParseInt(v, 10, 64) |
| 262 | if err != nil || secs > int64(maxRetryAfter/time.Second) { // err here is only ever "out of range" |
| 263 | return maxRetryAfter |
| 264 | } |
| 265 | d = time.Duration(secs) * time.Second |
| 266 | } else if t, err := http.ParseTime(v); err == nil && t.After(now) { |
| 267 | d = t.Sub(now) |
| 268 | } |
| 269 | if d > maxRetryAfter { |
| 270 | return maxRetryAfter |
| 271 | } |
| 272 | return d |
| 273 | } |
| 274 | |
amadan