Merge branch 'main' into uninstall-kill-watcher

changelog/fragments/1695035111-Resilient-handling-of-air-gapped-PGP-checks.yaml

@@ -0,0 +1,31 @@
+# Kind can be one of:
+# - breaking-change: a change to previously-documented behavior
+# - deprecation: functionality that is being removed in a later release
+# - bug-fix: fixes a problem in a previous version
+# - enhancement: extends functionality but does not break or fix existing behavior
+# - feature: new functionality
+# - known-issue: problems that we are aware of in a given version
+# - security: impacts on the security of a product or a user’s deployment.
+# - upgrade: important information for someone upgrading from a prior version
+# - other: does not fit into any of the other categories
+kind: bug-fix
+
+# Change summary; a 80ish characters long description of the change.
+summary: Resilient handling of air gapped PGP checks
+
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+description: Elastic Agent should not fail when remote PGP is specified (or official Elastic fallback PGP used) and remote is not available
+
+# Affected component; a word indicating the component this changeset affects.
+component: elastic-agent
+
+# PR number; optional; the PR number that added the changeset.
+# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
+# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
+# Please provide it if you are adding a fragment for a different PR.
+pr: 3427
+
+# Issue number; optional; the GitHub issue related to this changeset (either closes or is part of).
+# If not present is automatically filled by the tooling with the issue linked to the PR number.
+issue: 3368

internal/pkg/agent/application/upgrade/artifact/download/fs/verifier.go

