carlos / platform Public

Clone
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1package objstore
2
3import (
4 "context"
5 "crypto/md5"
6 "encoding/hex"
7 "errors"
8 "os"
9 "sort"
10 "strings"
11 "testing"
12 "time"
13
14 "github.com/carlosframework/platform/internal/release"
15)
16
17// memLister lists a MemProvisioner bucket, paginated, with real content-hash
18// ETags so the etag binding in Verdict is exercised rather than stubbed away.
19type memLister struct {
20 p *MemProvisioner
21 pageSize int
22 calls int
23 // modified is each key's last-modified time. It is half of the version a
24 // verdict binds to, so a lister that left it zero would make every sweep
25 // test blind to that half.
26 modified map[string]time.Time
27 // before runs at the top of each ListPage, so a test can make the world
28 // change underneath a sweep at the one moment that matters.
29 before func()
30}
31
32func (l *memLister) touch(key string, at time.Time) {
33 if l.modified == nil {
34 l.modified = map[string]time.Time{}
35 }
36 l.modified[key] = at
37}
38
39func (l *memLister) ListPage(_ context.Context, bucket, startAfter string, _ int32) ([]ScanObject, bool, error) {
40 l.calls++
41 if l.before != nil {
42 l.before()
43 }
44 keys := make([]string, 0, len(l.p.Buckets[bucket]))
45 for k := range l.p.Buckets[bucket] {
46 if k > startAfter {
47 keys = append(keys, k)
48 }
49 }
50 sort.Strings(keys)
51
52 size := l.pageSize
53 if size <= 0 {
54 size = sweepPageSize
55 }
56 truncated := len(keys) > size
57 if truncated {
58 keys = keys[:size]
59 }
60 objs := make([]ScanObject, 0, len(keys))
61 for _, k := range keys {
62 body := l.p.Buckets[bucket][k]
63 sum := md5.Sum(body)
64 mod := l.modified[k]
65 if mod.IsZero() {
66 mod = time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC)
67 }
68 objs = append(objs, ScanObject{
69 Key: k,
70 ETag: hex.EncodeToString(sum[:]),
71 Modified: mod,
72 Size: int64(len(body)),
73 })
74 }
75 return objs, truncated, nil
76}
77
78// refuseLister fails the test if it is ever asked for a page — for the cases
79// where the right amount of work is none.
80type refuseLister struct{ t *testing.T }
81
82func (l *refuseLister) ListPage(context.Context, string, string, int32) ([]ScanObject, bool, error) {
83 l.t.Fatal("listed a bucket that should never have been swept")
84 return nil, false, nil
85}
86
87// sweepFixture stands up a provisioned, scanning-enabled store with n objects.
88func sweepFixture(t *testing.T, n int) (release.Store, *MemProvisioner, *MemScanner, string) {
89 t.Helper()
90 s := release.NewDirStore(t.TempDir())
91 _, pub := testSealKeypair(t)
92 p := NewMemProvisioner("eu-west-1")
93 rec := rotateFixture(t, s, pub, p)
94 enableScanning(t, s, "bab", "titogo")
95
96 bucket := rec.Provisioned.Bucket
97 for i := 0; i < n; i++ {
98 p.Buckets[bucket][objectName(i)] = []byte(strings.Repeat("x", i+1))
99 }
100 return s, p, NewMemScanner(), bucket
101}
102
103func objectName(i int) string {
104 return "uploads/" + string(rune('a'+i)) + ".txt"
105}
106
107// sweepToCompletion drives RunSweep until a sweep finishes, so a paginated
108// test does not have to guess how many ticks that takes. It bounds the loop:
109// a sweep that never completes is a bug, not a slow test.
110func sweepToCompletion(t *testing.T, s release.Store, l ObjectLister, sc Scanner) *ScanSidecar {
111 t.Helper()
112 var side *ScanSidecar
113 for i := 0; i < 50; i++ {
114 var err error
115 side, err = RunSweep(context.Background(), s, l, sc, "bab", "titogo", 0)
116 if err != nil {
117 t.Fatalf("sweep %d: %v", i, err)
118 }
119 if side != nil && side.LastComplete != nil {
120 return side
121 }
122 }
123 t.Fatal("sweep never completed in 50 ticks")
124 return nil
125}
126
127// A synchronous engine's FIRST pass reports objects as scanned. Counting them
128// unscanned because "the next sweep reads the result" is GuardDuty's shape,
129// and applying it to ClamAV would make the audit sentence false for a full
130// re-sweep interval.
131func TestSweepCountsSynchronousVerdicts(t *testing.T) {
132 s, p, sc, _ := sweepFixture(t, 3)
133
134 side, err := RunSweep(context.Background(), s, &memLister{p: p}, sc, "bab", "titogo", time.Minute)
135 if err != nil {
136 t.Fatal(err)
137 }
138 if side.LastComplete == nil {
139 t.Fatal("a sweep that listed every object in one page did not complete")
140 }
141 got := side.LastComplete
142 if got.Objects != 3 || got.Scanned != 3 || got.Unscanned != 0 {
143 t.Fatalf("objects=%d scanned=%d unscanned=%d, want 3/3/0", got.Objects, got.Scanned, got.Unscanned)
144 }
145 if got.Engine != "clamav" || got.Signature == "" {
146 t.Fatalf("no provenance recorded: engine=%q signature=%q", got.Engine, got.Signature)
147 }
148 if side.Current != nil {
149 t.Fatal("a completed sweep left Current set")
150 }
151}
152
153// LastComplete survives the tick after completion. A single Sweep field would
154// overwrite the audit answer with page-one partials every cycle — the
155// compliance artifact destroyed on a timer.
156func TestSweepLastCompleteSurvivesNextTick(t *testing.T) {
157 s, p, sc, _ := sweepFixture(t, 5)
158 l := &memLister{p: p, pageSize: 2}
159
160 side := sweepToCompletion(t, s, l, sc)
161 if side.LastComplete.Objects != 5 {
162 t.Fatalf("completed sweep counted %d objects, want 5", side.LastComplete.Objects)
163 }
164
165 // One more tick: a fresh sweep starts and gets exactly one page in.
166 next, err := RunSweep(context.Background(), s, l, sc, "bab", "titogo", 0)
167 if err != nil {
168 t.Fatal(err)
169 }
170 if next.Current == nil || next.Current.Objects != 2 {
171 t.Fatalf("the next tick did not start a fresh partial sweep: %+v", next.Current)
172 }
173 if next.LastComplete == nil || next.LastComplete.Objects != 5 {
174 t.Fatalf("page-one partials overwrote the audit answer: %+v", next.LastComplete)
175 }
176}
177
178// Interrupted mid-bucket and re-run, a sweep reaches the same rollup without
179// double-counting what it already saw.
180func TestSweepResumesWithoutDoubleCounting(t *testing.T) {
181 s, p, sc, _ := sweepFixture(t, 5)
182 l := &memLister{p: p, pageSize: 2}
183
184 side := sweepToCompletion(t, s, l, sc)
185 if l.calls < 3 {
186 t.Fatalf("the fixture did not actually paginate: %d listing calls", l.calls)
187 }
188 if side.LastComplete.Objects != 5 || side.LastComplete.Scanned != 5 {
189 t.Fatalf("resumed sweep counted objects=%d scanned=%d, want 5/5",
190 side.LastComplete.Objects, side.LastComplete.Scanned)
191 }
192}
193
194// A fresh sweep over a bucket whose app is torn down between the listing and
195// the write must leave NOTHING behind: an orphan scan.json under a purged
196// prefix makes the app name permanently unclaimable, because the claim gate
197// refuses any name whose prefix still holds a key and box-side auto-purge
198// never reaches it.
199//
200// The first write of a fresh sweep is create-only by construction, so the CAS
201// retry guard cannot catch this on its own — only the pre-write re-read can.
202func TestSweepDoesNotResurrectTornDownApp(t *testing.T) {
203 ctx := context.Background()
204 s, p, sc, _ := sweepFixture(t, 3)
205
206 l := &memLister{p: p}
207 l.before = func() {
208 // Teardown lands while this sweep is listing.
209 if err := s.Delete(ctx, Key("bab", "titogo")); err != nil {
210 t.Fatal(err)
211 }
212 }
213
214 side, err := RunSweep(ctx, s, l, sc, "bab", "titogo", time.Minute)
215 if err != nil {
216 t.Fatalf("standing down is not an error: %v", err)
217 }
218 _ = side
219 if _, _, err := s.Get(ctx, ScanKey("bab", "titogo")); !errors.Is(err, os.ErrNotExist) {
220 t.Fatalf("the sweep wrote an orphan sidecar under a purged app: %v", err)
221 }
222}
223
224// A client-encrypted store is marked not_applicable and does no work. A
225// scanner cannot read ciphertext, and reporting every object clean would be a
226// true-looking sentence about bytes nobody examined.
227func TestSweepSkipsEncryptedStore(t *testing.T) {
228 ctx := context.Background()
229 s, _, sc, _ := sweepFixture(t, 3)
230
231 rec, etag, err := Get(ctx, s, "bab", "titogo")
232 if err != nil {
233 t.Fatal(err)
234 }
235 rec.Scanning.Requested = &ScanRequested{At: time.Now().UTC(), By: "m@example.com", Encrypted: true}
236 if _, err := Put(ctx, s, "bab", "titogo", rec, etag); err != nil {
237 t.Fatal(err)
238 }
239
240 side, err := RunSweep(ctx, s, &refuseLister{t: t}, sc, "bab", "titogo", time.Minute)
241 if err != nil {
242 t.Fatal(err)
243 }
244 if side == nil || !side.NotApplicable {
245 t.Fatalf("an encrypted store was not marked not_applicable: %+v", side)
246 }
247 if side.LastComplete != nil {
248 t.Fatal("coverage was claimed for a store nobody could read")
249 }
250}
251
252// A trashed app is not swept: it is destroyed within 30 days and scanning it
253// bills for egress on data on its way out.
254func TestSweepSkipsTrashedApp(t *testing.T) {
255 ctx := context.Background()
256 s, _, sc, _ := sweepFixture(t, 3)
257 if err := release.TrashApp(ctx, s, "bab", "titogo", "op@example.com", time.Now().UTC()); err != nil {
258 t.Fatal(err)
259 }
260
261 side, err := RunSweep(ctx, s, &refuseLister{t: t}, sc, "bab", "titogo", time.Minute)
262 if err != nil {
263 t.Fatal(err)
264 }
265 if side != nil {
266 t.Fatalf("a trashed app was swept: %+v", side)
267 }
268}
269
270// Skipped is not clean, and both are visible separately. An object over the
271// size ceiling was never judged; folding it into Scanned would make the audit
272// sentence false in exactly the direction that matters.
273func TestSweepCountsSkippedSeparately(t *testing.T) {
274 s, p, sc, _ := sweepFixture(t, 3)
275 sc.ScanAs(objectName(1), VerdictSkipped)
276
277 side, err := RunSweep(context.Background(), s, &memLister{p: p}, sc, "bab", "titogo", time.Minute)
278 if err != nil {
279 t.Fatal(err)
280 }
281 got := side.LastComplete
282 if got.Objects != 3 || got.Scanned != 2 || got.Skipped != 1 {
283 t.Fatalf("objects=%d scanned=%d skipped=%d, want 3/2/1", got.Objects, got.Scanned, got.Skipped)
284 }
285 if got.Unscanned != 1 {
286 t.Fatalf("unscanned=%d, want 1 — a skipped object is not covered", got.Unscanned)
287 }
288}
289
290// An infected object WAS scanned. We know exactly what it is; it is coverage,
291// not a gap. This is the one place the two counts could plausibly have been
292// wired the other way round.
293func TestSweepCountsInfectedAsScanned(t *testing.T) {
294 s, p, sc, _ := sweepFixture(t, 2)
295 sc.ScanAs(objectName(0), VerdictInfected)
296
297 side, err := RunSweep(context.Background(), s, &memLister{p: p}, sc, "bab", "titogo", time.Minute)
298 if err != nil {
299 t.Fatal(err)
300 }
301 got := side.LastComplete
302 if got.Infected != 1 || got.Scanned != 2 || got.Unscanned != 0 {
303 t.Fatalf("infected=%d scanned=%d unscanned=%d, want 1/2/0",
304 got.Infected, got.Scanned, got.Unscanned)
305 }
306}
307
308// A record naming another engine refuses rather than guessing. Verdicts in the
309// bucket were written under conventions this engine does not read, and
310// recounting them as unscanned would destroy the coverage history.
311func TestSweepRefusesOnEngineDrift(t *testing.T) {
312 ctx := context.Background()
313 s, p, sc, _ := sweepFixture(t, 1)
314
315 rec, etag, err := Get(ctx, s, "bab", "titogo")
316 if err != nil {
317 t.Fatal(err)
318 }
319 rec.Scanning.Enabled.Engine = "guardduty"
320 if _, err := Put(ctx, s, "bab", "titogo", rec, etag); err != nil {
321 t.Fatal(err)
322 }
323
324 _, err = RunSweep(ctx, s, &memLister{p: p}, sc, "bab", "titogo", time.Minute)
325 if err == nil {
326 t.Fatal("a sweep ran against a record naming a different engine")
327 }
328 if !strings.Contains(err.Error(), "guardduty") {
329 t.Fatalf("the refusal does not name the engine on the record: %v", err)
330 }
331}
332
333// An engine that goes away mid-page fails the pass and writes NOTHING. Every
334// verdict already reached is a tag in the bucket, so the re-run reads them
335// back without re-scanning a byte — but a half-written rollup would be a
336// number nobody could interpret.
337func TestSweepEngineFailureWritesNoRollup(t *testing.T) {
338 ctx := context.Background()
339 s, p, sc, _ := sweepFixture(t, 3)
340 sc.Fail("Scan", errors.New("dial unix /run/clamav/clamd.ctl: connection refused"))
341
342 if _, err := RunSweep(ctx, s, &memLister{p: p}, sc, "bab", "titogo", time.Minute); err == nil {
343 t.Fatal("an unreachable engine reported a successful sweep")
344 }
345 if _, _, err := s.Get(ctx, ScanKey("bab", "titogo")); !errors.Is(err, os.ErrNotExist) {
346 t.Fatalf("a failed sweep left a rollup behind: %v", err)
347 }
348}
349
350// A store with no scanning enabled is not swept at all — which is every store
351// on a deployment that has not opted in.
352func TestSweepIgnoresStoreWithoutScanning(t *testing.T) {
353 s := release.NewDirStore(t.TempDir())
354 _, pub := testSealKeypair(t)
355 p := NewMemProvisioner("eu-west-1")
356 rotateFixture(t, s, pub, p)
357
358 side, err := RunSweep(context.Background(), s, &refuseLister{t: t}, NewMemScanner(),
359 "bab", "titogo", time.Minute)
360 if err != nil {
361 t.Fatal(err)
362 }
363 if side != nil {
364 t.Fatalf("a store that never opted in was swept: %+v", side)
365 }
366}
367
368// The CAS retry guard, which the pre-write re-read does NOT cover.
369//
370// Two writes are in play. The FIRST write of a fresh sweep is create-only, and
371// the pre-write re-read is what protects that one. A LATER page's write is
372// conditional on an etag, and losing that CAS means someone else moved the
373// sidecar — another driver (the on-demand sweep verb races the hourly tick by
374// design), or a teardown deleting it. A conditional write against a DELETED
375// key also reads as ErrCASConflict, so a blind retry falls back to create-only
376// and resurrects the object teardown just removed.
377//
378// Here the sidecar is deleted between this sweep's page-two write attempt and
379// its conflict re-read. The sweep must stand down, not resurrect.
380func TestSweepStandsDownWhenSidecarVanishesMidRetry(t *testing.T) {
381 ctx := context.Background()
382 s, p, sc, _ := sweepFixture(t, 5)
383 l := &memLister{p: p, pageSize: 2}
384
385 // One page, so a real sidecar with a Current exists and the next write is
386 // a conditional replace rather than a create.
387 if _, err := RunSweep(ctx, s, l, sc, "bab", "titogo", 0); err != nil {
388 t.Fatal(err)
389 }
390 if _, _, err := s.Get(ctx, ScanKey("bab", "titogo")); err != nil {
391 t.Fatalf("setup: no sidecar to conflict against: %v", err)
392 }
393
394 // Now make the next write lose its CAS *and* find the key gone. Writing a
395 // different body under a new etag invalidates the caller's etag; deleting
396 // it afterwards is the teardown.
397 l.before = func() {
398 l.before = nil // once: only the second page races
399 if _, err := PutSidecar(ctx, s, "bab", "titogo",
400 &ScanSidecar{Current: &Sweep{After: "somewhere-else"}}, ""); err == nil {
401 t.Fatal("expected the create-only write to conflict with the existing sidecar")
402 }
403 // Delete it outright — the teardown case.
404 if err := s.Delete(ctx, ScanKey("bab", "titogo")); err != nil {
405 t.Fatal(err)
406 }
407 }
408
409 side, err := RunSweep(ctx, s, l, sc, "bab", "titogo", 0)
410 if err != nil {
411 t.Fatalf("standing down is not an error: %v", err)
412 }
413 _ = side
414 if _, _, err := s.Get(ctx, ScanKey("bab", "titogo")); !errors.Is(err, os.ErrNotExist) {
415 t.Fatalf("the sweep resurrected a sidecar that had been deleted: %v", err)
416 }
417}
418
419// Fresh and reused verdicts are counted apart. A clean tag is reused for as
420// long as the ETag matches, so a completed rollup can be built almost entirely
421// from judgements made months ago — against a different signature database
422// than the one the sidecar names. Without this split, "engine database" reads
423// as a claim about every object in the rollup, and it is not one.
424func TestSweepCountsFreshAndReusedVerdictsApart(t *testing.T) {
425 ctx := context.Background()
426 s, p, sc, _ := sweepFixture(t, 3)
427
428 first, err := RunSweep(ctx, s, &memLister{p: p}, sc, "bab", "titogo", time.Minute)
429 if err != nil {
430 t.Fatal(err)
431 }
432 if first.LastComplete.Fresh != 3 || first.LastComplete.Reused != 0 {
433 t.Fatalf("first sweep: fresh=%d reused=%d, want 3/0",
434 first.LastComplete.Fresh, first.LastComplete.Reused)
435 }
436
437 // Nothing changed in the bucket, so the second sweep reaches no new
438 // verdicts at all — it reads back the tags the first one wrote.
439 second, err := RunSweep(ctx, s, &memLister{p: p}, sc, "bab", "titogo", time.Minute)
440 if err != nil {
441 t.Fatal(err)
442 }
443 if second.LastComplete.Fresh != 0 || second.LastComplete.Reused != 3 {
444 t.Fatalf("second sweep: fresh=%d reused=%d, want 0/3 — every verdict was carried forward",
445 second.LastComplete.Fresh, second.LastComplete.Reused)
446 }
447 if second.LastComplete.Scanned != 3 {
448 t.Fatalf("scanned = %d, want 3 — reused verdicts are still coverage",
449 second.LastComplete.Scanned)
450 }
451}
452
453// An object that vanishes between the listing and the fetch is dropped, not
454// counted. It is neither covered nor a gap — it is not there.
455func TestSweepDropsObjectsThatDisappearMidPage(t *testing.T) {
456 ctx := context.Background()
457 s, p, _, _ := sweepFixture(t, 3)
458 sc := &goneScanner{gone: map[string]bool{objectName(1): true}}
459
460 side, err := RunSweep(ctx, s, &memLister{p: p}, sc, "bab", "titogo", time.Minute)
461 if err != nil {
462 t.Fatalf("a deleted object failed the sweep: %v", err)
463 }
464 if side.LastComplete.Objects != 2 {
465 t.Fatalf("objects = %d, want 2 — the deleted one must not be counted",
466 side.LastComplete.Objects)
467 }
468}
469
470// goneScanner answers ErrObjectGone for named keys and clean for the rest.
471type goneScanner struct {
472 MemScanner
473 gone map[string]bool
474}
475
476func (g *goneScanner) Verdict(context.Context, string, string, string) (Verdict, error) {
477 return VerdictUnknown, nil
478}
479
480func (g *goneScanner) Scan(_ context.Context, _, key, _ string, _ int64) (Verdict, bool, error) {
481 if g.gone[key] {
482 return VerdictUnknown, false, ErrObjectGone
483 }
484 return VerdictClean, true, nil
485}
486
487func (g *goneScanner) Provenance(context.Context) (string, string, error) {
488 return "clamav", "test/0/never", nil
489}
490
491// The sweep binds verdicts to the object VERSION, not the ETag alone.
492//
493// Same bytes, later last-modified: an overwrite that happened to produce the
494// same content hash. A chosen-prefix MD5 collision is exactly this shape, and
495// it is the case enforcement exists for — an app that accepts uploads and
496// serves them back. Passing only the ETag here would let the payload inherit
497// the clean verdict of the object it replaced, and every existing sweep test
498// would still have passed.
499func TestSweepRescansWhenOnlyLastModifiedChanges(t *testing.T) {
500 ctx := context.Background()
501 s, p, sc, bucket := sweepFixture(t, 1)
502 l := &memLister{p: p}
503
504 first, err := RunSweep(ctx, s, l, sc, "bab", "titogo", time.Minute)
505 if err != nil {
506 t.Fatal(err)
507 }
508 if first.LastComplete.Fresh != 1 {
509 t.Fatalf("setup: fresh = %d, want 1", first.LastComplete.Fresh)
510 }
511
512 // The bytes are untouched — so the ETag is identical — but the object was
513 // rewritten.
514 _ = bucket
515 l.touch(objectName(0), time.Date(2026, 9, 2, 0, 0, 0, 0, time.UTC))
516
517 second, err := RunSweep(ctx, s, l, sc, "bab", "titogo", time.Minute)
518 if err != nil {
519 t.Fatal(err)
520 }
521 if second.LastComplete.Fresh != 1 || second.LastComplete.Reused != 0 {
522 t.Fatalf("fresh=%d reused=%d — an overwritten object inherited its old verdict",
523 second.LastComplete.Fresh, second.LastComplete.Reused)
524 }
525}
526
527// Fresh and Reused split SCANNED, not the object count. An object that was
528// skipped was never judged, so counting its verdict as "judged during this
529// sweep" makes the printed arithmetic — "N of M scanned, of those X judged
530// now" — visibly wrong the moment one object is too large.
531func TestSweepFreshAndReusedSplitScannedNotObjects(t *testing.T) {
532 s, p, sc, _ := sweepFixture(t, 3)
533 sc.ScanAs(objectName(1), VerdictSkipped)
534
535 side, err := RunSweep(context.Background(), s, &memLister{p: p}, sc, "bab", "titogo", time.Minute)
536 if err != nil {
537 t.Fatal(err)
538 }
539 got := side.LastComplete
540 if got.Fresh+got.Reused != got.Scanned {
541 t.Fatalf("fresh(%d)+reused(%d) = %d, but scanned = %d of %d objects",
542 got.Fresh, got.Reused, got.Fresh+got.Reused, got.Scanned, got.Objects)
543 }
544 if got.Fresh != 2 {
545 t.Fatalf("fresh = %d, want 2 — the skipped object was never judged", got.Fresh)
546 }
547}
548
549// An engine enabled with no attestation on record is refused, not swept.
550// Sweeping reads tenant object bytes and the attestation is the consent for
551// that; guessing "probably plaintext" is not this walk's decision to make.
552func TestSweepRefusesWhenNoAttestationIsOnRecord(t *testing.T) {
553 ctx := context.Background()
554 s, p, sc, _ := sweepFixture(t, 1)
555
556 rec, etag, err := Get(ctx, s, "bab", "titogo")
557 if err != nil {
558 t.Fatal(err)
559 }
560 rec.Scanning.Requested = nil
561 if _, err := Put(ctx, s, "bab", "titogo", rec, etag); err != nil {
562 t.Fatal(err)
563 }
564
565 if _, err := RunSweep(ctx, s, &memLister{p: p}, sc, "bab", "titogo", time.Minute); err == nil {
566 t.Fatal("a store with no attestation on record was swept anyway")
567 }
568}
569