feat: support dual stack requests for aws clients (#52)

* feat: enable dual stack in aws clients

* refactor: optionally set custom endpoint resolver

* feat: update sts & iam client to utilize dualstack

* update aws dependencies to remove role policy bug
This commit is contained in:
Damian Debkowski authored and GitHub committed 2024-10-28 11:50:18 -07:00
1 parent ce22c1c46a
commit 899c62ce06
18 files changed
+349 -112

No files matched your search

+10
View File
@@ -19,6 +19,9 @@ import (
// map.
type CatalogAttributes struct {
*cred.CredentialAttributes
// DualStack is used for configuring how the aws client will resolve requests.
DualStack bool
}
func getCatalogAttributes(in *structpb.Struct) (*CatalogAttributes, error) {
@@ -31,6 +34,12 @@ func getCatalogAttributes(in *structpb.Struct) (*CatalogAttributes, error) {
return nil, err
}
dualStack, err := values.GetBoolValue(in, ConstAwsDualStack, false)
if err != nil {
badFields[fmt.Sprintf("attributes.%s", ConstAwsDualStack)] = err.Error()
}
delete(unknownFields, ConstAwsDualStack)
for s := range unknownFields {
switch s {
// Ignore knownFields from CredentialAttributes
@@ -57,6 +66,7 @@ func getCatalogAttributes(in *structpb.Struct) (*CatalogAttributes, error) {
return &CatalogAttributes{
CredentialAttributes: credAttributes,
DualStack: dualStack,
}, nil
}
+16
View File
@@ -43,6 +43,22 @@ func TestGetCatalogAttributes(t *testing.T) {
},
expectedErrContains: "attributes.bar: unrecognized field, attributes.foo: unrecognized field",
},
{
name: "with dual stack",
in: &structpb.Struct{
Fields: map[string]*structpb.Value{
"region": structpb.NewStringValue("us-west-2"),
"dual_stack": structpb.NewBoolValue(true),
},
},
expected: &CatalogAttributes{
CredentialAttributes: &credential.CredentialAttributes{
Region: "us-west-2",
DisableCredentialRotation: false,
},
DualStack: true,
},
},
{
name: "default",
in: &structpb.Struct{
+3
View File
@@ -10,4 +10,7 @@ const (
// ConstInstanceNameTagKey denotes the special AWS tag key that contains an EC2
// instance's name. This field is to be matched literally.
ConstInstanceNameTagKey = "Name"
// ConstAwsDualStack is the key for the dualstack flag in the aws s3 client.
ConstAwsDualStack = "dual_stack"
)
+39 -10
View File
@@ -51,7 +51,7 @@ func (p *HostPlugin) OnCreateCatalog(ctx context.Context, req *pb.OnCreateCatalo
if err != nil {
return nil, err
}
credConfig, err := credential.GetCredentialsConfig(catalog.GetSecrets(), catalogAttributes.CredentialAttributes)
credConfig, err := credential.GetCredentialsConfig(catalog.GetSecrets(), catalogAttributes.CredentialAttributes, catalogAttributes.DualStack)
if err != nil {
return nil, err
}
@@ -83,7 +83,11 @@ func (p *HostPlugin) OnCreateCatalog(ctx context.Context, req *pb.OnCreateCatalo
}
// perform dry run to ensure we can interact with AWS as expected.
if st := dryRunValidation(ctx, catalogState); st != nil {
opts := []ec2Option{}
if catalogAttributes.DualStack {
opts = append(opts, WithDualStack(catalogAttributes.DualStack))
}
if st := dryRunValidation(ctx, catalogState, opts); st != nil {
return nil, st.Err()
}
@@ -131,6 +135,7 @@ func (p *HostPlugin) OnUpdateCatalog(ctx context.Context, req *pb.OnUpdateCatalo
credState, err := credential.AwsCredentialPersistedStateFromProto(
req.GetPersisted().GetSecrets(),
oldCatalogAttributes.CredentialAttributes,
oldCatalogAttributes.DualStack,
p.testCredStateOpts...)
if err != nil {
return nil, errors.BadRequestStatus("error loading persisted state: %s", err)
@@ -139,7 +144,7 @@ func (p *HostPlugin) OnUpdateCatalog(ctx context.Context, req *pb.OnUpdateCatalo
// Verify the incoming credentials are valid and return any errors to the
// user if they're not. Note this doesn't validate the credentials against
// AWS - it only does logical validation on the fields.
updatedCredentials, err := credential.GetCredentialsConfig(newCatalog.GetSecrets(), newCatalogAttributes.CredentialAttributes)
updatedCredentials, err := credential.GetCredentialsConfig(newCatalog.GetSecrets(), newCatalogAttributes.CredentialAttributes, newCatalogAttributes.DualStack)
if err != nil {
return nil, err
}
@@ -166,7 +171,11 @@ func (p *HostPlugin) OnUpdateCatalog(ctx context.Context, req *pb.OnUpdateCatalo
if err != nil {
return nil, errors.BadRequestStatus("error loading persisted state: %s", err)
}
if st := dryRunValidation(ctx, newCatalogState); st != nil {
opts := []ec2Option{}
if newCatalogAttributes.DualStack {
opts = append(opts, WithDualStack(newCatalogAttributes.DualStack))
}
if st := dryRunValidation(ctx, newCatalogState, opts); st != nil {
return nil, st.Err()
}
@@ -208,7 +217,11 @@ func (p *HostPlugin) OnUpdateCatalog(ctx context.Context, req *pb.OnUpdateCatalo
}
// perform dry run to ensure we can interact with AWS as expected.
if st := dryRunValidation(ctx, catalogState); st != nil {
opts := []ec2Option{}
if newCatalogAttributes.DualStack {
opts = append(opts, WithDualStack(newCatalogAttributes.DualStack))
}
if st := dryRunValidation(ctx, catalogState, opts); st != nil {
return nil, st.Err()
}
@@ -248,6 +261,7 @@ func (p *HostPlugin) OnDeleteCatalog(ctx context.Context, req *pb.OnDeleteCatalo
credState, err := credential.AwsCredentialPersistedStateFromProto(
req.GetPersisted().GetSecrets(),
catalogAttributes.CredentialAttributes,
catalogAttributes.DualStack,
p.testCredStateOpts...)
if err != nil {
return nil, errors.BadRequestStatus("error loading persisted state: %s", err)
@@ -324,6 +338,7 @@ func (p *HostPlugin) OnCreateSet(ctx context.Context, req *pb.OnCreateSetRequest
credState, err := credential.AwsCredentialPersistedStateFromProto(
req.GetPersisted().GetSecrets(),
catalogAttributes.CredentialAttributes,
catalogAttributes.DualStack,
p.testCredStateOpts...)
if err != nil {
return nil, errors.BadRequestStatus("error loading persisted state: %s", err)
@@ -356,7 +371,11 @@ func (p *HostPlugin) OnCreateSet(ctx context.Context, req *pb.OnCreateSetRequest
return nil, errors.BadRequestStatus("error building set filters: %s", err)
}
if st := dryRunValidation(ctx, catalogState, describeInstanceFilters...); st != nil {
opts := []ec2Option{}
if catalogAttributes.DualStack {
opts = append(opts, WithDualStack(catalogAttributes.DualStack))
}
if st := dryRunValidation(ctx, catalogState, opts, describeInstanceFilters...); st != nil {
return nil, st.Err()
}
return &pb.OnCreateSetResponse{}, nil
@@ -382,6 +401,7 @@ func (p *HostPlugin) OnUpdateSet(ctx context.Context, req *pb.OnUpdateSetRequest
credState, err := credential.AwsCredentialPersistedStateFromProto(
req.GetPersisted().GetSecrets(),
catalogAttributes.CredentialAttributes,
catalogAttributes.DualStack,
p.testCredStateOpts...)
if err != nil {
return nil, errors.BadRequestStatus("error loading persisted state: %s", err)
@@ -416,7 +436,11 @@ func (p *HostPlugin) OnUpdateSet(ctx context.Context, req *pb.OnUpdateSetRequest
return nil, errors.BadRequestStatus("error building set filters: %s", err)
}
if st := dryRunValidation(ctx, catalogState, describeInstanceFilters...); st != nil {
opts := []ec2Option{}
if catalogAttributes.DualStack {
opts = append(opts, WithDualStack(catalogAttributes.DualStack))
}
if st := dryRunValidation(ctx, catalogState, opts, describeInstanceFilters...); st != nil {
return nil, st.Err()
}
@@ -450,6 +474,7 @@ func (p *HostPlugin) ListHosts(ctx context.Context, req *pb.ListHostsRequest) (*
credState, err := credential.AwsCredentialPersistedStateFromProto(
req.GetPersisted().GetSecrets(),
catalogAttributes.CredentialAttributes,
catalogAttributes.DualStack,
p.testCredStateOpts...)
if err != nil {
return nil, errors.BadRequestStatus("error loading persisted state: %s", err)
@@ -502,7 +527,11 @@ func (p *HostPlugin) ListHosts(ctx context.Context, req *pb.ListHostsRequest) (*
}
}
ec2Client, err := catalogState.EC2Client(ctx)
opts := []ec2Option{}
if catalogAttributes.DualStack {
opts = append(opts, WithDualStack(catalogAttributes.DualStack))
}
ec2Client, err := catalogState.EC2Client(ctx, opts...)
if err != nil {
return nil, errors.BadRequestStatus("error getting EC2 client: %s", err)
}
@@ -673,12 +702,12 @@ func awsInstanceToHost(instance types.Instance) (*pb.ListHostsResponseHost, erro
// credentials, the host listing functionality as well as the filters, if any
// are passed in. This function can therefore be used for both host catalog and
// host set validation.
func dryRunValidation(ctx context.Context, state *awsCatalogPersistedState, filters ...types.Filter) *status.Status {
func dryRunValidation(ctx context.Context, state *awsCatalogPersistedState, ec2Opts []ec2Option, filters ...types.Filter) *status.Status {
if state == nil {
return status.New(codes.InvalidArgument, "persisted state is required")
}
ec2Client, err := state.EC2Client(ctx)
ec2Client, err := state.EC2Client(ctx, ec2Opts...)
if err != nil {
return status.New(codes.InvalidArgument, fmt.Sprintf("error getting EC2 client: %s", err))
}
+5 -5
View File
@@ -2747,8 +2747,8 @@ func TestAppendDistinct(t *testing.T) {
}
func TestDryRunValidation(t *testing.T) {
t.Run("nilState", func(t *testing.T) {
st := dryRunValidation(context.Background(), nil)
t.Run("nil credential state", func(t *testing.T) {
st := dryRunValidation(context.Background(), nil, nil)
require.NotNil(t, st)
require.Equal(t, codes.InvalidArgument.String(), st.Code().String())
require.Equal(t, "persisted state is required", st.Message())
@@ -2762,7 +2762,7 @@ func TestDryRunValidation(t *testing.T) {
testEC2APIFunc: func(...aws.Config) (EC2API, error) {
return nil, fmt.Errorf("oops ec2 client err")
},
})
}, []ec2Option{})
require.NotNil(t, st)
require.Equal(t, codes.InvalidArgument.String(), st.Code().String())
require.Equal(t, "error getting EC2 client: oops ec2 client err", st.Message())
@@ -2777,7 +2777,7 @@ func TestDryRunValidation(t *testing.T) {
},
},
testEC2APIFunc: newTestMockEC2(nil, testMockEC2WithDescribeInstancesError(fmt.Errorf("oops describe instances error"))),
})
}, []ec2Option{})
require.NotNil(t, st)
require.Equal(t, codes.FailedPrecondition.String(), st.Code().String())
require.Equal(t, "aws describe instances failed: oops describe instances error", st.Message())
@@ -2792,7 +2792,7 @@ func TestDryRunValidation(t *testing.T) {
},
},
testEC2APIFunc: newTestMockEC2(nil, testMockEC2WithDescribeInstancesOutput(&ec2.DescribeInstancesOutput{})),
})
}, []ec2Option{})
require.Nil(t, st)
})
}
+57 -16
View File
@@ -7,12 +7,10 @@ import (
"context"
"errors"
"fmt"
"net"
"net/http"
"time"
"github.com/aws/aws-sdk-go-v2/aws"
"github.com/aws/aws-sdk-go-v2/service/ec2"
smithyEndpoints "github.com/aws/smithy-go/endpoints"
"github.com/hashicorp/boundary-plugin-aws/internal/credential"
pb "github.com/hashicorp/boundary/sdk/pbs/plugin"
"google.golang.org/protobuf/types/known/structpb"
@@ -22,17 +20,6 @@ type EC2API interface {
DescribeInstances(context.Context, *ec2.DescribeInstancesInput, ...func(*ec2.Options)) (*ec2.DescribeInstancesOutput, error)
}
var customClient = &http.Client{
Transport: &http.Transport{
Proxy: http.ProxyFromEnvironment,
DialContext: (&net.Dialer{
Timeout: 30 * time.Second,
KeepAlive: 30 * time.Second,
DualStack: true,
}).DialContext,
},
}
type awsCatalogPersistedState struct {
*credential.AwsCredentialPersistedState
@@ -86,7 +73,11 @@ func (s *awsCatalogPersistedState) toProto() (*pb.HostCatalogPersisted, error) {
// EC2Client returns a configured EC2 client based on the session
// information stored in the state.
func (s *awsCatalogPersistedState) EC2Client(ctx context.Context) (EC2API, error) {
func (s *awsCatalogPersistedState) EC2Client(ctx context.Context, opt ...ec2Option) (EC2API, error) {
opts, err := getOpts(opt...)
if err != nil {
return nil, fmt.Errorf("error parsing options when fetching EC2 client: %w", err)
}
awsCfg, err := s.GenerateCredentialChain(ctx)
if err != nil {
return nil, fmt.Errorf("error getting AWS configuration when fetching EC2 client: %w", err)
@@ -94,8 +85,58 @@ func (s *awsCatalogPersistedState) EC2Client(ctx context.Context) (EC2API, error
if awsCfg == nil {
return nil, fmt.Errorf("nil aws configuration")
}
var ec2Opts []func(*ec2.Options)
if opts.withDualStack {
ec2Opts = append(ec2Opts, ec2.WithEndpointResolverV2(&endpointResolver{
dualStack: opts.withDualStack,
}))
}
if s.testEC2APIFunc != nil {
return s.testEC2APIFunc(*awsCfg)
}
return ec2.NewFromConfig(*awsCfg), nil
return ec2.NewFromConfig(*awsCfg, ec2Opts...), nil
}
// getOpts iterates the inbound ec2Options and returns a struct
func getOpts(opt ...ec2Option) (ec2Options, error) {
opts := getDefaultEc2Options()
for _, o := range opt {
if o == nil {
continue
}
if err := o(&opts); err != nil {
return ec2Options{}, err
}
}
return opts, nil
}
// ec2Option - how ec2Options are passed as arguments
type ec2Option func(*ec2Options) error
// options = how options are represented
type ec2Options struct {
withDualStack bool
}
func getDefaultEc2Options() ec2Options {
return ec2Options{}
}
// WithDualStack sets the dual stack resolver
func WithDualStack(with bool) ec2Option {
return func(o *ec2Options) error {
o.withDualStack = with
return nil
}
}
type endpointResolver struct {
dualStack bool
}
func (e *endpointResolver) ResolveEndpoint(ctx context.Context, params ec2.EndpointParameters) (resolver smithyEndpoints.Endpoint, err error) {
params.UseDualStack = aws.Bool(e.dualStack)
resolver, err = ec2.NewDefaultEndpointResolverV2().ResolveEndpoint(ctx, params)
return
}
+10
View File
@@ -19,6 +19,9 @@ type StorageAttributes struct {
// EndpointUrl is used for configuring how the aws client will resolve requests.
EndpointUrl string
// DualStack is used for configuring how the aws client will resolve requests.
DualStack bool
}
func getStorageAttributes(in *structpb.Struct) (*StorageAttributes, error) {
@@ -37,6 +40,12 @@ func getStorageAttributes(in *structpb.Struct) (*StorageAttributes, error) {
}
delete(unknownFields, ConstAwsEndpointUrl)
dualStack, err := values.GetBoolValue(in, ConstAwsDualStack, false)
if err != nil {
badFields[fmt.Sprintf("attributes.%s", ConstAwsDualStack)] = err.Error()
}
delete(unknownFields, ConstAwsDualStack)
for s := range unknownFields {
switch s {
// Ignore knownFields from CredentialAttributes
@@ -64,5 +73,6 @@ func getStorageAttributes(in *structpb.Struct) (*StorageAttributes, error) {
return &StorageAttributes{
CredentialAttributes: credAttributes,
EndpointUrl: endpointUrl,
DualStack: dualStack,
}, nil
}
+15
View File
@@ -41,6 +41,21 @@ func TestGetStorageAttributes(t *testing.T) {
},
expectedErrContains: "attributes.bar: unrecognized field, attributes.foo: unrecognized field",
},
{
name: "good w/ dualstack",
in: &structpb.Struct{
Fields: map[string]*structpb.Value{
"region": structpb.NewStringValue("us-west-2"),
"dual_stack": structpb.NewBoolValue(true),
},
},
expected: &StorageAttributes{
CredentialAttributes: &cred.CredentialAttributes{
Region: "us-west-2",
},
DualStack: true,
},
},
{
name: "good w/ endpoint",
in: &structpb.Struct{
+3
View File
@@ -7,6 +7,9 @@ const (
// ConstAwsEndpointUrl is the key for the endpoint url in the aws s3 client.
ConstAwsEndpointUrl = "endpoint_url"
// ConstAwsDualStack is the key for the dualstack flag in the aws s3 client.
ConstAwsDualStack = "dual_stack"
// defaultStreamChunkSize is the recommened chunk size for sending data through a stream
defaultStreamChunkSize = 65536 // 64 KiB
)
+24 -10
View File
@@ -64,7 +64,7 @@ func (p *StoragePlugin) OnCreateStorageBucket(ctx context.Context, req *pb.OnCre
return nil, err
}
credConfig, err := cred.GetCredentialsConfig(bucket.GetSecrets(), storageAttributes.CredentialAttributes)
credConfig, err := cred.GetCredentialsConfig(bucket.GetSecrets(), storageAttributes.CredentialAttributes, storageAttributes.DualStack)
if err != nil {
return nil, err
}
@@ -147,6 +147,7 @@ func (p *StoragePlugin) OnUpdateStorageBucket(ctx context.Context, req *pb.OnUpd
credState, err := cred.AwsCredentialPersistedStateFromProto(
req.GetPersisted().GetData(),
oldStorageAttributes.CredentialAttributes,
oldStorageAttributes.DualStack,
p.testCredStateOpts...)
if err != nil {
return nil, errors.BadRequestStatus("error loading persisted state: %s", err)
@@ -155,7 +156,7 @@ func (p *StoragePlugin) OnUpdateStorageBucket(ctx context.Context, req *pb.OnUpd
// Verify the incoming credentials are valid and return any errors to the
// user if they're not. Note this doesn't validate the credentials against
// AWS - it only does logical validation on the fields.
updatedCredentials, err := cred.GetCredentialsConfig(newBucket.GetSecrets(), newStorageAttributes.CredentialAttributes)
updatedCredentials, err := cred.GetCredentialsConfig(newBucket.GetSecrets(), newStorageAttributes.CredentialAttributes, newStorageAttributes.DualStack)
if err != nil {
return nil, err
}
@@ -264,6 +265,7 @@ func (p *StoragePlugin) OnDeleteStorageBucket(ctx context.Context, req *pb.OnDel
credState, err := cred.AwsCredentialPersistedStateFromProto(
req.GetPersisted().GetData(),
storageAttributes.CredentialAttributes,
storageAttributes.DualStack,
p.testCredStateOpts...)
if err != nil {
return nil, errors.BadRequestStatus("error loading persisted state: %s", err)
@@ -318,7 +320,7 @@ func (p *StoragePlugin) HeadObject(ctx context.Context, req *pb.HeadObjectReques
return nil, err
}
credConfig, err := cred.GetCredentialsConfig(bucket.GetSecrets(), storageAttributes.CredentialAttributes)
credConfig, err := cred.GetCredentialsConfig(bucket.GetSecrets(), storageAttributes.CredentialAttributes, storageAttributes.DualStack)
if err != nil {
return nil, err
}
@@ -346,6 +348,9 @@ func (p *StoragePlugin) HeadObject(ctx context.Context, req *pb.HeadObjectReques
if storageAttributes.EndpointUrl != "" {
opts = append(opts, WithEndpoint(storageAttributes.EndpointUrl))
}
if storageAttributes.DualStack {
opts = append(opts, WithDualStack(storageAttributes.DualStack))
}
s3Client, err := storageState.S3Client(ctx, opts...)
if err != nil {
return nil, errors.BadRequestStatus("error getting S3 client: %s", err)
@@ -360,7 +365,7 @@ func (p *StoragePlugin) HeadObject(ctx context.Context, req *pb.HeadObjectReques
return nil, parseS3Error("head object", err, req).Err()
}
return &pb.HeadObjectResponse{
ContentLength: resp.ContentLength,
ContentLength: aws.ToInt64(resp.ContentLength),
LastModified: timestamppb.New(*resp.LastModified),
}, nil
}
@@ -386,7 +391,7 @@ func (p *StoragePlugin) ValidatePermissions(ctx context.Context, req *pb.Validat
return nil, err
}
credConfig, err := cred.GetCredentialsConfig(bucket.GetSecrets(), storageAttributes.CredentialAttributes)
credConfig, err := cred.GetCredentialsConfig(bucket.GetSecrets(), storageAttributes.CredentialAttributes, storageAttributes.DualStack)
if err != nil {
return nil, err
}
@@ -444,7 +449,7 @@ func (p *StoragePlugin) GetObject(req *pb.GetObjectRequest, stream pb.StoragePlu
return err
}
credConfig, err := cred.GetCredentialsConfig(bucket.GetSecrets(), storageAttributes.CredentialAttributes)
credConfig, err := cred.GetCredentialsConfig(bucket.GetSecrets(), storageAttributes.CredentialAttributes, storageAttributes.DualStack)
if err != nil {
return err
}
@@ -472,6 +477,9 @@ func (p *StoragePlugin) GetObject(req *pb.GetObjectRequest, stream pb.StoragePlu
if storageAttributes.EndpointUrl != "" {
opts = append(opts, WithEndpoint(storageAttributes.EndpointUrl))
}
if storageAttributes.DualStack {
opts = append(opts, WithDualStack(storageAttributes.DualStack))
}
s3Client, err := storageState.S3Client(stream.Context(), opts...)
if err != nil {
return errors.BadRequestStatus("error getting S3 client: %s", err)
@@ -555,7 +563,7 @@ func (p *StoragePlugin) PutObject(ctx context.Context, req *pb.PutObjectRequest)
return nil, err
}
credConfig, err := cred.GetCredentialsConfig(bucket.GetSecrets(), storageAttributes.CredentialAttributes)
credConfig, err := cred.GetCredentialsConfig(bucket.GetSecrets(), storageAttributes.CredentialAttributes, storageAttributes.DualStack)
if err != nil {
return nil, err
}
@@ -583,6 +591,9 @@ func (p *StoragePlugin) PutObject(ctx context.Context, req *pb.PutObjectRequest)
if storageAttributes.EndpointUrl != "" {
opts = append(opts, WithEndpoint(storageAttributes.EndpointUrl))
}
if storageAttributes.DualStack {
opts = append(opts, WithDualStack(storageAttributes.DualStack))
}
s3Client, err := storageState.S3Client(ctx, opts...)
if err != nil {
return nil, errors.BadRequestStatus("error getting S3 client: %s", err)
@@ -650,7 +661,7 @@ func (p *StoragePlugin) DeleteObjects(ctx context.Context, req *pb.DeleteObjects
return nil, err
}
credConfig, err := cred.GetCredentialsConfig(bucket.GetSecrets(), storageAttributes.CredentialAttributes)
credConfig, err := cred.GetCredentialsConfig(bucket.GetSecrets(), storageAttributes.CredentialAttributes, storageAttributes.DualStack)
if err != nil {
return nil, err
}
@@ -712,13 +723,13 @@ func (p *StoragePlugin) DeleteObjects(ctx context.Context, req *pb.DeleteObjects
res, err := client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
Bucket: aws.String(bucket.GetBucketName()),
Prefix: aws.String(prefix),
MaxKeys: maxkeys,
MaxKeys: aws.Int32(maxkeys),
ContinuationToken: conToken,
})
if err != nil {
return nil, parseS3Error("list objects", err, req).Err()
}
truncated = res.IsTruncated
truncated = aws.ToBool(res.IsTruncated)
conToken = res.NextContinuationToken
for _, o := range res.Contents {
objects = append(objects, types.ObjectIdentifier{
@@ -769,6 +780,9 @@ func dryRunValidation(ctx context.Context, state *awsStoragePersistedState, attr
if attrs.EndpointUrl != "" {
opts = append(opts, WithEndpoint(attrs.EndpointUrl))
}
if attrs.DualStack {
opts = append(opts, WithDualStack(attrs.DualStack))
}
client, err := state.S3Client(ctx, opts...)
if err != nil {
+6 -6
View File
@@ -2241,7 +2241,7 @@ func TestStoragePlugin_HeadObject(t *testing.T) {
newTestMockS3(
nil,
testMockS3WithHeadObjectOutput(&s3.HeadObjectOutput{
ContentLength: 1024,
ContentLength: aws.Int64(1024),
LastModified: aws.Time(createTime(t, "2006-01-02T15:04:05Z")),
}),
),
@@ -2272,7 +2272,7 @@ func TestStoragePlugin_HeadObject(t *testing.T) {
newTestMockS3(
nil,
testMockS3WithHeadObjectOutput(&s3.HeadObjectOutput{
ContentLength: 1024,
ContentLength: aws.Int64(1024),
LastModified: aws.Time(createTime(t, "2006-01-02T15:04:05Z")),
}),
),
@@ -3685,7 +3685,7 @@ func TestStoragePlugin_DeleteObjects(t *testing.T) {
newTestMockS3(
nil,
testMockS3WithListObjectsV2Output(&s3.ListObjectsV2Output{
IsTruncated: false,
IsTruncated: aws.Bool(false),
Contents: []s3types.Object{
{
Key: aws.String("abc/abc"),
@@ -3763,7 +3763,7 @@ func TestStoragePlugin_DeleteObjects(t *testing.T) {
newTestMockS3(
nil,
testMockS3WithListObjectsV2Output(&s3.ListObjectsV2Output{
IsTruncated: false,
IsTruncated: aws.Bool(false),
Contents: []s3types.Object{
s3types.Object{
Key: aws.String("abc/abc"),
@@ -3808,7 +3808,7 @@ func TestStoragePlugin_DeleteObjects(t *testing.T) {
newTestMockS3(
nil,
testMockS3WithListObjectsV2Output(&s3.ListObjectsV2Output{
IsTruncated: false,
IsTruncated: aws.Bool(false),
Contents: []s3types.Object{
{Key: aws.String("abc/abc1")},
{Key: aws.String("abc/abc2")},
@@ -3836,7 +3836,7 @@ func TestStoragePlugin_DeleteObjects(t *testing.T) {
newTestMockS3(
nil,
testMockS3WithListObjectsV2Output(&s3.ListObjectsV2Output{
IsTruncated: false,
IsTruncated: aws.Bool(false),
Contents: []s3types.Object{},
}),
// no deleted response since it shouldn't be called
+16 -20
View File
@@ -7,10 +7,7 @@ import (
"context"
"errors"
"fmt"
"net"
"net/http"
"net/url"
"time"
"github.com/aws/aws-sdk-go-v2/aws"
"github.com/aws/aws-sdk-go-v2/service/s3"
@@ -29,17 +26,6 @@ type S3API interface {
ListObjectsV2(ctx context.Context, params *s3.ListObjectsV2Input, optFns ...func(*s3.Options)) (*s3.ListObjectsV2Output, error)
}
var customClient = &http.Client{
Transport: &http.Transport{
Proxy: http.ProxyFromEnvironment,
DialContext: (&net.Dialer{
Timeout: 30 * time.Second,
KeepAlive: 30 * time.Second,
DualStack: true,
}).DialContext,
},
}
type awsStoragePersistedState struct {
*credential.AwsCredentialPersistedState
@@ -106,9 +92,10 @@ func (s *awsStoragePersistedState) S3Client(ctx context.Context, opt ...s3Option
return nil, fmt.Errorf("nil aws configuration")
}
var s3Opts []func(*s3.Options)
if opts.withEndpoint != "" {
if opts.withEndpoint != "" || opts.withDualStack {
s3Opts = append(s3Opts, s3.WithEndpointResolverV2(&endpointResolver{
endpoint: opts.withEndpoint,
endpoint: opts.withEndpoint,
dualStack: opts.withDualStack,
}))
}
if s.testS3APIFunc != nil {
@@ -136,7 +123,8 @@ type s3Option func(*s3Options) error
// options = how options are represented
type s3Options struct {
withEndpoint string
withEndpoint string
withDualStack bool
}
func getDefaultS3Options() s3Options {
@@ -151,18 +139,27 @@ func WithEndpoint(with string) s3Option {
}
}
// WithDualStack sets the dual stack resolver
func WithDualStack(with bool) s3Option {
return func(o *s3Options) error {
o.withDualStack = with
return nil
}
}
type endpointResolver struct {
endpoint string
endpoint string
dualStack bool
}
func (e *endpointResolver) ResolveEndpoint(ctx context.Context, params s3.EndpointParameters) (resolver smithyEndpoints.Endpoint, err error) {
var uri *url.URL
params.UseDualStack = aws.Bool(e.dualStack)
resolver, err = s3.NewDefaultEndpointResolverV2().ResolveEndpoint(ctx, params)
if err != nil {
return
}
if e.endpoint == "" {
// TODO: not sure if we should set an error or ignore the missing endpoint value and continue with the default resolver
return
}
uri, err = url.Parse(e.endpoint)
@@ -170,7 +167,6 @@ func (e *endpointResolver) ResolveEndpoint(ctx context.Context, params s3.Endpoi
return
}
if uri == nil {
// TODO: not sure if we should set an error or ignore the missing uri value and continue with the default resolver
return
}
resolver = smithyEndpoints.Endpoint{