corsaro_flowtuple.c

Go to the documentation of this file.

/* 
 * corsaro
 *
 * Alistair King, CAIDA, UC San Diego
 * corsaro-info@caida.org
 * 
 * Copyright (C) 2012 The Regents of the University of California.
 * 
 * This file is part of corsaro.
 *
 * corsaro is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * corsaro is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with corsaro.  If not, see <http://www.gnu.org/licenses/>.
 *
 */

#include "config.h"
#include "corsaro_int.h"

#include <arpa/inet.h>
#include <assert.h>
#include <inttypes.h>
#include <stdio.h>
#include <string.h>

#include "libtrace.h"

#include "utils.h"

#include "corsaro_io.h"
#include "corsaro_file.h"
#include "corsaro_log.h"
#include "corsaro_plugin.h"

#include "corsaro_flowtuple.h"

/* This magic number is a legacy number from when we used to call it the
   'sixtuple' */
#ifdef CORSARO_SLASH_EIGHT

#define CORSARO_FLOWTUPLE_MAGIC 0x53495854
#else

#define CORSARO_FLOWTUPLE_MAGIC 0x53495855
#endif

#define PLUGIN_NAME "flowtuple"

static corsaro_plugin_t corsaro_flowtuple_plugin = {
  PLUGIN_NAME,                                 /* name */
  CORSARO_PLUGIN_ID_FLOWTUPLE,                    /* id */
  CORSARO_FLOWTUPLE_MAGIC,                        /* magic */
  CORSARO_PLUGIN_GENERATE_PTRS(corsaro_flowtuple),   /* func ptrs */
  NULL,                                        /* next */
};

static const char *class_names[] = {
  "flowtuple_backscatter",
  "flowtuple_icmpreq",
  "flowtuple_other",
};

KSORT_INIT(sixt, corsaro_flowtuple_t*, corsaro_flowtuple_lt);

/*KHASH_INIT(sixt, corsaro_flowtuple_t*, khint32_t, 1, */
KHASH_INIT(sixt, corsaro_flowtuple_t*, char, 0, 
           corsaro_flowtuple_hash_func, corsaro_flowtuple_hash_equal);

struct corsaro_flowtuple_state_t {
  khash_t(sixt) *st_hash[CORSARO_FLOWTUPLE_CLASS_MAX+1];
  corsaro_file_t *outfile;
};

struct corsaro_flowtuple_in_state_t {
  corsaro_in_record_type_t expected_type;
  int tuple_total;
  int tuple_cnt;
};

#define STATE(corsaro)                                          \
  (CORSARO_PLUGIN_STATE(corsaro, flowtuple, CORSARO_PLUGIN_ID_FLOWTUPLE))

#define STATE_IN(corsaro)                                               \
  (CORSARO_PLUGIN_STATE(corsaro, flowtuple_in, CORSARO_PLUGIN_ID_FLOWTUPLE))

#define PLUGIN(corsaro)                                         \
  (CORSARO_PLUGIN_PLUGIN(corsaro, CORSARO_PLUGIN_ID_FLOWTUPLE))

static int flowtuple_classify_packet(corsaro_t *corsaro, libtrace_packet_t *packet)
{
  void *temp = NULL;
  uint16_t ethertype;
  uint8_t proto;
  uint32_t remaining;

  libtrace_tcp_t  *tcp_hdr  = NULL;
  libtrace_icmp_t *icmp_hdr = NULL;

  /* check for ipv4 */
  if((temp = trace_get_layer3(packet, &ethertype, &remaining)) == NULL ||
     ethertype != TRACE_ETHERTYPE_IP)
    {
      /* not an ip packet */
      corsaro_log(__func__, corsaro, "non-ip packet found (ethertype: %x)", 
                ethertype);
      return CORSARO_FLOWTUPLE_CLASS_OTHER;
    }

  /* get the transport header */
  if((temp = trace_get_transport(packet, &proto, &remaining)) == NULL)
    {
      /* not enough payload */
      return CORSARO_FLOWTUPLE_CLASS_OTHER;
    }

  /* check for tcp */
  if(proto == TRACE_IPPROTO_TCP && remaining >= 4)
    {
      tcp_hdr = (libtrace_tcp_t *)temp;

      /* look for SYNACK or RST */
      if((tcp_hdr->syn && tcp_hdr->ack) || tcp_hdr->rst)
        {
          return CORSARO_FLOWTUPLE_CLASS_BACKSCATTER;
        }
      else
        {
          return CORSARO_FLOWTUPLE_CLASS_OTHER;
        }
    }
  /* check for icmp */
  else if(proto == TRACE_IPPROTO_ICMP && remaining >= 2)
    {
      icmp_hdr = (libtrace_icmp_t *)temp;
      if(icmp_hdr->type == 0  ||
         icmp_hdr->type == 3  ||
         icmp_hdr->type == 4  ||
         icmp_hdr->type == 5  ||
         icmp_hdr->type == 11 ||
         icmp_hdr->type == 12 ||
         icmp_hdr->type == 14 ||
         icmp_hdr->type == 16 ||
         icmp_hdr->type == 18)
        {
          return CORSARO_FLOWTUPLE_CLASS_BACKSCATTER;
        }
      else
        {
          return CORSARO_FLOWTUPLE_CLASS_ICMPREQ;
        }
    }
  else
    {
      return CORSARO_FLOWTUPLE_CLASS_OTHER;
    }

  return -1;
}

