np_tcp.py

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

	self.dep_segs = self.dep_segs + 1
	self.dep_octs = self.dep_octs + plen
	self.snd_segs.append(p)

	self.check_segsz(p)
	self.check_segseq(p)
	# Check the segment against pending releases
	self.cmp_release(p)

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

    def dep_fin_pkt(self, p):

	if self.conn_state & FIN_SENT:
	    if self.trace:
		inform('%s#%d DUP FIN dep' % (self.nm, p.indx))

	self.conn_state = self.conn_state | FIN_SENT
	p.end =  seq_add(self.snd_nxt, 1)
	self.finack = p.end
	self.snd_nxt = p.end
	self.snd_segs.append(p)

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

    def departure(self, p):

	#flags = ord(p.flags)
	flags = p.flags

	if self.first == None:
	    #print '%s#%d setting as first' % (self.nm, p.indx)
	    self.first = p
	    p.markers = p.markers | P_FIRST

	if flags & TH_SYN:
	    self.dep_syn_pkt(p)
	    self.lastp = p

	if flags & TH_ACK:
	    self.dep_ack_pkt(p)

	if p.len:
	    self.dep_seg(p)
	    self.lastp = p

	if flags & TH_FIN:
	    self.dep_fin_pkt(p)

	#self.lastp = p

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

    def arr_syn(self, p):

	trace = 1

	#print '%s#%d SYN arr' % (self.nm, p.indx)

	# duplicate?
	if self.conn_state & SYN_RECD:
	    if p.seq == self.irs:
		# duplicate
		str = 'TCPConn #%d  DUP ARR SYN: %s#%d' % (self.connid, self.nm, p.indx)
		if trace and self.trace:
		    whoops(str)
		self.mach.logfun(str)
		p.remark =1
		self.dup_arr_syns = self.dup_arr_syns + 1
		if self.conn_state & SYN_ACK_SENT:
		    # seen the ACK arr q popped
		    pass
		else:
		    self.rcv_segs.pop(0)
		

	self.conn_state = self.conn_state | SYN_RECD
	self.irs = p.seq
	self.rcv_nxt = seq_add(self.irs, 1)
	self.snd_wnd = p.window

	self.snd_highack = self.irs

	self.rcv_segs.append(p)
	self.arr_synp = p

## 	#
## 	r = seq_Release(p.seq, self.rcv_nxt, p)
## 	r.rtt = 0

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

    #
    ## Identify and check SYNACK
    #
    def arr_synack(self, p):

	trace = 1

	if trace and self.trace:
	    print '%s#%d IN SYNACK' % (self.nm, p.indx)
	ack = p.ack
	self.arr_synackps.append(p)
	if self.conn_state & SYN_SENT:
	    # duplicate ?
	    if self.conn_state & SYN_ACK_RECD:
		str = 'TCPConn #%d DUP SYNACK: %s#%d' % (self.connid, self.nm, p.indx)
		if trace and self.trace:
		    whoops(str)
		self.mach.logfun(str)
		self.dup_arr_synacks = self.dup_arr_synacks + 1
		# pop the first one
		self.rcv_acks.pop(0)
		#assess delay
		self.dup_synack_delay = p.tm - self.arr_synackps[0].tm
		self.dup_syn_delay = 0 #ACK delay takes precedence
		
	    if p.ack == self.synack:
		if trace and self.trace:
		    print '%s#%d SYNACK 0K' % (self.nm, p.indx)
		self.conn_state = self.conn_state | SYN_ACK_RECD
		#self.connected = 1
		#self.rcv_highack = ack
		if trace and self.trace:
		    print '%s#%d SYNACK %d snd_segs first seq = %d iss = %d' % (self.nm, p.indx, len(self.snd_segs), self.snd_segs[0].seq, self.iss)
		if len(self.snd_segs) == 1 and self.snd_segs[0].seq == self.iss:
		    if self.synunack:
			synp = self.dep_synps[-1]
			if trace and self.trace:
			    print '%s#%d unACKed SYN #%d' % (self.nm, 
					    p.indx, synp.indx)
			lag =  self.arr_synackps[-1].tm - synp.tm
			self.syn_rrtt = lag
			#p.prtt = rtt
			p.lag = lag
