portscope.py

#!/usr/bin/env python3
"""portscope — a concurrent TCP port scanner with service and banner detection.

Performs a TCP connect scan against a host or CIDR range, identifies common
services from the port number, and grabs a service banner where one is offered.
Connect scans need no special privileges. Intended for the recon phase of an
authorized assessment of hosts you own or are permitted to test.

Standard library only.
"""
from __future__ import annotations

import argparse
import concurrent.futures
import ipaddress
import json
import socket
import ssl
import sys
import time
from typing import Optional

__version__ = "1.0.0"

# port -> service name, for the common cases nmap-services would cover.
SERVICES: dict[int, str] = {
    21: "ftp", 22: "ssh", 23: "telnet", 25: "smtp", 53: "dns",
    80: "http", 110: "pop3", 111: "rpcbind", 135: "msrpc", 139: "netbios-ssn",
    143: "imap", 161: "snmp", 389: "ldap", 443: "https", 445: "smb",
    465: "smtps", 587: "submission", 993: "imaps", 995: "pop3s",
    1433: "mssql", 1521: "oracle", 2049: "nfs", 2375: "docker", 3306: "mysql",
    3389: "rdp", 5432: "postgres", 5601: "kibana", 5900: "vnc", 5985: "winrm",
    6379: "redis", 8000: "http-alt", 8080: "http-proxy", 8443: "https-alt",
    9200: "elasticsearch", 11211: "memcached", 27017: "mongodb",
}

# A compact but practical default set: top services + common web/admin ports.
TOP_PORTS = [
    21, 22, 23, 25, 53, 80, 110, 111, 135, 139, 143, 161, 389, 443, 445,
    465, 587, 993, 995, 1433, 1521, 2049, 2375, 3306, 3389, 5432, 5601,
    5900, 5985, 6379, 8000, 8080, 8443, 9200, 11211, 27017,
]

TLS_PORTS = {443, 465, 993, 995, 8443}


class Colors:
    GREEN = "\033[32m"
    YELLOW = "\033[33m"
    CYAN = "\033[36m"
    DIM = "\033[2m"
    BOLD = "\033[1m"
    RESET = "\033[0m"

    @classmethod
    def disable(cls) -> None:
        for n in ("GREEN", "YELLOW", "CYAN", "DIM", "BOLD", "RESET"):
            setattr(cls, n, "")


def parse_ports(spec: str) -> list[int]:
    if spec.strip().lower() in ("top", "default"):
        return TOP_PORTS
    if spec.strip().lower() == "all":
        return list(range(1, 65536))
    out: set[int] = set()
    for part in spec.split(","):
        part = part.strip()
        if not part:
            continue
        if "-" in part:
            a, b = part.split("-", 1)
            lo, hi = int(a), int(b)
            out.update(range(min(lo, hi), max(lo, hi) + 1))
        else:
            out.add(int(part))
    return sorted(p for p in out if 1 <= p <= 65535)


def expand_targets(target: str) -> list[str]:
    """Accept a hostname, an IP, or a CIDR; return a list of host strings."""
    try:
        net = ipaddress.ip_network(target, strict=False)
        if net.num_addresses > 1:
            hosts = list(net.hosts())
            return [str(h) for h in hosts]
        return [str(net.network_address)]
    except ValueError:
        return [target]


def grab_banner(host: str, port: int, timeout: float, use_tls: bool) -> str:
    """Read a short banner; for HTTP/TLS ports nudge the server to respond."""
    try:
        raw = socket.create_connection((host, port), timeout=timeout)
    except OSError:
        return ""
    sock: socket.socket = raw
    try:
        if use_tls:
            ctx = ssl.create_default_context()
            ctx.check_hostname = False
            ctx.verify_mode = ssl.CERT_NONE
            sock = ctx.wrap_socket(raw, server_hostname=host)
        sock.settimeout(timeout)
        svc = SERVICES.get(port, "")
        if "http" in svc or port in (80, 8080, 8000):
            sock.sendall(b"HEAD / HTTP/1.0\r\nHost: %s\r\n\r\n" % host.encode())
        data = sock.recv(256)
        text = data.decode("utf-8", errors="ignore").strip()
        first = text.splitlines()[0] if text else ""
        return first[:120]
    except OSError:
        return ""
    finally:
        try:
            sock.close()
        except OSError:
            pass