static int sort_hash(corsaro_t *corsaro, kh_sixt_t *hash, corsaro_flowtuple_t ***sorted)
{
  khiter_t i;
  corsaro_flowtuple_t **ptr;

  if((ptr = malloc(sizeof(corsaro_flowtuple_t*)*kh_size(hash))) == NULL)
    {
      corsaro_log(__func__, corsaro, "could not malloc array for sorted keys");
      return -1;
    }
  *sorted = ptr;

  if(kh_size(hash) == 0)
    {
      /* no need to try and sort an empty hash */
      return 0;
    }
                                                
  for(i = kh_begin(hash); i != kh_end(hash); ++i)                       
    {   
      if(kh_exist(hash, i))                                     
        {
          *ptr = kh_key(hash, i);
          ptr++;
        }
    }

  ks_heapsort(sixt, kh_size(hash), *sorted);
  return 0;
}

static int binary_dump(corsaro_t *corsaro, corsaro_flowtuple_class_type_t dist)
{
  kh_sixt_t *h = STATE(corsaro)->st_hash[dist];

  int j;
  khiter_t i = 0;
  
  corsaro_flowtuple_t **sorted_keys;

  uint8_t hbuf[4+2+4];
  uint8_t *hptr = &hbuf[0];

  corsaro_flowtuple_t *flowtuple = NULL;;

  bytes_htonl(hptr, CORSARO_FLOWTUPLE_MAGIC);
  hptr+=4;

  bytes_htons(hptr, dist);
  hptr+=2;

  bytes_htonl(hptr, kh_size(h));

  if(corsaro_file_write(corsaro, STATE(corsaro)->outfile, &hbuf[0], 10) != 10)
    {
      corsaro_log(__func__, corsaro, "could not dump byte flowtuple header to file");
      return -1;
    }
  
  if(kh_size(h) > 0)
    {
      if(sort_hash(corsaro, h, &sorted_keys) != 0)
        {
          corsaro_log(__func__, corsaro, "could not sort keys");
          return -1;
        }

      for(j = 0; j < kh_size(h); j++)                   
        {               
          i = kh_get(sixt, h, sorted_keys[j]);
          if(kh_exist(h, i))
            {
              flowtuple = kh_key(h, i);
              /* store the value in the flowtuple structure */
              /*kh_key(h, i)->packet_cnt = htonl(kh_val(h, i));*/
              /*flowtuple->packet_cnt = htonl(flowtuple->packet_cnt);*/
              
              if(corsaro_file_write(corsaro, STATE(corsaro)->outfile,
                                  flowtuple, CORSARO_FLOWTUPLE_BYTECNT) != 
                 CORSARO_FLOWTUPLE_BYTECNT)
                {
                  corsaro_log(__func__, corsaro, 
                            "could not write flowtuple to file");
                  return -1;
                }
            }
        }
      free(sorted_keys);
    }

  if(corsaro_file_write(corsaro, STATE(corsaro)->outfile, &hbuf[0], 6) != 6)
    {
      corsaro_log(__func__, corsaro, "could not dump flowtuple trailer to file");
      return -1;
    }
  return 0;
}

