TCP Port Scanner (python, written by Codex)

Codex wrote this python project from the task below. It installed and ran on a clean Ubuntu 22.04 machine as written.

Task given to the agent:

TASK: TCP Port Scanner

Write a program that scans TCP ports on a target host to determine which ports are open, closed, or filtered, with support for service detection, concurrent scanning, and configurable scan ranges.

FUNCTIONAL REQUIREMENTS:
- Accept a target hostname or IP address as a command-line argument
- Scan a port range specified via --ports flag (e.g., --ports 1-1024, --ports 80,443,8080, or --ports 1-65535); default: well-known ports 1-1024
- Perform TCP connect scanning: attempt a full TCP handshake to determine if a port is open
- Support concurrent scanning with configurable thread count via --threads flag (default: 100) to speed up large range scans
- Implement configurable connection timeout via --timeout flag (default: 1 second per port)
- Detect common services on open ports by matching port numbers to known services (HTTP/80, HTTPS/443, SSH/22, FTP/21, SMTP/25, DNS/53, etc.) using a built-in service name database
- Attempt basic banner grabbing on open ports to identify the running service version via --banner flag
- Support scanning multiple hosts via --hosts flag (comma-separated or CIDR notation for a subnet)
- Print scan results to console in real-time as ports are discovered, with a final summary table showing: port number, state (open/closed/filtered), service name, and banner (if captured)
- Save the full scan report as JSON with --output flag (default: scan_report.json) including scan duration, host info, and per-port results
- If no target is given, scan localhost (127.0.0.1) on the common ports (21, 22, 25, 53, 80, 110, 143, 443, 993, 995, 3306, 5432, 6379, 8080, 8443) and display the results
- Handle errors: unresolvable hostnames, network unreachable, connection refused vs timeout distinction, and permission issues

Create a complete Python project for a clean Ubuntu 22.04 machine with only Python 3.10+ installed. Include:
- Source code
- requirements.txt with all dependencies (direct and transitive) pinned to exact versions
- README.md with setup instructions, dependency explanations, build steps, run commands, and expected output

[]

# TCP Port Scanner (Python)

Performs TCP connect scanning for one or many hosts with configurable port ranges, concurrency, timeout, optional banner grabbing, CIDR expansion, real-time output, and JSON reporting.

## Requirements

- Ubuntu 22.04
- Python 3.10+

## Dependencies

- Standard library only (`requirements.txt` is intentionally empty of installable packages)

## Run

```bash
python src/main.py scanme.nmap.org
python src/main.py 192.168.1.10 --ports 1-1024 --threads 200 --timeout 1
python src/main.py --hosts 192.168.1.0/24 --ports 22,80,443 --threads 100
python src/main.py example.com --banner --output scan_report.json
python src/main.py
```

## Notes

- Default scan range for explicit targets is `1-1024`.
- If no target/hosts are given, scanner checks localhost on common ports.
- `--hosts` accepts comma-separated hosts and CIDR notation.

# No external dependencies required (Python 3.10+ standard library only)

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import ipaddress
import json
import socket
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path

COMMON_LOCALHOST_PORTS = [21, 22, 25, 53, 80, 110, 143, 443, 993, 995, 3306, 5432, 6379, 8080, 8443]
SERVICES = {
    20: "ftp-data", 21: "ftp", 22: "ssh", 23: "telnet", 25: "smtp", 53: "dns", 80: "http",
    110: "pop3", 123: "ntp", 143: "imap", 443: "https", 465: "smtps", 587: "submission",
    993: "imaps", 995: "pop3s", 1433: "mssql", 1521: "oracle", 3306: "mysql", 3389: "rdp",
    5432: "postgresql", 6379: "redis", 8080: "http-alt", 8443: "https-alt",
}


def parse_ports(raw: str) -> list[int]:
    out: set[int] = set()
    for part in raw.split(","):
        p = part.strip()
        if not p:
            continue
        if "-" in p:
            a, b = p.split("-", 1)
            start = int(a)
            end = int(b)
            if start < 1 or end > 65535 or start > end:
                raise ValueError(f"Invalid range: {p}")
            out.update(range(start, end + 1))
        else:
            n = int(p)
            if n < 1 or n > 65535:
                raise ValueError(f"Invalid port: {p}")
            out.add(n)
    return sorted(out)


