talos/internal/app/lifecycle/container.go

// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at http://mozilla.org/MPL/2.0/.

package lifecycle

import (
	"bytes"
	"context"
	"crypto/rand"
	"encoding/hex"
	"fmt"
	"io"
	"log"
	"os"
	"strconv"

	"github.com/containerd/containerd/v2/client"
	"github.com/containerd/containerd/v2/core/leases"
	"github.com/containerd/containerd/v2/pkg/cio"
	"github.com/containerd/containerd/v2/pkg/oci"
	"github.com/containerd/errdefs"
	"github.com/opencontainers/runtime-spec/specs-go"
	"github.com/siderolabs/go-procfs/procfs"

	"github.com/siderolabs/talos/internal/app/internal/ctrhelper"
	"github.com/siderolabs/talos/internal/app/machined/pkg/system/pid"
	containerdrunner "github.com/siderolabs/talos/internal/app/machined/pkg/system/runner/containerd"
	"github.com/siderolabs/talos/internal/pkg/capability"
	"github.com/siderolabs/talos/internal/pkg/install"
	"github.com/siderolabs/talos/internal/pkg/selinux"
	"github.com/siderolabs/talos/pkg/machinery/api/common"
	configcore "github.com/siderolabs/talos/pkg/machinery/config"
	"github.com/siderolabs/talos/pkg/machinery/constants"
)

func generateContainerID() (string, error) {
	b := make([]byte, 8)
	if _, err := rand.Read(b); err != nil {
		return "", fmt.Errorf("failed to generate random ID: %w", err)
	}

	return fmt.Sprintf("installer-%s", hex.EncodeToString(b)), nil
}

// sendFunc is a callback to stream a message line back to the client.
type sendFunc func(msg string) error

// sendExitCodeFunc is a callback to stream the exit code back to the client.
type sendExitCodeFunc func(exitCode int32) error

// containerRunConfig holds all parameters needed to create and run the installer container.
type containerRunConfig struct {
	containerdInst *common.ContainerdInstance
	imageRef       string
	disk           string
	platform       string
	cfgContainer   configcore.Container
	opts           []install.Option

	send         sendFunc
	sendExitCode sendExitCodeFunc
}

