From 8c8950d9ef1ddea606974df897d3c885992d330b Mon Sep 17 00:00:00 2001 From: bircni Date: Fri, 6 Nov 2026 02:22:40 +0000 Subject: [PATCH] refactor(container): stream copies into the job and archives out of the host (#1340) Adds `CopyTarStream` to the job environments. It extracts directories and files into the job as they are written, kept below the destination and owned by the job user along with any directories it creates, so built-in actions can put large content in place without buffering it. The host's `GetContainerArchive` now streams instead of building the archive in memory. --------- Co-authored-by: silverwind <2021+silverwind@noreply.gitea.com> Co-authored-by: silverwind Co-authored-by: Lunny Xiao Reviewed-on: https://gitea.com/gitea/runner/pulls/1340 Reviewed-by: Lunny Xiao Reviewed-by: silverwind <2021+silverwind@noreply.gitea.com> Co-authored-by: bircni --- act/container/container_types.go | 1 + act/container/docker_run.go | 43 ++++++++++- act/container/docker_run_test.go | 45 +++++++++++ act/container/host_environment.go | 100 ++++++++++++++++++------- act/container/host_environment_test.go | 50 ++++++++++++- act/container/kubernetes.go | 15 ++-- act/container/kubernetes_test.go | 12 +++ act/container/tar.go | 83 ++++++++++++++++++++ act/runner/container_mock_test.go | 4 + 9 files changed, 313 insertions(+), 40 deletions(-) diff --git a/act/container/container_types.go b/act/container/container_types.go index c7deae30..0f399dc4 100644 --- a/act/container/container_types.go +++ b/act/container/container_types.go @@ -131,6 +131,7 @@ type Container interface { Copy(destPath string, files ...*FileEntry) common.Executor CopyDir(destPath, srcPath string, useGitIgnore, skipGitDir bool) common.Executor GetContainerArchive(ctx context.Context, srcPath string) (io.ReadCloser, error) + CopyTarStream(ctx context.Context, destPath string, write func(*TarWriter) error) error Inspect(ctx context.Context) (*Info, error) DumpLogs(ctx context.Context) error Pull(forcePull bool) common.Executor diff --git a/act/container/docker_run.go b/act/container/docker_run.go index 8dfb506f..303683f4 100644 --- a/act/container/docker_run.go +++ b/act/container/docker_run.go @@ -938,10 +938,7 @@ func (cr *containerReference) copyDir(dstPath, srcPath string, useGitIgnore, ski logger.Error(err) } }() - extractionRoot := "/" - if actPath := cr.GetActPath(); strings.HasPrefix(dstPath+"/", actPath+"/") { - extractionRoot = actPath // keeps tar entries from crossing the image's /var/run symlink - } + extractionRoot := cr.extractionRoot(dstPath) if err := writeDirTar(ctx, tarFile, path.Join("/", strings.TrimPrefix(dstPath, extractionRoot)), srcPath, useGitIgnore, skipGitDir, cr.UID, cr.GID, cr.markExecutable(ctx)); err != nil { return err } @@ -954,6 +951,44 @@ func (cr *containerReference) copyDir(dstPath, srcPath string, useGitIgnore, ski } } +func (cr *containerReference) extractionRoot(dstPath string) string { + if actPath := cr.GetActPath(); strings.HasPrefix(dstPath+"/", actPath+"/") { + return actPath // keeps tar entries from crossing the image's /var/run symlink + } + return "/" +} + +func (cr *containerReference) CopyTarStream(ctx context.Context, destPath string, write func(*TarWriter) error) error { + if common.Dryrun(ctx) { + return nil + } + if err := cr.connect()(ctx); err != nil { + return err + } + if err := cr.find()(ctx); err != nil { + return err + } + if cr.id == "" { + return cr.missingContainerError("copy to %s", destPath) + } + destPath = path.Clean(destPath) + root := cr.extractionRoot(destPath) + rel := func(fullPath string) string { return strings.TrimPrefix(strings.TrimPrefix(fullPath, root), "/") } + var parents []string // sent as entries so they belong to the job user, the daemon creates missing ones as root + for dir := destPath; dir != root && dir != "/" && dir != "."; dir = path.Dir(dir) { + if _, err := cr.cli.ContainerStatPath(ctx, cr.id, client.ContainerStatPathOptions{Path: dir}); err == nil { + break + } else if !cerrdefs.IsNotFound(err) { + return fmt.Errorf("stat %s: %w", dir, err) + } + parents = append(parents, rel(dir)) + } + slices.Reverse(parents) + content := pipeTar(cr.UID, cr.GID, rel(destPath), parents, write) + defer content.Close() + return cr.upload(ctx, root, content) +} + func (cr *containerReference) copyContent(dstPath string, files ...*FileEntry) common.Executor { return func(ctx context.Context) error { if cr.id == "" { diff --git a/act/container/docker_run_test.go b/act/container/docker_run_test.go index 1f5ee085..628c3954 100644 --- a/act/container/docker_run_test.go +++ b/act/container/docker_run_test.go @@ -122,6 +122,11 @@ func (m *mockDockerClient) CopyToContainer(ctx context.Context, id string, optio return args.Get(0).(mobyclient.CopyToContainerResult), args.Error(1) } +func (m *mockDockerClient) ContainerStatPath(ctx context.Context, id string, options mobyclient.ContainerStatPathOptions) (mobyclient.ContainerStatPathResult, error) { + args := m.Called(ctx, id, options) + return args.Get(0).(mobyclient.ContainerStatPathResult), args.Error(1) +} + func (m *mockDockerClient) ContainerInspect(ctx context.Context, id string, opts mobyclient.ContainerInspectOptions) (mobyclient.ContainerInspectResult, error) { args := m.Called(ctx, id, opts) return args.Get(0).(mobyclient.ContainerInspectResult), args.Error(1) @@ -539,6 +544,46 @@ func TestRejectsMissingContainer(t *testing.T) { check("Inspect after removal", err) } +func TestCopyTarStreamOwnsMissingParentsAndRejectsEscapes(t *testing.T) { + ctx := t.Context() + client := &mockDockerClient{} + client.On("ContainerInspect", ctx, "job", mobyclient.ContainerInspectOptions{}). + Return(mobyclient.ContainerInspectResult{Container: container.InspectResponse{ID: "job", Name: "/job"}}, nil) + stat := func(p string, err error) { + client.On("ContainerStatPath", ctx, "job", mobyclient.ContainerStatPathOptions{Path: p}). + Return(mobyclient.ContainerStatPathResult{}, err).Once() + } + stat("/w/out/a", cerrdefs.ErrNotFound) + stat("/w/out", cerrdefs.ErrNotFound) + stat("/w", nil) + var entries []string + client.On("CopyToContainer", ctx, "job", mock.Anything).Run(func(args mock.Arguments) { + opts := args.Get(2).(mobyclient.CopyToContainerOptions) + assert.Equal(t, "/", opts.DestinationPath) + reader := tar.NewReader(opts.Content) + for { + header, err := reader.Next() + if errors.Is(err, io.EOF) { + return + } + require.NoError(t, err) + assert.Equal(t, []any{1001, 1002, ""}, []any{header.Uid, header.Gid, header.Uname}, header.Name) + entries = append(entries, header.Name) + } + }).Return(mobyclient.CopyToContainerResult{}, nil).Once() + + cr := &containerReference{id: "job", cli: client, input: &NewContainerInput{Name: "job"}, UID: 1001, GID: 1002} + require.NoError(t, cr.CopyTarStream(ctx, "/w/out/a/", func(tw *TarWriter) error { + return tw.WriteHeader(&tar.Header{Typeflag: tar.TypeReg, Name: "dir/f.txt", Mode: 0o644, Uname: "nobody"}) + })) + assert.Equal(t, []string{"w/out/", "w/out/a/", "w/out/a/dir/", "w/out/a/dir/f.txt"}, entries) + client.AssertExpectations(t) + + require.ErrorContains(t, writeTarStream(io.Discard, 0, 0, "w/out/a", nil, func(tw *TarWriter) error { + return tw.WriteHeader(&tar.Header{Typeflag: tar.TypeReg, Name: "../../../etc/passwd"}) + }), "escapes") +} + func TestPublicCopyPipelineHandlesStaleID(t *testing.T) { source := t.TempDir() require.NoError(t, os.WriteFile(filepath.Join(source, "x"), nil, 0o644)) diff --git a/act/container/host_environment.go b/act/container/host_environment.go index 5a59dffc..18b97d90 100644 --- a/act/container/host_environment.go +++ b/act/container/host_environment.go @@ -6,7 +6,6 @@ package container import ( "archive/tar" - "bytes" "cmp" "context" "errors" @@ -17,6 +16,7 @@ import ( "os/exec" "path/filepath" "runtime" + "slices" "strings" "sync" "sync/atomic" @@ -117,17 +117,24 @@ func (e *HostEnvironment) Inspect(_ context.Context) (*Info, error) { } func (e *HostEnvironment) GetContainerArchive(ctx context.Context, srcPath string) (io.ReadCloser, error) { - buf := &bytes.Buffer{} - tw := tar.NewWriter(buf) - defer tw.Close() srcPath = filepath.Clean(srcPath) fi, err := os.Lstat(srcPath) if err != nil { return nil, err } - tc := &filecollector.TarCollector{ - TarWriter: tw, - } + reader, writer := io.Pipe() + go func() { + tw := tar.NewWriter(writer) + err := writeHostArchive(ctx, &filecollector.TarCollector{TarWriter: tw}, srcPath, fi) + if err == nil { + err = tw.Close() + } + writer.CloseWithError(err) + }() + return startedArchive(reader) +} + +func writeHostArchive(ctx context.Context, tc *filecollector.TarCollector, srcPath string, fi fs.FileInfo) error { if fi.IsDir() { srcPrefix := srcPath if !strings.HasSuffix(srcPrefix, string(filepath.Separator)) { @@ -138,31 +145,74 @@ func (e *HostEnvironment) GetContainerArchive(ctx context.Context, srcPath strin SrcPrefix: srcPrefix, Handler: tc, } - err = filepath.Walk(srcPath, fc.CollectFiles(ctx)) - if err != nil { - return nil, err + return filepath.Walk(srcPath, fc.CollectFiles(ctx)) + } + var f io.ReadCloser + var linkname string + var err error + if fi.Mode()&fs.ModeSymlink != 0 { + if linkname, err = os.Readlink(srcPath); err != nil { + return err } } else { - var f io.ReadCloser - var linkname string - if fi.Mode()&fs.ModeSymlink != 0 { - linkname, err = os.Readlink(srcPath) - if err != nil { - return nil, err - } + if f, err = os.Open(srcPath); err != nil { + return err + } + defer f.Close() + } + return tc.WriteFile(fi.Name(), fi, linkname, f) +} + +func (e *HostEnvironment) CopyTarStream(ctx context.Context, destPath string, write func(*TarWriter) error) error { + if err := os.MkdirAll(destPath, 0o777); err != nil { + return err + } + root, err := os.OpenRoot(destPath) + if err != nil { + return err + } + defer root.Close() + content := pipeTar(0, 0, "", nil, write) + defer content.Close() + tr := tar.NewReader(content) + var dirs []*tar.Header + for { + if err := ctx.Err(); err != nil { + return err + } + hdr, err := tr.Next() + if errors.Is(err, io.EOF) { + break + } else if err != nil { + return err + } + hdr.Name = filepath.FromSlash(strings.TrimSuffix(hdr.Name, "/")) + if hdr.Typeflag == tar.TypeDir { + err = root.MkdirAll(hdr.Name, 0o777) // modes apply after the contents, so read-only directories can be filled + dirs = append(dirs, hdr) } else { - f, err = os.Open(srcPath) - if err != nil { - return nil, err - } - defer f.Close() + err = writeRootFile(root, hdr.Name, hdr.FileInfo().Mode().Perm(), tr) } - err := tc.WriteFile(fi.Name(), fi, linkname, f) if err != nil { - return nil, err + return err } } - return io.NopCloser(buf), nil + for _, dir := range slices.Backward(dirs) { + if err := root.Chmod(dir.Name, dir.FileInfo().Mode().Perm()); err != nil { + return err + } + } + return nil +} + +func writeRootFile(root *os.Root, name string, mode fs.FileMode, body io.Reader) error { + _ = root.Remove(name) // replaces read-only files and applies mode, as the daemon and tar do + file, err := root.OpenFile(name, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, mode) + if err != nil { + return err + } + _, err = io.Copy(file, body) + return errors.Join(err, file.Close()) } func (e *HostEnvironment) Pull(_ bool) common.Executor { diff --git a/act/container/host_environment_test.go b/act/container/host_environment_test.go index cf3a8abe..5bcd4a3c 100644 --- a/act/container/host_environment_test.go +++ b/act/container/host_environment_test.go @@ -8,6 +8,7 @@ import ( "archive/tar" "bytes" "context" + "fmt" "io" "os" "path/filepath" @@ -76,7 +77,54 @@ func TestGetContainerArchive(t *testing.T) { require.NoError(t, err) assert.Equal(t, expectedContent, content) _, err = reader.Next() - assert.ErrorIs(t, err, io.EOF) + require.ErrorIs(t, err, io.EOF) + require.NoError(t, archive.Close()) + + _, err = e.GetContainerArchive(ctx, filepath.Join(e.Path, "missing")) + assert.ErrorIs(t, err, os.ErrNotExist) +} + +func TestHostCopyTarStream(t *testing.T) { + dest := filepath.Join(t.TempDir(), "out") + write := func(name string, mode int64, body string) func(*TarWriter) error { + return func(tw *TarWriter) error { + if err := tw.WriteHeader(&tar.Header{Typeflag: tar.TypeReg, Name: name, Mode: mode, Size: int64(len(body))}); err != nil { + return err + } + _, err := tw.Write([]byte(body)) + return err + } + } + e := &HostEnvironment{} + require.NoError(t, e.CopyTarStream(t.Context(), dest, write("dir/run.sh", 0o755, "#!/bin/sh\n"))) + content, err := os.ReadFile(filepath.Join(dest, "dir", "run.sh")) + require.NoError(t, err) + assert.Equal(t, "#!/bin/sh\n", string(content)) + if runtime.GOOS != "windows" { + require.NoError(t, os.Chmod(filepath.Join(dest, "dir", "run.sh"), 0o444)) + require.NoError(t, e.CopyTarStream(t.Context(), dest, func(tw *TarWriter) error { + if err := tw.WriteHeader(&tar.Header{Typeflag: tar.TypeDir, Name: "dir/", Mode: 0o555}); err != nil { + return err + } + return write("dir/run.sh", 0o755, "")(tw) + })) + t.Cleanup(func() { require.NoError(t, os.Chmod(filepath.Join(dest, "dir"), 0o755)) }) + for name, mode := range map[string]os.FileMode{"dir": 0o555, filepath.Join("dir", "run.sh"): 0o755} { + info, err := os.Stat(filepath.Join(dest, name)) + require.NoError(t, err) + assert.Equal(t, mode, info.Mode().Perm(), name) + } + require.NoError(t, os.Symlink(t.TempDir(), filepath.Join(dest, "link"))) + require.ErrorContains(t, e.CopyTarStream(t.Context(), dest, write("link/x", 0o644, "x")), "escapes") + } + + require.ErrorContains(t, e.CopyTarStream(t.Context(), dest, write("../escape", 0o644, "x")), "escapes") + assert.NoFileExists(t, filepath.Join(filepath.Dir(dest), "escape")) + require.Error(t, e.CopyTarStream(t.Context(), dest, func(_tw *TarWriter) error { return fmt.Errorf("read body: %w", io.EOF) })) + canceled, cancel := context.WithCancel(t.Context()) + cancel() + require.ErrorIs(t, e.CopyTarStream(canceled, dest, write("late", 0o644, "x")), context.Canceled) + assert.NoFileExists(t, filepath.Join(dest, "late")) } func TestHostEnvironmentExecExitCode(t *testing.T) { diff --git a/act/container/kubernetes.go b/act/container/kubernetes.go index a7953d42..9f554ff4 100644 --- a/act/container/kubernetes.go +++ b/act/container/kubernetes.go @@ -4,7 +4,6 @@ package container import ( - "bufio" "bytes" "cmp" "compress/gzip" @@ -554,6 +553,10 @@ func (c *kubernetesContainer) Copy(destPath string, files ...*FileEntry) common. } } +func (c *kubernetesContainer) CopyTarStream(ctx context.Context, destPath string, write func(*TarWriter) error) error { + return c.extract(ctx, destPath, true, func(writer io.Writer) error { return writeTarStream(writer, 0, 0, "", nil, write) }) // streams carry large content like artifacts +} + func (c *kubernetesContainer) CopyDir(destPath, srcPath string, useGitIgnore, skipGitDir bool) common.Executor { return func(ctx context.Context) error { return c.extract(ctx, destPath, true, func(writer io.Writer) error { // directories are actions and workspaces, megabytes of compressible sources @@ -573,15 +576,7 @@ func (c *kubernetesContainer) GetContainerArchive(ctx context.Context, srcPath s } writer.CloseWithError(err) }() - buffered := bufio.NewReader(reader) - if _, err := buffered.Peek(1); err != nil { - reader.Close() - return nil, err - } - return struct { - io.Reader - io.Closer - }{buffered, reader}, nil + return startedArchive(reader) } func (c *kubernetesContainer) UpdateFromEnv(srcPath string, env *map[string]string) common.Executor { diff --git a/act/container/kubernetes_test.go b/act/container/kubernetes_test.go index 04192955..48c731c4 100644 --- a/act/container/kubernetes_test.go +++ b/act/container/kubernetes_test.go @@ -309,6 +309,18 @@ func TestKubernetesPodRunsJobThroughAPI(t *testing.T) { assert.Contains(t, lastExec(), "command="+url.QueryEscape(workdir+"/act/actions/sha/")) assert.Contains(t, lastExec(), url.QueryEscape("gzip -dc |")) + require.NoError(t, job.CopyTarStream(ctx, workdir+"/download/new", func(tw *TarWriter) error { + if err := tw.WriteHeader(&tar.Header{Typeflag: tar.TypeReg, Name: "dir/data.txt", Mode: 0o644, Size: 7}); err != nil { + return err + } + _, err := tw.Write([]byte("payload")) + return err + })) + downloaded, err := os.ReadFile(workdir + "/download/new/dir/data.txt") + require.NoError(t, err) + assert.Equal(t, "payload", string(downloaded)) + assert.Contains(t, lastExec(), url.QueryEscape("gzip -dc |")) + pathEnv := map[string]string{} require.NoError(t, job.UpdateFromImageEnv(&pathEnv)(ctx)) assert.Equal(t, path, pathEnv["PATH"]) diff --git a/act/container/tar.go b/act/container/tar.go index 88a4ee51..f36a6329 100644 --- a/act/container/tar.go +++ b/act/container/tar.go @@ -6,8 +6,13 @@ package container import ( "archive/tar" + "bufio" "context" + "errors" + "fmt" "io" + "io/fs" + "path" "path/filepath" "strings" @@ -15,6 +20,84 @@ import ( "gitea.com/gitea/runner/act/filecollector" ) +// TarWriter takes directories and files named relative to the destination, owned by the job user. +type TarWriter struct { + writer *tar.Writer + uid, gid int + prefix string + dirs map[string]bool +} + +func (w *TarWriter) Write(data []byte) (int, error) { + return w.writer.Write(data) +} + +func (w *TarWriter) WriteHeader(hdr *tar.Header) error { + name := strings.TrimSuffix(hdr.Name, "/") + if !fs.ValidPath(name) { + return fmt.Errorf("tar entry %q escapes the destination", hdr.Name) + } + if hdr.Typeflag != tar.TypeDir && hdr.Typeflag != tar.TypeReg { + return fmt.Errorf("tar entry %q: unsupported type %q", hdr.Name, hdr.Typeflag) + } + for index := range len(name) { + if dir := name[:index]; name[index] == '/' && !w.dirs[dir] { + if err := w.write(&tar.Header{Typeflag: tar.TypeDir, Name: path.Join(w.prefix, dir), Mode: 0o755}); err != nil { + return err + } + w.dirs[dir] = true + } + } + if hdr.Typeflag == tar.TypeDir { + w.dirs[name] = true + } + entry := *hdr + entry.Name = path.Join(w.prefix, name) + return w.write(&entry) +} + +func (w *TarWriter) write(hdr *tar.Header) error { + hdr.Uid, hdr.Gid, hdr.Uname, hdr.Gname = w.uid, w.gid, "", "" // tar running as root prefers names over ids + if hdr.Typeflag == tar.TypeDir { + hdr.Name += "/" + } + return w.writer.WriteHeader(hdr) +} + +func writeTarStream(dst io.Writer, uid, gid int, prefix string, parents []string, write func(*TarWriter) error) error { + tw := &TarWriter{writer: tar.NewWriter(dst), uid: uid, gid: gid, prefix: prefix, dirs: map[string]bool{}} + for _, name := range parents { + if err := tw.write(&tar.Header{Typeflag: tar.TypeDir, Name: name, Mode: 0o755}); err != nil { + return err + } + } + if err := write(tw); errors.Is(err, io.EOF) { + return fmt.Errorf("tar stream: %v", err) // %v, as a wrapped io.EOF reads as the archive's end + } else if err != nil { + return err + } + return tw.writer.Close() +} + +func pipeTar(uid, gid int, prefix string, parents []string, write func(*TarWriter) error) *io.PipeReader { + reader, writer := io.Pipe() + go func() { writer.CloseWithError(writeTarStream(writer, uid, gid, prefix, parents, write)) }() + return reader +} + +// startedArchive returns reader once its first byte arrives, so failing to start archiving errors here, not on read. +func startedArchive(reader *io.PipeReader) (io.ReadCloser, error) { + buffered := bufio.NewReader(reader) + if _, err := buffered.Peek(1); err != nil { + reader.Close() + return nil, err + } + return struct { + io.Reader + io.Closer + }{buffered, reader}, nil +} + func writeFilesTar(ctx context.Context, w io.Writer, uid, gid int, files ...*FileEntry) error { tw := tar.NewWriter(w) for _, file := range files { diff --git a/act/runner/container_mock_test.go b/act/runner/container_mock_test.go index 377366de..4d585396 100644 --- a/act/runner/container_mock_test.go +++ b/act/runner/container_mock_test.go @@ -82,3 +82,7 @@ func (cm *containerMock) Inspect(ctx context.Context) (*container.Info, error) { err, _ := args.Get(1).(error) return info, err } + +func (cm *containerMock) CopyTarStream(ctx context.Context, destPath string, write func(*container.TarWriter) error) error { + return cm.Called(ctx, destPath, write).Error(0) +}