#5

src/ipscanner/__init__.py

@@ -0,0 +1,4 @@
+# src/ipscanner/__init__.py
+"""IP Scanner package for Python 3.8"""
+from .ipscanner import expand_cidr, expand_range, scan_hosts_parallel
+__all__ = ["expand_cidr", "expand_range", "scan_hosts_parallel"]

src/ipscanner/ipscanner.py

@@ -0,0 +1,63 @@
+# src/ipscanner/ipscanner.py
+"""
+Utilities to expand IP ranges/CIDR notation and scan hosts in parallel.
+Note: This does NOT use raw ICMP; it uses TCP connection attempts to a probe port
+(e.g., 80 or 443) to heuristically detect live hosts without sudo privileges.
+"""
+
+from typing import List, Iterable, Tuple
+import ipaddress
+from concurrent.futures import ThreadPoolExecutor, as_completed
+import socket
+import contextlib
+
+DEFAULT_PROBE_PORT = 80
+DEFAULT_TIMEOUT = 0.6
+
+def expand_cidr(cidr: str) -> List[str]:
+    """Return list of IP strings for a CIDR block (excluding network and broadcast for IPv4 if desired)."""
+    net = ipaddress.ip_network(cidr, strict=False)
+    return [str(ip) for ip in net.hosts()]
+
+def expand_range(range_spec: str) -> List[str]:
+    """
+    Expand a range like '192.168.1.1-192.168.1.10' into list of IPs.
+    Or single IP string returns [ip].
+    """
+    if "-" in range_spec:
+        start_s, end_s = range_spec.split("-", 1)
+        start = ipaddress.ip_address(start_s.strip())
+        end = ipaddress.ip_address(end_s.strip())
+        if start > end:
+            start, end = end, start
+        out = []
+        cur = int(start)
+        while cur <= int(end):
+            out.append(str(ipaddress.ip_address(cur)))
+            cur += 1
+        return out
+    else:
+        # Single address
+        return [str(ipaddress.ip_address(range_spec.strip()))]
+
+def _probe_host(host: str, port: int = DEFAULT_PROBE_PORT, timeout: float = DEFAULT_TIMEOUT) -> Tuple[str, bool]:
+    with contextlib.closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock:
+        sock.settimeout(timeout)
+        try:
+            sock.connect((host, port))
+            return (host, True)
+        except Exception:
+            return (host, False)
+
+def scan_hosts_parallel(hosts: Iterable[str],
+                        probe_port: int = DEFAULT_PROBE_PORT,
+                        timeout: float = DEFAULT_TIMEOUT,
+                        max_workers: int = 100) -> List[Tuple[str, bool]]:
+    """Return list of tuples (host, up_bool)."""
+    hosts = list(hosts)
+    results = []
+    with ThreadPoolExecutor(max_workers=max_workers) as ex:
+        futures = {ex.submit(_probe_host, h, probe_port, timeout): h for h in hosts}
+        for fut in as_completed(futures):
+            results.append(fut.result())
+    return sorted(results, key=lambda x: x[0])

src/mainScanner.py

@@ -1,110 +1,48 @@
-#!/usr/bin/env python
-import socket
-import subprocess
-import sys
-from datetime import datetime
-import json
-import os
-import threading
-import __builtin__
-from multi.scanner_thread import split_processing
-import logging
-from flask import Flask, render_template, request, redirect, url_for
-
-app = Flask(__name__)
-
-
-@app.route('/')
-def homepage():
-    return render_template('index.html')
-
-
-exc = getattr(__builtin__, "IOError", "FileNotFoundError")
-
-# Clear the screen
-# subprocess.call('clear', shell=True)
-
-
-@app.route('/input', methods=["POST"])
-def input():
-    # Ask for input
-    if request.method == "POST":
-        remoteServer = request.form["host"]
-        remoteServerIP = socket.gethostbyname(remoteServer)
-        range_low = int(request.form["range_low"])
-        range_high = int(request.form["range_high"])
+# src/mainScanner.py
+"""Multithreaded port scanning CLI - Python 3.8"""
+
+import argparse
+from src.multi.scanner_thread import threaded_port_scan
+from typing import List
+
+def parse_ports(spec: str) -> List[int]:
+    """Parse port specification strings like '22,80,443,1000-2000'"""
+    out = set()
+    for token in spec.split(","):
+        token = token.strip()
+        if not token:
+            continue
+        if "-" in token:
+            a,b = token.split("-",1)
+            out.update(range(int(a), int(b)+1))
+        else:
+            out.add(int(token))
+    return sorted(out)
+
+def main():
+    parser = argparse.ArgumentParser(description="Multithreaded Port Scanner (py3)")
+    parser.add_argument("host", help="Target host (ip or hostname)")
+    parser.add_argument("--ports", "-p", default="1-1024",
+                        help="Ports as comma-separated list and ranges e.g. 22,80,443,1000-2000")
+    parser.add_argument("--timeout", type=float, default=1.0)
+    parser.add_argument("--workers", type=int, default=100, help="Number of concurrent threads")
+    args = parser.parse_args()
+
+    ports = parse_ports(args.ports)
+    print(f"Scanning {args.host} on {len(ports)} ports with {args.workers} workers...")
+
+    results = threaded_port_scan(args.host, ports, timeout=args.timeout, max_workers=args.workers)
+
+    open_ports = [p for p, open_ in results if open_]
+    for port, open_ in results:
+        print(f"{port}: {'OPEN' if open_ else 'closed'}")
+
+    print("\nSummary:")
+    if open_ports:
+        print(f"Open ports: {', '.join(str(p) for p in open_ports)}")
     else:
