aether_scanner.py

#!/usr/bin/env python3
import argparse
import asyncio
import csv
import ipaddress
import json
import os
import platform
import re
import socket
import subprocess
import time
from concurrent.futures import ThreadPoolExecutor

# Default common ports to scan if none are specified
DEFAULT_PORTS = [
    21, 22, 23, 25, 53, 80, 110, 111, 135, 139, 143, 443, 445, 993, 995, 1723, 
    3306, 3389, 5900, 8080, 8000, 3000, 5000, 5432, 6379, 27017
]

def check_root():
    """Checks if the script is running with root/administrator privileges."""
    try:
        return os.getuid() == 0
    except AttributeError:
        # Windows administrator check
        try:
            import ctypes
            return ctypes.windll.shell32.IsUserAnAdmin() != 0
        except Exception:
            return False

def parse_ports(port_str):
    """Parses port argument into a list of integers."""
    if not port_str:
        return DEFAULT_PORTS
        
    ports = []
    # Split by comma
    parts = [p.strip() for p in port_str.split(',') if p.strip()]
    for part in parts:
        if '-' in part:
            try:
                start_p, end_p = map(int, part.split('-'))
                if 1 <= start_p <= end_p <= 65535:
                    ports.extend(range(start_p, end_p + 1))
                else:
                    print(f"[!] Invalid port range ignored: {part}")
            except ValueError:
                print(f"[!] Invalid port format ignored: {part}")
        else:
            try:
                p = int(part)
                if 1 <= p <= 65535:
                    ports.append(p)
                else:
                    print(f"[!] Port out of range ignored: {p}")
            except ValueError:
                print(f"[!] Invalid port ignored: {part}")
                
    return sorted(list(set(ports)))

def parse_targets(target_str):
    """Parses CIDR, IP ranges, domains, and individual IPs into a list of hosts."""
    targets = []
    parts = [p.strip() for p in target_str.split(',') if p.strip()]
    
    for part in parts:
        if '/' in part:
            try:
                network = ipaddress.ip_network(part, strict=False)
                if network.num_addresses > 2:
                    targets.extend([str(ip) for ip in network.hosts()])
                else:
                    targets.extend([str(ip) for ip in network])
            except ValueError:
                print(f"[!] Invalid CIDR range: {part}")
        elif '-' in part:
            match = re.match(r'^([\d\.]+)\-(\d+)$', part)
            if match:
                start_ip_str = match.group(1)
                end_num = int(match.group(2))
                try:
                    start_ip = ipaddress.ip_address(start_ip_str)
                    octets = start_ip_str.split('.')
                    start_num = int(octets[-1])
                    if start_num <= end_num <= 255:
                        prefix = '.'.join(octets[:-1])
                        for i in range(start_num, end_num + 1):
                            targets.append(f"{prefix}.{i}")
                    else:
                        print(f"[!] Invalid range limits: {part}")
                except ValueError:
                    print(f"[!] Invalid starting IP in range: {part}")
            else:
                print(f"[!] Invalid range format: {part}")
        else:
            try:
                ipaddress.ip_address(part)
                targets.append(part)
            except ValueError:
                try:
                    resolved_ip = socket.gethostbyname(part)
                    targets.append(resolved_ip)
                except socket.gaierror:
                    print(f"[!] Could not resolve hostname: {part}")
                    
    return list(dict.fromkeys(targets))

# --- HOST DISCOVERY ENGINES ---

def arp_scan(target_ips, timeout=2.0):
    """Performs a local network Layer 2 ARP scan using Scapy."""
    discovered = []
    try:
        from scapy.all import ARP, Ether, srp
        print(f"[*] Starting local ARP sweep on {len(target_ips)} targets...")
        ans, _ = srp(Ether(dst="ff:ff:ff:ff:ff:ff")/ARP(pdst=target_ips), timeout=timeout, verbose=False)
        for _, rcv in ans:
            discovered.append({
                "ip": rcv.psrc,
                "mac": rcv.hwsrc,
                "status": "up"
            })
    except ImportError:
        print("[!] Scapy is not installed. Falling back to Ping sweeps.")
    except Exception as e:
        print(f"[!] ARP sweep encountered an error: {e}. Falling back to Ping sweeps.")
    return discovered

