Update flowtuple4 to support schema version 2 (with spoofed+masscan)

Schema version is now declared in the baseavro class, as it
is used by multiple data formats.

src/pyavro_stardust/baseavro.pxd

@@ -58,6 +58,7 @@ cdef parsedNumericArrayBlock read_numeric_array(const unsigned char[:] buf,
 
 cdef class AvroRecord:
 
+    cdef public unsigned int schemaversion
     cdef long *attributes_l
     cdef char **attributes_s
     cdef long **attributes_na
@@ -78,9 +79,11 @@ cdef class AvroRecord:
             const unsigned int maxlen, const int attrind)
     cpdef vector[long] getNumericArray(self, const int attrind)
     cpdef void resetRecord(self)
+    cpdef void setSchemaVersion(self, const unsigned int schemaversion)
 
 
 cdef class AvroReader:
+    cdef unsigned int schemaversion
     cdef unsigned int nextblock
     cdef unsigned int unzip_offset
     cdef fh

src/pyavro_stardust/baseavro.pyx

@@ -133,6 +133,7 @@ cdef class AvroRecord:
         self.stringcount = 0
         self.numcount = 0
         self.numarraycount = 0
+        self.schemaversion = 0
 
     def __init__(self, numeric, strings, numarrays):
         cdef unsigned int i
@@ -297,6 +298,9 @@ cdef class AvroRecord:
                     self.attributes_na[i] = NULL
 
 
+    cpdef void setSchemaVersion(self, const unsigned int schemaversion):
+        self.schemaversion = schemaversion
+
 cdef class AvroReader:
     def __init__(self, filepath):
         self.filepath = filepath
@@ -308,6 +312,7 @@ cdef class AvroReader:
         self.unzip_offset = 0
         self.currentrec = None
         self.schemajson = None
+        self.schemaversion = 0
 
     def _readMore(self):
         try:

src/pyavro_stardust/flowtuple4.pxd

@@ -50,6 +50,9 @@ cpdef enum Flowtuple4AttributeNum:
     ATTR_FT4_FIRST_SYN_LEN = 11
     ATTR_FT4_FIRST_TCP_RWIN = 12
     ATTR_FT4_ASN = 13
+    ATTR_FT4_SPOOFED_COUNT = 14
+    ATTR_FT4_MASSCAN_COUNT = 15
+    ATTR_FT4_LAST_NUMERIC = 16
 
 cpdef enum Flowtuple4AttributeStr:
     ATTR_FT4_MAXMIND_CONTINENT = 0

src/pyavro_stardust/flowtuple4.pyx

@@ -41,7 +41,7 @@ from cpython cimport array
 @cython.final
 cdef class AvroFlowtuple4(AvroRecord):
     def __init__(self):
-        super().__init__(ATTR_FT4_ASN + 1, ATTR_FT4_LAST_STRING,
+        super().__init__(ATTR_FT4_LAST_NUMERIC, ATTR_FT4_LAST_STRING,
                 ATTR_FT4_LAST_NUM_ARRAY)
 
     def __str__(self):
@@ -87,6 +87,10 @@ cdef class AvroFlowtuple4(AvroRecord):
             "maxmind_country": self.attributes_s[<int>ATTR_FT4_MAXMIND_COUNTRY]
         }
 
+        if self.schemaversion == 2:
+            asdict["spoofed"] = self.attributes_l[<int>ATTR_FT4_SPOOFED_COUNT]
+            asdict["masscan"] = self.attributes_l[<int>ATTR_FT4_MASSCAN_COUNT]
+
         if needarrays:
             # XXX this feels like it could be faster, but not sure how
             # to improve this
@@ -106,6 +110,7 @@ cdef class AvroFlowtuple4(AvroRecord):
                 comm_ports.push_back(cv)
             asdict['common_src_ports'] = comm_ports
 
