SubScan.py

#!/usr/bin/env python3
"""
NetScanner - Network Scanning and Host Enumeration Tool
Requires: scapy (pip install scapy)
Run as root/admin for ICMP ping sweep
"""

import socket
import ipaddress
import threading
import sys
import os
import re
import platform
import subprocess
import datetime
from concurrent.futures import ThreadPoolExecutor, as_completed

try:
    from scapy.all import sr1, IP, ICMP, conf
    conf.verb = 0  # Suppress scapy output
    SCAPY_AVAILABLE = True
except ImportError:
    SCAPY_AVAILABLE = False

# ──────────────────────────────────────────────
# TOP 1000 COMMON PORTS (trimmed to most useful)
# ──────────────────────────────────────────────
COMMON_PORTS = [
    21, 22, 23, 25, 53, 80, 110, 111, 135, 139, 143, 443, 445,
    993, 995, 1723, 3306, 3389, 5900, 8080, 8443, 8888,
    # Extended common
    20, 69, 79, 88, 119, 123, 137, 138, 161, 162, 179, 194,
    389, 427, 465, 500, 514, 515, 543, 544, 548, 554, 587,
    631, 636, 646, 873, 990, 992, 1080, 1194, 1433, 1434,
    1521, 1701, 1723, 1812, 1813, 2049, 2082, 2083, 2086, 2087,
    2095, 2096, 2181, 2375, 2376, 2483, 2484, 3000, 3128, 3268,
    3269, 3306, 3389, 3690, 4333, 4444, 4848, 5000, 5001, 5432,
    5632, 5800, 5900, 5985, 5986, 6379, 6443, 6666, 7001, 7002,
    7070, 7443, 7474, 8000, 8008, 8009, 8080, 8081, 8083, 8088,
    8161, 8443, 8500, 8888, 8983, 9000, 9001, 9042, 9090, 9092,
    9200, 9300, 9418, 9999, 10000, 10250, 10255, 11211, 27017,
    27018, 27019, 50000, 50030, 50060, 50070, 50075, 50090
]

# Port service names
SERVICE_MAP = {
    20: "FTP-data", 21: "FTP", 22: "SSH", 23: "Telnet", 25: "SMTP",
    53: "DNS", 69: "TFTP", 79: "Finger", 80: "HTTP", 88: "Kerberos",
    110: "POP3", 111: "RPC", 119: "NNTP", 123: "NTP", 135: "MSRPC",
    137: "NetBIOS-ns", 138: "NetBIOS-dgm", 139: "NetBIOS-ssn",
    143: "IMAP", 161: "SNMP", 162: "SNMP-trap", 179: "BGP",
    389: "LDAP", 443: "HTTPS", 445: "SMB", 465: "SMTPS",
    500: "IKE", 514: "Syslog", 515: "LPD", 587: "SMTP-sub",
    631: "IPP", 636: "LDAPS", 873: "rsync", 990: "FTPS",
    993: "IMAPS", 995: "POP3S", 1080: "SOCKS", 1194: "OpenVPN",
    1433: "MSSQL", 1434: "MSSQL-UDP", 1521: "Oracle",
    1723: "PPTP", 2049: "NFS", 2375: "Docker", 2376: "Docker-TLS",
    3000: "Dev-Server", 3128: "Squid", 3268: "LDAP-GC",
    3306: "MySQL", 3389: "RDP", 3690: "SVN", 4444: "Metasploit",
    4848: "GlassFish", 5000: "Flask/UPnP", 5432: "PostgreSQL",
    5632: "PCAnywhere", 5800: "VNC-HTTP", 5900: "VNC",
    5985: "WinRM-HTTP", 5986: "WinRM-HTTPS", 6379: "Redis",
    6443: "K8s-API", 7001: "WebLogic", 8000: "HTTP-alt",
    8008: "HTTP-alt", 8080: "HTTP-proxy", 8081: "HTTP-alt",
    8443: "HTTPS-alt", 8888: "Jupyter", 9000: "PHP-FPM",
    9090: "Prometheus", 9092: "Kafka", 9200: "Elasticsearch",
    9300: "Elasticsearch-transport", 9418: "Git",
    10000: "Webmin", 10250: "Kubelet", 11211: "Memcached",
    27017: "MongoDB", 27018: "MongoDB-shard", 27019: "MongoDB-config",
    50000: "DB2",
}


