Fix remote ES client to use ssl.* settings (#3522)

nchaulet · web-flow · commit 0d09f4cd73a3 · 2024-05-07T09:31:42.000-04:00
diff --git a/Makefile b/Makefile
@@ -305,7 +305,7 @@ int-docker-start-async:
 .PHONY: int-docker-wait
 int-docker-wait:
 	@./dev-tools/integration/wait-for-elasticsearch.sh ${ELASTICSEARCH_USERNAME}:${ELASTICSEARCH_PASSWORD}@${TEST_ELASTICSEARCH_HOSTS}
-	@./dev-tools/integration/wait-for-elasticsearch.sh ${ELASTICSEARCH_USERNAME}:${ELASTICSEARCH_PASSWORD}@${TEST_REMOTE_ELASTICSEARCH_HOST}
+	@./dev-tools/integration/wait-for-elasticsearch.sh https://${ELASTICSEARCH_USERNAME}:${ELASTICSEARCH_PASSWORD}@${TEST_REMOTE_ELASTICSEARCH_HOST}
 
 # Start integration docker setup with wait for when the ES is ready
 .PHONY: int-docker-start
@@ -335,7 +335,8 @@ test-int: prepare-test-context  ## - Run integration tests with full setup (slow
 test-int-set: ## - Run integration tests without setup
 	# Initialize indices one before running all the tests
 	ELASTICSEARCH_SERVICE_TOKEN=$(shell ./dev-tools/integration/get-elasticsearch-servicetoken.sh ${ELASTICSEARCH_USERNAME}:${ELASTICSEARCH_PASSWORD}@${TEST_ELASTICSEARCH_HOSTS} "fleet-server") \
-	REMOTE_ELASTICSEARCH_SERVICE_TOKEN=$(shell ./dev-tools/integration/get-elasticsearch-servicetoken.sh ${ELASTICSEARCH_USERNAME}:${ELASTICSEARCH_PASSWORD}@${TEST_REMOTE_ELASTICSEARCH_HOST} "fleet-server-remote") \
+	REMOTE_ELASTICSEARCH_SERVICE_TOKEN=$(shell ./dev-tools/integration/get-elasticsearch-servicetoken.sh https://${ELASTICSEARCH_USERNAME}:${ELASTICSEARCH_PASSWORD}@${TEST_REMOTE_ELASTICSEARCH_HOST} "fleet-server-remote") \
+	REMOTE_ELASTICSEARCH_CA_CRT_BASE64="$(shell COMPOSE_PROJECT_NAME=integration docker compose  -f ./dev-tools/e2e/docker-compose.yml --env-file ./dev-tools/integration/.env exec elasticsearch-remote /bin/bash -c "cat /usr/share/elasticsearch/config/certs/ca/ca.crt" | base64)" \
 	ELASTICSEARCH_HOSTS=${TEST_ELASTICSEARCH_HOSTS} ELASTICSEARCH_USERNAME=${ELASTICSEARCH_USERNAME} ELASTICSEARCH_PASSWORD=${ELASTICSEARCH_PASSWORD} \
 	go test -v -tags=integration -count=1 -race -p 1 ./...
 
diff --git a/dev-tools/integration/docker-compose.yml b/dev-tools/integration/docker-compose.yml
@@ -1,6 +1,58 @@
   
 version: '2.3'
+volumes:
+  certs:
+    driver: local
+
 services:
+  setup:
+    image: docker.elastic.co/elasticsearch/elasticsearch:${ELASTICSEARCH_VERSION}-amd64
+    volumes:
+      - certs:/usr/share/elasticsearch/config/certs
+    user: "0"
+    healthcheck:
+      test: ["CMD-SHELL", "[ -f config/certs/es01/es01.crt ]"]
+      interval: 1s
+      timeout: 5s
+      retries: 120
+    command: >
+      bash -c '
+        if [ ! -f config/certs/ca.zip ]; then
+          echo "Creating CA";
+          bin/elasticsearch-certutil ca --silent --pem -out config/certs/ca.zip;
+          unzip config/certs/ca.zip -d config/certs;
+        fi;
+        if [ ! -f config/certs/certs.zip ]; then
+          echo "Creating certs";
+          echo -ne \
+          "instances:\n"\
+          "  - name: es01\n"\
+          "    dns:\n"\
+          "      - es01\n"\
+          "      - localhost\n"\
+          "    ip:\n"\
+          "      - 127.0.0.1\n"\
+          "  - name: es02\n"\
+          "    dns:\n"\
+          "      - es02\n"\
+          "      - localhost\n"\
+          "    ip:\n"\
+          "      - 127.0.0.1\n"\
+          "  - name: es03\n"\
+          "    dns:\n"\
+          "      - es03\n"\
+          "      - localhost\n"\
+          "    ip:\n"\
+          "      - 127.0.0.1\n"\
+          > config/certs/instances.yml;
+          bin/elasticsearch-certutil cert --silent --pem -out config/certs/certs.zip --in config/certs/instances.yml --ca-cert config/certs/ca/ca.crt --ca-key config/certs/ca/ca.key;
+          unzip config/certs/certs.zip -d config/certs;
+        fi;
+        echo "Setting file permissions"
+        chown -R root:root config/certs;
+        find . -type d -exec chmod 750 \{\} \;;
+        find . -type f -exec chmod 640 \{\} \;;
+      ';
   elasticsearch:
     image: "docker.elastic.co/elasticsearch/elasticsearch:${ELASTICSEARCH_VERSION}-amd64"
     container_name: elasticsearch
@@ -25,13 +77,30 @@ services:
       - 127.0.0.1:9200:9200
       
   elasticsearch-remote:
+    depends_on:
+      setup:
+        condition: service_healthy
     image: "docker.elastic.co/elasticsearch/elasticsearch:${ELASTICSEARCH_VERSION}-amd64"
     container_name: elasticsearch-remote
     environment:
       - node.name=es02
       - cluster.name=es-docker-cluster2
       - discovery.seed_hosts=elasticsearch
       - bootstrap.memory_lock=true
+      - xpack.security.enabled=true
+      - xpack.security.http.ssl.enabled=true
+      - xpack.security.http.ssl.key=certs/es02/es02.key
+      - xpack.security.http.ssl.certificate=certs/es02/es02.crt
+      - xpack.security.http.ssl.certificate_authorities=certs/ca/ca.crt
+      - xpack.security.transport.ssl.enabled=true
+      - xpack.security.transport.ssl.key=certs/es02/es02.key
+      - xpack.security.transport.ssl.certificate=certs/es02/es02.crt
+      - xpack.security.transport.ssl.certificate_authorities=certs/ca/ca.crt
+      - xpack.security.transport.ssl.verification_mode=certificate
+      - cluster.name="docker-cluster"
+      - network.host="0.0.0.0"
+      - xpack.security.authc.api_key.enabled="true"
+      - xpack.license.self_generated.type="trial"
       - "ES_JAVA_OPTS=-Xms1G -Xmx1G"
       - "ELASTIC_USERNAME=${ELASTICSEARCH_USERNAME}"
       - "ELASTIC_PASSWORD=${ELASTICSEARCH_PASSWORD}"
@@ -43,6 +112,7 @@ services:
           soft: 65536
           hard: 65536
     volumes:
+      - certs:/usr/share/elasticsearch/config/certs
       - ./elasticsearch.yml:/usr/share/elasticsearch/config/elasticsearch.yml
     ports:
       - 127.0.0.1:9201:9200
diff --git a/dev-tools/integration/get-elasticsearch-servicetoken.sh b/dev-tools/integration/get-elasticsearch-servicetoken.sh
@@ -3,7 +3,7 @@
 host="$1"
 account="$2"
 
-jsonBody="$(curl -sSL -XPOST "$host/_security/service/elastic/$account/credential/token/token1")"
+jsonBody="$(curl --insecure -sSL -XPOST "$host/_security/service/elastic/$account/credential/token/token1")"
 
 # use grep and sed to get the service token value as we may not have jq or a similar tool on the instance
 token=$(echo ${jsonBody} |  grep -Eo '"value"[^}]*' | grep -Eo ':.*' | sed -r "s/://" | sed -r 's/"//g')
diff --git a/dev-tools/integration/wait-for-elasticsearch.sh b/dev-tools/integration/wait-for-elasticsearch.sh
@@ -7,27 +7,27 @@ shift
 cmd="$@"
 
 
-until $(curl --output /dev/null --silent --head --fail "$host"); do
+until $(curl --insecure --output /dev/null --silent --head --fail "$host"); do
     printf '.'
     sleep 1
 done
 
 # First wait for ES to start...
-response=$(curl $host)
+response=$(curl --insecure $host)
 
 until [ "$response" = "200" ]; do
-    response=$(curl --write-out %{http_code} --silent --output /dev/null "$host")
+    response=$(curl --insecure --write-out %{http_code} --silent --output /dev/null "$host")
     echo '.'
     sleep 1
 done
 
 
 # next wait for ES status to turn to green
-health="$(curl -fsSL "$host/_cat/health?h=status")"
+health="$(curl --insecure -fsSL "$host/_cat/health?h=status")"
 health="$(echo "$health" | tr -d '[:space:]')"
 
 until [ "$health" = 'green' -o "$health" = 'yellow' ]; do
-    health="$(curl -fsSL "$host/_cat/health?h=status")"
+    health="$(curl --insecure -fsSL "$host/_cat/health?h=status")"
     echo $health
     health="$(echo "$health" | tr -d '[:space:]')"
     >&2 echo "Elasticsearch is unavailable - sleeping"
diff --git a/internal/pkg/bulk/engine.go b/internal/pkg/bulk/engine.go
@@ -19,6 +19,7 @@ import (
 	"github.com/elastic/fleet-server/v7/internal/pkg/config"
 	"github.com/elastic/fleet-server/v7/internal/pkg/es"
 	"github.com/elastic/fleet-server/v7/internal/pkg/logger"
+	"github.com/elastic/go-ucfg"
 
 	"github.com/elastic/go-elasticsearch/v8"
 	"github.com/elastic/go-elasticsearch/v8/esapi"
@@ -198,29 +199,24 @@ func (b *Bulker) CreateAndGetBulker(ctx context.Context, zlog zerolog.Logger, ou
 var newESClient = es.NewClient
 
 func (b *Bulker) createRemoteEsClient(ctx context.Context, outputName string, outputMap map[string]map[string]interface{}) (*elasticsearch.Client, error) {
-	hostsObj := outputMap[outputName]["hosts"]
-	hosts, ok := hostsObj.([]interface{})
-	if !ok {
-		return nil, fmt.Errorf("failed to get hosts from output: %v", hostsObj)
+	var esOutput config.Elasticsearch
+	esConfig, err := ucfg.NewFrom(outputMap[outputName], config.DefaultOptions...)
+	if err != nil {
+		return nil, err
 	}
-	hostsStrings := make([]string, len(hosts))
-	for i, host := range hosts {
-		hostsStrings[i], ok = host.(string)
-		if !ok {
-			return nil, fmt.Errorf("failed to get hosts from output: %v", host)
-		}
+	err = esConfig.Unpack(&esOutput)
+	if err != nil {
+		return nil, err
 	}
-	serviceToken, ok := outputMap[outputName]["service_token"].(string)
-	if !ok {
+	if len(esOutput.Hosts) == 0 {
+		return nil, fmt.Errorf("failed to get hosts from output: %v", outputName)
+	}
+	if esOutput.ServiceToken == "" {
 		return nil, fmt.Errorf("failed to get service token from output: %v", outputName)
 	}
-
 	cfg := config.Config{
 		Output: config.Output{
-			Elasticsearch: config.Elasticsearch{
-				Hosts:        hostsStrings,
-				ServiceToken: serviceToken,
-			},
+			Elasticsearch: esOutput,
 		},
 	}
 	es, err := newESClient(ctx, &cfg, false, elasticsearchOptions(
diff --git a/internal/pkg/server/remote_es_output_integration_test.go b/internal/pkg/server/remote_es_output_integration_test.go
@@ -8,6 +8,8 @@ package server
 
 import (
 	"context"
+	"crypto/tls"
+	"encoding/base64"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -28,6 +30,7 @@ import (
 
 const (
 	remoteESHost = "localhost:9201"
+	remoteESUrl  = "https://localhost:9201"
 )
 
 func Checkin(t *testing.T, ctx context.Context, srv *tserver, agentID, key string, shouldHaveRemoteES bool, actionType string) (string, string) {
@@ -96,6 +99,13 @@ func Checkin(t *testing.T, ctx context.Context, srv *tserver, agentID, key strin
 	return remoteAPIKey, actionID
 }
 
+func getRemoteElasticsearchCa(t *testing.T) string {
+	data, err := base64.StdEncoding.DecodeString(strings.Replace(os.Getenv("REMOTE_ELASTICSEARCH_CA_CRT_BASE64"), " ", "", -1))
+	require.NoError(t, err)
+
+	return string(data)
+}
+
 func Ack(t *testing.T, ctx context.Context, srv *tserver, actionID, agentID, key string) {
 	t.Logf("Fake an ack for action %s for agent %s", actionID, agentID)
 	body := fmt.Sprintf(`{
@@ -146,9 +156,11 @@ func Test_Agent_Remote_ES_Output(t *testing.T) {
 				"type": "elasticsearch",
 			},
 			"remoteES": {
-				"type":          "remote_elasticsearch",
-				"hosts":         []string{remoteESHost},
-				"service_token": os.Getenv("REMOTE_ELASTICSEARCH_SERVICE_TOKEN"),
+				"type":                        "remote_elasticsearch",
+				"hosts":                       []string{remoteESUrl},
+				"service_token":               os.Getenv("REMOTE_ELASTICSEARCH_SERVICE_TOKEN"),
+				"ssl.enabled":                 true,
+				"ssl.certificate_authorities": []string{getRemoteElasticsearchCa(t)},
 			},
 		},
 		OutputPermissions: json.RawMessage(`{"default": {}, "remoteES": {}}`),
@@ -255,13 +267,19 @@ func verifyRemoteAPIKey(t *testing.T, ctx context.Context, apiKeyID string, inva
 	// need to wait a bit before querying the api key
 	time.Sleep(time.Second)
 
-	requestURL := fmt.Sprintf("http://elastic:changeme@%s/_security/api_key?id=%s", remoteESHost, apiKeyID)
+	requestURL := fmt.Sprintf("https://elastic:changeme@%s/_security/api_key?id=%s", remoteESHost, apiKeyID)
 
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, requestURL, nil)
+	// Skip SSL verify as ES use self-signed certificate
+	tr := &http.Transport{
+		// #nosec G402
+		TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
+	}
+	client := &http.Client{Transport: tr}
 	if err != nil {
 		t.Fatal("error creating request for remote api key")
 	}
-	res, err := http.DefaultClient.Do(req)
+	res, err := client.Do(req)
 	if err != nil {
 		t.Fatal("error querying remote api key")
 	}
@@ -292,9 +310,11 @@ func Test_Agent_Remote_ES_Output_ForceUnenroll(t *testing.T) {
 				"type": "elasticsearch",
 			},
 			"remoteES": {
-				"type":          "remote_elasticsearch",
-				"hosts":         []string{remoteESHost},
-				"service_token": os.Getenv("REMOTE_ELASTICSEARCH_SERVICE_TOKEN"),
+				"type":                        "remote_elasticsearch",
+				"hosts":                       []string{remoteESUrl},
+				"service_token":               os.Getenv("REMOTE_ELASTICSEARCH_SERVICE_TOKEN"),
+				"ssl.enabled":                 true,
+				"ssl.certificate_authorities": []string{getRemoteElasticsearchCa(t)},
 			},
 		},
 		OutputPermissions: json.RawMessage(`{"default": {}, "remoteES": {}}`),
@@ -411,9 +431,11 @@ func Test_Agent_Remote_ES_Output_Unenroll(t *testing.T) {
 				"type": "elasticsearch",
 			},
 			"remoteES": {
-				"type":          "remote_elasticsearch",
-				"hosts":         []string{remoteESHost},
-				"service_token": os.Getenv("REMOTE_ELASTICSEARCH_SERVICE_TOKEN"),
+				"type":                        "remote_elasticsearch",
+				"hosts":                       []string{remoteESUrl},
+				"service_token":               os.Getenv("REMOTE_ELASTICSEARCH_SERVICE_TOKEN"),
+				"ssl.enabled":                 true,
+				"ssl.certificate_authorities": []string{getRemoteElasticsearchCa(t)},
 			},
 		},
 		OutputPermissions: json.RawMessage(`{"default": {}, "remoteES": {}}`),
