| 1 | // The sweep (spec §7): a resumable reconciliation pass over one bucket. |
| 2 | // |
| 3 | // It is a reconciler, not a pipeline. Nothing in CARLOS sees a PUT — the |
| 4 | // control plane hands an app its own credentials and then never touches the |
| 5 | // data plane — so there is no upload to hook. What a reconciler can do that a |
| 6 | // pipeline cannot is COUNT WHAT IT HAS NOT SEEN: a pipeline's misses are |
| 7 | // invisible by construction, while a walk that lists the bucket knows exactly |
| 8 | // how many objects carry no verdict. That number is the whole point. An audit |
| 9 | // answer needs two sentences, and the second one is the honest one. |
| 10 | package objstore |
| 11 | |
| 12 | import ( |
| 13 | "context" |
| 14 | "errors" |
| 15 | "fmt" |
| 16 | "time" |
| 17 | |
| 18 | awss3 "github.com/aws/aws-sdk-go-v2/service/s3" |
| 19 | |
| 20 | "github.com/carlosframework/platform/internal/release" |
| 21 | ) |
| 22 | |
| 23 | // sweepPageSize is how many keys one listing page asks for. The page is the |
| 24 | // unit of durable progress: everything on it is scanned, then the sidecar is |
| 25 | // written once. Smaller pages mean more sidecar writes; larger ones mean more |
| 26 | // work lost when a tick's budget runs out mid-page. |
| 27 | const sweepPageSize = 1000 |
| 28 | |
| 29 | // ScanObject is one key as the listing reports it. Size, ETag and |
| 30 | // last-modified all come free on the listing page, which is why the sweep |
| 31 | // never needs a HeadObject: the size decides the ceiling before any fetch, and |
| 32 | // the other two bind the verdict. |
| 33 | type ScanObject struct { |
| 34 | Key string |
| 35 | ETag string |
| 36 | Modified time.Time |
| 37 | Size int64 |
| 38 | } |
| 39 | |
| 40 | // Version is the token a verdict is bound to: this exact version of these |
| 41 | // exact bytes. |
| 42 | // |
| 43 | // Both halves earn their place. The ETag alone would be the content hash, and |
| 44 | // for a single-part upload that hash is MD5 — chosen-prefix MD5 collisions are |
| 45 | // practical, so an attacker controlling both uploads could give a payload the |
| 46 | // ETag of the clean object it replaces and inherit its verdict. Last-modified |
| 47 | // alone would be too coarse. Together they are cheap and neither is the |
| 48 | // uploader's to choose. |
| 49 | func (o ScanObject) Version() string { |
| 50 | return normalizeETag(o.ETag) + "@" + o.Modified.UTC().Format(time.RFC3339) |
| 51 | } |
| 52 | |
| 53 | // ObjectLister is the one listing call the sweep makes, narrowed so the walk |
| 54 | // is testable without S3 or credentials. |
| 55 | type ObjectLister interface { |
| 56 | // ListPage returns up to limit keys ordered after startAfter, and whether |
| 57 | // more remain. |
| 58 | ListPage(ctx context.Context, bucket, startAfter string, limit int32) (objs []ScanObject, truncated bool, err error) |
| 59 | } |
| 60 | |
| 61 | // s3Lister adapts an S3 client to ObjectLister. |
| 62 | type s3Lister struct{ s3 listPageAPI } |
| 63 | |
| 64 | type listPageAPI interface { |
| 65 | ListObjectsV2(ctx context.Context, in *awss3.ListObjectsV2Input, optFns ...func(*awss3.Options)) (*awss3.ListObjectsV2Output, error) |
| 66 | } |
| 67 | |
| 68 | // NewObjectLister adapts an S3 client to the sweep's listing seam. |
| 69 | func NewObjectLister(c listPageAPI) ObjectLister { return &s3Lister{s3: c} } |
| 70 | |
| 71 | func (l *s3Lister) ListPage(ctx context.Context, bucket, startAfter string, limit int32) ([]ScanObject, bool, error) { |
| 72 | in := &awss3.ListObjectsV2Input{Bucket: &bucket, MaxKeys: &limit} |
| 73 | // StartAfter, not ContinuationToken. A token's lifetime across processes |
| 74 | // is unspecified and, on Ceph, unmeasured; a key is a portable resume |
| 75 | // point that survives a console restart and a change of S3 vendor. |
| 76 | if startAfter != "" { |
| 77 | in.StartAfter = &startAfter |
| 78 | } |
| 79 | out, err := l.s3.ListObjectsV2(ctx, in) |
| 80 | if err != nil { |
| 81 | return nil, false, fmt.Errorf("objstore: list %s: %w", bucket, err) |
| 82 | } |
| 83 | objs := make([]ScanObject, 0, len(out.Contents)) |
| 84 | for _, o := range out.Contents { |
| 85 | obj := ScanObject{Key: aws2s(o.Key), ETag: normalizeETag(aws2s(o.ETag))} |
| 86 | if o.Size != nil { |
| 87 | obj.Size = *o.Size |
| 88 | } |
| 89 | if o.LastModified != nil { |
| 90 | obj.Modified = *o.LastModified |
| 91 | } |
| 92 | objs = append(objs, obj) |
| 93 | } |
| 94 | truncated := out.IsTruncated != nil && *out.IsTruncated |
| 95 | return objs, truncated, nil |
| 96 | } |
| 97 | |
| 98 | // errStandDown is not a failure. It is this sweep noticing that someone else — |
| 99 | // another driver, or a teardown — is ahead of it, and that the right move is |
| 100 | // to stop writing. RunSweep converts it to a nil error. |
| 101 | var errStandDown = errors.New("objstore: sweep standing down") |
| 102 | |
| 103 | // The two sweep failures an operator can actually act on, named so a door can |
| 104 | // tell them apart from a transport error and say which it was. Everything else |
| 105 | // stays a logged detail: bucket names, ARNs and request ids do not belong in a |
| 106 | // client sentence. |
| 107 | var ( |
| 108 | // ErrEngineUnreachable is the engine not answering at all. |
| 109 | ErrEngineUnreachable = errors.New("objstore: the scanning engine is not answering") |
| 110 | // ErrScanningDisabled is the deployment switch being off. Its own sentinel |
| 111 | // because a door must tell "nobody turned this on" apart from "the engine |
| 112 | // is broken": one is a command away, the other is an incident. |
| 113 | ErrScanningDisabled = errors.New("objstore: object scanning is disabled for this deployment") |
| 114 | // ErrEngineDrift is a record naming a different engine than the one |
| 115 | // configured. Refusing beats recounting another engine's verdicts as |
| 116 | // unscanned, which would destroy the coverage history. |
| 117 | ErrEngineDrift = errors.New("objstore: the record names a different scanning engine") |
| 118 | ) |
| 119 | |
| 120 | // RunSweep advances one bucket's coverage by up to budget of work and returns |
| 121 | // the sidecar as it now stands. |
| 122 | // |
| 123 | // A nil sidecar with a nil error means there was nothing to do: no store, no |
| 124 | // scanning enabled, or an app in the trash. |
| 125 | func claimRawRunSweep(ctx context.Context, s release.Store, lister ObjectLister, sc Scanner, |
| 126 | acct, app string, budget time.Duration) (*ScanSidecar, error) { |
| 127 | |
| 128 | rec, _, err := Get(ctx, s, acct, app) |
| 129 | if err != nil { |
| 130 | return nil, err |
| 131 | } |
| 132 | if rec == nil || rec.Provisioned == nil || rec.Scanning == nil || rec.Scanning.Enabled == nil { |
| 133 | return nil, nil |
| 134 | } |
| 135 | |
| 136 | // A trashed store is destroyed within 30 days. Scanning it bills for |
| 137 | // egress on data that is on its way out, and reports coverage for an app |
| 138 | // nobody can reach. |
| 139 | tomb, err := release.ReadAppTombstone(ctx, s, acct, app) |
| 140 | if err != nil { |
| 141 | return nil, err |
| 142 | } |
| 143 | if tomb != nil { |
| 144 | return nil, nil |
| 145 | } |
| 146 | |
| 147 | // A scanner cannot read ciphertext. Sweeping a client-encrypted store |
| 148 | // would report every object clean, which is worse than reporting nothing: |
| 149 | // it is a true-looking sentence about bytes nobody examined. Mark it and |
| 150 | // stop. |
| 151 | if rec.Scanning.Requested != nil && rec.Scanning.Requested.Encrypted { |
| 152 | return markNotApplicable(ctx, s, acct, app) |
| 153 | } |
| 154 | |
| 155 | // An engine enabled with no attestation on record is refused rather than |
| 156 | // swept. The grant door will not create that state any more, so this is |
| 157 | // unreachable in practice — but sweeping reads tenant object bytes, and |
| 158 | // the attestation is the consent for that. Guessing "probably plaintext" |
| 159 | // is not a decision this walk gets to make on the member's behalf. |
| 160 | if rec.Scanning.Requested == nil { |
| 161 | return nil, fmt.Errorf("objstore: %s/%s: scanning is enabled but no attestation is on record "+ |
| 162 | "— the member runs `carlos store scan request` to say whether these objects are readable", |
| 163 | acct, app) |
| 164 | } |
| 165 | |
| 166 | if sc == nil { |
| 167 | return nil, fmt.Errorf("objstore: %s/%s has scanning enabled but no engine is configured "+ |
| 168 | "(set CARLOS_CLAMD_ADDR)", acct, app) |
| 169 | } |
| 170 | |
| 171 | // The DEPLOYMENT switch, checked here and not only in the pass. |
| 172 | // |
| 173 | // A tick that goes quiet is not the same as scanning being off: the |
| 174 | // on-demand sweep verb reaches this function by its own path, so without |
| 175 | // this an operator could run `carlos system antivirus disable`, watch the |
| 176 | // hourly pass stop, and still have `carlos store scan sweep` read tenant |
| 177 | // bytes. Disable has to disable everywhere or it disables nothing. |
| 178 | on, err := release.AntivirusEnabled(ctx, s) |
| 179 | if err != nil { |
| 180 | return nil, err |
| 181 | } |
| 182 | if !on { |
| 183 | return nil, fmt.Errorf("%w (%s/%s)", ErrScanningDisabled, acct, app) |
| 184 | } |
| 185 | |
| 186 | engine, signature, err := sc.Provenance(ctx) |
| 187 | if err != nil { |
| 188 | return nil, fmt.Errorf("%w (%s/%s): %v", ErrEngineUnreachable, acct, app, err) |
| 189 | } |
| 190 | // Engine drift refuses rather than guesses. Verdicts already in the bucket |
| 191 | // were written by a different engine under its own conventions, and |
| 192 | // recounting them as unscanned would destroy the coverage history. An |
| 193 | // operator changes engine by revoking and re-granting, deliberately. |
| 194 | if rec.Scanning.Enabled.Engine != engine { |
| 195 | return nil, fmt.Errorf("%w (%s/%s): record says %q, configured engine is %q", |
| 196 | ErrEngineDrift, acct, app, rec.Scanning.Enabled.Engine, engine) |
| 197 | } |
| 198 | |
| 199 | side, etag, err := GetSidecar(ctx, s, acct, app) |
| 200 | if err != nil { |
| 201 | return nil, err |
| 202 | } |
| 203 | if side == nil { |
| 204 | side = &ScanSidecar{} |
| 205 | } |
| 206 | side.NotApplicable = false |
| 207 | if side.Current == nil { |
| 208 | side.Current = &Sweep{StartedAt: time.Now().UTC()} |
| 209 | } |
| 210 | side.Current.Engine = engine |
| 211 | side.Current.Signature = signature |
| 212 | |
| 213 | bucket := rec.Provisioned.Bucket |
| 214 | deadline := time.Now().Add(budget) |
| 215 | |
| 216 | for { |
| 217 | pageStart := side.Current.After |
| 218 | objs, truncated, err := lister.ListPage(ctx, bucket, pageStart, sweepPageSize) |
| 219 | if err != nil { |
| 220 | return nil, err |
| 221 | } |
| 222 | // Verdicts reached on THIS page versus verdicts carried forward from an |
| 223 | // earlier sweep. A clean tag is reused whenever the ETag still matches, |
| 224 | // so a completed rollup can consist almost entirely of judgements made |
| 225 | // months ago — against signatures that are not the ones the sidecar now |
| 226 | // names. Counting the two apart is what stops "engine database" being |
| 227 | // read as "every object below was judged against this". |
| 228 | var fresh, reused int64 |
| 229 | |
| 230 | for _, o := range objs { |
| 231 | if err := ctx.Err(); err != nil { |
| 232 | return nil, err |
| 233 | } |
| 234 | v, err := sc.Verdict(ctx, bucket, o.Key, o.Version()) |
| 235 | if errors.Is(err, ErrObjectGone) { |
| 236 | continue // deleted since the listing; not ours to count |
| 237 | } |
| 238 | if err != nil { |
| 239 | return nil, err |
| 240 | } |
| 241 | if v == VerdictUnknown { |
| 242 | // A scan error here is the ENGINE failing, not this object: |
| 243 | // the next object would fail identically. Abandon the page |
| 244 | // without writing. Nothing is lost but the listing — every |
| 245 | // verdict already reached is a tag in the bucket, so the |
| 246 | // re-run reads them back without re-scanning a byte. |
| 247 | // |
| 248 | // Per-object facts do NOT arrive here as errors. An object we |
| 249 | // are not allowed to read comes back as VerdictDenied and one |
| 250 | // that has been deleted as ErrObjectGone, precisely so that |
| 251 | // neither can stall the page: an error here abandons the page |
| 252 | // unwritten, and a page that fails identically every tick |
| 253 | // means the bucket's sweep never advances past it. |
| 254 | // The recorded flag is the on-demand drain's business, not |
| 255 | // the sweep's: Fresh and Reused count verdicts REACHED, and |
| 256 | // the sweep re-lists every cycle anyway, so an unpersisted |
| 257 | // verdict simply gets reached again. |
| 258 | v, _, err = sc.Scan(ctx, bucket, o.Key, o.Version(), o.Size) |
| 259 | if errors.Is(err, ErrObjectGone) { |
| 260 | continue |
| 261 | } |
| 262 | if err != nil { |
| 263 | return nil, err |
| 264 | } |
| 265 | if isCoverage(v) { |
| 266 | fresh++ |
| 267 | } |
| 268 | } else if isCoverage(v) { |
| 269 | reused++ |
| 270 | } |
| 271 | side.Current.count(v, o.Size) |
| 272 | } |
| 273 | |
| 274 | side.Current.Fresh += fresh |
| 275 | side.Current.Reused += reused |
| 276 | side.Current.At = time.Now().UTC() |
| 277 | if len(objs) > 0 { |
| 278 | side.Current.After = objs[len(objs)-1].Key |
| 279 | } |
| 280 | |
| 281 | last := !truncated |
| 282 | if last { |
| 283 | // The completed sweep becomes the answer, and Current clears. A |
| 284 | // single field would have the next tick overwrite the compliance |
| 285 | // artifact with page-one partials. |
| 286 | side.Current.After = "" |
| 287 | side.LastComplete = side.Current |
| 288 | side.Current = nil |
| 289 | } |
| 290 | |
| 291 | newETag, err := writeSidecar(ctx, s, acct, app, side, etag, pageStart) |
| 292 | if errors.Is(err, errStandDown) { |
| 293 | return side, nil |
| 294 | } |
| 295 | if err != nil { |
| 296 | return nil, err |
| 297 | } |
| 298 | etag = newETag |
| 299 | |
| 300 | if last || time.Now().After(deadline) { |
| 301 | return side, nil |
| 302 | } |
| 303 | } |
| 304 | } |
| 305 | |
| 306 | // isCoverage is a verdict that counts as Scanned. Fresh and Reused split THAT |
| 307 | // number, not the object count: an object that was skipped or denied was never |
| 308 | // judged at all, so calling its verdict "fresh" would make the printed |
| 309 | // arithmetic — "N of M scanned, of those X judged now" — visibly wrong the |
| 310 | // moment a single object is too large. |
| 311 | func isCoverage(v Verdict) bool { |
| 312 | return v == VerdictClean || v == VerdictInfected |
| 313 | } |
| 314 | |
| 315 | // count folds one object's verdict into the rollup. |
| 316 | // |
| 317 | // Scanned and Unscanned are not complements, and the arithmetic here is the |
| 318 | // reason the audit answer needs two sentences. An infected object WAS scanned |
| 319 | // — we know what it is. A skipped one was not judged at all. |
| 320 | func (sw *Sweep) count(v Verdict, size int64) { |
| 321 | sw.Objects++ |
| 322 | sw.Bytes += size |
| 323 | switch v { |
| 324 | case VerdictClean: |
| 325 | sw.Scanned++ |
| 326 | case VerdictInfected: |
| 327 | sw.Scanned++ |
| 328 | sw.Infected++ |
| 329 | case VerdictSkipped: |
| 330 | sw.Skipped++ |
| 331 | sw.Unscanned++ |
| 332 | case VerdictDenied: |
| 333 | sw.Denied++ |
| 334 | sw.Unscanned++ |
| 335 | case VerdictFailed: |
| 336 | sw.Failed++ |
| 337 | sw.Unscanned++ |
| 338 | default: |
| 339 | sw.Unscanned++ |
| 340 | } |
| 341 | } |
| 342 | |
| 343 | // markNotApplicable records that no coverage is claimed for this store, and |
| 344 | // does no listing at all. |
| 345 | func markNotApplicable(ctx context.Context, s release.Store, acct, app string) (*ScanSidecar, error) { |
| 346 | side, etag, err := GetSidecar(ctx, s, acct, app) |
| 347 | if err != nil { |
| 348 | return nil, err |
| 349 | } |
| 350 | if side != nil && side.NotApplicable && side.Current == nil { |
| 351 | return side, nil // already recorded; no write, no CAS churn |
| 352 | } |
| 353 | if side == nil { |
| 354 | side = &ScanSidecar{} |
| 355 | } |
| 356 | side.NotApplicable = true |
| 357 | side.Current = nil |
| 358 | if _, err := writeSidecar(ctx, s, acct, app, side, etag, ""); err != nil { |
| 359 | if errors.Is(err, errStandDown) { |
| 360 | return side, nil |
| 361 | } |
| 362 | return nil, err |
| 363 | } |
| 364 | return side, nil |
| 365 | } |
| 366 | |
| 367 | // writeSidecar performs the pre-write re-read and the CAS retry guard. |
| 368 | // |
| 369 | // The re-read closes a hole the retry guard alone cannot. A FRESH sweep's |
| 370 | // first write is create-only by construction — no sidecar exists, so the etag |
| 371 | // is "". Sequence: the sweep reads an enabled store.json, spends seconds |
| 372 | // listing and scanning a page, a teardown runs and PurgeApp deletes the whole |
| 373 | // prefix, and the sweep then writes scan.json create-only, which SUCCEEDS. The |
| 374 | // result is an orphan under a purged app, and that orphan is expensive: the |
| 375 | // app-name claim gate refuses any name whose prefix still holds a key, rename |
| 376 | // is refused, and box-side auto-purge never reaches it because the tombstone |
| 377 | // was written last. The app name becomes permanently unclaimable. |
| 378 | // |
| 379 | // The one-round-trip race this still leaves is the same one |
| 380 | // deleteRecordIfUnchanged already accepts. |
| 381 | func writeSidecar(ctx context.Context, s release.Store, acct, app string, |
| 382 | side *ScanSidecar, etag, pageStart string) (string, error) { |
| 383 | |
| 384 | rec, _, err := Get(ctx, s, acct, app) |
| 385 | if err != nil { |
| 386 | return "", err |
| 387 | } |
| 388 | if rec == nil || rec.Scanning == nil || rec.Scanning.Enabled == nil { |
| 389 | return "", errStandDown |
| 390 | } |
| 391 | |
| 392 | newETag, err := PutSidecar(ctx, s, acct, app, side, etag) |
| 393 | if err == nil { |
| 394 | return newETag, nil |
| 395 | } |
| 396 | if !errors.Is(err, release.ErrCASConflict) { |
| 397 | return "", err |
| 398 | } |
| 399 | |
| 400 | // Someone else wrote. A conditional write against a DELETED key also |
| 401 | // reads as ErrCASConflict, so a blind retry would fall back to a |
| 402 | // create-only write and resurrect a sidecar teardown just removed. Look |
| 403 | // before retrying. |
| 404 | current, freshETag, gerr := GetSidecar(ctx, s, acct, app) |
| 405 | if gerr != nil { |
| 406 | return "", gerr |
| 407 | } |
| 408 | if current == nil || current.Current == nil || current.Current.After != pageStart { |
| 409 | // Deleted, completed by another driver, or advanced past us. In every |
| 410 | // case this sweep is no longer the one making progress. |
| 411 | return "", errStandDown |
| 412 | } |
| 413 | newETag, err = PutSidecar(ctx, s, acct, app, side, freshETag) |
| 414 | if err != nil { |
| 415 | if errors.Is(err, release.ErrCASConflict) { |
| 416 | return "", errStandDown |
| 417 | } |
| 418 | return "", err |
| 419 | } |
| 420 | return newETag, nil |
| 421 | } |
| 422 | func RunSweep(ctx context.Context, s release.Store, lister ObjectLister, sc Scanner, acct, app string, budget time.Duration) (claimResult0 *ScanSidecar, claimResult1 error) { |
| 423 | claimResult1 = release.WithAppMutation(ctx, s, acct, app, "objstore.RunSweep", func(claimContext context.Context) error { |
| 424 | claimResult0, claimResult1 = claimRawRunSweep(claimContext, s, lister, sc, acct, app, budget) |
| 425 | return claimResult1 |
| 426 | }) |
| 427 | return |
| 428 | } |
| 429 | |
amadan