class Tee:
    """Writes output to both terminal and file simultaneously."""
    def __init__(self, filename):
        self.terminal = sys.stdout
        self.file = open(filename, 'w', encoding='utf-8')

    def write(self, message):
        self.terminal.write(message)
        self.file.write(message)

    def flush(self):
        self.terminal.flush()
        self.file.flush()

    def close(self):
        self.file.close()


# ──────────────────────────────────────────────
# COLORS
# ──────────────────────────────────────────────
class C:
    RED    = '\033[91m'
    GREEN  = '\033[92m'
    YELLOW = '\033[93m'
    CYAN   = '\033[96m'
    BOLD   = '\033[1m'
    DIM    = '\033[2m'
    RESET  = '\033[0m'

def banner():
    print(f"""{C.CYAN}{C.BOLD}
╔══════════════════════════════════════════════╗
║           NetScanner v1.0                   ║
║   Network Scanning & Host Enumeration Tool  ║
╚══════════════════════════════════════════════╝{C.RESET}
""")

# ──────────────────────────────────────────────
# AUTO DETECT IP + SUBNET
# ──────────────────────────────────────────────
def auto_detect_interfaces() -> list:
    """
    Runs ipconfig (Windows) or ip addr / ifconfig (Linux/Mac)
    Returns list of dicts: {iface, ip, subnet, prefix}
    """
    results = []
    os_type = platform.system()

    try:
        if os_type == "Windows":
            out = subprocess.check_output("ipconfig", shell=True, text=True, stderr=subprocess.DEVNULL)
            iface, ip, mask = None, None, None
            for line in out.splitlines():
                line = line.strip()
                if line.endswith(':') and not line.startswith(' '):
                    iface = line.rstrip(':')
                    ip, mask = None, None
                m = re.search(r'IPv4 Address[.\s]+:\s*([\d.]+)', line)
                if m: ip = m.group(1)
                m = re.search(r'Subnet Mask[.\s]+:\s*([\d.]+)', line)
                if m: mask = m.group(1)
                if ip and mask and iface:
                    try:
                        net = ipaddress.IPv4Network(f"{ip}/{mask}", strict=False)
                        results.append({"iface": iface, "ip": ip, "subnet": mask, "prefix": net.prefixlen, "network": net})
                        ip, mask = None, None
                    except: pass

        else:
            # Try `ip addr` first (modern Linux)
            try:
                out = subprocess.check_output(["ip", "addr"], text=True, stderr=subprocess.DEVNULL)
                iface = None
                for line in out.splitlines():
                    m = re.match(r'^\d+:\s+(\S+):', line)
                    if m: iface = m.group(1)
                    m = re.search(r'inet\s+([\d.]+)/(\d+)', line)
                    if m and iface:
                        ip, prefix = m.group(1), int(m.group(2))
                        if not ip.startswith('127'):
                            net = ipaddress.IPv4Network(f"{ip}/{prefix}", strict=False)
                            results.append({"iface": iface, "ip": ip, "subnet": str(net.netmask), "prefix": prefix, "network": net})
            except FileNotFoundError:
                # Fallback: ifconfig
                out = subprocess.check_output(["ifconfig"], text=True, stderr=subprocess.DEVNULL)
                iface = None
                for line in out.splitlines():
                    m = re.match(r'^(\S+)', line)
                    if m: iface = m.group(1).rstrip(':')
                    m = re.search(r'inet\s+([\d.]+)\s+netmask\s+([\d.]+)', line)
                    if m and iface:
                        ip, mask = m.group(1), m.group(2)
                        if not ip.startswith('127'):
                            net = ipaddress.IPv4Network(f"{ip}/{mask}", strict=False)
                            results.append({"iface": iface, "ip": ip, "subnet": mask, "prefix": net.prefixlen, "network": net})
    except Exception as e:
        print(f"{C.YELLOW}[!] Auto-detect failed: {e}{C.RESET}")

    return results


