britram/python-ipfix

IPFIX implementation for Python 3.x

python-ipfix

IPFIX implementation for Python 3.3.

This module provides a Python interface to IPFIX message streams, and provides tools for building IPFIX Exporting and Collecting Processes. It handles message framing and deframing, encoding and decoding IPFIX data records using templates, and a bridge between IPFIX ADTs and appropriate Python data types.

It should work on Python 2.7 as well; your mileage may vary. To install on Python 2.7, the following additional dependencies must be manually installed:

• pip install functools32
• pip install py2-ipaddress
• pip install pytz

python-ipfix's Issues

NetFlow version 5 stream source

Add a class to read a stream of NetFlow 5 PDUs (using the same interface as the NetFlow 9 PDU reader)

Complete sequence number support

Compliant IPFIX EPs should check sequence numbers.

Reading from socket

I have netflow traffic pointed to my device and want to listen on a specific port for netflow, does v9pdu support reading from sockets for netflow data and then parsing it? If so how would I go about doing this?

revisit time handling to ensure UTC everywhere

Timezone dependent test failures; recheck all time handling code to ensure all timestamps are forced to utc or to naive with implied UTC.

Two of the ipfix.message tests fail when I run them against master:

$ python test.py
**********************************************************************
File "/Users/rouge8/forks/python-ipfix/ipfix/message.py", line 140, in ipfix.message
Failed example:
    for rec in msg.namedict_iterator():
       print(sorted(rec.items()))
Expected:
    [('destinationIPv4Address', IPv4Address('10.5.6.7')), ('flowStartMilliseconds', datetime.datetime(2013, 6, 21, 12, 0, tzinfo=datetime.timezone.utc)), ('packetDeltaCount', 27), ('sourceIPv4Address', IPv4Address('10.1.2.3'))]
    [('destinationIPv4Address', IPv4Address('10.12.13.14')), ('flowStartMilliseconds', datetime.datetime(2013, 6, 21, 12, 0, 2, tzinfo=datetime.timezone.utc)), ('packetDeltaCount', 33), ('sourceIPv4Address', IPv4Address('10.8.9.11'))]
    [('flowStartMilliseconds', datetime.datetime(2013, 6, 21, 12, 0, 4, tzinfo=datetime.timezone.utc)), ('myNewInformationElement', "Grüezi, Y'all")]
Got:
    [('destinationIPv4Address', IPv4Address('10.5.6.7')), ('flowStartMilliseconds', datetime.datetime(2013, 6, 21, 18, 0, tzinfo=datetime.timezone.utc)), ('packetDeltaCount', 27), ('sourceIPv4Address', IPv4Address('10.1.2.3'))]
    [('destinationIPv4Address', IPv4Address('10.12.13.14')), ('flowStartMilliseconds', datetime.datetime(2013, 6, 21, 18, 0, 2, tzinfo=datetime.timezone.utc)), ('packetDeltaCount', 33), ('sourceIPv4Address', IPv4Address('10.8.9.11'))]
    [('flowStartMilliseconds', datetime.datetime(2013, 6, 21, 18, 0, 4, tzinfo=datetime.timezone.utc)), ('myNewInformationElement', "Grüezi, Y'all")]
**********************************************************************
File "/Users/rouge8/forks/python-ipfix/ipfix/message.py", line 155, in ipfix.message
Failed example:
    for rec in msg.tuple_iterator(ielist):
        print(rec)
Expected:
    (datetime.datetime(2013, 6, 21, 12, 0, tzinfo=datetime.timezone.utc), 27)
    (datetime.datetime(2013, 6, 21, 12, 0, 2, tzinfo=datetime.timezone.utc), 33)
Got:
    (datetime.datetime(2013, 6, 21, 18, 0, tzinfo=datetime.timezone.utc), 27)
    (datetime.datetime(2013, 6, 21, 18, 0, 2, tzinfo=datetime.timezone.utc), 33)
**********************************************************************
1 items had failures:
   2 of  40 in ipfix.message
***Test Failed*** 2 failures.

Add parse/deparse to IEs

Add the ability to associate parsers and deparsers with types (as default) and with IEs (as override), and tie this into csv2ipfix and ipfix2csv.

Is it possible to create a v9 flow?

I created a simple Netflow Exporter by following the examples at the top of message.py, and it works, but it gives me a v10 flow. Is it possible to create v9 flows?

Implement RFC 6313

Separate session state handling

Cleave session state (template and sequence number information) into a separate class which can be shared among multiple MessgaeBuffers and PDUBuffers.

Improve test coverage

