flow-import.c

netflow,抓包

 * functio: format_NFCollector1
 * 
 * import from Cisco NFCollector v1 ascii files
 */

/* Break line into fields */
static int ascii_fields(char **fp,int maxfp,char *inbuf,const char *delim)
{
  int n=0;
  while((*fp = strsep(&inbuf,delim)) != NULL && n < maxfp) {
    n++;
    fp++;
  }
  return ( n < maxfp ? n : 0 );
}

/* Map NFCollector recordtypes to flow-tools datatypes */
typedef enum { TYPE_IPV4, TYPE_16B, TYPE_32B, TYPE_8B , TYPE_DISCARD,
	       TYPE_LAST } cvt_t;

struct for2nat_st {
  u_int64 mask;
  cvt_t type;
  int offset;
};
struct nfcollector2flowtools {
  char *name;
  struct for2nat_st fcv[24];
};

struct nfcollector2flowtools NFC2ft[] = {
  { "AGGREGATION CallRecord", 
    { { FT_XFIELD_SRCADDR, TYPE_IPV4, offsetof(struct fts3rec_offsets,srcaddr) },
      { FT_XFIELD_DSTADDR, TYPE_IPV4, offsetof(struct fts3rec_offsets,dstaddr) },
      { FT_XFIELD_SRCPORT, TYPE_16B,  offsetof(struct fts3rec_offsets,srcport) },
      { FT_XFIELD_DSTPORT, TYPE_16B,  offsetof(struct fts3rec_offsets,dstport) },
      { FT_XFIELD_PROT   , TYPE_8B ,  offsetof(struct fts3rec_offsets,prot) },
      { FT_XFIELD_TOS    , TYPE_16B,  offsetof(struct fts3rec_offsets,tos) },
      { FT_XFIELD_DPKTS,   TYPE_32B,  offsetof(struct fts3rec_offsets,dPkts) },
      { FT_XFIELD_DOCTETS, TYPE_32B,  offsetof(struct fts3rec_offsets,dOctets) },
      { FT_XFIELD_DFLOWS,  TYPE_32B,  offsetof(struct fts3rec_offsets,dFlows) },
      /* We have pretty darn good uptime */
      { FT_XFIELD_FIRST, TYPE_32B,offsetof(struct fts3rec_offsets,First) },
      { FT_XFIELD_LAST   , TYPE_32B,  offsetof(struct fts3rec_offsets,Last) },
      { 0,                 TYPE_DISCARD, 0 },
      { 0,                 TYPE_LAST, 0 } }
  },
  { "AGGREGATION HostMatrix",
    { { FT_XFIELD_SRCADDR, TYPE_IPV4, offsetof(struct fts3rec_offsets,srcaddr) },
      { FT_XFIELD_DSTADDR, TYPE_IPV4, offsetof(struct fts3rec_offsets,dstaddr) },
      { FT_XFIELD_DPKTS,   TYPE_32B,  offsetof(struct fts3rec_offsets,dPkts) },
      { FT_XFIELD_DOCTETS, TYPE_32B,  offsetof(struct fts3rec_offsets,dOctets) },
      { FT_XFIELD_DFLOWS,  TYPE_32B,  offsetof(struct fts3rec_offsets,dFlows) },
      { 0,                 TYPE_LAST, 0 } }
  },
  { "AGGREGATION SourceNode",
    { { FT_XFIELD_SRCADDR, TYPE_IPV4, offsetof(struct fts3rec_offsets,srcaddr) },
      { FT_XFIELD_DPKTS,   TYPE_32B,  offsetof(struct fts3rec_offsets,dPkts) },
      { FT_XFIELD_DOCTETS, TYPE_32B,  offsetof(struct fts3rec_offsets,dOctets) },
      { FT_XFIELD_DFLOWS,  TYPE_32B,  offsetof(struct fts3rec_offsets,dFlows) },
      { 0,                 TYPE_LAST } }
  },
  { "AGGREGATION DestNode",
    { { FT_XFIELD_DSTADDR, TYPE_IPV4, offsetof(struct fts3rec_offsets,dstaddr) },
      { FT_XFIELD_DPKTS,   TYPE_32B,  offsetof(struct fts3rec_offsets,dPkts) },
      { FT_XFIELD_DOCTETS, TYPE_32B,  offsetof(struct fts3rec_offsets,dOctets) },
      { FT_XFIELD_DFLOWS,  TYPE_32B,  offsetof(struct fts3rec_offsets,dFlows) },
      { 0,                 TYPE_LAST } }
  },
    
  { NULL }
};


