np_tcp.py

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

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

    def writelog(self):

	#print 'writelog'
	if self.logwrite != None:
	    #print 'yes - %d' % (len(self.log_mssgs))
	    if len(self.log_mssgs):
		for m in self.log_mssgs:
		    #print 'LOG ' + m
		    self.logwrite(m)
	    else:
		pass
		#self.logwrite('#%d Zero messages\n' % self.conn.id)
	#raw_input('...')

	    for m in self.self_mssgs:
		self.logwrite(m)

	self.log_mssgs = []


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

    def buffer_log(self, s):
	
	self.log_mssgs.append(s)
	#print 'LOG buffer %s len now %d' % (s, len(self.log_mssgs))


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

    def do_timing(self):

	#stats = self.cstats = self.conn.stats = Connstats(self.conn)
	#stats = self.cstats = self.conn.stats = TCPStats(self.conn)
	#sstate = self.sstate
	#cstate = self.cstate
        return
	for p in self.conn.pktlist:
	    trig = p.trig
	    if p.dir == SERVER:
		stats.allsprtts.append((p.tm, p.prtt))
		if trig & TRIG_RTT:
		    # genuine RTT
		    stats.sprtts.append((p.tm, p.prtt))
		if trig & TRIG_RESP_FIRST:
                    # initial server delay
 		    stats.slat = (p.tm, int(p.delay))
		if trig & TRIG_RESP_DEL:
                    # subsequent response delay (HTTP-P)
 		    stats.splat.append((p.tm, int(p.delay)))
	    else:
 		stats.cprtts.append((p.tm, p.prtt))

        # apparent rtts
	t =  self.ttms[0]
	if t:
	    stats.aslat = (t.rps, t.rps-t.rqe)
	    for t in self.ttms[1:]:
		stats.asplat.append((t.rps, t.rps-t.rqe))
	


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


    def interp(self, x1, y1, x2, y2, cut):

	if x2 == x1:
	    return (y1)

	a = x2-cut
	b = y1-y2
	c = x2-x1
	y = (a*b)/c + y2

	return y


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


    def rtt_patch(self, start, r1, r2, curr):

	trace = 1
	trace = trace and self.trace

	if trace:
	    print 'rtt patch start=#%d ' % (start.indx),
	    if r1 != None:
		print 'r1=#%d prtt=%.3f ' % (r1.indx, r1.prtt/1000.0),
	    else:
		print 'r1 None ',
	    if r2 != None:
		print 'r2=#%d prtt=%.3f ' % (r2.indx, r2.prtt/1000.0),
	    else:
		print 'r2 None ',
	    if curr != None:
		print 'curr=#%d prtt=%.3f' % (curr.indx, curr.prtt/1000.0)
	    else:
		print 'curr None'
	
	if r1 == None:
	    # at most one to go on
	    if r2 == None:
		# nothing to go on
		base = self.tmp
	    else:
		# one to go onn
		base = r2.prtt
	else:
	    # got both to go on - interpolate in loop
	    base = None

	p = start
	while 1:
## 	    print p
## 	    print '#%d-%s' % (p.indx, trig_str(p))
	    if p == curr:
		break 
	    if (p.markers & P_FIRST):
		p = p.nxt
		continue
	    if trace:
		print '\tpatching #%d %s' % (p.indx, trig_str(p))
	    if base == None:
		prtt = self.interp(r1.tm, r1.prtt, r2.tm, r2.prtt, p.tm)
		#print 'interp (%.3f/%.3f -> %.3f ->  %.3f/%.3f) = %.3f' % (r1.tm/1000.0, r1.prtt/1000.0, p.tm/1000.0, r2.tm/1000.0, r2.prtt/1000.0, prtt/1000.0)
	    else:
		prtt = base
	    if p.trig & TRIG_RTT_DEL:
		# there's a delay inclusive of apparent prtt
		#print 'lag=',
		#print p.lag,
		#print ' prtt=',
		#print prtt
		p.delay = p.lag - prtt
	    elif p.trig & TRIG_DEL:
		# there's a delay not inclusive of apparent prtt 
		p.delay = p.lag
	    else:
		p.delay = 0

	    if p.delay < 0:
		str = 'TCPConn #%d NEGATIVE DELAY: #%d %.3f' % (self.conn.id, 
			       p.indx, p.delay/1000.0)
		if trace:
		    whoops(str)
		self.logfun(str)

	    p.prtt = prtt
	    if trace:
		print '\tpatched #%d now prtt %.3f delay %.3f %s' % (p.indx, 
		       p.prtt/1000.0, p.delay/1000.0, trig_str(p))
	    p = p.nxt
	    if p.nxt == p:
		# last pkt - exit
		break
