Create flows from a stream of packets

gistfile1.clj

;;WARNING: UNTESTED

(ns flows
  (:import jpcap.JpcapCaptor))

(defrecord flow-tuple :src-ip :dst-ip :src-port :dst-port :protocol)
(defrecord flow :flow-tuple :octets :first-update :last-update)

(comment
(defprotocol CaptureProtocol
  (packet-seq [capture]))

(defprotocol PacketProtocol
  (packet-to-flow [packet])
  (get-tuple [packet]))

(defprotocol FlowProtocol
  (merge [flow1 flow2]))

(defrecord Flow :flow-tuple :octets :first-update :last-update
  FlowProtocol
  (merge [flows1 flows2]))

(extend-type jpcap.packet.TCPPacket
  PacketProtocol
  (packet-to-flow [packet])
  (get-tuple [packet]))

(extend-type jpcap.packet.UDPPacket
  PacketProtocol
  (packet-to-flow [packet])
  (get-tuple [packet]))

(extend-type jpcap.packet.ICMPPacket
  PacketProtocol
  (packet-to-flow [packet])
  (get-tuple [packet]))

(extend-type jpcap.JpcapCaptor
  CaptureProtocol
  (packet-seq [capture])))

(defn get-tuple
  [packet]
  (struct flow-tuple 
    (.src_ip packet) 
    (.dst_ip packet) 
    (.src_port packet) 
    (.dst_port packet) 
    (.protocol packet)))

(defn packet-to-flow [packet] 
  (let [capture-ts [(.sec packet) (.usec packet)]]
    (struct flow (get-tuple packet) (.len packet) capture-ts capture-ts)))

(defn merge-flows [flows]
  (letfn [(merge2 [flow1 flow2]
            (struct flow
              (:tuple flow1)
              (+ (:octets flow1) (:octets flow2))
              (if (< (compare %1 %2) 0) (:first-update %1) (:first-update %2))
              (if (> (compare %1 %2) 0) (:last-update %1) (:last-update %2))))]
    (reduce merge2 flows)))

(defn packet-seq [captor]
  (let [packet (.getPacket captor)]
    (when-not (= (Packet/EOF) packet)
      (lazy-seq (cons packet (packet-seq captor))))))

(map #(reduce merge-flows %)
  (group-by :tuple 
    (map packet-to-flow (packet-seq (JpcapCaptor/openFile "capture.pcap")))))