Network Port Scanner with Banner Grabbing

Problem Statement

System administrators and security engineers frequently need to discover active services, exposed management interfaces, and unauthorized network listeners across enterprise subnets. While heavy-duty tools like Nmap exist, they are often absent in minimalist container images, restricted bastion hosts, or air-gapped production servers where installing non-standard binaries is prohibited or monitored strictly.

This script provides a lightweight, pure-Python solution for network reconnaissance and service identification without external third-party dependencies. It allows operators to quickly audit individual hosts or entire CIDR subnets, detect open TCP ports, and grab service identification banners to ensure exposed services match expected deployment baselines.

How It Works

The tool combines Python's standard socket module with concurrent execution via concurrent.futures.ThreadPoolExecutor. It performs the following data workflow:

1. Target Expansion: Host inputs (single IP, hostname, or CIDR network like 192.168.1.0/24) and port specifications (comma-separated lists or ranges like 22,80,100-200) are parsed into distinct IP/port combinations.
2. Concurrent Connection Probing: The script distributes tasks across a thread pool. For each target, a non-blocking connection attempt is made using socket.connect_ex().
3. Banner Grabbing: For ports returning a successful TCP hand-shake (0), a targeted protocol probe (HTTP request or protocol generic CRLF) is sent over the connection. The script collects initial response bytes, cleans printable ASCII metadata, and attributes the banner to the open port.
4. Result Aggregation: Results are safely gathered across threads and output directly to stdout in formatted tabular text or structured JSON.

+--------------------+      +-----------------------+      +-------------------------+
| Host/CIDR & Ports  | ---> | Host/Port Expansion   | ---> | ThreadPoolExecutor      |
+--------------------+      +-----------------------+      +-------------------------+
                                                                       |
                                                                       v
+--------------------+      +-----------------------+      +-------------------------+
| JSON / stdout      | <--- | Banner Cleaning &     | <--- | TCP connect_ex() +      |
| Formatted Output   |      | Aggregation           |      | Service Banner Probe    |
+--------------------+      +-----------------------+      +-------------------------+

Full Script

#!/usr/bin/env python3
"""
Network Port Scanner with Banner Grabbing

A fast, zero-dependency TCP port scanner capable of scanning single hosts or CIDR subnets,
performing concurrent TCP handshakes, and retrieving service banners.
"""

import argparse
import concurrent.futures
import ipaddress
import json
import os
import re
import socket
import sys
from typing import Any, Dict, List, Optional, Tuple

# Configuration via environment variables with sensible defaults
DEFAULT_TIMEOUT = float(os.environ.get("PORT_SCAN_TIMEOUT", "1.5"))
DEFAULT_THREADS = int(os.environ.get("PORT_SCAN_THREADS", "100"))


def parse_ports(port_input: str) -> List[int]:
    """Parse port strings including ranges and individual ports into a sorted list of unique ints.

    Example inputs: '22,80,443', '1-1024', '80,8080,9000-9005'
    """
    ports: set[int] = set()
    for part in port_input.split(","):
        part = part.strip()
        if not part:
            continue
        if "-" in part:
            try:
                start, end = part.split("-", 1)
                start_p, end_p = int(start), int(end)
                if start_p > end_p or start_p < 1 or end_p > 65535:
                    raise ValueError
                ports.update(range(start_p, end_p + 1))
            except ValueError:
                raise argparse.ArgumentTypeError(f"Invalid port range specified: '{part}'")
        else:
            try:
                p = int(part)
                if not (1 <= p <= 65535):
                    raise ValueError
                ports.add(p)
            except ValueError:
                raise argparse.ArgumentTypeError(f"Invalid port number specified: '{part}'")
    return sorted(list(ports))


