flowdmuxerdmux.py

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

     self.snaplen = self.pcap.snaplen()

     if self.npe:
         if self.linktype != DLT_NPROBE:
             s = 'Nprobe-errors specified but linktype is not DLT_NPROBE'
             print s
             self.log(s)
             sys.exit(1)
         else:
             self.get_atuple = self.get_err_atuple

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

    def make_proto_dirs(self, root):

        join = os.path.join
        dd = self.protd = {}
        dirs = []

        for p0, (s0, d, mp0) in ETHERTYPES_D.items():
            path0 = join(root, s0)
            dd[(p0, None)] = (path0, getattr(self, mp0)) 
            dirs.append(path0)
            for p1, (s1, mp1) in d.items():
                path1 = join(path0, str(p1))
                dd[(p0, p1)] = (path1, getattr(self, mp1)) 
                dirs.append(path1)

        for dir in dirs:
            try:
                os.makedirs(dir)
            except OSError, s:
                if str(s).find('File exists') < 0:
                    self.exception('can\'t create dir %s (%s)' % (dir, str(s)))
            
        
        
############################################################################

    def open_log(self):
        
        self.logfile = os.path.join(self.odir, 'log')
        try:
            self.logf = open(self.logfile, 'w')
        except IOError, s:
            self.exception('can\'t create  %s (%s)' % (logfile, str(s)))
        
        self.logf.write('\n\nLog for FlowDmuxerDmux:\n\n')
        self.logf.write('Run on file %s  by %s@%s on %s\n\n' % (self.infile, os.environ['LOGNAME'], os.environ['HOSTNAME'], asctime()))

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

    def log(self, s):

        self.logf.write(s + '\n')

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

    def report(self):

        for a in ('ipkts', 'dpkts', 'nosyn', 'truncated', 'frags', 'nconns', 'closed', 'to', 'resumed', 'active_at_end'):

            s = '%-15s %d' % (a, getattr(self, a))
            self.log(s)
            print s

        if self.npe:
            # rank errors
            by_e = {}
            by_h = {}
            all = []
            toterrs = 0

            for (host, err), n in self.errd.items():
                toterrs += n
                try:
                    by_e[err] += n
                except KeyError:
                    by_e[err] = n
                ## try:
##                     by_h[host] += n
##                 except KeyError:
##                     by_h[host] = n
                ehd = by_h.setdefault(host, [0, {}])
                ehd[0] += n
                ehd[1][err] = ehd[1].setdefault(err, 0) + n
                all.append((n, err, host))

            by_e = [(n, e) for e, n in by_e.items()]
            by_h_n = [(n, h, errs) for h, (n, errs) in by_h.items()]
            
            for l in [by_e, by_h_n, all]:
                l.sort()
                l.reverse()

            N = min(self.npe, len(by_e))
            s = '\nNprobe errors by error (top %d shown):\n\t%s %s %%\n' % (N, 'Errno.', 'No.'.center(6))
            print s
            self.log(s)
            for n, e in by_e[:N]:
                s = '\t%4d %6d %6.2f (%-s)' % (-e, n, (n*100.0)/toterrs, nprobe_errstring(e))
                print s
                self.log(s)

            N = min(self.npe, len(by_h_n))
            s = '\nNprobe errors by host (top %d shown):\n\t%s %s %s\n' % (N, 'Host'.center(18), 'Total'.center(6), 'Errors and number'.center(24))
            print s
            self.log(s)
            for n, h, errs in by_h_n[:self.npe]:
                s = '\t%18s %6d' % (ntoa(h), n)
                el = [(n, e) for e, n in errs.items()]
                el.sort()
                el.reverse()
                for ne, e in el:
                    s += ' (%d %d)' % (-e, ne)
                print s
                self.log(s)

            N = min(self.npe, len(all))
            s = '\nNprobe errors by host and error (top %d shown):\n\t%s %s %s\n' % (N, 'Errno.', 'Host'.center(18), 'No.'.center(6))
            print s
            self.log(s)
                
            for n, e, h in all[:self.npe]:
                s = '\t%4d %18s %6d' % (-e, ntoa(h), n)
                print s
                self.log(s)
            print

        print 'Dmux tree written in', self.odir

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

    def getopts(self, opts):

        
        dmuxer_opts = []

        try:
            optlist, args = getopt.getopt(opts, '', ['accept_nosyns', 'timeout=', 'float_timestamps', 'nprobe-errors'])
        except getopt.error, s:
            #print 'ouch'
            self.exception('FlowDmuxerDmux getopt() ' + str(s))

        for opt, par in optlist:

            if opt == '--accept_nosyns':
                self.accept_nosyns = 1

            elif opt == '--timeout':
                if self.npe:
                    raise FlowDmuxerException('Timeout specified with Nprobe error dump input')
                for unit, mult in (('us', 1), ('ms', 1e3), ('s', 1e6), ('m', 60*1e6), ('h', 60*60*1e6)):
                    pos = par.find(unit)
                    if pos >=0:
                        break
                try:
                    if pos > 0:
                        self.timeout = long(par[:pos])*long(mult)
                    else:
                        self.timeout = long(par)
                except ValueError, s:
                    self.exception('getopt() ' + str(s))
                self.ttimeo = TimeoDLL()
            else:
                if par:
                    eq = '='
                else:
                    eq = ''
                    dmuxer_opts.append(opt+eq+par)
                
        FlowDmuxer.getopts(self, dmuxer_opts)
        
