go-storage/s3.go

/*
MIT License

Copyright (c) 2024 Jonas Kaninda

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.
*/

package go_storage

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"net/http"
	"strings"
	"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/credentials"
	"github.com/aws/aws-sdk-go-v2/feature/s3/manager"
	"github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/aws/aws-sdk-go-v2/service/s3/types"
	"github.com/aws/smithy-go"
)

// S3 defaults.
const (
	DefaultS3PartSize    int64 = 64 << 20
	MinS3PartSize        int64 = 5 << 20
	DefaultS3Concurrency       = 2
	DefaultS3Region            = "us-east-1"
)

// S3Storage implements Storage for Amazon S3 and S3-compatible services
// (MinIO, Ceph, Wasabi, ...).
type S3Storage struct {
	client       *s3.Client
	uploader     *manager.Uploader //nolint:staticcheck // see NewS3Storage
	bucket       string
	region       string
	storageClass types.StorageClass
	sse          types.ServerSideEncryption
	sseKMSKeyID  string
}

// NewS3Storage creates a new S3 storage instance.
//
// Credentials: static keys (S3KeyID/S3Secret/S3SessionToken) when set,
// otherwise the default AWS credential chain (environment, shared profile
// S3Profile, web identity/IRSA, ECS/EC2 roles).
func NewS3Storage(config Config) (*S3Storage, error) {
	if err := validateConfig(config, map[string]string{
		"bucket": config.S3Bucket,
	}); err != nil {
		return nil, fmt.Errorf("s3: %w", err)
	}

	opts := []func(*awsconfig.LoadOptions) error{
		awsconfig.WithRequestChecksumCalculation(aws.RequestChecksumCalculationWhenRequired),
		awsconfig.WithResponseChecksumValidation(aws.ResponseChecksumValidationWhenRequired),
	}
	if config.S3Region != "" {
		opts = append(opts, awsconfig.WithRegion(config.S3Region))
	}
	if config.S3Profile != "" {
		opts = append(opts, awsconfig.WithSharedConfigProfile(config.S3Profile))
	}
	if config.S3KeyID != "" && config.S3Secret != "" {
		opts = append(opts, awsconfig.WithCredentialsProvider(
			credentials.NewStaticCredentialsProvider(config.S3KeyID, config.S3Secret, config.S3SessionToken)))
	}

	cfg, err := awsconfig.LoadDefaultConfig(context.Background(), opts...)
	if err != nil {
		return nil, fmt.Errorf("s3: failed to load AWS configuration: %w", err)
	}
	if cfg.Region == "" {
		cfg.Region = DefaultS3Region
	}

	endpoint := config.S3Endpoint
	if endpoint != "" && !strings.Contains(endpoint, "://") {
		if config.S3DisableTLS {
			endpoint = "http://" + endpoint
		} else {
			endpoint = "https://" + endpoint
		}
	}

	client := s3.NewFromConfig(cfg, func(o *s3.Options) {
		if endpoint != "" {
			o.BaseEndpoint = aws.String(endpoint)
		}
		o.UsePathStyle = config.S3ForcePath
	})

	partSize := config.S3PartSize
	if partSize <= 0 {
		partSize = DefaultS3PartSize
	}
	if partSize < MinS3PartSize {
		partSize = MinS3PartSize
	}
	concurrency := config.S3Concurrency
	if concurrency <= 0 {
		concurrency = DefaultS3Concurrency
	}

	//nolint:staticcheck // feature/s3/manager is deprecated in favour of the (pre-1.0) transfermanager.
	uploader := manager.NewUploader(client, func(u *manager.Uploader) {
		u.PartSize = partSize
		u.Concurrency = concurrency
		u.LeavePartsOnError = false
		u.RequestChecksumCalculation = aws.RequestChecksumCalculationWhenRequired
	})

	s := &S3Storage{
		client:       client,
		uploader:     uploader,
		bucket:       config.S3Bucket,
		region:       cfg.Region,
		storageClass: types.StorageClass(config.S3StorageClass),
	}
	switch strings.ToLower(config.S3SSE) {
	case "":
	case "aes256":
		s.sse = types.ServerSideEncryptionAes256
	case "aws:kms":
		s.sse = types.ServerSideEncryptionAwsKms
		s.sseKMSKeyID = config.S3SSEKMSKeyID
	default:
		s.sse = types.ServerSideEncryption(config.S3SSE)
		s.sseKMSKeyID = config.S3SSEKMSKeyID
	}
	return s, nil
}

// Client returns the underlying S3 client.
func (s *S3Storage) Client() *s3.Client { return s.client }

