np_obtree.py

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

#! /usr/bin/env python
###############################################################################
#                                                                             #
#   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 #
#                                                                             #
###############################################################################


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

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

import string
import glob
import os
import sys
from sys import argv
import getopt
from signal import *
import atexit
import types

import gc
from resource import getrusage, RUSAGE_SELF
import np_warnings

import Numeric
from traceback import print_stack

from print_col import upline, overprint, whoops

from nprobe import intoa_string, _free, malloc_report
from nprobe import tcp_hdrs, tcp_open
from nprobe import _inet_aton, tcp_open, tcp_hdrs, accept_conn, filter_help
from nprobe import REC_TCP_HTTP, REC_TCP_HTTP_OPEN, REC_TCP_HTTP_HDRS, \
     REC_UDP_DNS, OPEN_BIT, HDRS_BIT
from np_file_util import get_files
from np_http_util import allocate_http_reusable_objects, get_http_rec_and_trans
from np_ns_utils import NSLookup, get_ns_rec
from np_longutil import tv2l
from np_namecache import NameCache
from np_tfilter import TFilter
from np_filerec import FileRec
from np_ectrs import E_Counters
from np_statscollector import get_IPaddrs, NoIPAddrError
from np_treestats import TreeStats, NOB_THRESH

import scratch

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

def usage(scriptname):
    print "usage: " + scriptname + "rep-file-list"

    sys.exit(1)

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

def exitfun():
    
    from np_grabrec import grabobj
    if grabobj:
        grabobj.close()

    if namecache:
        namecache.file_cache()

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

def handle_sigint(n, f):
    
    print 'SIGINT'
    print_stack()
    exitfun()
    sys.exit(0)

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

def do_ns(lookups, connrec, nsrec, cachefn):
    
    rec = nsrec.ns_rec
    reqstate = nsrec.ns_rec.state
    if not rec.state & 0x3 == 0x3: # both request and response seen
	return

    cli = connrec.shost()

    lookup = NSLookup(cli, connrec, nsrec)
    
    if len(lookup.rrlist):
	# only interested if IP_ADDR or CNAME requests
        le = lookups.setdefault(cli, [])
        le.append(lookup)

	# append rr's to our name cache
	for rr in lookup.rrlist:
	    if rr.code == 1:
		cachefn(rr.addr, rr.req)
		#print 'adding'

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

#
# Null logging function (don't want particular examination to o'write logs
#

def write_null(arg):

    pass

    

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

#
# Aggregate cl list of clients wanted

def get_clis_wanted(s):
    clis = []
    for f in string.splitfields(s, ','):
	#print f
	#print '%d' % (string.atoi(f))
	clis.append(_inet_aton(f))

    return clis

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

def pre_pass_A(openfilelist, nrecs, spec, fspec, stats):

    def by_1(a, b):
        return b[1] - a[1]

    #
    # To get TCP records into
    #
    connrec, translist = allocate_http_reusable_objects()
    
    nread = 0
    done = 0
    do_del = 0
    rank = 0

    cyes = {}
    call = {}
    syes = {}
    sall = {}

    nconns = 0

    ectrs = stats.ectrs

    if spec == find_delays:
        why = 'finding potential delayed servers'
        do_del = 1
        stats.del_only = 1
    elif type(spec) == types.StringType:
        if spec.find('rank') == 0:
            rank = 1
            spec = spec.replace('rank', '')
            rend = spec[0].upper()
            try:
                nrank = int(spec[1:])
            except ValueError:
                print 'Pre_pass: don\'t understand int arg to', spec
                sys.exit(1)
            why = 'Ranking top %d %s' % (nrank, rend)
    else:
        print 'Pre_pass: don\'t understand spec', spec
        sys.exit(1)
        
        
        

    print 'Pre_pass:', why
    print
    
    for file in openfilelist:

        if done:
            break
        
        if len(openfilelist) > 1:
            print "\n~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~"
            print "File %s - %d records" % (file.fnm, file.counters.nrecords)
            print
            sys.stdout.flush()

        while not done:

            if nrecs and nread > nrecs:
                done = 1
                break

            rectype = file.next_type_rec(REC_TCP_HTTP)
                
            if rectype == -1: #EOF
                break

            elif rectype == REC_TCP_HTTP: #TCP/HTTP
                nread = nread + 1
                get_http_rec_and_trans(file, connrec, translist)

                if not accept_conn(connrec.flow_inner, fspec):
                    ectrs.cf_rej += 1
                    continue

                if do_del:
                    ca = connrec.shost()
                    sa = connrec.dhost()
                    call[ca] = 1
                    sall[sa] = 1
                    if spec(connrec):
                        syes[sa] = 1
                        cyes[ca] = 1
                elif rank:
                    nconns += 1
                    if rend == 'S':
                        sa = connrec.dhost()
                        try:
                            syes[sa] += 1
                        except KeyError:
                            syes[sa] = 1
                    elif rend == 'C':
                        ca = connrec.shost()
                        try:
                            syes[ca] += 1
                        except KeyError:
                            syes[ca] = 1
                    else:
                        print 'Pre_pass rank: unknown end', rend
                        sys.exit(1)
                else:
                    print 'Pre_pass: don\'t know spec', spec
                    sys.exit(1)

            else:
                file.advance()
                #print 'Pass1 - wrong record type', rectype
                #sys.exit(1)
                
            if not nread%100:
                overprint('Record %d' % (nread))

    overprint('%d relevant records' % (nread))

    if do_del:
        lsy = len(syes)
        lsa = len(sall)
        lcy = len(cyes)
        lca = len(call)
        s = 'Pre_pass - %d/%d Servers showing possible delays (%d/%d clients)    ' % (lsy, lsa, lcy, lca)
        print s
        stats.write_log(s)

    elif rank:
        nrconns = 0
        servs = syes.items()
        syes = {}
        servs.sort(by_1)
        if rend == 'S':
            d = syes
        else:
            d = cyes
        for s in servs[:nrank]:
            d[s[0]] = 1
            nrconns += s[1]
        s = 'Pre_pass - %d Servers seen (%d/%d conns)    ' % (len(servs),
                                                               nrconns, nconns)
        print s
        stats.write_log(s)

    #syes.update(cyes)
    return (syes, cyes)

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

