Async Port Scanner to identify open ports that are not listed in the Expected Ports. Default expected ports are: "80,443"

check_port_scanner.py

import sys
import socket
import argparse
from datetime import datetime
from typing import Any, List, Set

import asyncio
import pyfiglet


ascii_banner = pyfiglet.figlet_format("PORT SCANNER")
print(ascii_banner)

# Create argument parser
parser = argparse.ArgumentParser(description="Port Scanner")
parser.add_argument(
    "target",
    help="Hostname or IP address to scan",
    default="127.0.0.1",
)
parser.add_argument(
    "--expected_ports",
    help="Comma-separated list of expected open ports",
    default="80,443",
)
parser.add_argument(
    "--verbose",
    help="Print out all the ports while scanning and open ports when found",
    default=None,
)

# Parse arguments
args = parser.parse_args()

# Translate hostname to IPv4
target = socket.gethostbyname(args.target)

# Split the expected ports argument to get the expected ports
expected_ports = set(map(int, args.expected_ports.split(",")))

verbose = args.verbose

# Add Banner
print("-" * 50)
print("Scanning Target: " + target)
print("Scanning started at:" + str(datetime.now()))
print(f"Expected ports: {args.expected_ports}")
print("-" * 50)


async def scan_port(
    port: int, target: str, expected_ports: Set[int], unexpected_ports: List[int]
) -> None:
    """
    Asynchronously scans a single port on the target host.

    :param port: The port number to scan
    :param target: The target host (IP address or hostname)
    :param expected_ports: A set of expected open ports
    :param unexpected_ports: A list to collect unexpected open ports
    """
    try:
        reader, writer = await asyncio.open_connection(target, port)
        writer.close()
        await writer.wait_closed()
        if verbose:
            print(f"Port {port} is open")
        if port not in expected_ports:
            unexpected_ports.append(port)
    except:
        pass


async def main(target: str, expected_ports: Set[int], verbose: bool) -> None:
    """
    Asynchronously scans all ports on the target host, printing messages for open ports
    and collecting unexpected open ports.

    :param target: The target host (IP address or hostname)
    :param expected_ports: A set of expected open ports
    :param verbose: Whether to print a message for each port as it is scanned
    """
    unexpected_ports: List[int] = []
    tasks: List[Any] = []
    for port in range(1, 65535):
        if verbose:
            print(f"Review port: {port}")
        task = scan_port(port, target, expected_ports, unexpected_ports)
        tasks.append(task)

    await asyncio.gather(*tasks)

    if unexpected_ports:
        error_message = "Error: Unexpected open ports detected: " + ", ".join(
            map(str, unexpected_ports)
        )
        sys.exit(error_message)


# Run the asynchronous main function
asyncio.run(main(target, expected_ports, verbose))

console.sh

python check_port_scanner.py 127.0.0.1 --expected_ports=80,443 --verbose=true

requirements.txt

pyfiglet
asyncio