[filebeat] Add SSL and AUTH username support for Redis input (#40111)

* add ssl support

* add changelog entry

* add username support

* Update CHANGELOG.next.asciidoc

Co-authored-by: subham sarkar <sarkar.subhams2@gmail.com>

* add ssl in docs

* add username in docs

* add integration test

* goimports

* update docker compose redis service

* improve integration test

* keep old redis service

* address review

---------

Co-authored-by: subham sarkar <sarkar.subhams2@gmail.com>

CHANGELOG.next.asciidoc

@@ -318,6 +318,7 @@ https://github.com/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 - Implement Elastic Agent status and health reporting for Netflow Filebeat input. {pull}40080[40080]
 - Enhance input state reporting for CEL evaluations that return a single error object in events. {pull}40083[40083]
 - Allow absent credentials when using GCS with Application Default Credentials. {issue}39977[39977] {pull}40072[40072]
+- Add SSL and username support for Redis input, now the input includes support for Redis 6.0+. {pull}40111[40111]
 - Add scaling up support for Netflow input. {issue}37761[37761] {pull}40122[40122]
 - Update CEL mito extensions to v1.15.0. {pull}40294[40294]
 - Allow cross-region bucket configuration in s3 input. {issue}22161[22161] {pull}40309[40309]

filebeat/docker-compose.yml

@@ -10,6 +10,7 @@ services:
       kibana:        { condition: service_healthy }
       mosquitto:     { condition: service_healthy }
       redis:         { condition: service_healthy }
+      redis-tls:     { condition: service_healthy }
 
   elasticsearch:
     extends:
@@ -49,3 +50,15 @@ services:
     build: ${PWD}/input/redis/_meta
     ports:
       - 6379:6379
+
+  redis-tls:
+    build:
+      context: ${PWD}/input/redis/_meta
+      dockerfile: Dockerfile-tls
+    ports:
+      - 6380:6379
+    healthcheck:
+      test: ["CMD", "redis-cli", "-h", "localhost", "-p", "6379", "--tls", "--cert", "/certs/server-cert.pem", "--key", "/certs/server-key.pem", "--cacert", "/certs/root-ca.pem", "ping"]
+      interval: 10s
+      timeout: 5s
+      retries: 5

filebeat/docs/inputs/input-redis.asciidoc

@@ -39,6 +39,12 @@ The list of Redis hosts to connect to.
 
 The password to use when connecting to Redis.
 
+[float]
+[[redis-username]]
+===== `username`
+
+The username to use when connecting to Redis.
+
 [float]
 [[redis-scan_frequency]]
 ===== `scan_frequency`
@@ -79,3 +85,10 @@ include::../inputs/input-common-options.asciidoc[]
 
 :type!:
 
+[float]
+[[redis-ssl]]
+===== `ssl`
+
+Configuration options for SSL parameters like the certificate, key and the certificate authorities to use.
+
+See <<configuration-ssl>> for more information.

filebeat/input/redis/_meta/Dockerfile-tls

@@ -0,0 +1,16 @@
+FROM redis:latest
+
+# Copy the Redis configuration file
+COPY redis.conf /usr/local/etc/redis/redis.conf
+
+# Copy the certificates
+COPY certs /certs
+
+# Set the Redis password environment variable
+ENV REDIS_PASSWORD=password
+
+# Expose the Redis port
+EXPOSE 6379
+
+# Set the command to run Redis with the custom configuration
+CMD ["redis-server", "/usr/local/etc/redis/redis.conf"]

filebeat/input/redis/_meta/certs/root-ca.pem

@@ -0,0 +1,21 @@
+-----BEGIN CERTIFICATE-----
+MIIDdTCCAl2gAwIBAgIUPDh7kaMJ4tY3xCt5fkiRBMZS2cowDQYJKoZIhvcNAQEL
+BQAwSjELMAkGA1UEBhMCVVMxEzARBgNVBAgMCkNhbGlmb3JuaWExFDASBgNVBAcM
+C1NhbnRhIENsYXJhMRAwDgYDVQQDDAdmYWtlLUNBMB4XDTI0MDcwNDE5MTA1NVoX
+DTM0MDUxMzE5MTA1NVowSjELMAkGA1UEBhMCVVMxEzARBgNVBAgMCkNhbGlmb3Ju
+aWExFDASBgNVBAcMC1NhbnRhIENsYXJhMRAwDgYDVQQDDAdmYWtlLUNBMIIBIjAN
+BgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA0SFE9DkBPYxBqiCehTghPlSLN5pf
+BWNmrhWo2awpnarCIxwmiyEqBKPM8/nxnawyJoB9w3R8lqVHOG8IH7qhhyxXoPDZ
+yz8Tm9+vcFHY+fIC3o3/k3weM0McSSY+wIsir1SMLqdQ16MnYb5SlDHpsmUy4cXU
+YC1H4w+/dYa8CkBV7L9v+3PX2YRil5gkdOIbbPT+lvIsCjG2+gCmvygCpHl7vmg3
++bmphbv6Sd4UzCruksNTwf+wU4RtZEHvbE5wKgepIzFK+Cx+dgk0IC53fJM1pTiu
+ySO5zapi7lT14cqHA8So+QiT+HVxjfyajA84hCd9Ldwo3KmJNt7ei9Za6wIDAQAB
+o1MwUTAdBgNVHQ4EFgQUfofszMSp5sa1et/TaQolUVDrVG8wHwYDVR0jBBgwFoAU
+fofszMSp5sa1et/TaQolUVDrVG8wDwYDVR0TAQH/BAUwAwEB/zANBgkqhkiG9w0B
+AQsFAAOCAQEAQYR9ldRFcIFzWf8NBb219Zi7BvE2QmQ2c/A7hcjNbC7Nok4uIhai
+u2cShgsK+a7fSvDZWglcrLXniloL+zOOeHQGYGVYmYkMP8//tKZgVraAbq4xS6XJ
+DLQRKwoSDN1w9Ev6u5dLB0ml1sKGwhB73FhhJNl5fnB7/CICYlEeO0ShfO3FMP9q
+t1HhRGE9Deh56U0ZVFzqWByCv8UrqI4by2Ngm7n305EFpO8f7CcNFaM/t8ZjSpPm
+3hlVJAdKu+Bnf/VSrCDr85SQJGs6+q+CA1/L3a+UGgEe0eCKZBWW1x6jBXfOTyZ5
+GjWgBiuuGeZKKl6bAjcUuMNmB0LIKFBN2w==
+-----END CERTIFICATE-----

filebeat/input/redis/_meta/certs/server-cert.pem

@@ -0,0 +1,21 @@
+-----BEGIN CERTIFICATE-----
+MIIDcTCCAlmgAwIBAgIBATANBgkqhkiG9w0BAQsFADBKMQswCQYDVQQGEwJVUzET
+MBEGA1UECAwKQ2FsaWZvcm5pYTEUMBIGA1UEBwwLU2FudGEgQ2xhcmExEDAOBgNV
+BAMMB2Zha2UtQ0EwHhcNMjQwNzA0MTkxMDU3WhcNMzQwNTEzMTkxMDU3WjBOMQsw
+CQYDVQQGEwJVUzETMBEGA1UECAwKQ2FsaWZvcm5pYTEUMBIGA1UEBwwLU2FudGEg
+Q2xhcmExFDASBgNVBAMMC2Zha2Utc2VydmVyMIIBIjANBgkqhkiG9w0BAQEFAAOC
+AQ8AMIIBCgKCAQEAwTKEGs4jbbWUZfAvxyNpRl/evR4x9tarAfXYoZLXWVvoYElE
+ue7mm5wikm+YOt02Tq3fd4dUSB6j2/jap1nuuDbq/2pq0lOvkVqHoyN5kvBLtnDa
+617mabEjhvquQ6a3l3RP7ZWVuHOMAs6Ecgz7yLqtEmcX5dgHXRyoIYDhA4yN4uYp
+mcdofyxsWygMOME2v+h04Ap1aRgnCRy6kGaTUHpGEiIYwnNmbQQXPZBOLVwEa+TP
+u6DVkLOuFapLgjZ/AF0rYDkV6Q6FgYP5fsdRWqEfOpMyBZ4hxmikEN6CXrdWQb41
+SLKfCCeSy+D/14Nlv+MLv5OHSgQUMa7luxM4jQIDAQABo14wXDAaBgNVHREEEzAR
+hwQAAAAAgglsb2NhbGhvc3QwHQYDVR0OBBYEFO58Y193ie7o+8HMVYY6bIJDHKLg
+MB8GA1UdIwQYMBaAFH6H7MzEqebGtXrf02kKJVFQ61RvMA0GCSqGSIb3DQEBCwUA
+A4IBAQBNH9kNXgsld1uomByQbsOHKaw9iVJ63ZocvPTObnI3U5OJmvBFF7XLwWiO
+A50of4Rc2NF2lbTXB9/TK9ef7PWwwT0uwMdoI/c0alQzNxRkfUp2K1qykmRn5KkA
+0+v4IQxMp0bCa3cserK0ek/tr82hGY77CtirjpOu/inwJZ461oG7xKVQqW+3IX6H
+a2qgAQ6mMz2rD+iohi1/7FdNWzKHeOFL33g3klEJ5ZeYd+xla/nQbKNDxDUcFgNh
+046TaBEcv2OFBbbq6VYpRKrAe9XMqsy/fmV4Sk13GbfhMhSuc2nu+1yzSxohzF8A
+zhd0Srpssb9HtN4nBm4k3uUm78jo
+-----END CERTIFICATE-----

filebeat/input/redis/_meta/certs/server-key.pem

@@ -0,0 +1,28 @@
+-----BEGIN PRIVATE KEY-----
+MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQDBMoQaziNttZRl
+8C/HI2lGX969HjH21qsB9dihktdZW+hgSUS57uabnCKSb5g63TZOrd93h1RIHqPb
++NqnWe64Nur/amrSU6+RWoejI3mS8Eu2cNrrXuZpsSOG+q5DpreXdE/tlZW4c4wC
+zoRyDPvIuq0SZxfl2AddHKghgOEDjI3i5imZx2h/LGxbKAw4wTa/6HTgCnVpGCcJ
+HLqQZpNQekYSIhjCc2ZtBBc9kE4tXARr5M+7oNWQs64VqkuCNn8AXStgORXpDoWB
+g/l+x1FaoR86kzIFniHGaKQQ3oJet1ZBvjVIsp8IJ5LL4P/Xg2W/4wu/k4dKBBQx
+ruW7EziNAgMBAAECggEAAX4eN9pYXfS9cHsAeDmcBOibHw+06VgyBM/44JHKBqYD
+HM7JM053Amz/amHPYsMKi4g13sK/1oTl5uFLE7nhckI6i4bGy+oSSgYjAwDF0ljI
+4yYTNAPtEsK/NvQ8usHZppfANeJGOqcHKOvv3bVCeNVj2LNY+toDaF6ZSaxu2VkL
+JiB3ICYI4vAZP+ryEVwtD+pAncifWxkE9mAvRypgIJURvtOdTPJEiSoxRJoNh0Ld
+WsoELI8p7LSZF8I95BpVid3UF/NYf7mYBU69uCYIsSAeyq+fT7Lze0ZHbnPGMJOj
+UaaOecvlpIZSqxWLx5JmFONnY1Xq8VSCCbwrpj8Z+QKBgQDebvku9pej9/AXmKiy
+3j1gcrq4Xw7SazMRLHt1bSvlbkOQIdeA5bqT9mSrq8aRJbGIUvCm2YLzJ8PFhVNZ
+fVBPCu76ujpRvbUT2sJy++Q+GiFpPvlTLeiaO0U9A1VxKN63JBOz/mzaXisO0gOS
+MWjiGXomy0MPrYqCRHFChZbvtQKBgQDeWhgr0Tetst+Hntux8+gb+wudwnZX+RMN
+UeL0Cx/rOwR1EbJS6RvbhIOxWk9U7UTByngOcIiGldqmO1i4OxcQ49YnkU7qgyjx
+sguXjNv2dMu6lcesa4qZ3zXAm+GE81Vh2ew2rN58k6p9W8Kh0G/VyhWENV5+lk5u
+j+OHZce8eQKBgAJVx+fmTtE52RtmTt8R1jMdATjORqmO1opnnSQucTeHYM4yjMCd
+qMfE3mmu8/ayHpr/w+b8gZNr53I7ZBScbCtoQfn/2nzhMPV0ZnYujsbYH2Grd5KX
++MkltiRd1JfLhgsGJe7NzPa95lXRfpgaTK+S9OVTXPDdMYcMkOPR4zPBAoGAGedf
+F4O5O9gx6Gfeal3i9ZeKo+dqyBbxXETk7s94+XuXqlfUcYpMv4cxnHDL+zXlI7qF
+wBDmJt/AaEtTq6repg4U/ekUy4daNsYqSY6UdaLntSYL7A9fR0vUxEqkvEto8Axm
+U3xSMys02oPdKeLRlJOFbDCXgKHcI09KD5UQ7ukCgYEA0PwC985dSOyr0a8spGeb
+9LUfBQZ4dh8bOC+ACVu0P4SZyqxuPwxW7IMhmyK88VI6JyFAukSugs96As1H54hO
+P7rY82y+G3VwNIRDyDsEdb06i4r9t3mqvPKiOMX3NG+5z4Y067GQiR/rX5auSjMA
+2lo/JMIvnQTokcuLPooqUeA=
+-----END PRIVATE KEY-----

filebeat/input/redis/_meta/redis.conf

@@ -0,0 +1,9 @@
+port 0
+tls-port 6379
+tls-cert-file /certs/server-cert.pem
+tls-key-file /certs/server-key.pem
+tls-ca-cert-file /certs/root-ca.pem
+requirepass password
+user default on >password ~* &* +@all
+slowlog-log-slower-than 0
+slowlog-max-len 128

filebeat/input/redis/config.go

@@ -18,25 +18,31 @@
 package redis
 
 import (
+	"crypto/tls"
 	"time"
 
 	"github.com/elastic/beats/v7/filebeat/harvester"
+	"github.com/elastic/elastic-agent-libs/transport/tlscommon"
 )
 
-var defaultConfig = config{
-	ForwarderConfig: harvester.ForwarderConfig{
-		Type: "redis",
-	},
-	Network:  "tcp",
-	MaxConn:  10,
-	Password: "",
+func defaultConfig() config {
+	return config{
+		ForwarderConfig: harvester.ForwarderConfig{
+			Type: "redis",
+		},
+		Network: "tcp",
+		MaxConn: 10,
+	}
 }
 
 type config struct {
 	harvester.ForwarderConfig `config:",inline"`
-	Hosts                     []string      `config:"hosts" validate:"required"`
-	IdleTimeout               time.Duration `config:"idle_timeout"`
-	Network                   string        `config:"network"`
-	MaxConn                   int           `config:"maxconn" validate:"min=1"`
-	Password                  string        `config:"password"`
+	Hosts                     []string          `config:"hosts" validate:"required"`
+	IdleTimeout               time.Duration     `config:"idle_timeout"`
+	Network                   string            `config:"network"`
+	MaxConn                   int               `config:"maxconn" validate:"min=1"`
+	Username                  string            `config:"username"`
+	Password                  string            `config:"password"`
+	TLS                       *tlscommon.Config `config:"ssl"`
+	tlsConfig                 *tls.Config
 }

filebeat/input/redis/input.go

@@ -18,8 +18,6 @@
 package redis
 
 import (
-	"time"
-
 	rd "github.com/gomodule/redigo/redis"
 
 	"github.com/elastic/beats/v7/filebeat/channel"
@@ -29,6 +27,7 @@ import (
 	"github.com/elastic/beats/v7/libbeat/common/cfgwarn"
 	conf "github.com/elastic/elastic-agent-libs/config"
 	"github.com/elastic/elastic-agent-libs/logp"
+	"github.com/elastic/elastic-agent-libs/transport/tlscommon"
 )
 
 func init() {
@@ -51,13 +50,22 @@ type Input struct {
 func NewInput(cfg *conf.C, connector channel.Connector, context input.Context) (input.Input, error) {
 	cfgwarn.Experimental("Redis slowlog input is enabled.")
 
-	config := defaultConfig
+	config := defaultConfig()
 
 	err := cfg.Unpack(&config)
 	if err != nil {
 		return nil, err
 	}
 
+	if config.TLS.IsEnabled() {
+		tlsConfig, err := tlscommon.LoadTLSConfig(config.TLS)
+		if err != nil {
+			return nil, err
+		}
+
+		config.tlsConfig = tlsConfig.ToConfig()
+	}
+
 	out, err := connector.Connect(cfg)
 	if err != nil {
 		return nil, err
@@ -94,8 +102,7 @@ func (p *Input) Run() {
 
 	forwarder := harvester.NewForwarder(p.outlet)
 	for _, host := range p.config.Hosts {
-		pool := CreatePool(host, p.config.Password, p.config.Network,
-			p.config.MaxConn, p.config.IdleTimeout, p.config.IdleTimeout)
+		pool := CreatePool(host, p.config)
 
 		h, err := NewHarvester(pool.Get())
 		if err != nil {
@@ -121,28 +128,37 @@ func (p *Input) Wait() {}
 
 // CreatePool creates a redis connection pool
 // NOTE: This code is copied from the redis pool handling in metricbeat
-func CreatePool(
-	host, password, network string,
-	maxConn int,
-	idleTimeout, connTimeout time.Duration,
-) *rd.Pool {
+func CreatePool(host string, cfg config) *rd.Pool {
 	return &rd.Pool{
-		MaxIdle:     maxConn,
-		IdleTimeout: idleTimeout,
+		MaxIdle:     cfg.MaxConn,
+		IdleTimeout: cfg.IdleTimeout,
 		Dial: func() (rd.Conn, error) {
-			c, err := rd.Dial(network, host,
-				rd.DialConnectTimeout(connTimeout),
-				rd.DialReadTimeout(connTimeout),
-				rd.DialWriteTimeout(connTimeout))
+			dialOptions := []rd.DialOption{
+				rd.DialUsername(cfg.Username),
+				rd.DialConnectTimeout(cfg.IdleTimeout),
+				rd.DialReadTimeout(cfg.IdleTimeout),
+				rd.DialWriteTimeout(cfg.IdleTimeout),
+			}
+
+			if cfg.TLS.IsEnabled() && cfg.tlsConfig != nil {
+				dialOptions = append(dialOptions,
+					rd.DialUseTLS(true),
+					rd.DialTLSConfig(cfg.tlsConfig),
+				)
+			}
+
+			c, err := rd.Dial(cfg.Network, host, dialOptions...)
 			if err != nil {
 				return nil, err
 			}
-			if password != "" {
-				if _, err := c.Do("AUTH", password); err != nil {
+
+			if cfg.Password != "" {
+				if _, err := c.Do("AUTH", cfg.Password); err != nil {
 					c.Close()
 					return nil, err
 				}
 			}
+
 			return c, err
 		},
 	}