def ping_host(ip, timeout=1.0):
    """Pings a host using the native OS command or socket fallback."""
    system = platform.system().lower()
    if system == 'windows':
        cmd = ['ping', '-n', '1', '-w', str(int(timeout * 1000)), ip]
    elif system == 'darwin':
        cmd = ['ping', '-c', '1', '-t', str(max(1, int(timeout))), ip]
    else:
        cmd = ['ping', '-c', '1', '-W', str(max(1, int(timeout))), ip]
        
    try:
        res = subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        return ip, res.returncode == 0
    except Exception:
        # Fallback to connecting to port 80/443
        for port in (80, 443):
            try:
                s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                s.settimeout(timeout)
                s.connect((ip, port))
                s.close()
                return ip, True
            except Exception:
                pass
        return ip, False

def ping_sweep(ips, max_threads=100, timeout=1.5):
    """Performs a multi-threaded ping sweep across target hosts."""
    discovered = []
    print(f"[*] Starting multi-threaded Ping sweep on {len(ips)} targets...")
    with ThreadPoolExecutor(max_workers=min(len(ips), max_threads)) as executor:
        results = executor.map(lambda ip: ping_host(ip, timeout), ips)
        for ip, is_up in results:
            if is_up:
                discovered.append({
                    "ip": ip,
                    "mac": "Unknown",
                    "status": "up"
                })
    return discovered

# --- PORT SCANNING ENGINES ---

async def scan_tcp_port(ip, port, timeout=1.0):
    """Asynchronously scans a single TCP port using non-blocking connect."""
    try:
        conn = asyncio.open_connection(ip, port)
        _, writer = await asyncio.wait_for(conn, timeout=timeout)
        writer.close()
        try:
            await writer.wait_closed()
        except Exception:
            pass
        return port, "open"
    except asyncio.TimeoutError:
        return port, "filtered"
    except (ConnectionRefusedError, OSError):
        return port, "closed"

async def async_port_scan(ip, ports, concurrency=500, timeout=1.0):
    """Scans multiple ports asynchronously with concurrency throttling."""
    sem = asyncio.Semaphore(concurrency)
    
    async def scan_with_sem(port):
        async with sem:
            return await scan_tcp_port(ip, port, timeout)
            
    tasks = [scan_with_sem(port) for port in ports]
    results = await asyncio.gather(*tasks)
    return {port: status for port, status in results if status in ("open", "filtered")}

def scapy_syn_scan(ip, ports, timeout=1.0):
    """Performs a high-performance TCP SYN scan (requires root)."""
    from scapy.all import IP, TCP, sr1, send
    open_ports = {}
    
    def scan_port(port):
        try:
            pkt = IP(dst=ip)/TCP(dport=port, flags="S")
            resp = sr1(pkt, timeout=timeout, verbose=False)
            if resp is None:
                return port, "filtered"
            if resp.haslayer(TCP):
                tcp_layer = resp.getlayer(TCP)
                if tcp_layer.flags == 0x12: # SYN-ACK
                    # Send RST to close half-open connection cleanly
                    rst_pkt = IP(dst=ip)/TCP(dport=port, flags="R", seq=resp.ack)
                    send(rst_pkt, verbose=False)
                    return port, "open"
                elif tcp_layer.flags == 0x14: # RST-ACK
                    return port, "closed"
            return port, "closed"
        except Exception:
            return port, "closed"

    print(f"[*] Initiating TCP SYN stealth scan on {ip} for {len(ports)} ports...")
    with ThreadPoolExecutor(max_workers=100) as executor:
        results = executor.map(scan_port, ports)
        for port, status in results:
            if status in ("open", "filtered"):
                open_ports[port] = status
    return open_ports

# --- SERVICE DETECT & OS FINGERPRINTING ---

