flow-export.c

netflow,抓包

  long thiszone;
  char buf[1024];
  char *rec;

  if (ftio_check_xfield(ftio, FT_XFIELD_TOS | FT_XFIELD_PROT | 
    FT_XFIELD_SRCADDR | FT_XFIELD_DSTADDR | FT_XFIELD_SRCPORT |
    FT_XFIELD_DSTPORT)) {
    fterr_warnx("Flow record missing required field for format.");
    return -1;
  }

  ftio_get_ver(ftio, &ftv);
 
  fts3rec_compute_offsets(&fo, &ftv);

  if (gettimeofday(&now, &tz) < 0) {
    fterr_warnx("gettimeofday() failed");
    return -1;
  }

  bzero(&pfh, sizeof pfh);
  bzero(&pph, sizeof pph);
  bzero(&pd1, sizeof pd1);
  bzero(&pd2, sizeof pd2);
  bzero(&pd3, sizeof pd3);
  bzero(&pd4, sizeof pd4);
  bsize = 0;

  thiszone = tz.tz_minuteswest * -60;

  if (localtime((time_t *)&now.tv_sec)->tm_isdst)
    thiszone += 3600;

  pfh.magic = TCPDUMP_MAGIC;  
  pfh.version_major = TCPDUMP_VERSION_MAJOR;
  pfh.version_minor = TCPDUMP_VERSION_MINOR;
  pfh.thiszone = thiszone;
  pfh.sigfigs = 6;
  pfh.snaplen = 38; /* XXX TODO */
  pfh.linktype = 1;

  if (fwrite(&pfh, sizeof pfh, 1, stdout) != 1) {
    fterr_warnx("pcap header write failed");
    return -1;
  }

  pph.len = 58;
  pph.caplen = 58;

  pd1.eth_prot = 0x0008;
  pd2.version = 0x45;

  bcopy(&pph, buf, sizeof pph);
  bsize += sizeof pph;

  bcopy(&pd1, buf+bsize, sizeof pd1);
  bsize += sizeof pd1;

  while ((rec = ftio_read(ftio))) {

    cur.srcport = ((u_int16*)(rec+fo.srcport));
    cur.dstport = ((u_int16*)(rec+fo.dstport));
    cur.prot = ((u_int8*)(rec+fo.prot));
    cur.tos = ((u_int8*)(rec+fo.tos));
    cur.srcaddr = ((u_int32*)(rec+fo.srcaddr));
    cur.dstaddr = ((u_int32*)(rec+fo.dstaddr));


    pd2.tos = *cur.tos;
    pd2.prot = *cur.prot;
    pd2.srcaddr = *cur.srcaddr;
    pd2.dstaddr = *cur.dstaddr;

#if BYTE_ORDER == LITTLE_ENDIAN
    SWAPINT32(pd2.srcaddr);
    SWAPINT32(pd2.dstaddr);
#endif /* LITTLE_ENDIAN */

    good = 1;

    switch (pd2.prot) {

    case 6:

      pd3.srcport = *cur.srcport;
      pd3.dstport = *cur.dstport;

#if BYTE_ORDER == LITTLE_ENDIAN
      SWAPINT16(pd3.srcport);
      SWAPINT16(pd3.dstport);
#endif /* LITTLE_ENDIAN */

      bcopy(&pd2, buf+bsize, sizeof pd2);
      bcopy(&pd3, buf+bsize+sizeof pd2, sizeof pd3);
      bsize2 = bsize + sizeof pd2 + sizeof pd3;

      break;

    case 17:

      pd4.srcport = *cur.srcport;
      pd4.dstport = *cur.dstport;

#if BYTE_ORDER == LITTLE_ENDIAN
      SWAPINT16(pd4.srcport);
      SWAPINT16(pd4.dstport);
#endif /* LITTLE_ENDIAN */

      bcopy(&pd2, buf+bsize, sizeof pd2);
      bcopy(&pd4, buf+bsize+sizeof pd2, sizeof pd4);
      bsize2 = bsize + sizeof pd2 + sizeof pd4;

      break;

    default:
      good = 0;
      break;

    } /* switch */

    if (good) {
      if (fwrite(&buf, bsize2, 1, stdout) != 1) {
        fterr_warnx("pcap pkt write failed");
        return -1;
      }
    }

    ++opt->records;

  } /* while */

  return 0;
  
} /* format1 */

/*
 * function: format2
 *
 * export flows in ASCII CSV Format
 */
int format2(struct ftio *ftio, struct options *opt)
{
  struct fts3rec_offsets fo;
  struct ftver ftv;
  char fmt_buf[1024];
  char *rec;
  int len;

  ftio_get_ver(ftio, &ftv);

  /* remove invalid fields */
  opt->ft_mask &= ftrec_xfield(&ftv);

  fts3rec_compute_offsets(&fo, &ftv);

  fmt_xfields_type(fmt_buf, opt->ft_mask);

  printf("#:%s\n", fmt_buf);

  while ((rec = ftio_read(ftio))) {

    len = fmt_xfields_val(fmt_buf, rec, &fo, opt->ft_mask, 0);

    if (len)
      printf("%s\n", fmt_buf);

    ++opt->records;

  } /* while */

  return 0;
 
} /* format2 */ 

