913012443

pending completion

Build # 913012443

Build Type

Pull #869

gitlab-ci

Committed by

alfrunes

Commit Message

ci: Bump golangci-lint to v1.15.3 (go1.20)

Signed-off-by: Alf-Rune Siqveland <alf.rune@northern.tech>

Pull Request Pull Request #869: feat: New configuration parameter storage.proxy_uri

Run Details

179 of 253 new or added lines in 6 files covered. (70.75%)

2 existing lines in 2 files now uncovered.

7373 of 9353 relevant lines covered (78.83%)

33.73 hits per line

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

80.45

/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"
        "time"

        "github.com/aws/aws-sdk-go-v2/aws"
        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/deployments/model"
        "github.com/mendersoftware/deployments/storage"
)

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 {
        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: 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: 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 = 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: 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 (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
        }
        if date, err := time.Parse(
                req.SignedHeader.Get(paramAmzDate), paramAmzDateFormat,
        ); err == nil {
                signDate = date
        }

        link := &model.Link{
                Uri:    req.URL,
                Expire: signDate.Add(expireAfter),
                Method: http.MethodPut,
        }
        if opts.ProxyURI != nil {
                link, err = opts.RewriteLink(link)
        }
        return link, err
}

// 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")
        }
        if date, err := time.Parse(
                req.SignedHeader.Get(paramAmzDate), paramAmzDateFormat,
        ); err == nil {
                signDate = date
        }

        link := &model.Link{
                Uri:    req.URL,
                Expire: signDate.Add(expireAfter),
                Method: http.MethodGet,
        }
        if opts.ProxyURI != nil {
                link, err = opts.RewriteLink(link)
        }
        return link, err
}

// 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")
        }
        if date, err := time.Parse(
                req.SignedHeader.Get(paramAmzDate), paramAmzDateFormat,
        ); err == nil {
                signDate = date
        }
        link := &model.Link{
                Uri:    req.URL,
                Expire: signDate.Add(expireAfter),
                Method: http.MethodDelete,
        }
        if opts.ProxyURI != nil {
                link, err = opts.RewriteLink(link)
        }
        return link, err
}

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

mendersoftware / deployments / 913012443

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