static int ascii_dump(corsaro_t *corsaro, corsaro_flowtuple_class_type_t dist)
{
  khiter_t i;
  int j;
  corsaro_flowtuple_t *key;
  corsaro_flowtuple_t **sorted_keys;

  /*const char *name = class_names[dist];*/
  kh_sixt_t *h = STATE(corsaro)->st_hash[dist];

  corsaro_flowtuple_class_start_t class_start;
  corsaro_flowtuple_class_end_t class_end;

  class_start.magic = CORSARO_FLOWTUPLE_MAGIC;
  class_start.class_type = dist;
  class_start.count = kh_size(h);

  class_end.magic = CORSARO_FLOWTUPLE_MAGIC;
  class_end.class_type = dist;

  corsaro_flowtuple_class_start_fprint(corsaro, STATE(corsaro)->outfile, 
                                    &class_start);
  
  if(kh_size(h) > 0)
    {
      if(sort_hash(corsaro, h, &sorted_keys) != 0)
        {
          corsaro_log(__func__, corsaro, "could not sort keys");
          return -1;
        }
      for(j = 0; j < kh_size(h); j++)                   
        {               
          i = kh_get(sixt, h, sorted_keys[j]);
          if(kh_exist(h, i))
            {
              key = kh_key(h, i);
              /*key->packet_cnt = htonl(key->packet_cnt);*/
              corsaro_flowtuple_fprint(corsaro, STATE(corsaro)->outfile, key);  
            }
        }
      free(sorted_keys);
    }

  corsaro_flowtuple_class_end_fprint(corsaro, STATE(corsaro)->outfile, &class_end);
  return 0;
}

static int validate_class_start(corsaro_flowtuple_class_start_t *class)
{
  /* byteswap the values */
  class->magic = ntohl(class->magic);
  class->class_type = ntohs(class->class_type);
  class->count = ntohl(class->count);

  /* do some sanity checking */
  if(class->magic != CORSARO_FLOWTUPLE_MAGIC || 
     class->class_type > CORSARO_FLOWTUPLE_CLASS_MAX) 
    {
      return 0;
    }
  return 1;
}

static int read_class_start(corsaro_in_t *corsaro, 
                            corsaro_in_record_type_t *record_type,
                            corsaro_in_record_t *record)
{
  off_t bytes_read;

  if((bytes_read =
      corsaro_io_read_bytes(corsaro, record, 
                          sizeof(corsaro_flowtuple_class_start_t))) != 
     sizeof(corsaro_flowtuple_class_start_t))
    {
      *record_type = CORSARO_IN_RECORD_TYPE_NULL;
      return bytes_read;
    }

  if(validate_class_start((corsaro_flowtuple_class_start_t *)record->buffer) != 1)
    {
      corsaro_log_in(__func__, corsaro, "could not validate flowtuple class");
      corsaro_log_in(__func__, corsaro, 
                     "it is possible this flowtuple file was written "
#ifdef CORSARO_SLASH_EIGHT
                     "without /8 "
#else
                     "with /8 "
#endif
                     "optimizations enabled");
      corsaro_log_in(__func__, corsaro, "try rebuilding using the "
#ifdef CORSARO_SLASH_EIGHT
                    "--without-slash-eight "
#else
                    "--with-slash-eight=0 "
#endif
                    "configure option");
      *record_type = CORSARO_IN_RECORD_TYPE_NULL;
      return -1;
    }

  assert(bytes_read == sizeof(corsaro_flowtuple_class_start_t));

  *record_type = CORSARO_IN_RECORD_TYPE_FLOWTUPLE_CLASS_START;
  STATE_IN(corsaro)->tuple_total = ((corsaro_flowtuple_class_start_t *)
                                  record->buffer)->count;

  STATE_IN(corsaro)->expected_type = (STATE_IN(corsaro)->tuple_total == 0) ?
    CORSARO_IN_RECORD_TYPE_FLOWTUPLE_CLASS_END :
    CORSARO_IN_RECORD_TYPE_FLOWTUPLE_FLOWTUPLE;
  
  return bytes_read;
}

static int validate_class_end(corsaro_flowtuple_class_end_t *class)
{
  /* byteswap the values */
  class->magic = ntohl(class->magic);
  class->class_type = ntohs(class->class_type);

  /* do some sanity checking */
  if(class->magic != CORSARO_FLOWTUPLE_MAGIC || 
     class->class_type > CORSARO_FLOWTUPLE_CLASS_MAX) 
    {
      return 0;
    }
  return 1;
}

