| 1 | package console |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/rand" |
| 6 | "crypto/sha256" |
| 7 | "encoding/hex" |
| 8 | "encoding/json" |
| 9 | "errors" |
| 10 | "fmt" |
| 11 | "io" |
| 12 | "log" |
| 13 | "net/http" |
| 14 | "os" |
| 15 | "sort" |
| 16 | "strings" |
| 17 | "sync" |
| 18 | "time" |
| 19 | |
| 20 | "github.com/carlosframework/platform/internal/publicrelease" |
| 21 | "github.com/carlosframework/platform/internal/release" |
| 22 | "github.com/carlosframework/platform/internal/status" |
| 23 | ) |
| 24 | |
| 25 | // The console half of `carlos system` (spec §5). Every door here is for an |
| 26 | // OWNER of the ops account or an address on CARLOS_SYSTEM_OPERATORS, on a |
| 27 | // PERSON session, checked on every call (apiSystemRoller); uploading a local |
| 28 | // build is the owner's alone, and an admitted operator who asks for it gets a |
| 29 | // 403 (systemUploadOwnerOnly). A caller the gate refuses gets a bare 404 like |
| 30 | // the operator doors, so anyone else learns nothing. The console holds the bucket, the signing key and the egress; |
| 31 | // the operator's laptop holds none of them. |
| 32 | |
| 33 | const ( |
| 34 | systemUploadMaxBytes = int64(128 << 20) |
| 35 | systemAssetMaxBytes = int64(256 << 20) |
| 36 | systemJobTTL = time.Hour |
| 37 | ) |
| 38 | |
| 39 | // systemPlatformRe validates a "<goos>-<goarch>" the operator typed before it |
| 40 | // becomes part of an object key — the same shape and the same reason as |
| 41 | // apicli.go's targetRe, which is why it IS that expression rather than a |
| 42 | // second copy of it that could drift. |
| 43 | var systemPlatformRe = targetRe |
| 44 | |
| 45 | // systemJob is one promote in flight. Steps is the narration an operator |
| 46 | // polls; Err is set once, at the end. The mutex guards all three against the |
| 47 | // polling handler. |
| 48 | type systemJob struct { |
| 49 | mu sync.Mutex |
| 50 | Steps []string |
| 51 | Done bool |
| 52 | Err string |
| 53 | // Result is a job's own answer when it has one (a geoip publish's |
| 54 | // pointer, geoip spec §4.2): kept with the job so the poller sees what |
| 55 | // THIS job published, not whatever the current pointer later says. |
| 56 | Result any |
| 57 | at time.Time |
| 58 | } |
| 59 | |
| 60 | func (j *systemJob) setResult(v any) { |
| 61 | j.mu.Lock() |
| 62 | j.Result = v |
| 63 | j.mu.Unlock() |
| 64 | } |
| 65 | |
| 66 | func (j *systemJob) step(format string, args ...any) { |
| 67 | j.mu.Lock() |
| 68 | j.Steps = append(j.Steps, fmt.Sprintf(format, args...)) |
| 69 | j.mu.Unlock() |
| 70 | } |
| 71 | |
| 72 | func (j *systemJob) finish(err error) { |
| 73 | j.mu.Lock() |
| 74 | j.Done = true |
| 75 | if err != nil { |
| 76 | j.Err = err.Error() |
| 77 | } |
| 78 | j.mu.Unlock() |
| 79 | } |
| 80 | |
| 81 | func (j *systemJob) done() bool { |
| 82 | j.mu.Lock() |
| 83 | defer j.mu.Unlock() |
| 84 | return j.Done |
| 85 | } |
| 86 | |
| 87 | // snapshot copies the three published fields under the lock, so the caller |
| 88 | // can serialize them without holding it across a socket write. |
| 89 | func (j *systemJob) snapshot() (steps []string, done bool, errMsg string) { |
| 90 | j.mu.Lock() |
| 91 | defer j.mu.Unlock() |
| 92 | return append([]string(nil), j.Steps...), j.Done, j.Err |
| 93 | } |
| 94 | |
| 95 | func (j *systemJob) result() any { |
| 96 | j.mu.Lock() |
| 97 | defer j.mu.Unlock() |
| 98 | return j.Result |
| 99 | } |
| 100 | |
| 101 | // newSystemJob mints and records a job with no channel slot: for work whose |
| 102 | // own lock does the serialising (the geoip publish slot). |
| 103 | func (s *Server) newSystemJob() (string, *systemJob) { |
| 104 | s.systemMu.Lock() |
| 105 | defer s.systemMu.Unlock() |
| 106 | b := make([]byte, 8) |
| 107 | rand.Read(b) |
| 108 | id := hex.EncodeToString(b) |
| 109 | job := &systemJob{at: time.Now()} |
| 110 | if s.systemJobs == nil { |
| 111 | s.systemJobs = map[string]*systemJob{} |
| 112 | } |
| 113 | for k, j := range s.systemJobs { |
| 114 | if j.done() && time.Since(j.at) > systemJobTTL { |
| 115 | delete(s.systemJobs, k) |
| 116 | } |
| 117 | } |
| 118 | s.systemJobs[id] = job |
| 119 | return id, job |
| 120 | } |
| 121 | |
| 122 | // systemUpload is the one parked local build: `carlos system update --file` |
| 123 | // PUTs the bytes, then names the handle on the update request. There is no |
| 124 | // timestamp on it and no TTL sweep, because there is only ever ONE — the next |
| 125 | // upload evicts it — so the bound on what a forgotten one costs is the upload |
| 126 | // cap, not the age of the slot. |
| 127 | type systemUpload struct { |
| 128 | version, platform string |
| 129 | data []byte |
| 130 | } |
| 131 | |
| 132 | // apiSystemRoller is the gate every door in this file opens with: a person |
| 133 | // who is either an OWNER of the ops account or on SystemOperators |
| 134 | // (CARLOS_SYSTEM_OPERATORS, discussion #176). owner reports which, because |
| 135 | // the upload door and an update naming an upload are the owner's alone |
| 136 | // (SystemOperators' doc comment, server.go). |
| 137 | // |
| 138 | // A refusal is a bare 404, the same answer a nonexistent path gives, because |
| 139 | // whether this deployment has a system surface at all is not something a |
| 140 | // member needs to learn. The owner check comes first and a store error is |
| 141 | // not the end: a listed operator must still get in on a console whose role |
| 142 | // lookup failed, and a store error must never be what admits anyone. |
| 143 | func (s *Server) apiSystemRoller(w http.ResponseWriter, r *http.Request) (addr string, owner, ok bool) { |
| 144 | p, ok := principalFrom(r.Context()) |
| 145 | if !ok || p.Kind != PrincipalUser { |
| 146 | http.NotFound(w, r) |
| 147 | return "", false, false |
| 148 | } |
| 149 | role, err := s.store.Role(OpsAccountID, p.Address) |
| 150 | if err != nil { |
| 151 | log.Printf("console: api system: role: %v", err) |
| 152 | } |
| 153 | if err == nil && role == RoleOwner { |
| 154 | return p.Address, true, true |
| 155 | } |
| 156 | if s.systemOperator(p.Address) { |
| 157 | return p.Address, false, true |
| 158 | } |
| 159 | http.NotFound(w, r) |
| 160 | return "", false, false |
| 161 | } |
| 162 | |
| 163 | // systemOperator reports whether addr is on SystemOperators. It is |
| 164 | // relayOperator's exact-match rule (apirelay.go): the configured entry is |
| 165 | // trimmed and lowercased, the signed-in address is compared as it is, and |
| 166 | // never with EqualFold, whose Unicode folding admits homoglyphs. |
| 167 | func (s *Server) systemOperator(addr string) bool { |
| 168 | for _, op := range s.cfg.SystemOperators { |
| 169 | if strings.ToLower(strings.TrimSpace(op)) == addr { |
| 170 | return true |
| 171 | } |
| 172 | } |
| 173 | return false |
| 174 | } |
| 175 | |
| 176 | // auditSystem records an operator's platform action against the OPS account. |
| 177 | // |
| 178 | // It exists instead of s.audit because that one takes an *Account and |
| 179 | // dereferences it: these actions have no tenant account behind them — their |
| 180 | // subject is the deployment itself — so there is no *Account to pass and a |
| 181 | // nil would panic. The ops account is the right home for the row: its owners |
| 182 | // and the deployment's system operators (SystemOperators) are the only |
| 183 | // callers that ever reach these doors, and the row names which person it was. |
| 184 | // |
| 185 | // KNOW WHAT THIS IS AND IS NOT. These rows are keyed (OpsAccountID, app |
| 186 | // "platform"), and NO SURFACE READS THEM TODAY: Store.Events is asked for a |
| 187 | // real app name by the activity tab, and "platform" is not one. That is |
| 188 | // deliberate, not an oversight — the row is the DURABLE record of who moved |
| 189 | // the platform pointer and when, written where a reader can be added later |
| 190 | // (an operator activity view, `carlos system log`) without a migration. |
| 191 | // Until such a reader exists this is a write-only table, and the answer to |
| 192 | // "who promoted this?" comes from the pointer's own Audit list, which |
| 193 | // PromoteSystem signs into the bucket. Do not delete these rows on the |
| 194 | // grounds that nothing reads them; do not present them as an audit trail an |
| 195 | // operator can consult, because they cannot. |
| 196 | func (s *Server) auditSystem(addr, action, detail string) { |
| 197 | if err := s.store.RecordEvent(OpsAccountID, "platform", addr, action, detail, time.Now()); err != nil { |
| 198 | log.Printf("console: audit %s platform: %v", action, err) |
| 199 | } |
| 200 | } |
| 201 | |
| 202 | // systemBucket refuses on a console with no deployment bucket. Config.Releases |
| 203 | // == nil is a legal shape (see ensureAccountManifest and bumpSerial, which |
| 204 | // both guard for it), and every door here reads or writes the bucket — the |
| 205 | // pointer, the mirrored artifacts, the box heartbeats — so without this the |
| 206 | // answer to an ops owner on such a console is a nil-interface panic rather |
| 207 | // than a sentence. |
| 208 | func (s *Server) systemBucket(w http.ResponseWriter) bool { |
| 209 | if s.cfg.Releases == nil { |
| 210 | jsonError(w, http.StatusServiceUnavailable, |
| 211 | "this console serves no deployment bucket, so there is no platform pointer to read or move") |
| 212 | return false |
| 213 | } |
| 214 | return true |
| 215 | } |
| 216 | |
| 217 | // systemFeed is the release feed this console fetches from: cfg.SystemFeed |
| 218 | // when it is configured, the vendor feed otherwise. |
| 219 | // |
| 220 | // "Configured" means ALL THREE strings, not just one. A feed's three fields |
| 221 | // are a single fact — where to ask, where to download, and the key the |
| 222 | // checksum list must verify under — and any subset of them is not a feed at |
| 223 | // all: a DownloadBase with the vendor's PubKey would fetch one publisher's |
| 224 | // binaries and check them against another's signature, which cannot ever |
| 225 | // succeed and would read as "the release is corrupt" rather than as "this |
| 226 | // console is misconfigured". So a partial override is refused OUT LOUD and |
| 227 | // falls back, rather than being half-honoured. |
| 228 | func (s *Server) systemFeed() publicrelease.Feed { |
| 229 | f := s.cfg.SystemFeed |
| 230 | set := 0 |
| 231 | for _, v := range []string{f.LatestURL, f.DownloadBase, f.PubKey} { |
| 232 | if v != "" { |
| 233 | set++ |
| 234 | } |
| 235 | } |
| 236 | switch set { |
| 237 | case 0: |
| 238 | return publicrelease.Default() |
| 239 | case 3: |
| 240 | return f |
| 241 | default: |
| 242 | log.Printf("console: SystemFeed is only partly configured (latest_url=%t download_base=%t pub_key=%t) — "+ |
| 243 | "a feed is all three or none, so this console is using the vendor feed instead", |
| 244 | f.LatestURL != "", f.DownloadBase != "", f.PubKey != "") |
| 245 | return publicrelease.Default() |
| 246 | } |
| 247 | } |
| 248 | |
| 249 | // systemBoxes reads every heartbeat: the platform prefix and each fleet's. |
| 250 | // |
| 251 | // strict is what a fleet the console cannot enumerate MEANS, and the two |
| 252 | // callers need opposite answers. The promote (systemPlatforms) decides from |
| 253 | // this list which platforms to mirror, and a fleet it cannot see is a fleet |
| 254 | // whose architecture it would silently omit — so strict refuses rather than |
| 255 | // guesses. The status read only DESCRIBES, and an operator asking what is |
| 256 | // running is better served by the boxes that did answer plus an honest |
| 257 | // "partial" flag than by a blank page. The second return is that flag: true |
| 258 | // means some fleet's boxes are missing from the slice. |
| 259 | // |
| 260 | // NEITHER answer applies to a console with NO REGISTRY. cfg.Registry == "" |
| 261 | // and a registry file that has not been created yet are legal, ordinary |
| 262 | // shapes — a console before its first box, or one that serves no route |
| 263 | // registry at all — and they mean this deployment has no fleets, full stop. |
| 264 | // Treating them as "unreadable" would refuse every promote such a console |
| 265 | // could ever make and pin its status at partial forever, which is what the |
| 266 | // first cut of this function did. Only a registry that EXISTS and still |
| 267 | // cannot be opened, or one whose fleets cannot be listed, is a fleet set this |
| 268 | // console cannot see. |
| 269 | func (s *Server) systemBoxes(ctx context.Context, strict bool) ([]status.Box, bool, error) { |
| 270 | boxes, err := status.LoadBoxes(ctx, s.cfg.Releases) |
| 271 | if err != nil { |
| 272 | return nil, false, err |
| 273 | } |
| 274 | if s.cfg.Registry == "" { |
| 275 | return boxes, false, nil // this console serves no route registry: no fleets |
| 276 | } |
| 277 | // openPassRegistry reports the not-yet-created case by wrapping os.Stat's |
| 278 | // error, so errors.Is reaches it; the unconfigured case is the guard |
| 279 | // above, since that one is a plain message. Distinguished here rather |
| 280 | // than in openPassRegistry so its other callers keep the answers they |
| 281 | // already expect. |
| 282 | reg, err := s.openPassRegistry() |
| 283 | switch { |
| 284 | case errors.Is(err, os.ErrNotExist): |
| 285 | return boxes, false, nil // no registry file yet: no fleets |
| 286 | case err != nil: |
| 287 | return s.systemFleetsUnreadable(boxes, strict, err) |
| 288 | } |
| 289 | fleets, err := reg.ListFleets() |
| 290 | if err != nil { |
| 291 | return s.systemFleetsUnreadable(boxes, strict, err) |
| 292 | } |
| 293 | partial := false |
| 294 | for _, f := range fleets { |
| 295 | fb, err := status.LoadBoxes(ctx, release.NewPrefixStore(s.cfg.Releases, "fleets/"+f.Name+"/")) |
| 296 | if err != nil { |
| 297 | if strict { |
| 298 | return nil, false, fmt.Errorf("fleet %s: cannot read its box heartbeats, refusing to guess which platforms to mirror: %w", f.Name, err) |
| 299 | } |
| 300 | log.Printf("console: api system: fleet %s boxes: %v", f.Name, err) |
| 301 | partial = true |
| 302 | continue |
| 303 | } |
| 304 | boxes = append(boxes, fb...) |
| 305 | } |
| 306 | return boxes, partial, nil |
| 307 | } |
| 308 | |
| 309 | // systemFleetsUnreadable is the shared answer to "this console cannot say |
| 310 | // what fleets exist": a refusal under strict, a logged degradation otherwise. |
| 311 | func (s *Server) systemFleetsUnreadable(boxes []status.Box, strict bool, err error) ([]status.Box, bool, error) { |
| 312 | if strict { |
| 313 | return nil, false, fmt.Errorf("cannot list fleets, refusing to guess which platforms to mirror: %w", err) |
| 314 | } |
| 315 | log.Printf("console: api system: list fleets: %v — reporting the platform's own boxes only", err) |
| 316 | return boxes, true, nil |
| 317 | } |
| 318 | |
| 319 | type systemUpdateRequest struct { |
| 320 | Component string `json:"component"` |
| 321 | Channel string `json:"channel"` |
| 322 | Version string `json:"version"` |
| 323 | Upload string `json:"upload"` |
| 324 | Spread string `json:"spread"` |
| 325 | } |
| 326 | |
| 327 | // handleAPISystemUpdate starts the promote and answers with a job id. The |
| 328 | // work happens off the request because ingesting a release is minutes of |
| 329 | // downloading and mirroring, and an operator watching a spinner over a |
| 330 | // connection that may drop is not a place to hold it. |
| 331 | func (s *Server) handleAPISystemUpdate(w http.ResponseWriter, r *http.Request) { |
| 332 | addr, owner, ok := s.apiSystemRoller(w, r) |
| 333 | if !ok { |
| 334 | return |
| 335 | } |
| 336 | if !s.systemBucket(w) { |
| 337 | return |
| 338 | } |
| 339 | if s.cfg.Key == nil { |
| 340 | jsonError(w, http.StatusServiceUnavailable, "no signing key on this host — the console is running read-only") |
| 341 | return |
| 342 | } |
| 343 | var in systemUpdateRequest |
| 344 | if !readJSONStrictLimit(w, r, &in, 4<<10) { |
| 345 | return |
| 346 | } |
| 347 | // An upload handle is refused BEFORE a job starts: the job is what |
| 348 | // consumes the handle, so a check inside it would let an operator spend |
| 349 | // an owner's parked build on a refused request. |
| 350 | if in.Upload != "" && !owner { |
| 351 | systemUploadOwnerOnly(w) |
| 352 | return |
| 353 | } |
| 354 | if in.Component == "" { |
| 355 | in.Component = release.SystemComponentRouter |
| 356 | } |
| 357 | if in.Component != release.SystemComponentRouter { |
| 358 | jsonError(w, http.StatusBadRequest, "unknown component "+in.Component) |
| 359 | return |
| 360 | } |
| 361 | if in.Channel != release.RungCanary && in.Channel != release.RungStable { |
| 362 | jsonError(w, http.StatusBadRequest, "channel must be canary or stable") |
| 363 | return |
| 364 | } |
| 365 | spread := 5 * time.Minute |
| 366 | if in.Spread != "" { |
| 367 | d, err := time.ParseDuration(in.Spread) |
| 368 | if err != nil || d < 0 || d > 24*time.Hour { |
| 369 | jsonError(w, http.StatusBadRequest, "spread must be a duration between 0 and 24h") |
| 370 | return |
| 371 | } |
| 372 | spread = d |
| 373 | } |
| 374 | // A named release below the layout floor is refused here, before a |
| 375 | // job starts, while any box reports a nested, fenced or conflict row |
| 376 | // or no layout at all (layoutguard.go). Only a named version can be |
| 377 | // judged: "the latest public release" is resolved inside the job and |
| 378 | // is never below the floor, since the floor is a past release. |
| 379 | if in.Version != "" && in.Upload == "" { |
| 380 | if reason := s.layoutDowngradeBlocked(r.Context(), in.Version); reason != "" { |
| 381 | jsonError(w, http.StatusConflict, reason) |
| 382 | return |
| 383 | } |
| 384 | } |
| 385 | id, job, busy := s.startSystemJob(in.Component, in.Channel) |
| 386 | if busy != "" { |
| 387 | writeJSON(w, http.StatusConflict, map[string]string{ |
| 388 | "error": "a promote of " + in.Component + "/" + in.Channel + " is already running; watch it or wait for it to finish", |
| 389 | "job": busy, |
| 390 | }) |
| 391 | return |
| 392 | } |
| 393 | go s.runSystemUpdate(context.WithoutCancel(r.Context()), id, job, addr, in, spread) |
| 394 | writeJSON(w, http.StatusOK, map[string]string{"job": id}) |
| 395 | } |
| 396 | |
| 397 | // startSystemJob claims the ONE in-flight promote slot for a channel and |
| 398 | // mints the job that holds it, or reports the id of the job already there. |
| 399 | // |
| 400 | // One at a time per channel, and the reason is memory as much as |
| 401 | // correctness: a promote holds a whole release asset in the heap while it |
| 402 | // verifies and mirrors it, so N concurrent promotes are N copies, and the |
| 403 | // door is reachable by an ops owner with a script. Serializing per channel |
| 404 | // also removes the CAS race two promotes of different versions into the same |
| 405 | // channel would otherwise run — one of them would lose at PromoteSystem's |
| 406 | // Put and report a conflict the operator did not cause. |
| 407 | // |
| 408 | // It sweeps FINISHED jobs past the TTL on the way past. Keyed on Done rather |
| 409 | // than age: a promote that has legitimately been running for over an hour |
| 410 | // (a slow feed, a large asset) must not have its record swept out from under |
| 411 | // the operator watching it, which is exactly the moment they need it. |
| 412 | func (s *Server) startSystemJob(component, channel string) (id string, job *systemJob, busy string) { |
| 413 | s.systemMu.Lock() |
| 414 | defer s.systemMu.Unlock() |
| 415 | if running, ok := s.systemRunning[component+"/"+channel]; ok { |
| 416 | return "", nil, running |
| 417 | } |
| 418 | b := make([]byte, 8) |
| 419 | rand.Read(b) |
| 420 | id = hex.EncodeToString(b) |
| 421 | job = &systemJob{at: time.Now()} |
| 422 | if s.systemJobs == nil { |
| 423 | s.systemJobs = map[string]*systemJob{} |
| 424 | } |
| 425 | if s.systemRunning == nil { |
| 426 | s.systemRunning = map[string]string{} |
| 427 | } |
| 428 | for k, j := range s.systemJobs { |
| 429 | if j.done() && time.Since(j.at) > systemJobTTL { |
| 430 | delete(s.systemJobs, k) |
| 431 | } |
| 432 | } |
| 433 | s.systemJobs[id] = job |
| 434 | s.systemRunning[component+"/"+channel] = id |
| 435 | return id, job, "" |
| 436 | } |
| 437 | |
| 438 | // releaseSystemChannel hands the slot back, and only if this job is still the |
| 439 | // one holding it: a compare-and-delete on the id, so a late or repeated |
| 440 | // release can never free a LATER promote's slot and let a third run beside |
| 441 | // it. That makes it idempotent, which is what lets runSystemUpdate both defer |
| 442 | // it (the panic path) and call it explicitly at the point the ordering |
| 443 | // matters. |
| 444 | func (s *Server) releaseSystemChannel(component, channel, id string) { |
| 445 | key := component + "/" + channel |
| 446 | s.systemMu.Lock() |
| 447 | if s.systemRunning[key] == id { |
| 448 | delete(s.systemRunning, key) |
| 449 | } |
| 450 | s.systemMu.Unlock() |
| 451 | } |
| 452 | |
| 453 | // runSystemUpdate is the promote: resolve, ingest (or take the upload), |
| 454 | // mirror, sign, bump. Every step is idempotent, so a lost job is re-run. |
| 455 | func (s *Server) runSystemUpdate(ctx context.Context, id string, job *systemJob, actor string, in systemUpdateRequest, spread time.Duration) { |
| 456 | // Every exit hands the channel back, a panic included — without this a |
| 457 | // crashed promote would wedge its channel for the life of the process. |
| 458 | defer s.releaseSystemChannel(in.Component, in.Channel, id) |
| 459 | err := func() error { |
| 460 | var rel release.SystemRelease |
| 461 | if in.Upload != "" { |
| 462 | s.systemMu.Lock() |
| 463 | up, ok := s.systemUploads[in.Upload] |
| 464 | delete(s.systemUploads, in.Upload) |
| 465 | s.systemMu.Unlock() |
| 466 | if !ok { |
| 467 | return errors.New("upload handle unknown or already consumed") |
| 468 | } |
| 469 | if in.Version != "" && in.Version != up.version { |
| 470 | return fmt.Errorf("upload is %s, request says %s", up.version, in.Version) |
| 471 | } |
| 472 | job.step("using uploaded %s for %s", up.version, up.platform) |
| 473 | sum := sha256.Sum256(up.data) |
| 474 | sha := hex.EncodeToString(sum[:]) |
| 475 | rel = release.SystemRelease{Version: up.version, Source: "file", Artifacts: map[string]release.SystemArtifact{}} |
| 476 | if err := s.mirrorSystemArtifact(ctx, job, up.version, up.platform, up.data, sha); err != nil { |
| 477 | return err |
| 478 | } |
| 479 | rel.Artifacts[up.platform] = release.SystemArtifact{Key: release.SystemArtifactKey(up.version, up.platform), SHA256: sha} |
| 480 | } else { |
| 481 | feed := s.systemFeed() |
| 482 | tag := in.Version |
| 483 | if tag == "" { |
| 484 | var err error |
| 485 | if tag, err = feed.Latest(ctx); err != nil { |
| 486 | return err |
| 487 | } |
| 488 | } |
| 489 | if !publicrelease.TagRe.MatchString(tag) { |
| 490 | return fmt.Errorf("%q is not a release tag; use --file for a local build", tag) |
| 491 | } |
| 492 | job.step("resolved %s", tag) |
| 493 | platforms, err := s.systemPlatforms(ctx) |
| 494 | if err != nil { |
| 495 | return err |
| 496 | } |
| 497 | rel = release.SystemRelease{Version: tag, Source: "release:" + tag, Artifacts: map[string]release.SystemArtifact{}} |
| 498 | for _, plat := range platforms { |
| 499 | data, sha, err := feed.Asset(ctx, tag, publicrelease.AssetName(plat), systemAssetMaxBytes) |
| 500 | if err != nil { |
| 501 | return fmt.Errorf("%s: %w", plat, err) |
| 502 | } |
| 503 | job.step("verified %s (%d bytes)", publicrelease.AssetName(plat), len(data)) |
| 504 | if err := s.mirrorSystemArtifact(ctx, job, tag, plat, data, sha); err != nil { |
| 505 | return err |
| 506 | } |
| 507 | rel.Artifacts[plat] = release.SystemArtifact{Key: release.SystemArtifactKey(tag, plat), SHA256: sha} |
| 508 | } |
| 509 | } |
| 510 | p, err := s.promoteHostUpdate(ctx, rel, in.Component, in.Channel, |
| 511 | release.SystemPromoteOptions{Actor: actor, Spread: spread}) |
| 512 | if err != nil { |
| 513 | return err |
| 514 | } |
| 515 | job.step("pointed %s/%s at %s (spread %s)", in.Component, in.Channel, p.Current.Version, p.Spread) |
| 516 | s.auditSystem(actor, "system-update", "pointed "+in.Component+"/"+in.Channel+" at "+p.Current.Version) |
| 517 | s.bumpSerial(ctx) |
| 518 | job.step("serial bumped: boxes on %s start their spread", in.Channel) |
| 519 | return nil |
| 520 | }() |
| 521 | // And hand it back BEFORE marking the job done, so an operator who polls |
| 522 | // this job to completion and immediately promotes again is never told the |
| 523 | // channel is busy by the job they just watched finish. The deferred |
| 524 | // release above then finds nothing to do; it is there for the panic path. |
| 525 | s.releaseSystemChannel(in.Component, in.Channel, id) |
| 526 | job.finish(err) |
| 527 | } |
| 528 | |
| 529 | // systemPlatforms: the distinct platforms the fleet reports; linux-arm64 |
| 530 | // when nothing has reported the field yet. STRICT — a fleet this console |
| 531 | // cannot enumerate stops the promote, because a platform silently left out |
| 532 | // of the mirror is a fleet of boxes that will never adopt. |
| 533 | func (s *Server) systemPlatforms(ctx context.Context) ([]string, error) { |
| 534 | boxes, _, err := s.systemBoxes(ctx, true) |
| 535 | if err != nil { |
| 536 | return nil, err |
| 537 | } |
| 538 | set := map[string]bool{} |
| 539 | for _, b := range boxes { |
| 540 | if b.Metrics != nil && b.Metrics.Platform != "" { |
| 541 | set[b.Metrics.Platform] = true |
| 542 | } |
| 543 | } |
| 544 | if len(set) == 0 { |
| 545 | set["linux-arm64"] = true |
| 546 | } |
| 547 | var out []string |
| 548 | for p := range set { |
| 549 | out = append(out, p) |
| 550 | } |
| 551 | sort.Strings(out) |
| 552 | return out, nil |
| 553 | } |
| 554 | |
| 555 | // mirrorSystemArtifact: present with the same sha → skip; present with a |
| 556 | // different sha → refuse (versions are immutable, spec §3). |
| 557 | func (s *Server) mirrorSystemArtifact(ctx context.Context, job *systemJob, version, platform string, data []byte, sha string) error { |
| 558 | key := release.SystemArtifactKey(version, platform) |
| 559 | switch already, err := s.systemArtifactMatches(ctx, key, sha); { |
| 560 | case err != nil: |
| 561 | return err |
| 562 | case already: |
| 563 | job.step("%s already mirrored", key) |
| 564 | return nil |
| 565 | } |
| 566 | // Create-only: the empty ifMatch IS the CAS, so a concurrent promote of |
| 567 | // the same version loses rather than overwrites a byte a box may already |
| 568 | // have fetched. Losing that race is not a failure — the two channels' |
| 569 | // promotes of one version want the same bytes at the same key — so |
| 570 | // re-read and let the comparison above decide: identical is done, and |
| 571 | // different is the immutability refusal it always was. |
| 572 | if _, err := s.cfg.Releases.Put(ctx, key, data, ""); err != nil { |
| 573 | if !errors.Is(err, release.ErrCASConflict) { |
| 574 | return err |
| 575 | } |
| 576 | // The re-read answers every half: identical bytes means the other |
| 577 | // promote wrote what this one would have, different bytes comes back |
| 578 | // as the immutability refusal, and absent means the key was taken and |
| 579 | // released while this promote looked away. |
| 580 | already, cerr := s.systemArtifactMatches(ctx, key, sha) |
| 581 | if cerr != nil { |
| 582 | return cerr |
| 583 | } |
| 584 | if !already { |
| 585 | return fmt.Errorf("%s: another writer holds the key but it is not there now — re-run, every step is idempotent", key) |
| 586 | } |
| 587 | job.step("%s mirrored concurrently", key) |
| 588 | return nil |
| 589 | } |
| 590 | job.step("mirrored %s", key) |
| 591 | return nil |
| 592 | } |
| 593 | |
| 594 | // systemArtifactMatches answers "is sha already the object at key?" — |
| 595 | // absent is (false, nil), present and different is the immutability refusal, |
| 596 | // present and equal is (true, nil). |
| 597 | // |
| 598 | // The comparison STREAMS the object already in the bucket through a hasher |
| 599 | // rather than Getting it. A promote already holds one whole asset in the |
| 600 | // heap (the verified download, waiting to be Put); buffering the existing |
| 601 | // object beside it would make the peak two copies of a platform binary per |
| 602 | // in-flight promote, for no purpose beyond computing one sha256. |
| 603 | func (s *Server) systemArtifactMatches(ctx context.Context, key, sha string) (bool, error) { |
| 604 | got, err := s.objectSHA256(ctx, key) |
| 605 | switch { |
| 606 | case errors.Is(err, os.ErrNotExist): |
| 607 | return false, nil |
| 608 | case err != nil: |
| 609 | return false, err |
| 610 | } |
| 611 | if got == sha { |
| 612 | return true, nil |
| 613 | } |
| 614 | return false, fmt.Errorf("%s exists with different bytes — versions are immutable; pick a new version", key) |
| 615 | } |
| 616 | |
| 617 | // objectSHA256 streams the object at key through a hasher; absent is |
| 618 | // os.ErrNotExist. A ctx that ended during the read is its error, even on a |
| 619 | // store that read on regardless. |
| 620 | func (s *Server) objectSHA256(ctx context.Context, key string) (string, error) { |
| 621 | rc, _, err := release.OpenObject(ctx, s.cfg.Releases, key) |
| 622 | if err != nil { |
| 623 | return "", err |
| 624 | } |
| 625 | defer rc.Close() |
| 626 | h := sha256.New() |
| 627 | if _, err := io.Copy(h, rc); err != nil { |
| 628 | return "", fmt.Errorf("%s: reading the object to compare it: %w", key, err) |
| 629 | } |
| 630 | if err := ctx.Err(); err != nil { |
| 631 | return "", fmt.Errorf("%s: reading the object to compare it: %w", key, err) |
| 632 | } |
| 633 | return hex.EncodeToString(h.Sum(nil)), nil |
| 634 | } |
| 635 | |
| 636 | // bootstrapKey is the binary a new, rebuilt or failover box downloads at |
| 637 | // first boot. `carlos system update` mirrors each release under |
| 638 | // bootstrap/router/ and never writes this key, so it falls behind the fleet |
| 639 | // unless an operator copies stable's build over it (issue #316). |
| 640 | const bootstrapKey = "bootstrap/carlos" |
| 641 | |
| 642 | // bootstrapPlatform is the one platform bootstrap/carlos serves. |
| 643 | const bootstrapPlatform = "linux-arm64" |
| 644 | |
| 645 | // bootstrapReadBudget bounds the comparison. It reads the whole binary, and |
| 646 | // the status read it rides (also `carlos system update`'s preview) must |
| 647 | // answer inside the CLI's 30-second timeout on slow storage: past the budget |
| 648 | // the row reads unreadable and the rest of the response stands. |
| 649 | var bootstrapReadBudget = 10 * time.Second |
| 650 | |
| 651 | // bootstrapState is the status row for bootstrapKey against stable's signed |
| 652 | // linux-arm64 artifact, or nil when there is nothing to compare: no verified |
| 653 | // stable pointer, no linux-arm64 artifact in it, or no bootstrap object (a |
| 654 | // deployment that bootstraps another way). A read failure is reported, never |
| 655 | // fatal to the status read. |
| 656 | func (s *Server) bootstrapState(ctx context.Context, stable *release.SystemPointer) map[string]any { |
| 657 | if stable == nil { |
| 658 | return nil |
| 659 | } |
| 660 | art, ok := stable.Current.Artifacts[bootstrapPlatform] |
| 661 | if !ok || art.SHA256 == "" { |
| 662 | return nil |
| 663 | } |
| 664 | row := map[string]any{"version": stable.Current.Version, "key": art.Key} |
| 665 | ctx, cancel := context.WithTimeout(ctx, bootstrapReadBudget) |
| 666 | defer cancel() |
| 667 | got, err := s.objectSHA256(ctx, bootstrapKey) |
| 668 | switch { |
| 669 | case errors.Is(err, os.ErrNotExist): |
| 670 | return nil |
| 671 | case err != nil: |
| 672 | log.Printf("console: api system status: %s: %v", bootstrapKey, err) |
| 673 | row["state"], row["error"] = "unreadable", err.Error() |
| 674 | case strings.EqualFold(got, art.SHA256): |
| 675 | row["state"] = "current" |
| 676 | default: |
| 677 | row["state"] = "behind" |
| 678 | } |
| 679 | return row |
| 680 | } |
| 681 | |
| 682 | func (s *Server) handleAPISystemJob(w http.ResponseWriter, r *http.Request) { |
| 683 | if _, _, ok := s.apiSystemRoller(w, r); !ok { |
| 684 | return |
| 685 | } |
| 686 | s.systemMu.Lock() |
| 687 | job, ok := s.systemJobs[r.PathValue("id")] |
| 688 | s.systemMu.Unlock() |
| 689 | if !ok { |
| 690 | jsonError(w, http.StatusNotFound, "no such job — the console may have restarted; re-run, every step is idempotent") |
| 691 | return |
| 692 | } |
| 693 | steps, done, errMsg := job.snapshot() |
| 694 | out := map[string]any{"steps": steps, "done": done, "error": errMsg} |
| 695 | if res := job.result(); res != nil { |
| 696 | out["result"] = res |
| 697 | } |
| 698 | writeJSON(w, http.StatusOK, out) |
| 699 | } |
| 700 | |
| 701 | // handleAPISystemUpload parks one local build in memory for the job that |
| 702 | // names its handle. It is a streamed multipart read rather than a JSON body |
| 703 | // because the payload is a binary; the meta part comes FIRST so the version |
| 704 | // and platform are known — and refusable — before any bytes are buffered. |
| 705 | func (s *Server) handleAPISystemUpload(w http.ResponseWriter, r *http.Request) { |
| 706 | _, owner, ok := s.apiSystemRoller(w, r) |
| 707 | if !ok { |
| 708 | return |
| 709 | } |
| 710 | if !owner { |
| 711 | systemUploadOwnerOnly(w) |
| 712 | return |
| 713 | } |
| 714 | if !s.systemBucket(w) { |
| 715 | return // refuse before buffering bytes no promote could ever consume |
| 716 | } |
| 717 | r.Body = http.MaxBytesReader(w, r.Body, systemUploadMaxBytes+64<<10) |
| 718 | mr, err := r.MultipartReader() |
| 719 | if err != nil { |
| 720 | jsonError(w, http.StatusBadRequest, "multipart body required") |
| 721 | return |
| 722 | } |
| 723 | var meta struct { |
| 724 | Version string `json:"version"` |
| 725 | Platform string `json:"platform"` |
| 726 | } |
| 727 | first, err := mr.NextPart() |
| 728 | if err != nil || first.FormName() != "meta" { |
| 729 | jsonError(w, http.StatusBadRequest, "first part must be meta") |
| 730 | return |
| 731 | } |
| 732 | if err := json.NewDecoder(io.LimitReader(first, 4<<10)).Decode(&meta); err != nil { |
| 733 | jsonError(w, http.StatusBadRequest, "meta: "+err.Error()) |
| 734 | return |
| 735 | } |
| 736 | if !release.ValidVersion(meta.Version) || meta.Version == "dev" { |
| 737 | jsonError(w, http.StatusBadRequest, "meta.version is required and must be a version") |
| 738 | return |
| 739 | } |
| 740 | if !systemPlatformRe.MatchString(meta.Platform) { |
| 741 | jsonError(w, http.StatusBadRequest, "meta.platform is required (e.g. linux-arm64): the console cannot run a foreign binary to ask it") |
| 742 | return |
| 743 | } |
| 744 | part, err := mr.NextPart() |
| 745 | if err != nil || part.FormName() != "file" { |
| 746 | jsonError(w, http.StatusBadRequest, "second part must be file") |
| 747 | return |
| 748 | } |
| 749 | data, err := io.ReadAll(io.LimitReader(part, systemUploadMaxBytes+1)) |
| 750 | if err != nil { |
| 751 | jsonError(w, http.StatusBadRequest, "file: "+err.Error()) |
| 752 | return |
| 753 | } |
| 754 | if int64(len(data)) > systemUploadMaxBytes { |
| 755 | jsonError(w, http.StatusRequestEntityTooLarge, "file exceeds 128 MiB") |
| 756 | return |
| 757 | } |
| 758 | b := make([]byte, 8) |
| 759 | rand.Read(b) |
| 760 | handle := hex.EncodeToString(b) |
| 761 | s.systemMu.Lock() |
| 762 | if s.systemUploads == nil { |
| 763 | s.systemUploads = map[string]systemUpload{} |
| 764 | } |
| 765 | for k := range s.systemUploads { // one slot: a forgotten upload is not a memory leak |
| 766 | delete(s.systemUploads, k) |
| 767 | } |
| 768 | s.systemUploads[handle] = systemUpload{version: meta.Version, platform: meta.Platform, data: data} |
| 769 | s.systemMu.Unlock() |
| 770 | writeJSON(w, http.StatusOK, map[string]any{"upload": handle, "size": len(data)}) |
| 771 | } |
| 772 | |
| 773 | // systemUploadOwnerOnly is the one refusal a system operator meets: a local |
| 774 | // build is the ops owner's (SystemOperators' doc comment, server.go). A 403 |
| 775 | // rather than the gate's 404, because apiSystemRoller has already admitted |
| 776 | // the caller to these doors, so it teaches them only what THEY lack. |
| 777 | func systemUploadOwnerOnly(w http.ResponseWriter) { |
| 778 | jsonError(w, http.StatusForbidden, |
| 779 | "shipping a local build needs an owner of the ops account. as a system operator you can update to a signed release") |
| 780 | } |
| 781 | |
| 782 | // handleAPISystemRollback points a channel back at what it ran before. It |
| 783 | // answers inline rather than as a job: nothing is fetched or mirrored, |
| 784 | // because the previous release travels whole inside the pointer's history. |
| 785 | func (s *Server) handleAPISystemRollback(w http.ResponseWriter, r *http.Request) { |
| 786 | addr, _, ok := s.apiSystemRoller(w, r) |
| 787 | if !ok { |
| 788 | return |
| 789 | } |
| 790 | if !s.systemBucket(w) { |
| 791 | return |
| 792 | } |
| 793 | if s.cfg.Key == nil { |
| 794 | jsonError(w, http.StatusServiceUnavailable, "no signing key on this host — the console is running read-only") |
| 795 | return |
| 796 | } |
| 797 | var in struct { |
| 798 | Component string `json:"component"` |
| 799 | Channel string `json:"channel"` |
| 800 | } |
| 801 | if !readJSONStrictLimit(w, r, &in, 1<<10) { |
| 802 | return |
| 803 | } |
| 804 | // The same two checks the update door makes, and for the same reason: |
| 805 | // both arguments become part of an object key, and a door that validates |
| 806 | // what it writes but not what it rewinds is a door with two standards. |
| 807 | if in.Component == "" { |
| 808 | in.Component = release.SystemComponentRouter |
| 809 | } |
| 810 | if in.Component != release.SystemComponentRouter { |
| 811 | jsonError(w, http.StatusBadRequest, "unknown component "+in.Component) |
| 812 | return |
| 813 | } |
| 814 | if in.Channel != release.RungCanary && in.Channel != release.RungStable { |
| 815 | jsonError(w, http.StatusBadRequest, "channel must be canary or stable") |
| 816 | return |
| 817 | } |
| 818 | // The rollback target is the pointer's History[0], exactly what |
| 819 | // RollbackSystem will move to; it is read first so the layout guard |
| 820 | // can judge it before the pointer moves (design §8.2, "The console's |
| 821 | // rollback door"). A pointer that cannot be read here is left for |
| 822 | // RollbackSystem to report in its own words below. |
| 823 | if cur, _, err := release.GetSystemPointer(r.Context(), s.cfg.Releases, s.cfg.ReleasePub, in.Component, in.Channel); err == nil && len(cur.History) > 0 { |
| 824 | if reason := s.layoutDowngradeBlocked(r.Context(), cur.History[0].Version); reason != "" { |
| 825 | jsonError(w, http.StatusConflict, reason) |
| 826 | return |
| 827 | } |
| 828 | } |
| 829 | p, err := s.rollbackHostUpdate(r.Context(), in.Component, in.Channel, addr) |
| 830 | var held activationHeldError |
| 831 | switch { |
| 832 | case errors.Is(err, release.ErrCASConflict): |
| 833 | // M-6: the pointer moved between the read and the conditional |
| 834 | // write. No retry: a rollback pops history, so repeating it against |
| 835 | // a pointer the operator never saw could go back one release too far. |
| 836 | jsonError(w, http.StatusConflict, "the "+in.Channel+" pointer moved while this rollback was being prepared, so nothing was rolled back. Check carlos system status, then run the rollback again if it is still wanted") |
| 837 | return |
| 838 | case errors.As(err, &held): |
| 839 | jsonError(w, http.StatusConflict, held.Error()) |
| 840 | return |
| 841 | case errors.Is(err, release.ErrNoSystemHistory): |
| 842 | jsonError(w, http.StatusConflict, err.Error()) |
| 843 | return |
| 844 | case errors.Is(err, os.ErrNotExist): |
| 845 | jsonError(w, http.StatusNotFound, "no "+in.Channel+" pointer to roll back") |
| 846 | return |
| 847 | case err != nil: |
| 848 | jsonError(w, http.StatusInternalServerError, err.Error()) |
| 849 | return |
| 850 | } |
| 851 | s.auditSystem(addr, "system-rollback", "pointed "+in.Component+"/"+in.Channel+" back at "+p.Current.Version) |
| 852 | s.bumpSerial(r.Context()) |
| 853 | writeJSON(w, http.StatusOK, map[string]string{"channel": in.Channel, "version": p.Current.Version}) |
| 854 | } |
| 855 | |
| 856 | // handleAPISystemStatus is the whole picture in one read: what each channel |
| 857 | // points at, and what every box actually runs against it. |
| 858 | func (s *Server) handleAPISystemStatus(w http.ResponseWriter, r *http.Request) { |
| 859 | if _, _, ok := s.apiSystemRoller(w, r); !ok { |
| 860 | return |
| 861 | } |
| 862 | if !s.systemBucket(w) { |
| 863 | return |
| 864 | } |
| 865 | ctx := r.Context() |
| 866 | pointers := map[string]any{} |
| 867 | targets := map[string]string{} |
| 868 | var stable *release.SystemPointer |
| 869 | for _, ch := range []string{release.RungCanary, release.RungStable} { |
| 870 | p, _, err := release.GetSystemPointer(ctx, s.cfg.Releases, s.cfg.ReleasePub, release.SystemComponentRouter, ch) |
| 871 | switch { |
| 872 | case errors.Is(err, os.ErrNotExist): |
| 873 | // null means exactly one thing: this channel has never been |
| 874 | // promoted. It must not also mean "there IS a pointer and this |
| 875 | // console cannot verify it" — a tampered or wrongly-signed |
| 876 | // pointer reading as "never promoted" is the one failure an |
| 877 | // operator most needs to see, shown as the one thing they would |
| 878 | // ignore. |
| 879 | pointers[ch] = nil |
| 880 | case err != nil: |
| 881 | log.Printf("console: api system status: %s pointer: %v", ch, err) |
| 882 | pointers[ch] = map[string]any{"error": err.Error()} |
| 883 | default: |
| 884 | pointers[ch] = map[string]any{"version": p.Current.Version, "promoted_at": p.PromotedAt, "spread": p.Spread, "source": p.Current.Source, |
| 885 | "history": len(p.History)} |
| 886 | targets[ch] = p.Current.Version |
| 887 | if ch == release.RungStable { |
| 888 | stable = p |
| 889 | } |
| 890 | } |
| 891 | } |
| 892 | boxes, partial, err := s.systemBoxes(ctx, false) |
| 893 | if err != nil { |
| 894 | jsonError(w, http.StatusBadGateway, "box heartbeats unreadable: "+err.Error()) |
| 895 | return |
| 896 | } |
| 897 | now := time.Now() |
| 898 | rows := make([]map[string]any, 0, len(boxes)) |
| 899 | for _, b := range boxes { |
| 900 | row := map[string]any{"label": b.Label, "region": b.Region, "stale": status.BoxStale(b, now, status.SnapshotPeriod), |
| 901 | "platform": "", "channel": "", "release": "", "target": "", "system": nil} |
| 902 | if b.Metrics != nil { |
| 903 | ch := b.Metrics.SystemChannel |
| 904 | if ch == "" { |
| 905 | ch = release.RungStable |
| 906 | } |
| 907 | row["platform"], row["channel"], row["release"] = b.Metrics.Platform, ch, b.Metrics.Release |
| 908 | row["target"] = targets[ch] |
| 909 | if b.Metrics.System != nil { |
| 910 | row["system"] = b.Metrics.System |
| 911 | } |
| 912 | } |
| 913 | rows = append(rows, row) |
| 914 | } |
| 915 | sort.Slice(rows, func(i, j int) bool { return rows[i]["label"].(string) < rows[j]["label"].(string) }) |
| 916 | // partial says the box list is INCOMPLETE — some fleet's heartbeats |
| 917 | // could not be read. Always present, so a reader never has to tell a |
| 918 | // missing key from a false one, and so "no boxes on that fleet" and |
| 919 | // "could not ask that fleet" are different answers here. |
| 920 | // |
| 921 | // bootstrap is null when there is nothing to compare (bootstrapState). |
| 922 | writeJSON(w, http.StatusOK, map[string]any{"pointers": pointers, "boxes": rows, "partial": partial, |
| 923 | "bootstrap": s.bootstrapState(ctx, stable)}) |
| 924 | } |
| 925 | |
amadan