code factory

config.yml

@@ -477,6 +477,14 @@ multiplexer:
 #   tls-support: false
 #   # insecure skip verify
 #   tls-insecure: false
+#   # tls min version
+#   tls-min-version: 1.2
+#   # provide CA file to verify the server certificate
+#   ca-file: ""
+#   # provide client certificate file for mTLS
+#   cert-file: ""
+#   # provide client private key file for mTLS
+#   key-file: ""
 #   # how many DNS messages will be buffered before being sent
 #   buffer-size: 100
 #   # Channel buffer size for incoming packets, number of packet before to drop it.
@@ -492,6 +500,14 @@ multiplexer:
 #   tls-support: false
 #   # insecure skip verify
 #   tls-insecure: false
+#   # tls min version
+#   tls-min-version: 1.2
+#   # provide CA file to verify the server certificate
+#   ca-file: ""
+#   # provide client certificate file for mTLS
+#   cert-file: ""
+#   # provide client private key file for mTLS
+#   key-file: ""
 #   # bucket
 #   bucket: "db_dns"
 #   # Organization
@@ -520,6 +536,14 @@ multiplexer:
 #   proxy-url: ""
 #   # insecure skip verify
 #   tls-insecure: false
+#   # tls min version
+#   tls-min-version: 1.2
+#   # provide CA file to verify the server certificate
+#   ca-file: ""
+#   # provide client certificate file for mTLS
+#   cert-file: ""
+#   # provide client private key file for mTLS
+#   key-file: ""
 #   # basic auth login
 #   basic-auth-login: ""
 #   # basic auth password

dnsutils/config.go

