start.py

import scapy.all as scapy
import ipaddress
import socket
import threading
import argparse
from queue import Queue
import json
import logging
from typing import List, Dict, Tuple, Optional, Any

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[logging.FileHandler("nascar_scan.log"), logging.StreamHandler()]
)
logger = logging.getLogger("Nascar")

# ANSI Escape Codes for coloring
class Colors:
    GREEN = "\033[92m"
    BLUE = "\033[94m"
    YELLOW = "\033[93m"
    RED = "\033[91m"
    END = "\033[0m"
    BOLD = "\033[1m"

def get_args() -> argparse.Namespace:
    """Parses command line arguments."""
    parser = argparse.ArgumentParser(description="Nascar: Advanced WAN/LAN Network Scanner")
    parser.add_argument("-n", "--network", type=str, required=True, help="Target Domain, IP, or CIDR")
    parser.add_argument("-t", "--threads", type=int, default=10, help="Number of threads (default: 10)")
    parser.add_argument("-v", "--verbose", action="store_true", help="Enable verbose output")
    parser.add_argument("-s", "--silent", action="store_true", help="Run in silent mode")
    parser.add_argument("-lm", "--lateral", action="store_true", help="Enable lateral movement (port scan on alive hosts)")
    parser.add_argument("--timeout", type=float, default=1.0, help="Timeout for scan operations (default: 1.0s)")
    parser.add_argument("-o", "--output", type=str, help="Save results to a JSON file")
    return parser.parse_args()

def is_host_alive_icmp(ip: str, timeout: float) -> Tuple[bool, Optional[int]]:
    """Checks if a host is alive using ICMP (Ping). Returns (alive, ttl)."""
    try:
        icmp = scapy.IP(dst=ip)/scapy.ICMP()
        resp = scapy.sr1(icmp, timeout=timeout, verbose=0)
        if resp is not None:
            return True, int(resp.ttl)
        return False, None
    except Exception:
        return False, None

def guess_os(ttl: Optional[int]) -> str:
    """Guesses the OS based on the TTL value."""
    if ttl is None:
        return "Unknown"
    elif ttl >= 120:
        return "Windows"
    elif ttl >= 60:
        return "Linux/Unix"
    else:
        return "Unknown (Low TTL)"

def port_scan(ip: str, ports: List[int], timeout: float) -> List[int]:
    """Scans a list of ports on a given IP address."""
    open_ports = []
    for port in ports:
        try:
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
                s.settimeout(timeout / 2)
                if s.connect_ex((ip, port)) == 0:
                    open_ports.append(port)
        except Exception:
            pass
    return open_ports

def resolve_targets(target: str) -> List[ipaddress.IPv4Address]:
    """Resolves the target input into a list of IP addresses."""
    try:
        # Check if CIDR or single IP
        if "/" in target:
            net = ipaddress.ip_network(target, strict=False)
            return list(net.hosts())
        else:
            # Single IP
            return [ipaddress.ip_address(target)]
    except ValueError:
        # Not an IP or CIDR, try as domain
        try:
            _, _, ip_list = socket.gethostbyname_ex(target)
            return [ipaddress.ip_address(ip) for ip in ip_list]
        except Exception as e:
            logger.error(f"Could not resolve target '{target}': {e}")
            return []

