added python3 support and multithreading

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())

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)

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()
+        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)