// runInstallerContainer creates and runs the installer container synchronously,
// streaming output lines back to the client via the send callback.
//
//nolint:gocyclo,cyclop
func runInstallerContainer(ctx context.Context, pidRecorder pid.Recorder, rc *containerRunConfig) error {
	options := install.Options{Pull: true} //nolint:staticcheck
	if err := options.Apply(rc.opts...); err != nil {
		return fmt.Errorf("failed to apply install options: %w", err)
	}

	// connect to containerd
	ctx, detachedCtx, c8dClient, err := ctrhelper.ContainerdInstanceHelper(ctx, rc.containerdInst)
	if err != nil {
		return err
	}
	defer c8dClient.Close() //nolint:errcheck

	l, err := c8dClient.LeasesService().Create(ctx, leases.WithRandomID())
	if err != nil {
		return fmt.Errorf("failed to create lease: %w", err)
	}

	defer func() {
		if err := c8dClient.LeasesService().Delete(detachedCtx, l, leases.SynchronousDelete); err != nil {
			log.Printf("failed to delete lease %s: %v", l.ID, err)
		}
	}()

	ctx = leases.WithLease(ctx, l.ID)

	img, err := c8dClient.GetImage(ctx, rc.imageRef)
	if err != nil {
		return fmt.Errorf("installer image %q not found in containerd store: %w", rc.imageRef, err)
	}

	// build container spec
	mounts := buildMounts()
	args := buildInstallerArgs(rc.disk, rc.platform, &options)
	specOpts := buildSpecOpts(img, args, mounts)

	containerID, err := generateContainerID()
	if err != nil {
		return fmt.Errorf("failed to generate container ID: %v", err)
	}

	// create container
	ctr, err := c8dClient.NewContainer(ctx, containerID,
		client.WithImage(img),
		client.WithNewSnapshot(containerID, img),
		client.WithNewSpec(specOpts...),
	)
	if err != nil {
		return fmt.Errorf("failed to create container: %w", err)
	}

	defer func() {
		if cleanupErr := ctr.Delete(detachedCtx, client.WithSnapshotCleanup); cleanupErr != nil {
			log.Printf("failed to delete container %s: %v", ctr.ID(), cleanupErr)
		}
	}()

	// set up I/O: stdout/stderr -> pipe -> stream to client; stdin <- config bytes
	stdoutR, stdoutW := io.Pipe()

	var stdinReader interface {
		io.Reader
		WaitAndClose(context.Context, client.Task)
	}

	if rc.cfgContainer != nil {
		configBytes, cfgErr := rc.cfgContainer.Bytes()
		if cfgErr != nil {
			return fmt.Errorf("failed to serialize config: %w", cfgErr)
		}

		stdinReader = &containerdrunner.StdinCloser{
			Stdin:  bytes.NewReader(configBytes),
			Closer: make(chan struct{}),
		}
	}

	creator := cio.NewCreator(cio.WithStreams(stdinReader, stdoutW, stdoutW))

	task, err := ctr.NewTask(ctx, creator)
	if err != nil {
		stdoutW.Close() //nolint:errcheck

		return fmt.Errorf("failed to create task: %w", err)
	}

	defer func() {
		if _, delErr := task.Delete(detachedCtx, client.WithProcessKill); delErr != nil && !errdefs.IsNotFound(delErr) {
			log.Printf("failed to delete task: %v", delErr)
		}
	}()

	if stdinReader != nil {
		go stdinReader.WaitAndClose(ctx, task)
	}

	if err := task.Start(ctx); err != nil {
		stdoutW.Close() //nolint:errcheck

		return fmt.Errorf("failed to start task: %w", err)
	}

	if err := pidRecorder("installer", int32(task.Pid()), false); err != nil {
		return fmt.Errorf("failed to record installer PID: %w", err)
	}

	defer func() {
		if err := pidRecorder("installer", 0, true); err != nil {
			log.Printf("failed to clear installer PID record: %v", err)
		}
	}()

	statusC, err := task.Wait(detachedCtx)
	if err != nil {
		stdoutW.Close() //nolint:errcheck

		return fmt.Errorf("failed to wait for task: %w", err)
	}

	// stream output in a goroutine
	sendDone := make(chan error, 1)

	go func() {
		sendDone <- streamOutput(stdoutR, rc.send)
	}()

	// wait for task to exit
	exitStatus := <-statusC

	// close the write end so the reader gets EOF
	stdoutW.Close() //nolint:errcheck

	// wait for send loop to finish
	if sendErr := <-sendDone; sendErr != nil {
		log.Printf("error streaming output: %v", sendErr)
	}

	if exitStatus.Error() != nil {
		return fmt.Errorf("task exited with error: %w", exitStatus.Error())
	}

	exitCode := int32(exitStatus.ExitCode())

	// send exit code to client
	if err := rc.sendExitCode(exitCode); err != nil {
		return fmt.Errorf("failed to send exit code: %w", err)
	}

	if exitCode != 0 {
		log.Printf("installer container exited with code %d", exitCode)
	}

	return nil
}

// streamOutput reads from r line by line and sends each line via the send callback.
func streamOutput(r io.Reader, send sendFunc) error {
	buf := make([]byte, 512)

	for {
		n, err := r.Read(buf)
		if n > 0 {
			if sendErr := send(string(buf[:n])); sendErr != nil {
				return fmt.Errorf("failed to send message: %w", sendErr)
			}
		}

		if err != nil {
			if err == io.EOF {
				return nil
			}

			return fmt.Errorf("failed to read output: %w", err)
		}
	}
}

