| 1 | package objstore |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "io" |
| 8 | "log" |
| 9 | "strings" |
| 10 | "time" |
| 11 | |
| 12 | awss3 "github.com/aws/aws-sdk-go-v2/service/s3" |
| 13 | s3types "github.com/aws/aws-sdk-go-v2/service/s3/types" |
| 14 | "github.com/aws/smithy-go" |
| 15 | |
| 16 | "github.com/carlosframework/platform/internal/clamd" |
| 17 | ) |
| 18 | |
| 19 | // scanS3API is the subset of *s3.Client the scanner calls, narrowed the same |
| 20 | // way s3API is so a test can stand a stub in with no network or credentials. |
| 21 | // |
| 22 | // It is deliberately separate from s3API: the provisioner creates and destroys |
| 23 | // buckets and never reads an object, the scanner reads objects and never |
| 24 | // touches a bucket. Keeping them apart means neither can quietly grow the |
| 25 | // other's powers. |
| 26 | type scanS3API interface { |
| 27 | GetObject(ctx context.Context, in *awss3.GetObjectInput, optFns ...func(*awss3.Options)) (*awss3.GetObjectOutput, error) |
| 28 | HeadObject(ctx context.Context, in *awss3.HeadObjectInput, optFns ...func(*awss3.Options)) (*awss3.HeadObjectOutput, error) |
| 29 | GetObjectTagging(ctx context.Context, in *awss3.GetObjectTaggingInput, optFns ...func(*awss3.Options)) (*awss3.GetObjectTaggingOutput, error) |
| 30 | PutObjectTagging(ctx context.Context, in *awss3.PutObjectTaggingInput, optFns ...func(*awss3.Options)) (*awss3.PutObjectTaggingOutput, error) |
| 31 | } |
| 32 | |
| 33 | // streamScanner is the engine seam: what clamavScanner needs from a clamd. |
| 34 | // *clamd.Client satisfies it. The wire protocol is tested in internal/clamd, |
| 35 | // so tests here fake this interface rather than restaging a socket. |
| 36 | type streamScanner interface { |
| 37 | ScanStream(ctx context.Context, r io.Reader) (clamd.Result, error) |
| 38 | Version(ctx context.Context) (string, error) |
| 39 | } |
| 40 | |
| 41 | // clamavScanner scans objects by streaming them past a local clamd and records |
| 42 | // the verdict as object tags. |
| 43 | // |
| 44 | // Nothing about the bucket changes: there is no per-bucket plan to enable, no |
| 45 | // quota to consume and no per-bucket cost. That is the whole reason ClamAV is |
| 46 | // the primary engine — the bought alternative caps at 25 protected buckets per |
| 47 | // account per Region and cannot be raised. |
| 48 | type clamavScanner struct { |
| 49 | s3 scanS3API |
| 50 | engine streamScanner |
| 51 | maxBytes int64 |
| 52 | // signer is what makes a tag a verdict. See scantag.go: the app can write |
| 53 | // these tags on some gateways whatever the IAM policy says, so authenticity |
| 54 | // and not write-permission is the property. Never nil in a running |
| 55 | // deployment — the console refuses to construct a scanner without one. |
| 56 | signer *TagSigner |
| 57 | } |
| 58 | |
| 59 | var _ Scanner = (*clamavScanner)(nil) |
| 60 | |
| 61 | // NewClamAVScanner builds a Scanner against S3 in region and a clamd on |
| 62 | // socket. maxBytes is the size ceiling above which an object is recorded as |
| 63 | // skipped without being fetched; it should mirror clamd's own StreamMaxLength. |
| 64 | // |
| 65 | // This does NOT ping clamd. An engine that is down at boot — mid signature |
| 66 | // update, say — must not take the console with it: object scanning is an |
| 67 | // opt-in feature of one subsystem, and failing the whole control plane closed |
| 68 | // because a scanner is restarting would be a far worse outage than the one it |
| 69 | // prevents. An unreachable engine surfaces through Provenance, which is what |
| 70 | // `carlos store scan status` reports, and fails the pass loudly per tick. |
| 71 | func NewClamAVScanner(ctx context.Context, region, endpoint, socket string, maxBytes int64, signer *TagSigner) (Scanner, error) { |
| 72 | if signer == nil { |
| 73 | // A scanner with no signer would write verdicts nothing can verify and |
| 74 | // read every forgery as genuine. Refuse at construction, where an |
| 75 | // operator sees it, rather than at the first tag. |
| 76 | return nil, ErrNoTagKey |
| 77 | } |
| 78 | s3c, err := S3ClientFor(ctx, region, endpoint) |
| 79 | if err != nil { |
| 80 | return nil, fmt.Errorf("objstore: clamav scanner: %w", err) |
| 81 | } |
| 82 | return &clamavScanner{ |
| 83 | s3: s3c, |
| 84 | engine: clamd.New(socket, clamd.Timeouts{}), |
| 85 | maxBytes: maxBytes, |
| 86 | signer: signer, |
| 87 | }, nil |
| 88 | } |
| 89 | |
| 90 | // EnsureEnabled is a no-op. There is nothing on the bucket to configure: the |
| 91 | // engine is reached from the control plane, not attached to the store. |
| 92 | func (c *clamavScanner) EnsureEnabled(ctx context.Context, bucket string) error { return nil } |
| 93 | |
| 94 | // EnsureDisabled is a no-op for the same reason. Nothing was attached, so |
| 95 | // nothing leaks when a bucket is destroyed. |
| 96 | func (c *clamavScanner) EnsureDisabled(ctx context.Context, bucket string) error { return nil } |
| 97 | |
| 98 | // Capacity reports no limit. -1 is "unbounded", distinct from 0, which would |
| 99 | // mean a real engine with nothing left. |
| 100 | func (c *clamavScanner) Capacity(ctx context.Context) (remaining, capacity int, err error) { |
| 101 | return -1, -1, nil |
| 102 | } |
| 103 | |
| 104 | // Provenance names the engine and its signature database — the two facts an |
| 105 | // audit answer needs beside a count. A scan is only as good as the signatures |
| 106 | // it ran against, so "clean" without a database version is not evidence. |
| 107 | func (c *clamavScanner) Provenance(ctx context.Context) (engine, version string, err error) { |
| 108 | v, err := c.engine.Version(ctx) |
| 109 | if err != nil { |
| 110 | return "clamav", "", err |
| 111 | } |
| 112 | return "clamav", v, nil |
| 113 | } |
| 114 | |
| 115 | // Enforcement is the tag the store user's policy denies on. |
| 116 | func (c *clamavScanner) Enforcement() DenyTag { |
| 117 | return DenyTag{Key: TagVerdict, Value: string(VerdictInfected)} |
| 118 | } |
| 119 | |
| 120 | // Verdict reports what the engine previously said about this exact version of |
| 121 | // the object. |
| 122 | // |
| 123 | // A verdict is bound to the object VERSION it judged — see ObjectVersion: the |
| 124 | // ETag and the last-modified time together. |
| 125 | // |
| 126 | // Without that binding a key that once held something clean is a laundering |
| 127 | // channel: upload clean, get tagged clean, overwrite with the payload, keep |
| 128 | // the tag. The ETag alone is not quite enough for it, either. For a |
| 129 | // single-part upload the ETag is the content MD5, and chosen-prefix MD5 |
| 130 | // collisions are practical on ordinary hardware — so an attacker who controls |
| 131 | // both uploads could produce a payload with the ETag of the clean object it |
| 132 | // replaces. Last-modified moves on every overwrite and is not theirs to |
| 133 | // choose. A tag whose recorded version does not match reads as unknown, and |
| 134 | // the sweep scans it again. |
| 135 | func (c *clamavScanner) Verdict(ctx context.Context, bucket, key, version string) (Verdict, error) { |
| 136 | verdict, taggedVersion, at, stale, err := c.readSignedTag(ctx, bucket, key) |
| 137 | if err != nil { |
| 138 | return VerdictUnknown, err |
| 139 | } |
| 140 | if verdict == VerdictDenied && taggedVersion == "" { |
| 141 | // The tag set itself was unreadable. A fact about the object, not a |
| 142 | // transport failure, and it must not fail the pass. |
| 143 | return VerdictDenied, nil |
| 144 | } |
| 145 | out := c.interpretForSweep(verdict, taggedVersion, version) |
| 146 | if stale && out == verdict && verdict != "" { |
| 147 | // A verdict issued under the previous key, still valid, still about |
| 148 | // these bytes. Re-sign it so the old key can eventually be dropped — |
| 149 | // WITHOUT rescanning, and WITHOUT moving carlos-scan-at, because a |
| 150 | // re-sign judges nothing and must not claim a fresh scan. |
| 151 | c.resign(ctx, bucket, key, taggedVersion, verdict, at) |
| 152 | } |
| 153 | return out, nil |
| 154 | } |
| 155 | |
| 156 | // resign rewrites the signature of an existing verdict under the current key. |
| 157 | // |
| 158 | // Best effort, deliberately. The verdict is already valid under the previous |
| 159 | // key, so the cost of failing is that it is re-signed on some later tick; |
| 160 | // returning an error here would abandon the whole page over a housekeeping |
| 161 | // write and stall every object behind this one. |
| 162 | func (c *clamavScanner) resign(ctx context.Context, bucket, key, version string, v Verdict, at string) { |
| 163 | if _, err := c.tagWith(ctx, bucket, key, version, v, at); err != nil { |
| 164 | log.Printf("objstore: could not re-sign %s/%s under the current tag key (it stays valid under the previous one): %v", bucket, key, err) |
| 165 | } |
| 166 | } |
| 167 | |
| 168 | // readTag is the one place tags are read, shared by Verdict and Status so the |
| 169 | // two cannot drift on what a tag SAYS while differing, deliberately, on what |
| 170 | // it MEANS. |
| 171 | // |
| 172 | // A denied read comes back as (VerdictDenied, "", nil): unreadable tags are a |
| 173 | // fact about the object. |
| 174 | func (c *clamavScanner) readTag(ctx context.Context, bucket, key string) (Verdict, string, error) { |
| 175 | v, ver, _, _, err := c.readSignedTag(ctx, bucket, key) |
| 176 | return v, ver, err |
| 177 | } |
| 178 | |
| 179 | // readSignedTag is readTag with the two extra facts the re-signing path needs: |
| 180 | // the recorded timestamp, and whether the signature that vouched for all of it |
| 181 | // was issued under the PREVIOUS key. |
| 182 | // |
| 183 | // THIS IS THE TRUST BOUNDARY. Everything above it — Verdict, Status, the |
| 184 | // sweep, the drain, the status door — is reading a claim that has already been |
| 185 | // checked against the deployment's key. An unsigned or badly signed tag does |
| 186 | // not come back as a suspicious verdict to be handled carefully somewhere |
| 187 | // else; it comes back as no verdict at all, which is the same answer as a bare |
| 188 | // object and produces the same behaviour: scan it. |
| 189 | func (c *clamavScanner) readSignedTag(ctx context.Context, bucket, key string) (verdict Verdict, version, at string, stale bool, err error) { |
| 190 | out, err := c.s3.GetObjectTagging(ctx, &awss3.GetObjectTaggingInput{ |
| 191 | Bucket: &bucket, |
| 192 | Key: &key, |
| 193 | }) |
| 194 | if err != nil { |
| 195 | if isGone(err) { |
| 196 | return VerdictUnknown, "", "", false, ErrObjectGone |
| 197 | } |
| 198 | if isDenied(err) { |
| 199 | return VerdictDenied, "", "", false, nil |
| 200 | } |
| 201 | return VerdictUnknown, "", "", false, fmt.Errorf("objstore: read tags %s/%s: %w", bucket, key, err) |
| 202 | } |
| 203 | |
| 204 | var raw, sig string |
| 205 | for _, t := range out.TagSet { |
| 206 | switch aws2s(t.Key) { |
| 207 | case TagVerdict: |
| 208 | raw = aws2s(t.Value) |
| 209 | case TagVersion: |
| 210 | version = aws2s(t.Value) |
| 211 | case TagAt: |
| 212 | at = aws2s(t.Value) |
| 213 | case TagSignature: |
| 214 | sig = aws2s(t.Value) |
| 215 | } |
| 216 | } |
| 217 | if raw == "" { |
| 218 | return "", "", "", false, nil |
| 219 | } |
| 220 | ok, stale := c.signer.Verify(scanFacts{ |
| 221 | bucket: bucket, key: key, version: version, verdict: raw, at: at, |
| 222 | }, sig) |
| 223 | if !ok { |
| 224 | // Somebody else wrote this. On several gateways an app CAN, whatever |
| 225 | // the IAM policy says, so this is the expected shape of an attack and |
| 226 | // not an anomaly to shout about. Report exactly what a bare object |
| 227 | // reports. |
| 228 | return "", "", "", false, nil |
| 229 | } |
| 230 | return Verdict(raw), version, at, stale, nil |
| 231 | } |
| 232 | |
| 233 | // interpretForSweep applies the SWEEP's reading of a tag: unknown when it does |
| 234 | // not match the current version, and unknown for denied and failed so they are |
| 235 | // re-issued rather than inherited. |
| 236 | func (c *clamavScanner) interpretForSweep(verdict Verdict, taggedVersion, version string) Verdict { |
| 237 | if verdict == "" || taggedVersion == "" { |
| 238 | return VerdictUnknown |
| 239 | } |
| 240 | // denied and failed are RE-ISSUED every sweep, never carried forward. |
| 241 | // |
| 242 | // Both are transient by nature — a permissions fix, a Glacier restore, an |
| 243 | // engine that was having a bad day — and the tag is an audit trace of what |
| 244 | // happened last time, not a verdict to inherit. Reading them back as |
| 245 | // answers would make them permanent: the object is not rewritten, so its |
| 246 | // version never changes, so it would never be retried and the repair would |
| 247 | // never take. An SSE-KMS denial is the case that bites, because the |
| 248 | // tagging call needs no KMS access and therefore succeeds. |
| 249 | switch verdict { |
| 250 | case VerdictDenied, VerdictFailed: |
| 251 | return VerdictUnknown |
| 252 | } |
| 253 | if taggedVersion != version { |
| 254 | // The bytes changed under the verdict. Not a failure — just unknown |
| 255 | // again, which is exactly what makes the next sweep re-scan it. |
| 256 | return VerdictUnknown |
| 257 | } |
| 258 | return verdict |
| 259 | } |
| 260 | |
| 261 | // Status answers about the CURRENT version of one object, finding that version |
| 262 | // itself, and reports what the tag SAYS rather than what the sweep should do |
| 263 | // about it. See the Scanner interface for why those are different questions. |
| 264 | func (c *clamavScanner) Status(ctx context.Context, bucket, key string) (Verdict, ScanObject, error) { |
| 265 | head, err := c.s3.HeadObject(ctx, &awss3.HeadObjectInput{Bucket: &bucket, Key: &key}) |
| 266 | if err != nil { |
| 267 | if isGone(err) { |
| 268 | return VerdictUnknown, ScanObject{}, ErrObjectGone |
| 269 | } |
| 270 | if isDenied(err) { |
| 271 | // No version and no size, so a caller must not go on to Scan. |
| 272 | return VerdictDenied, ScanObject{}, nil |
| 273 | } |
| 274 | return VerdictUnknown, ScanObject{}, fmt.Errorf("objstore: head %s/%s: %w", bucket, key, err) |
| 275 | } |
| 276 | obj := ScanObject{Key: key, ETag: normalizeETag(aws2s(head.ETag))} |
| 277 | if head.LastModified != nil { |
| 278 | obj.Modified = *head.LastModified |
| 279 | } |
| 280 | if head.ContentLength != nil { |
| 281 | obj.Size = *head.ContentLength |
| 282 | } |
| 283 | |
| 284 | verdict, taggedVersion, err := c.readTag(ctx, bucket, key) |
| 285 | if err != nil { |
| 286 | return VerdictUnknown, obj, err |
| 287 | } |
| 288 | if verdict == VerdictDenied && taggedVersion == "" { |
| 289 | return VerdictDenied, obj, nil |
| 290 | } |
| 291 | if verdict == "" || taggedVersion == "" || taggedVersion != obj.Version() { |
| 292 | // No tag, or a tag about bytes that have since changed. Either way |
| 293 | // nothing is known about what is there now. |
| 294 | return VerdictUnknown, obj, nil |
| 295 | } |
| 296 | // Reported as written: denied and failed included. The re-issue rule is |
| 297 | // the sweep's, and applying it here would make a failed scan read as |
| 298 | // pending forever. |
| 299 | return verdict, obj, nil |
| 300 | } |
| 301 | |
| 302 | // Scan streams one object past the engine and records the result. |
| 303 | func (c *clamavScanner) Scan(ctx context.Context, bucket, key, version string, size int64) (Verdict, bool, error) { |
| 304 | // The ceiling is checked BEFORE the fetch. The size is already on the |
| 305 | // listing page that got us here, so paying egress to pull bytes clamd has |
| 306 | // told us it will refuse is pure waste — and on a bucket full of large |
| 307 | // media that waste is the entire bill. |
| 308 | if c.maxBytes > 0 && size > c.maxBytes { |
| 309 | recorded, err := c.tag(ctx, bucket, key, version, VerdictSkipped) |
| 310 | if err != nil { |
| 311 | return VerdictUnknown, false, err |
| 312 | } |
| 313 | return VerdictSkipped, recorded, nil |
| 314 | } |
| 315 | |
| 316 | obj, err := c.s3.GetObject(ctx, &awss3.GetObjectInput{Bucket: &bucket, Key: &key}) |
| 317 | if err != nil { |
| 318 | // An object we are not allowed to read is DENIED — a permanent fact |
| 319 | // about this object that a permissions fix repairs, not a retry. |
| 320 | // |
| 321 | // This classification is load-bearing, not tidiness. Returning a plain |
| 322 | // error here fails the whole page, unwritten; the next tick re-lists |
| 323 | // the same page, meets the same object and fails identically, and the |
| 324 | // bucket's sweep never advances past it. One SSE-KMS object the |
| 325 | // console's key policy does not cover, or one object owned by another |
| 326 | // account in an adopted bucket, would stall that store's coverage |
| 327 | // forever while every surface reported a sweep merely "in progress". |
| 328 | if isDenied(err) { |
| 329 | // Tagging is best effort and its failure is deliberately ignored: |
| 330 | // if we may not read the object we may well not be able to tag it, |
| 331 | // and the verdict still stands for this sweep's count. The tag is |
| 332 | // only an audit trace here — Verdict re-issues denied rather than |
| 333 | // reading it back, so nothing depends on this landing. |
| 334 | _, _ = c.tag(ctx, bucket, key, version, VerdictDenied) |
| 335 | // recorded is false whether or not the tag landed: nothing may |
| 336 | // depend on a trace we could not verify. |
| 337 | return VerdictDenied, false, nil |
| 338 | } |
| 339 | // Deleted between the listing and the fetch. Routine on a bucket |
| 340 | // anyone is using, and not a failure of anything — the sweep drops it |
| 341 | // rather than counting an object that no longer exists. |
| 342 | if isGone(err) { |
| 343 | return VerdictUnknown, false, ErrObjectGone |
| 344 | } |
| 345 | return VerdictUnknown, false, fmt.Errorf("objstore: get %s/%s: %w", bucket, key, err) |
| 346 | } |
| 347 | defer obj.Body.Close() |
| 348 | |
| 349 | // Straight from S3 into the socket. The object is never buffered: a |
| 350 | // gigabyte object costs a chunk of memory, not a gigabyte, and the console |
| 351 | // holds no copy of tenant data at any point. |
| 352 | res, err := c.engine.ScanStream(ctx, obj.Body) |
| 353 | if err != nil { |
| 354 | // The engine could not be reached or did not answer. That is a fact |
| 355 | // about the ENGINE, not about this object, and it must not be written |
| 356 | // as a per-object verdict: a clamd that is down for an hour would |
| 357 | // otherwise leave thousands of objects permanently tagged failed, |
| 358 | // with nothing to distinguish them from objects that genuinely broke |
| 359 | // the scanner. Fail the pass instead; the object stays unknown and |
| 360 | // the next sweep picks it up. |
| 361 | return VerdictUnknown, false, fmt.Errorf("objstore: scan %s/%s: %w", bucket, key, err) |
| 362 | } |
| 363 | |
| 364 | v := verdictFor(res.Status) |
| 365 | recorded, err := c.tag(ctx, bucket, key, version, v) |
| 366 | if err != nil { |
| 367 | return VerdictUnknown, false, err |
| 368 | } |
| 369 | return v, recorded, nil |
| 370 | } |
| 371 | |
| 372 | // verdictFor maps an engine status onto the recorded verdict. |
| 373 | // |
| 374 | // StatusSizeExceeded becomes VerdictSkipped, not VerdictFailed. Nothing |
| 375 | // failed: the object was simply never judged, and an audit that cannot tell |
| 376 | // "we looked and it was clean" from "we never looked" is not evidence. |
| 377 | func verdictFor(s clamd.Status) Verdict { |
| 378 | switch s { |
| 379 | case clamd.StatusClean: |
| 380 | return VerdictClean |
| 381 | case clamd.StatusInfected: |
| 382 | return VerdictInfected |
| 383 | case clamd.StatusSizeExceeded: |
| 384 | return VerdictSkipped |
| 385 | default: |
| 386 | return VerdictFailed |
| 387 | } |
| 388 | } |
| 389 | |
| 390 | // tag writes the verdict, the ETag it applies to, and when it was reached. |
| 391 | // |
| 392 | // PutObjectTagging REPLACES the whole tag set, so anything already on the |
| 393 | // object is preserved explicitly here rather than assumed absent. |
| 394 | // |
| 395 | // An earlier version of this reasoned that nothing could be there to lose: the |
| 396 | // store user has no s3:PutObjectTagging, and AWS requires that same permission |
| 397 | // to set tags via x-amz-tagging on upload, so an app cannot tag its own object |
| 398 | // by either route. That holds for a bucket CARLOS minted. It does not hold for |
| 399 | // an ADOPTED one — `store grant --bucket` takes over a bucket whose objects |
| 400 | // may carry tags written long before CARLOS scoped any credentials, by tooling |
| 401 | // that is still using them for lifecycle rules or cost allocation. Replacing |
| 402 | // the set blindly would delete them, silently, on first scan. |
| 403 | func (c *clamavScanner) tag(ctx context.Context, bucket, key, version string, v Verdict) (recorded bool, err error) { |
| 404 | // The timestamp is minted HERE, once, and threaded through — it goes in a |
| 405 | // tag and into the signature over it, and two calls to time.Now() would |
| 406 | // produce a signature over a moment the tag does not record. |
| 407 | return c.tagWith(ctx, bucket, key, version, v, time.Now().UTC().Format(time.RFC3339)) |
| 408 | } |
| 409 | |
| 410 | // tagWith writes a verdict stamped with a GIVEN time, so that re-signing an |
| 411 | // existing verdict after a key rotation can keep the moment the bytes were |
| 412 | // actually judged. tag() supplies now; resign() supplies what was already |
| 413 | // there. |
| 414 | func (c *clamavScanner) tagWith(ctx context.Context, bucket, key, version string, v Verdict, at string) (recorded bool, err error) { |
| 415 | scanTagSet := []s3types.Tag{ |
| 416 | {Key: s2aws(TagVerdict), Value: s2aws(string(v))}, |
| 417 | {Key: s2aws(TagVersion), Value: s2aws(version)}, |
| 418 | {Key: s2aws(TagAt), Value: s2aws(at)}, |
| 419 | // The one that makes the other three mean something. |
| 420 | {Key: s2aws(TagSignature), Value: s2aws(c.signer.Sign(scanFacts{ |
| 421 | bucket: bucket, key: key, version: version, verdict: string(v), at: at, |
| 422 | }))}, |
| 423 | } |
| 424 | // OURS FIRST, so that truncating to the limit below drops foreign tags |
| 425 | // and never one of the three this whole mechanism exists to write. |
| 426 | set := append([]s3types.Tag(nil), scanTagSet...) |
| 427 | // Read-merge-write. One extra call per object per LIFETIME, not per sweep: |
| 428 | // an object whose verdict still matches its version never reaches Scan |
| 429 | // again. |
| 430 | existing, err := c.s3.GetObjectTagging(ctx, &awss3.GetObjectTaggingInput{ |
| 431 | Bucket: &bucket, |
| 432 | Key: &key, |
| 433 | }) |
| 434 | if err != nil { //nolint:nestif // the three outcomes are the point |
| 435 | // A tag set we cannot read must NOT be written over. PutObjectTagging |
| 436 | // replaces the whole set, so proceeding here would destroy exactly the |
| 437 | // foreign tags this merge exists to protect — silently, and on the one |
| 438 | // path where we know least about what is there. The sweep read this |
| 439 | // same object's tags moments ago through Verdict, so a failure now is |
| 440 | // transient: fail the page and let the next tick retry. |
| 441 | if isDenied(err) || isGone(err) { |
| 442 | // Nothing we could ever write, and not a pass failure. NOT |
| 443 | // recorded, which is what stops a caller believing the object |
| 444 | // carries a verdict it does not. |
| 445 | return false, nil |
| 446 | } |
| 447 | return false, fmt.Errorf("objstore: read tags before writing %s/%s: %w", bucket, key, err) |
| 448 | } |
| 449 | for _, t := range existing.TagSet { |
| 450 | if !isScanTag(aws2s(t.Key)) { |
| 451 | set = append(set, t) |
| 452 | } |
| 453 | } |
| 454 | |
| 455 | // S3 caps an object at ten tags, and an object already carrying eight |
| 456 | // foreign ones cannot also hold our three. What happens next depends on |
| 457 | // the verdict, and the two cases are genuinely different. |
| 458 | // |
| 459 | // CLEAN (or failed, or skipped): yield. Erroring would abandon the page |
| 460 | // unwritten every single tick — the exact permanent stall the denied |
| 461 | // classification was added to remove — and the only cost of not recording |
| 462 | // is that the object is scanned again next time. A tenant's tags, which on |
| 463 | // an ADOPTED bucket may predate CARLOS entirely, are worth more than one |
| 464 | // avoided re-scan. |
| 465 | // |
| 466 | // INFECTED: do not yield. `scan enforce on` writes a Deny keyed on |
| 467 | // `s3:ExistingObjectTag/carlos-scan = infected` (aws.go), so the Deny only |
| 468 | // ever engages if this tag is WRITTEN. Yielding here means an app that |
| 469 | // attaches eight tags of its own to an upload makes its objects |
| 470 | // permanently unmarkable — verdict reached, write skipped, Deny never |
| 471 | // matches, malware readable through the app's own credential, and nothing |
| 472 | // anywhere alarms: the sweep counts the object as reached, the result |
| 473 | // object says recorded:false, and the status door reads no tag and answers |
| 474 | // unknown. |
| 475 | // |
| 476 | // So an infected verdict evicts foreign tags to fit. That is a real loss |
| 477 | // of a tenant's metadata and it is the right trade against a known- |
| 478 | // infected object that enforcement cannot touch. It takes the minimum the |
| 479 | // limit forces — our three go in, as many foreign tags as still fit are |
| 480 | // kept — because taking more than necessary would be gratuitous. |
| 481 | if len(set) > maxObjectTags { |
| 482 | if v != VerdictInfected { |
| 483 | return false, nil |
| 484 | } |
| 485 | log.Printf("objstore: %s/%s is infected and carries %d foreign tags — evicting %d of them so the verdict can be recorded and enforcement can see it", |
| 486 | bucket, key, len(set)-len(scanTagSet), len(set)-maxObjectTags) |
| 487 | // set[:maxObjectTags] keeps OUR tags, which are first, and as many |
| 488 | // foreign ones as still fit. The room left for a tenant is |
| 489 | // maxObjectTags-len(scanTagSet) — derived, because it changed from |
| 490 | // three to four the moment the signature tag landed and a hardcoded |
| 491 | // number would have quietly started writing an eleven-tag set. |
| 492 | set = set[:maxObjectTags] |
| 493 | } |
| 494 | |
| 495 | if _, err := c.s3.PutObjectTagging(ctx, &awss3.PutObjectTaggingInput{ |
| 496 | Bucket: &bucket, |
| 497 | Key: &key, |
| 498 | Tagging: &s3types.Tagging{TagSet: set}, |
| 499 | }); err != nil { |
| 500 | return false, fmt.Errorf("objstore: tag %s/%s: %w", bucket, key, err) |
| 501 | } |
| 502 | return true, nil |
| 503 | } |
| 504 | |
| 505 | // maxObjectTags is S3's per-object tag limit. |
| 506 | const maxObjectTags = 10 |
| 507 | |
| 508 | // isScanTag is the set this scanner owns and may overwrite. Everything else on |
| 509 | // the object belongs to whoever put it there. |
| 510 | func isScanTag(key string) bool { |
| 511 | return key == TagVerdict || key == TagVersion || key == TagAt || key == TagSignature |
| 512 | } |
| 513 | |
| 514 | // normalizeETag strips the quotes S3 wraps an ETag in. The listing and the tag |
| 515 | // must compare equal or every object re-scans forever, and only one of the two |
| 516 | // arrives quoted. |
| 517 | func normalizeETag(etag string) string { |
| 518 | return strings.Trim(strings.TrimSpace(etag), `"`) |
| 519 | } |
| 520 | |
| 521 | func aws2s(p *string) string { |
| 522 | if p == nil { |
| 523 | return "" |
| 524 | } |
| 525 | return *p |
| 526 | } |
| 527 | |
| 528 | func s2aws(s string) *string { return &s } |
| 529 | |
| 530 | // ErrObjectGone is an object that vanished between the listing page and the |
| 531 | // fetch. The sweep drops it rather than counting it: it is neither covered nor |
| 532 | // a gap, it simply is not there any more. |
| 533 | var ErrObjectGone = errors.New("objstore: object no longer exists") |
| 534 | |
| 535 | // deniedCodes are the S3 and KMS answers that mean "not allowed, and a retry |
| 536 | // will not change that". Matched on the API error code rather than the message |
| 537 | // so a wording change upstream does not silently reclassify them as failures. |
| 538 | var deniedCodes = map[string]bool{ |
| 539 | "AccessDenied": true, |
| 540 | "AccessDeniedException": true, // KMS, on an SSE-KMS object |
| 541 | "Forbidden": true, |
| 542 | "AllAccessDisabled": true, |
| 543 | "KMS.AccessDeniedException": true, |
| 544 | "InvalidObjectState": true, // GLACIER / DEEP_ARCHIVE: unreadable until restored |
| 545 | } |
| 546 | |
| 547 | func isDenied(err error) bool { |
| 548 | var ae smithy.APIError |
| 549 | if errors.As(err, &ae) { |
| 550 | return deniedCodes[ae.ErrorCode()] |
| 551 | } |
| 552 | return false |
| 553 | } |
| 554 | |
| 555 | // isGone is an object that is not there — deleted between the listing and the |
| 556 | // fetch, which happens routinely on a bucket in use. |
| 557 | func isGone(err error) bool { |
| 558 | var nsk *s3types.NoSuchKey |
| 559 | if errors.As(err, &nsk) { |
| 560 | return true |
| 561 | } |
| 562 | var ae smithy.APIError |
| 563 | if errors.As(err, &ae) { |
| 564 | code := ae.ErrorCode() |
| 565 | return code == "NoSuchKey" || code == "NotFound" |
| 566 | } |
| 567 | return false |
| 568 | } |
| 569 | |
amadan