diff --git a/src/ipscanner/__init__.py b/src/ipscanner/__init__.py new file mode 100644 index 0000000..77483f8 --- /dev/null +++ b/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"] diff --git a/src/ipscanner/ipscanner.py b/src/ipscanner/ipscanner.py new file mode 100644 index 0000000..073a006 --- /dev/null +++ b/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]) diff --git a/src/mainScanner.py b/src/mainScanner.py index ffca7c3..89fbddb 100644 --- a/src/mainScanner.py +++ b/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) diff --git a/src/multi/scanner_thread.py b/src/multi/scanner_thread.py index 549fee8..bc49eab 100644 --- a/src/multi/scanner_thread.py +++ b/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() \ No newline at end of file +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]) diff --git a/src/single/scanner.py b/src/single/scanner.py index b135557..9be33eb 100644 --- a/src/single/scanner.py +++ b/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