From ba3db0afc45dc7ecb1aea841444301ccc5b04cd1 Mon Sep 17 00:00:00 2001 From: Aryan Puri Date: Wed, 15 Oct 2025 14:46:39 +0530 Subject: [PATCH 1/2] added support for python3 in ipscanner --- src/ipscanner_py3.py | 80 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 80 insertions(+) create mode 100644 src/ipscanner_py3.py diff --git a/src/ipscanner_py3.py b/src/ipscanner_py3.py new file mode 100644 index 0000000..f9e4178 --- /dev/null +++ b/src/ipscanner_py3.py @@ -0,0 +1,80 @@ +#!/usr/bin/env python3 +import socket +from datetime import datetime +import json +import os +from multi.scanner_thread import split_processing + + +def get_absolute_path(relative_path: str) -> str: + """Resolve relative path (like '../config.json') to absolute path.""" + base_dir = os.path.dirname(os.path.abspath(__file__)) + normalized = os.path.normpath(relative_path) + return os.path.join(base_dir, normalized) + + +# ---------- Core scan function ---------- +def scan(addr: str, port: int = 135, timeout: float = 1.0) -> bool: + """Return True if given addr:port accepts a TCP connection.""" + try: + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.settimeout(timeout) + result = sock.connect_ex((addr, port)) + sock.close() + return result == 0 + except Exception: + return False + +# threading +def run1(ips: list, start: int, end: int) -> None: + """Scan a subset of IPs (ips[start:end]) for live hosts.""" + for ip in ips[start:end]: + if scan(ip): + print(f"{ip} is live") + +# ---------- Main program ---------- +if __name__ == "__main__": + # Input base IP (e.g. 192.168.1.1) + net1 = input("Enter the IP address: ").strip() + net2 = net1.split('.') + if len(net2) < 3: + print(" Please enter a valid IPv4 address (e.g. 192.168.1.1).") + raise SystemExit(1) + + # Build prefix like "192.168.1." + net3 = f"{net2[0]}.{net2[1]}.{net2[2]}." + + # Banner + print("-" * 60) + print(f"Please wait, scanning IP address.... {net3}XXX") + print("-" * 60) + + td1 = datetime.now() + + # Load configuration + try: + with open(get_absolute_path('../config.json'), 'r', encoding='utf-8') as config_file: + config = json.load(config_file) + + range_low = int(config['ipRange']['low']) + range_high = int(config['ipRange']['high']) + CONST_NUM_THREADS = int(config['thread']['count']) + + except FileNotFoundError: + print(" config.json file not found at", get_absolute_path('../config.json')) + raise + except (KeyError, ValueError, json.JSONDecodeError) as e: + print(" Error reading config.json:", e) + raise + + + ips = [f"{net3}{i}" for i in range(range_low, range_high + 1)] + + # Start threaded scanning + split_processing(ips, CONST_NUM_THREADS, run1) + + td2 = datetime.now() + total = td2 - td1 + print("-" * 60) + print(" Scanning completed in", total) + print("-" * 60) From 07e4cfae3799ad625c6da10e52e57230593e32d0 Mon Sep 17 00:00:00 2001 From: Aryan Puri Date: Wed, 15 Oct 2025 15:04:25 +0530 Subject: [PATCH 2/2] added python3 fix --- src/ipscanner.py | 161 ++++++++++++++++++++---------------- src/multi/scanner_thread.py | 39 ++++++--- src/scanner.py | 150 ++++++++++++++++++--------------- src/single/scanner.py | 137 ++++++++++++++++++++++-------- 4 files changed, 299 insertions(+), 188 deletions(-) diff --git a/src/ipscanner.py b/src/ipscanner.py index 09775ca..a543509 100644 --- a/src/ipscanner.py +++ b/src/ipscanner.py @@ -1,80 +1,95 @@ +#!/usr/bin/env python3 +"""Multithreaded IP scanner CLI. + +This script scans a range of IP addresses in a subnet to find live hosts using multiple threads. +It loads configuration from config.json for IP range and thread count. +""" +from __future__ import annotations + import socket from datetime import datetime import json import os +import sys +from typing import List from multi.scanner_thread import split_processing -# Ask for input -net1 = raw_input('Enter the IP address: ') -net2 = net1.split('.') -a = '.' -net3 = net2[0] + a + net2[1] + a + net2[2] + a - -# Print a nice banner with information on which host we are about to scan -print "-" * 60 -print "Please wait, scanning IP address....", net3+"XXX" -print "-" * 60 - -# Resolves the relative path to absolute path -# [BUG]: https://github.com/vinitshahdeo/PortScanner/issues/19 - - -def get_absolute_path(relative_path): - dir = os.path.dirname(os.path.abspath(__file__)) - split_path = relative_path.split("/") - absolute_path = os.path.join(dir, *split_path) - return absolute_path - - -# Check what time the scan started -td1 = datetime.now() - -try: - with open(get_absolute_path('../config.json')) as config_file: - config = json.load(config_file) - # print get_absolute_path('../config.json') - range_low = int(config['ipRange']['low']) - range_high = int(config['ipRange']['high']) - # defining number of threads running concurrently - CONST_NUM_THREADS = int(config['thread']['count']) - -except IOError: - print("config.json file not found") -except ValueError: - print("Kindly check the json file for appropriateness of range") - -ips = list(range(range_low, range_high, 1)) -# scanning the port only in range of (range_low, range_high) -range_high = range_high + 1 -# including the last address at 'range_high' - - -def scan(addr): - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - socket.setdefaulttimeout(1) - result = sock.connect_ex((addr, 135)) - # 'result' is used as error indicator, port 135 is used for Windows - # 137, 138, 139, 445 can also be used - if result == 0: - # tests if IP address is live + +def get_absolute_path(relative_path: str) -> str: + """Resolve relative path to absolute path.""" + base_dir = os.path.dirname(os.path.abspath(__file__)) + normalized = os.path.normpath(relative_path) + return os.path.join(base_dir, normalized) + + +def scan(addr: str, port: int = 135, timeout: float = 1.0) -> bool: + """Return True if addr:port accepts a TCP connection (host is live).""" + try: + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.settimeout(timeout) + result = sock.connect_ex((addr, port)) + sock.close() + return result == 0 + except Exception: + return False + + +def run1(ips: List[str], start: int, end: int, net_prefix: str) -> None: + """Scan a subset of IPs (ips[start:end]) for live hosts.""" + for i in range(start, end): + addr = f"{net_prefix}{ips[i]}" + if scan(addr): + print(f"{addr} is live") + + +def main() -> int: + # Ask for input + net1 = input("Enter the IP address (e.g., 192.168.1.1): ").strip() + net2 = net1.split('.') + if len(net2) != 4: + print("Please enter a valid IPv4 address.") return 1 - else: - return 0 - - -def run1(ips, range_low, range_high): - for ip in xrange(range_low, range_high): - addr = net3+str(ip) - # gets full address - if (scan(addr)): - print addr + " is live\n" - - -# calling function from scanner_thread.py for multithreading -split_processing(ips, CONST_NUM_THREADS, run1, range_low, range_high) -# Checking the time again -td2 = datetime.now() -# Calculates the difference of time, to see how long it took to run the script -total = td2-td1 -# Printing the information to screen -print "Scanning completed in ", total + + # Build prefix like "192.168.1." + net_prefix = f"{net2[0]}.{net2[1]}.{net2[2]}." + + # Print banner + print("-" * 60) + print(f"Please wait, scanning IP addresses: {net_prefix}XXX") + print("-" * 60) + + # Start time + td1 = datetime.now() + + # Load config + try: + config_path = get_absolute_path('../config.json') + with open(config_path, 'r', encoding='utf-8') as config_file: + config = json.load(config_file) + range_low = int(config['ipRange']['low']) + range_high = int(config['ipRange']['high']) + const_num_threads = int(config['thread']['count']) + except FileNotFoundError: + print("config.json file not found.") + return 1 + except (KeyError, ValueError, json.JSONDecodeError) as e: + print(f"Error reading config.json: {e}") + return 1 + + # Generate IP list (as strings for the last octet) + ips = [str(i) for i in range(range_low, range_high + 1)] + + # Multithreaded scan + split_processing(ips, const_num_threads, lambda p, s, e: run1(p, s, e, net_prefix), 0, len(ips)) + + # End time + td2 = datetime.now() + total = td2 - td1 + print("-" * 60) + print(f"Scanning completed in {total}") + print("-" * 60) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/multi/scanner_thread.py b/src/multi/scanner_thread.py index 549fee8..fd539ba 100644 --- a/src/multi/scanner_thread.py +++ b/src/multi/scanner_thread.py @@ -1,19 +1,32 @@ import threading +from typing import Callable, List, Tuple -def split_processing(ports, num_splits, scan, range_low, range_high): - split_size = (range_high-range_low) // num_splits - threads = [] +def split_data(range_low: int, range_high: int, num_splits: int) -> List[Tuple[int, int]]: + """Split a range into num_splits sub-ranges, returning list of (start, end) tuples.""" + split_size = (range_high - range_low) // num_splits + splits: List[Tuple[int, int]] = [] for i in range(num_splits): - # determine the indices of the list this thread will handle - start = i * split_size - # special case on the last chunk to account for uneven splits - end = range_high if i+1 == num_splits else (i+1) * split_size - # create the thread - threads.append( - threading.Thread(target=scan, args=(ports, start, end))) - threads[-1].start() # start the thread we just created + start = range_low + i * split_size + # Special case for the last chunk to account for uneven splits + end = range_high if i + 1 == num_splits else range_low + (i + 1) * split_size + splits.append((start, end)) + return splits - # wait for all threads to finish + +def run_threads(scan_func: Callable, ports, splits: List[Tuple[int, int]]) -> None: + """Run scan_func in separate threads for each split, passing (ports, start, end).""" + threads: List[threading.Thread] = [] + for start, end in splits: + t = threading.Thread(target=scan_func, args=(ports, start, end)) + threads.append(t) + t.start() + # Wait for all threads to finish for t in threads: - t.join() \ No newline at end of file + t.join() + + +def split_processing(ports, num_splits: int, scan: Callable, range_low: int, range_high: int) -> None: + """Split processing across multiple threads for the given range.""" + splits = split_data(range_low, range_high, num_splits) + run_threads(scan, ports, splits) \ No newline at end of file diff --git a/src/scanner.py b/src/scanner.py index 1cdf98b..2f7ced9 100644 --- a/src/scanner.py +++ b/src/scanner.py @@ -1,87 +1,105 @@ -#!/usr/bin/env python +#!/usr/bin/env python3 +"""Multithreaded port scanner CLI. + +This script scans a range of TCP ports on a single host using multiple threads. +It loads configuration from config.json for port range and thread count. +""" +from __future__ import annotations + import socket import subprocess import sys +import os from datetime import datetime import json -import os import threading -import __builtin__ +import builtins +from typing import List from multi.scanner_thread import split_processing -exc = getattr(__builtin__, "IOError", "FileNotFoundError") - -# Clear the screen -subprocess.call('clear', shell=True) - -# Ask for input -remoteServer = raw_input("Enter a remote host to scan: ") -remoteServerIP = socket.gethostbyname(remoteServer) - -# Print a nice banner with information on which host we are about to scan -print "-" * 60 -print "Please wait, scanning remote host....", remoteServerIP -print "-" * 60 - -# Resolves the relative path to absolute path -# [BUG]: https://github.com/vinitshahdeo/PortScanner/issues/19 -def get_absolute_path(relative_path): - dir = os.path.dirname(os.path.abspath(__file__)) - split_path = relative_path.split("/") - absolute_path = os.path.join(dir, *split_path) - return absolute_path - -# Check what time the scan started -t1 = datetime.now() - -# Getting port range values from config.json -try: - with open(get_absolute_path('../config.json')) as config_file: - config = json.load(config_file) - print get_absolute_path('../config.json') - range_high = int(config['range']['high']) - range_low = int(config['range']['low']) - # defining number of threads running concurrently - CONST_NUM_THREADS = int(config['thread']['count']) - -except IOError: - print("config.json file not found") -except ValueError: - print("Kindly check the json file for appropriateness of range") - -ports = list(range(range_low, range_high, 1)) -# scanning the port only in range of (range_low, range_high) - -def scan(ports, range_low, range_high): + +def clear_screen() -> None: + """Clear the terminal screen in a cross-platform way.""" + if os.name == "nt": + os.system("cls") + else: + os.system("clear") + + +def get_absolute_path(relative_path: str) -> str: + """Resolve relative path to absolute path.""" + base_dir = os.path.dirname(os.path.abspath(__file__)) + normalized = os.path.normpath(relative_path) + return os.path.join(base_dir, normalized) + + +def scan(ports: List[int], range_low: int, range_high: int, remote_ip: str) -> None: + """Scan ports from range_low to range_high on remote_ip.""" try: for port in range(range_low, range_high): sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - result = sock.connect_ex((remoteServerIP, port)) + sock.settimeout(1) # Add timeout + result = sock.connect_ex((remote_ip, port)) if result == 0: - print "Port {}: Open".format(port) + print(f"Port {port}: Open") sock.close() - except KeyboardInterrupt: - print "You pressed Ctrl+C" - sys.exit() - + print("Scan interrupted by user.") + sys.exit(130) except socket.gaierror: - print 'Hostname could not be resolved. Exiting' - sys.exit() - + print("Hostname could not be resolved. Exiting.") + sys.exit(1) except socket.error: - print "Couldn't connect to server" - sys.exit() + print("Couldn't connect to server. Exiting.") + sys.exit(1) -# calling function from scanner_thread.py for multithreading -split_processing(ports, CONST_NUM_THREADS, scan, range_low, range_high) +def main() -> int: + clear_screen() -# Checking the time again -t2 = datetime.now() + # Ask for input + remote_server = input("Enter a remote host to scan: ").strip() + try: + remote_server_ip = socket.gethostbyname(remote_server) + except socket.gaierror: + print("Error: Hostname could not be resolved.") + return 1 -# Calculates the difference of time, to see how long it took to run the script -total = t2 - t1 + # Print banner + print("-" * 60) + print(f"Please wait, scanning remote host {remote_server} ({remote_server_ip})") + print("-" * 60) -# Printing the information to screen -print 'Scanning Completed in: ', total + # Start time + t1 = datetime.now() + + # Load config + try: + config_path = get_absolute_path('../config.json') + with open(config_path, 'r', encoding='utf-8') as config_file: + config = json.load(config_file) + print(f"Loaded config from {config_path}") + range_high = int(config['range']['high']) + range_low = int(config['range']['low']) + const_num_threads = int(config['thread']['count']) + except FileNotFoundError: + print("config.json file not found.") + return 1 + except (KeyError, ValueError, json.JSONDecodeError) as e: + print(f"Error reading config.json: {e}") + return 1 + + ports = list(range(range_low, range_high + 1)) # Include range_high + + # Multithreaded scan + split_processing(ports, const_num_threads, lambda p, rl, rh: scan(p, rl, rh, remote_server_ip), range_low, range_high + 1) + + # End time + t2 = datetime.now() + total = t2 - t1 + print(f"Scanning completed in: {total}") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/single/scanner.py b/src/single/scanner.py index b135557..c1e9295 100644 --- a/src/single/scanner.py +++ b/src/single/scanner.py @@ -1,51 +1,116 @@ -#!/usr/bin/env python +#!/usr/bin/env python3 +"""Single-threaded port scanner CLI. + +This script scans a range of TCP ports on a single host without using +multithreading. It provides a small CLI to control host, ports and timeout. +""" +from __future__ import annotations + +import argparse import socket -import subprocess import sys +import time +import os from datetime import datetime +from typing import Iterable, List, Tuple + + +def clear_screen() -> None: + """Clear the terminal screen in a cross-platform way.""" + if os.name == "nt": + os.system("cls") + else: + os.system("clear") + + +def parse_ports(ports: str) -> Iterable[int]: + """Parse port argument. Accepts single port '80', comma list '22,80', or range '1-1024'.""" + ports = ports.strip() + if not ports: + return [] + result = set() + for part in ports.split(','): + part = part.strip() + if '-' in part: + start_s, end_s = part.split('-', 1) + start, end = int(start_s), int(end_s) + if start > end: + start, end = end, start + result.update(range(max(1, start), min(65535, end) + 1)) + else: + p = int(part) + if 1 <= p <= 65535: + result.add(p) + return sorted(result) + + +def scan_port(host: str, port: int, timeout: float) -> bool: + """Return True if TCP port is open on host, otherwise False.""" + try: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: + sock.settimeout(timeout) + # connect_ex returns 0 on success + return sock.connect_ex((host, port)) == 0 + except Exception: + return False + + +def scan_host(host: str, ports: Iterable[int], timeout: float) -> List[int]: + """Scan the given ports on host (single-threaded). Returns list of open ports.""" + open_ports: List[int] = [] + for port in ports: + if scan_port(host, port, timeout): + print(f"Port {port}: Open") + open_ports.append(port) + return open_ports -# Clear the screen -subprocess.call('clear', shell=True) -# Ask for input -remoteServer = raw_input("Enter a remote host to scan: ") -remoteServerIP = socket.gethostbyname(remoteServer) +def main(argv: List[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="Single-threaded TCP port scanner") + parser.add_argument("host", help="Hostname or IP to scan") + parser.add_argument("--ports", default="1-1024", + help="Ports to scan. Examples: '22', '22,80,443', '1-1024' (default)") + parser.add_argument("--timeout", type=float, default=0.5, + help="Connection timeout in seconds (default: 0.5)") + parser.add_argument("--no-clear", action="store_true", + help="Do not clear the screen before running") -# Print a nice banner with information on which host we are about to scan -print "-" * 60 -print "Please wait, scanning remote host....", remoteServerIP -print "-" * 60 + args = parser.parse_args(argv) -# Check what time the scan started -t1 = datetime.now() + try: + remote_ip = socket.gethostbyname(args.host) + except socket.gaierror: + print("Error: Hostname could not be resolved.") + return 2 -# scanning the port only in range of (1, 8888) + if not args.no_clear: + clear_screen() -try: - for port in range(1,8888): - sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - result = sock.connect_ex((remoteServerIP, port)) - if result == 0: - print "Port {}: Open".format(port) - sock.close() + print("-" * 60) + print(f"Scanning {args.host} ({remote_ip})") + print("-" * 60) -except KeyboardInterrupt: - print "You pressed Ctrl+C" - sys.exit() + try: + ports = list(parse_ports(args.ports)) + if not ports: + print("No valid ports to scan. Exiting.") + return 3 -except socket.gaierror: - print 'Hostname could not be resolved. Exiting' - sys.exit() + start_time = datetime.now() + open_ports = scan_host(remote_ip, ports, args.timeout) + duration = datetime.now() - start_time -except socket.error: - print "Couldn't connect to server" - sys.exit() + print("-" * 60) + print(f"Scan completed in {duration}") + print(f"Open ports: {len(open_ports)}") + if open_ports: + print(", ".join(str(p) for p in open_ports)) + return 0 -# Checking the time again -t2 = datetime.now() + except KeyboardInterrupt: + print("\nScan interrupted by user.") + return 130 -# Calculates the difference of time, to see how long it took to run the script -total = t2 - t1 -# Printing the information to screen -print 'Scanning Completed in: ', total +if __name__ == "__main__": + sys.exit(main())