Migrate session package from v1 to v2

Signed-off-by: Pankaj Walke <[email protected]>

Author: punkwalker

go.mod

@@ -20,6 +20,7 @@ require (
 	github.com/aws/aws-sdk-go v1.55.5
 	github.com/aws/aws-sdk-go-v2 v1.30.3
 	github.com/aws/aws-sdk-go-v2/config v1.27.11
+	github.com/aws/aws-sdk-go-v2/credentials v1.17.11
 	github.com/aws/aws-sdk-go-v2/service/s3 v1.53.1
 	github.com/aws/aws-sdk-go-v2/service/sts v1.28.6
 	github.com/aws/smithy-go v1.20.3
@@ -80,7 +81,6 @@ require (
 	github.com/antlr/antlr4/runtime/Go/antlr/v4 v4.0.0-20230305170008-8188dc5388df // indirect
 	github.com/asaskevich/govalidator v0.0.0-20200428143746-21a406dcc535 // indirect
 	github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.6.2 // indirect
-	github.com/aws/aws-sdk-go-v2/credentials v1.17.11 // indirect
 	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.1 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.15 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.15 // indirect

pkg/cloud/identityv2/identity.go

@@ -0,0 +1,180 @@
+/*
+Copyright 2021 The Kubernetes Authors.
+
+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 identity provides the AWSPrincipalTypeProvider interface and its implementations.
+package identityv2
+
+import (
+	"bytes"
+	"context"
+	"crypto/sha256"
+	"encoding/gob"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/credentials"
+	"github.com/aws/aws-sdk-go-v2/credentials/stscreds"
+	"github.com/aws/aws-sdk-go-v2/service/sts"
+	corev1 "k8s.io/api/core/v1"
+
+	infrav1 "sigs.k8s.io/cluster-api-provider-aws/v2/api/v1beta2"
+	"sigs.k8s.io/cluster-api-provider-aws/v2/pkg/logger"
+)
+
+// AWSPrincipalTypeProvider defines the interface for AWS Principal Type Provider.
+type AWSPrincipalTypeProvider interface {
+	aws.CredentialsProvider
+	// Hash returns a unique hash of the data forming the V2 credentials
+	// for this Principal
+	Hash() (string, error)
+	Name() string
+}
+
+// NewAWSStaticPrincipalTypeProvider will create a new AWSStaticPrincipalTypeProvider from a given AWSClusterStaticIdentity.
+func NewAWSStaticPrincipalTypeProvider(identity *infrav1.AWSClusterStaticIdentity, secret *corev1.Secret) *AWSStaticPrincipalTypeProvider {
+	accessKeyID := string(secret.Data["AccessKeyID"])
+	secretAccessKey := string(secret.Data["SecretAccessKey"])
+	sessionToken := string(secret.Data["SessionToken"])
+
+	credProvider := credentials.NewStaticCredentialsProvider(accessKeyID, secretAccessKey, sessionToken)
+	credCache := aws.NewCredentialsCache(credProvider)
+	return &AWSStaticPrincipalTypeProvider{
+		Principal:       identity,
+		credentials:     credCache,
+		AccessKeyID:     accessKeyID,
+		SecretAccessKey: secretAccessKey,
+		SessionToken:    sessionToken,
+	}
+}
+
+// GetAssumeRoleCredentialsCache will return the CredentialsCache of a given AWSRolePrincipalTypeProvider.
+func GetAssumeRoleCredentialsCache(roleIdentityProvider *AWSRolePrincipalTypeProvider, optFns []func(*config.LoadOptions) error) (*aws.CredentialsCache, error) {
+	cfg, err := config.LoadDefaultConfig(context.TODO(), optFns...)
+	if err != nil {
+		return nil, err
+	}
+
+	stsClient := sts.NewFromConfig(cfg)
+	credsProvider := stscreds.NewAssumeRoleProvider(stsClient, roleIdentityProvider.Principal.Spec.RoleArn, func(o *stscreds.AssumeRoleOptions) {
+		if roleIdentityProvider.Principal.Spec.ExternalID != "" {
+			o.ExternalID = aws.String(roleIdentityProvider.Principal.Spec.ExternalID)
+		}
+		o.RoleSessionName = roleIdentityProvider.Principal.Spec.SessionName
+		if roleIdentityProvider.Principal.Spec.InlinePolicy != "" {
+			o.Policy = aws.String(roleIdentityProvider.Principal.Spec.InlinePolicy)
+		}
+		o.Duration = time.Duration(roleIdentityProvider.Principal.Spec.DurationSeconds) * time.Second
+		// For testing
+		if roleIdentityProvider.stsClient != nil {
+			o.Client = roleIdentityProvider.stsClient
+		}
+	})
+
+	return aws.NewCredentialsCache(credsProvider), nil
+}
+
+// NewAWSRolePrincipalTypeProvider will create a new AWSRolePrincipalTypeProvider from an AWSClusterRoleIdentity.
+func NewAWSRolePrincipalTypeProvider(identity *infrav1.AWSClusterRoleIdentity, sourceProvider AWSPrincipalTypeProvider, region string, log logger.Wrapper) *AWSRolePrincipalTypeProvider {
+	return &AWSRolePrincipalTypeProvider{
+		credentials:    nil,
+		stsClient:      nil,
+		region:         region,
+		Principal:      identity,
+		sourceProvider: sourceProvider,
+		log:            log.WithName("AWSRolePrincipalTypeProvider"),
+	}
+}
+
+// AWSStaticPrincipalTypeProvider defines the specs for a static AWSPrincipalTypeProvider.
+type AWSStaticPrincipalTypeProvider struct {
+	Principal   *infrav1.AWSClusterStaticIdentity
+	credentials *aws.CredentialsCache
+	// these are for tests :/
+	AccessKeyID     string
+	SecretAccessKey string
+	SessionToken    string
+}
+
+// Hash returns the byte encoded AWSStaticPrincipalTypeProvider.
+func (p *AWSStaticPrincipalTypeProvider) Hash() (string, error) {
+	var roleIdentityValue bytes.Buffer
+	err := gob.NewEncoder(&roleIdentityValue).Encode(p)
+	if err != nil {
+		return "", err
+	}
+	hash := sha256.New()
+	return string(hash.Sum(roleIdentityValue.Bytes())), nil
+}
+
+// Retrieve returns the credential values for the AWSStaticPrincipalTypeProvider.
+func (p *AWSStaticPrincipalTypeProvider) Retrieve(ctx context.Context) (aws.Credentials, error) {
+	return p.credentials.Retrieve(ctx)
+}
+
+// Name returns the name of the AWSStaticPrincipalTypeProvider.
+func (p *AWSStaticPrincipalTypeProvider) Name() string {
+	return p.Principal.Name
+}
+
+// AWSRolePrincipalTypeProvider defines the specs for a AWSPrincipalTypeProvider with a role.
+type AWSRolePrincipalTypeProvider struct {
+	Principal      *infrav1.AWSClusterRoleIdentity
+	credentials    *aws.CredentialsCache
+	region         string
+	sourceProvider AWSPrincipalTypeProvider
+	log            logger.Wrapper
+	stsClient      stscreds.AssumeRoleAPIClient
+}
+
+// Hash returns the byte encoded AWSRolePrincipalTypeProvider.
+func (p *AWSRolePrincipalTypeProvider) Hash() (string, error) {
+	var roleIdentityValue bytes.Buffer
+	err := gob.NewEncoder(&roleIdentityValue).Encode(p)
+	if err != nil {
+		return "", err
+	}
+	hash := sha256.New()
+	return string(hash.Sum(roleIdentityValue.Bytes())), nil
+}
+
+// Name returns the name of the AWSRolePrincipalTypeProvider.
+func (p *AWSRolePrincipalTypeProvider) Name() string {
+	return p.Principal.Name
+}
+
+// Retrieve returns the credential values for the AWSRolePrincipalTypeProvider.
+func (p *AWSRolePrincipalTypeProvider) Retrieve(ctx context.Context) (aws.Credentials, error) {
+
+	if p.credentials == nil {
+		optFns := []func(*config.LoadOptions) error{config.WithRegion(p.region)}
+		if p.sourceProvider != nil {
+			sourceCreds, err := p.sourceProvider.Retrieve(ctx)
+			if err != nil {
+				return aws.Credentials{}, err
+			}
+			optFns = append(optFns, config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(sourceCreds.AccessKeyID, sourceCreds.SecretAccessKey, sourceCreds.SessionToken)))
+		}
+
+		creds, err := GetAssumeRoleCredentialsCache(p, optFns)
+		if err != nil {
+			return aws.Credentials{}, err
+		}
+		// Update credentials
+		p.credentials = creds
+	}
+	return p.credentials.Retrieve(ctx)
+}

pkg/cloud/interfaces.go

@@ -18,6 +18,7 @@ limitations under the License.
 package cloud
 
 import (
+	awsv2 "github.com/aws/aws-sdk-go-v2/aws"
 	awsclient "github.com/aws/aws-sdk-go/aws/client"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -32,6 +33,7 @@ import (
 // Session represents an AWS session.
 type Session interface {
 	Session() awsclient.ConfigProvider
+	SessionV2() awsv2.Config
 	ServiceLimiter(service string) *throttle.ServiceLimiter
 }
 

pkg/cloud/logs/logs.go

@@ -19,7 +19,6 @@ package logs
 
 import (
 	awsv2 "github.com/aws/aws-sdk-go-v2/aws"
-	"github.com/aws/aws-sdk-go-v2/config"
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/go-logr/logr"
 )
@@ -43,16 +42,16 @@ func GetAWSLogLevel(logger logr.Logger) aws.LogLevelType {
 }
 
 // GetAWSLogLevelV2 will return the log level of an AWS Logger.
-func GetAWSLogLevelV2(logger logr.Logger) config.LoadOptionsFunc {
+func GetAWSLogLevelV2(logger logr.Logger) awsv2.ClientLogMode {
 	if logger.V(logWithHTTPBody).Enabled() {
-		return config.WithClientLogMode(awsv2.LogRequestWithBody | awsv2.LogResponseWithBody)
+		return awsv2.LogRequestWithBody | awsv2.LogResponseWithBody
 	}
 
 	if logger.V(logWithHTTPHeader).Enabled() {
-		return config.WithClientLogMode(awsv2.LogRequest | awsv2.LogResponse)
+		return awsv2.LogRequest | awsv2.LogResponse
 	}
 
-	return nil
+	return awsv2.LogRequestEventMessage
 }
 
 // NewWrapLogr will create an AWS Logger wrapper.

pkg/cloud/scope/clients.go

@@ -22,7 +22,6 @@ import (
 	"fmt"
 
 	awsv2middleware "github.com/aws/aws-sdk-go-v2/aws/middleware"
-	"github.com/aws/aws-sdk-go-v2/config"
 	"github.com/aws/aws-sdk-go-v2/service/s3"
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/awserr"
@@ -209,31 +208,30 @@ func NewSSMClient(scopeUser cloud.ScopeUsage, session cloud.Session, logger logg
 
 // NewS3Client creates a new S3 API client for a given session.
 func NewS3Client(scopeUser cloud.ScopeUsage, session cloud.Session, logger logger.Wrapper, target runtime.Object) *s3.Client {
-	// TODO: Incorporate session into loading config
-	optFns := []func(*config.LoadOptions) error{config.WithLogger(logger.GetAWSLogger())}
-	if awslogs.GetAWSLogLevelV2(logger.GetLogger()) != nil {
-		optFns = append(optFns, awslogs.GetAWSLogLevelV2(logger.GetLogger()))
-	}
-	cfg, err := config.LoadDefaultConfig(context.TODO(), optFns...)
-	if err != nil {
-		panic(err)
-	}
-
-	cfg.APIOptions = append(
-		cfg.APIOptions,
-		func(stack *middleware.Stack) error {
-			return stack.Build.Add(getUserAgentHandlerV2(), middleware.Before)
+	// TODO: Implement EndpointResolverV2 for Service Endpoints
+	cfg := session.SessionV2()
+	s3Opts := []func(*s3.Options){
+		func(o *s3.Options) {
+			o.Logger = logger.GetAWSLogger()
 		},
-		func(stack *middleware.Stack) error {
-			return stack.Deserialize.Add(recordAWSPermissionsIssueV2(target), middleware.After)
+		func(o *s3.Options) {
+			o.ClientLogMode = awslogs.GetAWSLogLevelV2(logger.GetLogger())
 		},
-	)
+		s3.WithAPIOptions(
+			func(stack *middleware.Stack) error {
+				return stack.Build.Add(getUserAgentHandlerV2(), middleware.Before)
+			},
+			func(stack *middleware.Stack) error {
+				return stack.Deserialize.Add(recordAWSPermissionsIssueV2(target), middleware.After)
+			},
+		),
+	}
 	// TODO: https://docs.aws.amazon.com/sdk-for-go/v2/developer-guide/sdk-timing.html
 	// cfg.APIOptions = append(cfg.APIOptions, func(stack *middleware.Stack) error {
-	//	return stack.Deserialize.Add(awsmetrics.CaptureRequestMetricsV2(scopeUser.ControllerName()), middleware.Before)
+	// 	return stack.Deserialize.Add(awsmetrics.CaptureRequestMetricsV2(scopeUser.ControllerName()), middleware.Before)
 	// })
 
-	return s3.NewFromConfig(cfg)
+	return s3.NewFromConfig(cfg, s3Opts...)
 }
 
 func recordAWSPermissionsIssue(target runtime.Object) func(r *request.Request) {

pkg/cloud/scope/cluster.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"fmt"
 
+	awsv2 "github.com/aws/aws-sdk-go-v2/aws"
 	awsclient "github.com/aws/aws-sdk-go/aws/client"
 	"github.com/pkg/errors"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
@@ -45,6 +46,7 @@ type ClusterScopeParams struct {
 	ControllerName               string
 	Endpoints                    []ServiceEndpoint
 	Session                      awsclient.ConfigProvider
+	SessionV2                    awsv2.Config
 	TagUnmanagedNetworkResources bool
 }
 
@@ -77,13 +79,19 @@ func NewClusterScope(params ClusterScopeParams) (*ClusterScope, error) {
 		return nil, errors.Errorf("failed to create aws session: %v", err)
 	}
 
+	sessionv2, _, err := sessionForClusterWithRegionV2(params.Client, clusterScope, params.AWSCluster.Spec.Region, params.Endpoints, params.Logger)
+	if err != nil {
+		return nil, errors.Errorf("failed to create aws V2 session: %v", err)
+	}
+
 	helper, err := patch.NewHelper(params.AWSCluster, params.Client)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to init patch helper")
 	}
 
 	clusterScope.patchHelper = helper
 	clusterScope.session = session
+	clusterScope.sessionV2 = *sessionv2
 	clusterScope.serviceLimiters = serviceLimiters
 
 	return clusterScope, nil
@@ -99,6 +107,7 @@ type ClusterScope struct {
 	AWSCluster *infrav1.AWSCluster
 
 	session         awsclient.ConfigProvider
+	sessionV2       awsv2.Config
 	serviceLimiters throttle.ServiceLimiters
 	controllerName  string
 
@@ -351,6 +360,11 @@ func (s *ClusterScope) Session() awsclient.ConfigProvider {
 	return s.session
 }
 
+// Session returns the AWS SDK V2 session. Used for creating clients.
+func (s *ClusterScope) SessionV2() awsv2.Config {
+	return s.sessionV2
+}
+
 // ServiceLimiter returns the AWS SDK session. Used for creating clients.
 func (s *ClusterScope) ServiceLimiter(service string) *throttle.ServiceLimiter {
 	if sl, ok := s.serviceLimiters[service]; ok {