Read pcap file generated by tcpdump and store ip packet info to InfluxDB

store_pcap.go

package main

/* file: readPcap.go
 * brief: read pcap file with gopacket package
 * date: 2017-02-20
 * creator: thinkhy
 * reference:　http://www.devdungeon.com/content/packet-capture-injection-and-analysis-gopacket
 * TODO:
 * 	2017-02-22  Extract URL from HTTP packet
 */

import (
	"fmt"
	"github.com/google/gopacket"
	"github.com/google/gopacket/layers"
	"github.com/google/gopacket/pcap"
	// "github.com/davecgh/go-spew/spew"
	"github.com/influxdata/influxdb/client/v2"
	"log"
	"strings"
	"time"
)

var (
	pcapFile string = "tcpdump.pcap"
	handle   *pcap.Handle
	err      error
)

var (
	influxUser string = ""
	influxPwd string = ""
	addr string = "http:/ip:8086"
	dbName string = "android"
	measure string = "tcpdump"
	batchNumber int = 10  // Upload 1000 points for each batch
	influxPreciesion string = "ns"

)

type Metric struct {
	Src string
	Dst string
	SrcPort uint16
	DstPort uint16
	UrlPath string
	Length  int
	Timestamp time.Time
}

func FindDevices() {
	// Find all devices
	devices, err := pcap.FindAllDevs()
	if err != nil {
		log.Fatal(err)
	}

	// Print device information
	fmt.Println("Devices found:")
	for _, device := range devices {
		fmt.Println("\nName: ", device.Name)
		fmt.Println("Description: ", device.Description)
		fmt.Println("Devices addresses: ", device.Description)
		for _, address := range device.Addresses {
			fmt.Println("- IP address: ", address.IP)
			fmt.Println("- Subnet mask: ", address.Netmask)
		}
	}
}

func getMetricsFromPacket(packet gopacket.Packet) *Metric {
	var result Metric

	// Get metadata info defined at URL: https://github.com/google/gopacket/blob/master/packet.go#L83
	meta := packet.Metadata()
	result.Timestamp = meta.Timestamp
	result.Length = meta.Length
	fmt.Printf("Time: %v Length: %d\n", meta.Timestamp, meta.Length)

	// Let's see if the packet is IP (even though the ether type told us)
	ipLayer := packet.Layer(layers.LayerTypeIPv4)
	if ipLayer != nil {
		// fmt.Println("IPv4 layer detected.")
		ip, _ := ipLayer.(*layers.IPv4)

		result.Src = ip.SrcIP.String()
		result.Dst = ip.DstIP.String()
	}

	// Let's see if the packet is TCP
	tcpLayer := packet.Layer(layers.LayerTypeTCP)
	if tcpLayer != nil {
		// fmt.Println("TCP layer detected.")
		tcp, _ := tcpLayer.(*layers.TCP)

		result.SrcPort = uint16(tcp.SrcPort)
		result.DstPort = uint16(tcp.DstPort)
	}

	// Check for errors
	if err := packet.ErrorLayer(); err != nil {
		fmt.Println("Error decoding some part of the packet:", err)
	}

	return &result
}

