Add JMirror and Cisco encap format support to UDP sinks

Also:
 * use seqtrackerid to distribute CCs derived from vendmirrorid
   based intercepts amongst seqtracker threads, rather than
   always pushing them to thread 0.

Author: salcock

src/collector/alushim_parser.c

@@ -196,7 +196,8 @@ int check_alu_intercept(colthread_local_t *loc,
         }
 
         /* Create an appropriate IPCC and export it */
-        if (push_vendor_mirrored_ipcc_job(loc->zmq_pubsocks[0], &(alu->common),
+        if (push_vendor_mirrored_ipcc_job(
+                loc->zmq_pubsocks[alu->common.seqtrackerid], &(alu->common),
                 trace_get_timeval(packet), cin, direction, l3, bodylen) == 0) {
             /* for some reason, we failed to create or send the IPCC to
              * the sequencing thread? */

src/collector/cisco_parser.c

@@ -43,6 +43,30 @@ static inline uint32_t ciscomirror_get_intercept_id(ciscomirror_hdr_t *hdr) {
     return ntohl(hdr->interceptid);
 }
 
+uint8_t *decode_cisco_from_udp_payload(uint8_t *payload, uint32_t plen,
+        uint32_t *shimintid, uint32_t *bodylen) {
+
+    ciscomirror_hdr_t *header;
+    uint32_t rem;
+    uint8_t *l3;
+
+    if (payload == NULL || plen < sizeof(ciscomirror_hdr_t)) {
+        return NULL;
+    }
+    rem = plen;
+    header = (ciscomirror_hdr_t *)payload;
+    *shimintid = ciscomirror_get_intercept_id(header);
+
+    rem -= sizeof(ciscomirror_hdr_t);
+    l3 = ((uint8_t *)header) + sizeof(ciscomirror_hdr_t);
+
+    if (rem == 0) {
+        return 0;
+    }
+    *bodylen = rem;
+    return l3;
+}
+
 /** Converts a Cisco LI-mirrored packet directly into a CC encoding job for
  *  any intercepts that have requested its vendor mirror ID.
  *
@@ -64,31 +88,23 @@ int generate_cc_from_cisco(colthread_local_t *loc,
         libtrace_packet_t *packet, packet_info_t *pinfo,
         vendmirror_intercept_list_t *ciscomirrors) {
 
-    ciscomirror_hdr_t *hdr = NULL;
     void *payload;
     uint8_t *l3;
-    uint32_t rem, cept_id;
+    uint32_t rem, cept_id, bodylen;
     vendmirror_intercept_t *cept, *tmp;
     vendmirror_intercept_list_t *vmilist;
 
     payload = get_udp_payload(packet, &rem, NULL, NULL);
-    if (rem < sizeof(ciscomirror_hdr_t) || payload == NULL) {
+    l3 = decode_cisco_from_udp_payload(payload, rem, &cept_id, &bodylen);
+    if (l3 == NULL) {
         return 0;
     }
-    hdr = (ciscomirror_hdr_t *)payload;
-
-    cept_id = ciscomirror_get_intercept_id(hdr);
 
     HASH_FIND(hh, ciscomirrors, &cept_id, sizeof(cept_id), vmilist);
     if (vmilist == NULL) {
         return 0;
     }
-    rem -= sizeof(ciscomirror_hdr_t);
-    l3 = ((uint8_t *)payload) + sizeof(ciscomirror_hdr_t);
 
-    if (rem == 0) {
-        return 0;
-    }
     HASH_ITER(hh, vmilist->intercepts, cept, tmp) {
         if (pinfo->tv.tv_sec < cept->common.tostart_time) {
             continue;
@@ -98,9 +114,10 @@ int generate_cc_from_cisco(colthread_local_t *loc,
             continue;
         }
         /* Create an appropriate IPCC and export it */
-        if (push_vendor_mirrored_ipcc_job(loc->zmq_pubsocks[0], &(cept->common),
+        if (push_vendor_mirrored_ipcc_job(
+                loc->zmq_pubsocks[cept->common.seqtrackerid], &(cept->common),
                 trace_get_timeval(packet), cept_id, ETSI_DIR_INDETERMINATE,
-                l3, rem) == 0) {
+                l3, bodylen) == 0) {
             /* for some reason, we failed to create or send the IPCC to
              * the sequencing thread? */
 

src/collector/cisco_parser.h

@@ -32,6 +32,21 @@
 #include "coreserver.h"
 #include "intercept.h"
 
+/** Given the UDP payload of a Cisco LI-mirrored packet, this method will
+ *  skip past the shim header and return a pointer to the original packet
+ *  along with the intercept ID.
+ *
+ * @param payload       The UDP payload from the mirrored packet
+ * @param plen          The length of the UDP payload
+ * @param[out] shimintid    Updated to contain the intercept ID from the
+ *                          shim header
+ * @param[out] bodylen      The amount of bytes remaining after stripping
+ *                          the shim header
+ * @return a pointer to the first byte immediately after the shim header
+ */
+uint8_t *decode_cisco_from_udp_payload(uint8_t *payload, uint32_t plen,
+        uint32_t *shimintid, uint32_t *bodylen);
+
 /** Converts a Cisco LI-mirrored packet directly into a CC encoding job for
  *  any intercepts that have requested its vendor mirror ID.
  *

src/collector/jmirror_parser.c

@@ -44,42 +44,58 @@ static inline uint32_t jmirror_get_interceptid(jmirrorhdr_t *header) {
     return (ntohl(header->interceptid) & 0x3fffffff);
 }
 
+uint8_t *decode_jmirror_from_udp_payload(uint8_t *payload, uint32_t plen,
+        uint32_t *cin, uint32_t *shimintid, uint32_t *bodylen) {
+
+    jmirrorhdr_t *header;
+    uint32_t rem = plen;
+    uint8_t *l3;
+
+    header = (jmirrorhdr_t *)payload;
+    if (!header || rem < sizeof(jmirrorhdr_t)) {
+        return NULL;
+    }
+
+    *shimintid = jmirror_get_interceptid(header);
+    *cin = ntohl(header->sessionid);
+
+    rem -= sizeof(jmirrorhdr_t);
+    l3 = ((uint8_t *)header) + sizeof(jmirrorhdr_t);
+
+    if (rem == 0) {
+        return NULL;
+    }
+    *bodylen = rem;
+    return l3;
+}
+
 int check_jmirror_intercept(colthread_local_t *loc,
         libtrace_packet_t *packet, packet_info_t *pinfo,
         coreserver_t *jmirror_sources,
         vendmirror_intercept_list_t *jmirror_ints) {
 
     coreserver_t *cs;
-    jmirrorhdr_t *header = NULL;
-    uint32_t rem = 0, cept_id, cin;
+    uint32_t rem = 0, cept_id, cin, bodylen;
     vendmirror_intercept_t *cept, *tmp;
     vendmirror_intercept_list_t *vmilist;
-    char *l3;
+    uint8_t *l3, *start;
 
     if ((cs = match_packet_to_coreserver(jmirror_sources, pinfo, 1)) == NULL) {
         return 0;
     }
 
-    header = (jmirrorhdr_t *)get_udp_payload(packet, &rem, NULL, NULL);
-    if (rem < sizeof(jmirrorhdr_t) || header == NULL) {
+    start = get_udp_payload(packet, &rem, NULL, NULL);
+    l3 = decode_jmirror_from_udp_payload(start, rem, &cin, &cept_id,
+            &bodylen);
+    if (l3 == NULL) {
         return 0;
     }
 
-    cept_id = jmirror_get_interceptid(header);
-
     HASH_FIND(hh, jmirror_ints, &cept_id, sizeof(cept_id), vmilist);
     if (vmilist == NULL) {
         return 0;
     }
 
-    rem -= sizeof(jmirrorhdr_t);
-    l3 = ((char *)header) + sizeof(jmirrorhdr_t);
-
-    if (rem == 0) {
-        return 0;
-    }
-    cin = ntohl(header->sessionid);
-
     HASH_ITER(hh, vmilist->intercepts, cept, tmp) {
         if (pinfo->tv.tv_sec < cept->common.tostart_time) {
             continue;
@@ -89,9 +105,10 @@ int check_jmirror_intercept(colthread_local_t *loc,
             continue;
         }
                 /* Create an appropriate IPCC and export it */
-        if (push_vendor_mirrored_ipcc_job(loc->zmq_pubsocks[0], &(cept->common),
+        if (push_vendor_mirrored_ipcc_job(
+                loc->zmq_pubsocks[cept->common.seqtrackerid], &(cept->common),
                 trace_get_timeval(packet), cin, ETSI_DIR_INDETERMINATE,
-                l3, rem) == 0) {
+                l3, bodylen) == 0) {
             /* for some reason, we failed to create or send the IPCC to
              * the sequencing thread? */
 

src/collector/jmirror_parser.h

@@ -31,6 +31,9 @@
 #include "coreserver.h"
 #include "intercept.h"
 
+uint8_t *decode_jmirror_from_udp_payload(uint8_t *payload, uint32_t plen,
+        uint32_t *cin, uint32_t *shimintid, uint32_t *bodylen);
+
 int check_jmirror_intercept(colthread_local_t *loc,
         libtrace_packet_t *packet, packet_info_t *pinfo,
         coreserver_t *alusources, vendmirror_intercept_list_t *jmirrors);

src/collector/udp_sink_worker.c

@@ -31,6 +31,8 @@
 #include "collector_sync.h"
 #include "ipiri.h"
 #include "alushim_parser.h"
+#include "jmirror_parser.h"
+#include "cisco_parser.h"
 
 #include <zmq.h>
 #include <unistd.h>
@@ -368,8 +370,24 @@ static int process_udp_datagram(udp_sink_local_t *local, char *key) {
             return 0;
         }
 
+    } else if (local->encapfmt == INTERCEPT_UDP_ENCAP_FORMAT_JMIRROR) {
+        uint32_t shimintid = 0;
+        skipptr = decode_jmirror_from_udp_payload(recvbuf, got, &cin,
+                &shimintid, &iplen);
+        if (skipptr == NULL) {
+            return 0;
+        }
+        dir = local->direction;
+    } else if (local->encapfmt == INTERCEPT_UDP_ENCAP_FORMAT_CISCO) {
+        uint32_t shimintid = 0;
+        skipptr = decode_cisco_from_udp_payload(recvbuf, got, &shimintid,
+                &iplen);
+        if (skipptr == NULL) {
+            return 0;
+        }
+        dir = local->direction;
+        cin = local->cin;
     } else {
-        // TODO implement other encap methods
         return 0;
     }