def worker(ip_queue: Queue, results: List[Dict[str, Any]], results_lock: threading.Lock, args: argparse.Namespace, common_ports: List[int]):
    """Worker thread function to process IPs from the queue."""
    while not ip_queue.empty():
        try:
            ip = ip_queue.get_nowait()
        except:
            break
        
        ip_str = str(ip)
        mac = "N/A"
        hostname = "Unknown"
        alive = False
        os_name = "Unknown"

        is_local = ip.is_private

        if is_local:
            # ARP Scan for Local
            arp = scapy.ARP(pdst=ip_str)
            broadcast = scapy.Ether(dst="ff:ff:ff:ff:ff:ff")
            packet = broadcast / arp
            try:
                answered = scapy.srp(packet, timeout=args.timeout, verbose=False)[0]
                if answered:
                    alive = True
                    mac = answered[0][1].hwsrc
                    try:
                        hostname = socket.gethostbyaddr(answered[0][1].psrc)[0]
                    except Exception:
                        pass
                    _, ttl = is_host_alive_icmp(ip_str, args.timeout)
                    os_name = guess_os(ttl)
            except Exception as e:
                if args.verbose:
                    logger.debug(f"ARP failed for {ip_str}: {e}")
        
        # Fallback to ICMP if not found by ARP or if explicitly external
        if not alive:
            alive, ttl = is_host_alive_icmp(ip_str, args.timeout)
            if alive:
                os_name = guess_os(ttl)
                try:
                    hostname = socket.gethostbyaddr(ip_str)[0]
                except Exception:
                    pass

        if alive:
            info = {
                "ip": ip_str,
                "mac": mac,
                "hostname": hostname,
                "os": os_name
            }
            if args.lateral:
                info["open_ports"] = port_scan(ip_str, common_ports, args.timeout)
            
            with results_lock:
                results.append(info)
                if not args.silent:
                    msg = f"{Colors.GREEN}[+] Found:{Colors.END} {Colors.BOLD}{ip_str:<15}{Colors.END} | {os_name:<10} | {hostname}"
                    print(msg)
        elif args.verbose:
            print(f"{Colors.RED}[-]{Colors.END} No response from {ip_str}")
        
        ip_queue.task_done()

def main():
    args = get_args()
    if args.silent:
        logger.setLevel(logging.WARNING)

    common_ports = [22, 80, 443, 3389, 445, 139, 21, 23, 25, 53]
    
    targets = resolve_targets(args.network)
    if not targets:
        return

    ip_queue = Queue()
    for ip in targets:
        ip_queue.put(ip)

    results = []
    results_lock = threading.Lock()

    if not args.silent:
        print(f"\n{Colors.BLUE}{Colors.BOLD}[*] Nascar Scanner Initialized{Colors.END}")
        print(f"[*] Target: {args.network}")
        print(f"[*] Threads: {args.threads}")
        print(f"[*] Timeout: {args.timeout}s")
        print(f"[*] Scanning {len(targets)} potential hosts...\n")

    threads = []
    for _ in range(min(args.threads, len(targets))):
        t = threading.Thread(target=worker, args=(ip_queue, results, results_lock, args, common_ports))
        t.daemon = True
        t.start()
        threads.append(t)

    try:
        ip_queue.join()
    except KeyboardInterrupt:
        print(f"\n{Colors.YELLOW}[!] Scan interrupted by user. Displaying partial results...{Colors.END}")

    print(f"\n{Colors.BLUE}{Colors.BOLD}Scan Results:{Colors.END}")
    if args.lateral:
        header = f"{Colors.BOLD}{'IP Address':<16} {'MAC Address':<18} {'Hostname':<25} {'OS':<12} {'Open Ports'}{Colors.END}"
        print(header)
        print("-" * 95)
        for info in results:
            print("{:<16} {:<18} {:<25} {:<12} {}".format(
                info['ip'], info['mac'], info['hostname'], info['os'],
                ",".join(str(p) for p in info.get('open_ports', []))
            ))
    else:
        header = f"{Colors.BOLD}{'IP Address':<16} {'MAC Address':<18} {'Hostname':<25} {'OS':<12}{Colors.END}"
        print(header)
        print("-" * 75)
        for info in results:
            print("{:<16} {:<18} {:<25} {:<12}".format(info['ip'], info['mac'], info['hostname'], info['os']))

    if args.output:
        try:
            with open(args.output, 'w') as f:
                json.dump(results, f, indent=4)
            print(f"\n{Colors.GREEN}[!] Detailed report saved to {args.output}{Colors.END}")
        except Exception as e:
            logger.error(f"Failed to save output: {e}")

if __name__ == "__main__":
    main()