scanner.py

#!/usr/bin/env python3
"""
NetSleuth - A minimalist, high-performance CLI network port scanner.
Author: Portfolio Project
License: MIT
Usage: python scanner.py -t <target> -p <ports> [options]
"""

from __future__ import annotations

import argparse
import json
import socket
import sys
import time
from concurrent.futures import ThreadPoolExecutor, as_completed, Future
from dataclasses import dataclass, field, asdict
from datetime import datetime
from typing import Optional

from rich import box
from rich.console import Console
from rich.live import Live
from rich.panel import Panel
from rich.progress import (
    BarColumn,
    MofNCompleteColumn,
    Progress,
    SpinnerColumn,
    TaskID,
    TextColumn,
    TimeElapsedColumn,
)
from rich.table import Table
from rich.text import Text

# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------

BANNER = r"""
 _   _      _   ____  _            _   _
| \ | | ___| |_/ ___|| | ___ _   _| |_| |__
|  \| |/ _ \ __\___ \| |/ _ \ | | | __| '_ \
| |\  |  __/ |_ ___) | |  __/ |_| | |_| | | |
|_| \_|\___|\__|____/|_|\___|\__,_|\__|_| |_|

"""

DISCLAIMER = (
    "  ⚠  Educational Purposes Only — Authorized Scanning Only  ⚠"
)

# The 100 most commonly encountered TCP ports (nmap's default set, curated)
TOP_100_PORTS: list[int] = [
    21, 22, 23, 25, 53, 80, 81, 110, 111, 119,
    135, 139, 143, 179, 194, 389, 443, 445, 465, 514,
    515, 587, 631, 636, 873, 990, 993, 995, 1080, 1194,
    1433, 1521, 1723, 2049, 2082, 2083, 2086, 2087, 2095, 2096,
    2181, 2375, 2376, 3000, 3268, 3269, 3306, 3389, 3690, 4369,
    5000, 5432, 5900, 5984, 6379, 6443, 6660, 6661, 6662, 6663,
    6664, 6665, 6666, 6667, 6668, 6669, 7001, 7002, 7474, 8000,
    8008, 8080, 8081, 8443, 8888, 8983, 9000, 9042, 9090, 9200,
    9300, 9418, 10000, 11211, 15672, 27017, 27018, 27019, 28017, 50000,
    50070, 50075, 50090, 60000, 60001, 60002, 60003, 60004, 60005, 60010,
]

console = Console(highlight=False)


# ---------------------------------------------------------------------------
# Data classes
# ---------------------------------------------------------------------------

@dataclass
class ScanResult:
    """Holds the result of a single port scan attempt."""

    port: int
    is_open: bool
    service: str = "unknown"
    banner: str = ""


@dataclass
class ScanReport:
    """Aggregated report produced after a full scan session."""

    target: str
    resolved_ip: str
    start_time: str
    end_time: str
    duration_seconds: float
    total_ports_scanned: int
    open_ports: list[ScanResult] = field(default_factory=list)

    def to_dict(self) -> dict:
        """Serialise report to a plain Python dict (JSON-friendly).

        Returns:
            A dict representation of the report with open ports as dicts.
        """
        data = asdict(self)
        data["open_ports"] = [asdict(r) for r in self.open_ports]
        return data


# ---------------------------------------------------------------------------
# Core scanner class
# ---------------------------------------------------------------------------

