From bc4c3fb577532725883949d09b25bb8465e69136 Mon Sep 17 00:00:00 2001 From: Achref Ben Saad Date: Fri, 20 Dec 2024 15:00:27 +0000 Subject: [PATCH 1/3] enable ca support for elastic adapter Signed-off-by: Achref Ben Saad --- relay-server/elasticsearch/adapter.go | 20 +++++++++++++++++++- relay-server/main.go | 7 ++++++- 2 files changed, 25 insertions(+), 2 deletions(-) diff --git a/relay-server/elasticsearch/adapter.go b/relay-server/elasticsearch/adapter.go index e2983ab..c01da99 100644 --- a/relay-server/elasticsearch/adapter.go +++ b/relay-server/elasticsearch/adapter.go @@ -3,9 +3,12 @@ package elasticsearch import ( "bytes" "context" + "crypto/tls" "encoding/json" "fmt" "log" + "net/http" + "os" "strings" "sync" "sync/atomic" @@ -38,7 +41,16 @@ type ElasticsearchClient struct { // NewElasticsearchClient creates a new Elasticsearch client with the given Elasticsearch URL // and kubearmor LogClient with endpoint. It has a retry mechanism for certain HTTP status codes and a backoff function for retry delays. // It then creates a new NewBulkIndexer with the esClient -func NewElasticsearchClient(esURL string, esUser string, esPassword string) (*ElasticsearchClient, error) { +func NewElasticsearchClient(esURL string, esUser string, esPassword string, esCaCertPath string, esAllowInsecureTLS bool) (*ElasticsearchClient, error) { + + caCertBytes := []byte{} + if esCaCertPath != "" { + var err error + caCertBytes, err = os.ReadFile(esCaCertPath) + if err != nil { + return nil, fmt.Errorf("failed to open Elasticsearch CA file: %v", err) + } + } retryBackoff := backoff.NewExponentialBackOff() cfg := elasticsearch.Config{ Addresses: []string{esURL}, @@ -54,6 +66,12 @@ func NewElasticsearchClient(esURL string, esUser string, esPassword string) (*El return retryBackoff.NextBackOff() }, MaxRetries: 5, + Transport: &http.Transport{ + TLSClientConfig: &tls.Config{ + InsecureSkipVerify: esAllowInsecureTLS, + }, + }, + CACert: caCertBytes, } if len(esUser) != 0 && len(esPassword) != 0 { diff --git a/relay-server/main.go b/relay-server/main.go index 11a39f1..532516b 100644 --- a/relay-server/main.go +++ b/relay-server/main.go @@ -59,6 +59,11 @@ func main() { esUrl := os.Getenv("ES_URL") esUser := os.Getenv("ES_USERNAME") esPassword := os.Getenv("ES_PASSWORD") + esCaCertPath := os.Getenv("ES_CA_CERT_PATH") + esAllowInsecureTLS := false + if os.Getenv("ES_ALLOW_INSECURE_TLS") != "" { + esAllowInsecureTLS = true + } esAlertsIndex := os.Getenv("ES_ALERTS_INDEX") if esAlertsIndex == "" { esAlertsIndex = "kubearmor-alerts" @@ -90,7 +95,7 @@ func main() { // check and start an elasticsearch client if enableEsDashboards == "true" { - esCl, err := elasticsearch.NewElasticsearchClient(esUrl, esUser, esPassword) + esCl, err := elasticsearch.NewElasticsearchClient(esUrl, esUser, esPassword, esCaCertPath, esAllowInsecureTLS) if err != nil { kg.Warnf("Failed to start a Elasticsearch Client") return From 9a65100c5c008a1529520e5cdf516a9a4fa21902 Mon Sep 17 00:00:00 2001 From: Achref Ben Saad Date: Fri, 20 Dec 2024 15:40:47 +0000 Subject: [PATCH 2/3] uignore G304 Signed-off-by: Achref Ben Saad --- relay-server/Makefile | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/relay-server/Makefile b/relay-server/Makefile index 63153bf..de12e11 100644 --- a/relay-server/Makefile +++ b/relay-server/Makefile @@ -48,4 +48,4 @@ ifeq (, $(shell which gosec)) go install github.com/securego/gosec/v2/cmd/gosec@latest;\ } endif - cd $(CURDIR); gosec -exclude=G402 ./... + cd $(CURDIR); gosec -exclude=G402,G304 ./... From b9617bbcb25e93fa526ba234ea033b32100c6c2f Mon Sep 17 00:00:00 2001 From: Achref Ben Saad Date: Sat, 21 Dec 2024 08:19:11 +0000 Subject: [PATCH 3/3] fix bug Signed-off-by: Achref Ben Saad --- relay-server/elasticsearch/adapter.go | 24 +++++++++++++----------- relay-server/main.go | 2 +- 2 files changed, 14 insertions(+), 12 deletions(-) diff --git a/relay-server/elasticsearch/adapter.go b/relay-server/elasticsearch/adapter.go index c01da99..49e98c5 100644 --- a/relay-server/elasticsearch/adapter.go +++ b/relay-server/elasticsearch/adapter.go @@ -43,14 +43,6 @@ type ElasticsearchClient struct { // It then creates a new NewBulkIndexer with the esClient func NewElasticsearchClient(esURL string, esUser string, esPassword string, esCaCertPath string, esAllowInsecureTLS bool) (*ElasticsearchClient, error) { - caCertBytes := []byte{} - if esCaCertPath != "" { - var err error - caCertBytes, err = os.ReadFile(esCaCertPath) - if err != nil { - return nil, fmt.Errorf("failed to open Elasticsearch CA file: %v", err) - } - } retryBackoff := backoff.NewExponentialBackOff() cfg := elasticsearch.Config{ Addresses: []string{esURL}, @@ -71,7 +63,14 @@ func NewElasticsearchClient(esURL string, esUser string, esPassword string, esCa InsecureSkipVerify: esAllowInsecureTLS, }, }, - CACert: caCertBytes, + } + + if esCaCertPath != "" { + caCertBytes, err := os.ReadFile(esCaCertPath) + if err != nil { + return nil, fmt.Errorf("failed to open Elasticsearch CA file: %v", err) + } + cfg.CACert = caCertBytes } if len(esUser) != 0 && len(esPassword) != 0 { @@ -86,10 +85,13 @@ func NewElasticsearchClient(esURL string, esUser string, esPassword string, esCa bi, err := esutil.NewBulkIndexer(esutil.BulkIndexerConfig{ Client: esClient, // The Elasticsearch client FlushBytes: 1000000, // The flush threshold in bytes [1mb] - FlushInterval: 30 * time.Second, // The periodic flush interval [30 secs] + FlushInterval: 10 * time.Second, // The periodic flush interval [30 secs] + OnError: func(ctx context.Context, err error) { + log.Fatalf("Error creating the indexer: %v", err) + }, }) if err != nil { - log.Fatalf("Error creating the indexer: %s", err) + log.Fatalf("Error creating the indexer: %v", err) } alertCh := make(chan interface{}, 10000) return &ElasticsearchClient{bulkIndexer: bi, esClient: esClient, alertCh: alertCh}, nil diff --git a/relay-server/main.go b/relay-server/main.go index 532516b..3128e0a 100644 --- a/relay-server/main.go +++ b/relay-server/main.go @@ -97,7 +97,7 @@ func main() { if enableEsDashboards == "true" { esCl, err := elasticsearch.NewElasticsearchClient(esUrl, esUser, esPassword, esCaCertPath, esAllowInsecureTLS) if err != nil { - kg.Warnf("Failed to start a Elasticsearch Client") + kg.Warnf("Failed to start a Elasticsearch Client, %v", err) return } relayServer.ELKClient = esCl