diff --git a/probe/endpoint/dns_snooper.go b/probe/endpoint/dns_snooper.go new file mode 100644 index 000000000..3af147fd0 --- /dev/null +++ b/probe/endpoint/dns_snooper.go @@ -0,0 +1,215 @@ +package endpoint + +import ( + "bytes" + "fmt" + "math" + "time" + + log "github.com/Sirupsen/logrus" + "github.com/bluele/gcache" + "github.com/google/gopacket" + "github.com/google/gopacket/layers" + "github.com/google/gopacket/pcap" +) + +const ( + bufSize = 8 * 1024 * 1024 // 8MB + maxReverseDNSrecords = 10000 +) + +// DNSSnooper is a snopper of DNS queries +type DNSSnooper struct { + stop chan struct{} + pcapHandle *pcap.Handle + reverseDNSCache gcache.Cache +} + +// NewDNSSnooper creates a new snooper of DNS queries +func NewDNSSnooper() (*DNSSnooper, error) { + pcapHandle, err := newPcapHandle() + if err != nil { + return nil, err + } + reverseDNSCache := gcache.New(maxReverseDNSrecords).LRU().Build() + + s := &DNSSnooper{ + stop: make(chan struct{}), + pcapHandle: pcapHandle, + reverseDNSCache: reverseDNSCache, + } + go s.run() + return s, nil +} + +func newPcapHandle() (*pcap.Handle, error) { + // TODO: use specific interfaces instead? + inactive, err := pcap.NewInactiveHandle("any") + if err != nil { + return nil, err + } + defer inactive.CleanUp() + if err = inactive.SetPromisc(true); err != nil { + return nil, err + } + // TODO: reduce the size of packets being copied? maybe an overoptimization + // if err = inactive.SetSnapLen(snaplen); err != nil { + // return + // } + + // pcap timeout blackmagic copied from Weave Net to reduce CPU consumption + // see https://github.com/weaveworks/weave/commit/025315363d5ea8b8265f1b3ea800f24df2be51a4 + if err = inactive.SetTimeout(time.Duration(math.MaxInt64)); err != nil { + return nil, err + } + if err = inactive.SetImmediateMode(true); err != nil { + // If gopacket is compiled against an older pcap.h that + // doesn't have pcap_set_immediate_mode, it supplies a dummy + // definition that always returns PCAP_ERROR. That becomes + // "Generic error", which is not very helpful. The real + // pcap_set_immediate_mode never returns PCAP_ERROR, so this + // turns it into a more informative message. + if fmt.Sprint(err) == "Generic error" { + return nil, fmt.Errorf("compiled against an old version of libpcap; please compile against libpcap-1.5.0 or later") + } + + return nil, err + } + if err = inactive.SetBufferSize(bufSize); err != nil { + return nil, err + } + pcapHandle, err := inactive.Activate() + if err != nil { + return nil, err + } + if err := pcapHandle.SetDirection(pcap.DirectionIn); err != nil { + pcapHandle.Close() + return nil, err + } + if err := pcapHandle.SetBPFFilter("inbound and port 53"); err != nil { + pcapHandle.Close() + return nil, err + } + + return pcapHandle, nil +} + +// CachedNamesForIP obtains the domains associated to an IP, +// obtained while snooping A-record queries +func (s *DNSSnooper) CachedNamesForIP(ip string) []string { + result := []string{} + + domains, err := s.reverseDNSCache.Get(ip) + if err != nil { + return result + } + + for domain := range domains.(map[string]struct{}) { + result = append(result, domain) + } + + return result +} + +// Stop makes the snooper stop inspecting DNS communications +func (s *DNSSnooper) Stop() { + if s != nil { + close(s.stop) + } +} + +func (s *DNSSnooper) run() { + var ( + decodedLayers []gopacket.LayerType + dns layers.DNS + udp layers.UDP + tcp layers.TCP + ip4 layers.IPv4 + ip6 layers.IPv6 + eth layers.Ethernet + sll layers.LinuxSLL + ) + + // assumes that the "any" interface in being used (see https://wiki.wireshark.org/SLL) + packetParser := gopacket.NewDecodingLayerParser(layers.LayerTypeLinuxSLL, &sll, ð, &ip4, &ip6, &udp, &tcp, &dns) + + for { + select { + case <-s.stop: + s.pcapHandle.Close() + return + default: + } + + packet, _, err := s.pcapHandle.ZeroCopyReadPacketData() + if err != nil { + // TimeoutExpired is acceptable due to the Timeout black magic + // on the handle + if err != pcap.NextErrorTimeoutExpired { + log.Errorf("DNSSnooper: error reading packet data: %s", err) + } + continue + } + + if err := packetParser.DecodeLayers(packet, &decodedLayers); err != nil { + log.Errorf("DNSSnooper: error decoding packet: %s", err) + continue + } + + for _, layerType := range decodedLayers { + if layerType == layers.LayerTypeDNS { + s.processDNSMessage(&dns) + } + } + } +} + +func (s *DNSSnooper) processDNSMessage(dns *layers.DNS) { + + // Only consider responses to singleton, A-record questions + if !dns.QR || dns.ResponseCode != 0 || len(dns.Questions) != 1 { + return + } + question := dns.Questions[0] + if question.Type != layers.DNSTypeA || question.Class != layers.DNSClassIN { + return + } + + var ( + domainQueried = question.Name + records = append(dns.Answers, dns.Additionals...) + ips = map[string]struct{}{} + alias []byte + ) + + // Traverse records for a CNAME first since the DNS RFCs don't seem to guarantee it + // appearing before its A-records + for _, record := range records { + if record.Type == layers.DNSTypeCNAME && record.Class == layers.DNSClassIN && bytes.Equal(domainQueried, record.Name) { + alias = record.CNAME + break + } + } + + // Finally, get the answer + for _, record := range records { + if record.Type != layers.DNSTypeA || record.Class != layers.DNSClassIN { + continue + } + if bytes.Equal(domainQueried, record.Name) || (alias != nil && bytes.Equal(alias, record.Name)) { + ips[record.IP.String()] = struct{}{} + } + } + + log.Debugf("DNSSnooper: caught DNS lookup: %s -> %v", string(domainQueried), ips) + // Update cache + // TODO: Be smarter about the expiration of IPs with multiple reverse-domains + newDomain := string(domainQueried) + for ip := range ips { + if existingDomains, err := s.reverseDNSCache.Get(ip); err != nil { + existingDomains.(map[string]struct{})[newDomain] = struct{}{} + } else { + s.reverseDNSCache.Set(ip, map[string]struct{}{newDomain: {}}) + } + } +} diff --git a/probe/endpoint/reporter.go b/probe/endpoint/reporter.go index a9d61a6b9..58d59e50e 100644 --- a/probe/endpoint/reporter.go +++ b/probe/endpoint/reporter.go @@ -33,6 +33,7 @@ type Reporter struct { scanner procspy.ConnectionScanner natMapper natMapper reverseResolver *reverseResolver + dnsSnooper *DNSSnooper } // SpyDuration is an exported prometheus metric @@ -52,7 +53,7 @@ var SpyDuration = prometheus.NewSummaryVec( // on the host machine, at the granularity of host and port. That information // is stored in the Endpoint topology. It optionally enriches that topology // with process (PID) information. -func NewReporter(hostID, hostName string, spyProcs, useConntrack, walkProc bool, procRoot string, scanner procspy.ConnectionScanner) *Reporter { +func NewReporter(hostID, hostName string, spyProcs, useConntrack, walkProc bool, procRoot string, scanner procspy.ConnectionScanner, dnsSnooper *DNSSnooper) *Reporter { return &Reporter{ hostID: hostID, hostName: hostName, @@ -62,6 +63,7 @@ func NewReporter(hostID, hostName string, spyProcs, useConntrack, walkProc bool, natMapper: makeNATMapper(newConntrackFlowWalker(useConntrack, procRoot, "--any-nat")), reverseResolver: newReverseResolver(), scanner: scanner, + dnsSnooper: dnsSnooper, } } @@ -193,9 +195,11 @@ func (r *Reporter) makeEndpointNode(namespaceID string, addr string, port uint16 node := report.MakeNodeWith( report.MakeEndpointNodeID(r.hostID, namespaceID, addr, portStr), map[string]string{Addr: addr, Port: portStr}) - // In case we have a reverse resolution for the IP, we can use it for - // the name... - if names, err := r.reverseResolver.get(addr); err == nil { + names := r.dnsSnooper.CachedNamesForIP(addr) + if resolvedNames, err := r.reverseResolver.get(addr); err == nil { + append(names, resolvedNames) + } + if len(names) > 0 { node = node.WithSet(ReverseDNSNames, report.MakeStringSet(names...)) } if extra != nil { diff --git a/prog/probe.go b/prog/probe.go index 8562a5d8d..c3ba71496 100644 --- a/prog/probe.go +++ b/prog/probe.go @@ -145,7 +145,15 @@ func probeMain(flags probeFlags) { p.AddReporter(process.NewReporter(processCache, hostID, process.GetDeltaTotalJiffies)) } - endpointReporter := endpoint.NewReporter(hostID, hostName, flags.spyProcs, flags.useConntrack, flags.procEnabled, flags.procRoot, scanner) + // TODO: make the snooper optional + dnsSnooper, err := endpoint.NewDNSSnooper() + if err != nil { + log.Errorf("Fail to start DNS snooper: nodes for external services will be less accurate: %s", err) + } else { + defer dnsSnooper.Stop() + } + + endpointReporter := endpoint.NewReporter(hostID, hostName, flags.spyProcs, flags.useConntrack, flags.procEnabled, flags.procRoot, scanner, dnsSnooper) defer endpointReporter.Stop() p.AddReporter(endpointReporter)