amadan

carlos / platform Public

runs: stream staged inputs through a size limit; FIFO and ordering tests (task 2 review)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 6638782846a1ee21e0bd2f5ecbb6a12a1b5fa369 parent a6d2e0b
2 files changed, +131 −28
  • internal/runs/stage.go +39 −23
  • internal/runs/stage_linux_test.go +92 −5

CI cancelled — superseded by a newer push — run details

diff --git a/internal/runs/stage.go b/internal/runs/stage.go
index b73b124f8..79f1997b2 100644
--- a/internal/runs/stage.go
+++ b/internal/runs/stage.go
@@ -1,7 +1,6 @@
package runs
import (
- "bytes"
"context"
"crypto/sha256"
"encoding/hex"
@@ -17,12 +16,18 @@ import (
// 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.
+// beneath the instance's own directory, 0700, without following a symlink.
+// A directory that is already there is accepted as it is: the tenant shares
+// the agent's uid so its mode and owner prove nothing, the run id is
+// console-minted and unguessable, and a collision only denies the tenant's
+// own run. Anything else at either name (a symlink, a file, a FIFO) is a
+// refusal. Each file is created O_EXCL 0600 through instancefs, so a name
+// the tenant planted first is a refusal, never a write through it, and is
+// streamed from the store through a limit of its recorded size plus one
+// byte, so the read is bounded however large the object is; its length and
+// sha256 are checked against request.json as 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
@@ -48,27 +53,38 @@ func StageFiles(ctx context.Context, s release.Store, instanceDir string, req Re
return dir, nil
}
+// stageOne streams one input into dir/f.Name: created first, so a planted
+// name is refused before anything is fetched, then copied through a limit
+// of f.Size+1 into the file and a hasher at once. A length or hash that is
+// not the recorded one is an error; the caller removes the run directory.
+func stageOne(ctx context.Context, s release.Store, dir string, req Request, f File) error {
+ out, err := instancefs.OpenBeneath(dir, f.Name, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600)
+ if err != nil {
+ return err
+ }
+ defer out.Close()
+ rc, _, err := release.OpenObject(ctx, s, InputKey(req.Account, req.App, req.ID, f.Name))
+ if err != nil {
+ return fmt.Errorf("runs: read %s: %w", f.Name, err)
+ }
+ defer rc.Close()
+ h := sha256.New()
+ n, err := io.Copy(io.MultiWriter(out, h), io.LimitReader(rc, f.Size+1))
+ if err != nil {
+ return fmt.Errorf("runs: read %s: %w", f.Name, err)
+ }
+ if n != f.Size || hex.EncodeToString(h.Sum(nil)) != f.SHA256 {
+ return fmt.Errorf("runs: %s does not match its size and sha256", f.Name)
+ }
+ return out.Close()
+}
+
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 {
+ if err := stageOne(ctx, s, dir, req, f); err != nil {
return err
}
}
diff --git a/internal/runs/stage_linux_test.go b/internal/runs/stage_linux_test.go
index 964586cf6..0554598c8 100644
--- a/internal/runs/stage_linux_test.go
+++ b/internal/runs/stage_linux_test.go
@@ -7,9 +7,13 @@ import (
"crypto/sha256"
"encoding/hex"
"errors"
+ "io"
"os"
"path/filepath"
+ "sort"
+ "syscall"
"testing"
+ "time"
"github.com/carlosframework/platform/internal/instancefs"
"github.com/carlosframework/platform/internal/release"
@@ -90,11 +94,8 @@ 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))
- }
- }
+ sort.Slice(req.Files, func(i, j int) bool { return req.Files[i].Name < req.Files[j].Name })
+ req.Files[1].SHA256 = hex.EncodeToString(make([]byte, 32)) // b.csv fails after a.csv is staged
if _, err := StageFiles(context.Background(), s, inst, req); err == nil {
t.Fatal("a hash mismatch was staged")
}
@@ -304,3 +305,89 @@ func TestSweepRunDirsMovesPastAnInstanceItCannotClean(t *testing.T) {
t.Fatalf("the sweep stopped at the first failure: %v", err)
}
}
+
+// countingStore streams every object through a reader that counts the bytes
+// handed out, so a test can tell a bounded read from a whole-object one.
+type countingStore struct {
+ release.Store
+ read int64
+}
+
+type countingReader struct {
+ io.ReadCloser
+ n *int64
+}
+
+func (r countingReader) Read(p []byte) (int, error) {
+ n, err := r.ReadCloser.Read(p)
+ *r.n += int64(n)
+ return n, err
+}
+
+func (c *countingStore) Open(ctx context.Context, key string) (io.ReadCloser, string, error) {
+ rc, etag, err := release.OpenObject(ctx, c.Store, key)
+ if err != nil {
+ return nil, "", err
+ }
+ return countingReader{rc, &c.read}, etag, nil
+}
+
+func TestStageFilesBoundsTheReadByTheRecordedSize(t *testing.T) {
+ inner := release.NewDirStore(t.TempDir())
+ big := make([]byte, 1<<20)
+ if _, err := inner.Put(context.Background(), InputKey("bab", "titogo", tid, "a.csv"), big, ""); err != nil {
+ t.Fatal(err)
+ }
+ s := &countingStore{Store: inner}
+ sum := sha256.Sum256(big[:3])
+ req := Request{ID: tid, Account: "bab", App: "titogo",
+ Files: []File{{Name: "a.csv", Size: 3, SHA256: hex.EncodeToString(sum[:])}}}
+ inst := t.TempDir()
+ if _, err := StageFiles(context.Background(), s, inst, req); err == nil {
+ t.Fatal("an object larger than its recorded size was staged")
+ }
+ if s.read > 4 {
+ t.Fatalf("read %d bytes of an object recorded as 3", s.read)
+ }
+ if _, err := os.Lstat(filepath.Join(inst, DirName, tid)); !os.IsNotExist(err) {
+ t.Fatalf("the refused file's directory was left behind: %v", err)
+ }
+}
+
+func TestStageFilesRefusesAFIFOPromptly(t *testing.T) {
+ cases := map[string]func(inst string) string{
+ "at a file name": func(inst string) string {
+ if err := os.MkdirAll(filepath.Join(inst, DirName, tid), 0o700); err != nil {
+ t.Fatal(err)
+ }
+ return filepath.Join(inst, DirName, tid, "a.csv")
+ },
+ "at the run directory": func(inst string) string {
+ if err := os.MkdirAll(filepath.Join(inst, DirName), 0o700); err != nil {
+ t.Fatal(err)
+ }
+ return filepath.Join(inst, DirName, tid)
+ },
+ "at the run root": func(inst string) string { return filepath.Join(inst, DirName) },
+ }
+ for name, plant := range cases {
+ t.Run(name, func(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ req := stagedRequest(t, s, map[string]string{"a.csv": "1,2"})
+ inst := t.TempDir()
+ if err := syscall.Mkfifo(plant(inst), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ done := make(chan error, 1)
+ go func() { _, err := StageFiles(context.Background(), s, inst, req); done <- err }()
+ select {
+ case err := <-done:
+ if err == nil {
+ t.Fatal("staged through a FIFO")
+ }
+ case <-time.After(5 * time.Second):
+ t.Fatal("staging blocked on a FIFO")
+ }
+ })
+ }
+}