Merge commit '6d1f731ff76e21de2ab3c328bad52b6fea435dd5' as 'utils_v2'
This commit is contained in:
@@ -0,0 +1,173 @@
|
||||
"""
|
||||
|
||||
AUTHOR:
|
||||
|
||||
Khushal P Soonderji
|
||||
|
||||
DATE:
|
||||
|
||||
Friday, 12th Jul., 2024
|
||||
|
||||
OBJECTIVE:
|
||||
|
||||
To provide an easy way to get geolocation information of an IP address.
|
||||
|
||||
REFERENCES:
|
||||
|
||||
1) https://medium.com/@tubelwj/how-to-retrieve-ip-geolocation-information-in-python-929e15041e3e
|
||||
|
||||
DOWNLOADS:
|
||||
|
||||
N/A
|
||||
|
||||
"""
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** IMPORT ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# To make sibling directories accessible for imports:
|
||||
import sys
|
||||
sys.path.append(".")
|
||||
sys.path.append("..")
|
||||
|
||||
# For working with IP Addresses:
|
||||
import ipaddress
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** MACROS / ONE-TIME INIT ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# --- Nothing Yet
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** VARIABLES ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# --- Nothing Yet
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** FUNCTIONS ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
def ipv4_to_int(ip_string):
|
||||
|
||||
"""
|
||||
Converts an IP (v4) string to an integer value.
|
||||
:param ip_string: The IP address (v4) that you want to convert to integer format.
|
||||
:return: An integer representation of the IP (v4) address.
|
||||
"""
|
||||
|
||||
ip_numerical = int(ipaddress.IPv4Address(ip_string))
|
||||
return ip_numerical
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
def int_to_ipv4(ip_numerical):
|
||||
|
||||
"""
|
||||
Interprets the IP (v4) value from the given integer value.
|
||||
:param ip_numerical: The integer value that represents an IP (v4) address.
|
||||
:return:
|
||||
"""
|
||||
|
||||
ip_string = str(ipaddress.IPv4Address(ip_numerical))
|
||||
return ip_string
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
def ipv4_to_bin(ip_string):
|
||||
|
||||
"""
|
||||
Converts an input IP (v4) address to the binary string that represents the 32 bits.
|
||||
:param ip_string: The IP (v4) string in a format like "192.168.0.1"
|
||||
:return: The binary representation (as a string) of the input IP address.
|
||||
"""
|
||||
|
||||
ip_binary = bin(int(ipaddress.IPv4Address(ip_string)))[2:].zfill(32)
|
||||
return ip_binary
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
def bin_to_ipv4(ip_binary):
|
||||
|
||||
"""
|
||||
Interprets the IP (v4) value from the given binary string.
|
||||
:param ip_binary: The string of 1s and 0s that represents the IP (v4) address.
|
||||
:return: The IP (v4) address as a string.
|
||||
"""
|
||||
|
||||
ip_string = str(ipaddress.IPv4Address(int(ip_binary, 2)))
|
||||
return ip_string
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
def get_ipv4_range(ip_string, as_string = True):
|
||||
|
||||
"""
|
||||
Given a network description in the format "88.95.100.128/25", this function tells you the first and last IP
|
||||
addresses of that network. Useful for determining if an IP address lies in a network.
|
||||
:param ip_string: The input network description in the format "88.95.100.128/25"
|
||||
:param as_string: To select between integer and string formats for the IP range output.
|
||||
:return: The first and last IP addresses of the input network, and the count.
|
||||
"""
|
||||
|
||||
# Extract the components of the string:
|
||||
ip_components = ip_string.split("/")
|
||||
ip_addr = ipv4_to_int(ip_components[0])
|
||||
ip_bits = int(ip_components[1])
|
||||
|
||||
# Convert the mask number to binary representation:
|
||||
ip_mask = (1 << ip_bits) - 1
|
||||
ip_mask = ip_mask << (32 - ip_bits)
|
||||
inv_ip_mask = (~ip_mask) & 0xFFFF
|
||||
|
||||
# Figure out the start and end IP addresses:
|
||||
start_ip = ip_addr & ip_mask
|
||||
end_ip = ip_addr | inv_ip_mask
|
||||
count = end_ip - start_ip + 1
|
||||
|
||||
# If the IPs are needed as strings, we perform the conversion:
|
||||
if as_string:
|
||||
start_ip = int_to_ipv4(start_ip)
|
||||
end_ip = int_to_ipv4(end_ip)
|
||||
|
||||
# Done here:
|
||||
return start_ip, end_ip, count
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** MAIN PROGRAM ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
print(ipv4_to_int("255.255.255.255"))
|
||||
print(ipv4_to_int("x.x.x.x"))
|
||||
@@ -0,0 +1,201 @@
|
||||
"""
|
||||
|
||||
AUTHOR:
|
||||
|
||||
Khushal P Soonderji
|
||||
|
||||
DATE:
|
||||
|
||||
Friday, 28th Jun, 2024
|
||||
|
||||
OBJECTIVE:
|
||||
|
||||
To provide a way to ping a server and get the traceroute dump.
|
||||
|
||||
REFERENCES:
|
||||
|
||||
1) https://www.geeksforgeeks.org/traceroute-implementation-on-python/
|
||||
|
||||
DOWNLOADS:
|
||||
|
||||
N/A
|
||||
|
||||
"""
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** IMPORT ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# To make sibling directories accessible for imports:
|
||||
import sys
|
||||
sys.path.append(".")
|
||||
sys.path.append("..")
|
||||
|
||||
# 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
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** MACROS / ONE-TIME INIT ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# --- Nothing Yet
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** VARIABLES ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
# --- Nothing Yet
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** FUNCTIONS ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
def get_name_and_addr(destination):
|
||||
|
||||
# If the user provided the IP address:
|
||||
if (
|
||||
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
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
def tracert(
|
||||
destination,
|
||||
max_hops = 30,
|
||||
timeout = 2.0,
|
||||
port = 33434
|
||||
):
|
||||
|
||||
# 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,
|
||||
"hopNo": ttl - 1,
|
||||
"isDest": False,
|
||||
"hopAddr": None,
|
||||
"hopName": None,
|
||||
"ts": None
|
||||
}
|
||||
|
||||
# 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"] = 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:
|
||||
this_hop["isDest"] = True
|
||||
full_trace.append(this_hop)
|
||||
break
|
||||
|
||||
# Carry on to the next hop:
|
||||
full_trace.append(this_hop)
|
||||
ttl += 1
|
||||
|
||||
# Done here:
|
||||
return full_trace
|
||||
|
||||
|
||||
# *****************************************************************************************************************
|
||||
# ***** ****
|
||||
# *** MAIN PROGRAM ***
|
||||
# ***** ****
|
||||
# *****************************************************************************************************************
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
parser = argparse.ArgumentParser(description = "Traceroute Implementation in Python!")
|
||||
|
||||
parser.add_argument(
|
||||
"dest",
|
||||
help = "Destination (Name or IP address)."
|
||||
)
|
||||
parser.add_argument(
|
||||
"-m",
|
||||
"--max-hops",
|
||||
type = int,
|
||||
default = 30,
|
||||
help =
|
||||
"Maximum number of hops (default: 30)."
|
||||
)
|
||||
parser.add_argument(
|
||||
"-t",
|
||||
"--timeout",
|
||||
type = float,
|
||||
default = 2.0,
|
||||
help = "Timeout for each packet in seconds (default: 2.0)."
|
||||
)
|
||||
parser.add_argument(
|
||||
"-p",
|
||||
"--port",
|
||||
type = int,
|
||||
default = 33434,
|
||||
help = "The port to probe at the destination (default: 33434)."
|
||||
)
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
trace = tracert(
|
||||
destination = args.dest,
|
||||
max_hops = args.max_hops,
|
||||
timeout = args.timeout,
|
||||
port = args.port
|
||||
)
|
||||
print("TRACE:", json.to_string(trace))
|
||||
Reference in New Issue
Block a user