dnstap collector: new settings to disable dnsparser (#458)

* new option to disable dnsparser
* new test unit for pcap stdout

config.yml

@@ -109,6 +109,8 @@ multiplexer:
 #   reset-conn: true
 #   # Channel buffer size for incoming packets, number of packet before to drop it.
 #   chan-buffer-size: 65535
+#   # Disable the minimalist DNS parser
+#   disable-dnsparser: true
 
 # # dnstap proxifier with no protobuf decoding.
 # dnstap-proxifier:

dnsutils/config.go

@@ -186,6 +186,7 @@ type Config struct {
 			RcvBufSize        int    `yaml:"sock-rcvbuf"`
 			ResetConn         bool   `yaml:"reset-conn"`
 			ChannelBufferSize int    `yaml:"chan-buffer-size"`
+			DisableDNSParser  bool   `yaml:"disable-dnsparser"`
 		} `yaml:"dnstap"`
 		DnstapProxifier struct {
 			Enable        bool   `yaml:"enable"`
@@ -549,6 +550,7 @@ func (c *Config) SetDefault() {
 	c.Collectors.Dnstap.RcvBufSize = 0
 	c.Collectors.Dnstap.ResetConn = true
 	c.Collectors.Dnstap.ChannelBufferSize = 65535
+	c.Collectors.Dnstap.DisableDNSParser = false
 
 	c.Collectors.DnstapProxifier.Enable = false
 	c.Collectors.DnstapProxifier.ListenIP = ANY_IP

dnsutils/message.go

@@ -17,6 +17,7 @@ import (
 	"github.com/dmachard/go-dnstap-protobuf"
 	"github.com/google/gopacket"
 	"github.com/google/gopacket/layers"
+	"github.com/miekg/dns"
 	"github.com/nqd/flat"
 	"google.golang.org/protobuf/proto"
 )
@@ -731,7 +732,7 @@ func (dm *DnsMessage) ToPacketLayer() ([]gopacket.SerializableLayer, error) {
 		ip6.SrcIP = net.ParseIP(srcIp)
 		ip6.DstIP = net.ParseIP(dstIp)
 	default:
-		return nil, errors.New("family " + dm.NetworkInfo.Family + " not yet implemented")
+		return nil, errors.New("family (" + dm.NetworkInfo.Family + ") not yet implemented")
 	}
 
 	// set transport
@@ -830,3 +831,17 @@ func GetFakeDnsMessage() DnsMessage {
 	dm.DNS.Qtype = "A"
 	return dm
 }
+
+func GetFakeDnsMessageWithPayload() DnsMessage {
+	// fake dns query payload
+	dnsmsg := new(dns.Msg)
+	dnsmsg.SetQuestion("dnscollector.dev.", dns.TypeA)
+	dnsquestion, _ := dnsmsg.Pack()
+
+	dm := GetFakeDnsMessage()
+	dm.NetworkInfo.Family = PROTO_IPV4
+	dm.NetworkInfo.Protocol = PROTO_UDP
+	dm.DNS.Payload = dnsquestion
+	dm.DNS.Length = len(dnsquestion)
+	return dm
+}

docs/collectors/collector_dnstap.md

@@ -17,6 +17,7 @@ Options:
 - `sock-rcvbuf`: (integer) sets the socket receive buffer in bytes SO_RCVBUF, set to zero to use the default system value
 - `reset-conn`: (bool) Reset TCP connection on exit
 - `chan-buffer-size`: (integer) channel buffer size used on incoming packet, number of packet before to drop it.
+- `disable-dnsparser"`: (bool) disable the minimalist DNS parser
 
 Default values:
 
@@ -32,6 +33,7 @@ dnstap:
   sock-rcvbuf: 0
   reset-conn: true
   chan-buffer-size: 65535
+  disable-dnsparser: false
 ```
 
 ## DNS tap Proxifier

loggers/stdout.go

@@ -3,6 +3,7 @@ package loggers
 import (
 	"bytes"
 	"encoding/json"
+	"io"
 	"log"
 	"os"
 	"strings"
@@ -39,7 +40,8 @@ type StdOut struct {
 	config      *dnsutils.Config
 	configChan  chan *dnsutils.Config
 	logger      *logger.Logger
-	stdout      *log.Logger
+	writerText  *log.Logger
+	writerPcap  *pcapgo.Writer
 	name        string
 }
 
@@ -55,7 +57,7 @@ func NewStdOut(config *dnsutils.Config, console *logger.Logger, name string) *St
 		logger:      console,
 		config:      config,
 		configChan:  make(chan *dnsutils.Config),
-		stdout:      log.New(os.Stdout, "", 0),
+		writerText:  log.New(os.Stdout, "", 0),
 		name:        name,
 	}
 	o.ReadConfig()
@@ -70,6 +72,7 @@ func (c *StdOut) ReadConfig() {
 	if !IsStdoutValidMode(c.config.Loggers.Stdout.Mode) {
 		c.logger.Fatal("["+c.name+"] logger=stdout - invalid mode: ", c.config.Loggers.Stdout.Mode)
 	}
+
 	if len(c.config.Loggers.Stdout.TextFormat) > 0 {
 		c.textFormat = strings.Fields(c.config.Loggers.Stdout.TextFormat)
 	} else {
@@ -90,8 +93,18 @@ func (c *StdOut) LogError(msg string, v ...interface{}) {
 	c.logger.Error("["+c.name+"] logger=stdout - "+msg, v...)
 }
 
-func (o *StdOut) SetBuffer(b *bytes.Buffer) {
-	o.stdout.SetOutput(b)
+func (o *StdOut) SetTextWriter(b *bytes.Buffer) {
+	o.writerText = log.New(os.Stdout, "", 0)
+	o.writerText.SetOutput(b)
+}
+
+func (o *StdOut) SetPcapWriter(w io.Writer) {
+	o.LogInfo("init pcap writer")
+
+	o.writerPcap = pcapgo.NewWriter(w)
+	if err := o.writerPcap.WriteFileHeader(65536, layers.LinkTypeEthernet); err != nil {
+		o.logger.Fatal("["+o.name+"] logger=stdout - pcap init error: %e", err)
+	}
 }
 
 func (o *StdOut) Channel() chan dnsutils.DnsMessage {
@@ -140,7 +153,7 @@ RUN_LOOP:
 
 		case dm, opened := <-o.inputChan:
 			if !opened {
-				o.LogInfo("input channel closed!")
+				o.LogInfo("run: input channel closed!")
 				return
 			}
 
@@ -162,13 +175,8 @@ func (o *StdOut) Process() {
 	// standard output buffer
 	buffer := new(bytes.Buffer)
 
-	// pcap init ?
-	var writerPcap *pcapgo.Writer
-	if o.config.Loggers.Stdout.Mode == dnsutils.MODE_PCAP {
-		writerPcap = pcapgo.NewWriter(os.Stdout)
-		if err := writerPcap.WriteFileHeader(65536, layers.LinkTypeEthernet); err != nil {
-			o.LogError("pcap init error: %e", err)
-		}
+	if o.config.Loggers.Stdout.Mode == dnsutils.MODE_PCAP && o.writerPcap == nil {
+		o.SetPcapWriter(os.Stdout)
 	}
 
 	o.LogInfo("ready to process")
@@ -181,15 +189,20 @@ PROCESS_LOOP:
 
 		case dm, opened := <-o.outputChan:
 			if !opened {
-				o.LogInfo("output channel closed!")
+				o.LogInfo("process: output channel closed!")
 				return
 			}
 
 			switch o.config.Loggers.Stdout.Mode {
 			case dnsutils.MODE_PCAP:
+				if len(dm.DNS.Payload) == 0 {
+					o.LogError("process: no dns payload to encode, drop it")
+					continue
+				}
+
 				pkt, err := dm.ToPacketLayer()
 				if err != nil {
-					o.LogError("failed to encode to packet layer: %s", err)
+					o.LogError("unable to pack layer: %s", err)
 					continue
 				}
 
@@ -209,25 +222,25 @@ PROCESS_LOOP:
 					Length:        bufSize,
 				}
 
-				writerPcap.WritePacket(ci, buf.Bytes())
+				o.writerPcap.WritePacket(ci, buf.Bytes())
 
 			case dnsutils.MODE_TEXT:
-				o.stdout.Print(dm.String(o.textFormat,
+				o.writerText.Print(dm.String(o.textFormat,
 					o.config.Global.TextFormatDelimiter,
 					o.config.Global.TextFormatBoundary))
 
 			case dnsutils.MODE_JSON:
 				json.NewEncoder(buffer).Encode(dm)
-				o.stdout.Print(buffer.String())
+				o.writerText.Print(buffer.String())
 				buffer.Reset()
 
 			case dnsutils.MODE_FLATJSON:
 				flat, err := dm.Flatten()
 				if err != nil {
-					o.LogError("flattening DNS message failed: %e", err)
+					o.LogError("process: flattening DNS message failed: %e", err)
 				}
 				json.NewEncoder(buffer).Encode(flat)
-				o.stdout.Print(buffer.String())
+				o.writerText.Print(buffer.String())
 				buffer.Reset()
 			}
 		}

loggers/stdout_test.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/dmachard/go-dnscollector/dnsutils"
 	"github.com/dmachard/go-logger"
+	"github.com/google/gopacket/pcapgo"
 )
 
 func Test_StdoutTextMode(t *testing.T) {
@@ -68,7 +69,7 @@ func Test_StdoutTextMode(t *testing.T) {
 			cfg.Global.TextFormatBoundary = tc.boundary
 
 			g := NewStdOut(cfg, logger.New(false), "test")
-			g.SetBuffer(&stdout)
+			g.SetTextWriter(&stdout)
 
 			go g.Run()
 
@@ -112,7 +113,7 @@ func Test_StdoutJsonMode(t *testing.T) {
 			cfg := dnsutils.GetFakeConfig()
 			cfg.Loggers.Stdout.Mode = tc.mode
 			g := NewStdOut(cfg, logger.New(false), "test")
-			g.SetBuffer(&stdout)
+			g.SetTextWriter(&stdout)
 
 			go g.Run()
 
@@ -133,3 +134,74 @@ func Test_StdoutJsonMode(t *testing.T) {
 		})
 	}
 }
+
+func Test_StdoutPcapMode(t *testing.T) {
+	// redirect stdout output to bytes buffer
+	var pcap bytes.Buffer
+
+	// init logger and run
+	cfg := dnsutils.GetFakeConfig()
+	cfg.Loggers.Stdout.Mode = "pcap"
+
+	g := NewStdOut(cfg, logger.New(false), "test")
+	g.SetPcapWriter(&pcap)
+
+	go g.Run()
+
+	// send DNSMessage to channel
+	dm := dnsutils.GetFakeDnsMessageWithPayload()
+	g.Channel() <- dm
+
+	// stop logger
+	time.Sleep(time.Second)
+	g.Stop()
+
+	// check pcap output
+	pcapReader, err := pcapgo.NewReader(bytes.NewReader(pcap.Bytes()))
+	if err != nil {
+		t.Errorf("unable to read pcap: %s", err)
+		return
+	}
+	data, _, err := pcapReader.ReadPacketData()
+	if err != nil {
+		t.Errorf("unable to read packet: %s", err)
+		return
+	}
+	if len(data) < dm.DNS.Length {
+		t.Errorf("incorrect packet size: %d", len(data))
+	}
+}
+
+func Test_StdoutPcapMode_NoDNSPayload(t *testing.T) {
+	// redirect stdout output to bytes buffer
+	logger := logger.New(false)
+	var logs bytes.Buffer
+	logger.SetOutput(&logs)
+
+	var pcap bytes.Buffer
+
+	// init logger and run
+	cfg := dnsutils.GetFakeConfig()
+	cfg.Loggers.Stdout.Mode = "pcap"
+
+	g := NewStdOut(cfg, logger, "test")
+	g.SetPcapWriter(&pcap)
+
+	go g.Run()
+
+	// send DNSMessage to channel
+	dm := dnsutils.GetFakeDnsMessage()
+	g.Channel() <- dm
+
+	// stop logger
+	time.Sleep(time.Second)
+	g.Stop()
+
+	// check output
+	regxp := "ERROR:.*process: no dns payload to encode, drop it.*"
+	pattern := regexp.MustCompile(regxp)
+	ret := logs.String()
+	if !pattern.MatchString(ret) {
+		t.Errorf("stdout error want %s, got: %s", regxp, ret)
+	}
+}

processors/dnstap.go

@@ -260,20 +260,22 @@ RUN_LOOP:
 			dm.DnsTap.Timestamp = ts.UnixNano()
 			dm.DnsTap.TimestampRFC3339 = ts.UTC().Format(time.RFC3339Nano)
 
-			// decode the dns payload to get id, rcode and the number of question
-			// number of answer, ignore invalid packet
-			dnsHeader, err := dnsutils.DecodeDns(dm.DNS.Payload)
-			if err != nil {
-				// parser error
-				dm.DNS.MalformedPacket = true
-				d.LogInfo("dns parser malformed packet: %s", err)
-			}
+			if !d.config.Collectors.Dnstap.DisableDNSParser {
+				// decode the dns payload to get id, rcode and the number of question
+				// number of answer, ignore invalid packet
+				dnsHeader, err := dnsutils.DecodeDns(dm.DNS.Payload)
+				if err != nil {
+					// parser error
+					dm.DNS.MalformedPacket = true
+					d.LogInfo("dns parser malformed packet: %s", err)
+				}
 
-			if err = dnsutils.DecodePayload(&dm, &dnsHeader, d.config); err != nil {
-				// decoding error
-				if d.config.Global.Trace.LogMalformed {
-					d.LogError("%v - %v", err, dm)
-					d.LogError("dump invalid dns payload: %v", dm.DNS.Payload)
+				if err = dnsutils.DecodePayload(&dm, &dnsHeader, d.config); err != nil {
+					// decoding error
+					if d.config.Global.Trace.LogMalformed {
+						d.LogError("%v - %v", err, dm)
+						d.LogError("dump invalid dns payload: %v", dm.DNS.Payload)
+					}
 				}
 			}
 

processors/dnstap_test.go

@@ -149,3 +149,41 @@ func Test_DnstapProcessor_MalformedDnsAnswer(t *testing.T) {
 		t.Errorf("malformed packet not detected")
 	}
 }
+
+func Test_DnstapProcessor_DisableDNSParser(t *testing.T) {
+	logger := logger.New(true)
+	var o bytes.Buffer
+	logger.SetOutput(&o)
+
+	// init the dnstap consumer
+	cfg := dnsutils.GetFakeConfig()
+	cfg.Collectors.Dnstap.DisableDNSParser = true
+
+	consumer := NewDnstapProcessor(0, cfg, logger, "test", 512)
+	chan_to := make(chan dnsutils.DnsMessage, 512)
+
+	// prepare dns query
+	dnsmsg := new(dns.Msg)
+	dnsmsg.SetQuestion("www.google.fr.", dns.TypeA)
+	dnsquestion, _ := dnsmsg.Pack()
+
+	// prepare dnstap
+	dt := &dnstap.Dnstap{}
+	dt.Type = dnstap.Dnstap_Type.Enum(1)
+
+	dt.Message = &dnstap.Message{}
+	dt.Message.Type = dnstap.Message_Type.Enum(5)
+	dt.Message.QueryMessage = dnsquestion
+
+	data, _ := proto.Marshal(dt)
+
+	go consumer.Run([]chan dnsutils.DnsMessage{chan_to}, []string{"test"})
+	// add packet to consumer
+	consumer.GetChannel() <- data
+
+	// read dns message from dnstap consumer
+	dm := <-chan_to
+	if dm.DNS.Id != 0 {
+		t.Errorf("DNS ID should be equal to zero: %d", dm.DNS.Id)
+	}
+}