Add signature v2 support for S3 chunks client

Signed-off-by: Christian Simon <[email protected]>

Author: simonswine

CHANGELOG.md

@@ -45,6 +45,7 @@
   * `cortex_compactor_tenants_processing_succeeded`
   * `cortex_compactor_tenants_processing_failed`
 * [ENHANCEMENT] Added new experimental API endpoints: `POST /purger/delete_tenant` and `GET /purger/delete_tenant_status` for deleting all tenant data. Only works with blocks storage. #3549
+* [ENHANCEMENT] Chunks storage: add option to use V2 signatures for S3 authentication. #3560
 * [BUGFIX] Blocks storage ingester: fixed some cases leading to a TSDB WAL corruption after a partial write to disk. #3423
 * [BUGFIX] Blocks storage: Fix the race between ingestion and `/flush` call resulting in overlapping blocks. #3422
 * [BUGFIX] Querier: fixed `-querier.max-query-into-future` which wasn't correctly enforced on range queries. #3452

docs/configuration/config-file-reference.md

@@ -1259,6 +1259,11 @@ storage:
       # CLI flag: -ruler.storage.s3.http.insecure-skip-verify
       [insecure_skip_verify: <boolean> | default = false]
 
+    # The signature version to use for authenticating against S3. Supported
+    # values are: v4, v2.
+    # CLI flag: -ruler.storage.s3.signature-version
+    [signature_version: <string> | default = "v4"]
+
   swift:
     # Openstack authentication URL.
     # CLI flag: -ruler.storage.swift.auth-url
@@ -1577,6 +1582,11 @@ storage:
       # CLI flag: -alertmanager.storage.s3.http.insecure-skip-verify
       [insecure_skip_verify: <boolean> | default = false]
 
+    # The signature version to use for authenticating against S3. Supported
+    # values are: v4, v2.
+    # CLI flag: -alertmanager.storage.s3.signature-version
+    [signature_version: <string> | default = "v4"]
+
 # Enable the experimental alertmanager config api.
 # CLI flag: -experimental.alertmanager.enable-api
 [enable_api: <boolean> | default = false]
@@ -2054,6 +2064,11 @@ aws:
     # CLI flag: -s3.http.insecure-skip-verify
     [insecure_skip_verify: <boolean> | default = false]
 
+  # The signature version to use for authenticating against S3. Supported values
+  # are: v4, v2.
+  # CLI flag: -s3.signature-version
+  [signature_version: <string> | default = "v4"]
+
 azure:
   # Azure Cloud environment. Supported values are: AzureGlobal, AzureChinaCloud,
   # AzureGermanCloud, AzureUSGovernment.

go.mod