def pass1(openfilelist, stats, nrecs, clis, servs, wantdicts, scwd,
          connid,
          startconn, fspec,
          tfilt, cachefn):

    #
    # A dictionary of HTTP TCP connections in the log {conn_id: filerec}
    #
    obdict = stats.obdict

    #
    # Dictionarie of servers seen - {addr: [conn ids]}
    #
    stats.cdict = clidict = {}
    stats.sdict = servdict = {}
    
    #
    # List of DNS IP_ADDR and CNAME lookups seen
    #
    stats.nslookups = lookups = {}
    
    #
    # Dictionary of tcp connections keyed by conn_id - used to collect headers
    #
    conn_ids = {}
    
    #ectrs = E_Counters()
    ectrs = stats.ectrs

    if wantdicts:
        if wantdicts[0]:
            swd = wantdicts[0]
        else:
            swd = None
        if wantdicts[1]:
            cwd = wantdicts[1]
        else:
            cwd = None
    #
    # To get TCP records into
    #
    tcp_open_rec = tcp_open()
    hdrs = tcp_hdrs(1)
    connrec, translist = allocate_http_reusable_objects()
    
    nread = 0
    done = 0

    print 'Pass 1: collating client records'
    print
    
    for file in openfilelist:

        if done:
            break
        file.reset()

        flen = os.stat(file.fnm)[6]
        
        if len(openfilelist) > 1:
            print "\n~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~"
            print "File %s - %d records" % (file.fnm, file.counters.nrecords)
            print
            sys.stdout.flush()

        while not done:

            if nrecs and nread > nrecs:
                break

            rectype = file.find_next_http_and_hdrs_or_ns_record()
            
            offset = file.curr_offset # get before read advances
            #print 'offset %d' % (offset)
            if offset >= flen:
                print 'ERROR bad offset %d - file size %d' % (offset, flen)
                
            if rectype == -1: #EOF
                break

            elif rectype == REC_TCP_HTTP_OPEN: # notification of TCP open
                tcp_open_rec.get_open_rec(file)
                ectrs.nconns += 1
                id = tcp_open_rec.conn_id
                saddr = tcp_open_rec.dhost()
                caddr = tcp_open_rec.shost()

                if id < startconn:
                    continue

                elif not accept_conn(tcp_open_rec.flow, fspec):
                    ectrs.cf_rej += 1
                    continue
                
                if connid:
                    if id == connid:
                        conn_ids[id] = []
                        print 'Got conn #%d' % (connid)
                        print
                    else:
                        ectrs.id_rej += 1
                
                elif clis != None:
                    caddr = tcp_open_rec.shost()
                    if clis.count(caddr):
                        conn_ids[id] = [] #set up for list of headers
                    else:
                        ectrs.id_rej += 1
                
                elif servs != None:
                    if servs.count(saddr):
                        conn_ids[id] = [] #set up for list of headers
                    else:
                        ectrs.id_rej += 1

                elif wantdicts:
                    if swd:
                        if swd.has_key(saddr):
                            if cwd:
                                if cwd.has_key(caddr):
                                    conn_ids[id] = []
                                else:
                                    ectrs.id_rej += 1
                            else:
                                conn_ids[id] = []
                        else:
                            ectrs.id_rej += 1
                    else:
                        conn_ids[id] = []

                elif scwd:
                    try:
                        w = scwd[saddr][caddr]
                        conn_ids[id] = []
                    except KeyError:
                        ectrs.id_rej += 1
                else:
                    conn_ids[id] = []