corsaro.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 <assert.h>
#include <inttypes.h>
#include <stdlib.h>
#include <string.h>

#include "libtrace.h"

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

static corsaro_packet_t *corsaro_packet_alloc(corsaro_t *corsaro)
{
  corsaro_packet_t *pkt;

  if((pkt = malloc_zero(sizeof(corsaro_packet_t))) == NULL)
    {
      corsaro_log(__func__, corsaro, "could not malloc corsaro_packet");
      return NULL;
    }
  /* this corsaro packet still needs a libtrace packet to be loaded... */
  return pkt;
}

static void corsaro_packet_state_reset(corsaro_packet_t *packet)
{
  assert(packet != NULL);
  packet->state.flags = 0;
}

static void corsaro_packet_free(corsaro_packet_t *packet)
{
  /* we will assume that somebody else is taking care of the libtrace packet */
  if(packet != NULL)
    {
      free(packet);
    }
}

static void corsaro_free(corsaro_t *corsaro)
{
  corsaro_plugin_t *p = NULL;
  
  if(corsaro == NULL)
    {
      /* nothing to be done... */
      return;
    }

  if(corsaro->uridata != NULL)
    {
      free(corsaro->uridata);
      corsaro->uridata = NULL;
    }

  if(corsaro->monitorname != NULL)
    {
      free(corsaro->monitorname);
      corsaro->monitorname = NULL;
    }

  if(corsaro->template != NULL)
    {
      free(corsaro->template);
      corsaro->template = NULL;
    }
  
  if(corsaro->global_file != NULL)
    {
      corsaro_file_close(corsaro, corsaro->global_file);
      corsaro->global_file = NULL;
    }

  corsaro_log_close(corsaro);

  if(corsaro->packet != NULL)
    {
      corsaro_packet_free(corsaro->packet);
      corsaro->packet = NULL;
    }

  if(corsaro->plugin_manager != NULL)
    {
      while((p = corsaro_plugin_next(corsaro->plugin_manager, p)) != NULL)
        {
          p->close_output(corsaro);
        }

      corsaro_plugin_manager_free(corsaro->plugin_manager);
      corsaro->plugin_manager = NULL;
    }

  free(corsaro);

  return;
}

static void populate_interval(corsaro_interval_t *interval, uint32_t number,
                              uint32_t time)
{
  interval->corsaro_magic = CORSARO_MAGIC;
  interval->magic = CORSARO_MAGIC_INTERVAL;
  interval->number = number;
  interval->time = time;
}

static corsaro_t *corsaro_init(char *template, corsaro_file_mode_t mode)
{
  corsaro_t *e;

  if((e = malloc_zero(sizeof(corsaro_t))) == NULL)
    {
      corsaro_log(__func__, NULL, "could not malloc corsaro_t");
      return NULL;
    }

  /* what time is it? */
  gettimeofday_wrap(&e->init_time);

  /* uridata doesn't *need* to be set */

  /* monitorname also doesn't need to be set - we have a default from config */

  /* template does, however */
  /* check that it is valid */
  if(corsaro_io_validate_template(e, template) == 0)
    {
      corsaro_log(__func__, e, "invalid template %s", template);
      goto err;
    }
  if((e->template = strdup(template)) == NULL)
    {
      corsaro_log(__func__, e, 
                "could not duplicate template string (no memory?)");
      goto err;
    }

  /* as does the mode */
  e->output_mode = mode;

  /* check if compression should be used based on the file name */
  e->compress = corsaro_file_detect_compression(e, e->template);

  /* use the default compression level for now */
  e->compress_level = CORSARO_FILE_COMPRESS_LEVEL_DEFAULT;

  /* we have the template and file details, now we can open the global file */
  if((e->global_file = corsaro_io_prepare_file(e, CORSARO_IO_GLOBAL_NAME)) == NULL)
    {
      corsaro_log(__func__, e, "could not open global output file");
      goto err;
    }

  /* initialize the log file */
  if(corsaro_log_init(e) != 0)
    {
      corsaro_log(__func__, e, "could not initialize log file");
      goto err;
    }

  /* lets get us a wrapper packet ready */
  if((e->packet = corsaro_packet_alloc(e)) == NULL)
    {
      corsaro_log(__func__, e, "could not create corsaro packet");
      goto err;
    }

  /* ask the plugin manager to get us some plugins */
  /* this will init all compiled plugins but not start them, this gives
     us a chance to wait for the user to choose a subset to enable
     with corsaro_enable_plugin and then we'll re-init */
  if((e->plugin_manager = corsaro_plugin_manager_init(e->logfile)) == NULL)
    {
      corsaro_log(__func__, e, "could not initialize plugin manager");
      goto err;
    }

  /* interval doesn't need to be actively set, use the default for now */
  e->interval = CORSARO_INTERVAL_DEFAULT;

  /* initialize the current interval */
  populate_interval(&e->interval_start, 0, 0);

  /* the rest are zero, as they should be. */
 
  /* ready to rock and roll! */

  return e;

 err:
  corsaro_free(e);
  return NULL;
}