class PortScanner:
    """High-performance, threaded TCP port scanner.

    Attributes:
        target: Hostname or IP address supplied by the caller.
        timeout: Per-connection socket timeout in seconds.
        threads: Maximum number of concurrent worker threads.
    """

    def __init__(
        self,
        target: str,
        timeout: float = 1.0,
        threads: int = 50,
    ) -> None:
        """Initialise the scanner with connection parameters.

        Args:
            target: Hostname or IP address to scan.
            timeout: Socket connection timeout in seconds. Defaults to 1.0.
            threads: Thread-pool size for concurrent scanning. Defaults to 50.
        """
        self.target: str = target
        self.timeout: float = timeout
        self.threads: int = threads
        self.resolved_ip: str = ""
        self._results: list[ScanResult] = []
        self._interrupted: bool = False

    # ------------------------------------------------------------------
    # Public interface
    # ------------------------------------------------------------------

    def resolve_target(self) -> str:
        """Resolve the target hostname to an IPv4 address.

        Returns:
            The resolved IP address string.

        Raises:
            SystemExit: If DNS resolution fails.
        """
        try:
            ip = socket.gethostbyname(self.target)
            self.resolved_ip = ip
            return ip
        except socket.gaierror as exc:
            console.print(
                f"[bold red][!] DNS resolution failed for '{self.target}': {exc}[/]"
            )
            sys.exit(1)

    def scan_port(self, port: int) -> ScanResult:
        """Attempt a TCP connection to a single port and classify its state.

        Args:
            port: The TCP port number to probe (0–65535).

        Returns:
            A :class:`ScanResult` indicating whether the port is open and,
            if so, the probable service name.
        """
        try:
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
                sock.settimeout(self.timeout)
                connected = sock.connect_ex((self.resolved_ip, port))
                is_open = connected == 0
                service = self._get_service_name(port) if is_open else "unknown"
                return ScanResult(port=port, is_open=is_open, service=service)
        except (socket.error, OSError):
            return ScanResult(port=port, is_open=False)

    def run_scan(
        self,
        ports: list[int],
        progress: Progress,
        task_id: TaskID,
    ) -> list[ScanResult]:
        """Execute a concurrent scan over the given port list.

        Submits all port-scan jobs to a :class:`ThreadPoolExecutor` and
        collects results as they complete.  Supports graceful shutdown via
        ``KeyboardInterrupt``.

        Args:
            ports: Ordered list of port numbers to scan.
            progress: A *rich* :class:`Progress` instance used to update the
                progress bar in real time.
            task_id: The task handle inside *progress* to advance.

        Returns:
            List of :class:`ScanResult` objects sorted by port number.
        """
        results: list[ScanResult] = []
        futures: dict[Future[ScanResult], int] = {}

        with ThreadPoolExecutor(max_workers=self.threads) as executor:
            for port in ports:
                future: Future[ScanResult] = executor.submit(
                    self.scan_port, port
                )
                futures[future] = port

            try:
                for future in as_completed(futures):
                    result = future.result()
                    results.append(result)
                    progress.advance(task_id)
            except KeyboardInterrupt:
                self._interrupted = True
                console.print(
                    "\n[bold yellow][!] Interrupt received — stopping threads "
                    "gracefully…[/]"
                )
                executor.shutdown(wait=False, cancel_futures=True)

        self._results = sorted(results, key=lambda r: r.port)
        return self._results

    # ------------------------------------------------------------------
    # Output helpers
    # ------------------------------------------------------------------

    def build_report(
        self,
        ports: list[int],
        start_time: float,
    ) -> ScanReport:
        """Compile a :class:`ScanReport` from collected scan results.

        Args:
            ports: Full list of ports that were (or were attempted to be)
                scanned — used to calculate *total_ports_scanned*.
            start_time: Unix timestamp recorded just before scanning began.

        Returns:
            A populated :class:`ScanReport` instance.
        """
        end_ts = time.time()
        open_ports = [r for r in self._results if r.is_open]
        return ScanReport(
            target=self.target,
            resolved_ip=self.resolved_ip,
            start_time=datetime.fromtimestamp(start_time).isoformat(),
            end_time=datetime.fromtimestamp(end_ts).isoformat(),
            duration_seconds=round(end_ts - start_time, 3),
            total_ports_scanned=len(self._results),
            open_ports=open_ports,
        )

    @staticmethod
    def save_report(report: ScanReport, path: str) -> None:
        """Serialise and write a :class:`ScanReport` to a JSON file.

        Args:
            report: The scan report to persist.
            path: Filesystem path of the output JSON file.

        Raises:
            OSError: Propagated if the file cannot be written.
        """
        with open(path, "w", encoding="utf-8") as fh:
            json.dump(report.to_dict(), fh, indent=2)
        console.print(
            f"\n[bold green][✓] Report saved → [underline]{path}[/underline][/]"
        )

    # ------------------------------------------------------------------
    # Private utilities
    # ------------------------------------------------------------------

    @staticmethod
    def _get_service_name(port: int) -> str:
        """Look up the IANA service name for a given port number.

        Args:
            port: A TCP port number.

        Returns:
            The service name string (e.g. ``"http"``, ``"ssh"``) or
            ``"unknown"`` when no mapping exists.
        """
        try:
            return socket.getservbyport(port, "tcp")
        except OSError:
            return "unknown"


