914758580

pending completion

Build # 914758580

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

180 of 261 new or added lines in 6 files covered. (68.97%)

2 existing lines in 2 files now uncovered.

7374 of 9361 relevant lines covered (78.77%)

33.7 hits per line

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

80.0

/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 {
        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: 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	if s.settings.BucketName == nil {	5✔
NEW 158	return errors.New("s3: failed to initalize storage client: " +	×
NEW 159	"a bucket name is required")	×
NEW 160	}	×
161	hparams := &s3.HeadBucketInput{	5✔
162	Bucket: s.settings.BucketName,	5✔
163	}	5✔
164	var rspErr *awsHttp.ResponseError	5✔
165		5✔
166	_, err := s.client.HeadBucket(ctx, hparams)	5✔
167	if err == nil {	8✔
168	// bucket exists and have permission to access it	3✔
169	return nil	3✔
170	} else if errors.As(err, &rspErr) {	7✔
171	switch rspErr.Response.StatusCode {	2✔
172	case http.StatusNotFound:	2✔
173	err = nil // pass	2✔
174	case http.StatusForbidden:	×
175	err = fmt.Errorf(	×
176	"s3: insufficient permissions for accessing bucket '%s'",	×
NEW 177	*s.settings.BucketName,	×
178	)	×
179	}
180	}
181	if err != nil {	2✔
182	return err	×
183	}	×
184	cparams := &s3.CreateBucketInput{	2✔
185	Bucket: s.settings.BucketName,	2✔
186	}	2✔
187		2✔
188	_, err = s.client.CreateBucket(ctx, cparams, disableAccelerate)	2✔
189	if err != nil {	2✔
190	var errBucket *types.BucketAlreadyOwnedByYou	×
191	if !errors.As(err, errBucket) {	×
192	return errors.WithMessage(err, "s3: error creating bucket")	×
193	}	×
194	}
195	waitTime := time.Second * 30	2✔
196	if deadline, ok := ctx.Deadline(); ok {	4✔
197	waitTime = time.Until(deadline)	2✔
198	}	2✔
199	err = s3.NewBucketExistsWaiter(s.client).	2✔
200	Wait(ctx, hparams, waitTime)	2✔
201	return err	2✔
202	}
203
204	func (s *SimpleStorageService) optionsFromContext(
205	ctx context.Context,
206	) (settings *storageSettings, err error) {	5✔
207	ss, ok := storage.SettingsFromContext(ctx)	5✔
208	if ok && ss != nil {	7✔
209	err = ss.Validate()	2✔
210	if err == nil {	3✔
211	settings = newFromParent(&s.settings, ss)	1✔
212	}	1✔
213	} else {	4✔
214	settings = &s.settings	4✔
215	if settings.BucketName == nil {	4✔
NEW 216	err = ErrClientEmpty	×
NEW 217	}	×
218	}
219	return settings, err	5✔
220	}
221
222	func (s *SimpleStorageService) HealthCheck(ctx context.Context) error {	2✔
223	opts, err := s.optionsFromContext(ctx)	2✔
224	if err != nil {	2✔
225	return err	×
226	}	×
227	_, err = s.client.HeadBucket(ctx, &s3.HeadBucketInput{	2✔
228	Bucket: opts.BucketName,	2✔
229	}, opts.options)	2✔
230	return err	2✔
231	}
232
233	type objectReader struct {
234	io.ReadCloser
235	length int64
236	}
237
238	func (obj objectReader) Length() int64 {	×
239	return obj.length	×
240	}	×
241
242	func (s *SimpleStorageService) GetObject(
243	ctx context.Context,
244	path string,
245	) (io.ReadCloser, error) {	4✔
246	opts, err := s.optionsFromContext(ctx)	4✔
247	if err != nil {	5✔
248	return nil, err	1✔
249	}	1✔
250	params := &s3.GetObjectInput{	3✔
251	Bucket: opts.BucketName,	3✔
252	Key: aws.String(path),	3✔
253		3✔
254	RequestPayer: types.RequestPayerRequester,	3✔
255	}	3✔
256		3✔
257	out, err := s.client.GetObject(ctx, params, opts.options)	3✔
258	var rspErr *awsHttp.ResponseError	3✔
259	if errors.As(err, &rspErr) {	4✔
260	if rspErr.Response.StatusCode == http.StatusNotFound {	2✔
261	err = storage.ErrObjectNotFound	1✔
262	}	1✔
263	}
264	if err != nil {	4✔
265	return nil, errors.WithMessage(	1✔
266	err,	1✔
267	"s3: failed to get object",	1✔
268	)	1✔
269	}	1✔
270	return objectReader{	2✔
271	ReadCloser: out.Body,	2✔
272	length: out.ContentLength,	2✔
273	}, nil	2✔
274	}
275
276	// Delete removes deleted file from storage.
277	// Noop if ID does not exist.
278	func (s *SimpleStorageService) DeleteObject(ctx context.Context, path string) error {	1✔
279	opts, err := s.optionsFromContext(ctx)	1✔
280	if err != nil {	1✔
281	return err	×
282	}	×
283
284	params := &s3.DeleteObjectInput{	1✔
285	// Required	1✔
286	Bucket: opts.BucketName,	1✔
287	Key: aws.String(path),	1✔
288		1✔
289	// Optional	1✔
290	RequestPayer: types.RequestPayerRequester,	1✔
291	}	1✔
292		1✔
293	// ignore return response which contains charing info	1✔
294	// and file versioning data which are not in interest	1✔
295	_, err = s.client.DeleteObject(ctx, params, opts.options)	1✔
296	var rspErr *awsHttp.ResponseError	1✔
297	if errors.As(err, &rspErr) {	1✔
298	if rspErr.Response.StatusCode == http.StatusNotFound {	×
299	err = storage.ErrObjectNotFound	×
300	}	×
301	}
302	if err != nil {	1✔
303	return errors.WithMessage(err, "s3: error deleting object")	×
304	}	×
305
306	return nil	1✔
307	}
308
309	// Exists check if selected object exists in the storage
310	func (s *SimpleStorageService) StatObject(
311	ctx context.Context,
312	path string,
313	) (*storage.ObjectInfo, error) {	1✔
314		1✔
315	opts, err := s.optionsFromContext(ctx)	1✔
316	if err != nil {	1✔
317	return nil, err	×
318	}	×
319
320	params := &s3.HeadObjectInput{	1✔
321	Bucket: opts.BucketName,	1✔
322	Key: aws.String(path),	1✔
323	}	1✔
324	rsp, err := s.client.HeadObject(ctx, params, opts.options)	1✔
325	var rspErr *awsHttp.ResponseError	1✔
326	if errors.As(err, &rspErr) {	1✔
327	if rspErr.Response.StatusCode == http.StatusNotFound {	×
328	err = storage.ErrObjectNotFound	×
329	}	×
330	}
331	if err != nil {	1✔
332	return nil, errors.WithMessage(err, "s3: error getting object info")	×
333	}	×
334
335	return &storage.ObjectInfo{	1✔
336	Path: path,	1✔
337	LastModified: rsp.LastModified,	1✔
338	Size: &rsp.ContentLength,	1✔
339	}, nil	1✔
340	}
341
342	func fillBuffer(b []byte, r io.Reader) (int, error) {	1✔
343	var offset int	1✔
344	var err error	1✔
345	for n := 0; offset < len(b) && err == nil; offset += n {	2✔
346	n, err = r.Read(b[offset:])	1✔
347	}	1✔
348	return offset, err	1✔
349	}
350
351	// uploadMultipart uploads an artifact using the multipart API.
352	func (s *SimpleStorageService) uploadMultipart(
353	ctx context.Context,
354	buf []byte,
355	objectPath string,
356	artifact io.Reader,
357	) error {	1✔
358	const maxPartNum = 10000	1✔
359	var partNum int32 = 1	1✔
360	var rspUpload *s3.UploadPartOutput	1✔
361	opts, err := s.optionsFromContext(ctx)	1✔
362	if err != nil {	1✔
363	return err	×
364	}	×
365
366	// Pre-allocate 100 completed part (generous guesstimate)
367	completedParts := make([]types.CompletedPart, 0, 100)	1✔
368		1✔
369	// Initiate Multipart upload	1✔
370	createParams := &s3.CreateMultipartUploadInput{	1✔
371	Bucket: opts.BucketName,	1✔
372	Key: &objectPath,	1✔
373	ContentType: s.contentType,	1✔
374	}	1✔
375	rspCreate, err := s.client.CreateMultipartUpload(	1✔
376	ctx, createParams, opts.options,	1✔
377	)	1✔
378	if err != nil {	1✔
379	return err	×
380	}	×
381	uploadParams := &s3.UploadPartInput{	1✔
382	Bucket: opts.BucketName,	1✔
383	Key: &objectPath,	1✔
384	UploadId: rspCreate.UploadId,	1✔
385	PartNumber: partNum,	1✔
386	}	1✔
387		1✔
388	// Upload the first chunk already stored in buffer	1✔
389	r := bytes.NewReader(buf)	1✔
390	// Readjust upload parameters	1✔
391	uploadParams.Body = r	1✔
392	rspUpload, err = s.client.UploadPart(	1✔
393	ctx,	1✔
394	uploadParams,	1✔
395	opts.options,	1✔
396	)	1✔
397	if err != nil {	1✔
398	return err	×
399	}	×
400	completedParts = append(	1✔
401	completedParts,	1✔
402	types.CompletedPart{	1✔
403	ETag: rspUpload.ETag,	1✔
404	PartNumber: partNum,	1✔
405	},	1✔
406	)	1✔
407		1✔
408	// The following is loop is very similar to io.Copy except the	1✔
409	// destination is the s3 bucket.	1✔
410	for partNum++; partNum < maxPartNum; partNum++ {	2✔
411	// Read next chunk from stream (fill the whole buffer)	1✔
412	offset, eRead := fillBuffer(buf, artifact)	1✔
413	if offset > 0 {	2✔
414	r := bytes.NewReader(buf[:offset])	1✔
415	// Readjust upload parameters	1✔
416	uploadParams.PartNumber = partNum	1✔
417	uploadParams.Body = r	1✔
418	rspUpload, err = s.client.UploadPart(	1✔
419	ctx,	1✔
420	uploadParams,	1✔
421	opts.options,	1✔
422	)	1✔
423	if err != nil {	1✔
424	break	×
425	}
426	completedParts = append(	1✔
427	completedParts,	1✔
428	types.CompletedPart{	1✔
429	ETag: rspUpload.ETag,	1✔
430	PartNumber: partNum,	1✔
431	},	1✔
432	)	1✔
433	} else {	×
434	// Read did not return any bytes	×
435	break	×
436	}
437	if eRead != nil {	2✔
438	err = eRead	1✔
439	break	1✔
440	}
441	}
442	if err == nil \|\| err == io.EOF {	2✔
443	// Complete upload	1✔
444	uploadParams := &s3.CompleteMultipartUploadInput{	1✔
445	Bucket: opts.BucketName,	1✔
446	Key: &objectPath,	1✔
447	UploadId: rspCreate.UploadId,	1✔
448	MultipartUpload: &types.CompletedMultipartUpload{	1✔
449	Parts: completedParts,	1✔
450	},	1✔
451	}	1✔
452	_, err = s.client.CompleteMultipartUpload(	1✔
453	ctx,	1✔
454	uploadParams,	1✔
455	opts.options,	1✔
456	)	1✔
457	} else {	1✔
458	// Abort multipart upload!	×
459	uploadParams := &s3.AbortMultipartUploadInput{	×
NEW 460	Bucket: opts.BucketName,	×
461	Key: &objectPath,	×
462	UploadId: rspCreate.UploadId,	×
463	}	×
464	_, _ = s.client.AbortMultipartUpload(	×
465	ctx,	×
466	uploadParams,	×
NEW 467	opts.options,	×
468	)	×
469	}	×
470	return err	1✔
471	}
472
473	// UploadArtifact uploads given artifact into the file server (AWS S3 or minio)
474	// using objectID as a key. If the artifact is larger than 5 MiB, the file is
475	// uploaded using the s3 multipart API, otherwise the object is created in a
476	// single request.
477	func (s *SimpleStorageService) PutObject(
478	ctx context.Context,
479	path string,
480	src io.Reader,
481	) error {	1✔
482	var (	1✔
483	r io.Reader	1✔
484	l int64	1✔
485	n int	1✔
486	err error	1✔
487	buf []byte	1✔
488	)	1✔
489	if objReader, ok := src.(storage.ObjectReader); ok {	1✔
490	r = objReader	×
491	l = objReader.Length()	×
492	} else {	1✔
493	// Peek payload	1✔
494	buf = make([]byte, s.bufferSize)	1✔
495	n, err = fillBuffer(buf, src)	1✔
496	if err == io.EOF {	2✔
497	r = bytes.NewReader(buf[:n])	1✔
498	l = int64(n)	1✔
499	}	1✔
500	}
501
502	// If only one part, use PutObject API.
503	if r != nil {	2✔
504	var opts *storageSettings	1✔
505	opts, err = s.optionsFromContext(ctx)	1✔
506	if err != nil {	1✔
507	return err	×
508	}	×
509	// Ordinary single-file upload
510	uploadParams := &s3.PutObjectInput{	1✔
511	Body: r,	1✔
512	Bucket: opts.BucketName,	1✔
513	Key: &path,	1✔
514	ContentType: s.contentType,	1✔
515	ContentLength: l,	1✔
516	}	1✔
517	_, err = s.client.PutObject(	1✔
518	ctx,	1✔
519	uploadParams,	1✔
520	opts.options,	1✔
521	)	1✔
522	} else if err == nil {	2✔
523	err = s.uploadMultipart(ctx, buf, path, src)	1✔
524	}	1✔
525	return err	1✔
526	}
527
528	func (s *SimpleStorageService) PutRequest(
529	ctx context.Context,
530	path string,
531	expireAfter time.Duration,
532	) (*model.Link, error) {	1✔
533		1✔
534	expireAfter = capDurationToLimits(expireAfter).Truncate(time.Second)	1✔
535	opts, err := s.optionsFromContext(ctx)	1✔
536	if err != nil {	1✔
537	return nil, err	×
538	}	×
539
540	params := &s3.PutObjectInput{	1✔
541	// Required	1✔
542	Bucket: opts.BucketName,	1✔
543	Key: aws.String(path),	1✔
544	}	1✔
545		1✔
546	signDate := time.Now()	1✔
547	req, err := s.presignClient.PresignPutObject(	1✔
548	ctx,	1✔
549	params,	1✔
550	opts.presignOptions,	1✔
551	s3.WithPresignExpires(expireAfter),	1✔
552	)	1✔
553	if err != nil {	1✔
554	return nil, err	×
555	}	×
556	if date, err := time.Parse(	1✔
557	req.SignedHeader.Get(paramAmzDate), paramAmzDateFormat,	1✔
558	); err == nil {	1✔
559	signDate = date	×
560	}	×
561
562	link := &model.Link{	1✔
563	Uri: req.URL,	1✔
564	Expire: signDate.Add(expireAfter),	1✔
565	Method: http.MethodPut,	1✔
566	}	1✔
567	if opts.ProxyURI != nil {	1✔
NEW 568	link, err = opts.RewriteLink(link)	×
NEW 569	}	×
570	return link, err	1✔
571	}
572
573	// GetRequest duration is limited to 7 days (AWS limitation)
574	func (s *SimpleStorageService) GetRequest(
575	ctx context.Context,
576	objectPath string,
577	filename string,
578	expireAfter time.Duration,
579	) (*model.Link, error) {	1✔
580		1✔
581	expireAfter = capDurationToLimits(expireAfter).Truncate(time.Second)	1✔
582	opts, err := s.optionsFromContext(ctx)	1✔
583	if err != nil {	1✔
584	return nil, err	×
585	}	×
586
587	if _, err := s.StatObject(ctx, objectPath); err != nil {	1✔
588	return nil, errors.WithMessage(err, "s3: head object")	×
589	}	×
590
591	params := &s3.GetObjectInput{	1✔
592	Bucket: opts.BucketName,	1✔
593	Key: aws.String(objectPath),	1✔
594	ResponseContentType: s.contentType,	1✔
595	}	1✔
596		1✔
597	if filename != "" {	2✔
598	contentDisposition := fmt.Sprintf("attachment; filename=\"%s\"", filename)	1✔
599	params.ResponseContentDisposition = &contentDisposition	1✔
600	}	1✔
601
602	signDate := time.Now()	1✔
603	req, err := s.presignClient.PresignGetObject(ctx,	1✔
604	params,	1✔
605	opts.presignOptions,	1✔
606	s3.WithPresignExpires(expireAfter))	1✔
607	if err != nil {	1✔
608	return nil, errors.WithMessage(err, "s3: failed to sign GET request")	×
609	}	×
610	if date, err := time.Parse(	1✔
611	req.SignedHeader.Get(paramAmzDate), paramAmzDateFormat,	1✔
612	); err == nil {	1✔
613	signDate = date	×
614	}	×
615
616	link := &model.Link{	1✔
617	Uri: req.URL,	1✔
618	Expire: signDate.Add(expireAfter),	1✔
619	Method: http.MethodGet,	1✔
620	}	1✔
621	if opts.ProxyURI != nil {	1✔
NEW 622	link, err = opts.RewriteLink(link)	×
NEW 623	}	×
624	return link, err	1✔
625	}
626
627	// DeleteRequest returns a presigned deletion request
628	func (s *SimpleStorageService) DeleteRequest(
629	ctx context.Context,
630	path string,
631	expireAfter time.Duration,
632	) (*model.Link, error) {	1✔
633		1✔
634	expireAfter = capDurationToLimits(expireAfter).Truncate(time.Second)	1✔
635	opts, err := s.optionsFromContext(ctx)	1✔
636	if err != nil {	1✔
637	return nil, err	×
638	}	×
639
640	params := &s3.DeleteObjectInput{	1✔
641	Bucket: opts.BucketName,	1✔
642	Key: aws.String(path),	1✔
643	}	1✔
644		1✔
645	signDate := time.Now()	1✔
646	req, err := s.presignClient.PresignDeleteObject(ctx,	1✔
647	params,	1✔
648	opts.presignOptions,	1✔
649	s3.WithPresignExpires(expireAfter))	1✔
650	if err != nil {	1✔
651	return nil, errors.WithMessage(err, "s3: failed to sign DELETE request")	×
652	}	×
653	if date, err := time.Parse(	1✔
654	req.SignedHeader.Get(paramAmzDate), paramAmzDateFormat,	1✔
655	); err == nil {	1✔
656	signDate = date	×
657	}	×
658	link := &model.Link{	1✔
659	Uri: req.URL,	1✔
660	Expire: signDate.Add(expireAfter),	1✔
661	Method: http.MethodDelete,	1✔
662	}	1✔
663	if opts.ProxyURI != nil {	1✔
NEW 664	link, err = opts.RewriteLink(link)	×
NEW 665	}	×
666	return link, err	1✔
667	}
668
669	// presign requests are limited to 7 days
670	func capDurationToLimits(duration time.Duration) time.Duration {	1✔
671	if duration < ExpireMinLimit {	1✔
672	duration = ExpireMinLimit	×
673	} else if duration > ExpireMaxLimit {	1✔
674	duration = ExpireMaxLimit	×
675	}	×
676	return duration	1✔
677	}

mendersoftware / deployments / 914758580

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