[WIP] Update README to include pull request URL

.github/workflows/run_tests.yml

@@ -1,24 +1,28 @@
-name: Run tests
+name: Run Python unit tests
 
 on:
   push:
-    branches: [ master ]
+    branches: [ master, release ]
   pull_request:
-    branches: [ master ]
-
   workflow_dispatch:
 
 jobs:
-  test:
-    runs-on: ubuntu-latest
+  test-netflow:
+    runs-on: ubuntu-20.04
+    strategy:
+      matrix:
+        python:
+          - "3.5.3"  # Debian Stretch
+          - "3.7.3"  # Debian Buster
+          - "3.9.2"  # Debian Bullseye
+          - "3.11"   # Debian Bookworm uses 3.11.1, but it's in a newer pyenv release
     steps:
-      # Checks-out your repository under $GITHUB_WORKSPACE, so your job can access it
-      - uses: actions/checkout@v2
-
-      - name: Set up Python 3.5.3
-        uses: gabrielfalcao/pyenv-action@v7
+      - uses: actions/checkout@v3
+
+      - name: Set up Python with pyenv
+        uses: gabrielfalcao/pyenv-action@v11
         with:
-          default: '3.5.3'  # Debian Buster (stable)
-      
+          default: "${{ matrix.python }}"
+
       - name: Run Python unittests
         run: python3 -m unittest

README.md

@@ -1,9 +1,11 @@
 # Python NetFlow/IPFIX library
