AWS Flowlog local generator

network_flowlog.go

// This script generates flowlog stats for the local traffic 

package main

import (
    "fmt"
    "github.com/google/gopacket"
    "github.com/google/gopacket/layers"
    "github.com/google/gopacket/pcap"
    "log" 
    "strings"
    "strconv"
    "os"
    "time"
    "os/signal"
    "syscall"
    "flag"
    "runtime"
)

var (
    snapshotLen int32  = 96
    promiscuous bool   = false
    err         error
    timeout     time.Duration = 5 * time.Second
    handle      *pcap.Handle
    bw_registry = map[string]int{}
    pkt_registry = map[string]int{}
    starttime    int32 = 0 
    endtime    int32 = 0 
    accountid  *int
    device     *string
    filter     *string
)

func output_stats() {
   // FLOWLOG format (from http://docs.aws.amazon.com/AmazonVPC/latest/UserGuide/flow-logs.html#flow-log-records)
   // version account-id interface-id srcaddr dstaddr srcport dstport protocol packets bytes start end action log-status

    endtime = int32(time.Now().Unix())
    for key, val := range bw_registry {
           details := strings.Split(string(key),"-")
           fmt.Printf("1 %d %s %s %s %s %s 6 %d %d %d %d ACCEPT OK\n", 
                      *accountid, *device, details[0], details[1], details[2],details[3], pkt_registry[key],val,starttime, endtime)
    }

    starttime = endtime
}

func main() {
   device = flag.String("i", "en0", "interface to sniff")
   filter = flag.String("f", "tcp", "pcap filter - no udp") 
   interval := flag.Int("d", 30, "update interval")
   accountid = flag.Int("a", 0, "accountid")
   group := flag.Bool("g", false, "group both TCP channels together")
   flag.Parse()

   // Catch CTRL-C
   c := make(chan os.Signal, 2)
   signal.Notify(c, os.Interrupt, syscall.SIGTERM)
   go func() {
       <-c
       fmt.Println("\nBye!\n ")
       output_stats()  
       os.Exit(1)
   }()

   //set a x seconds ticker
   ticker := time.NewTicker(time.Duration(*interval) * time.Second)
   starttime = int32(time.Now().Unix())

   go func() {
     for t := range ticker.C {
         fmt.Println("\nFlowlog generated at", t)
         output_stats()
     }
   }()

   go func() {
     // Open device
     handle, err = pcap.OpenLive(*device, snapshotLen, promiscuous, timeout)
     if err != nil {panic(err) }
     defer handle.Close()

     // Set filter
     err = handle.SetBPFFilter(*filter)
     if err != nil {panic(err)}

     fmt.Printf("Capturing from %s, using filter %s\n",*device, *filter)
     packetSource := gopacket.NewPacketSource(handle, handle.LinkType())
     for packet := range packetSource.Packets() {
        printPacketInfo(packet, *group)
     }
  }() 

  log.Printf("Using Go version %s \n",runtime.Version() )
  // infite empty loop!
  for { }

}

func printPacketInfo(packet gopacket.Packet, group bool) {
    ipLayer := packet.Layer(layers.LayerTypeIPv4)
        ip, _ := ipLayer.(*layers.IPv4)
    tcpLayer := packet.Layer(layers.LayerTypeTCP)
        tcp, _ := tcpLayer.(*layers.TCP)
    if group {
       service_port := 0
       client := ""
       server := ""
       if int(tcp.SrcPort) < int(tcp.DstPort) {
          service_port = int(tcp.SrcPort)
          client = ip.DstIP.String()
          server = ip.SrcIP.String()
       } else {
          service_port = int(tcp.DstPort)
          client = ip.SrcIP.String()
          server = ip.DstIP.String()
       }
       aggregate_traffic := strings.Join([]string{ client, server, strconv.Itoa(service_port)},"-")
       bw_registry[aggregate_traffic] = int(ip.Length)
       pkt_registry[aggregate_traffic]++


    } else {
       traffic := strings.Join([]string{ip.SrcIP.String(),strconv.Itoa(int(tcp.SrcPort)),ip.DstIP.String(), strconv.Itoa(int(tcp.DstPort))},"-")
       bw_registry[traffic] = int(ip.Length)
       pkt_registry[traffic]++
       
    }

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

}