Repository: minio/minio-go
Pull request #1483 (2021-04-21 06:40:48)

Add ExistingObjectReplication setting to replication config
for https://github.com/minio/minio/pull/12109


Patch:
diff --git a/api-bucket-replication.go b/api-bucket-replication.go
index ef10355..41054e1 100644
--- a/api-bucket-replication.go
+++ b/api-bucket-replication.go
@@ -25,6 +25,7 @@ import (
 	"io/ioutil"
 	"net/http"
 	"net/url"
+	"time"
 
 	"github.com/minio/minio-go/v7/pkg/replication"
 	"github.com/minio/minio-go/v7/pkg/s3utils"
@@ -185,3 +186,43 @@ func (c Client) GetBucketReplicationMetrics(ctx context.Context, bucketName stri
 	}
 	return s, nil
 }
+
+// ResetBucketReplication kicks off replication of previously replicated objects if ExistingObjectReplication
+// is enabled in the replication config
+func (c Client) ResetBucketReplication(ctx context.Context, bucketName string, olderThan time.Duration) (resetID string, err error) {
+	// Input validation.
+	if err := s3utils.CheckValidBucketName(bucketName); err != nil {
+		return "", err
+	}
+	// Get resources properly escaped and lined up before
+	// using them in http request.
+	urlValues := make(url.Values)
+	urlValues.Set("replication-reset", "")
+	if olderThan > 0 {
+		urlValues.Set("older-than", olderThan.String())
+	}
+
+	// Execute GET on bucket to get replication config.
+	resp, err := c.executeMethod(ctx, http.MethodPut, requestMetadata{
+		bucketName:  bucketName,
+		queryValues: urlValues,
+	})
+
+	defer closeResponse(resp)
+	if err != nil {
+		return "", err
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		return "", httpRespToErrorResponse(resp, bucketName, "")
+	}
+	respBytes, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return "", err
+	}
+
+	if err := json.Unmarshal(respBytes, &resetID); err != nil {
+		return "", err
+	}
+	return resetID, nil
+}
diff --git a/pkg/credentials/iam_aws.go b/pkg/credentials/iam_aws.go
index 2ec0afc..ba918e9 100644
--- a/pkg/credentials/iam_aws.go
+++ b/pkg/credentials/iam_aws.go
@@ -59,6 +59,10 @@ const (
 	defaultECSRoleEndpoint      = "http://169.254.170.2"
 	defaultSTSRoleEndpoint      = "https://sts.amazonaws.com"
 	defaultIAMSecurityCredsPath = "/latest/meta-data/iam/security-credentials/"
+	tokenRequestTTLHeader       = "X-aws-ec2-metadata-token-ttl-seconds"
+	tokenPath                   = "/latest/api/token"
+	tokenTTL                    = "21600"
+	tokenRequestHeader          = "X-aws-ec2-metadata-token"
 )
 
 // NewIAM returns a pointer to a new Credentials object wrapping the IAM.