def select_interface(interfaces: list):
    """Let user pick an interface or enter manually."""
    if not interfaces:
        print(f"{C.YELLOW}[!] No interfaces auto-detected. Manual entry required.{C.RESET}\n")
        return None, None

    print(f"{C.BOLD}Detected Interfaces:{C.RESET}")
    for i, iface in enumerate(interfaces):
        print(f"  {C.YELLOW}[{i+1}]{C.RESET} {iface['iface']:<20} {C.GREEN}{iface['ip']}{C.RESET}/{iface['prefix']}  →  {iface['network']}")
    print(f"  {C.YELLOW}[M]{C.RESET} Manual entry\n")

    while True:
        choice = input("Select interface [1/2/.../M]: ").strip().upper()
        if choice == 'M':
            return None, None
        try:
            idx = int(choice) - 1
            if 0 <= idx < len(interfaces):
                sel = interfaces[idx]
                print(f"\n{C.GREEN}[✓] Using: {sel['iface']} — {sel['ip']}/{sel['prefix']}{C.RESET}\n")
                return sel['ip'], sel['prefix']
        except ValueError:
            pass
        print(f"{C.RED}  Invalid choice.{C.RESET}")


# ──────────────────────────────────────────────
# PING (ICMP)
# ──────────────────────────────────────────────
def ping_host_scapy(ip: str, timeout: float = 1) -> bool:
    pkt = sr1(IP(dst=ip)/ICMP(), timeout=timeout)
    return pkt is not None

def ping_host_socket(ip: str, timeout: float = 1) -> bool:
    """Fallback: TCP SYN to port 80 if scapy unavailable."""
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.settimeout(timeout)
        result = s.connect_ex((ip, 80))
        s.close()
        return result == 0
    except:
        return False

def ping_host(ip: str) -> bool:
    if SCAPY_AVAILABLE:
        return ping_host_scapy(ip)
    return ping_host_socket(ip)

# ──────────────────────────────────────────────
# PORT SCAN
# ──────────────────────────────────────────────
def scan_port(ip: str, port: int, timeout: float = 0.5) -> bool:
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.settimeout(timeout)
        result = s.connect_ex((ip, port))
        s.close()
        return result == 0
    except:
        return False

def get_service(port: int) -> str:
    if port in SERVICE_MAP:
        return SERVICE_MAP[port]
    try:
        return socket.getservbyport(port)
    except:
        return "unknown"

def get_banner(ip: str, port: int) -> str:
    try:
        s = socket.socket()
        s.settimeout(1)
        s.connect((ip, port))
        banner = s.recv(1024).decode(errors='ignore').strip()
        s.close()
        return banner[:60] if banner else ""
    except:
        return ""

def scan_ports(ip: str, ports: list, threads: int = 100) -> list:
    open_ports = []
    with ThreadPoolExecutor(max_workers=threads) as ex:
        futures = {ex.submit(scan_port, ip, p): p for p in ports}
        for f in as_completed(futures):
            if f.result():
                open_ports.append(futures[f])
    return sorted(open_ports)

# ──────────────────────────────────────────────
# HOST DISCOVERY
# ──────────────────────────────────────────────
def discover_hosts(network: ipaddress.IPv4Network, threads: int = 50) -> list:
    hosts = [str(h) for h in network.hosts()]
    active = []
    lock = threading.Lock()

    print(f"{C.CYAN}[*] Scanning {len(hosts)} hosts...{C.RESET}")

    def check(ip):
        if ping_host(ip):
            with lock:
                active.append(ip)
                print(f"  {C.GREEN}[+] {ip} — ALIVE{C.RESET}")

    with ThreadPoolExecutor(max_workers=threads) as ex:
        ex.map(check, hosts)

    return sorted(active, key=lambda x: ipaddress.IPv4Address(x))

# ──────────────────────────────────────────────
# PORT MODE SELECTION
# ──────────────────────────────────────────────
def select_port_mode() -> list:
    print(f"""
{C.BOLD}Port Scan Mode:{C.RESET}
  {C.YELLOW}[1]{C.RESET} Common Ports ({len(COMMON_PORTS)} ports)
  {C.YELLOW}[2]{C.RESET} Full Range   (1–65535)
  {C.YELLOW}[3]{C.RESET} Custom Range
""")
    while True:
        choice = input("Select [1/2/3]: ").strip()
        if choice == '1':
            return COMMON_PORTS
        elif choice == '2':
            print(f"{C.DIM}  Full scan may take a while...{C.RESET}")
            return list(range(1, 65536))
        elif choice == '3':
            try:
                start = int(input("  Start port: ").strip())
                end   = int(input("  End port:   ").strip())
                if 1 <= start <= end <= 65535:
                    return list(range(start, end + 1))
                print(f"{C.RED}  Invalid range.{C.RESET}")
            except ValueError:
                print(f"{C.RED}  Numbers only.{C.RESET}")
        else:
            print(f"{C.RED}  Choose 1, 2, or 3.{C.RESET}")

