Merge pull request #3 from negbie/master

SIP <-> RTCP correlation

Author: negbie

README.md

@@ -3,10 +3,10 @@ heplify is captagents little brother. While it offers a compareable performance
 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 
 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
-<img align="right" width="300" src="https://user-images.githubusercontent.com/20154956/30700149-0278a246-9ee7-11e7-8aef-8d68baef554a.png">
+
 On Debian/Ubuntu: sudo apt-get install libpcap-dev  
 On CentOS/RHEL: yum install libpcap-devel  
 On Windows: install WinPcap  

build_static.sh

@@ -0,0 +1,17 @@
+#!/bin/sh
+
+set -ex
+
+apk update
+apk add linux-headers musl-dev gcc go libpcap-dev ca-certificates  git
+
+mkdir /go
+export GOPATH=/go
+mkdir -p /go/src/github.com/negbie
+mkdir -p /mnt/out
+cp -a /mnt /go/src/github.com/negbie/heplify
+cd /go/src/github.com/negbie/heplify
+rm -f heplify*
+go get -v ./ ./
+go build --ldflags '-linkmode external -extldflags "-static -s -w"' -v ./
+cp ./heplify /mnt/out/

config/config.go

@@ -23,7 +23,7 @@ type InterfacesConfig struct {
 	WriteFile    string `config:"write_file"`
 	Snaplen      int    `config:"snaplen"`
 	BufferSizeMb int    `config:"buffer_size_mb"`
-	TopSpeed     bool   `config:"top_speed"`
+	ReadSpeed    bool   `config:"top_speed"`
 	OneAtATime   bool   `config:"one_at_a_time"`
 	Loop         int    `config:"loop"`
 }

decoder/decoder.go

