| 1 | package objstore |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "sync" |
| 6 | "time" |
| 7 | ) |
| 8 | |
| 9 | // MemScanner is the in-memory Scanner every walk test runs against, mirroring |
| 10 | // MemProvisioner: no network, no clamd, and a failure hook so a test can make |
| 11 | // any single method fail once and prove the walk survives it. |
| 12 | type MemScanner struct { |
| 13 | // onStatus is OnStatus's hook; see its comment. |
| 14 | onStatus func() |
| 15 | mu sync.Mutex |
| 16 | enabled map[string]bool |
| 17 | verdicts map[string]scanRecord |
| 18 | failures map[string]error |
| 19 | scanAs map[string]Verdict |
| 20 | // objects is what a listing would have said about a key. Status needs it: |
| 21 | // a caller holding only a key has no listing page to take a version from. |
| 22 | objects map[string]ScanObject |
| 23 | // unrecordable are keys whose verdict cannot be persisted — the ten-tag |
| 24 | // cap, in production. A hook rather than a stub, because the case is the |
| 25 | // reason Scan reports `recorded` at all. |
| 26 | unrecordable map[string]bool |
| 27 | // deniedHead are keys whose HeadObject is refused, so Status answers |
| 28 | // denied with a ZERO ScanObject — the one shape that leaves a caller with |
| 29 | // no version and no size. |
| 30 | deniedHead map[string]bool |
| 31 | // scans counts Scan calls per key, so a test can assert that deduping |
| 32 | // really stopped the engine reading the same object twice rather than |
| 33 | // merely folding two outcomes into one. |
| 34 | scans map[string]int |
| 35 | } |
| 36 | |
| 37 | // scanRecord is one object's verdict AND the object version it was judged |
| 38 | // against. The version is not decoration: Verdict answers Unknown when it |
| 39 | // disagrees, which is what stops a rewritten object inheriting the old bytes' |
| 40 | // verdict. |
| 41 | type scanRecord struct { |
| 42 | version string |
| 43 | verdict Verdict |
| 44 | } |
| 45 | |
| 46 | func NewMemScanner() *MemScanner { |
| 47 | return &MemScanner{ |
| 48 | enabled: map[string]bool{}, |
| 49 | verdicts: map[string]scanRecord{}, |
| 50 | failures: map[string]error{}, |
| 51 | } |
| 52 | } |
| 53 | |
| 54 | func memScanKey(bucket, key string) string { return bucket + "\x00" + key } |
| 55 | |
| 56 | // Fail arms a one-shot failure for the named method, the same shape |
| 57 | // MemProvisioner.takeFailure gives the provisioner walks. |
| 58 | func (s *MemScanner) Fail(method string, err error) { |
| 59 | s.mu.Lock() |
| 60 | defer s.mu.Unlock() |
| 61 | s.failures[method] = err |
| 62 | } |
| 63 | |
| 64 | // PutObject seeds what a listing would have said about a key, so Status can |
| 65 | // answer. Without it Status cannot tell an absent object from an unscanned |
| 66 | // one, and those are different answers. |
| 67 | func (s *MemScanner) PutObject(bucket, key, etag string, modified time.Time, size int64) { |
| 68 | s.mu.Lock() |
| 69 | defer s.mu.Unlock() |
| 70 | if s.objects == nil { |
| 71 | s.objects = map[string]ScanObject{} |
| 72 | } |
| 73 | s.objects[memScanKey(bucket, key)] = ScanObject{ |
| 74 | Key: key, ETag: etag, Modified: modified, Size: size, |
| 75 | } |
| 76 | } |
| 77 | |
| 78 | // Unrecordable makes Scan reach a verdict it cannot persist, the way the |
| 79 | // ten-tag cap does in production. |
| 80 | func (s *MemScanner) Unrecordable(key string) { |
| 81 | s.mu.Lock() |
| 82 | defer s.mu.Unlock() |
| 83 | if s.unrecordable == nil { |
| 84 | s.unrecordable = map[string]bool{} |
| 85 | } |
| 86 | s.unrecordable[key] = true |
| 87 | } |
| 88 | |
| 89 | // OnStatus installs a hook that runs at the top of Status, BEFORE any lock is |
| 90 | // taken, so a test can make something else happen while a caller is mid-read. |
| 91 | // |
| 92 | // It exists for one assertion that has no other seam: the status door |
| 93 | // registers its long-poll waiter before its first read, and the only way to |
| 94 | // catch the opposite order is to run a drain inside that read. The hook must |
| 95 | // not re-enter the scanner (Status will be called again beneath it), so a test |
| 96 | // uses sync.Once. |
| 97 | func (s *MemScanner) OnStatus(f func()) { |
| 98 | s.mu.Lock() |
| 99 | defer s.mu.Unlock() |
| 100 | s.onStatus = f |
| 101 | } |
| 102 | |
| 103 | // Status answers about the CURRENT version, and reports what the record SAYS |
| 104 | // — denied and failed included. The re-issue rule below is Verdict's, and |
| 105 | // applying it here would make a failed scan read as pending forever. |
| 106 | func (s *MemScanner) Status(ctx context.Context, bucket, key string) (Verdict, ScanObject, error) { |
| 107 | // The hook runs unlocked, or anything it does that touches this scanner |
| 108 | // deadlocks against the lock taken below. |
| 109 | s.mu.Lock() |
| 110 | hook := s.onStatus |
| 111 | s.mu.Unlock() |
| 112 | if hook != nil { |
| 113 | hook() |
| 114 | } |
| 115 | s.mu.Lock() |
| 116 | defer s.mu.Unlock() |
| 117 | if err := s.takeFailure("Status"); err != nil { |
| 118 | return VerdictUnknown, ScanObject{}, err |
| 119 | } |
| 120 | if s.deniedHead[memScanKey(bucket, key)] { |
| 121 | // HeadObject itself refused: no version, no size, nothing to scan |
| 122 | // with. The production shape is a bucket policy the console is not |
| 123 | // covered by, as against the SSE-KMS case where the head succeeds and |
| 124 | // the GET is what fails. |
| 125 | return VerdictDenied, ScanObject{}, nil |
| 126 | } |
| 127 | obj, ok := s.objects[memScanKey(bucket, key)] |
| 128 | if !ok { |
| 129 | return VerdictUnknown, ScanObject{}, ErrObjectGone |
| 130 | } |
| 131 | rec, ok := s.verdicts[memScanKey(bucket, key)] |
| 132 | if !ok || rec.version != obj.Version() { |
| 133 | return VerdictUnknown, obj, nil |
| 134 | } |
| 135 | return rec.verdict, obj, nil |
| 136 | } |
| 137 | |
| 138 | // Put seeds a verdict so a test can start from a partly-scanned bucket. |
| 139 | func (s *MemScanner) Put(bucket, key, version string, v Verdict) { |
| 140 | s.mu.Lock() |
| 141 | defer s.mu.Unlock() |
| 142 | s.verdicts[memScanKey(bucket, key)] = scanRecord{version: version, verdict: v} |
| 143 | } |
| 144 | |
| 145 | // IsEnabled lets Teardown's test prove the scanner was disabled while the |
| 146 | // record that named it still existed. |
| 147 | func (s *MemScanner) IsEnabled(bucket string) bool { |
| 148 | s.mu.Lock() |
| 149 | defer s.mu.Unlock() |
| 150 | return s.enabled[bucket] |
| 151 | } |
| 152 | |
| 153 | func (s *MemScanner) takeFailure(method string) error { |
| 154 | if err, ok := s.failures[method]; ok { |
| 155 | delete(s.failures, method) |
| 156 | return err |
| 157 | } |
| 158 | return nil |
| 159 | } |
| 160 | |
| 161 | func (s *MemScanner) EnsureEnabled(ctx context.Context, bucket string) error { |
| 162 | s.mu.Lock() |
| 163 | defer s.mu.Unlock() |
| 164 | if err := s.takeFailure("EnsureEnabled"); err != nil { |
| 165 | return err |
| 166 | } |
| 167 | s.enabled[bucket] = true // already enabled is a success |
| 168 | return nil |
| 169 | } |
| 170 | |
| 171 | func (s *MemScanner) EnsureDisabled(ctx context.Context, bucket string) error { |
| 172 | s.mu.Lock() |
| 173 | defer s.mu.Unlock() |
| 174 | if err := s.takeFailure("EnsureDisabled"); err != nil { |
| 175 | return err |
| 176 | } |
| 177 | delete(s.enabled, bucket) // already gone is a success |
| 178 | return nil |
| 179 | } |
| 180 | |
| 181 | func (s *MemScanner) Verdict(ctx context.Context, bucket, key, version string) (Verdict, error) { |
| 182 | s.mu.Lock() |
| 183 | defer s.mu.Unlock() |
| 184 | if err := s.takeFailure("Verdict"); err != nil { |
| 185 | return VerdictUnknown, err |
| 186 | } |
| 187 | rec, ok := s.verdicts[memScanKey(bucket, key)] |
| 188 | if !ok || rec.version != version { |
| 189 | return VerdictUnknown, nil |
| 190 | } |
| 191 | // The SWEEP's reading: denied and failed are re-issued rather than |
| 192 | // inherited, because both are transient and the object is not rewritten, |
| 193 | // so its version never changes and the repair would never take. Status |
| 194 | // deliberately does not do this — see the Scanner interface. |
| 195 | switch rec.verdict { |
| 196 | case VerdictDenied, VerdictFailed: |
| 197 | return VerdictUnknown, nil |
| 198 | } |
| 199 | return rec.verdict, nil |
| 200 | } |
| 201 | |
| 202 | func (s *MemScanner) Scan(ctx context.Context, bucket, key, version string, size int64) (Verdict, bool, error) { |
| 203 | s.mu.Lock() |
| 204 | defer s.mu.Unlock() |
| 205 | if err := s.takeFailure("Scan"); err != nil { |
| 206 | return VerdictUnknown, false, err |
| 207 | } |
| 208 | if s.scans == nil { |
| 209 | s.scans = map[string]int{} |
| 210 | } |
| 211 | s.scans[key]++ |
| 212 | v := VerdictClean |
| 213 | if got, ok := s.scanAs[key]; ok { |
| 214 | v = got |
| 215 | } |
| 216 | if s.unrecordable[key] { |
| 217 | // A verdict reached and not persisted: the engine judged the object, |
| 218 | // but nothing on the object says so. |
| 219 | return v, false, nil |
| 220 | } |
| 221 | s.verdicts[memScanKey(bucket, key)] = scanRecord{version: version, verdict: v} |
| 222 | return v, true, nil |
| 223 | } |
| 224 | |
| 225 | // DenyHead makes Status refuse the head of one key. |
| 226 | func (s *MemScanner) DenyHead(bucket, key string) { |
| 227 | s.mu.Lock() |
| 228 | defer s.mu.Unlock() |
| 229 | if s.deniedHead == nil { |
| 230 | s.deniedHead = map[string]bool{} |
| 231 | } |
| 232 | s.deniedHead[memScanKey(bucket, key)] = true |
| 233 | } |
| 234 | |
| 235 | // Scans is how many times the engine was asked to read one key. |
| 236 | func (s *MemScanner) Scans(key string) int { |
| 237 | s.mu.Lock() |
| 238 | defer s.mu.Unlock() |
| 239 | return s.scans[key] |
| 240 | } |
| 241 | |
| 242 | // ScanAs makes Scan answer v for key rather than clean, so a test can stage a |
| 243 | // skipped or infected object without a real engine. Skipped in particular is |
| 244 | // not a hypothetical: it is every object over the size ceiling. |
| 245 | func (s *MemScanner) ScanAs(key string, v Verdict) { |
| 246 | s.mu.Lock() |
| 247 | defer s.mu.Unlock() |
| 248 | if s.scanAs == nil { |
| 249 | s.scanAs = map[string]Verdict{} |
| 250 | } |
| 251 | s.scanAs[key] = v |
| 252 | } |
| 253 | |
| 254 | func (s *MemScanner) Enforcement() DenyTag { |
| 255 | return DenyTag{Key: TagVerdict, Value: string(VerdictInfected)} |
| 256 | } |
| 257 | |
| 258 | func (s *MemScanner) Capacity(ctx context.Context) (int, int, error) { return -1, -1, nil } |
| 259 | |
| 260 | // Provenance answers "clamav" rather than "memscanner" ON PURPOSE. The sweep |
| 261 | // refuses when the record's engine disagrees with the running one (spec §7.4), |
| 262 | // and test fixtures write Engine: "clamav" — a fake that named itself would |
| 263 | // make every one of those fixtures refuse instead of sweep. |
| 264 | func (s *MemScanner) Provenance(ctx context.Context) (string, string, error) { |
| 265 | s.mu.Lock() |
| 266 | defer s.mu.Unlock() |
| 267 | // Armable like the others: an engine that will not say what it is, is a |
| 268 | // state the drain and the sweep both have to survive, and a fake that |
| 269 | // could never be in it left that path untested. |
| 270 | if err := s.takeFailure("Provenance"); err != nil { |
| 271 | return "", "", err |
| 272 | } |
| 273 | return "clamav", "MemScanner/0/never", nil |
| 274 | } |
| 275 | |
| 276 | var _ Scanner = (*MemScanner)(nil) |
| 277 | |
amadan