static int read_class_end(corsaro_in_t *corsaro, 
                          corsaro_in_record_type_t *record_type,
                          corsaro_in_record_t *record)
{
  off_t bytes_read;

  if((bytes_read =
      corsaro_io_read_bytes(corsaro, record, 
                          sizeof(corsaro_flowtuple_class_end_t))) != 
     sizeof(corsaro_flowtuple_class_end_t))
    {
      *record_type = CORSARO_IN_RECORD_TYPE_NULL;
      return bytes_read;
    }

  if(validate_class_end((corsaro_flowtuple_class_end_t *)record->buffer) != 1)
    {
      corsaro_log_in(__func__, corsaro, "could not validate flowtuple class end");
      *record_type = CORSARO_IN_RECORD_TYPE_NULL;
      return -1;
    }

  assert(bytes_read == sizeof(corsaro_flowtuple_class_end_t));

  *record_type = CORSARO_IN_RECORD_TYPE_FLOWTUPLE_CLASS_END;
  STATE_IN(corsaro)->expected_type = (((corsaro_flowtuple_class_end_t *)
                                     record->buffer)->class_type == 
                                    (uint16_t)CORSARO_FLOWTUPLE_CLASS_MAX) ? 
    CORSARO_IN_RECORD_TYPE_IO_INTERVAL_END : 
    CORSARO_IN_RECORD_TYPE_FLOWTUPLE_CLASS_START;
  STATE_IN(corsaro)->tuple_total = 0;
  STATE_IN(corsaro)->tuple_cnt = 0;
  
  return bytes_read;
}

static int validate_flowtuple(corsaro_flowtuple_t *flowtuple)
{
  /* there is no real validation that can be done with this */
  return 1;
}

static int read_flowtuple(corsaro_in_t *corsaro, 
                         corsaro_in_record_type_t *record_type,
                         corsaro_in_record_t *record)
{
  off_t bytes_read;

  if((bytes_read = corsaro_io_read_bytes(corsaro, record, 
                                       sizeof(corsaro_flowtuple_t))) != 
     sizeof(corsaro_flowtuple_t))
    {
      *record_type = CORSARO_IN_RECORD_TYPE_NULL;
      return bytes_read;
    }

  if(validate_flowtuple((corsaro_flowtuple_t *)record->buffer) != 1)
    {
      corsaro_log_in(__func__, corsaro, "could not validate flowtuple");
      *record_type = CORSARO_IN_RECORD_TYPE_NULL;
      return -1;
    }

  *record_type = CORSARO_IN_RECORD_TYPE_FLOWTUPLE_FLOWTUPLE;

  if(++(STATE_IN(corsaro)->tuple_cnt) == STATE_IN(corsaro)->tuple_total)
    {
      STATE_IN(corsaro)->expected_type = CORSARO_IN_RECORD_TYPE_FLOWTUPLE_CLASS_END;
    }

  assert(bytes_read == sizeof(corsaro_flowtuple_t));

  return bytes_read;
}

/* == PUBLIC PLUGIN FUNCS BELOW HERE == */

corsaro_plugin_t *corsaro_flowtuple_alloc(corsaro_t *corsaro)
{
  return &corsaro_flowtuple_plugin;
}

int corsaro_flowtuple_probe_filename(const char *fname)
{
  /* look for 'corsaro_flowtuple' in the name */
  return corsaro_plugin_probe_filename(fname, &corsaro_flowtuple_plugin);
}

int corsaro_flowtuple_probe_magic(corsaro_in_t *corsaro, corsaro_file_in_t *file)
{
  char buffer[1024];
  int len;
  int i = CORSARO_IO_INTERVAL_HEADER_BYTE_LEN;

  len = corsaro_file_rpeek(corsaro, file, buffer, sizeof(buffer));

  /* an corsaro flowtuple file will have 'SIX[TU]' at 14, 15, 16, 17 */
  if(len >= CORSARO_IO_INTERVAL_HEADER_BYTE_LEN+4 && 
     buffer[i++] == 'S' && buffer[i++] == 'I' && 
     buffer[i++] == 'X' && (buffer[i] == 'T' || buffer[i] == 'U'))
    {
      return 1;
    }

  return 0;
}

