Merge branch 'feature/history'

grafolean · Mar 4, 2020 · 985ec0e · 985ec0e
2 parents 17cad94 + a59ef9d
commit 985ec0e
Show file tree

Hide file tree

Showing 3 changed files with 54 additions and 31 deletions.
diff --git a/docker-compose.yml b/docker-compose.yml
@@ -8,6 +8,10 @@ services:
   # the settings appropriately.
   #
 
+  # To run a NetFlow v5 simulator, use:
+  #  $ docker run --net=host --name nflow networkstatic/nflow-generator -t 127.0.0.1 -p 2055
+  # (replace the port appropriately)
+
   netflowbot:
     # If you wish to load an explicit version, change the next line. For example:
     #   image: grafolean/grafolean-netflow-bot:v1.0.0

diff --git a/netflowcollector.py b/netflowcollector.py
@@ -63,36 +63,54 @@ def process_netflow(netflow_port, named_pipe_filename):
                             "ts": ts,
                             "client": client_ip,
                             "seq": export.header.sequence,
-                            "flows": [{
-                                "IN_BYTES": data["IN_BYTES"],
-                                "PROTOCOL": data["PROTOCOL"],
-                                "DIRECTION": data["DIRECTION"],
-                                "INPUT_SNMP": data["INPUT_SNMP"],
-                                "OUTPUT_SNMP": data["OUTPUT_SNMP"],
-                                "L4_DST_PORT": data["L4_DST_PORT"],
-                                "L4_SRC_PORT": data["L4_SRC_PORT"],
-                                "IPV4_DST_ADDR": data["IPV4_DST_ADDR"],
-                                "IPV4_SRC_ADDR": data["IPV4_SRC_ADDR"],
-                            } for data in flows_data],
+                            "flows": [[
+                                # "IN_BYTES":
+                                data["IN_BYTES"],
+                                # "PROTOCOL":
+                                data["PROTOCOL"],
+                                # "DIRECTION":
+                                data["DIRECTION"],
+                                # "L4_DST_PORT":
+                                data["L4_DST_PORT"],
+                                # "L4_SRC_PORT":
+                                data["L4_SRC_PORT"],
+                                # "INPUT_SNMP":
+                                data["INPUT_SNMP"],
+                                # "OUTPUT_SNMP":
+                                data["OUTPUT_SNMP"],
+                                # "IPV4_DST_ADDR":
+                                data["IPV4_DST_ADDR"],
+                                # "IPV4_SRC_ADDR":
+                                data["IPV4_SRC_ADDR"],
+                             ] for data in flows_data],
                         }
                     elif export.header.version == 5:
                         entry = {
                             "ts": ts,
                             "client": client_ip,
                             "seq": export.header.sequence,
-                            "flows": [{
-                                "IN_BYTES": data["IN_OCTETS"],
-                                "PROTOCOL": data["PROTO"],
-                                "DIRECTION": DIRECTION_INGRESS,
-                                "INPUT_SNMP": data["INPUT"],
-                                "OUTPUT_SNMP": data["OUTPUT"],
-                                "L4_DST_PORT": data["DST_PORT"],
-                                "L4_SRC_PORT": data["SRC_PORT"],
+                            "flows": [[
+                                # "IN_BYTES":
+                                data["IN_OCTETS"],
+                                # "PROTOCOL":
+                                data["PROTO"],
+                                # "DIRECTION":
+                                DIRECTION_INGRESS,
+                                # "L4_DST_PORT":
+                                data["DST_PORT"],
+                                # "L4_SRC_PORT":
+                                data["SRC_PORT"],
+                                # "INPUT_SNMP":
+                                data["INPUT"],
+                                # "OUTPUT_SNMP":
+                                data["OUTPUT"],
                                 # netflow v5 IP addresses are decoded to integers, which is less suitable for us - pack
                                 # them back to bytes and transform them to strings:
-                                "IPV4_DST_ADDR": socket.inet_ntoa(struct.pack('!I', data["IPV4_DST_ADDR"])),
-                                "IPV4_SRC_ADDR": socket.inet_ntoa(struct.pack('!I', data["IPV4_SRC_ADDR"])),
-                            } for data in flows_data],
+                                # "IPV4_DST_ADDR":
+                                socket.inet_ntoa(struct.pack('!I', data["IPV4_DST_ADDR"])),
+                                # "IPV4_SRC_ADDR":
+                                socket.inet_ntoa(struct.pack('!I', data["IPV4_SRC_ADDR"])),
+                            ] for data in flows_data],
                         }
                     else:
                         log.error(f"Only Netflow v5 and v9 currently supported, ignoring record (version: [{export.header.version}])")

diff --git a/netflowwriter.py b/netflowwriter.py
@@ -85,15 +85,16 @@ def _get_data(record_db_seq, flows):
             for flow in flows:
                 yield (
                     record_db_seq,
-                    flow.get('IN_BYTES'),
-                    flow.get('PROTOCOL'),
-                    flow.get('DIRECTION'),
-                    flow.get('L4_DST_PORT'),
-                    flow.get('L4_SRC_PORT'),
-                    flow.get('INPUT_SNMP'),
-                    flow.get('OUTPUT_SNMP'),
-                    flow.get('IPV4_DST_ADDR'),
-                    flow.get('IPV4_SRC_ADDR'),
+                    *flow,
+                    # flow.get('IN_BYTES'),
+                    # flow.get('PROTOCOL'),
+                    # flow.get('DIRECTION'),
+                    # flow.get('L4_DST_PORT'),
+                    # flow.get('L4_SRC_PORT'),
+                    # flow.get('INPUT_SNMP'),
+                    # flow.get('OUTPUT_SNMP'),
+                    # flow.get('IPV4_DST_ADDR'),
+                    # flow.get('IPV4_SRC_ADDR'),
                 )
         data_iterator = _get_data(record_db_seq, j['flows'])
         psycopg2.extras.execute_values(