@@ -1,9 +1,13 @@
 package decoder
 
 import (
+	"bytes"
 	"hash"
 	"os"
+	"strconv"
+	"time"
 
+	"github.com/allegro/bigcache"
 	"github.com/cespare/xxhash"
 	"github.com/google/gopacket"
 	"github.com/google/gopacket/layers"
@@ -24,7 +28,10 @@ type Decoder struct {
 	tcpCount     int
 	dnsCount     int
 	unknownCount int
-	lru          *lru.ARCCache
+	IPFlow       gopacket.Flow
+	UDPFlow      gopacket.Flow
+	lru          *lru.Cache
+	bigcache     *bigcache.BigCache
 	hash         hash.Hash64
 }
 
@@ -40,19 +47,48 @@ type Packet struct {
 	Dport         uint16
 	CorrelationID []byte
 	Payload       []byte
+	Type          byte
 }
 
 func NewDecoder() *Decoder {
-
 	host, err := os.Hostname()
 	if err != nil {
 		host = "sniffer"
 	}
-	l, err := lru.NewARC(8192)
+
+	la, err := lru.New(8000)
 	if err != nil {
 		logp.Err("lru %v", err)
 	}
-	h := xxhash.New()
+
+	xh := xxhash.New()
+
+	bConf := 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: true,
+		// 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,
+	}
+
+	bc, err := bigcache.NewBigCache(bConf)
+	if err != nil {
+		logp.Err("bigcache %v", err)
+	}
+
 	d := &Decoder{
 		Host:         host,
 		defragger:    ip4defrag.NewIPv4Defragmenter(),
@@ -63,8 +99,9 @@ func NewDecoder() *Decoder {
 		tcpCount:     0,
 		dnsCount:     0,
 		unknownCount: 0,
-		lru:          l,
-		hash:         h,
+		lru:          la,
+		hash:         xh,
+		bigcache:     bc,
 	}
 	go d.flushFrag()
 	go d.printStats()
@@ -101,16 +138,18 @@ func (d *Decoder) Process(data []byte, ci *gopacket.CaptureInfo) (*Packet, error
 			}
 		}
 
+		d.IPFlow = ip4.NetworkFlow()
 		d.ip4Count++
+
 		pkt.Version = ip4.Version
 		pkt.Protocol = uint8(ip4.Protocol)
 		pkt.Srcip = ip2int(ip4.SrcIP)
 		pkt.Dstip = ip2int(ip4.DstIP)
 
 		ip4New, err := d.defragger.DefragIPv4(ip4)
 		if err != nil {
-			logp.Err("Error while de-fragmenting", err)
-			return nil, err
+			logp.Warn("Error while de-fragmenting", err)
+			return nil, nil
 		} else if ip4New == nil {
 			d.fragCount++
 			return nil, nil
@@ -133,21 +172,28 @@ func (d *Decoder) Process(data []byte, ci *gopacket.CaptureInfo) (*Packet, error
 			nextDecoder := ip4New.NextLayerType()
 			nextDecoder.Decode(ip4New.Payload, pb)
 		}
-		// TODO: generate a more meaningful CorrelationID
-		if config.Cfg.Mode == "DNS" || config.Cfg.Mode == "LOG" || config.Cfg.Mode == "TLS" {
-			pkt.CorrelationID = []byte(config.Cfg.Mode)
-		}
 	}
 
 	if udpLayer := packet.Layer(layers.LayerTypeUDP); udpLayer != nil {
 		udp, ok := udpLayer.(*layers.UDP)
 		if !ok {
 			return nil, nil
 		}
+
+		d.UDPFlow = udp.TransportFlow()
 		d.udpCount++
+
 		pkt.Sport = uint16(udp.SrcPort)
 		pkt.Dport = uint16(udp.DstPort)
 		pkt.Payload = udp.Payload
+		pkt.Type = 1
+
+		if config.Cfg.Mode == "SIPRTCP" {
+			d.cacheSDPIPPort(udp.Payload)
+			if (udp.Payload[0]&0xc0)>>6 == 2 && (udp.Payload[1] == 200 || udp.Payload[1] == 201) {
+				pkt.Payload, pkt.CorrelationID, pkt.Type = d.correlateRTCP(udp.Payload)
+			}
+		}
 
 	} else if tcpLayer := packet.Layer(layers.LayerTypeTCP); tcpLayer != nil {
 		tcp, ok := tcpLayer.(*layers.TCP)
@@ -158,6 +204,11 @@ func (d *Decoder) Process(data []byte, ci *gopacket.CaptureInfo) (*Packet, error
 		pkt.Sport = uint16(tcp.SrcPort)
 		pkt.Dport = uint16(tcp.DstPort)
 		pkt.Payload = tcp.Payload
+		pkt.Type = 1
+
+		if config.Cfg.Mode == "SIPRTCP" {
+			d.cacheSDPIPPort(tcp.Payload)
+		}
 	}
 
 	if dnsLayer := packet.Layer(layers.LayerTypeDNS); dnsLayer != nil {
@@ -186,3 +237,62 @@ func (d *Decoder) Process(data []byte, ci *gopacket.CaptureInfo) (*Packet, error
 	d.unknownCount++
 	return nil, nil
 }
+
+func (d *Decoder) cacheSDPIPPort(payload []byte) {
+	var SDPIP, RTCPPort string
+	var callID []byte
+
+	if posSDPIP, posSDPPort := bytes.Index(payload, []byte("c=IN IP4 ")), bytes.Index(payload, []byte("m=audio ")); posSDPIP >= 0 && posSDPPort >= 0 {
+		restIP := payload[posSDPIP:]
+		if posRestIP := bytes.Index(restIP, []byte("\r\n")); posRestIP >= 0 {
+			SDPIP = string(restIP[len("c=IN IP4 "):bytes.Index(restIP, []byte("\r\n"))])
+		} else {
+			logp.Warn("Couldn't find end of SDP IP in '%s'", string(restIP))
+		}
+
+		restPort := payload[posSDPPort:]
+		if posRestPort := bytes.Index(restIP, []byte(" RTP")); posRestPort >= 0 {
+			SDPPort, err := strconv.Atoi(string(restPort[len("m=audio "):bytes.Index(restPort, []byte(" RTP"))]))
+			if err != nil {
+				logp.Warn("%v", err)
+			}
+			RTCPPort = strconv.Itoa(SDPPort + 1)
+		} else {
+			logp.Warn("Couldn't find end of SDP Port in '%s'", string(restPort))
+		}
+
+		if posCallID := bytes.Index(payload, []byte("Call-ID: ")); posCallID >= 0 {
+			restCallID := payload[posCallID:]
+			if posRestCallID := bytes.Index(restIP, []byte("\r\n")); posRestCallID >= 0 {
+				callID = restCallID[len("Call-ID: "):bytes.Index(restCallID, []byte("\r\n"))]
+			} else {
+				logp.Warn("Couldn't find end of Call-ID in '%s'", string(restCallID))
+			}
+		} else if posID := bytes.Index(payload, []byte("i: ")); posID >= 0 {
+			restID := payload[posID:]
+			if posRestID := bytes.Index(restIP, []byte("\r\n")); posRestID >= 0 {
+				callID = restID[len("i: "):bytes.Index(restID, []byte("\r\n"))]
+			} else {
+				logp.Warn("Couldn't find end of Call-ID in '%s'", string(restID))
+			}
+		}
+		d.bigcache.Set(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
+	}
+
+	corrID, err := d.bigcache.Get(d.IPFlow.Src().String() + d.UDPFlow.Src().String())
+	if err != nil {
+		logp.Warn("%v", err)
+		return nil, nil, 0
+	}
+
+	logp.Debug("decoder", "RTCP JSON payload: %s", string(jsonRTCP))
+	return jsonRTCP, corrID, 5
+}

main.go

@@ -28,8 +28,8 @@ func parseFlags() {
 	flag.StringVar(&ifaceConfig.Type, "t", "pcap", "Capture types are [af_packet, pcap, file]")
 	flag.StringVar(&ifaceConfig.ReadFile, "rf", "", "Read packets from file. Please use -t file")
 	flag.StringVar(&ifaceConfig.WriteFile, "wf", "", "Write packets to file")
-	flag.IntVar(&ifaceConfig.Loop, "lp", 0, "Loop count over ReadFile")
-	flag.BoolVar(&ifaceConfig.TopSpeed, "ts", false, "Topspeed uses timestamps from packets")
+	flag.IntVar(&ifaceConfig.Loop, "lp", 1, "Loop count over ReadFile")
+	flag.BoolVar(&ifaceConfig.ReadSpeed, "rs", false, "Maximum read speed. Doesn't use packet timestamps")
 	flag.IntVar(&ifaceConfig.Snaplen, "s", 32768, "Snap length")
 	flag.IntVar(&ifaceConfig.BufferSizeMb, "b", 64, "Interface buffersize (MB)")
 	flag.IntVar(&keepLogFiles, "kl", 4, "Rotate the number of log files")
@@ -38,7 +38,7 @@ func parseFlags() {
 	flag.StringVar(&fileRotator.Path, "p", "./", "Log filepath")
 	flag.StringVar(&fileRotator.Name, "n", "heplify.log", "Log filename")
 	flag.Uint64Var(&rotateEveryKB, "r", 16384, "Log filesize (KB)")
-	flag.StringVar(&config.Cfg.Mode, "m", "SIP", "Capture modes [DNS, LOG, SIP, TLS]")
+	flag.StringVar(&config.Cfg.Mode, "m", "SIP", "Capture modes [DNS, LOG, SIP, RTCP, TLS]")
 	flag.BoolVar(&config.Cfg.Dedup, "dd", true, "Deduplicate packets")
 	flag.StringVar(&config.Cfg.Filter, "fi", "", "Filter out interesting packets like SIP INVITES, Handshakes ...")
 	flag.StringVar(&config.Cfg.Discard, "di", "", "Discard uninteresting packets like SIP OPTIONS, HTTP Requests ...")