Handle PRIVILEGE_LEVEL_CHANGE action (#10231)

Author: michalpristas

changelog/fragments/1759318728-support-for-privilege_level_change.yaml

@@ -0,0 +1,45 @@
+# REQUIRED
+# 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: feature
+
+# REQUIRED for all kinds
+# Change summary; a 80ish characters long description of the change.
+summary: Add support for downgrading a running Agent's privileges from Fleet
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+# description: 
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# impact:
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# action:
+
+# REQUIRED for all kinds
+# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+component: elastic-agent
+
+# AUTOMATED
+# OPTIONAL to manually add other PR URLs
+# PR URL: A link the PR 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: https://github.com/owner/repo/1234
+
+# AUTOMATED
+# OPTIONAL to manually add other issue URLs
+# Issue URL; 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: https://github.com/owner/repo/1234

internal/pkg/agent/application/actions/handlers/handle_action_privilege_level_change.go

@@ -0,0 +1,144 @@
+// 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 handlers
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/elastic/elastic-agent/internal/pkg/agent/application/coordinator"
+	"github.com/elastic/elastic-agent/internal/pkg/agent/application/paths"
+	"github.com/elastic/elastic-agent/internal/pkg/agent/application/reexec"
+	"github.com/elastic/elastic-agent/internal/pkg/agent/install"
+	"github.com/elastic/elastic-agent/internal/pkg/agent/install/componentvalidation"
+	"github.com/elastic/elastic-agent/internal/pkg/fleetapi"
+	"github.com/elastic/elastic-agent/internal/pkg/fleetapi/acker"
+	"github.com/elastic/elastic-agent/pkg/core/logger"
+)
+
+type reexecCoordinator interface {
+	ReExec(callback reexec.ShutdownCallbackFn, argOverrides ...string)
+}
+
+type PrivilegeLevelChange struct {
+	log   *logger.Logger
+	coord reexecCoordinator
+	ch    chan coordinator.ConfigChange
+}
+
+func NewPrivilegeLevelChange(
+	log *logger.Logger,
+	coord reexecCoordinator,
+	ch chan coordinator.ConfigChange,
+) *PrivilegeLevelChange {
+	return &PrivilegeLevelChange{
+		log:   log,
+		coord: coord,
+		ch:    ch,
+	}
+}
+
+// Handle handles PRIVILEGE_LEVEL_CHANGE action.
+func (h *PrivilegeLevelChange) Handle(ctx context.Context, a fleetapi.Action, acker acker.Acker) error {
+	return h.handle(ctx, a, acker)
+}
+
+func (h *PrivilegeLevelChange) handle(ctx context.Context, a fleetapi.Action, acker acker.Acker) (rerr error) {
+	action, ok := a.(*fleetapi.ActionPrivilegeLevelChange)
+	if !ok {
+		return fmt.Errorf("invalid type, expected ActionPrivilegeLevelChange and received %T", a)
+	}
+
+	defer func() {
+		if rerr != nil {
+			h.ackFailure(ctx, rerr, action, acker)
+		}
+	}()
+
+	if !action.Data.Unprivileged {
+		// only unprivileged supported at this point
+		return fmt.Errorf("unsupported action, ActionPrivilegeLevelChange supports only downgrading permissions")
+	}
+
+	// ensure no component issues
+	err := componentvalidation.EnsureNoServiceComponentIssues()
+	if err != nil {
+		h.log.Debugf("handlerPrivilegeLevelChange: found issues with components: %v", err)
+		return err
+	}
+
+	var username, groupname, password string
+	if action.Data.UserInfo != nil {
+		username = action.Data.UserInfo.Username
+		groupname = action.Data.UserInfo.Groupname
+		password = action.Data.UserInfo.Password
+	}
+	username, password = install.UnprivilegedUser(username, password)
+	groupname = install.UnprivilegedGroup(groupname)
+
+	// apply empty config to stop processing
+	stopComponents(ctx, h.ch, a, acker, nil)
+
+	// fix permissions
+	topPath := paths.Top()
+	_, err = install.SwitchServiceUser(topPath, &debugDescriber{h.log}, username, groupname, password)
+	if err != nil {
+		// error already adds context
+		return err
+	}
+
+	// ack
+	if err := acker.Ack(ctx, a); err != nil {
+		h.log.Errorw("failed to ACK an action",
+			"error.message", err,
+			"action", a)
+	}
+	if err := acker.Commit(ctx); err != nil {
+		h.log.Errorw("failed to commit ACK of an action",
+			"error.message", err,
+			"action", a)
+	}
+
+	// check everything is properly set up
+	userName, groupName, err := install.GetDesiredUser()
+	if err != nil {
+		return fmt.Errorf("failed to determine target user: %w", err)
+	}
+
+	if userName != "" || groupName != "" {
+		_, err = install.EnsureUserAndGroup(userName, groupName, &debugDescriber{h.log}, true)
+		if err != nil {
+			return fmt.Errorf("failed to setup user: %w", err)
+		}
+	}
+
+	// restart
+	h.coord.ReExec(nil)
+	return nil
+}
+
+func (h *PrivilegeLevelChange) ackFailure(ctx context.Context, err error, action *fleetapi.ActionPrivilegeLevelChange, acker acker.Acker) {
+	action.Err = err
+
+	if err := acker.Ack(ctx, action); err != nil {
+		h.log.Errorw("failed to ack privilege level change action",
+			"error.message", err,
+			"action", action)
+	}
+
+	if err := acker.Commit(ctx); err != nil {
+		h.log.Errorw("failed to commit privilege level change action",
+			"error.message", err,
+			"action", action)
+	}
+}
+
+type debugDescriber struct {
+	l *logger.Logger
+}
+
+func (d *debugDescriber) Describe(a string) {
+	d.l.Debug(a)
+}

