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 <[email protected]>
Co-authored-by: silverwind <[email protected]>
Co-authored-by: Lunny Xiao <[email protected]>
Reviewed-on: https://gitea.com/gitea/runner/pulls/1340
Reviewed-by: Lunny Xiao <[email protected]>
Reviewed-by: silverwind <[email protected]>
Co-authored-by: bircni <[email protected]>

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

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 == "" {

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))

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 {

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) {

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 {

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"])

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 {

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)
+}