jyapayne/traceroute
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

pep8 and pylint clean-up

Browse source
This commit is contained in:
ayeowch 2013-07-13 00:24:45 +10:00
commit 4ad54e9e56
1 changed files with 182 additions and 158 deletions
Download patch file Download diff file Expand all files Collapse all files

340
traceroute.py
View file

@ -1,34 +1,54 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# traceroute.py - Multi-source traceroute with geolocation information.
#
# Copyright (c) 2013 Addy Yeow Chin Heng <ayeowch@gmail.com>
#
# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish,
# distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to
# the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
# LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
# OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
# WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
"""
traceroute - Get traceroute results with associated geolocation information
for each hop for a specified host from geographically distant source(s).
Multi-source traceroute with geolocation information.
"""
__author__ = 'Dazzlepod (info@dazzlepod.com)'
__copyright__ = 'Copyright (c) 2013 Dazzlepod'
__version__ = '$Revision: #20 $'
import datetime
import json
import optparse
import os
import re
import signal
import socket
import sys
import urllib
import urllib2
from multiprocessing import Process
from subprocess import Popen, PIPE
USER_AGENT = "traceroute/1.0 (+https://github.com/ayeowch/traceroute)"
class Traceroute(object):
"""Traceroute instance."""
def __init__(self, ip_address='8.8.8.8', country='US', tmp_dir='/tmp', no_geo=False, timeout=120, debug=False):
"""
Multi-source traceroute instance.
"""
def __init__(self, ip_address="8.8.8.8", country="US", tmp_dir="/tmp",
no_geo=False, timeout=120, debug=False):
super(Traceroute, self).__init__()
self.ip_address = ip_address
self.country = country
self.tmp_dir = tmp_dir
@ -38,124 +58,133 @@ class Traceroute(object):
# Traceroute servers from traceroute.org
sources = {
# 'LO' for localhost in which case 'url' will be used as the command to execute traceroute locally
'LO': {
'url': 'traceroute %s' % self.ip_address,
'post_data': None,
'url': "traceroute {}".format(ip_address),
},
'BY': {
'url': 'http://by104.activeby.net/lg/',
'post_data': {'method': 'trace', 'host': self.ip_address, 'router': 'http://by104.activeby.net/lg/'}
'url': "http://by104.activeby.net/lg/",
'post_data': {
'method': "trace",
'host': ip_address,
},
},
'CH': {
'url': 'http://www.switch.ch/cgi-bin/network/nph-traceroute-opencms?destination=%s' % self.ip_address,
'post_data': None,
'url': ("http://www.switch.ch/cgi-bin/network/nph-traceroute"
"-opencms?destination={}".format(ip_address)),
},
'JP': {
'url': 'http://www.harenet.ad.jp/cgi-bin/harenet/traceroute/traceroute.cgi',
'post_data': {'host': self.ip_address},
'url': ("http://www.harenet.ad.jp/cgi-bin/harenet/traceroute/"
"traceroute.cgi"),
'post_data': {
'host': ip_address,
},
},
'RU': {
'url': 'http://ipnoc.zenon.net/pcgi/trace.pl?IP=%s' % self.ip_address,
'post_data': None
'url': "http://traceroute.rusnet.ru/?{}".format(ip_address)
},
'UK': {
'url': 'http://ab.newnet.co.uk/cgi-bin/traceroute?%s' % self.ip_address,
'post_data': None
'url': ("http://ab.newnet.co.uk/cgi-bin/traceroute?{}".format(
ip_address)),
},
'US': {
'url': 'http://www.net.princeton.edu/cgi-bin/traceroute.pl',
'post_data': {'target': self.ip_address},
'url': "http://www.net.princeton.edu/cgi-bin/traceroute.pl",
'post_data': {
'target': ip_address,
},
},
}
self.source = sources[self.country]
# cache geocoded IP addresses during the lifetime of this instance
self.locations = {}
self.locations = {} # Geo-coded locations
def traceroute(self):
"""Instead of running the actual traceroute command, we will fetch
"""
Instead of running the actual traceroute command, we will fetch
standard traceroute results from several publicly available webpages
that are listed at traceroute.org. For each hop, we will then attach
geolocation information to it."""
self.print_debug("ip_address = %s" % self.ip_address)
txt = os.path.join(self.tmp_dir, '%s.%s.txt' % (self.ip_address, self.country))
if not os.path.exists(txt):
if self.country == 'LO':
(status_code, traceroute) = self.execute_cmd(self.source['url'])
else:
(status_code, traceroute) = self.get_traceroute_output()
geolocation information to it.
"""
self.print_debug("ip_address={}".format(self.ip_address))
filename = "{}.{}.txt".format(self.ip_address, self.country)
filepath = os.path.join(self.tmp_dir, filename)
if not os.path.exists(filepath):
if self.country == "LO":
status_code, traceroute = self.execute_cmd(self.source['url'])
else:
status_code, traceroute = self.get_traceroute_output()
if status_code != 0 and status_code != 200:
return {'error': status_code}
open(filepath, "w").write(traceroute)
traceroute = open(filepath, "r").read()
f = open(txt, 'w')
f.write(traceroute)
f.close()
traceroute = open(txt, 'r').read()
# hops = dicts with keys: hop_num, hosts
# hop_num, hosts
hops = self.get_hops(traceroute)
# hops = dicts with keys: hop_num, hostname, ip_address, rtt
# hop_num, hostname, ip_address, rtt
hops = self.get_formatted_hops(hops)
if not self.no_geo:
# hops = dicts with keys: hop_num, hostname, ip_address, rtt, latitude, longitude
# hop_num, hostname, ip_address, rtt, latitude, longitude
hops = self.get_geocoded_hops(hops)
return hops
def get_traceroute_output(self):
"""Fetch traceroute output from a webpage."""
"""
Fetches traceroute output from a webpage.
"""
url = self.source['url']
(status_code, content) = self.urlopen(url, context = self.source['post_data'])
if 'post_data' in self.source:
context = self.source['post_data']
else:
context = None
status_code, content = self.urlopen(url, context=context)
content = content.strip()
pattern = re.compile(r'<pre.*?>(?P<traceroute>.*?)</pre>', re.DOTALL|re.IGNORECASE)
regex = r'<pre.*?>(?P<traceroute>.*?)</pre>'
pattern = re.compile(regex, re.DOTALL | re.IGNORECASE)
try:
traceroute = re.findall(pattern, content)[0].strip()
except IndexError:
# Manually append closing </pre> for partially downloaded page
content = '%s</pre>' % content
content = "{}</pre>".format(content)
traceroute = re.findall(pattern, content)[0].strip()
return (status_code, traceroute)
def get_hops(self, traceroute):
"""Get hops from a traceroute output and return the hops in an array
of dicts each representing hop number and the associated hosts data."""
"""
Returns hops from traceroute output in an array of dicts each
with hop number and the associated hosts data.
"""
hops = []
lines = traceroute.split('\n')
regex = r'^(?P<hop_num>\d+)(?P<hosts>.*?)$'
lines = traceroute.split("\n")
for line in lines:
line = line.strip()
hop = {}
if not line: continue
if not line:
continue
try:
hop = re.match(r'^(?P<hop_num>\d+)(?P<hosts>.*?)$', line).groupdict()
hop = re.match(regex, line).groupdict()
except AttributeError:
continue
self.print_debug(hop)
hops.append(hop)
return hops
def get_formatted_hops(self, hops):
"""hosts data from get_hops() is represented in a single string.
We use this function to better represent the hosts data in a dict."""
"""
Hosts data from get_hops() is represented in a single string.
We use this function to better represent the hosts data in a dict.
"""
formatted_hops = []
regex = r'(?P<h>[\w.-]+) \((?P<i>[\d.]+)\) (?P<r>\d{1,4}.\d{1,4} ms)'
for hop in hops:
hop_num = int(hop['hop_num'].strip())
hosts = hop['hosts'].replace(' ', ' ').strip()
# Using re.findall(), first we split the hosts, then for each host we store a tuple containing hostname, IP address and the first round-trip time
# [('<HOSTNAME>', '<IP_ADDRESS>', '<RTT1> ms'), ('<HOSTNAME_N>', '<IP_ADDRESS_N>', '<RTT1_N> ms')]
hosts = re.findall(r'(?P<hostname>[\w.-]+) \((?P<ip_address>[\d.]+)\) (?P<rtt>\d{1,4}.\d{1,4} ms)', hosts)
hosts = hop['hosts'].replace(" ", " ").strip()
# Using re.findall(), we split the hosts, then for each host,
# we store a tuple of hostname, IP address and the first RTT.
hosts = re.findall(regex, hosts)
for host in hosts:
hop_context = {
'hop_num': hop_num,
@ -165,24 +194,22 @@ class Traceroute(object):
}
self.print_debug(hop_context)
formatted_hops.append(hop_context)
return formatted_hops
def get_geocoded_hops(self, hops):
"""Return hops from get_formatted_hops() with geolocation information
for each hop."""
"""
Returns hops from get_formatted_hops() with geolocation information
for each hop.
"""
geocoded_hops = []
for hop in hops:
ip_address = hop['ip_address']
location = None
if self.locations.has_key(ip_address):
if ip_address in self.locations:
location = self.locations[ip_address]
else:
location = self.get_location(ip_address)
self.locations[ip_address] = location
if location:
geocoded_hops.append({
'hop_num': hop['hop_num'],
@ -192,140 +219,137 @@ class Traceroute(object):
'latitude': location['latitude'],
'longitude': location['longitude'],
})
return geocoded_hops
def get_location(self, ip_address):
"""Return geolocation information for the specified IP address, e.g.:
{"ip": "75.126.24.77",
"hostname": "web365.webfaction.com",
"isp": "SoftLayer Technologies",
"organization": "Client Intellect",
"country": "United States",
"region": "Texas",
"city": "Dallas",
"latitude": 32.9299,
"longitude": -96.8353}
"""
Returns geolocation information for the given IP address.
"""
location = None
url = "http://dazzlepod.com/ip/%s.json" % ip_address
(status_code, json_data) = self.urlopen(url)
url = "http://dazzlepod.com/ip/{}.json".format(ip_address)
status_code, json_data = self.urlopen(url)
if status_code == 200 and json_data:
tmp_location = json.loads(json_data)
if tmp_location.has_key('latitude') and tmp_location.has_key('longitude'):
if 'latitude' in tmp_location and 'longitude' in tmp_location:
location = tmp_location
return location
def execute_cmd(self, cmd):
"""Execute the specified command locally and return the resultant
return code and output."""
stdout = ''
"""
Executes given command using subprocess.Popen().
"""
stdout = ""
returncode = -1
p = Popen(cmd, shell=True, stdin=PIPE, stdout=PIPE, stderr=PIPE)
process = Popen(cmd, shell=True, stdin=PIPE, stdout=PIPE, stderr=PIPE)
try:
signal.signal(signal.SIGALRM, self.signal_handler)
signal.alarm(self.timeout)
stdout, stderr = p.communicate()
returncode = p.returncode
self.print_debug("cmd = %s, returncode = %d" % (cmd, returncode))
stdout, stderr = process.communicate()
returncode = process.returncode
self.print_debug("cmd={}, returncode={}".format(cmd, returncode))
if returncode != 0:
self.print_debug("stderr={}".format(stderr))
signal.alarm(0)
except Exception, e:
self.print_debug("%s" % str(e))
except Exception as err:
self.print_debug(str(err))
return (returncode, stdout)
def urlopen(self, url, context=None):
"""Perform HTTP GET/POST on the specified URL and return the resultant
status code and response."""
"""
Fetches webpage.
"""
status_code = 200
request = urllib2.Request(url = url)
request.add_header('User-Agent', 'traceroute/1.0 (+https://github.com/ayeowch/traceroute')
request = urllib2.Request(url=url)
request.add_header('User-Agent', USER_AGENT)
if context:
data = urllib.urlencode(context)
request.add_data(data)
content = ''
content = ""
try:
response = urllib2.urlopen(request)
self.print_debug("url = %s\nheader = %s" % (response.geturl(), response.info()))
self.print_debug("url={}".format(response.geturl()))
content = self.chunked_read(response)
except urllib2.HTTPError, e:
status_code = e.code
except urllib2.HTTPError as err:
status_code = err.code
except urllib2.URLError:
pass
try:
self.urlopen_count += 1
except AttributeError:
self.urlopen_count = 1
self.print_debug("[%d] url = %s, status_code = %d" % (self.urlopen_count, url, status_code))
return (status_code, content)
def chunked_read(self, response):
"""Read page response in chunks. A signal handler is attached to abort
reading after the set timeout.
Chunk size = 64 bytes, max. page size = 1MB
"""
content = ''
max_bytes = 1 * 1024 * 1024
completed_bytes = 0
bytes_per_read = 64
Fetches response in chunks. A signal handler is attached to abort
reading after set timeout.
"""
content = ""
max_bytes = 1 * 1024 * 1024 # Max. page size = 1MB
read_bytes = 0
bytes_per_read = 64 # Chunk size = 64 bytes
try:
signal.signal(signal.SIGALRM, self.signal_handler)
signal.alarm(self.timeout)
while completed_bytes <= max_bytes:
while read_bytes <= max_bytes:
data = response.read(bytes_per_read)
if not data:
break
content += data
completed_bytes += bytes_per_read
self.print_debug("completed_bytes = %d, %s" % (completed_bytes, data))
read_bytes += bytes_per_read
self.print_debug("read_bytes={}, {}".format(read_bytes, data))
signal.alarm(0)
except Exception, e:
self.print_debug("%s" % str(e))
except Exception as err:
self.print_debug(str(err))
return content
def signal_handler(self, signum, frame):
"""Signal handler that simply raises an exception when triggered."""
raise Exception("Caught signal %d" % signum)
def signal_handler(self, signum):
"""
Raises exception when signal is caught.
"""
raise Exception("Caught signal {}".format(signum))
def print_debug(self, msg):
"""Print debug message to standard output."""
"""
Prints debug message to standard output.
"""
if self.debug:
print "[DEBUG %s] %s" % (datetime.datetime.now(), msg)
print("[DEBUG {}] {}".format(datetime.datetime.now(), msg))
def main():
usage = """%prog --ip_address=IP_ADDRESS"""
cmdparser = optparse.OptionParser(usage, version=("traceroute " + __version__))
cmdparser.add_option("-i", "--ip_address", type="string", default="8.8.8.8", help="IP address of destination host (default: 8.8.8.8)")
cmdparser.add_option("-c", "--country", type="choice",
choices=['LO', 'BY', 'CH', 'JP', 'RU', 'UK', 'US',],
default="US",
help="Traceroute will be initiated from this country; choose 'LO' for localhost to run traceroute locally, 'BY' for Belarus, 'CH' for Switzerland, 'JP' for Japan, 'RU' for Russia, 'UK' for United Kingdom or 'US' for United States (default: US)")
cmdparser.add_option("-t", "--tmp_dir", type="string", default="/tmp", help="Temporary directory to store downloaded traceroute results (default: /tmp)")
cmdparser.add_option("-n", "--no_geo", action="store_true", default=False, help="No geolocation data (default: False)")
cmdparser.add_option("-s", "--timeout", type="int", default=120, help="Timeout in seconds for all downloads (default: 120)")
cmdparser.add_option("-d", "--debug", action="store_true", default=False, help="Show debug output (default: False)")
(options, args) = cmdparser.parse_args()
cmdparser = optparse.OptionParser("%prog --ip_address=IP_ADDRESS")
cmdparser.add_option(
"-i", "--ip_address", type="string", default="8.8.8.8",
help="IP address of destination host (default: 8.8.8.8)")
cmdparser.add_option(
"-c", "--country", type="choice", default="US",
choices=["LO", "BY", "CH", "JP", "RU", "UK", "US"],
help=("Traceroute will be initiated from this country; choose 'LO' "
"for localhost to run traceroute locally, 'BY' for Belarus, "
"'CH' for Switzerland, 'JP' for Japan, 'RU' for Russia, 'UK' "
"for United Kingdom or 'US' for United States (default: US)"))
cmdparser.add_option(
"-t", "--tmp_dir", type="string", default="/tmp",
help=("Temporary directory to store downloaded traceroute results "
"(default: /tmp)"))
cmdparser.add_option(
"-n", "--no_geo", action="store_true", default=False,
help="No geolocation data (default: False)")
cmdparser.add_option(
"-s", "--timeout", type="int", default=120,
help="Timeout in seconds for all downloads (default: 120)")
cmdparser.add_option(
"-d", "--debug", action="store_true", default=False,
help="Show debug output (default: False)")
options, _ = cmdparser.parse_args()
if options.ip_address:
traceroute = Traceroute(ip_address=options.ip_address,
country=options.country,
tmp_dir=options.tmp_dir,
no_geo=options.no_geo,
timeout=options.timeout,
debug=options.debug)
country=options.country,
tmp_dir=options.tmp_dir,
no_geo=options.no_geo,
timeout=options.timeout,
debug=options.debug)
hops = traceroute.traceroute()
hops = json.dumps(hops, indent=4)
print hops
print(json.dumps(hops, indent=4))
else:
cmdparser.print_usage()
return 0