| 1 | package console |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/hex" |
| 6 | "encoding/json" |
| 7 | "errors" |
| 8 | "io" |
| 9 | "net/http" |
| 10 | "strings" |
| 11 | "sync/atomic" |
| 12 | "testing" |
| 13 | "time" |
| 14 | |
| 15 | "github.com/carlosframework/platform/internal/objstore" |
| 16 | "github.com/carlosframework/platform/internal/registry" |
| 17 | "github.com/carlosframework/platform/internal/release" |
| 18 | ) |
| 19 | |
| 20 | type callbackRoundTripper func(*http.Request) (*http.Response, error) |
| 21 | |
| 22 | func (f callbackRoundTripper) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) } |
| 23 | |
| 24 | func callbackResponse(code int) *http.Response { |
| 25 | return &http.Response{StatusCode: code, Header: make(http.Header), Body: io.NopCloser(strings.NewReader("ignored"))} |
| 26 | } |
| 27 | |
| 28 | // deliveryArmed is a drain that can actually deliver: a ready route so the |
| 29 | // platform host resolves, a signer so a request may be built at all, and a |
| 30 | // transport the test controls. Anything less and the delivery path is skipped |
| 31 | // rather than exercised, which is how a test of delivery ends up proving |
| 32 | // nothing. |
| 33 | func deliveryArmed(t *testing.T, code string, rt http.RoundTripper) (*consoleFixture, *objstore.MemScanner) { |
| 34 | t.Helper() |
| 35 | return deliveryArmedWith(t, code, rt, callbackRoute(fixtureApp+"."+fixtureAcct+".carlosapp.dev")) |
| 36 | } |
| 37 | |
| 38 | // callbackRoute is a ready platform route of the fixture app, as the domains |
| 39 | // reconciler leaves it. |
| 40 | func callbackRoute(host string) registry.Route { |
| 41 | return registry.Route{Host: host, Kind: "instance", Status: "ready", App: fixtureApp, Account: fixtureAcct, Platform: true} |
| 42 | } |
| 43 | |
| 44 | // deliveryArmedWith is deliveryArmed with the platform routes chosen by the |
| 45 | // caller, added on top of the fixture's custom kass route. |
| 46 | func deliveryArmedWith(t *testing.T, code string, rt http.RoundTripper, routes ...registry.Route) (*consoleFixture, *objstore.MemScanner) { |
| 47 | t.Helper() |
| 48 | fx, sc, _ := drainFixture(t) |
| 49 | for _, r := range routes { |
| 50 | fx.addRoute(t, r) |
| 51 | } |
| 52 | enableStore(t, fx, fixtureAcct, fixtureApp, true) |
| 53 | fx.srv.cfg.AppsDomain = "carlosapp.dev" |
| 54 | fx.srv.cfg.CallbackSigner = callbackTestSigner(t, "master-for-delivery", "") |
| 55 | fx.srv.cfg.CallbackTransport = rt |
| 56 | sc.PutObject(drainBucket(fixtureAcct, fixtureApp), "uploads/x.pdf", "etag-x", time.Now(), 10) |
| 57 | if _, err := objstore.Enqueue(context.Background(), fx.rel, fx.srv.scanQueue, |
| 58 | fixtureAcct, fixtureApp, |
| 59 | &objstore.QueueItem{Key: "uploads/x.pdf", By: "test", CallbackPath: "/hook"}, |
| 60 | time.Now()); err != nil { |
| 61 | t.Fatal(err) |
| 62 | } |
| 63 | return fx, sc |
| 64 | } |
| 65 | |
| 66 | // TestDeliveryDoesNotBlockTheTick drives the REAL tick against an endpoint |
| 67 | // that never answers. |
| 68 | // |
| 69 | // The first version of this test called p.enqueue directly and asserted that |
| 70 | // the call returned — which proves the channel has a buffer, and would pass |
| 71 | // just as well if drainScanApp delivered synchronously somewhere else, because |
| 72 | // it never ran a tick at all. The claim in the name is about the TICK, so the |
| 73 | // tick is what has to be timed. |
| 74 | // |
| 75 | // The failure being ruled out is a tenant pointing a callback at a black hole |
| 76 | // and thereby holding up every other tenant's verdict: 50 items times a retry |
| 77 | // schedule inside a pass that sweeps apps one at a time. |
| 78 | func TestDeliveryDoesNotBlockTheTick(t *testing.T) { |
| 79 | park := make(chan struct{}) |
| 80 | defer close(park) |
| 81 | reached := make(chan struct{}, 1) |
| 82 | fx, _ := deliveryArmed(t, "code-deliver-block", callbackRoundTripper(func(*http.Request) (*http.Response, error) { |
| 83 | select { |
| 84 | case reached <- struct{}{}: |
| 85 | default: |
| 86 | } |
| 87 | <-park |
| 88 | return callbackResponse(204), nil |
| 89 | })) |
| 90 | |
| 91 | ctx, cancel := context.WithCancel(context.Background()) |
| 92 | defer cancel() |
| 93 | fx.srv.scanDelivery = newScanDeliveryPool(ctx, fx.srv) |
| 94 | |
| 95 | start := time.Now() |
| 96 | if err := fx.srv.scanDrainTick(ctx); err != nil { |
| 97 | t.Fatal(err) |
| 98 | } |
| 99 | if took := time.Since(start); took > 3*time.Second { |
| 100 | t.Fatalf("the tick took %v with one callback endpoint hanging — delivery is in the drain's path", took) |
| 101 | } |
| 102 | // And the delivery really was attempted, or the timing above measured a |
| 103 | // path that never made a request. |
| 104 | select { |
| 105 | case <-reached: |
| 106 | case <-time.After(5 * time.Second): |
| 107 | t.Fatal("no callback was attempted, so this test timed nothing") |
| 108 | } |
| 109 | } |
| 110 | |
| 111 | // TestCallbackGoesToPlatformHostOnly asserts where the DELIVERY actually goes. |
| 112 | // |
| 113 | // The first version called callbackTarget and checked its return, which is |
| 114 | // already covered by TestCallbackTargetIsBuiltStructurally in scanhook_test.go |
| 115 | // — it re-tested the helper rather than the caller. What matters here is that |
| 116 | // the pool builds its URL from the app's platform-assigned name and nothing |
| 117 | // from the queue item, so the request itself is what gets inspected. |
| 118 | func TestCallbackGoesToPlatformHostOnly(t *testing.T) { |
| 119 | got := make(chan string, 4) |
| 120 | fx, _ := deliveryArmed(t, "code-deliver-host", callbackRoundTripper(func(r *http.Request) (*http.Response, error) { |
| 121 | got <- r.URL.String() |
| 122 | return callbackResponse(204), nil |
| 123 | })) |
| 124 | |
| 125 | ctx, cancel := context.WithCancel(context.Background()) |
| 126 | defer cancel() |
| 127 | fx.srv.scanDelivery = newScanDeliveryPool(ctx, fx.srv) |
| 128 | if err := fx.srv.scanDrainTick(ctx); err != nil { |
| 129 | t.Fatal(err) |
| 130 | } |
| 131 | select { |
| 132 | case u := <-got: |
| 133 | want := "https://" + fixtureApp + "." + fixtureAcct + ".carlosapp.dev/hook" |
| 134 | if u != want { |
| 135 | t.Fatalf("posted to %q, want %q", u, want) |
| 136 | } |
| 137 | case <-time.After(5 * time.Second): |
| 138 | t.Fatal("no callback was delivered") |
| 139 | } |
| 140 | } |
| 141 | |
| 142 | func TestCallbackRefusesRedirects(t *testing.T) { |
| 143 | ctx, cancel := context.WithCancel(context.Background()) |
| 144 | p := newScanDeliveryPool(ctx, &Server{}) |
| 145 | defer func() { cancel(); p.stop() }() |
| 146 | req, _ := http.NewRequest(http.MethodPost, "https://app.example/hook", nil) |
| 147 | if err := p.client.CheckRedirect(req, nil); !errors.Is(err, http.ErrUseLastResponse) { |
| 148 | t.Fatalf("redirect error = %v", err) |
| 149 | } |
| 150 | } |
| 151 | |
| 152 | func TestTransportDoesNotSkipTLSVerification(t *testing.T) { |
| 153 | ctx, cancel := context.WithCancel(context.Background()) |
| 154 | p := newScanDeliveryPool(ctx, &Server{}) |
| 155 | cancel() |
| 156 | p.stop() |
| 157 | tr, ok := p.client.Transport.(*http.Transport) |
| 158 | if !ok { |
| 159 | t.Fatalf("production transport = %T", p.client.Transport) |
| 160 | } |
| 161 | if tr.TLSClientConfig != nil && tr.TLSClientConfig.InsecureSkipVerify { |
| 162 | t.Fatal("production callback transport disables TLS verification") |
| 163 | } |
| 164 | } |
| 165 | |
| 166 | // TestNeverPostsUnsigned. The first version used a bare &Server{}, which has |
| 167 | // no AppsDomain — so delivery returned at "no platform host" long before the |
| 168 | // signer mattered, and the test passed with the signer guard deleted. Worse, |
| 169 | // with the guard gone AND a real host it would nil-deref and panic a worker. |
| 170 | // |
| 171 | // The fixture therefore has a resolvable host and a real queued item, exactly |
| 172 | // as a delivering deployment does, and only the signer is missing. |
| 173 | func TestNeverPostsUnsigned(t *testing.T) { |
| 174 | var calls atomic.Int32 |
| 175 | fx, _ := deliveryArmed(t, "code-deliver-unsigned", callbackRoundTripper(func(*http.Request) (*http.Response, error) { |
| 176 | calls.Add(1) |
| 177 | return callbackResponse(204), nil |
| 178 | })) |
| 179 | fx.srv.cfg.CallbackSigner = nil // the one thing under test |
| 180 | |
| 181 | ctx, cancel := context.WithCancel(context.Background()) |
| 182 | defer cancel() |
| 183 | fx.srv.scanDelivery = newScanDeliveryPool(ctx, fx.srv) |
| 184 | if err := fx.srv.scanDrainTick(ctx); err != nil { |
| 185 | t.Fatal(err) |
| 186 | } |
| 187 | // Give any wrongly-made request time to arrive before concluding none did. |
| 188 | time.Sleep(300 * time.Millisecond) |
| 189 | if n := calls.Load(); n != 0 { |
| 190 | t.Fatalf("made %d request(s) with no signer — an unsigned callback is the hole §6.3 exists to close", n) |
| 191 | } |
| 192 | } |
| 193 | |
| 194 | // TestDeliveredSecretVerifiesTheHeaderItReceives is the round trip nothing |
| 195 | // covered, and the one an app author's first hour depends on. |
| 196 | // |
| 197 | // The secret handed to an app is HEX (EnvironmentFile values are text), while |
| 198 | // the MAC is keyed on the RAW 32 bytes. An author who feeds the environment |
| 199 | // string straight into HMAC gets a failure on every callback, and the natural |
| 200 | // next move is to stop verifying. So this asserts the whole path: the key as |
| 201 | // delivered, hex-decoded, verifies the header the pool actually sent. |
| 202 | func TestDeliveredSecretVerifiesTheHeaderItReceives(t *testing.T) { |
| 203 | type sent struct { |
| 204 | header string |
| 205 | body []byte |
| 206 | } |
| 207 | got := make(chan sent, 4) |
| 208 | fx, _ := deliveryArmed(t, "code-deliver-roundtrip", callbackRoundTripper(func(r *http.Request) (*http.Response, error) { |
| 209 | body, _ := io.ReadAll(r.Body) |
| 210 | got <- sent{header: r.Header.Get("Carlos-Signature"), body: body} |
| 211 | return callbackResponse(204), nil |
| 212 | })) |
| 213 | |
| 214 | ctx, cancel := context.WithCancel(context.Background()) |
| 215 | defer cancel() |
| 216 | fx.srv.scanDelivery = newScanDeliveryPool(ctx, fx.srv) |
| 217 | if err := fx.srv.scanDrainTick(ctx); err != nil { |
| 218 | t.Fatal(err) |
| 219 | } |
| 220 | select { |
| 221 | case s := <-got: |
| 222 | if s.header == "" { |
| 223 | t.Fatal("no Carlos-Signature header") |
| 224 | } |
| 225 | // Exactly what the app is given, and exactly what it must do with it. |
| 226 | delivered := fx.srv.cfg.CallbackSigner.DerivedKey(fixtureAcct, fixtureApp, 0) |
| 227 | key, err := hex.DecodeString(string(delivered)) |
| 228 | if err != nil { |
| 229 | t.Fatalf("the delivered secret is not hex: %v", err) |
| 230 | } |
| 231 | if err := VerifyCallback(key, s.header, s.body, time.Now()); err != nil { |
| 232 | t.Fatalf("the delivered secret does not verify the header we sent: %v", err) |
| 233 | } |
| 234 | case <-time.After(5 * time.Second): |
| 235 | t.Fatal("no callback was delivered") |
| 236 | } |
| 237 | } |
| 238 | |
| 239 | func TestCallbackRetriesAreBounded(t *testing.T) { |
| 240 | old := scanDeliveryRetryDelays |
| 241 | scanDeliveryRetryDelays = []time.Duration{0, 0, 0} |
| 242 | defer func() { scanDeliveryRetryDelays = old }() |
| 243 | var calls atomic.Int32 |
| 244 | signer := callbackTestSigner(t, "master", "") |
| 245 | p := &scanDeliveryPool{s: &Server{cfg: Config{CallbackSigner: signer}}, ctx: context.Background(), client: &http.Client{ |
| 246 | Transport: callbackRoundTripper(func(*http.Request) (*http.Response, error) { |
| 247 | calls.Add(1) |
| 248 | return callbackResponse(500), nil |
| 249 | }), |
| 250 | }} |
| 251 | job := scanDeliveryJob{ref: objstore.AppRef{Account: "bab", App: "app"}} |
| 252 | body := []byte(`{"verdict":"clean"}`) |
| 253 | for attempt := 0; ; attempt++ { |
| 254 | retry, _ := p.post(context.Background(), job, "https://app.bab.example/hook", body, 1) |
| 255 | if !retry || attempt == len(scanDeliveryRetryDelays) { |
| 256 | break |
| 257 | } |
| 258 | } |
| 259 | if calls.Load() != 4 { |
| 260 | t.Fatalf("attempts = %d, want initial plus three retries", calls.Load()) |
| 261 | } |
| 262 | } |
| 263 | |
| 264 | func TestDroppedCallbackStillReadableFromStatusDoor(t *testing.T) { |
| 265 | ctx := context.Background() |
| 266 | store := release.NewDirStore(t.TempDir()) |
| 267 | want := &objstore.ScanResult{Key: "uploads/x", Version: "opaque", Verdict: string(objstore.VerdictInfected)} |
| 268 | if err := objstore.WriteScanResult(ctx, store, "bab", "app", want); err != nil { |
| 269 | t.Fatal(err) |
| 270 | } |
| 271 | // A full queue takes the documented drop branch. |
| 272 | p := &scanDeliveryPool{ctx: ctx, jobs: make(chan scanDeliveryJob, 1)} |
| 273 | p.jobs <- scanDeliveryJob{} |
| 274 | p.enqueue(scanDeliveryJob{ref: objstore.AppRef{Account: "bab", App: "app"}, path: "/hook"}) |
| 275 | got, err := objstore.ReadScanResult(ctx, store, "bab", "app", "uploads/x", "opaque") |
| 276 | if err != nil || got == nil || got.Verdict != string(objstore.VerdictInfected) { |
| 277 | t.Fatalf("durable status after callback drop = %+v, err = %v", got, err) |
| 278 | } |
| 279 | } |
| 280 | |
| 281 | // TestCallbackBodyCarriesRecorded inspects the body the pool BUILDS, not one |
| 282 | // the test marshals itself. The first version constructed a literal and |
| 283 | // checked its JSON, which passes just as well against a deliver() that |
| 284 | // hard-codes Recorded: true — and Recorded is the field that tells an app |
| 285 | // whether an infected verdict is actually backed by the enforcement deny. |
| 286 | func TestCallbackBodyCarriesRecorded(t *testing.T) { |
| 287 | got := make(chan map[string]any, 4) |
| 288 | fx, sc := deliveryArmed(t, "code-deliver-recorded", callbackRoundTripper(func(r *http.Request) (*http.Response, error) { |
| 289 | raw, _ := io.ReadAll(r.Body) |
| 290 | var body map[string]any |
| 291 | _ = json.Unmarshal(raw, &body) |
| 292 | got <- body |
| 293 | return callbackResponse(204), nil |
| 294 | })) |
| 295 | // The ten-tag case: a verdict is reached and cannot be written onto the |
| 296 | // object, so the app must be told the deny is not in force. |
| 297 | sc.Unrecordable("uploads/x.pdf") |
| 298 | |
| 299 | ctx, cancel := context.WithCancel(context.Background()) |
| 300 | defer cancel() |
| 301 | fx.srv.scanDelivery = newScanDeliveryPool(ctx, fx.srv) |
| 302 | if err := fx.srv.scanDrainTick(ctx); err != nil { |
| 303 | t.Fatal(err) |
| 304 | } |
| 305 | select { |
| 306 | case body := <-got: |
| 307 | if rec, ok := body["recorded"].(bool); !ok || rec { |
| 308 | t.Fatalf("recorded = %v, want false — the verdict is not on the object, so no deny matches it", body["recorded"]) |
| 309 | } |
| 310 | if _, ok := body["version"]; ok { |
| 311 | t.Errorf("body carries the opaque version token: %v", body) |
| 312 | } |
| 313 | if body["etag"] == "" || body["etag"] == nil { |
| 314 | t.Errorf("body carries no etag, which is what an app matches on: %v", body) |
| 315 | } |
| 316 | case <-time.After(5 * time.Second): |
| 317 | t.Fatal("no callback was delivered") |
| 318 | } |
| 319 | } |
| 320 | |
| 321 | func TestCallbackBodyOmitsTheVersionToken(t *testing.T) { |
| 322 | b, err := json.Marshal(scanCallbackBody{Key: "x", ETag: "etag"}) |
| 323 | if err != nil { |
| 324 | t.Fatal(err) |
| 325 | } |
| 326 | if strings.Contains(string(b), "version") { |
| 327 | t.Fatalf("body exposes version token: %s", b) |
| 328 | } |
| 329 | } |
| 330 | |
| 331 | func TestNoPlatformHostIsNotAnError(t *testing.T) { |
| 332 | host, err := (&Server{}).scanCallbackHost(objstore.AppRef{Account: "bab", App: "app"}) |
| 333 | if err != nil || host != "" { |
| 334 | t.Fatalf("host = %q, err = %v", host, err) |
| 335 | } |
| 336 | } |
| 337 | |
| 338 | type callbackHostCase struct { |
| 339 | name string |
| 340 | routes []registry.Route |
| 341 | want string |
| 342 | } |
| 343 | |
| 344 | // callbackHostCases is shared by the helper test and the delivered POST so |
| 345 | // both are held to the same routes. The fixture already holds a ready custom |
| 346 | // kass route (kass.carlosapp.dev, not Platform), which must never win. |
| 347 | func callbackHostCases() []callbackHostCase { |
| 348 | const canon = fixtureApp + "." + fixtureAcct + ".carlosapp.dev" |
| 349 | with := func(mod func(*registry.Route)) registry.Route { |
| 350 | r := callbackRoute(canon) |
| 351 | mod(&r) |
| 352 | return r |
| 353 | } |
| 354 | return []callbackHostCase{ |
| 355 | {"canonical wins over an alphabetically earlier alternate", |
| 356 | []registry.Route{callbackRoute("a." + canon), callbackRoute(canon)}, canon}, |
| 357 | {"only an alternate platform apex uses its actual hostname", |
| 358 | []registry.Route{callbackRoute("kass.alt.example.test")}, "kass.alt.example.test"}, |
| 359 | {"alternates resolve lexicographically", |
| 360 | []registry.Route{callbackRoute("kass.zeta.example.test"), callbackRoute("kass.beta.example.test")}, "kass.beta.example.test"}, |
| 361 | {"custom-only has no host", nil, ""}, |
| 362 | {"wrong account cannot win", |
| 363 | []registry.Route{with(func(r *registry.Route) { r.Account = "zzzz" })}, ""}, |
| 364 | {"wrong app cannot win", |
| 365 | []registry.Route{with(func(r *registry.Route) { r.App = "other" })}, ""}, |
| 366 | {"pending cannot win", |
| 367 | []registry.Route{with(func(r *registry.Route) { r.Status = "pending" })}, ""}, |
| 368 | {"canary cannot win", |
| 369 | []registry.Route{with(func(r *registry.Route) { r.Canary = true })}, ""}, |
| 370 | {"sidecar cannot win", |
| 371 | []registry.Route{with(func(r *registry.Route) { r.Kind = "sidecar" })}, ""}, |
| 372 | {"ineligible canonical yields to an eligible alternate", |
| 373 | []registry.Route{with(func(r *registry.Route) { r.Canary = true }), callbackRoute("kass.alt.example.test")}, "kass.alt.example.test"}, |
| 374 | {"canonical-looking custom route is not platform ownership", |
| 375 | []registry.Route{with(func(r *registry.Route) { r.Platform = false })}, ""}, |
| 376 | } |
| 377 | } |
| 378 | |
| 379 | func TestCallbackHostSelection(t *testing.T) { |
| 380 | for _, tc := range callbackHostCases() { |
| 381 | t.Run(tc.name, func(t *testing.T) { |
| 382 | fx := newConsoleFixture(t) |
| 383 | fx.srv.cfg.AppsDomain = "carlosapp.dev" |
| 384 | for _, r := range tc.routes { |
| 385 | fx.addRoute(t, r) |
| 386 | } |
| 387 | host, err := fx.srv.scanCallbackHost(objstore.AppRef{Account: fixtureAcct, App: fixtureApp}) |
| 388 | if err != nil || host != tc.want { |
| 389 | t.Fatalf("host = %q, err = %v, want %q", host, err, tc.want) |
| 390 | } |
| 391 | }) |
| 392 | } |
| 393 | } |
| 394 | |
| 395 | func TestCallbackPostsToSelectedHost(t *testing.T) { |
| 396 | for _, tc := range callbackHostCases() { |
| 397 | t.Run(tc.name, func(t *testing.T) { |
| 398 | got := make(chan string, 4) |
| 399 | fx, _ := deliveryArmedWith(t, "code-deliver-select", callbackRoundTripper(func(r *http.Request) (*http.Response, error) { |
| 400 | got <- r.URL.String() |
| 401 | return callbackResponse(204), nil |
| 402 | }), tc.routes...) |
| 403 | ctx, cancel := context.WithCancel(context.Background()) |
| 404 | defer cancel() |
| 405 | fx.srv.scanDelivery = newScanDeliveryPool(ctx, fx.srv) |
| 406 | if err := fx.srv.scanDrainTick(ctx); err != nil { |
| 407 | t.Fatal(err) |
| 408 | } |
| 409 | if tc.want == "" { |
| 410 | select { |
| 411 | case u := <-got: |
| 412 | t.Fatalf("posted to %q with no eligible platform route", u) |
| 413 | case <-time.After(500 * time.Millisecond): |
| 414 | } |
| 415 | return |
| 416 | } |
| 417 | select { |
| 418 | case u := <-got: |
| 419 | if want := "https://" + tc.want + "/hook"; u != want { |
| 420 | t.Fatalf("posted to %q, want %q", u, want) |
| 421 | } |
| 422 | case <-time.After(5 * time.Second): |
| 423 | t.Fatal("no callback was delivered") |
| 424 | } |
| 425 | }) |
| 426 | } |
| 427 | } |
| 428 | |
amadan