feat: validate the machine configuration in the installer container

Talos validates machine configuration at boot time, and refuses to boot
if machine configuration is invalid.

As machine configuration validation rules might change over time, we
need to prevent a scenario when after an upgrade machine configuration
becomes invalid, as there's no way to roll back properly.

Machine configuration is submitted over stdin to the installer
container, and installer container validates it using the new version of
Talos (which is going to be installed).

If the config is not sent over stdin, installer assumes old version of
Talos and proceeds.

This should be backported to 0.9 to allow config validation on upgrade
to 0.10.

Fixes siderolabs#3419

Signed-off-by: Andrey Smirnov <smirnov.andrey@gmail.com>
(cherry picked from commit d5e2a45)

cmd/installer/cmd/install.go

@@ -5,13 +5,16 @@
 package cmd
 
 import (
+	"errors"
+	"fmt"
 	"log"
 
 	"github.com/spf13/cobra"
 
 	"github.com/talos-systems/talos/cmd/installer/pkg/install"
 	"github.com/talos-systems/talos/internal/app/machined/pkg/runtime"
 	"github.com/talos-systems/talos/internal/app/machined/pkg/runtime/v1alpha1/platform"
+	"github.com/talos-systems/talos/pkg/machinery/config/configloader"
 	"github.com/talos-systems/talos/pkg/version"
 )
 
@@ -49,9 +52,18 @@ func runInstallCmd() (err error) {
 		return err
 	}
 
-	if err = install.Install(p, seq, options); err != nil {
-		return err
+	config, err := configloader.NewFromStdin()
+	if err != nil {
+		if errors.Is(err, configloader.ErrNoConfig) {
+			log.Printf("machine configuration missing, skipping validation")
+		} else {
+			return fmt.Errorf("error loading machine configuration: %w", err)
+		}
+	} else {
+		if err = config.Validate(p.Mode()); err != nil {
+			return fmt.Errorf("machine configuration is invalid: %w", err)
+		}
 	}
 
-	return nil
+	return install.Install(p, seq, options)
 }

internal/app/machined/internal/install/install.go

