Python3 simple network port scanning tool

network_port_scan.py

import logging
from argparse import ArgumentParser
from concurrent.futures import ThreadPoolExecutor, as_completed
from ipaddress import ip_network
from multiprocessing import cpu_count
from socket import AF_INET, SOCK_STREAM, socket

from time import time

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')


def check_port(host: str, port: int, timeout: int = 1):
    """
    Check if a port is open on a host

    host: Host to check
    port: Port to check
    timeout: Timeout in seconds
    Returns: True if port is open, False otherwise
    """
    try:
        with socket(AF_INET, SOCK_STREAM) as s:
            s.settimeout(timeout)
            response = s.connect_ex((host, port))
            return response == 0
    except OSError as e:
        logging.error(f"Network error checking {host}:{port}: {e}")
        return False
    except ValueError as e:
        logging.error(f"Invalid argument for {host}:{port}: {e}")
        return False
    except Exception as e:
        logging.error(f"Unexpected error checking {host}:{port}: {e}")
        return False


def list_ip_addresses(cidr: str) -> list[str]:
    """
    List all IP addresses in a given CIDR range

    cidr: CIDR notation for the network
    Returns: List of IP addresses in the network
    """
    return [str(ip) for ip in ip_network(cidr, strict=True).hosts()]


if __name__ == "__main__":

    parser = ArgumentParser(description='Scan a network for open ports')
    parser.add_argument('-n', '--network', type=str, required=True, help='CIDR notation for network to scan (default: 192.168.228.0/24)')
    parser.add_argument('-p', '--port', type=int, required=True, help='Port to scan (default: 80)')
    parser.add_argument('-j', '--jobs', type=int, default=10, help='Job multiplier for workers (default: 4)')
    parser.add_argument('-o', '--output', type=str, help='Enable output (json, text, yaml)', default='')
    args = parser.parse_args()

    workers = cpu_count() * args.jobs
    result_addr = list[str]()
    start_time = time()

    with ThreadPoolExecutor(max_workers=workers) as executor:
        future_to_addr = {executor.submit(check_port, ip, args.port): ip for ip in list_ip_addresses(args.network)}

        for future in as_completed(future_to_addr):
            result = future.result()
            if result:
                addr = f"{future_to_addr[future]}:{args.port}"
                if args.output:
                    result_addr.append(addr)
                else:
                    logging.info(f"{addr} is open")

    if args.output == "text":
        print("\n".join(result_addr))
    elif args.output == "json":
        from json import dumps
        print(dumps(result_addr))
    elif args.output == "yaml":
        from yaml import dump
        print(dump(result_addr))
    else:
        elapsed_time = time() - start_time
        logging.info(f"Scan completed in {elapsed_time:.2f} seconds")