1622978334

Committed 13 Jan 2025 03:51PM UTC coverage: 72.802% (-3.8%) from 76.608%

Build # 1622978334

Build Type

Pull #300

gitlab-ci

Committed by

alfrunes

Commit Message

fix: Deployment device count should not exceed max devices

Added a condition to skip deployments when the device count reaches max
devices.

Changelog: Title
Ticket: MEN-7847
Signed-off-by: Alf-Rune Siqveland <alf.rune@northern.tech>

Pull Request Pull Request #300: fix: Deployment device count should not exceed max devices

Run Details

4251 of 6164 branches covered (68.96%)

Branch coverage included in aggregate %.

0 of 18 new or added lines in 1 file covered. (0.0%)

2544 existing lines in 83 files now uncovered.

42741 of 58384 relevant lines covered (73.21%)

21.49 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

26.38

/backend/services/deployments/storage/s3/s3.go

// Copyright 2023 Northern.tech AS
//
//    Licensed under the Apache License, Version 2.0 (the "License");
//    you may not use this file except in compliance with the License.
//    You may obtain a copy of the License at
//
//        http://www.apache.org/licenses/LICENSE-2.0
//
//    Unless required by applicable law or agreed to in writing, software
//    distributed under the License is distributed on an "AS IS" BASIS,
//    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//    See the License for the specific language governing permissions and
//    limitations under the License.

package s3

import (
        "bytes"
        "context"
        stderr "errors"
        "fmt"
        "io"
        "net/http"
        "net/url"
        "time"

        "github.com/aws/aws-sdk-go-v2/aws"
        v4 "github.com/aws/aws-sdk-go-v2/aws/signer/v4"
        awsHttp "github.com/aws/aws-sdk-go-v2/aws/transport/http"
        awsConfig "github.com/aws/aws-sdk-go-v2/config"
        "github.com/aws/aws-sdk-go-v2/service/s3"
        "github.com/aws/aws-sdk-go-v2/service/s3/types"
        validation "github.com/go-ozzo/ozzo-validation/v4"
        "github.com/pkg/errors"

        "github.com/mendersoftware/mender-server/services/deployments/model"
        "github.com/mendersoftware/mender-server/services/deployments/storage"
        "github.com/mendersoftware/mender-server/services/deployments/utils"
)

const (
        ExpireMaxLimit = 7 * 24 * time.Hour
        ExpireMinLimit = 1 * time.Minute

        MultipartMaxParts = 10000
        MultipartMinSize  = 5 * mib

        // Constants not exposed by aws-sdk-go
        // from /aws/signer/v4/internal/v4
        paramAmzDate       = "X-Amz-Date"
        paramAmzDateFormat = "20060102T150405Z"
)

var ErrClientEmpty = stderr.New("s3: storage client credentials not configured")

// SimpleStorageService - AWS S3 client.
// Data layer for file storage.
// Implements model.FileStorage interface
type SimpleStorageService struct {
        client        *s3.Client
        presignClient *s3.PresignClient
        settings      storageSettings
        bufferSize    int
        contentType   *string
}

type StaticCredentials struct {
        Key    string `json:"key"`
        Secret string `json:"secret"`
        Token  string `json:"token"`
}

func (creds StaticCredentials) Validate() error {
        return validation.ValidateStruct(&creds,
                validation.Field(&creds.Key, validation.Required),
                validation.Field(&creds.Secret, validation.Required),
        )
}

func (creds StaticCredentials) awsCredentials() aws.Credentials {
        return aws.Credentials{
                AccessKeyID:     creds.Key,
                SecretAccessKey: creds.Secret,
                SessionToken:    creds.Token,
                Source:          "mender:StaticCredentials",
        }
}

func (creds StaticCredentials) Retrieve(context.Context) (aws.Credentials, error) {
        return creds.awsCredentials(), nil
}

func newClient(
        ctx context.Context,
        withCredentials bool,
        opt *Options,
) (*SimpleStorageService, error) {
        if err := opt.Validate(); err != nil {
                return nil, errors.WithMessage(err, "s3: invalid configuration")
        }
        var (
                err error
                cfg aws.Config
        )

        if withCredentials {
                cfg, err = awsConfig.LoadDefaultConfig(ctx)
        } else {
                opt.StaticCredentials = nil
                cfg, err = awsConfig.LoadDefaultConfig(ctx,
                        awsConfig.WithCredentialsProvider(aws.AnonymousCredentials{}),
                )
        }
        if err != nil {
                return nil, err
        }

        clientOpts, presignOpts := opt.toS3Options()
        client := s3.NewFromConfig(cfg, clientOpts)
        presignClient := s3.NewPresignClient(client, presignOpts)

        return &SimpleStorageService{
                client:        client,
                presignClient: presignClient,

                bufferSize:  *opt.BufferSize,
                contentType: opt.ContentType,
                settings:    opt.storageSettings,
        }, nil
}

// NewEmpty initializes a new s3 client that does not implicitly load
// credentials from the environment. Credentials must be set using the
// StorageSettings provided with the Context.
func NewEmpty(ctx context.Context, opts ...*Options) (storage.ObjectStorage, error) {
        opt := NewOptions(opts...)
        return newClient(ctx, false, opt)
}

func New(ctx context.Context, opts ...*Options) (storage.ObjectStorage, error) {
        opt := NewOptions(opts...)

        s3c, err := newClient(ctx, true, opt)
        if err != nil {
                return nil, err
        }

        err = s3c.init(ctx)
        if err != nil {
                return nil, errors.WithMessage(err, "s3: failed to check bucket preconditions")
        }
        return s3c, nil
}

func disableAccelerate(opts *s3.Options) {
        opts.UseAccelerate = false
}

