920043239

pending completion

Build # 920043239

Build Type

Pull #872

gitlab-ci

Committed by

alfrunes

Commit Message

chore: Restrict tag character set

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

Pull Request Pull Request #872: MEN-6348: Add tags to releases

Run Details

220 of 229 new or added lines in 7 files covered. (96.07%)

223 existing lines in 7 files now uncovered.

7560 of 9480 relevant lines covered (79.75%)

34.07 hits per line

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

81.43

/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
        bucket        string
        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,
        }, 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, bucket string, opts ...*Options) (storage.ObjectStorage, error) {
        opt := NewOptions(opts...)
        s3c, err := newClient(ctx, true, opt)
        if err != nil {
                return nil, err
        }
        s3c.bucket = bucket

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

func (s *SimpleStorageService) init(ctx context.Context) error {
        hparams := &s3.HeadBucketInput{
                Bucket: aws.String(s.bucket),
        }
        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.bucket,
                        )
                }
        }
        if err != nil {
                return err
        }
        cparams := &s3.CreateBucketInput{
                Bucket: aws.String(s.bucket),
        }

        _, err = s.client.CreateBucket(ctx, cparams)
        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, &s3.HeadBucketInput{Bucket: &s.bucket}, waitTime)
        return err
}

func noOpts(*s3.Options) {
}

func (s *SimpleStorageService) optionsFromContext(
        ctx context.Context,
        presign bool,
) (bucket string, clientOptions func(*s3.Options), err error) {
        if settings := settingsFromContext(ctx); settings != nil {
                bucket = settings.Bucket
                clientOptions, err = settings.getOptions(presign)
        } else if s.bucket == "" {
                return "", nil, ErrClientEmpty
        } else {
                bucket = s.bucket
                clientOptions = noOpts
        }
        return bucket, clientOptions, err
}

func (s *SimpleStorageService) HealthCheck(ctx context.Context) error {
        bucket, opts, err := s.optionsFromContext(ctx, false)
        if err != nil {
                return err
        }
        _, err = s.client.HeadBucket(ctx, &s3.HeadBucketInput{
                Bucket: aws.String(bucket),
        }, opts)
        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) {
        bucket, opts, err := s.optionsFromContext(ctx, false)
        if err != nil {
                return nil, err
        }
        params := &s3.GetObjectInput{
                Bucket: aws.String(bucket),
                Key:    aws.String(path),

                RequestPayer: types.RequestPayerRequester,
        }

        out, err := s.client.GetObject(ctx, params, opts)
        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 {
        bucket, opts, err := s.optionsFromContext(ctx, false)
        if err != nil {
                return err
        }

        params := &s3.DeleteObjectInput{
                // Required
                Bucket: aws.String(bucket),
                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)
        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) {

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

        params := &s3.HeadObjectInput{
                Bucket: aws.String(bucket),
                Key:    aws.String(path),
        }
        rsp, err := s.client.HeadObject(ctx, params, opts)
        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
        bucket, opts, err := s.optionsFromContext(ctx, true)
        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:      &bucket,
                Key:         &objectPath,
                ContentType: s.contentType,
        }
        rspCreate, err := s.client.CreateMultipartUpload(
                ctx, createParams, opts,
        )
        if err != nil {
                return err
        }
        uploadParams := &s3.UploadPartInput{
                Bucket:     &bucket,
                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,
        )
        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,
                        )
                        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:   &bucket,
                        Key:      &objectPath,
                        UploadId: rspCreate.UploadId,
                        MultipartUpload: &types.CompletedMultipartUpload{
                                Parts: completedParts,
                        },
                }
                _, err = s.client.CompleteMultipartUpload(
                        ctx,
                        uploadParams,
                        opts,
                )
        } else {
                // Abort multipart upload!
                uploadParams := &s3.AbortMultipartUploadInput{
                        Bucket:   &bucket,
                        Key:      &objectPath,
                        UploadId: rspCreate.UploadId,
                }
                _, _ = s.client.AbortMultipartUpload(
                        ctx,
                        uploadParams,
                        opts,
                )
        }
        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 (
                        bucket string
                        opts   func(*s3.Options)
                )
                bucket, opts, err = s.optionsFromContext(ctx, true)
                if err != nil {
                        return err
                }
                // Ordinary single-file upload
                uploadParams := &s3.PutObjectInput{
                        Body:          r,
                        Bucket:        &bucket,
                        Key:           &path,
                        ContentType:   s.contentType,
                        ContentLength: l,
                }
                _, err = s.client.PutObject(
                        ctx,
                        uploadParams,
                        opts,
                )
        } 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)
        bucket, opts, err := s.optionsFromContext(ctx, true)
        if err != nil {
                return nil, err
        }

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

        signDate := time.Now()
        req, err := s.presignClient.PresignPutObject(
                ctx,
                params,
                s3.WithPresignExpires(expireAfter),
                s3.WithPresignClientFromClientOptions(opts),
        )
        if err != nil {
                return nil, err
        }
        if date, err := time.Parse(
                req.SignedHeader.Get(paramAmzDate), paramAmzDateFormat,
        ); err == nil {
                signDate = date
        }

        return &model.Link{
                Uri:    req.URL,
                Expire: signDate.Add(expireAfter),
                Method: http.MethodPut,
        }, nil
}

// 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)
        bucket, opts, err := s.optionsFromContext(ctx, true)
        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:              aws.String(bucket),
                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,
                s3.WithPresignExpires(expireAfter),
                s3.WithPresignClientFromClientOptions(opts))
        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
        }

        return &model.Link{
                Uri:    req.URL,
                Expire: signDate.Add(expireAfter),
                Method: http.MethodGet,
        }, nil
}

// 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)
        bucket, opts, err := s.optionsFromContext(ctx, true)
        if err != nil {
                return nil, err
        }

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

        signDate := time.Now()
        req, err := s.presignClient.PresignDeleteObject(ctx,
                params,
                s3.WithPresignExpires(expireAfter),
                s3.WithPresignClientFromClientOptions(opts))
        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
        }

        return &model.Link{
                Uri:    req.URL,
                Expire: signDate.Add(expireAfter),
                Method: http.MethodDelete,
        }, nil
}

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

mendersoftware / deployments / 920043239

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