(20241204) Minor changes.
This commit is contained in:
@@ -30,24 +30,20 @@
|
||||
|
||||
|
||||
# To make sibling directories accessible for imports:
|
||||
# ---
|
||||
import sys
|
||||
sys.path.append(".")
|
||||
sys.path.append("..")
|
||||
|
||||
# Utils:
|
||||
# ---
|
||||
from utils import json_utils
|
||||
from utils import time_utils
|
||||
from utils import regex_utils
|
||||
from utils_v2.string import json
|
||||
from utils_v2.string import regex
|
||||
from utils_v2.date_time import date_time
|
||||
|
||||
# For networking:
|
||||
# ---
|
||||
import socket
|
||||
from scapy.all import *
|
||||
|
||||
# For running the script from the terminal:
|
||||
# ---
|
||||
import argparse
|
||||
|
||||
|
||||
@@ -81,24 +77,21 @@ import argparse
|
||||
def get_name_and_addr(destination):
|
||||
|
||||
# If the user provided the IP address:
|
||||
# ---
|
||||
if (
|
||||
regex_utils.match(destination, regex_utils.REGEX_IPV4) or
|
||||
regex_utils.match(destination, regex_utils.REGEX_IPV6)
|
||||
regex.match(destination, regex.REGEX_IPV4) or
|
||||
regex.match(destination, regex.REGEX_IPV6)
|
||||
):
|
||||
try: destination_name = socket.gethostbyaddr(destination)[0]
|
||||
except Exception as exception: destination_name = "*"
|
||||
destination_ip = destination
|
||||
|
||||
# If the provided destination was the domain name:
|
||||
# ---
|
||||
else:
|
||||
destination_name = destination
|
||||
try: destination_ip = socket.gethostbyname(destination)
|
||||
except Exception as exception: destination_ip = "*"
|
||||
|
||||
# Done here:
|
||||
# ---
|
||||
return destination_name, destination_ip
|
||||
|
||||
|
||||
@@ -113,17 +106,14 @@ def tracert(
|
||||
):
|
||||
|
||||
# Initialize the variables:
|
||||
# ---
|
||||
destination_name, destination_ip = get_name_and_addr(destination)
|
||||
full_trace = []
|
||||
ttl = 1
|
||||
|
||||
# Keep noting hops till the limit is reached:
|
||||
# ---
|
||||
while ttl <= max_hops:
|
||||
|
||||
# Create a JSON for this stage:
|
||||
# ---
|
||||
this_hop = {
|
||||
"destAddr": destination_ip,
|
||||
"destName": destination_name,
|
||||
@@ -135,22 +125,18 @@ def tracert(
|
||||
}
|
||||
|
||||
# Create the IP and UDP headers and combine them:
|
||||
# ---
|
||||
ip_packet = IP(dst = destination, ttl = ttl)
|
||||
udp_packet = UDP(dport = port)
|
||||
trace_packet = ip_packet / udp_packet
|
||||
|
||||
# Send the packet and receive a reply and note down the timestamp:
|
||||
# ---
|
||||
reply = sr1(trace_packet, timeout = timeout, verbose = 0)
|
||||
this_hop["ts"] = time_utils.get_current_utc_datetime(as_string = True)
|
||||
this_hop["ts"] = date_time.get_current_utc_date_time(as_string = True)
|
||||
|
||||
# No response:
|
||||
# ---
|
||||
if reply is None: this_hop["hopAddr"] = this_hop["hopName"] = "*"
|
||||
|
||||
# If some response was received, we note the values and break out if this was the destination hop:
|
||||
# ---
|
||||
else:
|
||||
this_hop["hopName"], this_hop["hopAddr"] = get_name_and_addr(f"{reply.src}")
|
||||
if reply.type == 3:
|
||||
@@ -159,12 +145,10 @@ def tracert(
|
||||
break
|
||||
|
||||
# Carry on to the next hop:
|
||||
# ---
|
||||
full_trace.append(this_hop)
|
||||
ttl += 1
|
||||
|
||||
# Done here:
|
||||
# ---
|
||||
return full_trace
|
||||
|
||||
|
||||
@@ -203,7 +187,7 @@ if __name__ == "__main__":
|
||||
"--port",
|
||||
type = int,
|
||||
default = 33434,
|
||||
help = "Timeout for each packet in seconds (default: 33434)."
|
||||
help = "The port to probe at the destination (default: 33434)."
|
||||
)
|
||||
|
||||
args = parser.parse_args()
|
||||
@@ -214,5 +198,4 @@ if __name__ == "__main__":
|
||||
timeout = args.timeout,
|
||||
port = args.port
|
||||
)
|
||||
print("TRACE:")
|
||||
print(json_utils.to_json_string(trace))
|
||||
print("TRACE:", json.to_string(trace))
|
||||
|
||||
Reference in New Issue
Block a user