| 1 | // The scan queue: apps/<acct>/<app>/scan-queue/<id>.json, one object per item. |
| 2 | // |
| 3 | // ONE OBJECT PER ITEM, not one queue object. A shared queue object would put |
| 4 | // every enqueue in a compare-and-swap race with every other, and losing that |
| 5 | // race would be the normal case rather than the exception — the same problem |
| 6 | // the sidecar design avoided by having a single writer. Distinct keys never |
| 7 | // collide, so an enqueue needs no coordination at all. |
| 8 | // |
| 9 | // UNDER THE APP'S PREFIX, so PurgeApp removes the items with everything else |
| 10 | // and the name-claim gate can see them. A top-level scan-queue/<acct>/<app>/ |
| 11 | // would be found by one listing instead of an index — but PurgeApp lists only |
| 12 | // the app prefix, and the claim gate never looks outside it, so items would |
| 13 | // survive app deletion invisibly. The index below is the price of keeping them |
| 14 | // where deletion can find them. |
| 15 | // |
| 16 | // The walk lives here and the goroutine lives in console, mirroring how |
| 17 | // RunSweep and scanPass already divide: objstore knows what to do to a bucket, |
| 18 | // console knows when. |
| 19 | package objstore |
| 20 | |
| 21 | import ( |
| 22 | "context" |
| 23 | "crypto/rand" |
| 24 | "encoding/base32" |
| 25 | "encoding/binary" |
| 26 | "encoding/json" |
| 27 | "errors" |
| 28 | "fmt" |
| 29 | "log" |
| 30 | "os" |
| 31 | "sort" |
| 32 | "sync" |
| 33 | "time" |
| 34 | |
| 35 | "github.com/carlosframework/platform/internal/release" |
| 36 | ) |
| 37 | |
| 38 | // QueueSchema is pinned; an unknown schema refuses. |
| 39 | const QueueSchema = 1 |
| 40 | |
| 41 | // QueueTTL is how long an item may sit before it is dropped rather than |
| 42 | // scanned. One sweep cycle: after that the ordinary sweep has had its chance |
| 43 | // to reach the object, and a stale on-demand answer is worth less than none. |
| 44 | const QueueTTL = 24 * time.Hour |
| 45 | |
| 46 | // QueueItem is one app's request that a named object be scanned soon. |
| 47 | type QueueItem struct { |
| 48 | Schema int `json:"schema"` |
| 49 | Key string `json:"key"` |
| 50 | At time.Time `json:"at"` |
| 51 | By string `json:"by"` |
| 52 | // CallbackPath is the app's own path to POST the result to, validated at |
| 53 | // the door. Never a host: the platform knows where the app lives, and a |
| 54 | // caller-supplied host would be an SSRF against the console. |
| 55 | CallbackPath string `json:"callback_path,omitempty"` |
| 56 | } |
| 57 | |
| 58 | // AppRef names one app, for the index. |
| 59 | type AppRef struct{ Account, App string } |
| 60 | |
| 61 | // queueIDEncoding is lowercase base32-HEX without padding. |
| 62 | // |
| 63 | // The hex alphabet, not the standard one, and the difference is the whole |
| 64 | // point: base32's usual alphabet puts the digits 2-7 at values 26-31, ABOVE |
| 65 | // the letters, while in ASCII digits sort BELOW them. Encoded ids would then |
| 66 | // not sort in the same order as the timestamps they encode, and a prefix |
| 67 | // listing would stop being the queue in order. base32-hex is monotonic — |
| 68 | // 0-9 then a-v, in both value and byte order. |
| 69 | var queueIDEncoding = base32.NewEncoding("0123456789abcdefghijklmnopqrstuv").WithPadding(base32.NoPadding) |
| 70 | |
| 71 | // NewQueueID is a time-ordered id: milliseconds big-endian, then randomness. |
| 72 | // |
| 73 | // Hand-rolled because there is no ULID in this repo and adding a dependency is |
| 74 | // a decision, not a default. The properties that matter are that ids sort by |
| 75 | // time — so a prefix listing IS the queue in order — and that two enqueues in |
| 76 | // the same millisecond do not collide. |
| 77 | func NewQueueID(now time.Time) string { |
| 78 | var b [16]byte |
| 79 | binary.BigEndian.PutUint64(b[:8], uint64(now.UnixMilli())) |
| 80 | if _, err := rand.Read(b[8:]); err != nil { |
| 81 | // crypto/rand does not fail in practice; if it ever did, a |
| 82 | // time-only id still sorts correctly and merely risks a collision |
| 83 | // that the create-only write would catch. |
| 84 | binary.BigEndian.PutUint64(b[8:], uint64(now.UnixNano())) |
| 85 | } |
| 86 | return queueIDEncoding.EncodeToString(b[:]) |
| 87 | } |
| 88 | |
| 89 | // QueuePrefix is where one app's items live. |
| 90 | func QueuePrefix(acct, app string) string { |
| 91 | return release.AppPrefix(acct, app) + "scan-queue/" |
| 92 | } |
| 93 | |
| 94 | // QueueKey is one item's object key. |
| 95 | func QueueKey(acct, app, id string) string { return QueuePrefix(acct, app) + id + ".json" } |
| 96 | |
| 97 | // ---- the index -------------------------------------------------------------- |
| 98 | |
| 99 | // QueueIndex remembers which apps have work, so the drain lists only those |
| 100 | // prefixes instead of walking every app on the deployment every tick. |
| 101 | // |
| 102 | // SOFT STATE, deliberately. Losing it costs latency, never items: the boot |
| 103 | // rebuild finds anything stranded, and the ordinary sweep still reaches every |
| 104 | // object on its own cycle. What it must never do is claim an app is empty when |
| 105 | // it is not, which is what the generation counter below is for. |
| 106 | type QueueIndex struct { |
| 107 | mu sync.Mutex |
| 108 | gen map[AppRef]uint64 |
| 109 | } |
| 110 | |
| 111 | func NewQueueIndex() *QueueIndex { return &QueueIndex{gen: map[AppRef]uint64{}} } |
| 112 | |
| 113 | // Bump marks an app as having work. Called AFTER the item object is written: |
| 114 | // an index entry pointing at nothing is harmless, an item nothing points at is |
| 115 | // stranded until the next boot. |
| 116 | func (i *QueueIndex) Bump(ref AppRef) { |
| 117 | i.mu.Lock() |
| 118 | defer i.mu.Unlock() |
| 119 | if i.gen == nil { |
| 120 | i.gen = map[AppRef]uint64{} |
| 121 | } |
| 122 | i.gen[ref]++ |
| 123 | } |
| 124 | |
| 125 | // Observe reads the generation a drain is about to work against. |
| 126 | func (i *QueueIndex) Observe(ref AppRef) uint64 { |
| 127 | i.mu.Lock() |
| 128 | defer i.mu.Unlock() |
| 129 | return i.gen[ref] |
| 130 | } |
| 131 | |
| 132 | // Release drops an app's entry, and reports whether it did. |
| 133 | // |
| 134 | // TWO CONDITIONS, and the second is the one an earlier draft of this design |
| 135 | // got wrong. The generation must be unchanged — no enqueue landed while we |
| 136 | // worked — AND the listing must have come back empty. Generation alone is not |
| 137 | // enough: the drain caps how many items it handles per tick, so on a flooded |
| 138 | // app it deliberately leaves items behind with no enqueue to bump anything, |
| 139 | // and releasing there would delete the entry pointing at them. |
| 140 | func (i *QueueIndex) Release(ref AppRef, observed uint64, listedEmpty bool) bool { |
| 141 | if !listedEmpty { |
| 142 | return false |
| 143 | } |
| 144 | i.mu.Lock() |
| 145 | defer i.mu.Unlock() |
| 146 | if i.gen[ref] != observed { |
| 147 | return false |
| 148 | } |
| 149 | delete(i.gen, ref) |
| 150 | return true |
| 151 | } |
| 152 | |
| 153 | // Apps lists the apps with work, for the drain's round-robin. |
| 154 | func (i *QueueIndex) Apps() []AppRef { |
| 155 | i.mu.Lock() |
| 156 | defer i.mu.Unlock() |
| 157 | out := make([]AppRef, 0, len(i.gen)) |
| 158 | for ref := range i.gen { |
| 159 | out = append(out, ref) |
| 160 | } |
| 161 | sort.Slice(out, func(a, b int) bool { |
| 162 | if out[a].Account != out[b].Account { |
| 163 | return out[a].Account < out[b].Account |
| 164 | } |
| 165 | return out[a].App < out[b].App |
| 166 | }) |
| 167 | return out |
| 168 | } |
| 169 | |
| 170 | // Len is how many apps have work — the deployment-wide queue depth surface. |
| 171 | func (i *QueueIndex) Len() int { |
| 172 | i.mu.Lock() |
| 173 | defer i.mu.Unlock() |
| 174 | return len(i.gen) |
| 175 | } |
| 176 | |
| 177 | // ---- enqueue ---------------------------------------------------------------- |
| 178 | |
| 179 | // Enqueue writes one item and then marks the app. |
| 180 | // |
| 181 | // ORDER MATTERS: the object first, the index second. A crash between the two |
| 182 | // strands that item until the next boot rebuild, which is recoverable. The |
| 183 | // other order would mark an app whose item was never written, and a caller |
| 184 | // answered 202 in between would stop retrying an item nothing points at. |
| 185 | // |
| 186 | // ENQUEUE OWNS THE TIMESTAMP, which is why now is a parameter and At is not a |
| 187 | // field the caller fills in. The id is derived from At and the TTL is measured |
| 188 | // from it, so a door that forgot to stamp it would mint an id sorting at year |
| 189 | // one and an item the very next drain discards as expired — a 202 that means |
| 190 | // nothing. One writer, one clock. |
| 191 | func claimRawEnqueue(ctx context.Context, s release.Store, idx *QueueIndex, acct, app string, item *QueueItem, now time.Time) (string, error) { |
| 192 | item.Schema = QueueSchema |
| 193 | item.At = now.UTC() |
| 194 | body, err := json.MarshalIndent(item, "", " ") |
| 195 | if err != nil { |
| 196 | return "", err |
| 197 | } |
| 198 | id := NewQueueID(item.At) |
| 199 | if _, err := s.Put(ctx, QueueKey(acct, app, id), append(body, '\n'), ""); err != nil { |
| 200 | return "", fmt.Errorf("objstore: enqueue %s/%s: %w", acct, app, err) |
| 201 | } |
| 202 | idx.Bump(AppRef{Account: acct, App: app}) |
| 203 | return id, nil |
| 204 | } |
| 205 | |
| 206 | // ---- the boot rebuild ------------------------------------------------------- |
| 207 | |
| 208 | // RebuildQueue restores one app's index entry from what is actually in the |
| 209 | // bucket, and drops what expired while nobody was draining. |
| 210 | // |
| 211 | // THIS IS THE ONLY THING THAT COLLECTS A STRANDED ITEM, so it is not optional. |
| 212 | // Enqueue writes the object and then bumps the index; a crash between the two |
| 213 | // leaves an item in the bucket with nothing pointing at it, and the drain only |
| 214 | // ever looks at apps the index names. Nothing else would ever list that |
| 215 | // prefix again. |
| 216 | // |
| 217 | // UNCAPPED, unlike the drain's listing, and deliberately: the drain's cap is |
| 218 | // fairness between tenants within a tick, while this runs once and must see |
| 219 | // everything or the stranding it exists to fix survives it. |
| 220 | // |
| 221 | // live counts what a drain could still act on, and an item whose bytes could |
| 222 | // not be READ counts as live. An unreadable item is deleted (no tick will ever |
| 223 | // parse it), but a STORE error is the store having a bad moment: keeping the |
| 224 | // index entry costs one empty listing next tick, while dropping it strands the |
| 225 | // item until the next restart. Absence, never error — §7.5. |
| 226 | // |
| 227 | // Reads run outside any operation. The unreadable and expired items are |
| 228 | // collected afterwards in ONE app mutation, objstore.RebuildCollect, entered |
| 229 | // only when there is something to delete: the boot rebuild visits every app |
| 230 | // prefix, and an app with nothing to collect has nothing for it to write. |
| 231 | func RebuildQueue(ctx context.Context, s release.Store, idx *QueueIndex, acct, app string, now time.Time) (live int, err error) { |
| 232 | keys, err := s.List(ctx, QueuePrefix(acct, app)) |
| 233 | if err != nil { |
| 234 | if errors.Is(err, os.ErrNotExist) { |
| 235 | return 0, nil |
| 236 | } |
| 237 | return 0, err |
| 238 | } |
| 239 | // DEFERRED, so that an error partway through still points the index at what |
| 240 | // was found. The first version bumped only on the success path, so one |
| 241 | // failed Delete of an expired item returned early and left the app's LIVE |
| 242 | // items with nothing pointing at them — and since this rebuild is the only |
| 243 | // thing that collects a stranded item, they would have sat until the next |
| 244 | // restart. An index entry for an app with no work is harmless (one empty |
| 245 | // listing, then it is released); a live item with no entry is not. |
| 246 | defer func() { |
| 247 | if live > 0 { |
| 248 | idx.Bump(AppRef{Account: acct, App: app}) |
| 249 | } |
| 250 | }() |
| 251 | // A FAILED COLLECT MUST NOT ABORT THE INDEX PASS. The first version |
| 252 | // returned on the first Delete error, and because ids sort by time the |
| 253 | // EXPIRED items come first — so one refused delete returned before the |
| 254 | // live items had even been looked at, leaving them with nothing pointing |
| 255 | // at them until the next restart. Deleting is this pass's second job; |
| 256 | // indexing is its first, and the first must finish. |
| 257 | var stale []string |
| 258 | for _, qk := range keys { |
| 259 | item, rerr := readQueueItem(ctx, s, qk) |
| 260 | switch { |
| 261 | case errors.Is(rerr, os.ErrNotExist): |
| 262 | continue |
| 263 | case errors.Is(rerr, errQueueItemUnreadable): |
| 264 | stale = append(stale, qk) |
| 265 | continue |
| 266 | case rerr != nil: |
| 267 | log.Printf("objstore: scan queue rebuild: read %s: %v", qk, rerr) |
| 268 | live++ |
| 269 | continue |
| 270 | } |
| 271 | if now.Sub(item.At) > QueueTTL { |
| 272 | stale = append(stale, qk) |
| 273 | continue |
| 274 | } |
| 275 | live++ |
| 276 | } |
| 277 | if len(stale) == 0 { |
| 278 | return live, nil |
| 279 | } |
| 280 | if err := release.WithAppMutation(ctx, s, acct, app, "objstore.RebuildCollect", func(ctx context.Context) error { |
| 281 | return deleteQueueItems(ctx, s, stale) |
| 282 | }); err != nil { |
| 283 | // Still there, so still live: the next rebuild retries them. |
| 284 | log.Printf("objstore: scan queue rebuild: collect %s/%s: %v", acct, app, err) |
| 285 | return live + len(stale), err |
| 286 | } |
| 287 | return live, nil |
| 288 | } |
| 289 | |
| 290 | // QueueDepth is how many items are waiting for one app — the per-app half of |
| 291 | // the depth §4.4 and §10 both want on a status page. |
| 292 | // |
| 293 | // A full listing, not the index: the index counts APPS with work, which is the |
| 294 | // deployment-wide number, and says nothing about how much any one of them has. |
| 295 | // Absence is depth zero, not an error: an app that has never enqueued has no |
| 296 | // prefix. |
| 297 | func QueueDepth(ctx context.Context, s release.Store, acct, app string) (int, error) { |
| 298 | keys, err := s.List(ctx, QueuePrefix(acct, app)) |
| 299 | if err != nil { |
| 300 | if errors.Is(err, os.ErrNotExist) { |
| 301 | return 0, nil |
| 302 | } |
| 303 | return 0, err |
| 304 | } |
| 305 | return len(keys), nil |
| 306 | } |
| 307 | |
| 308 | // ---- draining --------------------------------------------------------------- |
| 309 | |
| 310 | // Drained is one item's outcome, everything console needs and nothing it does |
| 311 | // not: the long-poll wake, the callback body, and the per-tick meter fold. |
| 312 | type Drained struct { |
| 313 | Key, Version, ETag string |
| 314 | Verdict Verdict |
| 315 | // Recorded is whether the verdict reached the object itself. When false |
| 316 | // the caller has written a result object instead, and the app must be told |
| 317 | // so — an infected object with Recorded false is NOT covered by the |
| 318 | // enforcement deny, because the deny matches on a tag that is not there. |
| 319 | Recorded bool |
| 320 | // Gone is an object that was not there to scan. It is an OUTCOME, not an |
| 321 | // error: the app asked about a key, and "it does not exist" is the honest |
| 322 | // answer to that. It rides in the same list so a waiter parked on the key |
| 323 | // is woken — before this existed the drain deleted the items silently and |
| 324 | // a long-poll sat out its whole window for an answer that had already |
| 325 | // been settled. |
| 326 | Gone bool |
| 327 | // Scanned is whether the engine actually ran, as against the verdict being |
| 328 | // read back from an existing tag. The meter counts on-demand work; an |
| 329 | // answer the sweep already paid for is not the tenant's to be charged. |
| 330 | Scanned bool |
| 331 | Size int64 |
| 332 | Engine string |
| 333 | Signature string |
| 334 | // CallbackPaths is EVERY path that asked about this key in this page, not |
| 335 | // one. The drain dedupes by object key so the engine reads an object once, |
| 336 | // and an earlier draft deleted the duplicate items along with their |
| 337 | // callbacks: two requests for the same key with different paths returned |
| 338 | // two 202s and delivered one callback. Deduplicate the WORK, never the |
| 339 | // promises made about it. |
| 340 | CallbackPaths []string |
| 341 | } |
| 342 | |
| 343 | // ErrQueueNotPermitted is a queue that must not be drained: the record is |
| 344 | // gone, the app is in the trash, or the consent chain no longer holds. The |
| 345 | // caller deletes the items rather than retrying them. |
| 346 | var ErrQueueNotPermitted = errors.New("objstore: this store may not be scanned") |
| 347 | |
| 348 | // ErrNoEngine is a drain asked to run with no scanner behind it. |
| 349 | var ErrNoEngine = errors.New("objstore: no scanning engine") |
| 350 | |
| 351 | // DrainApp scans up to cap of one app's queued items. |
| 352 | // |
| 353 | // It owns Observe and Release. Console cannot: Observe must happen BEFORE the |
| 354 | // listing, and that ordering is exactly what the generation counter protects — |
| 355 | // a caller that could get it wrong would be a caller that eventually does. |
| 356 | // |
| 357 | // UNDER HOST CLAIMS, EVERY WRITE AND EVERY BACKEND CALL IS FENCED BY A SHORT |
| 358 | // EXCLUSIVE APP MUTATION OF ITS OWN, and nothing spans more than one object. |
| 359 | // Each scan group runs under objstore.DrainGroup, covering exactly the Scan |
| 360 | // call, its result-object write or delete, and the deletion of that group's |
| 361 | // items. Housekeeping (expired or unreadable items, a dropped queue) gets an |
| 362 | // operation only when there is something to delete. Reads — the listing, the |
| 363 | // consent chain, the items — stay outside. |
| 364 | // |
| 365 | // The shape matters more than it looks. One operation around the whole page |
| 366 | // made a failure on object 2 report object 1's finished, durable deletes as |
| 367 | // part of an uncertain effect, and held the app until an operator reconciled |
| 368 | // it; a scan group that completes is now complete on its own. A failure |
| 369 | // before any effect (a refused begin, a store read) settles not-applied and |
| 370 | // the app reopens; a Scan that fails after the boundary marked it issued |
| 371 | // settles uncertain and holds the app, which is the boundary's own ruling |
| 372 | // about a backend call whose outcome is unknown, and is not softened here. |
| 373 | // Known outcomes (object gone, consent withdrawn or the deployment disabled |
| 374 | // after the engine answered) are results of a completed effect, never the |
| 375 | // operation's error. |
| 376 | func DrainApp(ctx context.Context, s release.Store, sc Scanner, idx *QueueIndex, |
| 377 | acct, app string, cap int, now time.Time) (done []Drained, listedEmpty bool, err error) { |
| 378 | |
| 379 | // In words rather than as a nil dereference, the way RunSweep refuses |
| 380 | // (sweep.go). The console gates its goroutine on a wired engine, but a |
| 381 | // package that dereferences whatever it is handed is one call site away |
| 382 | // from taking the control plane down. |
| 383 | if sc == nil { |
| 384 | return nil, false, ErrNoEngine |
| 385 | } |
| 386 | ref := AppRef{Account: acct, App: app} |
| 387 | observed := idx.Observe(ref) |
| 388 | |
| 389 | keys, err := listQueue(ctx, s, acct, app, cap) |
| 390 | if err != nil { |
| 391 | return nil, false, err |
| 392 | } |
| 393 | listedEmpty = len(keys) == 0 |
| 394 | defer func() { |
| 395 | if err == nil { |
| 396 | idx.Release(ref, observed, listedEmpty) |
| 397 | } |
| 398 | }() |
| 399 | if listedEmpty { |
| 400 | return nil, true, nil |
| 401 | } |
| 402 | |
| 403 | // The consent chain, re-checked here and not only at the door: the record |
| 404 | // can change between enqueue and drain, and THIS is where bytes are read. |
| 405 | rec, refusal, err := QueuePermitted(ctx, s, acct, app) |
| 406 | if err != nil { |
| 407 | return nil, false, err |
| 408 | } |
| 409 | // The reversible refusal, kept apart from the permanent ones BY NAME now |
| 410 | // rather than by which return slot it arrived in. It used to come back as |
| 411 | // an error, so the `err != nil` line above caught it and the deletion |
| 412 | // below never saw it — correct, and correct by accident: any later |
| 413 | // refusal that returned an error would have been silently spared, and any |
| 414 | // that did not would have been silently deleted on. |
| 415 | if refusal == RefusalDisabled { |
| 416 | return nil, false, ErrScanningDisabled |
| 417 | } |
| 418 | if refusal.Permanent() { |
| 419 | // Not a retry — none of these can change back in a way that makes the |
| 420 | // items useful. Drop them and say so. |
| 421 | if derr := dropQueue(ctx, s, acct, app, keys); derr != nil { |
| 422 | return nil, false, derr |
| 423 | } |
| 424 | return nil, false, ErrQueueNotPermitted |
| 425 | } |
| 426 | |
| 427 | engine, signature, err := sc.Provenance(ctx) |
| 428 | if err != nil { |
| 429 | return nil, false, fmt.Errorf("%w (%s/%s): %v", ErrEngineUnreachable, acct, app, err) |
| 430 | } |
| 431 | if rec.Scanning.Enabled.Engine != engine { |
| 432 | return nil, false, fmt.Errorf("%w (%s/%s): record says %q, configured engine is %q", |
| 433 | ErrEngineDrift, acct, app, rec.Scanning.Enabled.Engine, engine) |
| 434 | } |
| 435 | |
| 436 | // The page-level bucket is only used to fail fast; the loop below re-reads |
| 437 | // the record and uses THAT bucket for every object. |
| 438 | _ = rec.Provisioned.Bucket |
| 439 | groups, stale := groupQueue(ctx, s, keys, now) |
| 440 | if len(stale) > 0 { |
| 441 | // Collected before any scan, in one operation, so a failure here |
| 442 | // abandons the page before a verdict exists to lose. |
| 443 | if err := release.WithAppMutation(ctx, s, acct, app, "objstore.DrainCollect", func(ctx context.Context) error { |
| 444 | return deleteQueueItems(ctx, s, stale) |
| 445 | }); err != nil { |
| 446 | return nil, false, err |
| 447 | } |
| 448 | } |
| 449 | for i, g := range groups { |
| 450 | // CONSENT IS RE-READ FOR EVERY OBJECT, not once for the page. |
| 451 | // |
| 452 | // An external security review found this: the check above happens |
| 453 | // once and the loop then reads up to fifty objects' bytes. A member |
| 454 | // who re-attests their store as client-encrypted, an operator who |
| 455 | // revokes the grant or disables the deployment, or a TrashApp landing |
| 456 | // mid-page, would all be honoured only from the NEXT tick — and every |
| 457 | // object still in this page would be fetched and scanned under a |
| 458 | // consent that had already been withdrawn. |
| 459 | // |
| 460 | // The cost is three small metadata reads per object against a full |
| 461 | // object body read and an AV scan, which is the right trade for the |
| 462 | // one property this whole feature rests on. A pre-call check still |
| 463 | // leaves an irreducible window between the check and the GET — that |
| 464 | // race cannot be closed without a lease the object store does not |
| 465 | // offer — but it is now one object wide instead of one page wide. |
| 466 | rec, refusal, cerr := QueuePermitted(ctx, s, acct, app) |
| 467 | if cerr != nil { |
| 468 | // A read error is not a refusal: keep the rest of the page and |
| 469 | // let the next tick have it. |
| 470 | return done, false, cerr |
| 471 | } |
| 472 | if refusal == RefusalDisabled { |
| 473 | return done, false, ErrScanningDisabled |
| 474 | } |
| 475 | if refusal.Permanent() { |
| 476 | // Everything not yet handled goes, on the same rule the |
| 477 | // page-level check applies — including this group, whose object |
| 478 | // has not been read. |
| 479 | if delErr := dropRemainingQueue(ctx, s, acct, app, groups[i:]); delErr != nil { |
| 480 | return done, false, delErr |
| 481 | } |
| 482 | return done, false, ErrQueueNotPermitted |
| 483 | } |
| 484 | // Engine drift too, for the same reason: the record can be moved to a |
| 485 | // different engine between one object and the next, and a verdict |
| 486 | // recorded under the wrong provenance is worse than no verdict. |
| 487 | if rec.Scanning.Enabled.Engine != engine { |
| 488 | return done, false, fmt.Errorf("%w (%s/%s): record says %q, configured engine is %q", |
| 489 | ErrEngineDrift, acct, app, rec.Scanning.Enabled.Engine, engine) |
| 490 | } |
| 491 | // ONE OPERATION PER GROUP. The known outcomes drainOne reports as |
| 492 | // errors — the object is gone, consent went away or the deployment |
| 493 | // was disabled after the engine answered — are results of a completed |
| 494 | // effect, so the group function returns nil for them and hands them |
| 495 | // out through outcome; the caller acts on them outside the operation. |
| 496 | // Anything else drainOne returns is a real failure, and the boundary |
| 497 | // settles it on its own evidence: not-applied before any effect, held |
| 498 | // after one. |
| 499 | var out Drained |
| 500 | var outcome error |
| 501 | err := release.WithAppMutation(ctx, s, acct, app, "objstore.DrainGroup", func(ctx context.Context) error { |
| 502 | // Acquiring the permit can wait behind a revoke or reprovision. |
| 503 | fresh, refusal, err := QueuePermitted(ctx, s, acct, app) |
| 504 | if err != nil { |
| 505 | return err |
| 506 | } |
| 507 | if refusal == RefusalDisabled { |
| 508 | outcome = ErrScanningDisabled |
| 509 | return nil |
| 510 | } |
| 511 | if refusal != RefusalNone { |
| 512 | outcome = ErrQueueNotPermitted |
| 513 | return nil |
| 514 | } |
| 515 | if fresh.Scanning.Enabled.Engine != engine { |
| 516 | return fmt.Errorf("%w (%s/%s): record says %q, configured engine is %q", |
| 517 | ErrEngineDrift, acct, app, fresh.Scanning.Enabled.Engine, engine) |
| 518 | } |
| 519 | o, derr := drainOne(ctx, s, sc, acct, app, fresh.Provisioned.Bucket, engine, signature, g) |
| 520 | switch { |
| 521 | case errors.Is(derr, ErrObjectGone): |
| 522 | outcome = derr |
| 523 | return deleteQueueItems(ctx, s, g.queueKeys) |
| 524 | case errors.Is(derr, ErrQueueNotPermitted), errors.Is(derr, ErrScanningDisabled): |
| 525 | // The items are the caller's to keep or drop. |
| 526 | outcome = derr |
| 527 | return nil |
| 528 | case derr != nil: |
| 529 | // An engine failure is about the ENGINE, not this item. Keep |
| 530 | // the item and the index entry, and let the next tick retry. |
| 531 | return derr |
| 532 | } |
| 533 | out = o |
| 534 | return deleteQueueItems(ctx, s, g.queueKeys) |
| 535 | }) |
| 536 | if err != nil { |
| 537 | return done, false, err |
| 538 | } |
| 539 | switch { |
| 540 | case errors.Is(outcome, ErrObjectGone): |
| 541 | done = append(done, Drained{Key: g.key, Gone: true, CallbackPaths: g.callbackPaths}) |
| 542 | continue |
| 543 | case errors.Is(outcome, ErrQueueNotPermitted): |
| 544 | // Consent went away between this group's check and its write. |
| 545 | // Same conclusion as the check above, so the error keeps meaning |
| 546 | // one thing to the caller: the queue was dropped, not deferred. |
| 547 | if delErr := dropRemainingQueue(ctx, s, acct, app, groups[i:]); delErr != nil { |
| 548 | return done, false, delErr |
| 549 | } |
| 550 | return done, false, outcome |
| 551 | case errors.Is(outcome, ErrScanningDisabled): |
| 552 | return done, false, outcome |
| 553 | } |
| 554 | done = append(done, out) |
| 555 | } |
| 556 | // A page that was fully consumed may still have more behind it. |
| 557 | listedEmpty = false |
| 558 | return done, listedEmpty, nil |
| 559 | } |
| 560 | |
| 561 | // dropRemainingQueue deletes the items of every group not yet handled, and the |
| 562 | // app's fallback results with them. |
| 563 | // |
| 564 | // One function because two places reach the same conclusion — the per-object |
| 565 | // consent check, and a refusal observed inside drainOne after the scan — and |
| 566 | // the caller must not be able to tell them apart. ErrQueueNotPermitted means |
| 567 | // "the queue was dropped" wherever it comes from. |
| 568 | func dropRemainingQueue(ctx context.Context, s release.Store, acct, app string, left []*scanGroup) error { |
| 569 | var keys []string |
| 570 | for _, g := range left { |
| 571 | keys = append(keys, g.queueKeys...) |
| 572 | } |
| 573 | return dropQueue(ctx, s, acct, app, keys) |
| 574 | } |
| 575 | |
| 576 | // dropQueue deletes the named items and the app's fallback results in ONE |
| 577 | // operation, objstore.DropQueue. The results go with the items: a result |
| 578 | // object is a verdict no door may answer from once the app may not be |
| 579 | // scanned, and it is the worse orphan of the two — a stranded queue item is |
| 580 | // drained next tick, while a stranded result is found by nothing, blocks the |
| 581 | // app-name claim gate, and resurrects a deleted app on the next boot. The raw |
| 582 | // results delete is called so that one begin covers both. |
| 583 | func dropQueue(ctx context.Context, s release.Store, acct, app string, keys []string) error { |
| 584 | return release.WithAppMutation(ctx, s, acct, app, "objstore.DropQueue", func(ctx context.Context) error { |
| 585 | if err := deleteQueueItems(ctx, s, keys); err != nil { |
| 586 | return err |
| 587 | } |
| 588 | return claimRawDeleteScanResults(ctx, s, acct, app) |
| 589 | }) |
| 590 | } |
| 591 | |
| 592 | // scanGroup is every queued request in this page that asks about ONE object |
| 593 | // key: the object is read once, and every request that asked gets its answer. |
| 594 | type scanGroup struct { |
| 595 | key string |
| 596 | // queueKeys is every item object standing for this key, all deleted |
| 597 | // together once the group has an outcome. |
| 598 | queueKeys []string |
| 599 | // callbackPaths is one entry per DISTINCT path that asked. Empty when |
| 600 | // every request was long-poll only. |
| 601 | callbackPaths []string |
| 602 | } |
| 603 | |
| 604 | // groupQueue reads a listed page into scan groups, naming what can never be |
| 605 | // acted on (stale: the caller deletes it) and KEEPING what a later tick might |
| 606 | // still read. |
| 607 | // |
| 608 | // The distinction is §7.5's, and getting it wrong loses a tenant's request for |
| 609 | // good: deletion is driven by ABSENCE and by content that no tick will ever |
| 610 | // parse, never by a store error. An earlier draft deleted on any read failure, |
| 611 | // so one transient 500 on a GET discarded an accepted request silently. |
| 612 | // |
| 613 | // It reads and never writes, so it runs outside any operation. |
| 614 | func groupQueue(ctx context.Context, s release.Store, keys []string, now time.Time) (groups []*scanGroup, stale []string) { |
| 615 | byKey := map[string]*scanGroup{} |
| 616 | for _, qk := range keys { |
| 617 | item, err := readQueueItem(ctx, s, qk) |
| 618 | switch { |
| 619 | case errors.Is(err, os.ErrNotExist): |
| 620 | continue // already gone; nothing to delete and nothing to do |
| 621 | case errors.Is(err, errQueueItemUnreadable): |
| 622 | // Unparseable, or a schema this build does not know. No tick will |
| 623 | // ever read it, and leaving it would hold the app's index entry |
| 624 | // open forever. |
| 625 | stale = append(stale, qk) |
| 626 | continue |
| 627 | case err != nil: |
| 628 | // A STORE error. Reversible: keep the item, keep the entry, and |
| 629 | // let the next tick have it. The rest of the page still drains — |
| 630 | // one bad round trip is not a reason to stall an app's queue. |
| 631 | log.Printf("objstore: scan queue: read %s: %v", qk, err) |
| 632 | continue |
| 633 | } |
| 634 | if now.Sub(item.At) > QueueTTL { |
| 635 | stale = append(stale, qk) |
| 636 | continue |
| 637 | } |
| 638 | g := byKey[item.Key] |
| 639 | if g == nil { |
| 640 | g = &scanGroup{key: item.Key} |
| 641 | byKey[item.Key] = g |
| 642 | groups = append(groups, g) |
| 643 | } |
| 644 | g.queueKeys = append(g.queueKeys, qk) |
| 645 | if item.CallbackPath != "" && !containsString(g.callbackPaths, item.CallbackPath) { |
| 646 | g.callbackPaths = append(g.callbackPaths, item.CallbackPath) |
| 647 | } |
| 648 | } |
| 649 | return groups, stale |
| 650 | } |
| 651 | |
| 652 | func containsString(list []string, want string) bool { |
| 653 | for _, s := range list { |
| 654 | if s == want { |
| 655 | return true |
| 656 | } |
| 657 | } |
| 658 | return false |
| 659 | } |
| 660 | |
| 661 | func deleteQueueItems(ctx context.Context, s release.Store, keys []string) error { |
| 662 | for _, k := range keys { |
| 663 | if err := s.Delete(ctx, k); err != nil && !errors.Is(err, os.ErrNotExist) { |
| 664 | return err |
| 665 | } |
| 666 | } |
| 667 | return nil |
| 668 | } |
| 669 | |
| 670 | // drainOne scans a single object and records the outcome wherever it can go. |
| 671 | func drainOne(ctx context.Context, s release.Store, sc Scanner, acct, app, bucket, engine, signature string, |
| 672 | g *scanGroup) (Drained, error) { |
| 673 | |
| 674 | verdict, obj, err := sc.Status(ctx, bucket, g.key) |
| 675 | if err != nil { |
| 676 | return Drained{}, err |
| 677 | } |
| 678 | // PROVENANCE IS NOT FILLED IN YET, and that is the whole of the fix below. |
| 679 | // engine and signature describe the engine THIS TICK is configured with; |
| 680 | // whether they describe the thing that judged this object is not known |
| 681 | // until we know whether the engine ran. |
| 682 | out := Drained{ |
| 683 | Key: g.key, Version: obj.Version(), ETag: obj.ETag, Size: obj.Size, |
| 684 | CallbackPaths: g.callbackPaths, |
| 685 | } |
| 686 | |
| 687 | // The drain applies the re-issue rule ITSELF: denied and failed are |
| 688 | // transient, and answering an on-demand request from such a tag forever |
| 689 | // would make this path strictly less able than the hourly sweep it exists |
| 690 | // to improve on. |
| 691 | needsScan := verdict == VerdictUnknown || verdict == VerdictFailed || verdict == VerdictDenied |
| 692 | if !needsScan { |
| 693 | // The object carries this verdict, at this version, right now — but |
| 694 | // NOTHING HERE KNOWS WHAT PRODUCED IT. Engine and signature stay |
| 695 | // empty, and the reason is worth stating because filling them in is |
| 696 | // the obvious-looking thing to do. |
| 697 | // |
| 698 | // The tag may have been written by the hourly sweep last week under a |
| 699 | // different signature database, and this deployment's engine may have |
| 700 | // been replaced since. Stamping today's provenance onto it produces a |
| 701 | // SIGNED callback body naming a signature database that never judged |
| 702 | // this object — the status door already refuses to make exactly that |
| 703 | // guess for a tag verdict (apiscanstatus.go), so the two surfaces |
| 704 | // would disagree and the one lying would be the audit record. |
| 705 | // |
| 706 | // A verdict with no provenance is honest and useful. A verdict with |
| 707 | // invented provenance is the failure this whole feature exists to |
| 708 | // avoid. The tag does carry its own `at`, which readSignedTag parses |
| 709 | // and Status discards; widening that seam is how these fields get |
| 710 | // filled in truthfully, and it is a separate change. |
| 711 | out.Verdict, out.Recorded = verdict, true |
| 712 | // And any fallback for this key is now superseded — §5.2.1 says the |
| 713 | // drain drops it as soon as a real verdict is persisted, and the |
| 714 | // sweep can be what persisted it. Without this the result object |
| 715 | // survives until the 24h age sweep even though the object it stands |
| 716 | // in for has carried a real answer since. Delete is idempotent, so |
| 717 | // the common case (no fallback) costs one cheap call. |
| 718 | if err := DeleteScanResult(ctx, s, acct, app, g.key); err != nil { |
| 719 | return Drained{}, err |
| 720 | } |
| 721 | return out, nil |
| 722 | } |
| 723 | if obj == (ScanObject{}) { |
| 724 | // HeadObject itself was denied, so there is no version and no size to |
| 725 | // scan with — and NOTHING WAS RECORDED. Saying recorded here would put |
| 726 | // "denied" in a callback beside a claim that the object carries it, |
| 727 | // when the console never even saw a tag; the enforcement deny matches |
| 728 | // on a tag that is not there, and the app would be told a durable |
| 729 | // answer exists that no door can produce. |
| 730 | out.Verdict, out.Recorded = verdict, false |
| 731 | return out, nil |
| 732 | } |
| 733 | |
| 734 | v, recorded, err := sc.Scan(ctx, bucket, g.key, obj.Version(), obj.Size) |
| 735 | if err != nil { |
| 736 | return Drained{}, err |
| 737 | } |
| 738 | // The engine ran, here, now — so this is the one place the configured |
| 739 | // engine and signature database really do describe what judged the bytes. |
| 740 | out.Verdict, out.Recorded = v, recorded |
| 741 | // Scanned is whether the bytes REACHED the engine, which the call alone |
| 742 | // does not say: skipped is an object over the size ceiling that was never |
| 743 | // read, and denied is a GET the console was refused. Counting either as |
| 744 | // scanned would meter a tenant for a full object nothing looked at. Clean, |
| 745 | // infected and failed are the answers an engine gives after reading. |
| 746 | out.Scanned = v == VerdictClean || v == VerdictInfected || v == VerdictFailed |
| 747 | out.Engine, out.Signature = engine, signature |
| 748 | |
| 749 | if recorded { |
| 750 | // A real verdict is on the object now, so any fallback must go. |
| 751 | if err := DeleteScanResult(ctx, s, acct, app, g.key); err != nil { |
| 752 | return Drained{}, err |
| 753 | } |
| 754 | return out, nil |
| 755 | } |
| 756 | |
| 757 | // The verdict could not be written to the object. Re-read the record |
| 758 | // first: the writeSidecar race applies, and a result object stranded under |
| 759 | // a purged prefix is found by nothing and resurrects the app. |
| 760 | // ANY refusal stops the write, permanent or not: the question here is not |
| 761 | // "will this item ever be useful" but "may this app be scanned right now", |
| 762 | // and a result object written after the answer turned to no is a verdict |
| 763 | // no door may answer from. |
| 764 | // |
| 765 | // AND IT IS RETURNED AS AN ERROR, which the first version got wrong. It |
| 766 | // returned (out, nil), so DrainApp read the group as a SUCCESS: it deleted |
| 767 | // the items, appended an outcome to done, and carried on scanning the rest |
| 768 | // of the page with positive knowledge that consent was gone. Three things |
| 769 | // followed, and the third is the worst — on a reversible `disabled` the |
| 770 | // items were deleted against §7.5, the remaining objects were read anyway, |
| 771 | // and `done` carried a verdict with Recorded false and NO result object |
| 772 | // behind it, so task 16 would deliver a callback saying "clean" about an |
| 773 | // object the status door will answer "pending" about forever. That is |
| 774 | // exactly the disagreement §5.2.1 introduced the result object to prevent. |
| 775 | if _, refusal, err := QueuePermitted(ctx, s, acct, app); err != nil { |
| 776 | return Drained{}, err |
| 777 | } else if refusal == RefusalDisabled { |
| 778 | return Drained{}, ErrScanningDisabled |
| 779 | } else if refusal != RefusalNone { |
| 780 | return Drained{}, ErrQueueNotPermitted |
| 781 | } |
| 782 | if err := WriteScanResult(ctx, s, acct, app, &ScanResult{ |
| 783 | Key: g.key, Verdict: string(v), Version: obj.Version(), |
| 784 | Engine: engine, Signature: signature, At: time.Now().UTC(), |
| 785 | }); err != nil { |
| 786 | return Drained{}, err |
| 787 | } |
| 788 | return out, nil |
| 789 | } |
| 790 | |
| 791 | // QueuePermitted is the consent chain, in one place so the door and the drain |
| 792 | // cannot answer it differently. |
| 793 | // |
| 794 | // EXPORTED because the door lives in another package. That is the whole |
| 795 | // reason, and it is worth stating: the alternative — console re-deriving the |
| 796 | // same six conditions from a record it fetched itself — is the exact shape |
| 797 | // §5.1.1 exists to forbid. One implementation, two callers, and the only |
| 798 | // difference between them is what they do with the answer. |
| 799 | func QueuePermitted(ctx context.Context, s release.Store, acct, app string) (*Record, QueueRefusal, error) { |
| 800 | rec, _, err := Get(ctx, s, acct, app) |
| 801 | if err != nil { |
| 802 | return nil, RefusalNone, err |
| 803 | } |
| 804 | if rec == nil || rec.Provisioned == nil { |
| 805 | return nil, RefusalNotGranted, nil |
| 806 | } |
| 807 | tomb, err := release.ReadAppTombstone(ctx, s, acct, app) |
| 808 | if err != nil { |
| 809 | return nil, RefusalNone, err |
| 810 | } |
| 811 | if tomb != nil { |
| 812 | return nil, RefusalTrashed, nil |
| 813 | } |
| 814 | switch { |
| 815 | case rec.Scanning == nil || rec.Scanning.Requested == nil: |
| 816 | return nil, RefusalNotRequested, nil |
| 817 | case rec.Scanning.Requested.Encrypted: |
| 818 | return nil, RefusalEncrypted, nil |
| 819 | case rec.Scanning.Enabled == nil: |
| 820 | return nil, RefusalNotEnabled, nil |
| 821 | } |
| 822 | on, err := release.AntivirusEnabled(ctx, s) |
| 823 | if err != nil { |
| 824 | return nil, RefusalNone, err |
| 825 | } |
| 826 | if !on { |
| 827 | // The record comes back because this refusal is REVERSIBLE and the |
| 828 | // caller may want to say something about the store it is still |
| 829 | // refusing to scan. Everything above returns nil: there is nothing |
| 830 | // there to talk about. |
| 831 | return rec, RefusalDisabled, nil |
| 832 | } |
| 833 | return rec, RefusalNone, nil |
| 834 | } |
| 835 | |
| 836 | // QueueRefusal is why the consent chain said no, and RefusalNone is the only |
| 837 | // value that means yes. |
| 838 | // |
| 839 | // It is a NAME rather than a bool because the enqueue door and the drain draw |
| 840 | // different conclusions from the same answer — the door turns it into an HTTP |
| 841 | // state (§5.1), the drain turns it into keep-or-delete (§7.5) — and a bool |
| 842 | // forces each of them to re-derive the reason from the record they were handed |
| 843 | // back. Two callers re-deriving the same fact from the same data is the exact |
| 844 | // shape in which they start disagreeing, and the whole point of §5.1.1 is that |
| 845 | // they must not. |
| 846 | // |
| 847 | // A read ERROR is not a refusal and never gets a name here. Nothing about |
| 848 | // consent was learned, and §7.5's "deletion is driven by absence, never by |
| 849 | // error" depends on those two staying separate: give a bad bucket moment a |
| 850 | // name and something will eventually delete on it. |
| 851 | type QueueRefusal string |
| 852 | |
| 853 | const ( |
| 854 | RefusalNone QueueRefusal = "" |
| 855 | RefusalNotGranted QueueRefusal = "not_granted" |
| 856 | RefusalTrashed QueueRefusal = "trashed" |
| 857 | RefusalNotRequested QueueRefusal = "not_requested" |
| 858 | RefusalEncrypted QueueRefusal = "encrypted" |
| 859 | RefusalNotEnabled QueueRefusal = "not_enabled" |
| 860 | RefusalDisabled QueueRefusal = "disabled" |
| 861 | ) |
| 862 | |
| 863 | // Permanent reports whether this refusal will still be true in a way that |
| 864 | // makes the queued item worthless — §7.5's delete-vs-keep split, in the one |
| 865 | // place both callers read it from. |
| 866 | // |
| 867 | // Everything upstream of the deployment switch is permanent: a store that was |
| 868 | // never granted, an app in the trash, scanning never requested, an attestation |
| 869 | // saying the bytes are ciphertext, an operator who never enabled it. None of |
| 870 | // those can change back in a way that makes THIS item useful — if the state is |
| 871 | // later repaired the sweep covers the object anyway, and the requester is long |
| 872 | // gone. |
| 873 | // |
| 874 | // The deployment switch is not. An operator runs `carlos system antivirus |
| 875 | // enable` and every item still in the queue is worth draining. A deployment |
| 876 | // disabled for longer than the TTL drops its queue, which is right: both doors |
| 877 | // answered `disabled` at the time. |
| 878 | // |
| 879 | // Written as an explicit list rather than `r != RefusalDisabled`, so that a |
| 880 | // refusal added later is permanent only because somebody said so. The default |
| 881 | // for a new name is "keep the item", which is the survivable mistake. |
| 882 | func (r QueueRefusal) Permanent() bool { |
| 883 | switch r { |
| 884 | case RefusalNotGranted, RefusalTrashed, RefusalNotRequested, |
| 885 | RefusalEncrypted, RefusalNotEnabled: |
| 886 | return true |
| 887 | } |
| 888 | return false |
| 889 | } |
| 890 | |
| 891 | func listQueue(ctx context.Context, s release.Store, acct, app string, cap int) ([]string, error) { |
| 892 | prefix := QueuePrefix(acct, app) |
| 893 | if pl, ok := s.(release.PageLister); ok { |
| 894 | return pl.ListPage(ctx, prefix, "", cap) |
| 895 | } |
| 896 | // A store that cannot page still works, just less cheaply — the same |
| 897 | // degradation storagePass accepts for SizeLister. |
| 898 | keys, err := s.List(ctx, prefix) |
| 899 | if err != nil { |
| 900 | return nil, err |
| 901 | } |
| 902 | sort.Strings(keys) |
| 903 | if cap > 0 && len(keys) > cap { |
| 904 | keys = keys[:cap] |
| 905 | } |
| 906 | return keys, nil |
| 907 | } |
| 908 | |
| 909 | // errQueueItemUnreadable marks an item whose BYTES are the problem — bad JSON, |
| 910 | // or a schema this build does not know. It is the one read failure that is |
| 911 | // permanent, and the drain deletes on it; every other read failure is the |
| 912 | // store having a bad moment and the item stays. Wrapped rather than compared |
| 913 | // so the caller can tell the two apart with errors.Is and nothing else. |
| 914 | var errQueueItemUnreadable = errors.New("objstore: unreadable queue item") |
| 915 | |
| 916 | func readQueueItem(ctx context.Context, s release.Store, key string) (*QueueItem, error) { |
| 917 | raw, _, err := s.Get(ctx, key) |
| 918 | if err != nil { |
| 919 | return nil, err |
| 920 | } |
| 921 | var item QueueItem |
| 922 | if err := json.Unmarshal(raw, &item); err != nil { |
| 923 | return nil, fmt.Errorf("%w %s: %v", errQueueItemUnreadable, key, err) |
| 924 | } |
| 925 | if item.Schema != QueueSchema { |
| 926 | return nil, fmt.Errorf("%w %s: schema %d is not %d", errQueueItemUnreadable, key, item.Schema, QueueSchema) |
| 927 | } |
| 928 | return &item, nil |
| 929 | } |
| 930 | func Enqueue(ctx context.Context, s release.Store, idx *QueueIndex, acct, app string, item *QueueItem, now time.Time) (claimResult0 string, claimResult1 error) { |
| 931 | claimResult1 = release.WithAppMutation(ctx, s, acct, app, "objstore.Enqueue", func(claimContext context.Context) error { |
| 932 | claimResult0, claimResult1 = claimRawEnqueue(claimContext, s, idx, acct, app, item, now) |
| 933 | return claimResult1 |
| 934 | }) |
| 935 | return |
| 936 | } |
| 937 | |
amadan