| 1 | package console |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "encoding/json" |
| 7 | "errors" |
| 8 | "io" |
| 9 | "log" |
| 10 | "net/http" |
| 11 | "strings" |
| 12 | "sync" |
| 13 | "time" |
| 14 | |
| 15 | "github.com/carlosframework/platform/internal/hostops" |
| 16 | "github.com/carlosframework/platform/internal/objstore" |
| 17 | "github.com/carlosframework/platform/internal/release" |
| 18 | ) |
| 19 | |
| 20 | const ( |
| 21 | scanDeliveryWorkers = 4 |
| 22 | scanDeliveryQueue = 64 |
| 23 | scanDeliveryTimeout = 10 * time.Second |
| 24 | ) |
| 25 | |
| 26 | var scanDeliveryRetryDelays = []time.Duration{time.Second, 4 * time.Second, 16 * time.Second} |
| 27 | |
| 28 | type scanDeliveryJob struct { |
| 29 | ref objstore.AppRef |
| 30 | done objstore.Drained |
| 31 | path string |
| 32 | gen int |
| 33 | } |
| 34 | |
| 35 | type scanCallbackBody struct { |
| 36 | Account string `json:"account"` |
| 37 | App string `json:"app"` |
| 38 | Key string `json:"key"` |
| 39 | ETag string `json:"etag"` |
| 40 | Verdict objstore.Verdict `json:"verdict"` |
| 41 | Recorded bool `json:"recorded"` |
| 42 | Engine string `json:"engine"` |
| 43 | Signature string `json:"signature"` |
| 44 | } |
| 45 | |
| 46 | // scanDeliveryPool keeps tenant HTTP outside the drain goroutine. Its queue is |
| 47 | // intentionally bounded: the status door is durable, while allowing callbacks |
| 48 | // to grow memory without limit would turn a best-effort convenience into a |
| 49 | // control-plane denial of service. |
| 50 | type scanDeliveryPool struct { |
| 51 | s *Server |
| 52 | ctx context.Context |
| 53 | jobs chan scanDeliveryJob |
| 54 | work chan scanDeliveryJob |
| 55 | done chan objstore.AppRef |
| 56 | wg sync.WaitGroup |
| 57 | client *http.Client |
| 58 | } |
| 59 | |
| 60 | func newScanDeliveryPool(ctx context.Context, s *Server) *scanDeliveryPool { |
| 61 | transport := s.cfg.CallbackTransport |
| 62 | if transport == nil { |
| 63 | // http.DefaultTransport verifies both the public CA chain and the target |
| 64 | // hostname. Do not clone it merely to set TLSClientConfig: that is where |
| 65 | // an InsecureSkipVerify regression can hide. |
| 66 | transport = http.DefaultTransport |
| 67 | } |
| 68 | p := &scanDeliveryPool{ |
| 69 | s: s, ctx: ctx, jobs: make(chan scanDeliveryJob, scanDeliveryQueue), |
| 70 | work: make(chan scanDeliveryJob), done: make(chan objstore.AppRef), |
| 71 | client: &http.Client{ |
| 72 | Transport: transport, |
| 73 | Timeout: scanDeliveryTimeout, |
| 74 | CheckRedirect: func(*http.Request, []*http.Request) error { |
| 75 | return http.ErrUseLastResponse |
| 76 | }, |
| 77 | }, |
| 78 | } |
| 79 | p.wg.Add(1) |
| 80 | go p.dispatch() |
| 81 | for range scanDeliveryWorkers { |
| 82 | p.wg.Add(1) |
| 83 | go p.worker() |
| 84 | } |
| 85 | return p |
| 86 | } |
| 87 | |
| 88 | func (p *scanDeliveryPool) stop() { p.wg.Wait() } |
| 89 | |
| 90 | func (p *scanDeliveryPool) enqueue(job scanDeliveryJob) { |
| 91 | if p == nil || job.path == "" { |
| 92 | return |
| 93 | } |
| 94 | select { |
| 95 | case p.jobs <- job: |
| 96 | case <-p.ctx.Done(): |
| 97 | default: |
| 98 | log.Printf("console: scan callback: %s/%s: dropped — delivery queue full", job.ref.Account, job.ref.App) |
| 99 | } |
| 100 | } |
| 101 | |
| 102 | func (p *scanDeliveryPool) dispatch() { |
| 103 | defer p.wg.Done() |
| 104 | busy := make(map[objstore.AppRef]bool) |
| 105 | queued := make(map[objstore.AppRef]int) |
| 106 | pending := make([]scanDeliveryJob, 0, scanDeliveryQueue) |
| 107 | for { |
| 108 | selected := -1 |
| 109 | for i, job := range pending { |
| 110 | if !busy[job.ref] { |
| 111 | selected = i |
| 112 | break |
| 113 | } |
| 114 | } |
| 115 | if selected >= 0 { |
| 116 | job := pending[selected] |
| 117 | select { |
| 118 | case p.work <- job: |
| 119 | busy[job.ref] = true |
| 120 | queued[job.ref]-- |
| 121 | pending = append(pending[:selected], pending[selected+1:]...) |
| 122 | case ref := <-p.done: |
| 123 | delete(busy, ref) |
| 124 | case <-p.ctx.Done(): |
| 125 | return |
| 126 | } |
| 127 | continue |
| 128 | } |
| 129 | select { |
| 130 | case job := <-p.jobs: |
| 131 | if len(pending) >= scanDeliveryQueue || queued[job.ref] >= 8 { |
| 132 | log.Printf("console: scan callback: %s/%s: dropped — delivery pending cap reached", job.ref.Account, job.ref.App) |
| 133 | } else { |
| 134 | pending = append(pending, job) |
| 135 | queued[job.ref]++ |
| 136 | } |
| 137 | case ref := <-p.done: |
| 138 | delete(busy, ref) |
| 139 | case <-p.ctx.Done(): |
| 140 | return |
| 141 | } |
| 142 | } |
| 143 | } |
| 144 | |
| 145 | func (p *scanDeliveryPool) worker() { |
| 146 | defer p.wg.Done() |
| 147 | for { |
| 148 | select { |
| 149 | case <-p.ctx.Done(): |
| 150 | return |
| 151 | case job := <-p.work: |
| 152 | p.deliver(job) |
| 153 | select { |
| 154 | case p.done <- job.ref: |
| 155 | case <-p.ctx.Done(): |
| 156 | return |
| 157 | } |
| 158 | } |
| 159 | } |
| 160 | } |
| 161 | |
| 162 | func (p *scanDeliveryPool) deliver(job scanDeliveryJob) { |
| 163 | if p.s.cfg.CallbackSigner == nil { |
| 164 | return // an unsigned POST is worse than a dropped best-effort callback |
| 165 | } |
| 166 | host, err := p.s.scanCallbackHost(job.ref) |
| 167 | if err != nil { |
| 168 | log.Printf("console: scan callback: %s/%s: %v", job.ref.Account, job.ref.App, err) |
| 169 | return |
| 170 | } |
| 171 | if host == "" { |
| 172 | return // no ready route means there is no platform host yet; the app polls |
| 173 | } |
| 174 | target, err := callbackTarget(host, job.path) |
| 175 | if err != nil { |
| 176 | log.Printf("console: scan callback: %s/%s: %v", job.ref.Account, job.ref.App, err) |
| 177 | return |
| 178 | } |
| 179 | verdict := job.done.Verdict |
| 180 | if job.done.Gone { |
| 181 | verdict = objstore.Verdict("gone") |
| 182 | } |
| 183 | body, err := json.Marshal(scanCallbackBody{ |
| 184 | Account: job.ref.Account, App: job.ref.App, Key: job.done.Key, |
| 185 | ETag: strings.Trim(job.done.ETag, `"`), Verdict: verdict, |
| 186 | Recorded: job.done.Recorded, Engine: job.done.Engine, Signature: job.done.Signature, |
| 187 | }) |
| 188 | if err != nil { |
| 189 | return |
| 190 | } |
| 191 | for attempt := 0; ; attempt++ { |
| 192 | retry, err := p.attempt(job, target.String(), body) |
| 193 | if errors.Is(err, errScanCallbackNotPermitted) { |
| 194 | return |
| 195 | } |
| 196 | if err != nil { |
| 197 | log.Printf("console: scan callback: %s/%s: %v", job.ref.Account, job.ref.App, err) |
| 198 | } |
| 199 | if !retry || attempt == len(scanDeliveryRetryDelays) { |
| 200 | return |
| 201 | } |
| 202 | t := time.NewTimer(scanDeliveryRetryDelays[attempt]) |
| 203 | select { |
| 204 | case <-p.ctx.Done(): |
| 205 | t.Stop() |
| 206 | return |
| 207 | case <-t.C: |
| 208 | } |
| 209 | } |
| 210 | } |
| 211 | |
| 212 | var errScanCallbackNotPermitted = errors.New("console: scan callback no longer permitted") |
| 213 | |
| 214 | // attempt is one POST as one app mutation, so a foreign exclusive operation |
| 215 | // on the app refuses the delivery. Consent and the callback generation are |
| 216 | // re-read inside it, before anything is sent. |
| 217 | func (p *scanDeliveryPool) attempt(job scanDeliveryJob, target string, body []byte) (bool, error) { |
| 218 | acct, app := job.ref.Account, job.ref.App |
| 219 | var retry bool |
| 220 | err := p.s.claimAppMutation(p.ctx, &Account{Sqid: acct}, app, "scan-callback-delivery", func(held context.Context) error { |
| 221 | if p.s.cfg.HostClaim { |
| 222 | appID, err := p.s.mutationAppID(held, acct, app) |
| 223 | if err == nil { |
| 224 | err = hostops.RequireAppMutation(held, p.s.cfg.Releases, acct, app, appID) |
| 225 | } |
| 226 | if err != nil { |
| 227 | return err |
| 228 | } |
| 229 | } |
| 230 | // BOUNDED. p.ctx is the pool's lifetime context and carries no deadline of |
| 231 | // its own, so a release store that stops answering would hold this worker |
| 232 | // for the life of the process — a quarter of the pool per hung read. |
| 233 | consentCtx, consentCancel := context.WithTimeout(held, storeTimeout) |
| 234 | defer consentCancel() |
| 235 | rec, refusal, err := objstore.QueuePermitted(consentCtx, p.s.cfg.Releases, acct, app) |
| 236 | if err != nil || refusal != objstore.RefusalNone || rec == nil || rec.Scanning == nil || rec.Scanning.WebhookGen != job.gen { |
| 237 | return errScanCallbackNotPermitted |
| 238 | } |
| 239 | var perr error |
| 240 | retry, perr = p.post(held, job, target, body, job.gen) |
| 241 | return perr |
| 242 | }) |
| 243 | return retry || errors.Is(err, hostops.ErrPending), err |
| 244 | } |
| 245 | |
| 246 | func (p *scanDeliveryPool) post(ctx context.Context, job scanDeliveryJob, target string, body []byte, gen int) (bool, error) { |
| 247 | req, err := http.NewRequestWithContext(ctx, http.MethodPost, target, bytes.NewReader(body)) |
| 248 | if err != nil { |
| 249 | return false, err |
| 250 | } |
| 251 | req.Header.Set("Content-Type", "application/json") |
| 252 | req.Header.Set("User-Agent", "carlos-scan-callback/1") |
| 253 | req.Header.Set("Carlos-Signature", p.s.cfg.CallbackSigner.Sign(job.ref.Account, job.ref.App, gen, body, time.Now())) |
| 254 | res, err := p.client.Do(req) |
| 255 | if err != nil { |
| 256 | return !errors.Is(err, context.Canceled), err |
| 257 | } |
| 258 | defer res.Body.Close() |
| 259 | _, _ = io.Copy(io.Discard, io.LimitReader(res.Body, 4<<10)) |
| 260 | if res.StatusCode >= 200 && res.StatusCode < 300 { |
| 261 | return false, nil |
| 262 | } |
| 263 | if res.StatusCode >= 300 && res.StatusCode < 400 { |
| 264 | // SAID OUT LOUD, because silence here is indistinguishable from |
| 265 | // success to the only person who can fix it. Redirects are refused by |
| 266 | // design (a tenant answering 302 to an internal address is the attack), |
| 267 | // but the ordinary cause is dull: a framework that redirects /hook to |
| 268 | // /hook/ turns every callback into a refusal, and nothing anywhere |
| 269 | // would otherwise say so. |
| 270 | log.Printf("console: scan callback %s/%s: %d redirect refused — a callback path must answer directly", |
| 271 | job.ref.Account, job.ref.App, res.StatusCode) |
| 272 | } |
| 273 | retry := res.StatusCode == http.StatusTooManyRequests || res.StatusCode >= 500 |
| 274 | return retry, nil |
| 275 | } |
| 276 | |
| 277 | // scanCallbackHost returns only a platform-assigned name that exists in the |
| 278 | // registry: the canonical app.account.apps-domain when it is a ready platform |
| 279 | // route, otherwise the lexicographically first ready platform route of the app. |
| 280 | func (s *Server) scanCallbackHost(ref objstore.AppRef) (string, error) { |
| 281 | if s.cfg.AppsDomain == "" || s.cfg.Registry == "" { |
| 282 | return "", nil |
| 283 | } |
| 284 | reg, err := openRegistry(s.cfg.Registry) |
| 285 | if err != nil { |
| 286 | return "", err |
| 287 | } |
| 288 | routes, err := reg.List() |
| 289 | if err != nil { |
| 290 | return "", err |
| 291 | } |
| 292 | canonical := ref.App + "." + ref.Account + "." + strings.TrimPrefix(s.cfg.AppsDomain, ".") |
| 293 | best := "" |
| 294 | for _, rt := range routes { |
| 295 | if rt.App != ref.App || release.RouteAccount(rt) != ref.Account || rt.Status != "ready" || |
| 296 | !rt.Platform || rt.Canary || rt.IsSidecar() { |
| 297 | continue |
| 298 | } |
| 299 | if rt.Host == canonical { |
| 300 | return canonical, nil |
| 301 | } |
| 302 | if best == "" || rt.Host < best { |
| 303 | best = rt.Host |
| 304 | } |
| 305 | } |
| 306 | return best, nil |
| 307 | } |
| 308 | |
amadan