mesh-controller/internal/builder/upstream_file.go

package builder

import (
	"archive/tar"
	"bufio"
	"bytes"
	"compress/bzip2"
	"compress/gzip"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"fmt"
	"io"
	"net/http"
	"os"
	"os/exec"
	"path"
	"path/filepath"
	"strings"
	"time"

	"github.com/novox/mesh-controller/internal/catalogue"
)

// An archive whose source is somebody else's published file, pinned by URL and sha256 (novox/hq ADR 0316
// §2, §6, to-be 56 §7): a mail client's release, an add-on's xpi.
//
// **Fetched by the build seat, checked against the pin before anything is read from it, and packed as any
// archive is packed.** What a machine receives is the mesh's own archive from the artifact store, by digest,
// exactly as for a directory of the repository; no machine ever asks the vendor. A file whose bytes are not
// the pin's fails the build and names both digests, because the address serving something else is the one
// thing the pin exists to catch.
//
// **Kept in the build seat's workspace under its sha256**, so it is fetched once: a vendor that serves only
// its newest version at one address (Owl) would otherwise break every later build of the same pin.

// maxUpstream is the largest upstream file fetched, and the largest a release may unpack to: the node-engine
// unpacks no archive larger than this either (mesh-host internal/apply/archive.go, maxArchive).
const maxUpstream = 512 << 20

// upstreamClient fetches upstream files: a variable so a test serves its own.
var upstreamClient = &http.Client{Timeout: 15 * time.Minute}

// upstreamDir is where the build seat keeps the upstream files it fetched, by sha256.
func upstreamDir(workspace string) string { return filepath.Join(workspace, "upstream") }

// upstreamArchive is the packed archive of an artifact pinned by URL: the file fetched (or kept from an
// earlier build) and checked, then laid out as the artifact says and packed.
func upstreamArchive(ctx context.Context, workspace string, a catalogue.Artifact,
	say func(step, format string, args ...any)) ([]byte, error) {
	file, err := upstreamFile(ctx, workspace, a, say)
	if err != nil {
		return nil, err
	}
	laid := filepath.Join(workspace, "upstream-"+a.Name)
	if err := os.RemoveAll(laid); err != nil {
		return nil, err
	}
	defer os.RemoveAll(laid)
	if err := os.MkdirAll(laid, 0o755); err != nil {
		return nil, err
	}
	if a.File != "" {
		body, err := os.ReadFile(file)
		if err != nil {
			return nil, err
		}
		if err := os.WriteFile(filepath.Join(laid, a.File), body, 0o644); err != nil {
			return nil, err
		}
	} else if err := unpackUpstream(ctx, file, laid, a.Within); err != nil {
		return nil, fmt.Errorf("%s (%s): %w", a.URL, a.Version, err)
	}
	return pack(laid)
}

// upstreamFile is the path of the pinned file in the workspace: kept from an earlier build when its bytes
// are still the pin's, else fetched now and kept only when they are.
func upstreamFile(ctx context.Context, workspace string, a catalogue.Artifact,
	say func(step, format string, args ...any)) (string, error) {
	want := strings.ToLower(a.SHA256)
	dir := upstreamDir(workspace)
	kept := filepath.Join(dir, want)
	if got, err := sumOf(kept); err == nil {
		if got == want {
			say("upstream", "%s %s: kept from an earlier build (sha256:%s)", a.Name, a.Version, short(want))
			return kept, nil
		}
		// Changed on disk since it was kept: fetched again rather than trusted.
		os.Remove(kept)
	}
	if err := os.MkdirAll(dir, 0o755); err != nil {
		return "", err
	}
	say("upstream", "fetching %s (%s)", a.URL, a.Version)
	request, err := http.NewRequestWithContext(ctx, http.MethodGet, a.URL, nil)
	if err != nil {
		return "", err
	}
	response, err := upstreamClient.Do(request)
	if err != nil {
		return "", fmt.Errorf("cannot fetch %s: %w", a.URL, err)
	}
	defer response.Body.Close()
	if response.StatusCode != http.StatusOK {
		return "", fmt.Errorf("%s answered %s", a.URL, response.Status)
	}
	part, err := os.CreateTemp(dir, want+".*.part")
	if err != nil {
		return "", err
	}
	defer os.Remove(part.Name())
	hash := sha256.New()
	n, err := io.Copy(io.MultiWriter(part, hash), io.LimitReader(response.Body, maxUpstream+1))
	if cerr := part.Close(); err == nil {
		err = cerr
	}
	if err != nil {
		return "", fmt.Errorf("fetching %s: %w", a.URL, err)
	}
	if n > maxUpstream {
		return "", fmt.Errorf("%s is larger than %d bytes, more than a machine unpacks; nothing was kept", a.URL, maxUpstream)
	}
	got := hex.EncodeToString(hash.Sum(nil))
	if got != want {
		return "", fmt.Errorf("%s was pinned as sha256:%s and what it served is sha256:%s (%d bytes); nothing "+
			"was kept. The address serves another file than the one reviewed: compare it, then move the pin "+
			"in a reviewed change (novox/hq ADR 0316)", a.URL, want, got, n)
	}
	if err := os.Rename(part.Name(), kept); err != nil {
		return "", err
	}
	say("upstream", "%s %s: %d bytes, sha256:%s as pinned", a.Name, a.Version, n, short(want))
	return kept, nil
}