def parse_targets(target_input: str) -> List[str]:
    """Parse IP address, hostname, or CIDR block into a list of individual target IP strings."""
    try:
        # Check if CIDR or valid IP
        net = ipaddress.ip_network(target_input, strict=False)
        if net.num_addresses == 1:
            return [str(net.network_address)]
        # Return usable host addresses for subnets
        return [str(ip) for ip in net.hosts()]
    except ValueError:
        # Attempt hostname resolution
        try:
            resolved_ip = socket.gethostbyname(target_input)
            return [resolved_ip]
        except socket.gaierror:
            raise argparse.ArgumentTypeError(f"Unable to resolve host or network: '{target_input}'")


def grab_banner(sock: socket.socket, port: int) -> str:
    """Send probe data where necessary and read initial service banner from open socket."""
    # Service probes for protocols that expect client greetings
    probes: Dict[int, bytes] = {
        80: b"HEAD / HTTP/1.1\r\nHost: localhost\r\nUser-Agent: Scanner\r\n\r\n",
        8080: b"HEAD / HTTP/1.1\r\nHost: localhost\r\nUser-Agent: Scanner\r\n\r\n",
        8000: b"HEAD / HTTP/1.1\r\nHost: localhost\r\nUser-Agent: Scanner\r\n\r\n",
        443: b"HEAD / HTTP/1.1\r\nHost: localhost\r\nUser-Agent: Scanner\r\n\r\n",
    }

    banner = ""
    try:
        probe = probes.get(port, b"\r\n")
        sock.sendall(probe)
        data = sock.recv(1024)
        if data:
            # Decode text safely and replace unprintable characters
            decoded = data.decode("utf-8", errors="replace").strip()
            # Collapse whitespace and control characters to single line
            clean = re.sub(r"[\r\n\t]+", " ", decoded)
            banner = "".join(char for char in clean if char.isprintable()).strip()
            # Truncate long banners
            if len(banner) > 80:
                banner = banner[:77] + "..."
    except (socket.timeout, SocketError, OSError):
        banner = ""

    return banner or "No banner returned"


# Socket error alias for exception catching
SocketError = socket.error


def scan_target(ip: str, port: int, timeout: float) -> Optional[Dict[str, Any]]:
    """Attempt TCP handshake on host:port, return target info if port is open."""
    try:
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
            sock.settimeout(timeout)
            result = sock.connect_ex((ip, port))
            if result == 0:
                # Port is open, grab banner
                banner = grab_banner(sock, port)
                return {"host": ip, "port": port, "state": "open", "banner": banner}
    except Exception:
        # Ignore individual socket errors during rapid scanning
        pass
    return None


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Fast concurrent TCP port scanner with service banner grabbing."
    )
    parser.add_argument(
        "--host",
        required=True,
        help="Target IP address, hostname, or CIDR network range (e.g., 192.168.1.0/24)",
    )
    parser.add_argument(
        "--ports",
        required=True,
        help="Comma-separated ports or ranges to scan (e.g., 22,80,443 or 21-100)",
    )
    parser.add_argument(
        "--timeout",
        type=float,
        default=DEFAULT_TIMEOUT,
        help=f"Socket timeout in seconds (default: {DEFAULT_TIMEOUT})",
    )
    parser.add_argument(
        "--threads",
        type=int,
        default=DEFAULT_THREADS,
        help=f"Max worker threads (default: {DEFAULT_THREADS})",
    )
    parser.add_argument(
        "--json",
        action="store_true",
        help="Output findings as structured JSON",
    )

    args = parser.parse_args()

    try:
        targets = parse_targets(args.host)
        ports = parse_ports(args.ports)
    except argparse.ArgumentTypeError as err:
        print(f"Error: {err}", file=sys.stderr)
        sys.exit(1)

    total_checks = len(targets) * len(ports)
    open_services: List[Dict[str, Any]] = []

    # Build work queue (host, port pairs)
    tasks = [(ip, port) for ip in targets for port in ports]

    with concurrent.futures.ThreadPoolExecutor(max_workers=args.threads) as executor:
        # Submit task map
        future_to_task = {
            executor.submit(scan_target, ip, port, args.timeout): (ip, port)
            for ip, port in tasks
        }

        for future in concurrent.futures.as_completed(future_to_task):
            res = future.result()
            if res:
                open_services.append(res)

    # Output formatting
    if args.json:
        print(json.dumps(open_services, indent=2))
    else:
        print(f"\nScan Complete. Checked {total_checks} host-port combinations.")
        print(f"Discovered {len(open_services)} open ports:\n")
        if open_services:
            print(f"{'HOST':<16} {'PORT':<8} {'STATE':<8} {'BANNER'}")
            print("-" * 70)
            # Sort findings by IP address, then port
            open_services.sort(key=lambda x: (x["host"], x["port"]))
            for item in open_services:
                print(
                    f"{item['host']:<16} {item['port']:<8} {item['state']:<8} {item['banner']}"
                )
        else:
            print("No open ports found.")