+
             pkt_sizes = self.getNumericArray(<int>ATTR_FT4_COMMON_PKT_SIZES)
             size_freqs = self.getNumericArray(<int>ATTR_FT4_COMMON_PKT_SIZE_FREQS)
             for i in range(pkt_sizes.size()):
@@ -114,6 +119,7 @@ cdef class AvroFlowtuple4(AvroRecord):
                 comm_sizes.push_back(cv)
             asdict['common_pkt_sizes'] = comm_sizes
 
+
             flags = self.getNumericArray(<int>ATTR_FT4_COMMON_TCP_FLAGS)
             flag_freqs = self.getNumericArray(<int>ATTR_FT4_COMMON_TCP_FLAG_FREQS)
             for i in range(flags.size()):
@@ -122,6 +128,7 @@ cdef class AvroFlowtuple4(AvroRecord):
                 comm_flags.push_back(cv)
             asdict['common_tcp_flags'] = comm_flags
 
+
         return asdict
 
 @cython.final
@@ -143,6 +150,14 @@ cdef class AvroFlowtuple4Reader(AvroReader):
 
         self.currentrec.resetRecord()
 
+        if self.schemaversion == 0:
+            self.schemaversion = 1
+            for f in self.schemajson['fields']:
+                if "masscan_packet_cnt" == f['name']:
+                    self.schemaversion = 2
+                    break
+            self.currentrec.setSchemaVersion(self.schemaversion)
+
         for i in range(0, ATTR_FT4_FIRST_TCP_RWIN + 1):
             offinc = self.currentrec.parseNumeric(buf[offset:],
                     maxlen - offset, i)
@@ -164,11 +179,20 @@ cdef class AvroFlowtuple4Reader(AvroReader):
                 return 0
             offset += offinc
 
-        offinc = self.currentrec.parseNumeric(buf[offset:], maxlen - offset,
-                ATTR_FT4_ASN)
-        if offinc <= 0:
+        if self.schemaversion == 1:
+            maxattr = ATTR_FT4_ASN + 1
+        elif self.schemaversion == 2:
+            maxattr = ATTR_FT4_LAST_NUMERIC
+        else:
             return 0
-        offset += offinc
+
+        for i in range(ATTR_FT4_ASN, maxattr):
+            offinc = self.currentrec.parseNumeric(buf[offset:],
+                    maxlen - offset, i)
+            if offinc <= 0:
+                return 0
+            offset += offinc
+
         return 1
 
 

src/pyavro_stardust/rsdos.pxd

@@ -64,18 +64,15 @@ cdef class AvroRsdos(AvroRecord):
 
     cdef unsigned char *packetcontent
     cdef public unsigned int pktcontentlen
-    cdef public unsigned int schemaversion
 
     cpdef dict asDict(self)
-    cpdef void setSchemaVersion(self, const unsigned int schemaversion)
     cpdef void resetRecord(self)
     cpdef bytes getRsdosPacketString(self)
     cpdef unsigned int getRsdosPacketSize(self)
     cpdef int setRsdosPacketString(self, const unsigned char[:] buf,
             const unsigned int maxlen)
 
 cdef class AvroRsdosReader(AvroReader):
-    cdef unsigned int schemaversion
     cdef int _parseNextRecord(self, const unsigned char[:] buf,
             const unsigned int maxlen)
 

src/pyavro_stardust/rsdos.pyx

@@ -174,16 +174,12 @@ cdef class AvroRsdos(AvroRecord):
         self.sizeinbuf += astr.toskip + astr.strlen
         return 1
 
-    cpdef void setSchemaVersion(self, const unsigned int schemaversion):
-        self.schemaversion = schemaversion
-
 @cython.final
 cdef class AvroRsdosReader(AvroReader):
 
     def __init__(self, filepath):
         super().__init__(filepath)
         self.currentrec = AvroRsdos()
-        self.schemaversion = 0
 
     cdef int _parseNextRecord(self, const unsigned char[:] buf,
             const unsigned int maxlen):