mirrorscript.py

#!/usr/bin/python3
# Mirrorscript By Padsala Tushal
import requests
import re
import subprocess
import threading
from shutil import copyfile
import os,sys
import socket
import time

# Check if user is root first.
if os.getuid() != 0:
	sys.exit("[!] Must run as root/sudo\n")

banner_text = '''
  __  __ _                        _____           _       _   
 |  \/  (_)                      / ____|         (_)     | |  
 | \  / |_ _ __ _ __ ___  _ __  | (___   ___ _ __ _ _ __ | |_ 
 | |\/| | | '__| '__/ _ \| '__|  \___ \ / __| '__| | '_ \| __|
 | |  | | | |  | | | (_) | |     ____) | (__| |  | | |_) | |_ 
 |_|  |_|_|_|  |_|  \___/|_|    |_____/ \___|_|  |_| .__/ \__|
                                                   | |        
                                                   |_|        
                                            by padsala tushal
'''
print(banner_text)
INFO = '\033[38;5;50m'+'Info'+'\033[0m'

hosts = []
mirrors = {}
threads = []

# Function for getting hosts
def get_hosts():
    print(f"[{INFO}] Getting mirror list ...")
    # Getting mirrors list 
    r = requests.get('https://http.kali.org/README.mirrorlist')

    if r.status_code!=200:
        sys.exit("[!] Failed to establish a connection to host\n")

    urls = re.findall(r'(?:href="http(?:s|))(.*)(?:/README")',r.text)[2:]
    # print(urls)

    # Getting Hostname of each of the url
    for url in urls:
        hostname = url.split("//")[-1].split("/")[0].split('?')[0]
        # print(hostname)
        hosts.append(hostname)
    # print(hosts)
get_hosts()


# Function for measure latency of host
def find_latency(hostname):
    host_domain = hostname  # Replace with the domain you want to measure
    try:
        start_time = time.time()
    # Create a socket and try to connect to the host's port 80 (HTTP)
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            s.settimeout(5)  # Set a timeout for the connection attempt
            s.connect((host_domain, 80))
            end_time = time.time()
            latency = (end_time - start_time) * 1000  # Convert to milliseconds
            mirrors[host_domain] = float(latency)
        # print(f"Latency to {host_domain}: {latency:.2f} ms")
    except socket.error as e:
        print(f"Error: {e}")


    # # Ping the host
    # p = subprocess.Popen(['ping','-c 3', hostname], stderr=subprocess.PIPE, stdout=subprocess.PIPE).communicate()
    # p = [str(x.decode('utf-8')) for x in p]
    # print(f"{p[0]} =")
    # # Finding the latency
    # if "100% packet loss" in p[0].strip():
    #     # average = "[!] Unable to check " + hostname + " latency, potentially host block ICMP request."
    #     pass
    #     # average  = 99999
    # else:
    #     average = p[0].strip().splitlines()[-1].split('=')[1].split('/')[1]
    #     print(hostname,average)
    #     mirrors[hostname] = float(average)

print(f"[{INFO}] Finding the best latency ...")
# Measuring each host latency
# Create Thread for each host
for host in hosts:
    t = threading.Thread(target=find_latency,args=(host,))
    threads.append(t)
    t.start()

# wait for all threads to complete
for t in threads:
    t.join()

# Sorting the host by latency
sorted_dictionary = dict(sorted(mirrors.items(), key=lambda item: item[1]))
# print(sorted_dictionary)

# Selecting Best mirror with lowest latency
first_element = next(iter(sorted_dictionary))
# print(first_element)
print(f"[{INFO}] Fastest Mirror : {first_element}")

print(f"[{INFO}] Making Backup of source.list file ...")
# Making Backup
copyfile('/etc/apt/sources.list', '/etc/apt/sources.list.bk')

print(f"[{INFO}] Updating sources.list with new entry ...")
# Commenting older entries 
with open('/etc/apt/sources.list','r') as f:
    lines = f.readlines()

with open('/etc/apt/sources.list','w') as f:
    for line in lines:
        f.write('#'+line)
    f.write(' \n')
    # Adding new entry
    f.write('#Autogenerated script by MirrorScripts \n')
    f.write(f"deb http://{first_element}/kali kali-rolling main contrib non-free \n")
    f.write(f"#deb-src http://{first_element}/kali kali-rolling main contrib non-free \n")

print("["+'\033[38;5;50m'+'Done'+'\033[0m'+"]")