Ensure that the connecting Elastic Agent is a supported version. (#239)

* Ensure that the connecting Elastic Agent is a supported version.

* Revert change to fleet-server.yml

* Fix added space in fleet-server.yml.

* Fixes from code review.

* Update go.mod.

Author: blakerouse

NOTICE.txt

@@ -26,9 +26,10 @@ import (
 	"github.com/elastic/fleet-server/v7/internal/pkg/policy"
 	"github.com/elastic/fleet-server/v7/internal/pkg/smap"
 	"github.com/elastic/fleet-server/v7/internal/pkg/sqn"
-	"github.com/miolini/datacounter"
 
+	"github.com/hashicorp/go-version"
 	"github.com/julienschmidt/httprouter"
+	"github.com/miolini/datacounter"
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
 )
@@ -67,6 +68,7 @@ func (rt Router) handleCheckin(w http.ResponseWriter, r *http.Request, ps httpro
 }
 
 type CheckinT struct {
+	verCon version.Constraints
 	cfg    *config.Server
 	cache  cache.Cache
 	bc     *BulkCheckin
@@ -79,6 +81,7 @@ type CheckinT struct {
 }
 
 func NewCheckinT(
+	verCon version.Constraints,
 	cfg *config.Server,
 	c cache.Cache,
 	bc *BulkCheckin,
@@ -96,6 +99,7 @@ func NewCheckinT(
 		Msg("Checkin install limits")
 
 	ct := &CheckinT{
+		verCon: verCon,
 		cfg:    cfg,
 		cache:  c,
 		bc:     bc,
@@ -124,6 +128,11 @@ func (ct *CheckinT) _handleCheckin(w http.ResponseWriter, r *http.Request, id st
 		return err
 	}
 
+	err = validateUserAgent(r, ct.verCon)
+	if err != nil {
+		return err
+	}
+
 	// Metrics; serenity now.
 	dfunc := cntCheckin.IncStart()
 	defer dfunc()

cmd/fleet/handleCheckin.go

@@ -24,6 +24,7 @@ import (
 
 	"github.com/elastic/go-elasticsearch/v8"
 	"github.com/gofrs/uuid"
+	"github.com/hashicorp/go-version"
 	"github.com/julienschmidt/httprouter"
 	"github.com/miolini/datacounter"
 	"github.com/rs/zerolog/log"
@@ -41,18 +42,20 @@ var (
 )
 
 type EnrollerT struct {
+	verCon version.Constraints
 	bulker bulk.Bulk
 	cache  cache.Cache
 	limit  *limit.Limiter
 }
 
-func NewEnrollerT(cfg *config.Server, bulker bulk.Bulk, c cache.Cache) (*EnrollerT, error) {
+func NewEnrollerT(verCon version.Constraints, cfg *config.Server, bulker bulk.Bulk, c cache.Cache) (*EnrollerT, error) {
 
 	log.Info().
 		Interface("limits", cfg.Limits.EnrollLimit).
 		Msg("Enroller install limits")
 
 	return &EnrollerT{
+		verCon: verCon,
 		limit:  limit.NewLimiter(&cfg.Limits.EnrollLimit),
 		bulker: bulker,
 		cache:  c,
@@ -113,6 +116,11 @@ func (et *EnrollerT) handleEnroll(r *http.Request) ([]byte, error) {
 		return nil, err
 	}
 
+	err = validateUserAgent(r, et.verCon)
+	if err != nil {
+		return nil, err
+	}
+
 	// Metrics; serenity now.
 	dfunc := cntEnroll.IncStart()
 	defer dfunc()

cmd/fleet/handleEnroll.go

@@ -31,6 +31,7 @@ import (
 
 	"github.com/elastic/elastic-agent-client/v7/pkg/client"
 	"github.com/elastic/elastic-agent-client/v7/pkg/proto"
+	"github.com/hashicorp/go-version"
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
 	"github.com/spf13/cobra"
@@ -348,7 +349,8 @@ func (a *AgentMode) OnError(err error) {
 }
 
 type FleetServer struct {
-	version  string
+	ver      string
+	verCon   version.Constraints
 	policyId string
 
 	cfg      *config.Config
@@ -358,9 +360,14 @@ type FleetServer struct {
 }
 
 // NewFleetServer creates the actual fleet server service.
-func NewFleetServer(cfg *config.Config, c cache.Cache, version string, reporter status.Reporter) (*FleetServer, error) {
+func NewFleetServer(cfg *config.Config, c cache.Cache, verStr string, reporter status.Reporter) (*FleetServer, error) {
+	verCon, err := buildVersionConstraint(verStr)
+	if err != nil {
+		return nil, err
+	}
 	return &FleetServer{
-		version:  version,
+		ver:      verStr,
+		verCon:   verCon,
 		cfg:      cfg,
 		cfgCh:    make(chan *config.Config, 1),
 		cache:    c,
@@ -513,7 +520,7 @@ func (f *FleetServer) runServer(ctx context.Context, cfg *config.Config) (err er
 	}
 
 	g.Go(loggedRunFunc(ctx, "Policy index monitor", pim.Run))
-	cord := coordinator.NewMonitor(cfg.Fleet, f.version, bulker, pim, coordinator.NewCoordinatorZero)
+	cord := coordinator.NewMonitor(cfg.Fleet, f.ver, bulker, pim, coordinator.NewCoordinatorZero)
 	g.Go(loggedRunFunc(ctx, "Coordinator policy monitor", cord.Run))
 
 	// Policy monitor
@@ -545,8 +552,8 @@ func (f *FleetServer) runServer(ctx context.Context, cfg *config.Config) (err er
 	bc := NewBulkCheckin(bulker)
 	g.Go(loggedRunFunc(ctx, "Bulk checkin", bc.Run))
 
-	ct := NewCheckinT(&f.cfg.Inputs[0].Server, f.cache, bc, pm, am, ad, tr, bulker)
-	et, err := NewEnrollerT(&f.cfg.Inputs[0].Server, bulker, f.cache)
+	ct := NewCheckinT(f.verCon, &f.cfg.Inputs[0].Server, f.cache, bc, pm, am, ad, tr, bulker)
+	et, err := NewEnrollerT(f.verCon, &f.cfg.Inputs[0].Server, bulker, f.cache)
 	if err != nil {
 		return err
 	}

cmd/fleet/main.go

@@ -36,7 +36,7 @@ var (
 
 func (f *FleetServer) initMetrics(ctx context.Context, cfg *config.Config) (*api.Server, error) {
 	registry := monitoring.GetNamespace("info").GetRegistry()
-	monitoring.NewString(registry, "version").Set(f.version)
+	monitoring.NewString(registry, "version").Set(f.ver)
 	monitoring.NewString(registry, "name").Set("fleet-server")
 	metrics.SetupMetrics("fleet-server")
 

cmd/fleet/metrics.go

@@ -33,14 +33,15 @@ func TestRunServer(t *testing.T) {
 	cfg.Host = "localhost"
 	cfg.Port = port
 
+	verCon := mustBuildConstraints("8.0.0")
 	c, err := cache.New(cache.Config{NumCounters: 100, MaxCost: 100000})
 	require.NoError(t, err)
 	bulker := ftesting.MockBulk{}
 	pim := mock.NewMockIndexMonitor()
 	pm := policy.NewMonitor(bulker, pim, 5*time.Millisecond)
 	bc := NewBulkCheckin(nil)
-	ct := NewCheckinT(cfg, c, bc, pm, nil, nil, nil, nil)
-	et, err := NewEnrollerT(cfg, nil, c)
+	ct := NewCheckinT(verCon, cfg, c, bc, pm, nil, nil, nil, nil)
+	et, err := NewEnrollerT(verCon, cfg, nil, c)
 	require.NoError(t, err)
 
 	router := NewRouter(bulker, ct, et, nil, nil, nil)

cmd/fleet/server_test.go

@@ -0,0 +1,78 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package fleet
+
+import (
+	"errors"
+	"fmt"
+	"math"
+	"net/http"
+	"strconv"
+	"strings"
+
+	"github.com/hashicorp/go-version"
+)
+
+const (
+	// MinVersion is the minimum version an Elastic Agent must be to communicate
+	MinVersion = "7.13"
+
+	userAgentPrefix = "elastic agent "
+)
+
+var (
+	ErrInvalidUserAgent   = errors.New("user-agent is invalid")
+	ErrUnsupportedVersion = errors.New("version is not supported")
+)
+
+// buildVersionConstraint turns the version into a constraint to ensure that the connecting Elastic Agent's are
+// a supported version.
+func buildVersionConstraint(verStr string) (version.Constraints, error) {
+	ver, err := version.NewVersion(verStr)
+	if err != nil {
+		return nil, err
+	}
+	verStr = maximizePatch(ver)
+	return version.NewConstraint(fmt.Sprintf(">= %s, <= %s", MinVersion, verStr))
+}
+
+// maximizePatch turns the version into a string that has the patch value set to the maximum integer.
+//
+// Used to allow the Elastic Agent to be at a higher patch version than the Fleet Server, but require that the
+// Elastic Agent is not higher in MAJOR or MINOR.
+func maximizePatch(ver *version.Version) string {
+	segments := ver.Segments()
+	if len(segments) > 2 {
+		segments = segments[:2]
+	}
+	segments = append(segments, math.MaxInt32)
+	segStrs := make([]string, 0, len(segments))
+	for _, segment := range segments {
+		segStrs = append(segStrs, strconv.Itoa(segment))
+	}
+	return strings.Join(segStrs, ".")
+}
+
+// validateUserAgent validates that the User-Agent of the connecting Elastic Agent is valid and that the version is
+// supported for this Fleet Server.
+func validateUserAgent(r *http.Request, verConst version.Constraints) error {
+	userAgent := r.Header.Get("User-Agent")
+	if userAgent == "" {
+		return ErrInvalidUserAgent
+	}
+	userAgent = strings.ToLower(userAgent)
+	if !strings.HasPrefix(userAgent, userAgentPrefix) {
+		return ErrInvalidUserAgent
+	}
+	verStr := strings.TrimSpace(strings.TrimSuffix(strings.TrimPrefix(userAgent, userAgentPrefix), "-snapshot"))
+	ver, err := version.NewVersion(verStr)
+	if err != nil {
+		return ErrInvalidUserAgent
+	}
+	if !verConst.Check(ver) {
+		return ErrUnsupportedVersion
+	}
+	return nil
+}

cmd/fleet/userAgent.go

@@ -0,0 +1,99 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package fleet
+
+import (
+	"net/http/httptest"
+	"testing"
+
+	"github.com/hashicorp/go-version"
+)
+
+func TestValidateUserAgent(t *testing.T) {
+	tests := []struct {
+		userAgent string
+		verCon    version.Constraints
+		err       error
+	}{
+		{
+			userAgent: "",
+			verCon:    nil,
+			err:       ErrInvalidUserAgent,
+		},
+		{
+			userAgent: "bad value",
+			verCon:    nil,
+			err:       ErrInvalidUserAgent,
+		},
+		{
+			userAgent: "eLaStIc AGeNt",
+			verCon:    nil,
+			err:       ErrInvalidUserAgent,
+		},
+		{
+			userAgent: "eLaStIc AGeNt v7.10.0",
+			verCon:    mustBuildConstraints("7.13.0"),
+			err:       ErrUnsupportedVersion,
+		},
+		{
+			userAgent: "eLaStIc AGeNt v7.11.1",
+			verCon:    mustBuildConstraints("7.13.0"),
+			err:       ErrUnsupportedVersion,
+		},
+		{
+			userAgent: "eLaStIc AGeNt v7.12.5",
+			verCon:    mustBuildConstraints("7.13.0"),
+			err:       ErrUnsupportedVersion,
+		},
+		{
+			userAgent: "eLaStIc AGeNt v7.13.0",
+			verCon:    mustBuildConstraints("7.13.0"),
+			err:       nil,
+		},
+		{
+			userAgent: "eLaStIc AGeNt v7.13.0",
+			verCon:    mustBuildConstraints("7.13.1"),
+			err:       nil,
+		},
+		{
+			userAgent: "eLaStIc AGeNt v7.13.1",
+			verCon:    mustBuildConstraints("7.13.0"),
+			err:       nil,
+		},
+		{
+			userAgent: "eLaStIc AGeNt v7.14.0",
+			verCon:    mustBuildConstraints("7.13.0"),
+			err:       ErrUnsupportedVersion,
+		},
+		{
+			userAgent: "eLaStIc AGeNt v8.0.0",
+			verCon:    mustBuildConstraints("7.13.0"),
+			err:       ErrUnsupportedVersion,
+		},
+		{
+			userAgent: "eLaStIc AGeNt v7.13.0",
+			verCon:    mustBuildConstraints("8.0.0"),
+			err:       nil,
+		},
+	}
+	for _, tr := range tests {
+		t.Run(tr.userAgent, func(t *testing.T) {
+			req := httptest.NewRequest("GET", "/", nil)
+			req.Header.Set("User-Agent", tr.userAgent)
+			res := validateUserAgent(req, tr.verCon)
+			if tr.err != res {
+				t.Fatalf("err mismatch: %v != %v", tr.err, res)
+			}
+		})
+	}
+}
+
+func mustBuildConstraints(verStr string) version.Constraints {
+	con, err := buildVersionConstraint(verStr)
+	if err != nil {
+		panic(err)
+	}
+	return con
+}

cmd/fleet/userAgent_test.go

@@ -12,6 +12,7 @@ require (
 	github.com/gofrs/uuid v3.3.0+incompatible
 	github.com/google/go-cmp v0.4.0
 	github.com/hashicorp/go-cleanhttp v0.5.1
+	github.com/hashicorp/go-version v1.3.0
 	github.com/hashicorp/golang-lru v0.5.2-0.20190520140433-59383c442f7d
 	github.com/julienschmidt/httprouter v1.3.0
 	github.com/miolini/datacounter v1.0.2

go.mod

@@ -449,8 +449,9 @@ github.com/hashicorp/go-retryablehttp v0.6.6 h1:HJunrbHTDDbBb/ay4kxa1n+dLmttUlnP
 github.com/hashicorp/go-retryablehttp v0.6.6/go.mod h1:vAew36LZh98gCBJNLH42IQ1ER/9wtLZZ8meHqQvEYWY=
 github.com/hashicorp/go-uuid v1.0.2 h1:cfejS+Tpcp13yd5nYHWDI6qVCny6wyX2Mt5SGur2IGE=
 github.com/hashicorp/go-uuid v1.0.2/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
-github.com/hashicorp/go-version v1.0.0 h1:21MVWPKDphxa7ineQQTrCU5brh7OuVVAzGOCnnCPtE8=
 github.com/hashicorp/go-version v1.0.0/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA=
+github.com/hashicorp/go-version v1.3.0 h1:McDWVJIU/y+u1BRV06dPaLfLCaT7fUTJLp5r04x7iNw=
+github.com/hashicorp/go-version v1.3.0/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA=
 github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
 github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
 github.com/hashicorp/golang-lru v0.5.2-0.20190520140433-59383c442f7d h1:Ft6PtvobE9vwkCsuoNO5DZDbhKkKuktAlSsiOi1X5NA=