add peer name text-format directive

Makefile

@@ -5,7 +5,7 @@ GO_VERSION := $(shell go env GOVERSION | sed -n 's/go\([0-9]\+\.[0-9]\+\).*/\1/p
 GO_LOGGER := 0.4.0
 GO_POWERDNS_PROTOBUF := 1.1.0
 GO_DNSTAP_PROTOBUF := 1.0.0
-GO_FRAMESTREAM := 0.7.0
+GO_FRAMESTREAM := 0.10.0
 GO_CLIENTSYSLOG := 0.3.0
 GO_TOPMAP := 1.0.0
 

collectors/dnstap.go

@@ -107,16 +107,6 @@ func (c *Dnstap) LogError(msg string, v ...interface{}) {
 	c.logger.Error(pkgutils.PrefixLogCollector+"["+c.name+" dnstap - "+msg, v...)
 }
 
-// func (c *Dnstap) LogConnInfo(connID int, msg string, v ...interface{}) {
-// 	prefix := fmt.Sprintf(pkgutils.PrefixLogCollector+"[%s] dnstap#%d - ", c.name, connID)
-// 	c.logger.Info(prefix+msg, v...)
-// }
-
-// func (c *Dnstap) LogConnError(connID int, msg string, v ...interface{}) {
-// 	prefix := fmt.Sprintf(pkgutils.PrefixLogCollector+"[%s] dnstap#%d - ", c.name, connID)
-// 	c.logger.Error(prefix+msg, v...)
-// }
-
 func (c *Dnstap) HandleConn(conn net.Conn) {
 	// close connection on function exit
 	defer conn.Close()
@@ -129,11 +119,13 @@ func (c *Dnstap) HandleConn(conn net.Conn) {
 
 	// get peer address
 	peer := conn.RemoteAddr().String()
-	c.LogInfo("new connection #%d from %s", connID, peer)
+	peerName := netlib.GetPeerName(peer)
+	c.LogInfo("new connection #%d from %s (%s)", connID, peer, peerName)
 
 	// start dnstap processor
 	dnstapProcessor := processors.NewDNSTapProcessor(
 		connID,
+		peerName,
 		c.config,
 		c.logger,
 		c.name,

collectors/dnstap_proxifier.go

@@ -133,8 +133,10 @@ func (c *DnstapProxifier) HandleConn(conn net.Conn) {
 
 	// process incoming frame and send it to recv channel
 	err := fs.ProcessFrame(recvChan)
-	if err != nil && !c.stopping {
-		c.LogError("transport error: %s", err)
+	if err != nil {
+		if !c.stopping {
+			c.LogError("transport error: %s", err)
+		}
 	}
 
 	close(recvChan)

collectors/file_ingestor.go

@@ -383,6 +383,7 @@ func (c *FileIngestor) Run() {
 	// start dnstap subprocessor
 	c.dnstapProcessor = processors.NewDNSTapProcessor(
 		0,
+		"",
 		c.config,
 		c.logger,
 		c.name,

collectors/powerdns.go

@@ -120,10 +120,18 @@ func (c *ProtobufPowerDNS) HandleConn(conn net.Conn) {
 
 	// get peer address
 	peer := conn.RemoteAddr().String()
-	c.LogInfo("new connection #%d from %s", connID, peer)
+	peerName := netlib.GetPeerName(peer)
+	c.LogInfo("new connection #%d from %s (%s)", connID, peer, peerName)
 
 	// start protobuf subprocessor
-	pdnsProc := processors.NewPdnsProcessor(connID, c.config, c.logger, c.name, c.config.Collectors.PowerDNS.ChannelBufferSize)
+	pdnsProc := processors.NewPdnsProcessor(
+		connID,
+		peerName,
+		c.config,
+		c.logger,
+		c.name,
+		c.config.Collectors.PowerDNS.ChannelBufferSize,
+	)
 	c.Lock()
 	c.pdnsProcessors = append(c.pdnsProcessors, &pdnsProc)
 	c.Unlock()

config.yml

@@ -26,6 +26,7 @@ global:
   # - timestamp-unixns: unix timestamp with nano support
   # - localtime: local time
   # - identity: dnstap identity
+  # - peer-name: hostname or ip address of the dnstap sender
   # - version: dnstap version
   # - extra: dnstap extra as string
   # - operation: dnstap operation
@@ -248,7 +249,7 @@ multiplexer:
 #   reset-conn: true
 #   # Channel buffer size for incoming packets, number of packet before to drop it.
 #   chan-buffer-size: 65535
-#   # Reconstruct DNS payload
+#   # Reconstruct and add raw DNS payload
 #   add-dns-payload: false
 
 # # tzsp (TaZmen Sniffer Protocol)

dnsutils/message.go

@@ -147,6 +147,7 @@ type DNSTap struct {
 	PolicyMatch      string  `json:"policy-match" msgpack:"policy-match"`
 	PolicyAction     string  `json:"policy-action" msgpack:"policy-action"`
 	PolicyValue      string  `json:"policy-value" msgpack:"policy-value"`
+	PeerName         string  `json:"peer-name" msgpack:"peer-name"`
 }
 
 type PowerDNS struct {
@@ -264,6 +265,7 @@ func (dm *DNSMessage) Init() {
 		PolicyMatch:      "-",
 		PolicyAction:     "-",
 		PolicyValue:      "-",
+		PeerName:         "-",
 	}
 
 	dm.DNS = DNS{
@@ -603,6 +605,8 @@ func (dm *DNSMessage) ToTextLine(format []string, fieldDelimiter string, fieldBo
 			s.WriteString(ts.Format("2006-01-02 15:04:05.999999999"))
 		case directive == "identity":
 			s.WriteString(dm.DNSTap.Identity)
+		case directive == "peer-name":
+			s.WriteString(dm.DNSTap.PeerName)
 		case directive == "version":
 			s.WriteString(dm.DNSTap.Version)
 		case directive == "extra":

dnsutils/message_test.go

@@ -204,7 +204,8 @@ func TestDnsMessage_Json_Reference(t *testing.T) {
 				  "policy-action": "-",
 				  "policy-match": "-",
 				  "policy-value": "-",
-				  "policy-rule": "-"
+				  "policy-rule": "-",
+				  "peer-name": "-"
 				}
 			}
 			`
@@ -260,6 +261,7 @@ func TestDnsMessage_JsonFlatten_Reference(t *testing.T) {
 					"dnstap.policy-action": "-",
 					"dnstap.policy-match": "-",
 					"dnstap.policy-value": "-",
+					"dnstap.peer-name": "-",
 					"edns.dnssec-ok": 0,
 					"edns.options": [],
 					"edns.rcode": 0,
@@ -590,6 +592,11 @@ func TestDnsMessage_TextFormat_DefaultDirectives(t *testing.T) {
 				PolicyValue: "value"}},
 			expected: "rule type action match value",
 		},
+		{
+			format:   "peer-name",
+			dm:       DNSMessage{DNSTap: DNSTap{PeerName: "testpeer"}},
+			expected: "testpeer",
+		},
 	}
 
 	for _, tc := range testcases {

docs/collectors.md

@@ -8,4 +8,4 @@
 | [XDP Sniffer](collectors/collector_xdp.md)            | Live capture on network interface with XDP |
 | [AF_PACKET Sniffer](collectors/collector_afpacket.md) | Live capture on network interface with AF_PACKET socket |
 | [File Ingestor](collectors/collector_fileingestor.md) | File ingestor like pcap |
-| [DNS Message](collectors/collector_dnsmessage.md)     | DNS Message like pcap |
+| [DNS Message](collectors/collector_dnsmessage.md)     | Internal DNS data structure |

docs/collectors/collector_powerdns.md

@@ -24,7 +24,7 @@ Settings:
 - `chan-buffer-size` (int) incoming channel size, number of packet before to drop it. Default to `65535`.
   > Specifies the maximum number of packets that can be buffered before dropping additional packets.
 - `add-dns-payload` (bool) generate and add fake DNS payload. Default to `false`.
-  > PowerDNS protobuf message does not contain a DNS payload; use this setting to add a fake DNS payload.
+  > PowerDNS protobuf message does not contain a DNS payload; use this setting to add a raw DNS payload.
 
 ## Custom text format
 

docs/configuration.md

@@ -67,6 +67,7 @@ Default directives:
 - `timestamp-unixns`: unix timestamp with nano support
 - `localtime`: local time
 - `identity`: dnstap identity
+- `peer-name`: hostname or ip address of the dnstap sender
 - `version`: dnstap version
 - `extra`: dnstap extra as string
 - `operation`: dnstap operation

docs/dnsjson.md

@@ -72,6 +72,7 @@ Example:
   "dnstap": {
     "operation": "CLIENT_RESPONSE",
     "identity": "dnsdist1",
+    "peer-name": "172.16.0.2",
     "version": "-",
     "extra": "-",
     "timestamp-rfc3339ns": "2021-12-27T14:33:44.559002118Z",
@@ -113,6 +114,7 @@ Using flat-json requires more processing on the host running go-dnscollector but
   "dns.resource-records.ar": [],
   "dns.resource-records.ns": [],
   "dnstap.identity": "foo",
+  "dnstap.peer-name": "172.16.0.2",
   "dnstap.latency": "0.000000",
   "dnstap.operation": "CLIENT_RESPONSE",
   "dnstap.timestamp-rfc3339ns": "2023-03-31T10:14:46.664534902Z",

netlib/conn.go

@@ -31,3 +31,30 @@ func Close(conn io.Closer, reset bool) error {
 	}
 	return nil
 }
+
+// GetPeerName returns the hostname associated with the given peer address.
+// If the peer address cannot be split into IP and port or if the hostname lookup fails,
+// it returns the peer address or IP itself.
+func GetPeerName(peerAddr string) string {
+	// Split the peer address into IP and port
+	peerIP, _, err := net.SplitHostPort(peerAddr)
+	if err != nil {
+		// If splitting fails, return the original peer address
+		return peerAddr
+	}
+
+	// Lookup hostname associated with the IP address
+	names, err := net.LookupAddr(peerIP)
+	if err != nil {
+		// If hostname lookup fails, return the IP address
+		return peerIP
+	}
+
+	// If hostname is found, return the first name in the list
+	if len(names) > 0 {
+		return names[0]
+	}
+
+	// If no hostname is found, return the IP address
+	return peerIP
+}

processors/dnstap.go

@@ -53,6 +53,7 @@ func GetFakeDNSTap(dnsquery []byte) *dnstap.Dnstap {
 
 type DNSTapProcessor struct {
 	ConnID         int
+	PeerName       string
 	doneRun        chan bool
 	stopRun        chan bool
 	doneMonitor    chan bool
@@ -68,11 +69,20 @@ type DNSTapProcessor struct {
 	droppedCount   map[string]int
 }
 
-func NewDNSTapProcessor(connID int, config *pkgconfig.Config, logger *logger.Logger, name string, size int) DNSTapProcessor {
+func NewDNSTapProcessor(
+	connID int,
+	peerName string,
+	config *pkgconfig.Config,
+	logger *logger.Logger,
+	name string,
+	size int,
+) DNSTapProcessor {
+
 	logger.Info(pkgutils.PrefixLogProcessor+"[%s] dnstap - conn #%d - initialization...", name, connID)
 
 	d := DNSTapProcessor{
 		ConnID:         connID,
+		PeerName:       peerName,
 		doneMonitor:    make(chan bool),
 		doneRun:        make(chan bool),
 		stopMonitor:    make(chan bool),
@@ -171,6 +181,8 @@ RUN_LOOP:
 			dm := dnsutils.DNSMessage{}
 			dm.Init()
 
+			dm.DNSTap.PeerName = d.PeerName
+
 			// init dns message with additionnals parts
 			transforms.InitDNSMessageFormat(&dm)
 

processors/dnstap_test.go

@@ -22,7 +22,7 @@ func Test_DnstapProcessor(t *testing.T) {
 	logger.SetOutput(&o)
 
 	// init the dnstap consumer
-	consumer := NewDNSTapProcessor(0, pkgconfig.GetFakeConfig(), logger, "test", 512)
+	consumer := NewDNSTapProcessor(0, "peertest", pkgconfig.GetFakeConfig(), logger, "test", 512)
 
 	// prepare dns query
 	dnsmsg := new(dns.Msg)
@@ -59,7 +59,7 @@ func Test_DnstapProcessor_MalformedDnsHeader(t *testing.T) {
 	logger.SetOutput(&o)
 
 	// init the dnstap consumer
-	consumer := NewDNSTapProcessor(0, pkgconfig.GetFakeConfig(), logger, "test", 512)
+	consumer := NewDNSTapProcessor(0, "peertest", pkgconfig.GetFakeConfig(), logger, "test", 512)
 	// chanTo := make(chan dnsutils.DNSMessage, 512)
 
 	// prepare dns query
@@ -98,7 +98,7 @@ func Test_DnstapProcessor_MalformedDnsQuestion(t *testing.T) {
 	logger.SetOutput(&o)
 
 	// init the dnstap consumer
-	consumer := NewDNSTapProcessor(0, pkgconfig.GetFakeConfig(), logger, "test", 512)
+	consumer := NewDNSTapProcessor(0, "peertest", pkgconfig.GetFakeConfig(), logger, "test", 512)
 	// chanTo := make(chan dnsutils.DNSMessage, 512)
 
 	// prepare dns query
@@ -136,7 +136,7 @@ func Test_DnstapProcessor_MalformedDnsAnswer(t *testing.T) {
 	logger.SetOutput(&o)
 
 	// init the dnstap consumer
-	consumer := NewDNSTapProcessor(0, pkgconfig.GetFakeConfig(), logger, "test", 512)
+	consumer := NewDNSTapProcessor(0, "peertest", pkgconfig.GetFakeConfig(), logger, "test", 512)
 	// chanTo := make(chan dnsutils.DNSMessage, 512)
 
 	// prepare dns query
@@ -178,7 +178,7 @@ func Test_DnstapProcessor_DisableDNSParser(t *testing.T) {
 	cfg := pkgconfig.GetFakeConfig()
 	cfg.Collectors.Dnstap.DisableDNSParser = true
 
-	consumer := NewDNSTapProcessor(0, cfg, logger, "test", 512)
+	consumer := NewDNSTapProcessor(0, "peertest", cfg, logger, "test", 512)
 
 	// prepare dns query
 	dnsmsg := new(dns.Msg)
@@ -219,7 +219,7 @@ func Test_DnstapProcessor_Extended(t *testing.T) {
 	cfg := pkgconfig.GetFakeConfig()
 	cfg.Collectors.Dnstap.ExtendedSupport = true
 
-	consumer := NewDNSTapProcessor(0, cfg, logger, "test", 512)
+	consumer := NewDNSTapProcessor(0, "peertest", cfg, logger, "test", 512)
 
 	// prepare dns query
 	dnsmsg := new(dns.Msg)
@@ -285,7 +285,7 @@ func Test_DnstapProcessor_BufferLoggerIsFull(t *testing.T) {
 	lg.SetOutputChannel((logsChan))
 
 	// init the dnstap consumer
-	consumer := NewDNSTapProcessor(0, pkgconfig.GetFakeConfig(), lg, "test", 512)
+	consumer := NewDNSTapProcessor(0, "peertest", pkgconfig.GetFakeConfig(), lg, "test", 512)
 
 	// prepare dns query
 	dnsmsg := new(dns.Msg)

processors/powerdns.go

@@ -29,6 +29,7 @@ var (
 
 type PdnsProcessor struct {
 	ConnID         int
+	PeerName       string
 	doneRun        chan bool
 	stopRun        chan bool
 	doneMonitor    chan bool
@@ -44,10 +45,11 @@ type PdnsProcessor struct {
 	droppedCount   map[string]int
 }
 
-func NewPdnsProcessor(connID int, config *pkgconfig.Config, logger *logger.Logger, name string, size int) PdnsProcessor {
+func NewPdnsProcessor(connID int, peerName string, config *pkgconfig.Config, logger *logger.Logger, name string, size int) PdnsProcessor {
 	logger.Info(pkgutils.PrefixLogProcessor+"[%s] powerdns - conn #%d - initialization...", name, connID)
 	d := PdnsProcessor{
 		ConnID:         connID,
+		PeerName:       peerName,
 		doneMonitor:    make(chan bool),
 		doneRun:        make(chan bool),
 		stopMonitor:    make(chan bool),