TCP Port Scanner (python, written by Claude Code)

Claude Code 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 (Trial 1)

A TCP port scanner with concurrent scanning, service detection, banner grabbing, and CIDR subnet support. Built with scapy for packet crafting and rich for terminal output.

## Dependencies

- **scapy** (2.5.0) - Packet crafting and SYN scanning
- **rich** (13.7.0) - Rich terminal output with tables and progress bars

## Installation

```bash
pip install -r requirements.txt
```

## Usage

```bash
# Scan default ports (1-1024) on a single host
python scanner.py 192.168.1.1

# Scan specific ports
python scanner.py 192.168.1.1 -p 22,80,443,8080

# Scan a port range
python scanner.py example.com -p 1-1024

# Scan a CIDR subnet
python scanner.py 192.168.1.0/24 -p 22,80,443

# SYN scan (requires root/admin)
python scanner.py 192.168.1.1 -p 1-1024 --scan-type syn

# Show closed ports and adjust concurrency
python scanner.py 10.0.0.1 -p 1-1024 --show-closed -w 200
```

## Features

- TCP connect scan and SYN scan modes
- Concurrent scanning with configurable worker threads
- Service detection via banner grabbing and port-based lookup
- CIDR subnet support for scanning entire networks
- Rich formatted output with progress bars and colored tables

scapy==2.5.0
rich==13.7.0

#!/usr/bin/env python3
"""
TCP Port Scanner with concurrent scanning, service detection, banner grabbing,
and CIDR subnet support. Uses scapy for packet crafting and rich for output.
"""

import sys
import ipaddress
import socket
import argparse
import concurrent.futures
import time
from typing import List, Dict, Optional, Tuple

from scapy.all import IP, TCP, sr1, conf
from rich.console import Console
from rich.table import Table
from rich.progress import Progress, SpinnerColumn, BarColumn, TextColumn
from rich.panel import Panel
from rich.text import Text

# Suppress scapy warnings
conf.verb = 0

console = Console()

# Well-known port to service mappings
WELL_KNOWN_SERVICES = {
    21: "FTP", 22: "SSH", 23: "Telnet", 25: "SMTP", 53: "DNS",
    80: "HTTP", 110: "POP3", 111: "RPCBind", 135: "MSRPC",
    139: "NetBIOS-SSN", 143: "IMAP", 443: "HTTPS", 445: "Microsoft-DS",
    993: "IMAPS", 995: "POP3S", 1433: "MSSQL", 1521: "Oracle",
    3306: "MySQL", 3389: "RDP", 5432: "PostgreSQL", 5900: "VNC",
    6379: "Redis", 8080: "HTTP-Proxy", 8443: "HTTPS-Alt",
    27017: "MongoDB", 6443: "Kubernetes",
}


def parse_ports(port_spec: str) -> List[int]:
    """Parse port specification string into a list of port numbers."""
    ports = set()
    for part in port_spec.split(","):
        part = part.strip()
        if "-" in part:
            start, end = part.split("-", 1)
            start, end = int(start.strip()), int(end.strip())
            if start > end:
                start, end = end, start
            ports.update(range(start, end + 1))
        else:
            ports.add(int(part))
    return sorted(p for p in ports if 1 <= p <= 65535)


def parse_targets(target_spec: str) -> List[str]:
    """Parse target specification which may include CIDR notation."""
    targets = []
    for part in target_spec.split(","):
        part = part.strip()
        if "/" in part:
            try:
                network = ipaddress.ip_network(part, strict=False)
                for host in network.hosts():
                    targets.append(str(host))
            except ValueError as e:
                console.print(f"[red]Invalid CIDR notation: {part} ({e})[/red]")
        else:
            try:
                ip = socket.gethostbyname(part)
                targets.append(ip)
            except socket.gaierror:
                console.print(f"[red]Cannot resolve hostname: {part}[/red]")
    return targets


def grab_banner(ip: str, port: int, timeout: float = 3.0) -> Optional[str]:
    """Attempt to grab a service banner from an open port."""
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(timeout)
        sock.connect((ip, port))

        # Send a probe for HTTP ports
        if port in (80, 8080, 8443, 443):
            sock.send(b"HEAD / HTTP/1.0\r\nHost: %b\r\n\r\n" % ip.encode())
        else:
            # For other services, just wait for the banner
            pass

        banner = sock.recv(1024).decode("utf-8", errors="replace").strip()
        sock.close()
        return banner[:200] if banner else None
    except Exception:
        return None


