diff --git a/changelog/fragments/1778540235-fix-artifact-access-control.yaml b/changelog/fragments/1778540235-fix-artifact-access-control.yaml new file mode 100644 index 0000000000..f6ef4aeb60 --- /dev/null +++ b/changelog/fragments/1778540235-fix-artifact-access-control.yaml @@ -0,0 +1,32 @@ +# 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: security + +# Change summary; a 80ish characters long description of the change. +summary: Enforce policy-based access control on artifact downloads + +# Long description; in case the summary is not enough to describe the change +# this field accommodate a description without length limits. +# NOTE: This field will be rendered only for breaking-change and known-issue kinds at the moment. +#description: + +# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc. +component: fleet-server + +# PR URL; 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: https://github.com/owner/repo/1234 + +# 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 diff --git a/internal/pkg/api/error.go b/internal/pkg/api/error.go index 2e27df6655..29aff2e62e 100644 --- a/internal/pkg/api/error.go +++ b/internal/pkg/api/error.go @@ -142,6 +142,24 @@ func NewHTTPErrResp(err error) HTTPErrResp { zerolog.WarnLevel, }, }, + { + ErrUnauthorizedArtifact, + HTTPErrResp{ + http.StatusForbidden, + "Forbidden", + "agent not authorized for artifact", + zerolog.WarnLevel, + }, + }, + { + ErrAgentPolicyIDMissing, + HTTPErrResp{ + http.StatusForbidden, + "Forbidden", + "agent has no policy ID", + zerolog.WarnLevel, + }, + }, { ErrorThrottle, HTTPErrResp{ diff --git a/internal/pkg/api/handleArtifacts.go b/internal/pkg/api/handleArtifacts.go index 1074af5bed..d0b22a60e1 100644 --- a/internal/pkg/api/handleArtifacts.go +++ b/internal/pkg/api/handleArtifacts.go @@ -25,6 +25,7 @@ import ( "github.com/elastic/fleet-server/v7/internal/pkg/dl" "github.com/elastic/fleet-server/v7/internal/pkg/logger" "github.com/elastic/fleet-server/v7/internal/pkg/model" + "github.com/elastic/fleet-server/v7/internal/pkg/policy" "github.com/elastic/fleet-server/v7/internal/pkg/throttle" "github.com/rs/zerolog" @@ -36,23 +37,27 @@ const ( ) var ( - ErrorThrottle = errors.New("cannot acquire throttle token") - ErrorBadSha2 = errors.New("malformed sha256") - ErrorRecord = errors.New("artifact record mismatch") - ErrorMismatchSha2 = errors.New("mismatched sha256") + ErrorThrottle = errors.New("cannot acquire throttle token") + ErrorBadSha2 = errors.New("malformed sha256") + ErrorRecord = errors.New("artifact record mismatch") + ErrorMismatchSha2 = errors.New("mismatched sha256") + ErrUnauthorizedArtifact = errors.New("agent not authorized for artifact") + ErrAgentPolicyIDMissing = errors.New("agent has no policy ID") ) type ArtifactT struct { bulker bulk.Bulk cache cache.Cache esThrottle *throttle.Throttle + pm policy.Monitor } -func NewArtifactT(cfg *config.Server, bulker bulk.Bulk, cache cache.Cache) *ArtifactT { +func NewArtifactT(cfg *config.Server, bulker bulk.Bulk, cache cache.Cache, pm policy.Monitor) *ArtifactT { return &ArtifactT{ bulker: bulker, cache: cache, esThrottle: throttle.NewThrottle(defaultMaxParallel), + pm: pm, } } @@ -138,19 +143,73 @@ func (at ArtifactT) processRequest(ctx context.Context, zlog zerolog.Logger, age return rdr, nil } -// TODO: Pull the policy record for this agent and validate that the -// requested artifact is assigned to this policy. This will prevent -// agents from retrieving artifacts that they do not have access to. -// Note that this is racy, the policy could have changed to allow an -// artifact before this instantiation of FleetServer has its local -// copy updated. Take the race conditions into consideration. +// authorizeArtifact checks that the requested artifact is listed in the agent's +// assigned policy. The policy is read from the in-memory cache maintained by the +// policy monitor, which introduces a staleness window between ES updates and +// cache refresh. This creates two race conditions: // -// Initial implementation is dependent on security by obscurity; ie. -// it should be difficult for an attacker to guess a guid. -func (at ArtifactT) authorizeArtifact(ctx context.Context, _ *model.Agent, _, _ string) error { - span, _ := apm.StartSpan(ctx, "authorizeArtifacts", "auth") // TODO return and use span ctx if this is ever not a nop +// - False negative (deny when should allow): an artifact was just added to a +// policy but the cache hasn't refreshed yet. This is self-healing — the agent +// will retry and succeed once the cache catches up. +// +// - False positive (allow when should deny): an artifact was just removed from +// a policy but the stale cache still lists it. The agent can download the +// artifact until the cache refreshes. This is the security-sensitive direction. +// A future improvement could verify against ES when the cache says "allow". +func (at ArtifactT) authorizeArtifact(ctx context.Context, agent *model.Agent, id, sha2 string) error { + span, ctx := apm.StartSpan(ctx, "authorizeArtifacts", "auth") defer span.End() - return nil // TODO + + // AgentPolicyID (agent_policy_id) is set at first checkin; PolicyID (policy_id) is set + // at enrollment. Fall back to PolicyID so newly enrolled agents that have not yet + // checked in can still download artifacts for their assigned policy. + policyID := agent.AgentPolicyID + if policyID == "" { + policyID = agent.PolicyID + } + if policyID == "" { + return ErrAgentPolicyIDMissing + } + + p, err := at.pm.GetPolicy(ctx, policyID) + if errors.Is(err, policy.ErrPolicyNotFound) { + return ErrUnauthorizedArtifact + } + if err != nil { + return fmt.Errorf("authorizeArtifact: %w", err) + } + + if p.Data != nil && policyHasArtifact(p.Data, id, sha2) { + return nil + } + + return ErrUnauthorizedArtifact +} + +func policyHasArtifact(pd *model.PolicyData, id, sha2 string) bool { + for _, input := range pd.Inputs { + if inputHasArtifact(input, id, sha2) { + return true + } + } + return false +} + +func inputHasArtifact(input map[string]any, id, sha2 string) bool { + manifestRaw, ok := input[dl.FieldArtifactManifest].(map[string]any) + if !ok { + return false + } + artifacts, ok := manifestRaw[dl.FieldArtifacts].(map[string]any) + if !ok { + return false + } + entry, ok := artifacts[id].(map[string]any) + if !ok { + return false + } + sha, _ := entry[dl.FieldDecodedSha256].(string) + return sha == sha2 } // Return artifact from cache by sha2 or fetch directly from Elastic. diff --git a/internal/pkg/api/handleArtifacts_test.go b/internal/pkg/api/handleArtifacts_test.go new file mode 100644 index 0000000000..5cdc3965cb --- /dev/null +++ b/internal/pkg/api/handleArtifacts_test.go @@ -0,0 +1,252 @@ +// 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. + +//go:build !integration + +package api + +import ( + "context" + "errors" + "testing" + + "github.com/elastic/fleet-server/v7/internal/pkg/model" + "github.com/elastic/fleet-server/v7/internal/pkg/policy" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type stubPolicyMonitor struct { + getPolicy func(ctx context.Context, policyID string) (*model.Policy, error) +} + +func (s *stubPolicyMonitor) Run(_ context.Context) error { return nil } +func (s *stubPolicyMonitor) Subscribe(_, _ string, _ int64) (policy.Subscription, error) { + return nil, nil +} +func (s *stubPolicyMonitor) Unsubscribe(_ policy.Subscription) error { return nil } +func (s *stubPolicyMonitor) LatestRev(_ context.Context, _ string) int64 { return 0 } +func (s *stubPolicyMonitor) GetPolicy(ctx context.Context, policyID string) (*model.Policy, error) { + return s.getPolicy(ctx, policyID) +} + +func TestPolicyHasArtifact(t *testing.T) { + policyData := &model.PolicyData{ + Inputs: []map[string]any{ + { + "type": "logfile", + "id": "logfile-1", + }, + { + "type": "endpoint", + "id": "endpoint-1", + "artifact_manifest": map[string]any{ + "manifest_version": "1.0.28", + "schema_version": "v1", + "artifacts": map[string]any{ + "endpoint-trustlist-windows-v1": map[string]any{ + "decoded_sha256": "74c2255ce31e0b48ada298ed6dacf6d1be7b0fb40c1bcb251d2da66f4b060acf", + "decoded_size": float64(338), + }, + "endpoint-trustlist-linux-v1": map[string]any{ + "decoded_sha256": "d801aa1fb7ddcc330a5e3173372ea6af4a3d08ec58074478e85aa5603e926658", + "decoded_size": float64(14), + }, + }, + }, + }, + }, + } + + tests := []struct { + name string + id string + sha2 string + want bool + }{ + { + name: "matching artifact", + id: "endpoint-trustlist-windows-v1", + sha2: "74c2255ce31e0b48ada298ed6dacf6d1be7b0fb40c1bcb251d2da66f4b060acf", + want: true, + }, + { + name: "matching linux artifact", + id: "endpoint-trustlist-linux-v1", + sha2: "d801aa1fb7ddcc330a5e3173372ea6af4a3d08ec58074478e85aa5603e926658", + want: true, + }, + { + name: "unknown artifact id", + id: "endpoint-blocklist-windows-v1", + sha2: "74c2255ce31e0b48ada298ed6dacf6d1be7b0fb40c1bcb251d2da66f4b060acf", + want: false, + }, + { + name: "wrong sha256 for known id", + id: "endpoint-trustlist-windows-v1", + sha2: "0000000000000000000000000000000000000000000000000000000000000000", + want: false, + }, + { + name: "empty id", + id: "", + sha2: "74c2255ce31e0b48ada298ed6dacf6d1be7b0fb40c1bcb251d2da66f4b060acf", + want: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := policyHasArtifact(policyData, tt.id, tt.sha2) + assert.Equal(t, tt.want, got) + }) + } +} + +func TestPolicyHasArtifact_NoInputs(t *testing.T) { + pd := &model.PolicyData{} + assert.False(t, policyHasArtifact(pd, "endpoint-trustlist-linux-v1", "abc123")) +} + +func TestPolicyHasArtifact_NoArtifactManifest(t *testing.T) { + pd := &model.PolicyData{ + Inputs: []map[string]any{ + {"type": "logfile"}, + }, + } + assert.False(t, policyHasArtifact(pd, "endpoint-trustlist-linux-v1", "abc123")) +} + +func TestPolicyHasArtifact_MultipleInputsWithArtifacts(t *testing.T) { + pd := &model.PolicyData{ + Inputs: []map[string]any{ + {"type": "logfile"}, + { + "type": "endpoint", + "artifact_manifest": map[string]any{ + "artifacts": map[string]any{ + "endpoint-trustlist-linux-v1": map[string]any{ + "decoded_sha256": "aaaa", + }, + }, + }, + }, + { + "type": "another-endpoint", + "artifact_manifest": map[string]any{ + "artifacts": map[string]any{ + "endpoint-blocklist-linux-v1": map[string]any{ + "decoded_sha256": "bbbb", + }, + }, + }, + }, + }, + } + assert.True(t, policyHasArtifact(pd, "endpoint-trustlist-linux-v1", "aaaa")) + assert.True(t, policyHasArtifact(pd, "endpoint-blocklist-linux-v1", "bbbb")) + assert.False(t, policyHasArtifact(pd, "endpoint-trustlist-linux-v1", "bbbb")) +} + +func TestAuthorizeArtifact(t *testing.T) { + const ( + policyID = "test-policy-id" + artifactID = "endpoint-trustlist-linux-v1" + sha2 = "d801aa1fb7ddcc330a5e3173372ea6af4a3d08ec58074478e85aa5603e926658" + ) + + policyWithArtifact := &model.Policy{ + Data: &model.PolicyData{ + Inputs: []map[string]any{ + { + "type": "endpoint", + "artifact_manifest": map[string]any{ + "artifacts": map[string]any{ + artifactID: map[string]any{ + "decoded_sha256": sha2, + }, + }, + }, + }, + }, + }, + } + + tests := []struct { + name string + agent *model.Agent + setupMock func(pm *mockPolicyMonitor) + wantErr error + }{ + { + name: "authorized: artifact in policy", + agent: &model.Agent{AgentPolicyID: policyID}, + setupMock: func(pm *mockPolicyMonitor) { + pm.On("GetPolicy", context.Background(), policyID).Return(policyWithArtifact, nil) + }, + wantErr: nil, + }, + { + name: "unauthorized: artifact not in policy", + agent: &model.Agent{AgentPolicyID: policyID}, + setupMock: func(pm *mockPolicyMonitor) { + pm.On("GetPolicy", context.Background(), policyID).Return(&model.Policy{ + Data: &model.PolicyData{}, + }, nil) + }, + wantErr: ErrUnauthorizedArtifact, + }, + { + name: "unauthorized: policy not found maps to 403", + agent: &model.Agent{AgentPolicyID: policyID}, + setupMock: func(pm *mockPolicyMonitor) { + pm.On("GetPolicy", context.Background(), policyID).Return(nil, policy.ErrPolicyNotFound) + }, + wantErr: ErrUnauthorizedArtifact, + }, + { + name: "error: GetPolicy returns unexpected error", + agent: &model.Agent{AgentPolicyID: policyID}, + setupMock: func(pm *mockPolicyMonitor) { + pm.On("GetPolicy", context.Background(), policyID).Return(nil, errors.New("elasticsearch unavailable")) + }, + wantErr: nil, // wrapped, so we check IsUnauthorized is false + }, + { + name: "authorized: uses PolicyID when AgentPolicyID is empty (pre-checkin)", + agent: &model.Agent{PolicyID: policyID}, + setupMock: func(pm *mockPolicyMonitor) { + pm.On("GetPolicy", context.Background(), policyID).Return(policyWithArtifact, nil) + }, + wantErr: nil, + }, + { + name: "forbidden: agent has no policy ID", + agent: &model.Agent{}, + setupMock: func(pm *mockPolicyMonitor) {}, + wantErr: ErrAgentPolicyIDMissing, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + pm := &mockPolicyMonitor{} + tt.setupMock(pm) + at := ArtifactT{pm: pm} + + err := at.authorizeArtifact(context.Background(), tt.agent, artifactID, sha2) + + if tt.name == "error: GetPolicy returns unexpected error" { + require.Error(t, err) + assert.False(t, errors.Is(err, ErrUnauthorizedArtifact)) + } else if tt.wantErr != nil { + require.ErrorIs(t, err, tt.wantErr) + } else { + require.NoError(t, err) + } + pm.AssertExpectations(t) + }) + } +} diff --git a/internal/pkg/api/handleCheckin_test.go b/internal/pkg/api/handleCheckin_test.go index f288bb069e..52d87f2e10 100644 --- a/internal/pkg/api/handleCheckin_test.go +++ b/internal/pkg/api/handleCheckin_test.go @@ -64,6 +64,12 @@ func (m *mockPolicyMonitor) LatestRev(ctx context.Context, id string) int64 { return args.Get(0).(int64) } +func (m *mockPolicyMonitor) GetPolicy(ctx context.Context, policyID string) (*model.Policy, error) { + args := m.Called(ctx, policyID) + p, _ := args.Get(0).(*model.Policy) + return p, args.Error(1) +} + func TestConvertActionData(t *testing.T) { tests := []struct { name string diff --git a/internal/pkg/dl/constants.go b/internal/pkg/dl/constants.go index 541efd2e94..5ca9dc27c4 100644 --- a/internal/pkg/dl/constants.go +++ b/internal/pkg/dl/constants.go @@ -65,10 +65,12 @@ const ( FieldAuditUnenrolledTime = "audit_unenrolled_time" FieldAuditUnenrolledReason = "audit_unenrolled_reason" - FieldDecodedSha256 = "decoded_sha256" - FieldIdentifier = "identifier" - FieldSharedID = "shared_id" - FieldEnrollmentID = "enrollment_id" + FieldArtifactManifest = "artifact_manifest" + FieldArtifacts = "artifacts" + FieldDecodedSha256 = "decoded_sha256" + FieldIdentifier = "identifier" + FieldSharedID = "shared_id" + FieldEnrollmentID = "enrollment_id" ) // Private constants diff --git a/internal/pkg/policy/monitor.go b/internal/pkg/policy/monitor.go index 36ed820aae..c38552f198 100644 --- a/internal/pkg/policy/monitor.go +++ b/internal/pkg/policy/monitor.go @@ -7,6 +7,7 @@ package policy import ( "context" "errors" + "fmt" "sync" "time" @@ -25,6 +26,8 @@ import ( const cloudPolicyID = "policy-elastic-agent-on-cloud" +var ErrPolicyNotFound = errors.New("policy not found") + /* Design should have the following properties @@ -65,6 +68,10 @@ type Monitor interface { // LatestRev returns the latest revision idx for the specified policy. LatestRev(ctx context.Context, policyID string) int64 + + // GetPolicy returns the cached policy for the given policy ID. + // If the policy is not in the cache, it forces a reload from Elasticsearch. + GetPolicy(ctx context.Context, policyID string) (*model.Policy, error) } type policyFetcher func(ctx context.Context, bulker bulk.Bulk, opt ...dl.Option) ([]model.Policy, error) @@ -591,3 +598,32 @@ func (m *monitorT) LatestRev(ctx context.Context, id string) int64 { } return p.pp.Policy.RevisionIdx } + +// GetPolicy returns the policy for the given policy ID from the in-memory cache. +// If the policy is not found in cache, all policies are reloaded from Elasticsearch. +func (m *monitorT) GetPolicy(ctx context.Context, policyID string) (*model.Policy, error) { + if policyID == "" { + return nil, errors.New("policy ID is empty") + } + + m.mut.Lock() + p, ok := m.policies[policyID] + m.mut.Unlock() + + if ok { + return &p.pp.Policy, nil + } + + if err := m.loadPolicies(ctx); err != nil { + return nil, fmt.Errorf("loading policies: %w", err) + } + + m.mut.Lock() + p, ok = m.policies[policyID] + m.mut.Unlock() + + if !ok { + return nil, fmt.Errorf("%w: %s", ErrPolicyNotFound, policyID) + } + return &p.pp.Policy, nil +} diff --git a/internal/pkg/server/fleet.go b/internal/pkg/server/fleet.go index f1c1ef5b48..c3b98b1295 100644 --- a/internal/pkg/server/fleet.go +++ b/internal/pkg/server/fleet.go @@ -533,7 +533,7 @@ func (f *Fleet) runSubsystems(ctx context.Context, cfg *config.Config, g *errgro return err } - at := api.NewArtifactT(&cfg.Inputs[0].Server, bulker, f.cache) + at := api.NewArtifactT(&cfg.Inputs[0].Server, bulker, f.cache, pm) ack := api.NewAckT(&cfg.Inputs[0].Server, bulker, f.cache) st := api.NewStatusT(&cfg.Inputs[0].Server, bulker, f.cache, api.WithSelfMonitor(sm), api.WithBuildInfo(f.bi)) oa := api.NewOpAMPT(ctx, &cfg.Inputs[0].Server, bulker, f.cache, bc) diff --git a/testing/e2e/api_version/client_api_2023_06_01.go b/testing/e2e/api_version/client_api_2023_06_01.go index 5dd7c97d3d..1164ab022d 100644 --- a/testing/e2e/api_version/client_api_2023_06_01.go +++ b/testing/e2e/api_version/client_api_2023_06_01.go @@ -305,13 +305,47 @@ func (tester *ClientAPITester20230601) TestFullFileUpload() { } func (tester *ClientAPITester20230601) TestArtifact() { - ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute) + ctx, cancel := context.WithTimeout(tester.T().Context(), 5*time.Minute) defer cancel() - _, agentKey := tester.Enroll(ctx, tester.enrollmentKey) + // Elastic Defend (endpoint) artifacts are only authorized for agents enrolled under a + // policy that has the Elastic Defend integration (security-policy). dummy-policy has no + // Elastic Defend input and therefore no artifact_manifest, so agents enrolled there + // would get 403 when downloading endpoint artifacts. + securityPolicyKey := tester.GetEnrollmentTokenForPolicyID(ctx, "security-policy") + _, agentKey := tester.Enroll(ctx, securityPolicyKey) tester.AddSecurityContainer(ctx) tester.AddSecurityContainerItem(ctx) hits := tester.FleetHasArtifacts(ctx) - tester.Artifact(ctx, agentKey, hits[0].Source.Identifier, hits[0].Source.DecodedSHA256, hits[0].Source.EncodedSHA256) + id, sha2, encodedSHA := hits[0].Source.Identifier, hits[0].Source.DecodedSHA256, hits[0].Source.EncodedSHA256 + // Wait for the policy document in ES to reference the artifact. This also gives the + // fleet-server policy monitor time to refresh its cache before we attempt the download. + tester.FleetPolicyHasArtifact(ctx, "security-policy", id, sha2) + // Retry on 403: even after the ES policy is updated, the in-memory cache in fleet-server + // may not have caught up yet. The monitor refreshes quickly but retrying is more robust. + apiClient, err := tester.getAPIClient(func(_ context.Context, req *http.Request) error { + req.Header.Set("Authorization", "ApiKey "+agentKey) + return nil + }) + tester.Require().NoError(err) + for { + if err := ctx.Err(); err != nil { + tester.Require().NoError(err, "context expired before artifact download succeeded") + } + resp, err := apiClient.ArtifactWithResponse(ctx, id, sha2, &api.ArtifactParams{}) + tester.Require().NoError(err) + if resp.StatusCode() == http.StatusForbidden { + select { + case <-ctx.Done(): + tester.Require().NoError(ctx.Err(), "context expired while retrying artifact download") + case <-time.After(time.Second): + } + continue + } + tester.Require().Equal(http.StatusOK, resp.StatusCode()) + hash := sha256.Sum256(resp.Body) + tester.Require().Equal(encodedSHA, fmt.Sprintf("%x", hash[:])) + break + } } diff --git a/testing/e2e/api_version/client_api_current.go b/testing/e2e/api_version/client_api_current.go index 4f6c587b6a..3b439680ac 100644 --- a/testing/e2e/api_version/client_api_current.go +++ b/testing/e2e/api_version/client_api_current.go @@ -405,15 +405,49 @@ func (tester *ClientAPITester) TestFullFileUpload() { } func (tester *ClientAPITester) TestArtifact() { - ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute) + ctx, cancel := context.WithTimeout(tester.T().Context(), 5*time.Minute) defer cancel() - _, agentKey := tester.Enroll(ctx, tester.enrollmentKey) + // Elastic Defend (endpoint) artifacts are only authorized for agents enrolled under a + // policy that has the Elastic Defend integration (security-policy). dummy-policy has no + // Elastic Defend input and therefore no artifact_manifest, so agents enrolled there + // would get 403 when downloading endpoint artifacts. + securityPolicyKey := tester.GetEnrollmentTokenForPolicyID(ctx, "security-policy") + _, agentKey := tester.Enroll(ctx, securityPolicyKey) tester.AddSecurityContainer(ctx) tester.AddSecurityContainerItem(ctx) hits := tester.FleetHasArtifacts(ctx) - tester.Artifact(ctx, agentKey, hits[0].Source.Identifier, hits[0].Source.DecodedSHA256, hits[0].Source.EncodedSHA256) + id, sha2, encodedSHA := hits[0].Source.Identifier, hits[0].Source.DecodedSHA256, hits[0].Source.EncodedSHA256 + // Wait for the policy document in ES to reference the artifact. This also gives the + // fleet-server policy monitor time to refresh its cache before we attempt the download. + tester.FleetPolicyHasArtifact(ctx, "security-policy", id, sha2) + // Retry on 403: even after the ES policy is updated, the in-memory cache in fleet-server + // may not have caught up yet. The monitor refreshes quickly but retrying is more robust. + for { + if err := ctx.Err(); err != nil { + tester.Require().NoError(err, "context expired before artifact download succeeded") + } + client, err := api.NewClientWithResponses(tester.endpoint, api.WithHTTPClient(tester.Client), api.WithRequestEditorFn(func(_ context.Context, req *http.Request) error { + req.Header.Set("Authorization", "ApiKey "+agentKey) + return nil + })) + tester.Require().NoError(err) + resp, err := client.ArtifactWithResponse(ctx, id, sha2, &api.ArtifactParams{}) + tester.Require().NoError(err) + if resp.StatusCode() == http.StatusForbidden { + select { + case <-ctx.Done(): + tester.Require().NoError(ctx.Err(), "context expired while retrying artifact download") + case <-time.After(time.Second): + } + continue + } + tester.Require().Equal(http.StatusOK, resp.StatusCode()) + hash := sha256.Sum256(resp.Body) + tester.Require().Equal(encodedSHA, fmt.Sprintf("%x", hash[:])) + break + } } func (tester *ClientAPITester) TestGetPGPKey() { diff --git a/testing/e2e/scaffold/scaffold.go b/testing/e2e/scaffold/scaffold.go index 4e214fc20c..cf6d68f5f1 100644 --- a/testing/e2e/scaffold/scaffold.go +++ b/testing/e2e/scaffold/scaffold.go @@ -592,6 +592,78 @@ func (s *Scaffold) FleetHasArtifacts(ctx context.Context) []ArtifactHit { } } +// FleetPolicyHasArtifact polls .fleet-policies until the most recent revision for the +// given policyID has the specified artifact (by identifier and decoded SHA256) in at +// least one input's artifact_manifest. This ensures the policy monitor will have +// up-to-date data before the caller attempts an artifact download. +func (s *Scaffold) FleetPolicyHasArtifact(ctx context.Context, policyID, identifier, decodedSha256 string) { + timer := time.NewTimer(time.Second) + for { + query := fmt.Sprintf(`{"query":{"term":{"policy_id":"%s"}},"sort":[{"revision_idx":{"order":"desc"}}],"size":1}`, policyID) + req, err := http.NewRequestWithContext(ctx, "POST", "http://localhost:9200/.fleet-policies/_search", strings.NewReader(query)) + s.Require().NoError(err) + req.SetBasicAuth(s.ElasticUser, s.ElasticPass) + req.Header.Set("Content-Type", "application/json") + select { + case <-ctx.Done(): + s.Require().NoError(ctx.Err(), "context expired before policy artifact_manifest was updated") + return + case <-timer.C: + resp, err := s.Client.Do(req) + s.Require().NoError(err) + if resp.StatusCode != http.StatusOK { + resp.Body.Close() + timer.Reset(time.Second) + continue + } + + var result struct { + Hits struct { + Hits []struct { + Source struct { + Data struct { + Inputs []map[string]any `json:"inputs"` + } `json:"data"` + } `json:"_source"` + } `json:"hits"` + Total struct { + Value int `json:"value"` + } `json:"total"` + } `json:"hits"` + } + err = json.NewDecoder(resp.Body).Decode(&result) + resp.Body.Close() + s.Require().NoError(err) + + if result.Hits.Total.Value > 0 { + for _, input := range result.Hits.Hits[0].Source.Data.Inputs { + if inputHasArtifact(input, identifier, decodedSha256) { + return + } + } + } + timer.Reset(time.Second) + } + } +} + +func inputHasArtifact(input map[string]any, identifier, decodedSha256 string) bool { + manifestRaw, ok := input["artifact_manifest"].(map[string]any) + if !ok { + return false + } + artifacts, ok := manifestRaw["artifacts"].(map[string]any) + if !ok { + return false + } + entry, ok := artifacts[identifier].(map[string]any) + if !ok { + return false + } + sha, _ := entry["decoded_sha256"].(string) + return sha == decodedSha256 +} + type logger struct { *testing.T }