@@ -366,6 +366,9 @@ type Config struct {
 			TlsSupport        bool   `yaml:"tls-support"`
 			TlsInsecure       bool   `yaml:"tls-insecure"`
 			TlsMinVersion     string `yaml:"tls-min-version"`
+			CAFile            string `yaml:"ca-file"`
+			CertFile          string `yaml:"cert-file"`
+			KeyFile           string `yaml:"key-file"`
 			Tag               string `yaml:"tag"`
 			BufferSize        int    `yaml:"buffer-size"`
 			ChannelBufferSize int    `yaml:"chan-buffer-size"`
@@ -377,6 +380,9 @@ type Config struct {
 			TlsSupport        bool   `yaml:"tls-support"`
 			TlsInsecure       bool   `yaml:"tls-insecure"`
 			TlsMinVersion     string `yaml:"tls-min-version"`
+			CAFile            string `yaml:"ca-file"`
+			CertFile          string `yaml:"cert-file"`
+			KeyFile           string `yaml:"key-file"`
 			Bucket            string `yaml:"bucket"`
 			Organization      string `yaml:"organization"`
 			ChannelBufferSize int    `yaml:"chan-buffer-size"`
@@ -393,6 +399,9 @@ type Config struct {
 			ProxyURL          string            `yaml:"proxy-url"`
 			TlsInsecure       bool              `yaml:"tls-insecure"`
 			TlsMinVersion     string            `yaml:"tls-min-version"`
+			CAFile            string            `yaml:"ca-file"`
+			CertFile          string            `yaml:"cert-file"`
+			KeyFile           string            `yaml:"key-file"`
 			BasicAuthLogin    string            `yaml:"basic-auth-login"`
 			BasicAuthPwd      string            `yaml:"basic-auth-pwd"`
 			BasicAuthPwdFile  string            `yaml:"basic-auth-pwd-file"`
@@ -465,6 +474,9 @@ type Config struct {
 			TlsSupport        bool   `yaml:"tls-support"`
 			TlsInsecure       bool   `yaml:"tls-insecure"`
 			TlsMinVersion     string `yaml:"tls-min-version"`
+			CAFile            string `yaml:"ca-file"`
+			CertFile          string `yaml:"cert-file"`
+			KeyFile           string `yaml:"key-file"`
 			SaslSupport       bool   `yaml:"sasl-support"`
 			SaslUsername      string `yaml:"sasl-username"`
 			SaslPassword      string `yaml:"sasl-password"`
@@ -692,6 +704,9 @@ func (c *Config) SetDefault() {
 	c.Loggers.Fluentd.TlsSupport = false
 	c.Loggers.Fluentd.TlsInsecure = false
 	c.Loggers.Fluentd.TlsMinVersion = TLS_v12
+	c.Loggers.Fluentd.CAFile = ""
+	c.Loggers.Fluentd.CertFile = ""
+	c.Loggers.Fluentd.KeyFile = ""
 	c.Loggers.Fluentd.Tag = "dns.collector"
 	c.Loggers.Fluentd.BufferSize = 100
 	c.Loggers.Fluentd.ChannelBufferSize = 65535
@@ -702,6 +717,9 @@ func (c *Config) SetDefault() {
 	c.Loggers.InfluxDB.TlsSupport = false
 	c.Loggers.InfluxDB.TlsInsecure = false
 	c.Loggers.InfluxDB.TlsMinVersion = TLS_v12
+	c.Loggers.InfluxDB.CAFile = ""
+	c.Loggers.InfluxDB.CertFile = ""
+	c.Loggers.InfluxDB.KeyFile = ""
 	c.Loggers.InfluxDB.Bucket = ""
 	c.Loggers.InfluxDB.Organization = ""
 	c.Loggers.InfluxDB.ChannelBufferSize = 65535
@@ -717,6 +735,9 @@ func (c *Config) SetDefault() {
 	c.Loggers.LokiClient.ProxyURL = ""
 	c.Loggers.LokiClient.TlsInsecure = false
 	c.Loggers.LokiClient.TlsMinVersion = TLS_v12
+	c.Loggers.LokiClient.CAFile = ""
+	c.Loggers.LokiClient.CertFile = ""
+	c.Loggers.LokiClient.KeyFile = ""
 	c.Loggers.LokiClient.BasicAuthLogin = ""
 	c.Loggers.LokiClient.BasicAuthPwd = ""
 	c.Loggers.LokiClient.BasicAuthPwdFile = ""
@@ -769,6 +790,9 @@ func (c *Config) SetDefault() {
 	c.Loggers.KafkaProducer.TlsSupport = false
 	c.Loggers.KafkaProducer.TlsInsecure = false
 	c.Loggers.KafkaProducer.TlsMinVersion = TLS_v12
+	c.Loggers.KafkaProducer.CAFile = ""
+	c.Loggers.KafkaProducer.CertFile = ""
+	c.Loggers.KafkaProducer.KeyFile = ""
 	c.Loggers.KafkaProducer.SaslSupport = false
 	c.Loggers.KafkaProducer.SaslUsername = ""
 	c.Loggers.KafkaProducer.SaslPassword = ""

docs/loggers/logger_fluentd.md

@@ -9,7 +9,7 @@ Fluentd client to remote server or unix socket.
 
 Options:
 
-* `transport`: (string) network transport to use: tcp|unix
+* `transport`: (string) network transport to use: `tcp`|`unix`
 * `listen-ip`: (string) remote address
 * `listen-port`: (integer) remote tcp port
 * `sock-path`: (string) unix socket path
@@ -20,6 +20,9 @@ Options:
 * `tls-support`: (boolean) enable tls
 * `tls-insecure`: (boolean) insecure skip verify
 * `tls-min-version`: (string) min tls version, default to 1.2
+* `ca-file`: (string) provide CA file to verify the server certificate
+* `cert-file`: (string) provide client certificate file for mTLS
+* `key-file`: (string) provide client private key file for mTLS
 * `buffer-size`: (integer) how many DNS messages will be buffered before being sent
 * `chan-buffer-size`: (integer) channel buffer size used on incoming dns message, number of messages before to drop it.
 
@@ -38,6 +41,9 @@ fluentd:
   tls-support: false
   tls-insecure: false
   tls-min-version: 1.2
+  ca-file: ""
+  cert-file: ""
+  key-file: ""
   buffer-size: 100
   chan-buffer-size: 65535
 ```

docs/loggers/logger_influxdb.md

@@ -1,5 +1,4 @@
 
-
 # Logger: InfluxDB client
 
 InfluxDB client to remote InfluxDB server
@@ -13,6 +12,9 @@ Options:
 - `tls-support`: (boolean) enable tls
 - `tls-insecure`: (boolean) insecure skip verify
 - `tls-min-version`: (string) min tls version
+- `ca-file`: (string) provide CA file to verify the server certificate
+- `cert-file`: (string) provide client certificate file for mTLS
+- `key-file`: (string) provide client private key file for mTLS
 - `chan-buffer-size`: (integer) channel buffer size used on incoming dns message, number of messages before to drop it.
 
 Default values:
@@ -26,5 +28,8 @@ influxdb:
   tls-support: false
   tls-insecure: false
   tls-min-version: 1.2
+  ca-file: ""
+  cert-file: ""
+  key-file: ""
   chan-buffer-size: 65535
 ```

docs/loggers/logger_kafka.md

@@ -12,11 +12,14 @@ Options:
 - `tls-support`: (boolean) enable tls
 - `tls-insecure`: (boolean) insecure skip verify
 - `tls-min-version`: (string) min tls version, default to 1.2
+- `ca-file`: (string) provide CA file to verify the server certificate
+- `cert-file`: (string) provide client certificate file for mTLS
+- `key-file`: (string) provide client private key file for mTLS
 - `sasl-support`: (boolean) enable SASL
 - `sasl-username`: (string) SASL username
 - `sasl-password`: (string) SASL password
-- `sasl-mechanism`: (string) SASL mechanism: PLAIN or SCRAM-SHA-512
-- `mode`: (string)  output format: text, json, or flat-json
+- `sasl-mechanism`: (string) SASL mechanism: `PLAIN` or `SCRAM-SHA-512`
+- `mode`: (string)  output format: `text`, `json`, or `flat-json`
 - `buffer-size`: (integer) how many DNS messages will be buffered before being sent
 - `topic`: (integer) kafka topic to forward messages to
 - `partition`: (integer) kafka partition
@@ -34,6 +37,9 @@ kafkaproducer:
   tls-support: false
   tls-insecure: false
   tls-min-version: 1.2
+  ca-file: ""
+  cert-file: ""
+  key-file: ""
   sasl-support: false
   sasl-mechanism: PLAIN
   sasl-username: ""

docs/loggers/logger_loki.md

@@ -6,15 +6,17 @@ Options:
 
 - `server-url`: (string) Loki server url
 - `job-name`: (string) Job name
-- `mode`: (string) output format: text, json, or flat-json
+- `mode`: (string) output format: `text`, `json`, or `flat-json`
 - `flush-interval`: (integer) flush batch every X seconds
 - `batch-size`: (integer) batch size for log entries in bytes
 - `retry-interval`: (integer) interval in second between before to retry to send batch
 - `text-format`: (string) output text format, please refer to the default text format to see all available directives, use this parameter if you want a specific format
 - `proxy-url`: (string) Proxy URL
-- `tls-support`: (boolean) enable tls
-- `tls-insecure`: (boolean) insecure skip verify
+- `tls-insecure`: (boolean) insecure  tls, skip certificate and hostname verify
 - `tls-min-version`: (string) min tls version
+- `ca-file`: (string) provide CA file to verify the server certificate
+- `cert-file`: (string) provide client certificate file for mTLS
+- `key-file`: (string) provide client private key file for mTLS
 - `basic-auth-login`: (string) basic auth login
 - `basic-auth-pwd`: (string) basic auth password
 - `basic-auth-pwd-file`: (string) path to a file containing the basic auth password
@@ -36,6 +38,9 @@ lokiclient:
   proxy-url: ""
   tls-insecure: false
   tls-min-version: 1.2
+  ca-file: ""
+  cert-file: ""
+  key-file: ""
   basic-auth-login: ""
   basic-auth-pwd: ""
   basic-auth-pwd-file: ""

loggers/fluentd.go

@@ -240,7 +240,7 @@ func (o *FluentdClient) Process() {
 	bufferDm := []dnsutils.DnsMessage{}
 
 	// init flust timer for buffer
-	flushInterval := time.Duration(o.config.Loggers.TcpClient.FlushInterval) * time.Second
+	flushInterval := time.Duration(o.config.Loggers.Fluentd.FlushInterval) * time.Second
 	flushTimer := time.NewTimer(flushInterval)
 
 	o.LogInfo("ready to process")
@@ -273,7 +273,7 @@ PROCESS_LOOP:
 			bufferDm = append(bufferDm, dm)
 
 			// buffer is full ?
-			if len(bufferDm) >= o.config.Loggers.TcpClient.BufferSize {
+			if len(bufferDm) >= o.config.Loggers.Fluentd.BufferSize {
 				o.FlushBuffer(&bufferDm)
 			}
 

loggers/influxdb.go

@@ -1,7 +1,6 @@
 package loggers
 
 import (
-	"crypto/tls"
 	"time"
 
 	"github.com/dmachard/go-dnscollector/dnsutils"
@@ -139,13 +138,18 @@ func (o *InfluxDBClient) Process() {
 	opts := influxdb2.DefaultOptions()
 	opts.SetUseGZip(true)
 	if o.config.Loggers.InfluxDB.TlsSupport {
-		tlsConfig := &tls.Config{
-			InsecureSkipVerify: false,
-			MinVersion:         tls.VersionTLS12,
+		tlsOptions := dnsutils.TlsOptions{
+			InsecureSkipVerify: o.config.Loggers.InfluxDB.TlsInsecure,
+			MinVersion:         o.config.Loggers.InfluxDB.TlsMinVersion,
+			CAFile:             o.config.Loggers.InfluxDB.CAFile,
+			CertFile:           o.config.Loggers.InfluxDB.CertFile,
+			KeyFile:            o.config.Loggers.InfluxDB.KeyFile,
 		}
 
-		tlsConfig.InsecureSkipVerify = o.config.Loggers.InfluxDB.TlsInsecure
-		tlsConfig.MinVersion = dnsutils.TLS_VERSION[o.config.Loggers.InfluxDB.TlsMinVersion]
+		tlsConfig, err := dnsutils.TlsClientConfig(tlsOptions)
+		if err != nil {
+			o.logger.Fatal("logger=influxdb - tls config failed:", err)
+		}
 
 		opts.SetTLSConfig(tlsConfig)
 	}

loggers/kafkaproducer.go

@@ -3,7 +3,6 @@ package loggers
 import (
 	"bytes"
 	"context"
-	"crypto/tls"
 	"encoding/json"
 	"strconv"
 	"strings"
@@ -62,11 +61,6 @@ func (c *KafkaProducer) GetName() string { return c.name }
 func (c *KafkaProducer) SetLoggers(loggers []dnsutils.Worker) {}
 
 func (o *KafkaProducer) ReadConfig() {
-
-	if o.config.Loggers.RedisPub.TlsSupport && !dnsutils.IsValidTLS(o.config.Loggers.RedisPub.TlsMinVersion) {
-		o.logger.Fatal("logger to kafka - invalid tls min version")
-	}
-
 	if len(o.config.Loggers.RedisPub.TextFormat) > 0 {
 		o.textFormat = strings.Fields(o.config.Loggers.RedisPub.TextFormat)
 	} else {
@@ -131,13 +125,18 @@ func (o *KafkaProducer) ConnectToKafka(ctx context.Context, readyTimer *time.Tim
 
 		// enable TLS
 		if o.config.Loggers.KafkaProducer.TlsSupport {
-			tlsConfig := &tls.Config{
-				MinVersion:         tls.VersionTLS12,
-				InsecureSkipVerify: false,
+			tlsOptions := dnsutils.TlsOptions{
+				InsecureSkipVerify: o.config.Loggers.KafkaProducer.TlsInsecure,
+				MinVersion:         o.config.Loggers.KafkaProducer.TlsMinVersion,
+				CAFile:             o.config.Loggers.KafkaProducer.CAFile,
+				CertFile:           o.config.Loggers.KafkaProducer.CertFile,
+				KeyFile:            o.config.Loggers.KafkaProducer.KeyFile,
 			}
-			tlsConfig.InsecureSkipVerify = o.config.Loggers.TcpClient.TlsInsecure
-			tlsConfig.MinVersion = dnsutils.TLS_VERSION[o.config.Loggers.TcpClient.TlsMinVersion]
 
+			tlsConfig, err := dnsutils.TlsClientConfig(tlsOptions)
+			if err != nil {
+				o.logger.Fatal("logger=kafka - tls config failed:", err)
+			}
 			dialer.TLS = tlsConfig
 		}