def expand_hosts(target: str | None, hosts: str | None) -> list[str]:
    seeds = []
    if target:
        seeds.append(target)
    if hosts:
        seeds.extend([h.strip() for h in hosts.split(",") if h.strip()])
    if not seeds:
        return ["127.0.0.1"]

    out: list[str] = []
    for h in seeds:
        if "/" in h:
            net = ipaddress.ip_network(h, strict=False)
            # limit very large ranges
            for i, host in enumerate(net.hosts()):
                if i >= 4096:
                    break
                out.append(str(host))
        else:
            out.append(h)
    return sorted(set(out))


def normalize_banner(text: str) -> str:
    return " ".join(text.strip().split())[:160]


def scan_port(host: str, ip: str, port: int, timeout: float, banner: bool) -> dict:
    result = {"port": port, "state": "filtered", "service": SERVICES.get(port, "unknown"), "banner": ""}
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(timeout)
    try:
        sock.connect((ip, port))
        result["state"] = "open"
        if banner:
            try:
                if port in {80, 8080, 8000, 443, 8443}:
                    sock.sendall(b"HEAD / HTTP/1.0\r\n\r\n")
                data = sock.recv(200)
                if data:
                    result["banner"] = normalize_banner(data.decode("utf-8", errors="ignore"))
            except Exception:
                pass
    except ConnectionRefusedError:
        result["state"] = "closed"
    except socket.timeout:
        result["state"] = "filtered"
    except OSError as exc:
        if getattr(exc, "errno", None) in {101, 113}:  # network unreachable/no route
            result["state"] = "filtered"
        else:
            result["state"] = "filtered"
            result["banner"] = str(exc)
    finally:
        sock.close()
    print(f"{host}:{port} {result['state']}" + (f" service={result['service']}" if result["state"] == "open" else "") + (f" banner=\"{result['banner']}\"" if result["banner"] else ""))
    return result


def scan_host(host: str, ports: list[int], threads: int, timeout: float, banner: bool) -> dict:
    try:
        ip = socket.gethostbyname(host)
    except socket.gaierror as exc:
        return {"host": host, "resolved_ip": "", "error": f"unresolvable host: {exc}", "results": []}

    rows: list[dict] = []
    with ThreadPoolExecutor(max_workers=max(1, threads)) as ex:
        futs = [ex.submit(scan_port, host, ip, p, timeout, banner) for p in ports]
        for fut in as_completed(futs):
            rows.append(fut.result())
    rows.sort(key=lambda r: r["port"])
    return {"host": host, "resolved_ip": ip, "results": rows}


def print_summary(report: dict) -> None:
    for host in report["hosts"]:
        print(f"\nHost: {host['host']} ({host.get('resolved_ip', '')})")
        if host.get("error"):
            print(f"Error: {host['error']}")
            continue
        print("Port | State | Service | Banner")
        print("---- | ----- | ------- | ------")
        for r in host["results"]:
            print(f"{r['port']} | {r['state']} | {r['service']} | {r['banner']}")


def parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(description="TCP Port Scanner")
    p.add_argument("target", nargs="?")
    p.add_argument("--ports", default="1-1024")
    p.add_argument("--threads", type=int, default=100)
    p.add_argument("--timeout", type=float, default=1.0)
    p.add_argument("--banner", action="store_true")
    p.add_argument("--hosts")
    p.add_argument("--output", default="scan_report.json")
    return p


def main() -> int:
    args = parser().parse_args()
    hosts = expand_hosts(args.target, args.hosts)
    ports = parse_ports(args.ports) if (args.target or args.hosts) else COMMON_LOCALHOST_PORTS

    started = time.time()
    host_reports = [scan_host(h, ports, args.threads, args.timeout, args.banner) for h in hosts]
    report = {
        "started_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(started)),
        "duration_sec": time.time() - started,
        "port_spec": args.ports if (args.target or args.hosts) else ",".join(map(str, COMMON_LOCALHOST_PORTS)),
        "hosts": host_reports,
    }

    print_summary(report)
    out = Path(args.output).resolve()
    out.write_text(json.dumps(report, indent=2), encoding="utf-8")
    print(f"\nReport saved: {out}")
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except Exception as exc:
        print(f"Error: {exc}")
        raise SystemExit(1)