func printPacketInfo(packet gopacket.Packet) {
	fmt.Println(packet)

	// Let's see if the packet is an ethernet packet
	ethernetLayer := packet.Layer(layers.LayerTypeEthernet)
	if ethernetLayer != nil {
		fmt.Println("Ethernet layer detected.")
		ethernetPacket, _ := ethernetLayer.(*layers.Ethernet)
		fmt.Println("Source MAC: ", ethernetPacket.SrcMAC)
		fmt.Println("Destination MAC: ", ethernetPacket.DstMAC)
		// Ethernet type is typically IPv4 but could be ARP or other
		fmt.Println("Ethernet type: ", ethernetPacket.EthernetType)
		fmt.Println()
	}

	// Let's see if the packet is IP (even though the ether type told us)
	ipLayer := packet.Layer(layers.LayerTypeIPv4)
	if ipLayer != nil {
		fmt.Println("IPv4 layer detected.")
		ip, _ := ipLayer.(*layers.IPv4)

		// IP layer variables:
		// Version (Either 4 or 6)
		// IHL (IP Header Length in 32-bit words)
		// TOS, Length, Id, Flags, FragOffset, TTL, Protocol (TCP?),
		// Checksum, SrcIP, DstIP
		fmt.Printf("Size: %d %d\n", ip.Length, ip.IHL)
		fmt.Printf("From %s to %s\n", ip.SrcIP, ip.DstIP)
		fmt.Println("Protocol: ", ip.Protocol)
		fmt.Println()
	}

	// Let's see if the packet is TCP
	tcpLayer := packet.Layer(layers.LayerTypeTCP)
	if tcpLayer != nil {
		fmt.Println("TCP layer detected.")
		tcp, _ := tcpLayer.(*layers.TCP)

		// TCP layer variables:
		// SrcPort, DstPort, Seq, Ack, DataOffset, Window, Checksum, Urgent
		// Bool flags: FIN, SYN, RST, PSH, ACK, URG, ECE, CWR, NS
		fmt.Printf("From port %d to %d\n", tcp.SrcPort, tcp.DstPort)
		fmt.Println("Sequence number: ", tcp.Seq)
		fmt.Println()
	}

	// Iterate over all layers, printing out each layer type
	fmt.Println("All packet layers:")
	for _, layer := range packet.Layers() {
		fmt.Println("- ", layer.LayerType())
	}

	// When iterating through packet.Layers() above,
	// if it lists Payload layer then that is the same as
	// this applicationLayer. applicationLayer contains the payload
	applicationLayer := packet.ApplicationLayer()
	if applicationLayer != nil {
		fmt.Println("Application layer/Payload found.")
		fmt.Printf("%s\n", applicationLayer.Payload())

		// Search for a string inside the payload
		if strings.Contains(string(applicationLayer.Payload()), "HTTP") {
			fmt.Println("HTTP found!")
		}
	}

	// Check for errors
	if err := packet.ErrorLayer(); err != nil {
		fmt.Println("Error decoding some part of the packet:", err)
	}
}

func ReadPcapFile() []Metric {
	var result []Metric

	// Open file instead of device
	handle, err = pcap.OpenOffline(pcapFile)
	if err != nil { log.Fatal(err) }
	defer handle.Close()

	// Set filter
	var filter string = "tcp and not host 115.28.243.196"
	err = handle.SetBPFFilter(filter)
	if err != nil {
		log.Fatal(err)
	}

	// Loop through packets in file
	packetSource := gopacket.NewPacketSource(handle, handle.LinkType())

	for packet := range packetSource.Packets() {
		// fmt.Println(packet)
		m := getMetricsFromPacket(packet)
		result = append(result, *m)
	}

	return result
}

func StoreMetrics(metrics []Metric) {
	// Connect to InfluxDB
	c, err := client.NewHTTPClient(client.HTTPConfig{
	    Addr:    addr,
	    Username: influxUser,
	    Password: influxPwd,
	})
	if err != nil {
		fmt.Println("Error creating InfluxDB Client: ", err.Error())
	}
	defer c.Close()
	fmt.Println("Connect successfully")

	q := client.NewQuery("CREATE DATABASE " + dbName, "", "")
	if response, err := c.Query(q); err == nil && response.Error() == nil {
		fmt.Println("Creating DATABASE " + dbName + " successfully!")
		fmt.Println(response.Results)
	} else {
		fmt.Println(err)
	}

	l := len(metrics)
	for i := 0; i < len(metrics); i += batchNumber {
		bp, _ := client.NewBatchPoints(client.BatchPointsConfig{
			Database: dbName,
			Precision: influxPreciesion,
		})
		for j := i; j < l && j < i + batchNumber; j++ {
			m := metrics[j]
			log.Printf("Process point #%d: %v\n", j, m)

			tags := map[string]string{
				"sn":  "eea29c8",
				"src": m.Src,
				"dst": m.Dst,
				"sport": fmt.Sprintf("%d", m.SrcPort),
				"dport": fmt.Sprintf("%d", m.DstPort),
				"path":  m.UrlPath,
			}

			// TODO: monitor http response code and tcp status code 2017-02-21
			fields := map[string]interface{}{
				"len": m.Length,
			}

			pt, _:= client.NewPoint(
				measure,
				tags,
				fields,
				m.Timestamp,
			)
			bp.AddPoint(pt)
		}

		err = c.Write(bp)
		if err != nil {
			fmt.Println("Error: ", err.Error())
		}
	}
}

func printDuplicateItems(metrics []Metric) {
	count := 0
	eye := make(map[time.Time]bool)
	for _, m := range metrics {
	if saw := eye[m.Timestamp]; saw {
	count++;
	}
	}
	fmt.Printf("Duplicate points: %d\n", count)
}

func main() {
	FindDevices()
	metrics := ReadPcapFile()
	StoreMetrics(metrics)
	//fmt.Println(metrics)
}