flow-cat.c

netflow,抓包

  /* preload headers */
  if (preload) {

    /* foreach list */
    for (i = 0; i < fte_entries; ++i) {

      /* foreach file in the list */
      FT_TAILQ_FOREACH(fty, &fte[i]->head, chain) {

        if (debug > 5)
          fterr_info("header load: file=%s", fty->name);

        /* stdin / real file? */
        if (fty->name[0]) {

          if ((in_fd = open(fty->name, O_RDONLY, 0)) == -1) {
            fterr_warn("open(%s)", fty->name);
            continue;
          }

        } else {

          continue;

        }

        /* initialize ftio stream */
        if (ftio_init(&ftio_in, in_fd, FT_IO_FLAG_READ |
          ((in_fd_plain && !disable_mmap) ? FT_IO_FLAG_MMAP : 0)) < 0)
          fterr_errx(1, "ftio_init(): failed");

        /* ensure required fields */
        if (ftio_check_xfield(&ftio_in,
          FT_XFIELD_UNIX_SECS|FT_XFIELD_UNIX_NSECS|FT_XFIELD_SYSUPTIME|
          FT_XFIELD_FIRST|FT_XFIELD_LAST))
          fterr_errx(1, "Flow record missing required field.");

        /* keep a record of all the possible fields */
        fields |= ftio_in.fth.fields;

        /* record smallest time */
        if (ftio_in.fth.fields & FT_FIELD_CAP_START) {
          time_tmp1 = ftio_get_cap_start(&ftio_in);
          if (time_tmp1 < time_start)
            time_start = time_tmp1;
        }

        /* record largest time */
        if (ftio_in.fth.fields & FT_FIELD_CAP_END) {
          time_tmp2 = ftio_get_cap_end(&ftio_in);
          if (time_tmp2 > time_end)
            time_end = time_tmp2;
        }

        /*
         * if coarse grained time filtering is enabled -- ie only filtering
         * files based on header time, ensure this file fits in the window
         */
        if (time_filter) {

          /* all flows between low and high */
          if (time_high && time_low) {

            time_delta = time_tmp2 - time_tmp1;

            if ((time_tmp1 < time_low) ||
                ((time_tmp2-time_delta) > time_high)) {
              fty->skip = 1;
              goto skip_file;
            }


          } else { /* between */

            /* all flows until low, ie -inf to low */
            if (time_low && (time_tmp1 > time_low)) {
              fty->skip = 1;
              goto skip_file;
            }

            /* all flows after high, ie high to +inf */
            if (time_high && (time_tmp2 < time_high)) {
              fty->skip = 1;
              goto skip_file;
            }

          } /* outside */

        } /* time_filter */

        /* total lost flows */
        lost_flows += ftio_get_lost(&ftio_in);

        /* total corrupt flows */
        corrupt_flows += ftio_get_corrupt(&ftio_in);

        /* total flows */
        total_flows += ftio_get_flows_count(&ftio_in);

skip_file:
        if (debug > 5)
          fterr_info("file=%s, status=%d", fty->name, fty->skip);

        ftio_close(&ftio_in);
      }
    }

    ftio_set_preloaded(&ftio_out, 1);
    if ((fields & (FT_FIELD_CAP_END|FT_FIELD_CAP_START)) == (
      FT_FIELD_CAP_START|FT_FIELD_CAP_END))
      ftio_set_cap_time(&ftio_out, time_start, time_end);
    ftio_set_flows_count(&ftio_out, total_flows);
    ftio_set_corrupt(&ftio_out, corrupt_flows);
    ftio_set_lost(&ftio_out, lost_flows);

    time_start = -1; /* MAXINT */
    time_end = 0;
    total_flows = 0;

  } /* preload */


  /* foreach list */
  for (i = 0; i < fte_entries; ++i) {

    /* foreach file in the list */
    FT_TAILQ_FOREACH(fty, &fte[i]->head, chain) {

      if (fty->skip)
        goto next_file;

      if (debug > 5)
        fterr_info("working file=%s", fty->name);

      /* stdin / real file? */
      if (fty->name[0]) {

        in_fd_plain = 1;

        if ((in_fd = open(fty->name, O_RDONLY, 0)) == -1) {
          fterr_warn("open(%s)", fty->name);
          continue;
        }

      } else {

        in_fd_plain = 0;
        in_fd = 0; /* stdin */

      }

      /* initialize ftio stream */
      if (ftio_init(&ftio_in, in_fd, FT_IO_FLAG_READ |
        ((in_fd_plain && !disable_mmap) ? FT_IO_FLAG_MMAP : 0)) < 0)
        fterr_errx(1, "ftio_init(): failed");

      /* get version from stream */
      ftio_get_ver(&ftio_in, &ftv2);

      /* successful process of a stream */
      ++ total_streams;

      /* record smallest time */
      time_tmp1 = ftio_get_cap_start(&ftio_in);
      if (time_tmp1 < time_start)
        time_start = time_tmp1;

      /* record largest time */
      time_tmp2 = ftio_get_cap_end(&ftio_in);
      if (time_tmp2 > time_end)
        time_end = time_tmp2;

      /* first time through loop? */
      if (!ftv.d_version) {

        /*
         * is this really the right thing to do here.  Reading a v1
         * stream gets handled by ftio_read(), but ftio_* leaves the
         * s_version at 1.
         */
        ftv2.s_version = FT_IO_SVERSION;

        /* set the version information in the io stream */
        if (ftio_set_ver(&ftio_out, &ftv2) < 0)
          fterr_errx(1, "ftio_set_ver(): failed");

        /* save for later compare */
        bcopy(&ftv2, &ftv, sizeof ftv);

        /* header first */
        if ((n = ftio_write_header(&ftio_out)) < 0)
          fterr_errx(1, "ftio_write_header(): failed");

        total_bytes += n;

        if (debug > 6)
          if (n)
            fterr_info("ftio_write_header()=%d", n);

      } else {

        /* ensure previous version == current version */
        if ((ftv.d_version != ftv2.d_version) ||
            (ftv.agg_method != ftv2.agg_method))
          fterr_errx(1, "data version or sub version changed!");

      }

      /* foreach flow record, copy it */
      while ((rec = ftio_read(&ftio_in))) {

        ++total_flows;

        if ((n = ftio_write(&ftio_out, rec)) < 0)
          fterr_errx(1, "ftio_write(): failed");

        total_bytes += n;

        if (debug > 6)
          if (n)
            fterr_info("ftio_write()=%d", n);

        /* interrupted? */
        if (done)
          break;

      } /* while copying */

      /* done with input stream */
      if (ftio_close(&ftio_in) < 0)
        fterr_errx(1, "ftio_close(): failed");

      /* interrupted? */
      if (done)
        break;

next_file:

    }  /* FOREACH filename in dir */

  } /* foreach dir bundle */

  /*
   * if the output file descriptor was a real file, re-write the
   * flow_header with the correct # of total flows
   */
  if (out_fd_plain) {

    ftio_set_cap_time(&ftio_out, time_start, time_end);
    ftio_set_flows_count(&ftio_out, total_flows);
    ftio_set_streaming(&ftio_out, 0);

    if ((n = ftio_write_header(&ftio_out)) < 0)
      fterr_errx(1, "ftio_write_header(): failed");

    total_bytes += n;

    if (debug > 6)
      if (n)
        fterr_info("ftio_write_header()=%d", n);

  } /* out_fd_plain */

  /* done with output stream */
  if ((n = ftio_close(&ftio_out)) < 0)
    fterr_errx(1, "ftio_close(): failed");

  total_bytes += n;

  if (debug > 6)
    if (n)
      fterr_info("ftio_close(ftio_out)=%d\n", n);

  if (debug > 0) {
    ftprof_end (&ftp, total_flows);
    ftprof_print(&ftp, argv[0], stderr);
  }

  if (debug > 1)
    fterr_info("Bytes written=%lu", total_bytes);

  /* free storage allocated to file list(s) */
  if (fte_entries) {
    for (i = 0; i < fte_entries; ++i) {
      ftfile_free(fte[i]);
      free(fte[i]);
    }
    free(fte);
  } /* fte_entries */

  /* no successful streams, then error */
  if (!total_streams) 
    return 1;
  else
    return 0;

} /* main */

void sig_quit(int sig)
{
  done = 1;
} /* sig_quit */

void usage(void) {

  fprintf(stderr, "Usage: flow-cat [-aghmp] [-b byte_order] [-C comment] [-d debug_level]\n");
  fprintf(stderr, "       [-o filename] [-t start_time] [-T end_time] [-z z_level]\n");
  fprintf(stderr, "       file|directory ...");
  fprintf(stderr, "\n");
  fprintf(stderr, "\n%s version %s: built by %s\n", PACKAGE, VERSION, FT_PROG_BUILD);


} /* usage */