## 	    if trace:
## 		print 'next = #%d' % (p.indx)

	    


##############################################################################
	    
	    
    def lag2prtt(self, p):

	trace = 1

	if p.dir == SERVER:
	    state = self.sstate
	else:
	    state = self.cstate

	if trace and self.trace:
	    print '%s#%d lag2prtt - %s' % (state.nm, p.indx, 
						trig_str(p))

	last = state.last_valid_prtt
	if p.trig & TRIG_RTT:
	    p.prtt = p.lag
	    if p != state.first:
		# have a valid partial RTT
		if last != None:
		    # already seen one
		    self.rtt_patch(last.nxt, last, p, p)
		else:
		    # this is the first
		    self.rtt_patch(state.first, None, p, p)
	    state.last_valid_prtt = p
		
	elif p.nxt == p and p != state.first:
	    # last packet
	    if last != None:
		# patch on basis of last valid partial RTT
 		self.rtt_patch(last.nxt, None, last, p)
		# and this one
 		self.rtt_patch(p, None, last, None)
	    else:
		# nothing to go on 
		self.tmp = state.find_min_prtt()
		self.rtt_patch(state.first, None, None, p)
		# and this one
		self.rtt_patch(p, None, None, None) 
		
		     
##############################################################################
##############################################################################

# Modelling failed exception
class TCPModelFailed:

    def __init__(self, state):
	self.state = state
##############################################################################

# Can't model exception
class TCPNoModel:

    pass
##############################################################################

# Can't model exception
class TCPModelPkts:

    pass
##############################################################################

# Can't model exception
class TCPModelNoTrans:

    pass
		     
##############################################################################
##############################################################################

class TCP_State:

    def __init__(self, mach, conn, dir, imp, reverse_state = None, trace=None):

	self.mach = mach
	self.dir = dir
	self.connid = conn.id
	self.trace = trace

	if dir == SERVER:
	    self.nm = 'S'
	    self.ndpkts = conn.sdpkts 
	    if imp.mss == None:
		self.segsz = conn.cmss # mss that can send
	    else:
		self.segsz = imp.mss
	    self.pl = conn.sbytes
	else:
	    self.nm = 'C'
	    self.ndpkts = conn.cdpkts
	    if imp.mss == None:
		self.segsz = conn.smss
	    else:
		self.segsz = imp.mss
	    self.pl = conn.cbytes

	#print '%s mss = %d' % (self.nm, self.segsz)

	self.imp = imp
	self.new_imp = None

	if reverse_state != None:
	    self.reverse = reverse_state
	    reverse_state.reverse = self

	self.IW = None
	# Initially set cwnd high - adjust when initial window known
	self.cwnd = MAX_IWF*self.segsz
	self.ssthresh = self.imp.ssthresh

	#print 'Imp is %s, IW_fact = %d' % (self.imp.name, self.imp.iw_fact)
	
	self.first = None

	self.conn_state = 0
	self.connected = 0

	self.sstart = 1
	
	self.maxseg = 0
	self.segiffy = 0
	self.segiffy_szs = {}

	self.dep_segs = 0
	self.dep_octs = 0

	self.snd_highack = -1L
	self.rcv_highack = -1L
	self.snd_nxt = -1L
	self.rcv_nxt = -1L

	self.rcv_acks = []
	self.snd_segs = []
	self.snd_synacks = []
	self.rcv_segs = []
	
	self.rcv_segheld = []
	self.snd_segheld = []
	self.too_high_acks = []
	self.too_low_acks = []

	self.releases = []

	self.rcv_olap_p = 0
	self.rcv_olap_o = 0
	self.rcv_rtmts_p = 0
	self.rcv_rtmts_o = 0

	self.snd_olap_p = 0
	self.snd_olap_o = 0
	self.snd_rtmts_p = 0
	self.snd_rtmts_o = 0

	self.dep_synps = []
	self.arr_synackps = []

	self.dup_dep_syns = 0
	self.dup_arr_syns = 0
	self.dup_dep_synacks = 0
	self.dup_arr_synacks = 0

	self.dup_syn_delay = 0
	self.dup_synack_delay = 0

	self.synunack = 0 # outstanding SYN
	self.synack = None # current
	self.finack = None # current

	self.lastp = None # last pkt to departure
	self.last_valid_prtt = None

	# current transaction times
	self.ttms = mach.ttms[0]
	self.reqi = self.repi = 0 #indices into trans times list


	return

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

    #
    # If no valid prtt found base calculations on minimum seen
    #
    def find_min_prtt(self):

	p = self.first
	min_rtt = BIGNUMBER
	while 1:
	    if p.prtt and p.prtt < min_rtt:
		min_rtt = p.prtt
	    if p.nxt == p:
		break
	    p = p.nxt

	if min_rtt == BIGNUMBER:
	    min_rtt = 0

	return min_rtt
	    

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

    def give_up(self):

	self.new_imp = None
	raise TCPNoModel()
	    

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

    def retry(self, new_imp):

	# Don't retry if not on auto
	if (self.dir == SERVER and self.mach.sauto) or (self.dir == CLIENT and self.mach.cauto):
	    self.new_imp = new_imp
	    raise TCPModelFailed(self)
	else:
	    del(new_imp)
	    return
	    

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

    def dep_syn_pkt(self, p):

	trace = 1

	if self.dir == SERVER and not (self.reverse.conn_state & SYN_SENT):
	    str = 'TCPConn #%d SERVER ORIGINATED CONN: %s#%d' % (self.connid, self.nm,
							 p.indx)
	    if trace and self.trace:
		whoops(str)
	    self.mach.logfun(str)
	    self.give_up()
	    

	# duplicate?
	if self.conn_state & SYN_SENT:
	    if p.seq == self.iss:
		# duplicate
		#print '%s#%d DUP DEP SYN' % (self.nm, p.indx)
		p.remark =1
		self.dup_dep_syns = self.dup_dep_syns + 1
		if self.conn_state & SYN_ACK_RECD:
		    # seen the ACK - dep q popped
		    pass
		else:
		    self.snd_segs.pop(0)
		# assess delay
		self.dup_syn_delay = p.tm - self.dep_synps[0].tm

	self.conn_state = self.conn_state | SYN_SENT
	self.snd_segs.append(p)

	self.iss = p.seq
	self.rcv_highack = self.iss
	self.snd_nxt = p.end = seq_add(self.iss, 1)
	self.synack = self.snd_nxt
	self.rcv_wnd = p.window
	self.release = self.iss

	self.dep_synps.append(p)
	self.synunack = 1