static int start_interval(corsaro_t *corsaro, struct timeval int_start)
{
  corsaro_plugin_t *tmp = NULL;

  /* record this so we know when the interval started */
  /* the interval number is already incremented by end_interval */
  corsaro->interval_start.time = int_start.tv_sec;

  /* ask each plugin to start a new interval */
  while((tmp = corsaro_plugin_next(corsaro->plugin_manager, tmp)) != NULL)
    {
      if(tmp->start_interval(corsaro, &corsaro->interval_start) != 0)
        {
          corsaro_log(__func__, corsaro, "%s failed to start interval at %ld", 
                    tmp->name, int_start.tv_sec);
          return -1;
        }
    }
  return 0;
}


static int end_interval(corsaro_t *corsaro, struct timeval int_end)
{
  corsaro_plugin_t *tmp = NULL;

  corsaro_interval_t interval_end;

  populate_interval(&interval_end, corsaro->interval_start.number,
                    int_end.tv_sec);

  /* write the global interval start header */
  if(corsaro_io_write_interval_start(corsaro, corsaro->global_file,
                                   &corsaro->interval_start) <= 0)
    {
      corsaro_log(__func__, corsaro, 
                "could not write global interval start headers at %ld",
                corsaro->interval_start.time);
      return -1;
    }
  /* ask each plugin to end the current interval */
  while((tmp = corsaro_plugin_next(corsaro->plugin_manager, tmp)) != NULL)
    {
      if(tmp->end_interval(corsaro, &interval_end) != 0)
        {
          corsaro_log(__func__, corsaro, "%s failed to end interval at %ld",
                    tmp->name, int_end.tv_sec);
          return -1;
        }
    }
  /* write the global interval end header */
  if(corsaro_io_write_interval_end(corsaro, corsaro->global_file, 
                                 &interval_end) <= 0)
    {
      corsaro_log(__func__, corsaro, 
                "could not write global interval end headers at %ld",
                interval_end.time);
      return -1;
    }
  return 0;
}

static void corsaro_in_free(corsaro_in_t *corsaro)
{
  if(corsaro == NULL)
    {
      /* nothing to be done */
      
      corsaro_log_in(__func__, corsaro, 
                   "WARNING: corsaro_in_free called on NULL object; "
                   "this could indicate a double-free");
      return;
    }

  /* free the uridata */
  free(corsaro->uridata);
  corsaro->uridata = NULL;

  /* close the plugin */
  if(corsaro->plugin != NULL)
    {
      corsaro->plugin->close_input(corsaro);
    }
  /*
  while((p = corsaro_plugin_next(corsaro->plugin_manager, p)) != NULL)
    {
      p->close_input(corsaro);
    }
  */
  
  /* free the plugin manager */
  if(corsaro->plugin_manager != NULL)
    {
      corsaro_plugin_manager_free(corsaro->plugin_manager);
      corsaro->plugin_manager = NULL;
    }

  /* close the input file */
  if(corsaro->file !=  NULL)
    {
      corsaro_file_rclose(corsaro, corsaro->file);
      corsaro->file = NULL;
    }

  corsaro->started = 0;

  free(corsaro);
}