def grab_banner(ip, port, timeout=2.0):
    """Attempts to grab a service banner from an open port."""
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.settimeout(timeout)
        s.connect((ip, port))
        
        # Web service checks
        if port in (80, 8080, 8000, 3000, 5000):
            s.sendall(b"GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n")
        elif port == 443:
            import ssl
            context = ssl.create_default_context()
            context.check_hostname = False
            context.verify_mode = ssl.CERT_NONE
            s = context.wrap_socket(s, server_hostname=ip)
            s.sendall(b"GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n")
            
        try:
            banner = s.recv(1024).decode('utf-8', errors='ignore').strip()
            if banner:
                if "HTTP/" in banner:
                    for line in banner.split('\n'):
                        if line.lower().startswith("server:"):
                            return line.split(':', 1)[1].strip()
                    return "Web Server (HTTP)"
                return banner.split('\n')[0].strip()
        except socket.timeout:
            pass
        finally:
            s.close()
    except Exception:
        pass
    return "Unknown"

def detect_os(ip, timeout=1.5):
    """Guesses the OS using TTL headers from ping or Scapy."""
    # Scapy raw packet check if root
    try:
        from scapy.all import IP, ICMP, sr1
        pkt = IP(dst=ip)/ICMP()
        resp = sr1(pkt, timeout=timeout, verbose=False)
        if resp and resp.haslayer(IP):
            ttl = resp.getlayer(IP).ttl
            return classify_ttl(ttl)
    except Exception:
        pass
        
    # Subprocess command check
    system = platform.system().lower()
    if system == 'windows':
        cmd = ['ping', '-n', '1', '-w', str(int(timeout * 1000)), ip]
    else:
        cmd = ['ping', '-c', '1', '-W', str(max(1, int(timeout))), ip]
        
    try:
        res = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
        if res.returncode == 0:
            match = re.search(r'ttl=(\d+)', res.stdout, re.IGNORECASE)
            if match:
                return classify_ttl(int(match.group(1)))
    except Exception:
        pass
    return "Unknown OS"

def classify_ttl(ttl):
    if ttl <= 64:
        return f"Linux/macOS/Unix"
    elif ttl <= 128:
        return f"Windows"
    else:
        return f"Network Device/Cisco"

# --- OUTPUT FORMATTERS ---

def print_table(results):
    """Outputs the scan results to stdout in a pretty table."""
    for host in results:
        print(f"\nHost: {host['ip']} ({host['mac']}) | OS: {host['os']} | Status: {host['status']}")
        print("PORT\t\tSTATE\t\tSERVICE BANNER")
        print("---------------------------------------------")
        if not host.get("ports"):
            print("No open/filtered ports found.")
        else:
            for port, details in host["ports"].items():
                print(f"{port}/tcp\t{details['state']}\t{details['banner']}")

def export_results(results, filepath, export_format):
    """Exports the results to a JSON or CSV file."""
    if export_format == "json":
        with open(filepath, 'w') as f:
            json.dump(results, f, indent=4)
        print(f"\n[+] Results successfully exported to JSON: {filepath}")
    elif export_format == "csv":
        with open(filepath, 'w', newline='') as f:
            writer = csv.writer(f)
            writer.writerow(["Host IP", "MAC Address", "Host Status", "Estimated OS", "Port", "Port State", "Service Banner"])
            for host in results:
                if not host.get("ports"):
                    writer.writerow([host["ip"], host["mac"], host["status"], host["os"], "None", "None", "None"])
                else:
                    for port, details in host["ports"].items():
                        writer.writerow([host["ip"], host["mac"], host["status"], host["os"], f"{port}/tcp", details["state"], details["banner"]])
        print(f"\n[+] Results successfully exported to CSV: {filepath}")

# --- MAIN CONTROLLER ---

