Support Client certificate authentication for opensearch/elastic search (#1406)

In addition to username and password, es/os support client-auth based
auth. See
[here](https://docs.opensearch.org/docs/latest/security/authentication-backends/client-auth/).
This pr wants to support that.

This change introduces client certificate authentication for Quesm when
connecting to Elasticsearch. It does not alter how users authenticate to
Quesm.

(this pr is to implement request of this
#1394)
<!-- A note on testing your PR -->
<!-- Basic unit test run is executed against each commit in the PR.
If you want to run a full integration test suite, you can trigger it by
commenting
     with '/run-integration-tests' or '/run-it' -->

---------

Co-authored-by: przemyslaw <[email protected]>

Author: hsuanyi

platform/backend_connectors/elasticsearch_backend_connector.go

@@ -6,7 +6,6 @@ package backend_connectors
 import (
 	"bytes"
 	"context"
-	"crypto/tls"
 	"fmt"
 	"github.com/QuesmaOrg/quesma/platform/config"
 	"github.com/QuesmaOrg/quesma/platform/elasticsearch"
@@ -32,32 +31,29 @@ type ElasticsearchBackendConnector struct {
 
 // NewElasticsearchBackendConnector is a constructor which uses old (v1) configuration object
 func NewElasticsearchBackendConnector(cfg config.ElasticsearchConfiguration) *ElasticsearchBackendConnector {
+	client := elasticsearch.NewHttpsClient(&cfg, esRequestTimeout)
 	conn := &ElasticsearchBackendConnector{
 		config: cfg,
-		client: &http.Client{
-			Transport: &http.Transport{
-				TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
-			},
-			Timeout: esRequestTimeout,
-		},
+		client: client,
 	}
 	return conn
 }
 
 // NewElasticsearchBackendConnectorFromDbConfig is an alternative constructor which uses the generic database configuration object
 func NewElasticsearchBackendConnectorFromDbConfig(cfg config.RelationalDbConfiguration) *ElasticsearchBackendConnector {
+	esConfig := config.ElasticsearchConfiguration{
+		Url:            cfg.Url,
+		User:           cfg.User,
+		Password:       cfg.Password,
+		ClientCertPath: cfg.ClientCertPath,
+		ClientKeyPath:  cfg.ClientKeyPath,
+		CACertPath:     cfg.CACertPath,
+	}
+
+	client := elasticsearch.NewHttpsClient(&esConfig, esRequestTimeout)
 	conn := &ElasticsearchBackendConnector{
-		config: config.ElasticsearchConfiguration{
-			Url:      cfg.Url,
-			User:     cfg.User,
-			Password: cfg.Password,
-		},
-		client: &http.Client{
-			Transport: &http.Transport{
-				TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
-			},
-			Timeout: esRequestTimeout,
-		},
+		config: esConfig,
+		client: client,
 	}
 	return conn
 }

platform/config/config_v2.go

@@ -94,6 +94,11 @@ type RelationalDbConfiguration struct {
 	ClusterName   string `koanf:"clusterName"` // When creating tables by Quesma - they'll use `ON CLUSTER ClusterName` clause
 	AdminUrl      *Url   `koanf:"adminUrl"`
 	DisableTLS    bool   `koanf:"disableTLS"`
+
+	// This supports es backend only.
+	ClientCertPath string `koanf:"clientCertPath"`
+	ClientKeyPath  string `koanf:"clientKeyPath"`
+	CACertPath     string `koanf:"caCertPath"`
 }
 
 func (c *RelationalDbConfiguration) IsEmpty() bool {
@@ -1069,9 +1074,12 @@ func (c *QuesmaNewConfiguration) getRelationalDBBackendConnector() (*BackendConn
 func (c *QuesmaNewConfiguration) getElasticsearchConfig() (ElasticsearchConfiguration, error) {
 	if esBackendConn := c.getElasticsearchBackendConnector(); esBackendConn != nil {
 		return ElasticsearchConfiguration{
-			Url:      esBackendConn.Config.Url,
-			User:     esBackendConn.Config.User,
-			Password: esBackendConn.Config.Password,
+			Url:            esBackendConn.Config.Url,
+			User:           esBackendConn.Config.User,
+			Password:       esBackendConn.Config.Password,
+			ClientCertPath: esBackendConn.Config.ClientCertPath,
+			ClientKeyPath:  esBackendConn.Config.ClientKeyPath,
+			CACertPath:     esBackendConn.Config.CACertPath,
 		}, nil
 	}
 	return ElasticsearchConfiguration{}, fmt.Errorf("elasticsearch backend connector must be configured")

platform/config/elasticsearch_config.go

@@ -7,4 +7,8 @@ type ElasticsearchConfiguration struct {
 	User     string `koanf:"user"`
 	Password string `koanf:"password"`
 	AdminUrl *Url   `koanf:"adminUrl"`
+
+	ClientCertPath string `koanf:"clientCertPath"`
+	ClientKeyPath  string `koanf:"clientKeyPath"`
+	CACertPath     string `koanf:"caCertPath"`
 }

platform/elasticsearch/client.go

@@ -6,10 +6,12 @@ import (
 	"bytes"
 	"context"
 	"crypto/tls"
+	"crypto/x509"
 	"fmt"
 	"github.com/QuesmaOrg/quesma/platform/config"
 	"github.com/QuesmaOrg/quesma/platform/logger"
 	"net/http"
+	"os"
 	"time"
 )
 
@@ -24,18 +26,53 @@ type SimpleClient struct {
 	config *config.ElasticsearchConfiguration
 }
 
-func NewSimpleClient(configuration *config.ElasticsearchConfiguration) *SimpleClient {
-	client := &http.Client{
+// NewHttpsClient should be merged with NewSimpleClient at some point -> TODO!
+func NewHttpsClient(configuration *config.ElasticsearchConfiguration, timeout time.Duration) *http.Client {
+	tlsConfig := &tls.Config{
+		MinVersion:         tls.VersionTLS12,
+		InsecureSkipVerify: true,
+	}
+
+	if configuration.CACertPath != "" {
+		caCert, err := os.ReadFile(configuration.CACertPath)
+		if err != nil {
+			logger.Warn().Msgf("failed to read CA certificate: %v. Fallback to skipping tls.", err)
+		} else {
+			caCertPool := x509.NewCertPool()
+			if !caCertPool.AppendCertsFromPEM(caCert) {
+				logger.Warn().Msgf("failed to append CA certificate: %v. Fallback to skipping tls.", err)
+			} else {
+				tlsConfig.RootCAs = caCertPool
+				tlsConfig.InsecureSkipVerify = false
+			}
+		}
+	}
+
+	if configuration.ClientCertPath != "" && configuration.ClientKeyPath != "" {
+		cert, err := tls.LoadX509KeyPair(configuration.ClientCertPath, configuration.ClientKeyPath)
+		if err != nil {
+			logger.Warn().Msgf("failed to load client certificate/key: %v. Fallback to certificate-less client.", err)
+		} else {
+			tlsConfig.Certificates = []tls.Certificate{cert}
+		}
+	}
+
+	return &http.Client{
 		Transport: &http.Transport{
-			TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
+			TLSClientConfig: tlsConfig,
 		},
-		Timeout: esRequestTimeout,
+		Timeout: timeout,
 	}
+}
+
+func NewSimpleClient(configuration *config.ElasticsearchConfiguration) *SimpleClient {
+	client := NewHttpsClient(configuration, esRequestTimeout)
 	return &SimpleClient{
 		client: client,
 		config: configuration,
 	}
 }
+
 func (es *SimpleClient) Request(ctx context.Context, method, endpoint string, body []byte) (*http.Response, error) {
 	return es.doRequest(ctx, method, endpoint, body, nil)
 }

platform/elasticsearch/client_test.go

@@ -4,13 +4,21 @@ package elasticsearch
 
 import (
 	"context"
+	"crypto/rand"
+	"crypto/rsa"
+	"crypto/x509"
+	"crypto/x509/pkix"
+	"encoding/pem"
 	"github.com/QuesmaOrg/quesma/platform/config"
 	"github.com/stretchr/testify/assert"
 	"io"
+	"math/big"
 	"net/http"
 	"net/http/httptest"
 	"net/url"
+	"os"
 	"testing"
+	"time"
 )
 
 const testPayload = "{'test': 'test'}"
@@ -108,3 +116,121 @@ func TestSimpleClient_RequestWithHeaders_OverwritesContentType(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Equal(t, resp.StatusCode, http.StatusOK)
 }
+
+func writeTempPEM(t *testing.T, prefix string, pemBytes []byte) string {
+	t.Helper()
+
+	tmpFile, err := os.CreateTemp("", prefix+"-*.pem")
+	if err != nil {
+		t.Fatalf("failed to create temp file: %v", err)
+	}
+	defer tmpFile.Close()
+
+	if _, err := tmpFile.Write(pemBytes); err != nil {
+		t.Fatalf("failed to write to temp file: %v", err)
+	}
+
+	t.Cleanup(func() {
+		os.Remove(tmpFile.Name())
+	})
+
+	return tmpFile.Name()
+}
+
+func generateTestCertAndKey(t *testing.T) (certPath, keyPath string) {
+	t.Helper()
+
+	private, err := rsa.GenerateKey(rand.Reader, 2048)
+	if err != nil {
+		t.Fatalf("failed to generate private key: %v", err)
+	}
+
+	template := x509.Certificate{
+		SerialNumber: big.NewInt(1),
+		Subject: pkix.Name{
+			Organization: []string{"Test Certificate"},
+		},
+		NotBefore:             time.Now(),
+		NotAfter:              time.Now().Add(365 * 24 * time.Hour),
+		KeyUsage:              x509.KeyUsageDigitalSignature | x509.KeyUsageKeyEncipherment,
+		ExtKeyUsage:           []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth, x509.ExtKeyUsageServerAuth},
+		BasicConstraintsValid: true,
+		IsCA:                  true,
+	}
+
+	certDER, err := x509.CreateCertificate(rand.Reader, &template, &template, &private.PublicKey, private)
+	if err != nil {
+		t.Fatalf("failed to create certificate: %v", err)
+	}
+
+	certPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: certDER})
+	keyPEM := pem.EncodeToMemory(&pem.Block{Type: "RSA PRIVATE KEY", Bytes: x509.MarshalPKCS1PrivateKey(private)})
+
+	certPath = writeTempPEM(t, "test-cert", certPEM)
+	keyPath = writeTempPEM(t, "test-key", keyPEM)
+	return certPath, keyPath
+}
+
+func TestNewHttpsClient_NoCerts(t *testing.T) {
+	conf := &config.ElasticsearchConfiguration{}
+	client := NewHttpsClient(conf, 5*time.Second)
+
+	tlsConfig := client.Transport.(*http.Transport).TLSClientConfig
+	assert.Nil(t, tlsConfig.RootCAs)
+	assert.Nil(t, tlsConfig.Certificates)
+	assert.True(t, tlsConfig.InsecureSkipVerify)
+}
+
+func TestNewHttpsClient_WithCACert(t *testing.T) {
+	caPath, _ := generateTestCertAndKey(t)
+
+	conf := &config.ElasticsearchConfiguration{
+		CACertPath: caPath,
+	}
+	client := NewHttpsClient(conf, 5*time.Second)
+
+	tlsConfig := client.Transport.(*http.Transport).TLSClientConfig
+	assert.NotNil(t, tlsConfig.RootCAs)
+	assert.Nil(t, tlsConfig.Certificates)
+	assert.False(t, tlsConfig.InsecureSkipVerify)
+}
+
+func TestNewHttpsClient_WithClientCert(t *testing.T) {
+	certPath, keyPath := generateTestCertAndKey(t) // real cert and key
+
+	conf := &config.ElasticsearchConfiguration{
+		ClientCertPath: certPath,
+		ClientKeyPath:  keyPath,
+	}
+	client := NewHttpsClient(conf, 5*time.Second)
+
+	tlsConfig := client.Transport.(*http.Transport).TLSClientConfig
+	assert.Nil(t, tlsConfig.RootCAs)
+	assert.NotNil(t, tlsConfig.Certificates)
+	assert.True(t, tlsConfig.InsecureSkipVerify)
+}
+
+func TestNewHttpsClient_InvalidCAPath(t *testing.T) {
+	conf := &config.ElasticsearchConfiguration{
+		CACertPath: "/nonexistent/file.pem",
+	}
+	client := NewHttpsClient(conf, 5*time.Second)
+
+	tlsConfig := client.Transport.(*http.Transport).TLSClientConfig
+	assert.Nil(t, tlsConfig.RootCAs)
+	assert.Nil(t, tlsConfig.Certificates)
+	assert.True(t, tlsConfig.InsecureSkipVerify)
+}
+
+func TestNewHttpsClient_InvalidClientCertPath(t *testing.T) {
+	conf := &config.ElasticsearchConfiguration{
+		ClientCertPath: "/invalid/cert.pem",
+		ClientKeyPath:  "/invalid/key.pem",
+	}
+	client := NewHttpsClient(conf, 5*time.Second)
+
+	tlsConfig := client.Transport.(*http.Transport).TLSClientConfig
+	assert.Nil(t, tlsConfig.RootCAs)
+	assert.Nil(t, tlsConfig.Certificates)
+	assert.True(t, tlsConfig.InsecureSkipVerify)
+}

platform/frontend_connectors/dispatcher.go

@@ -5,7 +5,6 @@ package frontend_connectors
 import (
 	"bytes"
 	"context"
-	"crypto/tls"
 	"errors"
 	"fmt"
 	"github.com/QuesmaOrg/quesma/platform/clickhouse"
@@ -32,6 +31,8 @@ import (
 	"time"
 )
 
+const httpClientTimeout = time.Minute
+
 func responseFromElastic(ctx context.Context, elkResponse *http.Response, w http.ResponseWriter) {
 	if id, ok := ctx.Value(tracing.RequestIdCtxKey).(string); ok {
 		logger.Debug().Str(logger.RID, id).Msgf("responding from Elasticsearch, status_code=%d", elkResponse.StatusCode)
@@ -90,14 +91,9 @@ func (r *Dispatcher) SetDependencies(deps quesma_api.Dependencies) {
 	r.debugInfoCollector = deps.DebugInfoCollector()
 	r.phoneHomeAgent = deps.PhoneHomeAgent()
 }
+
 func NewDispatcher(config *config.QuesmaConfiguration) *Dispatcher {
-	tr := &http.Transport{
-		TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
-	}
-	client := &http.Client{
-		Transport: tr,
-		Timeout:   time.Minute, // should be more configurable, 30s is Kibana default timeout
-	}
+	client := elasticsearch.NewHttpsClient(&config.Elasticsearch, httpClientTimeout)
 	requestProcessors := quesma_api.ProcessorChain{}
 	requestProcessors = append(requestProcessors, quesma_api.NewTraceIdPreprocessor())
 

platform/telemetry/phone_home.go

@@ -87,7 +87,8 @@ type agent struct {
 	recent            diag.PhoneHomeStats
 	telemetryEndpoint *config.Url
 
-	httpClient *http.Client
+	httpClient              *http.Client
+	elasticsearchHttpClient *http.Client
 }
 
 func generateInstanceID() string {
@@ -114,9 +115,7 @@ func NewPhoneHomeAgent(configuration *config.QuesmaConfiguration, clickHouseDb q
 	// TODO
 	// this is a question, maybe we should inherit context from the caller
 	// maybe the main function should be the one to cancel the context
-
 	ctx, cancel := context.WithCancel(context.Background())
-
 	return &agent{
 		ctx:                       ctx,
 		cancel:                    cancel,
@@ -141,6 +140,7 @@ func NewPhoneHomeAgent(configuration *config.QuesmaConfiguration, clickHouseDb q
 			},
 			Timeout: time.Minute,
 		},
+		elasticsearchHttpClient: elasticsearch.NewHttpsClient(&configuration.Elasticsearch, time.Minute),
 	}
 }
 
@@ -395,7 +395,7 @@ func (a *agent) callElastic(ctx context.Context, url *url.URL, response interfac
 		return err
 	}
 
-	resp, err := a.httpClient.Do(request)
+	resp, err := a.elasticsearchHttpClient.Do(request)
 	if err != nil {
 		logger.Error().Err(err).Msg("Error getting info from elasticsearch. ")
 		return err