# ---------------------------------------------------------------------------
# Argument parsing
# ---------------------------------------------------------------------------

def parse_ports(port_string: str) -> list[int]:
    """Convert a user-supplied port expression into a sorted list of integers.

    Accepts three input formats:

    * **Single port** – ``"80"``
    * **Comma-separated list** – ``"80,443,8080"``
    * **Range** – ``"1-1024"``

    Args:
        port_string: Raw string from the CLI ``--ports`` argument.

    Returns:
        Sorted, deduplicated list of port integers.

    Raises:
        argparse.ArgumentTypeError: If the expression is malformed or contains
            out-of-range values.
    """
    ports: list[int] = []
    try:
        if "-" in port_string and "," not in port_string:
            start, end = port_string.split("-", 1)
            ports = list(range(int(start), int(end) + 1))
        elif "," in port_string:
            ports = [int(p.strip()) for p in port_string.split(",")]
        else:
            ports = [int(port_string)]
    except ValueError:
        raise argparse.ArgumentTypeError(
            f"Invalid port expression: '{port_string}'. "
            "Use single (80), comma-separated (80,443), or range (1-1000)."
        )

    invalid = [p for p in ports if not 0 < p <= 65535]
    if invalid:
        raise argparse.ArgumentTypeError(
            f"Port(s) out of valid range (1–65535): {invalid[:5]}"
        )

    return sorted(set(ports))


def build_argument_parser() -> argparse.ArgumentParser:
    """Construct the CLI argument parser with all supported flags.

    Returns:
        A configured :class:`argparse.ArgumentParser` ready to call
        ``parse_args()`` on.
    """
    parser = argparse.ArgumentParser(
        prog="netsleuth",
        description="NetSleuth — Minimalist, high-performance TCP port scanner.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=(
            "Examples:\n"
            "  python scanner.py -t scanme.nmap.org\n"
            "  python scanner.py -t 192.168.1.1 -p 1-1024 --threads 100\n"
            "  python scanner.py -t example.com -p 80,443,8080 -o report.json\n"
        ),
    )
    parser.add_argument(
        "-t", "--target",
        required=True,
        metavar="HOST",
        help="Target hostname or IP address (required).",
    )
    parser.add_argument(
        "-p", "--ports",
        default=None,
        metavar="PORTS",
        help=(
            "Ports to scan. Accepts: single (80), "
            "comma-separated (80,443), or range (1-1024). "
            "Defaults to top-100 common ports."
        ),
    )
    parser.add_argument(
        "--timeout",
        type=float,
        default=1.0,
        metavar="SEC",
        help="Socket connection timeout in seconds (default: 1.0).",
    )
    parser.add_argument(
        "--threads",
        type=int,
        default=50,
        metavar="N",
        help="Number of concurrent threads (default: 50).",
    )
    parser.add_argument(
        "-o", "--output",
        default=None,
        metavar="FILE",
        help="Write results to a JSON file (e.g. results.json).",
    )
    return parser


# ---------------------------------------------------------------------------
# Display helpers
# ---------------------------------------------------------------------------

def print_banner() -> None:
    """Render the ASCII art banner and legal disclaimer to the terminal."""
    banner_text = Text(BANNER, style="bold cyan")
    disclaimer_text = Text(DISCLAIMER, style="bold yellow", justify="center")
    console.print(banner_text)
    console.print(
        Panel(disclaimer_text, border_style="yellow", expand=False, padding=(0, 2))
    )
    console.print()


def print_scan_header(target: str, resolved_ip: str, ports: list[int]) -> None:
    """Print a structured pre-scan info block.

    Args:
        target: The original target string provided by the user.
        resolved_ip: The resolved IPv4 address of the target.
        ports: Full list of ports to be scanned.
    """
    info_table = Table(box=box.SIMPLE, show_header=False, padding=(0, 1))
    info_table.add_column(style="bold cyan", no_wrap=True)
    info_table.add_column(style="white")

    port_summary = (
        f"{ports[0]}–{ports[-1]}" if len(ports) > 1 else str(ports[0])
    )
    info_table.add_row("Target", target)
    info_table.add_row("Resolved IP", resolved_ip)
    info_table.add_row("Ports", f"{port_summary}  ({len(ports)} total)")
    console.print(
        Panel(info_table, title="[bold]Scan Configuration[/]", border_style="cyan")
    )
    console.print()