def main():
    parser = argparse.ArgumentParser(description="Aether Network Scanner - A High-Performance Cross-Platform Security Tool")
    parser.add_argument("-t", "--target", required=True, help="Target IP address, range (e.g. 192.168.1.1-50), CIDR block (192.168.1.0/24), or hostname.")
    parser.add_argument("-p", "--ports", help="Ports to scan. Example: 22,80,443 or range 20-100. Default: 26 common ports.")
    parser.add_argument("-s", "--scan-type", choices=["syn", "connect", "auto"], default="auto", help="Scan type. SYN (root required) or Connect (asynchronous socket). Default: auto.")
    parser.add_argument("--ping", action="store_true", help="Perform Host Discovery sweep before port scanning.")
    parser.add_argument("--banner", action="store_true", help="Attempt service version banner grabbing on open ports.")
    parser.add_argument("--os", action="store_true", help="Estimate host Operating System via TTL fingerprinting.")
    parser.add_argument("--concurrency", type=int, default=500, help="Maximum concurrent sockets for Connect scan (Default: 500).")
    parser.add_argument("--timeout", type=float, default=1.0, help="Connection timeout in seconds (Default: 1.0).")
    parser.add_argument("-o", "--output", help="Write scan results to a file path.")
    parser.add_argument("-f", "--format", choices=["table", "json", "csv"], default="table", help="Output format. Default: table.")
    
    args = parser.parse_args()
    is_root = check_root()
    
    start_time = time.time()
    
    # 1. Parse Targets
    hosts = parse_targets(args.target)
    if not hosts:
        print("[!] No valid target hosts identified. Exiting.")
        return
        
    print(f"[*] Starting Aether Network Scanner at {time.strftime('%Y-%m-%d %H:%M:%S')}")
    print(f"[*] Targets parsed: {len(hosts)} hosts.")
    print(f"[*] Running with administrative privileges: {is_root}")
    
    # 2. Host Discovery
    alive_hosts = []
    if args.ping:
        # If root, do local ARP sweep for speed, otherwise fall back to ping sweeps
        if is_root and any(ipaddress.ip_address(h).is_private for h in hosts):
            # Attempt ARP sweep on private targets
            arp_results = arp_scan(hosts, timeout=args.timeout)
            alive_hosts.extend(arp_results)
            # Find targets missed by ARP (e.g. non-local or failed ARP)
            scanned_ips = {h["ip"] for h in arp_results}
            remaining_ips = [h for h in hosts if h not in scanned_ips]
            if remaining_ips:
                ping_results = ping_sweep(remaining_ips, timeout=args.timeout)
                alive_hosts.extend(ping_results)
        else:
            alive_hosts = ping_sweep(hosts, timeout=args.timeout)
            
        print(f"[+] Host discovery finished. Found {len(alive_hosts)} active hosts.")
    else:
        # Treat all targets as alive
        alive_hosts = [{"ip": h, "mac": "Unknown", "status": "up"} for h in hosts]

    # 3. Parse Ports
    ports_to_scan = parse_ports(args.ports)
    
    # 4. Port Scanning & Analysis
    scan_results = []
    for host in alive_hosts:
        ip = host["ip"]
        host_status = {
            "ip": ip,
            "mac": host["mac"],
            "status": host["status"],
            "os": detect_os(ip, args.timeout) if args.os else "Not Requested",
            "ports": {}
        }
        
        # Decide Scan Type
        actual_scan = args.scan_type
        if actual_scan == "auto":
            actual_scan = "syn" if is_root else "connect"
            
        if actual_scan == "syn" and not is_root:
            print("[!] TCP SYN scan requested but running without root. Falling back to Asynchronous Connect Scan.")
            actual_scan = "connect"
            
        # Execute Scan
        if actual_scan == "syn":
            open_ports = scapy_syn_scan(ip, ports_to_scan, args.timeout)
        else:
            print(f"[*] Scanning {ip} asynchronously (concurrency limit: {args.concurrency})...")
            open_ports = asyncio.run(async_port_scan(ip, ports_to_scan, args.concurrency, args.timeout))
            
        # Banner Grabbing
        for port, state in open_ports.items():
            banner = "Grab Not Requested"
            if args.banner and state == "open":
                banner = grab_banner(ip, port, args.timeout + 1.0)
            host_status["ports"][port] = {
                "state": state,
                "banner": banner
            }
            
        scan_results.append(host_status)
        
    # 5. Output Results
    if args.format == "table":
        print_table(scan_results)
    elif args.output:
        export_results(scan_results, args.output, args.format)
    else:
        # Print table if output format is specified but no output path
        print_table(scan_results)
        
    # Handle direct JSON/CSV output stdout redirection if file path is missing but format is JSON/CSV
    if not args.output and args.format != "table":
        if args.format == "json":
            print(json.dumps(scan_results, indent=4))
            
    print(f"\n[*] Scan completed in {time.time() - start_time:.2f} seconds.")

if __name__ == "__main__":
    main()