@@ -192,11 +196,14 @@ func getIAMRoleURL(endpoint string) (*url.URL, error) {
 // with the current EC2 service. If there are no credentials,
 // or there is an error making or receiving the request.
 // http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/iam-roles-for-amazon-ec2.html
-func listRoleNames(client *http.Client, u *url.URL) ([]string, error) {
+func listRoleNames(client *http.Client, u *url.URL, token string) ([]string, error) {
 	req, err := http.NewRequest(http.MethodGet, u.String(), nil)
 	if err != nil {
 		return nil, err
 	}
+	if token != "" {
+		req.Header.Add(tokenRequestHeader, token)
+	}
 	resp, err := client.Do(req)
 	if err != nil {
 		return nil, err
@@ -242,12 +249,35 @@ func getEcsTaskCredentials(client *http.Client, endpoint string) (ec2RoleCredRes
 	return respCreds, nil
 }
 
+func fetchIMDSToken(client *http.Client, endpoint string) (string, error) {
+	req, err := http.NewRequest(http.MethodPut, endpoint+tokenPath, nil)
+	if err != nil {
+		return "", err
+	}
+	req.Header.Add(tokenRequestTTLHeader, tokenTTL)
+	resp, err := client.Do(req)
+	if err != nil {
+		return "", err
+	}
+	defer resp.Body.Close()
+	data, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return "", err
+	}
+	if resp.StatusCode != http.StatusOK {
+		return "", errors.New(resp.Status)
+	}
+	return string(data), nil
+}
+
 // getCredentials - obtains the credentials from the IAM role name associated with
 // the current EC2 service.
 //
 // If the credentials cannot be found, or there is an error
 // reading the response an error will be returned.
 func getCredentials(client *http.Client, endpoint string) (ec2RoleCredRespBody, error) {
+	// https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/configuring-instance-metadata-service.html
+	token, _ := fetchIMDSToken(client, endpoint)
 
 	// http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/iam-roles-for-amazon-ec2.html
 	u, err := getIAMRoleURL(endpoint)
@@ -256,7 +286,7 @@ func getCredentials(client *http.Client, endpoint string) (ec2RoleCredRespBody,
 	}
 
 	// http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/iam-roles-for-amazon-ec2.html
-	roleNames, err := listRoleNames(client, u)
+	roleNames, err := listRoleNames(client, u, token)
 	if err != nil {
 		return ec2RoleCredRespBody{}, err
 	}
@@ -280,6 +310,9 @@ func getCredentials(client *http.Client, endpoint string) (ec2RoleCredRespBody,
 	if err != nil {
 		return ec2RoleCredRespBody{}, err
 	}
+	if token != "" {
+		req.Header.Add(tokenRequestHeader, token)
+	}
 
 	resp, err := client.Do(req)
 	if err != nil {
diff --git a/pkg/replication/replication.go b/pkg/replication/replication.go
index 280d25c..beacc71 100644
--- a/pkg/replication/replication.go
+++ b/pkg/replication/replication.go
@@ -46,20 +46,21 @@ const (
 
 // Options represents options to set a replication configuration rule
 type Options struct {
-	Op                     OptionType
-	ID                     string
-	Prefix                 string
-	RuleStatus             string
-	Priority               string
-	TagString              string
-	StorageClass           string
-	RoleArn                string
-	DestBucket             string
-	IsTagSet               bool
-	IsSCSet                bool
-	ReplicateDeletes       string // replicate versioned deletes
-	ReplicateDeleteMarkers string // replicate soft deletes
-	ReplicaSync            string // replicate replica metadata modifications
+	Op                      OptionType
+	ID                      string
+	Prefix                  string
+	RuleStatus              string
+	Priority                string
+	TagString               string
+	StorageClass            string
+	RoleArn                 string
+	DestBucket              string
+	IsTagSet                bool
+	IsSCSet                 bool
+	ReplicateDeletes        string // replicate versioned deletes
+	ReplicateDeleteMarkers  string // replicate soft deletes
+	ReplicaSync             string // replicate replica metadata modifications
+	ExistingObjectReplicate string
 }
 
 // Tags returns a slice of tags for a rule
@@ -195,6 +196,17 @@ func (c *Config) AddRule(opts Options) error {
 		return fmt.Errorf("replica metadata sync should be either [enable|disable]")
 	}
 
+	var existingStatus Status
+	if opts.ExistingObjectReplicate != "" {
+		switch opts.ExistingObjectReplicate {
+		case "enable":
+			existingStatus = Enabled
+		case "disable", "":
+			existingStatus = Disabled
+		default:
+			return fmt.Errorf("existingObjectReplicate should be either enable|disable")
+		}
+	}
 	newRule := Rule{
 		ID:       opts.ID,
 		Priority: priority,
@@ -214,6 +226,10 @@ func (c *Config) AddRule(opts Options) error {
 				Status: replicaSync,
 			},
 		},
+		// By default disable existing object replication unless selected
+		ExistingObjectReplication: ExistingObjectReplication{
+			Status: existingStatus,
+		},
 	}
 
 	// validate rule after overlaying priority for pre-existing rule being disabled.
@@ -335,7 +351,17 @@ func (c *Config) EditRule(opts Options) error {
 			return fmt.Errorf("replica metadata sync should be either [enable|disable]")
 		}
 	}
-
+	fmt.Println("opts.ExistingObjectReplicate>", opts.ExistingObjectReplicate)
+	if opts.ExistingObjectReplicate != "" {
+		switch opts.ExistingObjectReplicate {
+		case "enable":
+			newRule.ExistingObjectReplication.Status = Enabled
+		case "disable":
+			newRule.ExistingObjectReplication.Status = Disabled
+		default:
+			return fmt.Errorf("existingObjectsReplication state should be either [enable|disable]")
+		}
+	}
 	if opts.IsSCSet {
 		newRule.Destination.StorageClass = opts.StorageClass
 	}
@@ -400,15 +426,16 @@ func (c *Config) RemoveRule(opts Options) error {
 
 // Rule - a rule for replication configuration.
 type Rule struct {
-	XMLName                 xml.Name                `xml:"Rule" json:"-"`
-	ID                      string                  `xml:"ID,omitempty"`
-	Status                  Status                  `xml:"Status"`
-	Priority                int                     `xml:"Priority"`
-	DeleteMarkerReplication DeleteMarkerReplication `xml:"DeleteMarkerReplication"`
-	DeleteReplication       DeleteReplication       `xml:"DeleteReplication"`
-	Destination             Destination             `xml:"Destination"`
-	Filter                  Filter                  `xml:"Filter" json:"Filter"`
-	SourceSelectionCriteria SourceSelectionCriteria `xml:"SourceSelectionCriteria" json:"SourceSelectionCriteria"`
+	XMLName                   xml.Name                  `xml:"Rule" json:"-"`
+	ID                        string                    `xml:"ID,omitempty"`
+	Status                    Status                    `xml:"Status"`
+	Priority                  int                       `xml:"Priority"`
+	DeleteMarkerReplication   DeleteMarkerReplication   `xml:"DeleteMarkerReplication"`
+	DeleteReplication         DeleteReplication         `xml:"DeleteReplication"`
+	Destination               Destination               `xml:"Destination"`
+	Filter                    Filter                    `xml:"Filter" json:"Filter"`
+	SourceSelectionCriteria   SourceSelectionCriteria   `xml:"SourceSelectionCriteria" json:"SourceSelectionCriteria"`
+	ExistingObjectReplication ExistingObjectReplication `xml:"ExistingObjectReplication,omitempty" json:"ExistingObjectReplication,omitempty"`
 }
 
 // Validate validates the rule for correctness
@@ -430,8 +457,7 @@ func (r Rule) Validate() error {
 	if err := r.validateStatus(); err != nil {
 		return err
 	}
-
-	return nil
+	return r.ExistingObjectReplication.Validate()
 }
 
 // validateID - checks if ID is valid or not.
@@ -631,6 +657,27 @@ func (s SourceSelectionCriteria) Validate() error {
 	return nil
 }
 
+// ExistingObjectReplication - whether existing object replication is enabled
+type ExistingObjectReplication struct {
+	Status Status `xml:"Status"` // should be set to "Disabled" by default
+}
+
+// IsEmpty returns true if DeleteMarkerReplication is not set
+func (e ExistingObjectReplication) IsEmpty() bool {
+	return len(e.Status) == 0
+}
+
+// Validate validates whether the status is disabled.
+func (e ExistingObjectReplication) Validate() error {
+	if e.IsEmpty() {
+		return nil
+	}
+	if e.Status != Disabled && e.Status != Enabled {
+		return fmt.Errorf("invalid ExistingObjectReplication status")
+	}
+	return nil
+}
+
 // Metrics represents inline replication metrics
 // such as pending, failed and completed bytes in total for a bucket
 type Metrics struct {


Test patch:
diff --git a/pkg/credentials/iam_aws_test.go b/pkg/credentials/iam_aws_test.go
index 9f28d18..686721d 100644
--- a/pkg/credentials/iam_aws_test.go
+++ b/pkg/credentials/iam_aws_test.go
@@ -25,6 +25,7 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"os"
+	"strconv"
 	"testing"
 	"time"
 )
@@ -105,6 +106,40 @@ func initTestServer(expireOn string, failAssume bool) *httptest.Server {
 	return server
 }
 
+// Instance Metadata Service with V1 disabled.
+func initIMDSv2Server(expireOn string) *httptest.Server {
+	imdsToken := "IMDSTokenabc123=="
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		fmt.Println(r.URL.Path)
+		fmt.Println(r.Method)
+		if r.URL.Path == "/latest/api/token" && r.Method == "PUT" {
+			ttlHeader := r.Header.Get("X-aws-ec2-metadata-token-ttl-seconds")
+			ttl, err := strconv.ParseInt(ttlHeader, 10, 32)
+			if err != nil || ttl < 0 || ttl > 21600 {
+				http.Error(w, "", http.StatusBadRequest)
+				return
+			}
+			w.Header().Set("X-Aws-Ec2-Metadata-Token-Ttl-Seconds", ttlHeader)
+			w.Write([]byte(imdsToken))
+			return
+		}
+		token := r.Header.Get("X-aws-ec2-metadata-token")
+		if token != imdsToken {
+			http.Error(w, r.URL.Path, http.StatusUnauthorized)
+			return
+		}
+
+		if r.URL.Path == "/latest/meta-data/iam/security-credentials/" {
+			fmt.Fprintln(w, "RoleName")
+		} else if r.URL.Path == "/latest/meta-data/iam/security-credentials/RoleName" {
+			fmt.Fprintf(w, credsRespTmpl, expireOn)
+		} else {
+			http.Error(w, "bad request", http.StatusBadRequest)
+		}
+	}))
+	return server
+}
+
 func initEcsTaskTestServer(expireOn string) *httptest.Server {
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		fmt.Fprintf(w, credsRespEcsTaskTmpl, expireOn)
@@ -391,3 +426,15 @@ func TestStsCn(t *testing.T) {
 		t.Error("Expected creds to be expired.")
 	}
 }
+
+func TestIMDSv1Blocked(t *testing.T) {
+	server := initIMDSv2Server("2014-12-16T01:51:37Z")
+	p := &IAM{
+		Client:   http.DefaultClient,
+		Endpoint: server.URL,
+	}
+	_, err := p.Retrieve()
+	if err != nil {
+		t.Errorf("Unexpected IMDSv2 failure %s", err)
+	}
+}