flowdmuxerdmux.py

该软件根据网络数据生成NetFlow记录。NetFlow可用于网络规划、负载均衡、安全监控等

###############################################################################
#                                                                             #
#   Copyright 2005 University of Cambridge Computer Laboratory.               #
#                                                                             #
#   This file is part of Nprobe.                                              #
#                                                                             #
#   Nprobe is free software; you can redistribute it and/or modify            #
#   it under the terms of the GNU General Public License as published by      #
#   the Free Software Foundation; either version 2 of the License, or         #
#   (at your option) any later version.                                       #
#                                                                             #
#   Nprobe is distributed in the hope that it will be useful,                 #
#   but WITHOUT ANY WARRANTY; without even the implied warranty of            #
#   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the             #
#   GNU General Public License for more details.                              #
#                                                                             #
#   You should have received a copy of the GNU General Public License         #
#   along with Nprobe; if not, write to the Free Software                     #
#   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA #
#                                                                             #
###############################################################################


##############################################################################
## 
##
## 
## 
##
## 

############################################################################


from __future__ import generators

import os
import sys
import getopt
import socket
from struct import unpack as up
from time import asctime, sleep, time
import traceback
from commands import getstatusoutput

import re
import pcap

from DMuxer import FlowDmuxer, FlowDmuxerException
from pcap import ntoa, aton
from pcap import _ntohs as ntohs
from pcap import _ether_ntoa as entoa
from pcap import _ether_aton as eaton
from pcap import py_pcap, nprobe_errstring

from dirwalk import walk



############################################################################
#############################################################################

#
# Link types recognised
#

DLT_EN10MB = 1
DLT_IEEE802 = 6
DLT_RAW = 12
DLT_RAW_HIGH = 101

DLT_NPROBE = 12321

#
# Dictionary of link types and functions to extract 'atuple' used in
# identifying flows
#

linktypes = {
    DLT_EN10MB: 'get_eth_atuple',
    DLT_IEEE802: 'get_eth_atuple',
    DLT_NPROBE: 'get_eth_atuple',
    DLT_RAW: 'get_raw_ip_atuple',
    DLT_RAW_HIGH: 'get_raw_ip_atuple'
    }

#
# IP and TCP flag values
#

IP_MF = 0x2000

TH_FIN = 0x01
TH_SYN = 0x02
TH_RST = 0x04
TH_PUSH = 0x08
TH_ACK = 0x10
TH_URG = 0x20

#
# packet direction indicators
#

CLIENT = 0
SERVER = 1

#
# Connection state flags used in spotting TCP closes
#

NO_STATE = 0
CSYN = 0x1
SSYN = 0x2
CSYNA = 0x4
SSYNA = 0x8
CFIN = 0x10
SFIN = 0x20
CFINA = 0x40
SFINA = 0x80

SYNS = (CSYN, SSYN)
SYNAS = (CSYNA, SSYNA)
FINS = (CFIN, SFIN)
FINAS = (CFINA, SFINA)

FINACKED = CFINA | SFINA

#
# IP protocols recognised
#

IPPROTO_ICMP = 1
IPPROTO_IGMP = 2
IPPROTO_IPIP = 4
IPPROTO_TCP = 6
IPPROTO_UDP = 17

#
# Dictionary of IP protocols and functions to build paths to flow directories
#

IP_PROTS = {
    IPPROTO_ICMP: ('ICMP', 'ip_path'),
    IPPROTO_IGMP: ('IGMP', 'ip_path'),
    IPPROTO_IPIP: ('IPIP', 'ip_path'),
    IPPROTO_TCP: ('TCP', 'tcpudp_path'),
    IPPROTO_UDP: ('UDP', 'tcpudp_path'),
    'FRAGS': ('FRAGS', 'ip_fragpath'),
    'OTHER': ('OTHER', 'ip_path')
    }

#
# Ethernet protocols recognised
#

ETHERTYPE_IP = 0x0800
ETHERTYPE_ARP = 0x0806
ETHERTYPE_REVARP = 0x8035

