sniffersender.java

流量分析 可以对SNIFFER抓的包进行分析

package com.tianxun.NEI.sniffer;

import com.tianxun.NEI.sniffer.util.NetFlow;
import com.tianxun.NEI.sniffer.util.NetFlowHeader;
import com.tianxun.NEI.sniffer.util.NetFlowRecord;
import com.tianxun.NEI.sniffer.util.GscDiag;
import com.tianxun.NEI.sniffer.util.Logger;

import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.BindException;
import java.net.DatagramSocket;
import java.net.DatagramPacket;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;
import java.util.Iterator;
import java.util.TimerTask;

/**
 * 发送NetFlow包的线程.
 *
 * @author 聂军
 * @version 1.0 2004-9-13
 */

public class SnifferSender extends TimerTask {

    private static final DateFormat DATE_FORMAT =
        new SimpleDateFormat("yyyyMMdd");

    private static final DateFormat TIME_FORMAT =
        new SimpleDateFormat("HHmmss");

    private static final SimpleDateFormat sdf =
        new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    // 每个被发送的NetFlow包中包含的最大记录个数
    private static final int MAX_FLOW_NUM = 30;

    // 每个被发送的NetFlow包的最大字节长度
    private static final int MAX_FLOW_LEN = 1464;

    private static final int LOCAL_PORT =
        NSXSniffer.getConfigParam("Send.udp.local.port", 52521);

    private static final int DEST_PORT =
        NSXSniffer.getConfigParam("Send.udp.dst.port", 52523);

    private static final String DEST_IP =
        NSXSniffer.getConfigParam("Send.udp.dst.ip", "localhost");

    private static final boolean IS_EXPORT_FILE =
        NSXSniffer.getConfigParam("Write.flow.into.file", false);

    private static final String EXPORT_DIR_NAME =
        NSXSniffer.getConfigParam("Write.flow.into.dir.name", "../NetFlow");

    private static final String EXPORT_FILE_NAME =
        NSXSniffer.getConfigParam("Write.flow.into.file.name", "NetFlow");

    private static final String EXPORT_FILE_SUFFIX =
        NSXSniffer.getConfigParam("Write.flow.into.file.suffix", "rcd");

    private ByteArrayOutputStream bas = null;
    private DatagramSocket datagramSocket = null;
    private DatagramPacket datagramPacket = null;
    private NetFlowHeader flowHeader = new NetFlowHeader();

    /**
     * 构造器.
     */
    public SnifferSender() {
        try {
            String dstIp = DEST_IP;
            // 解析IP地址
            if ("localhost".equalsIgnoreCase(dstIp) ||
                "127.0.0.1".equalsIgnoreCase(dstIp)) {
                dstIp = InetAddress.getLocalHost().getHostAddress();
            }

            bas = new ByteArrayOutputStream(MAX_FLOW_LEN);

            for (int i = 0; i < 100; i++) {
                try {
                    datagramSocket = new DatagramSocket(LOCAL_PORT + i);
                    if (i == 99) {
                        System.exit(0);
                    } else {
                        break;
                    }
                } catch (BindException bex) {
                }
            }

            InetAddress inetAddress = InetAddress.getByAddress(
                NetFlow.ip2byte(dstIp));
            datagramPacket = new DatagramPacket(new byte[]{0}, 0,
                new InetSocketAddress(inetAddress, DEST_PORT));
        } catch (Exception ex) {
            ex.printStackTrace(System.out);
            GscDiag.printStackTrace(5, "SnifferSenderException:",ex);
            System.exit(0);
        }
    }

    /**
     * @see java.util.TimerTask#run()
     */
    public void run() {
        Map ht = NSXSniffer.getApp().getCurrentValidBuffer();
        int size = ht.size();
        if (size <= 0) {
            return;
        }
        int num = size / MAX_FLOW_NUM;
        int mod = size % MAX_FLOW_NUM;
        int count = 0;

        try {
            for (Iterator iter = ht.values().iterator(); iter.hasNext();) {

                count++;
                if (count % MAX_FLOW_NUM == 1) {
                    // 写入Flow包头
                    if (count > (num * MAX_FLOW_NUM)) {
                        flowHeader.setCount(mod);
                    } else {
                        flowHeader.setCount(MAX_FLOW_NUM);
                    }
                    flowHeader.writeBytes(bas);
                }
                // 写入Flow记录
                NetFlowRecord record = (NetFlowRecord) iter.next();
                /*GscDiag.log(0, "record:"+"Rpkt:"+record.getDPkts()+",RDoctes:"+record.getDOctets()+
                ",sIP:"+NetFlow.long2ip(record.getSrcAddr())+",dIP:"+NetFlow.long2ip(record.getDstAddr())+
                ",sPort:"+record.getSrcPort()+",dPort:"+record.getDstPort()+",Tcp:"+record.getTcpFlags()+
                ",time:"+sdf.format(new Date(System.currentTimeMillis())));*/

                /*if(NetFlow.long2ip(record.getDstAddr()).equals("192.168.1.104")){
                    System.out.println("record:"+"Rpkt:"+record.getDPkts()+",RDoctes:"+record.getDOctets()+
                    ",sIP:"+NetFlow.long2ip(record.getSrcAddr())+",dIP:"+NetFlow.long2ip(record.getDstAddr())+
                    ",sPort:"+record.getSrcPort()+",dPort:"+record.getDstPort()+",Tcp:"+record.getTcpFlags()+
                    ",time:"+sdf.format(new Date(System.currentTimeMillis())));
                }*/
                record.writeBytes(bas);
                if (count % MAX_FLOW_NUM == 0 || count == size) {
                    byte[] data = bas.toByteArray();
                    int length = bas.size();
                    datagramPacket.setData(data);
                    datagramPacket.setLength(length);
                    datagramSocket.send(datagramPacket);
                    bas.reset();
                }
            }
            ht.clear();
        } catch (Exception ex) {
            GscDiag.printStackTrace(5, "SnifferSenderException:",ex);
            ex.printStackTrace(System.out);
        } finally {
            bas.reset();
        }
    }

}