ftio.c

netflow,抓包

 * Get the version from a ftio stream
 */
void ftio_get_ver(struct ftio *ftio, struct ftver *ver)
{
  ver->d_version = ftio->fth.d_version;
  ver->s_version = ftio->fth.s_version;
  ver->agg_method = ftio->fth.agg_method;
  ver->agg_version = ftio->fth.agg_version;
}

/*
 * function: ftio_get_stime
 *
 * Get the starting time from a ftio stream
 */
u_int32 ftio_get_cap_start(struct ftio *ftio)
{
  return ftio->fth.cap_start;
}

/*
 * function: ftio_get_etime
 *
 * Get the ending time from a ftio stream
 */
u_int32 ftio_get_cap_end(struct ftio *ftio)
{
  return ftio->fth.cap_end;
}

/*
 * function: ftio_get_rec_total
 *
 * Get the total records processed from a ftio stream
 */
u_int64 ftio_get_rec_total(struct ftio *ftio)
{
  return ftio->rec_total;
}

/*
 * function: ftio_get_flows_count
 *
 * Get the total records processed from a ftio stream
 */
u_int32 ftio_get_flows_count(struct ftio *ftio)
{
  return ftio->fth.flows_count;
}


/*
 * function: ftio_get_debug
 *
 * Get the debug level from a ftio stream
 */
int ftio_get_debug(struct ftio *ftio)
{
  return ftio->debug;
}

/*
 * function: ftio_get_cap_hostname
 *
 * Get the header hostname from a ftio stream
 */
char *ftio_get_hostname(struct ftio *ftio)
{
  return ftio->fth.cap_hostname;
}

/*
 * function: ftio_get_comment
 *
 * Get the header comment from a ftio stream
 */
char *ftio_get_comment(struct ftio *ftio)
{
  return ftio->fth.comments;
}

/*
 * function ftio_get_corrupt
 *
 * Get header corrupt flows from a ftio stream
*/
u_int32 ftio_get_corrupt(struct ftio *ftio)
{
  return ftio->fth.pkts_corrupt;
}

/*
 * function ftio_get_lost
 *
 * Get header lost flows from a ftio stream
*/
u_int32 ftio_get_lost(struct ftio *ftio)
{
  return ftio->fth.flows_lost;
}

/*
 * function: ftio_close
 *
 * Free resources allocated with ftio_init()
 * Flush any non empty buffers if stream was initialized for WRITE
 * close file descriptor
 *
 * returns: <0   error
 *          >= 0 okay
 */
int ftio_close(struct ftio *ftio)
{
  int ret, err, n, nbytes;

  ret = -1;
  nbytes = 0;

  if (ftio->fth.fields & FT_FIELD_COMMENTS)
    free(ftio->fth.comments);

  if (ftio->fth.fields & FT_FIELD_CAP_HOSTNAME)
    free(ftio->fth.cap_hostname);

  if (ftio->fth.ftmap)
    ftmap_free(ftio->fth.ftmap);

  if (ftio->flags & FT_IO_FLAG_READ) {

    if (ftio->flags & FT_IO_FLAG_ZINIT)
      inflateEnd(&ftio->zs);

    if (ftio->z_buf)
      free (ftio->z_buf);

    if (ftio->d_buf)
      free (ftio->d_buf);

#if HAVE_MMAP

    if (ftio->mr)
      munmap(ftio->mr, (size_t)ftio->mr_size);

#endif /* HAVE_MMAP */

  } else if (ftio->flags & FT_IO_FLAG_WRITE) {

    /* compression enabled? */
    if (ftio->flags & FT_IO_FLAG_ZINIT) {

      ftio->zs.avail_in = 0;
        
      while (1) {

        err = deflate(&ftio->zs, Z_FINISH);

        /* if done compressing, do final write to disk */
        if (err == Z_STREAM_END)
          break;
    
        /* if anything other than Z_OK, then it's an error */
        if (err != Z_OK) {
          fterr_warnx("deflate(): failed");
          goto ftio_close_out;
        }   
    
        /* need to flush */
        if (!ftio->zs.avail_out) {

          n = writen(ftio->fd, ftio->z_buf, FT_Z_BUFSIZE);

          if (n < 0) {
            fterr_warn("writen()");
            goto ftio_close_out;
          }
  
          if (n == 0) {
            fterr_warnx("writen(): EOF");
            goto ftio_close_out;
          }

          nbytes += n;

          ftio->zs.next_out = (Bytef*)ftio->z_buf;
          ftio->zs.avail_out = FT_Z_BUFSIZE;

        } else
          break;
      } /* while 1 */

      n = writen(ftio->fd, ftio->z_buf, FT_Z_BUFSIZE-ftio->zs.avail_out);

      if (n < 0) {
        fterr_warn("writen()");
        goto ftio_close_out;
      }
  
      if (n == 0) {
        fterr_warnx("writen(): EOF");
        goto ftio_close_out;
      }

      nbytes += n;

      ret = 0;

    /* no compression */
    } else {

      if (ftio->d_start) {

        n = writen(ftio->fd, ftio->d_buf, ftio->d_start);

        if (n < 0) {
          fterr_warn("writen()");
          goto ftio_close_out;
        }
  
        if (n == 0) {
          fterr_warnx("writen(): EOF");
          goto ftio_close_out;
        }

        ftio->d_start = 0;

        nbytes += n;

        ret = 0;

      } /* buffer not empty */

      ret = 0;

    } /* compression */

  } /* io stream enabled for write */


ftio_close_out:

  if (ftio->flags & FT_IO_FLAG_WRITE) {

    if (ftio->flags & FT_IO_FLAG_ZINIT) {

      deflateEnd(&ftio->zs);
      ftio->flags &= ~FT_IO_FLAG_ZINIT;
      free(ftio->z_buf);

    } else {

      free (ftio->d_buf);

    }

  } /* FT_IO_FLAG_WRITE */

  /* don't lose error condition if close() is a success */
  if (ret < 0)
    ret = close(ftio->fd);
  else
    close(ftio->fd);

  /* no error and writing? then return bytes written */
  if ((ftio->flags & FT_IO_FLAG_WRITE) && (ret >= 0))
    ret = nbytes;

  return ret;
  
} /* ftio_close */

