amadan

carlos / platform Public

objstore: the Scanner seam and its in-memory fake

Mirrors the Provisioner seam: idempotent everywhere, because the sweep
that drives one re-runs after a half-failure at any point.

Two shapes are load-bearing. Verdict takes an etag and answers Unknown
when it disagrees, so a re-uploaded object cannot inherit the verdict of
the bytes it replaced. And Scan takes the object's size, which the sweep
already holds from the listing page, so an engine can refuse an
over-ceiling object without fetching it — the alternative is a HeadObject
per object per sweep, forever.

VerdictSkipped, VerdictDenied and VerdictFailed are separate from clean
on purpose: a store that is mostly skips has not been meaningfully
scanned, and evidence that blurs the two is worse than none.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev 1a68147ffd8ec268a020b8b7272b2807756656a0 parent 8108621
3 files changed, +312 −0
  • internal/objstore/memscanner.go +124 −0
  • internal/objstore/scanner.go +104 −0
  • internal/objstore/scanner_contract_test.go +84 −0
diff --git a/internal/objstore/memscanner.go b/internal/objstore/memscanner.go
new file mode 100644
index 000000000..3c19340f4
--- /dev/null
+++ b/internal/objstore/memscanner.go
@@ -0,0 +1,124 @@
+package objstore
+
+import (
+ "context"
+ "sync"
+)
+
+// MemScanner is the in-memory Scanner every walk test runs against, mirroring
+// MemProvisioner: no network, no clamd, and a failure hook so a test can make
+// any single method fail once and prove the walk survives it.
+type MemScanner struct {
+ mu sync.Mutex
+ enabled map[string]bool
+ verdicts map[string]scanRecord
+ failures map[string]error
+}
+
+// scanRecord is one object's verdict AND the etag it was judged against. The
+// etag is not decoration: Verdict answers Unknown when it disagrees, which is
+// what stops a rewritten object inheriting the old bytes' verdict.
+type scanRecord struct {
+ etag string
+ verdict Verdict
+}
+
+func NewMemScanner() *MemScanner {
+ return &MemScanner{
+ enabled: map[string]bool{},
+ verdicts: map[string]scanRecord{},
+ failures: map[string]error{},
+ }
+}
+
+func memScanKey(bucket, key string) string { return bucket + "\x00" + key }
+
+// Fail arms a one-shot failure for the named method, the same shape
+// MemProvisioner.takeFailure gives the provisioner walks.
+func (s *MemScanner) Fail(method string, err error) {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ s.failures[method] = err
+}
+
+// Put seeds a verdict so a test can start from a partly-scanned bucket.
+func (s *MemScanner) Put(bucket, key, etag string, v Verdict) {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ s.verdicts[memScanKey(bucket, key)] = scanRecord{etag: etag, verdict: v}
+}
+
+// IsEnabled lets Teardown's test prove the scanner was disabled while the
+// record that named it still existed.
+func (s *MemScanner) IsEnabled(bucket string) bool {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ return s.enabled[bucket]
+}
+
+func (s *MemScanner) takeFailure(method string) error {
+ if err, ok := s.failures[method]; ok {
+ delete(s.failures, method)
+ return err
+ }
+ return nil
+}
+
+func (s *MemScanner) EnsureEnabled(ctx context.Context, bucket string) error {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ if err := s.takeFailure("EnsureEnabled"); err != nil {
+ return err
+ }
+ s.enabled[bucket] = true // already enabled is a success
+ return nil
+}
+
+func (s *MemScanner) EnsureDisabled(ctx context.Context, bucket string) error {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ if err := s.takeFailure("EnsureDisabled"); err != nil {
+ return err
+ }
+ delete(s.enabled, bucket) // already gone is a success
+ return nil
+}
+
+func (s *MemScanner) Verdict(ctx context.Context, bucket, key, etag string) (Verdict, error) {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ if err := s.takeFailure("Verdict"); err != nil {
+ return VerdictUnknown, err
+ }
+ rec, ok := s.verdicts[memScanKey(bucket, key)]
+ if !ok || rec.etag != etag {
+ return VerdictUnknown, nil
+ }
+ return rec.verdict, nil
+}
+
+func (s *MemScanner) Scan(ctx context.Context, bucket, key, etag string, size int64) (Verdict, error) {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ if err := s.takeFailure("Scan"); err != nil {
+ return VerdictUnknown, err
+ }
+ s.verdicts[memScanKey(bucket, key)] = scanRecord{etag: etag, verdict: VerdictClean}
+ return VerdictClean, nil
+}
+
+func (s *MemScanner) Enforcement() DenyTag {
+ return DenyTag{Key: TagVerdict, Value: string(VerdictInfected)}
+}
+
+func (s *MemScanner) Capacity(ctx context.Context) (int, int, error) { return -1, -1, nil }
+
+// Provenance answers "clamav" rather than "memscanner" ON PURPOSE. The sweep
+// refuses when the record's engine disagrees with the running one (spec §7.4),
+// and test fixtures write Engine: "clamav" — a fake that named itself would
+// make every one of those fixtures refuse instead of sweep.
+func (s *MemScanner) Provenance(ctx context.Context) (string, string, error) {
+ return "clamav", "MemScanner/0/never", nil
+}
+
+var _ Scanner = (*MemScanner)(nil)
diff --git a/internal/objstore/scanner.go b/internal/objstore/scanner.go
new file mode 100644
index 000000000..4011d7501
--- /dev/null
+++ b/internal/objstore/scanner.go
@@ -0,0 +1,104 @@
+package objstore
+
+import "context"
+
+// Scanner is the scanning-engine seam (spec §3): everything the sweep needs an
+// engine to do, and nothing an app ever calls. Like Provisioner, every method
+// is idempotent — "already enabled" and "already gone" are successes — because
+// the sweep that drives one must be re-runnable after a half-failure.
+//
+// v1 ships the ClamAV implementation and this package's in-memory fake. A
+// GuardDuty scanner arrives here later without touching the sweep, once the
+// question in spec §4.2 has been measured: AWS tags only objects uploaded
+// after tagging was enabled, and every object in an existing store predates
+// its plan, so if an on-demand scan does not tag them then that engine needs a
+// different verdict source and not just a different implementation of this.
+type Scanner interface {
+ // EnsureEnabled turns scanning on for a bucket where the backend scans
+ // natively. Engines that scan by fetching have nothing to configure.
+ EnsureEnabled(ctx context.Context, bucket string) error
+ // EnsureDisabled releases whatever EnsureEnabled took. Teardown calls it
+ // BEFORE deleting the bucket (spec §7.3): the record is the only thing
+ // naming the backend estate, so releasing it after the record is gone
+ // leaks an estate nobody can find again.
+ EnsureDisabled(ctx context.Context, bucket string) error
+ // Verdict answers the recorded verdict for one object AT etag. A verdict
+ // recorded against a DIFFERENT etag is (VerdictUnknown, nil): the object
+ // has been rewritten since it was judged, and bytes nobody looked at have
+ // 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, etag string) (Verdict, 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
+ // it gets back, so a synchronous engine's FIRST pass is a pass full of
+ // real verdicts rather than one full of "unscanned".
+ //
+ // size comes from the listing page the sweep is already holding, so an
+ // engine can refuse an over-ceiling object without fetching it. Paying
+ // 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, etag string, size int64) (Verdict, 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
+ // and value, which is the one place the sweep's engine-blindness cannot
+ // hold.
+ Enforcement() DenyTag
+ // Capacity reports how many buckets this engine can still protect and the
+ // cap; (-1, -1) is uncapped. ClamAV is uncapped. GuardDuty caps at 25 per
+ // account per Region, not adjustable, against a platform that mints one
+ // bucket per app — which is why it is not the default engine.
+ Capacity(ctx context.Context) (remaining, cap int, err error)
+ // Provenance names what did the scanning — the engine and its signature
+ // database version — for the evidence record. "Scanned with what?" is an
+ // audit question, and a clean verdict from three-month-old signatures
+ // should not read like a fresh one.
+ Provenance(ctx context.Context) (engine, version string, err error)
+}
+
+// Verdict is one object's scan result. The zero value is "no recorded
+// verdict", which is a state and not an error.
+type Verdict string
+
+const (
+ VerdictUnknown Verdict = ""
+ VerdictClean Verdict = "clean"
+ // VerdictInfected is the only verdict the Deny statement acts on.
+ VerdictInfected Verdict = "infected"
+ // VerdictSkipped is too large, unsupported, or unextractable. It is NOT
+ // clean, and the evidence must never blur the two: a store that is mostly
+ // skips has not been meaningfully scanned.
+ VerdictSkipped Verdict = "skipped"
+ // VerdictDenied is the engine being unable to read the object. Distinct
+ // from Failed because the repair is a permissions fix, not a retry.
+ VerdictDenied Verdict = "denied"
+ // VerdictFailed is the engine reading the object and failing on it. It is
+ // NOT "the engine was unreachable" — an outage must never stamp thousands
+ // of objects with a verdict about their contents.
+ VerdictFailed Verdict = "failed"
+)
+
+// DenyTag is the object tag an engine writes for an infected object, and the
+// literal a Deny statement matches on (spec §8).
+//
+// It lives in the RECORD rather than being asked of a Scanner at policy-write
+// time, because Grant takes a Provisioner and has no Scanner to ask. That is
+// also what lets `scan enforce off` clear a deny on a deployment with no
+// engine built — without it, unsetting CARLOS_STORE_SCANNER on a deployment
+// holding any enforcing record would leave that store unrepairable.
+type DenyTag struct {
+ Key string `json:"key"`
+ Value string `json:"value"`
+}
+
+// The tags the ClamAV engine writes. TagETag is what binds a verdict to the
+// bytes it judged: without it a tag says only "some bytes under this key were
+// clean once", and a re-uploaded object would inherit its predecessor's
+// verdict.
+const (
+ TagVerdict = "carlos-scan"
+ TagETag = "carlos-scan-etag"
+ TagAt = "carlos-scan-at"
+)
diff --git a/internal/objstore/scanner_contract_test.go b/internal/objstore/scanner_contract_test.go
new file mode 100644
index 000000000..1359f8bab
--- /dev/null
+++ b/internal/objstore/scanner_contract_test.go
@@ -0,0 +1,84 @@
+package objstore
+
+import (
+ "context"
+ "testing"
+)
+
+// scannerContract is run against every Scanner implementation. Idempotence is
+// the property under test: every method is called twice and the second call
+// must be a success, because the sweep re-runs after a half-failure at any
+// point and a walk that only works once is not re-runnable.
+func scannerContract(t *testing.T, s Scanner, bucket string) {
+ t.Helper()
+ ctx := context.Background()
+
+ for i := 0; i < 2; i++ {
+ if err := s.EnsureEnabled(ctx, bucket); err != nil {
+ t.Fatalf("EnsureEnabled call %d: %v", i+1, err)
+ }
+ }
+
+ // An object nobody has scanned has no verdict, and that is a state rather
+ // than an error — the same idiom Get uses for an absent record.
+ v, err := s.Verdict(ctx, bucket, "absent.txt", "etag-1")
+ if err != nil {
+ t.Fatalf("Verdict on an unscanned object: %v", err)
+ }
+ if v != VerdictUnknown {
+ t.Fatalf("Verdict on an unscanned object = %q, want %q", v, VerdictUnknown)
+ }
+
+ for i := 0; i < 2; i++ {
+ if _, err := s.Scan(ctx, bucket, "a.txt", "etag-1", 10); err != nil {
+ t.Fatalf("Scan call %d: %v", i+1, err)
+ }
+ }
+
+ // A verdict is bound to the bytes it judged. The same key at a different
+ // etag has no verdict, so the sweep rescans rather than letting a
+ // re-uploaded object inherit its predecessor's result.
+ if v, err := s.Verdict(ctx, bucket, "a.txt", "etag-2"); err != nil || v != VerdictUnknown {
+ t.Fatalf("Verdict at a changed etag = (%q, %v), want (%q, nil)", v, err, VerdictUnknown)
+ }
+
+ for i := 0; i < 2; i++ {
+ if err := s.EnsureDisabled(ctx, bucket); err != nil {
+ t.Fatalf("EnsureDisabled call %d: %v", i+1, err)
+ }
+ }
+}
+
+func TestMemScannerContract(t *testing.T) {
+ scannerContract(t, NewMemScanner(), "carlos-store-tito-bab-titogo")
+}
+
+// The seeded-verdict path gets the same etag treatment as the scanned one, so
+// a fixture cannot accidentally set up a verdict that outlives its bytes.
+func TestMemScannerSeededVerdictIsETagBound(t *testing.T) {
+ ctx := context.Background()
+ s := NewMemScanner()
+ s.Put("b", "k", "etag-1", VerdictInfected)
+
+ if v, _ := s.Verdict(ctx, "b", "k", "etag-1"); v != VerdictInfected {
+ t.Fatalf("same etag = %q, want %q", v, VerdictInfected)
+ }
+ if v, _ := s.Verdict(ctx, "b", "k", "etag-2"); v != VerdictUnknown {
+ t.Fatalf("changed etag = %q, want %q", v, VerdictUnknown)
+ }
+}
+
+// The failure hook arms exactly one failure, so a walk test can prove the
+// sweep survives a mid-flight engine error and does not retry forever.
+func TestMemScannerFailureIsOneShot(t *testing.T) {
+ ctx := context.Background()
+ s := NewMemScanner()
+ s.Fail("EnsureEnabled", context.DeadlineExceeded)
+
+ if err := s.EnsureEnabled(ctx, "b"); err == nil {
+ t.Fatal("armed failure did not fire")
+ }
+ if err := s.EnsureEnabled(ctx, "b"); err != nil {
+ t.Fatalf("failure fired twice: %v", err)
+ }
+}