static corsaro_in_t *corsaro_in_init(char *corsarouri)
{
  corsaro_in_t *e;

  if((e = malloc_zero(sizeof(corsaro_in_t))) == NULL)
    {
      corsaro_log_in(__func__, NULL, "could not malloc corsaro_t");
      return NULL;
    }

  if((e->uridata = strdup(corsarouri)) == NULL)
    {
      corsaro_log_in(__func__, e, 
                "could not duplicate uri string (no memory?)");
      goto err;
    }

  /* set to null until we know if this is a global file or a plugin */
  e->expected_type = CORSARO_IN_RECORD_TYPE_NULL;
  
  /* initialize the logging */
  if(corsaro_log_in_init(e) != 0)
    {
      corsaro_log_in(__func__, e, "could not initialize log file");
      goto err;
    }

  /* ask the plugin manager to get us some plugins */
  if((e->plugin_manager = corsaro_plugin_manager_init(NULL)) == 0)
    {
      corsaro_log_in(__func__, e, "could not initialize plugins");
      goto err;
    }

  /* do not init plugins here, we will init only the one needed */

  /* delay opening the input file until we 'start' */

  return e;

 err:
  corsaro_in_free(e);
  return NULL;
}

static int process_packet(corsaro_t *corsaro, corsaro_packet_t *packet)
{
  corsaro_plugin_t *tmp = NULL;
  while((tmp = corsaro_plugin_next(corsaro->plugin_manager, tmp)) != NULL)
    {
      if(tmp->process_packet(corsaro, packet) < 0)
        {
          corsaro_log(__func__, corsaro, "%s failed to process packet", 
                    tmp->name);
          return -1;
        }
    }
  return 0;
}

static int check_global_filename(char *fname)
{
  if(strstr(fname, CORSARO_IO_GLOBAL_NAME) != NULL)
    {
      return 1;
    }
  return 0;
}

static int check_global_magic(corsaro_in_t *corsaro, corsaro_file_in_t *file)
{
  char buffer[1024];
  int len;

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

  /* an corsaro global file will have 'EDGRHEAD' as the first 8 bytes */
  if(len < 8)
    {
      return 0;
    }

  /* lets make this easy and just do a string compare */
  buffer[8] = '\0';

  if(strncmp(&buffer[0], "EDGRHEAD", 8) == 0)
    {
      return 1;
    }
  return 0;
}

static int is_plugin_data_or_interval(corsaro_in_t *corsaro)
{
  char buffer[1024];
  int len;

  /* we need to peek and see if there is any plugin data */
  len = corsaro_file_rpeek(corsaro, corsaro->file, buffer, 
                           sizeof(buffer));
  if(len < sizeof(corsaro_plugin_data_t) 
     && len < sizeof(corsaro_interval_t))
    {
      corsaro_log_in(__func__, corsaro, 
                     "invalid corsaro global file");
      return -1;
    }
          
  /* a plugin data record will have 'EDGRDATA' */
  /* an interval start record will have 'EDGRINTR' */
  /* either way, use strncmp */
  buffer[8] = '\0';
  if(strncmp(buffer, "EDGRDATA", 8) == 0)
    {
      return 1;
    }
  else if(strncmp(buffer, "EDGRINTR", 8) == 0)
    {
      return 0;

    }
  else
    {
      return -1;
    }
}

static int is_trailer_or_interval(corsaro_in_t *corsaro)
{
  char buffer[1024];
  int len;

  /* we need to peek and see if there is any plugin data */
  len = corsaro_file_rpeek(corsaro, corsaro->file, buffer, 
                           sizeof(buffer));
  if(len < sizeof(corsaro_trailer_t) 
     && len < sizeof(corsaro_interval_t))
    {
      corsaro_log_in(__func__, corsaro, 
                     "invalid corsaro global file");
      return -1;
    }
          
  /* a plugin data record will have 'EDGRFOOT' */
  /* an interval start record will have 'EDGRINTR' */
  /* either way, use strncmp */
  buffer[8] = '\0';
  if(strncmp(buffer, "EDGRFOOT", 8) == 0)
    {
      return 1;
    }
  else if(strncmp(buffer, "EDGRINTR", 8) == 0)
    {
      return 0;
    }
  else
    {
      return -1;
    }
}

