amadan

carlos / platform Public

console: the carlos system doors — status, update job, upload, rollback

Ops-account owners on a person session, bare 404 otherwise. The
update is a job: resolve, ingest from the signed public feed (or an
upload), mirror per reported platform, sign the pointer, bump the
serial. Versions are immutable in the bucket.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015QCUCpfxbCyEAVQ2m51dH5
Paul Campbell pushed by paul@keymail.dev 6758dde170096eae82d7fc04834ca3da0e729835 parent 1c33565
5 files changed, +855 −4
  • internal/console/apisystem.go +527 −0
  • internal/console/apisystem_test.go +273 −0
  • internal/console/authcli.go +16 −0
  • internal/console/server.go +14 −0
  • internal/console/svcroutes_test.go +25 −4
diff --git a/internal/console/apisystem.go b/internal/console/apisystem.go
new file mode 100644
index 000000000..a99cc3f89
--- /dev/null
+++ b/internal/console/apisystem.go
@@ -0,0 +1,527 @@
+package console
+
+import (
+ "context"
+ "crypto/rand"
+ "crypto/sha256"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "log"
+ "net/http"
+ "os"
+ "sort"
+ "sync"
+ "time"
+
+ "github.com/carlosframework/platform/internal/publicrelease"
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/status"
+)
+
+// The console half of `carlos system` (spec §5). Every door here is for an
+// OWNER of the ops account on a PERSON session, checked on every call;
+// refusal is a bare 404 like the operator doors, so a non-owner learns
+// nothing. The console holds the bucket, the signing key and the egress;
+// the operator's laptop holds none of them.
+
+const (
+ systemUploadMaxBytes = int64(128 << 20)
+ systemAssetMaxBytes = int64(256 << 20)
+ systemJobTTL = time.Hour
+)
+
+// systemPlatformRe validates a "<goos>-<goarch>" the operator typed before it
+// becomes part of an object key — the same shape and the same reason as
+// apicli.go's targetRe, which is why it IS that expression rather than a
+// second copy of it that could drift.
+var systemPlatformRe = targetRe
+
+// systemJob is one promote in flight. Steps is the narration an operator
+// polls; Err is set once, at the end. The mutex guards all three against the
+// polling handler.
+type systemJob struct {
+ mu sync.Mutex
+ Steps []string
+ Done bool
+ Err string
+ at time.Time
+}
+
+func (j *systemJob) step(format string, args ...any) {
+ j.mu.Lock()
+ j.Steps = append(j.Steps, fmt.Sprintf(format, args...))
+ j.mu.Unlock()
+}
+
+func (j *systemJob) finish(err error) {
+ j.mu.Lock()
+ j.Done = true
+ if err != nil {
+ j.Err = err.Error()
+ }
+ j.mu.Unlock()
+}
+
+// systemUpload is the one parked local build: `carlos system update --file`
+// PUTs the bytes, then names the handle on the update request. There is no
+// timestamp on it and no TTL sweep, because there is only ever ONE — the next
+// upload evicts it — so the bound on what a forgotten one costs is the upload
+// cap, not the age of the slot.
+type systemUpload struct {
+ version, platform string
+ data []byte
+}
+
+// apiSystemOwner is the gate every door in this file opens with: a person,
+// and an OWNER of the ops account. A refusal is a bare 404 — the same answer
+// a nonexistent path gives — because whether this deployment has a system
+// surface at all is not something a member needs to learn.
+func (s *Server) apiSystemOwner(w http.ResponseWriter, r *http.Request) (string, bool) {
+ p, ok := principalFrom(r.Context())
+ if !ok || p.Kind != PrincipalUser {
+ http.NotFound(w, r)
+ return "", false
+ }
+ role, err := s.store.Role(OpsAccountID, p.Address)
+ if err != nil {
+ log.Printf("console: api system: role: %v", err)
+ http.NotFound(w, r)
+ return "", false
+ }
+ if role != RoleOwner {
+ http.NotFound(w, r)
+ return "", false
+ }
+ return p.Address, true
+}
+
+// auditSystem records an operator's platform action against the OPS account.
+//
+// It exists instead of s.audit because that one takes an *Account and
+// dereferences it: these actions have no tenant account behind them — their
+// subject is the deployment itself — so there is no *Account to pass and a
+// nil would panic. The ops account is the right home for the row: its owners
+// are the only callers that ever reach these doors.
+func (s *Server) auditSystem(addr, action, detail string) {
+ if err := s.store.RecordEvent(OpsAccountID, "platform", addr, action, detail, time.Now()); err != nil {
+ log.Printf("console: audit %s platform: %v", action, err)
+ }
+}
+
+// systemBucket refuses on a console with no deployment bucket. Config.Releases
+// == nil is a legal shape (see ensureAccountManifest and bumpSerial, which
+// both guard for it), and every door here reads or writes the bucket — the
+// pointer, the mirrored artifacts, the box heartbeats — so without this the
+// answer to an ops owner on such a console is a nil-interface panic rather
+// than a sentence.
+func (s *Server) systemBucket(w http.ResponseWriter) bool {
+ if s.cfg.Releases == nil {
+ jsonError(w, http.StatusServiceUnavailable,
+ "this console serves no deployment bucket, so there is no platform pointer to read or move")
+ return false
+ }
+ return true
+}
+
+func (s *Server) systemFeed() publicrelease.Feed {
+ if s.cfg.SystemFeed.DownloadBase == "" {
+ return publicrelease.Default()
+ }
+ return s.cfg.SystemFeed
+}
+
+// systemBoxes reads every heartbeat: the platform prefix and each fleet's.
+func (s *Server) systemBoxes(ctx context.Context) ([]status.Box, error) {
+ boxes, err := status.LoadBoxes(ctx, s.cfg.Releases)
+ if err != nil {
+ return nil, err
+ }
+ reg, err := s.openPassRegistry()
+ if err != nil {
+ return boxes, nil // no registry: platform boxes only
+ }
+ fleets, err := reg.ListFleets()
+ if err != nil {
+ return boxes, nil
+ }
+ for _, f := range fleets {
+ fb, err := status.LoadBoxes(ctx, release.NewPrefixStore(s.cfg.Releases, "fleets/"+f.Name+"/"))
+ if err != nil {
+ log.Printf("console: api system: fleet %s boxes: %v", f.Name, err)
+ continue
+ }
+ boxes = append(boxes, fb...)
+ }
+ return boxes, nil
+}
+
+type systemUpdateRequest struct {
+ Component string `json:"component"`
+ Channel string `json:"channel"`
+ Version string `json:"version"`
+ Upload string `json:"upload"`
+ Spread string `json:"spread"`
+}
+
+// handleAPISystemUpdate starts the promote and answers with a job id. The
+// work happens off the request because ingesting a release is minutes of
+// downloading and mirroring, and an operator watching a spinner over a
+// connection that may drop is not a place to hold it.
+func (s *Server) handleAPISystemUpdate(w http.ResponseWriter, r *http.Request) {
+ addr, ok := s.apiSystemOwner(w, r)
+ if !ok {
+ return
+ }
+ if !s.systemBucket(w) {
+ return
+ }
+ if s.cfg.Key == nil {
+ jsonError(w, http.StatusServiceUnavailable, "no signing key on this host — the console is running read-only")
+ return
+ }
+ var in systemUpdateRequest
+ if !readJSONStrictLimit(w, r, &in, 4<<10) {
+ return
+ }
+ if in.Component == "" {
+ in.Component = release.SystemComponentRouter
+ }
+ if in.Component != release.SystemComponentRouter {
+ jsonError(w, http.StatusBadRequest, "unknown component "+in.Component)
+ return
+ }
+ if in.Channel != release.RungCanary && in.Channel != release.RungStable {
+ jsonError(w, http.StatusBadRequest, "channel must be canary or stable")
+ return
+ }
+ spread := 5 * time.Minute
+ if in.Spread != "" {
+ d, err := time.ParseDuration(in.Spread)
+ if err != nil || d < 0 || d > 24*time.Hour {
+ jsonError(w, http.StatusBadRequest, "spread must be a duration between 0 and 24h")
+ return
+ }
+ spread = d
+ }
+ id, job := s.newSystemJob()
+ go s.runSystemUpdate(context.WithoutCancel(r.Context()), job, addr, in, spread)
+ writeJSON(w, http.StatusOK, map[string]string{"job": id})
+}
+
+// newSystemJob mints a job id and its record, sweeping jobs older than the
+// TTL on the way past — the registry is process memory, and an operator who
+// promoted an hour ago is not still polling.
+func (s *Server) newSystemJob() (string, *systemJob) {
+ b := make([]byte, 8)
+ rand.Read(b)
+ id := hex.EncodeToString(b)
+ job := &systemJob{at: time.Now()}
+ s.systemMu.Lock()
+ defer s.systemMu.Unlock()
+ if s.systemJobs == nil {
+ s.systemJobs = map[string]*systemJob{}
+ }
+ for k, j := range s.systemJobs {
+ if time.Since(j.at) > systemJobTTL {
+ delete(s.systemJobs, k)
+ }
+ }
+ s.systemJobs[id] = job
+ return id, job
+}
+
+// runSystemUpdate is the promote: resolve, ingest (or take the upload),
+// mirror, sign, bump. Every step is idempotent, so a lost job is re-run.
+func (s *Server) runSystemUpdate(ctx context.Context, job *systemJob, actor string, in systemUpdateRequest, spread time.Duration) {
+ err := func() error {
+ var rel release.SystemRelease
+ if in.Upload != "" {
+ s.systemMu.Lock()
+ up, ok := s.systemUploads[in.Upload]
+ delete(s.systemUploads, in.Upload)
+ s.systemMu.Unlock()
+ if !ok {
+ return errors.New("upload handle unknown or already consumed")
+ }
+ if in.Version != "" && in.Version != up.version {
+ return fmt.Errorf("upload is %s, request says %s", up.version, in.Version)
+ }
+ job.step("using uploaded %s for %s", up.version, up.platform)
+ sum := sha256.Sum256(up.data)
+ sha := hex.EncodeToString(sum[:])
+ rel = release.SystemRelease{Version: up.version, Source: "file", Artifacts: map[string]release.SystemArtifact{}}
+ if err := s.mirrorSystemArtifact(ctx, job, up.version, up.platform, up.data, sha); err != nil {
+ return err
+ }
+ rel.Artifacts[up.platform] = release.SystemArtifact{Key: release.SystemArtifactKey(up.version, up.platform), SHA256: sha}
+ } else {
+ feed := s.systemFeed()
+ tag := in.Version
+ if tag == "" {
+ var err error
+ if tag, err = feed.Latest(ctx); err != nil {
+ return err
+ }
+ }
+ if !publicrelease.TagRe.MatchString(tag) {
+ return fmt.Errorf("%q is not a release tag; use --file for a local build", tag)
+ }
+ job.step("resolved %s", tag)
+ platforms, err := s.systemPlatforms(ctx)
+ if err != nil {
+ return err
+ }
+ rel = release.SystemRelease{Version: tag, Source: "release:" + tag, Artifacts: map[string]release.SystemArtifact{}}
+ for _, plat := range platforms {
+ data, sha, err := feed.Asset(ctx, tag, publicrelease.AssetName(plat), systemAssetMaxBytes)
+ if err != nil {
+ return fmt.Errorf("%s: %w", plat, err)
+ }
+ job.step("verified %s (%d bytes)", publicrelease.AssetName(plat), len(data))
+ if err := s.mirrorSystemArtifact(ctx, job, tag, plat, data, sha); err != nil {
+ return err
+ }
+ rel.Artifacts[plat] = release.SystemArtifact{Key: release.SystemArtifactKey(tag, plat), SHA256: sha}
+ }
+ }
+ p, err := release.PromoteSystem(ctx, s.cfg.Releases, release.RootSigner(s.cfg.Key), in.Component, in.Channel, rel,
+ release.SystemPromoteOptions{Actor: actor, Spread: spread})
+ if err != nil {
+ return err
+ }
+ job.step("pointed %s/%s at %s (spread %s)", in.Component, in.Channel, p.Current.Version, p.Spread)
+ s.auditSystem(actor, "system-update", "pointed "+in.Component+"/"+in.Channel+" at "+p.Current.Version)
+ s.bumpSerial(ctx)
+ job.step("serial bumped: boxes on %s start their spread", in.Channel)
+ return nil
+ }()
+ job.finish(err)
+}
+
+// systemPlatforms: the distinct platforms the fleet reports; linux-arm64
+// when nothing has reported the field yet.
+func (s *Server) systemPlatforms(ctx context.Context) ([]string, error) {
+ boxes, err := s.systemBoxes(ctx)
+ if err != nil {
+ return nil, err
+ }
+ set := map[string]bool{}
+ for _, b := range boxes {
+ if b.Metrics != nil && b.Metrics.Platform != "" {
+ set[b.Metrics.Platform] = true
+ }
+ }
+ if len(set) == 0 {
+ set["linux-arm64"] = true
+ }
+ var out []string
+ for p := range set {
+ out = append(out, p)
+ }
+ sort.Strings(out)
+ return out, nil
+}
+
+// mirrorSystemArtifact: present with the same sha → skip; present with a
+// different sha → refuse (versions are immutable, spec §3).
+func (s *Server) mirrorSystemArtifact(ctx context.Context, job *systemJob, version, platform string, data []byte, sha string) error {
+ key := release.SystemArtifactKey(version, platform)
+ existing, _, err := s.cfg.Releases.Get(ctx, key)
+ switch {
+ case err == nil:
+ sum := sha256.Sum256(existing)
+ if hex.EncodeToString(sum[:]) == sha {
+ job.step("%s already mirrored", key)
+ return nil
+ }
+ return fmt.Errorf("%s exists with different bytes — versions are immutable; pick a new version", key)
+ case errors.Is(err, os.ErrNotExist):
+ default:
+ return err
+ }
+ // Create-only: the empty ifMatch IS the CAS, so a concurrent promote of
+ // the same version loses rather than overwrites a byte a box may already
+ // have fetched.
+ if _, err := s.cfg.Releases.Put(ctx, key, data, ""); err != nil {
+ return err
+ }
+ job.step("mirrored %s", key)
+ return nil
+}
+
+func (s *Server) handleAPISystemJob(w http.ResponseWriter, r *http.Request) {
+ if _, ok := s.apiSystemOwner(w, r); !ok {
+ return
+ }
+ s.systemMu.Lock()
+ job, ok := s.systemJobs[r.PathValue("id")]
+ s.systemMu.Unlock()
+ if !ok {
+ jsonError(w, http.StatusNotFound, "no such job — the console may have restarted; re-run, every step is idempotent")
+ return
+ }
+ job.mu.Lock()
+ defer job.mu.Unlock()
+ writeJSON(w, http.StatusOK, map[string]any{"steps": append([]string(nil), job.Steps...), "done": job.Done, "error": job.Err})
+}
+
+// handleAPISystemUpload parks one local build in memory for the job that
+// names its handle. It is a streamed multipart read rather than a JSON body
+// because the payload is a binary; the meta part comes FIRST so the version
+// and platform are known — and refusable — before any bytes are buffered.
+func (s *Server) handleAPISystemUpload(w http.ResponseWriter, r *http.Request) {
+ if _, ok := s.apiSystemOwner(w, r); !ok {
+ return
+ }
+ if !s.systemBucket(w) {
+ return // refuse before buffering bytes no promote could ever consume
+ }
+ r.Body = http.MaxBytesReader(w, r.Body, systemUploadMaxBytes+64<<10)
+ mr, err := r.MultipartReader()
+ if err != nil {
+ jsonError(w, http.StatusBadRequest, "multipart body required")
+ return
+ }
+ var meta struct {
+ Version string `json:"version"`
+ Platform string `json:"platform"`
+ }
+ first, err := mr.NextPart()
+ if err != nil || first.FormName() != "meta" {
+ jsonError(w, http.StatusBadRequest, "first part must be meta")
+ return
+ }
+ if err := json.NewDecoder(io.LimitReader(first, 4<<10)).Decode(&meta); err != nil {
+ jsonError(w, http.StatusBadRequest, "meta: "+err.Error())
+ return
+ }
+ if !release.ValidVersion(meta.Version) || meta.Version == "dev" {
+ jsonError(w, http.StatusBadRequest, "meta.version is required and must be a version")
+ return
+ }
+ if !systemPlatformRe.MatchString(meta.Platform) {
+ jsonError(w, http.StatusBadRequest, "meta.platform is required (e.g. linux-arm64): the console cannot run a foreign binary to ask it")
+ return
+ }
+ part, err := mr.NextPart()
+ if err != nil || part.FormName() != "file" {
+ jsonError(w, http.StatusBadRequest, "second part must be file")
+ return
+ }
+ data, err := io.ReadAll(io.LimitReader(part, systemUploadMaxBytes+1))
+ if err != nil {
+ jsonError(w, http.StatusBadRequest, "file: "+err.Error())
+ return
+ }
+ if int64(len(data)) > systemUploadMaxBytes {
+ jsonError(w, http.StatusRequestEntityTooLarge, "file exceeds 128 MiB")
+ return
+ }
+ b := make([]byte, 8)
+ rand.Read(b)
+ handle := hex.EncodeToString(b)
+ s.systemMu.Lock()
+ if s.systemUploads == nil {
+ s.systemUploads = map[string]systemUpload{}
+ }
+ for k := range s.systemUploads { // one slot: a forgotten upload is not a memory leak
+ delete(s.systemUploads, k)
+ }
+ s.systemUploads[handle] = systemUpload{version: meta.Version, platform: meta.Platform, data: data}
+ s.systemMu.Unlock()
+ writeJSON(w, http.StatusOK, map[string]any{"upload": handle, "size": len(data)})
+}
+
+// handleAPISystemRollback points a channel back at what it ran before. It
+// answers inline rather than as a job: nothing is fetched or mirrored,
+// because the previous release travels whole inside the pointer's history.
+func (s *Server) handleAPISystemRollback(w http.ResponseWriter, r *http.Request) {
+ addr, ok := s.apiSystemOwner(w, r)
+ if !ok {
+ return
+ }
+ if !s.systemBucket(w) {
+ return
+ }
+ if s.cfg.Key == nil {
+ jsonError(w, http.StatusServiceUnavailable, "no signing key on this host — the console is running read-only")
+ return
+ }
+ var in struct {
+ Component string `json:"component"`
+ Channel string `json:"channel"`
+ }
+ if !readJSONStrictLimit(w, r, &in, 1<<10) {
+ return
+ }
+ if in.Component == "" {
+ in.Component = release.SystemComponentRouter
+ }
+ p, err := release.RollbackSystem(r.Context(), s.cfg.Releases, release.RootSigner(s.cfg.Key), in.Component, in.Channel, addr)
+ switch {
+ case errors.Is(err, release.ErrNoSystemHistory):
+ jsonError(w, http.StatusConflict, err.Error())
+ return
+ case errors.Is(err, os.ErrNotExist):
+ jsonError(w, http.StatusNotFound, "no "+in.Channel+" pointer to roll back")
+ return
+ case err != nil:
+ jsonError(w, http.StatusInternalServerError, err.Error())
+ return
+ }
+ s.auditSystem(addr, "system-rollback", "pointed "+in.Component+"/"+in.Channel+" back at "+p.Current.Version)
+ s.bumpSerial(r.Context())
+ writeJSON(w, http.StatusOK, map[string]string{"channel": in.Channel, "version": p.Current.Version})
+}
+
+// handleAPISystemStatus is the whole picture in one read: what each channel
+// points at, and what every box actually runs against it.
+func (s *Server) handleAPISystemStatus(w http.ResponseWriter, r *http.Request) {
+ if _, ok := s.apiSystemOwner(w, r); !ok {
+ return
+ }
+ if !s.systemBucket(w) {
+ return
+ }
+ ctx := r.Context()
+ pointers := map[string]any{}
+ targets := map[string]string{}
+ for _, ch := range []string{release.RungCanary, release.RungStable} {
+ p, _, err := release.GetSystemPointer(ctx, s.cfg.Releases, s.cfg.ReleasePub, release.SystemComponentRouter, ch)
+ if err != nil {
+ pointers[ch] = nil
+ continue
+ }
+ pointers[ch] = map[string]any{"version": p.Current.Version, "promoted_at": p.PromotedAt, "spread": p.Spread, "source": p.Current.Source,
+ "history": len(p.History)}
+ targets[ch] = p.Current.Version
+ }
+ boxes, err := s.systemBoxes(ctx)
+ if err != nil {
+ jsonError(w, http.StatusBadGateway, "box heartbeats unreadable: "+err.Error())
+ return
+ }
+ now := time.Now()
+ rows := make([]map[string]any, 0, len(boxes))
+ for _, b := range boxes {
+ row := map[string]any{"label": b.Label, "region": b.Region, "stale": status.BoxStale(b, now, status.SnapshotPeriod),
+ "platform": "", "channel": "", "release": "", "target": "", "system": nil}
+ if b.Metrics != nil {
+ ch := b.Metrics.SystemChannel
+ if ch == "" {
+ ch = release.RungStable
+ }
+ row["platform"], row["channel"], row["release"] = b.Metrics.Platform, ch, b.Metrics.Release
+ row["target"] = targets[ch]
+ if b.Metrics.System != nil {
+ row["system"] = b.Metrics.System
+ }
+ }
+ rows = append(rows, row)
+ }
+ sort.Slice(rows, func(i, j int) bool { return rows[i]["label"].(string) < rows[j]["label"].(string) })
+ writeJSON(w, http.StatusOK, map[string]any{"pointers": pointers, "boxes": rows})
+}
diff --git a/internal/console/apisystem_test.go b/internal/console/apisystem_test.go
new file mode 100644
index 000000000..308de5e1e
--- /dev/null
+++ b/internal/console/apisystem_test.go
@@ -0,0 +1,273 @@
+package console
+
+import (
+ "bytes"
+ "context"
+ "crypto/ed25519"
+ "crypto/sha256"
+ "encoding/base64"
+ "encoding/hex"
+ "encoding/json"
+ "fmt"
+ "mime/multipart"
+ "net/http"
+ "net/http/httptest"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/publicrelease"
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/status"
+)
+
+// systemFeed serves a signed fake release from an httptest server and
+// points the fixture's console at it.
+func systemFeed(t *testing.T, fx *consoleFixture, assets map[string][]byte) {
+ t.Helper()
+ pub, priv, _ := ed25519.GenerateKey(nil)
+ var sums strings.Builder
+ for name, body := range assets {
+ s := sha256.Sum256(body)
+ fmt.Fprintf(&sums, "%s %s\n", hex.EncodeToString(s[:]), name)
+ }
+ sig := base64.StdEncoding.EncodeToString(ed25519.Sign(priv, []byte(sums.String())))
+ mux := http.NewServeMux()
+ mux.HandleFunc("/latest.txt", func(w http.ResponseWriter, r *http.Request) { fmt.Fprintln(w, "v0.18.0") })
+ mux.HandleFunc("/dl/v0.18.0/SHA256SUMS.txt", func(w http.ResponseWriter, r *http.Request) { fmt.Fprint(w, sums.String()) })
+ mux.HandleFunc("/dl/v0.18.0/SHA256SUMS.txt.sig", func(w http.ResponseWriter, r *http.Request) { fmt.Fprintln(w, sig) })
+ for name, body := range assets {
+ body := body
+ mux.HandleFunc("/dl/v0.18.0/"+name, func(w http.ResponseWriter, r *http.Request) { w.Write(body) })
+ }
+ ts := httptest.NewServer(mux)
+ t.Cleanup(ts.Close)
+ fx.srv.cfg.SystemFeed = publicrelease.Feed{LatestURL: ts.URL + "/latest.txt", DownloadBase: ts.URL + "/dl", PubKey: base64.StdEncoding.EncodeToString(pub)}
+}
+
+func waitJob(t *testing.T, fx *consoleFixture, token, id string) map[string]any {
+ t.Helper()
+ for i := 0; i < 100; i++ {
+ st, out := getJSON(t, fx.ts.URL+"/api/cli/system/jobs/"+id, token)
+ if st != http.StatusOK {
+ t.Fatalf("job = %d %v", st, out)
+ }
+ if out["done"] == true {
+ return out
+ }
+ time.Sleep(20 * time.Millisecond)
+ }
+ t.Fatal("job never finished")
+ return nil
+}
+
+func seedBox(t *testing.T, fx *consoleFixture, label, platform, channel, rel string) {
+ t.Helper()
+ b := status.CollectBox(nil, status.Probes{Release: rel, Platform: platform, SystemChannel: channel}, time.Now())
+ b.Label = label
+ if err := status.WriteBox(context.Background(), fx.rel, b); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func TestSystemDoorsAreForOpsOwnersOnly(t *testing.T) {
+ fx := newConsoleFixture(t)
+ if err := fx.st.AddMember(OpsAccountID, "member@keymail.dev", RoleMember, time.Now()); err != nil {
+ t.Fatal(err)
+ }
+ member := cliToken(t, fx, "member@keymail.dev", "code-sys-member")
+ // The refusal is the BARE 404 the operator doors give — raw, not
+ // jsonError'd — so it is byte-identical to what a path that does not
+ // exist answers. That identity is the point: a member must not be able
+ // to learn that this deployment has a system surface at all, and the
+ // nonexistent path below is the baseline every refusal is measured
+ // against, which is why the assertions read the body and not just the
+ // status.
+ _, want := getRaw(t, fx.ts.URL+"/api/cli/system/no-such-door", member)
+ for _, u := range []string{"/api/cli/system/status", "/api/cli/system/jobs/deadbeef"} {
+ st, body := getRaw(t, fx.ts.URL+u, member)
+ if st != http.StatusNotFound || body != want {
+ t.Errorf("GET %s as a member = %d %q, want 404 %q", u, st, body, want)
+ }
+ }
+ for _, u := range []string{"/api/cli/system/update", "/api/cli/system/rollback"} {
+ st, body := postRaw(t, fx.ts.URL+u, map[string]string{"channel": "canary"}, member)
+ if st != http.StatusNotFound || body != want {
+ t.Errorf("POST %s as a member = %d %q, want 404 %q", u, st, body, want)
+ }
+ }
+ if st, _ := getRaw(t, fx.ts.URL+"/api/cli/system/status", "not-a-token"); st == http.StatusOK {
+ t.Error("unauthenticated status answered 200")
+ }
+}
+
+// TestSystemDoorsWithoutABucketRefuse: Config.Releases == nil is a legal
+// deployment shape, and every door here reads or writes the bucket. An owner
+// on such a console must get a sentence, not the nil-interface panic
+// status.LoadBoxes and release.GetSystemPointer would otherwise give.
+func TestSystemDoorsWithoutABucketRefuse(t *testing.T) {
+ fx := newConsoleFixture(t)
+ owner := cliToken(t, fx, "paul@keymail.dev", "code-sys-nobucket")
+ fx.srv.cfg.Releases = nil
+ if st, out := getJSON(t, fx.ts.URL+"/api/cli/system/status", owner); st != http.StatusServiceUnavailable {
+ t.Errorf("status without a bucket = %d %v, want 503", st, out)
+ }
+ if st, out := postJSON(t, fx.ts.URL+"/api/cli/system/update", map[string]any{"channel": "canary"}, owner); st != http.StatusServiceUnavailable {
+ t.Errorf("update without a bucket = %d %v, want 503", st, out)
+ }
+ if st, out := postJSON(t, fx.ts.URL+"/api/cli/system/rollback", map[string]string{"channel": "canary"}, owner); st != http.StatusServiceUnavailable {
+ t.Errorf("rollback without a bucket = %d %v, want 503", st, out)
+ }
+}
+
+func TestSystemUpdateMirrorsSignsAndBumps(t *testing.T) {
+ fx := newConsoleFixture(t)
+ arm := []byte("ELF arm64")
+ systemFeed(t, fx, map[string][]byte{"carlos-linux-arm64": arm, "carlos-linux-amd64": []byte("ELF amd64")})
+ seedBox(t, fx, "ie", "linux-arm64", "canary", "v0.17.0")
+ owner := cliToken(t, fx, "paul@keymail.dev", "code-sys-owner")
+ before, _, _ := release.ReadSerial(context.Background(), fx.rel)
+ st, out := postJSON(t, fx.ts.URL+"/api/cli/system/update", map[string]any{"channel": "canary", "spread": "2m"}, owner)
+ if st != http.StatusOK {
+ t.Fatalf("update = %d %v", st, out)
+ }
+ job := waitJob(t, fx, owner, str(t, out, "job"))
+ if job["error"] != "" {
+ t.Fatalf("job failed: %v", job)
+ }
+ p, _, err := release.GetSystemPointer(context.Background(), fx.rel, fx.pub, "router", "canary")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if p.Current.Version != "v0.18.0" || p.Spread != "2m0s" || p.Current.Source != "release:v0.18.0" {
+ t.Fatalf("pointer = %+v", p)
+ }
+ a, ok := p.Current.Artifacts["linux-arm64"]
+ if !ok {
+ t.Fatalf("no arm64 artifact: %+v", p.Current.Artifacts)
+ }
+ if _, ok := p.Current.Artifacts["linux-amd64"]; ok {
+ t.Fatal("mirrored a platform no box reports")
+ }
+ body, etag, err := fx.rel.Get(context.Background(), a.Key)
+ if err != nil || !bytes.Equal(body, arm) {
+ t.Fatalf("artifact not mirrored: %v", err)
+ }
+ s := sha256.Sum256(arm)
+ if a.SHA256 != hex.EncodeToString(s[:]) {
+ t.Fatalf("sha = %s", a.SHA256)
+ }
+ after, _, _ := release.ReadSerial(context.Background(), fx.rel)
+ if after.Serial == before.Serial {
+ t.Fatal("serial not bumped")
+ }
+ // A second promote of the same version to stable copies nothing and
+ // refuses different bytes under the same version.
+ st, out = postJSON(t, fx.ts.URL+"/api/cli/system/update", map[string]any{"channel": "stable", "version": "v0.18.0"}, owner)
+ if st != http.StatusOK {
+ t.Fatalf("second update = %d %v", st, out)
+ }
+ if job := waitJob(t, fx, owner, str(t, out, "job")); job["error"] != "" {
+ t.Fatalf("second job: %v", job)
+ }
+ // Overwrite the object out from under the pointer — a CAS-guarded Put
+ // with the etag we just read, because an empty ifMatch is create-only
+ // and would refuse an existing key. A re-promote of that version must
+ // then refuse: versions are immutable.
+ if _, err := fx.rel.Put(context.Background(), a.Key, []byte("different"), etag); err != nil {
+ t.Fatalf("overwrite the mirrored artifact: %v", err)
+ }
+ st, out = postJSON(t, fx.ts.URL+"/api/cli/system/update", map[string]any{"channel": "canary", "version": "v0.18.0"}, owner)
+ if st != http.StatusOK {
+ t.Fatalf("re-promote = %d %v", st, out)
+ }
+ job = waitJob(t, fx, owner, str(t, out, "job"))
+ if e, _ := job["error"].(string); e == "" || !strings.Contains(e, "immutable") {
+ t.Fatalf("re-promote over different bytes: %v", job)
+ }
+}
+
+func TestSystemUpdateFromAnUpload(t *testing.T) {
+ fx := newConsoleFixture(t)
+ owner := cliToken(t, fx, "paul@keymail.dev", "code-sys-upload")
+ var buf bytes.Buffer
+ mw := multipart.NewWriter(&buf)
+ meta, _ := mw.CreateFormField("meta")
+ meta.Write([]byte(`{"version":"66587fa","platform":"linux-arm64"}`))
+ file, _ := mw.CreateFormFile("file", "carlos-linux-arm64")
+ file.Write([]byte("ELF local"))
+ mw.Close()
+ req, _ := http.NewRequest(http.MethodPut, fx.ts.URL+"/api/cli/system/upload", &buf)
+ req.Header.Set("Content-Type", mw.FormDataContentType())
+ req.Header.Set("Authorization", "Bearer "+owner)
+ res, err := http.DefaultClient.Do(req)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var up map[string]any
+ if res.StatusCode != http.StatusOK {
+ t.Fatalf("upload = %d", res.StatusCode)
+ }
+ if err := json.NewDecoder(res.Body).Decode(&up); err != nil {
+ t.Fatal(err)
+ }
+ st, out := postJSON(t, fx.ts.URL+"/api/cli/system/update", map[string]any{"channel": "canary", "upload": up["upload"]}, owner)
+ if st != http.StatusOK {
+ t.Fatalf("update = %d %v", st, out)
+ }
+ if job := waitJob(t, fx, owner, str(t, out, "job")); job["error"] != "" {
+ t.Fatalf("job: %v", job)
+ }
+ p, _, err := release.GetSystemPointer(context.Background(), fx.rel, fx.pub, "router", "canary")
+ if err != nil || p.Current.Version != "66587fa" || p.Current.Source != "file" {
+ t.Fatalf("pointer = %+v %v", p, err)
+ }
+ // Upload without a platform is refused at the door.
+ var buf2 bytes.Buffer
+ mw2 := multipart.NewWriter(&buf2)
+ m2, _ := mw2.CreateFormField("meta")
+ m2.Write([]byte(`{"version":"abc"}`))
+ f2, _ := mw2.CreateFormFile("file", "x")
+ f2.Write([]byte("x"))
+ mw2.Close()
+ req2, _ := http.NewRequest(http.MethodPut, fx.ts.URL+"/api/cli/system/upload", &buf2)
+ req2.Header.Set("Content-Type", mw2.FormDataContentType())
+ req2.Header.Set("Authorization", "Bearer "+owner)
+ res2, _ := http.DefaultClient.Do(req2)
+ if res2.StatusCode != http.StatusBadRequest {
+ t.Fatalf("upload without platform = %d", res2.StatusCode)
+ }
+}
+
+func TestSystemStatusAndRollback(t *testing.T) {
+ fx := newConsoleFixture(t)
+ owner := cliToken(t, fx, "paul@keymail.dev", "code-sys-status")
+ seedBox(t, fx, "ie", "linux-arm64", "canary", "v1")
+ sg := release.RootSigner(fx.priv)
+ rel := func(v string) release.SystemRelease {
+ return release.SystemRelease{Version: v, Artifacts: map[string]release.SystemArtifact{"linux-arm64": {Key: release.SystemArtifactKey(v, "linux-arm64"), SHA256: strings.Repeat("a", 64)}}}
+ }
+ release.PromoteSystem(context.Background(), fx.rel, sg, "router", "canary", rel("v1"), release.SystemPromoteOptions{})
+ time.Sleep(1100 * time.Millisecond)
+ release.PromoteSystem(context.Background(), fx.rel, sg, "router", "canary", rel("v2"), release.SystemPromoteOptions{})
+ st, out := getJSON(t, fx.ts.URL+"/api/cli/system/status", owner)
+ if st != http.StatusOK {
+ t.Fatalf("status = %d %v", st, out)
+ }
+ boxes := out["boxes"].([]any)
+ if len(boxes) != 1 {
+ t.Fatalf("boxes = %v", boxes)
+ }
+ b := boxes[0].(map[string]any)
+ if b["label"] != "ie" || b["release"] != "v1" || b["target"] != "v2" || b["channel"] != "canary" {
+ t.Fatalf("box row = %v", b)
+ }
+ time.Sleep(1100 * time.Millisecond)
+ st, out = postJSON(t, fx.ts.URL+"/api/cli/system/rollback", map[string]string{"channel": "canary"}, owner)
+ if st != http.StatusOK || str(t, out, "version") != "v1" {
+ t.Fatalf("rollback = %d %v", st, out)
+ }
+ if st, out = postJSON(t, fx.ts.URL+"/api/cli/system/rollback", map[string]string{"channel": "canary"}, owner); st != http.StatusConflict {
+ t.Fatalf("second rollback = %d %v (nothing left in history)", st, out)
+ }
+}
diff --git a/internal/console/authcli.go b/internal/console/authcli.go
index c44a82887..9b8a6acd9 100644
--- a/internal/console/authcli.go
+++ b/internal/console/authcli.go
@@ -746,6 +746,22 @@ func (s *Server) registerCLIRoutes() {
s.cliRoute("POST /api/cli/apps/{account}/{app}/store/grant", svcrole.Allow(svcrole.Admin), s.handleAPIStoreGrant)
s.cliRoute("POST /api/cli/apps/{account}/{app}/store/rotate", svcrole.Allow(svcrole.Admin), s.handleAPIStoreRotate)
s.cliRoute("POST /api/cli/apps/{account}/{app}/store/rotate/finish", svcrole.Allow(svcrole.Admin), s.handleAPIStoreRotateFinish)
+ // `carlos system` (apisystem.go, spec §5) — the operator's promote of
+ // the PLATFORM's own binary: what is running where, roll the fleet
+ // forward onto a signed public release (or a local build), and roll it
+ // back. Every one of them is the ZERO Set, and for the sharpest version
+ // of the escalation cut in this file: the subject is the substrate every
+ // tenant runs inside. A credential that reached these doors could point
+ // every box in the deployment at a binary of its choosing, which is not
+ // a widening of any workload's authority — it is the end of the
+ // boundary the roles are drawn inside. The owner check itself is in the
+ // handlers (apiSystemOwner), because cliRoute's Set can only narrow
+ // SERVICE principals and a person always passes through it.
+ s.cliRoute("GET /api/cli/system/status", svcrole.Set{}, s.handleAPISystemStatus)
+ s.cliRoute("POST /api/cli/system/update", svcrole.Set{}, s.handleAPISystemUpdate)
+ s.cliRoute("GET /api/cli/system/jobs/{id}", svcrole.Set{}, s.handleAPISystemJob)
+ s.cliRoute("PUT /api/cli/system/upload", svcrole.Set{}, s.handleAPISystemUpload)
+ s.cliRoute("POST /api/cli/system/rollback", svcrole.Set{}, s.handleAPISystemRollback)
}
// clientIP is who the rate limiter counts against.
diff --git a/internal/console/server.go b/internal/console/server.go
index fc2aacd4e..fc37ef855 100644
--- a/internal/console/server.go
+++ b/internal/console/server.go
@@ -38,6 +38,7 @@ import (
"github.com/carlosframework/platform/internal/mail"
"github.com/carlosframework/platform/internal/objstore"
"github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/publicrelease"
"github.com/carlosframework/platform/internal/registry"
"github.com/carlosframework/platform/internal/release"
"github.com/carlosframework/platform/internal/status"
@@ -446,6 +447,9 @@ type Config struct {
// (view.go) — the same cached handle every reader here already uses.
Registry string
Releases release.Store // the deployment bucket
+ // SystemFeed is where `carlos system update` fetches a public release
+ // from. Zero value = the vendor feed; tests point it at a stub.
+ SystemFeed publicrelease.Feed
// Bucket is that same deployment bucket's NAME — CARLOS_DEPLOYMENT_BUCKET,
// the one bucket relationship this console can state as fact.
// /fleet/architecture draws the box -> bucket edge from it (archgraph.go's
@@ -727,6 +731,16 @@ type Server struct {
// TTL can be proven without a bucket, the same way storeSizer and
// ceCosts stand in for AWS above.
foldEdgeIPs func(ctx context.Context) []string
+ // The `carlos system` promote's two in-memory registries (apisystem.go):
+ // systemJobs is the running/finished update jobs an operator polls, and
+ // systemUploads is the single upload slot a `--file` promote parks its
+ // bytes in between the PUT that carries them and the job that consumes
+ // them. Both are process-lifetime and deliberately NOT durable: a
+ // console restart loses a job's step list, and re-running the promote is
+ // the recovery, because every step of it is idempotent.
+ systemMu sync.Mutex
+ systemJobs map[string]*systemJob
+ systemUploads map[string]systemUpload
}
// edgeFoldTTL is how long one fold of the deployment's edges files stands
diff --git a/internal/console/svcroutes_test.go b/internal/console/svcroutes_test.go
index d738e5573..1a2943cde 100644
--- a/internal/console/svcroutes_test.go
+++ b/internal/console/svcroutes_test.go
@@ -64,10 +64,18 @@ import (
// raises the tenant's daily cap — the bound the tenant sends inside. A
// credential must no more raise its own sending cap than rewrite the
// manifest it runs under.
+//
+// systemDoors is the `carlos system` family (apisystem.go), and it is the
+// widest subject of the four: the PLATFORM's own binary on every box in the
+// deployment. A credential reaching one of these could point the whole fleet
+// at a binary of its choosing, which is not a wider version of any workload's
+// authority — it is the end of the boundary the roles are drawn inside. One
+// substring covers all five doors, and no role may ever reach any of them.
var (
mintDoors = []string{"/instance-token", "/api/cli/services/", "/email/credentials"}
fleetDoors = []string{"/api/cli/fleets/"}
policyDoors = []string{"/instances-manifest", "/{app}/routes", "/email/pause", "/build-probe"}
+ systemDoors = []string{"/api/cli/system/"}
)
// closedDoors is the exact set of /api/cli/ patterns registered with the zero
@@ -146,6 +154,16 @@ var closedDoors = []string{
// over a process that never cycled, which is the precise failure the
// probe exists to prevent. Setting it is a person's one-time act.
"POST /api/cli/apps/{account}/{app}/build-probe",
+ // `carlos system` (apisystem.go): what the whole deployment RUNS. The
+ // same cut as the mint and fleet doors, taken over the substrate every
+ // tenant sits inside rather than over one account's credentials, so it
+ // is shut to every role including admin. The owner check is in the
+ // handlers, because a person passes cliRoute's Set regardless.
+ "GET /api/cli/system/status",
+ "POST /api/cli/system/update",
+ "GET /api/cli/system/jobs/{id}",
+ "PUT /api/cli/system/upload",
+ "POST /api/cli/system/rollback",
}
// TestEveryCLIDoorNamesItsRoles is the default-deny guarantee for the
@@ -189,14 +207,16 @@ func TestEveryCLIDoorNamesItsRoles(t *testing.T) {
// TestNoRoleReachesAMintFleetOrPolicyDoor is the escalation cut: a credential
// must never be able to mint or rotate a credential — including itself —
-// must never enrol a box, and must never rewrite the policy that bounds what
-// its own app may be brought up as.
+// must never enrol a box, must never rewrite the policy that bounds what its
+// own app may be brought up as, and must never move the platform binary the
+// whole deployment runs (systemDoors).
func TestNoRoleReachesAMintFleetOrPolicyDoor(t *testing.T) {
s := newConsoleFixture(t).srv
var families []string
families = append(families, mintDoors...)
families = append(families, fleetDoors...)
families = append(families, policyDoors...)
+ families = append(families, systemDoors...)
for pattern, set := range s.cliRoles {
for _, closed := range families {
if !strings.Contains(pattern, closed) {
@@ -685,6 +705,7 @@ func TestEveryClosedDoorRefusesAServiceCredential(t *testing.T) {
"{label}", "box1",
"{name}", "hub",
"{host}", "web.go.tito.io",
+ "{id}", "deadbeefdeadbeef",
).Replace(path)
if strings.Contains(u, "{") {
t.Fatalf("closedDoors entry %q has a wildcard this test does not fill", pattern)
@@ -702,8 +723,8 @@ func TestEveryClosedDoorRefusesAServiceCredential(t *testing.T) {
}
seen++
}
- if seen != len(closedDoors) || seen != 19 {
- t.Errorf("walked %d closed doors, want all %d (and 19 today: instance-token, the routes door, instances-manifest, the route-channel move, six fleets, four services, two email credentials, email pause, the build probe, the live tenancy read)",
+ if seen != len(closedDoors) || seen != 24 {
+ t.Errorf("walked %d closed doors, want all %d (and 24 today: instance-token, the routes door, instances-manifest, the route-channel move, six fleets, four services, two email credentials, email pause, the build probe, the live tenancy read, five system doors)",
seen, len(closedDoors))
}
}