/*
 * function: format3
 *
 * export flows into MySQL Database
 */
int format3(struct ftio *ftio, struct options *opt)
{
#ifdef HAVE_MYSQL
  struct fts3rec_offsets fo;
  struct ftver ftv;
  char fields[1024], values[1024], query[3*1024];
  char *rec;
  char *db_host, *db_name, *db_table, *db_user, *db_pwd, *db_tmp, *tmp;
  int db_port;
  int len;

  MYSQL mysql;

  db_host = DB_DEFAULT_DBHOST;
  db_name = DB_DEFAULT_DBNAME;
  db_port = DB_DEFAULT_DBPORT;
  db_user = DB_DEFAULT_DBUSER;
  db_table = DB_DEFAULT_DBTABLE;
  db_pwd = DB_DEFAULT_DBPWD;

  /* parse URI string */

  if (strlen(opt->dbaseURI)) {

    tmp = opt->dbaseURI;

    db_user = strsep(&tmp, ":");
    db_pwd = strsep(&tmp, ":");
    db_host = strsep(&tmp, ":");
    db_tmp = strsep(&tmp, ":");
    db_name = strsep(&tmp, ":");
    db_table = strsep(&tmp, ":");
    db_port = atoi(db_tmp);

    if (!db_user || !db_pwd || !db_host || !db_tmp || !db_name || !db_table) {
      fterr_warnx("Missing field in dbaseURI, expecting user:pwd:host:port:name:table.");
      return -1;
    }

  } /* dbaseURI */

  ftio_get_ver(ftio, &ftv);

  fts3rec_compute_offsets(&fo, &ftv);

  /* remove invalid fields */
  opt->ft_mask &= ftrec_xfield(&ftv);

  /* generate the field names once */
  fmt_xfields_type(fields, opt->ft_mask);

  /* open MySQL database */
  if (!(mysql_init(&mysql)))
    fterr_errx(1, "mysql_init(): failed");

  if (mysql_options(&mysql, MYSQL_READ_DEFAULT_GROUP, "simple"))
    fterr_errx(1, "mysql_options(): %s", mysql_error(&mysql));

  if (mysql_real_connect(&mysql, db_host, db_user, db_pwd, 
	db_name, db_port, NULL, 0) == NULL) 
    fterr_errx(1,"mysql_real_connect(): %s\n", mysql_error(&mysql));

  /* foreach flow */
  while ((rec = ftio_read(ftio))) {

    len = fmt_xfields_val(values, rec, &fo, opt->ft_mask, 1);

    /* form SQL query and execute it */
    if (len) {
      strcpy (query, "INSERT INTO ");
      strcat (query, db_table);
      strcat (query, "(");
      strcat (query, fields);
      strcat (query, ") VALUES (");
      strcat (query, values);
      strcat (query, ")");
printf("field=%s\n val=%s\n query=%s\n", fields, values, query);

      if (mysql_real_query(&mysql, query, strlen(query)) != 0) 
        fterr_warnx("mysql_real_query(): %s", mysql_error(&mysql));

    }

    ++opt->records;

  } /* while */

  /* close database */
  mysql_close(&mysql);

#else /* MYSQL */

  fterr_warnx("Format not supported");

#endif /* MYSQL */

  return 0;
 
} /* format3 */ 

/*
 * function: format4
 *
 * export flows in wire format
*/
int format4(struct ftio *ftio, struct options *opt)
{
  struct ftver ftv;
  struct ftencode fte;
  char *rec;
  int ret;

  /* initialize encode struct */
  ftencode_init(&fte, 0);

  /* copy version from io stream */
  ftio_get_ver(ftio, &ftv);

  bcopy(&ftv, &fte.ver, sizeof ftv);

  /* foreach flow */
  while ((rec = ftio_read(ftio))) {

retry:

    ret = fts3rec_pdu_encode(&fte, rec);
/*   ret == 0 then send and clear out buffer
 *   ret > 0 then encode another
 *   ret < 0 then this encoding failed, send and clear out buffer
*/
  
    if (ret <= 0) { 
    
      /* convert pdu to network byte order */
#if BYTE_ORDER == LITTLE_ENDIAN
      ftpdu_swap(fte.buf_enc, BYTE_ORDER);
#endif /* BYTE_ORDER == LITTLE_ENDIAN */

      if (fwrite(&fte.buf, fte.buf_size, 1, stdout) != 1)
        fterr_err(1, "fwrite()");

      /* reset encode buffer */
      ftencode_reset(&fte);
 
      /* if ret < 0 then the current record was not encoded */   
      if (ret < 0)
        goto retry;
    }

    ++opt->records;

  }

  /* any left over? */
  if (fte.buf_size) {

    /* convert pdu to network byte order */
    ftpdu_swap(fte.buf_enc, BYTE_ORDER);

    if (fwrite(&fte.buf, fte.buf_size, 1, stdout) != 1)
      fterr_err(1, "fwrite()");

  } /* fte.buf_size */

  return 0;

} /* format4 */