func (s *SimpleStorageService) init(ctx context.Context) error {
        if s.settings.BucketName == nil {
                return errors.New("s3: failed to initalize storage client: " +
                        "a bucket name is required")
        }
        hparams := &s3.HeadBucketInput{
                Bucket: s.settings.BucketName,
        }
        var rspErr *awsHttp.ResponseError

        _, err := s.client.HeadBucket(ctx, hparams)
        if err == nil {
                // bucket exists and have permission to access it
                return nil
        } else if errors.As(err, &rspErr) {
                switch rspErr.Response.StatusCode {
                case http.StatusNotFound:
                        err = nil // pass
                case http.StatusForbidden:
                        err = fmt.Errorf(
                                "s3: insufficient permissions for accessing bucket '%s'",
                                *s.settings.BucketName,
                        )
                }
        }
        if err != nil {
                return err
        }
        cparams := &s3.CreateBucketInput{
                Bucket: s.settings.BucketName,
        }

        _, err = s.client.CreateBucket(ctx, cparams, disableAccelerate)
        if err != nil {
                var errBucket *types.BucketAlreadyOwnedByYou
                if !errors.As(err, errBucket) {
                        return errors.WithMessage(err, "s3: error creating bucket")
                }
        }
        waitTime := time.Second * 30
        if deadline, ok := ctx.Deadline(); ok {
                waitTime = time.Until(deadline)
        }
        err = s3.NewBucketExistsWaiter(s.client).
                Wait(ctx, hparams, waitTime)
        return err
}

func (s *SimpleStorageService) optionsFromContext(
        ctx context.Context,
) (settings *storageSettings, err error) {
        ss, ok := storage.SettingsFromContext(ctx)
        if ok && ss != nil {
                err = ss.Validate()
                if err == nil {
                        settings = newFromParent(&s.settings, ss)
                }
        } else {
                settings = &s.settings
                if settings.BucketName == nil {
                        err = ErrClientEmpty
                }
        }
        return settings, err
}

func (s *SimpleStorageService) HealthCheck(ctx context.Context) error {
        opts, err := s.optionsFromContext(ctx)
        if err != nil {
                return err
        }
        _, err = s.client.HeadBucket(ctx, &s3.HeadBucketInput{
                Bucket: opts.BucketName,
        }, opts.options)
        return err
}

type objectReader struct {
        io.ReadCloser
        length int64
}

func (obj objectReader) Length() int64 {
        return obj.length
}

func (s *SimpleStorageService) GetObject(
        ctx context.Context,
        path string,
) (io.ReadCloser, error) {
        opts, err := s.optionsFromContext(ctx)
        if err != nil {
                return nil, err
        }
        params := &s3.GetObjectInput{
                Bucket: opts.BucketName,
                Key:    aws.String(path),

                RequestPayer: types.RequestPayerRequester,
        }

        out, err := s.client.GetObject(ctx, params, opts.options)
        var rspErr *awsHttp.ResponseError
        if errors.As(err, &rspErr) {
                if rspErr.Response.StatusCode == http.StatusNotFound {
                        err = storage.ErrObjectNotFound
                }
        }
        if err != nil {
                return nil, errors.WithMessage(
                        err,
                        "s3: failed to get object",
                )
        }
        return objectReader{
                ReadCloser: out.Body,
                length:     *out.ContentLength,
        }, nil
}

// Delete removes deleted file from storage.
// Noop if ID does not exist.
func (s *SimpleStorageService) DeleteObject(ctx context.Context, path string) error {
        opts, err := s.optionsFromContext(ctx)
        if err != nil {
                return err
        }

        params := &s3.DeleteObjectInput{
                // Required
                Bucket: opts.BucketName,
                Key:    aws.String(path),

                // Optional
                RequestPayer: types.RequestPayerRequester,
        }

        // ignore return response which contains charing info
        // and file versioning data which are not in interest
        _, err = s.client.DeleteObject(ctx, params, opts.options)
        var rspErr *awsHttp.ResponseError
        if errors.As(err, &rspErr) {
                if rspErr.Response.StatusCode == http.StatusNotFound {
                        err = storage.ErrObjectNotFound
                }
        }
        if err != nil {
                return errors.WithMessage(err, "s3: error deleting object")
        }

        return nil
}

// Exists check if selected object exists in the storage
func (s *SimpleStorageService) StatObject(
        ctx context.Context,
        path string,
) (*storage.ObjectInfo, error) {

        opts, err := s.optionsFromContext(ctx)
        if err != nil {
                return nil, err
        }

        params := &s3.HeadObjectInput{
                Bucket: opts.BucketName,
                Key:    aws.String(path),
        }
        rsp, err := s.client.HeadObject(ctx, params, opts.options)
        var rspErr *awsHttp.ResponseError
        if errors.As(err, &rspErr) {
                if rspErr.Response.StatusCode == http.StatusNotFound {
                        err = storage.ErrObjectNotFound
                }
        }
        if err != nil {
                return nil, errors.WithMessage(err, "s3: error getting object info")
        }

        return &storage.ObjectInfo{
                Path:         path,
                LastModified: rsp.LastModified,
                Size:         rsp.ContentLength,
        }, nil
}

func fillBuffer(b []byte, r io.Reader) (int, error) {
        var offset int
        var err error
        for n := 0; offset < len(b) && err == nil; offset += n {
                n, err = r.Read(b[offset:])
        }
        return offset, err
}

