portquiz.net concurrent Python asyncio port connectivity tester

Readme.md

Check outbound port connectivity using portquiz.net and Python

Demonstrates use of concurrent.futures.ThreadPoolExecutor and asyncio.Semaphore in separate examples to check outbound port connectivity.

In both examples, the maximum number of concurrent connections is limited to 5 (user option). This examples shows that Asyncio using Semaphore with AIOHTTP can be 10-15% faster than ThreadPoolExecutor with urllib.request.

This can be useful to quickly check why a certain web service (say SSH) doesn't work on a network connection. Some public networks only allow traffic on say ports 80 and 443.

Solutions to blocked ports for SSH include using SSH ProxyJump with an intermediate TRUSTED server on an allowed port. Some remote SSH systems actually require this, where they the desired server has only LAN access, and a gateway SSH server with no privileges is used as the SSH ProxyJump by network design.

Another straightforward solution to outbound port blocking is to use a VPN, unless VPNs are also blocked.

portquiz.py

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.10"
# dependencies = [
#     "aiohttp",
# ]
# ///

"""
AsyncIO AIOHTTP may be faster and/or use less resources than ThreadPool version
"""

import argparse
import asyncio

import aiohttp

DEFAULT_PORTS = [22, 80, 443, 143, 8080, 3389]
TIMEOUT = 2  # seconds
MAX_CONCURRENT = 5  # limit simultaneous requests


async def check_port(
    session: aiohttp.ClientSession,
    port: int,
    timeout: aiohttp.ClientTimeout,
    semaphore: asyncio.Semaphore,
) -> int:
    """Check if a port is accessible via portquiz.net"""
    async with semaphore:
        url = f"http://portquiz.net:{port}"
        async with session.get(url, timeout=timeout) as response:
            return response.status


def _port_type(raw: str) -> int:
    port = int(raw)
    if not 1 <= port <= 65535:
        raise argparse.ArgumentTypeError("port must be 1-65535")
    return port


async def main(ports: list[int], timeout: int, max_concurrent: int):

    to = aiohttp.ClientTimeout(total=timeout)
    semaphore = asyncio.Semaphore(max_concurrent)

    async with aiohttp.ClientSession() as session:
        tasks = [check_port(session, port, to, semaphore) for port in ports]
        results = await asyncio.gather(*tasks, return_exceptions=True)

        for port, result in zip(ports, results):
            if isinstance(result, aiohttp.ClientError):
                print(f"FAIL: {port} {result}")
            elif isinstance(result, asyncio.TimeoutError):
                print(f"FAIL: {port} timeout")
            elif isinstance(result, Exception):
                print(f"FAIL: {port} {result}")
            elif result != 200:
                print(f"FAIL: {port} returned HTTP {result}")
            else:
                print(f"OK: {port} is open")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Port quiz tester")
    parser.add_argument(
        "-p",
        "--ports",
        nargs="+",
        type=_port_type,
        help="Ports to test",
        default=DEFAULT_PORTS,
    )
    parser.add_argument(
        "-t",
        "--timeout",
        type=int,
        help="Timeout for each port check in seconds",
        default=TIMEOUT,
    )
    parser.add_argument(
        "-c",
        "--concurrent",
        type=int,
        help="Maximum concurrent requests",
        default=MAX_CONCURRENT,
    )
    args = parser.parse_args()

    asyncio.run(main(args.ports, args.timeout, args.concurrent))

portquiz_ThreadPool.py

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.10"
# dependencies = []
# ///

"""
based on https://gist.github.com/fopina/a8ed5559aa28ef59cf95d172a4fea045
"""

import argparse
import concurrent.futures
import urllib.request
import urllib.error

DEFAULT_PORTS = [22, 80, 443, 143, 8080, 3389]
TIMEOUT = 2  # seconds


def check_port(port: int, timeout: int) -> int:
    r = urllib.request.Request(f"http://portquiz.net:{port}")
    with urllib.request.urlopen(r, timeout=timeout) as conn:
        return conn.status


def _port_type(raw: str) -> int:
    port = int(raw)
    if not 1 <= port <= 65535:
        raise argparse.ArgumentTypeError("port must be 1-65535")
    return port


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Port quiz tester")
    parser.add_argument(
        "-p",
        "--ports",
        nargs="+",
        type=_port_type,
        help="Ports to test",
        default=DEFAULT_PORTS,
    )
    parser.add_argument(
        "-t",
        "--timeout",
        type=int,
        help="Timeout for each port check in seconds",
        default=TIMEOUT,
    )
    args = parser.parse_args()

    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        future_to_port = {
            executor.submit(check_port, port, args.timeout): port for port in args.ports
        }
        for future in concurrent.futures.as_completed(future_to_port):
            port = future_to_port[future]
            try:
                status = future.result()
                print(f"OK: {port} is open")
            except urllib.error.HTTPError as e:
                print(f"FAIL: {port} HTTP {e.code} {e.reason}")
            except urllib.error.URLError as e:
                print(f"FAIL: {port} {e.reason}")
            except TimeoutError:
                print(f"FAIL: {port} timeout")