[Ingest Manager] Agent verifies packages before using them (elastic#1…

…8876)

[Ingest Manager] Agent verifies packages before using them (elastic#18876)

dev-tools/packaging/packages.yml

@@ -60,6 +60,12 @@ shared:
       /etc/{{.BeatName}}/data/downloads/metricbeat-{{ beat_version }}{{if .Snapshot}}-SNAPSHOT{{end}}-{{.GOOS}}-{{.AgentArchName}}.tar.gz:
         source: '{{ elastic_beats_dir }}/x-pack/metricbeat/build/distributions/metricbeat-{{ beat_version }}{{if .Snapshot}}-SNAPSHOT{{end}}-{{.GOOS}}-{{.AgentArchName}}.tar.gz'
         mode: 0644
+      /etc/{{.BeatName}}/data/downloads/filebeat-{{ beat_version }}{{if .Snapshot}}-SNAPSHOT{{end}}-{{.GOOS}}-{{.AgentArchName}}.tar.gz.sha512:
+        source: '{{ elastic_beats_dir }}/x-pack/filebeat/build/distributions/filebeat-{{ beat_version }}{{if .Snapshot}}-SNAPSHOT{{end}}-{{.GOOS}}-{{.AgentArchName}}.tar.gz.sha512'
+        mode: 0644
+      /etc/{{.BeatName}}/data/downloads/metricbeat-{{ beat_version }}{{if .Snapshot}}-SNAPSHOT{{end}}-{{.GOOS}}-{{.AgentArchName}}.tar.gz.sha512:
+        source: '{{ elastic_beats_dir }}/x-pack/metricbeat/build/distributions/metricbeat-{{ beat_version }}{{if .Snapshot}}-SNAPSHOT{{end}}-{{.GOOS}}-{{.AgentArchName}}.tar.gz.sha512'
+        mode: 0644
 
 
   # MacOS pkg spec for community beats.
@@ -103,6 +109,12 @@ shared:
       /etc/{{.BeatName}}/data/downloads/metricbeat-{{ beat_version }}{{if .Snapshot}}-SNAPSHOT{{end}}-{{.GOOS}}-{{.AgentArchName}}.tar.gz:
         source: '{{ elastic_beats_dir }}/x-pack/metricbeat/build/distributions/metricbeat-{{ beat_version }}{{if .Snapshot}}-SNAPSHOT{{end}}-{{.GOOS}}-{{.AgentArchName}}.tar.gz'
         mode: 0644
+      /etc/{{.BeatName}}/data/downloads/filebeat-{{ beat_version }}{{if .Snapshot}}-SNAPSHOT{{end}}-{{.GOOS}}-{{.AgentArchName}}.tar.gz.sha512:
+        source: '{{ elastic_beats_dir }}/x-pack/filebeat/build/distributions/filebeat-{{ beat_version }}{{if .Snapshot}}-SNAPSHOT{{end}}-{{.GOOS}}-{{.AgentArchName}}.tar.gz.sha512'
+        mode: 0644
+      /etc/{{.BeatName}}/data/downloads/metricbeat-{{ beat_version }}{{if .Snapshot}}-SNAPSHOT{{end}}-{{.GOOS}}-{{.AgentArchName}}.tar.gz.sha512:
+        source: '{{ elastic_beats_dir }}/x-pack/metricbeat/build/distributions/metricbeat-{{ beat_version }}{{if .Snapshot}}-SNAPSHOT{{end}}-{{.GOOS}}-{{.AgentArchName}}.tar.gz.sha512'
+        mode: 0644
 
   - &agent_binary_files
     '{{.BeatName}}{{.BinaryExt}}':
@@ -137,6 +149,13 @@ shared:
       'data/downloads/metricbeat-{{ beat_version }}{{if .Snapshot}}-SNAPSHOT{{end}}-{{.GOOS}}-{{.AgentArchName}}.tar.gz':
         source: '{{ elastic_beats_dir }}/x-pack/metricbeat/build/distributions/metricbeat-{{ beat_version }}{{if .Snapshot}}-SNAPSHOT{{end}}-{{.GOOS}}-{{.AgentArchName}}.tar.gz'
         mode: 0644
+      <<: *agent_binary_files
+      'data/downloads/filebeat-{{ beat_version }}{{if .Snapshot}}-SNAPSHOT{{end}}-{{.GOOS}}-{{.AgentArchName}}.tar.gz.sha512':
+        source: '{{ elastic_beats_dir }}/x-pack/filebeat/build/distributions/filebeat-{{ beat_version }}{{if .Snapshot}}-SNAPSHOT{{end}}-{{.GOOS}}-{{.AgentArchName}}.tar.gz.sha512'
+        mode: 0644
+      'data/downloads/metricbeat-{{ beat_version }}{{if .Snapshot}}-SNAPSHOT{{end}}-{{.GOOS}}-{{.AgentArchName}}.tar.gz.sha512':
+        source: '{{ elastic_beats_dir }}/x-pack/metricbeat/build/distributions/metricbeat-{{ beat_version }}{{if .Snapshot}}-SNAPSHOT{{end}}-{{.GOOS}}-{{.AgentArchName}}.tar.gz.sha512'
+        mode: 0644
 
   # Binary package spec (zip for windows) for community beats.
   - &agent_windows_binary_spec
@@ -155,6 +174,12 @@ shared:
       'data/downloads/metricbeat-{{ beat_version }}{{if .Snapshot}}-SNAPSHOT{{end}}-{{.GOOS}}-{{.AgentArchName}}.zip':
         source: '{{ elastic_beats_dir }}/x-pack/metricbeat/build/distributions/metricbeat-{{ beat_version }}{{if .Snapshot}}-SNAPSHOT{{end}}-{{.GOOS}}-{{.AgentArchName}}.zip'
         mode: 0644
+      'data/downloads/filebeat-{{ beat_version }}{{if .Snapshot}}-SNAPSHOT{{end}}-{{.GOOS}}-{{.AgentArchName}}.zip.sha512':
+        source: '{{ elastic_beats_dir }}/x-pack/filebeat/build/distributions/filebeat-{{ beat_version }}{{if .Snapshot}}-SNAPSHOT{{end}}-{{.GOOS}}-{{.AgentArchName}}.zip.sha512'
+        mode: 0644
+      'data/downloads/metricbeat-{{ beat_version }}{{if .Snapshot}}-SNAPSHOT{{end}}-{{.GOOS}}-{{.AgentArchName}}.zip.sha512':
+        source: '{{ elastic_beats_dir }}/x-pack/metricbeat/build/distributions/metricbeat-{{ beat_version }}{{if .Snapshot}}-SNAPSHOT{{end}}-{{.GOOS}}-{{.AgentArchName}}.zip.sha512'
+        mode: 0644
 
   - &agent_docker_spec
     <<: *agent_binary_spec

x-pack/elastic-agent/CHANGELOG.asciidoc

@@ -68,3 +68,4 @@
 - Pick up version from libbeat {pull}18350[18350]
 - Use shorter hash for application differentiator {pull}18770[18770]
 - When not port are specified and the https is used fallback to 443 {pull}18844[18844]
+- Agent verifies packages before using them {pull}18876[18876]

x-pack/elastic-agent/pkg/agent/application/stream.go

@@ -78,6 +78,11 @@ func newOperator(ctx context.Context, log *logger.Logger, id routingKey, config
 	}
 
 	fetcher := downloader.NewDownloader(operatorConfig.DownloadConfig)
+	verifier, err := downloader.NewVerifier(operatorConfig.DownloadConfig)
+	if err != nil {
+		return nil, errors.New(err, "initiating verifier")
+	}
+
 	installer, err := install.NewInstaller(operatorConfig.DownloadConfig)
 	if err != nil {
 		return nil, errors.New(err, "initiating installer")
@@ -94,6 +99,7 @@ func newOperator(ctx context.Context, log *logger.Logger, id routingKey, config
 		id,
 		config,
 		fetcher,
+		verifier,
 		installer,
 		stateResolver,
 		r,

x-pack/elastic-agent/pkg/agent/operation/common_test.go

@@ -46,14 +46,15 @@ func getTestOperator(t *testing.T, installPath string) (*Operator, *operatorCfg.
 	l := getLogger()
 
 	fetcher := &DummyDownloader{}
+	verifier := &DummyVerifier{}
 	installer := &DummyInstaller{}
 
 	stateResolver, err := stateresolver.NewStateResolver(l)
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	operator, err := NewOperator(context.Background(), l, "p1", cfg, fetcher, installer, stateResolver, nil, noop.NewMonitor())
+	operator, err := NewOperator(context.Background(), l, "p1", cfg, fetcher, verifier, installer, stateResolver, nil, noop.NewMonitor())
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -81,18 +82,24 @@ type TestConfig struct {
 	TestFile string
 }
 
-type DummyDownloader struct {
-}
+type DummyDownloader struct{}
 
 func (*DummyDownloader) Download(_ context.Context, p, v string) (string, error) {
 	return "", nil
 }
 
 var _ download.Downloader = &DummyDownloader{}
 
-type DummyInstaller struct {
+type DummyVerifier struct{}
+
+func (*DummyVerifier) Verify(p, v string) (bool, error) {
+	return true, nil
 }
 
+var _ download.Verifier = &DummyVerifier{}
+
+type DummyInstaller struct{}
+
 func (*DummyInstaller) Install(p, v, _ string) error {
 	return nil
 }

x-pack/elastic-agent/pkg/agent/operation/monitoring_test.go

@@ -116,6 +116,7 @@ func getMonitorableTestOperator(t *testing.T, installPath string, m monitoring.M
 	l := getLogger()
 
 	fetcher := &DummyDownloader{}
+	verifier := &DummyVerifier{}
 	installer := &DummyInstaller{}
 
 	stateResolver, err := stateresolver.NewStateResolver(l)
@@ -124,7 +125,7 @@ func getMonitorableTestOperator(t *testing.T, installPath string, m monitoring.M
 	}
 	ctx := context.Background()
 
-	operator, err := NewOperator(ctx, l, "p1", cfg, fetcher, installer, stateResolver, nil, m)
+	operator, err := NewOperator(ctx, l, "p1", cfg, fetcher, verifier, installer, stateResolver, nil, m)
 	if err != nil {
 		t.Fatal(err)
 	}

x-pack/elastic-agent/pkg/agent/operation/operation_verify.go

@@ -6,18 +6,35 @@ package operation
 
 import (
 	"context"
+	"fmt"
+	"os"
 
 	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/errors"
+	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/agent/operation/config"
+	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/artifact"
+	"github.com/elastic/beats/v7/x-pack/elastic-agent/pkg/artifact/download"
 )
 
 // operationVerify verifies downloaded artifact for correct signature
 // skips if artifact is already installed
 type operationVerify struct {
 	eventProcessor callbackHooks
+	program        Descriptor
+	operatorConfig *config.Config
+	verifier       download.Verifier
 }
 
-func newOperationVerify(eventProcessor callbackHooks) *operationVerify {
-	return &operationVerify{eventProcessor: eventProcessor}
+func newOperationVerify(
+	program Descriptor,
+	operatorConfig *config.Config,
+	verifier download.Verifier,
+	eventProcessor callbackHooks) *operationVerify {
+	return &operationVerify{
+		program:        program,
+		operatorConfig: operatorConfig,
+		eventProcessor: eventProcessor,
+		verifier:       verifier,
+	}
 }
 
 // Name is human readable name identifying an operation
@@ -30,7 +47,18 @@ func (o *operationVerify) Name() string {
 // - Start does not need to run if process is running
 // - Fetch does not need to run if package is already present
 func (o *operationVerify) Check() (bool, error) {
-	return false, nil
+	downloadConfig := o.operatorConfig.DownloadConfig
+	fullPath, err := artifact.GetArtifactPath(o.program.BinaryName(), o.program.Version(), downloadConfig.OS(), downloadConfig.Arch(), downloadConfig.TargetDirectory)
+	if err != nil {
+		return false, err
+	}
+
+	if _, err := os.Stat(fullPath); os.IsNotExist(err) {
+		return false, errors.New(errors.TypeApplication,
+			fmt.Sprintf("%s.%s package does not exist in %s. Skipping operation %s", o.program.BinaryName(), o.program.Version(), fullPath, o.Name()))
+	}
+
+	return true, err
 }
 
 // Run runs the operation
@@ -45,5 +73,18 @@ func (o *operationVerify) Run(ctx context.Context, application Application) (err
 		}
 	}()
 
+	isVerified, err := o.verifier.Verify(o.program.BinaryName(), o.program.Version())
+	if err != nil {
+		return errors.New(err,
+			fmt.Sprintf("operation '%s' failed to verify %s.%s", o.Name(), o.program.BinaryName(), o.program.Version()),
+			errors.TypeSecurity)
+	}
+
+	if !isVerified {
+		return errors.New(err,
+			fmt.Sprintf("operation '%s' marked '%s.%s' corrupted", o.Name(), o.program.BinaryName(), o.program.Version()),
+			errors.TypeSecurity)
+	}
+
 	return nil
 }

x-pack/elastic-agent/pkg/agent/operation/operator.go

@@ -51,6 +51,7 @@ type Operator struct {
 	appsLock sync.Mutex
 
 	downloader download.Downloader
+	verifier   download.Verifier
 	installer  install.Installer
 }
 
@@ -63,6 +64,7 @@ func NewOperator(
 	pipelineID string,
 	config *config.Config,
 	fetcher download.Downloader,
+	verifier download.Verifier,
 	installer install.Installer,
 	stateResolver *stateresolver.StateResolver,
 	eventProcessor callbackHooks,
@@ -87,6 +89,7 @@ func NewOperator(
 		pipelineID:     pipelineID,
 		logger:         logger,
 		downloader:     fetcher,
+		verifier:       verifier,
 		installer:      installer,
 		stateResolver:  stateResolver,
 		apps:           make(map[string]Application),
@@ -155,7 +158,7 @@ func (o *Operator) HandleConfig(cfg configrequest.Request) error {
 func (o *Operator) start(p Descriptor, cfg map[string]interface{}) (err error) {
 	flow := []operation{
 		newOperationFetch(o.logger, p, o.config, o.downloader, o.eventProcessor),
-		newOperationVerify(o.eventProcessor),
+		newOperationVerify(p, o.config, o.verifier, o.eventProcessor),
 		newOperationInstall(o.logger, p, o.config, o.installer, o.eventProcessor),
 		newOperationStart(o.logger, p, o.config, cfg, o.eventProcessor),
 		newOperationConfig(o.logger, o.config, cfg, o.eventProcessor),

x-pack/elastic-agent/pkg/artifact/download/fs/downloader.go

@@ -50,8 +50,10 @@ func (e *Downloader) Download(_ context.Context, programName, version string) (s
 	path, err := e.download(e.config.OS(), programName, version)
 	if err != nil {
 		os.Remove(path)
+		return "", err
 	}
 
+	_, err = e.downloadHash(e.config.OS(), programName, version)
 	return path, err
 }
 
@@ -66,6 +68,27 @@ func (e *Downloader) download(operatingSystem, programName, version string) (str
 		return "", errors.New(err, "generating package path failed")
 	}
 
+	return e.downloadFile(filename, fullPath)
+}
+
+func (e *Downloader) downloadHash(operatingSystem, programName, version string) (string, error) {
+	filename, err := artifact.GetArtifactName(programName, version, operatingSystem, e.config.Arch())
+	if err != nil {
+		return "", errors.New(err, "generating package name failed")
+	}
+
+	fullPath, err := artifact.GetArtifactPath(programName, version, operatingSystem, e.config.Arch(), e.config.TargetDirectory)
+	if err != nil {
+		return "", errors.New(err, "generating package path failed")
+	}
+
+	filename = filename + ".sha512"
+	fullPath = fullPath + ".sha512"
+
+	return e.downloadFile(filename, fullPath)
+}
+
+func (e *Downloader) downloadFile(filename, fullPath string) (string, error) {
 	sourcePath := filepath.Join(e.dropPath, filename)
 	sourceFile, err := os.Open(sourcePath)
 	if err != nil {

x-pack/elastic-agent/pkg/artifact/download/fs/verifier.go

@@ -5,11 +5,16 @@
 package fs
 
 import (
+	"bufio"
 	"bytes"
+	"crypto/sha512"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"os"
 	"path/filepath"
+	"strings"
+	"sync"
 
 	"golang.org/x/crypto/openpgp"
 
@@ -35,10 +40,6 @@ func NewVerifier(config *artifact.Config) (*Verifier, error) {
 		config: config,
 	}
 
-	if err := v.loadPGP(config.PgpFile); err != nil {
-		return nil, errors.New(err, "loading PGP")
-	}
-
 	return v, nil
 }
 
@@ -52,6 +53,63 @@ func (v *Verifier) Verify(programName, version string) (bool, error) {
 
 	fullPath := filepath.Join(v.config.TargetDirectory, filename)
 
+	return v.verifyHash(filename, fullPath)
+}
+
+func (v *Verifier) verifyHash(filename, fullPath string) (bool, error) {
+	hashFilePath := fullPath + ".sha512"
+	hashFileHandler, err := os.Open(hashFilePath)
+	if err != nil {
+		return false, err
+	}
+	defer hashFileHandler.Close()
+
+	// get hash
+	// content of a file is in following format
+	// hash  filename
+	var expectedHash string
+	scanner := bufio.NewScanner(hashFileHandler)
+	for scanner.Scan() {
+		line := scanner.Text()
+		if !strings.HasSuffix(line, filename) {
+			continue
+		}
+
+		expectedHash = strings.TrimSpace(strings.TrimSuffix(line, filename))
+	}
+
+	if expectedHash == "" {
+		return false, fmt.Errorf("hash for '%s' not found", filename)
+	}
+
+	// compute file hash
+	fileReader, err := os.OpenFile(fullPath, os.O_RDONLY, 0666)
+	if err != nil {
+		return false, errors.New(err, errors.TypeFilesystem, errors.M(errors.MetaKeyPath, fullPath))
+	}
+	defer fileReader.Close()
+
+	hash := sha512.New()
+	if _, err := io.Copy(hash, fileReader); err != nil {
+		return false, err
+	}
+	computedHash := fmt.Sprintf("%x", hash.Sum(nil))
+
+	return expectedHash == computedHash, nil
+}
+
+func (v *Verifier) verifyAsc(filename, fullPath string) (bool, error) {
+	var err error
+	var pgpBytesLoader sync.Once
+
+	pgpBytesLoader.Do(func() {
+		err = v.loadPGP(v.config.PgpFile)
+	})
+
+	if err != nil {
+		return false, errors.New(err, "loading PGP")
+	}
+
 	ascBytes, err := v.getPublicAsc(filename)
 	if err != nil {
 		return false, err