| 1 | // The scan result object: apps/<acct>/<app>/scan-results/<sha256(key)>.json. |
| 2 | // |
| 3 | // It exists for one narrow case, and the case is not hypothetical. tag() |
| 4 | // declines to write a verdict when the object already carries eight foreign |
| 5 | // tags (S3 caps an object at ten) or when the pre-write tag read is denied. |
| 6 | // Both refusals are deliberate — they stop one object stalling a whole sweep — |
| 7 | // and the sweep tolerates them because it re-lists every 24h. |
| 8 | // |
| 9 | // A queue item does not. The on-demand drain scans, gets a real verdict, and |
| 10 | // deletes the item; without somewhere else to put that verdict the status door |
| 11 | // would answer "pending" forever while the callback said "clean". The app's |
| 12 | // declared source of truth would never arrive, and the two answers it did get |
| 13 | // would disagree. |
| 14 | // |
| 15 | // KEYED BY sha256(key), NOT by a queue id: the door asks by key, and ULID-named |
| 16 | // queue items would mean listing the whole prefix on every poll. |
| 17 | // |
| 18 | // AND VERSION-BOUND, which is the part that has been got wrong three times in |
| 19 | // this design's reviews. Without it: v1 scanned, tag not written because the |
| 20 | // cap fired, result says clean; the tenant overwrites with v2; the tag is still |
| 21 | // absent, the door falls through to the result object, and answers clean for |
| 22 | // bytes nobody has read. That is precisely the laundering the verdict tag's own |
| 23 | // version binding exists to stop, reintroduced by the mechanism meant to close |
| 24 | // a gap in it. |
| 25 | package objstore |
| 26 | |
| 27 | import ( |
| 28 | "context" |
| 29 | "crypto/sha256" |
| 30 | "encoding/hex" |
| 31 | "encoding/json" |
| 32 | "errors" |
| 33 | "fmt" |
| 34 | "os" |
| 35 | "time" |
| 36 | |
| 37 | "github.com/carlosframework/platform/internal/release" |
| 38 | ) |
| 39 | |
| 40 | // ScanResultSchema is pinned; an unknown schema refuses rather than guessing. |
| 41 | const ScanResultSchema = 1 |
| 42 | |
| 43 | // ScanResult is a verdict that was reached but could not be written onto the |
| 44 | // object it describes. |
| 45 | type ScanResult struct { |
| 46 | Schema int `json:"schema"` |
| 47 | Key string `json:"key"` |
| 48 | Verdict string `json:"verdict"` |
| 49 | // Version is the object version this verdict judged. A result whose |
| 50 | // version does not match the object's current one is not an answer about |
| 51 | // the object that is there now. |
| 52 | Version string `json:"version"` |
| 53 | Engine string `json:"engine,omitempty"` |
| 54 | Signature string `json:"signature,omitempty"` |
| 55 | At time.Time `json:"at"` |
| 56 | } |
| 57 | |
| 58 | // ScanResultKey is where one key's unrecorded verdict lives. |
| 59 | // |
| 60 | // Under the APP's prefix, so PurgeApp removes it with everything else and the |
| 61 | // name-claim gate can see it. A top-level results prefix would survive app |
| 62 | // deletion invisibly, which is the same argument that keeps the queue here. |
| 63 | func ScanResultKey(acct, app, objectKey string) string { |
| 64 | sum := sha256.Sum256([]byte(objectKey)) |
| 65 | return release.AppPrefix(acct, app) + "scan-results/" + hex.EncodeToString(sum[:]) + ".json" |
| 66 | } |
| 67 | |
| 68 | // ScanResultsPrefix is the whole set, for the age sweep and for deletion when |
| 69 | // an app stops being scannable. |
| 70 | func ScanResultsPrefix(acct, app string) string { |
| 71 | return release.AppPrefix(acct, app) + "scan-results/" |
| 72 | } |
| 73 | |
| 74 | // WriteScanResult records a verdict the object itself could not carry. |
| 75 | // |
| 76 | // READ-THEN-CAS, not create-only. release.Store.Put has exactly two modes, and |
| 77 | // the second write for a key is a REPLACE: v1 unrecordable, tenant uploads v2, |
| 78 | // v2 also unrecordable. A create-only write there returns ErrCASConflict and |
| 79 | // silently drops the v2 verdict, leaving the door on "pending" until a sweep |
| 80 | // happens to tag it. |
| 81 | // |
| 82 | // The caller must have re-read store.json immediately before calling this. The |
| 83 | // race is the one writeSidecar documents — read the record, scan for seconds, |
| 84 | // a teardown and PurgeApp land, and a write under the purged prefix succeeds — |
| 85 | // but the consequence here is worse than a stranded sidecar. A stranded queue |
| 86 | // item is drained on the next tick because its index entry survives; a stranded |
| 87 | // result object is found by nothing, blocks the app-name claim gate, and |
| 88 | // because materialize counts any key under an app prefix and fails toward |
| 89 | // keeping the row, it RESURRECTS the deleted app on the next console boot. |
| 90 | func claimRawWriteScanResult(ctx context.Context, s release.Store, acct, app string, res *ScanResult) error { |
| 91 | res.Schema = ScanResultSchema |
| 92 | body, err := json.MarshalIndent(res, "", " ") |
| 93 | if err != nil { |
| 94 | return err |
| 95 | } |
| 96 | body = append(body, '\n') |
| 97 | key := ScanResultKey(acct, app, res.Key) |
| 98 | |
| 99 | for attempt := 0; attempt < 2; attempt++ { |
| 100 | _, etag, err := s.Get(ctx, key) |
| 101 | if err != nil && !errors.Is(err, os.ErrNotExist) { |
| 102 | return fmt.Errorf("objstore: read scan result %s/%s: %w", acct, app, err) |
| 103 | } |
| 104 | if errors.Is(err, os.ErrNotExist) { |
| 105 | etag = "" // create-only |
| 106 | } |
| 107 | if _, err := s.Put(ctx, key, body, etag); err != nil { |
| 108 | if errors.Is(err, release.ErrCASConflict) { |
| 109 | continue // somebody moved under us; re-read and try once more |
| 110 | } |
| 111 | return fmt.Errorf("objstore: write scan result %s/%s: %w", acct, app, err) |
| 112 | } |
| 113 | return nil |
| 114 | } |
| 115 | return fmt.Errorf("objstore: write scan result %s/%s: lost the race twice", acct, app) |
| 116 | } |
| 117 | |
| 118 | // ReadScanResult answers only about currentVersion. A result recorded for |
| 119 | // different bytes is ABSENT, not stale-but-usable: the comparison lives here so |
| 120 | // no door and no app ever handles the version token, and so nobody can forget |
| 121 | // to make it. |
| 122 | func ReadScanResult(ctx context.Context, s release.Store, acct, app, objectKey, currentVersion string) (*ScanResult, error) { |
| 123 | raw, _, err := s.Get(ctx, ScanResultKey(acct, app, objectKey)) |
| 124 | if err != nil { |
| 125 | if errors.Is(err, os.ErrNotExist) { |
| 126 | return nil, nil |
| 127 | } |
| 128 | return nil, err |
| 129 | } |
| 130 | var res ScanResult |
| 131 | if err := json.Unmarshal(raw, &res); err != nil { |
| 132 | return nil, fmt.Errorf("objstore: scan result %s/%s: %w", acct, app, err) |
| 133 | } |
| 134 | if res.Schema != ScanResultSchema { |
| 135 | return nil, fmt.Errorf("objstore: scan result %s/%s: schema %d is not %d", |
| 136 | acct, app, res.Schema, ScanResultSchema) |
| 137 | } |
| 138 | if res.Version != currentVersion { |
| 139 | // Recorded about bytes that are no longer there. |
| 140 | return nil, nil |
| 141 | } |
| 142 | return &res, nil |
| 143 | } |
| 144 | |
| 145 | // DeleteScanResult drops one key's result — called when a later scan DOES |
| 146 | // persist a verdict onto the object, so the fallback cannot outlive the real |
| 147 | // answer. |
| 148 | func claimRawDeleteScanResult(ctx context.Context, s release.Store, acct, app, objectKey string) error { |
| 149 | err := s.Delete(ctx, ScanResultKey(acct, app, objectKey)) |
| 150 | if err != nil && !errors.Is(err, os.ErrNotExist) { |
| 151 | return err |
| 152 | } |
| 153 | return nil |
| 154 | } |
| 155 | |
| 156 | // DeleteScanResults drops an app's whole fallback set — called when the app |
| 157 | // may no longer be scanned at all. A result object outliving consent is a |
| 158 | // verdict no door is allowed to answer from, and it is the orphan that blocks |
| 159 | // the app-name claim gate. |
| 160 | func claimRawDeleteScanResults(ctx context.Context, s release.Store, acct, app string) error { |
| 161 | keys, err := s.List(ctx, ScanResultsPrefix(acct, app)) |
| 162 | if err != nil { |
| 163 | if errors.Is(err, os.ErrNotExist) { |
| 164 | return nil |
| 165 | } |
| 166 | return err |
| 167 | } |
| 168 | for _, k := range keys { |
| 169 | if err := s.Delete(ctx, k); err != nil && !errors.Is(err, os.ErrNotExist) { |
| 170 | return err |
| 171 | } |
| 172 | } |
| 173 | return nil |
| 174 | } |
| 175 | |
| 176 | // CollectScanResults deletes an app's results older than before, and reports |
| 177 | // how many it removed. |
| 178 | // |
| 179 | // WHY AGE AT ALL. A result exists because a verdict could not be written onto |
| 180 | // the object, and the conditions that cause that — a ten-tag object, a denied |
| 181 | // tagging call — do not clear themselves. Nothing else deletes these: the |
| 182 | // drain only removes one when a LATER scan records a real verdict on the same |
| 183 | // key, which for a permanently untaggable object never happens. Without a |
| 184 | // collector an app accumulates one small object per such key forever. |
| 185 | // |
| 186 | // The TTL is the queue's, for the same reason the queue's is what it is: after |
| 187 | // a sweep cycle the ordinary walk has had its chance at the object, and the |
| 188 | // door's answer falls back to reading the tag itself. |
| 189 | // |
| 190 | // ModLister is optional (release/store.go:161), exactly as storagePass treats |
| 191 | // SizeLister: a store that cannot answer is skipped, not an error — the caller |
| 192 | // logs once rather than failing a tick over a capability a test fake lacks. |
| 193 | // |
| 194 | // The listing and the age filter run outside any operation; the app mutation |
| 195 | // objstore.CollectScanResults is entered only when at least one result is |
| 196 | // past its TTL. This runs once per visited app per tick and once per app at |
| 197 | // boot, and an exclusive operation for an app with nothing to delete would |
| 198 | // be most of them. |
| 199 | func CollectScanResults(ctx context.Context, s release.Store, acct, app string, before time.Time) (int, error) { |
| 200 | expired, err := expiredScanResults(ctx, s, acct, app, before) |
| 201 | if err != nil || len(expired) == 0 { |
| 202 | return 0, err |
| 203 | } |
| 204 | revisions := make(map[string]string, len(expired)) |
| 205 | for _, key := range expired { |
| 206 | _, etag, err := s.Get(ctx, key) |
| 207 | if errors.Is(err, os.ErrNotExist) { |
| 208 | continue |
| 209 | } |
| 210 | if err != nil { |
| 211 | return 0, err |
| 212 | } |
| 213 | revisions[key] = etag |
| 214 | } |
| 215 | if len(revisions) == 0 { |
| 216 | return 0, nil |
| 217 | } |
| 218 | n := 0 |
| 219 | err = release.WithAppMutation(ctx, s, acct, app, "objstore.CollectScanResults", func(ctx context.Context) error { |
| 220 | // Recheck age after acquiring authority, and delete only the revision |
| 221 | // observed earlier. A replacement also survives with claims disabled. |
| 222 | stillExpired, err := expiredScanResults(ctx, s, acct, app, before) |
| 223 | if err != nil { |
| 224 | return err |
| 225 | } |
| 226 | for _, k := range stillExpired { |
| 227 | etag, observed := revisions[k] |
| 228 | if !observed { |
| 229 | continue |
| 230 | } |
| 231 | err := release.DeleteIfMatch(ctx, s, k, etag) |
| 232 | if errors.Is(err, release.ErrCASConflict) || errors.Is(err, os.ErrNotExist) { |
| 233 | continue |
| 234 | } |
| 235 | if err != nil { |
| 236 | return err |
| 237 | } |
| 238 | n++ |
| 239 | } |
| 240 | return nil |
| 241 | }) |
| 242 | return n, err |
| 243 | } |
| 244 | |
| 245 | // expiredScanResults names an app's results modified before the cut-off. |
| 246 | func expiredScanResults(ctx context.Context, s release.Store, acct, app string, before time.Time) ([]string, error) { |
| 247 | ml, ok := s.(release.ModLister) |
| 248 | if !ok { |
| 249 | return nil, ErrNoModLister |
| 250 | } |
| 251 | entries, err := ml.ListMod(ctx, ScanResultsPrefix(acct, app)) |
| 252 | if err != nil { |
| 253 | if errors.Is(err, os.ErrNotExist) { |
| 254 | return nil, nil |
| 255 | } |
| 256 | return nil, err |
| 257 | } |
| 258 | var expired []string |
| 259 | for _, e := range entries { |
| 260 | if e.Mod.Before(before) { |
| 261 | expired = append(expired, e.Key) |
| 262 | } |
| 263 | } |
| 264 | return expired, nil |
| 265 | } |
| 266 | |
| 267 | // ErrNoModLister is a store that cannot rank by modification time. The caller |
| 268 | // skips the age sweep and says so once, rather than treating a missing |
| 269 | // capability as a failure. |
| 270 | var ErrNoModLister = errors.New("objstore: this store cannot list modification times") |
| 271 | |
| 272 | func DeleteScanResult(ctx context.Context, s release.Store, acct, app, objectKey string) (claimResult0 error) { |
| 273 | claimResult0 = release.WithAppMutation(ctx, s, acct, app, "objstore.DeleteScanResult", func(claimContext context.Context) error { |
| 274 | claimResult0 = claimRawDeleteScanResult(claimContext, s, acct, app, objectKey) |
| 275 | return claimResult0 |
| 276 | }) |
| 277 | return |
| 278 | } |
| 279 | func DeleteScanResults(ctx context.Context, s release.Store, acct, app string) (claimResult0 error) { |
| 280 | claimResult0 = release.WithAppMutation(ctx, s, acct, app, "objstore.DeleteScanResults", func(claimContext context.Context) error { |
| 281 | claimResult0 = claimRawDeleteScanResults(claimContext, s, acct, app) |
| 282 | return claimResult0 |
| 283 | }) |
| 284 | return |
| 285 | } |
| 286 | func WriteScanResult(ctx context.Context, s release.Store, acct, app string, res *ScanResult) (claimResult0 error) { |
| 287 | claimResult0 = release.WithAppMutation(ctx, s, acct, app, "objstore.WriteScanResult", func(claimContext context.Context) error { |
| 288 | claimResult0 = claimRawWriteScanResult(claimContext, s, acct, app, res) |
| 289 | return claimResult0 |
| 290 | }) |
| 291 | return |
| 292 | } |
| 293 | |
amadan