# ──────────────────────────────────────────────
# MAIN
# ──────────────────────────────────────────────
def main():
    # Setup output file
    timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
    outfile = f"scan_{timestamp}.txt"
    tee = Tee(outfile)
    sys.stdout = tee

    banner()
    print(f"{C.DIM}Output saving to: {outfile}{C.RESET}\n")

    if not SCAPY_AVAILABLE:
        print(f"{C.YELLOW}[!] scapy not found — using TCP fallback for host discovery{C.RESET}")
        print(f"{C.DIM}    pip install scapy for ICMP ping sweep{C.RESET}\n")

    # ── Get IP & Subnet (Auto-detect) ──
    try:
        print(f"{C.CYAN}[*] Detecting network interfaces...{C.RESET}\n")
        interfaces = auto_detect_interfaces()
        local_ip, prefix = select_interface(interfaces)

        # Manual fallback
        if local_ip is None:
            local_ip = input(f"{C.BOLD}Enter your IP address: {C.RESET}").strip()
            subnet   = input(f"{C.BOLD}Enter subnet mask (e.g. 255.255.255.0 or /24): {C.RESET}").strip()
            if subnet.startswith('/'):
                prefix = int(subnet[1:])
            else:
                prefix = ipaddress.IPv4Network(f"0.0.0.0/{subnet}").prefixlen

        network = ipaddress.IPv4Network(f"{local_ip}/{prefix}", strict=False)

    except Exception as e:
        print(f"{C.RED}[!] Invalid input: {e}{C.RESET}")
        sys.stdout = tee.terminal
        tee.close()
        return

    print(f"\n{C.CYAN}[*] Network: {network}{C.RESET}")
    print(f"{C.CYAN}[*] Hosts in range: {network.num_addresses - 2}{C.RESET}\n")

    # ── Port Mode ──
    ports = select_port_mode()
    print(f"\n{C.CYAN}[*] Port range: {ports[0]}–{ports[-1]} ({len(ports)} ports){C.RESET}\n")

    start_time = datetime.datetime.now()
    print(f"{C.DIM}Scan started: {start_time.strftime('%Y-%m-%d %H:%M:%S')}{C.RESET}\n")
    print("─" * 50)

    # ── Host Discovery ──
    print(f"\n{C.BOLD}[ PHASE 1 — Host Discovery ]{C.RESET}")
    active_hosts = discover_hosts(network)

    if not active_hosts:
        print(f"\n{C.RED}[!] No active hosts found.{C.RESET}")
        sys.stdout = tee.terminal
        tee.close()
        return

    print(f"\n{C.GREEN}[*] {len(active_hosts)} active host(s) found.{C.RESET}\n")

    # ── Port Scan ──
    print(f"{C.BOLD}[ PHASE 2 — Port Scanning ]{C.RESET}\n")

    results = {}
    for host in active_hosts:
        print(f"{C.CYAN}[*] Scanning {host}...{C.RESET}")
        open_ports = scan_ports(host, ports)
        results[host] = open_ports

        if open_ports:
            for p in open_ports:
                svc = get_service(p)
                bnr = get_banner(host, p)
                bnr_str = f"  {C.DIM}» {bnr}{C.RESET}" if bnr else ""
                print(f"    {C.GREEN}[OPEN]{C.RESET} {p:<6} {C.YELLOW}{svc}{C.RESET}{bnr_str}")
        else:
            print(f"    {C.DIM}No open ports found.{C.RESET}")
        print()

    # ── Summary ──
    end_time = datetime.datetime.now()
    elapsed  = (end_time - start_time).seconds

    print("─" * 50)
    print(f"\n{C.BOLD}[ SCAN SUMMARY ]{C.RESET}")
    print(f"  Network   : {network}")
    print(f"  Hosts up  : {len(active_hosts)}")
    print(f"  Duration  : {elapsed}s")
    print(f"  Completed : {end_time.strftime('%Y-%m-%d %H:%M:%S')}\n")

    for host, ports_open in results.items():
        tag = f"({len(ports_open)} open)" if ports_open else "(no open ports)"
        print(f"  {C.GREEN}{host}{C.RESET}  {tag}")

    print(f"\n{C.DIM}Results saved to: {outfile}{C.RESET}\n")

    sys.stdout = tee.terminal
    tee.close()


if __name__ == "__main__":
    main()