// uploadMultipart uploads an artifact using the multipart API.
func (s *SimpleStorageService) uploadMultipart(
        ctx context.Context,
        buf []byte,
        objectPath string,
        artifact io.Reader,
) error {
        const maxPartNum = 10000
        var partNum int32 = 1
        var rspUpload *s3.UploadPartOutput
        opts, err := s.optionsFromContext(ctx)
        if err != nil {
                return err
        }

        // Pre-allocate 100 completed part (generous guesstimate)
        completedParts := make([]types.CompletedPart, 0, 100)

        // Initiate Multipart upload
        createParams := &s3.CreateMultipartUploadInput{
                Bucket:      opts.BucketName,
                Key:         &objectPath,
                ContentType: s.contentType,
        }
        rspCreate, err := s.client.CreateMultipartUpload(
                ctx, createParams, opts.options,
        )
        if err != nil {
                return err
        }
        uploadParams := &s3.UploadPartInput{
                Bucket:     opts.BucketName,
                Key:        &objectPath,
                UploadId:   rspCreate.UploadId,
                PartNumber: aws.Int32(partNum),
        }

        // Upload the first chunk already stored in buffer
        r := bytes.NewReader(buf)
        // Readjust upload parameters
        uploadParams.Body = r
        rspUpload, err = s.client.UploadPart(
                ctx,
                uploadParams,
                opts.options,
        )
        if err != nil {
                return err
        }
        completedParts = append(
                completedParts,
                types.CompletedPart{
                        ETag:       rspUpload.ETag,
                        PartNumber: aws.Int32(partNum),
                },
        )

        // The following is loop is very similar to io.Copy except the
        // destination is the s3 bucket.
        for partNum++; partNum < maxPartNum; partNum++ {
                // Read next chunk from stream (fill the whole buffer)
                offset, eRead := fillBuffer(buf, artifact)
                if offset > 0 {
                        r := bytes.NewReader(buf[:offset])
                        // Readjust upload parameters
                        uploadParams.PartNumber = aws.Int32(partNum)
                        uploadParams.Body = r
                        rspUpload, err = s.client.UploadPart(
                                ctx,
                                uploadParams,
                                opts.options,
                        )
                        if err != nil {
                                break
                        }
                        completedParts = append(
                                completedParts,
                                types.CompletedPart{
                                        ETag:       rspUpload.ETag,
                                        PartNumber: aws.Int32(partNum),
                                },
                        )
                } else {
                        // Read did not return any bytes
                        break
                }
                if eRead != nil {
                        err = eRead
                        break
                }
        }
        if err == nil || err == io.EOF {
                // Complete upload
                uploadParams := &s3.CompleteMultipartUploadInput{
                        Bucket:   opts.BucketName,
                        Key:      &objectPath,
                        UploadId: rspCreate.UploadId,
                        MultipartUpload: &types.CompletedMultipartUpload{
                                Parts: completedParts,
                        },
                }
                _, err = s.client.CompleteMultipartUpload(
                        ctx,
                        uploadParams,
                        opts.options,
                )
        } else {
                // Abort multipart upload!
                uploadParams := &s3.AbortMultipartUploadInput{
                        Bucket:   opts.BucketName,
                        Key:      &objectPath,
                        UploadId: rspCreate.UploadId,
                }
                _, _ = s.client.AbortMultipartUpload(
                        ctx,
                        uploadParams,
                        opts.options,
                )
        }
        return err
}

// UploadArtifact uploads given artifact into the file server (AWS S3 or minio)
// using objectID as a key. If the artifact is larger than 5 MiB, the file is
// uploaded using the s3 multipart API, otherwise the object is created in a
// single request.
func (s *SimpleStorageService) PutObject(
        ctx context.Context,
        path string,
        src io.Reader,
) error {
        var (
                r   io.Reader
                l   int64
                n   int
                err error
                buf []byte
        )
        if objReader, ok := src.(storage.ObjectReader); ok {
                r = objReader
                l = objReader.Length()
        } else {
                // Peek payload
                buf = make([]byte, s.bufferSize)
                n, err = fillBuffer(buf, src)
                if err == io.EOF {
                        r = bytes.NewReader(buf[:n])
                        l = int64(n)
                }
        }

        // If only one part, use PutObject API.
        if r != nil {
                var opts *storageSettings
                opts, err = s.optionsFromContext(ctx)
                if err != nil {
                        return err
                }
                // Ordinary single-file upload
                uploadParams := &s3.PutObjectInput{
                        Body:          r,
                        Bucket:        opts.BucketName,
                        Key:           &path,
                        ContentType:   s.contentType,
                        ContentLength: &l,
                }
                _, err = s.client.PutObject(
                        ctx,
                        uploadParams,
                        opts.options,
                )
        } else if err == nil {
                err = s.uploadMultipart(ctx, buf, path, src)
        }
        return err
}

func buildLink(
        req *v4.PresignedHTTPRequest,
        signDate time.Time,
        expireAfter time.Duration,
        proxyURL *url.URL,
) (*model.Link, error) {
        signURL, err := url.Parse(req.URL)
        if err != nil {
                return nil, fmt.Errorf("s3: failed to parse signed URL: %w", err)
        }
        if date, err := time.Parse(
                req.SignedHeader.Get(paramAmzDate), paramAmzDateFormat,
        ); err == nil {
                signDate = date
        }
        signURL, err = utils.RewriteProxyURL(signURL, proxyURL)
        if err != nil {
                return nil, fmt.Errorf("s3: failed to rewrite signed URL to proxy: %w", err)
        }

        return &model.Link{
                Uri:    signURL.String(),
                Expire: signDate.Add(expireAfter),
                Method: req.Method,
        }, nil
}

func (s *SimpleStorageService) PutRequest(
        ctx context.Context,
        path string,
        expireAfter time.Duration,
) (*model.Link, error) {

        expireAfter = capDurationToLimits(expireAfter).Truncate(time.Second)
        opts, err := s.optionsFromContext(ctx)
        if err != nil {
                return nil, err
        }

        params := &s3.PutObjectInput{
                // Required
                Bucket: opts.BucketName,
                Key:    aws.String(path),
        }

        signDate := time.Now()
        req, err := s.presignClient.PresignPutObject(
                ctx,
                params,
                opts.presignOptions,
                s3.WithPresignExpires(expireAfter),
        )
        if err != nil {
                return nil, err
        }
        return buildLink(req, signDate, expireAfter, opts.ProxyURI)
}

