Add network traffic example (#415)

* Add network traffic example

* Remove unused import

examples/datasets.yml

@@ -105,3 +105,9 @@ data:
     title: 'Solar Radiation NREL - NSRDB Meta'
     files:
       - NSRDB_StationsMeta.csv
+
+  - url: http://s3.amazonaws.com/datashader-data/maccdc2012_graph.zip
+    title: 'National CyberWatch Mid-Atlantic Collegiate Cyber Defense Competition'
+    files:
+      - maccdc2012_nodes.parq
+      - maccdc2012_edges.parq

examples/packet_capture_graph.ipynb

@@ -0,0 +1,259 @@
+{
+ "cells": [
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "# Graphing network packets"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## Preparing data"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "The data source comes from a publicly available network forensics repository: http://www.netresec.com/?page=PcapFiles. The selected file is https://download.netresec.com/pcap/maccdc-2012/maccdc2012_00000.pcap.gz. We will select only TCP traffic.\n",
+    "\n",
+    "```\n",
+    "tcpdump -qns 0 -r maccdc2012_00000.pcap | grep tcp > maccdc2012_00000.txt\n",
+    "```"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "For example, here is a snapshot of the resulting output:"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "```\n",
+    "09:30:07.780000 IP 192.168.202.68.8080 > 192.168.24.100.1038: tcp 1380\n",
+    "09:30:07.780000 IP 192.168.24.100.1038 > 192.168.202.68.8080: tcp 0\n",
+    "09:30:07.780000 IP 192.168.202.68.8080 > 192.168.24.100.1038: tcp 1380\n",
+    "09:30:07.780000 IP 192.168.202.68.8080 > 192.168.24.100.1038: tcp 1380\n",
+    "09:30:07.780000 IP 192.168.27.100.37877 > 192.168.204.45.41936: tcp 0\n",
+    "09:30:07.780000 IP 192.168.24.100.1038 > 192.168.202.68.8080: tcp 0\n",
+    "09:30:07.780000 IP 192.168.202.68.8080 > 192.168.24.100.1038: tcp 1380\n",
+    "09:30:07.780000 IP 192.168.202.68.8080 > 192.168.24.100.1038: tcp 1380\n",
+    "09:30:07.780000 IP 192.168.202.68.8080 > 192.168.24.100.1038: tcp 1380\n",
+    "09:30:07.780000 IP 192.168.202.68.8080 > 192.168.24.100.1038: tcp 1380\n",
+    "```"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "Given the directional nature of network traffic and the numerous ports per node, we will simplify the graph by treating traffic between nodes as undirected and ignorning the distinction between ports. The graph edges will have weights represented by the total number of bytes across both nodes in either direction."
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "```\n",
+    "python pcap_to_parquet.py maccdc2012_00000.txt\n",
+    "```"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "The resulting output will be two Parquet dataframes, `maccdc2012_nodes.parq` and `maccdc2012_edges.parq`."
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## Loading data"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "import pandas as pd\n",
+    "import datashader as ds\n",
+    "import datashader.transfer_functions as tf\n",
+    "\n",
+    "from colorcet import fire\n",
+    "from datashader.bundling import hammer_bundle\n",
+    "from datashader.layout import circular_layout\n",
+    "\n",
+    "from dask.distributed import Client\n",
+    "from fastparquet import ParquetFile\n",
+    "\n",
+    "client = Client()\n",
+    "width, height = 2000, 2000\n",
+    "x_range = (-0.01, 1.01)\n",
+    "y_range = (-0.01, 1.01)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "nodes_df = ParquetFile('maccdc2012_nodes.parq').to_pandas()\n",
+    "len(nodes_df)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "edges_df = ParquetFile('maccdc2012_edges.parq').to_pandas()\n",
+    "len(edges_df)"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## Edge bundling"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "collapsed": true
+   },
+   "outputs": [],
+   "source": [
+    "positioned_df = circular_layout(nodes_df, edges_df)\n",
+    "positioned_df = positioned_df.set_index('id')"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "collapsed": true
+   },
+   "outputs": [],
+   "source": [
+    "bundled_df = hammer_bundle(positioned_df, edges_df)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "collapsed": true
+   },
+   "outputs": [],
+   "source": [
+    "cvs = ds.Canvas(width, height, x_range, y_range)\n",
+    "img = tf.shade(cvs.points(bundled_df, 'x', 'y'), cmap=fire)\n",
+    "bundled_img = tf.set_background(img, color='black')"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "bundled_img"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## Showing nodes with active traffic"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "active_edges_df = edges_df[edges_df['weight'] > 0]"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "collapsed": true
+   },
+   "outputs": [],
+   "source": [
+    "active_nodes = active_edges_df.source.append(active_edges_df.target, ignore_index=True).unique()\n",
+    "active_nodes_df = pd.DataFrame(active_nodes, columns=['id'])\n",
+    "active_nodes_df = active_nodes_df.set_index('id')\n",
+    "active_nodes_df = active_nodes_df.join(positioned_df)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "cvs = ds.Canvas(width, height, x_range, y_range)\n",
+    "agg = cvs.points(active_nodes_df, 'x', 'y')\n",
+    "nodes_img = tf.spread(tf.shade(agg, cmap='red'), px=5)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "bundled_img + nodes_img"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "collapsed": true
+   },
+   "outputs": [],
+   "source": []
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "Python 3",
+   "language": "python",
+   "name": "python3"
+  },
+  "language_info": {
+   "codemirror_mode": {
+    "name": "ipython",
+    "version": 3
+   },
+   "file_extension": ".py",
+   "mimetype": "text/x-python",
+   "name": "python",
+   "nbconvert_exporter": "python",
+   "pygments_lexer": "ipython3",
+   "version": "3.6.1"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 2
+}

examples/pcap_to_parquet.py

@@ -0,0 +1,77 @@
+#!/usr/bin/env python
+
+"""
+Convert PCAP output to undirected graph and save in Parquet format.
+"""
+
+from __future__ import print_function
+
+import re
+import socket
+import struct
+import sys
+
+import fastparquet as fp
+import numpy as np
+import pandas as pd
+
+
+def ip_to_integer(s):
+    return struct.unpack("!I", socket.inet_aton(s))[0]
+
+
+def to_parquet(filename, prefix="maccdc2012"):
+    with open(filename) as f:
+        traffic = {}
+        nodes = set()
+
+        for line in f.readlines():
+            fields = line.split()
+            if not fields:
+                continue
+            try:
+                addresses = []
+
+                # Extract source IP address and convert to integer
+                m = re.match(r'\d+\.\d+\.\d+\.\d+', fields[2])
+                if not m:
+                    continue
+                addresses.append(ip_to_integer(m.group(0)))
+
+                # Extract target IP address and convert to integer
+                m = re.match(r'\d+\.\d+\.\d+\.\d+', fields[4])
+                if not m:
+                    continue
+                addresses.append(ip_to_integer(m.group(0)))
+
+                nodes = nodes.union(addresses)
+                key = tuple(sorted(addresses))
+
+                # Extract packet size
+                nbytes = int(fields[-1])
+
+                if key in traffic:
+                    traffic[key] += nbytes
+                else:
+                    traffic[key] = nbytes
+            except:
+                pass
+
+        # Anonymize IP addresses by subtracting minimum from all integers
+        min_node_id = min(nodes)
+        edges = []
+        for key in traffic:
+            edge = [key[0] - min_node_id, key[1] - min_node_id, traffic[key]]
+            edges.append(edge)
+
+        nodes_df = pd.DataFrame(np.array(list(nodes)) - min_node_id, columns=['id'])
+        edges_df = pd.DataFrame(np.array(edges), columns=['source', 'target', 'weight'])
+
+        fp.write('{}_nodes.parq'.format(prefix), nodes_df)
+        fp.write('{}_edges.parq'.format(prefix), edges_df)
+
+if __name__ == '__main__':
+    if len(sys.argv) >= 2:
+        to_parquet(sys.argv[1], prefix=sys.argv[2])
+    else:
+        to_parquet(sys.argv[1])