int corsaro_flowtuple_init_output(corsaro_t *corsaro)
{
  int i;
  struct corsaro_flowtuple_state_t *state;
  corsaro_plugin_t *plugin = PLUGIN(corsaro);
  assert(plugin != NULL);

  if((state = malloc_zero(sizeof(struct corsaro_flowtuple_state_t))) == NULL)
    {
      corsaro_log(__func__, corsaro, 
                "could not malloc corsaro_flowtuple_state_t");
      goto err;
    }
  corsaro_plugin_register_state(corsaro->plugin_manager, plugin, state);

  /* open the output file */
  if((state->outfile = corsaro_io_prepare_file(corsaro, plugin->name)) == NULL)
    {
      corsaro_log(__func__, corsaro, "could not open %s output file", 
                plugin->name);
      goto err;
    }

  for(i = 0; i <= CORSARO_FLOWTUPLE_CLASS_MAX; i++)
    {
      assert(state->st_hash[i] == NULL);
      state->st_hash[i] = kh_init(sixt);
      assert(state->st_hash[i] != NULL);
    }
  
  return 0;

 err:
  corsaro_flowtuple_close_output(corsaro);
  return -1;
}

int corsaro_flowtuple_init_input(corsaro_in_t *corsaro)
{
  struct corsaro_flowtuple_in_state_t *state;
  corsaro_plugin_t *plugin = PLUGIN(corsaro);
  assert(plugin != NULL);

  if((state = malloc_zero(sizeof(struct corsaro_flowtuple_in_state_t))) == NULL)
    {
      corsaro_log_in(__func__, corsaro, 
                "could not malloc corsaro_flowtuple_state_t");
      goto err;
    }
  corsaro_plugin_register_state(corsaro->plugin_manager, plugin, state);

  /* we initially expect an corsaro interval record */
  state->expected_type = CORSARO_IN_RECORD_TYPE_IO_INTERVAL_START;

  /* don't set the tuple_cnt until we actually see a class start record */
  
  return 0;

 err:
  corsaro_flowtuple_close_input(corsaro);
  return -1;
}

int corsaro_flowtuple_close_input(corsaro_in_t *corsaro)
{
  struct corsaro_flowtuple_in_state_t *state = STATE_IN(corsaro);

  if(state != NULL)
    {
      corsaro_plugin_free_state(corsaro->plugin_manager, PLUGIN(corsaro));
    }
  return 0;
}

int corsaro_flowtuple_close_output(corsaro_t *corsaro)
{
  int i;
  struct corsaro_flowtuple_state_t *state = STATE(corsaro);

  if(state != NULL)
    {
      for(i = 0; i <= CORSARO_FLOWTUPLE_CLASS_MAX; i++)
        {
          if(state->st_hash[i] != NULL)
            {
              kh_free(sixt, state->st_hash[i], &corsaro_flowtuple_free);
              kh_destroy(sixt, state->st_hash[i]);
              state->st_hash[i] = NULL;
            }
        }

      if(state->outfile != NULL)
        {
          corsaro_file_close(corsaro, state->outfile);
          state->outfile = NULL;
        }
      corsaro_plugin_free_state(corsaro->plugin_manager, PLUGIN(corsaro));
    }
  return 0;
}

off_t corsaro_flowtuple_read_record(struct corsaro_in *corsaro, 
                               corsaro_in_record_type_t *record_type, 
                               corsaro_in_record_t *record)
{
  
  struct corsaro_flowtuple_in_state_t *state = STATE_IN(corsaro);

  off_t bytes_read = -1;

  /* we have 5 different types of records that could be in this file */
  switch(state->expected_type)
    {
    case CORSARO_IN_RECORD_TYPE_IO_INTERVAL_START:
      /* ask the io subsystem to read it for us */
      bytes_read = corsaro_io_read_interval_start(corsaro, corsaro->file, 
                                          record_type, record);
      if(bytes_read == sizeof(corsaro_interval_t))
        {
          state->expected_type = CORSARO_IN_RECORD_TYPE_FLOWTUPLE_CLASS_START;
        }
      break;

    case CORSARO_IN_RECORD_TYPE_FLOWTUPLE_CLASS_START:
      /* we'll handle this one */
      bytes_read = read_class_start(corsaro, record_type, record);
      break;

    case CORSARO_IN_RECORD_TYPE_FLOWTUPLE_FLOWTUPLE:
      /* we'll handle this too */
      bytes_read = read_flowtuple(corsaro, record_type, record);
      break;

    case CORSARO_IN_RECORD_TYPE_FLOWTUPLE_CLASS_END:
      /* as with this */
      bytes_read = read_class_end(corsaro, record_type, record);
      break;

    case CORSARO_IN_RECORD_TYPE_IO_INTERVAL_END:
      /* ask the io subsystem to read it for us */
      bytes_read = corsaro_io_read_interval_end(corsaro, corsaro->file,
                                              record_type, record);
      if(bytes_read == sizeof(corsaro_interval_t))
        {
          state->expected_type = CORSARO_IN_RECORD_TYPE_IO_INTERVAL_START;
        }
      break;

    default:
      corsaro_log_in(__func__, corsaro, "invalid expected record type");
    }

  return bytes_read;
}

