scanner.py

import asyncio
import socket
from typing import List, Tuple

class Scanner:
    def __init__(self):
        pass

    async def scan_port(self, ip: str, port: int, timeout: float = 1.0) -> Tuple[str, int, bool]:
        """
        Scans a specific port on an IP address.
        Returns (ip, port, is_open).
        """
        conn = None
        try:
            conn = await asyncio.wait_for(
                asyncio.open_connection(ip, port),
                timeout=timeout
            )
            return ip, port, True
        except (asyncio.TimeoutError, ConnectionRefusedError, OSError):
            return ip, port, False
        finally:
            if conn:
                reader, writer = conn
                writer.close()
                await writer.wait_closed()

    async def scan_ports(self, ip: str, ports: List[int]) -> List[Tuple[str, int, bool]]:
        """
        Scans a list of ports on a single IP.
        """
        tasks = [self.scan_port(ip, port) for port in ports]
        results = await asyncio.gather(*tasks)
        return results

    async def scan_range(self, ips: List[str], ports: List[int]) -> List[Tuple[str, int, bool]]:
        """
        Scans a list of IPs for the given ports.
        """
        tasks = []
        for ip in ips:
            tasks.append(self.scan_ports(ip, ports))
        
        results_nested = await asyncio.gather(*tasks)
        # Flatten results
    def parse_ports(self, ports_str: str) -> List[int]:
        """
        Parses a string of ports (e.g., "80, 443, 8000-8010") into a list of integers.
        """
        ports = set()
        parts = ports_str.split(',')
        for part in parts:
            part = part.strip()
            if not part:
                continue
            if '-' in part:
                try:
                    start, end = map(int, part.split('-'))
                    ports.update(range(start, end + 1))
                except ValueError:
                    continue
            else:
                try:
                    ports.add(int(part))
                except ValueError:
                    continue
        return sorted(list(ports))

    def parse_ips(self, ip_str: str) -> List[str]:
        """
        Parses a string of IPs (e.g., "192.168.1.1, 10.0.0.1") into a list of strings.
        For simplicity, this version handles comma-separated IPs.
        Future improvement: Support CIDR or ranges like 192.168.1.1-10.
        """
        # Basic implementation for now
        ips = [ip.strip() for ip in ip_str.split(',') if ip.strip()]
        return ips