amadan
console: callback delivery, bounded and backstopped by the door
Task 16, the security-critical one. Written by codex (gpt-5.6-sol); the two rewritten tests below, the mutation testing and this message are mine. Its run was killed for memory partway through writing its report, but the code and tests were already on disk and complete. A BOUNDED WORKER POOL, not delivery in the drain's goroutine. Inline delivery puts a tenant's hanging endpoint directly in the path of every other tenant's verdict — fifty items times a retry schedule, inside a pass that sweeps apps one at a time — which is the starvation §7.2 and §7.3 exist to prevent and something a tenant can trigger by pointing a callback at a black hole. Four workers, a queue of 64 that DROPS and logs on overflow rather than growing, and retries scheduled by the pool. Dropping is defensible only because pull is the source of truth: the status door still answers, so a lost callback costs one poll. The security properties are the point of the task. The host is built from the app's platform-assigned name and nothing from the queue item; every redirect is refused with ErrUseLastResponse, or a tenant answering 302 -> http://10.x/ turns a signed POST into a GET at an internal address; the transport is http.DefaultTransport and is NOT cloned to set TLSClientConfig, with a test asserting InsecureSkipVerify's absence, because TLS name verification is the second lock that makes DNS rebinding irrelevant; nothing is ever posted unsigned; the response body is drained through a 4KiB LimitReader and discarded. The body carries the bare ETag and never the opaque version token — §11.7, the property the version binding exists to protect. TWO TESTS OVERCLAIMED THEIR NAMES AND WERE REWRITTEN. TestDeliveryDoesNotBlockTheTick called p.enqueue directly and asserted the call returned. That proves the channel has a buffer. It never ran a tick, so it would have passed against a drain that delivered synchronously somewhere else — and the claim in the name is about the tick. It now drives the real scanDrainTick against an endpoint that never answers and times it, and separately asserts a request was actually attempted so the timing measured something. TestCallbackGoesToPlatformHostOnly called callbackTarget and checked its return — which scanhook_test.go already covers. It re-tested the helper rather than the caller. It now inspects the URL of the request the pool actually makes. Both were caught by asking a question a green run cannot answer: would this fail if the behaviour were removed? Mutating delivery to run inline now hangs the tick; mutating the host to a foreign name now fails on the request URL; following redirects fails too. Race detector clean over the delivery and drain tests together. The delivery path also settles §6.2's corrected condition in code: scanCallbackHost returns a host only once a ready route proves the domains loop has an upstream to mint one from, so an app with no ready route is skipped rather than posted to https:///hook.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
6 files changed,
+507
−1
internal/console/apiscan.go+10 −0internal/console/apiscanqueue_test.go+3 −0internal/console/scandeliver.go+253 −0internal/console/scandeliver_test.go+225 −0internal/console/scandrain.go+11 −0internal/console/server.go+5 −1
CI passed — run details
diff --git a/internal/console/apiscan.go b/internal/console/apiscan.go| index afd44fc2e..57bad93e4 100644 |
| --- a/internal/console/apiscan.go |
| +++ b/internal/console/apiscan.go |
| @@ -915,6 +915,16 @@ func (s *Server) handleAPIScanEnqueue(w http.ResponseWriter, r *http.Request) { |
| "scan callbacks are not configured for this deployment (CARLOS_SCAN_WEBHOOK_KEY)") |
| return |
| } |
| + host, err := s.scanCallbackHost(objstore.AppRef{Account: acct.Sqid, App: app}) |
| + if err != nil { |
| + log.Printf("console: api scan enqueue %s/%s: platform host: %v", acct.Sqid, app, err) |
| + jsonError(w, http.StatusBadGateway, "the app's platform route could not be read — try again") |
| + return |
| + } |
| + if host == "" { |
| + jsonError(w, http.StatusConflict, "this app has no ready route yet — omit callback_path and poll scan status") |
| + return |
| + } |
| } |
| if s.cfg.Releases == nil { |
| scanNoEngineState(w) |
diff --git a/internal/console/apiscanqueue_test.go b/internal/console/apiscanqueue_test.go| index 4001640f6..3ebf595f1 100644 |
| --- a/internal/console/apiscanqueue_test.go |
| +++ b/internal/console/apiscanqueue_test.go |
| @@ -8,6 +8,7 @@ import ( |
| "time" |
| "github.com/carlosframework/platform/internal/objstore" |
| + "github.com/carlosframework/platform/internal/registry" |
| "github.com/carlosframework/platform/internal/release" |
| "github.com/carlosframework/platform/internal/svcrole" |
| ) |
| @@ -49,6 +50,8 @@ func enqueueArmed(t *testing.T, code string) (*consoleFixture, string) { |
| // while the deployment switch is off (it would be enabling a control that |
| // is not running). |
| fx.srv.cfg.StoreScanner = objstore.NewMemScanner() |
| + fx.srv.cfg.AppsDomain = "oncarlos.test" |
| + fx.addRoute(t, registry.Route{Host: "custom.example.test", Kind: "instance", Status: "ready", App: "titogo", Account: release.OpsAccount}) |
| enableDeploymentScanning(t, fx) |
| fx.srv.cfg.StoreOperators = []string{"alice@keymail.dev"} |
| if s, o := postJSON(t, scanURL(fx, "titogo")+"/grant", map[string]any{}, token); s != http.StatusOK { |
diff --git a/internal/console/scandeliver.go b/internal/console/scandeliver.go| new file mode 100644 |
| index 000000000..00388b8a9 |
| --- /dev/null |
| +++ b/internal/console/scandeliver.go |
| @@ -0,0 +1,253 @@ |
| +package console |
| + |
| +import ( |
| + "bytes" |
| + "context" |
| + "encoding/json" |
| + "errors" |
| + "io" |
| + "log" |
| + "net/http" |
| + "strings" |
| + "sync" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/objstore" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +const ( |
| + scanDeliveryWorkers = 4 |
| + scanDeliveryQueue = 64 |
| + scanDeliveryTimeout = 10 * time.Second |
| +) |
| + |
| +var scanDeliveryRetryDelays = []time.Duration{time.Second, 4 * time.Second, 16 * time.Second} |
| + |
| +type scanDeliveryJob struct { |
| + ref objstore.AppRef |
| + done objstore.Drained |
| + path string |
| + gen int |
| +} |
| + |
| +type scanCallbackBody struct { |
| + Account string `json:"account"` |
| + App string `json:"app"` |
| + Key string `json:"key"` |
| + ETag string `json:"etag"` |
| + Verdict objstore.Verdict `json:"verdict"` |
| + Recorded bool `json:"recorded"` |
| + Engine string `json:"engine"` |
| + Signature string `json:"signature"` |
| +} |
| + |
| +// scanDeliveryPool keeps tenant HTTP outside the drain goroutine. Its queue is |
| +// intentionally bounded: the status door is durable, while allowing callbacks |
| +// to grow memory without limit would turn a best-effort convenience into a |
| +// control-plane denial of service. |
| +type scanDeliveryPool struct { |
| + s *Server |
| + ctx context.Context |
| + jobs chan scanDeliveryJob |
| + work chan scanDeliveryJob |
| + done chan objstore.AppRef |
| + wg sync.WaitGroup |
| + client *http.Client |
| +} |
| + |
| +func newScanDeliveryPool(ctx context.Context, s *Server) *scanDeliveryPool { |
| + transport := s.cfg.CallbackTransport |
| + if transport == nil { |
| + // http.DefaultTransport verifies both the public CA chain and the target |
| + // hostname. Do not clone it merely to set TLSClientConfig: that is where |
| + // an InsecureSkipVerify regression can hide. |
| + transport = http.DefaultTransport |
| + } |
| + p := &scanDeliveryPool{ |
| + s: s, ctx: ctx, jobs: make(chan scanDeliveryJob, scanDeliveryQueue), |
| + work: make(chan scanDeliveryJob), done: make(chan objstore.AppRef), |
| + client: &http.Client{ |
| + Transport: transport, |
| + Timeout: scanDeliveryTimeout, |
| + CheckRedirect: func(*http.Request, []*http.Request) error { |
| + return http.ErrUseLastResponse |
| + }, |
| + }, |
| + } |
| + p.wg.Add(1) |
| + go p.dispatch() |
| + for range scanDeliveryWorkers { |
| + p.wg.Add(1) |
| + go p.worker() |
| + } |
| + return p |
| +} |
| + |
| +func (p *scanDeliveryPool) stop() { p.wg.Wait() } |
| + |
| +func (p *scanDeliveryPool) enqueue(job scanDeliveryJob) { |
| + if p == nil || job.path == "" { |
| + return |
| + } |
| + select { |
| + case p.jobs <- job: |
| + case <-p.ctx.Done(): |
| + default: |
| + log.Printf("console: scan callback: %s/%s: dropped — delivery queue full", job.ref.Account, job.ref.App) |
| + } |
| +} |
| + |
| +func (p *scanDeliveryPool) dispatch() { |
| + defer p.wg.Done() |
| + busy := make(map[objstore.AppRef]bool) |
| + queued := make(map[objstore.AppRef]int) |
| + pending := make([]scanDeliveryJob, 0, scanDeliveryQueue) |
| + for { |
| + selected := -1 |
| + for i, job := range pending { |
| + if !busy[job.ref] { |
| + selected = i |
| + break |
| + } |
| + } |
| + if selected >= 0 { |
| + job := pending[selected] |
| + select { |
| + case p.work <- job: |
| + busy[job.ref] = true |
| + queued[job.ref]-- |
| + pending = append(pending[:selected], pending[selected+1:]...) |
| + case ref := <-p.done: |
| + delete(busy, ref) |
| + case <-p.ctx.Done(): |
| + return |
| + } |
| + continue |
| + } |
| + select { |
| + case job := <-p.jobs: |
| + if len(pending) >= scanDeliveryQueue || queued[job.ref] >= 8 { |
| + log.Printf("console: scan callback: %s/%s: dropped — delivery pending cap reached", job.ref.Account, job.ref.App) |
| + } else { |
| + pending = append(pending, job) |
| + queued[job.ref]++ |
| + } |
| + case ref := <-p.done: |
| + delete(busy, ref) |
| + case <-p.ctx.Done(): |
| + return |
| + } |
| + } |
| +} |
| + |
| +func (p *scanDeliveryPool) worker() { |
| + defer p.wg.Done() |
| + for { |
| + select { |
| + case <-p.ctx.Done(): |
| + return |
| + case job := <-p.work: |
| + p.deliver(job) |
| + select { |
| + case p.done <- job.ref: |
| + case <-p.ctx.Done(): |
| + return |
| + } |
| + } |
| + } |
| +} |
| + |
| +func (p *scanDeliveryPool) deliver(job scanDeliveryJob) { |
| + if p.s.cfg.CallbackSigner == nil { |
| + return // an unsigned POST is worse than a dropped best-effort callback |
| + } |
| + host, err := p.s.scanCallbackHost(job.ref) |
| + if err != nil { |
| + log.Printf("console: scan callback: %s/%s: %v", job.ref.Account, job.ref.App, err) |
| + return |
| + } |
| + if host == "" { |
| + return // no ready route means there is no platform host yet; the app polls |
| + } |
| + target, err := callbackTarget(host, job.path) |
| + if err != nil { |
| + log.Printf("console: scan callback: %s/%s: %v", job.ref.Account, job.ref.App, err) |
| + return |
| + } |
| + rec, refusal, err := objstore.QueuePermitted(p.ctx, p.s.cfg.Releases, job.ref.Account, job.ref.App) |
| + if err != nil || refusal != objstore.RefusalNone || rec == nil || rec.Scanning == nil || rec.Scanning.WebhookGen != job.gen { |
| + return |
| + } |
| + verdict := job.done.Verdict |
| + if job.done.Gone { |
| + verdict = objstore.Verdict("gone") |
| + } |
| + body, err := json.Marshal(scanCallbackBody{ |
| + Account: job.ref.Account, App: job.ref.App, Key: job.done.Key, |
| + ETag: strings.Trim(job.done.ETag, `"`), Verdict: verdict, |
| + Recorded: job.done.Recorded, Engine: job.done.Engine, Signature: job.done.Signature, |
| + }) |
| + if err != nil { |
| + return |
| + } |
| + for attempt := 0; ; attempt++ { |
| + retry, err := p.post(job, target.String(), body, job.gen) |
| + if err != nil { |
| + log.Printf("console: scan callback: %s/%s: %v", job.ref.Account, job.ref.App, err) |
| + } |
| + if !retry || attempt == len(scanDeliveryRetryDelays) { |
| + return |
| + } |
| + t := time.NewTimer(scanDeliveryRetryDelays[attempt]) |
| + select { |
| + case <-p.ctx.Done(): |
| + t.Stop() |
| + return |
| + case <-t.C: |
| + } |
| + } |
| +} |
| + |
| +func (p *scanDeliveryPool) post(job scanDeliveryJob, target string, body []byte, gen int) (bool, error) { |
| + req, err := http.NewRequestWithContext(p.ctx, http.MethodPost, target, bytes.NewReader(body)) |
| + if err != nil { |
| + return false, err |
| + } |
| + req.Header.Set("Content-Type", "application/json") |
| + req.Header.Set("User-Agent", "carlos-scan-callback/1") |
| + req.Header.Set("Carlos-Signature", p.s.cfg.CallbackSigner.Sign(job.ref.Account, job.ref.App, gen, body, time.Now())) |
| + res, err := p.client.Do(req) |
| + if err != nil { |
| + return !errors.Is(err, context.Canceled), err |
| + } |
| + defer res.Body.Close() |
| + _, drainErr := io.Copy(io.Discard, io.LimitReader(res.Body, 4<<10)) |
| + if res.StatusCode >= 200 && res.StatusCode < 300 { |
| + return drainErr != nil, drainErr |
| + } |
| + retry := res.StatusCode == http.StatusTooManyRequests || res.StatusCode >= 500 |
| + return retry, nil |
| +} |
| + |
| +// scanCallbackHost returns only the platform-assigned name, but only after a |
| +// ready route proves the domains loop has an upstream from which to mint it. |
| +func (s *Server) scanCallbackHost(ref objstore.AppRef) (string, error) { |
| + if s.cfg.AppsDomain == "" || s.cfg.Registry == "" { |
| + return "", nil |
| + } |
| + reg, err := openRegistry(s.cfg.Registry) |
| + if err != nil { |
| + return "", err |
| + } |
| + routes, err := reg.List() |
| + if err != nil { |
| + return "", err |
| + } |
| + for _, rt := range routes { |
| + if rt.App == ref.App && release.RouteAccount(rt) == ref.Account && rt.Status == "ready" { |
| + return ref.App + "." + ref.Account + "." + strings.TrimPrefix(s.cfg.AppsDomain, "."), nil |
| + } |
| + } |
| + return "", nil |
| +} |
diff --git a/internal/console/scandeliver_test.go b/internal/console/scandeliver_test.go| new file mode 100644 |
| index 000000000..a9e4bb9cc |
| --- /dev/null |
| +++ b/internal/console/scandeliver_test.go |
| @@ -0,0 +1,225 @@ |
| +package console |
| + |
| +import ( |
| + "context" |
| + "encoding/json" |
| + "errors" |
| + "io" |
| + "net/http" |
| + "strings" |
| + "sync/atomic" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/objstore" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +type callbackRoundTripper func(*http.Request) (*http.Response, error) |
| + |
| +func (f callbackRoundTripper) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) } |
| + |
| +func callbackResponse(code int) *http.Response { |
| + return &http.Response{StatusCode: code, Header: make(http.Header), Body: io.NopCloser(strings.NewReader("ignored"))} |
| +} |
| + |
| +// deliveryArmed is a drain that can actually deliver: a ready route so the |
| +// platform host resolves, a signer so a request may be built at all, and a |
| +// transport the test controls. Anything less and the delivery path is skipped |
| +// rather than exercised, which is how a test of delivery ends up proving |
| +// nothing. |
| +func deliveryArmed(t *testing.T, code string, rt http.RoundTripper) (*consoleFixture, *objstore.MemScanner) { |
| + t.Helper() |
| + fx, sc, _ := drainFixture(t) |
| + enableStore(t, fx, fixtureAcct, fixtureApp, true) |
| + fx.srv.cfg.AppsDomain = "carlosapp.dev" |
| + fx.srv.cfg.CallbackSigner = callbackTestSigner(t, "master-for-delivery", "") |
| + fx.srv.cfg.CallbackTransport = rt |
| + sc.PutObject(drainBucket(fixtureAcct, fixtureApp), "uploads/x.pdf", "etag-x", time.Now(), 10) |
| + if _, err := objstore.Enqueue(context.Background(), fx.rel, fx.srv.scanQueue, |
| + fixtureAcct, fixtureApp, |
| + &objstore.QueueItem{Key: "uploads/x.pdf", By: "test", CallbackPath: "/hook"}, |
| + time.Now()); err != nil { |
| + t.Fatal(err) |
| + } |
| + return fx, sc |
| +} |
| + |
| +// TestDeliveryDoesNotBlockTheTick drives the REAL tick against an endpoint |
| +// that never answers. |
| +// |
| +// The first version of this test called p.enqueue directly and asserted that |
| +// the call returned — which proves the channel has a buffer, and would pass |
| +// just as well if drainScanApp delivered synchronously somewhere else, because |
| +// it never ran a tick at all. The claim in the name is about the TICK, so the |
| +// tick is what has to be timed. |
| +// |
| +// The failure being ruled out is a tenant pointing a callback at a black hole |
| +// and thereby holding up every other tenant's verdict: 50 items times a retry |
| +// schedule inside a pass that sweeps apps one at a time. |
| +func TestDeliveryDoesNotBlockTheTick(t *testing.T) { |
| + park := make(chan struct{}) |
| + defer close(park) |
| + reached := make(chan struct{}, 1) |
| + fx, _ := deliveryArmed(t, "code-deliver-block", callbackRoundTripper(func(*http.Request) (*http.Response, error) { |
| + select { |
| + case reached <- struct{}{}: |
| + default: |
| + } |
| + <-park |
| + return callbackResponse(204), nil |
| + })) |
| + |
| + ctx, cancel := context.WithCancel(context.Background()) |
| + defer cancel() |
| + fx.srv.scanDelivery = newScanDeliveryPool(ctx, fx.srv) |
| + |
| + start := time.Now() |
| + if err := fx.srv.scanDrainTick(ctx); err != nil { |
| + t.Fatal(err) |
| + } |
| + if took := time.Since(start); took > 3*time.Second { |
| + t.Fatalf("the tick took %v with one callback endpoint hanging — delivery is in the drain's path", took) |
| + } |
| + // And the delivery really was attempted, or the timing above measured a |
| + // path that never made a request. |
| + select { |
| + case <-reached: |
| + case <-time.After(5 * time.Second): |
| + t.Fatal("no callback was attempted, so this test timed nothing") |
| + } |
| +} |
| + |
| +// TestCallbackGoesToPlatformHostOnly asserts where the DELIVERY actually goes. |
| +// |
| +// The first version called callbackTarget and checked its return, which is |
| +// already covered by TestCallbackTargetIsBuiltStructurally in scanhook_test.go |
| +// — it re-tested the helper rather than the caller. What matters here is that |
| +// the pool builds its URL from the app's platform-assigned name and nothing |
| +// from the queue item, so the request itself is what gets inspected. |
| +func TestCallbackGoesToPlatformHostOnly(t *testing.T) { |
| + got := make(chan string, 4) |
| + fx, _ := deliveryArmed(t, "code-deliver-host", callbackRoundTripper(func(r *http.Request) (*http.Response, error) { |
| + got <- r.URL.String() |
| + return callbackResponse(204), nil |
| + })) |
| + |
| + ctx, cancel := context.WithCancel(context.Background()) |
| + defer cancel() |
| + fx.srv.scanDelivery = newScanDeliveryPool(ctx, fx.srv) |
| + if err := fx.srv.scanDrainTick(ctx); err != nil { |
| + t.Fatal(err) |
| + } |
| + select { |
| + case u := <-got: |
| + want := "https://" + fixtureApp + "." + fixtureAcct + ".carlosapp.dev/hook" |
| + if u != want { |
| + t.Fatalf("posted to %q, want %q", u, want) |
| + } |
| + case <-time.After(5 * time.Second): |
| + t.Fatal("no callback was delivered") |
| + } |
| +} |
| + |
| +func TestCallbackRefusesRedirects(t *testing.T) { |
| + ctx, cancel := context.WithCancel(context.Background()) |
| + p := newScanDeliveryPool(ctx, &Server{}) |
| + defer func() { cancel(); p.stop() }() |
| + req, _ := http.NewRequest(http.MethodPost, "https://app.example/hook", nil) |
| + if err := p.client.CheckRedirect(req, nil); !errors.Is(err, http.ErrUseLastResponse) { |
| + t.Fatalf("redirect error = %v", err) |
| + } |
| +} |
| + |
| +func TestTransportDoesNotSkipTLSVerification(t *testing.T) { |
| + ctx, cancel := context.WithCancel(context.Background()) |
| + p := newScanDeliveryPool(ctx, &Server{}) |
| + cancel() |
| + p.stop() |
| + tr, ok := p.client.Transport.(*http.Transport) |
| + if !ok { |
| + t.Fatalf("production transport = %T", p.client.Transport) |
| + } |
| + if tr.TLSClientConfig != nil && tr.TLSClientConfig.InsecureSkipVerify { |
| + t.Fatal("production callback transport disables TLS verification") |
| + } |
| +} |
| + |
| +func TestNeverPostsUnsigned(t *testing.T) { |
| + var calls atomic.Int32 |
| + s := &Server{cfg: Config{CallbackTransport: callbackRoundTripper(func(*http.Request) (*http.Response, error) { |
| + calls.Add(1) |
| + return callbackResponse(204), nil |
| + })}} |
| + p := &scanDeliveryPool{s: s} |
| + p.deliver(scanDeliveryJob{}) |
| + if calls.Load() != 0 { |
| + t.Fatal("made an unsigned request") |
| + } |
| +} |
| + |
| +func TestCallbackRetriesAreBounded(t *testing.T) { |
| + old := scanDeliveryRetryDelays |
| + scanDeliveryRetryDelays = []time.Duration{0, 0, 0} |
| + defer func() { scanDeliveryRetryDelays = old }() |
| + var calls atomic.Int32 |
| + signer := callbackTestSigner(t, "master", "") |
| + p := &scanDeliveryPool{s: &Server{cfg: Config{CallbackSigner: signer}}, ctx: context.Background(), client: &http.Client{ |
| + Transport: callbackRoundTripper(func(*http.Request) (*http.Response, error) { |
| + calls.Add(1) |
| + return callbackResponse(500), nil |
| + }), |
| + }} |
| + job := scanDeliveryJob{ref: objstore.AppRef{Account: "bab", App: "app"}} |
| + body := []byte(`{"verdict":"clean"}`) |
| + for attempt := 0; ; attempt++ { |
| + retry, _ := p.post(job, "https://app.bab.example/hook", body, 1) |
| + if !retry || attempt == len(scanDeliveryRetryDelays) { |
| + break |
| + } |
| + } |
| + if calls.Load() != 4 { |
| + t.Fatalf("attempts = %d, want initial plus three retries", calls.Load()) |
| + } |
| +} |
| + |
| +func TestDroppedCallbackStillReadableFromStatusDoor(t *testing.T) { |
| + ctx := context.Background() |
| + store := release.NewDirStore(t.TempDir()) |
| + want := &objstore.ScanResult{Key: "uploads/x", Version: "opaque", Verdict: string(objstore.VerdictInfected)} |
| + if err := objstore.WriteScanResult(ctx, store, "bab", "app", want); err != nil { |
| + t.Fatal(err) |
| + } |
| + // A full queue takes the documented drop branch. |
| + p := &scanDeliveryPool{ctx: ctx, jobs: make(chan scanDeliveryJob, 1)} |
| + p.jobs <- scanDeliveryJob{} |
| + p.enqueue(scanDeliveryJob{ref: objstore.AppRef{Account: "bab", App: "app"}, path: "/hook"}) |
| + got, err := objstore.ReadScanResult(ctx, store, "bab", "app", "uploads/x", "opaque") |
| + if err != nil || got == nil || got.Verdict != string(objstore.VerdictInfected) { |
| + t.Fatalf("durable status after callback drop = %+v, err = %v", got, err) |
| + } |
| +} |
| + |
| +func TestCallbackBodyCarriesRecorded(t *testing.T) { |
| + b, err := json.Marshal(scanCallbackBody{Recorded: true}) |
| + if err != nil || !strings.Contains(string(b), `"recorded":true`) { |
| + t.Fatalf("body = %s, err = %v", b, err) |
| + } |
| +} |
| + |
| +func TestCallbackBodyOmitsTheVersionToken(t *testing.T) { |
| + b, err := json.Marshal(scanCallbackBody{Key: "x", ETag: "etag"}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if strings.Contains(string(b), "version") { |
| + t.Fatalf("body exposes version token: %s", b) |
| + } |
| +} |
| + |
| +func TestNoPlatformHostIsNotAnError(t *testing.T) { |
| + host, err := (&Server{}).scanCallbackHost(objstore.AppRef{Account: "bab", App: "app"}) |
| + if err != nil || host != "" { |
| + t.Fatalf("host = %q, err = %v", host, err) |
| + } |
| +} |
diff --git a/internal/console/scandrain.go b/internal/console/scandrain.go| index 39aeae949..82cff6ff9 100644 |
| --- a/internal/console/scandrain.go |
| +++ b/internal/console/scandrain.go |
| @@ -55,6 +55,11 @@ func (s *Server) StartScanDrain(ctx context.Context) { |
| if s.cfg.Releases == nil || s.cfg.StoreScanner == nil { |
| return // no bucket, or no engine: nothing to drain into |
| } |
| + s.scanDelivery = newScanDeliveryPool(ctx, s) |
| + defer func() { |
| + s.scanDelivery.stop() |
| + s.scanDelivery = nil |
| + }() |
| rebuilt := true |
| if err := s.rebuildScanQueue(ctx); err != nil { |
| rebuilt = false |
| @@ -198,6 +203,7 @@ func (s *Server) drainScanApp(ctx context.Context, ref objstore.AppRef, now time |
| ctx, cancel := context.WithTimeout(ctx, scanDrainAppBudget) |
| defer cancel() |
| + startRec, _, _ := objstore.Get(ctx, s.cfg.Releases, ref.Account, ref.App) |
| done, _, err := objstore.DrainApp(ctx, s.cfg.Releases, s.cfg.StoreScanner, s.scanQueue, |
| ref.Account, ref.App, scanDrainPerApp, now) |
| @@ -208,6 +214,11 @@ func (s *Server) drainScanApp(ctx context.Context, ref objstore.AppRef, now time |
| // that already exists. |
| for _, d := range done { |
| s.scanNotify(ref, d.Key) |
| + for _, path := range d.CallbackPaths { |
| + if startRec != nil && startRec.Scanning != nil { |
| + s.scanDelivery.enqueue(scanDeliveryJob{ref: ref, done: d, path: path, gen: startRec.Scanning.WebhookGen}) |
| + } |
| + } |
| } |
| switch { |
diff --git a/internal/console/server.go b/internal/console/server.go| index 99c54f4cc..355887863 100644 |
| --- a/internal/console/server.go |
| +++ b/internal/console/server.go |
| @@ -561,6 +561,9 @@ type Config struct { |
| // after grant. nil is a supported deployment shape; callback enrollment and |
| // key delivery refuse rather than ever falling back to unsigned delivery. |
| CallbackSigner *CallbackSigner |
| + // CallbackTransport is the scan callback test seam. nil uses the production |
| + // transport, including ordinary TLS certificate and hostname verification. |
| + CallbackTransport http.RoundTripper |
| // StoreDeployment is the deployment-wide half of every store: which |
| // backend, which region and endpoint buckets are made in, and the prefix |
| // minted bucket names carry. Threaded into every walk so this config is |
| @@ -698,7 +701,8 @@ type Server struct { |
| // goroutine while the drain reads it from its own. A nil map assigned on |
| // first use would be the race, and a Server that always has one cannot |
| // have it. |
| - scanQueue *objstore.QueueIndex |
| + scanQueue *objstore.QueueIndex |
| + scanDelivery *scanDeliveryPool |
| // scanWaits is the in-process wake the status door's long-poll selects on |
| // (scandrain.go). Zero value is usable, which is why it is a value rather |
| // than a pointer: a Server built by a test that never starts the drain |