added redis sub

capture.go

@@ -13,7 +13,7 @@ import (
 )
 
 
-func captureToBuffer() bytes.Buffer {
+func captureToBuffer(req Capmsg)  {
 
     var (
         deviceName  string = "eth0"
@@ -26,6 +26,8 @@ func captureToBuffer() bytes.Buffer {
         packetCount int = 0
     )
 
+    fmt.Println("Capturing on interface: " + req.Interface)
+
     var f bytes.Buffer
     w := pcapgo.NewWriter(&f)
     w.WriteFileHeader(uint32(snapshotLen), layers.LinkTypeEthernet)
@@ -53,5 +55,6 @@ func captureToBuffer() bytes.Buffer {
     }
 
     fmt.Println("Returning from capture")
-    return f
+    postBufferCloudshark(*csschemePtr, *cshostPtr, *cstokenPtr, f)
+    return
 }

chanBuffer.go

@@ -0,0 +1,25 @@
+package main
+
+// BUFFER FOR CAPTURE MESSAGES
+
+type RingBufferCapmsg struct {
+    inputChannel  <-chan Capmsg
+    outputChannel chan Capmsg
+}
+
+
+func NewRingBufferCapmsg(inputChannel <-chan Capmsg, outputChannel chan Capmsg) *RingBufferCapmsg {
+    return &RingBufferCapmsg{inputChannel, outputChannel}
+}
+
+func (r *RingBufferCapmsg) Run() {
+    for v := range r.inputChannel {
+        select {
+        case r.outputChannel <- v:
+        default:
+            <-r.outputChannel
+            r.outputChannel <- v
+        }
+    }
+    close(r.outputChannel)
+}

main.go

@@ -2,22 +2,35 @@ package main
 
 import (
   "flag"
+  "fmt"
 )
 
+var cshostPtr *string
+var cstokenPtr *string
+var csschemePtr *string
+var upPtr *bool
+
 func main() {
 
-    cshostPtr := flag.String("cshost", "localhost", "cloushark host")
-    cstokenPtr := flag.String("cstoken", "xxxxxxx", "cloushark api token")
-    csschemePtr := flag.String("csscheme", "https", "cloushark scheme http|https")
-    upPtr := flag.Bool("upload", false, "Upload pcap")
-    flag.Parse()
+    cshostPtr = flag.String("cshost", "localhost", "cloushark host")
+    cstokenPtr = flag.String("cstoken", "xxxxxxx", "cloushark api token")
+    csschemePtr = flag.String("csscheme", "https", "cloushark scheme http|https")
+    redisnode := flag.String("redisnode", "127.0.0.1", "Hostname|IP of redis server.  Default localhost")
+    redisport := flag.Int("redisport", 6379, "Port of redis server. Default 6379")
+    redischannel := flag.String("redischannel", "capture", "Redis channel to subscribe to.  Default capture")
 
-    if(*upPtr)  {
-        pbuf := captureToBuffer();
-        postBufferCloudshark(*csschemePtr, *cshostPtr, *cstokenPtr, pbuf) 
-    }
+    upPtr = flag.Bool("upload", false, "Upload pcap")
+    flag.Parse()
 
-
+    // Channel for thread sync
+    done := make(chan bool)
 
+    go func()  {
+        fmt.Println("Starting Redis Thread")
+        subToRedis(*redisnode, *redisport, *redischannel)
+        done <- true
+    }()
 
+    fmt.Println("Exiting?")
+    <- done
 }

msg.json

@@ -0,0 +1,11 @@
+{
+    "node": "builder1.lab5",
+    "interface": "bond1",
+    "tags": "blah,tagme,stuff",
+    "bpf": "dst ip 10.0.0.1",
+    "customer": "paypal",
+    "snap": 1500,
+    "packets": 50,
+    "alertid": 655443,
+    "duration": 15
+}

publish.msg

@@ -0,0 +1,2 @@
+PUBLISH capture '{"duration":15,"node":"builder1.lab5","interface":"bond1","tags":"blah,tagme,stuff","bpf":"dst ip 10.0.0.1","customer":"paypal","snap":1500,"packets":50,"alertid":655443}'
+

redis.go

@@ -0,0 +1,46 @@
+package main
+
+import (
+    "fmt"
+    "strconv"
+    "encoding/json"
+    "github.com/garyburd/redigo/redis"
+)
+
+var c redis.Conn
+var gerr error
+
+func subToRedis(server string, port int, subchannel string) {
+
+    fmt.Println("Attempting connect to " + server) 
+    c, gerr = redis.Dial("tcp", server + ":" + strconv.Itoa(port))
+    if gerr != nil {
+        fmt.Printf("Error connecting: %s\n", gerr)
+    }
+    fmt.Println("Connected to " + server) 
+    psc := redis.PubSubConn{c}
+    psc.Subscribe(subchannel)
+
+    for {
+        var msg Capmsg
+        switch v := psc.Receive().(type) {
+        case redis.Message:
+            fmt.Printf("%s: message: %s\n", v.Channel, v.Data)
+            if err := json.Unmarshal(v.Data, &msg); err != nil {
+                fmt.Println(err)
+            } else {
+                if(*upPtr)  {
+                    go func() {
+                        captureToBuffer(msg);
+                    }()
+                }
+            }
+        case redis.Subscription:
+            fmt.Printf("%s: %s %d\n", v.Channel, v.Kind, v.Count)
+        case error:
+            fmt.Printf("Error: %s\n", v)
+        }
+
+    }
+
+}

structs.go

@@ -0,0 +1,19 @@
+package main
+
+import (
+    "time"
+)
+
+type Capmsg struct  {
+    Node            string          `json:"node,omitempty"`
+    Interface       string          `json:"interface,omitempty"`
+    Tags            string          `json:"tags,omitempty"`
+    Bpf             string          `json:"bpf,omitempty"`
+    Customer        string          `json:"customer,omitempty"`
+    Snap            int             `json:"snap"`
+    Packets         int             `json:"packets"`
+    Alertid         int             `json:"alertid,omitempty"`
+    Duration        time.Duration   `json:"duration,omitempty"`
+}
+
+type Capmsgs []Capmsg