#
# Dictionary of Ethernet protocols and functions to build paths to
# flow directories
#

ETHERTYPES_D = {
    ETHERTYPE_IP: ('IP', IP_PROTS, 'nullpath'),
    ETHERTYPE_ARP: ('ARP', {}, 'arp_path'),
    ETHERTYPE_REVARP: ('REVARP', {}, 'arp_path'),
    'OTHER': ('OTHER', {}, 'arp_path')
    }

ETHER_HDR_LEN = 14
ETHERMTU = 1500

LLC_SNAP_HDR_LEN = 8
ETH_LLC_HDR_LEN = ETHER_HDR_LEN + LLC_SNAP_HDR_LEN
# Relative to packet start
#MAC_OFF = 0
#MAC_END = 14
ETH_PROT_OFF = 12
ETH_PROT_END = ETH_PROT_OFF + 2
SNAP_PROT_OFF = 20
SNAP_PROT_END = SNAP_PROT_OFF + 2

ETH_DH_OFF = 0
ETH_DH_END = ETH_DH_OFF + 6
ETH_SH_OFF = 6
ETH_SH_END = ETH_SH_OFF + 6

#
# Constant offsets and byte lengths (relative to hdr start) of IP header fields
#

IP_HDR_LEN = 20
IP_HL_OFF = 0
IP_IPLEN_OFF = 2
IP_IPLEN_END = IP_IPLEN_OFF + 2
IP_IPID_OFF = 4
IP_IPID_END = IP_IPID_OFF + 2
IP_IPOFF_OFF = 6
IP_IPOFF_END = IP_IPOFF_OFF + 2
IP_PROT_OFF = 9
IP_SA_OFF = 12
IP_SA_END = IP_SA_OFF + 4
IP_DA_OFF = 16
IP_DA_END = IP_DA_OFF + 4

#
# Constant offsets and byte lengths (relative to hdr start) of TCP/UDP
# port fields
#

TCPUDP_HDR_LEN = 20

TCPUDP_SP_OFF = 0
TCPUDP_SP_END = TCPUDP_SP_OFF + 2
TCPUDP_DP_OFF = 2
TCPUDP_DP_END = TCPUDP_DP_OFF + 2

#
# Constant offsets and byte lengths (relative to hdr start) of other TCP
# header fields used
#

TCP_SEQ_OFF = 4
TCP_SEQ_END = TCP_SEQ_OFF + 4
TCP_ACK_OFF = 8
TCP_ACK_END = TCP_ACK_OFF + 4

TCP_FLAGS_OFF = 13

ERRCODE_OFF = 4



############################################################################
#############################################################################

#
# A DLL to implement the flow timeout queue
#

class TimeoDLL:

    def __init__(self):

        self.n = self.p = self
        self.last = None
        self.len = 0

    def __len__(self):

        return self.len

    def ins_a(self, preob, ob):

        ob.n = preob.n
        preob.n = ob
        ob.p = preob
        ob.n.p = ob
        self.len += 1

    def ins_b(self, postob, ob):

        ob.n = postob
        ob.p = postob.p
        postob.p = ob
        ob.p.n = ob
        self.len += 1

    def prepend(self, ob):

        self.n.p = ob
        ob.n = self.n
        ob.p = self
        self.n = ob
        self.len += 1

    def append(self, ob):

        self.p.n = ob
        ob.p = self.p
        ob.n = self
        self.p = ob
        self.len += 1

    def rem(self, ob):

        
        ob.n.p = ob.p
        ob.p.n = ob.n
        self.len -= 1

    def toend(self, ob):

        self.rem(ob)
        self.append(ob)

    def isempty(self):

        return self.n == self

    def head(self):

        return self.n

    def tail(self):

        return self.p

    def pop(self):

        o = self.head()
        self.rem(o)
        return (o)

    def items(self):

        c = self.n
        while c != self:
            yield c
            c = c.n

    def check(self):

        for o in self.items():
            try:
                assert o.n.p == o
                assert o.p.n == o
            except AssertionError:
                print 'obj', o, 'o.n', o.n, 'o.p', o.p, 'p.n', o.p.n, 'n.p', o.n.p, 'otq', o.otq, 'ontimes', o.onotq, 'offtimes', o.offotq
                raise
            #if o.p.last:
                #assert o.p.last < o.last