## 			if synp.prtt:
## 			    p.trtt = rtt + synp.prtt
		    #self.IW = self.segsz*self.imp.iw_fact#XX
		    #self.cwnd = self.IW#XX
		    #self.cwnd = self.segsz*self.imp.iw_fact
		    self.sstart = 1
		    p.markers = p.markers | P_PHASE_SS
		    if trace and self.trace: 
			print '%s#%d SYNACK_RECD - lag = %.3f' % (self.nm, p.indx, p.lag/1000.0)

		    if self.reverse.conn_state & SYN_ACK_RECD:
			self.conn_state |= CONNECTED
			self.reverse.conn_state |= CONNECTED
			if trace and self.trace: 
			    print '%s#%d Marking connected' % (self.nm, p.indx)
		else:
		    p.remark = 1
		    str = 'TCPConn #%d UNMATCHED SYNACK: %s#%d' % (self.connid, 
							   self.nm, p.indx)
		    if trace and self.trace:
			whoops(str)
		    self.mach.logfun(str)
		    self.giveup()
		    #sys.exit(1)
	    else:
		p.remark = 1
		str = 'TCPConn #%d SYNACK wrong: %s#%d expected %d - got %d' % (self.connid, self.nm, p.indx, seq_add(self.iss, 1), ack)
		if trace and self.trace:
		    whoops(str)
		self.mach.logfun(str)
		    #sys.exit(0)
	else:
	    p.remark = 1
	    str = 'TCPConn #%d  non SYNACK ack (unconnected): %s#%d' % (self.connid, 
				   self.nm, p.indx)
	    if trace and self.trace:
		whoops(str)

	self.synunack = 0
	

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

    def arr_ack(self, p):
	#print '%s #%d ARR_ACK %d self.highack = %d' % (self.nm, p.indx, p.ack, self.rcv_highack)
	cwnd_trace = 1
	rel_trace = 1
	ack_trace = 1

	ack = p.ack

	# looks reasonable?
	if seq_lt(ack, self.rcv_highack):
	    p.remark = 1
	    str = 'TCPConn #%d LOW_ACK ARRIVED: %s#%d - got %d expected >= %d' % (self.connid, self.nm, p.indx, ack, self.rcv_highack)
	    if ack_trace and self.trace:
		whoops(str)
	    self.mach.logfun(str)
	    self.too_low_acks.append(p)
	    p.markers |= P_BOGUS_ACK
	    return
	    #sys.exit(0)
	    #raw_input('........anything to continue\n')

	if seq_gt(ack, self.snd_nxt):
	    p.remark = 1
	    str = 'TCPConn #%d  HIGH ACK ARRIVED: %s#%d - got  %d expected <= %d' % (self.connid, self.nm, p.indx, ack, self.snd_nxt)
	    if ack_trace and self.trace:
		whoops(str)
	    self.mach.logfun(str)
	    #sys.exit(0);
	    #raw_input('........anything to continue\n')
	    # TODO - something with this
	    self.too_high_acks.append(p)
	    p.markers |= P_BOGUS_ACK
	    return

	self.rcv_acks.append(p)

	self.snd_wnd = p.window 

	# First data ACK'd - establish initial window
	if self.IW == None and (self.conn_state & CONNECTED) and self.dep_octs:
	    self.set_IW(p)
	
 	if self.synunack or not self.conn_state & SYN_ACK_RECD:
	    # Check is SYNACK
	    self.arr_synack(p)
	elif ack != self.rcv_highack:
	# ACK increased
	    # Adjust cwnd
	    if cwnd_trace and self.trace:
		print '%s#%d adj cwnd + ' % (self.nm, p.indx),
	    if self.cwnd < self.ssthresh:
		if  cwnd_trace and self.trace:
		    print '%d (ss)' % (self.segsz)
		self.cwnd += self.segsz
		self.sstart = 1
		p.markers |= P_PHASE_SS
	    else:
		if cwnd_trace and self.trace:
		    print '%d (ca)' % ((self.segsz*self.segsz)/self.cwnd)
		self.cwnd += ((self.segsz*self.segsz)/self.cwnd)
		self.sstart = 0

	if ack != self.rcv_highack:
	    ## remove acked pkts from sent list
	    if ack_trace and self.trace:
		print '%s#%d arr ack ' % (self.nm, p.indx),
	    p_acked = 0
	    llen = len(self.snd_segs)
	    if llen == 0:
		if ack_trace and self.trace:
		    print ' - no sent packets'
	    while llen:
		snt = self.snd_segs[0]
		send  = snt.end
		if seq_lte(send, ack):
		    if ack_trace and self.trace:
			print 'of %d ' % (snt.indx),
		    if ack == send:
			# pkt is last acked by this ack
			if not p.len:
			    #not piggy-backed
			    p.trig = TRIG_ACK
			    p.lag = p.tm - snt.tm
			    p.pkt_acked = snt
			if snt.rel!= None:
			    if self.trace and ack_trace:
				print '(rel by %d) ' % (snt.rel.ack_pkt.indx),
				print 'lag = %.3f popped %d' % (p.lag/1000.0, snt.indx)
		    self.snd_segs.pop(0)
		    llen = llen -1
		    p_acked += 1
		else:
		    break
	    if ack_trace and self.trace:
		print 'acked = %d' % (p_acked)
	    if p_acked < 2:
		if p.ack == self.synack:
		    p.trig = TRIG_SYNACK
		elif p.ack == self.finack:
		    p.trig = TRIG_FINACK
		elif not p.len:
		    # tmp conditional
		    if p.lag:
			p.trig = TRIG_DEL_ACK
            elif p_acked > 2 and p.ack != self.finack:
                str = 'TCPConn #%d MULTI-ACK: %s#%d - %d segs' % (self.connid, self.nm, p.indx, p_acked)
                self.mach.logfun(str)

	else:
	    if not p.len:
		# repeat ack - don't use for rtt calc
		#p.dcause = p.dcause | P_RTT_INVALID
		p.trig = TRIG_RTT_INVALID

		

    ## 	if p_acked:
    ## 	    print '%s#%d arr acked %d seg_rtt = %.3f' % (self.nm, 
    ## 						  p.indx, p_acked, rtt/1000.0)



	self.rcv_highack = ack
	    

	## Highest seq released by this ack
	if self.snd_wnd < self.cwnd:
	    w = self.snd_wnd 
	    wboss = 0
	    bs = 'snd_wnd'
	else:
	    w = self.cwnd
	    wboss = P_REL_BOSS_CWND
	    bs = 'cwnd'

	
	release = seq_add(ack,w)
	#if release > self.release:
	if seq_gt(release, self.release):
	    # push onto pending releases
	    if self.trace:
		p.highrel = release
		if rel_trace:
		    print '%s#%d release %d-%d (%+d) boss %s' % (self.nm, 
			   p.indx, self.release, release, 
			       seq_diff(release, self.release), bs)
	    r = seq_Release(self.release, release, p)
	    self.releases.append(r)
	    #p.rel = r
	    #p.highrel = release
	    p.markers = p.markers | wboss
	    self.release = release 
	
	
	


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

    def arr_seg(self, p):

	seq = p.seq
	len = p.len
	end = seq_add(seq, len)

	self.rcv_segs.append(p)


	if seq == self.rcv_nxt:
	    self.rcv_nxt = end # TODO - check len within rcv wnd
	elif seq_gt(seq, self.rcv_nxt):
	    # there's a gap
	    p.markers = p.markers | P_SEQ_HELD
	    self.rcv_segheld.append(p)
	elif seq_gt(end, self.rcv_nxt):
	    # some o'lap
	    self.rcv_olap_p = self.rcv_olap_p + 1 
	    self.rcv_olap_o = self.rcv_olap_o + seq_sub(self.rcv_nxt, seq)
	    self.rcv_nxt = end 
	else:
	    # totally duplicated
	    self.rcv_rtmts_p = self.rcv_rtmts_p + 1
	    self.rcv_rtmts_o = self.rcv_rtmts_o + len
	    # remove from received list
	    self.rcv_segs.pop()
	    

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


    def arr_fin_pkt(self, p):

	if self.conn_state & FIN_RECD:
	    str = 'TCPConn #%d DUP FIN: %s#%d' % (self.connid, self.nm, p.indx)
	    if self.trace:
		whoops(str)
	    self.mach.logfun(str)
	    
	self.conn_state = self.conn_state | FIN_RECD

	#r = seq_Release(p.seq, seq_add(p.seq, 1), p)
	new = seq_add(self.release, 1)
	r = seq_Release(self.release, new, p)
	self.releases.append(r)
	if self.trace:
	    print '%s#%d FIN release appended %d-%d' % (self.nm, p.indx, 
			      self.release, new)
	self.release = new
	
	    

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

    def arrival(self, p):

	flags = p.flags

	if flags & TH_SYN:
	    self.arr_syn(p)

	if flags & TH_ACK:
	    self.arr_ack(p)

	if p.len:
	    self.arr_seg(p)

	if flags & TH_FIN:
	    self.arr_fin_pkt(p)

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

    def set_IW(self, p):

	#print '%s#%d set IWF @ %d/%d octs' % (self.nm, p.indx, self.dep_octs, self.pl)
	trace = 1

        try:
            rel = self.releases[-1]
        except IndexError:
            #print '%s#%d set IWF no releases' % (self.nm, p.indx)
            return
            
	rp = rel.ack_pkt
	octs = self.dep_octs
	if octs == self.pl:
	    # all data sent - can't tell anything
	    self.imp = self.imp.clone([('iw_fact', 0)])
            self.IW = 0
	    return
	
	q, r = divmod(self.dep_octs, self.segsz)
	if r:
	    q += 1
	self.IW = self.cwnd = q*self.segsz
        #print 'set IW', self.IW
	imp = self.imp
	if q != imp.iw_fact:
	    self.imp = imp.clone([('iw_fact', q)])
	#print '%s#%d set IWF = %d' % (self.nm, p.indx, q)
	#self.print_releases()
	#print self.snd_nxt
	if len(self.releases) > 1:
	    str = 'TCPConn #%d Set_IW - too many initial releases: %s#%d' % (self.connid, self.nm, p.indx)
	    if self.trace:
		whoops(str)
		self.print_releases()
		#raw_input('...')
	    self.mach.logfun(str)
	if rel.highseq != self.snd_nxt and not(seq_add(rel.highseq, 1) == self.snd_nxt and (self.conn_state & FIN_SENT)):
	    str = 'TCPConn #%d Set_IW - rel_highseq = %d snd_nxt = %d FIN_RECD = %x: %s#%d' % (self.connid, rel.highseq, self.snd_nxt, self.conn_state & FIN_SENT, self.nm, p.indx)
	    if self.trace:
		whoops(str)
		self.print_releases()
		#raw_input('...')
	    self.mach.logfun(str)
	    #sys.exit(1)

	self.releases = []
	sw = rp.window

	if sw < self.cwnd:
	    w = sw 
	    wboss = 0
	    bs = 'snd_wnd'
	else:
	    w = self.cwnd
	    wboss = P_REL_BOSS_CWND
	    bs = 'cwnd'

	
	self.release = seq_add(rp.ack,w)
	if self.trace:
	    rp.highrel = self.release
	    if trace:
		    print '%s resetting #%d release %d boss %s' % (self.nm, 
			   rp.indx, self.release, bs)
	    rp.markers = rp.markers | wboss 
	

	    

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

    def print_releases(self):

	print 'current releases:'
	for r in self.releases:

	    print '\t %d-%d by %d highrel %d' % (r.start, 
		r.end, r.ack_pkt.indx, r.highseq)

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