## 	# dummy
## 	r = seq_Release(p.seq, self.snd_nxt, p)
## 	r.rtt = 0
## 	p.rel = r

## 	p.nrtt = 0

 	self.release = self.snd_nxt

## 	## # is this a reverse syn - if so get nrtt for this packet
## 	if self.conn_state:
## 	    p.nrtt = p.tm - self.reverse.dep_synps[-1].tm
## 	    # 

## 	if self.conn_state & SYN_RECD:
## 	    # this is a reverse
## 	    p.prtt = p.tm - self.reverse.dep_synps[-1].tm

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

    def dep_ack_pkt(self, p):

	trace = 1
	ack = p.ack

	# looks reasonable?
	if seq_lt(ack, self.snd_highack):
	    str = 'TCPConn #%d LOW ACK departing: %s#%d - got  %d expected >= %d' % (self.connid, self.nm, p.indx, ack, self.snd_highack)
	    if trace and self.trace:
		whoops(str)
	    self.mach.logfun(str)
	    #raw_input('anything to continue..........\n')
	    return

	if self.conn_state & SYN_RECD and not self.conn_state & SYN_ACK_SENT:
	    if ack == seq_add(self.irs, 1):
		if len(self.rcv_segs) == 1 and self.rcv_segs[0].seq == self.irs:
		    self.conn_state =  self.conn_state | SYN_ACK_SENT
		    self.snd_synacks.append(p)
		    rtt =  p.tm - self.rcv_segs[0].tm
		    self.rcv_syn_rtt = rtt
		    
		    p.trig = TRIG_SYNACK
		    p.lag = p.tm - self.rcv_segs[0].tm
		    p.prtt = p.lag
		    #self.last_valid_prtt = p
		    #p.nrtt = rtt
		    self.rcv_segs.pop(0)
		    if self.reverse.conn_state & SYN_ACK_SENT:
			# catch the case of a departing seg riding on a SYNACK
			self.conn_state |= CONNECTED
			self.reverse.conn_state |= CONNECTED 
			#print '%s#%d Marking connected' % (self.nm, p.indx)
		    #print '%s#%d SYN_ACK_SENT rcv_syn_rtt %.3f' % (self.nm, p.indx, self.rcv_syn_rtt/1000.0)
	    else:
		p.remark = 1
		str = 'TCPConn #%d UNMATCHED SYNACK: %#d %s -  expected %d got %d' % (self.connid, p.indx, self.nm, seq_add(self.irs, 1), ack)
		if trace and self.trace: 
		    whoops(str)