diff --git a/JNC-Scan_Launcher.sh b/JNC-Scan_Launcher.sh new file mode 100644 index 0000000..ed8f97b --- /dev/null +++ b/JNC-Scan_Launcher.sh @@ -0,0 +1,28 @@ +#!/bin/bash + +# Obtener el directorio donde está el script +DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )" +cd "$DIR" + +# Comprobar si se está ejecutando en una terminal +if [ -t 0 ]; then + # Estamos en una terminal, pedir sudo directamente + echo "Iniciando JNC-Scan..." + sudo ./venv/bin/python3 main.py +else + # No estamos en una terminal (doble clic), intentar abrir una + if command -v gnome-terminal &> /dev/null; then + gnome-terminal -- bash -c "sudo $DIR/venv/bin/python3 main.py; exec bash" + elif command -v x-terminal-emulator &> /dev/null; then + x-terminal-emulator -e "sudo $DIR/venv/bin/python3 main.py" + elif command -v xterm &> /dev/null; then + xterm -e "sudo $DIR/venv/bin/python3 main.py" + else + # Fallback gráfico si no hay terminal detectada (usando pkexec si existe) + if command -v pkexec &> /dev/null; then + pkexec $DIR/venv/bin/python3 main.py + else + notify-send "Error" "No se encontró una terminal compatible para pedir contraseña de root." + fi + fi +fi diff --git a/main.py b/main.py new file mode 100644 index 0000000..9e4e604 --- /dev/null +++ b/main.py @@ -0,0 +1,28 @@ +import sys +import os +from PyQt6.QtWidgets import QApplication +from src.gui.main_window import MainWindow +from src.gui.styles import apply_styles + +def main(): + # Check for root privileges + if os.geteuid() != 0: + print("This application requires root privileges to manage network interfaces.") + print("Please run with sudo.") + # In a real GUI app we might want to show a dialog before exiting, + # but for now we'll just print to stdout. + # We can also attempt to restart with sudo, but that's complex for now. + sys.exit(1) + + app = QApplication(sys.argv) + + # Apply premium styles + apply_styles(app) + + window = MainWindow() + window.show() + + sys.exit(app.exec()) + +if __name__ == "__main__": + main() diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..edcf0f6 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,6 @@ +PyQt6 +pyroute2 +scapy +python-nmap +netifaces +psutil diff --git a/src/__init__.py b/src/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/__pycache__/__init__.cpython-312.pyc b/src/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..83d69a7 Binary files /dev/null and b/src/__pycache__/__init__.cpython-312.pyc differ diff --git a/src/core/__pycache__/diagnostics.cpython-312.pyc b/src/core/__pycache__/diagnostics.cpython-312.pyc new file mode 100644 index 0000000..5704771 Binary files /dev/null and b/src/core/__pycache__/diagnostics.cpython-312.pyc differ diff --git a/src/core/__pycache__/network_manager.cpython-312.pyc b/src/core/__pycache__/network_manager.cpython-312.pyc new file mode 100644 index 0000000..13cf926 Binary files /dev/null and b/src/core/__pycache__/network_manager.cpython-312.pyc differ diff --git a/src/core/__pycache__/scanner.cpython-312.pyc b/src/core/__pycache__/scanner.cpython-312.pyc new file mode 100644 index 0000000..ed9f817 Binary files /dev/null and b/src/core/__pycache__/scanner.cpython-312.pyc differ diff --git a/src/core/__pycache__/sniffer.cpython-312.pyc b/src/core/__pycache__/sniffer.cpython-312.pyc new file mode 100644 index 0000000..7159626 Binary files /dev/null and b/src/core/__pycache__/sniffer.cpython-312.pyc differ diff --git a/src/core/__pycache__/vlan_scanner.cpython-312.pyc b/src/core/__pycache__/vlan_scanner.cpython-312.pyc new file mode 100644 index 0000000..7c7311b Binary files /dev/null and b/src/core/__pycache__/vlan_scanner.cpython-312.pyc differ diff --git a/src/core/diagnostics.py b/src/core/diagnostics.py new file mode 100644 index 0000000..ff97676 --- /dev/null +++ b/src/core/diagnostics.py @@ -0,0 +1,58 @@ +import subprocess +import platform + +class DiagnosticsManager: + def __init__(self): + pass + + def run_ping(self, host, count=4): + """ + Executes the ping command. + Returns a generator that yields output lines. + """ + param = '-n' if platform.system().lower() == 'windows' else '-c' + command = ['ping', param, str(count), host] + + return self._run_command(command) + + def run_traceroute(self, host): + """ + Executes the tracepath command (as replacement for traceroute). + Returns a generator that yields output lines. + """ + # Using tracepath as it's more commonly available without root on modern Linux + command = ['tracepath', host] + + return self._run_command(command) + + def run_dns_lookup(self, host, query_type='A'): + """ + Executes the dig command for DNS lookup. + Returns a generator that yields output lines. + """ + command = ['dig', host, query_type] + + return self._run_command(command) + + def _run_command(self, command): + """ + Helper to run a command and yield its output line by line. + """ + try: + process = subprocess.Popen( + command, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + bufsize=1, + universal_newlines=True + ) + + for line in process.stdout: + yield line.strip() + + process.stdout.close() + process.wait() + + except Exception as e: + yield f"Error executing command: {str(e)}" diff --git a/src/core/network_manager.py b/src/core/network_manager.py new file mode 100644 index 0000000..bf4f726 --- /dev/null +++ b/src/core/network_manager.py @@ -0,0 +1,374 @@ +import socket +import fcntl +import struct +import psutil +import subprocess +import json +from pyroute2 import IPRoute + +class NetworkManager: + def __init__(self): + self.ipr = IPRoute() + + def list_interfaces(self): + """ + Returns a list of dictionaries containing interface details. + """ + interfaces = [] + + # Get basic stats using psutil + stats = psutil.net_if_stats() + addrs = psutil.net_if_addrs() + + # Get more detailed info using pyroute2 + links = self.ipr.get_links() + + for link in links: + if_name = link.get_attr('IFLA_IFNAME') + if if_name == 'lo': + continue + + if_index = link['index'] + mac_address = link.get_attr('IFLA_ADDRESS') + operstate = link.get_attr('IFLA_OPERSTATE') + + # Get IP addresses for this interface + ip_addrs = self.ipr.get_addr(index=if_index) + ipv4 = [] + ipv6 = [] + + for addr in ip_addrs: + family = addr['family'] + ip = addr.get_attr('IFA_ADDRESS') + prefixlen = addr['prefixlen'] + + if family == socket.AF_INET: + ipv4.append(f"{ip}/{prefixlen}") + elif family == socket.AF_INET6: + ipv6.append(f"{ip}/{prefixlen}") + + # Determine type (Wired/Wireless) - heuristic + # In a real scenario, we might check /sys/class/net//wireless or use iw + is_wireless = False + try: + # Simple check if wireless extension exists + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s: + fcntl.ioctl(s.fileno(), 0x8B01, struct.pack('256s', if_name.encode()[:15])) + is_wireless = True + except OSError: + pass + + interfaces.append({ + 'name': if_name, + 'mac': mac_address, + 'status': operstate, + 'ipv4': ipv4, + 'ipv6': ipv6, + 'type': 'Wireless' if is_wireless else 'Wired', + 'is_up': operstate == 'UP', + 'config': self.get_interface_config(if_name) # Add config details + }) + + return interfaces + + def get_interface_config(self, iface_name): + """ + Uses nmcli to get the current configuration of the interface. + Returns a dict with mode (auto/manual), ip, gateway, dns. + """ + config = { + 'method': 'unknown', + 'ip': '', + 'prefix': '', + 'gateway': '', + 'dns': '' + } + + try: + # 1. Get Connection Method (from connection profile) + cmd_conn = ["nmcli", "-t", "-f", "GENERAL.CONNECTION", "device", "show", iface_name] + res_conn = subprocess.run(cmd_conn, capture_output=True, text=True) + conn_name = res_conn.stdout.strip().split(':')[1] if ':' in res_conn.stdout else res_conn.stdout.strip() + + if conn_name: + cmd_method = ["nmcli", "-t", "-f", "ipv4.method", "connection", "show", conn_name] + res_method = subprocess.run(cmd_method, capture_output=True, text=True) + config['method'] = res_method.stdout.strip().split(':')[1] if ':' in res_method.stdout else res_method.stdout.strip() + + # 2. Get Dynamic/Applied Config (from device show) + # Fields: IP4.ADDRESS[1], IP4.GATEWAY, IP4.DNS[1] + cmd_dev = ["nmcli", "-t", "-f", "IP4.ADDRESS,IP4.GATEWAY,IP4.DNS", "device", "show", iface_name] + result = subprocess.run(cmd_dev, capture_output=True, text=True) + + lines = result.stdout.strip().split('\n') + for line in lines: + if ':' not in line: continue + key, value = line.split(':', 1) + value = value.strip() + + if key.startswith("IP4.ADDRESS"): + if '/' in value: + config['ip'], config['prefix'] = value.split('/') + else: + config['ip'] = value + elif key == "IP4.GATEWAY": + config['gateway'] = value + elif key.startswith("IP4.DNS"): + # Append multiple DNS if needed, for now just take the first or comma separate + if config['dns']: + config['dns'] += ", " + value + else: + config['dns'] = value + + except Exception: + pass + + return config + + def apply_interface_config(self, iface_name, method, ip=None, prefix=None, gateway=None, dns=None): + """ + Uses nmcli to apply configuration. + method: 'auto' (DHCP) or 'manual' (Static) + """ + try: + # Get connection name + cmd = ["nmcli", "-t", "-f", "GENERAL.CONNECTION", "device", "show", iface_name] + result = subprocess.run(cmd, capture_output=True, text=True) + conn_name = result.stdout.strip().split(':')[1] if ':' in result.stdout else result.stdout.strip() + + if not conn_name: + # Try to create a new connection if none exists? + # For now, assume one exists or fail. + return False, "No NetworkManager connection found for this interface." + + cmds = ["nmcli", "connection", "modify", conn_name, "ipv4.method", method] + + if method == 'manual': + if not ip or not prefix: + return False, "IP and Prefix required for manual mode." + cmds.extend(["ipv4.addresses", f"{ip}/{prefix}"]) + if gateway: + cmds.extend(["ipv4.gateway", gateway]) + if dns: + cmds.extend(["ipv4.dns", dns]) + else: + # Clear static settings when switching to auto + cmds.extend(["ipv4.addresses", "", "ipv4.gateway", "", "ipv4.dns", ""]) + + # Apply changes + subprocess.run(cmds, check=True) + + # Bring up the connection to apply + subprocess.run(["nmcli", "connection", "up", conn_name], check=True) + + return True, f"Configuration applied to {iface_name}" + + except subprocess.CalledProcessError as e: + return False, f"Failed to apply config: {str(e)}" + except Exception as e: + return False, str(e) + + def set_ip_config(self, iface_name, ip_address, prefix_len, gateway=None): + # Legacy method wrapper + return self.apply_interface_config(iface_name, 'manual', ip_address, prefix_len, gateway) + + def set_dhcp(self, iface_name): + # Legacy method wrapper + return self.apply_interface_config(iface_name, 'auto') + + def create_vlan_bridge(self, trunk_iface, vlan_id, access_iface): + """ + Creates a bridge between a VLAN on the trunk interface and an access interface. + 1. Create VLAN interface (e.g., eth0.35) + 2. Create Bridge (e.g., br35) + 3. Add VLAN interface and Access interface to Bridge + 4. Bring everything UP + """ + vlan_iface_name = f"{trunk_iface}.{vlan_id}" + bridge_name = f"br{vlan_id}" + + try: + # 1. Create VLAN interface + trunk_idx = self.ipr.link_lookup(ifname=trunk_iface)[0] + self.ipr.link('add', ifname=vlan_iface_name, kind='vlan', + link=trunk_idx, vlan_id=int(vlan_id)) + + # 2. Create Bridge + self.ipr.link('add', ifname=bridge_name, kind='bridge') + + # 3. Add ports to bridge + vlan_idx = self.ipr.link_lookup(ifname=vlan_iface_name)[0] + access_idx = self.ipr.link_lookup(ifname=access_iface)[0] + bridge_idx = self.ipr.link_lookup(ifname=bridge_name)[0] + + self.ipr.link('set', index=vlan_idx, master=bridge_idx) + self.ipr.link('set', index=access_idx, master=bridge_idx) + + # 4. Bring up interfaces + self.ipr.link('set', index=vlan_idx, state='up') + self.ipr.link('set', index=access_idx, state='up') + self.ipr.link('set', index=bridge_idx, state='up') + + return True, f"Bridge {bridge_name} created successfully." + + except Exception as e: + return False, str(e) + + def create_simple_bridge(self, iface1, iface2): + """ + Creates a simple bridge between two interfaces (Passthrough). + """ + bridge_name = "br_pass" + try: + # Check if bridge exists + if self.ipr.link_lookup(ifname=bridge_name): + return False, f"Bridge {bridge_name} already exists." + + # Create bridge + self.ipr.link('add', ifname=bridge_name, kind='bridge') + br_idx = self.ipr.link_lookup(ifname=bridge_name)[0] + + # Get indices + idx1 = self.ipr.link_lookup(ifname=iface1)[0] + idx2 = self.ipr.link_lookup(ifname=iface2)[0] + + # Helper to prepare interface + def prepare_interface(iface_name): + # Check if wireless + is_wireless = False + try: + with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s: + fcntl.ioctl(s.fileno(), 0x8B01, struct.pack('256s', iface_name.encode()[:15])) + is_wireless = True + except OSError: + pass + + if is_wireless: + # Try to enable 4addr mode (WDS) for bridging + try: + subprocess.run(["iw", "dev", iface_name, "set", "4addr", "on"], check=True, capture_output=True) + except subprocess.CalledProcessError: + # If 4addr fails, we proceed but warn/expect failure + pass + except FileNotFoundError: + # iw command not installed + pass + + prepare_interface(iface1) + prepare_interface(iface2) + + # Add ports to bridge + try: + self.ipr.link('set', index=idx1, master=br_idx) + self.ipr.link('set', index=idx2, master=br_idx) + except Exception as e: + if "95" in str(e) or "Operation not supported" in str(e): + # Cleanup + self.ipr.link('del', index=br_idx) + return False, (f"Bridging failed (Error 95). Wi-Fi interfaces in Client mode cannot be bridged " + f"unless they support '4addr' (WDS) mode. Your card/driver may not support this.") + raise e + + # Bring everything up + self.ipr.link('set', index=br_idx, state='up') + self.ipr.link('set', index=idx1, state='up') + self.ipr.link('set', index=idx2, state='up') + + return True, f"Bridge {bridge_name} created between {iface1} and {iface2}" + + except Exception as e: + # Cleanup if failed + try: + br_idx = self.ipr.link_lookup(ifname=bridge_name) + if br_idx: + self.ipr.link('del', index=br_idx[0]) + except: + pass + return False, str(e) + + def start_dhcp_server(self, interface, ip, range_start, range_end): + """ + Starts a dnsmasq DHCP server on the specified interface. + """ + try: + # 1. Configure IP on the interface + self.apply_interface_config(interface, 'manual', ip, '24') # Assuming /24 for simplicity + + # 2. Kill existing dnsmasq for this interface (simple cleanup) + subprocess.run(["pkill", "-f", f"dnsmasq.*{interface}"], check=False) + + # 3. Start dnsmasq + # --interface=br_pass --dhcp-range=192.168.x.x,192.168.x.y,12h --bind-interfaces + cmd = [ + "dnsmasq", + "--no-daemon", + f"--interface={interface}", + f"--dhcp-range={range_start},{range_end},12h", + f"--listen-address={ip}", + "--bind-interfaces" + ] + + # Run in background + subprocess.Popen(cmd) + + return True, f"DHCP Server started on {interface}" + + except Exception as e: + return False, str(e) + + def delete_bridge(self, bridge_name): + try: + # Cleanup DHCP if running + subprocess.run(["pkill", "-f", f"dnsmasq.*{bridge_name}"], check=False) + + idx = self.ipr.link_lookup(ifname=bridge_name) + if not idx: + return False, "Bridge not found" + + self.ipr.link('del', index=idx[0]) + return True, "Bridge deleted" + except Exception as e: + return False, str(e) + + def setup_wifi_ap(self, wifi_iface, ssid, password, bridge_name): + """ + Sets up a Wi-Fi Access Point bridged to the specified bridge. + Uses hostapd. + """ + import subprocess + import os + + # 1. Create hostapd config + hostapd_conf = f""" +interface={wifi_iface} +bridge={bridge_name} +driver=nl80211 +ssid={ssid} +hw_mode=g +channel=7 +wmm_enabled=0 +macaddr_acl=0 +auth_algs=1 +ignore_broadcast_ssid=0 +wpa=2 +wpa_passphrase={password} +wpa_key_mgmt=WPA-PSK +wpa_pairwise=TKIP +rsn_pairwise=CCMP +""" + conf_path = f"/tmp/hostapd_{wifi_iface}.conf" + with open(conf_path, "w") as f: + f.write(hostapd_conf) + + # 2. Kill existing hostapd on this interface + subprocess.run(["pkill", "-f", f"hostapd_{wifi_iface}.conf"], check=False) + + # 3. Start hostapd + try: + subprocess.Popen(["hostapd", conf_path]) + return True, "Wi-Fi AP started." + except Exception as e: + return False, str(e) + + def close(self): + self.ipr.close() diff --git a/src/core/scanner.py b/src/core/scanner.py new file mode 100644 index 0000000..a5abc16 --- /dev/null +++ b/src/core/scanner.py @@ -0,0 +1,128 @@ +import nmap +import subprocess +import shlex +import tempfile +import os +from PyQt6.QtCore import QThread, pyqtSignal + +import re + +class ScannerThread(QThread): + scan_finished = pyqtSignal(dict) + scan_error = pyqtSignal(str) + scan_progress = pyqtSignal(str) + port_found = pyqtSignal(int, str, str, str) # port, proto, state, service + + def __init__(self, target, arguments): + super().__init__() + self.target = target + self.arguments = arguments + self.process = None + self.temp_xml = None + + def run(self): + try: + # Create temp file for XML output + fd, self.temp_xml = tempfile.mkstemp(suffix='.xml') + os.close(fd) + + # Construct command: nmap -oX -v + # -v is important to get progress output on stdout + cmd = ["nmap", "-oX", self.temp_xml, "-v"] + shlex.split(self.arguments) + [self.target] + + self.scan_progress.emit(f"Executing: {' '.join(cmd)}\n") + + self.process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1) + + # Regex for discovered ports: "Discovered open port 80/tcp on 192.168.1.1" + port_regex = re.compile(r"Discovered open port (\d+)/(\w+) on") + + # Read stdout line by line for progress + while True: + line = self.process.stdout.readline() + if not line and self.process.poll() is not None: + break + if line: + self.scan_progress.emit(line.strip()) + # Check for discovered port + match = port_regex.search(line) + if match: + port = int(match.group(1)) + proto = match.group(2).upper() + self.port_found.emit(port, proto, "open", "") + + stdout, stderr = self.process.communicate() + + if self.process.returncode == 0: + # Parse XML output from file + nm = nmap.PortScanner() + try: + with open(self.temp_xml, 'r') as f: + xml_content = f.read() + + if not xml_content: + self.scan_error.emit("Scan produced no output.") + return + + result = nm.analyse_nmap_xml_scan(xml_content) + if result is None: + result = nm.scan_result + self.scan_finished.emit(result) + except Exception as e: + self.scan_error.emit(f"Error parsing scan results: {str(e)}") + else: + # If killed or failed + if self.process.returncode == -9: # SIGKILL + self.scan_progress.emit("\nScan stopped by user.") + self.scan_error.emit("Scan stopped by user.") + else: + self.scan_error.emit(f"Scan failed: {stderr}") + + except Exception as e: + self.scan_error.emit(str(e)) + finally: + # Cleanup temp file + if self.temp_xml and os.path.exists(self.temp_xml): + try: + os.remove(self.temp_xml) + except: + pass + + def stop(self): + if self.process: + self.process.kill() + self.wait() + +class NetworkScanner: + def __init__(self): + pass + + def scan(self, target, ports=None, scan_type="TCP", skip_discovery=False): + """ + Initiates a scan. Returns the thread object. + target: IP or range (e.g., '192.168.1.1' or '192.168.1.0/24') + ports: String of ports (e.g., '80,443' or '1-1000') + scan_type: 'TCP' or 'UDP' + skip_discovery: If True, adds -Pn (treat all as online). If False, performs host discovery (ARP/Ping). + """ + # -T4: Aggressive timing + # --stats-every 5s: Print progress line every 5 seconds + # --min-rate 1000: Send packets at least 1000 per second (Speed up) + arguments = "-T4 --stats-every 5s --min-rate 1000" + + if skip_discovery: + arguments += " -Pn" + + if scan_type == "UDP": + arguments += " -sU" + else: + arguments += " -sT" # Connect scan (safer for non-root) + + if ports: + arguments += f" -p {ports}" + else: + arguments += " -F" # Fast scan if no ports specified + + thread = ScannerThread(target, arguments) + thread.start() + return thread diff --git a/src/core/sniffer.py b/src/core/sniffer.py new file mode 100644 index 0000000..5d33643 --- /dev/null +++ b/src/core/sniffer.py @@ -0,0 +1,138 @@ +from scapy.all import sniff, IP, TCP, UDP, ARP, ICMP +from PyQt6.QtCore import QThread, pyqtSignal +import datetime + +class SnifferThread(QThread): + packet_received = pyqtSignal(dict) + error_occurred = pyqtSignal(str) + + def __init__(self, interface, filter_str=""): + super().__init__() + self.interface = interface + self.filter_str = filter_str + self.running = True + + def run(self): + try: + # Check if interface is up + if not self.is_interface_up(self.interface): + raise Exception(f"Interface {self.interface} is DOWN. Please enable it first.") + + sniff(iface=self.interface, + prn=self.process_packet, + filter=self.filter_str, + store=0, + stop_filter=lambda x: not self.running) + except Exception as e: + self.error_occurred.emit(str(e)) + + def is_interface_up(self, interface): + try: + import psutil + stats = psutil.net_if_stats() + if interface in stats: + return stats[interface].isup + return False + except: + return True # Assume up if check fails to avoid blocking + + def process_packet(self, packet): + if not self.running: + return + + try: + pkt_info = { + 'time': datetime.datetime.now().strftime("%H:%M:%S.%f")[:-3], + 'src': 'Unknown', + 'dst': 'Unknown', + 'proto': 'Unknown', + 'info': packet.summary() + } + + if IP in packet: + pkt_info['src'] = packet[IP].src + pkt_info['dst'] = packet[IP].dst + pkt_info['proto'] = packet[IP].proto + + # Printer Ports + printer_ports = [631, 515, 9100] + + if TCP in packet: + pkt_info['proto'] = 'TCP' + flags = packet[TCP].flags + flag_str = str(flags) + + sport = packet[TCP].sport + dport = packet[TCP].dport + + info = f"{sport} -> {dport} [{flag_str}]" + + # Check for Printer Traffic + if sport in printer_ports or dport in printer_ports: + info = f"[PRINTER] {info}" + + # Check for potential issues (simple heuristics) + if 'R' in flag_str: # RST + info += " [RESET]" + pkt_info['status'] = 'Error' + elif 'S' in flag_str and 'A' not in flag_str: # SYN only + info += " [SYN]" + pkt_info['status'] = 'Connecting' + else: + pkt_info['status'] = 'OK' + + pkt_info['info'] = info + + elif UDP in packet: + pkt_info['proto'] = 'UDP' + sport = packet[UDP].sport + dport = packet[UDP].dport + info = f"{sport} -> {dport}" + + if sport in printer_ports or dport in printer_ports: + info = f"[PRINTER] {info}" + + pkt_info['info'] = info + pkt_info['status'] = 'OK' + + elif ICMP in packet: + pkt_info['proto'] = 'ICMP' + icmp_type = packet[ICMP].type + icmp_code = packet[ICMP].code + + if icmp_type == 3: # Destination Unreachable + pkt_info['info'] = f"Dest Unreachable (Code: {icmp_code})" + pkt_info['status'] = 'Error' + else: + pkt_info['info'] = f"Type: {icmp_type} Code: {icmp_code}" + pkt_info['status'] = 'OK' + + elif ARP in packet: + pkt_info['proto'] = 'ARP' + pkt_info['src'] = packet[ARP].psrc + pkt_info['dst'] = packet[ARP].pdst + pkt_info['info'] = f"Who has {packet[ARP].pdst}? Tell {packet[ARP].psrc}" + pkt_info['status'] = 'OK' + + self.packet_received.emit(pkt_info) + + except Exception: + pass + + def stop(self): + self.running = False + +class NetworkSniffer: + def __init__(self): + self.thread = None + + def start_sniffing(self, interface, filter_str=""): + self.thread = SnifferThread(interface, filter_str) + self.thread.start() + return self.thread + + def stop_sniffing(self): + if self.thread: + self.thread.stop() + self.thread.wait() + self.thread = None diff --git a/src/core/vlan_scanner.py b/src/core/vlan_scanner.py new file mode 100644 index 0000000..380dce4 --- /dev/null +++ b/src/core/vlan_scanner.py @@ -0,0 +1,98 @@ +from PyQt6.QtCore import QThread, pyqtSignal +import subprocess +import re +import time + +class VLANScannerThread(QThread): + vlan_detected = pyqtSignal(int, int) # vlan_id, count + untagged_detected = pyqtSignal(int) # count + scan_finished = pyqtSignal() + error_occurred = pyqtSignal(str) + + def __init__(self, interface, timeout=10): + super().__init__() + self.interface = interface + self.timeout = timeout + self.running = True + self.process = None + self.detected_vlans = {} # vlan_id -> count + self.untagged_count = 0 + + def run(self): + try: + # Use tcpdump to capture ALL packets to detect untagged vs tagged + # -i: interface + # -e: print link-level header (to see vlan) + # -n: don't resolve names + # -l: buffered output + # not arp and not stp: filter out some noise if desired, but let's keep it simple + cmd = ["tcpdump", "-i", self.interface, "-e", "-n", "-l"] + + self.process = subprocess.Popen( + cmd, + stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, + text=True, + bufsize=1 + ) + + start_time = time.time() + + while self.running: + # Check timeout + if time.time() - start_time > self.timeout: + break + + # Non-blocking read line + line = self.process.stdout.readline() + if not line and self.process.poll() is not None: + break + + if line: + self.parse_line(line) + + except Exception as e: + self.error_occurred.emit(str(e)) + finally: + if self.process: + self.process.terminate() + try: + self.process.wait(timeout=1) + except: + self.process.kill() + self.scan_finished.emit() + + def parse_line(self, line): + # Example Tagged: ... ethertype 802.1Q (0x8100), length 100: vlan 35, p 0, ethertype IPv4 ... + # Example Untagged: ... ethertype IPv4 (0x0800), length 100: ... + + # Regex to find "vlan " + match = re.search(r"vlan (\d+)", line) + if match: + vlan_id = int(match.group(1)) + if vlan_id not in self.detected_vlans: + self.detected_vlans[vlan_id] = 0 + + self.detected_vlans[vlan_id] += 1 + self.vlan_detected.emit(vlan_id, self.detected_vlans[vlan_id]) + else: + # It's an untagged packet (or at least not 802.1Q) + # Filter out empty lines or weird output + if "ethertype" in line: + self.untagged_count += 1 + self.untagged_detected.emit(self.untagged_count) + + def stop(self): + self.running = False + if self.process: + self.process.kill() # Force kill to break readline() + self.wait() + +class VLANScanner: + def __init__(self): + pass + + def scan(self, interface, timeout=10): + thread = VLANScannerThread(interface, timeout) + thread.start() + return thread diff --git a/src/gui/__pycache__/main_window.cpython-312.pyc b/src/gui/__pycache__/main_window.cpython-312.pyc new file mode 100644 index 0000000..941de87 Binary files /dev/null and b/src/gui/__pycache__/main_window.cpython-312.pyc differ diff --git a/src/gui/__pycache__/styles.cpython-312.pyc b/src/gui/__pycache__/styles.cpython-312.pyc new file mode 100644 index 0000000..b99b5d9 Binary files /dev/null and b/src/gui/__pycache__/styles.cpython-312.pyc differ diff --git a/src/gui/main_window.py b/src/gui/main_window.py new file mode 100644 index 0000000..01576c8 --- /dev/null +++ b/src/gui/main_window.py @@ -0,0 +1,129 @@ +from PyQt6.QtWidgets import (QMainWindow, QWidget, QVBoxLayout, QTabWidget, + QLabel, QHBoxLayout, QStatusBar) +from PyQt6.QtCore import Qt +from src.core.network_manager import NetworkManager +from src.gui.widgets.interface_list import InterfaceListWidget +from src.gui.widgets.ip_config import IPConfigWidget +from src.gui.widgets.vlan_bridge import VLANBridgeWidget +from src.gui.widgets.scanner_view import ScannerWidget +from src.gui.widgets.sniffer_view import SnifferWidget +from src.gui.widgets.diagnostics_view import DiagnosticsWidget +from src.gui.widgets.port_scanner import PortScannerWidget +from src.utils.report_generator import ReportGenerator +from PyQt6.QtWidgets import QFileDialog + +class MainWindow(QMainWindow): + def __init__(self): + super().__init__() + self.setWindowTitle("JNC-NetTools // ADA version") + self.setMinimumSize(1000, 700) # Allow resizing, set minimum size + + self.nm = NetworkManager() + + # Central Widget + central_widget = QWidget() + self.setCentralWidget(central_widget) + main_layout = QHBoxLayout(central_widget) + + # Sidebar (Interface List) + self.interface_list = InterfaceListWidget(self.nm) + self.interface_list.setFixedWidth(300) + self.interface_list.interface_selected.connect(self.on_interface_selected) + main_layout.addWidget(self.interface_list) + + # Main Content Area (Tabs) + self.tabs = QTabWidget() + main_layout.addWidget(self.tabs) + + # IP Configuration Tab + self.ip_config_widget = IPConfigWidget(self.nm) + self.tabs.addTab(self.ip_config_widget, "IP Configuration") + + # VLAN & Bridge Tab + self.vlan_bridge_widget = VLANBridgeWidget(self.nm) + self.tabs.addTab(self.vlan_bridge_widget, "VLAN / Bridge") + + # Scanner Tab (Network Discovery) + self.scanner_widget = ScannerWidget() + self.tabs.addTab(self.scanner_widget, "Net Scanner") + + # Port Scanner Tab (Dedicated) + self.port_scanner_widget = PortScannerWidget() + self.tabs.addTab(self.port_scanner_widget, "Port Scanner") + + # Sniffer Tab + self.sniffer_widget = SnifferWidget(self.nm) + self.tabs.addTab(self.sniffer_widget, "Sniffer") + + # Diagnostics Tab + self.diagnostics_widget = DiagnosticsWidget() + self.tabs.addTab(self.diagnostics_widget, "Diagnostics") + + # Report Button (Top Right or Bottom) + # Let's add a toolbar for global actions + toolbar = self.addToolBar("Main") + toolbar.addAction("Generate Report", self.generate_report) + + # Status Bar + self.status_bar = QStatusBar() + self.setStatusBar(self.status_bar) + self.status_bar.showMessage("Ready") + + def on_interface_selected(self, iface_data): + self.status_bar.showMessage(f"Selected Interface: {iface_data['name']}") + self.ip_config_widget.set_interface(iface_data) + + def closeEvent(self, event): + self.nm.close() + self.scanner_widget.close() + self.port_scanner_widget.close() + self.vlan_bridge_widget.close() + event.accept() + + def generate_report(self): + # Gather data + interfaces = self.nm.list_interfaces() + + # Get scan results from scanner widget + scan_results = [] + for row in range(self.scanner_widget.table.rowCount()): + scan_results.append({ + 'ip': self.scanner_widget.table.item(row, 0).text(), + 'mac': self.scanner_widget.table.item(row, 1).text(), + 'vendor': self.scanner_widget.table.item(row, 2).text(), + 'status': self.scanner_widget.table.item(row, 3).text() + }) + + # Get sniffer results (All Traffic) + sniff_all = [] + for row in range(self.sniffer_widget.table_all.rowCount()): + sniff_all.append({ + 'time': self.sniffer_widget.table_all.item(row, 0).text(), + 'src': self.sniffer_widget.table_all.item(row, 1).text(), + 'dst': self.sniffer_widget.table_all.item(row, 2).text(), + 'proto': self.sniffer_widget.table_all.item(row, 3).text(), + 'info': self.sniffer_widget.table_all.item(row, 4).text(), + 'status': self.sniffer_widget.table_all.item(row, 5).text() + }) + + # Get sniffer results (Failed/Anomalies) + sniff_failed = [] + for row in range(self.sniffer_widget.table_failed.rowCount()): + sniff_failed.append({ + 'time': self.sniffer_widget.table_failed.item(row, 0).text(), + 'src': self.sniffer_widget.table_failed.item(row, 1).text(), + 'dst': self.sniffer_widget.table_failed.item(row, 2).text(), + 'proto': self.sniffer_widget.table_failed.item(row, 3).text(), + 'info': self.sniffer_widget.table_failed.item(row, 4).text(), + 'status': self.sniffer_widget.table_failed.item(row, 5).text() + }) + + generator = ReportGenerator() + filename, _ = QFileDialog.getSaveFileName(self, "Save Report", "report.html", "HTML Files (*.html)") + + if filename: + success, msg = generator.generate_html_report(interfaces, scan_results, sniff_all, sniff_failed, filename) + if success: + self.status_bar.showMessage(f"Report saved to {msg}") + else: + self.status_bar.showMessage(f"Error saving report: {msg}") diff --git a/src/gui/styles.py b/src/gui/styles.py new file mode 100644 index 0000000..e44c519 --- /dev/null +++ b/src/gui/styles.py @@ -0,0 +1,165 @@ +from PyQt6.QtWidgets import QApplication +from PyQt6.QtGui import QPalette, QColor + +def apply_styles(app: QApplication): + """Applies a premium dark theme to the application.""" + app.setStyle("Fusion") + + palette = QPalette() + + # Colors + dark_bg = QColor(30, 30, 30) + darker_bg = QColor(20, 20, 20) + text_color = QColor(230, 230, 230) + accent_color = QColor(0, 120, 215) # A nice blue + accent_hover = QColor(0, 140, 240) + disabled_text = QColor(127, 127, 127) + + palette.setColor(QPalette.ColorRole.Window, dark_bg) + palette.setColor(QPalette.ColorRole.WindowText, text_color) + palette.setColor(QPalette.ColorRole.Base, darker_bg) + palette.setColor(QPalette.ColorRole.AlternateBase, dark_bg) + palette.setColor(QPalette.ColorRole.ToolTipBase, text_color) + palette.setColor(QPalette.ColorRole.ToolTipText, text_color) + palette.setColor(QPalette.ColorRole.Text, text_color) + palette.setColor(QPalette.ColorRole.Button, dark_bg) + palette.setColor(QPalette.ColorRole.ButtonText, text_color) + palette.setColor(QPalette.ColorRole.BrightText, Qt.GlobalColor.red) + palette.setColor(QPalette.ColorRole.Link, accent_color) + palette.setColor(QPalette.ColorRole.Highlight, accent_color) + palette.setColor(QPalette.ColorRole.HighlightedText, Qt.GlobalColor.black) + + app.setPalette(palette) + + # Stylesheet for specific widgets to add that "premium" feel + # Stylesheet for specific widgets to add that "premium" feel + app.setStyleSheet(""" + QMainWindow { + background-color: #181818; + } + QWidget { + font-family: 'Segoe UI', 'Roboto', sans-serif; + font-size: 13px; + color: #e0e0e0; + } + QTabWidget::pane { + border: 1px solid #333333; + background: #252526; + border-radius: 4px; + top: -1px; + } + QTabBar::tab { + background: #1e1e1e; + color: #999999; + padding: 10px 20px; + border-top-left-radius: 4px; + border-top-right-radius: 4px; + margin-right: 2px; + font-weight: 600; + border: 1px solid transparent; + } + QTabBar::tab:selected { + background: #252526; + color: #ffffff; + border-top: 2px solid #007acc; + border-bottom: 1px solid #252526; + } + QTabBar::tab:hover { + background: #2d2d2d; + color: #ffffff; + } + QPushButton { + background-color: #007acc; + color: white; + border: none; + padding: 8px 16px; + border-radius: 4px; + font-weight: 600; + } + QPushButton:hover { + background-color: #0062a3; + } + QPushButton:pressed { + background-color: #004c80; + } + QPushButton:disabled { + background-color: #333333; + color: #666666; + } + QLineEdit, QComboBox, QSpinBox { + background-color: #333333; + border: 1px solid #444444; + color: #f0f0f0; + padding: 6px; + border-radius: 3px; + } + QLineEdit:focus, QComboBox:focus, QSpinBox:focus { + border: 1px solid #007acc; + background-color: #3c3c3c; + } + QTableWidget { + background-color: #1e1e1e; + border: 1px solid #333333; + border-radius: 4px; + gridline-color: #333333; + selection-background-color: #264f78; + selection-color: white; + alternate-background-color: #252526; + } + QHeaderView::section { + background-color: #252526; + color: #cccccc; + padding: 6px; + border: none; + border-right: 1px solid #333333; + border-bottom: 1px solid #333333; + font-weight: 600; + } + QListWidget { + background-color: #1e1e1e; + border: 1px solid #333333; + border-radius: 4px; + } + QListWidget::item { + padding: 8px; + border-bottom: 1px solid #252526; + } + QListWidget::item:selected { + background-color: #37373d; + color: white; + border-left: 2px solid #007acc; + } + QGroupBox { + border: 1px solid #444444; + border-radius: 6px; + margin-top: 20px; + padding-top: 15px; + font-weight: 600; + } + QGroupBox::title { + subcontrol-origin: margin; + subcontrol-position: top left; + padding: 0 5px; + color: #007acc; + left: 10px; + } + QStatusBar { + background-color: #007acc; + color: white; + } + QProgressBar { + border: 1px solid #444444; + border-radius: 4px; + text-align: center; + background-color: #333333; + } + QProgressBar::chunk { + background-color: #007acc; + width: 10px; + } + QSplitter::handle { + background-color: #333333; + } + """) + +from PyQt6.QtCore import Qt diff --git a/src/gui/widgets/__pycache__/diagnostics_view.cpython-312.pyc b/src/gui/widgets/__pycache__/diagnostics_view.cpython-312.pyc new file mode 100644 index 0000000..f494d62 Binary files /dev/null and b/src/gui/widgets/__pycache__/diagnostics_view.cpython-312.pyc differ diff --git a/src/gui/widgets/__pycache__/interface_list.cpython-312.pyc b/src/gui/widgets/__pycache__/interface_list.cpython-312.pyc new file mode 100644 index 0000000..7e3cc37 Binary files /dev/null and b/src/gui/widgets/__pycache__/interface_list.cpython-312.pyc differ diff --git a/src/gui/widgets/__pycache__/ip_config.cpython-312.pyc b/src/gui/widgets/__pycache__/ip_config.cpython-312.pyc new file mode 100644 index 0000000..da359fb Binary files /dev/null and b/src/gui/widgets/__pycache__/ip_config.cpython-312.pyc differ diff --git a/src/gui/widgets/__pycache__/port_scanner.cpython-312.pyc b/src/gui/widgets/__pycache__/port_scanner.cpython-312.pyc new file mode 100644 index 0000000..acdaa04 Binary files /dev/null and b/src/gui/widgets/__pycache__/port_scanner.cpython-312.pyc differ diff --git a/src/gui/widgets/__pycache__/scanner_view.cpython-312.pyc b/src/gui/widgets/__pycache__/scanner_view.cpython-312.pyc new file mode 100644 index 0000000..3bf19de Binary files /dev/null and b/src/gui/widgets/__pycache__/scanner_view.cpython-312.pyc differ diff --git a/src/gui/widgets/__pycache__/sniffer_view.cpython-312.pyc b/src/gui/widgets/__pycache__/sniffer_view.cpython-312.pyc new file mode 100644 index 0000000..7eee041 Binary files /dev/null and b/src/gui/widgets/__pycache__/sniffer_view.cpython-312.pyc differ diff --git a/src/gui/widgets/__pycache__/vlan_bridge.cpython-312.pyc b/src/gui/widgets/__pycache__/vlan_bridge.cpython-312.pyc new file mode 100644 index 0000000..320f527 Binary files /dev/null and b/src/gui/widgets/__pycache__/vlan_bridge.cpython-312.pyc differ diff --git a/src/gui/widgets/diagnostics_view.py b/src/gui/widgets/diagnostics_view.py new file mode 100644 index 0000000..4e6c3bd --- /dev/null +++ b/src/gui/widgets/diagnostics_view.py @@ -0,0 +1,182 @@ +from PyQt6.QtWidgets import (QWidget, QVBoxLayout, QHBoxLayout, QLabel, + QLineEdit, QPushButton, QTextEdit, QComboBox, + QTabWidget, QSpinBox, QGroupBox) +from PyQt6.QtCore import QThread, pyqtSignal, Qt +from src.core.diagnostics import DiagnosticsManager + +class DiagnosticsWorker(QThread): + output_signal = pyqtSignal(str) + finished_signal = pyqtSignal() + + def __init__(self, func, *args): + super().__init__() + self.func = func + self.args = args + self.is_running = True + + def run(self): + try: + for line in self.func(*self.args): + if not self.is_running: + break + self.output_signal.emit(line) + except Exception as e: + self.output_signal.emit(f"Error: {str(e)}") + finally: + self.finished_signal.emit() + + def stop(self): + self.is_running = False + +class DiagnosticsWidget(QWidget): + def __init__(self): + super().__init__() + self.dm = DiagnosticsManager() + self.worker = None + self.init_ui() + + def init_ui(self): + layout = QVBoxLayout(self) + + # Tabs for different tools + self.tabs = QTabWidget() + layout.addWidget(self.tabs) + + # Ping Tab + self.ping_tab = QWidget() + self.init_ping_tab() + self.tabs.addTab(self.ping_tab, "Ping") + + # Traceroute Tab + self.traceroute_tab = QWidget() + self.init_traceroute_tab() + self.tabs.addTab(self.traceroute_tab, "Traceroute") + + # DNS Tab + self.dns_tab = QWidget() + self.init_dns_tab() + self.tabs.addTab(self.dns_tab, "DNS Lookup") + + def init_ping_tab(self): + layout = QVBoxLayout(self.ping_tab) + + # Controls + controls_layout = QHBoxLayout() + + controls_layout.addWidget(QLabel("Host:")) + self.ping_host_input = QLineEdit() + self.ping_host_input.setPlaceholderText("e.g., google.com or 8.8.8.8") + controls_layout.addWidget(self.ping_host_input) + + controls_layout.addWidget(QLabel("Count:")) + self.ping_count_input = QSpinBox() + self.ping_count_input.setRange(1, 100) + self.ping_count_input.setValue(4) + controls_layout.addWidget(self.ping_count_input) + + self.ping_btn = QPushButton("Ping") + self.ping_btn.clicked.connect(self.start_ping) + controls_layout.addWidget(self.ping_btn) + + layout.addLayout(controls_layout) + + # Output + self.ping_output = QTextEdit() + self.ping_output.setReadOnly(True) + layout.addWidget(self.ping_output) + + def init_traceroute_tab(self): + layout = QVBoxLayout(self.traceroute_tab) + + # Controls + controls_layout = QHBoxLayout() + + controls_layout.addWidget(QLabel("Host:")) + self.trace_host_input = QLineEdit() + self.trace_host_input.setPlaceholderText("e.g., google.com") + controls_layout.addWidget(self.trace_host_input) + + self.trace_btn = QPushButton("Trace") + self.trace_btn.clicked.connect(self.start_traceroute) + controls_layout.addWidget(self.trace_btn) + + layout.addLayout(controls_layout) + + # Output + self.trace_output = QTextEdit() + self.trace_output.setReadOnly(True) + layout.addWidget(self.trace_output) + + def init_dns_tab(self): + layout = QVBoxLayout(self.dns_tab) + + # Controls + controls_layout = QHBoxLayout() + + controls_layout.addWidget(QLabel("Host:")) + self.dns_host_input = QLineEdit() + self.dns_host_input.setPlaceholderText("e.g., google.com") + controls_layout.addWidget(self.dns_host_input) + + controls_layout.addWidget(QLabel("Type:")) + self.dns_type_input = QComboBox() + self.dns_type_input.addItems(["A", "AAAA", "MX", "NS", "TXT", "CNAME", "SOA"]) + controls_layout.addWidget(self.dns_type_input) + + self.dns_btn = QPushButton("Lookup") + self.dns_btn.clicked.connect(self.start_dns) + controls_layout.addWidget(self.dns_btn) + + layout.addLayout(controls_layout) + + # Output + self.dns_output = QTextEdit() + self.dns_output.setReadOnly(True) + layout.addWidget(self.dns_output) + + def start_ping(self): + host = self.ping_host_input.text().strip() + if not host: + self.ping_output.append("Please enter a host.") + return + + count = self.ping_count_input.value() + self.ping_output.clear() + self.ping_output.append(f"Pinging {host}...") + self.ping_btn.setEnabled(False) + + self.worker = DiagnosticsWorker(self.dm.run_ping, host, count) + self.worker.output_signal.connect(self.ping_output.append) + self.worker.finished_signal.connect(lambda: self.ping_btn.setEnabled(True)) + self.worker.start() + + def start_traceroute(self): + host = self.trace_host_input.text().strip() + if not host: + self.trace_output.append("Please enter a host.") + return + + self.trace_output.clear() + self.trace_output.append(f"Tracing route to {host}...") + self.trace_btn.setEnabled(False) + + self.worker = DiagnosticsWorker(self.dm.run_traceroute, host) + self.worker.output_signal.connect(self.trace_output.append) + self.worker.finished_signal.connect(lambda: self.trace_btn.setEnabled(True)) + self.worker.start() + + def start_dns(self): + host = self.dns_host_input.text().strip() + if not host: + self.dns_output.append("Please enter a host.") + return + + query_type = self.dns_type_input.currentText() + self.dns_output.clear() + self.dns_output.append(f"Looking up {query_type} records for {host}...") + self.dns_btn.setEnabled(False) + + self.worker = DiagnosticsWorker(self.dm.run_dns_lookup, host, query_type) + self.worker.output_signal.connect(self.dns_output.append) + self.worker.finished_signal.connect(lambda: self.dns_btn.setEnabled(True)) + self.worker.start() diff --git a/src/gui/widgets/interface_list.py b/src/gui/widgets/interface_list.py new file mode 100644 index 0000000..1e9b2ea --- /dev/null +++ b/src/gui/widgets/interface_list.py @@ -0,0 +1,100 @@ +from PyQt6.QtWidgets import (QWidget, QVBoxLayout, QListWidget, QListWidgetItem, + QLabel, QHBoxLayout, QFrame, QPushButton, QStyle) +from PyQt6.QtCore import Qt, pyqtSignal +from PyQt6.QtGui import QIcon, QColor, QFont + +class InterfaceListWidget(QWidget): + interface_selected = pyqtSignal(dict) + + def __init__(self, network_manager): + super().__init__() + self.nm = network_manager + self.layout = QVBoxLayout(self) + + # Header + header_layout = QHBoxLayout() + title = QLabel("Network Interfaces") + title.setFont(QFont("Segoe UI", 16, QFont.Weight.Bold)) + header_layout.addWidget(title) + header_layout.addStretch() + + # Refresh Button + self.refresh_btn = QPushButton() + self.refresh_btn.setIcon(self.style().standardIcon(QStyle.StandardPixmap.SP_BrowserReload)) + self.refresh_btn.setToolTip("Refresh Interfaces") + self.refresh_btn.setFixedSize(30, 30) + self.refresh_btn.clicked.connect(self.refresh_interfaces) + header_layout.addWidget(self.refresh_btn) + + self.layout.addLayout(header_layout) + + # List + self.list_widget = QListWidget() + self.list_widget.itemClicked.connect(self.on_item_clicked) + self.layout.addWidget(self.list_widget) + + self.refresh_interfaces() + + def refresh_interfaces(self): + self.list_widget.clear() + interfaces = self.nm.list_interfaces() + + for iface in interfaces: + item = QListWidgetItem() + + # Create custom widget for the item + item_widget = QWidget() + item_layout = QVBoxLayout(item_widget) + item_layout.setContentsMargins(5, 5, 5, 5) + + # Top row: Name + Type + Status + top_row = QHBoxLayout() + name_label = QLabel(iface['name']) + name_label.setFont(QFont("Segoe UI", 12, QFont.Weight.Bold)) + + type_label = QLabel(iface['type']) + type_label.setStyleSheet("color: #aaaaaa; font-size: 10px;") + + status_color = "#4caf50" if iface['is_up'] else "#f44336" + status_label = QLabel("● " + iface['status']) + status_label.setStyleSheet(f"color: {status_color}; font-weight: bold;") + + # Config Method Badge (DHCP/Static) + config_method = iface.get('config', {}).get('method', 'unknown') + method_text = "DHCP" if config_method == 'auto' else "Static" if config_method == 'manual' else config_method.upper() + method_label = QLabel(f"[{method_text}]") + method_label.setStyleSheet("color: #e0e0e0; font-size: 10px; background-color: #3e3e3e; padding: 2px 4px; border-radius: 3px;") + + top_row.addWidget(name_label) + top_row.addWidget(type_label) + top_row.addWidget(method_label) + top_row.addStretch() + top_row.addWidget(status_label) + + # Bottom row: IP + MAC + bottom_row = QHBoxLayout() + + ip_text = iface['ipv4'][0] if iface['ipv4'] else "No IP" + ip_label = QLabel(f"IP: {ip_text}") + ip_label.setStyleSheet("color: #0078d7;") + + mac_label = QLabel(f"MAC: {iface['mac']}") + mac_label.setStyleSheet("color: #888888;") + + bottom_row.addWidget(ip_label) + bottom_row.addStretch() + bottom_row.addWidget(mac_label) + + item_layout.addLayout(top_row) + item_layout.addLayout(bottom_row) + + item.setSizeHint(item_widget.sizeHint()) + self.list_widget.addItem(item) + self.list_widget.setItemWidget(item, item_widget) + + # Store data in item + item.setData(Qt.ItemDataRole.UserRole, iface) + + def on_item_clicked(self, item): + iface_data = item.data(Qt.ItemDataRole.UserRole) + self.interface_selected.emit(iface_data) diff --git a/src/gui/widgets/ip_config.py b/src/gui/widgets/ip_config.py new file mode 100644 index 0000000..592843d --- /dev/null +++ b/src/gui/widgets/ip_config.py @@ -0,0 +1,165 @@ +from PyQt6.QtWidgets import (QWidget, QVBoxLayout, QHBoxLayout, QLabel, + QLineEdit, QPushButton, QRadioButton, QButtonGroup, + QGroupBox, QFormLayout, QMessageBox) +from PyQt6.QtCore import Qt +import ipaddress + +class IPConfigWidget(QWidget): + def __init__(self, network_manager): + super().__init__() + self.nm = network_manager + self.current_iface = None + + self.layout = QVBoxLayout(self) + + # IP Configuration Group + config_group = QGroupBox("IP Configuration") + config_layout = QVBoxLayout(config_group) + + # Mode Selection + mode_layout = QHBoxLayout() + self.dhcp_radio = QRadioButton("DHCP") + self.static_radio = QRadioButton("Static") + self.static_radio.setChecked(True) + + self.mode_group = QButtonGroup() + self.mode_group.addButton(self.dhcp_radio) + self.mode_group.addButton(self.static_radio) + + mode_layout.addWidget(self.dhcp_radio) + mode_layout.addWidget(self.static_radio) + mode_layout.addStretch() + config_layout.addLayout(mode_layout) + + # Static IP Fields + self.form_layout = QFormLayout() + self.ip_input = QLineEdit() + self.cidr_input = QLineEdit() + self.cidr_input.setPlaceholderText("24") + self.gateway_input = QLineEdit() + + self.form_layout.addRow("IP Address:", self.ip_input) + self.form_layout.addRow("CIDR Prefix:", self.cidr_input) + self.form_layout.addRow("Gateway:", self.gateway_input) + + # DNS Field + self.dns_input = QLineEdit() + self.form_layout.addRow("DNS:", self.dns_input) + + config_layout.addLayout(self.form_layout) + + # Apply Button + self.apply_btn = QPushButton("Apply Configuration") + self.apply_btn.clicked.connect(self.apply_config) + config_layout.addWidget(self.apply_btn) + + self.layout.addWidget(config_group) + + # Subnet Calculator Group + calc_group = QGroupBox("Subnet Calculator") + calc_layout = QVBoxLayout(calc_group) + + calc_form = QFormLayout() + self.calc_ip = QLineEdit() + self.calc_cidr = QLineEdit() + self.calc_cidr.setPlaceholderText("24") + self.calc_netmask = QLineEdit() + self.calc_netmask.setReadOnly(True) + self.calc_network = QLineEdit() + self.calc_network.setReadOnly(True) + self.calc_broadcast = QLineEdit() + self.calc_broadcast.setReadOnly(True) + + calc_form.addRow("IP Address:", self.calc_ip) + calc_form.addRow("CIDR:", self.calc_cidr) + calc_form.addRow("Netmask:", self.calc_netmask) + calc_form.addRow("Network:", self.calc_network) + calc_form.addRow("Broadcast:", self.calc_broadcast) + + calc_layout.addLayout(calc_form) + + # Calculate Button (or auto-calc) + self.calc_btn = QPushButton("Calculate") + self.calc_btn.clicked.connect(self.calculate_subnet) + calc_layout.addWidget(self.calc_btn) + + self.layout.addWidget(calc_group) + self.layout.addStretch() + + # Connect signals + self.dhcp_radio.toggled.connect(self.toggle_inputs) + self.static_radio.toggled.connect(self.toggle_inputs) + + def set_interface(self, iface_data): + self.current_iface = iface_data + + # Get real config from NM + config = iface_data.get('config', {}) + method = config.get('method', 'auto') + + if method == 'manual': + self.static_radio.setChecked(True) + self.ip_input.setText(config.get('ip', '')) + self.cidr_input.setText(config.get('prefix', '')) + self.gateway_input.setText(config.get('gateway', '')) + self.dns_input.setText(config.get('dns', '')) + else: + self.dhcp_radio.setChecked(True) + # Clear fields or show current lease? + # For now, let's show current IP even if DHCP + if iface_data['ipv4']: + ip, prefix = iface_data['ipv4'][0].split('/') + self.ip_input.setText(ip) + self.cidr_input.setText(prefix) + + # Update calculator + self.calc_ip.setText(self.ip_input.text()) + self.calc_cidr.setText(self.cidr_input.text()) + self.calculate_subnet() + + def toggle_inputs(self): + enabled = self.static_radio.isChecked() + self.ip_input.setEnabled(enabled) + self.cidr_input.setEnabled(enabled) + self.gateway_input.setEnabled(enabled) + self.dns_input.setEnabled(enabled) + + def calculate_subnet(self): + ip = self.calc_ip.text() + cidr = self.calc_cidr.text() + + if not ip or not cidr: + return + + try: + network = ipaddress.IPv4Interface(f"{ip}/{cidr}") + self.calc_netmask.setText(str(network.netmask)) + self.calc_network.setText(str(network.network)) + self.calc_broadcast.setText(str(network.network.broadcast_address)) + except ValueError as e: + self.calc_netmask.setText("Invalid IP/CIDR") + + def apply_config(self): + if not self.current_iface: + QMessageBox.warning(self, "Error", "No interface selected") + return + + iface_name = self.current_iface['name'] + + try: + if self.dhcp_radio.isChecked(): + success, msg = self.nm.apply_interface_config(iface_name, 'auto') + else: + ip = self.ip_input.text() + cidr = self.cidr_input.text() + gateway = self.gateway_input.text() + dns = self.dns_input.text() + success, msg = self.nm.apply_interface_config(iface_name, 'manual', ip, cidr, gateway, dns) + + if success: + QMessageBox.information(self, "Success", msg) + else: + QMessageBox.critical(self, "Error", msg) + + except Exception as e: + QMessageBox.critical(self, "Error", str(e)) diff --git a/src/gui/widgets/port_scanner.py b/src/gui/widgets/port_scanner.py new file mode 100644 index 0000000..c8fa850 --- /dev/null +++ b/src/gui/widgets/port_scanner.py @@ -0,0 +1,324 @@ +from PyQt6.QtWidgets import (QWidget, QVBoxLayout, QHBoxLayout, QLabel, + QLineEdit, QPushButton, QComboBox, QTableWidget, + QTableWidgetItem, QHeaderView, QMessageBox, QGroupBox, QTextEdit) +from PyQt6.QtCore import Qt +from src.core.scanner import NetworkScanner + +class PortScannerWidget(QWidget): + def __init__(self): + super().__init__() + self.scanner = NetworkScanner() + self.scan_thread = None + + self.layout = QVBoxLayout(self) + + # Controls Group + controls_group = QGroupBox("Target Configuration") + controls_layout = QVBoxLayout(controls_group) + + # Target IP + target_layout = QHBoxLayout() + target_layout.addWidget(QLabel("Target IP:")) + self.target_input = QLineEdit() + self.target_input.setPlaceholderText("e.g., 192.168.1.50") + target_layout.addWidget(self.target_input) + controls_layout.addLayout(target_layout) + + # Ports + ports_layout = QHBoxLayout() + ports_layout.addWidget(QLabel("Ports:")) + self.ports_input = QLineEdit() + self.ports_input.setPlaceholderText("e.g., 80, 443, 1000-2000") + ports_layout.addWidget(self.ports_input) + controls_layout.addLayout(ports_layout) + + # Protocol & Button + action_layout = QHBoxLayout() + action_layout.addWidget(QLabel("Protocol:")) + self.protocol_combo = QComboBox() + self.protocol_combo.addItems(["TCP", "UDP", "Both"]) + action_layout.addWidget(self.protocol_combo) + + self.scan_btn = QPushButton("Start Port Scan") + self.scan_btn.clicked.connect(self.start_scan) + action_layout.addWidget(self.scan_btn) + + controls_layout.addLayout(action_layout) + + self.layout.addWidget(controls_group) + + # Filter & Search Group + filter_layout = QHBoxLayout() + + filter_layout.addWidget(QLabel("Filter:")) + self.filter_combo = QComboBox() + self.filter_combo.addItems(["All", "Open", "Closed"]) + self.filter_combo.currentTextChanged.connect(self.apply_filters) + filter_layout.addWidget(self.filter_combo) + + filter_layout.addWidget(QLabel("Search Port:")) + self.search_input = QLineEdit() + self.search_input.setPlaceholderText("Search port number...") + self.search_input.textChanged.connect(self.apply_filters) + filter_layout.addWidget(self.search_input) + + self.layout.addLayout(filter_layout) + + # Results Table + self.table = QTableWidget() + self.table.setColumnCount(4) + self.table.setHorizontalHeaderLabels(["Port", "Protocol", "State", "Service"]) + self.table.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeMode.Stretch) + self.layout.addWidget(self.table) + + # Progress Log + log_group = QGroupBox("Scan Progress") + log_layout = QVBoxLayout(log_group) + self.log_output = QTextEdit() + self.log_output.setReadOnly(True) + self.log_output.setMaximumHeight(150) + self.log_output.setStyleSheet("background-color: #1e1e1e; color: #00ff00; font-family: monospace;") + log_layout.addWidget(self.log_output) + self.layout.addWidget(log_group) + + def start_scan(self): + if self.scan_btn.text() == "Stop Scan": + self.stop_scan() + return + + target = self.target_input.text().strip() + ports = self.ports_input.text().strip() + protocol = self.protocol_combo.currentText() + + if not target: + QMessageBox.warning(self, "Error", "Please enter a target IP.") + return + + if not ports: + QMessageBox.warning(self, "Error", "Please enter ports to scan.") + return + + self.scan_btn.setText("Stop Scan") + self.scan_btn.setStyleSheet("background-color: #800000; color: white;") # Red for stop + self.table.setRowCount(0) + self.log_output.clear() + + self.current_target = target + self.current_ports = ports + self.chain_udp = False + + if protocol == "Both": + self.chain_udp = True + self.run_scan(target, ports, "TCP") + else: + self.run_scan(target, ports, protocol) + + def run_scan(self, target, ports, protocol): + self.log_output.append(f"--- Starting {protocol} Scan ---\n") + # Port Scanner should use skip_discovery=True (-Pn) to ensure it scans even if ping blocked + self.scan_thread = self.scanner.scan(target, ports, protocol, skip_discovery=True) + self.scan_thread.scan_finished.connect(self.on_scan_finished) + self.scan_thread.scan_error.connect(self.on_scan_error) + self.scan_thread.scan_progress.connect(self.on_scan_progress) + self.scan_thread.port_found.connect(self.on_port_found) + + def stop_scan(self): + self.chain_udp = False # Cancel chain + if self.scan_thread: + self.scan_thread.stop() + self.scan_btn.setEnabled(False) + self.scan_btn.setText("Stopping...") + + def on_scan_progress(self, msg): + self.log_output.append(msg) + sb = self.log_output.verticalScrollBar() + sb.setValue(sb.maximum()) + + def on_port_found(self, port, proto, state, service): + # Check if already exists + items = self.table.findItems(str(port), Qt.MatchFlag.MatchExactly) + for item in items: + if item.column() == 0: + row = item.row() + if self.table.item(row, 1).text() == proto: + return + + self.add_result_row(port, proto, state, service) + + def on_scan_finished(self, result): + # 1. Process XML results + found_ports = set() + if result and 'scan' in result: + for ip, data in result['scan'].items(): + if 'tcp' in data: + for port, info in data['tcp'].items(): + self.update_or_add_row(port, "TCP", info['state'], info.get('name', '')) + found_ports.add((int(port), "TCP")) + if 'udp' in data: + for port, info in data['udp'].items(): + self.update_or_add_row(port, "UDP", info['state'], info.get('name', '')) + found_ports.add((int(port), "UDP")) + + # 2. Chain UDP if needed + if self.chain_udp: + self.chain_udp = False + self.run_scan(self.current_target, self.current_ports, "UDP") + return + + # 3. Finalize (only after all scans done) + self.finalize_scan(found_ports) + + def finalize_scan(self, found_ports_unused): + self.scan_btn.setEnabled(True) + self.scan_btn.setText("Start Port Scan") + self.scan_btn.setStyleSheet("") + + # Disable updates to prevent freeze + self.table.setUpdatesEnabled(False) + + try: + # 1. Build set of ALREADY FOUND ports from the table + existing_ports = set() + for row in range(self.table.rowCount()): + port_item = self.table.item(row, 0) + proto_item = self.table.item(row, 1) + if port_item and proto_item: + existing_ports.add((int(port_item.text()), proto_item.text())) + + # 2. Parse requested ports + requested_ports_str = self.ports_input.text().strip() + requested_proto = self.protocol_combo.currentText() + + target_ports = set() + try: + parts = requested_ports_str.split(',') + for part in parts: + if '-' in part: + start, end = map(int, part.split('-')) + target_ports.update(range(start, end + 1)) + else: + target_ports.add(int(part)) + except: + pass + + # Determine protocols + protos = [] + if requested_proto == "TCP" or requested_proto == "Both": + protos.append("TCP") + if requested_proto == "UDP" or requested_proto == "Both": + protos.append("UDP") + + # 3. Add closed rows for missing ports + # Only show closed ports if the total number of ports is reasonable (< 500) + # to prevent UI freezing and overwhelming the user. + if len(target_ports) > 500: + self.log_output.append(f"\n[INFO] Large range ({len(target_ports)} ports). Hiding closed ports.") + else: + self.table.setSortingEnabled(False) + + for port in target_ports: + for proto in protos: + if (port, proto) not in existing_ports: + # Pass update_filter=False for bulk addition + self.add_result_row(port, proto, "closed", "", update_filter=False) + + self.table.setSortingEnabled(True) + + self.apply_filters() # Apply once at the end + + finally: + self.table.setUpdatesEnabled(True) + + def row_exists(self, port, proto): + # Deprecated / Unused in optimized version but kept for safety if needed elsewhere + items = self.table.findItems(str(port), Qt.MatchFlag.MatchExactly) + for item in items: + if item.column() == 0: + if self.table.item(item.row(), 1).text() == proto: + return True + return False + + def update_or_add_row(self, port, proto, state, service): + # Find row + items = self.table.findItems(str(port), Qt.MatchFlag.MatchExactly) + found_row = -1 + for item in items: + if item.column() == 0: + if self.table.item(item.row(), 1).text() == proto: + found_row = item.row() + break + + if found_row >= 0: + # Update + self.update_row_style(found_row, state) + self.table.setItem(found_row, 3, QTableWidgetItem(service)) + else: + # Add + self.add_result_row(port, proto, state, service) + + def add_result_row(self, port, proto, state, service, update_filter=True): + row = self.table.rowCount() + self.table.insertRow(row) + self.table.setItem(row, 0, QTableWidgetItem(str(port))) + self.table.setItem(row, 1, QTableWidgetItem(proto)) + + state_item = QTableWidgetItem(state) + self.table.setItem(row, 2, state_item) + self.table.setItem(row, 3, QTableWidgetItem(service)) + + self.update_row_style(row, state) + + if update_filter: + self.apply_filters() # Re-apply filters to new row + + def update_row_style(self, row, state): + state_item = self.table.item(row, 2) + if state == 'open': + state_item.setForeground(Qt.GlobalColor.green) + state_item.setText("open") + elif state == 'closed': + state_item.setForeground(Qt.GlobalColor.red) + state_item.setText("closed") + else: + state_item.setForeground(Qt.GlobalColor.yellow) + state_item.setText(state) + + def apply_filters(self): + filter_state = self.filter_combo.currentText().lower() + search_text = self.search_input.text().strip() + + for i in range(self.table.rowCount()): + show = True + + # State Filter + state_item = self.table.item(i, 2) + state = state_item.text() + if filter_state != "all": + if filter_state == "open" and state != "open": + show = False + elif filter_state == "closed" and state != "closed": + show = False + + # Search Filter + if show and search_text: + port_item = self.table.item(i, 0) + if search_text not in port_item.text(): + show = False + + self.table.setRowHidden(i, not show) + + def on_scan_error(self, msg): + self.chain_udp = False # Stop chain on error + self.scan_btn.setEnabled(True) + self.scan_btn.setText("Start Port Scan") + self.scan_btn.setStyleSheet("") + + if "kill" in str(msg) or "None" in str(msg) or "stopped by user" in str(msg): + return + + QMessageBox.critical(self, "Error", f"Scan failed: {msg}") + + def closeEvent(self, event): + if self.scan_thread and self.scan_thread.isRunning(): + self.scan_thread.stop() + event.accept() diff --git a/src/gui/widgets/scanner_view.py b/src/gui/widgets/scanner_view.py new file mode 100644 index 0000000..8361031 --- /dev/null +++ b/src/gui/widgets/scanner_view.py @@ -0,0 +1,163 @@ +from PyQt6.QtWidgets import (QWidget, QVBoxLayout, QHBoxLayout, QLabel, + QLineEdit, QPushButton, QTableWidget, QTableWidgetItem, + QHeaderView, QMessageBox, QProgressBar, QComboBox, QTextEdit, QGroupBox) +from src.core.scanner import NetworkScanner + +class ScannerWidget(QWidget): + def __init__(self): + super().__init__() + self.scanner = NetworkScanner() + self.scan_thread = None + + self.layout = QVBoxLayout(self) + + # Controls + controls_layout = QHBoxLayout() + + self.target_input = QLineEdit() + self.target_input.setPlaceholderText("Target Network (e.g. 192.168.1.0/24)") + + self.scan_btn = QPushButton("Scan Network") + self.scan_btn.clicked.connect(self.start_scan) + + self.filter_combo = QComboBox() + self.filter_combo.addItems(["All", "Up", "Down"]) + self.filter_combo.currentTextChanged.connect(self.apply_filters) + + controls_layout.addWidget(QLabel("Target:")) + controls_layout.addWidget(self.target_input) + controls_layout.addWidget(QLabel("Filter:")) + controls_layout.addWidget(self.filter_combo) + controls_layout.addWidget(self.scan_btn) + + self.layout.addLayout(controls_layout) + + # Progress Bar + self.progress = QProgressBar() + self.progress.setRange(0, 0) # Indeterminate + self.progress.hide() + self.layout.addWidget(self.progress) + + # Results Table + self.table = QTableWidget() + self.table.setColumnCount(5) + self.table.setHorizontalHeaderLabels(["IP", "MAC", "Vendor", "Status", "Open Ports"]) + self.table.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeMode.Stretch) + self.layout.addWidget(self.table) + + # Progress Log + log_group = QGroupBox("Scan Progress") + log_layout = QVBoxLayout(log_group) + self.log_output = QTextEdit() + self.log_output.setReadOnly(True) + self.log_output.setMaximumHeight(150) + self.log_output.setStyleSheet("background-color: #1e1e1e; color: #00ff00; font-family: monospace;") + log_layout.addWidget(self.log_output) + self.layout.addWidget(log_group) + + def start_scan(self): + if self.scan_btn.text() == "Stop Scan": + if self.scan_thread: + self.scan_thread.stop() + self.scan_btn.setEnabled(False) + self.scan_btn.setText("Stopping...") + return + + target = self.target_input.text() + + if not target: + QMessageBox.warning(self, "Error", "Please enter a target IP range.") + return + + self.scan_btn.setText("Stop Scan") + self.scan_btn.setStyleSheet("background-color: #800000; color: white;") + self.progress.show() + self.table.setRowCount(0) + self.log_output.clear() + + # Use default scan (TCP, Fast ports) for discovery + # skip_discovery=False ensures we ONLY scan hosts that are UP (ARP/Ping response) + self.scan_thread = self.scanner.scan(target, ports=None, scan_type="TCP", skip_discovery=False) + self.scan_thread.scan_finished.connect(self.on_scan_finished) + self.scan_thread.scan_error.connect(self.on_scan_error) + self.scan_thread.scan_progress.connect(self.on_scan_progress) + + def on_scan_progress(self, msg): + self.log_output.append(msg) + sb = self.log_output.verticalScrollBar() + sb.setValue(sb.maximum()) + + def on_scan_finished(self, result): + self.scan_btn.setEnabled(True) + self.scan_btn.setText("Scan Network") + self.scan_btn.setStyleSheet("") + self.progress.hide() + + scan_data = result.get('scan', {}) + self.table.setRowCount(len(scan_data)) + + for i, (ip, data) in enumerate(scan_data.items()): + mac = "Unknown" + vendor = "Unknown" + status = data['status']['state'] + + if 'addresses' in data and 'mac' in data['addresses']: + mac = data['addresses']['mac'] + if 'vendor' in data and mac in data['vendor']: + vendor = data['vendor'][mac] + + # Collect Open Ports + open_ports = [] + if 'tcp' in data: + for port, info in data['tcp'].items(): + if info['state'] == 'open': + open_ports.append(f"{port}/tcp") + if 'udp' in data: + for port, info in data['udp'].items(): + if info['state'] == 'open': + open_ports.append(f"{port}/udp") + + ports_str = ", ".join(open_ports) if open_ports else "" + + self.table.setItem(i, 0, QTableWidgetItem(ip)) + self.table.setItem(i, 1, QTableWidgetItem(mac)) + self.table.setItem(i, 2, QTableWidgetItem(vendor)) + self.table.setItem(i, 3, QTableWidgetItem(status)) + self.table.setItem(i, 4, QTableWidgetItem(ports_str)) + + self.apply_filters() + + def apply_filters(self): + filter_text = self.filter_combo.currentText().lower() + + for row in range(self.table.rowCount()): + status_item = self.table.item(row, 3) + if not status_item: + continue + + status = status_item.text().lower() + + if filter_text == "all": + self.table.setRowHidden(row, False) + elif filter_text == "up" and status == "up": + self.table.setRowHidden(row, False) + elif filter_text == "down" and status == "down": + self.table.setRowHidden(row, False) + else: + self.table.setRowHidden(row, True) + + def on_scan_error(self, error): + self.scan_btn.setEnabled(True) + self.scan_btn.setText("Scan Network") + self.scan_btn.setStyleSheet("") + self.progress.hide() + + if "stopped by user" in str(error): + return + + QMessageBox.critical(self, "Scan Error", error) + + def closeEvent(self, event): + if self.scan_thread and self.scan_thread.isRunning(): + self.scan_thread.stop() + event.accept() diff --git a/src/gui/widgets/sniffer_view.py b/src/gui/widgets/sniffer_view.py new file mode 100644 index 0000000..bcf8da0 --- /dev/null +++ b/src/gui/widgets/sniffer_view.py @@ -0,0 +1,162 @@ +from PyQt6.QtWidgets import (QWidget, QVBoxLayout, QHBoxLayout, QLabel, + QComboBox, QPushButton, QTableWidget, QTableWidgetItem, + QHeaderView, QLineEdit, QMessageBox, QCheckBox, QSplitter, QGroupBox) +from PyQt6.QtGui import QColor +from PyQt6.QtCore import Qt +from src.core.sniffer import NetworkSniffer + +class SnifferWidget(QWidget): + def __init__(self, network_manager): + super().__init__() + self.nm = network_manager + self.sniffer = NetworkSniffer() + self.is_sniffing = False + + self.layout = QVBoxLayout(self) + + # Controls + controls_layout = QHBoxLayout() + + self.iface_combo = QComboBox() + self.refresh_interfaces() + + self.filter_input = QLineEdit() + self.filter_input.setPlaceholderText("BPF Filter (e.g. tcp port 80)") + + self.ip_filter_input = QLineEdit() + self.ip_filter_input.setPlaceholderText("Filter IP (Src/Dst)") + self.ip_filter_input.textChanged.connect(self.apply_ip_filter) + + self.start_btn = QPushButton("Start Sniffing") + self.start_btn.clicked.connect(self.toggle_sniffing) + + controls_layout.addWidget(QLabel("Interface:")) + controls_layout.addWidget(self.iface_combo) + controls_layout.addWidget(QLabel("BPF:")) + controls_layout.addWidget(self.filter_input) + controls_layout.addWidget(QLabel("IP Filter:")) + controls_layout.addWidget(self.ip_filter_input) + controls_layout.addWidget(self.start_btn) + + self.layout.addLayout(controls_layout) + + # Splitter for Dual Consoles + self.splitter = QSplitter(Qt.Orientation.Vertical) + self.layout.addWidget(self.splitter) + + # Top Console: All Traffic + self.all_traffic_group = QGroupBox("All Traffic") + self.all_traffic_layout = QVBoxLayout(self.all_traffic_group) + self.table_all = self.create_table() + self.all_traffic_layout.addWidget(self.table_all) + self.splitter.addWidget(self.all_traffic_group) + + # Bottom Console: Failed / Slow / Anomalies + self.failed_traffic_group = QGroupBox("Failed / Slow / Anomalies") + self.failed_traffic_layout = QVBoxLayout(self.failed_traffic_group) + self.table_failed = self.create_table() + self.failed_traffic_layout.addWidget(self.table_failed) + self.splitter.addWidget(self.failed_traffic_group) + + # Set initial sizes + self.splitter.setSizes([400, 200]) + + def create_table(self): + table = QTableWidget() + table.setColumnCount(6) + table.setHorizontalHeaderLabels(["Time", "Source", "Destination", "Protocol", "Info", "Status"]) + table.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeMode.Interactive) + table.horizontalHeader().setStretchLastSection(True) + table.setAlternatingRowColors(True) + return table + + def refresh_interfaces(self): + self.iface_combo.clear() + interfaces = self.nm.list_interfaces() + for iface in interfaces: + self.iface_combo.addItem(iface['name']) + + def toggle_sniffing(self): + if self.is_sniffing: + self.sniffer.stop_sniffing() + self.start_btn.setText("Start Sniffing") + self.start_btn.setStyleSheet("background-color: #007acc;") + self.is_sniffing = False + else: + iface = self.iface_combo.currentText() + filter_str = self.filter_input.text() + + self.table_all.setRowCount(0) + self.table_failed.setRowCount(0) + + self.sniffer_thread = self.sniffer.start_sniffing(iface, filter_str) + self.sniffer_thread.packet_received.connect(self.add_packet) + self.sniffer_thread.error_occurred.connect(self.on_error) + + self.start_btn.setText("Stop Sniffing") + self.start_btn.setStyleSheet("background-color: #d32f2f;") + self.is_sniffing = True + + def add_packet(self, pkt): + # Add to All Traffic + self.insert_packet_row(self.table_all, pkt) + + # Check if failed/slow/anomaly + status = pkt.get('status', '') + if status in ['Error', 'Connecting']: + self.insert_packet_row(self.table_failed, pkt) + + def insert_packet_row(self, table, pkt): + row = table.rowCount() + table.insertRow(row) + table.setItem(row, 0, QTableWidgetItem(pkt['time'])) + table.setItem(row, 1, QTableWidgetItem(pkt['src'])) + table.setItem(row, 2, QTableWidgetItem(pkt['dst'])) + table.setItem(row, 3, QTableWidgetItem(str(pkt['proto']))) + table.setItem(row, 4, QTableWidgetItem(pkt['info'])) + + status_item = QTableWidgetItem(pkt.get('status', '')) + if pkt.get('status') == 'Error': + status_item.setForeground(QColor('#ff5252')) # Red + elif pkt.get('status') == 'Connecting': + status_item.setForeground(QColor('#ffab40')) # Orange + elif pkt.get('status') == 'OK': + status_item.setForeground(QColor('#69f0ae')) # Green + + table.setItem(row, 5, status_item) + + # Check filter immediately + self.check_filter_for_row(table, row, pkt['src'], pkt['dst']) + + if not table.isRowHidden(row): + table.scrollToBottom() + + def apply_ip_filter(self): + filter_ip = self.ip_filter_input.text().strip() + + for table in [self.table_all, self.table_failed]: + for row in range(table.rowCount()): + src = table.item(row, 1).text() + dst = table.item(row, 2).text() + + if not filter_ip: + table.setRowHidden(row, False) + elif filter_ip in src or filter_ip in dst: + table.setRowHidden(row, False) + else: + table.setRowHidden(row, True) + + def check_filter_for_row(self, table, row, src, dst): + filter_ip = self.ip_filter_input.text().strip() + if not filter_ip: + table.setRowHidden(row, False) + return + + if filter_ip in src or filter_ip in dst: + table.setRowHidden(row, False) + else: + table.setRowHidden(row, True) + + def on_error(self, msg): + self.toggle_sniffing() + QMessageBox.critical(self, "Sniffer Error", msg) diff --git a/src/gui/widgets/vlan_bridge.py b/src/gui/widgets/vlan_bridge.py new file mode 100644 index 0000000..a500eb6 --- /dev/null +++ b/src/gui/widgets/vlan_bridge.py @@ -0,0 +1,510 @@ +from PyQt6.QtWidgets import (QWidget, QVBoxLayout, QHBoxLayout, QLabel, + QComboBox, QSpinBox, QPushButton, QGroupBox, + QMessageBox, QFormLayout, QLineEdit, QTableWidget, + QTableWidgetItem, QHeaderView, QCheckBox, QSplitter) +from PyQt6.QtCore import Qt, QTimer +from src.core.vlan_scanner import VLANScanner + +class VLANBridgeWidget(QWidget): + def __init__(self, network_manager): + super().__init__() + self.nm = network_manager + self.vlan_scanner = VLANScanner() + self.scan_thread = None + + self.layout = QVBoxLayout(self) + + # Create a horizontal layout for the top part (Bridge + Wifi) + self.splitter = QSplitter(Qt.Orientation.Vertical) + self.layout.addWidget(self.splitter) + + # Top Container Widget + top_widget = QWidget() + top_layout = QHBoxLayout(top_widget) + top_layout.setContentsMargins(0, 0, 0, 0) + + # Left Column: Wired Bridging + left_col = QVBoxLayout() + + # Wired Bridging Group + wired_group = QGroupBox("Wired Bridging") + wired_layout = QVBoxLayout(wired_group) + wired_layout.setSpacing(5) + wired_layout.setContentsMargins(10, 15, 10, 10) + + # Mode Selection + mode_layout = QHBoxLayout() + mode_layout.addWidget(QLabel("Mode:")) + self.mode_combo = QComboBox() + self.mode_combo.addItems(["VLAN Trunk (Trunk -> Access)", "Passthrough (Untagged -> Untagged)"]) + self.mode_combo.currentIndexChanged.connect(self.toggle_mode_inputs) + mode_layout.addWidget(self.mode_combo) + wired_layout.addLayout(mode_layout) + + # Interface Selection + form_layout = QFormLayout() + form_layout.setSpacing(5) + self.trunk_combo = QComboBox() # Input + self.trunk_combo.currentIndexChanged.connect(self.on_trunk_combo_changed) + self.access_combo = QComboBox() # Output + + self.lbl_input = QLabel("Input Interface (Trunk):") + self.lbl_output = QLabel("Output Interface (Access):") + + form_layout.addRow(self.lbl_input, self.trunk_combo) + form_layout.addRow(self.lbl_output, self.access_combo) + + # VLAN ID Input + self.vlan_input = QSpinBox() + self.vlan_input.setRange(1, 4094) + self.lbl_vlan = QLabel("VLAN ID:") + form_layout.addRow(self.lbl_vlan, self.vlan_input) + + wired_layout.addLayout(form_layout) + + # IP / DHCP Configuration + ip_group = QGroupBox("Bridge IP & DHCP Server") + ip_layout = QFormLayout(ip_group) + ip_layout.setSpacing(5) + + self.enable_dhcp_cb = QCheckBox("Enable DHCP Server & Static IP") + self.enable_dhcp_cb.toggled.connect(self.toggle_dhcp_inputs) + ip_layout.addRow(self.enable_dhcp_cb) + + self.bridge_ip = QLineEdit("192.168.50.1") + self.dhcp_start = QLineEdit("192.168.50.10") + self.dhcp_end = QLineEdit("192.168.50.100") + + ip_layout.addRow("Bridge IP (Gateway):", self.bridge_ip) + ip_layout.addRow("DHCP Start:", self.dhcp_start) + ip_layout.addRow("DHCP End:", self.dhcp_end) + + wired_layout.addWidget(ip_group) + self.ip_group = ip_group + + # Initialize inputs state + self.toggle_dhcp_inputs(False) + + self.bridge_btn = QPushButton("Create Bridge") + self.bridge_btn.clicked.connect(self.create_bridge) + wired_layout.addWidget(self.bridge_btn) + + # Bridge Management Group + manage_group = QGroupBox("Manage Bridges") + manage_layout = QHBoxLayout(manage_group) + manage_layout.setContentsMargins(5, 15, 5, 5) + + self.active_bridges_combo = QComboBox() + self.delete_bridge_btn = QPushButton("Delete Bridge") + self.delete_bridge_btn.setStyleSheet("background-color: #800000; color: white;") + self.delete_bridge_btn.clicked.connect(self.delete_bridge) + + manage_layout.addWidget(QLabel("Active Bridges:")) + manage_layout.addWidget(self.active_bridges_combo) + manage_layout.addWidget(self.delete_bridge_btn) + + wired_layout.addWidget(manage_group) + + left_col.addWidget(wired_group) + top_layout.addLayout(left_col) + + # Right Column: Wi-Fi Bridging + right_col = QVBoxLayout() + wifi_group = QGroupBox("Wi-Fi AP Bridging") + wifi_layout = QVBoxLayout(wifi_group) + wifi_layout.setSpacing(5) + wifi_layout.setContentsMargins(10, 15, 10, 10) + + wifi_form = QFormLayout() + wifi_form.setSpacing(5) + self.wifi_iface_combo = QComboBox() + self.wifi_iface_combo.currentIndexChanged.connect(self.on_wifi_iface_changed) + self.ssid_input = QLineEdit() + self.password_input = QLineEdit() + self.password_input.setEchoMode(QLineEdit.EchoMode.Password) + self.bridge_select_combo = QComboBox() # Select which bridge to attach to + + wifi_form.addRow("Wi-Fi Interface:", self.wifi_iface_combo) + wifi_form.addRow("SSID:", self.ssid_input) + wifi_form.addRow("Password:", self.password_input) + wifi_form.addRow("Attach to Bridge:", self.bridge_select_combo) + + wifi_layout.addLayout(wifi_form) + + # Wi-Fi DHCP Configuration + wifi_dhcp_group = QGroupBox("Wi-Fi DHCP Server") + wifi_dhcp_layout = QFormLayout(wifi_dhcp_group) + wifi_dhcp_layout.setSpacing(5) + + self.wifi_enable_dhcp_cb = QCheckBox("Enable DHCP on Bridge") + self.wifi_enable_dhcp_cb.toggled.connect(self.toggle_wifi_dhcp_inputs) + wifi_dhcp_layout.addRow(self.wifi_enable_dhcp_cb) + + self.wifi_bridge_ip = QLineEdit("192.168.60.1") + self.wifi_dhcp_start = QLineEdit("192.168.60.10") + self.wifi_dhcp_end = QLineEdit("192.168.60.100") + + wifi_dhcp_layout.addRow("Bridge IP:", self.wifi_bridge_ip) + wifi_dhcp_layout.addRow("DHCP Start:", self.wifi_dhcp_start) + wifi_dhcp_layout.addRow("DHCP End:", self.wifi_dhcp_end) + + wifi_layout.addWidget(wifi_dhcp_group) + self.wifi_dhcp_group = wifi_dhcp_group + + # Initialize Wi-Fi DHCP inputs state + self.toggle_wifi_dhcp_inputs(False) + + wifi_btns_layout = QHBoxLayout() + self.wifi_btn = QPushButton("Start Wi-Fi AP") + self.wifi_btn.clicked.connect(self.start_wifi) + wifi_btns_layout.addWidget(self.wifi_btn) + + self.stop_wifi_btn = QPushButton("Stop Wi-Fi AP") + self.stop_wifi_btn.clicked.connect(self.stop_wifi) + self.stop_wifi_btn.setEnabled(False) + wifi_btns_layout.addWidget(self.stop_wifi_btn) + + wifi_layout.addLayout(wifi_btns_layout) + + # Connected Clients + self.clients_group = QGroupBox("Connected Clients") + clients_layout = QVBoxLayout(self.clients_group) + clients_layout.setContentsMargins(5, 15, 5, 5) + self.clients_table = QTableWidget() + self.clients_table.setColumnCount(3) + self.clients_table.setHorizontalHeaderLabels(["MAC", "IP", "Hostname"]) + self.clients_table.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeMode.Stretch) + self.clients_table.setMaximumHeight(150) + clients_layout.addWidget(self.clients_table) + wifi_layout.addWidget(self.clients_group) + + right_col.addWidget(wifi_group) + top_layout.addLayout(right_col) + + # Client Refresh Timer + self.client_timer = QTimer() + self.client_timer.timeout.connect(self.refresh_clients) + self.client_timer.setInterval(5000) # 5 seconds + + self.splitter.addWidget(top_widget) + + # Bottom Part: VLAN Scanner + scanner_group = QGroupBox("Passive VLAN Scanner") + scanner_layout = QVBoxLayout(scanner_group) + scanner_layout.setContentsMargins(10, 15, 10, 10) + + scan_controls = QHBoxLayout() + self.scan_iface_combo = QComboBox() + self.scan_btn = QPushButton("Start Scan (10s)") + self.scan_btn.clicked.connect(self.start_scan) + + scan_controls.addWidget(QLabel("Interface:")) + scan_controls.addWidget(self.scan_iface_combo) + scan_controls.addWidget(self.scan_btn) + + scanner_layout.addLayout(scan_controls) + + self.scan_table = QTableWidget() + self.scan_table.setColumnCount(2) + self.scan_table.setHorizontalHeaderLabels(["VLAN ID", "Packets Detected"]) + self.scan_table.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeMode.Stretch) + self.scan_table.setMinimumHeight(200) # Ensure it has space + scanner_layout.addWidget(self.scan_table) + + self.port_type_label = QLabel("Port Type: Unknown") + self.port_type_label.setStyleSheet("font-weight: bold; color: #aaaaaa;") + scanner_layout.addWidget(self.port_type_label) + + self.splitter.addWidget(scanner_group) + + # Set initial sizes for splitter (Top, Bottom) + self.splitter.setSizes([400, 300]) + + self.refresh_interfaces() + self.toggle_mode_inputs() # Initialize UI based on default mode + + def refresh_interfaces(self): + interfaces = self.nm.list_interfaces() + self.trunk_combo.clear() + self.access_combo.clear() + self.scan_iface_combo.clear() + self.wifi_iface_combo.clear() + + for iface in interfaces: + name = iface['name'] + self.trunk_combo.addItem(name) + self.access_combo.addItem(name) + self.scan_iface_combo.addItem(name) + if iface['type'] == 'Wireless': + self.wifi_iface_combo.addItem(name) + + # Populate bridges + self.bridge_select_combo.clear() + self.active_bridges_combo.clear() + + bridges = [iface['name'] for iface in interfaces if iface['name'].startswith('br')] + + for br in bridges: + self.bridge_select_combo.addItem(br) + self.active_bridges_combo.addItem(br) + + # Add manual entry option for Wi-Fi bridge selection if needed + self.bridge_select_combo.setEditable(True) + + def toggle_mode_inputs(self): + mode = self.mode_combo.currentText() + if "Passthrough" in mode: + self.lbl_input.setText("Interface 1:") + self.lbl_output.setText("Interface 2:") + self.lbl_vlan.setVisible(False) + self.vlan_input.setVisible(False) + else: + self.lbl_input.setText("Input Interface (Trunk):") + self.lbl_output.setText("Output Interface (Access):") + self.lbl_vlan.setVisible(True) + self.lbl_vlan.setVisible(True) + self.vlan_input.setVisible(True) + + def toggle_dhcp_inputs(self, checked): + self.bridge_ip.setEnabled(checked) + self.dhcp_start.setEnabled(checked) + self.dhcp_end.setEnabled(checked) + + def toggle_wifi_dhcp_inputs(self, checked): + self.wifi_bridge_ip.setEnabled(checked) + self.wifi_dhcp_start.setEnabled(checked) + self.wifi_dhcp_end.setEnabled(checked) + + def on_trunk_combo_changed(self, index): + iface_name = self.trunk_combo.currentText() + self.autofill_dhcp(iface_name, self.bridge_ip, self.dhcp_start, self.dhcp_end) + + def on_wifi_iface_changed(self, index): + iface_name = self.wifi_iface_combo.currentText() + self.autofill_dhcp(iface_name, self.wifi_bridge_ip, self.wifi_dhcp_start, self.wifi_dhcp_end) + + def autofill_dhcp(self, iface_name, ip_field, start_field, end_field): + if not iface_name: + return + + # Get config to auto-fill DHCP + config = self.nm.get_interface_config(iface_name) + ip = config.get('ip', '') + + if ip: + # Suggest a subnet based on this IP + parts = ip.split('.') + if len(parts) == 4: + base = ".".join(parts[:3]) + ip_field.setText(ip) + start_field.setText(f"{base}.100") + end_field.setText(f"{base}.200") + + def create_bridge(self): + mode = self.mode_combo.currentText() + iface1 = self.trunk_combo.currentText() + iface2 = self.access_combo.currentText() + + if iface1 == iface2: + QMessageBox.warning(self, "Error", "Interfaces must be different") + return + + success = False + msg = "" + bridge_name = "" + + if "Passthrough" in mode: + success, msg = self.nm.create_simple_bridge(iface1, iface2) + bridge_name = "br_pass" + else: + vlan_id = self.vlan_input.value() + success, msg = self.nm.create_vlan_bridge(iface1, vlan_id, iface2) + bridge_name = f"br{vlan_id}" + + if success: + # Handle IP/DHCP + if self.enable_dhcp_cb.isChecked(): + ip = self.bridge_ip.text() + start = self.dhcp_start.text() + end = self.dhcp_end.text() + + # Start DHCP + d_success, d_msg = self.nm.start_dhcp_server(bridge_name, ip, start, end) + if d_success: + msg += f"\nDHCP Server running on {ip}" + else: + msg += f"\nDHCP Error: {d_msg}" + + QMessageBox.information(self, "Success", msg) + self.bridge_select_combo.addItem(bridge_name) + else: + QMessageBox.critical(self, "Error", msg) + + def start_wifi(self): + wifi_iface = self.wifi_iface_combo.currentText() + ssid = self.ssid_input.text() + password = self.password_input.text() + bridge = self.bridge_select_combo.currentText() + + if not wifi_iface or not ssid or not password or not bridge: + QMessageBox.warning(self, "Error", "Please fill all Wi-Fi fields.") + return + + success, msg = self.nm.setup_wifi_ap(wifi_iface, ssid, password, bridge) + + if success: + # Handle DHCP for Wi-Fi + if self.wifi_enable_dhcp_cb.isChecked(): + ip = self.wifi_bridge_ip.text() + start = self.wifi_dhcp_start.text() + end = self.wifi_dhcp_end.text() + + # Start DHCP on the bridge + d_success, d_msg = self.nm.start_dhcp_server(bridge, ip, start, end) + if d_success: + msg += f"\nDHCP Server running on {ip}" + else: + msg += f"\nDHCP Error: {d_msg}" + + self.wifi_btn.setEnabled(False) + self.stop_wifi_btn.setEnabled(True) + self.client_timer.start() + QMessageBox.information(self, "Success", msg) + else: + QMessageBox.critical(self, "Error", f"Failed to start Wi-Fi AP: {msg}") + + def delete_bridge(self): + bridge = self.active_bridges_combo.currentText() + if not bridge: + return + + confirm = QMessageBox.question(self, "Confirm Deletion", + f"Are you sure you want to delete bridge '{bridge}'?\nThis will stop any associated DHCP servers.", + QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No) + + if confirm == QMessageBox.StandardButton.Yes: + success, msg = self.nm.delete_bridge(bridge) + if success: + QMessageBox.information(self, "Success", msg) + self.refresh_interfaces() + else: + QMessageBox.critical(self, "Error", msg) + + def stop_wifi(self): + iface = self.wifi_iface_combo.currentText() + success, msg = self.nm.stop_wifi_ap(iface) + + if success: + self.wifi_btn.setEnabled(True) + self.stop_wifi_btn.setEnabled(False) + self.client_timer.stop() + self.clients_table.setRowCount(0) + QMessageBox.information(self, "Success", msg) + else: + QMessageBox.critical(self, "Error", msg) + + def refresh_clients(self): + iface = self.wifi_iface_combo.currentText() + clients = self.nm.get_wifi_clients(iface) + + self.clients_table.setRowCount(len(clients)) + for i, client in enumerate(clients): + self.clients_table.setItem(i, 0, QTableWidgetItem(client['mac'])) + self.clients_table.setItem(i, 1, QTableWidgetItem(client['ip'])) + self.clients_table.setItem(i, 2, QTableWidgetItem(client['hostname'])) + + def start_scan(self): + if self.scan_btn.text() == "Stop Scan": + if self.scan_thread: + self.scan_thread.stop() + self.scan_btn.setEnabled(False) + self.scan_btn.setText("Stopping...") + return + + iface = self.scan_iface_combo.currentText() + self.scan_table.setRowCount(0) + self.port_type_label.setText("Port Type: Scanning...") + self.port_type_label.setStyleSheet("font-weight: bold; color: #e0e000;") + + self.scan_btn.setText("Stop Scan") + self.scan_btn.setStyleSheet("background-color: #800000; color: white;") + + self.scan_thread = self.vlan_scanner.scan(iface) + self.scan_thread.vlan_detected.connect(self.on_vlan_detected) + self.scan_thread.untagged_detected.connect(self.on_untagged_detected) + self.scan_thread.scan_finished.connect(self.on_scan_finished) + self.scan_thread.error_occurred.connect(self.on_scan_error) + + def on_untagged_detected(self, count): + self.update_table_row("Untagged (Native)", count) + self.update_port_analysis() + + def on_vlan_detected(self, vlan_id, count): + self.update_table_row(str(vlan_id), count) + self.update_port_analysis() + + def update_table_row(self, label, count): + # Check if row exists + found = False + for row in range(self.scan_table.rowCount()): + if self.scan_table.item(row, 0).text() == label: + self.scan_table.setItem(row, 1, QTableWidgetItem(str(count))) + found = True + break + + if not found: + row = self.scan_table.rowCount() + self.scan_table.insertRow(row) + self.scan_table.setItem(row, 0, QTableWidgetItem(label)) + self.scan_table.setItem(row, 1, QTableWidgetItem(str(count))) + + def update_port_analysis(self): + # Analyze table content to guess port type + has_untagged = False + vlan_count = 0 + + for row in range(self.scan_table.rowCount()): + label = self.scan_table.item(row, 0).text() + if label == "Untagged (Native)": + has_untagged = True + else: + vlan_count += 1 + + if vlan_count > 1: + msg = f"Trunk Port (Detected {vlan_count} VLANs)" + color = "#00e000" # Green + elif vlan_count == 1 and has_untagged: + msg = "Hybrid Port (Tagged + Untagged)" + color = "#e0e000" # Yellow + elif vlan_count == 1: + msg = "Trunk Port (Single VLAN)" + color = "#00e000" + elif has_untagged: + msg = "Access Port (Untagged Traffic Only)" + color = "#00e0e0" # Cyan + else: + msg = "Unknown / No Traffic" + color = "#aaaaaa" + + self.port_type_label.setText(f"Port Type: {msg}") + self.port_type_label.setStyleSheet(f"font-weight: bold; color: {color};") + + def on_scan_finished(self): + self.scan_btn.setEnabled(True) + self.scan_btn.setText("Start Scan (10s)") + self.scan_btn.setStyleSheet("") + # Don't show popup if just stopped, but hard to distinguish here without flag. + # Let's just show it, or maybe check if it was manual stop? + # For now, simple message is fine. + # QMessageBox.information(self, "Scan Finished", "VLAN Scan completed.") + + def on_scan_error(self, msg): + self.scan_btn.setEnabled(True) + self.scan_btn.setText("Start Scan (10s)") + self.scan_btn.setStyleSheet("") + QMessageBox.critical(self, "Error", msg) + + def closeEvent(self, event): + if self.scan_thread and self.scan_thread.isRunning(): + self.scan_thread.stop() + event.accept() diff --git a/src/utils/__pycache__/report_generator.cpython-312.pyc b/src/utils/__pycache__/report_generator.cpython-312.pyc new file mode 100644 index 0000000..9f2b044 Binary files /dev/null and b/src/utils/__pycache__/report_generator.cpython-312.pyc differ diff --git a/src/utils/report_generator.py b/src/utils/report_generator.py new file mode 100644 index 0000000..4517c85 --- /dev/null +++ b/src/utils/report_generator.py @@ -0,0 +1,104 @@ +import datetime +import os + +class ReportGenerator: + def __init__(self): + pass + + def generate_html_report(self, interfaces, scan_results, sniff_all, sniff_failed, filename="report.html"): + """ + Generates an HTML report with the provided data. + """ + + html = f""" + + + + JNC-NetTools Report + + + +
+