Doctest testing isn't full coverage (for some things: reader, writer, ieutils, this is okay; for others, it isn't). Add additional tests to the documentation or supplemental tests to increase coverage.

How are option templates created?

The example for a regular template is given as something like:

ipfix.ie.use_iana_default()

tmpl = ipfix.template.for_specs(261, "flowStartMilliseconds", 
                                     "flowEndMilliseconds", 
                                     "sourceIPv4Address", 
                                     "destinationIPv4Address",
                                     "sourceTransportPort",
                                     "destinationTransportPort",
                                     "protocolIdentifier",
                                     "octetDeltaCount", 
                                     "packetDeltaCount")

msg = ipfix.message.MessageBuffer()
msg.begin_export(8304)
msg.add_template(tmpl)
msg.export_new_set(261)
msg.export_namedict({ 'flowStartMilliseconds':    iso8601('2012-10-22 09:29:07.170000'),
                      'flowEndMilliseconds':      iso8601('2012-10-22 09:29:33.916000'),
                      'sourceIPv4Address':        ip_address('192.0.2.11'),
                      'destinationIPv4Address':   ip_address('192.0.2.212'),
                      'sourceTransportPort':      32798,
                      'destinationTransportPort': 80,
                      'protocolIdentifier':       6,
                      'packetDeltaCount':         17,
                      'octetDeltaCount':          3329})

But how would a message containing a option template be constructed?
The below creates the template and data records, but how is the template identified as an option template and the scope length set to 1?

ipfix.ie.use_iana_default()

tmpl = ipfix.template.for_specs(262, "applicationId", 
                                     "applicationName")

msg = ipfix.message.MessageBuffer()
msg.begin_export(8304)
msg.add_template(tmpl)
msg.export_new_set(262)
msg.export_namedict({ 'applicationId':    2345,
                      'applicationName':      'foobar'})

Encoding and decoding fails for template with first ie element being varlen

Hello,

PROBLEM:

If we define a template with the first ie element being a varlen, then the decoding does not seem to work.
Please see the below to reproduce the issue -

import ipfix.template
import ipfix.ie
import hexdump
from datetime import datetime
from ipaddress import ip_address

msg = ipfix.message.MessageBuffer()

msg.begin_export(8304)

ipfix.ie.use_iana_default()

tmpl = ipfix.template.from_ielist(256,
ipfix.ie.spec_list(("applicationName",
"flowStartMilliseconds",
"sourceIPv4Address",
"destinationIPv4Address",
"packetDeltaCount")))

msg.add_template(tmpl)

msg.export_ensure_set(256)

rec = { "applicationName" : "testing123",
"flowStartMilliseconds" : datetime.strptime("2013-06-21 14:00:00",
"%Y-%m-%d %H:%M:%S"),
"sourceIPv4Address" : ip_address("10.1.2.3"),
"destinationIPv4Address" : ip_address("10.5.6.7"),
"packetDeltaCount" : 27}

msg.export_namedict(rec)

print(msg)
print(hexdump.dump(msg.to_bytes()))

###################################################

READER

###################################################

msg1 = ipfix.message.MessageBuffer()
msg1.from_bytes(msg.to_bytes())

for rec in msg1.namedict_iterator():
print(sorted(rec.items()))

FIX:

The root cause of the issue is in template.py - encode_to() and decode_from() methods.

diff --git a/ipfix/template.py b/ipfix/template.py
index 1203e55..87a7f5b 100644
--- a/ipfix/template.py
+++ b/ipfix/template.py
@@ -187,7 +187,7 @@ class Template(object):
offset += packplan.st.size

     # short circuit on no varlen

•    if not self.varlenslice:
  

•    if self.varlenslice == None:
         return (vals, offset)
  
     # direct iteration over remaining IEs
  

@@ -239,7 +239,7 @@ class Template(object):
offset += packplan.st.size

     # shortcircuit no varlen

•    if not self.varlenslice:
  

•    if self.varlenslice == None:
         return offset
  
     # direct iteration over remaining IEs
  

If you are ok with this fix. Please let me know how to commit this change to master.
I have also attached the diff file.

Thanks
Shravan

diff.txt

NTP-based times are based on wrong epoch

NTP times are incorrectly being based with Jan 1, 1970 as the epoch date, instead of Jan 1, 1900. 1970 is correct for datetimeSeconds and datetimeMilliseconds, but incorrect for datetimeMicroseconds and datetimeNanoseconds.

Python 2.7 backport

coordinate with / investigate joelparker fork to backport to 2.7

SCTP support

Add support for SCTP based on https://github.com/philpraxis/pysctp

NetFlow V9 (read-only) support

Add support for reading of NetFlow V9 PDUs from streams. Needs a new MessageStreamReader-like class, since V9 framing is broken.