def print_results_table(report: ScanReport) -> None:
    """Render the final scan results as a formatted Rich table.

    Args:
        report: Completed :class:`ScanReport` containing open port data.
    """
    table = Table(
        title=f"[bold]Open Ports on {report.resolved_ip}[/]",
        box=box.ROUNDED,
        border_style="cyan",
        header_style="bold cyan",
        show_lines=True,
    )
    table.add_column("PORT", style="bold green", justify="right", min_width=6)
    table.add_column("PROTOCOL", justify="center", min_width=10)
    table.add_column("SERVICE", style="bold blue", min_width=14)
    table.add_column("STATUS", justify="center", min_width=10)

    if not report.open_ports:
        console.print(
            Panel(
                "[yellow]No open ports found in the scanned range.[/]",
                border_style="yellow",
            )
        )
        return

    for result in sorted(report.open_ports, key=lambda r: r.port):
        table.add_row(
            str(result.port),
            "TCP",
            result.service.upper(),
            "[bold green]OPEN[/]",
        )

    console.print(table)


def print_scan_footer(report: ScanReport, interrupted: bool) -> None:
    """Print a summary footer with scan statistics.

    Args:
        report: The completed scan report.
        interrupted: Whether the scan was stopped early by the user.
    """
    status_msg = (
        "[bold yellow]⚡ Scan interrupted by user[/]"
        if interrupted
        else "[bold green]✓ Scan completed[/]"
    )
    summary_table = Table(box=box.SIMPLE, show_header=False, padding=(0, 1))
    summary_table.add_column(style="bold cyan", no_wrap=True)
    summary_table.add_column(style="white")
    summary_table.add_row("Ports Scanned", str(report.total_ports_scanned))
    summary_table.add_row("Open Ports Found", str(len(report.open_ports)))
    summary_table.add_row("Duration", f"{report.duration_seconds}s")

    console.print()
    console.print(
        Panel(
            summary_table,
            title=status_msg,
            border_style="green" if not interrupted else "yellow",
        )
    )


# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------

def main() -> None:
    """Parse arguments, coordinate the scan, and render output."""
    print_banner()

    parser = build_argument_parser()
    args = parser.parse_args()

    # ---- Resolve ports ------------------------------------------------
    if args.ports:
        try:
            ports = parse_ports(args.ports)
        except argparse.ArgumentTypeError as exc:
            console.print(f"[bold red][!] {exc}[/]")
            sys.exit(1)
    else:
        ports = TOP_100_PORTS
        console.print(
            "[dim]No port range specified — scanning top-100 common ports.[/dim]\n"
        )

    # ---- Initialise scanner -------------------------------------------
    scanner = PortScanner(
        target=args.target,
        timeout=args.timeout,
        threads=args.threads,
    )

    # ---- DNS resolution -----------------------------------------------
    console.print(f"[cyan]Resolving[/] [bold]{args.target}[/] …")
    resolved_ip = scanner.resolve_target()
    console.print(f"[green]  ✓ Resolved → {resolved_ip}[/]\n")

    print_scan_header(args.target, resolved_ip, ports)

    # ---- Progress bar setup ------------------------------------------
    progress = Progress(
        SpinnerColumn(),
        TextColumn("[progress.description]{task.description}"),
        BarColumn(bar_width=40),
        MofNCompleteColumn(),
        TimeElapsedColumn(),
        console=console,
        transient=True,
    )

    start_time = time.time()

    with Live(progress, console=console, refresh_per_second=20):
        task_id = progress.add_task(
            f"[cyan]Scanning {resolved_ip}[/]", total=len(ports)
        )
        scanner.run_scan(ports, progress, task_id)

    # ---- Build report -------------------------------------------------
    report = scanner.build_report(ports, start_time)

    # ---- Display results ----------------------------------------------
    print_results_table(report)
    print_scan_footer(report, scanner._interrupted)

    # ---- Optional JSON export ----------------------------------------
    if args.output:
        try:
            PortScanner.save_report(report, args.output)
        except OSError as exc:
            console.print(f"[bold red][!] Could not write output file: {exc}[/]")


if __name__ == "__main__":
    main()