func (s *S3Storage) putInput(key string, body io.Reader, opts *PutOptions) *s3.PutObjectInput {
	in := &s3.PutObjectInput{
		Bucket: aws.String(s.bucket),
		Key:    aws.String(key),
		Body:   body,
	}
	if opts != nil {
		if opts.ContentType != "" {
			in.ContentType = aws.String(opts.ContentType)
		}
		if len(opts.Metadata) > 0 {
			in.Metadata = opts.Metadata
		}
	}
	if s.storageClass != "" {
		in.StorageClass = s.storageClass
	}
	if s.sse != "" {
		in.ServerSideEncryption = s.sse
		if s.sseKMSKeyID != "" {
			in.SSEKMSKeyId = aws.String(s.sseKMSKeyID)
		}
	}
	return in
}

// Put streams r to key using a multipart upload (aborted on error).
func (s *S3Storage) Put(ctx context.Context, key string, r io.Reader, opts *PutOptions) (Item, error) {
	k, err := cleanKey(key)
	if err != nil {
		return Item{}, fmt.Errorf("s3: %w", err)
	}
	cr := &countingReader{r: ctxReader{ctx: ctx, r: r}}
	//nolint:staticcheck // see NewS3Storage
	out, err := s.uploader.Upload(ctx, s.putInput(k, cr, opts))
	if err != nil {
		return Item{}, fmt.Errorf("s3: failed to upload %s to bucket %s: %w", k, s.bucket, err)
	}
	return Item{
		Key:          k,
		Size:         cr.n,
		ModifiedTime: time.Now().UTC(),
		ETag:         aws.ToString(out.ETag),
		VersionID:    aws.ToString(out.VersionID),
	}, nil
}

// PutIf performs a conditional single-request PutObject (If-Match /
// If-None-Match: *). The body is buffered and limited to 16 MiB.
func (s *S3Storage) PutIf(ctx context.Context, key string, r io.Reader, cond Condition, opts *PutOptions) (Item, error) {
	k, err := cleanKey(key)
	if err != nil {
		return Item{}, fmt.Errorf("s3: %w", err)
	}
	if err := validateCondition(cond); err != nil {
		return Item{}, fmt.Errorf("s3: %w", err)
	}
	data, err := io.ReadAll(io.LimitReader(ctxReader{ctx: ctx, r: r}, maxConditionalPutSize+1))
	if err != nil {
		return Item{}, fmt.Errorf("s3: failed to read data: %w", err)
	}
	if len(data) > maxConditionalPutSize {
		return Item{}, fmt.Errorf("s3: conditional put of %s exceeds %d bytes", k, maxConditionalPutSize)
	}
	in := s.putInput(k, bytes.NewReader(data), opts)
	in.ContentLength = aws.Int64(int64(len(data)))
	if cond.IfMatch != "" {
		in.IfMatch = aws.String(cond.IfMatch)
	}
	if cond.IfNoneMatch {
		in.IfNoneMatch = aws.String("*")
	}
	out, err := s.client.PutObject(ctx, in)
	if err != nil {
		// If-Match on a missing key yields 404 NoSuchKey on AWS.
		if isS3Precondition(err) || (cond.IfMatch != "" && s3StatusCode(err) == http.StatusNotFound) {
			return Item{}, fmt.Errorf("s3: conditional put of %s: %w", k, ErrPreconditionFailed)
		}
		return Item{}, fmt.Errorf("s3: failed to put %s: %w", k, err)
	}
	return Item{
		Key:          k,
		Size:         int64(len(data)),
		ModifiedTime: time.Now().UTC(),
		ETag:         aws.ToString(out.ETag),
		VersionID:    aws.ToString(out.VersionId),
	}, nil
}

// Get streams the object body.
func (s *S3Storage) Get(ctx context.Context, key string) (io.ReadCloser, Item, error) {
	k, err := cleanKey(key)
	if err != nil {
		return nil, Item{}, fmt.Errorf("s3: %w", err)
	}
	out, err := s.client.GetObject(ctx, &s3.GetObjectInput{Bucket: aws.String(s.bucket), Key: aws.String(k)})
	if err != nil {
		return nil, Item{}, s.mapErr("get", k, err)
	}
	return out.Body, Item{
		Key:          k,
		Size:         aws.ToInt64(out.ContentLength),
		ModifiedTime: aws.ToTime(out.LastModified),
		ETag:         aws.ToString(out.ETag),
		VersionID:    aws.ToString(out.VersionId),
	}, nil
}