int fmt_xfields_type(char *buf, u_int64 xfield)
{
  int comma;

  buf[0] = 0;

  if (xfield & FT_XFIELD_UNIX_SECS) {
    strcat(buf, FT_XFIELD_ASC_UNIX_SECS);
    comma = 1;
  }


  if (xfield & FT_XFIELD_UNIX_NSECS) {
    if (comma) strcat(buf, ",");
    strcat(buf, FT_XFIELD_ASC_UNIX_NSECS);
    comma = 1;
  }

  if (xfield & FT_XFIELD_SYSUPTIME) {
    if (comma) strcat(buf, ",");
    strcat(buf, FT_XFIELD_ASC_SYSUPTIME);
    comma = 1;
  }

  if (xfield & FT_XFIELD_EXADDR) {
    if (comma) strcat(buf, ",");
    strcat(buf, FT_XFIELD_ASC_EXADDR);
    comma = 1;
  }

  if (xfield & FT_XFIELD_DFLOWS) {
    if (comma) strcat(buf, ",");
    strcat(buf, FT_XFIELD_ASC_DFLOWS);
    comma = 1;
  }

  if (xfield & FT_XFIELD_DPKTS) {
    if (comma) strcat(buf, ",");
    strcat(buf, FT_XFIELD_ASC_DPKTS);
    comma = 1;
  }

  if (xfield & FT_XFIELD_DOCTETS) {
    if (comma) strcat(buf, ",");
    strcat(buf, FT_XFIELD_ASC_DOCTETS);
    comma = 1;
  }

  if (xfield & FT_XFIELD_FIRST) {
    if (comma) strcat(buf, ",");
    strcat(buf, FT_XFIELD_ASC_FIRST);
    comma = 1;
  }

  if (xfield & FT_XFIELD_LAST) {
    if (comma) strcat(buf, ",");
    strcat(buf, FT_XFIELD_ASC_LAST);
    comma = 1;
  }

  if (xfield & FT_XFIELD_ENGINE_TYPE) {
    if (comma) strcat(buf, ",");
    strcat(buf, FT_XFIELD_ASC_ENGINE_TYPE);
    comma = 1;
  }

  if (xfield & FT_XFIELD_ENGINE_ID) {
    if (comma) strcat(buf, ",");
    strcat(buf, FT_XFIELD_ASC_ENGINE_ID);
    comma = 1;
  }

  if (xfield & FT_XFIELD_SRCADDR) {
    if (comma) strcat(buf, ",");
    strcat(buf, FT_XFIELD_ASC_SRCADDR);
    comma = 1;
  }

  if (xfield & FT_XFIELD_DSTADDR) {
    if (comma) strcat(buf, ",");
    strcat(buf, FT_XFIELD_ASC_DSTADDR);
    comma = 1;
  }

  if (xfield & FT_XFIELD_NEXTHOP) {
    if (comma) strcat(buf, ",");
    strcat(buf, FT_XFIELD_ASC_NEXTHOP);
    comma = 1;
  }

  if (xfield & FT_XFIELD_INPUT) {
    if (comma) strcat(buf, ",");
    strcat(buf, FT_XFIELD_ASC_INPUT);
    comma = 1;
  }

  if (xfield & FT_XFIELD_OUTPUT) {
    if (comma) strcat(buf, ",");
    strcat(buf, FT_XFIELD_ASC_OUTPUT);
    comma = 1;
  }

  if (xfield & FT_XFIELD_SRCPORT) {
    if (comma) strcat(buf, ",");
    strcat(buf, FT_XFIELD_ASC_SRCPORT);
    comma = 1;
  }

  if (xfield & FT_XFIELD_DSTPORT) {
    if (comma) strcat(buf, ",");
    strcat(buf, FT_XFIELD_ASC_DSTPORT);
    comma = 1;
  }

  if (xfield & FT_XFIELD_PROT) {
    if (comma) strcat(buf, ",");
    strcat(buf, FT_XFIELD_ASC_PROT);
    comma = 1;
  }

  if (xfield & FT_XFIELD_TOS) {
    if (comma) strcat(buf, ",");
    strcat(buf, FT_XFIELD_ASC_TOS);
    comma = 1;
  }

  if (xfield & FT_XFIELD_TCP_FLAGS) {
    if (comma) strcat(buf, ",");
    strcat(buf, FT_XFIELD_ASC_TCP_FLAGS);
    comma = 1;
  }

  if (xfield & FT_XFIELD_SRC_MASK) {
    if (comma) strcat(buf, ",");
    strcat(buf, FT_XFIELD_ASC_SRC_MASK);
    comma = 1;
  }

  if (xfield & FT_XFIELD_DST_MASK) {
    if (comma) strcat(buf, ",");
    strcat(buf, FT_XFIELD_ASC_DST_MASK);
    comma = 1;
  }