<s> """<STR_LIT>""" <EOL> from __future__ import absolute_import <EOL> from __future__ import print_function <EOL> from __future__ import unicode_literals <EOL> from __future__ import division <EOL> from socket import ntohl <EOL> from pycopia import aid <EOL> TCP_CONNECTIONS_FILE = "<STR_LIT>" <EOL> TCP_STATES = aid . Enums ( TCP_ESTABLISHED = <NUM_LIT:1> , TCP_SYN_SENT = <NUM_LIT:2> , TCP_SYN_RECV = <NUM_LIT:3> , TCP_FIN_WAIT1 = <NUM_LIT:4> , TCP_FIN_WAIT2 = <NUM_LIT:5> , <EOL> TCP_TIME_WAIT = <NUM_LIT:6> , TCP_CLOSE = <NUM_LIT:7> , TCP_CLOSE_WAIT = <NUM_LIT:8> , TCP_LAST_ACK = <NUM_LIT:9> , TCP_LISTEN = <NUM_LIT:10> , TCP_CLOSING = <NUM_LIT:11> ) <EOL> ( TCP_ESTABLISHED , TCP_SYN_SENT , TCP_SYN_RECV , TCP_FIN_WAIT1 , TCP_FIN_WAIT2 , <EOL> TCP_TIME_WAIT , TCP_CLOSE , TCP_CLOSE_WAIT , TCP_LAST_ACK , TCP_LISTEN , TCP_CLOSING ) = TCP_STATES <EOL> def itoa ( address ) : <EOL> return "<STR_LIT>" % ( ( address >> <NUM_LIT> ) & <NUM_LIT> , <EOL> ( ( address & <NUM_LIT> ) >> <NUM_LIT:16> ) , <EOL> ( ( address & <NUM_LIT> ) >> <NUM_LIT:8> ) , <EOL> ( address & <NUM_LIT> ) ) <EOL> class NetstatTCPEntry ( object ) : <EOL> def __init__ ( self , local_address , rem_address , st , queues , tr_tm_when , retrnsmt , uid , timeout , inode ) : <EOL> self . local_address , self . local_port = self . _split_addrport ( local_address ) <EOL> self . rem_address , self . rem_port = self . _split_addrport ( rem_address ) <EOL> self . state = TCP_STATES . find ( int ( st , <NUM_LIT:16> ) ) <EOL> self . tx_queue , self . rx_queue = [ int ( p , <NUM_LIT:16> ) for p in queues . split ( "<STR_LIT::>" ) ] <EOL> self . tr_tm_when = tr_tm_when <EOL> self . retrnsmt = int ( retrnsmt , <NUM_LIT:16> ) <EOL> self . uid = int ( uid ) <EOL> self . timeout = int ( timeout ) <EOL> self . inode = int ( inode ) <EOL> def _split_addrport ( self , addrtext ) : <EOL> addr_s , port_s = addrtext . split ( "<STR_LIT::>" ) <EOL> return ntohl ( int ( addr_s , <NUM_LIT:16> ) ) , int ( port_s , <NUM_LIT:16> ) <EOL> def __str__ ( self ) : <EOL> return "<STR_LIT>" . format ( <EOL> itoa ( self . local_address ) , self . local_port , itoa ( self . rem_address ) , self . rem_port , self . state ) <EOL> class TCPTable ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . _slots = [ ] <EOL> def update ( self ) : <EOL> self . _slots = slots = [ ] <EOL> lines = open ( TCP_CONNECTIONS_FILE ) . readlines ( ) <EOL> for line in lines [ <NUM_LIT:1> : ] : <EOL> parts = line . split ( ) <EOL> slots . append ( NetstatTCPEntry ( * parts [ <NUM_LIT:1> : <NUM_LIT:10> ] ) ) <EOL> def __str__ ( self ) : <EOL> s = [ """<STR_LIT>""" ] <EOL> for nse in self . _slots : <EOL> s . append ( str ( nse ) ) <EOL> return "<STR_LIT:\n>" . join ( s ) <EOL> def __getitem__ ( self , idx ) : <EOL> return self . _slots [ idx ] <EOL> def __iter__ ( self ) : <EOL> return iter ( self . _slots ) <EOL> def get_listeners ( self ) : <EOL> for slot in self . _slots : <EOL> if slot . state == TCP_LISTEN : <EOL> yield slot <EOL> def listening_on_port ( self , port ) : <EOL> for slot in self . _slots : <EOL> if slot . state == TCP_LISTEN and slot . local_port == port : <EOL> return True <EOL> return False <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> tcp = TCPTable ( ) <EOL> tcp . update ( ) <EOL> print ( tcp ) <EOL> print ( "<STR_LIT>" ) <EOL> for slot in tcp . get_listeners ( ) : <EOL> print ( "<STR_LIT:U+0020>" , slot ) <EOL> print ( "<STR_LIT>" , tcp . listening_on_port ( <NUM_LIT> ) ) <EOL> print ( "<STR_LIT>" , tcp . listening_on_port ( <NUM_LIT> ) ) </s>
