flowdmuxerdmux.py

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

                prot = (ETHERTYPE_IP, p1)
                start += iphl
            elif self.protd.has_key((ETHERTYPE_IP, p1)):
                prot = (ETHERTYPE_IP, p1)
            else:
                prot = (ETHERTYPE_IP, 'OTHER')
                
            return ((prot, sa, err, None, None), 0, start, None)

        else:
            #
            # No point in looking at others as no MAC-level addressing
            # supplied
            #
            return (None, None, None, None)
        
############################################################################

    def get_raw_ip_atuple(self, pkt):
        
        d = pkt[2]
        start = 0

        len = min(pkt[1], self.snaplen)
        
        if len < IP_HDR_LEN:
            self.truncated += 1
            return (None, None, None, None)
            
        sa = up('i', d[IP_SA_OFF:IP_SA_END])[0]
        da = up('i',d[IP_DA_OFF:IP_DA_END])[0]
        iphl = (ord(d[IP_HL_OFF]) & 0x0f) << 2
        ipo = up('!H',d[IP_IPOFF_OFF:IP_IPOFF_END])[0]
        ipoff = (ipo & 0x1fff) << 3
        if ipoff != 0:
            id = up('!H',d[IP_IPID_OFF:IP_IPID_END])[0]
            iplen = up('!H', d[IP_IPLEN_OFF:IP_IPLEN_END])[0]
            return (((ETHERTYPE_IP, 'FRAGS'), sa, id, None, None), 0, None, (da, id, ipoff, iplen-iphl, ipo & IP_MF))

        elif ipo & IP_MF:
            id = up('!H',d[IP_IPID_OFF:IP_IPID_END])[0]
            iplen = up('!H', d[IP_IPLEN_OFF:IP_IPLEN_END])[0]
            fragt = (da, id, ipoff, iplen-iphl, 1)
        else:
            fragt = None

        p1 = ord(d[IP_PROT_OFF])

        if p1 == IPPROTO_TCP or p1 == IPPROTO_UDP:
            prot = (ETHERTYPE_IP, p1)
            start = iphl
            if len < start + TCPUDP_HDR_LEN:
                self.truncated += 1
                return (None, None, None, None)
            sp = up('!H', d[start+TCPUDP_SP_OFF: start+TCPUDP_SP_END])[0]
            dp = up('!H', d[start+TCPUDP_DP_OFF: start+TCPUDP_DP_END])[0]
        elif self.protd.has_key((ETHERTYPE_IP, p1)):
            prot = (ETHERTYPE_IP, p1)
            sp = dp = None
        else:
            prot = (ETHERTYPE_IP, 'OTHER')
            sp = dp = None

        return ((prot, sa, da, sp, dp), p1 == IPPROTO_TCP, start, fragt)
        