// buildMounts constructs the OCI mounts for the installer container.
func buildMounts() []specs.Mount {
	mounts := []specs.Mount{
		{Type: "bind", Destination: "/dev", Source: "/dev", Options: []string{"rbind", "rshared", "rw"}},
	}

	if _, err := os.Stat(constants.MachineSocketPath); err == nil {
		mounts = append(mounts, specs.Mount{
			Type: "bind", Destination: constants.MachineSocketPath,
			Source: constants.MachineSocketPath, Options: []string{"rbind", "rshared", "ro"},
		})
	}

	if _, err := os.Stat(constants.EFIVarsMountPoint); err == nil {
		mounts = append(mounts, specs.Mount{
			Type: "efivarfs", Source: "efivarfs",
			Destination: constants.EFIVarsMountPoint,
			Options:     []string{"rw", "nosuid", "nodev", "noexec", "relatime"},
		})
	}

	if _, err := os.Stat(constants.SDStubDynamicInitrdPath); err == nil {
		mounts = append(mounts, specs.Mount{
			Type: "bind", Destination: constants.SDStubDynamicInitrdPath,
			Source: constants.SDStubDynamicInitrdPath, Options: []string{"rbind", "rshared", "ro"},
		})
	}

	return mounts
}

// buildInstallerArgs constructs the command-line arguments for the installer binary.
func buildInstallerArgs(disk, platform string, options *install.Options) []string {
	config := constants.ConfigNone
	if c := procfs.ProcCmdline().Get(constants.KernelParamConfig).First(); c != nil {
		config = *c
	}

	args := []string{
		"/bin/installer",
		"install",
		"--disk=" + disk,
		"--platform=" + platform,
		"--config=" + config,
		"--upgrade=" + strconv.FormatBool(options.Upgrade),
		"--force=" + strconv.FormatBool(options.Force),
		"--zero=" + strconv.FormatBool(options.Zero),
	}

	for _, arg := range options.ExtraKernelArgs {
		args = append(args, "--extra-kernel-arg", arg)
	}

	for _, preservedArg := range []string{
		constants.KernelParamSideroLink,
		constants.KernelParamEventsSink,
		constants.KernelParamLoggingKernel,
		constants.KernelParamEquinixMetalEvents,
		constants.KernelParamAuditdDisabled,
		constants.KernelParamDashboardDisabled,
		constants.KernelParamNetIfnames,
		constants.KernelParamEnforceModuleSigVerify,
	} {
		if c := procfs.ProcCmdline().Get(preservedArg).First(); c != nil {
			args = append(args, "--extra-kernel-arg", fmt.Sprintf("%s=%s", preservedArg, *c))
		}
	}

	return args
}

// buildSpecOpts constructs the OCI spec options for the installer container.
func buildSpecOpts(img client.Image, args []string, mounts []specs.Mount) []oci.SpecOpts {
	specOpts := []oci.SpecOpts{
		oci.WithImageConfig(img),
		oci.WithProcessArgs(args...),
		oci.WithHostNamespace(specs.NetworkNamespace),
		oci.WithHostNamespace(specs.PIDNamespace),
		oci.WithMounts(mounts),
		oci.WithHostHostsFile,
		oci.WithHostResolvconf,
		oci.WithParentCgroupDevices,
		oci.WithCapabilities(capability.AllGrantableCapabilities()),
		oci.WithMaskedPaths(nil),
		oci.WithReadonlyPaths(nil),
		oci.WithWriteableSysfs,
		oci.WithWriteableCgroupfs,
		oci.WithApparmorProfile(""),
		oci.WithSeccompUnconfined,
		oci.WithAllDevicesAllowed,
	}

	if selinux.IsEnabled() {
		specOpts = append(specOpts, oci.WithSelinuxLabel(constants.SelinuxLabelInstaller))
	}

	return specOpts
}