1635240308

Committed 21 Jan 2025 01:35PM UTC coverage: 76.574% (+0.002%) from 76.572%

Build # 1635240308

Build Type

push

gitlab-ci

Committed by

web-flow

Commit Message

Merge pull request #346 from alfrunes/MEN-7939

fix: Use internal URLs for storage backend when generating artifacts

Run Details

4299 of 6257 branches covered (68.71%)

Branch coverage included in aggregate %.

37 of 57 new or added lines in 5 files covered. (64.91%)

5 existing lines in 2 files now uncovered.

45366 of 58602 relevant lines covered (77.41%)

20.12 hits per line

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

80.29

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

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

package s3

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

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

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

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

        MultipartMaxParts = 10000
        MultipartMinSize  = 5 * mib

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

type objectReader struct {
        io.ReadCloser
        length int64
}

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

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

                RequestPayer: types.RequestPayerRequester,
        }

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

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

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

                // Optional
                RequestPayer: types.RequestPayerRequester,
        }

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

        return nil
}

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

mendersoftware / mender-server / 1635240308

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