Merge pull request #27 from NBISweden/feature/es-tls

Add TLS config for ES

config.go

@@ -46,10 +46,10 @@ func getCLflags() ClFlags {
 
 	pflag.CommandLine.AddGoFlagSet(flag.CommandLine)
 	pflag.Parse()
-    err := viper.BindPFlags(pflag.CommandLine)
-    if err != nil {
-        log.Fatalf("Could not bind process flags for commandline: %v", err)
-    }
+	err := viper.BindPFlags(pflag.CommandLine)
+	if err != nil {
+		log.Fatalf("Could not bind process flags for commandline: %v", err)
+	}
 
 	action := viper.GetString("action")
 	batches := viper.GetInt("batches")
@@ -92,8 +92,12 @@ func configS3Storage() S3Config {
 // configElastic populates a ElasticConfig
 func configElastic() ElasticConfig {
 	elastic := ElasticConfig{}
-	elastic.User = viper.GetString("elastic.user")
-	elastic.Password = viper.GetString("elastic.password")
+	elastic.user = viper.GetString("elastic.user")
+	elastic.password = viper.GetString("elastic.password")
+	elastic.verifyPeer = viper.GetBool("elastic.verifypeer")
+	elastic.caCert = viper.GetString("elastic.cacert")
+	elastic.clientCert = viper.GetString("elastic.clientcert")
+	elastic.clientKey = viper.GetString("elastic.clientkey")
 
 	return elastic
 }

config.yaml

@@ -10,3 +10,7 @@ s3:
 elastic:
   user: "elastic"
   password: "elastic"
+  verifypeer: false
+  cacert: ""
+  clientcert: ""
+  clientkey: ""

elastic.go

@@ -3,8 +3,13 @@ package main
 import (
 	"bytes"
 	"context"
+	"crypto/tls"
+	"crypto/x509"
 	"fmt"
 	"io"
+	"io/ioutil"
+	"net/http"
+	"reflect"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -18,8 +23,62 @@ import (
 
 // ElasticConfig is a Struct that holds ElasticSearch config
 type ElasticConfig struct {
-	User     string
-	Password string
+	user       string
+	password   string
+	verifyPeer bool
+	caCert     string
+	clientCert string
+	clientKey  string
+}
+
+// transportConfigES is a helper method to setup TLS for the ES client.
+func transportConfigES(config ElasticConfig) http.RoundTripper {
+	cfg := new(tls.Config)
+
+	// Enforce TLS1.2 or higher
+	cfg.MinVersion = 2
+
+	// Read system CAs
+	var systemCAs, _ = x509.SystemCertPool()
+	if reflect.DeepEqual(systemCAs, x509.NewCertPool()) {
+		log.Debug("creating new CApool")
+		systemCAs = x509.NewCertPool()
+	}
+	cfg.RootCAs = systemCAs
+
+	if config.caCert != "" {
+		cacert, e := ioutil.ReadFile(config.caCert)
+		if e != nil {
+			log.Fatalf("failed to append %q to RootCAs: %v", cacert, e)
+		}
+		if ok := cfg.RootCAs.AppendCertsFromPEM(cacert); !ok {
+			log.Debug("no certs appended, using system certs only")
+		}
+	}
+
+	if config.verifyPeer {
+		if config.clientCert == "" || config.clientKey == "" {
+			log.Fatalf("No client cert or key were provided")
+		}
+
+		cert, e := ioutil.ReadFile(config.clientCert)
+		if e != nil {
+			log.Fatalf("failed to append client cert %q: %v", config.clientCert, e)
+		}
+		key, e := ioutil.ReadFile(config.clientKey)
+		if e != nil {
+			log.Fatalf("failed to append key %q: %v", config.clientKey, e)
+		}
+		if certs, e := tls.X509KeyPair(cert, key); e == nil {
+			cfg.Certificates = append(cfg.Certificates, certs)
+		}
+	}
+
+	var trConfig http.RoundTripper = &http.Transport{
+		TLSClientConfig:   cfg,
+		ForceAttemptHTTP2: true}
+
+	return trConfig
 }
 
 func readResponse(r io.Reader) string {

main.go

@@ -24,12 +24,14 @@ func main() {
 
 	retryBackoff := backoff.NewExponentialBackOff()
 
+	tr := transportConfigES(conf.Elastic)
+
 	c, err := elastic.NewClient(elasticsearch.Config{
 		Addresses: []string{
 			flags.instance,
 		},
-		Username:      conf.Elastic.User,
-		Password:      conf.Elastic.Password,
+		Username:      conf.Elastic.user,
+		Password:      conf.Elastic.password,
 		RetryOnStatus: []int{502, 503, 504, 429},
 		RetryBackoff: func(i int) time.Duration {
 			if i == 1 {
@@ -38,6 +40,7 @@ func main() {
 			return retryBackoff.NextBackOff()
 		},
 		MaxRetries: 5,
+		Transport:  tr,
 	})
 
 	if err != nil {