notebooks/Introducing Flows.ipynb

{
 "metadata": {
  "name": "",
  "signature": "sha256:1475994bad370dab9f57aacf9b90c1ac9c561ff8a7c3ad201c1b632f4325cec2"
 },
 "nbformat": 3,
 "nbformat_minor": 0,
 "worksheets": [
  {
   "cells": [
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "# Introducing Flows\n",
      "\n",
      "In this notebook, we'll use the QoF flow meter, Python `ipfix` module, and Pandas to explore the characteristics of flow data.\n",
      "\n",
      "Flow data analysis is in general somewhat difficult to experiment with, as it involves passive observation of network traffic data, which carries with it network end-user privacy risk and may additionally come with stringent regulatory and legal requirements. For this tutorial, we'll be using publicly available [WIDE MAWI traces](http://mawi.wide.ad.jp/mawi/), which, being collected from a transpacific backbone link, are not really representative of the types of traffic in access, enterprise, or academic networks; and, having their IP addresses anonymized without preserving structure, are not really useful for looking at the structure of the networks on either side of the backbone. We'll put up with these inconveniences for the sake of having real data to play with, though.\n",
      "\n",
      "If you have access to your own network traces, you can run these through [QoF](https://github.com/britram/qof): \n",
      "```\n",
      "qof --yaml qof-config.yaml --in my-trace-file.gz --out my-ipfix-file.ipfix\n",
      "```\n",
      "then point this notebook at the resulting IPFIX file. (Note, of course, that commentary in this notebook is based on the set of MAWI trace data used in the course, and will probably not match the results in your own data.)\n",
      "\n",
      "The QoF command we used to create the trace used in this notebook is shown below:\n",
      "```\n",
      "[brian@magpie ~]$ qof --verbose --yaml qof-simple-uniflow.yaml --in mawi-0330-30min.pcap.gz \\\n",
      "                    | gzip > mawi-0330-30min-uniflow.ipfix.gz\n",
      "[2014-06-23 15:57:45] qof 0.9.0 (\"Albula\") starting\n",
      "[2014-06-23 15:59:09] Processed 66397621 packets into 7634742 flows:\n",
      "[2014-06-23 15:59:09]   Mean flow rate 94245.55/s.\n",
      "[2014-06-23 15:59:09]   Mean packet rate 819632.16/s.\n",
      "[2014-06-23 15:59:09]   Virtual bandwidth 5044.6538 Mbps.\n",
      "[2014-06-23 15:59:09]   Maximum flow table size 159240.\n",
      "[2014-06-23 15:59:09]   579 flush events.\n",
      "[2014-06-23 15:59:09]   4453490 asymmetric/unidirectional flows detected (58.33%)\n",
      "[2014-06-23 15:59:09] Assembled 33813 fragments into 16810 packets:\n",
      "[2014-06-23 15:59:09]   Expired 26 incomplete fragmented packets. (0.00%)\n",
      "[2014-06-23 15:59:09]   Maximum fragment table size 23.\n",
      "[2014-06-23 15:59:09] Rejected 65071 packets during decode: (0.10%)\n",
      "[2014-06-23 15:59:09]   65071 due to incomplete headers: (0.10%)\n",
      "[2014-06-23 15:59:09]     52931 incomplete IPv6 extension headers. (0.08%)\n",
      "[2014-06-23 15:59:09]     12140 incomplete transport headers. (0.02%)\n",
      "[2014-06-23 15:59:09]     (Use a larger snaplen to reduce incomplete headers.)\n",
      "[2014-06-23 15:59:09] qof terminating\n",
      "```\n",
      "\n",
      "This notebook uses the [Pandas](http://pandas.pydata.org/) data analysis framework to explore a collection of flow data. So first, run the following code to set up the environment:"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "import ipfix\n",
      "import panfix\n",
      "import gzip\n",
      "import bz2\n",
      "\n",
      "import numpy as np\n",
      "import pandas as pd\n",
      "import matplotlib.pyplot as plt\n",
      "\n",
      "%matplotlib inline\n",
      "\n",
      "plt.rcParams['figure.figsize'] = (12, 6)\n",
      "\n",
      "ipfix.ie.use_iana_default()\n",
      "ipfix.types.use_integer_ipv4()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Now we can import some flows into a dataframe. This might take a while."
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# Set the name of the IPFIX file to work on here.\n",
      "ipfix_filename = \"../../mawi-0330-30min-uniflow.ipfix.gz\"\n",
      "# change to gzip.open, bz2.open or open, as appropriate\n",
      "ipfix_file_fn = gzip.open\n",
      "# Change to None for no limit, or set a limit to reduce memory requirements\n",
      "ipfix_max_flows = 1000000\n",
      "# Skip the first N flows when limiting (prevents short flow bias)\n",
      "ipfix_skip_flows = 2000000\n",
      "\n",
      "df = panfix.dataframe_from_ipfix(ipfix_filename, (\n",
      "                                 \"flowStartMilliseconds\",  \"flowEndMilliseconds\", \n",
      "                                 \"sourceIPv4Address\",      \"sourceTransportPort\",\n",
      "                                 \"destinationIPv4Address\", \"destinationTransportPort\", \n",
      "                                 \"protocolIdentifier\",     \"flowEndReason\",\n",
      "                                 \"octetDeltaCount\",        \"packetDeltaCount\"), \n",
      "                                 count=ipfix_max_flows, skip=ipfix_skip_flows,\n",
      "                                 open_fn=ipfix_file_fn)\n",
      "print(\"Loaded \"+str(len(df))+\" flows.\")"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Let's fix up the dataframe a bit:"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "df = panfix.coerce_timestamps(df) # fixup timestamp datatypes\n",
      "df = panfix.derive_duration(df)   # calculate duration of each flow \n",
      "df[\"flowDeltaCount\"] = 1          # one flow per flow (useful for aggregating by flows)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Let's have a look at a time series of flow counts:"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "flows_ts = df[\"flowDeltaCount\"].copy()\n",
      "flows_ts.index = df[\"flowEndMilliseconds\"]\n",
      "flows_ts.resample(how=\"sum\", rule=\"1s\").plot()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "To explain what's going on here, we first have understand two fundamental configuration parameters of flow meters that are important to intepreting results. First is the *passive timeout* (or *idle timeout*). Any flow for which no packets are seen during this interval will be expired and exported. Second is the *active timeout* which can also be thought of as the maximum duration of the flow record. The passive timeout ensures that flows without a \"natural\" end (i.e., all non-TCP flows) are eventually flushed and exported. The active timeout sets a maximum delay between the time the first packet of the flow is seen, and the time the first record representing packets in that flow are exported.\n",
      "\n",
      "Exported flows are ordered neither by start nor end time. Naturally terminated (i.e., through TCP FIN or RST) and actively timed-out flows are available for export immediately after the last packet, while passively timed-out flows will have export times up to one passive timeout in the past.\n",
      "\n",
      "QoF's passive timeout is set to 30 seconds by default, and its active timeout to 300 seconds.\n",
      "\n",
      "Given this knowledge, we can see three distinct intervals in the series of flow end times.\n",
      "\n",
      "- We see only passively timed-out flows for the first passive timeout interval (the 30 seconds between about 04:08:40 and 04:09:10).\n",
      "- The capture starts at (export time) 04:09:10. From here we see the steady state: flows terminated for all reasons. Here we see a flow export rate between 3900 and 4900 flows per second.\n",
      "- The capture ends at (export time) 04:13:10. For the last passive timeout interval, no passively timed out flows can be seen, only those timed out actively or naturally.\n",
      "\n",
      "These intervals are also reflected in the time series of start times:"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "flows_ts.index = df[\"flowStartMilliseconds\"]\n",
      "flows_ts.resample(how=\"sum\", rule=\"1s\").plot()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "## Layer 4 protocol analysis\n",
      "\n",
      "Now, let's have a look at the protocols in use in the trace file. To do this, we'll group by protocols and sum flow counts:"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "df.groupby(\"protocolIdentifier\")['flowDeltaCount'].aggregate(\"sum\").plot(kind=\"bar\")"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "This is probably easier to read if we get names for the protocols for the system protocols database. First, we need to load the protocol names into a dataframe."
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "proto_df = pd.read_table(\"../data/protocols\", header=None, index_col=1, usecols=(0,1), \n",
      "                         names=[\"protocolName\", \"protocolNumber\"])[\"protocolName\"]"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Now we can create a new data frame containing the aggregate (which will be indexed by protocol identifier), join it to the protocol numbers table, reindex the data frame by protocol name, then plot with names: "
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "proto_flows = pd.DataFrame(df.groupby(\"protocolIdentifier\")['flowDeltaCount'].aggregate(\"sum\"))\n",
      "proto_flows = proto_flows.join(proto_df)\n",
      "proto_flows.index = proto_flows[\"protocolName\"]\n",
      "proto_flows[\"flowDeltaCount\"].plot(kind=\"bar\")"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Here we see TCP, ICMP, and UDP dominating flow counts, with some ipv6 in ipv4 encapsulation, and negligble amounts of ipcencap (ipv4 in ipv4 encapsulation), GRE (tunneling), ESP (IPsec), and PIM (multicast encapsulation). Now let's look at the same breakdown by bytes:"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "proto_bytes = pd.DataFrame(df.groupby(\"protocolIdentifier\")['octetDeltaCount'].aggregate(\"sum\"))\n",
      "proto_bytes = proto_bytes.join(proto_df)\n",
      "proto_bytes.index = proto_bytes[\"protocolName\"]\n",
      "proto_bytes[\"octetDeltaCount\"].plot(kind=\"bar\")"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Here we see a very different picture: by bytes, almost all the traffic is TCP. We can also calculate this proportion numerically, by summing the number of bytes in TCP flows and dividing by the number of bytes in all flows:"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "sum(df[df[\"protocolIdentifier\"] == 6][\"octetDeltaCount\"]) / sum(df[\"octetDeltaCount\"])"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "## Counting Ports\n",
      "\n",
      "Let's have a look at port numbers by transport protocol. First, let's select UDP and TCP flows separately."
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "udp_df = df[df[\"protocolIdentifier\"] == 17]\n",
      "tcp_df = df[df[\"protocolIdentifier\"] == 6]"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Since we're looking at uniflows, and are interested in services that actually responded, we'll look at source addresses. We'll use the value_counts() shortcut for counting top N ports by flows count. First for UDP:"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "udp_df[\"sourceTransportPort\"].value_counts()[:10].plot(kind=\"bar\")"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Hm. Here we see DNS (53) and NTP (123), along with quite a lot of flows on a high port (58534). We can dig into those to try and figure out what's going on there:"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "udp_58534_df = udp_df[udp_df[\"sourceTransportPort\"] == 58534]\n",
      "udp_58534_df[\"sourceIPv4Address\"].value_counts()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Almost all of these flows come from a single source, which probably indicates scanning. Let's turn our attention to TCP now:"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "tcp_df[\"sourceTransportPort\"].value_counts()[:10].plot(kind=\"bar\")"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Here, Port 80 (HTTP) and 443 (HTTP over TLS) dominate, as expected. There's a little Port 22 (SSH) as well."
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "## Elephants and Mice\n",
      "\n",
      "We can have a more in-depth view of the relationship of flow rates and durations by plotting these on two-dimensional histogram. Here we'll have duration on the X axis and data rate (in nominal bits per second) on the Y axis. The weight of each bin will show the count of flows, the count of packets in flows, or the count of bytes in flows, falling into that bin. First execute the following to define the function we'll use here:"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "def plot_rate_duration_uniflow(df, by, filename=None):\n",
      "    plt.figure(figsize=(9,7))\n",
      "    plt.hexbin(x = df[\"durationSeconds\"],\n",
      "           y = (df[\"octetDeltaCount\"] * 8) / (df[\"durationSeconds\"] + 0.001), \n",
      "           C = df[by],\n",
      "           reduce_C_function = np.sum,\n",
      "           yscale='log',\n",
      "           bins='log',\n",
      "           cmap = plt.cm.binary)\n",
      "    cb = plt.colorbar()\n",
      "    cb.set_label(\"log10(\"+by+\")\")\n",
      "    plt.xlabel(\"duration (s)\")\n",
      "    plt.ylabel(\"data rate (bps)\")\n",
      "    if filename:\n",
      "        plt.savefig(filename)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Now let's look at the shape of all flows by flow count:"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "plot_rate_duration_uniflow(df, by=\"flowDeltaCount\")"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "The flow counts are dominated by very short flows with rates between about 10kbps and 10Mbps. There are a few long-duration low rate flows, and even fewer long duration high rate flows. Given a flow, there is a very high chance it is very short (a \"mouse\")."
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "plot_rate_duration_uniflow(df, by=\"packetDeltaCount\")"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Looking at packet counts tells a different story: here, the packet counts are dominated by longer flows in the 1Mbps - 10Mbps range (\"elephants\"), many of which are maximum duration (recalling that the QoF active timeout is 300 seconds; flows longer than that will be represented by multiple records)."
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "plot_rate_duration_uniflow(df, by=\"octetDeltaCount\")"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Viewed in terms of octets, the reign of the elephants is even more pronounced."
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "We can also look at this in terms of the size of the packets. Bulk transfer applications tend to have larger packets (closer to MTU, usually around 1500 bytes as set by Ethernet), while machine-to-machine applications use smaller packets, and various constant bit-rate protocols (usually media). We don't have per-packet size information in flow data, but we can approximate it by dividing the number of bytes in the flow by the number of packets."
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "(df[\"octetDeltaCount\"] / df[\"packetDeltaCount\"]).hist(range=(0,1600),bins=150, weights=df[\"packetDeltaCount\"])"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "The distribution is quite bimodal, with small packets clustered around 40 bpp, and large packets clustered around 1500. We can also see some dependency between bytes per packet and application (approximated by port) by plotting these in two dimensions:"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "def plot_port_bpp_uniflow(df, by=\"packetDeltaCount\", portrange=(0,65535), filename=None):\n",
      "    plt.figure(figsize=(9,7))\n",
      "    plt.hexbin(x = df[\"octetDeltaCount\"] / df[\"packetDeltaCount\"],\n",
      "           y = df[\"sourceTransportPort\"], \n",
      "           C = df[by],\n",
      "           reduce_C_function = np.sum,\n",
      "           bins='log',\n",
      "           cmap = plt.cm.binary,\n",
      "           extent=(0,1500,portrange[0],portrange[1])),\n",
      "    cb = plt.colorbar()\n",
      "    cb.set_label(\"log10(\"+by+\")\")\n",
      "    plt.xlabel(\"mean octets/packet\")\n",
      "    plt.ylabel(\"port\")\n",
      "    if filename:\n",
      "        plt.savefig(filename)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "plot_port_bpp_uniflow(tcp_df)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "plot_port_bpp_uniflow(tcp_df, portrange=(0,512))"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "plot_port_bpp_uniflow(udp_df)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "plot_port_bpp_uniflow(udp_df, portrange=(0,512))"
     ],
     "language": "python",
     "metadata": {},
     "outputs": []
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "- - - - - -\n",
      "This notebook is &copy; 2013-2014 Brian Trammell, and is licensed under a [Creative Commons Attribution-ShareAlike 4.0 International License](http://creativecommons.org/licenses/by-sa/4.0/)."
     ]
    }
   ],
   "metadata": {}
  }
 ]
}