Enable IAM based auth to ES for AWS clients (jaegertracing#465)

Allows clients to specify AWS IAM configuration details for their connection to ElasticSearch. We allow this to enable an additional layer of security for those clients running on AWS. More details on connecting to ElasticSearch on AWS here: https://docs.aws.amazon.com/elasticsearch-service/latest/developerguide/es-vpc.html Signed-off-by: Wesley Kim <[email protected]>
wesleyk · Aug 30, 2018 · 213e52e · 213e52e
1 parent ddf9b22
commit 213e52e
Show file tree

Hide file tree

Showing 4 changed files with 116 additions and 20 deletions.
diff --git a/glide.lock b/glide.lock
diff --git a/pkg/es/config/config.go b/pkg/es/config/config.go
@@ -19,7 +19,10 @@ import (
 	"sync"
 	"time"
 
+	"github.com/aws/aws-sdk-go/aws/credentials"
+	"github.com/aws/aws-sdk-go/aws/signer/v4"
 	"github.com/pkg/errors"
+	"github.com/sha1sum/aws_signing_client"
 	"github.com/uber/jaeger-lib/metrics"
 	"go.uber.org/zap"
 	"gopkg.in/olivere/elastic.v5"
@@ -42,6 +45,14 @@ type Configuration struct {
 	BulkActions       int
 	BulkFlushInterval time.Duration
 	IndexPrefix       string
+	AwsConfig         AwsIamConfiguration
+}
+
+// AwsIamConfiguration describes the AWS-specific configuration needed to connect to ElasticSearch with IAM authn
+type AwsIamConfiguration struct {
+	AccessKeyID     string
+	SecretAccessKey string
+	Region          string
 }
 
 // ClientBuilder creates new es.Client
@@ -58,7 +69,13 @@ func (c *Configuration) NewClient(logger *zap.Logger, metricsFactory metrics.Fac
 	if len(c.Servers) < 1 {
 		return nil, errors.New("No servers specified")
 	}
-	rawClient, err := elastic.NewClient(c.GetConfigs()...)
+
+	clientOptionFuncs, err := c.GetConfigs()
+	if err != nil {
+		return nil, err
+	}
+
+	rawClient, err := elastic.NewClient(clientOptionFuncs...)
 	if err != nil {
 		return nil, err
 	}
@@ -143,6 +160,15 @@ func (c *Configuration) ApplyDefaults(source *Configuration) {
 	if c.BulkFlushInterval == 0 {
 		c.BulkFlushInterval = source.BulkFlushInterval
 	}
+	if c.AwsConfig.AccessKeyID == "" {
+		c.AwsConfig.AccessKeyID = source.AwsConfig.AccessKeyID
+	}
+	if c.AwsConfig.SecretAccessKey == "" {
+		c.AwsConfig.SecretAccessKey = source.AwsConfig.SecretAccessKey
+	}
+	if c.AwsConfig.Region == "" {
+		c.AwsConfig.Region = source.AwsConfig.Region
+	}
 }
 
 // GetNumShards returns number of shards from Configuration
@@ -166,10 +192,27 @@ func (c *Configuration) GetIndexPrefix() string {
 }
 
 // GetConfigs wraps the configs to feed to the ElasticSearch client init
-func (c *Configuration) GetConfigs() []elastic.ClientOptionFunc {
+func (c *Configuration) GetConfigs() ([]elastic.ClientOptionFunc, error) {
 	options := make([]elastic.ClientOptionFunc, 3)
 	options[0] = elastic.SetURL(c.Servers...)
 	options[1] = elastic.SetBasicAuth(c.Username, c.Password)
 	options[2] = elastic.SetSniff(c.Sniffer)
-	return options
+
+	// if AWS IAM credentials are provided, instantiate the elastic client with
+	// a signing client created via the AWS SDK
+	if c.AwsConfig.AccessKeyID != "" && c.AwsConfig.SecretAccessKey != "" && c.AwsConfig.Region != "" {
+		creds := credentials.NewStaticCredentials(
+			c.AwsConfig.AccessKeyID,
+			c.AwsConfig.SecretAccessKey,
+			"",
+		)
+		signer := v4.NewSigner(creds)
+		awsSigningClient, err := aws_signing_client.New(signer, nil, "es", c.AwsConfig.Region)
+		if err != nil {
+			return nil, err
+		}
+		options = append(options, elastic.SetHttpClient(awsSigningClient))
+	}
+
+	return options, nil
 }
diff --git a/plugin/storage/es/options.go b/plugin/storage/es/options.go
@@ -25,18 +25,21 @@ import (
 )
 
 const (
-	suffixUsername          = ".username"
-	suffixPassword          = ".password"
-	suffixSniffer           = ".sniffer"
-	suffixServerURLs        = ".server-urls"
-	suffixMaxSpanAge        = ".max-span-age"
-	suffixNumShards         = ".num-shards"
-	suffixNumReplicas       = ".num-replicas"
-	suffixBulkSize          = ".bulk.size"
-	suffixBulkWorkers       = ".bulk.workers"
-	suffixBulkActions       = ".bulk.actions"
-	suffixBulkFlushInterval = ".bulk.flush-interval"
-	suffixIndexPrefix       = ".index-prefix"
+	suffixUsername              = ".username"
+	suffixPassword              = ".password"
+	suffixSniffer               = ".sniffer"
+	suffixServerURLs            = ".server-urls"
+	suffixMaxSpanAge            = ".max-span-age"
+	suffixNumShards             = ".num-shards"
+	suffixNumReplicas           = ".num-replicas"
+	suffixBulkSize              = ".bulk.size"
+	suffixBulkWorkers           = ".bulk.workers"
+	suffixBulkActions           = ".bulk.actions"
+	suffixBulkFlushInterval     = ".bulk.flush-interval"
+	suffixIndexPrefix           = ".index-prefix"
+	suffixAwsIamAccessKeyID     = ".aws.access_key_id"
+	suffixAwsIamSecretAccessKey = ".aws.secret_access_key"
+	suffixAwsRegion             = ".aws.region"
 )
 
 // TODO this should be moved next to config.Configuration struct (maybe ./flags package)
@@ -75,6 +78,11 @@ func NewOptions(primaryNamespace string, otherNamespaces ...string) *Options {
 				BulkWorkers:       1,
 				BulkActions:       1000,
 				BulkFlushInterval: time.Millisecond * 200,
+				AwsConfig: config.AwsIamConfiguration{
+					AccessKeyID:     "",
+					SecretAccessKey: "",
+					Region:          "",
+				},
 			},
 			servers:   "http://127.0.0.1:9200",
 			namespace: primaryNamespace,
@@ -146,6 +154,18 @@ func addFlags(flagSet *flag.FlagSet, nsConfig *namespaceConfig) {
 		nsConfig.namespace+suffixIndexPrefix,
 		nsConfig.IndexPrefix,
 		"Optional prefix of Jaeger indices. For example \"production\" creates \"production:jaeger-*\".")
+	flagSet.String(
+		nsConfig.namespace+suffixAwsIamAccessKeyID,
+		nsConfig.AwsConfig.AccessKeyID,
+		"The access key name to use if using AWS ElasticSearch and connecting with IAM authentication")
+	flagSet.String(
+		nsConfig.namespace+suffixAwsIamSecretAccessKey,
+		nsConfig.AwsConfig.SecretAccessKey,
+		"The secret access key to use if using AWS ElasticSearch and connecting with IAM authentication")
+	flagSet.String(
+		nsConfig.namespace+suffixAwsRegion,
+		nsConfig.AwsConfig.Region,
+		"The AWS region to use if using AWS ElasticSearch and connecting with IAM authentication")
 }
 
 // InitFromViper initializes Options with properties from viper
@@ -169,6 +189,9 @@ func initFromViper(cfg *namespaceConfig, v *viper.Viper) {
 	cfg.BulkActions = v.GetInt(cfg.namespace + suffixBulkActions)
 	cfg.BulkFlushInterval = v.GetDuration(cfg.namespace + suffixBulkFlushInterval)
 	cfg.IndexPrefix = v.GetString(cfg.namespace + suffixIndexPrefix)
+	cfg.AwsConfig.AccessKeyID = v.GetString(cfg.namespace + suffixAwsIamAccessKeyID)
+	cfg.AwsConfig.SecretAccessKey = v.GetString(cfg.namespace + suffixAwsIamSecretAccessKey)
+	cfg.AwsConfig.Region = v.GetString(cfg.namespace + suffixAwsRegion)
 }
 
 // GetPrimary returns primary configuration.

diff --git a/plugin/storage/es/options_test.go b/plugin/storage/es/options_test.go
@@ -33,11 +33,15 @@ func TestOptions(t *testing.T) {
 	assert.Equal(t, int64(1), primary.NumReplicas)
 	assert.Equal(t, 72*time.Hour, primary.MaxSpanAge)
 	assert.False(t, primary.Sniffer)
+	assert.Equal(t, "", primary.AwsConfig.AccessKeyID)
+	assert.Equal(t, "", primary.AwsConfig.SecretAccessKey)
+	assert.Equal(t, "", primary.AwsConfig.Region)
 
 	aux := opts.Get("archive")
 	assert.Equal(t, primary.Username, aux.Username)
 	assert.Equal(t, primary.Password, aux.Password)
 	assert.Equal(t, primary.Servers, aux.Servers)
+	assert.Equal(t, primary.AwsConfig, aux.AwsConfig)
 }
 
 func TestOptionsWithFlags(t *testing.T) {
@@ -50,7 +54,9 @@ func TestOptionsWithFlags(t *testing.T) {
 		"--es.sniffer=true",
 		"--es.max-span-age=48h",
 		"--es.num-shards=20",
-		"--es.num-replicas=10",
+		"--es.aws.access_key_id=access-key",
+		"--es.aws.secret_access_key=the-secret",
+		"--es.aws.region=us-west-2",
 		// a couple overrides
 		"--es.aux.server-urls=3.3.3.3,4.4.4.4",
 		"--es.aux.max-span-age=24h",
@@ -62,6 +68,9 @@ func TestOptionsWithFlags(t *testing.T) {
 	assert.Equal(t, "hello", primary.Username)
 	assert.Equal(t, []string{"1.1.1.1", "2.2.2.2"}, primary.Servers)
 	assert.Equal(t, 48*time.Hour, primary.MaxSpanAge)
+	assert.Equal(t, "access-key", primary.AwsConfig.AccessKeyID)
+	assert.Equal(t, "the-secret", primary.AwsConfig.SecretAccessKey)
+	assert.Equal(t, "us-west-2", primary.AwsConfig.Region)
 	assert.True(t, primary.Sniffer)
 
 	aux := opts.Get("es.aux")