-        return EnvironmentError
-
-    # 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')
-        # 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)
-
-    portnum = []
-
-    def scan(ports, range_low, range_high):
-        try:
-            for port in range(range_low, range_high):
-                sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-                result = sock.connect_ex((remoteServerIP, port))
-                if result == 0:
-                    print "Port {}: 	 Open".format(port)
-                    portnum.append("Port "+str(port))
-                sock.close()
-
-        except KeyboardInterrupt:
-            print "You pressed Ctrl+C"
-            sys.exit()
-
-        except socket.gaierror:
-            print 'Hostname could not be resolved. Exiting'
-            sys.exit()
-
-        except socket.error:
-            print "Couldn't connect to server"
-            sys.exit()
-
-    # calling function from scanner_thread.py for multithreading
-    split_processing(ports, CONST_NUM_THREADS, scan, range_low, range_high)
-
-    # Checking the time again
-    t2 = datetime.now()
-
-    # 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
-    return render_template('index.html', portnum=portnum, host=remoteServerIP, range_low=range_low, range_high=range_high, total=total)
+        print("No open ports found.")
 
+if __name__ == "__main__":
+    main()
 
-if __name__ == '__main__':
-    app.run(debug=True)

src/multi/scanner_thread.py

@@ -1,19 +1,40 @@
-import threading
+# src/multi/scanner_thread.py
+"""Thread-worker utilities for multithreaded port scanning (Python 3.8)."""
 
+from typing import Tuple, List, Iterable, Callable
+import socket
+import contextlib
+from concurrent.futures import ThreadPoolExecutor, as_completed
 
-def split_processing(ports, num_splits, scan, range_low, range_high):
-    split_size = (range_high-range_low) // num_splits
-    threads = []
-    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
+DEFAULT_TIMEOUT = 1.0
 
-    # wait for all threads to finish
-    for t in threads:
-        t.join()
+def is_port_open(host: str, port: int, timeout: float = DEFAULT_TIMEOUT) -> bool:
+    """Return True if a TCP connection to host:port succeeds."""
+    with contextlib.closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock:
+        sock.settimeout(timeout)
+        try:
+            sock.connect((host, port))
+            return True
+        except Exception:
+            return False
+
+def _scan_one(args: Tuple[str, int, float]) -> Tuple[int, bool]:
+    host, port, timeout = args
+    return (port, is_port_open(host, port, timeout))
+
+def threaded_port_scan(host: str,
+                       ports: Iterable[int],
+                       timeout: float = DEFAULT_TIMEOUT,
+                       max_workers: int = 50) -> List[Tuple[int, bool]]:
+    """
+    Scan ports concurrently on given host.
+    Returns list of (port, is_open), order sorted by port.
+    """
+    ports = list(ports)
+    results = []
+    with ThreadPoolExecutor(max_workers=max_workers) as ex:
+        futures = {ex.submit(_scan_one, (host, p, timeout)): p for p in ports}
+        for fut in as_completed(futures):
+            port, open_ = fut.result()
+            results.append((port, open_))
+    return sorted(results, key=lambda x: x[0])

src/single/scanner.py

@@ -1,51 +1,51 @@
-#!/usr/bin/env python
-import socket
-import subprocess
-import sys
-from datetime import datetime
-
-# 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
-
-# Check what time the scan started
-t1 = datetime.now()
-
-# scanning the port only in range of (1, 8888)
+# src/single/scanner.py
+"""Single-threaded port scanner (Python 3.8)
+Reference implementation: scans ports sequentially.
+Used as fallback or for small scans.
+"""
 
-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()
-
-except KeyboardInterrupt:
-    print "You pressed Ctrl+C"
-    sys.exit()
-
-except socket.gaierror:
-    print 'Hostname could not be resolved. Exiting'
-    sys.exit()
-
-except socket.error:
-    print "Couldn't connect to server"
-    sys.exit()
-
-# Checking the time again
-t2 = datetime.now()
-
-# Calculates the difference of time, to see how long it took to run the script
-total =  t2 - t1
+from typing import Iterable, List, Tuple
+import socket
+import contextlib
+
+DEFAULT_TIMEOUT = 1.0  # seconds
+
+def is_port_open(host: str, port: int, timeout: float = DEFAULT_TIMEOUT) -> bool:
+    """Return True if host:port accepts TCP connection."""
+    with contextlib.closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock:
+        sock.settimeout(timeout)
+        try:
+            sock.connect((host, port))
+            return True
+        except Exception:
+            return False
+
+def scan_ports(host: str, ports: Iterable[int], timeout: float = DEFAULT_TIMEOUT) -> List[Tuple[int, bool]]:
+    results = []
+    for p in ports:
+        open_ = is_port_open(host, p, timeout=timeout)
+        results.append((p, open_))
+    return results
+
+if __name__ == "__main__":
+    import argparse
+    parser = argparse.ArgumentParser(description="Single-threaded port scanner (py3)")
+    parser.add_argument("host", help="Target host (IP or hostname)")
+    parser.add_argument("--ports", help="Comma-separated ports or range e.g. 20-25,80", default="1-1024")
+    parser.add_argument("--timeout", type=float, default=DEFAULT_TIMEOUT)
+    args = parser.parse_args()
+
+    def parse_ports(s: str):
+        out = []
+        for part in s.split(","):
+            if "-" in part:
+                a,b = part.split("-",1)
+                out.extend(range(int(a), int(b)+1))
+            else:
+                out.append(int(part))
+        return out
+
+    ports = parse_ports(args.ports)
+    for port, open_ in scan_ports(args.host, ports, timeout=args.timeout):
+        print(f"{port}: {'OPEN' if open_ else 'closed'}")
 
-# Printing the information to screen
-print 'Scanning Completed in: ', total