amadan

carlos / platform Public

host: carlos system apply — the refuse path

Lock, one O_NOFOLLOW-walked request under /data, the channel from
root's own env, the pointer re-verified, the per-channel replay
watermark, the no-op, and the candidate copied into a 0700 staging
dir and verified there before it is made executable.
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 e77c1863a5671ea19c8159543d6ae367531a0a2d parent 1378dec
2 files changed, +679 −0
  • internal/host/systemapply.go +310 −0
  • internal/host/systemapply_test.go +369 −0
diff --git a/internal/host/systemapply.go b/internal/host/systemapply.go
new file mode 100644
index 000000000..78e2df5ad
--- /dev/null
+++ b/internal/host/systemapply.go
@@ -0,0 +1,310 @@
+package host
+
+import (
+ "context"
+ "crypto/ed25519"
+ "crypto/sha256"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "os"
+ "path/filepath"
+ "strings"
+ "syscall"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// The root half of the platform self-update (spec §6). Runs as root from
+// carlos-system-update.service (or --recover from carlos-system-recover).
+// Every box-changing step is journaled first; every run begins with
+// recovery; root reads ONE file under /data and treats it as a hint.
+
+const (
+ systemEdgeUnit = "carlos-edge"
+ systemLockFile = "lock"
+ systemMaxRequest = 4 << 10
+ systemGateWait = 20 * time.Second
+ systemGateSoak = 15 * time.Second
+ systemVersionWait = 30 * time.Second
+ systemBootstrapWait = 120 * time.Second
+)
+
+// systemQuiesceUnits are the root units that exec /opt/carlos/carlos
+// against the running edge. Stopped (services too — stopping a timer does
+// not stop the pass it already started) before the rename, re-armed after
+// the gate.
+var systemQuiesceUnits = []string{
+ "carlos-adopt.timer", "carlos-adopt.path", "carlos-stage.path", "carlos-restart.path",
+ "carlos-adopt.service", "carlos-config-sync.service", "carlos-stage.service", "carlos-restart.service",
+}
+
+var (
+ errSystemRefused = errors.New("refused")
+ errSystemDeferred = errors.New("deferred")
+ errNotYetImplemented = errors.New("system apply: install path not implemented")
+)
+
+// sysOps is the seam between the apply and the box: systemctl, exec with a
+// deadline, the clock, sleep, the edge probe. Tests record and answer.
+type sysOps struct {
+ systemctl func(args ...string) (string, error)
+ run func(ctx context.Context, timeout time.Duration, name string, args ...string) (string, error)
+ now func() time.Time
+ sleep func(time.Duration)
+ probeEdge func() error
+ edgeStartedThisBoot func() bool
+ logf func(format string, args ...any)
+}
+
+type systemApplier struct {
+ ops sysOps
+ store release.Store
+ pub ed25519.PublicKey
+ channel string
+ optDir string
+ dataDir string
+ etcDir string // "/etc" in production; a temp dir in tests
+ build string // version.Build of THIS process
+ platform string // "<goos>-<goarch>"; the caller sets it from runtime
+ gateWait, soak time.Duration
+}
+
+func (a *systemApplier) rootDir() string { return systemRootDir(a.optDir) }
+func (a *systemApplier) stagingDir() string { return filepath.Join(a.rootDir(), "staging") }
+func (a *systemApplier) lockPath() string { return filepath.Join(a.rootDir(), systemLockFile) }
+func (a *systemApplier) file(name string) string { return filepath.Join(a.rootDir(), name) }
+
+// Run is the whole oneshot. It returns an error only for a failure the
+// caller should exit non-zero on (an apply that rolled back, a rollback
+// that failed, a broken box); refusals and no-ops are recorded in
+// result.json and return nil.
+func (a *systemApplier) Run(ctx context.Context, recover bool) error {
+ if err := os.MkdirAll(a.rootDir(), 0o755); err != nil {
+ return err
+ }
+ lock, err := os.OpenFile(a.lockPath(), os.O_CREATE|os.O_RDWR, 0o600)
+ if err != nil {
+ return err
+ }
+ defer lock.Close()
+ if recover {
+ if err := flockExclusive(lock); err != nil {
+ return err
+ }
+ } else if err := flockExclusiveNonblock(lock); err != nil {
+ a.ops.logf("system apply: another apply is running; leaving it to finish")
+ return nil
+ }
+ defer os.RemoveAll(a.stagingDir())
+
+ if err := a.recoverIfNeeded(ctx); err != nil {
+ return err
+ }
+ if recover {
+ return nil
+ }
+ req, err := a.readRequest()
+ if errors.Is(err, os.ErrNotExist) {
+ return nil
+ }
+ if err != nil {
+ a.ops.logf("system apply: request: %v", err)
+ return nil // nothing to record against: the request itself is unreadable
+ }
+ p, err := a.verifyPointer(ctx)
+ if err != nil {
+ return a.record(req, err)
+ }
+ if req.Version != p.Current.Version || !req.PromotedAt.Equal(p.PromotedAt) {
+ return a.record(req, fmt.Errorf("%w: request names %s@%s, pointer is %s@%s", errSystemRefused,
+ req.Version, req.PromotedAt.Format(time.RFC3339), p.Current.Version, p.PromotedAt.Format(time.RFC3339)))
+ }
+ seen, err := readSystemSeen(a.optDir)
+ if err != nil {
+ return a.record(req, fmt.Errorf("%w: seen.json unreadable: %v", errSystemRefused, err))
+ }
+ if wm, ok := seen[a.channel]; ok && p.PromotedAt.Before(wm) {
+ return a.record(req, fmt.Errorf("%w: replay — pointer promoted %s is older than the %s watermark %s", errSystemRefused,
+ p.PromotedAt.Format(time.RFC3339), a.channel, wm.Format(time.RFC3339)))
+ }
+ seen[a.channel] = p.PromotedAt
+ if err := writeJSONDurable(a.file("seen.json"), seen, 0o644); err != nil {
+ return err
+ }
+ if req.NoOp || p.Current.Version == a.onDiskBuild(ctx) {
+ return a.writeResult(req, systemOutcomeNoted, "")
+ }
+ staged, err := a.stageCandidate(p, req)
+ if err != nil {
+ return a.record(req, fmt.Errorf("%w: %v", errSystemRefused, err))
+ }
+ return a.install(ctx, req, p, staged) // Task 8b
+}
+
+func (a *systemApplier) install(ctx context.Context, req systemRequest, p *release.SystemPointer, staged string) error {
+ return errNotYetImplemented
+}
+
+func (a *systemApplier) recoverIfNeeded(ctx context.Context) error {
+ j, err := readSystemJournal(a.optDir)
+ if err != nil || j == nil {
+ return err
+ }
+ return fmt.Errorf("system apply: journal open at %s — recovery not implemented", j.Phase) // Task 8c
+}
+
+// record turns a refusal or deferral into result.json and a nil return.
+func (a *systemApplier) record(req systemRequest, err error) error {
+ outcome := systemOutcomeRefused
+ if errors.Is(err, errSystemDeferred) {
+ outcome = systemOutcomeDeferred
+ }
+ a.ops.logf("system apply: %s %s: %s: %v", a.channel, req.Version, outcome, err)
+ return a.writeResult(req, outcome, err.Error())
+}
+
+func (a *systemApplier) writeResult(req systemRequest, outcome, reason string) error {
+ return writeJSONDurable(a.file("result.json"), systemResult{
+ Channel: a.channel, Version: req.Version, PromotedAt: req.PromotedAt,
+ Outcome: outcome, Reason: reason, At: a.ops.now().UTC(),
+ }, 0o644)
+}
+
+// readRequest is the one read under /data: a fixed path reached without
+// following a single symlink, a regular file under 4 KiB, a version that
+// passes validSystemVersion. Anything else is an error, and root acts on
+// nothing.
+func (a *systemApplier) readRequest() (systemRequest, error) {
+ f, err := openBeneath(a.dataDir, "system", systemComponent, "requested.json")
+ if err != nil {
+ return systemRequest{}, err
+ }
+ defer f.Close()
+ st, err := f.Stat()
+ if err != nil {
+ return systemRequest{}, err
+ }
+ if !st.Mode().IsRegular() || st.Size() > systemMaxRequest {
+ return systemRequest{}, fmt.Errorf("request: not a regular file under %d bytes", systemMaxRequest)
+ }
+ var req systemRequest
+ if err := json.NewDecoder(io.LimitReader(f, systemMaxRequest)).Decode(&req); err != nil {
+ return systemRequest{}, fmt.Errorf("request: %w", err)
+ }
+ if !validSystemVersion(req.Version) {
+ return systemRequest{}, fmt.Errorf("request: version %q is not a version", req.Version)
+ }
+ if req.PromotedAt.IsZero() {
+ return systemRequest{}, errors.New("request: promoted_at unset")
+ }
+ return req, nil
+}
+
+// openBeneath opens root/parts... one component at a time with
+// O_DIRECTORY|O_NOFOLLOW for every directory and O_NOFOLLOW for the leaf,
+// so no component under a user-owned tree can be a symlink.
+func openBeneath(root string, parts ...string) (*os.File, error) {
+ fd, err := syscall.Open(root, syscall.O_RDONLY|syscall.O_DIRECTORY|syscall.O_CLOEXEC, 0)
+ if err != nil {
+ return nil, fmt.Errorf("open %s: %w", root, err)
+ }
+ for i, part := range parts {
+ flags := syscall.O_RDONLY | syscall.O_NOFOLLOW | syscall.O_CLOEXEC
+ if i < len(parts)-1 {
+ flags |= syscall.O_DIRECTORY
+ }
+ next, err := syscall.Openat(fd, part, flags, 0)
+ syscall.Close(fd)
+ if err != nil {
+ return nil, fmt.Errorf("open %s: %w", filepath.Join(append([]string{root}, parts[:i+1]...)...), err)
+ }
+ fd = next
+ }
+ return os.NewFile(uintptr(fd), filepath.Join(append([]string{root}, parts...)...)), nil
+}
+
+// verifyPointer fetches THIS box's channel pointer and verifies it with
+// the release public key. The channel is root's own (host.env), never the
+// request's. A store error is deferred, a bad pointer refused.
+func (a *systemApplier) verifyPointer(ctx context.Context) (*release.SystemPointer, error) {
+ p, _, err := release.GetSystemPointer(ctx, a.store, a.pub, release.SystemComponentRouter, a.channel)
+ switch {
+ case err == nil:
+ return p, nil
+ case errors.Is(err, os.ErrNotExist):
+ return nil, fmt.Errorf("%w: no %s pointer for this box's channel", errSystemRefused, a.channel)
+ case strings.Contains(err.Error(), "system pointer"):
+ return nil, fmt.Errorf("%w: %v", errSystemRefused, err)
+ default:
+ return nil, fmt.Errorf("%w: %v", errSystemDeferred, err)
+ }
+}
+
+// onDiskBuild asks the installed binary what it is, with a deadline. Not
+// version.Build: after an interrupted apply the two can differ.
+func (a *systemApplier) onDiskBuild(ctx context.Context) string {
+ out, err := a.ops.run(ctx, systemVersionWait, systemBinaryPath(a.optDir), "version")
+ if err != nil {
+ return ""
+ }
+ return strings.TrimSpace(strings.TrimPrefix(strings.TrimSpace(out), "carlos "))
+}
+
+// stageCandidate copies the agent's cache into staging/ (0700) as 0600,
+// verifies the copy's sha against the pointer, makes it executable, and
+// runs its `version`. Nothing under /data is read after the copy.
+func (a *systemApplier) stageCandidate(p *release.SystemPointer, req systemRequest) (string, error) {
+ art, ok := p.Current.Artifacts[a.platform]
+ if !ok {
+ return "", fmt.Errorf("pointer names no artifact for %s", a.platform)
+ }
+ os.RemoveAll(a.stagingDir())
+ if err := os.MkdirAll(a.stagingDir(), 0o700); err != nil {
+ return "", err
+ }
+ if err := os.Chmod(a.stagingDir(), 0o700); err != nil {
+ return "", err
+ }
+ src, err := openBeneath(a.dataDir, "system", systemComponent, req.Version, "carlos")
+ if err != nil {
+ return "", fmt.Errorf("candidate: %w", err)
+ }
+ defer src.Close()
+ if st, err := src.Stat(); err != nil || !st.Mode().IsRegular() {
+ return "", errors.New("candidate: not a regular file")
+ }
+ staged := filepath.Join(a.stagingDir(), "carlos")
+ dst, err := os.OpenFile(staged, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
+ if err != nil {
+ return "", err
+ }
+ h := sha256.New()
+ n, err := io.Copy(io.MultiWriter(dst, h), io.LimitReader(src, systemArtifactLimit+1))
+ dst.Close()
+ if err != nil {
+ return "", err
+ }
+ if n > systemArtifactLimit {
+ return "", fmt.Errorf("candidate: larger than %d bytes", systemArtifactLimit)
+ }
+ if got := hex.EncodeToString(h.Sum(nil)); got != art.SHA256 {
+ return "", fmt.Errorf("candidate: sha256 %s does not match the signed %s", got, art.SHA256)
+ }
+ if err := os.Chmod(staged, 0o755); err != nil {
+ return "", err
+ }
+ out, err := a.ops.run(context.Background(), systemVersionWait, staged, "version")
+ if err != nil || !strings.Contains(out, "carlos "+p.Current.Version) {
+ return "", fmt.Errorf("candidate reports %q, not %s", strings.TrimSpace(out), p.Current.Version)
+ }
+ return staged, nil
+}
+
+func flockExclusive(f *os.File) error { return syscall.Flock(int(f.Fd()), syscall.LOCK_EX) }
+func flockExclusiveNonblock(f *os.File) error {
+ return syscall.Flock(int(f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB)
+}
diff --git a/internal/host/systemapply_test.go b/internal/host/systemapply_test.go
new file mode 100644
index 000000000..26ffb937d
--- /dev/null
+++ b/internal/host/systemapply_test.go
@@ -0,0 +1,369 @@
+package host
+
+import (
+ "context"
+ "crypto/ed25519"
+ "crypto/sha256"
+ "encoding/hex"
+ "errors"
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// applyFixture is a fake box: temp opt/data/etc roots, a DirStore with a
+// signed pointer, and a sysOps that records every call and answers like a
+// healthy systemd. The candidate "binary" is answered by the fake run.
+type applyFixture struct {
+ t *testing.T
+ a *systemApplier
+ store *release.DirStore
+ priv ed25519.PrivateKey
+ calls []string // "systemctl is-active carlos-edge", "run <base> version", …
+ answer map[string]string // "version:<path>" → stdout; "plan:<path>" → JSON plan
+ now time.Time
+ edgeUp bool
+ nrestarts string
+ activeEnter string
+ probeErr error
+ edgeStartedThisBoot bool
+}
+
+func newApplyFixture(t *testing.T, channel, build string) *applyFixture {
+ t.Helper()
+ pub, priv, err := ed25519.GenerateKey(nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ f := &applyFixture{t: t, store: release.NewDirStore(t.TempDir()), priv: priv, answer: map[string]string{},
+ now: time.Date(2026, 9, 6, 11, 0, 0, 0, time.UTC), edgeUp: true, nrestarts: "0", activeEnter: "1000", edgeStartedThisBoot: true}
+ f.a = &systemApplier{
+ store: f.store, pub: pub, channel: channel,
+ optDir: t.TempDir(), dataDir: t.TempDir(), etcDir: t.TempDir(), build: build,
+ platform: "linux-arm64",
+ }
+ f.a.ops = sysOps{
+ systemctl: func(args ...string) (string, error) {
+ f.calls = append(f.calls, "systemctl "+strings.Join(args, " "))
+ switch {
+ case args[0] == "is-active" && args[1] == systemEdgeUnit:
+ if f.edgeUp {
+ return "active\n", nil
+ }
+ return "failed\n", errors.New("inactive")
+ case args[0] == "show":
+ return f.nrestarts + "\n" + f.activeEnter + "\n", nil
+ }
+ return "", nil
+ },
+ run: func(ctx context.Context, timeout time.Duration, name string, args ...string) (string, error) {
+ f.calls = append(f.calls, "run "+filepath.Base(name)+" "+strings.Join(args, " "))
+ if len(args) == 1 && args[0] == "version" {
+ if v, ok := f.answer["version:"+name]; ok {
+ return v, nil
+ }
+ return "", errors.New("no such binary")
+ }
+ if len(args) >= 3 && args[0] == "bootstrap" && args[1] == "-root" {
+ plan, ok := f.answer["plan:"+name]
+ if !ok {
+ return "", errors.New("bootstrap failed")
+ }
+ units := strings.Split(f.answer["units:"+name], " ")
+ dir := filepath.Join(args[2], "etc", "systemd", "system")
+ os.MkdirAll(dir, 0o755)
+ for _, u := range units {
+ if u != "" {
+ os.WriteFile(filepath.Join(dir, u), []byte("[Unit]\n# "+filepath.Base(name)+"\n"), 0o644)
+ }
+ }
+ os.MkdirAll(filepath.Join(args[2], "etc", "sysusers.d"), 0o755)
+ os.WriteFile(filepath.Join(args[2], "etc", "sysusers.d", "carlos.conf"), []byte("g carlos-sockets -\n"), 0o644)
+ os.MkdirAll(filepath.Join(args[2], "etc", "tmpfiles.d"), 0o755)
+ os.WriteFile(filepath.Join(args[2], "etc", "tmpfiles.d", "carlos.conf"), []byte("d /run/carlos\n"), 0o644)
+ os.MkdirAll(filepath.Join(args[2], "data"), 0o755)
+ os.WriteFile(filepath.Join(args[2], "data", "registry.db"), nil, 0o640)
+ return plan, nil
+ }
+ return "", errors.New("unexpected command")
+ },
+ now: func() time.Time { return f.now },
+ sleep: func(d time.Duration) { f.now = f.now.Add(d) },
+ probeEdge: func() error { return f.probeErr },
+ edgeStartedThisBoot: func() bool { return f.edgeStartedThisBoot },
+ logf: t.Logf,
+ }
+ os.MkdirAll(systemRootDir(f.a.optDir), 0o755)
+ os.MkdirAll(filepath.Join(f.a.etcDir, "systemd", "system"), 0o755)
+ // The on-disk binary answers `version` with build.
+ os.WriteFile(systemBinaryPath(f.a.optDir), []byte("#!/bin/sh\necho carlos "+build+"\n"), 0o755)
+ f.answer["version:"+systemBinaryPath(f.a.optDir)] = "carlos " + build + "\n"
+ f.answer["plan:"+systemBinaryPath(f.a.optDir)] = `{"enabled":["carlos-edge.service","carlos-adopt.timer","carlos-adopt.path","carlos-stage.path","carlos-restart.path","carlos-system-update.path","carlos-system-recover.service"],"inert":["carlos-restore-verify.timer"]}`
+ f.answer["units:"+systemBinaryPath(f.a.optDir)] = "carlos-edge.service carlos-adopt.timer carlos-adopt.path carlos-stage.path carlos-restart.path carlos-system-update.path carlos-system-update.service carlos-system-recover.service carlos-old-only.timer"
+ return f
+}
+
+// promote signs a pointer for f.a.channel naming version with body as the
+// linux-arm64 artifact, stages body into the agent's cache, and writes the
+// request the agent would have written. Returns the request.
+func (f *applyFixture) promote(version string, body []byte, noop bool) systemRequest {
+ f.t.Helper()
+ sum := sha256.Sum256(body)
+ rel := release.SystemRelease{Version: version, Artifacts: map[string]release.SystemArtifact{
+ "linux-arm64": {Key: release.SystemArtifactKey(version, "linux-arm64"), SHA256: hex.EncodeToString(sum[:])}}}
+ f.store.Put(context.Background(), rel.Artifacts["linux-arm64"].Key, body, "")
+ p, err := release.PromoteSystem(context.Background(), f.store, release.RootSigner(f.priv), "router", f.a.channel, rel, release.SystemPromoteOptions{})
+ if err != nil {
+ f.t.Fatal(err)
+ }
+ req := systemRequest{Version: version, PromotedAt: p.PromotedAt, NoOp: noop, SeenAt: f.now}
+ os.MkdirAll(systemDataDir(f.a.dataDir), 0o755)
+ writeJSONDurable(systemRequestPath(f.a.dataDir), req, 0o644)
+ if !noop {
+ os.MkdirAll(filepath.Dir(systemCachePath(f.a.dataDir, version)), 0o755)
+ os.WriteFile(systemCachePath(f.a.dataDir, version), body, 0o644)
+ }
+ return req
+}
+
+func (f *applyFixture) result() *systemResult { r, _ := readSystemResult(f.a.optDir); return r }
+
+func (f *applyFixture) called(prefix string) bool {
+ for _, c := range f.calls {
+ if strings.HasPrefix(c, prefix) {
+ return true
+ }
+ }
+ return false
+}
+
+func TestApplyNoRequestIsNothingToDo(t *testing.T) {
+ f := newApplyFixture(t, "stable", "v1")
+ if err := f.a.Run(context.Background(), false); err != nil {
+ t.Fatal(err)
+ }
+ if f.result() != nil || len(f.calls) != 0 {
+ t.Fatalf("result=%+v calls=%v", f.result(), f.calls)
+ }
+}
+
+func TestApplyLockKeepsTwoRunsApart(t *testing.T) {
+ f := newApplyFixture(t, "stable", "v1")
+ f.promote("v2", []byte("x"), false)
+ // Hold the lock as another process would.
+ held, err := os.OpenFile(f.a.lockPath(), os.O_CREATE|os.O_RDWR, 0o600)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer held.Close()
+ if err := flockExclusiveNonblock(held); err != nil {
+ t.Fatal(err)
+ }
+ if err := f.a.Run(context.Background(), false); err != nil {
+ t.Fatalf("a locked-out path run must exit 0 quietly: %v", err)
+ }
+ if f.result() != nil {
+ t.Fatal("a locked-out run wrote a result")
+ }
+}
+
+func TestApplyRequestIsAHintNotAnInstruction(t *testing.T) {
+ f := newApplyFixture(t, "stable", "v1")
+ // A symlinked parent under /data.
+ os.MkdirAll(filepath.Join(f.a.dataDir, "elsewhere"), 0o755)
+ os.Symlink(filepath.Join(f.a.dataDir, "elsewhere"), filepath.Join(f.a.dataDir, "system"))
+ os.MkdirAll(filepath.Join(f.a.dataDir, "elsewhere", "router"), 0o755)
+ os.WriteFile(filepath.Join(f.a.dataDir, "elsewhere", "router", "requested.json"), []byte(`{"version":"v2"}`), 0o644)
+ if _, err := f.a.readRequest(); err == nil {
+ t.Fatal("followed a symlinked parent under /data")
+ }
+ os.Remove(filepath.Join(f.a.dataDir, "system"))
+ // A symlinked request file.
+ os.MkdirAll(systemDataDir(f.a.dataDir), 0o755)
+ os.WriteFile(filepath.Join(f.a.dataDir, "real.json"), []byte(`{"version":"v2"}`), 0o644)
+ os.Symlink(filepath.Join(f.a.dataDir, "real.json"), systemRequestPath(f.a.dataDir))
+ if _, err := f.a.readRequest(); err == nil {
+ t.Fatal("followed a symlinked request")
+ }
+ os.Remove(systemRequestPath(f.a.dataDir))
+ // A traversal version.
+ writeJSONDurable(systemRequestPath(f.a.dataDir), systemRequest{Version: "../../etc/passwd", PromotedAt: f.now}, 0o644)
+ if _, err := f.a.readRequest(); err == nil {
+ t.Fatal("accepted a traversal version")
+ }
+ // Oversized.
+ os.WriteFile(systemRequestPath(f.a.dataDir), []byte(`{"version":"`+strings.Repeat("a", 5000)+`"}`), 0o644)
+ if _, err := f.a.readRequest(); err == nil {
+ t.Fatal("accepted a 5 KiB request")
+ }
+}
+
+func TestApplyChannelComesFromRootNotTheRequest(t *testing.T) {
+ // The box is on stable; a valid CANARY pointer exists and the agent
+ // requests its version. Root reads control/system/router/stable.json
+ // (missing) and refuses — a signed canary promote does not authorise a
+ // stable box.
+ f := newApplyFixture(t, "stable", "v1")
+ f.a.channel = "canary"
+ req := f.promote("v2", []byte("x"), false) // signs the CANARY pointer
+ f.a.channel = "stable"
+ if err := f.a.Run(context.Background(), false); err != nil {
+ t.Fatal(err)
+ }
+ r := f.result()
+ if r == nil || r.Outcome != systemOutcomeRefused || !r.matches("stable", "v2", req.PromotedAt) {
+ t.Fatalf("result = %+v", r)
+ }
+ if _, err := os.Stat(systemBinaryPath(f.a.optDir) + ".prev"); err == nil {
+ t.Fatal("touched the box")
+ }
+}
+
+func TestApplyStaleTupleIsRefusedAgainstTheRequestsTuple(t *testing.T) {
+ f := newApplyFixture(t, "stable", "v1")
+ old := f.promote("v2", []byte("x"), false)
+ time.Sleep(1100 * time.Millisecond)
+ f.promote("v3", []byte("y"), false)
+ // Put the OLD request back: the agent's spread started under v2.
+ writeJSONDurable(systemRequestPath(f.a.dataDir), old, 0o644)
+ if err := f.a.Run(context.Background(), false); err != nil {
+ t.Fatal(err)
+ }
+ r := f.result()
+ if r == nil || r.Outcome != systemOutcomeRefused || r.Version != "v2" || !r.PromotedAt.Equal(old.PromotedAt) {
+ t.Fatalf("result = %+v (must name the REQUEST's tuple so v3 is not suppressed)", r)
+ }
+}
+
+func TestApplyReplayGuardIsPerChannelAndEqualIsAllowed(t *testing.T) {
+ f := newApplyFixture(t, "stable", "v1")
+ req := f.promote("v2", []byte("x"), true)
+ // Older than the watermark → refused, seen unchanged.
+ writeJSONDurable(filepath.Join(systemRootDir(f.a.optDir), "seen.json"), systemSeen{"stable": req.PromotedAt.Add(time.Hour), "canary": time.Time{}}, 0o644)
+ if err := f.a.Run(context.Background(), false); err != nil {
+ t.Fatal(err)
+ }
+ if r := f.result(); r == nil || r.Outcome != systemOutcomeRefused || !strings.Contains(r.Reason, "replay") {
+ t.Fatalf("result = %+v", r)
+ }
+ // Equal → allowed (a retry of the same promote).
+ writeJSONDurable(filepath.Join(systemRootDir(f.a.optDir), "seen.json"), systemSeen{"stable": req.PromotedAt}, 0o644)
+ if err := f.a.Run(context.Background(), false); err != nil {
+ t.Fatal(err)
+ }
+ if r := f.result(); r == nil || r.Outcome != systemOutcomeNoted {
+ t.Fatalf("result = %+v", r)
+ }
+ // Another channel's newer watermark does not block this channel.
+ writeJSONDurable(filepath.Join(systemRootDir(f.a.optDir), "seen.json"), systemSeen{"canary": req.PromotedAt.Add(time.Hour)}, 0o644)
+ if err := f.a.Run(context.Background(), false); err != nil {
+ t.Fatal(err)
+ }
+ if r := f.result(); r == nil || r.Outcome != systemOutcomeNoted {
+ t.Fatalf("result = %+v", r)
+ }
+ seen, _ := readSystemSeen(f.a.optDir)
+ if !seen["stable"].Equal(req.PromotedAt) {
+ t.Fatalf("seen not advanced: %v", seen)
+ }
+}
+
+func TestApplyNoOpNotesAndTouchesNothing(t *testing.T) {
+ f := newApplyFixture(t, "stable", "v2")
+ f.promote("v2", []byte("x"), true)
+ if err := f.a.Run(context.Background(), false); err != nil {
+ t.Fatal(err)
+ }
+ if r := f.result(); r == nil || r.Outcome != systemOutcomeNoted {
+ t.Fatalf("result = %+v", r)
+ }
+ if _, err := os.Stat(f.a.stagingDir()); err == nil {
+ t.Fatal("a no-op created staging/")
+ }
+ if len(f.calls) != 0 {
+ t.Fatalf("a no-op ran commands: %v", f.calls)
+ }
+}
+
+func TestApplyStoreErrorIsDeferredNotRefused(t *testing.T) {
+ f := newApplyFixture(t, "stable", "v1")
+ req := f.promote("v2", []byte("x"), false)
+ f.a.store = unreadableStore{f.store}
+ if err := f.a.Run(context.Background(), false); err != nil {
+ t.Fatal(err)
+ }
+ if r := f.result(); r == nil || r.Outcome != systemOutcomeDeferred || !r.matches("stable", "v2", req.PromotedAt) {
+ t.Fatalf("result = %+v", r)
+ }
+}
+
+// unreadableStore is the bucket-outage fixture for reads: the door is
+// restarting, so the pointer cannot be fetched at all.
+type unreadableStore struct{ release.Store }
+
+func (unreadableStore) Get(context.Context, string) ([]byte, string, error) {
+ return nil, "", errors.New("door restarting")
+}
+
+func TestApplyStagesTheCandidateFromARootOwnedCopy(t *testing.T) {
+ f := newApplyFixture(t, "stable", "v1")
+ body := []byte("#!/bin/sh\necho carlos v2\n")
+ f.promote("v2", body, false)
+ f.answer["version:"+filepath.Join(f.a.stagingDir(), "carlos")] = "carlos v2\n"
+ p, err := f.a.verifyPointer(context.Background())
+ if err != nil {
+ t.Fatal(err)
+ }
+ req, _ := f.a.readRequest()
+ staged, err := f.a.stageCandidate(p, req)
+ if err != nil {
+ t.Fatal(err)
+ }
+ st, _ := os.Stat(staged)
+ if st.Mode().Perm() != 0o755 {
+ t.Fatalf("staged mode = %v after verification", st.Mode().Perm())
+ }
+ if dst, _ := os.Stat(f.a.stagingDir()); dst.Mode().Perm() != 0o700 {
+ t.Fatalf("staging dir mode = %v", dst.Mode().Perm())
+ }
+ // Swapping the cache after staging changes nothing about the copy.
+ os.WriteFile(systemCachePath(f.a.dataDir, "v2"), []byte("evil"), 0o644)
+ got, _ := os.ReadFile(staged)
+ if string(got) != string(body) {
+ t.Fatal("staged copy follows the cache")
+ }
+ // A sha mismatch is refused BEFORE the copy is made executable and
+ // staging/ is cleared.
+ f2 := newApplyFixture(t, "stable", "v1")
+ f2.promote("v2", body, false)
+ os.WriteFile(systemCachePath(f2.a.dataDir, "v2"), []byte("evil"), 0o644)
+ p2, _ := f2.a.verifyPointer(context.Background())
+ req2, _ := f2.a.readRequest()
+ if _, err := f2.a.stageCandidate(p2, req2); err == nil || !strings.Contains(err.Error(), "sha256") {
+ t.Fatalf("err = %v", err)
+ }
+ // A candidate that lies about its version is refused.
+ f3 := newApplyFixture(t, "stable", "v1")
+ f3.promote("v2", body, false)
+ f3.answer["version:"+filepath.Join(f3.a.stagingDir(), "carlos")] = "carlos v9\n"
+ p3, _ := f3.a.verifyPointer(context.Background())
+ req3, _ := f3.a.readRequest()
+ if _, err := f3.a.stageCandidate(p3, req3); err == nil || !strings.Contains(err.Error(), "reports") {
+ t.Fatalf("err = %v", err)
+ }
+ // A symlinked cache file is refused.
+ f4 := newApplyFixture(t, "stable", "v1")
+ f4.promote("v2", body, false)
+ os.Remove(systemCachePath(f4.a.dataDir, "v2"))
+ os.WriteFile(filepath.Join(f4.a.dataDir, "real"), body, 0o644)
+ os.Symlink(filepath.Join(f4.a.dataDir, "real"), systemCachePath(f4.a.dataDir, "v2"))
+ p4, _ := f4.a.verifyPointer(context.Background())
+ req4, _ := f4.a.readRequest()
+ if _, err := f4.a.stageCandidate(p4, req4); err == nil {
+ t.Fatal("followed a symlinked cache file")
+ }
+}