############################################################################
#############################################################################

#
# A dummy timeout queue for operation without flow timeouts
#
                

class TimeoNull:

    def __init__(self):

        self.ins_a = self.ins_b = self.prepend = self.append = self.nullf
        self.rem = self.toend = self.nullf

        self.head = self.tail = self.pop = self.check = self.nullf

    def nullf(self, *args):

        return None

    def isempty(self):

        return 1

    def items(self):

        return []

    def __len__(self):

        return 0

############################################################################
#############################################################################

#
# One of these for each flow
#
        
class FlowOb:

    def __init__(self, atuple, ts):

        self.atuple = atuple # flow identifier
        self.start = ts
        self.file = None
        self.written = 0
        self.last = 0
        self.pn = -1
        
        self.state = NO_STATE   # used for TCP conns
        self.seqs = [None, None]# ditto
        
        self.held = [] # file/offset or un-dumped packets
        self.alts = [] # alternative identifiers to this flow (e.g. for IP frags)


############################################################################
#############################################################################

class FlowDmuxerDmux(FlowDmuxer):

    '''
    Class FlowDmuxerDmux (file FlowDmuxerDmux.py) - sub-class of FlowDmuxer

    Dmux all flows passing packet filter into tree (one file per flow) ordered
    by protocol, address, port etc. as appropriate.

    Options:

      --timeout=<t[us/ms/s/m/h]> Time out quiescent flows after t us/ms/s/m/h
                                 (no unit - us) - default is no timeout.
      --accept_nosyns            Accept TCP flows without SYN (default reject
                                 flows not commencing with at least one SYN).
      --check                    Check consistency of dmux tree on completion.
      --nprobe-errors=<n>        Denotes input file is Nprobe error packet dump
                                 - dmuxes by Nprobe error No. and originating
                                 host. Rank top n occurences by error, host and
                                 (error, host)

    A log file is written into the dmux tree base directory.
    
      '''

    def __init__(self, infile, filtexp, opath=None, raise_natural_exceptions=0, save_check_info=0, nprobe_errors=0):

        FlowDmuxer.__init__(self, infile, filtexp, opath, raise_natural_exceptions=raise_natural_exceptions)

        self.accept_nosyns = 0
        self.save_check_info = save_check_info
        self.npe = nprobe_errors

        self.pcap_next = self.pcap.next_o_tts

        self.flowdir = {}
        self.fragdir = {}
        self.pending_dumps = []

        self.ttimeo = TimeoNull()
        self.timeout = long(time())*1000000

        self.odir = os.path.join(self.odir, self.ofile) #always a dir
        if self.npe:
            str = '_nprobe_errors'
            self.errd = {}
        else:
            str = '_dmux'
        if not self.ofile:
            self.odir = os.path.join(self.odir, os.path.split(self.infile)[1]+str)
        else:
            self.odir = os.path.join(self.odir, os.path.split(self.infile)[1]+str)

        self.make_proto_dirs(self.odir)

        self.open_log()

        self.set_link()

        self.ipkts = 0
        self.dpkts = 0
        self.nconns = 0
        self.nosyn = 0
        self.truncated = 0 
        self.nopen = 0
        self.to = 0
        self.resumed = 0
        self.closed = 0
        self.frags = 0
        self.active_at_end = 0
        if self.save_check_info:
            self.oots = {}

        
############################################################################

    def __del__(self):
        
        
        FlowDmuxer.__del__(self)

        
############################################################################

    def set_link(self):

     try:
         self.get_atuple = getattr(self, linktypes[self.linktype])
     except KeyError:
         s = 'Can\'t recognise data link type (0x%x)' % (self.linktype)
         self.log(s)
         print s
         sys.exit(1)