@@ -5,6 +5,7 @@
 package install
 
 import (
+	"bytes"
 	"context"
 	"fmt"
 	"log"
@@ -21,6 +22,7 @@ import (
 	"golang.org/x/sys/unix"
 
 	"github.com/talos-systems/talos/internal/app/machined/pkg/runtime"
+	containerdrunner "github.com/talos-systems/talos/internal/app/machined/pkg/system/runner/containerd"
 	"github.com/talos-systems/talos/internal/pkg/containers/image"
 	"github.com/talos-systems/talos/internal/pkg/kmsg"
 	machineapi "github.com/talos-systems/talos/pkg/machinery/api/machine"
@@ -31,7 +33,7 @@ import (
 // RunInstallerContainer performs an installation via the installer container.
 //
 //nolint:gocyclo
-func RunInstallerContainer(disk, platform, ref string, reg config.Registries, opts ...Option) error {
+func RunInstallerContainer(disk, platform, ref string, configBytes []byte, reg config.Registries, opts ...Option) error {
 	options := DefaultInstallOptions()
 
 	for _, opt := range opts {
@@ -40,7 +42,10 @@ func RunInstallerContainer(disk, platform, ref string, reg config.Registries, op
 		}
 	}
 
-	ctx := namespaces.WithNamespace(context.Background(), constants.SystemContainerdNamespace)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	ctx = namespaces.WithNamespace(ctx, constants.SystemContainerdNamespace)
 
 	client, err := containerd.New(constants.SystemContainerdAddress)
 	if err != nil {
@@ -134,13 +139,20 @@ func RunInstallerContainer(disk, platform, ref string, reg config.Registries, op
 
 	w := &kmsg.Writer{KmsgWriter: f}
 
-	creator := cio.NewCreator(cio.WithStreams(nil, w, w))
+	configR := &containerdrunner.StdinCloser{
+		Stdin:  bytes.NewReader(configBytes),
+		Closer: make(chan struct{}),
+	}
+
+	creator := cio.NewCreator(cio.WithStreams(configR, w, w))
 
 	t, err := container.NewTask(ctx, creator)
 	if err != nil {
 		return err
 	}
 
+	go configR.WaitAndClose(ctx, t)
+
 	defer t.Delete(ctx) //nolint:errcheck
 
 	if err = t.Start(ctx); err != nil {

internal/app/machined/pkg/runtime/v1alpha1/v1alpha1_sequencer_tasks.go

@@ -1471,11 +1471,17 @@ func Upgrade(seq runtime.Sequence, data interface{}) (runtime.TaskExecutionFunc,
 
 		logger.Printf("performing upgrade via %q", in.GetImage())
 
+		configBytes, err := r.Config().Bytes()
+		if err != nil {
+			return fmt.Errorf("error marshaling configuration: %w", err)
+		}
+
 		// We pull the installer image when we receive an upgrade request. No need
 		// to pull it again.
 		err = install.RunInstallerContainer(
 			devname, r.State().Platform().Name(),
 			in.GetImage(),
+			configBytes,
 			r.Config().Machine().Registries(),
 			install.OptionsFromUpgradeRequest(r, in)...,
 		)
@@ -1698,6 +1704,11 @@ func UnmountEphemeralPartition(seq runtime.Sequence, data interface{}) (runtime.
 // Install mounts or installs the system partitions.
 func Install(seq runtime.Sequence, data interface{}) (runtime.TaskExecutionFunc, string) {
 	return func(ctx context.Context, logger *log.Logger, r runtime.Runtime) (err error) {
+		configBytes, err := r.Config().Bytes()
+		if err != nil {
+			return fmt.Errorf("error marshaling configuration: %w", err)
+		}
+
 		switch {
 		case !r.State().Machine().Installed():
 			installerImage := r.Config().Machine().Install().Image()
@@ -1709,6 +1720,7 @@ func Install(seq runtime.Sequence, data interface{}) (runtime.TaskExecutionFunc,
 				r.Config().Machine().Install().Disk(),
 				r.State().Platform().Name(),
 				installerImage,
+				configBytes,
 				r.Config().Machine().Registries(),
 				install.WithForce(true),
 				install.WithZero(r.Config().Machine().Install().Zero()),
@@ -1734,6 +1746,7 @@ func Install(seq runtime.Sequence, data interface{}) (runtime.TaskExecutionFunc,
 			err = install.RunInstallerContainer(
 				devname, r.State().Platform().Name(),
 				r.State().Machine().StagedInstallImageRef(),
+				configBytes,
 				r.Config().Machine().Registries(),
 				install.WithOptions(options),
 			)

internal/app/machined/pkg/system/runner/containerd/containerd.go

@@ -5,9 +5,11 @@
 package containerd
 
 import (
+	"bytes"
 	"context"
 	"fmt"
 	"io"
+	"io/ioutil"
 	"os"
 	"syscall"
 	"time"
@@ -33,7 +35,7 @@ type containerdRunner struct {
 	client      *containerd.Client
 	ctx         context.Context
 	container   containerd.Container
-	stdinCloser *stdinCloser
+	stdinCloser *StdinCloser
 }
 
 // NewRunner creates runner.Runner that runs a container in containerd.
@@ -151,15 +153,7 @@ func (c *containerdRunner) Run(eventSink events.Recorder) error {
 
 	if r != nil {
 		// See https://github.com/containerd/containerd/issues/4489.
-		go func() {
-			select {
-			case <-c.ctx.Done():
-				return
-			case <-c.stdinCloser.closer:
-				//nolint:errcheck
-				task.CloseIO(c.ctx, containerd.WithStdinCloser)
-			}
-		}()
+		go c.stdinCloser.WaitAndClose(c.ctx, task)
 	}
 
 	defer task.Delete(c.ctx) //nolint:errcheck
@@ -259,24 +253,17 @@ func (c *containerdRunner) StdinReader() (io.Reader, error) {
 		return nil, err
 	}
 
-	c.stdinCloser = &stdinCloser{
-		stdin:  c.opts.Stdin,
-		closer: make(chan struct{}),
+	// copy the input buffer as containerd API seems to be buggy:
+	//  * if the task fails to start, IO loop is not stopped properly, so after a restart there are two goroutines concurrently reading from stdin
+	contents, err := ioutil.ReadAll(c.opts.Stdin)
+	if err != nil {
+		return nil, err
 	}
 
-	return c.stdinCloser, nil
-}
-
-type stdinCloser struct {
-	stdin  io.Reader
-	closer chan struct{}
-}
-
-func (s *stdinCloser) Read(p []byte) (int, error) {
-	n, err := s.stdin.Read(p)
-	if err == io.EOF {
-		close(s.closer)
+	c.stdinCloser = &StdinCloser{
+		Stdin:  bytes.NewReader(contents),
+		Closer: make(chan struct{}),
 	}
 
-	return n, err
+	return c.stdinCloser, nil
 }

internal/app/machined/pkg/system/runner/containerd/stdin.go

@@ -0,0 +1,38 @@
+// 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 containerd
+
+import (
+	"context"
+	"io"
+
+	"github.com/containerd/containerd"
+)
+
+// StdinCloser wraps io.Reader providing a signal when reader is read till EOF.
+type StdinCloser struct {
+	Stdin  io.Reader
+	Closer chan struct{}
+}
+
+func (s *StdinCloser) Read(p []byte) (int, error) {
+	n, err := s.Stdin.Read(p)
+	if err == io.EOF {
+		close(s.Closer)
+	}
+
+	return n, err
+}
+
+// WaitAndClose closes containerd task stdin when StdinCloser is exhausted.
+func (s *StdinCloser) WaitAndClose(ctx context.Context, task containerd.Task) {
+	select {
+	case <-ctx.Done():
+		return
+	case <-s.Closer:
+		//nolint:errcheck
+		task.CloseIO(ctx, containerd.WithStdinCloser)
+	}
+}

pkg/machinery/config/configloader/configloader.go

@@ -7,6 +7,7 @@ package configloader
 
 import (
 	"bytes"
+	"errors"
 	"fmt"
 	"io"
 	"io/ioutil"
@@ -17,6 +18,9 @@ import (
 	"github.com/talos-systems/talos/pkg/machinery/config/types/v1alpha1"
 )
 
+// ErrNoConfig is returned when no configuration was found in the input.
+var ErrNoConfig = errors.New("config not found")
+
 // newConfig initializes and returns a Configurator.
 func newConfig(source []byte) (config config.Provider, err error) {
 	dec := decoder.NewDecoder(source)
@@ -34,7 +38,7 @@ func newConfig(source []byte) (config config.Provider, err error) {
 		}
 	}
 
-	return nil, fmt.Errorf("config not found")
+	return nil, ErrNoConfig
 }
 
 // NewFromFile will take a filepath and attempt to parse a config file from it.
@@ -58,7 +62,7 @@ func NewFromStdin() (config.Provider, error) {
 
 	config, err := NewFromBytes(buf.Bytes())
 	if err != nil {
-		return nil, fmt.Errorf("failed load config from stdin: %v", err)
+		return nil, fmt.Errorf("failed load config from stdin: %w", err)
 	}
 
 	return config, nil