// 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 websiteScheme string websiteDomain string websitePort string hostStyle bool checksumDisable bool PartSize int64 Concurrency int debug bool versioningEnabled bool azureTests bool windowsTests bool sidecarTests 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, CheckRedirect: func(req *http.Request, via []*http.Request) error { return http.ErrUseLastResponse }, } 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 WithWebsiteScheme(scheme string) Option { return func(s *S3Conf) { s.websiteScheme = scheme } } func WithWebsiteDomain(d string) Option { return func(s *S3Conf) { s.websiteDomain = d } } func WithWebsitePort(p string) Option { return func(s *S3Conf) { s.websitePort = p } } 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 WithWindowsMode() Option { return func(s *S3Conf) { s.windowsTests = true } } func WithSidecarMode() Option { return func(s *S3Conf) { s.sidecarTests = 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...) }