def scan_port(host: str, port: int, timeout: float, banner: bool) -> Optional[dict]:
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.settimeout(timeout)
    try:
        if s.connect_ex((host, port)) != 0:
            return None
    except OSError:
        return None
    finally:
        s.close()
    result = {"port": port, "service": SERVICES.get(port, "unknown"), "banner": ""}
    if banner:
        result["banner"] = grab_banner(host, port, timeout, port in TLS_PORTS)
    return result


def scan_host(host: str, ports: list[int], timeout: float, workers: int,
              banner: bool) -> list[dict]:
    open_ports: list[dict] = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as pool:
        futures = {pool.submit(scan_port, host, p, timeout, banner): p for p in ports}
        for fut in concurrent.futures.as_completed(futures):
            r = fut.result()
            if r:
                open_ports.append(r)
    return sorted(open_ports, key=lambda r: r["port"])


def is_alive(host: str, ports: list[int], timeout: float) -> bool:
    """Cheap liveness probe: try a couple of likely-open ports."""
    probes = [p for p in (443, 80, 22, 445, 3389) if p in ports] or ports[:3]
    for p in probes:
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.settimeout(timeout)
        try:
            if s.connect_ex((host, p)) == 0:
                return True
        except OSError:
            pass
        finally:
            s.close()
    return False


def print_host(host: str, results: list[dict], c: type[Colors]) -> None:
    print(f"\n{c.BOLD}{host}{c.RESET}")
    if not results:
        print(f"  {c.DIM}no open ports in the scanned range{c.RESET}")
        return
    print(f"  {c.DIM}{'PORT':<11}{'SERVICE':<16}BANNER{c.RESET}")
    for r in results:
        port = f"{r['port']}/tcp"
        svc = r["service"]
        banner = r["banner"]
        line = f"  {c.GREEN}{port:<11}{c.RESET}{c.CYAN}{svc:<16}{c.RESET}"
        if banner:
            line += f"{c.DIM}{banner}{c.RESET}"
        print(line)


def main() -> int:
    ap = argparse.ArgumentParser(
        prog="portscope",
        description="Concurrent TCP connect scanner with service and banner detection.",
    )
    ap.add_argument("target", help="hostname, IP, or CIDR (e.g. 10.0.0.0/24)")
    ap.add_argument("-p", "--ports", default="top",
                    help="ports: 'top', 'all', a list (22,80,443) or range (1-1024)")
    ap.add_argument("-t", "--timeout", type=float, default=1.0,
                    help="per-connection timeout in seconds (default 1.0)")
    ap.add_argument("-w", "--workers", type=int, default=200,
                    help="concurrent connections per host (default 200)")
    ap.add_argument("--no-banner", action="store_true", help="skip banner grabbing")
    ap.add_argument("--open-only", action="store_true",
                    help="only print hosts that have at least one open port")
    ap.add_argument("--json", action="store_true", help="emit JSON instead of text")
    ap.add_argument("--no-color", action="store_true", help="disable ANSI colors")
    args = ap.parse_args()

    if args.no_color or not sys.stdout.isatty() or args.json:
        Colors.disable()
    c = Colors

    try:
        ports = parse_ports(args.ports)
    except ValueError:
        print("invalid --ports specification", file=sys.stderr)
        return 2
    if not ports:
        print("no ports to scan", file=sys.stderr)
        return 2

    hosts = expand_targets(args.target)
    sweep = len(hosts) > 1
    started = time.time()

    if not args.json:
        print(f"{c.DIM}portscope {__version__} - scanning {len(hosts)} host(s), "
              f"{len(ports)} port(s) each{c.RESET}")

    report: list[dict] = []
    for host in hosts:
        # On a subnet sweep, skip hosts that show no life to save time.
        if sweep and not is_alive(host, ports, min(args.timeout, 0.6)):
            continue
        results = scan_host(host, ports, args.timeout, args.workers,
                            not args.no_banner)
        if args.open_only and not results:
            continue
        report.append({"host": host, "open_ports": results})
        if not args.json:
            print_host(host, results, c)

    elapsed = time.time() - started
    if args.json:
        print(json.dumps({"target": args.target, "ports_scanned": len(ports),
                          "elapsed_s": round(elapsed, 2), "hosts": report}, indent=2))
    else:
        total_open = sum(len(h["open_ports"]) for h in report)
        print(f"\n{c.DIM}done in {elapsed:.1f}s - {total_open} open port(s) "
              f"across {len(report)} responsive host(s){c.RESET}")
    return 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyboardInterrupt:
        print("\ninterrupted", file=sys.stderr)
        sys.exit(130)