wrapper for tcpflow which enhances its stdout output to include easily parsed timestamp format, incl. milliseconds, tz offset, and packaged in flat json one-object-per-row ideal for mapreducing on

tcpflow-1.4.4_wrapper.coffee

#!/usr/bin/env coffee
#
# install on ubuntu:
#   sudo apt-get install automake autoconf libpcap-dev zlib1g-dev libboost-dev libcairo2-dev
#
#   cd /tmp/
#   git clone --recursive -b tcpflow-1.4.4 git@github.com:simsong/tcpflow.git tcpflow/
#   cd tcpflow/
#
#   sh bootstrap.sh
#   ./configure
#   make
#   sudo make install
#   which tcpflow
#   tcpflow --version
#
# usage:
#   sudo tcpflow -i eth0 -c -B -d5 "(tcp port 80)" 2>&1 | ./tcpreflow.coffee >> /var/log/server.log
#
# produces output like:
#   {"src":"010.069.000.015","sprt":"32912","dst":"166.070.123.071",
#    "dprt":"00443","data":"%xt%o%mi%1116821%42%19%\u0000",
#    "dir":"in","ts":"2014-5-18 0:12:14.369 6"}
#

stream = require 'stream'

class TcpFlowParser
  @create: (emit) ->
    ws = stream.Writable()
    msg = {}
    buf = ''

    pop = (type, data) ->
      if type is 'tcpflow'
        # ignoring for now
        # TODO: output JSON connection events
      else if type is 'packet'
        data.dir =
          # NOTICE: hard-coded for now
          if /172\.016\.000\.\d{3}/.test(data.src) and /00443/.test(data.sprt)
            'out'
          else
            'in'
        emit data
      return

    ws._write = (chunk, enc, next) ->
      buf += chunk.toString 'utf8'
      while (m = buf.match /\n+/) isnt null
        line = buf.substr 0, m.index + (m[0].length - 1)
        buf = buf.substr m.index + m[0].length

        # determine type of line
        if /^tcpflow: /.test line
          # can pop these immediately and ignore that they interrupt messages
          pop 'tcpflow', data: line
        else if (m = line.match /^(\d{3}\.\d{3}\.\d{3}\.\d{3})\.(\d{5})-(\d{3}\.\d{3}\.\d{3}\.\d{3})\.(\d{5}): (.+)$/) isnt null
          # don't pop any messages until a new line type is identified
          if msg.data
            pop 'packet', msg
          msg = src: m[1], sprt: m[2], dst: m[3], dprt: m[4], data: m[5]
        else # append to previous message
          msg.data += line
      next()
      return

    return ws

out = (event) ->
  d = new Date
  event.ts = "#{d.getFullYear()}-#{d.getMonth()+1}-#{d.getDate()} #{d.getHours()}:#{d.getMinutes()}:#{d.getSeconds()}.#{d.getMilliseconds()} #{d.getTimezoneOffset()/60}"
  process.stdout.write "#{JSON.stringify event}\n"

process.stdin.setEncoding 'utf8'
process.stdin.resume()
process.stdin
  .pipe TcpFlowParser.create (event) ->
    out event
    return