import os import socket import datetime import threading from time import sleep right_now = datetime.datetime.now class NetScanner: def __init__(self): self.found = {} self.scanned = 0 self.total_ips = 0 self.total_ports = 0 self.current_addr = None self.current_port = None def _reset(self): self.__init__() print() # ======================================================== # # # Host Scanning # # ======================================================== # def ping_sweep_classC(self, net_addr: str): BASE = '.'.join(net_addr.split('.')[:3]) + '.' print(f"For address: {BASE}__ <--\n") start, end = self.get_ip_range() if (start > 0 and start < 255) and \ (end > start and end <= 255): choice = input(f"\nScan {BASE}{start} - {BASE}{end} [y/n]?: ") self.total_ips = len(range(start, end+1)) if choice.lower() == 'y': print("Scanning...\n") loader = LoaderThread(self) loader.start() t1 = right_now() for n in range(start, end+1): self.scanned += 1 addr = f"{BASE}{n}" self.current_addr = addr if self._ping_addr_found(addr): self.found[n] = addr t2 = right_now() loader.terminate() total = t2 - t1 print(f"\nScan completed in: {str(total).split('.')[0]}") self._print_results() self._reset() else: print("Ping Sweep canceled.\n") else: print("Incorrect Input! Try again...\n") self.ping_sweep_classC(net_addr) self._reset() def thread_ping_scan(self, net_addr: str, threads=26): BASE = '.'.join(net_addr.split('.')[:3]) + '.' print(f"For address: {BASE}__ <--\n") start, end = self.get_ip_range() if threads > self.total_ips: threads = self.total_ips // 3 t1 = right_now() ips_per_thread = self.total_ips // threads chunk_start = start chunk_end = start + ips_per_thread thread_list = [] for i in range(threads): if i == (threads-1): t = PingThread(BASE, chunk_start, int(end), self) else: t = PingThread(BASE, chunk_start, chunk_end, self) t.start() thread_list.append(t) chunk_start = (chunk_end + 1) chunk_end = (chunk_start + ips_per_thread) print(f"[*]Number of active threads: {threading.activeCount()}\n") loader = LoaderThread(self) loader.start() for t in thread_list: t.join() loader.join() print("\n[*]Scan 100% Complete! Exiting Main Thread\n") t2 = right_now() self._print_results() total = t2 - t1 print(f"\nScan completed in: {str(total).split('.')[0]}") self._reset() def tcp_scan(self, net_addr: str): BASE = '.'.join(net_addr.split('.')[:3]) + '.' print(f"For address: {BASE}__ <--\n") start, end = self.get_ip_range() if (start > 0 and start < 255) and \ (end > start and end <= 255): choice = input(f"\nScan {BASE}{start} - {BASE}{end} [y/n]?: ") if choice.lower() == 'y': print("Scanning...\n") t1 = right_now() for n in range(start, end+1): if self._scan_tcp(f"{BASE}{n}"): self.found[n] = f"{BASE}{n}" t2 = right_now() total = t2 - t1 self._print_results() print(f"\nScan completed in: {str(total).split('.')[0]}") else: print("TCP Scan canceled.\n") else: print("Incorrect Input! Try again...\n") self.tcp_scan(net_addr) self._reset() def thread_tcp_scan(self, net_addr: str, threads=26): BASE = '.'.join(net_addr.split('.')[:3]) + '.' print(f"For address: {BASE}__ <--\n") start, end = self.get_ip_range() if (start > 0 and start < 255) and \ (end > start and end <= 255): choice = input(f"\nScan {BASE}{start} - {BASE}{end} [y/n]?: ") if choice.lower() == 'y': print("Scanning...\n") if threads > self.total_ips: threads = self.total_ips // 3 t1 = right_now() ips_per_thread = self.total_ips // threads chunk_start = start chunk_end = start + ips_per_thread thread_list = [] loader = LoaderThread(self) for i in range(threads): if i == (threads-1): t = TCPScanThread(BASE, chunk_start, int(end), self, loader) else: t = TCPScanThread(BASE, chunk_start, chunk_end, self, loader) t.start() thread_list.append(t) chunk_start = (chunk_end + 1) chunk_end = (chunk_start + ips_per_thread) print(f"[*]Number of active threads: {threading.activeCount() - 1}\n") loader.start() for t in thread_list: t.join() loader.join() print("\n[*]Scan 100% Complete! Exiting Main Thread\n") t2 = right_now() total = t2 - t1 self._print_results() print(f"\nScan completed in: {str(total).split('.')[0]}") self._reset() def thread_port_scan(self, ip_addr: str, threads=15): use_default = input("Scan all ports [y/n]?: ").lower() if use_default == 'y': port_range = range(1,65000+1) else: s = int(input("Enter start of port range: ")) e = int(input("Enter end of port range: ")) port_range = range(s,e+1) self.total_ports = len(port_range) if self.total_ports < threads: if self.total_ports > 5: threads = 5 else: threads = self.total_ports elif self.total_ports >= 1000 and self.total_ports <= 9999: threads = 25 elif self.total_ports >= 10000 and self.total_ports <= 24999: threads = 50 elif self.total_ports >= 25000 and self.total_ports <= 65000: threads = 100 ports_per_thread = self.total_ports // threads chunk_start = 0 chunk_end = (chunk_start + ports_per_thread) t1 = right_now() loader = PortLoaderThread(self) thread_list = [] for worker in range(threads): if worker == threads - 1: t = PortScanThread(ip_addr, port_range[chunk_start], port_range[-1], self, loader) else: t = PortScanThread(ip_addr, port_range[chunk_start], port_range[chunk_end], self, loader) t.start() thread_list.append(t) chunk_start = (chunk_end + 1) chunk_end = (chunk_start + ports_per_thread) print(f"[*]Number of active threads: {threading.activeCount() - 1}\n") loader.start() for t in thread_list: t.join() loader.join() print("\n[*]Scan 100% Complete! Exiting Main Thread\n") t2 = right_now() total = t2 - t1 self._print_port_results() print(f"\nScan completed in: {str(total).split('.')[0]}") self._reset() def _scan_tcp(self, addr: str): with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: socket.setdefaulttimeout(1) ports = [20,21,22,80,135,138,139,443,8080] #print(f"\nCurrent Host: {addr}") for port in ports: self.current_port = port #print(f"\rChecking port: {port}...", end='') if sock.connect_ex((addr, port)) == 0: return 1 return 0 def _print_results(self): print("Results:\n========\n") if self.found: for n in sorted(self.found.keys()): print("--- " + f"{self.found[n]}".ljust(14, ' ') + "---> LIVE") else: print("No Results Found.") def _print_port_results(self): print("Results:".center(21, ' ')) print('='*21) if self.found: print("PORT".ljust(10, ' ') + "SERVICE") for n in sorted(self.found.keys()): print(f"{n}".ljust(10, ' ') + f"{self.found[n]}") else: print("No Results Found.\n") def get_ip_range(self): """Returns start and end 4th octects to scan in List""" start = int(input("Enter start of range: ")) end = int(input("Enter end of range: ")) self.total_ips = (len(range(start, end+1))) return [start, end] @staticmethod def _ping_addr_found(addr): """Returns True if able to ping given address""" res = os.popen(f"ping -c 1 {addr}") for line in res.readlines(): if "ttl" in line.lower(): return 1 return 0 @staticmethod def _get_port_service(port: int): """Returns port service if known, "unkwn" otherwise.""" try: return socket.getservbyport(port) except OSError: return "unkwn" # ======================================================== # # # Threads # # ======================================================== # class PingThread(threading.Thread): def __init__(self, base, _start, _end, scanner_obj): super().__init__() self.base = base self.scanner = scanner_obj self._start = _start self._end = _end def run(self): self._scan(self._start, self._end) def _scan(self, start, end): for n in range(start, end+1): addr = f"{self.base}{n}" self.scanner.current_addr = addr #print(f"Scanning {addr}...") if self.scanner._ping_addr_found(addr): self.scanner.found[n] = addr self.scanner.scanned += 1 class TCPScanThread(threading.Thread): def __init__(self, base, _start, _end, scanner_obj, loader_obj): super().__init__() self.base = base self._start = _start self._end = _end self.scanner = scanner_obj self.loader = loader_obj def run(self): self._scan(self._start, self._end) def _scan(self, start, end): for n in range(start, end+1): addr = f"{self.base}{n}" self.scanner.current_addr = addr #print(f"Scanning {addr}...") try: if self.scanner._scan_tcp(addr): #print(f"{addr} ---> LIVE") self.scanner.found[n] = addr except socket.gaierror: print(f"[!]No Host Associated with {addr}[!]") self.loader.terminate() self.scanner.scanned += 1 class PortScanThread(threading.Thread): def __init__(self, addr, start_port, end_port, scanner_obj: NetScanner, loader_obj): super().__init__() self.addr = addr self.begin = start_port self.end = end_port self.scanner = scanner_obj self.loader = loader_obj def run(self): self._scan() def _scan(self): self.scanner.current_addr = self.addr with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: socket.setdefaulttimeout(.5) for port in range(self.begin, self.end+1): self.scanner.current_port = port try: if sock.connect_ex((self.addr, port)) == 0: self.scanner.found[port] = self.scanner._get_port_service(port) except socket.gaierror: print("[!]") continue self.scanner.scanned += 1 self.loader.terminate() class LoaderThread(threading.Thread): def __init__(self, scanner_obj: NetScanner): super().__init__() self.scanner = scanner_obj self._running = True def run(self): self.load() def load(self): """Prints Completion Percentage Info""" while self.scanner.scanned != self.scanner.total_ips and self._running: print(f"{(self.scanner.scanned/self.scanner.total_ips) * 100:.2f}% Scanning: " + f"{self.scanner.current_addr}:{self.scanner.current_port}".ljust(18, ' ') + '|' + ('#'*((self.scanner.scanned * 20) // self.scanner.total_ips)).ljust(20, ' ') + '|', end='\r') sleep(.5) def terminate(self): """Stops Loader""" self._running = False class PortLoaderThread(threading.Thread): def __init__(self, scanner_obj: NetScanner): super().__init__() self.scanner = scanner_obj self._running = True def run(self): self.load() def load(self): """Prints Completion Percentage Info""" while self.scanner.scanned != self.scanner.total_ports and self._running: print(f"{(self.scanner.scanned/self.scanner.total_ports) * 100:.2f}% Scanning: {self.scanner.current_addr}:" + f"{self.scanner.current_port}".ljust(6, ' ') + '|' + ('#'*((self.scanner.scanned * 20) // self.scanner.total_ports)).ljust(20, ' ') + '|', end='\r') sleep(.5) def terminate(self): """Stops Loader""" self._running = False # ======================================================== # # # Main Loop # # ======================================================== # def main(): """Main Program Loop""" ns = NetScanner() options = { 1: ns.thread_ping_scan, 2: ns.thread_tcp_scan, 3: ns.thread_port_scan } print("Scan Type:\n==========") print(""" Host Scanning ------------- [1] Threaded Ping Sweep [2] Threaded TCP Scan Port Scanning ------------- [3] Threaded Port Scan """) choice = input('> ') if int(choice) in options.keys(): addr = input("\nEnter network address: ") options[int(choice)](addr) else: print("\nScan Type Option Not Availible.\n") main() if __name__ == "__main__": print("Net Scanner\n\n") while True: main()

netscanner_v3.py