diff --git a/.github/workflows/test-go.yaml b/.github/workflows/test-go.yaml index ae165c368d..0cd3b79003 100644 --- a/.github/workflows/test-go.yaml +++ b/.github/workflows/test-go.yaml @@ -49,7 +49,7 @@ jobs: # Pre-starting dependencies here means they are ready to go when we need them. - name: Start Infra Dependencies # Use & to background this - run: FLEET_MYSQL_IMAGE=${{ matrix.mysql }} docker-compose up -d mysql_test redis redis-cluster-1 redis-cluster-2 redis-cluster-3 redis-cluster-4 redis-cluster-5 redis-cluster-6 redis-cluster-setup & + run: FLEET_MYSQL_IMAGE=${{ matrix.mysql }} docker-compose up -d mysql_test redis redis-cluster-1 redis-cluster-2 redis-cluster-3 redis-cluster-4 redis-cluster-5 redis-cluster-6 redis-cluster-setup minio & # It seems faster not to cache Go dependencies - name: Install Go Dependencies @@ -73,6 +73,7 @@ jobs: NETWORK_TEST=1 \ REDIS_TEST=1 \ MYSQL_TEST=1 \ + MINIO_STORAGE_TEST=1 \ make test-go 2>&1 | tee /tmp/gotest.log | gotestfmt - name: Upload to Codecov diff --git a/cmd/fleet/serve.go b/cmd/fleet/serve.go index 725b4e6bd0..80c2b3f90c 100644 --- a/cmd/fleet/serve.go +++ b/cmd/fleet/serve.go @@ -199,7 +199,7 @@ the way that the Fleet server works. } if config.S3.Bucket != "" { - carveStore, err = s3.New(config.S3, ds) + carveStore, err = s3.NewCarveStore(config.S3, ds) if err != nil { initFatal(err, "initializing S3 carvestore") } diff --git a/docs/Contributing/Testing.md b/docs/Contributing/Testing.md index 5149a54310..05a82e77a8 100644 --- a/docs/Contributing/Testing.md +++ b/docs/Contributing/Testing.md @@ -50,7 +50,7 @@ REDIS_TEST=1 MYSQL_TEST=1 make test To run all Go unit tests, run the following: ``` -REDIS_TEST=1 MYSQL_TEST=1 make test-go +REDIS_TEST=1 MYSQL_TEST=1 MINIO_STORAGE_TEST=1 make test-go ``` ### Go linters diff --git a/server/datastore/s3/carves.go b/server/datastore/s3/carves.go index d582d6199f..110fbe92d8 100644 --- a/server/datastore/s3/carves.go +++ b/server/datastore/s3/carves.go @@ -11,6 +11,7 @@ import ( "github.com/aws/aws-sdk-go/aws/awserr" "github.com/aws/aws-sdk-go/service/s3" + "github.com/fleetdm/fleet/v4/server/config" "github.com/fleetdm/fleet/v4/server/contexts/ctxerr" "github.com/fleetdm/fleet/v4/server/fleet" ) @@ -24,49 +25,66 @@ const ( timePrefixFormat = "2006/01/02/15" ) +// CarveStore is a type implementing the CarveStore interface +// relying on AWS S3 storage +type CarveStore struct { + *s3store + metadatadb fleet.CarveStore +} + +// NewCarveStore creates a new store with the given config +func NewCarveStore(config config.S3Config, metadatadb fleet.CarveStore) (*CarveStore, error) { + s3store, err := newS3store(config) + if err != nil { + return nil, err + } + + return &CarveStore{s3store, metadatadb}, nil +} + // generateS3Key builds S3 key from carve metadata // all keys are prefixed by date so that they can easily be listed chronologically -func (d *Datastore) generateS3Key(metadata *fleet.CarveMetadata) string { +func (c *CarveStore) generateS3Key(metadata *fleet.CarveMetadata) string { simpleDateHour := metadata.CreatedAt.Format(timePrefixFormat) - return fmt.Sprintf("%s%s/%s", d.prefix, simpleDateHour, metadata.Name) + return fmt.Sprintf("%s%s/%s", c.prefix, simpleDateHour, metadata.Name) } // NewCarve initializes a new file carving session -func (d *Datastore) NewCarve(ctx context.Context, metadata *fleet.CarveMetadata) (*fleet.CarveMetadata, error) { - objectKey := d.generateS3Key(metadata) - res, err := d.s3client.CreateMultipartUpload(&s3.CreateMultipartUploadInput{ - Bucket: &d.bucket, +func (c *CarveStore) NewCarve(ctx context.Context, metadata *fleet.CarveMetadata) (*fleet.CarveMetadata, error) { + objectKey := c.generateS3Key(metadata) + res, err := c.s3client.CreateMultipartUpload(&s3.CreateMultipartUploadInput{ + Bucket: &c.bucket, Key: &objectKey, }) if err != nil { return nil, ctxerr.Wrap(ctx, err, "s3 multipart carve create") } metadata.SessionId = *res.UploadId - return d.metadatadb.NewCarve(ctx, metadata) + return c.metadatadb.NewCarve(ctx, metadata) } // UpdateCarve updates carve definition in database // Only max_block and expired are updatable -func (d *Datastore) UpdateCarve(ctx context.Context, metadata *fleet.CarveMetadata) error { - return d.metadatadb.UpdateCarve(ctx, metadata) +func (c *CarveStore) UpdateCarve(ctx context.Context, metadata *fleet.CarveMetadata) error { + return c.metadatadb.UpdateCarve(ctx, metadata) } // listS3Carves lists all keys up to a given one or if the passed max number // of keys has been reached; keys are returned in a set-like map -func (d *Datastore) listS3Carves(lastPrefix string, maxKeys int) (map[string]bool, error) { +func (c *CarveStore) listS3Carves(lastPrefix string, maxKeys int) (map[string]bool, error) { var err error var continuationToken string result := make(map[string]bool) if maxKeys <= 0 { maxKeys = defaultMaxS3Keys } - if !strings.HasPrefix(lastPrefix, d.prefix) { - lastPrefix = d.prefix + lastPrefix + if !strings.HasPrefix(lastPrefix, c.prefix) { + lastPrefix = c.prefix + lastPrefix } for { - carveFilesPage, err := d.s3client.ListObjectsV2(&s3.ListObjectsV2Input{ - Bucket: &d.bucket, - Prefix: &d.prefix, + carveFilesPage, err := c.s3client.ListObjectsV2(&s3.ListObjectsV2Input{ + Bucket: &c.bucket, + Prefix: &c.prefix, ContinuationToken: &continuationToken, }) if err != nil { @@ -90,10 +108,10 @@ func (d *Datastore) listS3Carves(lastPrefix string, maxKeys int) (map[string]boo // lifecycle configurations provided by AWS. This will compare a portion of the // metadata present in the database and mark as expired the carves no longer // available in S3 (ignores the `now` argument) -func (d *Datastore) CleanupCarves(ctx context.Context, now time.Time) (int, error) { +func (c *CarveStore) CleanupCarves(ctx context.Context, now time.Time) (int, error) { var err error // Get the 1000 oldest carves - nonExpiredCarves, err := d.ListCarves(ctx, fleet.CarveListOptions{ + nonExpiredCarves, err := c.ListCarves(ctx, fleet.CarveListOptions{ ListOptions: fleet.ListOptions{PerPage: cleanupSize}, Expired: false, }) @@ -102,17 +120,17 @@ func (d *Datastore) CleanupCarves(ctx context.Context, now time.Time) (int, erro } // List carves in S3 up to a hour+1 prefix lastCarveNextHour := nonExpiredCarves[len(nonExpiredCarves)-1].CreatedAt.Add(time.Hour) - lastCarvePrefix := d.prefix + lastCarveNextHour.Format(timePrefixFormat) - carveKeys, err := d.listS3Carves(lastCarvePrefix, 2*cleanupSize) + lastCarvePrefix := c.prefix + lastCarveNextHour.Format(timePrefixFormat) + carveKeys, err := c.listS3Carves(lastCarvePrefix, 2*cleanupSize) if err != nil { return 0, ctxerr.Wrap(ctx, err, "s3 carve cleanup") } // Compare carve metadata in DB with S3 listing and update expiration flag cleanCount := 0 for _, carve := range nonExpiredCarves { - if _, ok := carveKeys[d.generateS3Key(carve)]; !ok { + if _, ok := carveKeys[c.generateS3Key(carve)]; !ok { carve.Expired = true - err = d.UpdateCarve(ctx, carve) + err = c.UpdateCarve(ctx, carve) cleanCount++ } } @@ -120,33 +138,33 @@ func (d *Datastore) CleanupCarves(ctx context.Context, now time.Time) (int, erro } // Carve returns carve metadata by ID -func (d *Datastore) Carve(ctx context.Context, carveID int64) (*fleet.CarveMetadata, error) { - return d.metadatadb.Carve(ctx, carveID) +func (c *CarveStore) Carve(ctx context.Context, carveID int64) (*fleet.CarveMetadata, error) { + return c.metadatadb.Carve(ctx, carveID) } // CarveBySessionId returns carve metadata by session ID -func (d *Datastore) CarveBySessionId(ctx context.Context, sessionID string) (*fleet.CarveMetadata, error) { - return d.metadatadb.CarveBySessionId(ctx, sessionID) +func (c *CarveStore) CarveBySessionId(ctx context.Context, sessionID string) (*fleet.CarveMetadata, error) { + return c.metadatadb.CarveBySessionId(ctx, sessionID) } // CarveByName returns carve metadata by name -func (d *Datastore) CarveByName(ctx context.Context, name string) (*fleet.CarveMetadata, error) { - return d.metadatadb.CarveByName(ctx, name) +func (c *CarveStore) CarveByName(ctx context.Context, name string) (*fleet.CarveMetadata, error) { + return c.metadatadb.CarveByName(ctx, name) } // ListCarves returns a list of the currently available carves -func (d *Datastore) ListCarves(ctx context.Context, opt fleet.CarveListOptions) ([]*fleet.CarveMetadata, error) { - return d.metadatadb.ListCarves(ctx, opt) +func (c *CarveStore) ListCarves(ctx context.Context, opt fleet.CarveListOptions) ([]*fleet.CarveMetadata, error) { + return c.metadatadb.ListCarves(ctx, opt) } // listCompletedParts returns a list of the parts in a multipart updaload given a key and uploadID // results are wrapped into the s3.CompletedPart struct -func (d *Datastore) listCompletedParts(objectKey, uploadID string) ([]*s3.CompletedPart, error) { +func (c *CarveStore) listCompletedParts(objectKey, uploadID string) ([]*s3.CompletedPart, error) { var res []*s3.CompletedPart var partMarker int64 for { - parts, err := d.s3client.ListParts(&s3.ListPartsInput{ - Bucket: &d.bucket, + parts, err := c.s3client.ListParts(&s3.ListPartsInput{ + Bucket: &c.bucket, Key: &objectKey, UploadId: &uploadID, PartNumberMarker: &partMarker, @@ -169,12 +187,12 @@ func (d *Datastore) listCompletedParts(objectKey, uploadID string) ([]*s3.Comple } // NewBlock uploads a new block for a specific carve -func (d *Datastore) NewBlock(ctx context.Context, metadata *fleet.CarveMetadata, blockID int64, data []byte) error { - objectKey := d.generateS3Key(metadata) +func (c *CarveStore) NewBlock(ctx context.Context, metadata *fleet.CarveMetadata, blockID int64, data []byte) error { + objectKey := c.generateS3Key(metadata) partNumber := blockID + 1 // PartNumber is 1-indexed - _, err := d.s3client.UploadPart(&s3.UploadPartInput{ + _, err := c.s3client.UploadPart(&s3.UploadPartInput{ Body: bytes.NewReader(data), - Bucket: &d.bucket, + Bucket: &c.bucket, Key: &objectKey, PartNumber: &partNumber, UploadId: &metadata.SessionId, @@ -184,18 +202,18 @@ func (d *Datastore) NewBlock(ctx context.Context, metadata *fleet.CarveMetadata, } if metadata.MaxBlock < blockID { metadata.MaxBlock = blockID - if err = d.UpdateCarve(ctx, metadata); err != nil { + if err = c.UpdateCarve(ctx, metadata); err != nil { return ctxerr.Wrap(ctx, err, "s3 multipart carve upload") } } if blockID >= metadata.BlockCount-1 { // The last block was reached, multipart upload can be completed - parts, err := d.listCompletedParts(objectKey, metadata.SessionId) + parts, err := c.listCompletedParts(objectKey, metadata.SessionId) if err != nil { return ctxerr.Wrap(ctx, err, "s3 multipart carve upload") } - _, err = d.s3client.CompleteMultipartUpload(&s3.CompleteMultipartUploadInput{ - Bucket: &d.bucket, + _, err = c.s3client.CompleteMultipartUpload(&s3.CompleteMultipartUploadInput{ + Bucket: &c.bucket, Key: &objectKey, UploadId: &metadata.SessionId, MultipartUpload: &s3.CompletedMultipartUpload{Parts: parts}, @@ -208,15 +226,15 @@ func (d *Datastore) NewBlock(ctx context.Context, metadata *fleet.CarveMetadata, } // GetBlock returns a block of data for a carve -func (d *Datastore) GetBlock(ctx context.Context, metadata *fleet.CarveMetadata, blockID int64) ([]byte, error) { - objectKey := d.generateS3Key(metadata) +func (c *CarveStore) GetBlock(ctx context.Context, metadata *fleet.CarveMetadata, blockID int64) ([]byte, error) { + objectKey := c.generateS3Key(metadata) // blockID is 0-indexed and sequential so can be perfectly used for evaluating ranges // range extremes are inclusive as for RFC-2616 (section 14.35) // no need to cap the rangeEnd to the carve size as S3 will do that by itself rangeStart := blockID * metadata.BlockSize rangeString := fmt.Sprintf("bytes=%d-%d", rangeStart, rangeStart+metadata.BlockSize-1) - res, err := d.s3client.GetObject(&s3.GetObjectInput{ - Bucket: &d.bucket, + res, err := c.s3client.GetObject(&s3.GetObjectInput{ + Bucket: &c.bucket, Key: &objectKey, Range: &rangeString, }) @@ -225,7 +243,7 @@ func (d *Datastore) GetBlock(ctx context.Context, metadata *fleet.CarveMetadata, if errors.As(err, &awsErr) && awsErr.Code() == s3.ErrCodeNoSuchKey { // The carve does not exists in S3, mark expired metadata.Expired = true - if updateErr := d.UpdateCarve(ctx, metadata); err != nil { + if updateErr := c.UpdateCarve(ctx, metadata); err != nil { err = ctxerr.Wrap(ctx, err, updateErr.Error()) } } diff --git a/server/datastore/s3/installer.go b/server/datastore/s3/installer.go new file mode 100644 index 0000000000..47c451cfac --- /dev/null +++ b/server/datastore/s3/installer.go @@ -0,0 +1,62 @@ +package s3 + +import ( + "context" + "fmt" + "io" + "path" + + "github.com/aws/aws-sdk-go/service/s3" + "github.com/fleetdm/fleet/v4/server/config" + "github.com/fleetdm/fleet/v4/server/contexts/ctxerr" + "github.com/fleetdm/fleet/v4/server/fleet" +) + +const ( + desktopPath = "desktop" + executable = "fleet-osquery" +) + +// InstallerStore contains methods to retrieve installers from S3 +type InstallerStore struct { + *s3store +} + +// NewInstallerStore creates a new instance with the given S3 config +func NewInstallerStore(config config.S3Config) (*InstallerStore, error) { + s3store, err := newS3store(config) + if err != nil { + return nil, err + } + return &InstallerStore{s3store}, nil +} + +// Exists checks if an installer exists in the S3 bucket +func (i *InstallerStore) Exists(ctx context.Context, installer fleet.Installer) (bool, error) { + key := i.keyForInstaller(installer) + _, err := i.s3client.HeadObject(&s3.HeadObjectInput{Bucket: &i.bucket, Key: &key}) + if err != nil { + return false, err + } + return true, nil +} + +// Get retrieves the requested installer from S3 +func (i *InstallerStore) Get(ctx context.Context, installer fleet.Installer) (io.ReadCloser, error) { + key := i.keyForInstaller(installer) + req, err := i.s3client.GetObject(&s3.GetObjectInput{Bucket: &i.bucket, Key: &key}) + if err != nil { + return nil, ctxerr.Wrap(ctx, err, "get installer from storage") + } + return req.Body, nil +} + +// keyForInstaller builds an S3 key to search for the installer +func (i *InstallerStore) keyForInstaller(installer fleet.Installer) string { + file := fmt.Sprintf("%s.%s", executable, installer.Kind) + dir := "" + if installer.Desktop { + dir = desktopPath + } + return path.Join(i.prefix, installer.EnrollSecret, dir, file) +} diff --git a/server/datastore/s3/installer_test.go b/server/datastore/s3/installer_test.go new file mode 100644 index 0000000000..fd75fa8119 --- /dev/null +++ b/server/datastore/s3/installer_test.go @@ -0,0 +1,82 @@ +package s3 + +import ( + "context" + "io" + "testing" + + "github.com/fleetdm/fleet/v4/server/fleet" + "github.com/stretchr/testify/require" +) + +func TestInstallerExists(t *testing.T) { + ctx := context.Background() + store := setupInstallerStore(t, "installers", "random-prefix") + + t.Run("returns true for existing installers", func(t *testing.T) { + installers := seedInstallerStore(t, store, "enroll-secret") + + for _, i := range installers { + exists, err := store.Exists(ctx, *i) + require.NoError(t, err) + require.Equal(t, exists, true) + } + }) + + t.Run("returns false for non-existing installers", func(t *testing.T) { + i := fleet.Installer{ + EnrollSecret: "non-existent", + Kind: "pkg", + Desktop: false, + } + exists, err := store.Exists(ctx, i) + require.Error(t, err) + require.Equal(t, exists, false) + + i = fleet.Installer{ + EnrollSecret: "non-existent", + Kind: "pkg", + Desktop: true, + } + exists, err = store.Exists(ctx, i) + require.Error(t, err) + require.Equal(t, exists, false) + }) +} + +func TestGetInstaller(t *testing.T) { + ctx := context.Background() + store := setupInstallerStore(t, "installers", "random-prefix") + + t.Run("gets a blob with the file contents for each installer", func(t *testing.T) { + installers := seedInstallerStore(t, store, "enroll-secret") + + for _, i := range installers { + blob, err := store.Get(ctx, *i) + require.NoError(t, err) + contents, err := io.ReadAll(blob) + require.NoError(t, err) + require.Equal(t, "mock", string(contents)) + } + }) + + t.Run("returns an error for non-existing installers", func(t *testing.T) { + i := fleet.Installer{ + EnrollSecret: "non-existent", + Kind: "pkg", + Desktop: false, + } + blob, err := store.Get(ctx, i) + require.Error(t, err) + require.Nil(t, blob) + + i = fleet.Installer{ + EnrollSecret: "non-existent", + Kind: "pkg", + Desktop: true, + } + blob, err = store.Get(ctx, i) + require.Error(t, err) + require.Nil(t, blob) + }) +} diff --git a/server/datastore/s3/s3.go b/server/datastore/s3/s3.go index 598e39ddd0..bfa8e7c7b2 100644 --- a/server/datastore/s3/s3.go +++ b/server/datastore/s3/s3.go @@ -11,22 +11,18 @@ import ( "github.com/aws/aws-sdk-go/service/s3" "github.com/aws/aws-sdk-go/service/s3/s3manager" "github.com/fleetdm/fleet/v4/server/config" - "github.com/fleetdm/fleet/v4/server/fleet" ) const awsRegionHint = "us-east-1" -// Datastore is a type implementing the CarveStore interface -// relying on AWS S3 storage -type Datastore struct { - metadatadb fleet.CarveStore - s3client *s3.S3 - bucket string - prefix string +type s3store struct { + s3client *s3.S3 + bucket string + prefix string } -// New initializes an S3 Datastore -func New(config config.S3Config, metadatadb fleet.CarveStore) (*Datastore, error) { +// newS3store initializes an S3 Datastore +func newS3store(config config.S3Config) (*s3store, error) { conf := &aws.Config{} // Use default auth provire if no static credentials were provided @@ -69,10 +65,9 @@ func New(config config.S3Config, metadatadb fleet.CarveStore) (*Datastore, error config.Region = region } - return &Datastore{ - metadatadb: metadatadb, - s3client: s3.New(sess, &aws.Config{Region: &config.Region}), - bucket: config.Bucket, - prefix: config.Prefix, + return &s3store{ + s3client: s3.New(sess, &aws.Config{Region: &config.Region}), + bucket: config.Bucket, + prefix: config.Prefix, }, nil } diff --git a/server/datastore/s3/testing_utils.go b/server/datastore/s3/testing_utils.go new file mode 100644 index 0000000000..994d05fad6 --- /dev/null +++ b/server/datastore/s3/testing_utils.go @@ -0,0 +1,107 @@ +package s3 + +import ( + "os" + "strings" + "testing" + + "github.com/aws/aws-sdk-go/aws" + "github.com/aws/aws-sdk-go/service/s3" + "github.com/fleetdm/fleet/v4/server/config" + "github.com/fleetdm/fleet/v4/server/fleet" + "github.com/stretchr/testify/require" +) + +const ( + accessKeyID = "minio" + secretAccessKey = "minio123!" + testEndpoint = "localhost:9000" + mockInstallerContents = "mock" +) + +func setupInstallerStore(tb testing.TB, bucket, prefix string) *InstallerStore { + checkEnv(tb) + + store, err := NewInstallerStore(config.S3Config{ + Bucket: bucket, + Prefix: prefix, + Region: "minio", + EndpointURL: testEndpoint, + AccessKeyID: accessKeyID, + SecretAccessKey: secretAccessKey, + ForceS3PathStyle: true, + DisableSSL: true, + }) + require.Nil(tb, err) + + store.s3client.CreateBucket(&s3.CreateBucketInput{ + Bucket: &bucket, + CreateBucketConfiguration: &s3.CreateBucketConfiguration{}, + }) + + tb.Cleanup(func() { cleanupStore(tb, store) }) + + return store +} + +func seedInstallerStore(tb testing.TB, store *InstallerStore, enrollSecret string) []*fleet.Installer { + checkEnv(tb) + installers := []*fleet.Installer{ + {EnrollSecret: enrollSecret, Kind: "pkg", Desktop: false}, + {EnrollSecret: enrollSecret, Kind: "msi", Desktop: false}, + {EnrollSecret: enrollSecret, Kind: "deb", Desktop: false}, + {EnrollSecret: enrollSecret, Kind: "rpm", Desktop: false}, + {EnrollSecret: enrollSecret, Kind: "pkg", Desktop: true}, + {EnrollSecret: enrollSecret, Kind: "msi", Desktop: true}, + {EnrollSecret: enrollSecret, Kind: "deb", Desktop: true}, + {EnrollSecret: enrollSecret, Kind: "rpm", Desktop: true}, + } + + for _, i := range installers { + uploadMockInstaller(tb, store, i) + } + + return installers +} + +func uploadMockInstaller(tb testing.TB, store *InstallerStore, installer *fleet.Installer) { + checkEnv(tb) + _, err := store.s3client.PutObject(&s3.PutObjectInput{ + Bucket: &store.bucket, + Body: aws.ReadSeekCloser(strings.NewReader(mockInstallerContents)), + Key: aws.String(store.keyForInstaller(*installer)), + }) + require.NoError(tb, err) +} + +func cleanupStore(tb testing.TB, store *InstallerStore) { + checkEnv(tb) + resp, err := store.s3client.ListObjects(&s3.ListObjectsInput{ + Bucket: &store.bucket, + }) + require.NoError(tb, err) + + var objs []*s3.ObjectIdentifier + for _, o := range resp.Contents { + objs = append(objs, &s3.ObjectIdentifier{Key: o.Key}) + } + _, err = store.s3client.DeleteObjects(&s3.DeleteObjectsInput{ + Bucket: &store.bucket, + Delete: &s3.Delete{ + Objects: objs, + }, + }) + require.NoError(tb, err) + + _, err = store.s3client.DeleteBucket(&s3.DeleteBucketInput{ + Bucket: &store.bucket, + }) + require.NoError(tb, err) +} + +func checkEnv(tb testing.TB) { + if _, ok := os.LookupEnv("MINIO_STORAGE_TEST"); !ok { + tb.Skip("set MINIO_STORAGE_TEST environment variable to run S3-based tests") + + } +} diff --git a/server/fleet/installer.go b/server/fleet/installer.go new file mode 100644 index 0000000000..303f86455b --- /dev/null +++ b/server/fleet/installer.go @@ -0,0 +1,8 @@ +package fleet + +// Installer describes an installer in an S3 bucket +type Installer struct { + EnrollSecret string + Kind string + Desktop bool +}