int format_NFCollector1(struct ftio *ftio, struct options *opt)
{
  struct for2nat_st *fcv;
  struct fts3rec_offsets fo;
  u_char buf[FT_IO_MAXREC];
  char inbuf[1024];
  char *rec;
  char *fields[20];
  u_int64 dmask, imask;
  int ret;
  int debug=ftio_get_debug(ftio);

  rec = (char*)&buf;

  fts3rec_compute_offsets(&fo, &opt->ftv);

  dmask=0;
  if(opt->ftv.d_version == 8)
    dmask = ( opt->ftv.agg_method > 0 && opt->ftv.agg_method < sizeof(v8mask)-1 ?
	      v8mask[opt->ftv.agg_method] : 0);
  else
    dmask = ( opt->ftv.d_version > 0 && opt->ftv.d_version < sizeof(vXmask)-1 ?
	      vXmask[opt->ftv.d_version] : 0);
  if(!dmask) {
    fterr_warnx("unsupported export version");
    return -1;
  }

  /* First line has format descriptor */
  if(fgets(inbuf,sizeof(inbuf),stdin) == NULL) {
    fterr_warnx("No header");
    return -1;
  }
  if(ascii_fields(fields,sizeof(fields),inbuf,"|") != 9) {
    /* XXX complain about unregocnized format */
    fterr_warnx("Unregconized format line");
    return -1;
  }
  if(strcmp("FORMAT A",fields[1]) != 0) {
    fterr_warnx("Unregocnized format \"%s\"",fields[1]);
    return -1;
  }
  fcv=NULL;
  for(ret=0;NFC2ft[ret].name != NULL;ret++) {
    if(strcmp(NFC2ft[ret].name,fields[2]) == 0) {
      fcv=(NFC2ft[ret].fcv);
      if(debug)
	fprintf(stderr,"Input format %s\n",fields[2]);
      break;
    }
  } 
  if(fcv == NULL) {
    fterr_warnx("Unsupported aggrecation scheme \"%s\"",fields[2]);
    return -1;
  }
  /* Compute imask from conversion table */
  imask=0;
  for(ret = 0; (fcv+ret)->type != TYPE_LAST; ret++) {
    imask = imask | (fcv+ret)->mask;
  }
  /* check compatibility of imask & dmask XXX */
  if((imask & dmask) == 0) {
    fterr_warnx("Incompatible input and destination (no common fields)");
    return -1;
  }

  ret = 1;
  while (!(feof(stdin))) {
    char **fp;
    u_int32 val=0;
    int off=0,n,i;
    if(fgets(inbuf,sizeof(inbuf),stdin) == NULL)
      continue;
    ++ret;
    n=ascii_fields(fields,sizeof(fields),inbuf,"|");
    fp=fields;
    bzero(rec,sizeof(buf));
    for(i=0;i < n && *fp;i++) {
      if(dmask & fcv[i].mask) {
	off=*(short *)(((char *)&(fo))+fcv[i].offset);
	switch(fcv[i].type) {
	case TYPE_32B:
	case TYPE_16B:
	case TYPE_8B:
	  val = strtoul(*fp,(char **)0L,10);
	  break;
	case TYPE_IPV4:
	  val = scan_ip(*fp);
	  /* Fall thru */
	case TYPE_DISCARD:
	case TYPE_LAST:
	default:
	  break;
	}
	switch(fcv[i].type) {
	case TYPE_16B:
	  *((u_int16*)(rec+off)) = (u_int16)val;
	  break;
	case TYPE_IPV4:
	case TYPE_32B:
	  *((u_int32*)(rec+off)) = (u_int32)val;
	  break;
	case TYPE_8B:
	  *((u_int8*)(rec+off)) = (u_int8)val;
	case TYPE_DISCARD:
	case TYPE_LAST:
	  break; /* Make sun Cpro happy */
	}
      }
      fp++;
    }
    /* XXX Something sane here */
    if(fp && fcv[i].type == TYPE_LAST) {
      if(ftio_write(ftio,rec) < 0) {
	fterr_warnx("ftio_write(): failed");
	break;
      }
      ++opt->records;
    } else
      fterr_warnx("Broken record at line %d (not output)",ret);
  }
  return 1;
}