############################################################################
            
   
    def process_pkt(self, pkt):

        self.ipkts += 1

        ts = pkt[3]
        ts = ts[0]*1000000 + ts[1]
        self.tm = ts
        d = pkt[2]

        atuple, tcp, start, fragt = self.get_atuple(pkt)
        if not atuple:
            return

        if tcp:
            flags = ord(d[start+TCP_FLAGS_OFF]) & 0x3f
            SYN = flags & TH_SYN
        else:
            SYN = 1

        flow, way = self.get_flow(ts, atuple, SYN, tcp)

        if fragt: # fragmented IP datagram
            self.frags += 1
            sa = atuple[0][1]
            da, id, ipoff, len, MF = fragt
            if not ipoff: # first frag
                at = ((ETHERTYPE_IP, 'FRAGS'), sa, id, None, None)
                if self.flowdir.has_key(at): # some trailing frags already got
                    fragflow = self.flowdir[at]
                    if fragflow != flow and ts - self.timeout < fragflow.last:
                        # not already merged and not timed out
                        # merge received frags
                        flow = self.pre_frags(self.flowdir[at], flow, ts)
                        self.nconns -= 1
                else:
                    # this is first to arrive - mark for trailing frags
                    self.flowdir[at] = flow
                    flow.alts.append(at)
                    
        if not flow:
            return

        flow.pn += 1
        if flow.last > ts:
            self.log('Unordered time stamp %s Pkt. No. %d' % (flow.path, flow.pn))
            if self.save_check_info:
                self.oots[hash(tuple(d))] = 1
        flow.last = ts
        f = flow.file

        if f:
            self.pcap.alt_dump(f)
            flow.written = 1
            self.dpkts += 1
        else:
            if self.stdin:
                f = self.reopen_flow(flow)
                self.pcap.alt_dump(f)
                self.dpkts += 1
            else:
                flow.held.append(pkt[0])

        if tcp:

            if flags & TH_FIN:
                flow.state |= FINS[way]
                flow.seqs[way] = up('!I', d[start+TCP_SEQ_OFF:start+TCP_SEQ_END])[0]

            if (flow.state & FINS[not way]) and (flags & TH_ACK):
                if (up('!I', d[start+TCP_ACK_OFF:start+TCP_ACK_END])[0] == flow.seqs[not way] + 1):
                    flow.state |= FINAS[not way]

            if flow.state & FINACKED:
                self.closed += 1
                self.end_flow(flow)
        
