amadan

carlos / platform Public

runs: stage a run's files beneath its instance, and the boot sweep (run spec §5.2, §5.8, codex 6)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude a6d2e0b8ee053d9e93226b7840881be52a0b6bdc parent 39e359f
2 files changed, +419 −0
  • internal/runs/stage.go +113 −0
  • internal/runs/stage_linux_test.go +306 −0
diff --git a/internal/runs/stage.go b/internal/runs/stage.go
new file mode 100644
index 000000000..b73b124f8
--- /dev/null
+++ b/internal/runs/stage.go
@@ -0,0 +1,113 @@
+package runs
+
+import (
+ "bytes"
+ "context"
+ "crypto/sha256"
+ "encoding/hex"
+ "errors"
+ "fmt"
+ "io"
+ "os"
+ "path/filepath"
+
+ "github.com/carlosframework/platform/internal/instancefs"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// StageFiles makes <instanceDir>/.carlos-run/<id>/ and downloads the run's
+// files into it (run spec §5.2 step 3). Both directories are created
+// beneath the instance's own directory, 0700, without following a
+// symlink; each file is created O_EXCL 0600 through instancefs, so a name
+// the tenant planted first is a refusal, never a write through it; each is
+// checked against request.json's size and sha256 before it is written and
+// its name against PlainName. The directory exists even when there are no
+// files: it is CARLOS_RUN_DIR.
+//
+// When a file fails after the run directory is in place, the directory is
+// removed again (no symlink followed), so a refused staging leaves nothing
+// half-written; a failure to remove it is joined onto the error returned.
+func StageFiles(ctx context.Context, s release.Store, instanceDir string, req Request) (string, error) {
+ if !ValidID(req.ID) {
+ return "", fmt.Errorf("runs: %q is not a run id", req.ID)
+ }
+ root := filepath.Join(instanceDir, DirName)
+ if _, err := instancefs.EnsureDirBeneath(instanceDir, DirName, 0o700); err != nil {
+ return "", err
+ }
+ if _, err := instancefs.EnsureDirBeneath(root, req.ID, 0o700); err != nil {
+ return "", err
+ }
+ dir := filepath.Join(root, req.ID)
+ if err := stageEach(ctx, s, dir, req); err != nil {
+ if rerr := instancefs.RemoveAllBeneath(root, req.ID); rerr != nil {
+ err = errors.Join(err, rerr)
+ }
+ return "", err
+ }
+ return dir, nil
+}
+
+func stageEach(ctx context.Context, s release.Store, dir string, req Request) error {
+ for _, f := range req.Files {
+ if !PlainName(f.Name) {
+ return fmt.Errorf("runs: %q is not a plain file name", f.Name)
+ }
+ raw, _, err := s.Get(ctx, InputKey(req.Account, req.App, req.ID, f.Name))
+ if err != nil {
+ return fmt.Errorf("runs: read %s: %w", f.Name, err)
+ }
+ sum := sha256.Sum256(raw)
+ if int64(len(raw)) != f.Size || hex.EncodeToString(sum[:]) != f.SHA256 {
+ return fmt.Errorf("runs: %s does not match its size and sha256", f.Name)
+ }
+ out, err := instancefs.OpenBeneath(dir, f.Name, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600)
+ if err != nil {
+ return err
+ }
+ _, werr := io.Copy(out, bytes.NewReader(raw))
+ serr := out.Sync()
+ cerr := out.Close()
+ if err := errors.Join(werr, serr, cerr); err != nil {
+ return err
+ }
+ }
+ return nil
+}
+
+// RemoveRunDir deletes one run's directory, following no symlink. A tenant
+// can make a tree uncleanable (by depth or mode): the caller logs the error
+// and carries on, it never fails a run.
+func RemoveRunDir(instanceDir, id string) error {
+ if !ValidID(id) {
+ return fmt.Errorf("runs: %q is not a run id", id)
+ }
+ return instancefs.RemoveAllBeneath(filepath.Join(instanceDir, DirName), id)
+}
+
+// SweepRunDirs deletes every instance's .carlos-run directory (run spec
+// §5.8, the box sweep). It runs at agent start, before any run is picked:
+// the edge's stop took every cmd unit with it (BindsTo=carlos-edge.service),
+// so no run on this box is open. One instance it cannot clean (a tenant can
+// make a tree uncleanable) does not stop it: it moves on to the next and
+// returns every failure joined, for the caller to log.
+func SweepRunDirs(dataDir string) error {
+ base := filepath.Join(dataDir, "instances")
+ ents, err := os.ReadDir(base)
+ if errors.Is(err, os.ErrNotExist) {
+ return nil
+ }
+ if err != nil {
+ return err
+ }
+ var errs []error
+ for _, e := range ents {
+ if !e.IsDir() {
+ continue
+ }
+ if err := instancefs.RemoveAllBeneath(filepath.Join(base, e.Name()), DirName); err != nil {
+ errs = append(errs, err)
+ }
+ }
+ return errors.Join(errs...)
+}
diff --git a/internal/runs/stage_linux_test.go b/internal/runs/stage_linux_test.go
new file mode 100644
index 000000000..964586cf6
--- /dev/null
+++ b/internal/runs/stage_linux_test.go
@@ -0,0 +1,306 @@
+//go:build linux
+
+package runs
+
+import (
+ "context"
+ "crypto/sha256"
+ "encoding/hex"
+ "errors"
+ "os"
+ "path/filepath"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/instancefs"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+func stagedRequest(t *testing.T, s release.Store, files map[string]string) Request {
+ t.Helper()
+ req := Request{ID: tid, Account: "bab", App: "titogo"}
+ for name, body := range files {
+ sum := sha256.Sum256([]byte(body))
+ req.Files = append(req.Files, File{Name: name, Size: int64(len(body)), SHA256: hex.EncodeToString(sum[:])})
+ if _, err := s.Put(context.Background(), InputKey("bab", "titogo", tid, name), []byte(body), ""); err != nil {
+ t.Fatal(err)
+ }
+ }
+ return req
+}
+
+func TestStageFilesWritesTheUploads(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ inst := t.TempDir()
+ req := stagedRequest(t, s, map[string]string{"a.csv": "1,2", "b.json": "{}"})
+ dir, err := StageFiles(context.Background(), s, inst, req)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if dir != filepath.Join(inst, DirName, tid) {
+ t.Fatalf("dir %q", dir)
+ }
+ st, err := os.Stat(dir)
+ if err != nil || st.Mode().Perm() != 0o700 {
+ t.Fatalf("run dir mode %v, %v", st.Mode(), err)
+ }
+ for name, want := range map[string]string{"a.csv": "1,2", "b.json": "{}"} {
+ raw, err := os.ReadFile(filepath.Join(dir, name))
+ fi, _ := os.Stat(filepath.Join(dir, name))
+ if err != nil || string(raw) != want || fi.Mode().Perm() != 0o600 {
+ t.Errorf("%s: %q %v mode %v", name, raw, err, fi.Mode())
+ }
+ }
+}
+
+func TestStageFilesMakesTheDirectoryWithNoFiles(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ inst := t.TempDir()
+ dir, err := StageFiles(context.Background(), s, inst, Request{ID: tid, Account: "bab", App: "titogo"})
+ if err != nil {
+ t.Fatal(err)
+ }
+ if st, err := os.Stat(dir); err != nil || !st.IsDir() {
+ t.Fatalf("CARLOS_RUN_DIR must always exist: %v", err)
+ }
+}
+
+func TestStageFilesChecksSizeAndHash(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ inst := t.TempDir()
+ req := stagedRequest(t, s, map[string]string{"a.csv": "1,2"})
+ req.Files[0].SHA256 = hex.EncodeToString(make([]byte, 32))
+ if _, err := StageFiles(context.Background(), s, inst, req); err == nil {
+ t.Fatal("a hash mismatch was staged")
+ }
+ if _, err := os.Lstat(filepath.Join(inst, DirName, tid, "a.csv")); !os.IsNotExist(err) {
+ t.Fatalf("a mismatched file was left behind: %v", err)
+ }
+ s = release.NewDirStore(t.TempDir()) // the store refuses a second Put of the same input
+ req = stagedRequest(t, s, map[string]string{"a.csv": "1,2"})
+ req.Files[0].Size = 2
+ if err := RemoveRunDir(inst, tid); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := StageFiles(context.Background(), s, inst, req); err == nil {
+ t.Fatal("a size mismatch was staged")
+ }
+}
+
+func TestStageFilesRemovesWhatItStagedWhenALaterFileFails(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ inst := t.TempDir()
+ req := stagedRequest(t, s, map[string]string{"a.csv": "1,2", "b.csv": "3,4"})
+ for i := range req.Files {
+ if req.Files[i].Name == "b.csv" {
+ req.Files[i].SHA256 = hex.EncodeToString(make([]byte, 32))
+ }
+ }
+ if _, err := StageFiles(context.Background(), s, inst, req); err == nil {
+ t.Fatal("a hash mismatch was staged")
+ }
+ if _, err := os.Lstat(filepath.Join(inst, DirName, tid)); !os.IsNotExist(err) {
+ t.Fatalf("a failed staging left its directory behind: %v", err)
+ }
+}
+
+func TestStageFilesRefusesAPlantedFile(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ inst := t.TempDir()
+ req := stagedRequest(t, s, map[string]string{"a.csv": "1,2"})
+ if err := os.MkdirAll(filepath.Join(inst, DirName, tid), 0o700); err != nil {
+ t.Fatal(err)
+ }
+ outside := filepath.Join(t.TempDir(), "victim")
+ if err := os.WriteFile(outside, []byte("keep"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Symlink(outside, filepath.Join(inst, DirName, tid, "a.csv")); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := StageFiles(context.Background(), s, inst, req); err == nil {
+ t.Fatal("staged over a planted symlink")
+ }
+ if raw, _ := os.ReadFile(outside); string(raw) != "keep" {
+ t.Fatal("the planted symlink's target was written")
+ }
+}
+
+func TestStageFilesRefusesAPlantedRegularFile(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ inst := t.TempDir()
+ req := stagedRequest(t, s, map[string]string{"a.csv": "1,2"})
+ planted := filepath.Join(inst, DirName, tid, "a.csv")
+ if err := os.MkdirAll(filepath.Dir(planted), 0o700); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(planted, []byte("tenant"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := StageFiles(context.Background(), s, inst, req); err == nil {
+ t.Fatal("staged over a planted file")
+ }
+}
+
+func TestStageFilesRefusesASymlinkedRunRoot(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ inst, outside := t.TempDir(), t.TempDir()
+ if err := os.Symlink(outside, filepath.Join(inst, DirName)); err != nil {
+ t.Fatal(err)
+ }
+ _, err := StageFiles(context.Background(), s, inst, Request{ID: tid, Account: "bab", App: "titogo"})
+ if !errors.Is(err, instancefs.ErrSymlink) {
+ t.Fatalf("err %v, want ErrSymlink", err)
+ }
+ if ents, _ := os.ReadDir(outside); len(ents) != 0 {
+ t.Fatalf("the symlink's target was written: %v", ents)
+ }
+}
+
+func TestStageFilesRefusesASymlinkedRunDirectory(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ inst, outside := t.TempDir(), t.TempDir()
+ req := stagedRequest(t, s, map[string]string{"a.csv": "1,2"})
+ if err := os.Mkdir(filepath.Join(inst, DirName), 0o700); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Symlink(outside, filepath.Join(inst, DirName, tid)); err != nil {
+ t.Fatal(err)
+ }
+ _, err := StageFiles(context.Background(), s, inst, req)
+ if !errors.Is(err, instancefs.ErrSymlink) {
+ t.Fatalf("err %v, want ErrSymlink", err)
+ }
+ if ents, _ := os.ReadDir(outside); len(ents) != 0 {
+ t.Fatalf("the symlink's target was written: %v", ents)
+ }
+}
+
+func TestStageFilesRefusesANameThatIsNotPlain(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ req := Request{ID: tid, Account: "bab", App: "titogo", Files: []File{{Name: "../x", Size: 1}}}
+ if _, err := StageFiles(context.Background(), s, t.TempDir(), req); err == nil {
+ t.Fatal("a path name was staged (spec §7: checked again by the agent)")
+ }
+}
+
+func TestStageFilesRefusesAnIDThatIsNotARunID(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ inst := t.TempDir()
+ if _, err := StageFiles(context.Background(), s, inst, Request{ID: "../x", Account: "bab", App: "titogo"}); err == nil {
+ t.Fatal("a bad run id was staged")
+ }
+ if err := RemoveRunDir(inst, ".."); err == nil {
+ t.Fatal("a bad run id was removed")
+ }
+}
+
+func TestRemoveRunDirFollowsNoSymlink(t *testing.T) {
+ inst, outside := t.TempDir(), t.TempDir()
+ if err := os.WriteFile(filepath.Join(outside, "keep"), []byte("k"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ dir := filepath.Join(inst, DirName, tid)
+ if err := os.MkdirAll(dir, 0o700); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Symlink(outside, filepath.Join(dir, "link")); err != nil {
+ t.Fatal(err)
+ }
+ if err := RemoveRunDir(inst, tid); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := os.Lstat(dir); !os.IsNotExist(err) {
+ t.Fatalf("run dir survived: %v", err)
+ }
+ if _, err := os.Stat(filepath.Join(outside, "keep")); err != nil {
+ t.Fatalf("RemoveRunDir followed a symlink: %v", err)
+ }
+ if err := RemoveRunDir(inst, tid); err != nil {
+ t.Fatalf("removing a missing run dir: %v", err)
+ }
+}
+
+func TestSweepRunDirsRemovesEveryRunDirectory(t *testing.T) {
+ data := t.TempDir()
+ for _, host := range []string{"a.example.com", "b.example.com"} {
+ if err := os.MkdirAll(filepath.Join(data, "instances", host, DirName, tid, "sub"), 0o700); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(filepath.Join(data, "instances", host, "app.db"), []byte("db"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if err := SweepRunDirs(data); err != nil {
+ t.Fatal(err)
+ }
+ for _, host := range []string{"a.example.com", "b.example.com"} {
+ if _, err := os.Lstat(filepath.Join(data, "instances", host, DirName)); !os.IsNotExist(err) {
+ t.Errorf("%s: .carlos-run survived", host)
+ }
+ if _, err := os.Stat(filepath.Join(data, "instances", host, "app.db")); err != nil {
+ t.Errorf("%s: the sweep touched the database", host)
+ }
+ }
+}
+
+func TestSweepRunDirsWithNoInstancesIsNotAnError(t *testing.T) {
+ if err := SweepRunDirs(t.TempDir()); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func TestSweepRunDirsFollowsNoSymlink(t *testing.T) {
+ data, outside := t.TempDir(), t.TempDir()
+ if err := os.WriteFile(filepath.Join(outside, "keep"), []byte("k"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ // One instance whose .carlos-run is a link; one instance that is itself a link.
+ if err := os.MkdirAll(filepath.Join(data, "instances", "a.example.com"), 0o700); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Symlink(outside, filepath.Join(data, "instances", "a.example.com", DirName)); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Mkdir(filepath.Join(outside, DirName), 0o700); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Symlink(outside, filepath.Join(data, "instances", "b.example.com")); err != nil {
+ t.Fatal(err)
+ }
+ if err := SweepRunDirs(data); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := os.Lstat(filepath.Join(data, "instances", "a.example.com", DirName)); !os.IsNotExist(err) {
+ t.Errorf("the planted link survived: %v", err)
+ }
+ if _, err := os.Stat(filepath.Join(outside, "keep")); err != nil {
+ t.Errorf("the sweep followed a link: %v", err)
+ }
+ if _, err := os.Stat(filepath.Join(outside, DirName)); err != nil {
+ t.Errorf("the sweep followed a linked instance: %v", err)
+ }
+}
+
+func TestSweepRunDirsMovesPastAnInstanceItCannotClean(t *testing.T) {
+ if os.Geteuid() == 0 {
+ t.Skip("root reads a mode 000 directory")
+ }
+ data := t.TempDir()
+ stuck := filepath.Join(data, "instances", "a.example.com", DirName, tid)
+ if err := os.MkdirAll(filepath.Join(stuck, "sub"), 0o700); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.Chmod(stuck, 0); err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { _ = os.Chmod(stuck, 0o700) })
+ if err := os.MkdirAll(filepath.Join(data, "instances", "b.example.com", DirName, tid), 0o700); err != nil {
+ t.Fatal(err)
+ }
+ if err := SweepRunDirs(data); err == nil {
+ t.Fatal("an uncleanable instance went unreported")
+ }
+ if _, err := os.Lstat(filepath.Join(data, "instances", "b.example.com", DirName)); !os.IsNotExist(err) {
+ t.Fatalf("the sweep stopped at the first failure: %v", err)
+ }
+}