Enable Travis CI?

When I was developing #2, I added a .travis-ci.yml file for running the tests on Python 2.7, 3.3, and 3.4. However, it's only enabled on my fork. If you want to have this, you just need to go here, sign in with GitHub, and enable it:

https://travis-ci.org/britram/python-ipfix

If not, I'd suggest removing the config file.

compat.py not in pip install ipfix

compat.py does not exist in the pip install ipfix library version 0.9.7

IPFIX "Illegal message lengthX" Error

ilmarh's example from: #17 works great for Netflow v9 (at least with my limited testing).

When I add import ipfix.reader, and change his r = ipfix.v9pdu.from_stream(self.rfile) to say r = ipfix.reader.from_stream(self.rfile) instead, it always throws an exception for me. The traceback says:

...
for rec in r.namedict_iterator():
...
ipfix.template.IpfixDecodeError: ('Illegal message length2',)

The 2 isn't always a 2.

The flow is coming from a Mikrotik device.
Am I doing something wrong? Or is the Mikrotik sending some improperly-formatted packets? PRTG seemed to have no issues collecting IPFIX data from the Mikrotik.

My full code looks like this:

#!/usr/bin/env python3

import socketserver
#import ipfix.v9pdu
import ipfix.reader
import ipfix.ie

import argparse

ap = argparse.ArgumentParser(description="Dump IPFIX files collected over UDP")
ap.add_argument('--spec', metavar='specfile', help='iespec file to read')
args = ap.parse_args()

ipfix.ie.use_iana_default()
ipfix.ie.use_5103_default()

if args.spec:
    ipfix.ie.use_specfile(args.spec)

msgcount = 0
templates = {}
accepted_tids = set()

def dummy1(o,b):
    pass

class CollectorDictHandler(socketserver.DatagramRequestHandler):
    def handle(self):
        global msgcount
        msgcount = msgcount + 1
        reccount = 0

        print("connection from " + str(self.client_address))
        r = ipfix.reader.from_stream(self.rfile)

        try:
            r.templates = templates[str(self.client_address)]
        except KeyError as e:
            templates[str(self.client_address)] = {}
            r.templates = templates[str(self.client_address)]

        r.accepted_tids = accepted_tids
        r.unknown_data_set_hook = dummy1

        for rec in r.namedict_iterator():
            print("--- record %u in message %u from %s---" %
                 (reccount, msgcount, str(self.client_address)))
            reccount += 1
            for key in rec:
                 print("  %30s => %s" % (key, str(rec[key])))

        print("reccount = "+str(reccount))

ss = socketserver.UDPServer(("", 1234), CollectorDictHandler)
ss.serve_forever()

Sequence number & export time order switched

When I send IPFIX constructed with MessageBuffer.to_bytes, I see the following in wireshark:

The value recorded as the seqno is actually the export timestamp, and vice versa. I found the relevant code in message.py, in MessageBuffer.to_bytes:

        # Update message header in buffer
        _msghdr_st.pack_into(self.mbuf, 0, 10, self.length,
                             self.sequence, self.export_epoch, self.odid)

As expected, sequence precedes export_epoch in the pack_into call. This is also the case with read_message:

(version, self.length, self.sequence, self.export_epoch, self.odid) = \
                _msghdr_st.unpack_from(self.mbuf, 0)

Wireshark's expectations match RFC 7011.

I'm fixing this (and adding a test) in my 2.7 branch (private for the moment). I'd be happy to cherry-pick those changes into a pull request when they're done -- or I can hand that off for now; either way is fine.

UDP support in command-line tool(s)

Add UDP support to ipfix2csv and future tools. Ensure that UDP template management works. Will require extensions to serversocket for binary access to udp packets.

Do something reasonable for multiple IEs in a template

Fix timestamp type implementation

Python handling of timestamps and timezones is (at least as far as I can tell) brutally internally consistent in precisely the wrong way; timestamps will be corrected multiple times, causing errors, in the present master revision of the library.

Fix this.

old collector_test.py?

Couldn't make collector_test.py work until line 27 change:

•    for rec in r.namedict_iterator():
  

•    for rec in r.records_as_dict():
  

I suppose it's just an old example, because sources have new usage description.

btw, just want to ask if it is possible to parse netflow with this library. i thought, that it should, but saw version != 10 condition, and now confused. thank you

Add UDP support to ipfix.writer.MessageStreamWriter

UDP is still the most common protocol understood by collectors.

Template caching

As I want to use this library for IPFIX message parsing , where I am pushing data every second. So to reduce overhead I am sending Template only once or at the beginning of communication begin.