if __name__ == "__main__":
    main()

Configuration

The script accepts execution parameters via CLI options or environment variables:

Usage Examples

1. Scanning Specific Ports Across a Subnet

Scan standard management and web service ports on a /24 subnet:

python3 port_scan.py --host 192.168.1.0/24 --ports 22,80,443,8080 --timeout 1.0

Output:

Scan Complete. Checked 1016 host-port combinations.
Discovered 3 open ports:

HOST             PORT     STATE    BANNER
----------------------------------------------------------------------
192.168.1.1      80       open     HTTP/1.1 200 OK Server: lighttpd/1.4.59
192.168.1.15     22       open     SSH-2.0-OpenSSH_8.9p1 Ubuntu-3ubuntu0.1
192.168.1.150    8080     open     HTTP/1.1 404 Not Found Server: Werkzeug/2.2.2

2. Port Range Scan with JSON Output

Audit host port range 20-100 and pipe structured results to jq or write to disk:

python3 port_scan.py --host 10.0.0.5 --ports 20-100 --json

Output:

[
  {
    "host": "10.0.0.5",
    "port": 21,
    "state": "open",
    "banner": "220 (vsFTPd 3.0.3)"
  },
  {
    "host": "10.0.0.5",
    "port": 22,
    "state": "open",
    "banner": "SSH-2.0-OpenSSH_8.2p1 Ubuntu-4ubuntu0.5"
  },
  {
    "host": "10.0.0.5",
    "port": 80,
    "state": "open",
    "banner": "HTTP/1.1 200 OK Server: nginx/1.18.0"
  }
]

Error Handling

The script handles operational edge cases natively:

• Host Resolution Failures: Unresolvable domain names trigger an immediate argparse.ArgumentTypeError and exit cleanly without raising Python tracebacks.
• Network Timeouts & Dropped Packets: Firewalled ports that drop packets (drop/DROP rules) will safely hit the connection --timeout limit and return None without blocking other worker threads.
• Control Characters in Banners: Unprintable byte sequences returned by binary protocols (e.g., MySQL, RDP) are filtered using regular expressions (re.sub) and standard string sanitization before terminal display.
• Thread Exhaustion / Memory Limits: Thread counts default to safe bounds (100), preventing OS thread creation exhaustion (OS Error: cannot allocate memory).

Extending Ideas

Here are four ways to build upon this tool:

1. TLS Banner Grabbing: Wrap open sockets with standard library ssl.create_default_context().wrap_socket() for TLS ports (e.g., 443, 8443, 636) to parse SSL certificate fields (Issuer, Expiry, Subject Alternative Names).
2. Service Version Detection: Parse returned banner strings using regular expressions to automatically extract product names and version strings (e.g., identifying OpenSSH version or Apache releases).
3. UDP Port Scanning Support: Incorporate socket.SOCK_DGRAM handling for common UDP services like DNS (53), SNMP (161), or NTP (123).
4. Nmap XML Export: Structure output dictionary objects into Nmap-compatible XML syntax for direct consumption by vulnerability management frameworks like DefectDojo or OpenVAS.