<s> from twisted . internet import defer <EOL> from twisted . python import usage <EOL> from ooni . templates import scapyt <EOL> from ooni . utils import log <EOL> from ooni . utils . txscapy import MPTraceroute <EOL> from ooni . settings import config <EOL> class UsageOptions ( usage . Options ) : <EOL> optParameters = [ <EOL> [ '<STR_LIT>' , '<STR_LIT:b>' , None , '<STR_LIT>' ] , <EOL> [ '<STR_LIT>' , '<STR_LIT:t>' , <NUM_LIT:5> , '<STR_LIT>' ] , <EOL> [ '<STR_LIT>' , '<STR_LIT:m>' , <NUM_LIT:30> , <EOL> '<STR_LIT>' ] , <EOL> [ '<STR_LIT>' , '<STR_LIT:d>' , None , <EOL> '<STR_LIT>' ] , <EOL> [ '<STR_LIT>' , '<STR_LIT:i>' , None , <EOL> '<STR_LIT>' ] , <EOL> [ '<STR_LIT>' , '<STR_LIT:n>' , None , <EOL> '<STR_LIT>' ] , <EOL> ] <EOL> class Traceroute ( scapyt . BaseScapyTest ) : <EOL> name = "<STR_LIT>" <EOL> description = "<STR_LIT>" "<STR_LIT>" <EOL> requiredTestHelpers = { '<STR_LIT>' : '<STR_LIT>' } <EOL> requiresRoot = True <EOL> requiresTor = False <EOL> usageOptions = UsageOptions <EOL> dst_ports = [ <NUM_LIT:0> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> , <NUM_LIT> ] <EOL> version = "<STR_LIT>" <EOL> def setUp ( self ) : <EOL> self . report [ '<STR_LIT>' ] = dict ( <EOL> [ ( '<STR_LIT>' % d , [ ] ) for d in self . dst_ports ] ) <EOL> self . report [ '<STR_LIT>' ] = dict ( <EOL> [ ( '<STR_LIT>' % d , [ ] ) for d in self . dst_ports ] ) <EOL> self . report [ '<STR_LIT>' ] = { '<STR_LIT>' : [ ] } <EOL> @ defer . inlineCallbacks <EOL> def run_traceroute ( self , protocol ) : <EOL> st = MPTraceroute ( ) <EOL> if self . localOptions [ '<STR_LIT>' ] : <EOL> st . ttl_max = int ( self . localOptions [ '<STR_LIT>' ] ) <EOL> if self . localOptions [ '<STR_LIT>' ] : <EOL> st . dst_ports = [ int ( self . localOptions [ '<STR_LIT>' ] ) ] <EOL> if self . localOptions [ '<STR_LIT>' ] : <EOL> st . interval = float ( self . localOptions [ '<STR_LIT>' ] ) <EOL> log . msg ( "<STR_LIT>" % ( protocol , <EOL> self . localOptions [ '<STR_LIT>' ] ) ) <EOL> log . msg ( "<STR_LIT>" % st . timeout ) <EOL> config . scapyFactory . registerProtocol ( st ) <EOL> traceroute = getattr ( st , protocol + '<STR_LIT>' ) <EOL> yield traceroute ( self . localOptions [ '<STR_LIT>' ] ) <EOL> st . stopListening ( ) <EOL> st . matchResponses ( ) <EOL> for packet in st . sent_packets : <EOL> self . report [ '<STR_LIT>' ] . append ( packet ) <EOL> for packet in st . matched_packets . values ( ) : <EOL> self . report [ '<STR_LIT>' ] . extend ( packet ) <EOL> for ttl in xrange ( st . ttl_min , st . ttl_max ) : <EOL> matchedPackets = filter ( <EOL> lambda x : x . ttl == ttl , <EOL> st . matched_packets . keys ( ) ) <EOL> for packet in matchedPackets : <EOL> for response in st . matched_packets [ packet ] : <EOL> self . addToReport ( packet , response ) <EOL> def test_icmp_traceroute ( self ) : <EOL> return self . run_traceroute ( '<STR_LIT>' ) <EOL> def test_tcp_traceroute ( self ) : <EOL> return self . run_traceroute ( '<STR_LIT>' ) <EOL> def test_udp_traceroute ( self ) : <EOL> return self . run_traceroute ( '<STR_LIT>' ) <EOL> def addToReport ( self , packet , response ) : <EOL> if packet . proto == <NUM_LIT:1> : <EOL> self . report [ '<STR_LIT>' ] [ '<STR_LIT>' ] . append ( <EOL> { '<STR_LIT>' : packet . ttl , '<STR_LIT>' : response . time - packet . time , <EOL> '<STR_LIT:address>' : response . src } ) <EOL> elif packet . proto == <NUM_LIT:6> : <EOL> self . report [ '<STR_LIT>' ] [ <EOL> '<STR_LIT>' % packet . dport ] . append ( <EOL> { '<STR_LIT>' : packet . ttl , '<STR_LIT>' : response . time - packet . time , <EOL> '<STR_LIT:address>' : response . src , '<STR_LIT>' : response . sport } ) <EOL> else : <EOL> self . report [ '<STR_LIT>' ] [ <EOL> '<STR_LIT>' % packet . dport ] . append ( <EOL> { '<STR_LIT>' : packet . ttl , '<STR_LIT>' : response . time - packet . time , <EOL> '<STR_LIT:address>' : response . src , '<STR_LIT>' : response . sport } ) </s>
