versitygw/tests/integration/s3conf.go

// Copyright 2023 Versity Software
// This file is 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 integration

import (
	"context"
	"crypto/tls"
	"io"
	"log"
	"net/http"
	"os"

	"github.com/aws/aws-sdk-go-v2/aws"
	v4 "github.com/aws/aws-sdk-go-v2/aws/signer/v4"
	"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/transfermanager"
	"github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/aws/smithy-go/middleware"
)

type S3Conf struct {
	awsID             string
	awsSecret         string
	awsRegion         string
	endpoint          string
	hostStyle         bool
	checksumDisable   bool
	PartSize          int64
	Concurrency       int
	debug             bool
	versioningEnabled bool
	azureTests        bool
	tlsStatus         bool
	httpClient        *http.Client
}

func NewS3Conf(opts ...Option) *S3Conf {
	s := &S3Conf{}

	for _, opt := range opts {
		opt(s)
	}

	customTransport := &http.Transport{
		TLSClientConfig: &tls.Config{
			InsecureSkipVerify: s.tlsStatus,
		},
	}

	customHTTPClient := &http.Client{
		Transport: customTransport,
		Timeout:   shortTimeout,
	}

	s.httpClient = customHTTPClient

	return s
}

type Option func(*S3Conf)

func WithAccess(ak string) Option {
	return func(s *S3Conf) { s.awsID = ak }
}
func WithSecret(sk string) Option {
	return func(s *S3Conf) { s.awsSecret = sk }
}
func WithRegion(r string) Option {
	return func(s *S3Conf) { s.awsRegion = r }
}
func WithEndpoint(e string) Option {
	return func(s *S3Conf) { s.endpoint = e }
}
func WithDisableChecksum() Option {
	return func(s *S3Conf) { s.checksumDisable = true }
}
func WithHostStyle() Option {
	return func(s *S3Conf) { s.hostStyle = true }
}
func WithPartSize(p int64) Option {
	return func(s *S3Conf) { s.PartSize = p }
}
func WithConcurrency(c int) Option {
	return func(s *S3Conf) { s.Concurrency = c }
}
func WithDebug() Option {
	return func(s *S3Conf) { s.debug = true }
}
func WithVersioningEnabled() Option {
	return func(s *S3Conf) { s.versioningEnabled = true }
}
func WithAzureMode() Option {
	return func(s *S3Conf) { s.azureTests = true }
}
func WithTLSStatus(ts bool) Option {
	return func(s *S3Conf) { s.tlsStatus = ts }
}

func (c *S3Conf) getCreds() credentials.StaticCredentialsProvider {
	// TODO support token/IAM
	if c.awsSecret == "" {
		c.awsSecret = os.Getenv("AWS_SECRET_ACCESS_KEY")
	}
	if c.awsSecret == "" {
		log.Fatal("no AWS_SECRET_ACCESS_KEY found")
	}

	return credentials.NewStaticCredentialsProvider(c.awsID, c.awsSecret, "")
}

func (c *S3Conf) GetClient() *s3.Client {
	return s3.NewFromConfig(c.Config(), func(o *s3.Options) {
		if c.hostStyle {
			o.BaseEndpoint = &c.endpoint
			o.UsePathStyle = false
		}
	})
}

func (c *S3Conf) GetPresignClient() *s3.PresignClient {
	return s3.NewPresignClient(c.GetClient())
}

func (c *S3Conf) GetAnonymousClient() *s3.Client {
	cfg := c.Config()
	cfg.Credentials = aws.AnonymousCredentials{}
	return s3.NewFromConfig(cfg, func(o *s3.Options) {
		if c.hostStyle {
			o.BaseEndpoint = &c.endpoint
			o.UsePathStyle = false
		}
	})
}

func (cfg *S3Conf) getUserClient(usr user) *s3.Client {
	config := *cfg
	config.awsID = usr.access
	config.awsSecret = usr.secret

	return config.GetClient()
}

func (c *S3Conf) Config() aws.Config {
	creds := c.getCreds()

	opts := []func(*config.LoadOptions) error{
		config.WithRegion(c.awsRegion),
		config.WithCredentialsProvider(creds),
		config.WithHTTPClient(c.httpClient),
		config.WithRetryMaxAttempts(1),
	}

	opts = append(opts, config.WithHTTPClient(c.httpClient))

	if c.checksumDisable {
		opts = append(opts,
			config.WithAPIOptions([]func(*middleware.Stack) error{v4.SwapComputePayloadSHA256ForUnsignedPayloadMiddleware}))
	}

	if c.debug {
		opts = append(opts,
			config.WithClientLogMode(aws.LogSigning|aws.LogRetries|aws.LogRequest|aws.LogResponse|aws.LogRequestEventMessage|aws.LogResponseEventMessage))
	}

	cfg, err := config.LoadDefaultConfig(
		context.TODO(), opts...)
	if err != nil {
		log.Fatalln("error:", err)
	}

	if c.endpoint != "" && c.endpoint != "aws" {
		cfg.BaseEndpoint = &c.endpoint
	}

	return cfg
}

func (c *S3Conf) UploadData(r io.Reader, bucket, object string) error {
	uploader := transfermanager.New(c.GetClient(),
		func(options *transfermanager.Options) {
			options.PartSizeBytes = c.PartSize
			options.Concurrency = c.Concurrency
		})

	upinfo := &transfermanager.UploadObjectInput{
		Body:   r,
		Bucket: &bucket,
		Key:    &object,
	}

	_, err := uploader.UploadObject(context.Background(), upinfo)
	return err
}

func (c *S3Conf) DownloadData(w io.WriterAt, bucket, object string) (int64, error) {
	downloader := transfermanager.New(c.GetClient(),
		func(options *transfermanager.Options) {
			options.PartSizeBytes = c.PartSize
			options.Concurrency = c.Concurrency
		})

	downinfo := &transfermanager.DownloadObjectInput{
		Bucket:   &bucket,
		Key:      &object,
		WriterAt: w,
	}

	out, err := downloader.DownloadObject(context.Background(), downinfo)
	if err != nil {
		return 0, err
	}

	return *out.ContentLength, nil
}

func (c *S3Conf) getAdminCommand(args ...string) []string {
	if c.tlsStatus {
		return append([]string{"admin", "--allow-insecure"}, args...)
	}

	return append([]string{"admin"}, args...)
}