Allow agent to override fleet settings

internal/pkg/agent/application/coordinator/coordinator.go

@@ -8,6 +8,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"os"
 	"reflect"
 	"strings"
 	"sync"
@@ -614,7 +615,7 @@ func (c *Coordinator) Migrate(
 	backoffFactory func(done <-chan struct{}) backoff.Backoff,
 	notifyFn func(context.Context, *fleetapi.ActionMigrate) error,
 ) error {
-	if c.specs.Platform().OS == component.Container {
+	if c.isContainerizedEnvironment() {
 		return ErrContainerNotSupported
 	}
 	if !c.isManaged {
@@ -1609,6 +1610,14 @@ func (c *Coordinator) processConfigAgent(ctx context.Context, cfg *config.Config
 		c.logger.Errorf("failed to add secret markers: %v", err)
 	}
 
+	// override retrieved config from Fleet with persisted config from AgentConfig file
+
+	if c.caps != nil {
+		if err := applyPersistedConfig(cfg, paths.ConfigFile(), c.isContainerizedEnvironment, c.caps.AllowFleetOverride); err != nil {
+			return fmt.Errorf("could not apply persisted configuration: %w", err)
+		}
+	}
+
 	// perform and verify ast translation
 	m, err := cfg.ToMapStr()
 	if err != nil {
@@ -1696,6 +1705,34 @@ func (c *Coordinator) generateAST(cfg *config.Config, m map[string]interface{})
 	return nil
 }
 
+func applyPersistedConfig(cfg *config.Config, configFile string, checkFns ...func() bool) error {
+	for _, checkFn := range checkFns {
+		if !checkFn() {
+			// Feature is disabled, nothing to do
+			return nil
+		}
+	}
+
+	f, err := os.OpenFile(configFile, os.O_RDONLY, 0)
+	if err != nil && os.IsNotExist(err) {
+		return nil
+	} else if err != nil {
+		return fmt.Errorf("opening config file: %w", err)
+	}
+	defer f.Close()
+
+	persisted, err := config.NewConfigFrom(f)
+	if err != nil {
+		return fmt.Errorf("parsing persisted config: %w", err)
+	}
+
+	err = cfg.Merge(persisted)
+	if err != nil {
+		return fmt.Errorf("merging persisted config: %w", err)
+	}
+	return nil
+}
+
 // observeASTVars identifies the variables that are referenced in the computed AST and passed to
 // the varsMgr so it knows what providers are being referenced. If a providers is not being
 // referenced then the provider does not need to be running.
@@ -2362,3 +2399,7 @@ func computeEnrollOptions(ctx context.Context, cfgPath string, cfgFleetPath stri
 	options = enroll.FromFleetConfig(cfg.Fleet)
 	return options, nil
 }
+
+func (c *Coordinator) isContainerizedEnvironment() bool {
+	return c.specs.Platform().OS == component.Container
+}

internal/pkg/agent/application/coordinator/coordinator_test.go

@@ -1011,6 +1011,47 @@ func TestCoordinator_UpgradeDetails(t *testing.T) {
 	require.Equal(t, expectedErr.Error(), coord.state.UpgradeDetails.Metadata.ErrorMsg)
 }
 
+func Test_ApplyPersistedConfig(t *testing.T) {
+	cfgFile := filepath.Join(".", "testdata", "overrides.yml")
+
+	testCases := []struct {
+		name               string
+		featureEnable      bool
+		expectedLogs       bool
+		expectedOutputType string
+	}{
+		{name: "enabled", featureEnable: true, expectedLogs: false, expectedOutputType: "kafka"},
+		{name: "disabled", featureEnable: false, expectedLogs: true, expectedOutputType: "elasticsearch"},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			cfg, err := config.LoadFile(filepath.Join(".", "testdata", "config.yaml"))
+			require.NoError(t, err)
+
+			err = applyPersistedConfig(cfg, cfgFile, func() bool { return tc.featureEnable })
+			require.NoError(t, err)
+
+			c := &configuration.Configuration{}
+			require.NoError(t, cfg.Agent.Unpack(&c))
+
+			require.Equal(t, tc.expectedLogs, c.Settings.MonitoringConfig.MonitorLogs)
+			require.True(t, c.Settings.MonitoringConfig.MonitorMetrics)
+			require.True(t, c.Settings.MonitoringConfig.Enabled)
+
+			// make sure output is not kafka
+			oc, err := cfg.Agent.Child("outputs", -1)
+			require.NoError(t, err)
+
+			do, err := oc.Child("default", -1)
+			require.NoError(t, err)
+			outputType, err := do.String("type", -1)
+			require.NoError(t, err)
+			assert.Equal(t, tc.expectedOutputType, outputType, "output type should be %s, got %s", tc.expectedOutputType, outputType)
+		})
+	}
+}
+
 func BenchmarkCoordinator_generateComponentModel(b *testing.B) {
 	// load variables
 	varsMaps := []map[string]any{}

internal/pkg/agent/application/coordinator/testdata/overrides.yml

@@ -0,0 +1,15 @@
+agent:
+  monitoring:
+    enabled: true
+    use_output: default
+    logs: false
+    metrics: true
+    http:
+      port: 6774
+outputs:
+  default:
+    type: kafka
+    hosts: localhost
+    api_key: ""
+    username: ""
+    password: ""

internal/pkg/agent/cmd/run.go

@@ -286,6 +286,11 @@ func runElasticAgent(
 			ownership.GID = gid
 		}
 
+		// check capabilities permissions before fixing them
+		if err := checkCapabilitiesPerms(paths.AgentCapabilitiesPath(), userName, ownership.UID); err != nil {
+			return fmt.Errorf("invalid capabilities file permissions: %w", err)
+		}
+
 		topPath := paths.Top()
 		err = perms.FixPermissions(topPath, perms.WithOwnership(ownership))
 		if err != nil {

internal/pkg/agent/cmd/run_darwin.go

@@ -8,6 +8,7 @@ package cmd
 
 import (
 	"fmt"
+	"os"
 	"syscall"
 
 	"github.com/elastic/elastic-agent/pkg/core/logger"
@@ -39,3 +40,18 @@ func dropRootPrivileges(l *logger.Logger, ownership utils.FileOwner) error {
 
 	return nil
 }
+
+func checkCapabilitiesPerms(agentCapabilitiesPath string, userName string, uid int) error {
+	var capabilitiesUID int
+	if userName != "" {
+		capabilitiesUID = uid
+	} else {
+		capabilitiesUID = os.Getuid()
+	}
+	if err := utils.HasStrictExecPermsAndOwnership(agentCapabilitiesPath, capabilitiesUID); err != nil && !os.IsNotExist(err) {
+		// capabilities are corrupted, we should not proceed
+		return fmt.Errorf("invalid capabilities file permissions: %w", err)
+	}
+
+	return nil
+}

internal/pkg/agent/cmd/run_linux.go

@@ -8,6 +8,7 @@ package cmd
 
 import (
 	"fmt"
+	"os"
 	"syscall"
 
 	"github.com/elastic/elastic-agent/pkg/core/logger"
@@ -41,3 +42,18 @@ func dropRootPrivileges(l *logger.Logger, ownership utils.FileOwner) error {
 	return nil
 
 }
+
+func checkCapabilitiesPerms(agentCapabilitiesPath string, userName string, uid int) error {
+	var capabilitiesUID int
+	if userName != "" {
+		capabilitiesUID = uid
+	} else {
+		capabilitiesUID = os.Getuid()
+	}
+	if err := utils.HasStrictExecPermsAndOwnership(agentCapabilitiesPath, capabilitiesUID); err != nil && !os.IsNotExist(err) {
+		// capabilities are corrupted, we should not proceed
+		return fmt.Errorf("invalid capabilities file permissions: %w", err)
+	}
+
+	return nil
+}

internal/pkg/agent/cmd/run_windows.go

@@ -26,3 +26,8 @@ func logExternal(msg string) {
 }
 
 func dropRootPrivileges(_ *logger.Logger, _ utils.FileOwner) error { return nil }
+
+func checkCapabilitiesPerms(_ string, _ string, _ string) error {
+	// not implemented on Windows
+	return nil
+}

internal/pkg/capabilities/capabilities.go

@@ -19,13 +19,15 @@ type Capabilities interface {
 	AllowUpgrade(version string, sourceURI string) bool
 	AllowInput(name string) bool
 	AllowOutput(name string) bool
+	AllowFleetOverride() bool
 }
 
 type capabilitiesManager struct {
-	log          *logger.Logger
-	inputChecks  []*stringMatcher
-	outputChecks []*stringMatcher
-	upgradeCaps  []*upgradeCapability
+	log               *logger.Logger
+	inputChecks       []*stringMatcher
+	outputChecks      []*stringMatcher
+	upgradeCaps       []*upgradeCapability
+	fleetOverrideCaps []*fleetOverrideCapability
 }
 
 func (cm *capabilitiesManager) AllowInput(inputType string) bool {
@@ -40,6 +42,10 @@ func (cm *capabilitiesManager) AllowUpgrade(version string, uri string) bool {
 	return allowUpgrade(cm.log, version, uri, cm.upgradeCaps)
 }
 
+func (cm *capabilitiesManager) AllowFleetOverride() bool {
+	return allowFleetOverride(cm.log, cm.fleetOverrideCaps)
+}
+
 func LoadFile(capsFile string, log *logger.Logger) (Capabilities, error) {
 	// load capabilities from file
 	fd, err := os.Open(capsFile)
@@ -68,8 +74,10 @@ func Load(capsReader io.Reader, log *logger.Logger) (Capabilities, error) {
 	caps := spec.Capabilities
 
 	return &capabilitiesManager{
-		inputChecks:  caps.inputChecks,
-		outputChecks: caps.outputChecks,
-		upgradeCaps:  caps.upgradeChecks,
+		log:               log,
+		inputChecks:       caps.inputChecks,
+		outputChecks:      caps.outputChecks,
+		upgradeCaps:       caps.upgradeChecks,
+		fleetOverrideCaps: caps.fleetOverrideChecks,
 	}, nil
 }

internal/pkg/capabilities/fleet_override.go

@@ -0,0 +1,42 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License 2.0;
+// you may not use this file except in compliance with the Elastic License 2.0.
+
+package capabilities
+
+import "github.com/elastic/elastic-agent/pkg/core/logger"
+
+type fleetOverrideCapability struct {
+	// Whether a successful condition check lets an upgrade proceed or blocks it
+	rule allowOrDeny
+}
+
+func newFleetOverrideCapability(rule allowOrDeny) *fleetOverrideCapability {
+	return &fleetOverrideCapability{
+		rule: rule,
+	}
+}
+
+func allowFleetOverride(
+	log *logger.Logger,
+	fleetOverrideCaps []*fleetOverrideCapability,
+) bool {
+	// first match wins
+	for _, cap := range fleetOverrideCaps {
+		if cap == nil {
+			// being defensive here, should not happen
+			continue
+		}
+
+		switch cap.rule {
+		case ruleTypeAllow:
+			log.Debugf("Fleet override allowed by capability")
+			return true
+		case ruleTypeDeny:
+			log.Debugf("Fleet override denied by capability")
+			return false
+		}
+	}
+	// No matching capability found, disable by default
+	return false
+}

internal/pkg/capabilities/spec.go

@@ -13,9 +13,10 @@ import (
 // capabilitiesList deserializes a YAML list of capabilities into organized
 // arrays based on their type, for easy use by capabilitiesManager.
 type capabilitiesList struct {
-	inputChecks   []*stringMatcher
-	outputChecks  []*stringMatcher
-	upgradeChecks []*upgradeCapability
+	inputChecks         []*stringMatcher
+	outputChecks        []*stringMatcher
+	upgradeChecks       []*upgradeCapability
+	fleetOverrideChecks []*fleetOverrideCapability
 }
 
 // a type for capability values that must equal "allow" or "deny", enforced
@@ -81,6 +82,15 @@ func (r *capabilitiesList) UnmarshalYAML(unmarshal func(interface{}) error) erro
 				return err
 			}
 			r.upgradeChecks = append(r.upgradeChecks, cap)
+		} else if _, found = mm["fleet_override"]; found {
+			spec := struct {
+				Type allowOrDeny `yaml:"rule"`
+			}{}
+			if err := yaml.Unmarshal(partialYaml, &spec); err != nil {
+				return err
+			}
+			cap := newFleetOverrideCapability(spec.Type)
+			r.fleetOverrideChecks = append(r.fleetOverrideChecks, cap)
 		} else {
 			return fmt.Errorf("unexpected capability type for definition number '%d'", i)
 		}

pkg/component/runtime/command.go

@@ -363,13 +363,12 @@ func (c *commandRuntime) start(comm Communicator) error {
 	}
 	env = append(env, fmt.Sprintf("%s=%s", envAgentComponentID, c.current.ID))
 	env = append(env, fmt.Sprintf("%s=%s", envAgentComponentType, c.getSpecType()))
-	uid := os.Geteuid()
 	workDir := c.current.WorkDirPath(paths.Run())
 	path, err := filepath.Abs(c.getSpecBinaryPath())
 	if err != nil {
 		return fmt.Errorf("failed to determine absolute path: %w", err)
 	}
-	err = utils.HasStrictExecPerms(path, uid)
+	err = utils.HasStrictExecPerms(path)
 	if err != nil {
 		return fmt.Errorf("execution of component prevented: %w", err)
 	}