amadan
scan: what three reviews found in tasks 11-13
Two independent adversarial reviews (one on the landed code, one an
external security pass) plus a design review of the unwritten half.
Everything below was found by them, not by me, and every fix has a test
that fails without it.
TWO BLOCKERS, both about reading tenant bytes.
CONSENT WAS CHECKED ONCE PER PAGE, NOT PER OBJECT. DrainApp checked the
chain and then scanned up to fifty objects under that one answer. A
member re-attesting their store as client-encrypted, an operator
revoking the grant, a TrashApp — all were honoured only from the NEXT
tick, and every object still in the page was fetched and scanned under a
consent that had already been withdrawn. It is re-read for every object
now, and the bucket comes from the re-read record so a re-provisioned
store is never read at its old address. Three small metadata reads
against a full object body and an AV scan is the right trade for the one
property this feature rests on. The residual race between the check and
the GET is irreducible without a lease the object store does not offer,
and is now one object wide instead of one page wide.
DRAINONE SWALLOWED A REFUSAL IT HAD JUST OBSERVED. On the unrecordable
path it re-read consent, found it gone, and returned (out, nil) — so
DrainApp read the group as a SUCCESS, deleted the items, and carried on
scanning. On a reversible `disabled` that deleted items §7.5 says to
keep; worse, `done` carried a verdict with Recorded false and no result
object behind it, which task 16 would have turned into a callback saying
"clean" about an object the status door answers "pending" about forever
— the exact disagreement §5.2.1 introduced the result object to prevent.
THE PATH VALIDATOR HAD A SECOND DOOR, which is AGENTS.md's own rule
turned on us. It enforced "one leading slash, no backslash" on the raw
string only, and url.Parse percent-decodes: `/%5c%5cevil/x`,
`/%2f%2fevil/x`, `/safe/..;ignored/admin` and `/%EF%BC%8E%EF%BC%8E/admin`
were all accepted. The first two are refused spellings of shapes it
already refused; the third is a servlet-style receiver stripping
";params" before normalising; the fourth is Unicode compatibility
normalisation turning fullwidth stops into "..". Every rule now runs on
the decoded path as well, non-ASCII is refused outright rather than
enumerating confusables, and callbackTarget re-parses its own output and
requires the scheme and host to survive the round trip — the one check
that does not depend on having thought of the right hostile shape first.
The rest: the boot rebuild no longer abandons the index pass when a
collect fails (ids sort by time, so the expired items come FIRST — one
refused delete returned before the live items were looked at, stranding
them until the next restart); a rebuild that fails at boot is retried
each tick instead of stranding everything for the life of the process;
one app's turn is time-bounded, because fifty slow objects could hold
the whole deployment's drain; the read door is rate limited, since only
PARKED waiters were capped and a no-wait poll loop was free; a vanished
object is an outcome that wakes its waiters rather than silence; a tag
answer collects the fallback it supersedes; and no_engine now carries
§5.1's {"state","reason"} shape at the two app-facing doors, which is
what the spec told clients to branch on.
ONE TEST WAS VACUOUS: the wait-cap test staged a CLEAN object, so the
door answered on its first read and returned before the caps could
matter — it passed against a door that ignored them entirely. It stages
a pending object now, with a control proving the same request DOES park
when under the cap.
AND ONE CORRECTION TO A REVIEW. It suggested the long-poll's
register-before-read order was testable with a hook running a drain
inside the first read. Implemented, then mutation-tested: it does not
discriminate, because the hook fires at the top of Status and the read
returns the state the drain just produced, so the door answers from the
read in either order. The test is renamed to what it actually proves — a
drain landing mid-read costs neither the answer nor the window — and the
ordering is still recorded as unasserted, now with the right reason.
Nineteen mutations across the fixes; the two that survived were both too
weak to mean anything and were rewritten until they bit.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01ExnZtUYmt2h8QFcymjPaSL
9 files changed,
+641
−60
internal/console/apiscan.go+14 −2internal/console/apiscanstatus.go+32 −8internal/console/apiscanstatus_test.go+133 −7internal/console/scandrain.go+34 −0internal/console/scanhook.go+72 −21internal/console/scanhook_test.go+16 −0internal/objstore/memscanner.go+24 −0internal/objstore/queue.go+140 −10internal/objstore/queue_test.go+176 −12
CI cancelled — superseded by a newer push — run details
diff --git a/internal/console/apiscan.go b/internal/console/apiscan.go| index 1fd16dc21..c3d1e2f6a 100644 |
| --- a/internal/console/apiscan.go |
| +++ b/internal/console/apiscan.go |
| @@ -51,6 +51,18 @@ func scanNoEngine(w http.ResponseWriter) { |
| "no virus-scanning engine is configured for this deployment (CARLOS_CLAMD_ADDR)") |
| } |
| +// scanNoEngineState is scanNoEngine in §5.1's reply SHAPE, for the two |
| +// app-facing doors. |
| +// |
| +// The older operator verbs keep the plain {"error": ...}: a person reading a |
| +// CLI message wants the sentence. An APP branches on `reason`, and §5.1 and |
| +// §5.2 both promise it one — a door that answered a bare error would leave a |
| +// client that did exactly what the spec told it with nothing to switch on. |
| +func scanNoEngineState(w http.ResponseWriter) { |
| + scanUnavailable(w, http.StatusServiceUnavailable, "no_engine", |
| + "no virus-scanning engine is configured for this deployment (CARLOS_CLAMD_ADDR)") |
| +} |
| + |
| // scanDisabled is the 503 for a deployment whose engine is wired but whose |
| // scanning is switched off. Names the command, not a variable: this one is |
| // restartless. |
| @@ -761,7 +773,7 @@ func (s *Server) handleAPIScanEnqueue(w http.ResponseWriter, r *http.Request) { |
| } |
| } |
| if s.cfg.Releases == nil { |
| - scanNoEngine(w) |
| + scanNoEngineState(w) |
| return |
| } |
| ctx, cancel := context.WithTimeout(r.Context(), storeTimeout) |
| @@ -793,7 +805,7 @@ func (s *Server) handleAPIScanEnqueue(w http.ResponseWriter, r *http.Request) { |
| // ever drain: with no engine wired the item would sit until its TTL and |
| // the 202 would have been a lie. |
| if s.cfg.StoreScanner == nil { |
| - scanNoEngine(w) |
| + scanNoEngineState(w) |
| return |
| } |
| if _, err := objstore.Enqueue(ctx, s.cfg.Releases, s.scanQueue, acct.Sqid, app, |
diff --git a/internal/console/apiscanstatus.go b/internal/console/apiscanstatus.go| index 96e0656a3..032a6da2a 100644 |
| --- a/internal/console/apiscanstatus.go |
| +++ b/internal/console/apiscanstatus.go |
| @@ -119,6 +119,20 @@ func (s *Server) handleAPIScanObject(w http.ResponseWriter, r *http.Request) { |
| if !ok { |
| return |
| } |
| + // RATE LIMITED, and on the same budget as the enqueue door on purpose. |
| + // |
| + // §5.2 caps PARKED waiters, which bounds nothing for a client that polls |
| + // with no `wait` at all: each of those costs a HeadObject and a |
| + // GetObjectTagging on the console's own credentials, plus a release-store |
| + // read on the pending path. Sharing one per-principal budget with enqueue |
| + // bounds an app's total cost to this console rather than letting it spend |
| + // twice; an app that long-polls properly uses two requests a minute and |
| + // never meets it. |
| + p, _ := principalFrom(r.Context()) |
| + if !s.allowWrite(p, time.Now()) { |
| + jsonError(w, http.StatusTooManyRequests, "too many requests — wait a minute and try again") |
| + return |
| + } |
| key := strings.TrimSpace(r.URL.Query().Get("key")) |
| if key == "" { |
| jsonError(w, http.StatusBadRequest, `name the object, e.g. ?key=uploads/x.pdf`) |
| @@ -130,7 +144,7 @@ func (s *Server) handleAPIScanObject(w http.ResponseWriter, r *http.Request) { |
| return |
| } |
| if s.cfg.Releases == nil { |
| - scanNoEngine(w) |
| + scanNoEngineState(w) |
| return |
| } |
| ctx, cancel := context.WithTimeout(r.Context(), wait+storeTimeout) |
| @@ -157,7 +171,7 @@ func (s *Server) handleAPIScanObject(w http.ResponseWriter, r *http.Request) { |
| return |
| } |
| if s.cfg.StoreScanner == nil { |
| - scanNoEngine(w) |
| + scanNoEngineState(w) |
| return |
| } |
| bucket := rec.Provisioned.Bucket |
| @@ -166,12 +180,22 @@ func (s *Server) handleAPIScanObject(w http.ResponseWriter, r *http.Request) { |
| // and the registration would wake nobody, and this request would then park |
| // for its whole window on an answer that had already landed. |
| // |
| - // NOT ASSERTED BY ANY TEST, and said so rather than left looking covered: |
| - // catching the wrong order needs the drain to land inside the gap between |
| - // the two statements, and there is no seam here to hold it there. The |
| - // window is real but small, and the cost of losing the race is one wait |
| - // that expires rather than a wrong answer — the door re-reads the object |
| - // when its timer fires and never answers from the notification itself. |
| + // NOT ASSERTED BY ANY TEST, and said so rather than left looking covered. |
| + // |
| + // A review suggested this was testable with a hook that runs a drain |
| + // inside the door's first read. It was tried, and it does not |
| + // discriminate: the hook fires at the top of Status, so the read returns |
| + // the state the drain just produced and the door answers from the read in |
| + // EITHER order — the mutation that moves this block below the read |
| + // survives the test. Catching the wrong order needs the drain to land |
| + // after the read has taken its value and before the select, and there is |
| + // no seam that holds it there. |
| + // |
| + // The window is real but small, and the cost of losing the race is one |
| + // wait that expires rather than a wrong answer — the door re-reads the |
| + // object when its timer fires and never answers from the notification |
| + // itself. TestLongPollSurvivesADrainDuringItsFirstRead covers what can be |
| + // covered. |
| var woken <-chan struct{} |
| if wait > 0 { |
| ch, release, ok := s.scanWait(objstore.AppRef{Account: acct.Sqid, App: app}, key) |
diff --git a/internal/console/apiscanstatus_test.go b/internal/console/apiscanstatus_test.go| index 8164a5d86..265aa24b8 100644 |
| --- a/internal/console/apiscanstatus_test.go |
| +++ b/internal/console/apiscanstatus_test.go |
| @@ -3,7 +3,11 @@ package console |
| import ( |
| "context" |
| "encoding/json" |
| + "errors" |
| + "fmt" |
| + "io" |
| "net/http" |
| + "sync/atomic" |
| "testing" |
| "time" |
| @@ -45,6 +49,33 @@ func getStatus(t *testing.T, fx *consoleFixture, tok, query string) (int, map[st |
| return status, out |
| } |
| +// getStatusSafe is getStatus for a NON-TEST goroutine. rawReq calls t.Fatal on |
| +// a transport error, and FailNow from any goroutine but the test's own is |
| +// undefined behaviour that go vet's testinggoroutine check exists to catch. |
| +func getStatusSafe(fx *consoleFixture, tok, query string) (map[string]any, error) { |
| + req, err := http.NewRequest(http.MethodGet, enqueueURL(fx, "titogo")+"?"+query, nil) |
| + if err != nil { |
| + return nil, err |
| + } |
| + req.Header.Set("Authorization", "Bearer "+tok) |
| + resp, err := http.DefaultClient.Do(req) |
| + if err != nil { |
| + return nil, err |
| + } |
| + defer resp.Body.Close() |
| + raw, err := io.ReadAll(resp.Body) |
| + if err != nil { |
| + return nil, err |
| + } |
| + out := map[string]any{} |
| + if len(raw) > 0 { |
| + if err := json.Unmarshal(raw, &out); err != nil { |
| + return nil, fmt.Errorf("decode %q: %w", raw, err) |
| + } |
| + } |
| + return out, nil |
| +} |
| + |
| // seedObject puts an object in the engine's world and returns its version, so |
| // a test can stage a tag or a result against the same bytes. |
| func seedObject(sc *objstore.MemScanner, bucket, key, etag string) string { |
| @@ -193,6 +224,13 @@ func TestStatusDoorSharesTheConsentChain(t *testing.T) { |
| enableDeploymentScanning(t, fx) |
| tok := appToken(t, fx, "bot") |
| + // ARMED TO FAIL ON ANY READ. MemScanner counts Scans, not Status calls, so |
| + // "no bytes were read" cannot be asserted by counting — but an engine that |
| + // errors on Status turns any read into a 502, which a 409 proves did not |
| + // happen. That is what makes this a test of ORDER and not just of wording. |
| + sc := fx.srv.cfg.StoreScanner.(*objstore.MemScanner) |
| + sc.Fail("Status", errors.New("the engine must not have been consulted")) |
| + |
| status, out := getStatus(t, fx, tok, "key=uploads/x.pdf") |
| if status != http.StatusConflict || out["reason"] != "not_requested" { |
| t.Fatalf("status door = %d %v, want the enqueue door's 409 not_requested", status, out) |
| @@ -200,6 +238,19 @@ func TestStatusDoorSharesTheConsentChain(t *testing.T) { |
| if out["state"] != "unavailable" { |
| t.Errorf("state = %v, want unavailable", out["state"]) |
| } |
| + |
| + // And the reversible refusal, which is a different status code — one row |
| + // proved nothing about the mapping being shared, only about one entry. |
| + requestScanning(t, fx, member, false) |
| + fx.srv.cfg.StoreOperators = []string{"alice@keymail.dev"} |
| + if s, o := postJSON(t, scanURL(fx, "titogo")+"/grant", map[string]any{}, member); s != http.StatusOK { |
| + t.Fatalf("scan grant = %d %v", s, o) |
| + } |
| + setDeploymentScanning(t, fx, false) |
| + status, out = getStatus(t, fx, tok, "key=uploads/x.pdf") |
| + if status != http.StatusServiceUnavailable || out["reason"] != "disabled" { |
| + t.Fatalf("disabled = %d %v, want 503 disabled", status, out) |
| + } |
| } |
| // TestStatusDoorNeedsAKey and rejects a malformed wait. |
| @@ -272,9 +323,14 @@ func TestLongPollWakesOnDrain(t *testing.T) { |
| state string |
| } |
| done := make(chan reply, 1) |
| + errs := make(chan error, 1) |
| go func() { |
| start := time.Now() |
| - _, out := getStatus(t, fx, tok, "key=uploads/x.pdf&wait=25s") |
| + out, err := getStatusSafe(fx, tok, "key=uploads/x.pdf&wait=25s") |
| + if err != nil { |
| + errs <- err |
| + return |
| + } |
| st, _ := out["state"].(string) |
| done <- reply{out: out, took: time.Since(start), state: st} |
| }() |
| @@ -294,6 +350,8 @@ func TestLongPollWakesOnDrain(t *testing.T) { |
| if got.took > 20*time.Second { |
| t.Errorf("took %v against a 25s wait — it sat out the window rather than being woken", got.took) |
| } |
| + case err := <-errs: |
| + t.Fatalf("long-poll request: %v", err) |
| case <-time.After(20 * time.Second): |
| t.Fatal("the parked request was never woken by the drain") |
| } |
| @@ -403,8 +461,12 @@ func TestReleaseIsIdempotent(t *testing.T) { |
| // skipped, so a full console degrades to polling rather than to errors. |
| func TestStatusDoorAnswersWhenTheWaitCapIsFull(t *testing.T) { |
| fx, sc, tok, bucket := statusArmed(t, "code-status-full") |
| - v := seedObject(sc, bucket, "uploads/x.pdf", "etag-x") |
| - sc.Put(bucket, "uploads/x.pdf", v, objstore.VerdictClean) |
| + // PENDING, not clean. The first version staged a CLEAN object, so the door |
| + // answered on its first read and returned before the wait mattered at all |
| + // — it passed against a door that ignored the caps entirely AND against |
| + // one that parked forever when refused. The object has to be one the door |
| + // would WANT to wait for. |
| + seedObject(sc, bucket, "uploads/x.pdf", "etag-x") |
| ref := objstore.AppRef{Account: release.OpsAccount, App: "titogo"} |
| var rels []func() |
| @@ -421,10 +483,74 @@ func TestStatusDoorAnswersWhenTheWaitCapIsFull(t *testing.T) { |
| start := time.Now() |
| status, out := getStatus(t, fx, tok, "key=uploads/x.pdf&wait=10s") |
| - if status != http.StatusOK || out["state"] != "clean" { |
| - t.Fatalf("over the cap = %d %v, want the truthful 200 clean", status, out) |
| + if status != http.StatusOK || out["state"] != "pending" { |
| + t.Fatalf("over the cap = %d %v, want the truthful 200 pending", status, out) |
| + } |
| + if took := time.Since(start); took > 3*time.Second { |
| + t.Fatalf("took %v against a 10s wait — a request that cannot park must answer at once, not park anyway", took) |
| + } |
| + |
| + // THE CONTROL, without which "answered immediately" proves nothing: under |
| + // the cap, the same request on the same pending object waits. |
| + for _, rel := range rels { |
| + rel() |
| + } |
| + rels = nil |
| + start = time.Now() |
| + if _, out := getStatus(t, fx, tok, "key=uploads/x.pdf&wait=400ms"); out["state"] != "pending" { |
| + t.Fatalf("under the cap = %v, want pending", out) |
| + } |
| + if took := time.Since(start); took < 350*time.Millisecond { |
| + t.Errorf("under the cap it returned after %v — it did not park, so the test above says nothing", took) |
| + } |
| +} |
| + |
| +// TestLongPollSurvivesADrainDuringItsFirstRead runs a whole drain INSIDE the |
| +// door's first Status call and requires a prompt, correct answer. |
| +// |
| +// WHAT IT DOES NOT PROVE, stated because the obvious reading of it is wrong: it |
| +// does not pin the register-before-read ORDER. That was the intent, and a |
| +// mutation that moves registration after the first read SURVIVES it. The |
| +// reason is that the hook fires at the top of Status, so by the time the read |
| +// returns the drain has already tagged the object and the door answers "clean" |
| +// from the read itself — in both orderings, without the wake mattering at all. |
| +// |
| +// The ordering therefore remains unasserted, and the comment on the door says |
| +// so. What this does pin is that a drain landing mid-read cannot make the door |
| +// lose the answer or hang, which is worth having on its own. |
| +func TestLongPollSurvivesADrainDuringItsFirstRead(t *testing.T) { |
| + ctx := context.Background() |
| + fx, sc, tok, bucket := statusArmed(t, "code-status-order2") |
| + seedObject(sc, bucket, "uploads/x.pdf", "etag-x") |
| + if _, err := objstore.Enqueue(ctx, fx.rel, fx.srv.scanQueue, release.OpsAccount, "titogo", |
| + &objstore.QueueItem{Key: "uploads/x.pdf", By: "test"}, time.Now()); err != nil { |
| + t.Fatal(err) |
| + } |
| + |
| + // FIRE ONCE, and with a compare-and-swap rather than sync.Once: the drain |
| + // this hook starts calls Status itself, so the hook re-enters — and |
| + // Once.Do is not reentrant, it DEADLOCKS the second caller against the |
| + // first. A CAS lets the inner call see "already fired" and return. |
| + var fired atomic.Bool |
| + sc.OnStatus(func() { |
| + if !fired.CompareAndSwap(false, true) { |
| + return |
| + } |
| + if err := fx.srv.scanDrainTick(ctx); err != nil { |
| + t.Errorf("drain inside the door's read: %v", err) |
| + } |
| + }) |
| + |
| + start := time.Now() |
| + status, out := getStatus(t, fx, tok, "key=uploads/x.pdf&wait=20s") |
| + took := time.Since(start) |
| + if status != http.StatusOK { |
| + t.Fatalf("status = %d %v", status, out) |
| + } |
| + if out["state"] != "clean" { |
| + t.Fatalf("state = %v, want clean — the drain ran during the first read", out["state"]) |
| } |
| - if took := time.Since(start); took > 5*time.Second { |
| - t.Errorf("took %v — a request that cannot park should answer at once", took) |
| + if took > 15*time.Second { |
| + t.Errorf("took %v against a 20s wait — a drain during the read must not cost the whole window", took) |
| } |
| } |
diff --git a/internal/console/scandrain.go b/internal/console/scandrain.go| index f28ea490a..39aeae949 100644 |
| --- a/internal/console/scandrain.go |
| +++ b/internal/console/scandrain.go |
| @@ -37,6 +37,14 @@ const scanDrainInterval = 30 * time.Second |
| // thousand queued objects delay every other tenant's single urgent upload. |
| const scanDrainPerApp = 50 |
| +// scanDrainAppBudget bounds one app's turn in one tick — the time half of the |
| +// fairness rule, where scanDrainPerApp is the count half. |
| +// |
| +// Longer than a tick on purpose: a drain that overran its interval should |
| +// finish the objects it started rather than be cut off mid-scan, and the |
| +// ticker's next fire simply waits. What it must not do is run unbounded. |
| +const scanDrainAppBudget = 90 * time.Second |
| + |
| // StartScanDrain rebuilds the index and then drains every 30s until ctx is |
| // done. serve() starts it; constructing a Server starts nothing. |
| // |
| @@ -47,7 +55,9 @@ func (s *Server) StartScanDrain(ctx context.Context) { |
| if s.cfg.Releases == nil || s.cfg.StoreScanner == nil { |
| return // no bucket, or no engine: nothing to drain into |
| } |
| + rebuilt := true |
| if err := s.rebuildScanQueue(ctx); err != nil { |
| + rebuilt = false |
| // Not fatal. A console that cannot rebuild still drains what it is |
| // told about from here on, and the next restart tries again; refusing |
| // to start the goroutine would turn a listing hiccup into an outage |
| @@ -57,6 +67,19 @@ func (s *Server) StartScanDrain(ctx context.Context) { |
| t := time.NewTicker(scanDrainInterval) |
| defer t.Stop() |
| for { |
| + // RETRIED UNTIL IT SUCCEEDS ONCE. One listing failure at boot used to |
| + // strand every pre-restart item for the life of the process, which on |
| + // a console that stays up for weeks is weeks — and §7.1 calls this |
| + // pass "the only thing that collects a stranded item, so it is not |
| + // optional". A retry per tick costs one listing on a deployment that |
| + // is already failing them. |
| + if !rebuilt { |
| + if err := s.rebuildScanQueue(ctx); err != nil { |
| + log.Printf("console: scan queue rebuild (retry): %v", err) |
| + } else { |
| + rebuilt = true |
| + } |
| + } |
| if err := s.scanDrainTick(ctx); err != nil { |
| log.Printf("console: scan drain: %v", err) |
| } |
| @@ -164,6 +187,17 @@ func (s *Server) scanDrainTick(ctx context.Context) error { |
| // not end the tick for the apps after it in the round-robin, which is the |
| // posture scanPass already takes for the same reason. |
| func (s *Server) drainScanApp(ctx context.Context, ref objstore.AppRef, now time.Time) { |
| + // ONE APP'S TURN IS BOUNDED. clamd's own read timeout is minutes, and the |
| + // object body stream carries no deadline of its own, so fifty slow objects |
| + // belonging to one tenant could hold the whole deployment's drain for |
| + // hours — starving every other tenant's single urgent item, which is the |
| + // precise failure §7.3's cap exists to prevent and which the cap alone |
| + // does not close. The sweep budgets per app for the same reason; this is |
| + // that budget. An app that runs out keeps its items and its index entry |
| + // and gets the next tick. |
| + ctx, cancel := context.WithTimeout(ctx, scanDrainAppBudget) |
| + defer cancel() |
| + |
| done, _, err := objstore.DrainApp(ctx, s.cfg.Releases, s.cfg.StoreScanner, s.scanQueue, |
| ref.Account, ref.App, scanDrainPerApp, now) |
diff --git a/internal/console/scanhook.go b/internal/console/scanhook.go| index af27c8d1f..3b9a4ffbc 100644 |
| --- a/internal/console/scanhook.go |
| +++ b/internal/console/scanhook.go |
| @@ -30,6 +30,15 @@ var errCallbackPath = errors.New("console: callback path") |
| // validCallbackPath is the rule. It runs at the DOOR, on the way in, so that |
| // nothing hostile is ever stored — a validator that runs at delivery time |
| // leaves every already-accepted item in the bucket to be found later. |
| +// |
| +// IT CHECKS BOTH SPELLINGS OF EVERY RULE: the raw string the app sent AND the |
| +// decoded path url.Parse produces. The first version of this function checked |
| +// only the raw string, and an external security review walked straight past it |
| +// with "/%5c%5cevil/x" and "/%2f%2fevil/x" — percent-encoded forms of two |
| +// shapes it explicitly refused in plaintext. That is AGENTS.md's own rule |
| +// turned on us: a rule stated as an absence can only be checked by looking for |
| +// the absent thing, and cannot see a second door. url.Parse IS the second |
| +// door. |
| func validCallbackPath(p string) error { |
| switch { |
| case p == "": |
| @@ -38,16 +47,21 @@ func validCallbackPath(p string) error { |
| return fmt.Errorf("%w: longer than %d bytes", errCallbackPath, maxCallbackPath) |
| } |
| - // Control bytes first, before anything tries to interpret the string. A |
| - // raw CR or LF is header injection into the request the console is about |
| - // to build; a NUL truncates in anything that later hands this to C. |
| - if i := strings.IndexAny(p, "\r\n\x00"); i >= 0 { |
| - return fmt.Errorf("%w: control byte at %d", errCallbackPath, i) |
| - } |
| - // And the percent-encoded forms, which survive to whatever decodes them. |
| - // Checked on the raw string because url.Parse will happily keep them in |
| - // RawQuery, and the comparison is case-insensitive because %0D and %0d are |
| - // the same byte. |
| + // ASCII ONLY, and this is the rule that closes the normalisation family. |
| + // A receiver that applies Unicode compatibility normalisation turns |
| + // U+FF0E FULLWIDTH FULL STOP into ".", so "/．．/admin" becomes a |
| + // traversal on arrival while looking like nothing here. We cannot know |
| + // what the receiving stack normalises, and the set of confusables is not |
| + // something to enumerate — so a callback path is printable ASCII, and an |
| + // app that wants anything else has to say it in the query string, where |
| + // it stays percent-encoded and is never a path segment. |
| + for i := 0; i < len(p); i++ { |
| + if c := p[i]; c < 0x20 || c > 0x7e { |
| + return fmt.Errorf("%w: byte %#02x at %d is not printable ASCII", errCallbackPath, c, i) |
| + } |
| + } |
| + // Percent-encoded control bytes, which survive to whatever decodes them. |
| + // Case-insensitive, because %0D and %0d are the same byte. |
| lower := strings.ToLower(p) |
| for _, enc := range []string{"%0d", "%0a", "%00"} { |
| if strings.Contains(lower, enc) { |
| @@ -55,11 +69,8 @@ func validCallbackPath(p string) error { |
| } |
| } |
| - // Exactly one leading slash, and no backslash anywhere. "//evil/x" is |
| - // protocol-relative and "/\evil/x" is treated as protocol-relative by |
| - // enough parsers to count; a backslash has no legitimate place in a path |
| - // we are about to build a URL from, so it is refused outright rather than |
| - // enumerated. |
| + // The raw shape. "//evil/x" is protocol-relative and "/\evil/x" is treated |
| + // as protocol-relative by enough parsers to count. |
| switch { |
| case !strings.HasPrefix(p, "/"): |
| return fmt.Errorf("%w: must begin with /", errCallbackPath) |
| @@ -81,19 +92,45 @@ func validCallbackPath(p string) error { |
| if u.Scheme != "" || u.Host != "" || u.User != nil || u.Opaque != "" { |
| return fmt.Errorf("%w: names a host or scheme", errCallbackPath) |
| } |
| + |
| + // NOW THE SAME RULES AGAIN, on the decoded path. Everything below is a |
| + // shape that is refused above in plaintext and was reachable through |
| + // percent-encoding until an external review pointed it out. |
| + switch { |
| + case !strings.HasPrefix(u.Path, "/"): |
| + return fmt.Errorf("%w: decodes to something that is not a path", errCallbackPath) |
| + case strings.HasPrefix(u.Path, "//"): |
| + return fmt.Errorf("%w: decodes to a protocol-relative path", errCallbackPath) |
| + case strings.ContainsRune(u.Path, '\\'): |
| + return fmt.Errorf("%w: decodes to a path containing a backslash", errCallbackPath) |
| + case strings.ContainsAny(u.Path, "#?"): |
| + return fmt.Errorf("%w: decodes to a path containing a delimiter", errCallbackPath) |
| + } |
| + for i := 0; i < len(u.Path); i++ { |
| + if c := u.Path[i]; c < 0x21 || c > 0x7e { |
| + return fmt.Errorf("%w: decoded path byte %#02x at %d is not printable ASCII", |
| + errCallbackPath, c, i) |
| + } |
| + } |
| + // SEMICOLONS. A servlet-style receiver strips ";params" from each segment |
| + // BEFORE it normalises the path, so "/safe/..;ignored/admin" arrives as |
| + // "/safe/../admin" and traverses — while every check that splits on "/" |
| + // and compares to ".." sees the harmless segment "..;ignored". Refused |
| + // outright: a callback path has no legitimate use for one. |
| + if strings.ContainsRune(u.Path, ';') { |
| + return fmt.Errorf("%w: ';' in the path", errCallbackPath) |
| + } |
| // An '@' in the first segment is how an authority is smuggled past a |
| // check that only looks at u.Host: "/x@evil/y" parses as a path, but |
| // anything that later re-parses a concatenated form can read it as |
| // userinfo. We build structurally and would survive it — this refuses it |
| - // anyway, because a callback path has no reason to hold one and the next |
| - // consumer of this field may not build structurally. |
| + // anyway, because the next consumer of this field may not. |
| if strings.Contains(u.Path, "@") { |
| return fmt.Errorf("%w: '@' in the path", errCallbackPath) |
| } |
| // Traversal, on the DECODED path, which is the whole point: "%2e%2e" is |
| - // ".." to every server that will receive it. u.Path is already decoded, so |
| - // this catches both spellings with one check. |
| + // ".." to every server that will receive it. |
| for _, seg := range strings.Split(u.Path, "/") { |
| if seg == ".." { |
| return fmt.Errorf("%w: '..' segment", errCallbackPath) |
| @@ -120,10 +157,24 @@ func callbackTarget(appHost, p string) (*url.URL, error) { |
| if err != nil { |
| return nil, fmt.Errorf("%w: %v", errCallbackPath, err) |
| } |
| - return &url.URL{ |
| + out := &url.URL{ |
| Scheme: "https", |
| Host: appHost, |
| Path: u.Path, |
| RawQuery: u.RawQuery, |
| - }, nil |
| + } |
| + // THE ROUND TRIP. Everything above reasons about what url.URL will render; |
| + // this checks it. Re-parsing our own output and requiring the scheme, host |
| + // and port to be exactly what we set is the one assertion that does not |
| + // depend on having thought of the right hostile shape in advance — and the |
| + // review that prompted it found four we had not. |
| + back, err := url.Parse(out.String()) |
| + if err != nil { |
| + return nil, fmt.Errorf("%w: built a URL that will not re-parse: %v", errCallbackPath, err) |
| + } |
| + if back.Scheme != "https" || back.Host != appHost || back.User != nil { |
| + return nil, fmt.Errorf("%w: built target re-parses as %q, not a path on %q", |
| + errCallbackPath, back.String(), appHost) |
| + } |
| + return out, nil |
| } |
diff --git a/internal/console/scanhook_test.go b/internal/console/scanhook_test.go| index 051b203b5..57e335225 100644 |
| --- a/internal/console/scanhook_test.go |
| +++ b/internal/console/scanhook_test.go |
| @@ -46,6 +46,22 @@ func TestCallbackPathValidation(t *testing.T) { |
| {"fragment", "/hooks#frag"}, |
| {"NUL", "/hooks\x00"}, |
| {"over 1KB", "/" + strings.Repeat("a", 1024)}, |
| + // The DECODED forms of shapes already refused above. url.Parse |
| + // percent-decodes into u.Path, so a rule enforced only on the raw |
| + // string is a rule with a second door — the exact lesson AGENTS.md |
| + // now carries. Found by an external security review. |
| + {"encoded backslash", "/%5cevil.example/x"}, |
| + {"encoded double backslash", "/%5c%5cevil.example/x"}, |
| + {"encoded protocol-relative", "/%2f%2fevil.example/x"}, |
| + {"encoded single extra slash", "/%2fevil.example/x"}, |
| + // A servlet-style receiver strips ;params BEFORE normalising the |
| + // path, which turns this into /admin. |
| + {"path parameter hiding a traversal", "/safe/..;ignored/admin"}, |
| + {"bare path parameter", "/hooks;jsessionid=1/scan"}, |
| + // Fullwidth full stops. A receiver applying Unicode compatibility |
| + // normalisation turns these into "..". |
| + {"fullwidth traversal", "/\uff0e\uff0e/admin"}, |
| + {"encoded fullwidth traversal", "/%EF%BC%8E%EF%BC%8E/admin"}, |
| } |
| for _, tc := range refused { |
diff --git a/internal/objstore/memscanner.go b/internal/objstore/memscanner.go| index 6f2281cc3..f0b8e0d59 100644 |
| --- a/internal/objstore/memscanner.go |
| +++ b/internal/objstore/memscanner.go |
| @@ -10,6 +10,8 @@ import ( |
| // MemProvisioner: no network, no clamd, and a failure hook so a test can make |
| // any single method fail once and prove the walk survives it. |
| type MemScanner struct { |
| + // onStatus is OnStatus's hook; see its comment. |
| + onStatus func() |
| mu sync.Mutex |
| enabled map[string]bool |
| verdicts map[string]scanRecord |
| @@ -84,10 +86,32 @@ func (s *MemScanner) Unrecordable(key string) { |
| s.unrecordable[key] = true |
| } |
| +// OnStatus installs a hook that runs at the top of Status, BEFORE any lock is |
| +// taken, so a test can make something else happen while a caller is mid-read. |
| +// |
| +// It exists for one assertion that has no other seam: the status door |
| +// registers its long-poll waiter before its first read, and the only way to |
| +// catch the opposite order is to run a drain inside that read. The hook must |
| +// not re-enter the scanner (Status will be called again beneath it), so a test |
| +// uses sync.Once. |
| +func (s *MemScanner) OnStatus(f func()) { |
| + s.mu.Lock() |
| + defer s.mu.Unlock() |
| + s.onStatus = f |
| +} |
| + |
| // Status answers about the CURRENT version, and reports what the record SAYS |
| // — denied and failed included. The re-issue rule below is Verdict's, and |
| // applying it here would make a failed scan read as pending forever. |
| func (s *MemScanner) Status(ctx context.Context, bucket, key string) (Verdict, ScanObject, error) { |
| + // The hook runs unlocked, or anything it does that touches this scanner |
| + // deadlocks against the lock taken below. |
| + s.mu.Lock() |
| + hook := s.onStatus |
| + s.mu.Unlock() |
| + if hook != nil { |
| + hook() |
| + } |
| s.mu.Lock() |
| defer s.mu.Unlock() |
| if err := s.takeFailure("Status"); err != nil { |
diff --git a/internal/objstore/queue.go b/internal/objstore/queue.go| index eb5895426..28666e556 100644 |
| --- a/internal/objstore/queue.go |
| +++ b/internal/objstore/queue.go |
| @@ -231,6 +231,25 @@ func RebuildQueue(ctx context.Context, s release.Store, idx *QueueIndex, acct, a |
| } |
| return 0, err |
| } |
| + // DEFERRED, so that an error partway through still points the index at what |
| + // was found. The first version bumped only on the success path, so one |
| + // failed Delete of an expired item returned early and left the app's LIVE |
| + // items with nothing pointing at them — and since this rebuild is the only |
| + // thing that collects a stranded item, they would have sat until the next |
| + // restart. An index entry for an app with no work is harmless (one empty |
| + // listing, then it is released); a live item with no entry is not. |
| + defer func() { |
| + if live > 0 { |
| + idx.Bump(AppRef{Account: acct, App: app}) |
| + } |
| + }() |
| + // A FAILED COLLECT MUST NOT ABORT THE INDEX PASS. The first version |
| + // returned on the first Delete error, and because ids sort by time the |
| + // EXPIRED items come first — so one refused delete returned before the |
| + // live items had even been looked at, leaving them with nothing pointing |
| + // at them until the next restart. Deleting is this pass's second job; |
| + // indexing is its first, and the first must finish. |
| + var failed error |
| for _, qk := range keys { |
| item, rerr := readQueueItem(ctx, s, qk) |
| switch { |
| @@ -238,7 +257,9 @@ func RebuildQueue(ctx context.Context, s release.Store, idx *QueueIndex, acct, a |
| continue |
| case errors.Is(rerr, errQueueItemUnreadable): |
| if derr := s.Delete(ctx, qk); derr != nil && !errors.Is(derr, os.ErrNotExist) { |
| - return live, derr |
| + // Still there, so still live: the next rebuild retries it. |
| + log.Printf("objstore: scan queue rebuild: delete unreadable %s: %v", qk, derr) |
| + failed, live = derr, live+1 |
| } |
| continue |
| case rerr != nil: |
| @@ -248,16 +269,14 @@ func RebuildQueue(ctx context.Context, s release.Store, idx *QueueIndex, acct, a |
| } |
| if now.Sub(item.At) > QueueTTL { |
| if derr := s.Delete(ctx, qk); derr != nil && !errors.Is(derr, os.ErrNotExist) { |
| - return live, derr |
| + log.Printf("objstore: scan queue rebuild: delete expired %s: %v", qk, derr) |
| + failed, live = derr, live+1 |
| } |
| continue |
| } |
| live++ |
| } |
| - if live > 0 { |
| - idx.Bump(AppRef{Account: acct, App: app}) |
| - } |
| - return live, nil |
| + return live, failed |
| } |
| // QueueDepth is how many items are waiting for one app — the per-app half of |
| @@ -290,6 +309,13 @@ type Drained struct { |
| // so — an infected object with Recorded false is NOT covered by the |
| // enforcement deny, because the deny matches on a tag that is not there. |
| Recorded bool |
| + // Gone is an object that was not there to scan. It is an OUTCOME, not an |
| + // error: the app asked about a key, and "it does not exist" is the honest |
| + // answer to that. It rides in the same list so a waiter parked on the key |
| + // is woken — before this existed the drain deleted the items silently and |
| + // a long-poll sat out its whole window for an answer that had already |
| + // been settled. |
| + Gone bool |
| // Scanned is whether the engine actually ran, as against the verdict being |
| // read back from an existing tag. The meter counts on-demand work; an |
| // answer the sweep already paid for is not the tenant's to be charged. |
| @@ -389,17 +415,74 @@ func DrainApp(ctx context.Context, s release.Store, sc Scanner, idx *QueueIndex, |
| ErrEngineDrift, acct, app, rec.Scanning.Enabled.Engine, engine) |
| } |
| - bucket := rec.Provisioned.Bucket |
| + // The page-level bucket is only used to fail fast; the loop below re-reads |
| + // the record and uses THAT bucket for every object. |
| + _ = rec.Provisioned.Bucket |
| groups, err := groupQueue(ctx, s, keys, now) |
| if err != nil { |
| return nil, false, err |
| } |
| - for _, g := range groups { |
| + for i, g := range groups { |
| + // CONSENT IS RE-READ FOR EVERY OBJECT, not once for the page. |
| + // |
| + // An external security review found this: the check above happens |
| + // once and the loop then reads up to fifty objects' bytes. A member |
| + // who re-attests their store as client-encrypted, an operator who |
| + // revokes the grant or disables the deployment, or a TrashApp landing |
| + // mid-page, would all be honoured only from the NEXT tick — and every |
| + // object still in this page would be fetched and scanned under a |
| + // consent that had already been withdrawn. |
| + // |
| + // The cost is three small metadata reads per object against a full |
| + // object body read and an AV scan, which is the right trade for the |
| + // one property this whole feature rests on. A pre-call check still |
| + // leaves an irreducible window between the check and the GET — that |
| + // race cannot be closed without a lease the object store does not |
| + // offer — but it is now one object wide instead of one page wide. |
| + rec, refusal, cerr := QueuePermitted(ctx, s, acct, app) |
| + if cerr != nil { |
| + // A read error is not a refusal: keep the rest of the page and |
| + // let the next tick have it. |
| + return done, false, cerr |
| + } |
| + if refusal == RefusalDisabled { |
| + return done, false, ErrScanningDisabled |
| + } |
| + if refusal.Permanent() { |
| + // Everything not yet handled goes, on the same rule the |
| + // page-level check applies — including this group, whose object |
| + // has not been read. |
| + if delErr := dropRemainingQueue(ctx, s, acct, app, groups[i:]); delErr != nil { |
| + return done, false, delErr |
| + } |
| + return done, false, ErrQueueNotPermitted |
| + } |
| + // Engine drift too, for the same reason: the record can be moved to a |
| + // different engine between one object and the next, and a verdict |
| + // recorded under the wrong provenance is worse than no verdict. |
| + if rec.Scanning.Enabled.Engine != engine { |
| + return done, false, fmt.Errorf("%w (%s/%s): record says %q, configured engine is %q", |
| + ErrEngineDrift, acct, app, rec.Scanning.Enabled.Engine, engine) |
| + } |
| + // And the bucket comes from the RE-READ record, so a store that was |
| + // re-provisioned mid-page is never read at its old address. |
| + bucket := rec.Provisioned.Bucket |
| + |
| out, derr := drainOne(ctx, s, sc, acct, app, bucket, engine, signature, g) |
| + if errors.Is(derr, ErrQueueNotPermitted) { |
| + // Consent went away between this group's check and its write. |
| + // Same conclusion as the check above, so the error keeps meaning |
| + // one thing to the caller: the queue was dropped, not deferred. |
| + if delErr := dropRemainingQueue(ctx, s, acct, app, groups[i:]); delErr != nil { |
| + return done, false, delErr |
| + } |
| + return done, false, derr |
| + } |
| if errors.Is(derr, ErrObjectGone) { |
| if delErr := deleteQueueItems(ctx, s, g.queueKeys); delErr != nil { |
| return done, false, delErr |
| } |
| + done = append(done, Drained{Key: g.key, Gone: true, CallbackPaths: g.callbackPaths}) |
| continue |
| } |
| if derr != nil { |
| @@ -417,6 +500,29 @@ func DrainApp(ctx context.Context, s release.Store, sc Scanner, idx *QueueIndex, |
| return done, listedEmpty, nil |
| } |
| +// dropRemainingQueue deletes the items of every group not yet handled, and the |
| +// app's fallback results with them. |
| +// |
| +// One function because two places reach the same conclusion — the per-object |
| +// consent check, and a refusal observed inside drainOne after the scan — and |
| +// the caller must not be able to tell them apart. ErrQueueNotPermitted means |
| +// "the queue was dropped" wherever it comes from. |
| +func dropRemainingQueue(ctx context.Context, s release.Store, acct, app string, left []*scanGroup) error { |
| + var keys []string |
| + for _, g := range left { |
| + keys = append(keys, g.queueKeys...) |
| + } |
| + if err := deleteQueueItems(ctx, s, keys); err != nil { |
| + return err |
| + } |
| + // And the fallback results. A result object is a verdict no door may |
| + // answer from once the app may not be scanned, and it is the worse orphan |
| + // of the two: a stranded queue item is drained next tick, while a stranded |
| + // result is found by nothing, blocks the app-name claim gate, and |
| + // resurrects a deleted app on the next boot. |
| + return DeleteScanResults(ctx, s, acct, app) |
| +} |
| + |
| // scanGroup is every queued request in this page that asks about ONE object |
| // key: the object is read once, and every request that asked gets its answer. |
| type scanGroup struct { |
| @@ -518,6 +624,15 @@ func drainOne(ctx context.Context, s release.Store, sc Scanner, acct, app, bucke |
| if !needsScan { |
| // The object carries this verdict, at this version, right now. |
| out.Verdict, out.Recorded = verdict, true |
| + // And any fallback for this key is now superseded — §5.2.1 says the |
| + // drain drops it as soon as a real verdict is persisted, and the |
| + // sweep can be what persisted it. Without this the result object |
| + // survives until the 24h age sweep even though the object it stands |
| + // in for has carried a real answer since. Delete is idempotent, so |
| + // the common case (no fallback) costs one cheap call. |
| + if err := DeleteScanResult(ctx, s, acct, app, g.key); err != nil { |
| + return Drained{}, err |
| + } |
| return out, nil |
| } |
| if obj == (ScanObject{}) { |
| @@ -552,8 +667,23 @@ func drainOne(ctx context.Context, s release.Store, sc Scanner, acct, app, bucke |
| // "will this item ever be useful" but "may this app be scanned right now", |
| // and a result object written after the answer turned to no is a verdict |
| // no door may answer from. |
| - if _, refusal, err := QueuePermitted(ctx, s, acct, app); err != nil || refusal != RefusalNone { |
| - return out, err |
| + // |
| + // AND IT IS RETURNED AS AN ERROR, which the first version got wrong. It |
| + // returned (out, nil), so DrainApp read the group as a SUCCESS: it deleted |
| + // the items, appended an outcome to done, and carried on scanning the rest |
| + // of the page with positive knowledge that consent was gone. Three things |
| + // followed, and the third is the worst — on a reversible `disabled` the |
| + // items were deleted against §7.5, the remaining objects were read anyway, |
| + // and `done` carried a verdict with Recorded false and NO result object |
| + // behind it, so task 16 would deliver a callback saying "clean" about an |
| + // object the status door will answer "pending" about forever. That is |
| + // exactly the disagreement §5.2.1 introduced the result object to prevent. |
| + if _, refusal, err := QueuePermitted(ctx, s, acct, app); err != nil { |
| + return Drained{}, err |
| + } else if refusal == RefusalDisabled { |
| + return Drained{}, ErrScanningDisabled |
| + } else if refusal != RefusalNone { |
| + return Drained{}, ErrQueueNotPermitted |
| } |
| if err := WriteScanResult(ctx, s, acct, app, &ScanResult{ |
| Key: g.key, Verdict: string(v), Version: obj.Version(), |
diff --git a/internal/objstore/queue_test.go b/internal/objstore/queue_test.go| index 833ca50cb..f8f4384a1 100644 |
| --- a/internal/objstore/queue_test.go |
| +++ b/internal/objstore/queue_test.go |
| @@ -491,8 +491,19 @@ func TestDrainAppDropsGoneObjects(t *testing.T) { |
| if err != nil { |
| t.Fatalf("a vanished object failed the drain: %v", err) |
| } |
| - if len(done) != 0 { |
| - t.Fatalf("outcomes for an object that is not there: %+v", done) |
| + // IT IS AN OUTCOME, and this test used to assert the opposite. "There is |
| + // no such object" is the honest answer to the question the app asked, and |
| + // reporting nothing meant a long-poll parked on that key sat out its whole |
| + // window for an answer that had already been settled — the drain deleted |
| + // the item silently and woke nobody. |
| + if len(done) != 1 { |
| + t.Fatalf("outcomes = %+v, want the one that says the object is gone", done) |
| + } |
| + if !done[0].Gone { |
| + t.Errorf("outcome %+v is not marked Gone, so a caller cannot tell it from a verdict", done[0]) |
| + } |
| + if done[0].Verdict != "" || done[0].Recorded || done[0].Scanned { |
| + t.Errorf("a gone object produced a verdict-shaped outcome: %+v", done[0]) |
| } |
| left, _ := s.List(ctx, QueuePrefix("bab", "titogo")) |
| if len(left) != 0 { |
| @@ -839,11 +850,18 @@ func TestResultObjectNotWrittenUnderPurgedPrefix(t *testing.T) { |
| sc.Unrecordable(objectName(0)) // the ten-tag cap, in production |
| enqueueN(t, s, idx, 1) |
| - // The teardown lands between the drain's own consent check (the first read |
| - // of store.json) and drainOne's re-read before writing (the second). The |
| - // re-read is the guard; without it the create-only write SUCCEEDS under a |
| - // prefix nothing will ever list again. |
| - staged := &recordRaceStore{Store: s, key: Key("bab", "titogo"), before: 2, race: func() { |
| + // The teardown lands between the PER-OBJECT consent check (the second read |
| + // of store.json, added when a review pointed out that one check per page |
| + // is not one check per object) and drainOne's re-read before writing (the |
| + // third). The re-read is the guard being tested here; without it the |
| + // create-only write SUCCEEDS under a prefix nothing will ever list again. |
| + // |
| + // The count is 3 rather than 2 because there are now three reads of this |
| + // key in a drain, not two. If it ever drifts again, the tell is this test |
| + // failing with ErrQueueNotPermitted — that means the race moved to an |
| + // EARLIER guard, which is the case TestPurgeBeforeTheObjectIsReadIsRefused |
| + // below covers on purpose. |
| + staged := &recordRaceStore{Store: s, key: Key("bab", "titogo"), before: 3, race: func() { |
| // What PurgeApp leaves behind: the prefix, emptied. Done directly |
| // because PurgeApp itself refuses over a provisioned record, and the |
| // teardown that clears it is not what this test is about. |
| @@ -859,18 +877,73 @@ func TestResultObjectNotWrittenUnderPurgedPrefix(t *testing.T) { |
| } |
| }} |
| - if _, _, err := DrainApp(ctx, staged, sc, idx, "bab", "titogo", 10, time.Now()); err != nil { |
| - t.Fatalf("drain: %v", err) |
| + done, _, err := DrainApp(ctx, staged, sc, idx, "bab", "titogo", 10, time.Now()) |
| + // REPORTED, not swallowed. The first version returned (out, nil) here, so |
| + // the drain read the group as a success: it deleted the items, put an |
| + // outcome in done, and carried on scanning the rest of the page knowing |
| + // consent was gone. The outcome it produced had Recorded false with no |
| + // result object behind it, which task 16 would have turned into a callback |
| + // saying "clean" about an object the status door answers "pending" about |
| + // forever. |
| + if !errors.Is(err, ErrQueueNotPermitted) { |
| + t.Fatalf("drain err = %v, want ErrQueueNotPermitted", err) |
| } |
| - left, err := s.List(ctx, release.AppPrefix("bab", "titogo")) |
| - if err != nil { |
| - t.Fatal(err) |
| + if len(done) != 0 { |
| + t.Errorf("done carries %d outcome(s) for a store that may not be scanned: %+v", len(done), done) |
| + } |
| + left, lerr := s.List(ctx, release.AppPrefix("bab", "titogo")) |
| + if lerr != nil { |
| + t.Fatal(lerr) |
| } |
| if len(left) != 0 { |
| t.Fatalf("the purged app's prefix still holds %v — its name can never be claimed again", left) |
| } |
| } |
| +// TestPurgeBeforeTheObjectIsReadIsRefused is the guard the per-object consent |
| +// check adds, and it is the better of the two outcomes: the teardown lands |
| +// before this object's bytes are read at all, so nothing is fetched, nothing |
| +// is scanned and no result can be stranded. |
| +// |
| +// The page-level check alone could not do this. It runs once and then the loop |
| +// reads up to fifty objects, so a purge — or a revoked grant, or a member |
| +// re-attesting the store as client-encrypted — was honoured only from the next |
| +// tick, and every object still in the page was read under a consent that had |
| +// already been withdrawn. Found by an external security review. |
| +func TestPurgeBeforeTheObjectIsReadIsRefused(t *testing.T) { |
| + ctx := context.Background() |
| + s, sc, idx, _ := queueFixture(t, 1) |
| + enqueueN(t, s, idx, 1) |
| + |
| + staged := &recordRaceStore{Store: s, key: Key("bab", "titogo"), before: 2, race: func() { |
| + keys, err := s.List(ctx, release.AppPrefix("bab", "titogo")) |
| + if err != nil { |
| + t.Error(err) |
| + return |
| + } |
| + for _, k := range keys { |
| + if err := s.Delete(ctx, k); err != nil { |
| + t.Error(err) |
| + } |
| + } |
| + }} |
| + |
| + _, _, err := DrainApp(ctx, staged, sc, idx, "bab", "titogo", 10, time.Now()) |
| + if !errors.Is(err, ErrQueueNotPermitted) { |
| + t.Fatalf("err = %v, want ErrQueueNotPermitted — consent was gone before this object was read", err) |
| + } |
| + if n := sc.Scans(objectName(0)); n != 0 { |
| + t.Errorf("scanned %d time(s) after consent was withdrawn mid-page", n) |
| + } |
| + left, lerr := s.List(ctx, release.AppPrefix("bab", "titogo")) |
| + if lerr != nil { |
| + t.Fatal(lerr) |
| + } |
| + if len(left) != 0 { |
| + t.Fatalf("the purged app's prefix still holds %v", left) |
| + } |
| +} |
| + |
| // recordRaceStore runs race() just before the Nth read of one key, so a test |
| // can put a competing writer at a chosen point in a walk. Same shape as |
| // walk_test.go's raceOnReadStore, which fires on the first read only. |
| @@ -1059,3 +1132,94 @@ func disableDeployment(t *testing.T, s release.Store) { |
| t.Fatal(err) |
| } |
| } |
| + |
| +// readErrorStore fails Get for one exact key, and behaves for everything else. |
| +// A transient bad moment on one object, which is the case §7.5 separates from |
| +// absence. |
| +type readErrorStore struct { |
| + release.Store |
| + key string |
| + err error |
| +} |
| + |
| +func (r *readErrorStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| + if key == r.key { |
| + return nil, "", r.err |
| + } |
| + return r.Store.Get(ctx, key) |
| +} |
| + |
| +// TestRebuildKeepsAnItemItCouldNotRead pins the claim RebuildQueue's own |
| +// comment makes and nothing tested: a STORE error is the store having a bad |
| +// moment, so the item counts as live and the index still points at it. |
| +// |
| +// The opposite — treating an unreadable item as gone — silently strands it, |
| +// because this rebuild is the only thing that would ever have listed that |
| +// prefix again. §7.5's rule is that deletion is driven by absence, never by |
| +// error, and this is that rule on the read side. |
| +func TestRebuildKeepsAnItemItCouldNotRead(t *testing.T) { |
| + ctx := context.Background() |
| + s, _, idx, _ := queueFixture(t, 1) |
| + enqueueN(t, s, idx, 1) |
| + |
| + keys, err := s.List(ctx, QueuePrefix("bab", "titogo")) |
| + if err != nil || len(keys) != 1 { |
| + t.Fatalf("staging: %v %v", keys, err) |
| + } |
| + broken := &readErrorStore{Store: s, key: keys[0], err: errors.New("s3 had a moment")} |
| + |
| + fresh := NewQueueIndex() |
| + live, err := RebuildQueue(ctx, broken, fresh, "bab", "titogo", time.Now()) |
| + if err != nil { |
| + t.Fatalf("a read error failed the whole rebuild: %v", err) |
| + } |
| + if live != 1 { |
| + t.Errorf("live = %d, want 1 — an item we could not read is not an item we know is gone", live) |
| + } |
| + if fresh.Len() != 1 { |
| + t.Error("the index does not point at the app, so nothing will ever list that prefix again") |
| + } |
| + left, lerr := s.List(ctx, QueuePrefix("bab", "titogo")) |
| + if lerr != nil || len(left) != 1 { |
| + t.Errorf("the item was deleted on a store error: %v %v", left, lerr) |
| + } |
| +} |
| + |
| +// TestRebuildStillIndexesWhenADeleteFails is the blocker an adversarial review |
| +// found: the bump used to happen only on the success path, so one failed |
| +// Delete of an EXPIRED item returned early and left the app's LIVE items with |
| +// nothing pointing at them — until the next restart, since this pass is the |
| +// only thing that collects a stranded item. |
| +func TestRebuildStillIndexesWhenADeleteFails(t *testing.T) { |
| + ctx := context.Background() |
| + s, _, idx, _ := queueFixture(t, 2) |
| + |
| + // One expired item and one live one. |
| + if _, err := Enqueue(ctx, s, idx, "bab", "titogo", |
| + &QueueItem{Key: objectName(0), By: "app"}, time.Now().Add(-QueueTTL-time.Hour)); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := Enqueue(ctx, s, idx, "bab", "titogo", |
| + &QueueItem{Key: objectName(1), By: "app"}, time.Now()); err != nil { |
| + t.Fatal(err) |
| + } |
| + broken := &deleteErrorStore{Store: s, err: errors.New("s3 refused the delete")} |
| + |
| + fresh := NewQueueIndex() |
| + live, err := RebuildQueue(ctx, broken, fresh, "bab", "titogo", time.Now()) |
| + if err == nil { |
| + t.Fatal("the failed delete was not reported") |
| + } |
| + _ = live |
| + if fresh.Len() != 1 { |
| + t.Fatalf("index holds %d app(s) after a failed delete, want 1 — the live item is stranded until the next restart", fresh.Len()) |
| + } |
| +} |
| + |
| +// deleteErrorStore fails every Delete. |
| +type deleteErrorStore struct { |
| + release.Store |
| + err error |
| +} |
| + |
| +func (d *deleteErrorStore) Delete(context.Context, string) error { return d.err } |