// Stat returns object metadata (HeadObject).
func (s *S3Storage) Stat(ctx context.Context, key string) (Item, error) {
	k, err := cleanKey(key)
	if err != nil {
		return Item{}, fmt.Errorf("s3: %w", err)
	}
	out, err := s.client.HeadObject(ctx, &s3.HeadObjectInput{Bucket: aws.String(s.bucket), Key: aws.String(k)})
	if err != nil {
		return Item{}, s.mapErr("stat", k, err)
	}
	return Item{
		Key:          k,
		Size:         aws.ToInt64(out.ContentLength),
		ModifiedTime: aws.ToTime(out.LastModified),
		ETag:         aws.ToString(out.ETag),
		VersionID:    aws.ToString(out.VersionId),
	}, nil
}

// Upload uploads a file to S3.
func (s *S3Storage) Upload(ctx context.Context, localPath string, remotePath string) error {
	if err := uploadFile(ctx, s, localPath, remotePath); err != nil {
		return fmt.Errorf("s3: %w", err)
	}
	return nil
}

// Download downloads a file from S3.
func (s *S3Storage) Download(ctx context.Context, remotePath string, localPath string) error {
	if err := downloadFile(ctx, s, remotePath, localPath); err != nil {
		return fmt.Errorf("s3: %w", err)
	}
	return nil
}

// List lists objects (recursively) with the given prefix.
func (s *S3Storage) List(ctx context.Context, prefix string) ([]Item, error) {
	p, err := cleanPrefix(prefix)
	if err != nil {
		return nil, fmt.Errorf("s3: %w", err)
	}
	return s.list(ctx, p, "")
}

// ListDir lists the objects directly under dir.
func (s *S3Storage) ListDir(ctx context.Context, dir string) ([]Item, error) {
	d, err := cleanDir(dir)
	if err != nil {
		return nil, fmt.Errorf("s3: %w", err)
	}
	return s.list(ctx, dirPrefix(d), "/")
}

func (s *S3Storage) list(ctx context.Context, prefix, delimiter string) ([]Item, error) {
	in := &s3.ListObjectsV2Input{Bucket: aws.String(s.bucket), Prefix: aws.String(prefix)}
	if delimiter != "" {
		in.Delimiter = aws.String(delimiter)
	}
	items := make([]Item, 0)
	p := s3.NewListObjectsV2Paginator(s.client, in)
	for p.HasMorePages() {
		page, err := p.NextPage(ctx)
		if err != nil {
			return nil, fmt.Errorf("s3: failed to list objects in bucket %s: %w", s.bucket, err)
		}
		for _, obj := range page.Contents {
			key := aws.ToString(obj.Key)
			if delimiter != "" && strings.HasSuffix(key, "/") {
				continue // directory marker
			}
			items = append(items, Item{
				Key:          key,
				ModifiedTime: aws.ToTime(obj.LastModified),
				Size:         aws.ToInt64(obj.Size),
				ETag:         aws.ToString(obj.ETag),
			})
		}
	}
	return items, nil
}

// Remove deletes an object. A missing key returns ErrNotFound (checked with
// HeadObject because DeleteObject is idempotent on S3).
func (s *S3Storage) Remove(ctx context.Context, remotePath string) error {
	k, err := cleanKey(remotePath)
	if err != nil {
		return fmt.Errorf("s3: %w", err)
	}
	if _, err := s.Stat(ctx, k); err != nil {
		return err
	}
	if _, err := s.client.DeleteObject(ctx, &s3.DeleteObjectInput{Bucket: aws.String(s.bucket), Key: aws.String(k)}); err != nil {
		return s.mapErr("remove", k, err)
	}
	return nil
}

// Close releases resources (no-op).
func (s *S3Storage) Close() error {
	return nil
}

func (s *S3Storage) mapErr(op, key string, err error) error {
	if s3StatusCode(err) == http.StatusNotFound {
		return fmt.Errorf("s3: %s %s in bucket %s: %w", op, key, s.bucket, ErrNotFound)
	}
	var nsk *types.NoSuchKey
	var nf *types.NotFound
	if errors.As(err, &nsk) || errors.As(err, &nf) {
		return fmt.Errorf("s3: %s %s in bucket %s: %w", op, key, s.bucket, ErrNotFound)
	}
	return fmt.Errorf("s3: %s %s in bucket %s: %w", op, key, s.bucket, err)
}

func s3StatusCode(err error) int {
	var re *awshttp.ResponseError
	if errors.As(err, &re) {
		return re.HTTPStatusCode()
	}
	return 0
}

func isS3Precondition(err error) bool {
	var apiErr smithy.APIError
	if errors.As(err, &apiErr) {
		switch apiErr.ErrorCode() {
		case "PreconditionFailed", "ConditionalRequestConflict":
			return true
		}
	}
	switch s3StatusCode(err) {
	case http.StatusPreconditionFailed, http.StatusConflict:
		return true
	}
	return false
}