static off_t read_record(corsaro_in_t *corsaro, 
                         corsaro_in_record_type_t *record_type, 
                         corsaro_in_record_t *record)
{
  off_t bytes_read = -1;
  int rc;

  /* this code is adapted from corsaro_flowtuple.c */
  switch(corsaro->expected_type)
    {
    case CORSARO_IN_RECORD_TYPE_IO_HEADER:
      /* ask the io subsystem to read it for us */
      bytes_read = corsaro_io_read_header(corsaro, corsaro->file, record_type,
                                          record);
      if(bytes_read > 0)
        {
          corsaro->expected_type = CORSARO_IN_RECORD_TYPE_IO_INTERVAL_START;
        }
      break;

    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))
        {
          rc = is_plugin_data_or_interval(corsaro);
          if(rc == 1)
            {
              corsaro->expected_type = CORSARO_IN_RECORD_TYPE_IO_PLUGIN_START;
            }
          else if(rc == 0)
            {
              corsaro->expected_type = CORSARO_IN_RECORD_TYPE_IO_INTERVAL_END;
            }
          else
            {
              *record_type = CORSARO_IN_RECORD_TYPE_NULL;
            }
        }
      break;

    case CORSARO_IN_RECORD_TYPE_IO_PLUGIN_START:
      /* ask the io subsystem to read it for us */
      bytes_read = corsaro_io_read_plugin_start(corsaro, corsaro->file, 
                                                record_type, record);
      if(bytes_read == sizeof(corsaro_plugin_data_t))
        {
          /* which plugin is this? */
          if((corsaro->plugin = corsaro_plugin_get_by_id(
                         corsaro->plugin_manager,
                         ((corsaro_plugin_data_t*)record->buffer)->plugin_id)
              )
             == NULL)
            {
              corsaro_log_in(__func__, corsaro, "invalid plugin id detected");
              corsaro_log_in(__func__, corsaro, "is corsaro built with all "
                             "necessary plugins?");
              *record_type = CORSARO_IN_RECORD_TYPE_NULL;
            }
          else
            {
              /* we'll pass these over to the plugin */
              corsaro->expected_type = CORSARO_IN_RECORD_TYPE_INTERNAL_REDIRECT; 
            }     
        }
      else
        {
          corsaro_log_in(__func__, corsaro, 
                         "failed to read plugin data start");
          *record_type = CORSARO_IN_RECORD_TYPE_NULL;
        }
      break;

    case CORSARO_IN_RECORD_TYPE_INTERNAL_REDIRECT:
      /* pass this over to the plugin */
      assert(corsaro->plugin != NULL);
      bytes_read = corsaro->plugin->read_global_data_record(corsaro, record_type, record);

      if(bytes_read >= 0)
        {
          corsaro->expected_type = CORSARO_IN_RECORD_TYPE_IO_PLUGIN_END;
        }
      break;

    case CORSARO_IN_RECORD_TYPE_IO_PLUGIN_END:
      /* ask the io subsystem to read it for us */
      bytes_read = corsaro_io_read_plugin_end(corsaro, corsaro->file,
                                              record_type, record);
      if(bytes_read == sizeof(corsaro_plugin_data_t))
        {
          /* check if there is more plugin data */
          rc = is_plugin_data_or_interval(corsaro);
          if(rc == 1)
            {
              corsaro->expected_type = CORSARO_IN_RECORD_TYPE_IO_PLUGIN_START;
            }
          else if(rc == 0)
            {
              corsaro->expected_type = CORSARO_IN_RECORD_TYPE_IO_INTERVAL_END;
            }
          else
            {
              *record_type = CORSARO_IN_RECORD_TYPE_NULL;
            }
        }
      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))
        {
          /* check for an interval start, or for a trailer */
          rc = is_trailer_or_interval(corsaro);
          if(rc == 0)
            {
              corsaro->expected_type = CORSARO_IN_RECORD_TYPE_IO_INTERVAL_START;
            }
          else if(rc == 1)
            {
              corsaro->expected_type = CORSARO_IN_RECORD_TYPE_IO_TRAILER;
            }
          else
            {
              *record_type = CORSARO_IN_RECORD_TYPE_NULL;
            }
        }
      break;

    case CORSARO_IN_RECORD_TYPE_IO_TRAILER:
      /* ask the io subsystem to read it for us */
      bytes_read = corsaro_io_read_trailer(corsaro, corsaro->file, record_type,
                                           record);
      if(bytes_read == sizeof(corsaro_trailer_t))
        {
          corsaro->expected_type = CORSARO_IN_RECORD_TYPE_IO_HEADER;
        }
      break;

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

  return bytes_read;
}

/* == PUBLIC CORSARO API FUNCS BELOW HERE == */