def detect_service(port: int, banner: Optional[str] = None) -> str:
    """Detect the service running on a port using banner and known services."""
    if banner:
        banner_lower = banner.lower()
        if "ssh" in banner_lower:
            return "SSH"
        elif "http" in banner_lower:
            return "HTTP"
        elif "ftp" in banner_lower:
            return "FTP"
        elif "smtp" in banner_lower:
            return "SMTP"
        elif "mysql" in banner_lower:
            return "MySQL"
        elif "postgresql" in banner_lower or "postgres" in banner_lower:
            return "PostgreSQL"
        elif "redis" in banner_lower:
            return "Redis"
        elif "imap" in banner_lower:
            return "IMAP"
        elif "pop3" in banner_lower or "pop" in banner_lower:
            return "POP3"

    return WELL_KNOWN_SERVICES.get(port, "Unknown")


def syn_scan_port(ip: str, port: int, timeout: float = 2.0) -> Dict:
    """Perform a SYN scan on a single port using scapy."""
    result = {
        "port": port,
        "state": "closed",
        "service": "Unknown",
        "banner": None,
    }

    try:
        pkt = IP(dst=ip) / TCP(dport=port, flags="S")
        resp = sr1(pkt, timeout=timeout, verbose=0)

        if resp is not None:
            if resp.haslayer(TCP):
                tcp_layer = resp.getlayer(TCP)
                if tcp_layer.flags == 0x12:  # SYN-ACK
                    result["state"] = "open"
                    # Send RST to close the connection
                    rst = IP(dst=ip) / TCP(dport=port, flags="R")
                    sr1(rst, timeout=1, verbose=0)

                    # Try banner grabbing on open ports
                    banner = grab_banner(ip, port)
                    result["banner"] = banner
                    result["service"] = detect_service(port, banner)
                elif tcp_layer.flags == 0x14:  # RST-ACK
                    result["state"] = "closed"
        else:
            result["state"] = "filtered"

    except Exception as e:
        result["state"] = "error"
        result["banner"] = str(e)

    return result


def connect_scan_port(ip: str, port: int, timeout: float = 2.0) -> Dict:
    """Perform a TCP connect scan on a single port."""
    result = {
        "port": port,
        "state": "closed",
        "service": "Unknown",
        "banner": None,
    }

    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(timeout)
        code = sock.connect_ex((ip, port))
        if code == 0:
            result["state"] = "open"
            banner = grab_banner(ip, port)
            result["banner"] = banner
            result["service"] = detect_service(port, banner)
        sock.close()
    except socket.timeout:
        result["state"] = "filtered"
    except Exception as e:
        result["state"] = "error"
        result["banner"] = str(e)

    return result


def scan_host(
    ip: str,
    ports: List[int],
    scan_type: str = "connect",
    max_workers: int = 100,
    timeout: float = 2.0,
) -> List[Dict]:
    """Scan all specified ports on a single host concurrently."""
    results = []
    scan_func = syn_scan_port if scan_type == "syn" else connect_scan_port

    with Progress(
        SpinnerColumn(),
        TextColumn("[progress.description]{task.description}"),
        BarColumn(),
        TextColumn("[progress.percentage]{task.percentage:>3.0f}%"),
        TextColumn("({task.completed}/{task.total})"),
        console=console,
    ) as progress:
        task = progress.add_task(f"Scanning {ip}", total=len(ports))

        with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
            future_to_port = {
                executor.submit(scan_func, ip, port, timeout): port
                for port in ports
            }

            for future in concurrent.futures.as_completed(future_to_port):
                result = future.result()
                results.append(result)
                progress.advance(task)

    results.sort(key=lambda r: r["port"])
    return results


