S3 - Encryption context header cannot be null

CHANGELOG.md

@@ -12,6 +12,7 @@ We use *breaking :warning:* to mark changes that are not backward compatible (re
 
 ### Added
 
+- [#5089](https://github.com/thanos-io/thanos/pull/5089) S3: Create an empty map in the case SSE-KMS is used and no KMSEncryptionContext is passed.
 - [#4970](https://github.com/thanos-io/thanos/pull/4970) Added a new flag `exclude-delete` to `tools bucket ls`, which excludes blocks marked for deletion.
 - [#4903](https://github.com/thanos-io/thanos/pull/4903) Compactor: Added tracing support for compaction.
 - [#4909](https://github.com/thanos-io/thanos/pull/4909) Compactor: Add flag --max-time / --min-time to filter blocks that are ready to be compacted.

docs/components/compact.md

@@ -2,7 +2,7 @@
 
 The `thanos compact` command applies the compaction procedure of the Prometheus 2.0 storage engine to block data stored in object storage. It is generally not semantically concurrency safe and must be deployed as a singleton against a bucket.
 
-Compactor is also responsible for downsampling of data. There is a time delay before downsampling at a given resolution is possible. This is necessary because downsampled chunks will have fewer samples in them, and as chunks are fixed size, data spanning more time will be required to fill them. 
+Compactor is also responsible for downsampling of data. There is a time delay before downsampling at a given resolution is possible. This is necessary because downsampled chunks will have fewer samples in them, and as chunks are fixed size, data spanning more time will be required to fill them.
 * Creating 5m downsampling for blocks older than **40 hours** (2d)
 * Creating 1h downsampling for blocks older than **10 days** (2w)
 

pkg/objstore/s3/s3.go

@@ -273,6 +273,12 @@ func NewBucketWithConfig(logger log.Logger, config Config, component string) (*B
 	if config.SSEConfig.Type != "" {
 		switch config.SSEConfig.Type {
 		case SSEKMS:
+			// If the KMSEncryptionContext is a nil map the header that is
+			// constructed by the encrypt.ServerSide object will be base64
+			// encoded "nil" which is not accepted by AWS.
+			if config.SSEConfig.KMSEncryptionContext == nil {
+				config.SSEConfig.KMSEncryptionContext = make(map[string]string)
+			}
 			sse, err = encrypt.NewSSEKMS(config.SSEConfig.KMSKeyID, config.SSEConfig.KMSEncryptionContext)
 			if err != nil {
 				return nil, errors.Wrap(err, "initialize s3 client SSE-KMS")

pkg/objstore/s3/s3_test.go

@@ -5,6 +5,8 @@ package s3
 
 import (
 	"context"
+	"encoding/base64"
+	"encoding/json"
 	"io"
 	"io/ioutil"
 	"net/http"
@@ -18,6 +20,8 @@ import (
 	"github.com/thanos-io/thanos/pkg/testutil"
 )
 
+const endpoint string = "localhost:80"
+
 func TestParseConfig(t *testing.T) {
 	input := []byte(`bucket: abcd
 insecure: false`)
@@ -304,7 +308,7 @@ list_objects_version: "abcd"`)
 func TestBucket_getServerSideEncryption(t *testing.T) {
 	// Default config should return no SSE config.
 	cfg := DefaultConfig
-	cfg.Endpoint = "localhost:80"
+	cfg.Endpoint = endpoint
 	bkt, err := NewBucketWithConfig(log.NewNopLogger(), cfg, "test")
 	testutil.Ok(t, err)
 
@@ -314,7 +318,7 @@ func TestBucket_getServerSideEncryption(t *testing.T) {
 
 	// If SSE is configured in the client config it should be used.
 	cfg = DefaultConfig
-	cfg.Endpoint = "localhost:80"
+	cfg.Endpoint = endpoint
 	cfg.SSEConfig = SSEConfig{Type: SSES3}
 	bkt, err = NewBucketWithConfig(log.NewNopLogger(), cfg, "test")
 	testutil.Ok(t, err)
@@ -323,9 +327,56 @@ func TestBucket_getServerSideEncryption(t *testing.T) {
 	testutil.Ok(t, err)
 	testutil.Equals(t, encrypt.S3, sse.Type())
 
+	// SSE-KMS can be configured in the client config with an optional
+	// KMSEncryptionContext - In this case the encryptionContextHeader should be
+	// a base64 encoded string which represents a string-string map "{}"
+	cfg = DefaultConfig
+	cfg.Endpoint = endpoint
+	cfg.SSEConfig = SSEConfig{
+		Type:     SSEKMS,
+		KMSKeyID: "key",
+	}
+	bkt, err = NewBucketWithConfig(log.NewNopLogger(), cfg, "test")
+	testutil.Ok(t, err)
+
+	sse, err = bkt.getServerSideEncryption(context.Background())
+	testutil.Ok(t, err)
+	testutil.Equals(t, encrypt.KMS, sse.Type())
+
+	encryptionContextHeader := "X-Amz-Server-Side-Encryption-Context"
+	headers := make(http.Header)
+	sse.Marshal(headers)
+	wantJson, err := json.Marshal(make(map[string]string))
+	testutil.Ok(t, err)
+	want := base64.StdEncoding.EncodeToString(wantJson)
+	testutil.Equals(t, want, headers.Get(encryptionContextHeader))
+
+	// If the KMSEncryptionContext is set then the header should reflect it's
+	// value.
+	cfg = DefaultConfig
+	cfg.Endpoint = endpoint
+	cfg.SSEConfig = SSEConfig{
+		Type:                 SSEKMS,
+		KMSKeyID:             "key",
+		KMSEncryptionContext: map[string]string{"foo": "bar"},
+	}
+	bkt, err = NewBucketWithConfig(log.NewNopLogger(), cfg, "test")
+	testutil.Ok(t, err)
+
+	sse, err = bkt.getServerSideEncryption(context.Background())
+	testutil.Ok(t, err)
+	testutil.Equals(t, encrypt.KMS, sse.Type())
+
+	headers = make(http.Header)
+	sse.Marshal(headers)
+	wantJson, err = json.Marshal(cfg.SSEConfig.KMSEncryptionContext)
+	testutil.Ok(t, err)
+	want = base64.StdEncoding.EncodeToString(wantJson)
+	testutil.Equals(t, want, headers.Get(encryptionContextHeader))
+
 	// If SSE is configured in the context it should win.
 	cfg = DefaultConfig
-	cfg.Endpoint = "localhost:80"
+	cfg.Endpoint = endpoint
 	cfg.SSEConfig = SSEConfig{Type: SSES3}
 	override, err := encrypt.NewSSEKMS("test", nil)
 	testutil.Ok(t, err)