-This package contains libraries and tools for **NetFlow versions 1, 5 and 9, and IPFIX**.
+This package contains libraries and tools for **NetFlow versions 1, 5 and 9, and IPFIX**. It is available [on PyPI as "netflow"](https://pypi.org/project/netflow/).
 
 Version 9 is the first NetFlow version using templates. Templates make dynamically sized and configured NetFlow data flowsets possible, which makes the collector's job harder. The library provides the `netflow.parse_packet()` function as the main API point (see below). By importing `netflow.v1`, `netflow.v5` or `netflow.v9` you have direct access to the respective parsing objects, but at the beginning you probably will have more success by running the reference collector (example below) and look into its code. IPFIX (IP Flow Information Export) is based on NetFlow v9 and standardized by the IETF. All related classes are contained in `netflow.ipfix`. 
 
-Copyright 2016-2020 Dominik Pataky <software+pynetflow@dpataky.eu>
+![Data flow diagram](nf-workflow.png)
+
+Copyright 2016-2023 Dominik Pataky <software+pynetflow@dpataky.eu>
 
 Licensed under MIT License. See LICENSE.
 
@@ -34,7 +36,7 @@ assert p.header.version == 5  # NetFlow v5 packet
 assert p.flows[0].PROTO == 1  # ICMP flow
 ```
 
-In NetFlow v9 and IPFIX, templates are used instead of a fixed set of fields (like `PROTO`). See `collector.py` on how to handle these.
+In NetFlow v9 and IPFIX, templates are used instead of a fixed set of fields (like `PROTO`). See `collector.py` on how to handle these. You **must** store received templates in between exports and pass them to the parser when new packets arrive. Not storing the templates will always result in parsing failures.
 
 ## Using the collector and analyzer
 Since v0.9.0 the `netflow` library also includes reference implementations of a collector and an analyzer as CLI tools.
@@ -67,20 +69,20 @@ The test files contain tests for all use cases in the library, based on real sof
 
   1. Run tcpdump/Wireshark on your public-facing interface (with tcpdump, save the pcap to disk).
   2. Produce some sample flows, e.g. surf the web and refresh your mail client. With Wireshark, save the captured packets to disk.
-  4. Run tcpdump/Wireshark again on a local interface.
+  3. Run tcpdump/Wireshark again on a local interface.
   4. Run `softflowd` with the `-r <pcap_file>` flag. softflowd reads the captured traffic, produces the flows and exports them. Use the interface you are capturing packets on to send the exports to. E.g. capture on the localhost interface (with `-i lo` or on loopback) and then let softflowd export to `127.0.0.1:1337`.
   5. Examine the captured traffic. Use Wireshark and set the `CFLOW` "decode as" dissector on the export packets (e.g. based on the port). The `data` fields should then be shown correctly as Netflow payload.
   6. Extract this payload as hex stream. Anonymize the IP addresses with a hex editor if necessary. A recommended hex editor is [bless](https://github.com/afrantzis/bless).
 
 Second, a Docker way:
 
-  2. Run a softflowd daemon in the background inside a Docker container, listening on `eth0` and exporting to e.g. `172.17.0.1:1337`.
-  3. On your host start Wireshark to listen on the Docker bridge.
-  4. Create some traffic from inside the container.
-  5. Check the softflow daemon with `softflowctl dump-flows`.
-  6. If you have some flows shown to you, export them with `softflowctl expire-all`.
-  7. Your Wireshark should have picked up the epxort packets (it does not matter if there's a port unreachable error).
-  8. Set the decoder for the packets to `CFLOW` and copy the hex value from the NetFlow packet.
+  1. Run a softflowd daemon in the background inside a Docker container, listening on `eth0` and exporting to e.g. `172.17.0.1:1337`.
+  2. On your host start Wireshark to listen on the Docker bridge.
+  3. Create some traffic from inside the container.
+  4. Check the softflow daemon with `softflowctl dump-flows`.
+  5. If you have some flows shown to you, export them with `softflowctl expire-all`.
+  6. Your Wireshark should have picked up the export packets (it does not matter if there's a port unreachable error).
+  7. Set the decoder for the packets to `CFLOW` and copy the hex value from the NetFlow packet.
 
 Your exported hex string should begin with `0001`, `0005`, `0009` or `000a`, depending on the version.
 

netflow/analyzer.py

@@ -204,7 +204,7 @@ def total_packets(self):
     if args.match_host:
         try:
             match_host = ipaddress.ip_address(args.match_host)
-        except ValueError as ex:
+        except ValueError:
             exit("IP address '{}' is neither IPv4 nor IPv6".format(args.match_host))
 
     # Using a file and using stdin differ in their further usage for gzip.open

netflow/collector.py

@@ -19,9 +19,9 @@
 import time
 from collections import namedtuple
 
-from .ipfix import IPFIXTemplateNotRecognized
-from .utils import UnknownExportVersion, parse_packet
-from .v9 import V9TemplateNotRecognized
+from netflow.ipfix import IPFIXTemplateNotRecognized
+from netflow.utils import UnknownExportVersion, parse_packet
+from netflow.v9 import V9TemplateNotRecognized
 
 RawPacket = namedtuple('RawPacket', ['ts', 'client', 'data'])
 ParsedPacket = namedtuple('ParsedPacket', ['ts', 'client', 'export'])

netflow/ipfix.py

@@ -514,9 +514,9 @@ def get_type_unpack(cls, key: Union[int, str]) -> Optional[DataType]:
         :return:
         """
         item = None
-        if type(key) == int:
+        if type(key) is int:
             item = cls.by_id(key)
-        elif type(key) == str:
+        elif type(key) is str:
             item = cls.by_name(key)
         if not item:
             return None
@@ -575,7 +575,7 @@ def is_signed(cls, dt: Union[DataType, str]) -> bool:
         :return:
         """
         fields = ["signed8", "signed16", "signed32", "signed64"]
-        if type(dt) == DataType:
+        if type(dt) is DataType:
             return dt.type in fields
         return dt in fields
 
@@ -587,7 +587,7 @@ def is_float(cls, dt: Union[DataType, str]) -> bool:
         :return:
         """
         fields = ["float32", "float64"]
-        if type(dt) == DataType:
+        if type(dt) is DataType:
             return dt.type in fields
         return dt in fields
 
@@ -601,7 +601,7 @@ def is_bytes(cls, dt: Union[DataType, str]) -> bool:
         fields = ["octetArray", "string",
                   "macAddress", "ipv4Address", "ipv6Address",
                   "dateTimeMicroseconds", "dateTimeNanoseconds"]
-        if type(dt) == DataType:
+        if type(dt) is DataType:
             return dt.type in fields
         return dt in fields
 
@@ -635,6 +635,10 @@ class IPFIXTemplateNotRecognized(KeyError):
     pass
 
 
+class PaddingCalculationError(Exception):
+    pass
+
+
 class IPFIXHeader:
     """The header of the IPFIX export packet
     """
@@ -663,9 +667,6 @@ def __init__(self, data):
         offset += offset_add
         if len(self.fields) != self.field_count:
             raise IPFIXMalformedRecord
-
-        # TODO: if padding is needed, implement here
-
         self._length = offset
 
     def get_length(self):
@@ -697,8 +698,6 @@ def __init__(self, data):
             raise IPFIXMalformedRecord
         offset += offset_add
 
-        # TODO: if padding is needed, implement here
-
         self._length = offset
 
     def get_length(self):
@@ -739,7 +738,7 @@ def __init__(self, data, template: List[Union[TemplateField, TemplateFieldEnterp
                 raise NotImplementedError("Field type with ID {} is not implemented".format(field_type_id))
 
             datatype = field_type.type  # type: str
-            discovered_fields.append((field_type.name, field_type_id))
+            discovered_fields.append((datatype, field_type_id))
 
             # Catch fields which are meant to be raw bytes and skip the rest
             if IPFIXDataTypes.is_bytes(datatype):
@@ -766,15 +765,18 @@ def __init__(self, data, template: List[Union[TemplateField, TemplateFieldEnterp
         pack = struct.unpack(unpacker, data[0:offset])
 
         # Iterate through template again, but taking the unpacked values this time
-        for index, ((field_type_name, field_type_id), value) in enumerate(zip(discovered_fields, pack)):
+        for index, ((field_datatype, field_type_id), value) in enumerate(zip(discovered_fields, pack)):
             if type(value) is bytes:
                 # Check if value is raw bytes, so no conversion happened in struct.unpack
-                if field_type_name in ["string"]:
-                    value = str(value)
+                if field_datatype in ["string"]:
+                    try:
+                        value = value.decode()
+                    except UnicodeDecodeError:
+                        value = str(value)
                 # TODO: handle octetArray (= does not have to be unicode encoded)
-                elif field_type_name in ["boolean"]:
+                elif field_datatype in ["boolean"]:
                     value = True if value == 1 else False  # 2 = false per RFC
-                elif field_type_name in ["dateTimeMicroseconds", "dateTimeNanoseconds"]:
+                elif field_datatype in ["dateTimeMicroseconds", "dateTimeNanoseconds"]:
                     seconds = value[:4]
                     fraction = value[4:]
                     value = (int.from_bytes(seconds, "big"), int.from_bytes(fraction, "big"))
@@ -809,17 +811,29 @@ def __init__(self, data: bytes, templates):
         self.records = []
         self._templates = {}
 
-        offset = IPFIXSetHeader.size
+        offset = IPFIXSetHeader.size  # fixed size
+
         if self.header.set_id == 2:  # template set
             while offset < self.header.length:  # length of whole set
                 template_record = IPFIXTemplateRecord(data[offset:])
                 self.records.append(template_record)
                 if template_record.field_count == 0:
+                    # Should not happen, since RFC says "one or more"
                     self._templates[template_record.template_id] = None
                 else:
                     self._templates[template_record.template_id] = template_record.fields
                 offset += template_record.get_length()
 
+                # If the rest of the data is deemed to be too small for another
+                # template record, check existence of padding
+                if (
+                        offset != self.header.length
+                        and self.header.length - offset <= 16  # 16 is chosen as a guess
+                        and rest_is_padding_zeroes(data[:self.header.length], offset)
+                ):
+                    # Rest should be padding zeroes
+                    break
+
         elif self.header.set_id == 3:  # options template
             while offset < self.header.length:
                 optionstemplate_record = IPFIXOptionsTemplateRecord(data[offset:])
@@ -831,16 +845,47 @@ def __init__(self, data: bytes, templates):
                         optionstemplate_record.scope_fields + optionstemplate_record.fields
                 offset += optionstemplate_record.get_length()
 
+                # If the rest of the data is deemed to be too small for another
+                # options template record, check existence of padding
+                if (
+                        offset != self.header.length
+                        and self.header.length - offset <= 16  # 16 is chosen as a guess
+                        and rest_is_padding_zeroes(data[:self.header.length], offset)
+                ):
+                    # Rest should be padding zeroes
+                    break
+
         elif self.header.set_id >= 256:  # data set, set_id is template id
-            while offset < self.header.length:
-                template = templates.get(
-                    self.header.set_id)  # type: List[Union[TemplateField, TemplateFieldEnterprise]]
-                if not template:
-                    raise IPFIXTemplateNotRecognized
-                data_record = IPFIXDataRecord(data[offset:], template)
+            # First, get the template behind the ID. Returns a list of fields or raises an exception
+            template_fields = templates.get(
+                self.header.set_id)  # type: List[Union[TemplateField, TemplateFieldEnterprise]]
+            if not template_fields:
+                raise IPFIXTemplateNotRecognized
+
+            # All template fields have a known length. Add them all together to get the length of the data set.
+            dataset_length = functools.reduce(lambda a, x: a + x.length, template_fields, 0)
+
+            # This is the last possible offset value possible if there's no padding.
+            # If there is padding, this value marks the beginning of the padding.
+            # Two cases possible:
+            # 1. No padding: then (4 + x * dataset_length) == self.header.length
+            # 2. Padding: then (4 + x * dataset_length + p) == self.header.length,
+            #    where p is the remaining length of padding zeroes. The modulo calculates p
+            no_padding_last_offset = self.header.length - ((self.header.length - IPFIXSetHeader.size) % dataset_length)
+
+            while offset < no_padding_last_offset:
+                data_record = IPFIXDataRecord(data[offset:], template_fields)
                 self.records.append(data_record)
                 offset += data_record.get_length()
-        self._length = offset
+
+            # Safety check
+            if (
+                    offset != self.header.length
+                    and not rest_is_padding_zeroes(data[:self.header.length], offset)
+            ):
+                raise PaddingCalculationError
+
+        self._length = self.header.length
 
     def get_length(self):
         return self._length
@@ -922,7 +967,7 @@ def __init__(self, data: bytes, templates: Dict[int, list]):
             raise IPFIXMalformedPacket
 
     @property
-    def contains_new_templates(self):
+    def contains_new_templates(self) -> bool:
         return self._contains_new_templates
 
     @property
@@ -950,11 +995,11 @@ def parse_fields(data: bytes, count: int) -> (list, int):
     offset = 0
     fields = []  # type: List[Union[TemplateField, TemplateFieldEnterprise]]
     for ctr in range(count):
-        if data[offset] & 1 << 7 != 0:  # enterprise flag set
+        if (data[offset] & (1 << 7)) != 0:  # enterprise flag set. Bitwise AND checks bit only in the first byte/octet
             pack = struct.unpack("!HHI", data[offset:offset + 8])
             fields.append(
                 TemplateFieldEnterprise(
-                    id=pack[0] & ~(1 << 7),  # ID, clear enterprise flag bit
+                    id=(pack[0] & ~(1 << 15)),  # clear enterprise flag bit. Bitwise AND and INVERT work on two bytes
                     length=pack[1],  # field length
                     enterprise_number=pack[2]  # enterprise number
                 )
@@ -963,7 +1008,21 @@ def parse_fields(data: bytes, count: int) -> (list, int):
         else:
             pack = struct.unpack("!HH", data[offset:offset + 4])
             fields.append(
-                TemplateField(id=pack[0], length=pack[1])
+                TemplateField(
+                    id=pack[0],
+                    length=pack[1]
+                )
             )
             offset += 4
     return fields, offset
+
+
+def rest_is_padding_zeroes(data: bytes, offset: int) -> bool:
+    if offset <= len(data):
+        # padding zeros, so rest of bytes must be summed to 0
+        if sum(data[offset:]) != 0:
+            return False
+        return True
+
+    # If offset > len(data) there is an error
+    raise ValueError("netflow.ipfix.rest_is_padding_zeroes received a greater offset value than there is data")

netflow/utils.py

@@ -31,8 +31,8 @@ def get_export_version(data):
     return struct.unpack('!H', data[:2])[0]
 
 
-def parse_packet(data: Union[str, bytes], templates: Dict = None) -> Union[V1ExportPacket, V5ExportPacket,
-                                                                           V9ExportPacket, IPFIXExportPacket]:
+def parse_packet(data: Union[str, bytes], templates: Dict = None) \
+        -> Union[V1ExportPacket, V5ExportPacket, V9ExportPacket, IPFIXExportPacket]:
     """
     Parse an exported packet, either from string (hex) or from bytes.
 
@@ -66,10 +66,10 @@ def parse_packet(data: Union[str, bytes], templates: Dict = None) -> Union[V1Exp
     :param templates: The templates dictionary with keys 'netflow' and 'ipfix' (created if not existing).
     :return: The parsed packet, or an exception.
     """
-    if type(data) == str:
+    if type(data) is str:
         # hex dump as string
         data = bytes.fromhex(data)
-    elif type(data) == bytes:
+    elif type(data) is bytes:
         # check representation based on utf-8 decoding result
         try:
             # hex dump as bytes, but not hex
@@ -83,8 +83,8 @@ def parse_packet(data: Union[str, bytes], templates: Dict = None) -> Union[V1Exp
 
     if version in [9, 10] and templates is None:
         raise ValueError("{} packet detected, but no templates dict was passed! For correct parsing of packets with "
-                         "templates, create a 'templates' dict and pass it into the 'parse_packet' function.".format(
-            "NetFlow v9" if version == 9 else "IPFIX"))
+                         "templates, create a 'templates' dict and pass it into the 'parse_packet' function."
+                         .format("NetFlow v9" if version == 9 else "IPFIX"))
 
     if version == 1:
         return V1ExportPacket(data)