off_t corsaro_flowtuple_read_global_data_record(struct corsaro_in *corsaro, 
                              enum corsaro_in_record_type *record_type, 
                              struct corsaro_in_record *record)
{
  /* we write nothing to the global file. someone messed up */
  return -1;
}

int corsaro_flowtuple_start_interval(corsaro_t *corsaro, corsaro_interval_t *int_start)
{
  int i;
  for(i = 0; i <= CORSARO_FLOWTUPLE_CLASS_MAX; i++)
    {
      if(STATE(corsaro)->st_hash[i] == NULL)
        {
          corsaro_log(__func__, corsaro,
                    "corsaro_flowtuple_init must be called first");
          return -1;
        }
      else
        {
          /* free all of the flowtuples */
          kh_free(sixt, STATE(corsaro)->st_hash[i], &corsaro_flowtuple_free);
          kh_clear(sixt, STATE(corsaro)->st_hash[i]);
        }
    }
  return 0;
}

int corsaro_flowtuple_end_interval(corsaro_t *corsaro, corsaro_interval_t *int_end)
{
  int i;

  corsaro_io_write_interval_start(corsaro, STATE(corsaro)->outfile, 
                                &corsaro->interval_start);

  for(i = 0; i <= CORSARO_FLOWTUPLE_CLASS_MAX; i++)
    {
      if((CORSARO_FILE_MODE(STATE(corsaro)->outfile) == CORSARO_FILE_MODE_ASCII && 
          ascii_dump(corsaro, i) != 0) ||
         (CORSARO_FILE_MODE(STATE(corsaro)->outfile) == CORSARO_FILE_MODE_BINARY && 
          binary_dump(corsaro, i) != 0))
        {
          corsaro_log(__func__, corsaro, "could not dump hash");
          return -1;
        }
    }

  corsaro_io_write_interval_end(corsaro, STATE(corsaro)->outfile, int_end);

  return 0;
}

int corsaro_flowtuple_process_packet(corsaro_t *corsaro,
                                  corsaro_packet_t *packet)
{
  libtrace_packet_t *ltpacket = LT_PKT(packet);
  libtrace_ip_t *ip_hdr = NULL;
  libtrace_icmp_t *icmp_hdr = NULL;
  libtrace_tcp_t *tcp_hdr = NULL;
  corsaro_flowtuple_t  t;
  int class;

  if((ip_hdr = trace_get_ip(ltpacket)) != NULL)
    {
      /* finish building the flowtuple */
      t.ip_len = ip_hdr->ip_len;

      t.src_ip = ip_hdr->ip_src.s_addr;
      CORSARO_FLOWTUPLE_IP_TO_SIXT(ip_hdr->ip_dst.s_addr, &t);

      t.protocol = ip_hdr->ip_p;
      t.tcp_flags = 0; /* in case we don't find a tcp header */

      t.ttl = ip_hdr->ip_ttl;

      if(ip_hdr->ip_p == TRACE_IPPROTO_ICMP && 
         (icmp_hdr = trace_get_icmp(ltpacket)) != NULL)
        {
          t.src_port = htons(icmp_hdr->type);
          t.dst_port = htons(icmp_hdr->code);
        }
      else
        {
          if(ip_hdr->ip_p == TRACE_IPPROTO_TCP && 
             (tcp_hdr = trace_get_tcp(ltpacket)) != NULL)
            {
              /* we have ignore the NS flag because it doesn't fit in
                 an 8 bit field. blame alberto (ak - 2/2/12) */
              t.tcp_flags = (
                             (tcp_hdr->cwr << 7) |
                             (tcp_hdr->ece << 6) |
                             (tcp_hdr->urg << 5) |
                             (tcp_hdr->ack << 4) |
                             (tcp_hdr->psh << 3) |
                             (tcp_hdr->rst << 2) |
                             (tcp_hdr->syn << 1) |
                             (tcp_hdr->fin << 0)
                             );
            }

          t.src_port = htons(trace_get_source_port(ltpacket));
          t.dst_port = htons(trace_get_destination_port(ltpacket));
        }

      /* classify this packet and increment the appropriate hash */
      if((class = flowtuple_classify_packet(corsaro, ltpacket)) < 0)
        {
          corsaro_log(__func__, corsaro, "could not classify packet");
          return -1;
        }

      if(class == CORSARO_FLOWTUPLE_CLASS_BACKSCATTER)
        {
          packet->state.flags |= CORSARO_PACKET_STATE_FLAG_BACKSCATTER;
        }
      
      if(corsaro_flowtuple_add_inc(STATE(corsaro)->st_hash[class], &t, 1) != 0)
        {
          corsaro_log(__func__, corsaro, "could not increment value for flowtuple");
          return -1;
        }
    }
  /*
  else
    {
      corsaro_log(__func__, corsaro, "packet found without IPv4 header\n");
    }
  */
  return 0;
}