corsaro_t *corsaro_alloc_output(char *template, corsaro_file_mode_t mode)
{
  corsaro_t *corsaro;

  /* quick sanity check that folks aren't trying to write to stdout */
  if(template == NULL || strcmp(template, "-") == 0)
    {
      corsaro_log(__func__, NULL, "writing to stdout not supported");
      return NULL;
    }
  
  /* initialize the corsaro object */
  if((corsaro = corsaro_init(template, mode)) == NULL)
    {
      corsaro_log(__func__, NULL, "could not initialize corsaro object");
      return NULL;
    }

  return corsaro;
}

int corsaro_start_output(corsaro_t *corsaro)
{
  corsaro_plugin_t *p = NULL;

  assert(corsaro != NULL);

  /* ask the plugin manager to start up */
  /* this allows it to remove disabled plugins */
  if(corsaro_plugin_manager_start(corsaro->plugin_manager) != 0)
    {
      corsaro_log(__func__, corsaro, "could not start plugin manager");
      corsaro_free(corsaro);
      return -1;
    }

  /* now, ask each plugin to open its output file */
  /* we need to do this here so that the corsaro object is fully set up
     that is, the traceuri etc is set */
  while((p = corsaro_plugin_next(corsaro->plugin_manager, p)) != NULL)
    {
      if(p->init_output(corsaro) != 0)
        {
          corsaro_log(__func__, corsaro, "plugin could not init output");
          corsaro_free(corsaro);
          return -1;
        }
    }

  /* write headers to the global file */
  if(corsaro_io_write_header(corsaro, corsaro->global_file, NULL) <= 0)
    {
      corsaro_log(__func__, corsaro, "could not write global headers");
      corsaro_free(corsaro);
      return -1;
    }

  corsaro->started = 1;

  return 0;
}

void corsaro_set_interval(corsaro_t *corsaro, int i)
{
  assert(corsaro != NULL);
  
  /* you can't set the interval once corsaro has been started */
  assert(corsaro->started == 0);

  corsaro->interval = i;
}

int corsaro_set_traceuri(corsaro_t *corsaro, char *uri)
{
  assert(corsaro != NULL);

  if(corsaro->started != 0)
    {
      corsaro_log(__func__, corsaro, 
                "trace uri can only be set before "
                "corsaro_start_output is called");
      return -1;
    }

  if(corsaro->uridata != NULL)
    {
      corsaro_log(__func__, corsaro, "WARNING: updating trace uri from %s to %s",
                corsaro->uridata, uri);
    }
  if((corsaro->uridata = strdup(uri)) == NULL)
    {
      corsaro_log(__func__, corsaro, 
                "could not duplicate uri string (no memory?)");
      return -1;
    }
  corsaro_log(__func__, corsaro, "%s", corsaro->uridata);
  return 0;
}

int corsaro_enable_plugin(corsaro_t *corsaro, const char *plugin_name)
{
  assert(corsaro != NULL);
  assert(corsaro->plugin_manager != NULL);
  
  if(corsaro->started != 0)
    {
      corsaro_log(__func__, corsaro, 
                  "trace uri can only be set before "
                  "corsaro_start_output is called");
      return -1;
    }

  return corsaro_plugin_enable_plugin(corsaro->plugin_manager, plugin_name);
  
}

const char *corsaro_get_traceuri(corsaro_t *corsaro)
{
  return corsaro->uridata;
}

int corsaro_set_monitorname(corsaro_t *corsaro, char *name)
{
  assert(corsaro != NULL);

  if(corsaro->started != 0)
    {
      corsaro_log(__func__, corsaro, 
                "monitor name can only be set before "
                "corsaro_start_output is called");
      return -1;
    }

  if(corsaro->monitorname != NULL)
    {
      corsaro_log(__func__, corsaro, 
                  "WARNING: updating monitor name from %s to %s",
                  corsaro->monitorname, name);
    }

  if((corsaro->monitorname = strdup(name)) == NULL)
    {
      corsaro_log(__func__, corsaro, 
                "could not duplicate monitor name string (no memory?)");
      return -1;
    }
  corsaro_log(__func__, corsaro, "%s", corsaro->monitorname);
  return 0;
}

const char *corsaro_get_monitorname(corsaro_t *corsaro)
{
  if(corsaro->monitorname != NULL)
    {
      return corsaro->monitorname;
    }
  else
    {
      return STR(CORSARO_MONITOR_NAME);
    }
}