It decode well the first message as it has template as well data set. But then after it not able to decode the message.
Currently after reading message from UDP socket storing into a local Queue, not passing stream directly to any library decoding method.

Code snippet, how I am using it:
ipfix.message.MessageBuffer()
msg.from_bytes(byteBuffer)
for rec in msg.namedict_iterator()
XXXXXXXXXX

But it is not caching the templates it has processed previously.

UDP IPFIX Collector not Working

collector_test.py seems to be old, and uses TCP. I needed to do almost exactly the same thing it was trying to demo, but using UDP.

This is what I got:

#!/usr/bin/env python3

import socketserver
import ipfix.reader
import ipfix.ie

ipfix.ie.use_iana_default()
ipfix.ie.use_5103_default()

class CollectorDictHandler(socketserver.DatagramRequestHandler):
    def handle(self):
        print("=" * 80) 
        r = ipfix.reader.from_stream(self.rfile)

        for rec in r.namedict_iterator():
            print('-' * 80) 
            for key in rec:
                print("%s: %s" % (key, rec[key]))

ss = socketserver.UDPServer(("0.0.0.0", 1234), CollectorDictHandler)
ss.serve_forever()

The device exporting the flows is a Mikrotik. The code above runs, but r.namedict_iterator() never returns any records, so all I get is '====...' across my screen every time it receives a packet, but no actual data.

From what I understand, ipfix.reader is smart enough to read templates from the stream, and use them when trying to decode incoming data. Is this correct?

Am I doing something else wrong?

Better support for fixed length strings in SVG message drawing code.

If the basic example of a string is changed to a fixed length of 13 (which exactly matches the string length passed to the data record) it works just fine:

import ipfix
import ipfix.vis
from IPython.display import SVG

def draw_message(msg, length=256):
    return SVG(ipfix.vis.MessageBufferRenderer(msg, raster=4).render(length=length))

def draw_template(tmpl):
    ofd = ipfix.vis.OctetFieldDrawing(raster=4)
    ipfix.vis.draw_template(ofd, tmpl)
    return SVG(ofd.render((90,30)))

ipfix.ie.use_iana_default()
vtmpl = ipfix.template.for_specs(262, "wlanSSID[13]")
draw_template(vtmpl)

msg = ipfix.message.MessageBuffer()
msg.begin_export(8304)
msg.add_template(vtmpl)
msg.export_new_set(262)
msg.export_namedict({'wlanSSID':                 'ietf-a-v6only'})
draw_message(msg)

If it's changed to 12 it also works just fine, truncating the string to fit.

If it's changed to 14, or any other larger value, it crashes.
I'm guessing that it needs to correctly zero pad the string.

---------------------------------------------------------------------------
ExpatError                                Traceback (most recent call last)
<ipython-input-3-9dedd178bbfb> in <module>()
      5 msg.export_namedict({'wlanSSID':                 'ietf-a-v6only',
      6                      'protocolIdentifier':       6})
----> 7 draw_message(msg)

<ipython-input-1-ffa57e3f6760> in draw_message(msg, length)
      5 
      6 def draw_message(msg, length=256):
----> 7     return SVG(ipfix.vis.MessageBufferRenderer(msg, raster=4).render(length=length))
      8 
      9 def draw_template(tmpl):

/usr/lib/python3/dist-packages/IPython/core/display.py in __init__(self, data, url, filename)
    388                 data = None
    389 
--> 390         self.data = data
    391         self.url = url
    392         self.filename = None if filename is None else unicode_type(filename)

/usr/lib/python3/dist-packages/IPython/core/display.py in data(self, svg)
    496         from xml.dom import minidom
    497         svg = cast_bytes_py2(svg)
--> 498         x = minidom.parseString(svg)
    499         # get svg tag (should be 1)
    500         found_svg = x.getElementsByTagName('svg')

/usr/lib/python3.5/xml/dom/minidom.py in parseString(string, parser)
   1966     if parser is None:
   1967         from xml.dom import expatbuilder
-> 1968         return expatbuilder.parseString(string)
   1969     else:
   1970         from xml.dom import pulldom

/usr/lib/python3.5/xml/dom/expatbuilder.py in parseString(string, namespaces)
    923     else:
    924         builder = ExpatBuilder()
--> 925     return builder.parseString(string)
    926 
    927 

/usr/lib/python3.5/xml/dom/expatbuilder.py in parseString(self, string)
    221         parser = self.getParser()
    222         try:
--> 223             parser.Parse(string, True)
    224             self._setup_subset(string)
    225         except ParseEscape:

ExpatError: not well-formed (invalid token): line 1, column 4720