############################################################################

    def make_path(self, t, ts):

        #
        # Call one of the fn.s below (as determined by protd dir lookup)
        # to construct the appropriate dir name for a flow
        #

        ph =  self.protd[t[0]]

        return ph[1](ph[0], t, ts)

    #
    # create protocol-dependant dmux tree dir names 
    #
    

    def tcpudp_path(self, dir, t, ts):

        if self.npe:
            dir = '%s/%d' % (dir, -t[2])
            p = '%s/%s' % (dir, ntoa(t[1]))
        else:
            dir = '%s/%s/%s' % (dir, ntoa(t[1]), ntoa(t[2]))
            p = '%s/%u.%06u-%u-%u' % (dir, ts/1000000, ts%1000000, t[3], t[4])
            
        if not os.path.isdir(dir):
            os.makedirs(dir)
            
        return p

    def ip_fragpath(self, dir, t, ts):

        dir = '%s/%s/%d' % (dir, ntoa(t[1]), t[2])
        if not os.path.isdir(dir):
            os.makedirs(dir)
        p = '%s/%u.%06u' % (dir, ts/1000000, ts%1000000)
        return p

    def ip_path(self, dir, t, ts):

        if self.npe:
            dir = '%s/%d' % (dir, -t[2])
            p = '%s/%s' % (dir, ntoa(t[1]))
        else:
            dir = '%s/%s/%s' % (dir, ntoa(t[1]), ntoa(t[2]))
            p = '%s/%u.%06u' % (dir, ts/1000000, ts%1000000)
            
        if not os.path.isdir(dir):
            os.makedirs(dir)
            
        return p

    def arp_path(self, dir, t, ts):
        dir = '%s/%s/%s' % (dir, entoa(t[1]), entoa(t[2]))
        #print dir
        if not os.path.isdir(dir):
            os.makedirs(dir)
        p = '%s/%u.%u' % (dir, ts/1000000, ts%1000000)
        return p
            

    def nullpath(dir, t, ts):

        return None
        