int corsaro_per_packet(corsaro_t *corsaro, libtrace_packet_t *ltpacket)
{
  struct timeval ts;
  struct timeval report;

  assert(corsaro != NULL);

  if(corsaro->started != 1)
    {
      corsaro_log(__func__, corsaro, "corsaro_start_output must be called before"
                "packets can be processed");
      return -1;
    }

  /* poke this ltpacket into our corsaro packet */
  corsaro->packet->ltpacket = ltpacket;

  /* ensure that the state is clear */
  corsaro_packet_state_reset(corsaro->packet);
  
  /* this is now the latest packet we have seen */
  corsaro->last_ts = ts = trace_get_timeval(ltpacket);

  /* if this is the first packet we record, keep the timestamp */
  if(corsaro->packet_cnt == 0)
    {
      corsaro->first_ts = ts;
      if(start_interval(corsaro, ts) != 0)
        {
          corsaro_log(__func__, corsaro, "could not start interval at %ld", 
                    ts.tv_sec);
          return -1;
        }
      corsaro->next_report = ts.tv_sec + corsaro->interval;
    }

  /* using an interval value of less than zero disables intervals
     such that only one distribution will be generated upon completion */
  while(corsaro->interval >= 0 && (uint32_t)ts.tv_sec >= corsaro->next_report)
    {
      /* we want to mark the end of the interval such that all pkt times are <=
         the time of the end of the interval. 
         because we deal in second granularity, we simply subtract one from the
         time */
      report.tv_sec = corsaro->next_report-1;
      
      if(end_interval(corsaro, report) != 0)
        {
          corsaro_log(__func__, corsaro, "could not end interval at %ld", 
                    ts.tv_sec);
          /* we don't free in case the client wants to try to carry on */
          return -1;
        }

      /* we now add the second back on to the time to get the start time */
      report.tv_sec = corsaro->next_report;
      if(start_interval(corsaro, report) != 0)
        {
          corsaro_log(__func__, corsaro, "could not start interval at %ld", 
                    ts.tv_sec);
          /* we don't free in case the client wants to try to carry on */
          return -1;
        }
      corsaro->interval_start.number++;
      corsaro->next_report += corsaro->interval;
    }

  /* count this packet for our overall packet count */
  corsaro->packet_cnt++;

  /* ask each plugin to process this packet */
  return process_packet(corsaro, corsaro->packet);
}

int corsaro_finalize_output(corsaro_t *corsaro)
{
  if(corsaro->started != 0)
    {
      if(end_interval(corsaro, corsaro->last_ts) != 0)
        {
          corsaro_log(__func__, corsaro, "could not end interval at %ld", 
                      corsaro->last_ts.tv_sec);
          corsaro_free(corsaro);
          return -1;
        }
      
      /* write headers to the global file */
      if(corsaro_io_write_trailer(corsaro, corsaro->global_file, NULL) <= 0)
        {
          corsaro_log(__func__, corsaro, "could not write global trailers");
          corsaro_free(corsaro);
          return -1;
        }
    }
  
  corsaro_free(corsaro);
  return 0;
}

/* ===== corsaro_in API functions ===== */

corsaro_in_t *corsaro_alloc_input(char *corsarouri)
{
  corsaro_in_t *corsaro;
  
  /* initialize the corsaro object */
  if((corsaro = corsaro_in_init(corsarouri)) == NULL)
    {
      corsaro_log_in(__func__, NULL, "could not initialize corsaro_in object");
      return NULL;
    }

  return corsaro;
}