// GetRequest duration is limited to 7 days (AWS limitation)
func (s *SimpleStorageService) GetRequest(
        ctx context.Context,
        objectPath string,
        filename string,
        expireAfter time.Duration,
) (*model.Link, error) {

        expireAfter = capDurationToLimits(expireAfter).Truncate(time.Second)
        opts, err := s.optionsFromContext(ctx)
        if err != nil {
                return nil, err
        }

        if _, err := s.StatObject(ctx, objectPath); err != nil {
                return nil, errors.WithMessage(err, "s3: head object")
        }

        params := &s3.GetObjectInput{
                Bucket:              opts.BucketName,
                Key:                 aws.String(objectPath),
                ResponseContentType: s.contentType,
        }

        if filename != "" {
                contentDisposition := fmt.Sprintf("attachment; filename=\"%s\"", filename)
                params.ResponseContentDisposition = &contentDisposition
        }

        signDate := time.Now()
        req, err := s.presignClient.PresignGetObject(ctx,
                params,
                opts.presignOptions,
                s3.WithPresignExpires(expireAfter))
        if err != nil {
                return nil, errors.WithMessage(err, "s3: failed to sign GET request")
        }
        return buildLink(req, signDate, expireAfter, opts.ProxyURI)
}

// DeleteRequest returns a presigned deletion request
func (s *SimpleStorageService) DeleteRequest(
        ctx context.Context,
        path string,
        expireAfter time.Duration,
) (*model.Link, error) {

        expireAfter = capDurationToLimits(expireAfter).Truncate(time.Second)
        opts, err := s.optionsFromContext(ctx)
        if err != nil {
                return nil, err
        }

        params := &s3.DeleteObjectInput{
                Bucket: opts.BucketName,
                Key:    aws.String(path),
        }

        signDate := time.Now()
        req, err := s.presignClient.PresignDeleteObject(ctx,
                params,
                opts.presignOptions,
                s3.WithPresignExpires(expireAfter))
        if err != nil {
                return nil, errors.WithMessage(err, "s3: failed to sign DELETE request")
        }
        return buildLink(req, signDate, expireAfter, opts.ProxyURI)
}

// presign requests are limited to 7 days
func capDurationToLimits(duration time.Duration) time.Duration {
        if duration < ExpireMinLimit {
                duration = ExpireMinLimit
        } else if duration > ExpireMaxLimit {
                duration = ExpireMaxLimit
        }
        return duration
}

