Merge pull request #5 from negbie/master

Simple LRU cache, more checks

Author: negbie

README.md

@@ -1,15 +1,11 @@
 # heplify
 heplify is captagents little brother. While it offers a compareable performance the design goal was simplicity.
-It's a single binary which you can place on your linux or windows machine. Just run it to capture packets and 
-send them to Homer. Right now heplify is able to send SIP, DNS, LOG or TLS handshakes into homer. It's able to 
+It's a single binary which you can run to capture packets and send them to Homer. 
+Right now heplify is able to send SIP, correlated RTCP and very basic DNS, LOG or TLS handshakes into homer. It's able to 
 handle fragmented and duplicate packets out of the box.
 <img align="right" width="300" src="https://user-images.githubusercontent.com/20154956/30700149-0278a246-9ee7-11e7-8aef-8d68baef554a.png">
 ### Requirements
-* libpcap
-
-On Debian/Ubuntu: sudo apt-get install libpcap-dev  
-On CentOS/RHEL: yum install libpcap-devel  
-On Windows: install WinPcap  
+* None if you use the binary from the [releases](https://github.com/sipcapture/heplify/releases)  
 
 ### Installation
 Simply grab it from the [releases](https://github.com/sipcapture/heplify/releases)  
@@ -18,35 +14,36 @@ chmod +x heplify
 ### Usage
 ```bash
   -i    Listen on interface
-  -t    Capture types are [af_packet, pcap, file] (default "pcap")
-  -m    Capture modes [DNS, LOG, SIP, TLS] (default "SIP")
+  -t    Capture types are [pcap, af_packet] (default "pcap")
+  -m    Capture modes [DNS, LOG, SIP, SIPRTCP, TLS] (default "SIP")
+  -pr   Portrange to capture SIP (default "5060-5090")
   -hs   HEP Server address (default "127.0.0.1:9060")
-  -di   Discard uninteresting packets like SIP OPTIONS, HTTP Requests ...
-  -fi   Filter out interesting packets like SIP INVITES, Handshakes ...
-  -rf   Read packets from file. Please use -t file
-  -wf   Write packets to file
+  -di   Discard uninteresting packets
+  -fi   Filter interesting packets
+  -rf   Read packets from pcap file
+  -wf   Write packets to pcap file
   -e    Log to stderr and disable syslog/file output
   -l    Log level [debug, info, warning, error] (default "info")
 ```
 
 ### Examples
 ```bash
-# Capture SIP packets on eth2 and send them to Homer under 192.168.1.1:9060
-./heplify -i eth2 -hs "192.168.1.1:9060"
+# Capture SIP packets on eth2 and send them to 192.168.1.1:9060
+./heplify -i eth2 -hs 192.168.1.1:9060 &
 
-# Print default log level to stdout
-./heplify -i eth2 -hs "192.168.1.1:9060" -e
+# Capture SIP packets on eth2 and send them to 192.168.1.1:9060. Print debug log level to stdout
+./heplify -i eth2 -hs 192.168.1.1:9060 -e -l debug
 
-# Print debug log level to stdout
-./heplify -i eth2 -hs "192.168.1.1:9060" -e -l debug
+# Capture SIP packets with custom port range on eth2 and send them to 192.168.1.1:9060
+./heplify -i eth2 -pr 6000-6010 -hs 192.168.1.1:9060 &
 
-# Capture LOG packets on eth2 and send them to Homer under 192.168.1.1:9060
-./heplify -i eth2 -hs "192.168.1.1:9060" -m LOG
+# Use af_packet to capture SIP and correlated RTCP packets on eth2 and send them to 192.168.1.1:9060
+./heplify -i eth2 -hs 192.168.1.1:9060 -t af_packet -m SIPRTCP &
 
 # Capture SIP packets on eth2 and save them to pcap into current folder
-./heplify -i eth2 -wf capture.pcap
+./heplify -i eth2 -wf capture.pcap -t af_packet &
 
-# Read pcap file from current folder and send it's content to Homer under 192.168.1.1:9060
-./heplify -i eth2 -t file -rf capture.pcap -hs "192.168.1.1:9060"
+# Read example/rtp_rtcp_sip.pcap and send SIP and correlated RTCP packets to 192.168.1.1:9060
+./heplify -rf example/rtp_rtcp_sip.pcap -m SIPRTCP -hs 192.168.1.1:9060 &
 
 ```

decoder/decoder.go

@@ -5,13 +5,10 @@ import (
 	"hash"
 	"os"
 	"strconv"
-	"time"
 
-	"github.com/allegro/bigcache"
 	"github.com/cespare/xxhash"
 	"github.com/google/gopacket"
 	"github.com/google/gopacket/layers"
-	"github.com/hashicorp/golang-lru"
 	"github.com/negbie/heplify/config"
 	"github.com/negbie/heplify/ip4defrag"
 	"github.com/negbie/heplify/logp"
@@ -28,11 +25,12 @@ type Decoder struct {
 	tcpCount     int
 	dnsCount     int
 	unknownCount int
-	IPFlow       gopacket.Flow
-	UDPFlow      gopacket.Flow
+	FlowSrcIP    string
+	FlowSrcPort  string
 	SIPHash      hash.Hash64
-	SIPCache     *lru.Cache
-	RTCPCache    *bigcache.BigCache
+	SIPCache     *Cache
+	SDPCache     *Cache
+	RTCPCache    *Cache
 }
 
 type Packet struct {
@@ -56,38 +54,10 @@ func NewDecoder() *Decoder {
 		host = "sniffer"
 	}
 
-	sh := xxhash.New()
-
-	sc, err := lru.New(8000)
-	if err != nil {
-		logp.Err("lru %v", err)
-	}
-
-	rcConf := bigcache.Config{
-		// number of shards (must be a power of 2)
-		Shards: 1024,
-		// time after which entry can be evicted
-		LifeWindow: 180 * time.Minute,
-		// rps * lifeWindow, used only in initial memory allocation
-		MaxEntriesInWindow: 1000 * 180 * 60,
-		// max entry size in bytes, used only in initial memory allocation
-		MaxEntrySize: 300,
-		// prints information about additional memory allocation
-		Verbose: false,
-		// cache will not allocate more memory than this limit, value in MB
-		// if value is reached then the oldest entries can be overridden for the new ones
-		// 0 value means no size limit
-		HardMaxCacheSize: 512,
-		// callback fired when the oldest entry is removed because of its
-		// expiration time or no space left for the new entry. Default value is nil which
-		// means no callback and it prevents from unwrapping the oldest entry.
-		OnRemove: nil,
-	}
-
-	rc, err := bigcache.NewBigCache(rcConf)
-	if err != nil {
-		logp.Err("bigcache %v", err)
-	}
+	hSIP := xxhash.New()
+	cSIP := NewLRUCache(4000)
+	cSDP := NewLRUCache(10000)
+	cRTCP := NewLRUCache(100000)
 
 	d := &Decoder{
 		Host:         host,
@@ -99,9 +69,10 @@ func NewDecoder() *Decoder {
 		tcpCount:     0,
 		dnsCount:     0,
 		unknownCount: 0,
-		SIPHash:      sh,
-		SIPCache:     sc,
-		RTCPCache:    rc,
+		SIPHash:      hSIP,
+		SIPCache:     cSIP,
+		SDPCache:     cSDP,
+		RTCPCache:    cRTCP,
 	}
 	go d.flushFrag()
 	go d.printStats()
@@ -127,8 +98,7 @@ func (d *Decoder) Process(data []byte, ci *gopacket.CaptureInfo) (*Packet, error
 
 		if config.Cfg.Dedup {
 			d.SIPHash.Write(ip4.Payload)
-			//key := fastHash(ip4.Payload)
-			key := d.SIPHash.Sum64()
+			key := strconv.FormatUint(d.SIPHash.Sum64(), 10)
 			d.SIPHash.Reset()
 			_, dup := d.SIPCache.Get(key)
 			d.SIPCache.Add(key, nil)
@@ -138,7 +108,7 @@ func (d *Decoder) Process(data []byte, ci *gopacket.CaptureInfo) (*Packet, error
 			}
 		}
 
-		d.IPFlow = ip4.NetworkFlow()
+		d.FlowSrcIP = ip4.NetworkFlow().Src().String()
 		d.ip4Count++
 
 		pkt.Version = ip4.Version
@@ -180,7 +150,7 @@ func (d *Decoder) Process(data []byte, ci *gopacket.CaptureInfo) (*Packet, error
 			return nil, nil
 		}
 
-		d.UDPFlow = udp.TransportFlow()
+		d.FlowSrcPort = udp.TransportFlow().Src().String()
 		d.udpCount++
 
 		pkt.Sport = uint16(udp.SrcPort)
@@ -190,7 +160,7 @@ func (d *Decoder) Process(data []byte, ci *gopacket.CaptureInfo) (*Packet, error
 
 		if config.Cfg.Mode == "SIPRTCP" {
 			d.cacheSDPIPPort(udp.Payload)
-			if (udp.Payload[0]&0xc0)>>6 == 2 && (udp.Payload[1] == 200 || udp.Payload[1] == 201) {
+			if (udp.Payload[0]&0xc0)>>6 == 2 && udp.SrcPort%2 != 0 && udp.DstPort%2 != 0 && (udp.Payload[1] == 200 || udp.Payload[1] == 201) {
 				pkt.Payload, pkt.CorrelationID, pkt.Type = d.correlateRTCP(udp.Payload)
 			}
 		}
@@ -276,23 +246,28 @@ func (d *Decoder) cacheSDPIPPort(payload []byte) {
 				logp.Warn("Couldn't find end of Call-ID in '%s'", string(restID))
 			}
 		}
-		d.RTCPCache.Set(SDPIP+RTCPPort, callID)
+		d.SDPCache.Add(SDPIP+RTCPPort, callID)
 	}
 }
 
 func (d *Decoder) correlateRTCP(payload []byte) ([]byte, []byte, byte) {
-	jsonRTCP, err := protos.ParseRTCP(payload)
-	if err != nil {
-		logp.Warn("%v", err)
-		return nil, nil, 0
+	jsonRTCP, info := protos.ParseRTCP(payload)
+	if info != "" {
+		logp.Info("%v", info)
+		if jsonRTCP == nil {
+			return nil, nil, 0
+		}
 	}
 
-	corrID, err := d.RTCPCache.Get(d.IPFlow.Src().String() + d.UDPFlow.Src().String())
-	if err != nil {
-		logp.Warn("%v", err)
-		return nil, nil, 0
+	if corrID, ok := d.SDPCache.Get(d.FlowSrcIP + d.FlowSrcPort); ok {
+		logp.Debug("decoder", "SDPCache RTCP JSON payload: %s", string(jsonRTCP))
+		d.RTCPCache.Add(d.FlowSrcIP+d.FlowSrcPort, corrID)
+		return jsonRTCP, corrID, 5
+	} else if corrID, ok := d.RTCPCache.Get(d.FlowSrcIP + d.FlowSrcPort); ok {
+		logp.Debug("decoder", "RTCPCache RTCP JSON payload: %s", string(jsonRTCP))
+		return jsonRTCP, corrID, 5
 	}
 
-	logp.Debug("decoder", "RTCP JSON payload: %s", string(jsonRTCP))
-	return jsonRTCP, corrID, 5
+	logp.Info("Couldn't find RTCP correlation value for key=%v", d.FlowSrcIP+d.FlowSrcPort)
+	return nil, nil, 0
 }

decoder/util.go

@@ -1,8 +1,10 @@
 package decoder
 
 import (
+	"container/list"
 	"encoding/binary"
 	"net"
+	"sync"
 	"time"
 
 	"github.com/negbie/heplify/logp"
@@ -46,3 +48,68 @@ func (d *Decoder) printStats() {
 		}()
 	}
 }
+
+type cacheValue struct {
+	key   string
+	bytes []byte
+}
+
+// Just an estimate
+func (v *cacheValue) size() uint64 {
+	return uint64(len([]byte(v.key)) + len(v.bytes))
+}
+
+type Cache struct {
+	sync.Mutex
+	Size     uint64
+	capacity uint64
+	list     *list.List
+	table    map[string]*list.Element
+}
+
+// NewLRUCache with a maximum size of capacity bytes.
+func NewLRUCache(capacity uint64) *Cache {
+	return &Cache{
+		capacity: capacity,
+		list:     list.New(),
+		table:    make(map[string]*list.Element),
+	}
+}
+
+// Set some {key, document} into the cache. Doesn't do anything if the key is already present.
+func (c *Cache) Add(key string, document []byte) {
+	c.Lock()
+	defer c.Unlock()
+
+	_, ok := c.table[key]
+	if ok {
+		return
+	}
+	v := &cacheValue{key, document}
+	elt := c.list.PushFront(v)
+	c.table[key] = elt
+	c.Size += v.size()
+	for c.Size > c.capacity {
+		elt := c.list.Back()
+		if elt == nil {
+			return
+		}
+		v := c.list.Remove(elt).(*cacheValue)
+		delete(c.table, v.key)
+		c.Size -= v.size()
+	}
+}
+
+// Get retrieves a value from the cache and returns the value and an indicator boolean to show whether it was
+// present.
+func (c *Cache) Get(key string) (document []byte, ok bool) {
+	c.Lock()
+	defer c.Unlock()
+
+	elt, ok := c.table[key]
+	if !ok {
+		return nil, false
+	}
+	c.list.MoveToFront(elt)
+	return elt.Value.(*cacheValue).bytes, true
+}