/* ==== FlowTuple External Convenience Functions ==== */

uint32_t corsaro_flowtuple_get_source_ip(corsaro_flowtuple_t *flowtuple)
{
  assert(flowtuple != NULL);

  return flowtuple->src_ip;
}

uint32_t corsaro_flowtuple_get_destination_ip(corsaro_flowtuple_t *flowtuple)
{
  assert(flowtuple != NULL);

  return CORSARO_FLOWTUPLE_SIXT_TO_IP(flowtuple);
}

off_t corsaro_flowtuple_fprint(corsaro_t *corsaro, corsaro_file_t *file,
                           corsaro_flowtuple_t *flowtuple)
{
  char ip_a[INET_ADDRSTRLEN];
  char ip_b[INET_ADDRSTRLEN];
  uint32_t tmp;

  assert(corsaro != NULL);
  assert(file != NULL);
  assert(flowtuple != NULL);

  tmp = flowtuple->src_ip;
  inet_ntop(AF_INET,&tmp, &ip_a[0], 16);
  tmp = CORSARO_FLOWTUPLE_SIXT_TO_IP(flowtuple);
  inet_ntop(AF_INET, &tmp, &ip_b[0], 16);
  
  return corsaro_file_printf(corsaro, file, "%s|%s"
                           "|%"PRIu16"|%"PRIu16
                           "|%"PRIu8"|%"PRIu8"|0x%02"PRIx8
                           "|%"PRIu16
                           ",%"PRIu32"\n",                                      
                           ip_a, ip_b,
                           ntohs(flowtuple->src_port), 
                           ntohs(flowtuple->dst_port),
                           flowtuple->protocol, 
                           flowtuple->ttl,
                           flowtuple->tcp_flags,
                           ntohs(flowtuple->ip_len), 
                           ntohl(flowtuple->packet_cnt)); 
}

void corsaro_flowtuple_print(corsaro_flowtuple_t *flowtuple)
{
  char ip_a[16];
  char ip_b[16];
  uint32_t tmp;

  assert(flowtuple != NULL);

  tmp = flowtuple->src_ip;
  inet_ntop(AF_INET,&tmp, &ip_a[0], 16);
  tmp = CORSARO_FLOWTUPLE_SIXT_TO_IP(flowtuple);
  inet_ntop(AF_INET, &tmp, &ip_b[0], 16);

  fprintf(stdout, "%s|%s"
         "|%"PRIu16"|%"PRIu16
         "|%"PRIu8"|%"PRIu8"|0x%02"PRIx8
         "|%"PRIu16
         ",%"PRIu32"\n",                                        
         ip_a, ip_b,
         ntohs(flowtuple->src_port), 
         ntohs(flowtuple->dst_port),
         flowtuple->protocol, 
         flowtuple->ttl,
         flowtuple->tcp_flags,
         ntohs(flowtuple->ip_len), 
         ntohl(flowtuple->packet_cnt));
}

off_t corsaro_flowtuple_class_start_fprint(corsaro_t *corsaro, corsaro_file_t *file, 
                                        corsaro_flowtuple_class_start_t *class)
{
  return corsaro_file_printf(corsaro, file, 
                           "START %s %"PRIu32"\n", 
                           class_names[class->class_type], 
                           class->count);
}