1	// Copyright 2023 Northern.tech AS
2	//
3	// Licensed under the Apache License, Version 2.0 (the "License");
4	// you may not use this file except in compliance with the License.
5	// You may obtain a copy of the License at
6	//
7	// http://www.apache.org/licenses/LICENSE-2.0
8	//
9	// Unless required by applicable law or agreed to in writing, software
10	// distributed under the License is distributed on an "AS IS" BASIS,
11	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12	// See the License for the specific language governing permissions and
13	// limitations under the License.
14
15	package s3
16
17	import (
18	"bytes"
19	"context"
20	stderr "errors"
21	"fmt"
22	"io"
23	"net/http"
24	"net/url"
25	"time"
26
27	"github.com/aws/aws-sdk-go-v2/aws"
28	v4 "github.com/aws/aws-sdk-go-v2/aws/signer/v4"
29	awsHttp "github.com/aws/aws-sdk-go-v2/aws/transport/http"
30	awsConfig "github.com/aws/aws-sdk-go-v2/config"
31	"github.com/aws/aws-sdk-go-v2/service/s3"
32	"github.com/aws/aws-sdk-go-v2/service/s3/types"
33	validation "github.com/go-ozzo/ozzo-validation/v4"
34	"github.com/pkg/errors"
35
36	"github.com/mendersoftware/mender-server/services/deployments/model"
37	"github.com/mendersoftware/mender-server/services/deployments/storage"
38	"github.com/mendersoftware/mender-server/services/deployments/utils"
39	)
40
41	const (
42	ExpireMaxLimit = 7 * 24 * time.Hour
43	ExpireMinLimit = 1 * time.Minute
44
45	MultipartMaxParts = 10000
46	MultipartMinSize = 5 * mib
47
48	// Constants not exposed by aws-sdk-go
49	// from /aws/signer/v4/internal/v4
50	paramAmzDate = "X-Amz-Date"
51	paramAmzDateFormat = "20060102T150405Z"
52	)
53
54	var ErrClientEmpty = stderr.New("s3: storage client credentials not configured")
55
56	// SimpleStorageService - AWS S3 client.
57	// Data layer for file storage.
58	// Implements model.FileStorage interface
59	type SimpleStorageService struct {
60	client *s3.Client
61	presignClient *s3.PresignClient
62	settings storageSettings
63	bufferSize int
64	contentType *string
65	}
66
67	type StaticCredentials struct {
68	Key string `json:"key"`
69	Secret string `json:"secret"`
70	Token string `json:"token"`
71	}
72
73	func (creds StaticCredentials) Validate() error {	1✔
74	return validation.ValidateStruct(&creds,	1✔
75	validation.Field(&creds.Key, validation.Required),	1✔
76	validation.Field(&creds.Secret, validation.Required),	1✔
77	)	1✔
78	}	1✔
79
80	func (creds StaticCredentials) awsCredentials() aws.Credentials {	1✔
81	return aws.Credentials{	1✔
82	AccessKeyID: creds.Key,	1✔
83	SecretAccessKey: creds.Secret,	1✔
84	SessionToken: creds.Token,	1✔
85	Source: "mender:StaticCredentials",	1✔
86	}	1✔
87	}	1✔
88
89	func (creds StaticCredentials) Retrieve(context.Context) (aws.Credentials, error) {	1✔
90	return creds.awsCredentials(), nil	1✔
91	}	1✔
92
93	func newClient(
94	ctx context.Context,
95	withCredentials bool,
96	opt *Options,
97	) (*SimpleStorageService, error) {	1✔
98	if err := opt.Validate(); err != nil {	1✔
99	return nil, errors.WithMessage(err, "s3: invalid configuration")	×
100	}	×
101	var (	1✔
102	err error	1✔
103	cfg aws.Config	1✔
104	)	1✔
105		1✔
106	if withCredentials {	2✔
107	cfg, err = awsConfig.LoadDefaultConfig(ctx)	1✔
108	} else {	1✔
UNCOV 109	opt.StaticCredentials = nil	×
UNCOV 110	cfg, err = awsConfig.LoadDefaultConfig(ctx,	×
UNCOV 111	awsConfig.WithCredentialsProvider(aws.AnonymousCredentials{}),	×
UNCOV 112	)	×
UNCOV 113	}	×
114	if err != nil {	1✔
115	return nil, err	×
116	}	×
117
118	clientOpts, presignOpts := opt.toS3Options()	1✔
119	client := s3.NewFromConfig(cfg, clientOpts)	1✔
120	presignClient := s3.NewPresignClient(client, presignOpts)	1✔
121		1✔
122	return &SimpleStorageService{	1✔
123	client: client,	1✔
124	presignClient: presignClient,	1✔
125		1✔
126	bufferSize: *opt.BufferSize,	1✔
127	contentType: opt.ContentType,	1✔
128	settings: opt.storageSettings,	1✔
129	}, nil	1✔
130	}
131
132	// NewEmpty initializes a new s3 client that does not implicitly load
133	// credentials from the environment. Credentials must be set using the
134	// StorageSettings provided with the Context.
UNCOV 135	func NewEmpty(ctx context.Context, opts ...*Options) (storage.ObjectStorage, error) {	×
UNCOV 136	opt := NewOptions(opts...)	×
UNCOV 137	return newClient(ctx, false, opt)	×
UNCOV 138	}	×
139
140	func New(ctx context.Context, opts ...*Options) (storage.ObjectStorage, error) {	1✔
141	opt := NewOptions(opts...)	1✔
142		1✔
143	s3c, err := newClient(ctx, true, opt)	1✔
144	if err != nil {	1✔
145	return nil, err	×
146	}	×
147
148	err = s3c.init(ctx)	1✔
149	if err != nil {	1✔
150	return nil, errors.WithMessage(err, "s3: failed to check bucket preconditions")	×
151	}	×
152	return s3c, nil	1✔
153	}
154
155	func disableAccelerate(opts *s3.Options) {	1✔
156	opts.UseAccelerate = false	1✔
157	}	1✔
158
159	func (s *SimpleStorageService) init(ctx context.Context) error {	1✔
160	if s.settings.BucketName == nil {	1✔
161	return errors.New("s3: failed to initalize storage client: " +	×
162	"a bucket name is required")	×
163	}	×
164	hparams := &s3.HeadBucketInput{	1✔
165	Bucket: s.settings.BucketName,	1✔
166	}	1✔
167	var rspErr *awsHttp.ResponseError	1✔
168		1✔
169	_, err := s.client.HeadBucket(ctx, hparams)	1✔
170	if err == nil {	2✔
171	// bucket exists and have permission to access it	1✔
172	return nil	1✔
173	} else if errors.As(err, &rspErr) {	3✔
174	switch rspErr.Response.StatusCode {	1✔
175	case http.StatusNotFound:	1✔
176	err = nil // pass	1✔
177	case http.StatusForbidden:	×
178	err = fmt.Errorf(	×
179	"s3: insufficient permissions for accessing bucket '%s'",	×
180	*s.settings.BucketName,	×
181	)	×
182	}
183	}
184	if err != nil {	1✔
185	return err	×
186	}	×
187	cparams := &s3.CreateBucketInput{	1✔
188	Bucket: s.settings.BucketName,	1✔
189	}	1✔
190		1✔
191	_, err = s.client.CreateBucket(ctx, cparams, disableAccelerate)	1✔
192	if err != nil {	1✔
193	var errBucket *types.BucketAlreadyOwnedByYou	×
194	if !errors.As(err, errBucket) {	×
195	return errors.WithMessage(err, "s3: error creating bucket")	×
196	}	×
197	}
198	waitTime := time.Second * 30	1✔
199	if deadline, ok := ctx.Deadline(); ok {	2✔
200	waitTime = time.Until(deadline)	1✔
201	}	1✔
202	err = s3.NewBucketExistsWaiter(s.client).	1✔
203	Wait(ctx, hparams, waitTime)	1✔
204	return err	1✔
205	}
206
207	func (s *SimpleStorageService) optionsFromContext(
208	ctx context.Context,
209	) (settings *storageSettings, err error) {	1✔
210	ss, ok := storage.SettingsFromContext(ctx)	1✔
211	if ok && ss != nil {	2✔
212	err = ss.Validate()	1✔
213	if err == nil {	1✔
UNCOV 214	settings = newFromParent(&s.settings, ss)	×
UNCOV 215	}	×
216	} else {	1✔
217	settings = &s.settings	1✔
218	if settings.BucketName == nil {	1✔
219	err = ErrClientEmpty	×
220	}	×
221	}
222	return settings, err	1✔
223	}
224
225	func (s *SimpleStorageService) HealthCheck(ctx context.Context) error {	1✔
226	opts, err := s.optionsFromContext(ctx)	1✔
227	if err != nil {	1✔
228	return err	×
229	}	×
230	_, err = s.client.HeadBucket(ctx, &s3.HeadBucketInput{	1✔
231	Bucket: opts.BucketName,	1✔
232	}, opts.options)	1✔
233	return err	1✔
234	}
235
236	type objectReader struct {
237	io.ReadCloser
238	length int64
239	}
240
241	func (obj objectReader) Length() int64 {	×
242	return obj.length	×
243	}	×
244
245	func (s *SimpleStorageService) GetObject(
246	ctx context.Context,
247	path string,
248	) (io.ReadCloser, error) {	1✔
249	opts, err := s.optionsFromContext(ctx)	1✔
250	if err != nil {	2✔
251	return nil, err	1✔
252	}	1✔
253	params := &s3.GetObjectInput{	1✔
254	Bucket: opts.BucketName,	1✔
255	Key: aws.String(path),	1✔
256		1✔
257	RequestPayer: types.RequestPayerRequester,	1✔
258	}	1✔
259		1✔
260	out, err := s.client.GetObject(ctx, params, opts.options)	1✔
261	var rspErr *awsHttp.ResponseError	1✔
262	if errors.As(err, &rspErr) {	2✔
263	if rspErr.Response.StatusCode == http.StatusNotFound {	2✔
264	err = storage.ErrObjectNotFound	1✔
265	}	1✔
266	}
267	if err != nil {	2✔
268	return nil, errors.WithMessage(	1✔
269	err,	1✔
270	"s3: failed to get object",	1✔
271	)	1✔
272	}	1✔
273	return objectReader{	1✔
274	ReadCloser: out.Body,	1✔
275	length: *out.ContentLength,	1✔
276	}, nil	1✔
277	}
278
279	// Delete removes deleted file from storage.
280	// Noop if ID does not exist.
UNCOV 281	func (s *SimpleStorageService) DeleteObject(ctx context.Context, path string) error {	×
UNCOV 282	opts, err := s.optionsFromContext(ctx)	×
UNCOV 283	if err != nil {	×
284	return err	×
285	}	×
286
UNCOV 287	params := &s3.DeleteObjectInput{	×
UNCOV 288	// Required	×
UNCOV 289	Bucket: opts.BucketName,	×
UNCOV 290	Key: aws.String(path),	×
UNCOV 291		×
UNCOV 292	// Optional	×
UNCOV 293	RequestPayer: types.RequestPayerRequester,	×
UNCOV 294	}	×
UNCOV 295		×
UNCOV 296	// ignore return response which contains charing info	×
UNCOV 297	// and file versioning data which are not in interest	×
UNCOV 298	_, err = s.client.DeleteObject(ctx, params, opts.options)	×
UNCOV 299	var rspErr *awsHttp.ResponseError	×
UNCOV 300	if errors.As(err, &rspErr) {	×
301	if rspErr.Response.StatusCode == http.StatusNotFound {	×
302	err = storage.ErrObjectNotFound	×
303	}	×
304	}
UNCOV 305	if err != nil {	×
306	return errors.WithMessage(err, "s3: error deleting object")	×
307	}	×
308
UNCOV 309	return nil	×
310	}
311
312	// Exists check if selected object exists in the storage
313	func (s *SimpleStorageService) StatObject(
314	ctx context.Context,
315	path string,
UNCOV 316	) (*storage.ObjectInfo, error) {	×
UNCOV 317		×
UNCOV 318	opts, err := s.optionsFromContext(ctx)	×
UNCOV 319	if err != nil {	×
320	return nil, err	×
321	}	×
322
UNCOV 323	params := &s3.HeadObjectInput{	×
UNCOV 324	Bucket: opts.BucketName,	×
UNCOV 325	Key: aws.String(path),	×
UNCOV 326	}	×
UNCOV 327	rsp, err := s.client.HeadObject(ctx, params, opts.options)	×
UNCOV 328	var rspErr *awsHttp.ResponseError	×
UNCOV 329	if errors.As(err, &rspErr) {	×
330	if rspErr.Response.StatusCode == http.StatusNotFound {	×
331	err = storage.ErrObjectNotFound	×
332	}	×
333	}
UNCOV 334	if err != nil {	×
335	return nil, errors.WithMessage(err, "s3: error getting object info")	×
336	}	×
337
UNCOV 338	return &storage.ObjectInfo{	×
UNCOV 339	Path: path,	×
UNCOV 340	LastModified: rsp.LastModified,	×
UNCOV 341	Size: rsp.ContentLength,	×
UNCOV 342	}, nil	×
343	}
344
UNCOV 345	func fillBuffer(b []byte, r io.Reader) (int, error) {	×
UNCOV 346	var offset int	×
UNCOV 347	var err error	×
UNCOV 348	for n := 0; offset < len(b) && err == nil; offset += n {	×
UNCOV 349	n, err = r.Read(b[offset:])	×
UNCOV 350	}	×
UNCOV 351	return offset, err	×
352	}
353
354	// uploadMultipart uploads an artifact using the multipart API.
355	func (s *SimpleStorageService) uploadMultipart(
356	ctx context.Context,
357	buf []byte,
358	objectPath string,
359	artifact io.Reader,
UNCOV 360	) error {	×
UNCOV 361	const maxPartNum = 10000	×
UNCOV 362	var partNum int32 = 1	×
UNCOV 363	var rspUpload *s3.UploadPartOutput	×
UNCOV 364	opts, err := s.optionsFromContext(ctx)	×
UNCOV 365	if err != nil {	×
366	return err	×
367	}	×
368
369	// Pre-allocate 100 completed part (generous guesstimate)
UNCOV 370	completedParts := make([]types.CompletedPart, 0, 100)	×
UNCOV 371		×
UNCOV 372	// Initiate Multipart upload	×
UNCOV 373	createParams := &s3.CreateMultipartUploadInput{	×
UNCOV 374	Bucket: opts.BucketName,	×
UNCOV 375	Key: &objectPath,	×
UNCOV 376	ContentType: s.contentType,	×
UNCOV 377	}	×
UNCOV 378	rspCreate, err := s.client.CreateMultipartUpload(	×
UNCOV 379	ctx, createParams, opts.options,	×
UNCOV 380	)	×
UNCOV 381	if err != nil {	×
382	return err	×
383	}	×
UNCOV 384	uploadParams := &s3.UploadPartInput{	×
UNCOV 385	Bucket: opts.BucketName,	×
UNCOV 386	Key: &objectPath,	×
UNCOV 387	UploadId: rspCreate.UploadId,	×
UNCOV 388	PartNumber: aws.Int32(partNum),	×
UNCOV 389	}	×
UNCOV 390		×
UNCOV 391	// Upload the first chunk already stored in buffer	×
UNCOV 392	r := bytes.NewReader(buf)	×
UNCOV 393	// Readjust upload parameters	×
UNCOV 394	uploadParams.Body = r	×
UNCOV 395	rspUpload, err = s.client.UploadPart(	×
UNCOV 396	ctx,	×
UNCOV 397	uploadParams,	×
UNCOV 398	opts.options,	×
UNCOV 399	)	×
UNCOV 400	if err != nil {	×
401	return err	×
402	}	×
UNCOV 403	completedParts = append(	×
UNCOV 404	completedParts,	×
UNCOV 405	types.CompletedPart{	×
UNCOV 406	ETag: rspUpload.ETag,	×
UNCOV 407	PartNumber: aws.Int32(partNum),	×
UNCOV 408	},	×
UNCOV 409	)	×
UNCOV 410		×
UNCOV 411	// The following is loop is very similar to io.Copy except the	×
UNCOV 412	// destination is the s3 bucket.	×
UNCOV 413	for partNum++; partNum < maxPartNum; partNum++ {	×
UNCOV 414	// Read next chunk from stream (fill the whole buffer)	×
UNCOV 415	offset, eRead := fillBuffer(buf, artifact)	×
UNCOV 416	if offset > 0 {	×
UNCOV 417	r := bytes.NewReader(buf[:offset])	×
UNCOV 418	// Readjust upload parameters	×
UNCOV 419	uploadParams.PartNumber = aws.Int32(partNum)	×
UNCOV 420	uploadParams.Body = r	×
UNCOV 421	rspUpload, err = s.client.UploadPart(	×
UNCOV 422	ctx,	×
UNCOV 423	uploadParams,	×
UNCOV 424	opts.options,	×
UNCOV 425	)	×
UNCOV 426	if err != nil {	×
427	break	×
428	}
UNCOV 429	completedParts = append(	×
UNCOV 430	completedParts,	×
UNCOV 431	types.CompletedPart{	×
UNCOV 432	ETag: rspUpload.ETag,	×
UNCOV 433	PartNumber: aws.Int32(partNum),	×
UNCOV 434	},	×
UNCOV 435	)	×
436	} else {	×
437	// Read did not return any bytes	×
438	break	×
439	}
UNCOV 440	if eRead != nil {	×
UNCOV 441	err = eRead	×
UNCOV 442	break	×
443	}
444	}
UNCOV 445	if err == nil \|\| err == io.EOF {	×
UNCOV 446	// Complete upload	×
UNCOV 447	uploadParams := &s3.CompleteMultipartUploadInput{	×
UNCOV 448	Bucket: opts.BucketName,	×
UNCOV 449	Key: &objectPath,	×
UNCOV 450	UploadId: rspCreate.UploadId,	×
UNCOV 451	MultipartUpload: &types.CompletedMultipartUpload{	×
UNCOV 452	Parts: completedParts,	×
UNCOV 453	},	×
UNCOV 454	}	×
UNCOV 455	_, err = s.client.CompleteMultipartUpload(	×
UNCOV 456	ctx,	×
UNCOV 457	uploadParams,	×
UNCOV 458	opts.options,	×
UNCOV 459	)	×
UNCOV 460	} else {	×
461	// Abort multipart upload!	×
462	uploadParams := &s3.AbortMultipartUploadInput{	×
463	Bucket: opts.BucketName,	×
464	Key: &objectPath,	×
465	UploadId: rspCreate.UploadId,	×
466	}	×
467	_, _ = s.client.AbortMultipartUpload(	×
468	ctx,	×
469	uploadParams,	×
470	opts.options,	×
471	)	×
472	}	×
UNCOV 473	return err	×
474	}
475
476	// UploadArtifact uploads given artifact into the file server (AWS S3 or minio)
477	// using objectID as a key. If the artifact is larger than 5 MiB, the file is
478	// uploaded using the s3 multipart API, otherwise the object is created in a
479	// single request.
480	func (s *SimpleStorageService) PutObject(
481	ctx context.Context,
482	path string,
483	src io.Reader,
UNCOV 484	) error {	×
UNCOV 485	var (	×
UNCOV 486	r io.Reader	×
UNCOV 487	l int64	×
UNCOV 488	n int	×
UNCOV 489	err error	×
UNCOV 490	buf []byte	×
UNCOV 491	)	×
UNCOV 492	if objReader, ok := src.(storage.ObjectReader); ok {	×
493	r = objReader	×
494	l = objReader.Length()	×
UNCOV 495	} else {	×
UNCOV 496	// Peek payload	×
UNCOV 497	buf = make([]byte, s.bufferSize)	×
UNCOV 498	n, err = fillBuffer(buf, src)	×
UNCOV 499	if err == io.EOF {	×
UNCOV 500	r = bytes.NewReader(buf[:n])	×
UNCOV 501	l = int64(n)	×
UNCOV 502	}	×
503	}
504
505	// If only one part, use PutObject API.
UNCOV 506	if r != nil {	×
UNCOV 507	var opts *storageSettings	×
UNCOV 508	opts, err = s.optionsFromContext(ctx)	×
UNCOV 509	if err != nil {	×
510	return err	×
511	}	×
512	// Ordinary single-file upload
UNCOV 513	uploadParams := &s3.PutObjectInput{	×
UNCOV 514	Body: r,	×
UNCOV 515	Bucket: opts.BucketName,	×
UNCOV 516	Key: &path,	×
UNCOV 517	ContentType: s.contentType,	×
UNCOV 518	ContentLength: &l,	×
UNCOV 519	}	×
UNCOV 520	_, err = s.client.PutObject(	×
UNCOV 521	ctx,	×
UNCOV 522	uploadParams,	×
UNCOV 523	opts.options,	×
UNCOV 524	)	×
UNCOV 525	} else if err == nil {	×
UNCOV 526	err = s.uploadMultipart(ctx, buf, path, src)	×
UNCOV 527	}	×
UNCOV 528	return err	×
529	}
530
531	func buildLink(
532	req *v4.PresignedHTTPRequest,
533	signDate time.Time,
534	expireAfter time.Duration,
535	proxyURL *url.URL,
UNCOV 536	) (*model.Link, error) {	×
UNCOV 537	signURL, err := url.Parse(req.URL)	×
UNCOV 538	if err != nil {	×
539	return nil, fmt.Errorf("s3: failed to parse signed URL: %w", err)	×
540	}	×
UNCOV 541	if date, err := time.Parse(	×
UNCOV 542	req.SignedHeader.Get(paramAmzDate), paramAmzDateFormat,	×
UNCOV 543	); err == nil {	×
544	signDate = date	×
545	}	×
UNCOV 546	signURL, err = utils.RewriteProxyURL(signURL, proxyURL)	×
UNCOV 547	if err != nil {	×
548	return nil, fmt.Errorf("s3: failed to rewrite signed URL to proxy: %w", err)	×
549	}	×
550
UNCOV 551	return &model.Link{	×
UNCOV 552	Uri: signURL.String(),	×
UNCOV 553	Expire: signDate.Add(expireAfter),	×
UNCOV 554	Method: req.Method,	×
UNCOV 555	}, nil	×
556	}
557
558	func (s *SimpleStorageService) PutRequest(
559	ctx context.Context,
560	path string,
561	expireAfter time.Duration,
UNCOV 562	) (*model.Link, error) {	×
UNCOV 563		×
UNCOV 564	expireAfter = capDurationToLimits(expireAfter).Truncate(time.Second)	×
UNCOV 565	opts, err := s.optionsFromContext(ctx)	×
UNCOV 566	if err != nil {	×
567	return nil, err	×
568	}	×
569
UNCOV 570	params := &s3.PutObjectInput{	×
UNCOV 571	// Required	×
UNCOV 572	Bucket: opts.BucketName,	×
UNCOV 573	Key: aws.String(path),	×
UNCOV 574	}	×
UNCOV 575		×
UNCOV 576	signDate := time.Now()	×
UNCOV 577	req, err := s.presignClient.PresignPutObject(	×
UNCOV 578	ctx,	×
UNCOV 579	params,	×
UNCOV 580	opts.presignOptions,	×
UNCOV 581	s3.WithPresignExpires(expireAfter),	×
UNCOV 582	)	×
UNCOV 583	if err != nil {	×
584	return nil, err	×
585	}	×
UNCOV 586	return buildLink(req, signDate, expireAfter, opts.ProxyURI)	×
587	}
588
589	// GetRequest duration is limited to 7 days (AWS limitation)
590	func (s *SimpleStorageService) GetRequest(
591	ctx context.Context,
592	objectPath string,
593	filename string,
594	expireAfter time.Duration,
UNCOV 595	) (*model.Link, error) {	×
UNCOV 596		×
UNCOV 597	expireAfter = capDurationToLimits(expireAfter).Truncate(time.Second)	×
UNCOV 598	opts, err := s.optionsFromContext(ctx)	×
UNCOV 599	if err != nil {	×
600	return nil, err	×
601	}	×
602
UNCOV 603	if _, err := s.StatObject(ctx, objectPath); err != nil {	×
604	return nil, errors.WithMessage(err, "s3: head object")	×
605	}	×
606
UNCOV 607	params := &s3.GetObjectInput{	×
UNCOV 608	Bucket: opts.BucketName,	×
UNCOV 609	Key: aws.String(objectPath),	×
UNCOV 610	ResponseContentType: s.contentType,	×
UNCOV 611	}	×
UNCOV 612		×
UNCOV 613	if filename != "" {	×
UNCOV 614	contentDisposition := fmt.Sprintf("attachment; filename=\"%s\"", filename)	×
UNCOV 615	params.ResponseContentDisposition = &contentDisposition	×
UNCOV 616	}	×
617
UNCOV 618	signDate := time.Now()	×
UNCOV 619	req, err := s.presignClient.PresignGetObject(ctx,	×
UNCOV 620	params,	×
UNCOV 621	opts.presignOptions,	×
UNCOV 622	s3.WithPresignExpires(expireAfter))	×
UNCOV 623	if err != nil {	×
624	return nil, errors.WithMessage(err, "s3: failed to sign GET request")	×
625	}	×
UNCOV 626	return buildLink(req, signDate, expireAfter, opts.ProxyURI)	×
627	}
628
629	// DeleteRequest returns a presigned deletion request
630	func (s *SimpleStorageService) DeleteRequest(
631	ctx context.Context,
632	path string,
633	expireAfter time.Duration,
UNCOV 634	) (*model.Link, error) {	×
UNCOV 635		×
UNCOV 636	expireAfter = capDurationToLimits(expireAfter).Truncate(time.Second)	×
UNCOV 637	opts, err := s.optionsFromContext(ctx)	×
UNCOV 638	if err != nil {	×
639	return nil, err	×
640	}	×
641
UNCOV 642	params := &s3.DeleteObjectInput{	×
UNCOV 643	Bucket: opts.BucketName,	×
UNCOV 644	Key: aws.String(path),	×
UNCOV 645	}	×
UNCOV 646		×
UNCOV 647	signDate := time.Now()	×
UNCOV 648	req, err := s.presignClient.PresignDeleteObject(ctx,	×
UNCOV 649	params,	×
UNCOV 650	opts.presignOptions,	×
UNCOV 651	s3.WithPresignExpires(expireAfter))	×
UNCOV 652	if err != nil {	×
653	return nil, errors.WithMessage(err, "s3: failed to sign DELETE request")	×
654	}	×
UNCOV 655	return buildLink(req, signDate, expireAfter, opts.ProxyURI)	×
656	}
657
658	// presign requests are limited to 7 days
UNCOV 659	func capDurationToLimits(duration time.Duration) time.Duration {	×
UNCOV 660	if duration < ExpireMinLimit {	×
661	duration = ExpireMinLimit	×
UNCOV 662	} else if duration > ExpireMaxLimit {	×
663	duration = ExpireMaxLimit	×
664	}	×
UNCOV 665	return duration	×
666	}

mendersoftware / mender-server / 1622978334

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous