np_tcp.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 #
#                                                                             #
###############################################################################


#
# Implement TCP state as observed from mid-point packet collection
#

##############################################################################
##############################################################################
import sys
#from nprobe import *
from nprobe import TRANS_INCOMPLETE, TRANS_FINISHED, SERVER, CLIENT
from np_packet_markers import *
#from np_obnode import *
from  np_longutil import ul2l
from np_seq import  SEQ_MAX, seq_add, seq_sub, seq_diff, seq_gt, seq_gte, \
     seq_lt, seq_lte
from print_col import *
from TCP_Imp import TCP_Imps, MAX_IWF
#from np_Connstats import Connstats
#from np_tcpstats import TCPStats
from minmax import BIGNUMBER, MIN, MAX, MAX3, MAX4

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

#
# TCP flags
#

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

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

#
# TCP state connection_state values
#

SYN_RECD = 0x1
SYN_SENT = 0x2
SYN_ACK_RECD = 0x4
SYN_ACK_SENT = 0x8

FIN_RECD = 0x10
FIN_SENT = 0x20
FIN_ACK_RECD = 0x40
FIN_ACK_SENT = 0x80

CONNECTED = 0x100

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

#
# All pkts run through machine?
#

PASSES_INCOMPLETE = 0
PASSES_COMPLETE = 1

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

CPKTS_THRESH = 1
SPKTS_THRESH = 1

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


def trig_str(p):

    r = 'trig='
    c = p.trig

    if c == 0: r += 'none '
    elif (c & TRIG_RTT_INVALID): r += 'rtt inval '
    elif (c & TRIG_CAUSE_LATER): r += 'cause later '
    elif (c & TRIG_ACK): r += 'ACK '
    elif (c & TRIG_SYNACK): r += 'SYNACK '
    elif (c & TRIG_FINACK): r += 'FINACK '
    elif (c & TRIG_DEL_ACK): r += 'del ACK '
    elif (c & TRIG_REL): r += 'ACK-> '
    elif (c & TRIG_PREV): r += 'prev-> '
    elif (c & TRIG_RESP_DEL): r += 'req-> '
    elif (c & TRIG_RESP_Q): r += 'reply_q-> '
    elif (c & TRIG_RESP_FIRST): r += 'first '
    elif (c & TRIG_RESP_REL): r += 'ACK-> '
    elif (c & TRIG_REQ_DEL): r += 'reply-> '
    elif (c & TRIG_REQ_Q): r += 'req_q-> '
    elif (c & TRIG_REQ_FIRST): r += 'first '
    elif (c & TRIG_REQ_REL): r += 'ACK-> '
    elif (c & TRIG_SERV_DEL): r += 'serv del '
    elif (c & TRIG_CLI_DEL): r += 'cli del '
    else:
	r += 'unknown (%x) ' % (c)

    r += '(%x) ' % (c)
    r += 'lag='
    if p.lag != None:
	r += '%.3f' % (p.lag/1000.0)
    else:
	r += 'none'

    return r


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

#
# Identifiers for transaction events - value order counts for sorting
#

REQSTART = 0x1
REQEND = 0x2
REPSTART = 0x4
REPEND = 0x8

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

#
# Sequence range released by an ACK arrival
#

class seq_Release:

    def __init__(self, start, end, pkt):

	self.start = start
	self.end = end
	self.range = seq_diff(end, start)
	self.ack_pkt = pkt
	self.pkts_released = 0
	self.highseq = start
	self.rtt = 0

	#print '#%d releasing %d - %d' % (pkt.indx, start, end)


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

    # How does a segment compare with a release
    # - return start and end differences 
    def seq_cmp(self, start, end):

	return (seq_diff(self.highseq, start), seq_diff(self.end, end))
	
	    
		     
##############################################################################
##############################################################################

#
# Significant transaction timings
#

class Trans_Tms_t:

    def __init__(self, trans, i, open):

	self.rqs = ul2l(trans.http_reqstart_us()) + open
	self.rqe = ul2l(trans.http_reqend_us()) + open
	self.rps = ul2l(trans.http_repstart_us()) + open
	self.rpe = ul2l(trans.http_repend_us()) + open
        
	self.obsz = trans.http_obj_bytes()

	self.sstatus = trans.http_serv_status()
	self.cstatus = trans.http_cli_status()
	self.indx = i

        #
        # XXX TODO - why does rep end for non-body responses, eg code 304
        #  show as following packet time on pers conns
        # - s/be same as start time
        if (self.sstatus & TRANS_INCOMPLETE) \
           and (self.sstatus & TRANS_FINISHED) \
           and self.obsz == 0:
            self.rpe = self.rps

	#if self.sstatus & TRANS_INCOMPLETE:
	    #self.rpe = -1 # make sure never matched

    def __str__(self):

        return 'Trans_Tms[%d] = reqs %.3f reqe %.3f reps %.3f repe %.3f sz %d' % (self.indx, self.rqs/1000.0, self.rqe/1000.0, self.rps/1000.0, self.rpe/1000.0, self.obsz)

    def printself(self):

	print 'Trans_Tms[%d] = reqs %.3f reqe %.3f reps %.3f repe %.3f sz %d' % (self.indx, self.rqs/1000.0, self.rqe/1000.0, self.rps/1000.0, self.rpe/1000.0, self.obsz)
#

class Trans_Tms_T(Trans_Tms_t):

    def __init__(self, Trans, i):

	self.rqs = Trans.reqstart
	self.rqe = Trans.reqend
	self.rps = Trans.repstart
	self.rpe = Trans.repend
        
	self.obsz = Trans.nbytes

	self.sstatus = Trans.sstatus
	self.cstatus = Trans.cstatus
	self.indx = i
        
        if (self.sstatus & TRANS_INCOMPLETE) \
           and (self.sstatus & TRANS_FINISHED) \
           and self.obsz == 0:
            self.rpe = self.rps

    

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

#
# Set up TCP state and run through packets
#

class TCP_Machine:

    def __init__(self, conn, imps, logwrite, trace=0,
                 cpkts_thresh=CPKTS_THRESH, spkts_thresh=SPKTS_THRESH):

        if conn.sdpkts < spkts_thresh or conn.cdpkts < cpkts_thresh:
            raise TCPModelPkts 
	
	self.trace = trace
	if trace:
	    conn.printself()
	#else:
	    #print 'Conn #%d' % (conn.id)

	try:
	    print conn.notes
	except AttributeError:
	    pass

	#print logwrite
	self.logwrite = logwrite

	if logwrite == None:
	    self.logfun = self.null_f
	else:
	    self.logfun = self.buffer_log

	# Only write log messages to file when found final solution
	self.log_mssgs = []
	self.self_mssgs = []

	self.conn = conn
 	self.imps = imps
	self.made_imps = []

	self.sauto = 0
	self.cauto = 0

	self.valid = 1

	#
	# Get any info we have on usage of the connection
	#
        self.ttms = conn.ttms

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

    def null_f(self, arg):

        pass


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

    #
    # suggested imps already tried? - return 1 if so
    #
    def imps_tried(self, simp, cimp):
	#print 'tried %s %s?' % (simp.name, cimp.name)
	#print simp,; print cimp
	for imps in self.imps_attempted:
	    #print '\t%s %s' % (imps[0].name, imps[1].name)
	    #print imps[0],; print imps[1]
	    if simp == imps[0] and cimp == imps[1]:
		#print 'yes'
		return 1

	#print 'no'
	return 0


##############################################################################
    #
    #  Run the packets through the machine
    #
    def run_pkts(self):

	sstate = self.sstate
	cstate = self.cstate

	i = 0

	for p in self.conn.pktlist:
	    p.indx = i
	    i = i+1
	    p.markers = 0
	    p.trig = 0
	    p.remark = 0
	    p.rel = None
	    p.highrel = None
	    p.lag = None
	    p.trig = 0
	    p.prtt = p.trtt = p.delay = 0
	    p.tstr = ''
	    #print '#%d' % (i),

	    if p.dir == SERVER:
		depstate = self.sstate
		arrstate = self.cstate
	    else:
		depstate = self.cstate
		arrstate = self.sstate

	    try:
		depstate.departure(p)
		arrstate.arrival(p)
	    except TCPModelFailed, e:
		# goint to retry with new model
		state = e.state
		self.imps_attempted[-1][2] = PASSES_INCOMPLETE
		break
	    except TCPNoModel:
		# can't do - log it
		self.writelog()
		raise

	    self.lag2prtt(p)

	

##############################################################################
    #
    #  Run the connection through the machine - attemp to find best
    #   implementation fit if running on auto
    #
    #  - return any good implementations found
    #
    def do_pkts(self, recursing = 0):

	trace = 1
	conn = self.conn
	#conn.make_clusters()


	if not recursing:

	    if trace and self.trace:
		print 'Starting with imps %s %s' % (self.imps.simp_curr.name, 
						    self.imps.cimp_curr.name)
						
	    if self.imps.simp_curr.name == 'Auto':
		self.sauto = 1
		self.simp = self.imps.simp_default
		self.imps.simp_curr = self.simp
	    else:
		self.simp = self.imps.simp_curr
	    if self.imps.cimp_curr.name == 'Auto':
		self.cauto = 1
		self.cimp = self.imps.cimp_default
		self.imps.cimp_curr = self.cimp
	    else:
		self.cimp = self.imps.cimp_curr

	    #entry fields are [simp, cimp, complete, confidence]
	self.imps_attempted = []

	simp = self.simp
	cimp = self.cimp
	ret_imps = [None, None]
	
	while 1:
	    if trace and self.trace:
		inform('trying S imp %s C imp %s' % (simp.make_sig(), cimp.make_sig()))
	    if not recursing and self.imps_tried(simp, cimp):
		# looping - run out of ideas for improved fit
		if trace and self.trace:
		    inform('- already tried')
		break
## 	    if simp == self.imps.give_up or cimp == self.imps.give_up:
## 		break
	    self.log_mssgs = [] # clear
	    self.imps_attempted.append([simp, cimp, PASSES_COMPLETE, 0])
	    sstate = self.sstate = conn.sstate = TCP_State(self, conn, SERVER, simp, trace=self.trace)
	    cstate = self.cstate = conn.cstate = TCP_State(self, conn, CLIENT, cimp, conn.sstate, trace=self.trace)
	    sstate.new_imp = None
	    cstate.new_imp = None

	    self.run_pkts()
	    if trace and self.trace:
		print

	    
	    if sstate.new_imp != None:
		if self.sauto:
		    simp = self.imps.have_equiv(sstate.new_imp)
		    simp = simp.already_have(self.made_imps)
		    if trace and self.trace:
			inform('S new imp \'%s\'' % (simp.make_sig()))
		    continue
	    else:
		self.imps_attempted[-1][3] = self.imps_attempted[-1][3] + 50
		ret_imps[0] = sstate.imp
		

	    if cstate.new_imp != None:
		if self.cauto:
		    cimp = self.imps.have_equiv(cstate.new_imp)
		    cimp = cimp.already_have(self.made_imps)
		    if trace and self.trace:
			inform('C new imp \'%s\'' % (cimp.make_sig()))
		    continue
	    else:
		self.imps_attempted[-1][3] = self.imps_attempted[-1][3] + 50
		ret_imps[1] = cstate.imp

	    break


	if not recursing:
	    best_imp = self.imps_attempted[0]
	    for imp in self.imps_attempted:
		if imp[3] > best_imp[3]:
		    best_imp = imp
		
	    if trace and self.trace:    
		print 'Conn #%d TCP implementations tried:' % (conn.id)
	    for imp in self.imps_attempted:
		if trace and self.trace:
		    print '\tServer %s Client %s confidence %d' % (imp[0].make_sig(), imp[1].make_sig(), imp[3]),
		    if imp == best_imp:
			print ' - accepted',
			if imp[2] == PASSES_COMPLETE:
			    print ' - complete'
		    print

	    if best_imp[2] != PASSES_COMPLETE:
		# run it right through
		simp = self.simp = best_imp[0]
		cimp = self.cimp = best_imp[1]
		ret_imps = [simp, cimp]
		self.sauto = self.cauto = 0
		if trace and self.trace:
		    print 'redoing state with best fit %s %s' % (simp.make_sig(), cimp.make_sig())
		self.do_pkts(recursing = 1)

	    #for d in self.sdels:
		#print '#%4d %.3f %.3f' % (self.conn.id, d[0]/1000.0, d[1]/1000.0)
	    #self.do_rtts()
	    self.do_timing()
	    # write out any messages from final model
	    self.writelog()

	return ret_imps[0], ret_imps[1]