int corsaro_start_input(corsaro_in_t *corsaro)
{
  corsaro_plugin_t *p = NULL;

  assert(corsaro != NULL);
  assert(corsaro->started == 0);
  assert(corsaro->plugin == NULL);

  /* open the file! */
  if((corsaro->file = corsaro_file_ropen(corsaro, corsaro->uridata)) == NULL)
    {
      corsaro_log_in(__func__, corsaro, "could not open input file %s", 
                   corsaro->uridata);
      /* ak comments the following, leave it up to the caller to free
         the state object */
      /*corsaro_in_free(corsaro);*/
      return -1;
    }

  /* determine the plugin which created this file */
  while((p = corsaro_plugin_next(corsaro->plugin_manager, p)) != NULL && 
        corsaro->plugin == NULL)
    {
      if(p->probe_filename(corsaro->uridata) == 1)
        {
          corsaro_log_in(__func__, corsaro, 
                       "%s plugin selected to read %s (using file name)", 
                       p->name, corsaro->uridata);
          corsaro->plugin = p;
        }
    }

  /* if the previous method for detection failed, lets try peeking into
     the file */
  p = NULL;
  while(corsaro->plugin == NULL && 
        (p = corsaro_plugin_next(corsaro->plugin_manager, p)) != NULL)
    {
      if(p->probe_magic(corsaro, corsaro->file) == 1)
        {
          corsaro_log_in(__func__, corsaro, 
                       "%s plugin selected to read %s (using magic)",
                       p->name, corsaro->uridata);
          corsaro->plugin = p;
        }
    }

  /* if corsaro->plugin is still NULL, see if this is the global output */
  if(corsaro->plugin == NULL)
    {
      if(check_global_filename(corsaro->uridata) != 1 && 
         check_global_magic(corsaro, corsaro->file) != 1)
        {
          /* we have no idea what this file was created by */
          corsaro_log_in(__func__, corsaro, "unable to find plugin to decode %s\n"
                       " - is this a corsaro file?\n"
                       " - is corsaro compiled with all needed plugins?", 
                       corsaro->uridata);
          return -1;
        }
      else
        {
          /* this the corsaro global output */
          /* we initially expect an corsaro header record in a global file */
          corsaro->expected_type = CORSARO_IN_RECORD_TYPE_IO_HEADER;
          corsaro_log_in(__func__, corsaro, "corsaro_global selected to read %s",
                         corsaro->uridata);
        }
    }
  else
    {
      /* start up the plugin we detected */
      if(corsaro->plugin->init_input(corsaro) != 0)
        {
          corsaro_log_in(__func__, corsaro, "could not initialize %s", 
                       corsaro->plugin->name);
          return -1;
        }
    }

  corsaro->started = 1;

  return 0;
}

corsaro_in_record_t *corsaro_in_alloc_record(corsaro_in_t *corsaro)
{
  corsaro_in_record_t *record = NULL;

  if((record = malloc(sizeof(corsaro_in_record_t))) == NULL)
    {
      corsaro_log_in(__func__, corsaro, "could not malloc corsaro_in_record_t");
      return NULL;
    }

  record->corsaro = corsaro;

  /* pre-allocate some memory for the buffer */
  if((record->buffer = malloc(sizeof(uint8_t)*
                              CORSARO_IN_RECORD_DEFAULT_BUFFER_LEN)) == NULL)
    {
      corsaro_log_in(__func__, corsaro, "could not malloc record buffer");
      corsaro_in_free_record(record);
      return NULL;
    }

  record->buffer_len = CORSARO_IN_RECORD_DEFAULT_BUFFER_LEN;

  record->type = -1;

  return record;
}

void corsaro_in_free_record(corsaro_in_record_t *record)
{
  if(record == NULL)
    {
      corsaro_log_file(__func__, NULL, "possible double free of record pointer");
      return;
    }

  if(record->buffer != NULL)
    {
      free(record->buffer);
      record->buffer = NULL;
    }
  record->buffer_len = -1;

  record->type = -1;

  free(record);
}

off_t corsaro_in_read_record(corsaro_in_t *corsaro,
                         corsaro_in_record_type_t *record_type,
                         corsaro_in_record_t *record)
{
  /* the first check makes sure we don't have a plugin that we have assigned
     the file to, the second ensures that if there is a plugin, we will
     only directly use it when we're not in global mode */
  if(corsaro->plugin != NULL 
     && corsaro->expected_type == CORSARO_IN_RECORD_TYPE_NULL)
    {
      return corsaro->plugin->read_record(corsaro, record_type, record);
    }
  else
    {
      /* this is the global plugin, handle it ourselves */
      return read_record(corsaro, record_type, record);
    }
  
  return -1;
}

void *corsaro_in_get_record_data(corsaro_in_t *corsaro, corsaro_in_record_t *record)
{
  return record->buffer;
}

int corsaro_finalize_input(corsaro_in_t *corsaro)
{
  corsaro_in_free(corsaro);
  return 0;
}