amadan
objstore: Status answers about an object, Verdict decides retries
Two questions that look like one, and conflating them is why a failed scan would have read as pending forever. Verdict is SWEEP RETRY POLICY: a stored denied or failed comes back as unknown so the sweep re-scans, because both are transient — a permissions fix, a Glacier restore, an engine having a bad day — and the object is not rewritten, so its version never changes and the repair would never take. That is right, and it is the wrong answer to "what happened to this object". The on-demand drain deletes an item once it has been scanned; with the re-issue rule in the read path, the status door would answer pending for a failed object until the end of time, while the callback said something else. So Status reports what the tag says, denied and failed included. Both read the same tag through one readTag helper, so they cannot drift on what a tag SAYS while differing deliberately on what it MEANS. Mutation-checked: applying the re-issue rule in Status fails both halves of TestStatusDoesNotApplyTheReIssueRule. Status also finds the current version itself and returns the object. A caller holding only a key — the drain, the status door — has no listing page to take a version from, and asking it to reconstruct the token would leak an internal format and hand it the laundering bug as a correctness bug. A denied HeadObject gives a zero ScanObject, which is how a caller knows not to go on and Scan. And Scan now reports whether the verdict was PERSISTED. tag() declines to write in two places — an object already carrying eight foreign tags, and a denied pre-write read — both deliberate, both fine for a sweep that re-lists every 24h, and both fatal for a queue item that gets deleted once scanned. It returned a bare nil on those paths, which made "I wrote it" and "I decided not to" indistinguishable to every caller. The sweep takes _ for the flag: Fresh and Reused count verdicts REACHED, so its numbers do not move.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
7 files changed,
+432
−51
internal/objstore/clamav.go+102 −30internal/objstore/clamav_test.go+218 −12internal/objstore/memscanner.go+69 −3internal/objstore/scanner.go+34 −1internal/objstore/scanner_contract_test.go+1 −1internal/objstore/sweep.go+5 −1internal/objstore/sweep_test.go+3 −3
diff --git a/internal/objstore/clamav.go b/internal/objstore/clamav.go| index 8a5470f48..7fa184d0e 100644 |
| --- a/internal/objstore/clamav.go |
| +++ b/internal/objstore/clamav.go |
| @@ -24,6 +24,7 @@ import ( |
| // other's powers. |
| type scanS3API interface { |
| GetObject(ctx context.Context, in *awss3.GetObjectInput, optFns ...func(*awss3.Options)) (*awss3.GetObjectOutput, error) |
| + HeadObject(ctx context.Context, in *awss3.HeadObjectInput, optFns ...func(*awss3.Options)) (*awss3.HeadObjectOutput, error) |
| GetObjectTagging(ctx context.Context, in *awss3.GetObjectTaggingInput, optFns ...func(*awss3.Options)) (*awss3.GetObjectTaggingOutput, error) |
| PutObjectTagging(ctx context.Context, in *awss3.PutObjectTaggingInput, optFns ...func(*awss3.Options)) (*awss3.PutObjectTaggingOutput, error) |
| } |
| @@ -119,21 +120,37 @@ func (c *clamavScanner) Enforcement() DenyTag { |
| // choose. A tag whose recorded version does not match reads as unknown, and |
| // the sweep scans it again. |
| func (c *clamavScanner) Verdict(ctx context.Context, bucket, key, version string) (Verdict, error) { |
| + verdict, taggedVersion, err := c.readTag(ctx, bucket, key) |
| + if err != nil { |
| + return VerdictUnknown, err |
| + } |
| + if verdict == VerdictDenied && taggedVersion == "" { |
| + // The tag set itself was unreadable. A fact about the object, not a |
| + // transport failure, and it must not fail the pass. |
| + return VerdictDenied, nil |
| + } |
| + return c.interpretForSweep(verdict, taggedVersion, version), nil |
| +} |
| + |
| +// readTag is the one place tags are read, shared by Verdict and Status so the |
| +// two cannot drift on what a tag SAYS while differing, deliberately, on what |
| +// it MEANS. |
| +// |
| +// A denied read comes back as (VerdictDenied, "", nil): unreadable tags are a |
| +// fact about the object. |
| +func (c *clamavScanner) readTag(ctx context.Context, bucket, key string) (Verdict, string, error) { |
| out, err := c.s3.GetObjectTagging(ctx, &awss3.GetObjectTaggingInput{ |
| Bucket: &bucket, |
| Key: &key, |
| }) |
| if err != nil { |
| if isGone(err) { |
| - return VerdictUnknown, ErrObjectGone |
| + return VerdictUnknown, "", ErrObjectGone |
| } |
| if isDenied(err) { |
| - // We cannot read this object's tags and never will without a |
| - // permissions change. That is a fact about the object, not a |
| - // transport failure, and it must not fail the pass. |
| - return VerdictDenied, nil |
| + return VerdictDenied, "", nil |
| } |
| - return VerdictUnknown, fmt.Errorf("objstore: read tags %s/%s: %w", bucket, key, err) |
| + return VerdictUnknown, "", fmt.Errorf("objstore: read tags %s/%s: %w", bucket, key, err) |
| } |
| var verdict, taggedVersion string |
| @@ -145,8 +162,15 @@ func (c *clamavScanner) Verdict(ctx context.Context, bucket, key, version string |
| taggedVersion = aws2s(t.Value) |
| } |
| } |
| + return Verdict(verdict), taggedVersion, nil |
| +} |
| + |
| +// interpretForSweep applies the SWEEP's reading of a tag: unknown when it does |
| +// not match the current version, and unknown for denied and failed so they are |
| +// re-issued rather than inherited. |
| +func (c *clamavScanner) interpretForSweep(verdict Verdict, taggedVersion, version string) Verdict { |
| if verdict == "" || taggedVersion == "" { |
| - return VerdictUnknown, nil |
| + return VerdictUnknown |
| } |
| // denied and failed are RE-ISSUED every sweep, never carried forward. |
| // |
| @@ -157,29 +181,71 @@ func (c *clamavScanner) Verdict(ctx context.Context, bucket, key, version string |
| // version never changes, so it would never be retried and the repair would |
| // never take. An SSE-KMS denial is the case that bites, because the |
| // tagging call needs no KMS access and therefore succeeds. |
| - switch Verdict(verdict) { |
| + switch verdict { |
| case VerdictDenied, VerdictFailed: |
| - return VerdictUnknown, nil |
| + return VerdictUnknown |
| } |
| if taggedVersion != version { |
| // The bytes changed under the verdict. Not a failure — just unknown |
| // again, which is exactly what makes the next sweep re-scan it. |
| - return VerdictUnknown, nil |
| + return VerdictUnknown |
| } |
| - return Verdict(verdict), nil |
| + return verdict |
| +} |
| + |
| +// Status answers about the CURRENT version of one object, finding that version |
| +// itself, and reports what the tag SAYS rather than what the sweep should do |
| +// about it. See the Scanner interface for why those are different questions. |
| +func (c *clamavScanner) Status(ctx context.Context, bucket, key string) (Verdict, ScanObject, error) { |
| + head, err := c.s3.HeadObject(ctx, &awss3.HeadObjectInput{Bucket: &bucket, Key: &key}) |
| + if err != nil { |
| + if isGone(err) { |
| + return VerdictUnknown, ScanObject{}, ErrObjectGone |
| + } |
| + if isDenied(err) { |
| + // No version and no size, so a caller must not go on to Scan. |
| + return VerdictDenied, ScanObject{}, nil |
| + } |
| + return VerdictUnknown, ScanObject{}, fmt.Errorf("objstore: head %s/%s: %w", bucket, key, err) |
| + } |
| + obj := ScanObject{Key: key, ETag: normalizeETag(aws2s(head.ETag))} |
| + if head.LastModified != nil { |
| + obj.Modified = *head.LastModified |
| + } |
| + if head.ContentLength != nil { |
| + obj.Size = *head.ContentLength |
| + } |
| + |
| + verdict, taggedVersion, err := c.readTag(ctx, bucket, key) |
| + if err != nil { |
| + return VerdictUnknown, obj, err |
| + } |
| + if verdict == VerdictDenied && taggedVersion == "" { |
| + return VerdictDenied, obj, nil |
| + } |
| + if verdict == "" || taggedVersion == "" || taggedVersion != obj.Version() { |
| + // No tag, or a tag about bytes that have since changed. Either way |
| + // nothing is known about what is there now. |
| + return VerdictUnknown, obj, nil |
| + } |
| + // Reported as written: denied and failed included. The re-issue rule is |
| + // the sweep's, and applying it here would make a failed scan read as |
| + // pending forever. |
| + return verdict, obj, nil |
| } |
| // Scan streams one object past the engine and records the result. |
| -func (c *clamavScanner) Scan(ctx context.Context, bucket, key, version string, size int64) (Verdict, error) { |
| +func (c *clamavScanner) Scan(ctx context.Context, bucket, key, version string, size int64) (Verdict, bool, error) { |
| // The ceiling is checked BEFORE the fetch. The size is already on the |
| // listing page that got us here, so paying egress to pull bytes clamd has |
| // told us it will refuse is pure waste — and on a bucket full of large |
| // media that waste is the entire bill. |
| if c.maxBytes > 0 && size > c.maxBytes { |
| - if err := c.tag(ctx, bucket, key, version, VerdictSkipped); err != nil { |
| - return VerdictUnknown, err |
| + recorded, err := c.tag(ctx, bucket, key, version, VerdictSkipped) |
| + if err != nil { |
| + return VerdictUnknown, false, err |
| } |
| - return VerdictSkipped, nil |
| + return VerdictSkipped, recorded, nil |
| } |
| obj, err := c.s3.GetObject(ctx, &awss3.GetObjectInput{Bucket: &bucket, Key: &key}) |
| @@ -200,16 +266,18 @@ func (c *clamavScanner) Scan(ctx context.Context, bucket, key, version string, s |
| // and the verdict still stands for this sweep's count. The tag is |
| // only an audit trace here — Verdict re-issues denied rather than |
| // reading it back, so nothing depends on this landing. |
| - _ = c.tag(ctx, bucket, key, version, VerdictDenied) |
| - return VerdictDenied, nil |
| + _, _ = c.tag(ctx, bucket, key, version, VerdictDenied) |
| + // recorded is false whether or not the tag landed: nothing may |
| + // depend on a trace we could not verify. |
| + return VerdictDenied, false, nil |
| } |
| // Deleted between the listing and the fetch. Routine on a bucket |
| // anyone is using, and not a failure of anything — the sweep drops it |
| // rather than counting an object that no longer exists. |
| if isGone(err) { |
| - return VerdictUnknown, ErrObjectGone |
| + return VerdictUnknown, false, ErrObjectGone |
| } |
| - return VerdictUnknown, fmt.Errorf("objstore: get %s/%s: %w", bucket, key, err) |
| + return VerdictUnknown, false, fmt.Errorf("objstore: get %s/%s: %w", bucket, key, err) |
| } |
| defer obj.Body.Close() |
| @@ -225,14 +293,15 @@ func (c *clamavScanner) Scan(ctx context.Context, bucket, key, version string, s |
| // with nothing to distinguish them from objects that genuinely broke |
| // the scanner. Fail the pass instead; the object stays unknown and |
| // the next sweep picks it up. |
| - return VerdictUnknown, fmt.Errorf("objstore: scan %s/%s: %w", bucket, key, err) |
| + return VerdictUnknown, false, fmt.Errorf("objstore: scan %s/%s: %w", bucket, key, err) |
| } |
| v := verdictFor(res.Status) |
| - if err := c.tag(ctx, bucket, key, version, v); err != nil { |
| - return VerdictUnknown, err |
| + recorded, err := c.tag(ctx, bucket, key, version, v) |
| + if err != nil { |
| + return VerdictUnknown, false, err |
| } |
| - return v, nil |
| + return v, recorded, nil |
| } |
| // verdictFor maps an engine status onto the recorded verdict. |
| @@ -266,7 +335,7 @@ func verdictFor(s clamd.Status) Verdict { |
| // may carry tags written long before CARLOS scoped any credentials, by tooling |
| // that is still using them for lifecycle rules or cost allocation. Replacing |
| // the set blindly would delete them, silently, on first scan. |
| -func (c *clamavScanner) tag(ctx context.Context, bucket, key, version string, v Verdict) error { |
| +func (c *clamavScanner) tag(ctx context.Context, bucket, key, version string, v Verdict) (recorded bool, err error) { |
| set := []s3types.Tag{ |
| {Key: s2aws(TagVerdict), Value: s2aws(string(v))}, |
| {Key: s2aws(TagVersion), Value: s2aws(version)}, |
| @@ -279,7 +348,7 @@ func (c *clamavScanner) tag(ctx context.Context, bucket, key, version string, v |
| Bucket: &bucket, |
| Key: &key, |
| }) |
| - if err != nil { |
| + if err != nil { //nolint:nestif // the three outcomes are the point |
| // A tag set we cannot read must NOT be written over. PutObjectTagging |
| // replaces the whole set, so proceeding here would destroy exactly the |
| // foreign tags this merge exists to protect — silently, and on the one |
| @@ -287,9 +356,12 @@ func (c *clamavScanner) tag(ctx context.Context, bucket, key, version string, v |
| // same object's tags moments ago through Verdict, so a failure now is |
| // transient: fail the page and let the next tick retry. |
| if isDenied(err) || isGone(err) { |
| - return nil // nothing we could ever write, and not a pass failure |
| + // Nothing we could ever write, and not a pass failure. NOT |
| + // recorded, which is what stops a caller believing the object |
| + // carries a verdict it does not. |
| + return false, nil |
| } |
| - return fmt.Errorf("objstore: read tags before writing %s/%s: %w", bucket, key, err) |
| + return false, fmt.Errorf("objstore: read tags before writing %s/%s: %w", bucket, key, err) |
| } |
| for _, t := range existing.TagSet { |
| if !isScanTag(aws2s(t.Key)) { |
| @@ -304,7 +376,7 @@ func (c *clamavScanner) tag(ctx context.Context, bucket, key, version string, v |
| // and move on; the object is simply re-scanned next time rather than |
| // blocking every object behind it. |
| if len(set) > maxObjectTags { |
| - return nil |
| + return false, nil |
| } |
| if _, err := c.s3.PutObjectTagging(ctx, &awss3.PutObjectTaggingInput{ |
| @@ -312,9 +384,9 @@ func (c *clamavScanner) tag(ctx context.Context, bucket, key, version string, v |
| Key: &key, |
| Tagging: &s3types.Tagging{TagSet: set}, |
| }); err != nil { |
| - return fmt.Errorf("objstore: tag %s/%s: %w", bucket, key, err) |
| + return false, fmt.Errorf("objstore: tag %s/%s: %w", bucket, key, err) |
| } |
| - return nil |
| + return true, nil |
| } |
| // maxObjectTags is S3's per-object tag limit. |
diff --git a/internal/objstore/clamav_test.go b/internal/objstore/clamav_test.go| index b0ee1490a..ac9c123df 100644 |
| --- a/internal/objstore/clamav_test.go |
| +++ b/internal/objstore/clamav_test.go |
| @@ -25,6 +25,7 @@ type stubScanS3 struct { |
| t *testing.T |
| getObject func(*awss3.GetObjectInput) (*awss3.GetObjectOutput, error) |
| + headObject func(*awss3.HeadObjectInput) (*awss3.HeadObjectOutput, error) |
| getTagging func(*awss3.GetObjectTaggingInput) (*awss3.GetObjectTaggingOutput, error) |
| putTagging func(*awss3.PutObjectTaggingInput) (*awss3.PutObjectTaggingOutput, error) |
| } |
| @@ -36,6 +37,21 @@ func (s *stubScanS3) GetObject(_ context.Context, in *awss3.GetObjectInput, _ .. |
| return s.getObject(in) |
| } |
| +func (s *stubScanS3) HeadObject(_ context.Context, in *awss3.HeadObjectInput, _ ...func(*awss3.Options)) (*awss3.HeadObjectOutput, error) { |
| + if s.headObject == nil { |
| + s.t.Fatal("unexpected HeadObject call") |
| + } |
| + return s.headObject(in) |
| +} |
| + |
| +// headOK makes HeadObject describe an object of the given shape, which is what |
| +// Status needs in order to know the CURRENT version. |
| +func (s *stubScanS3) headOK(etag string, modified time.Time, size int64) { |
| + s.headObject = func(*awss3.HeadObjectInput) (*awss3.HeadObjectOutput, error) { |
| + return &awss3.HeadObjectOutput{ETag: &etag, LastModified: &modified, ContentLength: &size}, nil |
| + } |
| +} |
| + |
| func (s *stubScanS3) GetObjectTagging(_ context.Context, in *awss3.GetObjectTaggingInput, _ ...func(*awss3.Options)) (*awss3.GetObjectTaggingOutput, error) { |
| if s.getTagging == nil { |
| s.t.Fatal("unexpected GetObjectTagging call") |
| @@ -115,7 +131,7 @@ func TestClamAVSkipsOversizeWithoutFetching(t *testing.T) { |
| eng := &fakeEngine{} |
| c := &clamavScanner{s3: s3, engine: eng, maxBytes: 10} |
| - v, err := c.Scan(context.Background(), "b", "big.iso", "abc", 11) |
| + v, _, err := c.Scan(context.Background(), "b", "big.iso", "abc", 11) |
| if err != nil { |
| t.Fatal(err) |
| } |
| @@ -146,7 +162,7 @@ func TestClamAVTagsVerdictWithETag(t *testing.T) { |
| eng := &fakeEngine{res: clamd.Result{Status: clamd.StatusClean}} |
| c := &clamavScanner{s3: s3, engine: eng, maxBytes: 1 << 20} |
| - v, err := c.Scan(context.Background(), "b", "k", "d41d8cd9@2026-09-01T10:00:00Z", 5) |
| + v, _, err := c.Scan(context.Background(), "b", "k", "d41d8cd9@2026-09-01T10:00:00Z", 5) |
| if err != nil { |
| t.Fatal(err) |
| } |
| @@ -175,7 +191,7 @@ func TestClamAVDialFailureWritesNoVerdict(t *testing.T) { |
| eng := &fakeEngine{err: errors.New("dial unix /run/clamav/clamd.ctl: connection refused")} |
| c := &clamavScanner{s3: s3, engine: eng, maxBytes: 1 << 20} |
| - v, err := c.Scan(context.Background(), "b", "k", "e", 5) |
| + v, _, err := c.Scan(context.Background(), "b", "k", "e", 5) |
| if err == nil { |
| t.Fatal("an unreachable engine reported success") |
| } |
| @@ -199,7 +215,7 @@ func TestClamAVSizeExceededMapsToSkipped(t *testing.T) { |
| eng := &fakeEngine{res: clamd.Result{Status: clamd.StatusSizeExceeded}} |
| c := &clamavScanner{s3: s3, engine: eng, maxBytes: 0} // no local ceiling: the engine's own |
| - v, err := c.Scan(context.Background(), "b", "k", "e", 999) |
| + v, _, err := c.Scan(context.Background(), "b", "k", "e", 999) |
| if err != nil { |
| t.Fatal(err) |
| } |
| @@ -227,7 +243,7 @@ func TestClamAVEngineErrorIsRecordedAsFailed(t *testing.T) { |
| eng := &fakeEngine{res: clamd.Result{Status: clamd.StatusError, Raw: "stream: broken ERROR"}} |
| c := &clamavScanner{s3: s3, engine: eng, maxBytes: 1 << 20} |
| - v, err := c.Scan(context.Background(), "b", "k", "e", 5) |
| + v, _, err := c.Scan(context.Background(), "b", "k", "e", 5) |
| if err != nil { |
| t.Fatal(err) |
| } |
| @@ -252,7 +268,7 @@ func TestClamAVInfectedVerdictMatchesEnforcementTag(t *testing.T) { |
| eng := &fakeEngine{res: clamd.Result{Status: clamd.StatusInfected, Signature: "Eicar-Signature"}} |
| c := &clamavScanner{s3: s3, engine: eng, maxBytes: 1 << 20} |
| - if _, err := c.Scan(context.Background(), "b", "k", "e", 6); err != nil { |
| + if _, _, err := c.Scan(context.Background(), "b", "k", "e", 6); err != nil { |
| t.Fatal(err) |
| } |
| deny := c.Enforcement() |
| @@ -404,7 +420,7 @@ func TestClamAVUnreadableObjectIsDeniedNotAPassFailure(t *testing.T) { |
| } |
| c := &clamavScanner{s3: s3, engine: &fakeEngine{}, maxBytes: 1 << 20} |
| - v, err := c.Scan(context.Background(), "b", "k", "e", 5) |
| + v, _, err := c.Scan(context.Background(), "b", "k", "e", 5) |
| if err != nil { |
| t.Fatalf("an unreadable object failed the pass: %v", err) |
| } |
| @@ -431,7 +447,7 @@ func TestClamAVDeletedObjectIsGone(t *testing.T) { |
| } |
| c := &clamavScanner{s3: s3, engine: &fakeEngine{}, maxBytes: 1 << 20} |
| - if _, err := c.Scan(context.Background(), "b", "k", "e", 5); !errors.Is(err, ErrObjectGone) { |
| + if _, _, err := c.Scan(context.Background(), "b", "k", "e", 5); !errors.Is(err, ErrObjectGone) { |
| t.Fatalf("err = %v, want ErrObjectGone", err) |
| } |
| } |
| @@ -463,7 +479,7 @@ func TestClamAVTransportErrorStillFailsThePass(t *testing.T) { |
| } |
| c := &clamavScanner{s3: s3, engine: &fakeEngine{}, maxBytes: 1 << 20} |
| - v, err := c.Scan(context.Background(), "b", "k", "e", 5) |
| + v, _, err := c.Scan(context.Background(), "b", "k", "e", 5) |
| if err == nil { |
| t.Fatal("a transport failure was recorded as a verdict") |
| } |
| @@ -490,7 +506,7 @@ func TestClamAVPreservesForeignTags(t *testing.T) { |
| } |
| c := &clamavScanner{s3: s3, engine: &fakeEngine{res: clamd.Result{Status: clamd.StatusClean}}, maxBytes: 1 << 20} |
| - if _, err := c.Scan(context.Background(), "b", "k", "fresh", 5); err != nil { |
| + if _, _, err := c.Scan(context.Background(), "b", "k", "fresh", 5); err != nil { |
| t.Fatal(err) |
| } |
| if wrote["cost-centre"] != "research" { |
| @@ -562,7 +578,7 @@ func TestClamAVDoesNotBlindWriteWhenTagReadFails(t *testing.T) { |
| } |
| c := &clamavScanner{s3: s3, engine: &fakeEngine{res: clamd.Result{Status: clamd.StatusClean}}, maxBytes: 1 << 20} |
| - if _, err := c.Scan(context.Background(), "b", "k", "v1", 5); err == nil { |
| + if _, _, err := c.Scan(context.Background(), "b", "k", "v1", 5); err == nil { |
| t.Fatal("an unreadable tag set was written over anyway") |
| } |
| } |
| @@ -583,7 +599,7 @@ func TestClamAVSkipsTagWriteRatherThanStallAtTheTagLimit(t *testing.T) { |
| s3.allowTagRead(foreign...) |
| c := &clamavScanner{s3: s3, engine: &fakeEngine{res: clamd.Result{Status: clamd.StatusClean}}, maxBytes: 1 << 20} |
| - v, err := c.Scan(context.Background(), "b", "k", "v1", 5) |
| + v, _, err := c.Scan(context.Background(), "b", "k", "v1", 5) |
| if err != nil { |
| t.Fatalf("a heavily tagged object stalled the sweep: %v", err) |
| } |
| @@ -591,3 +607,193 @@ func TestClamAVSkipsTagWriteRatherThanStallAtTheTagLimit(t *testing.T) { |
| t.Fatalf("verdict = %q, want clean — it was scanned, just not recorded", v) |
| } |
| } |
| + |
| +// Status finds the CURRENT version itself and hands back the object, so a |
| +// caller holding only a key can go on to Scan without a HeadObject of its own. |
| +// |
| +// That second return is the whole point: Scan needs a version and a size, the |
| +// on-demand drain has no listing page, and asking it to reconstruct the token |
| +// would leak an internal format and hand it the laundering bug as a |
| +// correctness bug. |
| +func TestStatusReturnsObjectSoCallerCanScan(t *testing.T) { |
| + mod := time.Date(2026, 9, 2, 8, 0, 0, 0, time.UTC) |
| + s3 := &stubScanS3{t: t} |
| + s3.headOK(`"d41d8cd9"`, mod, 4096) |
| + s3.getTagging = func(*awss3.GetObjectTaggingInput) (*awss3.GetObjectTaggingOutput, error) { |
| + return taggingOutput(), nil // no tag yet |
| + } |
| + c := &clamavScanner{s3: s3, engine: &fakeEngine{}} |
| + |
| + v, obj, err := c.Status(context.Background(), "b", "k") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if v != VerdictUnknown { |
| + t.Fatalf("verdict = %q, want unknown for an untagged object", v) |
| + } |
| + if obj.Size != 4096 || obj.ETag != "d41d8cd9" || !obj.Modified.Equal(mod) { |
| + t.Fatalf("object = %+v — a caller cannot Scan without these", obj) |
| + } |
| + if obj.Version() == "" { |
| + t.Fatal("no version token to scan against") |
| + } |
| +} |
| + |
| +// Status reports what the tag SAYS. Verdict decides what the SWEEP does about |
| +// it. Conflating them makes a failed scan read as pending forever: the drain |
| +// deletes the item as scanned, and nothing would ever change the answer. |
| +func TestStatusDoesNotApplyTheReIssueRule(t *testing.T) { |
| + mod := time.Date(2026, 9, 2, 8, 0, 0, 0, time.UTC) |
| + version := ScanObject{ETag: "abc", Modified: mod}.Version() |
| + |
| + for _, stuck := range []Verdict{VerdictDenied, VerdictFailed} { |
| + t.Run(string(stuck), func(t *testing.T) { |
| + s3 := &stubScanS3{t: t} |
| + s3.headOK("abc", mod, 10) |
| + s3.getTagging = func(*awss3.GetObjectTaggingInput) (*awss3.GetObjectTaggingOutput, error) { |
| + return taggingOutput(TagVerdict, string(stuck), TagVersion, version), nil |
| + } |
| + c := &clamavScanner{s3: s3, engine: &fakeEngine{}} |
| + |
| + // Status: as written. |
| + got, _, err := c.Status(context.Background(), "b", "k") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if got != stuck { |
| + t.Fatalf("Status = %q, want %q reported as itself", got, stuck) |
| + } |
| + |
| + // Verdict: re-issued, so the sweep retries it. |
| + sweepView, err := c.Verdict(context.Background(), "b", "k", version) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if sweepView != VerdictUnknown { |
| + t.Fatalf("Verdict = %q, want unknown so the sweep re-scans", sweepView) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +// A tag about bytes that have since changed is not an answer about the bytes |
| +// that are there now. |
| +func TestStatusIgnoresStaleTag(t *testing.T) { |
| + mod := time.Date(2026, 9, 2, 8, 0, 0, 0, time.UTC) |
| + s3 := &stubScanS3{t: t} |
| + s3.headOK("new-etag", mod, 10) |
| + s3.getTagging = func(*awss3.GetObjectTaggingInput) (*awss3.GetObjectTaggingOutput, error) { |
| + return taggingOutput(TagVerdict, "clean", |
| + TagVersion, ScanObject{ETag: "old-etag", Modified: mod}.Version()), nil |
| + } |
| + c := &clamavScanner{s3: s3, engine: &fakeEngine{}} |
| + |
| + v, _, err := c.Status(context.Background(), "b", "k") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if v != VerdictUnknown { |
| + t.Fatalf("a rewritten object reported %q from its predecessor's tag", v) |
| + } |
| +} |
| + |
| +// A denied HeadObject leaves no version and no size, so the caller must not go |
| +// on to Scan — the zero ScanObject is how it knows. |
| +func TestStatusOnDeniedHead(t *testing.T) { |
| + s3 := &stubScanS3{t: t} |
| + s3.headObject = func(*awss3.HeadObjectInput) (*awss3.HeadObjectOutput, error) { |
| + return nil, apiErr("AccessDenied") |
| + } |
| + c := &clamavScanner{s3: s3, engine: &fakeEngine{}} |
| + |
| + v, obj, err := c.Status(context.Background(), "b", "k") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if v != VerdictDenied { |
| + t.Fatalf("verdict = %q, want denied", v) |
| + } |
| + if obj != (ScanObject{}) { |
| + t.Fatalf("object = %+v, want the zero value so nobody tries to scan it", obj) |
| + } |
| +} |
| + |
| +func TestStatusOnMissingObject(t *testing.T) { |
| + s3 := &stubScanS3{t: t} |
| + s3.headObject = func(*awss3.HeadObjectInput) (*awss3.HeadObjectOutput, error) { |
| + return nil, apiErr("NotFound") |
| + } |
| + c := &clamavScanner{s3: s3, engine: &fakeEngine{}} |
| + |
| + if _, _, err := c.Status(context.Background(), "b", "k"); !errors.Is(err, ErrObjectGone) { |
| + t.Fatalf("err = %v, want ErrObjectGone", err) |
| + } |
| +} |
| + |
| +// A verdict reached but not persisted says so. Without the flag the on-demand |
| +// drain would report clean, delete the item, and the status door would answer |
| +// pending forever — the app's source of truth never arriving while a callback |
| +// said otherwise. |
| +func TestScanReportsUnrecordedAtTheTagLimit(t *testing.T) { |
| + var foreign []string |
| + for i := 0; i < 8; i++ { |
| + foreign = append(foreign, "own-"+string(rune('a'+i)), "v") |
| + } |
| + s3 := &stubScanS3{t: t} // putTagging unset: an 11-tag write fails the test |
| + s3.getObject = func(*awss3.GetObjectInput) (*awss3.GetObjectOutput, error) { |
| + return objectBody("hello"), nil |
| + } |
| + s3.allowTagRead(foreign...) |
| + c := &clamavScanner{s3: s3, engine: &fakeEngine{res: clamd.Result{Status: clamd.StatusClean}}, maxBytes: 1 << 20} |
| + |
| + v, recorded, err := c.Scan(context.Background(), "b", "k", "v1", 5) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if v != VerdictClean { |
| + t.Fatalf("verdict = %q, want clean — it WAS scanned", v) |
| + } |
| + if recorded { |
| + t.Fatal("recorded = true, but nothing was written to the object") |
| + } |
| +} |
| + |
| +// The ordinary path records, so `recorded` is not simply always false. |
| +func TestScanReportsRecordedOnTheOrdinaryPath(t *testing.T) { |
| + s3 := &stubScanS3{t: t} |
| + s3.getObject = func(*awss3.GetObjectInput) (*awss3.GetObjectOutput, error) { |
| + return objectBody("hello"), nil |
| + } |
| + s3.allowTagRead() |
| + s3.putTagging = func(*awss3.PutObjectTaggingInput) (*awss3.PutObjectTaggingOutput, error) { |
| + return &awss3.PutObjectTaggingOutput{}, nil |
| + } |
| + c := &clamavScanner{s3: s3, engine: &fakeEngine{res: clamd.Result{Status: clamd.StatusClean}}, maxBytes: 1 << 20} |
| + |
| + v, recorded, err := c.Scan(context.Background(), "b", "k", "v1", 5) |
| + if err != nil || v != VerdictClean || !recorded { |
| + t.Fatalf("= %q, %v, %v", v, recorded, err) |
| + } |
| +} |
| + |
| +// A denied fetch is never "recorded", whether or not the best-effort tag |
| +// landed: nothing may depend on a trace we could not verify. |
| +func TestScanDeniedIsNeverRecorded(t *testing.T) { |
| + s3 := &stubScanS3{t: t} |
| + s3.getObject = func(*awss3.GetObjectInput) (*awss3.GetObjectOutput, error) { |
| + return nil, apiErr("AccessDenied") |
| + } |
| + s3.allowTagRead() |
| + s3.putTagging = func(*awss3.PutObjectTaggingInput) (*awss3.PutObjectTaggingOutput, error) { |
| + return &awss3.PutObjectTaggingOutput{}, nil |
| + } |
| + c := &clamavScanner{s3: s3, engine: &fakeEngine{}, maxBytes: 1 << 20} |
| + |
| + v, recorded, err := c.Scan(context.Background(), "b", "k", "v1", 5) |
| + if err != nil || v != VerdictDenied { |
| + t.Fatalf("= %q, %v", v, err) |
| + } |
| + if recorded { |
| + t.Fatal("a denied object reported its verdict as recorded") |
| + } |
| +} |
diff --git a/internal/objstore/memscanner.go b/internal/objstore/memscanner.go| index fb0701435..c70221f5c 100644 |
| --- a/internal/objstore/memscanner.go |
| +++ b/internal/objstore/memscanner.go |
| @@ -3,6 +3,7 @@ package objstore |
| import ( |
| "context" |
| "sync" |
| + "time" |
| ) |
| // MemScanner is the in-memory Scanner every walk test runs against, mirroring |
| @@ -14,6 +15,13 @@ type MemScanner struct { |
| verdicts map[string]scanRecord |
| failures map[string]error |
| scanAs map[string]Verdict |
| + // objects is what a listing would have said about a key. Status needs it: |
| + // a caller holding only a key has no listing page to take a version from. |
| + objects map[string]ScanObject |
| + // unrecordable are keys whose verdict cannot be persisted — the ten-tag |
| + // cap, in production. A hook rather than a stub, because the case is the |
| + // reason Scan reports `recorded` at all. |
| + unrecordable map[string]bool |
| } |
| // scanRecord is one object's verdict AND the object version it was judged |
| @@ -43,6 +51,51 @@ func (s *MemScanner) Fail(method string, err error) { |
| s.failures[method] = err |
| } |
| +// PutObject seeds what a listing would have said about a key, so Status can |
| +// answer. Without it Status cannot tell an absent object from an unscanned |
| +// one, and those are different answers. |
| +func (s *MemScanner) PutObject(bucket, key, etag string, modified time.Time, size int64) { |
| + s.mu.Lock() |
| + defer s.mu.Unlock() |
| + if s.objects == nil { |
| + s.objects = map[string]ScanObject{} |
| + } |
| + s.objects[memScanKey(bucket, key)] = ScanObject{ |
| + Key: key, ETag: etag, Modified: modified, Size: size, |
| + } |
| +} |
| + |
| +// Unrecordable makes Scan reach a verdict it cannot persist, the way the |
| +// ten-tag cap does in production. |
| +func (s *MemScanner) Unrecordable(key string) { |
| + s.mu.Lock() |
| + defer s.mu.Unlock() |
| + if s.unrecordable == nil { |
| + s.unrecordable = map[string]bool{} |
| + } |
| + s.unrecordable[key] = true |
| +} |
| + |
| +// Status answers about the CURRENT version, and reports what the record SAYS |
| +// — denied and failed included. The re-issue rule below is Verdict's, and |
| +// applying it here would make a failed scan read as pending forever. |
| +func (s *MemScanner) Status(ctx context.Context, bucket, key string) (Verdict, ScanObject, error) { |
| + s.mu.Lock() |
| + defer s.mu.Unlock() |
| + if err := s.takeFailure("Status"); err != nil { |
| + return VerdictUnknown, ScanObject{}, err |
| + } |
| + obj, ok := s.objects[memScanKey(bucket, key)] |
| + if !ok { |
| + return VerdictUnknown, ScanObject{}, ErrObjectGone |
| + } |
| + rec, ok := s.verdicts[memScanKey(bucket, key)] |
| + if !ok || rec.version != obj.Version() { |
| + return VerdictUnknown, obj, nil |
| + } |
| + return rec.verdict, obj, nil |
| +} |
| + |
| // Put seeds a verdict so a test can start from a partly-scanned bucket. |
| func (s *MemScanner) Put(bucket, key, version string, v Verdict) { |
| s.mu.Lock() |
| @@ -96,21 +149,34 @@ func (s *MemScanner) Verdict(ctx context.Context, bucket, key, version string) ( |
| if !ok || rec.version != version { |
| return VerdictUnknown, nil |
| } |
| + // The SWEEP's reading: denied and failed are re-issued rather than |
| + // inherited, because both are transient and the object is not rewritten, |
| + // so its version never changes and the repair would never take. Status |
| + // deliberately does not do this — see the Scanner interface. |
| + switch rec.verdict { |
| + case VerdictDenied, VerdictFailed: |
| + return VerdictUnknown, nil |
| + } |
| return rec.verdict, nil |
| } |
| -func (s *MemScanner) Scan(ctx context.Context, bucket, key, version string, size int64) (Verdict, error) { |
| +func (s *MemScanner) Scan(ctx context.Context, bucket, key, version string, size int64) (Verdict, bool, error) { |
| s.mu.Lock() |
| defer s.mu.Unlock() |
| if err := s.takeFailure("Scan"); err != nil { |
| - return VerdictUnknown, err |
| + return VerdictUnknown, false, err |
| } |
| v := VerdictClean |
| if got, ok := s.scanAs[key]; ok { |
| v = got |
| } |
| + if s.unrecordable[key] { |
| + // A verdict reached and not persisted: the engine judged the object, |
| + // but nothing on the object says so. |
| + return v, false, nil |
| + } |
| s.verdicts[memScanKey(bucket, key)] = scanRecord{version: version, verdict: v} |
| - return v, nil |
| + return v, true, nil |
| } |
| // ScanAs makes Scan answer v for key rather than clean, so a test can stage a |
diff --git a/internal/objstore/scanner.go b/internal/objstore/scanner.go| index d3c1bbee3..21c4803a8 100644 |
| --- a/internal/objstore/scanner.go |
| +++ b/internal/objstore/scanner.go |
| @@ -28,6 +28,28 @@ type Scanner interface { |
| // no verdict. Absent is (VerdictUnknown, nil) too — the same idiom Get |
| // uses for an absent record, because "not scanned" is a state. |
| Verdict(ctx context.Context, bucket, key, version string) (Verdict, error) |
| + // Status answers what is known about the CURRENT version of one object, |
| + // and returns the object as a listing page would have described it so the |
| + // caller can go on to Scan it without a HeadObject of its own. |
| + // |
| + // TWO DIFFERENCES FROM Verdict, and both matter. |
| + // |
| + // It finds the current version itself, because a caller holding only a key |
| + // — the on-demand drain, the status door — has no listing page to take one |
| + // from. Handing them the token to reconstruct would leak an internal |
| + // format and hand them the laundering bug as a correctness bug. |
| + // |
| + // And it does NOT apply Verdict's re-issue rule. A stored `denied` or |
| + // `failed` comes back as itself here. That rule is SWEEP RETRY POLICY — |
| + // re-scan the transient ones — and it is the wrong answer to "what |
| + // happened to this object": left in place, a failed scan would read as |
| + // pending forever, because the drain deletes the item as scanned and |
| + // nothing would ever change the answer. |
| + // |
| + // ErrObjectGone when the object is not there. A denied HeadObject is |
| + // (VerdictDenied, ScanObject{}, nil) — there is no version or size to |
| + // scan with, so a caller must not go on to Scan a zero ScanObject. |
| + Status(ctx context.Context, bucket, key string) (Verdict, ScanObject, error) |
| // Scan scans one object and records its verdict against etag. It returns |
| // the verdict when the engine is synchronous (ClamAV) and VerdictUnknown |
| // when the engine scans asynchronously (GuardDuty). The sweep counts what |
| @@ -39,7 +61,18 @@ type Scanner interface { |
| // egress for bytes the scanner will reject is pure waste, and the |
| // alternative — a HeadObject per object per sweep — is a round trip on |
| // every object forever. |
| - Scan(ctx context.Context, bucket, key, version string, size int64) (Verdict, error) |
| + // recorded reports whether the verdict was actually PERSISTED to the |
| + // object. It is not always: tag() declines to write when the object |
| + // already carries eight foreign tags (S3 caps an object at ten) or when |
| + // the pre-write tag read is denied. Both are deliberate — they stop one |
| + // object stalling a whole sweep — and the sweep tolerates them because it |
| + // re-lists every 24h. |
| + // |
| + // A queue item does not. Without this flag the on-demand drain would |
| + // report clean, delete the item, and the status door would answer pending |
| + // forever, because the tag is not there — the app's declared source of |
| + // truth never arriving while a callback said otherwise. |
| + Scan(ctx context.Context, bucket, key, version string, size int64) (verdict Verdict, recorded bool, err error) |
| // Enforcement names the object tag this engine writes for an infected |
| // object, so the IAM policy writer can deny reads on it (spec §8). |
| // Engine-specific by necessity: a Deny statement must name a literal key |
diff --git a/internal/objstore/scanner_contract_test.go b/internal/objstore/scanner_contract_test.go| index 1359f8bab..02049e5e0 100644 |
| --- a/internal/objstore/scanner_contract_test.go |
| +++ b/internal/objstore/scanner_contract_test.go |
| @@ -30,7 +30,7 @@ func scannerContract(t *testing.T, s Scanner, bucket string) { |
| } |
| for i := 0; i < 2; i++ { |
| - if _, err := s.Scan(ctx, bucket, "a.txt", "etag-1", 10); err != nil { |
| + if _, _, err := s.Scan(ctx, bucket, "a.txt", "etag-1", 10); err != nil { |
| t.Fatalf("Scan call %d: %v", i+1, err) |
| } |
| } |
diff --git a/internal/objstore/sweep.go b/internal/objstore/sweep.go| index e3e47df6f..0af7fa4a2 100644 |
| --- a/internal/objstore/sweep.go |
| +++ b/internal/objstore/sweep.go |
| @@ -251,7 +251,11 @@ func RunSweep(ctx context.Context, s release.Store, lister ObjectLister, sc Scan |
| // neither can stall the page: an error here abandons the page |
| // unwritten, and a page that fails identically every tick |
| // means the bucket's sweep never advances past it. |
| - v, err = sc.Scan(ctx, bucket, o.Key, o.Version(), o.Size) |
| + // The recorded flag is the on-demand drain's business, not |
| + // the sweep's: Fresh and Reused count verdicts REACHED, and |
| + // the sweep re-lists every cycle anyway, so an unpersisted |
| + // verdict simply gets reached again. |
| + v, _, err = sc.Scan(ctx, bucket, o.Key, o.Version(), o.Size) |
| if errors.Is(err, ErrObjectGone) { |
| continue |
| } |
diff --git a/internal/objstore/sweep_test.go b/internal/objstore/sweep_test.go| index 2ea2ade96..627f6c0cb 100644 |
| --- a/internal/objstore/sweep_test.go |
| +++ b/internal/objstore/sweep_test.go |
| @@ -477,11 +477,11 @@ func (g *goneScanner) Verdict(context.Context, string, string, string) (Verdict, |
| return VerdictUnknown, nil |
| } |
| -func (g *goneScanner) Scan(_ context.Context, _, key, _ string, _ int64) (Verdict, error) { |
| +func (g *goneScanner) Scan(_ context.Context, _, key, _ string, _ int64) (Verdict, bool, error) { |
| if g.gone[key] { |
| - return VerdictUnknown, ErrObjectGone |
| + return VerdictUnknown, false, ErrObjectGone |
| } |
| - return VerdictClean, nil |
| + return VerdictClean, true, nil |
| } |
| func (g *goneScanner) Provenance(context.Context) (string, string, error) { |