br: support s3 profile config (#62199)

ref #46609

Author: takaidohigasi

br/pkg/storage/BUILD.bazel

@@ -99,6 +99,7 @@ go_test(
         "locking_test.go",
         "memstore_test.go",
         "parse_test.go",
+        "s3_flags_test.go",
         "s3_test.go",
         "storage_test.go",
         "writer_test.go",
@@ -120,6 +121,7 @@ go_test(
         "@com_github_pingcap_errors//:errors",
         "@com_github_pingcap_failpoint//:failpoint",
         "@com_github_pingcap_kvproto//pkg/brpb",
+        "@com_github_spf13_pflag//:pflag",
         "@com_github_stretchr_testify//require",
         "@org_golang_x_sync//errgroup",
         "@org_uber_go_mock//gomock",

br/pkg/storage/parse_test.go

@@ -310,6 +310,131 @@ func TestIsLocal(t *testing.T) {
 	}
 }
 
+func TestS3ProfileOption(t *testing.T) {
+	// Test parsing profile from URL query parameter
+	testProfile := "my-test-profile"
+	s, err := ParseBackend(
+		fmt.Sprintf("s3://bucket/prefix/?profile=%s", url.QueryEscape(testProfile)),
+		nil,
+	)
+	require.NoError(t, err)
+	s3 := s.GetS3()
+	require.NotNil(t, s3)
+	require.Equal(t, "bucket", s3.Bucket)
+	require.Equal(t, "prefix", s3.Prefix)
+	require.Equal(t, testProfile, s3.Profile)
+
+	// Test with BackendOptions
+	s3opt := &BackendOptions{
+		S3: S3BackendOptions{
+			Profile: "profile-from-options",
+		},
+	}
+	s, err = ParseBackend("s3://bucket2/prefix/", s3opt)
+	require.NoError(t, err)
+	s3 = s.GetS3()
+	require.NotNil(t, s3)
+	require.Equal(t, "bucket2", s3.Bucket)
+	require.Equal(t, "prefix", s3.Prefix)
+	require.Equal(t, "profile-from-options", s3.Profile)
+
+	// Test profile with other S3 options
+	s, err = ParseBackend(
+		"s3://bucket3/prefix/?profile=dev-profile&region=us-west-2&endpoint=https://s3.example.com",
+		nil,
+	)
+	require.NoError(t, err)
+	s3 = s.GetS3()
+	require.NotNil(t, s3)
+	require.Equal(t, "bucket3", s3.Bucket)
+	require.Equal(t, "prefix", s3.Prefix)
+	require.Equal(t, "dev-profile", s3.Profile)
+	require.Equal(t, "us-west-2", s3.Region)
+	require.Equal(t, "https://s3.example.com", s3.Endpoint)
+
+	// Test empty profile (should remain empty)
+	s, err = ParseBackend("s3://bucket4/prefix/", nil)
+	require.NoError(t, err)
+	s3 = s.GetS3()
+	require.NotNil(t, s3)
+	require.Equal(t, "bucket4", s3.Bucket)
+	require.Equal(t, "prefix", s3.Prefix)
+	require.Equal(t, "", s3.Profile) // Should be empty when not specified
+}
+
+func TestS3ProfileCredentialsValidation(t *testing.T) {
+	// Test that profile makes credentials optional in URL parsing
+
+	// Case 1: Profile without credentials - should be allowed
+	s, err := ParseBackend("s3://bucket/prefix/?profile=production&region=us-west-2", nil)
+	require.NoError(t, err, "Should not require credentials when using profile")
+	s3 := s.GetS3()
+	require.NotNil(t, s3)
+	require.Equal(t, "production", s3.Profile)
+	require.Equal(t, "us-west-2", s3.Region)
+	require.Equal(t, "", s3.AccessKey)
+	require.Equal(t, "", s3.SecretAccessKey)
+
+	// Case 2: Profile with partial credentials - should be allowed
+	s, err = ParseBackend("s3://bucket/prefix/?profile=dev&access-key=override-key", nil)
+	require.NoError(t, err, "Should allow partial credentials with profile")
+	s3 = s.GetS3()
+	require.NotNil(t, s3)
+	require.Equal(t, "dev", s3.Profile)
+	require.Equal(t, "override-key", s3.AccessKey)
+	require.Equal(t, "", s3.SecretAccessKey) // No secret key, but should be OK with profile
+
+	// Case 3: Profile with explicit credentials override
+	s, err = ParseBackend("s3://bucket/prefix/?profile=staging&access-key=explicit-access&secret-access-key=explicit-secret", nil)
+	require.NoError(t, err)
+	s3 = s.GetS3()
+	require.NotNil(t, s3)
+	require.Equal(t, "staging", s3.Profile)
+	require.Equal(t, "explicit-access", s3.AccessKey)
+	require.Equal(t, "explicit-secret", s3.SecretAccessKey)
+}
+
+func TestS3NoProfileCredentialsValidation(t *testing.T) {
+	// Test that without profile, credential validation still applies
+
+	// Case 1: No profile, partial credentials - should fail
+	s3opt := &BackendOptions{
+		S3: S3BackendOptions{
+			AccessKey: "only-access-key",
+			// Missing SecretAccessKey
+		},
+	}
+	_, err := ParseBackend("s3://bucket/prefix/", s3opt)
+	require.Error(t, err, "Should fail when access key provided without secret key")
+	require.Contains(t, err.Error(), "secret_access_key not found")
+
+	// Case 2: No profile, both credentials - should be valid
+	s3opt2 := &BackendOptions{
+		S3: S3BackendOptions{
+			AccessKey:       "test-access",
+			SecretAccessKey: "test-secret",
+		},
+	}
+	s, err := ParseBackend("s3://bucket/prefix/", s3opt2)
+	require.NoError(t, err)
+	s3Backend2 := s.GetS3()
+	require.Equal(t, "test-access", s3Backend2.AccessKey)
+	require.Equal(t, "test-secret", s3Backend2.SecretAccessKey)
+
+	// Case 3: No profile, no credentials - should be valid (IAM role, etc.)
+	s3opt3 := &BackendOptions{
+		S3: S3BackendOptions{
+			Region: "us-east-1",
+			// No credentials
+		},
+	}
+	s, err = ParseBackend("s3://bucket/prefix/", s3opt3)
+	require.NoError(t, err, "Should allow no credentials when no profile (for IAM roles, etc.)")
+	s3Backend3 := s.GetS3()
+	require.Equal(t, "", s3Backend3.AccessKey)
+	require.Equal(t, "", s3Backend3.SecretAccessKey)
+}
+
 func TestParseBackend(t *testing.T) {
 	{
 		backendOptions := &BackendOptions{

br/pkg/storage/s3.go

@@ -56,6 +56,7 @@ const (
 	s3ProviderOption     = "s3.provider"
 	s3RoleARNOption      = "s3.role-arn"
 	s3ExternalIDOption   = "s3." + S3ExternalID
+	s3ProfileOption      = "s3.profile"
 	notFound             = "NotFound"
 	// number of retries to make of operations.
 	maxRetries = 7
@@ -180,6 +181,7 @@ type S3BackendOptions struct {
 	UseAccelerateEndpoint bool   `json:"use-accelerate-endpoint" toml:"use-accelerate-endpoint"`
 	RoleARN               string `json:"role-arn" toml:"role-arn"`
 	ExternalID            string `json:"external-id" toml:"external-id"`
+	Profile               string `json:"profile" toml:"profile"`
 	ObjectLockEnabled     bool   `json:"object-lock-enabled" toml:"object-lock-enabled"`
 }
 
@@ -203,11 +205,14 @@ func (options *S3BackendOptions) Apply(s3 *backuppb.S3) error {
 		options.UseAccelerateEndpoint {
 		options.ForcePathStyle = false
 	}
-	if options.AccessKey == "" && options.SecretAccessKey != "" {
-		return errors.Annotate(berrors.ErrStorageInvalidConfig, "access_key not found")
-	}
-	if options.AccessKey != "" && options.SecretAccessKey == "" {
-		return errors.Annotate(berrors.ErrStorageInvalidConfig, "secret_access_key not found")
+	// When not using a profile, if either key is provided, both must be provided
+	if options.Profile == "" {
+		if options.AccessKey == "" && options.SecretAccessKey != "" {
+			return errors.Annotate(berrors.ErrStorageInvalidConfig, "access_key not found")
+		}
+		if options.AccessKey != "" && options.SecretAccessKey == "" {
+			return errors.Annotate(berrors.ErrStorageInvalidConfig, "secret_access_key not found")
+		}
 	}
 
 	s3.Endpoint = strings.TrimSuffix(options.Endpoint, "/")
@@ -224,6 +229,8 @@ func (options *S3BackendOptions) Apply(s3 *backuppb.S3) error {
 	s3.RoleArn = options.RoleARN
 	s3.ExternalId = options.ExternalID
 	s3.Provider = options.Provider
+	s3.Profile = options.Profile
+
 	return nil
 }
 
@@ -241,6 +248,8 @@ func defineS3Flags(flags *pflag.FlagSet) {
 	flags.String(s3ProviderOption, "", "(experimental) Set the S3 provider, e.g. aws, alibaba, ceph")
 	flags.String(s3RoleARNOption, "", "(experimental) Set the ARN of the IAM role to assume when accessing AWS S3")
 	flags.String(s3ExternalIDOption, "", "(experimental) Set the external ID when assuming the role to access AWS S3")
+	flags.String(s3ProfileOption, "", "(experimental) Set the AWS profile to use for AWS S3 authentication. "+
+		"Command line options take precedence over profile settings")
 }
 
 // parseFromFlags parse S3BackendOptions from command line flags.
@@ -284,6 +293,11 @@ func (options *S3BackendOptions) parseFromFlags(flags *pflag.FlagSet) error {
 	if err != nil {
 		return errors.Trace(err)
 	}
+	options.Profile, err = flags.GetString(s3ProfileOption)
+	if err != nil {
+		return errors.Trace(err)
+	}
+
 	return nil
 }
 
@@ -362,17 +376,26 @@ func NewS3Storage(ctx context.Context, backend *backuppb.S3, opts *ExternalStora
 	if opts.HTTPClient != nil {
 		awsConfig.WithHTTPClient(opts.HTTPClient)
 	}
-	cred, err := autoNewCred(&qs)
-	if err != nil {
-		return nil, errors.Trace(err)
-	}
-	if cred != nil {
-		awsConfig.WithCredentials(cred)
+	// When using a profile, let AWS SDK handle credentials through the profile
+	// Don't call autoNewCred as it interferes with profile-based authentication
+	if qs.Profile == "" {
+		cred, err := autoNewCred(&qs)
+		if err != nil {
+			return nil, errors.Trace(err)
+		}
+		if cred != nil {
+			awsConfig.WithCredentials(cred)
+		}
 	}
 	// awsConfig.WithLogLevel(aws.LogDebugWithSigning)
 	awsSessionOpts := session.Options{
 		Config: *awsConfig,
 	}
+	if qs.Profile != "" {
+		awsSessionOpts.Profile = qs.Profile
+		// Use default credential chain when profile is specified
+		awsSessionOpts.SharedConfigState = session.SharedConfigEnable
+	}
 	ses, err := session.NewSessionWithOptions(awsSessionOpts)
 	if err != nil {
 		return nil, errors.Trace(err)