// sumOf is the sha256 of a file, in hexadecimal.
func sumOf(file string) (string, error) {
	f, err := os.Open(file)
	if err != nil {
		return "", err
	}
	defer f.Close()
	hash := sha256.New()
	if _, err := io.Copy(hash, f); err != nil {
		return "", err
	}
	return hex.EncodeToString(hash.Sum(nil)), nil
}

// unpackUpstream unpacks a tar archive, compressed with gzip, bzip2 or xz (told by its first bytes), into a
// directory: of its entries only those within `within` (all, when empty), with that prefix taken off.
//
// **Only files and directories, and nothing outside.** A link, a device or an entry naming a path outside
// the archive, or outside `within`, refuses the build by name: the node-engine unpacks only files and
// directories, and an archive quietly unpacked without what it held would be a release that is not the one
// pinned.
func unpackUpstream(ctx context.Context, file, into, within string) error {
	f, err := os.Open(file)
	if err != nil {
		return err
	}
	defer f.Close()
	buffered := bufio.NewReader(f)
	magic, _ := buffered.Peek(6)
	var stream io.Reader
	var wait func() error
	switch {
	case bytes.HasPrefix(magic, []byte{0x1f, 0x8b}):
		zipped, err := gzip.NewReader(buffered)
		if err != nil {
			return err
		}
		defer zipped.Close()
		stream = zipped
	case bytes.HasPrefix(magic, []byte("BZh")):
		stream = bzip2.NewReader(buffered)
	case bytes.HasPrefix(magic, []byte{0xfd, '7', 'z', 'X', 'Z', 0x00}):
		// Go's standard library reads no xz; the build seat's image carries xz (cmd/mesh-builder/Dockerfile).
		if _, err := exec.LookPath("xz"); err != nil {
			return errors.New("it is compressed with xz, and this build seat has no xz to read it")
		}
		cmd := exec.CommandContext(ctx, "xz", "-dc")
		cmd.Stdin = buffered
		var stderr bytes.Buffer
		cmd.Stderr = &stderr
		out, err := cmd.StdoutPipe()
		if err != nil {
			return err
		}
		if err := cmd.Start(); err != nil {
			return err
		}
		stream = out
		wait = func() error {
			if err := cmd.Wait(); err != nil {
				return fmt.Errorf("xz could not read it: %v %s", err, strings.TrimSpace(stderr.String()))
			}
			return nil
		}
		defer func() {
			if wait != nil {
				_ = cmd.Process.Kill()
				_ = cmd.Wait()
			}
		}()
	default:
		return errors.New("it is not a tar archive compressed with gzip, bzip2 or xz; a single file is kept " +
			"whole by naming it in `file`")
	}
	if err := untar(stream, into, within); err != nil {
		return err
	}
	if wait != nil {
		w := wait
		wait = nil
		return w()
	}
	return nil
}

// untar writes a tar stream's files and directories within `within` into a directory.
func untar(stream io.Reader, into, within string) error {
	reader := tar.NewReader(stream)
	prefix := ""
	if within != "" {
		prefix = path.Clean(within) + "/"
	}
	var total int64
	files := 0
	for {
		header, err := reader.Next()
		if err == io.EOF {
			break
		}
		if err != nil {
			return err
		}
		name := path.Clean(strings.TrimPrefix(header.Name, "./"))
		if path.IsAbs(name) || name == ".." || strings.HasPrefix(name, "../") {
			return fmt.Errorf("%s names a path outside the archive", header.Name)
		}
		if name == "." || name+"/" == prefix {
			continue
		}
		if prefix != "" {
			if !strings.HasPrefix(name, prefix) {
				return fmt.Errorf("%s is outside %s, the directory the artifact is taken from", header.Name, within)
			}
			name = strings.TrimPrefix(name, prefix)
		}
		target := filepath.Join(into, filepath.FromSlash(name))
		switch header.Typeflag {
		case tar.TypeDir:
			if err := os.MkdirAll(target, 0o755); err != nil {
				return err
			}
		case tar.TypeReg:
			total += header.Size
			if total > maxUpstream {
				return fmt.Errorf("it unpacks to more than %d bytes, more than a machine unpacks", maxUpstream)
			}
			if err := os.MkdirAll(filepath.Dir(target), 0o755); err != nil {
				return err
			}
			mode := os.FileMode(0o644)
			if header.Mode&0o111 != 0 {
				mode = 0o755
			}
			out, err := os.OpenFile(target, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, mode)
			if err != nil {
				return err
			}
			if _, err := io.Copy(out, io.LimitReader(reader, header.Size)); err != nil {
				out.Close()
				return err
			}
			if err := out.Close(); err != nil {
				return err
			}
			files++
		default:
			return fmt.Errorf("%s is a %c, and an archive carries only files and directories", header.Name,
				header.Typeflag)
		}
	}
	if files == 0 {
		if within != "" {
			return fmt.Errorf("it holds no file within %s", within)
		}
		return errors.New("it holds no file")
	}
	return nil
}