| 1 | // The scan pass (spec §7.1): the goroutine that drives objstore.RunSweep over |
| 2 | // every store that has opted in. |
| 3 | // |
| 4 | // Its own ticker, deliberately. Draft 2 wanted to ride storagePass; that was |
| 5 | // wrong — storagePass returns once the day's usage object exists |
| 6 | // (economicspass.go:137-140), so it runs once per UTC day, and a sweep that |
| 7 | // advances one page a day would never finish a bucket of any size. It is also |
| 8 | // not folded into economicsPass, whose single goroutine already runs census, |
| 9 | // statements and billing and already worries about a slow step delaying the |
| 10 | // others. |
| 11 | package console |
| 12 | |
| 13 | import ( |
| 14 | "context" |
| 15 | "errors" |
| 16 | "log" |
| 17 | "strings" |
| 18 | "time" |
| 19 | |
| 20 | "github.com/carlosframework/platform/internal/objstore" |
| 21 | "github.com/carlosframework/platform/internal/release" |
| 22 | ) |
| 23 | |
| 24 | // scanPassInterval is how often the pass ticks. The re-sweep FLOOR is a |
| 25 | // separate number (Config.ScanInterval): ticking hourly is what lets a |
| 26 | // part-finished sweep make progress every hour, while a finished one is left |
| 27 | // alone until it is genuinely stale. |
| 28 | const scanPassInterval = time.Hour |
| 29 | |
| 30 | const ( |
| 31 | defaultScanBudget = time.Minute |
| 32 | defaultScanInterval = 24 * time.Hour |
| 33 | ) |
| 34 | |
| 35 | // StartScanPass runs scanPass immediately and then hourly until ctx is done — |
| 36 | // StartInfraPass's shape (infrapass.go:49). |
| 37 | func (s *Server) StartScanPass(ctx context.Context) { |
| 38 | if s.cfg.Releases == nil || s.cfg.StoreScanner == nil { |
| 39 | return // no bucket, or nobody opted in: nothing to tick for |
| 40 | } |
| 41 | t := time.NewTicker(scanPassInterval) |
| 42 | defer t.Stop() |
| 43 | for { |
| 44 | if err := s.scanPass(ctx); err != nil { |
| 45 | log.Printf("console: scan pass: %v", err) |
| 46 | } |
| 47 | select { |
| 48 | case <-ctx.Done(): |
| 49 | return |
| 50 | case <-t.C: |
| 51 | } |
| 52 | } |
| 53 | } |
| 54 | |
| 55 | // antivirusEnabled reads the deployment record, seeding it once when the |
| 56 | // retired CARLOS_STORE_SCANNER supplied this deployment's wiring. |
| 57 | // |
| 58 | // The seed carries an intent the operator already expressed. Refusing to seed |
| 59 | // would not withhold consent — it would silently withdraw consent already |
| 60 | // given, which is the failure cmd/console refuses to permit elsewhere: "a typo |
| 61 | // that silently disabled scanning would leave an operator believing a |
| 62 | // compliance control is on when nothing is running." |
| 63 | // |
| 64 | // A read error is neither "enabled" nor "disabled": it is an error. Seeding on |
| 65 | // a failed read would enable scanning on a deployment that had disabled it. |
| 66 | func (s *Server) antivirusEnabled(ctx context.Context) (bool, error) { |
| 67 | av, _, err := release.GetAntivirus(ctx, s.cfg.Releases) |
| 68 | if err != nil { |
| 69 | return false, err |
| 70 | } |
| 71 | if av != nil { |
| 72 | return av.Enabled, nil |
| 73 | } |
| 74 | if !s.cfg.ScanSeedFromDeprecatedEnv { |
| 75 | // Wired with no record: nothing scans until an operator enables it. |
| 76 | // Said once per tick rather than silently, because the difference |
| 77 | // between "off" and "nobody told it to be on" is the whole point. |
| 78 | log.Printf("console: a clamd address is configured but object scanning is not enabled — run `carlos system antivirus enable`") |
| 79 | return false, nil |
| 80 | } |
| 81 | // Create-only: two consoles, or a restart loop, converge on one seed, and |
| 82 | // an operator who later runs `disable` is never overridden on the next |
| 83 | // restart. |
| 84 | _, err = release.PutAntivirus(ctx, s.cfg.Releases, &release.AntivirusSettings{ |
| 85 | Enabled: true, |
| 86 | Engine: "clamav", |
| 87 | At: time.Now().UTC(), |
| 88 | By: "migration from CARLOS_STORE_SCANNER", |
| 89 | }, "") |
| 90 | if errors.Is(err, release.ErrCASConflict) { |
| 91 | // Somebody else got there first — and NOT NECESSARILY WITH A SEED. |
| 92 | // The race that matters is an operator running `disable` in the same |
| 93 | // instant: the read above saw no record, the operator's write landed, |
| 94 | // and answering "enabled" here would run a whole pass over tenant |
| 95 | // bytes against the switch the operator had just turned off. So read |
| 96 | // the winner rather than assuming it agrees with us. |
| 97 | av, _, rerr := release.GetAntivirus(ctx, s.cfg.Releases) |
| 98 | if rerr != nil { |
| 99 | return false, rerr |
| 100 | } |
| 101 | if av == nil { |
| 102 | // Created and gone again between the two calls. Neither on nor |
| 103 | // off: say off and let the next tick decide. |
| 104 | return false, nil |
| 105 | } |
| 106 | return av.Enabled, nil |
| 107 | } |
| 108 | if err != nil { |
| 109 | return false, err |
| 110 | } |
| 111 | return true, nil |
| 112 | } |
| 113 | |
| 114 | func (s *Server) scanBudget() time.Duration { |
| 115 | if s.cfg.ScanBudget > 0 { |
| 116 | return s.cfg.ScanBudget |
| 117 | } |
| 118 | return defaultScanBudget |
| 119 | } |
| 120 | |
| 121 | func (s *Server) scanInterval() time.Duration { |
| 122 | if s.cfg.ScanInterval > 0 { |
| 123 | return s.cfg.ScanInterval |
| 124 | } |
| 125 | return defaultScanInterval |
| 126 | } |
| 127 | |
| 128 | // scanPass gives every enabled store one budget of scanning. |
| 129 | // |
| 130 | // Enumeration is storagePass's (economicspass.go:150-181): ListDirs("apps/"), |
| 131 | // per-account ListDirs, per-app objstore.Get. Hourly, that is 24x that pass's |
| 132 | // listing cost — accepted for v1 and named here so it is a known cost rather |
| 133 | // than a surprise. An index is the fix if it bites. |
| 134 | // |
| 135 | // One app's failure does not stop the others. A store whose engine is |
| 136 | // unreachable, whose record names a different engine, or whose bucket has gone |
| 137 | // is logged and skipped: the pass exists to make progress on what it can. |
| 138 | func (s *Server) scanPass(ctx context.Context) error { |
| 139 | if s.cfg.Releases == nil || s.cfg.StoreScanner == nil { |
| 140 | return nil |
| 141 | } |
| 142 | // Policy is read every tick, so `carlos system antivirus enable` takes |
| 143 | // effect without a restart. Disabled means this tick does nothing — not |
| 144 | // that there is no goroutine, which is what would make enabling need one. |
| 145 | enabled, err := s.antivirusEnabled(ctx) |
| 146 | if err != nil { |
| 147 | return err |
| 148 | } |
| 149 | if !enabled { |
| 150 | return nil |
| 151 | } |
| 152 | acctPrefixes, err := s.cfg.Releases.ListDirs(ctx, "apps/") |
| 153 | if err != nil { |
| 154 | return err |
| 155 | } |
| 156 | budget := s.scanBudget() |
| 157 | interval := s.scanInterval() |
| 158 | |
| 159 | for _, acctPrefix := range acctPrefixes { |
| 160 | acct := strings.TrimSuffix(strings.TrimPrefix(acctPrefix, "apps/"), "/") |
| 161 | appPrefixes, err := s.cfg.Releases.ListDirs(ctx, acctPrefix) |
| 162 | if err != nil { |
| 163 | return err |
| 164 | } |
| 165 | for _, appPrefix := range appPrefixes { |
| 166 | if err := ctx.Err(); err != nil { |
| 167 | return err |
| 168 | } |
| 169 | app := strings.TrimSuffix(strings.TrimPrefix(appPrefix, acctPrefix), "/") |
| 170 | if err := s.scanOneApp(ctx, acct, app, budget, interval); err != nil { |
| 171 | log.Printf("console: scan pass: %s/%s: %v", acct, app, err) |
| 172 | } |
| 173 | } |
| 174 | } |
| 175 | return nil |
| 176 | } |
| 177 | |
| 178 | // scanOneApp decides whether this app is due and, if so, sweeps it. |
| 179 | func (s *Server) scanOneApp(ctx context.Context, acct, app string, budget, interval time.Duration) error { |
| 180 | rec, _, err := objstore.Get(ctx, s.cfg.Releases, acct, app) |
| 181 | if err != nil { |
| 182 | return err |
| 183 | } |
| 184 | // Absent is (nil, "", nil) — objstore.Get's own idiom — so this is "no |
| 185 | // store", not a failure. |
| 186 | if rec == nil || rec.Provisioned == nil || rec.Scanning == nil || rec.Scanning.Enabled == nil { |
| 187 | return nil |
| 188 | } |
| 189 | |
| 190 | due, err := s.storeIsDue(ctx, rec, acct, app, interval) |
| 191 | if err != nil { |
| 192 | return err |
| 193 | } |
| 194 | if !due { |
| 195 | return nil |
| 196 | } |
| 197 | |
| 198 | lister, err := s.storeLister(ctx, rec.Provisioned) |
| 199 | if err != nil { |
| 200 | return err |
| 201 | } |
| 202 | _, err = objstore.RunSweep(ctx, s.cfg.Releases, lister, s.cfg.StoreScanner, acct, app, budget) |
| 203 | return err |
| 204 | } |
| 205 | |
| 206 | // storeIsDue answers whether this store wants work now. |
| 207 | // |
| 208 | // A sweep already in flight is always due — that is how a bucket bigger than |
| 209 | // one budget ever finishes. Only a COMPLETED sweep waits out the interval, so |
| 210 | // LastComplete stays stable long enough to be quoted in an audit answer |
| 211 | // instead of being rebuilt from scratch every hour. |
| 212 | func (s *Server) storeIsDue(ctx context.Context, rec *objstore.Record, acct, app string, interval time.Duration) (bool, error) { |
| 213 | side, _, err := objstore.GetSidecar(ctx, s.cfg.Releases, acct, app) |
| 214 | if err != nil { |
| 215 | // A sidecar we cannot read is not a reason to re-sweep blindly: it is |
| 216 | // a schema or ownership refusal that wants an operator, and sweeping |
| 217 | // anyway would overwrite whatever it was trying to protect. |
| 218 | return false, err |
| 219 | } |
| 220 | if side == nil { |
| 221 | return true, nil // never swept |
| 222 | } |
| 223 | if side.Current != nil { |
| 224 | return true, nil // mid-sweep: keep going |
| 225 | } |
| 226 | if side.NotApplicable { |
| 227 | // Client-encrypted — but the RECORD decides that, not the sidecar. |
| 228 | // |
| 229 | // A member can re-attest from --client-encrypted to --plaintext, which |
| 230 | // rewrites store.json and leaves the stale not_applicable sidecar |
| 231 | // behind. Reading only the sidecar here meant that store was never |
| 232 | // swept again by any tick, for as long as it existed, while |
| 233 | // `store status` cheerfully reported "scanning: clamav". Only the |
| 234 | // on-demand sweep verb could unstick it, and nothing told anyone to |
| 235 | // run it. |
| 236 | return recordSaysEncrypted(rec) == false, nil |
| 237 | } |
| 238 | if side.LastComplete == nil { |
| 239 | return true, nil |
| 240 | } |
| 241 | return time.Since(side.LastComplete.At) >= interval, nil |
| 242 | } |
| 243 | |
| 244 | // storeLister routes through the test seam or builds a real S3 lister, the |
| 245 | // way sizeStore already does for the storage sweep. |
| 246 | func (s *Server) storeLister(ctx context.Context, p *objstore.Provisioned) (objstore.ObjectLister, error) { |
| 247 | if s.storeListerFor != nil { |
| 248 | return s.storeListerFor(ctx, p) |
| 249 | } |
| 250 | client, err := objstore.S3ClientFor(ctx, p.Region, p.Endpoint) |
| 251 | if err != nil { |
| 252 | return nil, err |
| 253 | } |
| 254 | return objstore.NewObjectLister(client), nil |
| 255 | } |
| 256 | |
| 257 | // errNoScanner is what the doors return when scanning was never configured. |
| 258 | var errNoScanner = errors.New("console: no scanning engine is configured on this deployment") |
| 259 | |
| 260 | // errNoStore is a sweep asked for on an app that has no provisioned store. |
| 261 | var errNoStore = errors.New("console: this app has no object store") |
| 262 | |
| 263 | // sweepNow drives one sweep on demand, for the operator verb. It shares |
| 264 | // everything with the ticker except the schedule — spec §7.1's guarantee is |
| 265 | // "at most one driver RECORDS", not that two never run: the sidecar's CAS |
| 266 | // guard makes the loser stand down. |
| 267 | func (s *Server) sweepNow(ctx context.Context, acct, app string) (*objstore.ScanSidecar, error) { |
| 268 | if s.cfg.StoreScanner == nil { |
| 269 | return nil, errNoScanner |
| 270 | } |
| 271 | rec, _, err := objstore.Get(ctx, s.cfg.Releases, acct, app) |
| 272 | if err != nil { |
| 273 | return nil, err |
| 274 | } |
| 275 | if rec == nil || rec.Provisioned == nil { |
| 276 | return nil, errNoStore |
| 277 | } |
| 278 | lister, err := s.storeLister(ctx, rec.Provisioned) |
| 279 | if err != nil { |
| 280 | return nil, err |
| 281 | } |
| 282 | return objstore.RunSweep(ctx, s.cfg.Releases, lister, s.cfg.StoreScanner, acct, app, s.scanBudget()) |
| 283 | } |
| 284 | |
| 285 | // recordSaysEncrypted is the single reading of the member's attestation. The |
| 286 | // record is the authority; the sidecar is a cache of a decision made from it. |
| 287 | func recordSaysEncrypted(rec *objstore.Record) bool { |
| 288 | return rec != nil && rec.Scanning != nil && |
| 289 | rec.Scanning.Requested != nil && rec.Scanning.Requested.Encrypted |
| 290 | } |
| 291 | |
amadan