void corsaro_flowtuple_class_start_print(corsaro_flowtuple_class_start_t *class)
{
  fprintf(stdout, "START %s %"PRIu32"\n", class_names[class->class_type], 
          class->count);
}

off_t corsaro_flowtuple_class_end_fprint(corsaro_t *corsaro, corsaro_file_t *file, 
                                      corsaro_flowtuple_class_end_t *class)
{
  return corsaro_file_printf(corsaro, file, "END %s\n", 
                           class_names[class->class_type]);
}

void corsaro_flowtuple_class_end_print(corsaro_flowtuple_class_end_t *class)
{
  fprintf(stdout, "END %s\n", class_names[class->class_type]);
}

off_t corsaro_flowtuple_record_fprint(corsaro_t *corsaro, corsaro_file_t *file, 
                                   corsaro_in_record_type_t record_type,
                                   corsaro_in_record_t *record)
{
  switch(record_type)
    {
    case CORSARO_IN_RECORD_TYPE_FLOWTUPLE_CLASS_START:
      return corsaro_flowtuple_class_start_fprint(corsaro, file, 
                            (corsaro_flowtuple_class_start_t *)record->buffer);
      break;

    case CORSARO_IN_RECORD_TYPE_FLOWTUPLE_CLASS_END:
      return corsaro_flowtuple_class_end_fprint(corsaro, file,
                            (corsaro_flowtuple_class_end_t *)record->buffer);
      break;

    case CORSARO_IN_RECORD_TYPE_FLOWTUPLE_FLOWTUPLE:
      return corsaro_flowtuple_fprint(corsaro, file, 
                                   (corsaro_flowtuple_t *)record->buffer);
      break;

    default:
      corsaro_log(__func__, corsaro, "record_type %d not a flowtuple record",
                record_type);
      return -1;
      break;
    }

  return -1;
}

int corsaro_flowtuple_record_print(corsaro_in_record_type_t record_type,
                                corsaro_in_record_t *record)
{
  switch(record_type)
    {
    case CORSARO_IN_RECORD_TYPE_FLOWTUPLE_CLASS_START:
      corsaro_flowtuple_class_start_print(
                        (corsaro_flowtuple_class_start_t *)record->buffer);
      break;

    case CORSARO_IN_RECORD_TYPE_FLOWTUPLE_CLASS_END:
      corsaro_flowtuple_class_end_print(
                            (corsaro_flowtuple_class_end_t *)record->buffer);
      break;

    case CORSARO_IN_RECORD_TYPE_FLOWTUPLE_FLOWTUPLE:
      corsaro_flowtuple_print((corsaro_flowtuple_t *)record->buffer);
      break;

    default:
      corsaro_log_file(__func__, NULL, "record_type %d not a flowtuple record",
                record_type);
      return -1;
      break;
    }

  return 0;
}

void corsaro_flowtuple_free(corsaro_flowtuple_t *t)
{
  free(t);
}

int corsaro_flowtuple_add_inc(void *h, corsaro_flowtuple_t *t, int increment)
{
  kh_sixt_t *hash = (kh_sixt_t *)h;
  int khret;
  khiter_t khiter;
  corsaro_flowtuple_t *new_6t = NULL;

  assert(hash != NULL);
  
  /* check if this is in the hash already */
  if((khiter = kh_get(sixt, hash, t)) == kh_end(hash))
    {
      /* create a new tuple struct */
      if((new_6t = malloc(sizeof(corsaro_flowtuple_t))) == NULL)
        {
          corsaro_log_file(__func__, NULL, "malloc failed");
          return -1;
        }

      /* fill it */
      memcpy(new_6t, t, sizeof(corsaro_flowtuple_t));

      /* add it to the hash */
      khiter = kh_put(sixt, hash, new_6t, &khret);
      /* set the count to one */
      /*kh_value(hash, khiter) = increment;*/
      new_6t->packet_cnt = htonl(increment);
    }
  else
    {
      /* simply increment the existing one */
      /*kh_value(hash, khiter)+=increment;*/
      new_6t = kh_key(hash, khiter);
      new_6t->packet_cnt = htonl(ntohl(new_6t->packet_cnt)+increment);
    }
  return 0;
}