/*
 * function: format4
 *
 * raw packet format
*/
int format4(struct ftio *ftio, struct options *opt)
{
  struct ftpdu_header ftheader, ftheader2;
  struct ftpdu ftpdu;
  struct ftseq ftseq;
  size_t rlen, len;
  void (*xlate)(void *in_rec, void *out_rec);
  int ret, n, i, offset;
  char xl_rec[FT_IO_MAXREC], *out_rec;

  bzero (&ftpdu, sizeof ftpdu);
  bzero (&ftseq, sizeof ftseq);
  xlate = (void*)0L;

  while (!(feof(stdin))) {

    ret = -1;

    /* find out the PDU version and flow count */
    if ((rlen = fread(&ftheader, sizeof (ftheader), 1, stdin) != 1))
      goto done;

    /* copy to swap bytes */
    bcopy(&ftheader, &ftheader2, sizeof ftheader);

#if BYTE_ORDER == LITTLE_ENDIAN
  SWAPINT16(ftheader2.version);
  SWAPINT16(ftheader2.count);
#endif /* LITTLE_ENDIAN */

    switch (ftheader2.version) {

      case 1:
        len = ftheader2.count*sizeof(struct ftrec_v1);
        /* v1 does not have sequence# and engine* */
        len -= 8;
        break;

      case 5:
        len = ftheader2.count*sizeof(struct ftrec_v5);
        break;

      case 6:
        len = ftheader2.count*sizeof(struct ftrec_v6);
        break;

      case 7:
        len = ftheader2.count*sizeof(struct ftrec_v7);
        break;

      case 8:

        switch (ftheader2.aggregation) {

          case 1:
            len = ftheader2.count*sizeof(struct ftrec_v8_1);
            break;

          case 2:
            len = ftheader2.count*sizeof(struct ftrec_v8_2);
            break;

          case 3:
            len = ftheader2.count*sizeof(struct ftrec_v8_3);
            break;

          case 4:
            len = ftheader2.count*sizeof(struct ftrec_v8_4);
            break;

          case 5:
            len = ftheader2.count*sizeof(struct ftrec_v8_5);
            break;

          case 6:
            len = ftheader2.count*sizeof(struct ftrec_v8_6);
            break;

          case 7:
            len = ftheader2.count*sizeof(struct ftrec_v8_7);
            break;

          case 8:
            len = ftheader2.count*sizeof(struct ftrec_v8_8);
            break;

          case 9:
            len = ftheader2.count*sizeof(struct ftrec_v8_9);
            break;

          case 10:
            len = ftheader2.count*sizeof(struct ftrec_v8_10);
            break;

          case 11:
            len = ftheader2.count*sizeof(struct ftrec_v8_11);
            break;

          case 12:
            len = ftheader2.count*sizeof(struct ftrec_v8_12);
            break;

          case 13:
            len = ftheader2.count*sizeof(struct ftrec_v8_13);
            break;

          case 14:
            len = ftheader2.count*sizeof(struct ftrec_v8_14);
            break;

          default:
            fterr_errx(1, "Unrecognized PDU version: %d aggregation %d.",
              ftheader2.version, ftheader2.aggregation);
            break;
        } /* switch */
        break;

      default:
        fterr_errx(1, "Unrecognized PDU version: %d.", ftheader2.version);
        break;

    } /* switch */

    ftpdu.bused = len + sizeof (ftheader);

    /* read in the rest of the PDU */
    if ((rlen = fread(ftpdu.buf+sizeof (ftheader), len, 1, stdin) != 1)) {
      fterr_errx(1, "fread(): failed - expecting to read %d bytes.", len);
      goto done;
    }

    /* copy in the read header */
    bcopy(&ftheader, &ftpdu.buf, sizeof (ftheader));

    /* verify integrity, get version */
    if (ftpdu_verify(&ftpdu) < 0) {
      fterr_warnx("ftpdu_verify(): failed.");
      goto done;
    }

    /* first flow or no configured destination? */
    if (!opt->ftv.set) {

      bcopy(&ftpdu.ftv, &opt->ftv, sizeof ftpdu.ftv);
      opt->ftv.set = 1;

    } else {

      /* translation among v8 aggregation methods not possible */
      if ((opt->ftv.d_version == 8) &&
        ((opt->ftv.agg_method != ftpdu.ftv.agg_method)
        || (opt->ftv.agg_version != ftpdu.ftv.agg_version))) {

        fterr_errx(1, "Unexpected PDU: oagg=%d agg=%d over=%d ver=%d",
          opt->ftv.agg_method, ftpdu.ftv.agg_method,
          opt->ftv.agg_version, ftpdu.ftv.agg_version);
      }

      if (opt->ftv.d_version != ftpdu.ftv.d_version)
        xlate = ftrec_xlate_func(&ftpdu.ftv, &opt->ftv);

    }

    /* verify sequence # */
    if (ftpdu_check_seq(&ftpdu, &ftseq) < 0) {

      fterr_warnx("ftpdu_seq_check(): expecting=%lu received=%lu lost=%lu",
        (u_long)ftseq.seq_exp,
        (u_long)ftseq.seq_rcv,
        (u_long)ftseq.seq_lost);

    }

    /* decode */
    ftpdu.ftd.byte_order = opt->ftset.byte_order;

    n = fts3rec_pdu_decode(&ftpdu);

    /* update the exporter stats */
    opt->records += n;

    /* write */
    for (i = 0, offset = 0; i < ftpdu.ftd.count; 
      ++i, offset += ftpdu.ftd.rec_size) {

      if (xlate) {

        xlate(ftpdu.ftd.buf+offset, &xl_rec);
        out_rec = (char*)&xl_rec;

      } else {

        out_rec = (char*)ftpdu.ftd.buf+offset;

      }

      if ((n = ftio_write(ftio, out_rec)) < 0)
        fterr_errx(1, "ftio_write(): failed");

    }

  } /* while */

done:

  return ret;

} /* format4 */

void usage(void) {

  fprintf(stderr, "Usage: flow-import [-h] [-b big|little] [-d debug_level] [-f format]\n");
  fprintf(stderr, "       [-m mask_fields] [-V pdu_version] [-z z_level]\n");


  fprintf(stderr, "\n%s version %s: built by %s\n", PACKAGE, VERSION, FT_PROG_BUILD);

} /* usage */