/*
 * function: ftio_zstat_print
 *
 * Print resources utilization associated with ftio zlib usage
 *
 * returns: <0   error
 *          >= 0 okay
 */
void ftio_zstat_print(struct ftio *ftio, FILE *std)
{

  double d;

  d = ((double)ftio->zs.total_out) / ((double)ftio->zs.total_in);
  fprintf(std, "compression: total_in=%lu total_out=%lu  %3.3f:1\n",
    ftio->zs.total_in, ftio->zs.total_out, d);

}

/*
 * function: ftio_read
 *
 * Return the next fts3rec_* in the ftio stream, or 0L for EOF
 *
 * Record is returned in host byte order
 *
 * Stream must be first initialized with ftio_init() 
 *
 */
void *ftio_read(struct ftio *ftio)
{

  int n, err;
  void *ret;
  struct fts1rec_compat *compat;
  u_int32 bleft, boff;

  ret = (void*)0L;

#if HAVE_MMAP
  /* mmap enabled? */
  if (ftio->flags & FT_IO_FLAG_MMAP) {

    /* compressed ? */
    if (ftio->fth.flags & FT_HEADER_FLAG_COMPRESS) {

      /* EOF? */
      if (!ftio->zs.avail_in)
        goto ftio_read_out;

      err = inflate(&ftio->zs, Z_SYNC_FLUSH);

      if ((err != Z_OK) && (err != Z_STREAM_END)) {
        fterr_warnx("inflate(): failed");
        goto ftio_read_out;
      }

      /* if avail_out == 0, then a full record was inflated -- return it */
      if (!ftio->zs.avail_out) {

        /* XXX check for interrupt record */

        /* reset zlib for next call */
        ftio->zs.avail_out = ftio->rec_size;
        ftio->zs.next_out = (Bytef*)ftio->d_buf;

        ret = (void*)ftio->d_buf;
        goto ftio_read_out;

      } else {

        /* should never happen - partial decode */
        if (ftio->zs.avail_out != ftio->rec_size)
          fterr_warnx("Warning, partial inflated record before EOF");

        /* signal EOF to caller */
        goto ftio_read_out;

      } /* ftio->zs.avail_out */

    /* not compressed */
    } else {

      /* bytes left */
      bleft = ftio->d_end - ftio->d_start;

      /* enough bytes in d_buf to return a record? */
      if (bleft >= ftio->rec_size) {
        boff = ftio->d_start;
        ftio->d_start += ftio->rec_size;

        ret = (char*)ftio->mr+boff;
        goto ftio_read_out;
      }

      /* signal EOF? */
      if (!bleft)
        goto ftio_read_out;

      /* shouldn't happen */
      fterr_warnx("Warning, partial record before EOF");
      goto ftio_read_out;

    } /* not compressed and mmap */

  } /* mmap */

#endif /* HAVE_MMAP */

  /* processed compressed stream */
  if (ftio->fth.flags & FT_HEADER_FLAG_COMPRESS) {

    while (1) {

      /*
       * if the inflate buffer is empty, perform a read()
      */

      if (!ftio->zs.avail_in) {

        n = read(ftio->fd, (char*)ftio->z_buf, FT_Z_BUFSIZE);

        /* EOF and inflate buffer is empty -- done. */
        if (!n) {

          /*
           * check for partial record inflated.  This would never
           * happen on a uncorrupted stream
           */
          if (ftio->zs.avail_out != ftio->rec_size)
            fterr_warnx("Warning, partial inflated record before EOF");

          /* signal EOF to caller */
          goto ftio_read_out;

        }

        /* read error -- done. */
        if (n == -1) {
          fterr_warn("read()");
          goto ftio_read_out;
        }

        ftio->zs.avail_in = n;
        ftio->zs.next_in = (Bytef*)ftio->z_buf;

      } /* if inflate buffer empty */

      /*
       * inflate stream, attempt to get a record
       */

      err = inflate(&ftio->zs, Z_SYNC_FLUSH);

      if ((err != Z_OK) && (err != Z_STREAM_END)) {
        fterr_warnx("inflate(): failed");
        goto ftio_read_out;
      }

      /* if avail_out == 0, then a full record was inflated -- return it */
      if (!ftio->zs.avail_out) {

        /* XXX check for interrupt record */

        /* reset zlib for next call */
        ftio->zs.avail_out = ftio->rec_size;
        ftio->zs.next_out = (Bytef*)ftio->d_buf;

        ret = (void*)ftio->d_buf;
        goto ftio_read_out;

      } /* bytes available for inflate */
    } /* while 1 */
  } /* compressed stream */

  /*
   * uncompressed stream 
   */

  /*
   * if there are not enough bytes between the start pointer and the end
   * of d_buf for a full record, perform a read()
   */

  /* while(1) loop to optimize for normal case of returning bytes from d_buf */
  while (1) {

    bleft = ftio->d_end - ftio->d_start;

    /* enough bytes in d_buf to return a record? */
    if (bleft >= ftio->rec_size) {

      boff = ftio->d_start;
      ftio->d_start += ftio->rec_size;

      /* XXX check for interrupt record */
      /* ftio_interrupt(ftio); */
      /* continue */

      ret = (char*)ftio->d_buf+boff;
      goto ftio_read_out;

    /* no, perform a read() to try for more */
    } else {

      /* move trailing partial record to top of buffer */
      if (bleft)
        bcopy(ftio->d_buf+ftio->d_start, ftio->d_buf, bleft);

      ftio->d_end = bleft;
      ftio->d_start = 0;

      n = read(ftio->fd, (char*)ftio->d_buf+ftio->d_end,
        FT_D_BUFSIZE - ftio->d_end);

      /* read failed? */
      if (n < 0) {
        fterr_warn("read()");
        goto ftio_read_out;
      }

      /* eof? */
      if (n == 0) {
        if (ftio->d_start)
          fterr_warnx("Warning, partial record before EOF");
        goto ftio_read_out;
      }

      ftio->d_end += n;

    } /* need a read() */
  } /* while 1 */

ftio_read_out:

  if (ret) {

      /* fix byte ordering */
#if BYTE_ORDER == BIG_ENDIAN
      if (ftio->fth.byte_order == FT_HEADER_LITTLE_ENDIAN)
        ftio->swapf((void*)ret);
#endif /* BYTE_ORDER == BIG_ENDIAN */
  
#if BYTE_ORDER == LITTLE_ENDIAN
      if (ftio->fth.byte_order == FT_HEADER_BIG_ENDIAN)
        ftio->swapf((void*)ret);
#endif /* BYTE_ORDER == LITTLE_ENDIAN */

    /* increment total records processed */
    ftio->rec_total ++;

    /*
     * backwards compatability hack.  Map the stream version 1 into
     * a stream version 2 
     */
    if (ftio->fth.s_version == 1) {

      if (ftio->fth.d_version == 1) {

        compat = ret;

        ftio->compat_v1.sysUpTime = 0;
        ftio->compat_v1.unix_secs = compat->unix_secs;
        ftio->compat_v1.unix_nsecs = compat->unix_msecs * 1000000;
        ftio->compat_v1.srcaddr = compat->srcaddr;
        ftio->compat_v1.dstaddr = compat->dstaddr;
        ftio->compat_v1.nexthop = compat->nexthop;
        ftio->compat_v1.input = compat->input;
        ftio->compat_v1.output = compat->output;
        ftio->compat_v1.dPkts = compat->dPkts;
        ftio->compat_v1.dOctets = compat->dOctets;
        ftio->compat_v1.Last = compat->Last;
        ftio->compat_v1.First = compat->First;
        ftio->compat_v1.srcport = compat->srcport;
        ftio->compat_v1.dstport = compat->dstport;
        ftio->compat_v1.prot = compat->prot;
        ftio->compat_v1.tos = compat->tos;
        ftio->compat_v1.tcp_flags = compat->flags;

        ret = (void*)&ftio->compat_v1;

      } else if (ftio->fth.d_version == 5) {

        compat = ret;

        ftio->compat_v5.sysUpTime = 0;
        ftio->compat_v5.unix_secs = compat->unix_secs;
        ftio->compat_v5.unix_nsecs = compat->unix_msecs * 1000000;
        ftio->compat_v5.srcaddr = compat->srcaddr;
        ftio->compat_v5.dstaddr = compat->dstaddr;
        ftio->compat_v5.nexthop = compat->nexthop;