@@ -37,6 +37,7 @@ require (
 	github.com/json-iterator/go v1.1.10
 	github.com/lann/builder v0.0.0-20180802200727-47ae307949d0 // indirect
 	github.com/lib/pq v1.3.0
+	github.com/minio/minio-go/v7 v7.0.2
 	github.com/mitchellh/go-wordwrap v1.0.0
 	github.com/ncw/swift v1.0.50
 	github.com/oklog/ulid v1.3.1

pkg/alertmanager/multitenant.go

@@ -125,6 +125,14 @@ func (cfg *MultitenantAlertmanagerConfig) RegisterFlags(f *flag.FlagSet) {
 	cfg.Store.RegisterFlags(f)
 }
 
+// Validate config and returns error on failure
+func (cfg *MultitenantAlertmanagerConfig) Validate() error {
+	if err := cfg.Store.Validate(); err != nil {
+		return errors.Wrap(err, "invalid storage config")
+	}
+	return nil
+}
+
 type multitenantAlertmanagerMetrics struct {
 	lastReloadSuccessful          *prometheus.GaugeVec
 	lastReloadSuccessfulTimestamp *prometheus.GaugeVec

pkg/alertmanager/storage.go

@@ -5,6 +5,8 @@ import (
 	"flag"
 	"fmt"
 
+	"github.com/pkg/errors"
+
 	"github.com/cortexproject/cortex/pkg/alertmanager/alerts"
 	"github.com/cortexproject/cortex/pkg/alertmanager/alerts/configdb"
 	"github.com/cortexproject/cortex/pkg/alertmanager/alerts/local"
@@ -43,6 +45,14 @@ func (cfg *AlertStoreConfig) RegisterFlags(f *flag.FlagSet) {
 	cfg.S3.RegisterFlagsWithPrefix("alertmanager.storage.", f)
 }
 
+// Validate config and returns error on failure
+func (cfg *AlertStoreConfig) Validate() error {
+	if err := cfg.S3.Validate(); err != nil {
+		return errors.Wrap(err, "invalid S3 Storage config")
+	}
+	return nil
+}
+
 // NewAlertStore returns a new rule storage backend poller and store
 func NewAlertStore(cfg AlertStoreConfig) (AlertStore, error) {
 	switch cfg.Type {

pkg/chunk/aws/dynamodb_storage_client.go

@@ -87,6 +87,14 @@ func (cfg *StorageConfig) RegisterFlags(f *flag.FlagSet) {
 	cfg.S3Config.RegisterFlags(f)
 }
 
+// Validate config and returns error on failure
+func (cfg *StorageConfig) Validate() error {
+	if err := cfg.S3Config.Validate(); err != nil {
+		return errors.Wrap(err, "invalid S3 Storage config")
+	}
+	return nil
+}
+
 type dynamoDBStorageClient struct {
 	cfg       DynamoDBConfig
 	schemaCfg chunk.SchemaConfig

pkg/chunk/aws/s3_storage_client.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/tls"
 	"flag"
+	"fmt"
 	"hash/fnv"
 	"io"
 	"net"
@@ -14,18 +15,32 @@ import (
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/awserr"
 	"github.com/aws/aws-sdk-go/aws/credentials"
+	"github.com/aws/aws-sdk-go/aws/request"
 	"github.com/aws/aws-sdk-go/aws/session"
+	v4 "github.com/aws/aws-sdk-go/aws/signer/v4"
 	"github.com/aws/aws-sdk-go/service/s3"
 	"github.com/aws/aws-sdk-go/service/s3/s3iface"
+	"github.com/minio/minio-go/v7/pkg/signer"
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
 	awscommon "github.com/weaveworks/common/aws"
 	"github.com/weaveworks/common/instrument"
 
 	"github.com/cortexproject/cortex/pkg/chunk"
+	"github.com/cortexproject/cortex/pkg/util"
 	"github.com/cortexproject/cortex/pkg/util/flagext"
 )
 
+const (
+	SignatureVersionV4 = "v4"
+	SignatureVersionV2 = "v2"
+)
+
+var (
+	supportedSignatureVersions     = []string{SignatureVersionV4, SignatureVersionV2}
+	errUnsupportedSignatureVersion = errors.New("unsupported signature version")
+)
+
 var (
 	s3RequestDuration = instrument.NewHistogramCollector(prometheus.NewHistogramVec(prometheus.HistogramOpts{
 		Namespace: "cortex",
@@ -48,14 +63,15 @@ type S3Config struct {
 	S3               flagext.URLValue
 	S3ForcePathStyle bool
 
-	BucketNames     string
-	Endpoint        string     `yaml:"endpoint"`
-	Region          string     `yaml:"region"`
-	AccessKeyID     string     `yaml:"access_key_id"`
-	SecretAccessKey string     `yaml:"secret_access_key"`
-	Insecure        bool       `yaml:"insecure"`
-	SSEEncryption   bool       `yaml:"sse_encryption"`
-	HTTPConfig      HTTPConfig `yaml:"http_config"`
+	BucketNames      string
+	Endpoint         string     `yaml:"endpoint"`
+	Region           string     `yaml:"region"`
+	AccessKeyID      string     `yaml:"access_key_id"`
+	SecretAccessKey  string     `yaml:"secret_access_key"`
+	Insecure         bool       `yaml:"insecure"`
+	SSEEncryption    bool       `yaml:"sse_encryption"`
+	HTTPConfig       HTTPConfig `yaml:"http_config"`
+	SignatureVersion string     `yaml:"signature_version"`
 
 	Inject InjectRequestMiddleware `yaml:"-"`
 }
@@ -89,6 +105,15 @@ func (cfg *S3Config) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
 	f.DurationVar(&cfg.HTTPConfig.IdleConnTimeout, prefix+"s3.http.idle-conn-timeout", 90*time.Second, "The maximum amount of time an idle connection will be held open.")
 	f.DurationVar(&cfg.HTTPConfig.ResponseHeaderTimeout, prefix+"s3.http.response-header-timeout", 0, "If non-zero, specifies the amount of time to wait for a server's response headers after fully writing the request.")
 	f.BoolVar(&cfg.HTTPConfig.InsecureSkipVerify, prefix+"s3.http.insecure-skip-verify", false, "Set to false to skip verifying the certificate chain and hostname.")
+	f.StringVar(&cfg.SignatureVersion, prefix+"s3.signature-version", SignatureVersionV4, fmt.Sprintf("The signature version to use for authenticating against S3. Supported values are: %s.", strings.Join(supportedSignatureVersions, ", ")))
+}
+
+// Validate config and returns error on failure
+func (cfg *S3Config) Validate() error {
+	if !util.StringsContain(supportedSignatureVersions, cfg.SignatureVersion) {
+		return errUnsupportedSignatureVersion
+	}
+	return nil
 }
 
 type S3ObjectClient struct {
@@ -111,6 +136,10 @@ func NewS3ObjectClient(cfg S3Config) (*S3ObjectClient, error) {
 
 	s3Client := s3.New(sess)
 
+	if cfg.SignatureVersion == SignatureVersionV2 {
+		s3Client.Handlers.Sign.Swap(v4.SignRequestHandler.Name, v2SignRequestHandler(cfg))
+	}
+
 	var sseEncryption *string
 	if cfg.SSEEncryption {
 		sseEncryption = aws.String("AES256")
@@ -124,6 +153,28 @@ func NewS3ObjectClient(cfg S3Config) (*S3ObjectClient, error) {
 	return &client, nil
 }
 
+func v2SignRequestHandler(cfg S3Config) request.NamedHandler {
+	return request.NamedHandler{
+		Name: "v2.SignRequestHandler",
+		Fn: func(req *request.Request) {
+			credentials, err := req.Config.Credentials.GetWithContext(req.Context())
+			if err != nil {
+				if err != nil {
+					req.Error = err
+					return
+				}
+			}
+
+			req.HTTPRequest = signer.SignV2(
+				*req.HTTPRequest,
+				credentials.AccessKeyID,
+				credentials.SecretAccessKey,
+				!cfg.S3ForcePathStyle,
+			)
+		},
+	}
+}
+
 func buildS3Config(cfg S3Config) (*aws.Config, []string, error) {
 	var s3Config *aws.Config
 	var err error

pkg/chunk/storage/factory.go

@@ -117,6 +117,9 @@ func (cfg *Config) Validate() error {
 	if err := cfg.AzureStorageConfig.Validate(); err != nil {
 		return errors.Wrap(err, "invalid Azure Storage config")
 	}
+	if err := cfg.AWSStorageConfig.Validate(); err != nil {
+		return errors.Wrap(err, "invalid AWS Storage config")
+	}
 	return nil
 }
 

pkg/cortex/cortex.go

@@ -213,6 +213,9 @@ func (c *Config) Validate(log log.Logger) error {
 	if err := c.Compactor.Validate(); err != nil {
 		return errors.Wrap(err, "invalid compactor config")
 	}
+	if err := c.Alertmanager.Validate(); err != nil {
+		return errors.Wrap(err, "invalid alertmanager config")
+	}
 
 	if c.Storage.Engine == storage.StorageEngineBlocks && c.Querier.SecondStoreEngine != storage.StorageEngineChunks && len(c.Schema.Configs) > 0 {
 		level.Warn(log).Log("schema configuration is not used by the blocks storage engine, and will have no effect")

pkg/ruler/storage.go

@@ -54,6 +54,9 @@ func (cfg *RuleStoreConfig) Validate() error {
 	if err := cfg.Azure.Validate(); err != nil {
 		return errors.Wrap(err, "invalid Azure Storage config")
 	}
+	if err := cfg.S3.Validate(); err != nil {
+		return errors.Wrap(err, "invalid S3 Storage config")
+	}
 	return nil
 }