def display_results(ip: str, results: List[Dict], show_closed: bool = False):
    """Display scan results in a formatted table using rich."""
    open_ports = [r for r in results if r["state"] == "open"]
    filtered_ports = [r for r in results if r["state"] == "filtered"]
    closed_ports = [r for r in results if r["state"] == "closed"]

    console.print()
    header = Text(f"Scan Results for {ip}", style="bold cyan")
    console.print(Panel(header, expand=False))

    table = Table(show_header=True, header_style="bold magenta")
    table.add_column("Port", style="cyan", width=8)
    table.add_column("State", width=10)
    table.add_column("Service", style="green", width=15)
    table.add_column("Banner", style="dim", max_width=60)

    display_results_list = open_ports + filtered_ports
    if show_closed:
        display_results_list += closed_ports
    display_results_list.sort(key=lambda r: r["port"])

    for r in display_results_list:
        state_style = {
            "open": "[bold green]open[/bold green]",
            "closed": "[red]closed[/red]",
            "filtered": "[yellow]filtered[/yellow]",
            "error": "[red]error[/red]",
        }.get(r["state"], r["state"])

        banner_text = r["banner"][:60] if r["banner"] else ""
        table.add_row(str(r["port"]), state_style, r["service"], banner_text)

    if len(display_results_list) > 0:
        console.print(table)
    else:
        console.print("[yellow]No open or filtered ports found.[/yellow]")

    console.print(
        f"\n[bold]Summary:[/bold] {len(open_ports)} open, "
        f"{len(filtered_ports)} filtered, {len(closed_ports)} closed "
        f"(Total: {len(results)} ports scanned)"
    )


def main():
    parser = argparse.ArgumentParser(
        description="TCP Port Scanner with concurrent scanning, "
        "service detection, and banner grabbing",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  %(prog)s 192.168.1.1
  %(prog)s 192.168.1.0/24 -p 22,80,443
  %(prog)s example.com -p 1-1024 --scan-type syn
  %(prog)s 10.0.0.1 -p 80,443,8080 --banner --show-closed
        """,
    )
    parser.add_argument(
        "target",
        help="Target IP, hostname, or CIDR range (comma-separated for multiple)",
    )
    parser.add_argument(
        "-p", "--ports",
        default="1-1024",
        help="Port specification (e.g., 22,80,443 or 1-1024). Default: 1-1024",
    )
    parser.add_argument(
        "--scan-type",
        choices=["connect", "syn"],
        default="connect",
        help="Scan type: connect (default) or syn (requires root)",
    )
    parser.add_argument(
        "-w", "--workers",
        type=int,
        default=100,
        help="Maximum concurrent workers (default: 100)",
    )
    parser.add_argument(
        "-t", "--timeout",
        type=float,
        default=2.0,
        help="Timeout per port in seconds (default: 2.0)",
    )
    parser.add_argument(
        "--banner",
        action="store_true",
        help="Enable banner grabbing (default: enabled for open ports)",
    )
    parser.add_argument(
        "--show-closed",
        action="store_true",
        help="Show closed ports in results",
    )

    args = parser.parse_args()

    console.print(
        Panel(
            "[bold cyan]TCP Port Scanner[/bold cyan]\n"
            "Concurrent scanning | Service detection | Banner grabbing",
            expand=False,
        )
    )

    targets = parse_targets(args.target)
    if not targets:
        console.print("[red]No valid targets specified.[/red]")
        sys.exit(1)

    ports = parse_ports(args.ports)
    if not ports:
        console.print("[red]No valid ports specified.[/red]")
        sys.exit(1)

    console.print(f"[bold]Targets:[/bold] {len(targets)} host(s)")
    console.print(f"[bold]Ports:[/bold] {len(ports)} port(s)")
    console.print(f"[bold]Scan type:[/bold] {args.scan_type}")
    console.print(f"[bold]Workers:[/bold] {args.workers}")
    console.print(f"[bold]Timeout:[/bold] {args.timeout}s")

    start_time = time.time()

    for target_ip in targets:
        results = scan_host(
            target_ip,
            ports,
            scan_type=args.scan_type,
            max_workers=args.workers,
            timeout=args.timeout,
        )
        display_results(target_ip, results, show_closed=args.show_closed)

    elapsed = time.time() - start_time
    console.print(f"\n[bold green]Scan completed in {elapsed:.2f} seconds[/bold green]")


if __name__ == "__main__":
    main()