JNC-NetTools Diagnostic Report

+

Generated on: {datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")}

+ +

Network Interfaces

+ + + """ + + for iface in interfaces: + ip = iface['ipv4'][0] if iface['ipv4'] else "N/A" + status_class = "status-up" if iface['is_up'] else "status-down" + html += f"" + + html += """ +
NameTypeStatusIP AddressMAC Address
{iface['name']}{iface['type']}{iface['status']}{ip}{iface['mac']}
+ +

Port Scan Results

+ + + """ + + if not scan_results: + html += "" + else: + for item in scan_results: + html += f"" + + html += """ +
IPMACVendorStatus
No scan results available.
{item['ip']}{item['mac']}{item['vendor']}{item['status']}
+ +

Anomalies / Failed Traffic (Errors & Retransmissions)

+ + + """ + + if not sniff_failed: + html += "" + else: + for pkt in sniff_failed: + html += f"" + + html += """ +
TimeSourceDestinationProtocolInfoStatus
No anomalies detected.
{pkt['time']}{pkt['src']}{pkt['dst']}{pkt['proto']}{pkt['info']}{pkt['status']}
+ +

Traffic Log (Last 100 Packets)

+ + + """ + + # Limit to last 100 for readability + for pkt in sniff_all[-100:]: + html += f"" + + html += """ +
TimeSourceDestinationProtocolInfoStatus
{pkt['time']}{pkt['src']}{pkt['dst']}{pkt['proto']}{pkt['info']}{pkt['status']}
+
+ + + """ + + try: + with open(filename, "w") as f: + f.write(html) + return True, os.path.abspath(filename) + except Exception as e: + return False, str(e) diff --git a/src/walkthrough.md.resolved b/src/walkthrough.md.resolved new file mode 100644 index 0000000..6396e13 --- /dev/null +++ b/src/walkthrough.md.resolved @@ -0,0 +1,86 @@ +# JNC-Scan Walkthrough + +This guide explains how to run and verify the **JNC-Scan** network diagnostic tool. + +## Prerequisites + +Ensure you have the required system dependencies: +```bash +sudo apt install python3-pyqt6 nmap tcpdump bridge-utils hostapd dnsmasq python3-venv libxcb-cursor0 ethtool +``` + +## Installation (Virtual Environment) + +Since modern Linux distributions manage Python packages externally, we should use a virtual environment: + +1. **Create a virtual environment**: + ```bash + python3 -m venv venv + ``` + +2. **Activate it and install dependencies**: + ```bash + source venv/bin/activate + pip install -r requirements.txt + ``` + +## Running the Application + +**CRITICAL**: The application requires **root privileges**. When using a virtual environment, point `sudo` to the venv's Python executable: + +```bash +sudo ./venv/bin/python3 main.py +``` + +## Verification Steps + +### 1. Interface Management +- **Launch**: Verify the app starts and lists your network interfaces. +- **Details**: Select an interface to see its IP, MAC, and status. + +### 2. IP Configuration & Calculator +- **Static IP**: Go to "IP Configuration". Select an interface, choose "Static", enter an IP (e.g., `192.168.99.1`) and CIDR (`24`). Click "Apply". Verify with `ip addr` in a terminal. +- **Calculator**: Enter an IP and CIDR in the calculator section. Click "Calculate" to see the Netmask and Broadcast address. + +### 3. VLAN & Bridging +- **Wired Bridge**: + 1. Go to "VLAN / Bridge". + 2. Select your main Ethernet (Trunk) and a secondary Ethernet (Access). + 3. Enter a VLAN ID (e.g., `35`). + 4. Click "Create Bridge". + 5. Verify connectivity on the secondary interface. +- **Wi-Fi AP**: + 1. Select a Wireless interface. + 2. Enter SSID and Password. + 3. Select the bridge you just created (e.g., `br35`). + 4. Click "Start Wi-Fi AP". +- **VLAN Scanner**: + 1. In "VLAN / Bridge", look for "Passive VLAN Scanner". + 2. Select an interface. + 3. Click "Start Scan". + 4. Wait 10 seconds. The table will show any VLAN tags detected on the wire. + +### 4. Network Scanner +- Go to "Scanner". +- Enter a range (e.g., `192.168.1.0/24`). +- Click "Scan Network". +- Verify the list of devices appears. + +### 5. Packet Sniffer +- Go to "Sniffer". +- Select an interface (e.g., the bridge `br35`). +- Click "Start Sniffing". +- Generate traffic (ping, browse). +- Verify packets appear in the table. + +### 6. Reporting +- Click "Generate Report" in the toolbar. +- Save the HTML file. +- Open it in a browser to view the summary of interfaces, scan results, and captured packets. + +## Troubleshooting + +- **"No module named..."**: Ensure you are running inside the venv: `sudo ./venv/bin/python3 main.py`. +- **"xcb plugin error"**: Install `libxcb-cursor0`. +- **"Permission denied"**: Ensure you are running with `sudo`. +- **NetworkManager errors**: Ensure `nmcli` is installed (`sudo apt install network-manager`).