internal/pkg/agent/application/actions/handlers/handler_action_unenroll.go

@@ -10,7 +10,6 @@ import (
 	"time"
 
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/coordinator"
-	"github.com/elastic/elastic-agent/internal/pkg/config"
 	"github.com/elastic/elastic-agent/internal/pkg/fleetapi"
 	"github.com/elastic/elastic-agent/internal/pkg/fleetapi/acker"
 	"github.com/elastic/elastic-agent/pkg/core/logger"
@@ -92,20 +91,16 @@ func (h *Unenroll) handle(ctx context.Context, a fleetapi.Action, acker acker.Ac
 		a = nil
 	}
 
-	// Generate empty policy change, this removing all the running components
-	unenrollPolicy := newPolicyChange(ctx, config.New(), a, acker, true, false)
-	h.ch <- unenrollPolicy
-
 	// backup action for future start to avoid starting fleet gateway loop
-	h.stateStore.SetAction(a)
-	if err := h.stateStore.Save(); err != nil {
-		h.log.Warnf("Failed to update state store: %v", err)
+	backupCallback := func() {
+		h.stateStore.SetAction(a)
+		if err := h.stateStore.Save(); err != nil {
+			h.log.Warnf("Failed to update state store: %v", err)
+		}
 	}
 
-	unenrollCtx, cancel := context.WithTimeout(ctx, timeout)
-	defer cancel()
-
-	unenrollPolicy.WaitAck(unenrollCtx)
+	// Generate empty policy change, this removing all the running components
+	stopComponents(ctx, h.ch, a, acker, backupCallback)
 
 	// close fleet gateway loop
 	for _, c := range h.closers {

internal/pkg/agent/application/actions/handlers/handler_helpers.go

@@ -15,8 +15,10 @@ import (
 
 	"github.com/elastic/elastic-agent/internal/pkg/agent/application/coordinator"
 	"github.com/elastic/elastic-agent/internal/pkg/agent/errors"
+	"github.com/elastic/elastic-agent/internal/pkg/config"
 	"github.com/elastic/elastic-agent/internal/pkg/core/backoff"
 	"github.com/elastic/elastic-agent/internal/pkg/fleetapi"
+	"github.com/elastic/elastic-agent/internal/pkg/fleetapi/acker"
 	"github.com/elastic/elastic-agent/pkg/component"
 	"github.com/elastic/elastic-agent/pkg/component/runtime"
 )
@@ -43,6 +45,18 @@ type unitWithComponent struct {
 	component component.Component
 }
 
+func stopComponents(ctx context.Context, ch chan coordinator.ConfigChange, a fleetapi.Action, acker acker.Acker, backupCallback func()) {
+	unenrollPolicy := newPolicyChange(ctx, config.New(), a, acker, true, false)
+	ch <- unenrollPolicy
+
+	backupCallback()
+
+	unenrollCtx, cancel := context.WithTimeout(ctx, unenrollTimeout)
+	defer cancel()
+
+	unenrollPolicy.WaitAck(unenrollCtx)
+}
+
 func findMatchingUnitsByActionType(state coordinator.State, typ string) []unitWithComponent {
 	ucs := make([]unitWithComponent, 0)
 	for _, comp := range state.Components {

internal/pkg/agent/application/application.go

@@ -39,6 +39,7 @@ import (
 	"github.com/elastic/elastic-agent/internal/pkg/fleetapi/acker/lazy"
 	"github.com/elastic/elastic-agent/internal/pkg/fleetapi/acker/retrier"
 	fleetclient "github.com/elastic/elastic-agent/internal/pkg/fleetapi/client"
+	otelconfig "github.com/elastic/elastic-agent/internal/pkg/otel/config"
 	otelmanager "github.com/elastic/elastic-agent/internal/pkg/otel/manager"
 	"github.com/elastic/elastic-agent/internal/pkg/queue"
 	"github.com/elastic/elastic-agent/internal/pkg/release"
@@ -269,7 +270,7 @@ func New(
 	otelManager, err := otelmanager.NewOTelManager(
 		log.Named("otel_manager"),
 		logLevel, baseLogger,
-		otelmanager.SubprocessExecutionMode,
+		otelconfig.SubprocessExecutionMode,
 		agentInfo,
 		cfg.Settings.Collector,
 		monitor.ComponentMonitoringConfig,

internal/pkg/agent/application/managed_mode.go

@@ -402,6 +402,11 @@ func (m *managedConfigManager) initDispatcher(canceller context.CancelFunc) *han
 		handlers.NewMigrate(m.log, m.agentInfo, m.coord),
 	)
 
+	m.dispatcher.MustRegister(
+		&fleetapi.ActionPrivilegeLevelChange{},
+		handlers.NewPrivilegeLevelChange(m.log, m.coord, m.ch),
+	)
+
 	m.dispatcher.MustRegister(
 		&fleetapi.ActionUnknown{},
 		handlers.NewUnknown(m.log),