-
Notifications
You must be signed in to change notification settings - Fork 0
/
traceroute.py
104 lines (83 loc) · 3.02 KB
/
traceroute.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
import socket
import random
import time
import struct
from IPv4_Header import *
"""
Simple Traceroute script with ICMP support.
TODO:
TTL calibrations
"""
class Traceroute(object):
def __init__(self, dst, hops=15):
"""
Initializes a new tracer object
"""
self.dst = dst
self.ip_dst = socket.gethostbyname(dst)
self.hops = hops
self.ttl = 1
self.port = 33434 # starting port
def run(self):
"""
Run the tracer
"""
print('Tracing to {} [{}], {} hops max'.format(self.dst, self.ip_dst, self.hops))
while True:
timer_start = time.time()
receiver = self.create_receiver()
sender = self.create_sender()
sender.sendto(b'', (self.ip_dst, self.port))
addr = None
timer_end = None
try:
data, addr = receiver.recvfrom(1024)
timer_end = time.time()
# print source and destination of IP
ip_header = IPv4_Header(data[:20]) # first 20 Bytes
print("\t {} -> {}".format(ip_header.src_address, ip_header.dst_address))
start_icmp = ip_header.ihl * 4 # Calculate start of ICMP
icmp_header = ICMP(data[start_icmp:start_icmp + sizeof(ICMP)])
print("\t type:{} code:{} {}".format(icmp_header.type, icmp_header.code, icmp_header.show_results()))
# type 11 , code 0 is correct answer from server.
except socket.error as e:
# Handling failed connection.
pass
finally:
receiver.close()
sender.close()
if addr:
time_to_dst = round((timer_end - timer_start) * 1000, 2)
print('{:<4} {} {} ms'.format(self.ttl, addr[0], time_to_dst))
if addr[0] == self.ip_dst:
print("Destination reached.")
break
else:
print('{:<4} *'.format(self.ttl))
self.ttl += 1
self.port += 1
if self.ttl > self.hops:
break
def create_receiver(self):
"""
Creates a receiver socket
:return: a socket instance
"""
s = socket.socket(family=socket.AF_INET, type=socket.SOCK_RAW, proto=socket.IPPROTO_ICMP)
# include headers in packets
s.setsockopt(socket.IPPROTO_IP, socket.IP_HDRINCL, 1)
timeout = struct.pack("ll", 10, 0)
s.setsockopt(socket.SOL_SOCKET, socket.SO_RCVTIMEO, timeout)
s.bind(('', self.port))
return s
def create_sender(self):
"""
Creates a sender socket
:returns: a socket instance
"""
s = socket.socket(family=socket.AF_INET, type=socket.SOCK_DGRAM, proto=socket.IPPROTO_UDP)
s.setsockopt(socket.SOL_IP, socket.IP_TTL, self.ttl)
return s
if __name__ == '__main__':
traceroute = Traceroute("www.onet.pl", hops=15)
traceroute.run()