@@ -124,10 +124,11 @@ func (v *Verifier) verifyAsc(fullPath string, skipDefaultPgp bool, pgpSources ..
 		if len(check) == 0 {
 			continue
 		}
-		raw, err := download.PgpBytesFromSource(check, v.client)
+		raw, err := download.PgpBytesFromSource(v.log, check, v.client)
 		if err != nil {
 			return err
 		}
+
 		if len(raw) == 0 {
 			continue
 		}

internal/pkg/agent/application/upgrade/artifact/download/fs/verifier_test.go

@@ -173,50 +173,60 @@ func prepareFetchVerifyTests(dropPath, targetDir, targetFilePath, hashTargetFile
 }
 
 func TestVerify(t *testing.T) {
-	log, _ := logger.New("", false)
-	targetDir, err := ioutil.TempDir(os.TempDir(), "")
-	if err != nil {
-		t.Fatal(err)
+	tt := []struct {
+		Name             string
+		RemotePGPUris    []string
+		UnreachableCount int
+	}{
+		{"default", nil, 0},
+		{"unreachable local path", []string{download.PgpSourceURIPrefix + "https://127.0.0.1:2874/path/does/not/exist"}, 1},
 	}
 
-	timeout := 30 * time.Second
-
-	config := &artifact.Config{
-		TargetDirectory: targetDir,
-		DropPath:        filepath.Join(targetDir, "drop"),
-		OperatingSystem: "linux",
-		Architecture:    "32",
-		HTTPTransportSettings: httpcommon.HTTPTransportSettings{
-			Timeout: timeout,
-		},
+	for _, tc := range tt {
+		t.Run(tc.Name, func(t *testing.T) {
+			log, obs := logger.NewTesting("TestVerify")
+			targetDir, err := ioutil.TempDir(os.TempDir(), "")
+			require.NoError(t, err)
+
+			timeout := 30 * time.Second
+
+			config := &artifact.Config{
+				TargetDirectory: targetDir,
+				DropPath:        filepath.Join(targetDir, "drop"),
+				OperatingSystem: "linux",
+				Architecture:    "32",
+				HTTPTransportSettings: httpcommon.HTTPTransportSettings{
+					Timeout: timeout,
+				},
+			}
+
+			err = prepareTestCase(beatSpec, version, config)
+			require.NoError(t, err)
+
+			testClient := NewDownloader(config)
+			artifact, err := testClient.Download(context.Background(), beatSpec, version)
+			require.NoError(t, err)
+
+			t.Cleanup(func() {
+				os.Remove(artifact)
+				os.Remove(artifact + ".sha512")
+				os.RemoveAll(config.DropPath)
+			})
+
+			_, err = os.Stat(artifact)
+			require.NoError(t, err)
+
+			testVerifier, err := NewVerifier(log, config, true, nil)
+			require.NoError(t, err)
+
+			err = testVerifier.Verify(beatSpec, version, false, tc.RemotePGPUris...)
+			require.NoError(t, err)
+
+			// log message informing remote PGP was skipped
+			logs := obs.FilterMessageSnippet("Skipped remote PGP located at")
+			require.Equal(t, tc.UnreachableCount, logs.Len())
+		})
 	}
-
-	if err := prepareTestCase(beatSpec, version, config); err != nil {
-		t.Fatal(err)
-	}
-
-	testClient := NewDownloader(config)
-	artifact, err := testClient.Download(context.Background(), beatSpec, version)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	_, err = os.Stat(artifact)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	testVerifier, err := NewVerifier(log, config, true, nil)
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	err = testVerifier.Verify(beatSpec, version, false)
-	require.NoError(t, err)
-
-	os.Remove(artifact)
-	os.Remove(artifact + ".sha512")
-	os.RemoveAll(config.DropPath)
 }
 
 func prepareTestCase(a artifact.Artifact, version string, cfg *artifact.Config) error {

internal/pkg/agent/application/upgrade/artifact/download/http/verifier.go

@@ -125,10 +125,11 @@ func (v *Verifier) verifyAsc(a artifact.Artifact, version string, skipDefaultPgp
 		if len(check) == 0 {
 			continue
 		}
-		raw, err := download.PgpBytesFromSource(check, v.client)
+		raw, err := download.PgpBytesFromSource(v.log, check, v.client)
 		if err != nil {
 			return err
 		}
+
 		if len(raw) == 0 {
 			continue
 		}

internal/pkg/agent/application/upgrade/artifact/download/verifier.go

@@ -20,10 +20,11 @@ import (
 	"strings"
 	"time"
 
-	"github.com/elastic/elastic-agent/internal/pkg/agent/application/upgrade/artifact"
+	"github.com/hashicorp/go-multierror"
 
 	"golang.org/x/crypto/openpgp" //nolint:staticcheck // crypto/openpgp is only receiving security updates.
 
+	"github.com/elastic/elastic-agent/internal/pkg/agent/application/upgrade/artifact"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/errors"
 )
 
@@ -32,6 +33,17 @@ const (
 	PgpSourceURIPrefix = "pgp_uri:"
 )
 
+var (
+	ErrRemotePGPDownloadFailed = errors.New("Remote PGP download failed")
+	ErrInvalidLocation         = errors.New("Remote PGP location is invalid")
+)
+
+// warnLogger is a logger that only needs to implement Warnf, as that is the only functions
+// that the downloadProgressReporter uses.
+type warnLogger interface {
+	Warnf(format string, args ...interface{})
+}
+
 // ChecksumMismatchError indicates the expected checksum for a file does not
 // match the computed checksum.
 type ChecksumMismatchError struct {
@@ -168,13 +180,17 @@ func VerifyGPGSignature(file string, asciiArmorSignature, publicKey []byte) erro
 	return nil
 }
 
-func PgpBytesFromSource(source string, client http.Client) ([]byte, error) {
+func PgpBytesFromSource(log warnLogger, source string, client http.Client) ([]byte, error) {
 	if strings.HasPrefix(source, PgpSourceRawPrefix) {
 		return []byte(strings.TrimPrefix(source, PgpSourceRawPrefix)), nil
 	}
 
 	if strings.HasPrefix(source, PgpSourceURIPrefix) {
-		return fetchPgpFromURI(strings.TrimPrefix(source, PgpSourceURIPrefix), client)
+		pgpBytes, err := fetchPgpFromURI(strings.TrimPrefix(source, PgpSourceURIPrefix), client)
+		if errors.Is(err, ErrRemotePGPDownloadFailed) || errors.Is(err, ErrInvalidLocation) {
+			log.Warnf("Skipped remote PGP located at %q because it's unavailable: %v", strings.TrimPrefix(source, PgpSourceURIPrefix), err)
+		}
+		return pgpBytes, nil
 	}
 
 	return nil, errors.New("unknown pgp source")
@@ -187,7 +203,7 @@ func CheckValidDownloadUri(rawURI string) error {
 	}
 
 	if !strings.EqualFold(uri.Scheme, "https") {
-		return fmt.Errorf("failed to check URI %q: HTTPS is required", rawURI)
+		return multierror.Append(fmt.Errorf("failed to check URI %q: HTTPS is required", rawURI), ErrInvalidLocation)
 	}
 
 	return nil
@@ -207,7 +223,7 @@ func fetchPgpFromURI(uri string, client http.Client) ([]byte, error) {
 	}
 	resp, err := http.DefaultClient.Do(req)
 	if err != nil {
-		return nil, err
+		return nil, multierror.Append(err, ErrRemotePGPDownloadFailed)
 	}
 	defer resp.Body.Close()
 

testing/integration/upgrade_test.go

@@ -162,10 +162,6 @@ func testUpgradeFleetManagedElasticAgent(t *testing.T, ctx context.Context, info
 	t.Log(`Waiting for enrolled Agent status to be "online"...`)
 	require.Eventually(t, tools.WaitForAgentStatus(t, kibClient, policy.ID, "online"), 10*time.Minute, 15*time.Second, "Agent status is not online")
 
-	// Upgrade Watcher check disabled until
-	// https://github.com/elastic/elastic-agent/issues/2977 is resolved.
-	// checkUpgradeWatcherRan(t, s.agentFixture)
-
 	// We remove the `-SNAPSHOT` suffix because, post-upgrade, the version reported
 	// by the Agent will not contain this suffix, even if a `-SNAPSHOT`-suffixed
 	// version was used as the target version for the upgrade.
@@ -212,7 +208,7 @@ func TestStandaloneUpgrade(t *testing.T) {
 			parsedUpgradeVersion, err := version.ParseVersion(define.Version())
 			require.NoErrorf(t, err, "define.Version() %q cannot be parsed as agent version", define.Version())
 			skipVerify := version_8_7_0.Less(*parsedVersion)
-			testStandaloneUpgrade(ctx, t, agentFixture, parsedVersion, parsedUpgradeVersion, "", skipVerify, true, false, "")
+			testStandaloneUpgrade(ctx, t, agentFixture, parsedVersion, parsedUpgradeVersion, "", skipVerify, true, false, CustomPGP{})
 		})
 	}
 }
@@ -251,13 +247,66 @@ func TestStandaloneUpgradeWithGPGFallback(t *testing.T) {
 
 	_, defaultPGP := release.PGP()
 	firstSeven := string(defaultPGP[:7])
-	customPGP := strings.Replace(
+	newPgp := strings.Replace(
+		string(defaultPGP),
+		firstSeven,
+		"abcDEFg",
+		1,
+	)
+
+	customPGP := CustomPGP{
+		PGP: newPgp,
+	}
+
+	testStandaloneUpgrade(ctx, t, agentFixture, fromVersion, toVersion, "", false, false, true, customPGP)
+}
+
+func TestStandaloneUpgradeWithGPGFallbackOneRemoteFailing(t *testing.T) {
+	define.Require(t, define.Requirements{
+		Local: false, // requires Agent installation
+		Sudo:  true,  // requires Agent installation
+	})
+
+	minVersion := version_8_10_0_SNAPSHOT
+	fromVersion, err := version.ParseVersion(define.Version())
+	require.NoError(t, err)
+
+	if fromVersion.Less(*minVersion) {
+		t.Skipf("Version %s is lower than min version %s", define.Version(), minVersion)
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	// previous
+	toVersion, err := fromVersion.GetPreviousMinor()
+	require.NoError(t, err, "failed to get previous minor")
+	agentFixture, err := define.NewFixture(
+		t,
+		define.Version(),
+	)
+	require.NoError(t, err, "error creating fixture")
+
+	err = agentFixture.Prepare(ctx)
+	require.NoError(t, err, "error preparing agent fixture")
+
+	err = agentFixture.Configure(ctx, []byte(fastWatcherCfg))
+	require.NoError(t, err, "error configuring agent fixture")
+
+	_, defaultPGP := release.PGP()
+	firstSeven := string(defaultPGP[:7])
+	newPgp := strings.Replace(
 		string(defaultPGP),
 		firstSeven,
 		"abcDEFg",
 		1,
 	)
 
+	customPGP := CustomPGP{
+		PGP:    newPgp,
+		PGPUri: "https://127.0.0.1:3456/non/existing/path",
+	}
+
 	testStandaloneUpgrade(ctx, t, agentFixture, fromVersion, toVersion, "", false, false, true, customPGP)
 }
 
@@ -334,8 +383,7 @@ func TestStandaloneDowngradeToPreviousSnapshotBuild(t *testing.T) {
 	t.Logf("Targeting upgrade to version %+v", upgradeInputVersion)
 	parsedFromVersion, err := version.ParseVersion(define.Version())
 	require.NoErrorf(t, err, "define.Version() %q cannot be parsed as agent version", define.Version())
-	testStandaloneUpgrade(ctx, t, agentFixture, parsedFromVersion, upgradeInputVersion, expectedAgentHashAfterUpgrade, false, true, false, "")
-
+	testStandaloneUpgrade(ctx, t, agentFixture, parsedFromVersion, upgradeInputVersion, expectedAgentHashAfterUpgrade, false, true, false, CustomPGP{})
 }
 
 func getUpgradableVersions(ctx context.Context, t *testing.T, upgradeToVersion string) (upgradableVersions []*version.ParsedSemVer) {
@@ -410,7 +458,7 @@ func testStandaloneUpgrade(
 	allowLocalPackage bool,
 	skipVerify bool,
 	skipDefaultPgp bool,
-	customPgp string,
+	customPgp CustomPGP,
 ) {
 
 	var nonInteractiveFlag bool
@@ -462,8 +510,16 @@ func testStandaloneUpgrade(
 		upgradeCmdArgs = append(upgradeCmdArgs, "--skip-default-pgp")
 	}
 
-	if len(customPgp) > 0 {
-		upgradeCmdArgs = append(upgradeCmdArgs, "--pgp", customPgp)
+	if len(customPgp.PGP) > 0 {
+		upgradeCmdArgs = append(upgradeCmdArgs, "--pgp", customPgp.PGP)
+	}
+
+	if len(customPgp.PGPUri) > 0 {
+		upgradeCmdArgs = append(upgradeCmdArgs, "--pgp-uri", customPgp.PGPUri)
+	}
+
+	if len(customPgp.PGPPath) > 0 {
+		upgradeCmdArgs = append(upgradeCmdArgs, "--pgp-path", customPgp.PGPPath)
 	}
 
 	upgradeTriggerOutput, err := f.Exec(ctx, upgradeCmdArgs)
@@ -811,6 +867,9 @@ func TestUpgradeBrokenPackageVersion(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
+	err = f.Configure(ctx, []byte(fastWatcherCfg))
+	require.NoError(t, err, "error configuring agent fixture")
+
 	output, err := tools.InstallStandaloneAgent(f)
 	t.Logf("Agent installation output: %q", string(output))
 	require.NoError(t, err)
@@ -995,3 +1054,9 @@ inputs:
 		return checkAgentHealthAndVersion(t, ctx, agentFixture, upgradeFromVersion.CoreVersion(), upgradeFromVersion.IsSnapshot(), "")
 	}, 2*time.Minute, 10*time.Second, "Rolled back Agent never became healthy")
 }
+
+type CustomPGP struct {
+	PGP     string
+	PGPUri  string
+	PGPPath string
+}