############################################################################

    #
    # Optional fn called at end to check consistency of dmux tree contents
    #
            
    def check(self, quiet=1):

        #
        # Find any log entries pertinent to a check fail
        #

        def get_log(ff, quiet=1):

            try:
                lf = open(self.logfile, 'r')
            except IOError, s:
                self.exception('Check get_log() ' + str(s))
            lines = []
            for l in lf.readlines():
                if l.find(ff) >= 0:
                    lines.append(l)

            if not quiet:
                if lines:
                    print 'Pertinent log entries:-'
                    for l in lines:
                        print l
                else:
                    print 'No pertinent log entries'

            return lines

        #
        # Translate directory path matching re groups to consistency check data
        #

        def base_ip_addrs_ports(m):

            return (aton(m.group('a1')), aton(m.group('a2')), int(m.group('tss'))*1000000 +  int(m.group('tsus')), int(m.group('p1')), int(m.group('p2')))

        def base_ip_addrs(m):

            return (aton(m.group('a1')), aton(m.group('a2')), int(m.group('tss'))*1000000 +  int(m.group('tsus')))

        def base_ip_frags(m):

            return (aton(m.group('a1')), int(m.group('a2')), int(m.group('tss'))*1000000 +  int(m.group('tsus')))

        def base_eth_addrs(m):

            return (eaton(m.group('a1')), eaton(m.group('a2')), int(m.group('tss'))*1000000 +  int(m.group('tsus')))

        def base_errs(m):
            
            return(int(m.group('err')), aton(m.group('a')))


        #
        # Flow-type consistency checking routines
        #

        def check_ip_addrs_ports((a1, a2, fts, p1, p2), pkt, ts, np):

            t = self.get_atuple(pkt)
            sa = t[0][1]
            da = t[0][2]
            fragt = t[3]

            if fragt:
                ipoff = fragt[2]
                da = fragt[0]
            else:
                ipoff = 0
                
            if ipoff == 0:
                sp = t[0][3]
                dp = t[0][4]
                
            try:
                if np == 0 and ipoff == 0:
                    assert ts == fts
                    assert sa == a1
                    assert da == a2
                    assert sp == p1
                    assert dp == p2
                elif ipoff == 0:
                    assert sa == a1 or sa == a2
                    assert da == a1 or da == a2
                    assert sp ==p1 or sp == p2
                    assert dp == p1 or dp == p2
                else:
                    assert sa == a1 or sa == a2
                    assert da == a1 or da == a2
                    
            except AssertionError:
                tbs = traceback.format_list(traceback.extract_tb(sys.exc_info()[2]))
                #self.exception('Check failed file %s packet No. %d\n%s' % (p, np, tbs[0]))
                print 'Check failed file %s packet No. %d\n%s' % (p, np, tbs[0])
                get_log(p)
                return 1
            return 0

        def check_ip_addrs((a1, a2, fts), pkt, ts, np):

            t = self.get_atuple(pkt)
            sa = t[0][1]
            da = t[0][2]

            try:
                if np == 0:
                    assert ts == fts
                    assert sa == a1
                    assert da == a2
                else:
                    assert sa == a1 or sa == a2
                    assert da == a1 or da == a2
            except AssertionError:
                tbs = traceback.format_list(traceback.extract_tb(sys.exc_info()[2]))
                print 'Check failed file %s packet No. %d\n%s' % (p, np, tbs[0])
                get_log(p)
                return 1
            return 0

        def check_ip_frags((a, id, fts), pkt, ts, np):

            t = self.get_atuple(pkt)
            sa = t[0][1]
            sid = t[3][1]

            try:
                assert sa == a
                assert sid == id
                if np == 0:
                    assert ts == fts
            except AssertionError:
                tbs = traceback.format_list(traceback.extract_tb(sys.exc_info()[2]))
                print 'Check failed file %s packet No. %d\n%s' % (p, np, tbs[0])
                get_log(p)
                return 1
            return 0

        def check_eth_addrs((a1, a2, fts), pkt, ts, np):


            d = pkt[2]
            da = d[ETH_DH_OFF:ETH_DH_END]
            sa = d[ETH_SH_OFF:ETH_SH_END]

            try:
                if np == 0:
                    assert ts == fts
                    assert sa == a1
                    assert da == a2
                else:
                    assert sa == a1 or sa == a2
                    assert da == a1 or da == a2
            except AssertionError:
                tbs = traceback.format_list(traceback.extract_tb(sys.exc_info()[2]))
                print 'Check failed file %s packet No. %d\n%s' % (p, np, tbs[0])
                get_log(p)
                raw_input('...')
                return 1
            return 0

        def check_errs((err, addr), pkt, ts, np):

            t = self.get_atuple(pkt)
            try:
                assert err == -t[0][2]
                assert addr == t[0][1]
            except AssertionError:
                tbs = traceback.format_list(traceback.extract_tb(sys.exc_info()[2]))
                print 'Check failed file %s packet No. %d\n%s' % (p, np, tbs[0])
                return 1
            return 0

        #
        # Main check fn starts here
        #

        print 'Checking dmux tree for consistency'

        nconns = 0
        npkts = 0
        nerrors = 0

        frag_nto = []

        #
        # re.s to match flow-type dependant directory paths
        #
        
        ip_addrs_ports = re.compile('.*/IP/(6|(17))/(?P<a1>(\d+\.){3}\d+)/(?P<a2>(\d+\.){3}\d+)/(?P<tss>\d+)\.(?P<tsus>\d+)-(?P<p1>\d+)-(?P<p2>\d+)')
        ip_addrs = re.compile('.*/IP/(\d+|OTHER)/(?P<a1>(\d+\.){3}\d+)/(?P<a2>(\d+\.){3}\d+)/(?P<tss>\d+)\.(?P<tsus>\d+)(?P<p1>)(?P<p2>)')
        ip_frags = re.compile('.*/IP/FRAGS/(?P<a1>(\d+\.){3}\d+)/(?P<a2>\d+)/(?P<tss>\d+)\.(?P<tsus>\d+)(?P<p1>)(?P<p2>)')
        
        eth_addrs = re.compile('.*/(ARP|REVARP|OTHER)/(?P<a1>([\dabcdef]+:){5}[\dabcdef]+)/(?P<a2>([\dabcdef]+:){5}[\dabcdef]+)/(?P<tss>\d+)\.(?P<tsus>\d+)(?P<p1>)(?P<p2>)')

        err_dirs = re.compile('.*/IP/(\d+|OTHER)/(?P<err>\d+)/(?P<a>(\d+\.){3}\d+)')

        #
        # Dictionary of directory->data mapping and packet-checking fns
        # keyed by path re match
        #

        if self.npe:
            checks = [(err_dirs, base_errs, check_errs)]
        else:
            checks = [
                (ip_addrs_ports, base_ip_addrs_ports, check_ip_addrs_ports),
                (ip_addrs, base_ip_addrs, check_ip_addrs),
                (ip_frags, base_ip_frags, check_ip_frags),
                (eth_addrs, base_eth_addrs, check_eth_addrs)
                ]

        for root, dirs, files in walk(self.odir):

            for f in files:

                if f == 'log':
                    continue

                ppc = pcap.py_pcap()
                next = ppc.next_o_tts
                nconns += 1
                nthis = 0
                p = os.path.join(root, f)
                
                m = None

                #
                # find the path-matching re, path translation and check
                # functions
                #
                for match, base, check in checks:
                    m = match.match(p)
                    if m:
                        break
                    
                if not m:
                    print 'File %s does not meet naming convention - not checked' % (p)
                    nerrors += 1
                    continue

                stuff = base(m)
                if self.npe:
                    fts = 0
                else:
                    fts = stuff[2]
                
                try:
                    ppc.open_offline(p)
                except (IOError, TypeError), s:
                    print
                    print p
                    self.exception(str(s))

                try:
                    ppc.set_filter("")
                except TypeError, s:
                    print
                    self.exception(str(s))

                np = -1

                #
                # Loop through packets and check'm
                #
                while 1:
                    try:
                        pkt = next()
                        ts = pkt[3]
                        npkts += 1
                        np += 1
                        nthis += 1
                    except EOFError:
                        break
                    except TypeError, s:
                        print
                        self.exception(str(s))

                    ts = (ts[0]*1000000 + ts[1])
                    nerrors += check(stuff, pkt, ts, np)
                    try:
                        assert ts >= fts
                        assert ts <= fts + self.timeout
                    except AssertionError:
                        #
                        # Collecting packet filter may have misordered packets
                        # in the dump - check this from the log before
                        # identifying an error
                        #
                        lines = get_log(p, quiet=quiet)
                        found = 0
                        for l in lines:
                            m2 = re.match('Unordered time stamp.*?No\. (?P<pno>\d+)', l)
                            if m2 and int(m2.group('pno')) == np:
                                found += 1
                                break
                        if not found:
                            # fallback - check on hash of pkt contents
                            if self.oots.has_key(hash(tuple(pkt[2]))):
                                found = 1
                        if not found:
                            tbs = traceback.format_list(traceback.extract_tb(sys.exc_info()[2]))
                            print 'Check failed  - time stamp ordering (current %u.%06u, last %u.%06u) file %s pkt. No. %d\n%s' % (ts/1000000, ts%1000000, fts/1000000, fts%1000000, p, np, tbs[0])
                            nerrors += 1

                    try:
                        assert ts <= fts + self.timeout
                    except AssertionError:
                        if is_frag(data):
                            frag_nto.append((p, np))
                        else:
                            tbs = traceback.format_list(traceback.extract_tb(sys.exc_info()[2]))
                            print 'Check failed  - timeout missed (current %u.%06u, last %u.%06u) file %s pkt. No. %d\n%s' % (ts/1000000, ts%1000000, fts/1000000, fts%1000000, p, np, tbs[0])
                            nerrors += 1
                        
                    fts = ts
                    

        try:
            assert npkts == self.ipkts - self.nosyn -self.truncated
            assert nconns == self.nconns + self.resumed
        except AssertionError:
            nerrors += 1
            print 'Check failed  - inconsistent No. of packets/flows (%d/%d)/(%d/%d)' % (npkts, self.ipkts, nconns, self.nconns + self.resumed)

        if not nerrors:
            print 'Check passed'
        else:
            print '%d fails' % (nerrors)

        if frag_nto:
            print 'Fragment timeouts not regarded as errors:'
            for f, pn in frag_nto:
                print f, 'packet No.', pn
            
            
        
############################################################################
############################################################################