v5_flow.java

接收网络设备上NetFlow工具导出的NetFlow数据

package cai.flow.packets;

import java.sql.PreparedStatement;
import java.sql.SQLException;

import cai.flow.packets.v9.FieldDefinition;
import cai.flow.packets.v9.Template;
import cai.flow.struct.Address;
import cai.flow.struct.Prefix;
import cai.flow.struct.Scheme_DataASMatrix;
import cai.flow.struct.Scheme_DataDstAS;
import cai.flow.struct.Scheme_DataHostMatrix;
import cai.flow.struct.Scheme_DataInterface;
import cai.flow.struct.Scheme_DataInterfaceMatrix;
import cai.flow.struct.Scheme_DataNode;
import cai.flow.struct.Scheme_DataPrefix;
import cai.flow.struct.Scheme_DataPrefixMatrix;
import cai.flow.struct.Scheme_DataProtocol;
import cai.flow.struct.Scheme_DataSrcAS;
import cai.sql.SQL;
import cai.utils.*;

/*
 V5 Flow structure，一个流的结构
 V9的data flowset记录也用这个结构入库

 *-------*-----------*----------------------------------------------------------*
 | Bytes | Contents  | Description                                              |
 *-------*-----------*----------------------------------------------------------*
 | 0-3   | srcaddr   | Source IP address                                        |
 *-------*-----------*----------------------------------------------------------*
 | 4-7   | dstaddr   | Destination IP address                                   |
 *-------*-----------*----------------------------------------------------------*
 | 8-11  | nexthop   | IP address of next hop router                            |
 *-------*-----------*----------------------------------------------------------*
 | 12-13 | input     | Interface index (ifindex) of input interface             |
 *-------*-----------*----------------------------------------------------------*
 | 14-15 | output    | Interface index (ifindex) of output interface            |
 *-------*-----------*----------------------------------------------------------*
 | 16-19 | dPkts     | Packets in the flow                                      |
 *-------*-----------*----------------------------------------------------------*
 | 20-23 | dOctets   | Total number of Layer 3 bytes in the packets of the flow |
 *-------*-----------*----------------------------------------------------------*
 | 24-27 | First     | SysUptime at start of flow                               |
 *-------*-----------*----------------------------------------------------------*
 | 28-31 | Last      | SysUptime at the time the last packet of the flow was    |
 |       |           | received                                                 |
 *-------*-----------*----------------------------------------------------------*
 | 32-33 | srcport   | TCP/UDP source port number or equivalent                 |
 *-------*-----------*----------------------------------------------------------*
 | 34-35 | dstport   | TCP/UDP destination port number or equivalent            |
 *-------*-----------*----------------------------------------------------------*
 | 36    | pad1      | Unused (zero) bytes                                      |
 *-------*-----------*----------------------------------------------------------*
 | 37    | tcp_flags | Cumulative OR of TCP flags                               |
 *-------*-----------*----------------------------------------------------------*
 | 38    | prot      | IP protocol type (for example, TCP = 6; UDP = 17)        |
 *-------*-----------*----------------------------------------------------------*
 | 39    | tos       | IP type of service (ToS)                                 |
 *-------*-----------*----------------------------------------------------------*
 | 40-41 | src_as    | Autonomous system number of the source, either origin or |
 |       |           | peer                                                     |
 *-------*-----------*----------------------------------------------------------*
 | 42-43 | dst_as    | Autonomous system number of the destination, either      |
 |       |           | origin or peer                                           |
 *-------*-----------*----------------------------------------------------------*
 | 44    | src_mask  | Source address prefix mask bits                          |
 *-------*-----------*----------------------------------------------------------*
 | 45    | dst_mask  | Destination address prefix mask bits                     |
 *-------*-----------*----------------------------------------------------------*
 | 46-47 | pad2      | Unused (zero) bytes                                      |
 *-------*-----------*----------------------------------------------------------*

 */

public class V5_Flow extends Flow {
    String srcaddr = "", dstaddr = "", nexthop = "";

    Prefix srcprefix, dstprefix;

    long input = -1, output = -1;

    long dPkts = 0, dOctets = 0, First = 0, Last = 0;

