| 1 | // The on-demand READ door (spec §5.2): what is known about one object right |
| 2 | // now, optionally waiting a little for it. |
| 3 | // |
| 4 | // Its own file rather than more of apiscan.go, because the scan family there |
| 5 | // is about POLICY — who may scan, whether enforcement is on — and this one |
| 6 | // answers a question about a single object for the app that owns it. They |
| 7 | // share the consent chain and nothing else. |
| 8 | package console |
| 9 | |
| 10 | import ( |
| 11 | "context" |
| 12 | "errors" |
| 13 | "log" |
| 14 | "net/http" |
| 15 | "strings" |
| 16 | "time" |
| 17 | |
| 18 | "github.com/carlosframework/platform/internal/objstore" |
| 19 | ) |
| 20 | |
| 21 | // scanWaitMax is the long-poll ceiling. Beyond this a client should be |
| 22 | // polling: a request parked longer than half a minute is holding a console |
| 23 | // goroutine against an answer that may never come in this process at all. |
| 24 | const scanWaitMax = 30 * time.Second |
| 25 | |
| 26 | // scanWaitersPerApp and scanWaitersGlobal cap parked requests. |
| 27 | // |
| 28 | // BOTH, and the per-app one is not implied by the global one: without it a |
| 29 | // single tenant fills the global budget and every other tenant's wait |
| 30 | // degrades to polling. Without the GLOBAL one, enough tenants at their |
| 31 | // per-app limit still add up to more parked goroutines than a console should |
| 32 | // hold. Neither is a refusal — over the cap the door answers `pending` |
| 33 | // immediately, which is TRUE, and the client polls. |
| 34 | const ( |
| 35 | scanWaitersPerApp = 32 |
| 36 | scanWaitersGlobal = 256 |
| 37 | ) |
| 38 | |
| 39 | // scanStatusOf reads what is known about one object, in the order §5.2 sets: |
| 40 | // the object's own tag first, the fallback result second. |
| 41 | // |
| 42 | // THE TAG WINS, and the order is not a preference. A result object exists only |
| 43 | // because a verdict could not be written onto the object; if the object now |
| 44 | // carries one, the object is the more recent and more authoritative answer, |
| 45 | // and the drain deletes the result the moment that happens. |
| 46 | // |
| 47 | // The version comparison happens inside the scanner and inside |
| 48 | // ReadScanResult. It never reaches this door and never reaches an app: the |
| 49 | // token is an internal format, and an app comparing it wrongly gets the |
| 50 | // laundering bug as a correctness bug — upload v2 over a key whose v1 was |
| 51 | // tagged clean, read the stale tag, serve the payload. |
| 52 | func (s *Server) scanStatusOf(ctx context.Context, acct, app, bucket, key string) (map[string]any, error) { |
| 53 | verdict, obj, err := s.cfg.StoreScanner.Status(ctx, bucket, key) |
| 54 | if errors.Is(err, objstore.ErrObjectGone) { |
| 55 | return map[string]any{"state": "gone", "recorded": false}, nil |
| 56 | } |
| 57 | if err != nil { |
| 58 | return nil, err |
| 59 | } |
| 60 | if verdict != objstore.VerdictUnknown { |
| 61 | // REPORTED AS THEMSELVES. `failed` and `denied` are verdicts, not |
| 62 | // absences: the landed tag reader maps them back to unknown so the |
| 63 | // SWEEP retries them, which is right as retry policy and wrong as an |
| 64 | // answer to a question about an object. Left applied here, a failed |
| 65 | // scan would read as `pending` forever — the drain deletes the item |
| 66 | // as handled, and nothing changes the answer until a sweep re-fails |
| 67 | // it. |
| 68 | // |
| 69 | // `denied` is not claimed as recorded: Status also answers it for a |
| 70 | // refused HeadObject or tag read, where no tag was seen at all. |
| 71 | return map[string]any{"state": string(verdict), "recorded": verdict != objstore.VerdictDenied}, nil |
| 72 | } |
| 73 | // No usable tag. The object may still have an answer that could not be |
| 74 | // written onto it. |
| 75 | if obj == (objstore.ScanObject{}) { |
| 76 | // A denied HeadObject: no version, so no result object can be matched |
| 77 | // against it either. Status already said what this is. |
| 78 | return map[string]any{"state": "pending", "recorded": false}, nil |
| 79 | } |
| 80 | res, err := objstore.ReadScanResult(ctx, s.cfg.Releases, acct, app, key, obj.Version()) |
| 81 | if err != nil { |
| 82 | return nil, err |
| 83 | } |
| 84 | if res == nil { |
| 85 | // Absent, or recorded against different bytes — ReadScanResult makes |
| 86 | // no distinction and neither should this door. Nothing is known about |
| 87 | // the object that is there NOW. |
| 88 | return map[string]any{"state": "pending", "recorded": false}, nil |
| 89 | } |
| 90 | // The verdict is on a result object rather than the object itself, which |
| 91 | // the app has to know: an infected object answered this way is NOT covered |
| 92 | // by the enforcement Deny, because that matches on a tag that is not |
| 93 | // there. |
| 94 | out := map[string]any{"state": res.Verdict, "recorded": false} |
| 95 | // Provenance, but only where it is actually known. A result object records |
| 96 | // the engine and signature that judged it; a TAG, through the Status seam |
| 97 | // as it stands, yields the verdict alone. Reporting this deployment's |
| 98 | // current engine for a tag would be a guess dressed as a record — the |
| 99 | // engine may have been replaced since — so the tag branch above says |
| 100 | // nothing rather than something plausible. |
| 101 | if res.Engine != "" { |
| 102 | out["engine"] = res.Engine |
| 103 | } |
| 104 | if res.Signature != "" { |
| 105 | out["signature"] = res.Signature |
| 106 | } |
| 107 | if !res.At.IsZero() { |
| 108 | out["at"] = res.At.UTC().Format(time.RFC3339) |
| 109 | } |
| 110 | return out, nil |
| 111 | } |
| 112 | |
| 113 | // scanStatusPermitted runs the consent chain and answers any refusal itself. |
| 114 | // The door calls it before EVERY read, so a revoke or disable that lands while |
| 115 | // a request is parked stops the next read rather than being answered past. |
| 116 | func (s *Server) scanStatusPermitted(ctx context.Context, w http.ResponseWriter, r *http.Request, acct, app string) (bucket string, ok bool) { |
| 117 | rec, refusal, err := objstore.QueuePermitted(ctx, s.cfg.Releases, acct, app) |
| 118 | if err != nil { |
| 119 | log.Printf("console: api scan object %s/%s: consent: %v", acct, app, err) |
| 120 | jsonError(w, http.StatusBadGateway, "the store record could not be read — try again") |
| 121 | return "", false |
| 122 | } |
| 123 | // The SAME mapping the enqueue door uses, which is the point of it being a |
| 124 | // table: two doors that answered the same refusal differently would let an |
| 125 | // app conclude something about its store from the disagreement. |
| 126 | if ans := scanRefusalAnswer(refusal); ans.refused() { |
| 127 | if ans.unknown { |
| 128 | log.Printf("console: api scan object %s/%s: unhandled refusal %q", acct, app, refusal) |
| 129 | } |
| 130 | if ans.notFound { |
| 131 | http.NotFound(w, r) |
| 132 | return "", false |
| 133 | } |
| 134 | scanUnavailable(w, ans.status, string(refusal), ans.msg) |
| 135 | return "", false |
| 136 | } |
| 137 | return rec.Provisioned.Bucket, true |
| 138 | } |
| 139 | |
| 140 | // handleAPIScanObject answers GET .../store/scan/object?key=&wait=. |
| 141 | // |
| 142 | // `skipped`, `gone` and `unavailable` are NOT `clean`, and they are distinct |
| 143 | // states rather than an error field, because an app treating any of them as |
| 144 | // success reproduces exactly the dishonesty the whole feature exists to |
| 145 | // prevent: reporting coverage that does not exist. |
| 146 | func (s *Server) handleAPIScanObject(w http.ResponseWriter, r *http.Request) { |
| 147 | _, acct, app, ok := s.apiAccountApp(w, r) |
| 148 | if !ok { |
| 149 | return |
| 150 | } |
| 151 | // RATE LIMITED, and on the same budget as the enqueue door on purpose. |
| 152 | // |
| 153 | // §5.2 caps PARKED waiters, which bounds nothing for a client that polls |
| 154 | // with no `wait` at all: each of those costs a HeadObject and a |
| 155 | // GetObjectTagging on the console's own credentials, plus a release-store |
| 156 | // read on the pending path. Sharing one per-principal budget with enqueue |
| 157 | // bounds an app's total cost to this console rather than letting it spend |
| 158 | // twice; an app that long-polls properly uses two requests a minute and |
| 159 | // never meets it. |
| 160 | p, _ := principalFrom(r.Context()) |
| 161 | if !s.allowWrite(p, time.Now()) { |
| 162 | jsonError(w, http.StatusTooManyRequests, "too many requests — wait a minute and try again") |
| 163 | return |
| 164 | } |
| 165 | // Queried byte for byte: a trimmed key names a different object. |
| 166 | key := r.URL.Query().Get("key") |
| 167 | if strings.TrimSpace(key) == "" { |
| 168 | jsonError(w, http.StatusBadRequest, `name the object, e.g. ?key=uploads/x.pdf`) |
| 169 | return |
| 170 | } |
| 171 | wait, err := scanWaitFor(r.URL.Query().Get("wait")) |
| 172 | if err != nil { |
| 173 | jsonError(w, http.StatusBadRequest, err.Error()) |
| 174 | return |
| 175 | } |
| 176 | if s.cfg.Releases == nil { |
| 177 | scanNoEngineState(w) |
| 178 | return |
| 179 | } |
| 180 | ctx, cancel := context.WithTimeout(r.Context(), wait+storeTimeout) |
| 181 | defer cancel() |
| 182 | |
| 183 | bucket, ok := s.scanStatusPermitted(ctx, w, r, acct.Sqid, app) |
| 184 | if !ok { |
| 185 | return |
| 186 | } |
| 187 | if s.cfg.StoreScanner == nil { |
| 188 | scanNoEngineState(w) |
| 189 | return |
| 190 | } |
| 191 | |
| 192 | // REGISTERED BEFORE THE FIRST READ. A drain that finishes between the read |
| 193 | // and the registration would wake nobody, and this request would then park |
| 194 | // for its whole window on an answer that had already landed. |
| 195 | // |
| 196 | // NOT ASSERTED BY ANY TEST, and said so rather than left looking covered. |
| 197 | // |
| 198 | // A review suggested this was testable with a hook that runs a drain |
| 199 | // inside the door's first read. It was tried, and it does not |
| 200 | // discriminate: the hook fires at the top of Status, so the read returns |
| 201 | // the state the drain just produced and the door answers from the read in |
| 202 | // EITHER order — the mutation that moves this block below the read |
| 203 | // survives the test. Catching the wrong order needs the drain to land |
| 204 | // after the read has taken its value and before the select, and there is |
| 205 | // no seam that holds it there. |
| 206 | // |
| 207 | // The window is real but small, and the cost of losing the race is one |
| 208 | // wait that expires rather than a wrong answer — the door re-reads the |
| 209 | // object when its timer fires and never answers from the notification |
| 210 | // itself. TestLongPollSurvivesADrainDuringItsFirstRead covers what can be |
| 211 | // covered. |
| 212 | var woken <-chan struct{} |
| 213 | if wait > 0 { |
| 214 | ch, release, ok := s.scanWait(objstore.AppRef{Account: acct.Sqid, App: app}, key) |
| 215 | if ok { |
| 216 | defer release() |
| 217 | woken = ch |
| 218 | } |
| 219 | // Not ok means the caps are full. Nothing is refused: the read below |
| 220 | // still happens and still answers truthfully; only the WAITING is |
| 221 | // skipped. |
| 222 | } |
| 223 | |
| 224 | deadline := time.NewTimer(wait) |
| 225 | defer deadline.Stop() |
| 226 | for { |
| 227 | out, err := s.scanStatusOf(ctx, acct.Sqid, app, bucket, key) |
| 228 | if err != nil { |
| 229 | log.Printf("console: api scan object %s/%s %q: %v", acct.Sqid, app, key, err) |
| 230 | jsonError(w, http.StatusBadGateway, "the verdict could not be read — try again") |
| 231 | return |
| 232 | } |
| 233 | if out["state"] != "pending" || woken == nil { |
| 234 | writeJSON(w, http.StatusOK, out) |
| 235 | return |
| 236 | } |
| 237 | select { |
| 238 | case <-woken: |
| 239 | // Round again and re-read. The wake says "something happened", |
| 240 | // never what — the answer comes from the object, not from the |
| 241 | // notification, so a spurious wake costs one read and nothing |
| 242 | // else. |
| 243 | case <-deadline.C: |
| 244 | // RE-READ once more, rather than answering with the value captured |
| 245 | // before parking: the drain notifies only after a whole app page |
| 246 | // returns, so an object scanned early in a busy page would |
| 247 | // otherwise be answered `pending` by a waiter that could have seen |
| 248 | // its answer. The loop answers whatever that read says. |
| 249 | woken = nil |
| 250 | case <-ctx.Done(): |
| 251 | // Nothing was read now, and the earlier `pending` is not an answer. |
| 252 | jsonError(w, http.StatusBadGateway, "the verdict could not be read — try again") |
| 253 | return |
| 254 | } |
| 255 | // Consent again before the next read: a revoke, trash or disable that |
| 256 | // landed while parked must stop it. |
| 257 | if bucket, ok = s.scanStatusPermitted(ctx, w, r, acct.Sqid, app); !ok { |
| 258 | return |
| 259 | } |
| 260 | } |
| 261 | } |
| 262 | |
| 263 | // scanWaitFor parses the wait parameter and bounds it. |
| 264 | // |
| 265 | // An over-long wait is CLAMPED rather than refused: a client asking for five |
| 266 | // minutes wants the longest wait it can have, and answering 400 turns a |
| 267 | // reasonable request into an error the app author has to special-case. A |
| 268 | // malformed one is a 400, because that is a bug in the caller rather than an |
| 269 | // ambitious number. |
| 270 | func scanWaitFor(raw string) (time.Duration, error) { |
| 271 | raw = strings.TrimSpace(raw) |
| 272 | if raw == "" { |
| 273 | return 0, nil |
| 274 | } |
| 275 | d, err := time.ParseDuration(raw) |
| 276 | if err != nil { |
| 277 | return 0, errors.New(`wait must be a duration such as "15s"`) |
| 278 | } |
| 279 | if d < 0 { |
| 280 | return 0, errors.New("wait cannot be negative") |
| 281 | } |
| 282 | if d > scanWaitMax { |
| 283 | return scanWaitMax, nil |
| 284 | } |
| 285 | return d, nil |
| 286 | } |
| 287 | |
amadan