############################################################################

    def tidy(self):

        # dump any timed out flows
        if len(self.ttimeo):
            ts = self.ttimeo.tail().last
            self.timeo(ts)
            self.active_at_end = len(self.ttimeo)
            remainder = self.ttimeo.items()
        else:
            self.active_at_end = len(self.flowdir)
            remainder = self.flowdir.itervalues()
            

        freeable = 0
        #for o in self.ttimeo.items():
        for o in remainder:
            if o.file:
                freeable = 1
                break

        if len(self.ttimeo):
            ts = self.tm + self.timeout + 1
            self.timeo(ts)
            self.to -= self.active_at_end
        else:
            self.dump_flowdir()

        if freeable:
            assert len(self.pending_dumps) == 0
        else:
            for flow in self.pending_dumps:
                self.dump_flow(flow)

        self.pending_dumps = []

        try:
            assert len(self.flowdir) == len(self.ttimeo)
            assert not self.flowdir
            assert self.ttimeo.isempty()
            assert not self.pending_dumps
            assert self.nopen == 0
            assert self.dpkts == self.ipkts - self.nosyn -self.truncated
        except AssertionError, s:
            tbs = traceback.format_list(traceback.extract_tb(sys.exc_info()[2]))
            self.exception('tidy - assertion error\n%s' % (tbs[0]))


        fragsdir = self.protd[(ETHERTYPE_IP, 'FRAGS')][0]
        frags = 0
        for root, dirs, files in walk(fragsdir, topdown=0):
            if not (files or dirs):
                os.rmdir(root)
            else:
                frags += len(files)

        if frags:
            str = 'Caution - unresolved IP fragments (%d)' % (frags)
            print str
            self.log(str)


        resdir = self.odir
        for root, dirs, files in walk(resdir, topdown=0):
            if not (files or dirs):
                os.rmdir(root)

            
            

        self.logf.flush()

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

    def dump_held(self, flow, f):

        pcap = self.pcap
        seek = pcap.fseek
        next = self.pcap_next
        dump = pcap.alt_dump

        osave = pcap.ftell()

        for off in flow.held:
            try:
                #print 'off', off
                seek(off)
                next()
                dump(f)
                self.dpkts += 1
            except (TypeError, IOError), s:
                self.exception('dump_held() loop ' + str(s))

        seek(osave)
        flow.held = []
        

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

    def dump_flow(self, flow):

        try:
            if flow.written:
                f = open(flow.path, 'a')
            else:
                f = self.pcap.alt_dump_open(flow.path)
                flow.written = 1
        except IOError, s:
            self.exception(str(s))

        self.dump_held(flow, f)
        
        f.close()

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

    def dump_flowdir(self): 

        freed = 0
        dump = self.dump_flow
        dir = self.flowdir
        pending = self.pending_dumps

        npe = self.npe
        if npe:
            ed = self.errd

        for flow in dir.itervalues():

            try:
                if flow.dumped:
                    continue
            except AttributeError:
                flow.dumped = 1

            if npe:
                fa = flow.atuple
                ed[(fa[1], fa[2])] = flow.pn + 1
            
            if flow.file:
                flow.file.close()
                flow.file = None
                self.nopen -= 1
                freed += 1
            elif flow.held:
                if freed > 0:
                    dump(flow)
                else:
                    pending.append(flow)

        if freed:
            for fl in pending:
                dump(fl)
            pending = []

        self.flowdir = {}

        return freed

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

    def timeo(self, ts): 

        tto = self.ttimeo
        empty = tto.isempty
        head = tto.head
        rem = tto.rem
        to = self.timeout
        freed = 0
        dump = self.dump_flow
        dir = self.flowdir
        pending = self.pending_dumps

        while not empty():

            flow = head()
            if flow.last > ts - to:
                break
            
            rem(flow)
            del dir[flow.atuple]
            for ref in flow.alts:
                del dir[ref]
            self.to += 1
            
            if flow.file:
                flow.file.close()
                flow.file = None
                self.nopen -= 1
                freed += 1
            elif flow.held:
                if freed > 0:
                    dump(flow)
                else:
                    pending.append(flow)

        if freed:
            for fl in pending:
                dump(fl)
            pending = []

        return freed
                

        
        
############################################################################
            
   
    def timesubst(self, f1, f2):

        if f1.atuple[0][1] == 'FRAGS':
            fragpath = f1.path
            flowpath = f2.path
        elif f2.atuple[0][1] == 'FRAGS':
            fragpath = f2.path
            flowpath = f1.path
        else:
            raise FlowDmuxerException('timesubst(): can\'t identify fragments file')

        m = re.match('.*/IP/FRAGS/(\d+\.){3}\d+/\d+/(?P<ts>\d+\.\d+)', fragpath)
        if not m:
            raise FlowDmuxerException('timesubst() - couldn\'t extract timestamp from frags file path')
        frag_ts = m.group('ts')

        m = re.match('.*/IP/(6|(17))/(\d+\.){3}\d+/(\d+\.){3}\d+/(?P<ts>\d+\.\d+)-\d+-\d+', flowpath)
        if not m:
            raise FlowDmuxerException('timesubst() - couldn\'t extract timestamp from flow file path')
        flow_ts = m.group('ts')

        fgts = float(frag_ts)
        flts = float(flow_ts)

        if flts > fgts:
            return re.sub(flow_ts, frag_ts, flowpath)
        else:
            return flowpath 
                

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