    long srcport = -1, dstport = -1;

    byte tcp_flags = 0, prot = -1, tos = 0;

    long src_as = -1, dst_as = -1;

    byte src_mask = 0, dst_mask = 0;

    String RouterIP = "";

    long src_addr = 0, dst_addr = 0, next_hop = 0;

    public V5_Flow(String RouterIP, final byte[] buf, int off, Template t) throws
            DoneException {
        this.RouterIP = RouterIP;
        if (buf.length < t.getTypeOffset( -1)) { // 多做判断，去掉垃圾包
            throw new DoneException("包长度不符合template" + t.getTemplateId() + "要求");
        }
        // 考虑offset和长度为非法值的问题的问题
        int currOffset = 0, currLen = 0;
        currOffset = t.getTypeOffset(FieldDefinition.IPV4_SRC_ADDR);
        currLen = t.getTypeLen(FieldDefinition.IPV4_SRC_ADDR);
        if (currOffset >= 0 && currLen > 0) {
            srcaddr = Util.str_addr(src_addr = Util.to_number(buf, off
                    + currOffset, currLen));
            if (Params.isSrcExcludes(src_addr)) {
                throw new DoneException(""); //放弃这个流
            }
            if (!Params.isSrcIncludes(src_addr)) {
                throw new DoneException("");
            }
            if (srcaddr.startsWith("0.")) {
                System.err.println("ERROR:Template is " + t.getTemplateId() +
                                   " router is " + t.getRouterIp() +
                                   " has srcaddr like 0.");
                throw new DoneException("savePacketF_"+RouterIP+"_"+t.getTemplateId());
            }
        }
        currOffset = t.getTypeOffset(FieldDefinition.IPV4_DST_ADDR);
        currLen = t.getTypeLen(FieldDefinition.IPV4_DST_ADDR);
        if (currOffset >= 0 && currLen > 0) {
            dstaddr = Util.str_addr(dst_addr = Util.to_number(buf, off
                    + currOffset, currLen));
            if (Params.isDstExcludes(dst_addr)) {
                throw new DoneException("");
            }
            if (!Params.isDstIncludes(dst_addr)) {
                throw new DoneException("savePacketF_"+RouterIP+"_"+t.getTemplateId());
            }
            if (srcaddr.startsWith("0.")) {
                System.err.println("ERROR:Template is " + t.getTemplateId() +
                                   " router is " + t.getRouterIp() +
                                   " has srcaddr like 0.");
                throw new DoneException("");
            }
        }
        currOffset = t.getTypeOffset(FieldDefinition.IPV4_NEXT_HOP);
        currLen = t.getTypeLen(FieldDefinition.IPV4_NEXT_HOP);
        if (currOffset >= 0 && currLen > 0) {
            nexthop = Util.str_addr(next_hop = Util.to_number(buf, off
                    + currOffset, currLen));
        }
        currOffset = t.getTypeOffset(FieldDefinition.INPUT_SNMP);
        currLen = t.getTypeLen(FieldDefinition.INPUT_SNMP);
        if (currOffset >= 0 && currLen > 0) {
            input = Util.to_number(buf, off + currOffset, currLen);
        }
        currOffset = t.getTypeOffset(FieldDefinition.OUTPUT_SNMP);
        currLen = t.getTypeLen(FieldDefinition.OUTPUT_SNMP);
        if (currOffset >= 0 && currLen > 0) {
            output = Util.to_number(buf, off + currOffset, currLen);
        }
        currOffset = t.getTypeOffset(FieldDefinition.InPKTS_32);
        currLen = t.getTypeLen(FieldDefinition.InPKTS_32);
        if (currOffset >= 0 && currLen > 0) {
            dPkts = Util.to_number(buf, off + currOffset, currLen) *
                    t.getSamplingRate();
        }
        currOffset = t.getTypeOffset(FieldDefinition.InBYTES_32);
        currLen = t.getTypeLen(FieldDefinition.InBYTES_32);
        if (currOffset >= 0 && currLen > 0) {
            dOctets = Util.to_number(buf, off + currOffset, currLen) *
                      t.getSamplingRate();
        }
        currOffset = t.getTypeOffset(FieldDefinition.FIRST_SWITCHED);
        currLen = t.getTypeLen(FieldDefinition.FIRST_SWITCHED);
        if (currOffset >= 0 && currLen > 0) {
            First = Util.to_number(buf, off + currOffset,
                                   currLen);
            if (!Variation.getInstance().judgeVary(First)) {
                throw new DoneException("Error:Time MisMatch");
            }
        }
        currOffset = t.getTypeOffset(FieldDefinition.LAST_SWITCHED);
        currLen = t.getTypeLen(FieldDefinition.LAST_SWITCHED);
        if (currOffset >= 0 && currLen > 0) {
            try {
                Last = Util.to_number(buf, off + currOffset, currLen);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        currOffset = t.getTypeOffset(FieldDefinition.L4_SRC_PORT);
        currLen = t.getTypeLen(FieldDefinition.L4_SRC_PORT);
        if (currOffset >= 0 && currLen > 0) {
            srcport = Util.to_number(buf, off + currOffset, currLen);
        }
        currOffset = t.getTypeOffset(FieldDefinition.L4_DST_PORT);
        currLen = t.getTypeLen(FieldDefinition.L4_DST_PORT);
        if (currOffset >= 0 && currLen > 0) {
            dstport = Util.to_number(buf, off + currOffset, currLen);
        }
        currOffset = t.getTypeOffset(FieldDefinition.TCP_FLAGS);
        currLen = t.getTypeLen(FieldDefinition.TCP_FLAGS);
        if (currOffset >= 0 && currLen > 0) {
            tcp_flags = buf[off + currOffset];
        }
        currOffset = t.getTypeOffset(FieldDefinition.PROT);
        currLen = t.getTypeLen(FieldDefinition.PROT);
        if (currOffset >= 0 && currLen > 0) {
            prot = buf[off + currOffset];
        }
        currOffset = t.getTypeOffset(FieldDefinition.SRC_TOS);
        currLen = t.getTypeLen(FieldDefinition.SRC_TOS);
        if (currOffset >= 0 && currLen > 0) {
            tos = buf[off + currOffset];
        }
        currOffset = t.getTypeOffset(FieldDefinition.SRC_AS);
        currLen = t.getTypeLen(FieldDefinition.SRC_AS);
        if (currOffset >= 0 && currLen > 0) {
            src_as = Util.to_number(buf, off + currOffset, currLen);
        }
        currOffset = t.getTypeOffset(FieldDefinition.DST_AS);
        currLen = t.getTypeLen(FieldDefinition.DST_AS);
        if (currOffset >= 0 && currLen > 0) {
            dst_as = Util.to_number(buf, off + currOffset, currLen);
        }
        currOffset = t.getTypeOffset(FieldDefinition.SRC_MASK);
        currLen = t.getTypeLen(FieldDefinition.SRC_MASK);
        if (currOffset >= 0 && currLen > 0) {
            src_mask = buf[off + currOffset];
        }
        currOffset = t.getTypeOffset(FieldDefinition.DST_MASK);
        currLen = t.getTypeLen(FieldDefinition.DST_MASK);
        if (currOffset >= 0 && currLen > 0) {
            dst_mask = buf[off + currOffset];
        }
        if (src_addr != 0 || src_mask != 0) {
            srcprefix = new Prefix(src_addr, src_mask);
            dstprefix = new Prefix(dst_addr, dst_mask);
        }
        if (dPkts + dOctets <= 0) { // 如果没有流量
            throw new DoneException("空流量");
        }
    }

    public V5_Flow(String RouterIP, byte[] buf, int off) throws DoneException {
        this.RouterIP = RouterIP;

        srcaddr = Util.str_addr(src_addr = Util.to_number(buf, off + 0, 4));
        dstaddr = Util.str_addr(dst_addr = Util.to_number(buf, off + 4, 4));
        nexthop = Util.str_addr(next_hop = Util.to_number(buf, off + 8, 4));

        input = Util.to_number(buf, off + 12, 2);
        output = Util.to_number(buf, off + 14, 2);
        dPkts = Util.to_number(buf, off + 16, 4);
        dOctets = Util.to_number(buf, off + 20, 4);