diff --git a/src/__pycache__/__init__.cpython-312.pyc b/src/__pycache__/__init__.cpython-312.pyc index 83d69a7..f896b42 100644 Binary files a/src/__pycache__/__init__.cpython-312.pyc and b/src/__pycache__/__init__.cpython-312.pyc differ diff --git a/src/core/__pycache__/advanced_router.cpython-312.pyc b/src/core/__pycache__/advanced_router.cpython-312.pyc new file mode 100644 index 0000000..e4ba1bd Binary files /dev/null and b/src/core/__pycache__/advanced_router.cpython-312.pyc differ diff --git a/src/core/__pycache__/diagnostics.cpython-312.pyc b/src/core/__pycache__/diagnostics.cpython-312.pyc index 5704771..c4d4850 100644 Binary files a/src/core/__pycache__/diagnostics.cpython-312.pyc 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 index 13cf926..ec8f65d 100644 Binary files a/src/core/__pycache__/network_manager.cpython-312.pyc 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 index ed9f817..6dd2ae8 100644 Binary files a/src/core/__pycache__/scanner.cpython-312.pyc 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 index 7159626..2760e6d 100644 Binary files a/src/core/__pycache__/sniffer.cpython-312.pyc and b/src/core/__pycache__/sniffer.cpython-312.pyc differ diff --git a/src/core/__pycache__/vlan35_checker.cpython-312.pyc b/src/core/__pycache__/vlan35_checker.cpython-312.pyc new file mode 100644 index 0000000..23bf009 Binary files /dev/null and b/src/core/__pycache__/vlan35_checker.cpython-312.pyc differ diff --git a/src/core/__pycache__/vlan_scanner.cpython-312.pyc b/src/core/__pycache__/vlan_scanner.cpython-312.pyc index 7c7311b..ccd245a 100644 Binary files a/src/core/__pycache__/vlan_scanner.cpython-312.pyc and b/src/core/__pycache__/vlan_scanner.cpython-312.pyc differ diff --git a/src/core/advanced_router.py b/src/core/advanced_router.py new file mode 100644 index 0000000..56053ba --- /dev/null +++ b/src/core/advanced_router.py @@ -0,0 +1,996 @@ +import json +import socket +import subprocess +from typing import Dict, List, Optional, Tuple + +from pyroute2 import IPRoute + +from src.core.network_manager import NetworkManager + + +class AdvancedRouter: + """ + Capa de alto nivel sobre pyroute2 / iptables / nftables + para ofrecer primitivas de routing avanzadas al GUI. + """ + + def __init__(self, network_manager: NetworkManager): + self.nm = network_manager + # Reutilizamos el mismo IPRoute que usa NetworkManager + self.ipr: IPRoute = network_manager.ipr + + # Estado en memoria para poder mostrar en la UI y guardar/cargar + self.vlans: Dict[str, Dict] = {} # clave: ifname VLAN (ej. eth0.10) + self.vlan_redirections: List[Dict] = [] + self.dhcp_servers: Dict[str, Dict] = {} # clave: interfaz sobre la que corre dnsmasq + self.gateways: Dict[str, Dict] = {} # clave: nombre interfaz o VLAN ifname + self.routes: List[Dict] = [] + self.routing_rules: List[Dict] = [] + self.nat_rules: List[Dict] = [] + self.firewall_rules: List[Dict] = [] + self.vlan_visibility: Dict[Tuple[int, int], bool] = {} # (src_vlan, dst_vlan) -> allowed + + self.firewall_backend = self.detect_firewall_backend() + + # Descubrir VLANs existentes en el sistema (por ejemplo eth0.10 creadas en sesiones anteriores) + self._discover_existing_vlans() + + def _discover_existing_vlans(self) -> None: + """ + Recorre los enlaces de pyroute2 y añade a self.vlans + todas las interfaces cuyo nombre tenga formato base.VLAN + (ej: enp1s0.10). Esto permite que VLANs creadas en una + sesión anterior se muestren y puedan borrarse. + """ + try: + links = self.ipr.get_links() + except Exception: + return + + for link in links: + ifname = link.get_attr("IFLA_IFNAME") + if not ifname: + continue + + if "." not in ifname: + continue + + base, dot, vlan_str = ifname.partition(".") + try: + vlan_id = int(vlan_str) + except ValueError: + continue + + if vlan_id < 1 or vlan_id > 4094: + continue + + # No machacamos una definición existente en memoria (por si viene de un load_config) + if ifname in self.vlans: + continue + + self.vlans[ifname] = { + "parent": base, + "vlan_id": vlan_id, + "mode": "trunk", + } + + # ------------------------------------------------------------------ + # VLANs y Trunks + # ------------------------------------------------------------------ + + def create_vlan_trunk(self, interface: str, vlan_ids: List[int]) -> Tuple[bool, str]: + """ + Crea múltiples interfaces VLAN tagged sobre una interfaz trunk. + No crea bridges todavía; eso se hará desde otras primitivas. + """ + try: + trunk_idx = self.ipr.link_lookup(ifname=interface)[0] + except IndexError: + return False, f"Interface {interface} not found" + + # Verificar que la interfaz esté UP antes de crear VLANs + try: + link_info = self.ipr.get_links(ifname=interface)[0] + operstate = link_info.get_attr("IFLA_OPERSTATE") + if operstate != "UP": + return False, f"La interfaz {interface} debe estar UP antes de crear VLANs. Estado actual: {operstate}" + except Exception: + pass # Si no podemos verificar el estado, continuamos de todas formas + + created = [] + for vid in vlan_ids: + if vid < 1 or vid > 4094: + return False, f"Invalid VLAN ID {vid}" + + vlan_ifname = f"{interface}.{vid}" + try: + # Si ya existe, simplemente la marcamos como conocida + if self.ipr.link_lookup(ifname=vlan_ifname): + self.vlans[vlan_ifname] = { + "parent": interface, + "vlan_id": vid, + "mode": "trunk", + } + created.append(vlan_ifname) + continue + + # Asegurar que vlan_id es un entero válido en el rango correcto + vlan_id_int = int(vid) + if vlan_id_int < 1 or vlan_id_int > 4094: + return False, f"VLAN ID {vid} fuera de rango válido (1-4094)" + + # Crear la VLAN usando la misma sintaxis que network_manager + self.ipr.link( + "add", + ifname=vlan_ifname, + kind="vlan", + link=trunk_idx, + vlan_id=vlan_id_int, + ) + vlan_idx = self.ipr.link_lookup(ifname=vlan_ifname)[0] + self.ipr.link("set", index=vlan_idx, state="up") + + self.vlans[vlan_ifname] = { + "parent": interface, + "vlan_id": vid, + "mode": "trunk", + } + created.append(vlan_ifname) + except IndexError: + # La interfaz no se creó correctamente + return False, f"No se pudo crear la interfaz VLAN {vlan_ifname}. Verifica que la interfaz base {interface} soporte VLANs." + except Exception as e: + error_msg = str(e) + # Mejorar el mensaje de error para casos comunes + if "34" in error_msg or "out of range" in error_msg.lower() or "ERANGE" in error_msg: + # Obtener lista de interfaces disponibles que podrían soportar VLANs + available_ifaces = [iface["name"] for iface in self.nm.list_interfaces() + if iface["name"] != interface and iface["type"] == "Wired"] + suggestions = "" + if available_ifaces: + suggestions = f"\n\nInterfaces alternativas disponibles que podrían soportar VLANs:\n" + ", ".join(available_ifaces[:5]) + + return False, ( + f"❌ La interfaz '{interface}' NO soporta VLAN tagging 802.1Q.\n\n" + f"Motivo: El driver de esta interfaz (probablemente USB Ethernet) no implementa " + f"soporte para VLANs a nivel de kernel.\n\n" + f"Soluciones:\n" + f"• Usa una interfaz Ethernet tradicional (no USB) como eth0, enp*, etc.\n" + f"• Si necesitas conectar redes, usa la pestaña 'VLAN / Puente' para crear " + f"bridges simples sin VLAN tagging.\n" + f"• Verifica el soporte de VLANs con: 'cat /proc/net/vlan/config'" + f"{suggestions}" + ) + return False, f"Error creating VLAN {vid} on {interface}: {error_msg}" + + return True, f"Created/validated VLANs on {interface}: {', '.join(created)}" + + def create_vlan_access(self, interface: str, vlan_id: int) -> Tuple[bool, str]: + """ + Configura una interfaz como acceso untagged a una VLAN. + A nivel Linux esto normalmente se modela como bridge + VLAN filter. + Aquí simplemente registramos el mapeo lógico; el bridge real se + gestiona en create_bridge_with_vlan_filters(). + """ + if vlan_id < 1 or vlan_id > 4094: + return False, "Invalid VLAN ID" + + self.vlans[interface] = { + "parent": interface, + "vlan_id": vlan_id, + "mode": "access", + } + return True, f"Interface {interface} marked as access for VLAN {vlan_id}" + + def delete_vlan(self, vlan_ifname: str) -> Tuple[bool, str]: + """ + Elimina una interfaz VLAN del sistema y limpia su estado interno. + También limpia redirecciones y configuraciones asociadas. + """ + try: + # Verificar si existe en el sistema + vlan_idx = self.ipr.link_lookup(ifname=vlan_ifname) + if vlan_idx: + # Primero bajar la interfaz + self.ipr.link("set", index=vlan_idx[0], state="down") + # Luego eliminarla + self.ipr.link("del", index=vlan_idx[0]) + + # Limpiar del estado interno + if vlan_ifname in self.vlans: + self.vlans.pop(vlan_ifname) + + # Limpiar redirecciones que usen esta VLAN + self.vlan_redirections = [ + r for r in self.vlan_redirections + if r.get("src_iface") != vlan_ifname + and r.get("dst_iface") != vlan_ifname + and not (r.get("src_vlan") and f"{r.get('src_iface', '')}.{r.get('src_vlan')}" == vlan_ifname) + and not (r.get("dst_vlan") and f"{r.get('dst_iface', '')}.{r.get('dst_vlan')}" == vlan_ifname) + ] + + # Limpiar gateways y DHCP asociados + if vlan_ifname in self.gateways: + self.gateways.pop(vlan_ifname) + if vlan_ifname in self.dhcp_servers: + # Detener DHCP si está corriendo + subprocess.run(["pkill", "-f", f"dnsmasq.*{vlan_ifname}"], check=False) + self.dhcp_servers.pop(vlan_ifname) + + return True, f"VLAN {vlan_ifname} eliminada correctamente" + except IndexError: + # La interfaz no existe en el sistema, pero la limpiamos del estado + if vlan_ifname in self.vlans: + self.vlans.pop(vlan_ifname) + return True, f"VLAN {vlan_ifname} eliminada del estado (no existía en el sistema)" + except Exception as e: + return False, f"Error eliminando VLAN {vlan_ifname}: {str(e)}" + + def redirect_vlan_tagged_to_untagged( + self, trunk_iface: str, vlan_id: int, access_iface: str + ) -> Tuple[bool, str]: + """ + Redirige una VLAN tagged de un trunk a una interfaz de acceso + creando (si es necesario) un bridge y asociando ambos puertos. + """ + return self.redirect_vlan_between_interfaces( + src_iface=trunk_iface, + src_vlan_id=vlan_id, + dst_iface=access_iface, + dst_vlan_id=None, + ) + + def redirect_vlan_untagged_to_tagged( + self, access_iface: str, vlan_id: int, trunk_iface: str + ) -> Tuple[bool, str]: + """ + Variante inversa: tráfico untagged de access_iface se mete en VLAN + concreta sobre el trunk. + """ + # Si la VLAN ya existe, usar método directo sin intentar crear nada + vlan_ifname = f"{trunk_iface}.{vlan_id}" + if self.ipr.link_lookup(ifname=vlan_ifname): + # La VLAN existe, crear bridge directo + return self._create_bridge_between_interfaces(access_iface, vlan_ifname) + else: + # Intentar método genérico + return self.redirect_vlan_between_interfaces( + src_iface=access_iface, + src_vlan_id=None, + dst_iface=trunk_iface, + dst_vlan_id=vlan_id, + ) + + def _create_bridge_between_interfaces( + self, iface1: str, iface2: str, bridge_name: Optional[str] = None + ) -> Tuple[bool, str]: + """ + Crea un bridge simple entre dos interfaces existentes. + No intenta crear VLANs ni nada adicional. + """ + try: + if not bridge_name: + bridge_name = f"br_{iface1.replace('.', '_')}_to_{iface2.replace('.', '_')}" + + # Verificar que ambas interfaces existen + idx1_list = self.ipr.link_lookup(ifname=iface1) + idx2_list = self.ipr.link_lookup(ifname=iface2) + + if not idx1_list: + return False, f"Interfaz '{iface1}' no existe" + if not idx2_list: + return False, f"Interfaz '{iface2}' no existe" + + idx1 = idx1_list[0] + idx2 = idx2_list[0] + + # Crear bridge si no existe + br_list = self.ipr.link_lookup(ifname=bridge_name) + if not br_list: + self.ipr.link("add", ifname=bridge_name, kind="bridge") + + br_idx = self.ipr.link_lookup(ifname=bridge_name)[0] + + # Añadir interfaces al bridge + self.ipr.link("set", index=idx1, master=br_idx) + self.ipr.link("set", index=idx2, master=br_idx) + + # Activar todo + 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} creado entre {iface1} y {iface2}" + except Exception as e: + return False, f"Error creando bridge: {str(e)}" + + def redirect_vlan_between_interfaces( + self, + src_iface: str, + src_vlan_id: Optional[int], + dst_iface: str, + dst_vlan_id: Optional[int], + ) -> Tuple[bool, str]: + """ + Redirección genérica entre interfaces (+/- VLAN). Implementación + basada en bridges: se crea un bridge específico y se conectan + las interfaces/ subinterfaces correspondientes. + + Si la VLAN destino ya existe (ej: enp122s0.10), la usa directamente + sin intentar crearla. Esto permite conectar interfaces que no soportan + VLANs (como USB Ethernet) a VLANs existentes. + """ + if src_iface == dst_iface and src_vlan_id == dst_vlan_id: + return False, "Source and destination cannot be identical" + + try: + # Determinar nombres de interfaces finales + src_ifname = src_iface + dst_ifname = dst_iface + + # Origen: NUNCA intentar crear VLAN si src_vlan_id es None (untagged) + # Si src_vlan_id está especificado, verificar si existe primero + if src_vlan_id is not None: + vlan_ifname = f"{src_iface}.{src_vlan_id}" + # Verificar si ya existe + if self.ipr.link_lookup(ifname=vlan_ifname): + src_ifname = vlan_ifname + else: + # Intentar crear solo si la interfaz base soporta VLANs + # Si falla, usamos la interfaz base directamente + ok, msg = self.create_vlan_trunk(src_iface, [src_vlan_id]) + if ok: + src_ifname = vlan_ifname + # Si falla, dejamos src_ifname = src_iface (untagged) + # No retornamos error aquí, simplemente usamos la base + + # Destino: verificar si la VLAN ya existe ANTES de intentar crearla + if dst_vlan_id is not None: + vlan_ifname = f"{dst_iface}.{dst_vlan_id}" + # Verificar si ya existe + existing_vlan = self.ipr.link_lookup(ifname=vlan_ifname) + if existing_vlan: + # La VLAN ya existe, usarla directamente + dst_ifname = vlan_ifname + else: + # La VLAN no existe, intentar crearla + ok, msg = self.create_vlan_trunk(dst_iface, [dst_vlan_id]) + if ok: + dst_ifname = vlan_ifname + else: + return False, f"No se pudo crear/verificar VLAN destino {vlan_ifname}: {msg}" + + # Si el origen es untagged y el destino es una VLAN existente, + # usar método directo de bridge simple + if src_vlan_id is None and dst_vlan_id is not None: + vlan_ifname = f"{dst_iface}.{dst_vlan_id}" + if self.ipr.link_lookup(ifname=vlan_ifname): + # La VLAN destino existe, usar método directo + ok, msg = self._create_bridge_between_interfaces(src_iface, vlan_ifname) + if ok: + rule = { + "src_iface": src_iface, + "src_vlan": None, + "dst_iface": dst_iface, + "dst_vlan": dst_vlan_id, + "bridge": msg.split(" ")[1] if "Bridge" in msg else f"br_{src_iface}_to_{vlan_ifname.replace('.', '_')}", + } + self.vlan_redirections.append(rule) + return ok, msg + + # Verificar que ambas interfaces existen antes de crear el bridge + try: + src_idx_list = self.ipr.link_lookup(ifname=src_ifname) + dst_idx_list = self.ipr.link_lookup(ifname=dst_ifname) + + if not src_idx_list: + return False, f"Interfaz origen '{src_ifname}' no existe" + if not dst_idx_list: + return False, f"Interfaz destino '{dst_ifname}' no existe" + + src_idx = src_idx_list[0] + dst_idx = dst_idx_list[0] + except (IndexError, Exception) as e: + return False, f"Error verificando interfaces: origen={src_ifname}, destino={dst_ifname}, error={str(e)}" + + # Crear bridge para esta redirección + bridge_name = f"br_{src_ifname.replace('.', '_')}_to_{dst_ifname.replace('.', '_')}" + + # Verificar si el bridge ya existe + existing_bridge = self.ipr.link_lookup(ifname=bridge_name) + if not existing_bridge: + self.ipr.link("add", ifname=bridge_name, kind="bridge") + + br_idx = self.ipr.link_lookup(ifname=bridge_name)[0] + + # Añadir interfaces al bridge + self.ipr.link("set", index=src_idx, master=br_idx) + self.ipr.link("set", index=dst_idx, master=br_idx) + + # Activar todo + self.ipr.link("set", index=br_idx, state="up") + self.ipr.link("set", index=src_idx, state="up") + self.ipr.link("set", index=dst_idx, state="up") + + rule = { + "src_iface": src_iface, + "src_vlan": src_vlan_id, + "dst_iface": dst_iface, + "dst_vlan": dst_vlan_id, + "bridge": bridge_name, + } + self.vlan_redirections.append(rule) + return True, f"Bridge {bridge_name} creado: {src_ifname} → {dst_ifname}" + except Exception as e: + error_str = str(e) + # Mejorar mensaje de error + if "34" in error_str or "ERANGE" in error_str or "out of range" in error_str.lower(): + return False, ( + f"Error: No se pudo crear el bridge. Posibles causas:\n" + f"• La interfaz origen '{src_iface}' no puede ser parte de un bridge con VLANs\n" + f"• La interfaz destino '{dst_iface}' no soporta la operación\n" + f"• Intenta usar la pestaña 'VLAN / Puente' para crear un bridge simple" + ) + return False, f"Error en redirección: {error_str}" + + # ------------------------------------------------------------------ + # Bridges y VLAN filters (esqueleto, se extenderá en tareas siguientes) + # ------------------------------------------------------------------ + + def create_bridge_with_vlan_filters( + self, bridge_name: str, interfaces: List[str], vlan_filters: List[Dict] + ) -> Tuple[bool, str]: + """ + Crea un bridge y aplica filtros VLAN en sus puertos. + vlan_filters: lista de dicts con claves: + { 'iface': str, 'vid': int, 'pvid': bool, 'untagged': bool } + """ + try: + if not self.ipr.link_lookup(ifname=bridge_name): + self.ipr.link("add", ifname=bridge_name, kind="bridge") + + br_idx = self.ipr.link_lookup(ifname=bridge_name)[0] + + for iface in interfaces: + idx = self.ipr.link_lookup(ifname=iface)[0] + self.ipr.link("set", index=idx, master=br_idx) + self.ipr.link("set", index=idx, state="up") + + self.ipr.link("set", index=br_idx, state="up") + + # Aplicar filtros VLAN sobre cada puerto + for vf in vlan_filters: + try: + port_idx = self.ipr.link_lookup(ifname=vf["iface"])[0] + vid = int(vf["vid"]) + flags = [] + if vf.get("pvid"): + flags.append("pvid") + if vf.get("untagged"): + flags.append("untagged") + + vlan_info = {"vid": vid} + if flags: + vlan_info["flags"] = flags + + # pyroute2 expone vlan_filter desde ipr + self.ipr.vlan_filter( + "add", + index=port_idx, + vlan_info=vlan_info, + ) + except Exception: + # No abortamos por un fallo puntual de filtro; se puede depurar luego + continue + + return True, f"Bridge {bridge_name} created with VLAN filters" + except Exception as e: + return False, str(e) + + def add_vlan_filter_to_bridge_port( + self, bridge_name: str, port_iface: str, vlan_id: int, pvid: bool = False, untagged: bool = False + ) -> Tuple[bool, str]: + try: + if not self.ipr.link_lookup(ifname=bridge_name): + return False, f"Bridge {bridge_name} not found" + + port_idx = self.ipr.link_lookup(ifname=port_iface)[0] + flags = [] + if pvid: + flags.append("pvid") + if untagged: + flags.append("untagged") + vlan_info = {"vid": int(vlan_id)} + if flags: + vlan_info["flags"] = flags + + self.ipr.vlan_filter("add", index=port_idx, vlan_info=vlan_info) + return True, "VLAN filter added" + except Exception as e: + return False, str(e) + + def remove_vlan_filter_from_bridge_port( + self, bridge_name: str, port_iface: str, vlan_id: int + ) -> Tuple[bool, str]: + try: + if not self.ipr.link_lookup(ifname=bridge_name): + return False, f"Bridge {bridge_name} not found" + + port_idx = self.ipr.link_lookup(ifname=port_iface)[0] + self.ipr.vlan_filter("del", index=port_idx, vlan_info={"vid": int(vlan_id)}) + return True, "VLAN filter removed" + except Exception as e: + return False, str(e) + + # ------------------------------------------------------------------ + # DHCP y Gateways + # ------------------------------------------------------------------ + + def start_dhcp_server_for_vlan( + self, + vlan_interface: str, + ip: str, + range_start: str, + range_end: str, + gateway: Optional[str] = None, + dns: Optional[str] = None, + ) -> Tuple[bool, str]: + """ + Envuelve a NetworkManager.start_dhcp_server y además registra gateway/dns. + """ + ok, msg = self.nm.start_dhcp_server(vlan_interface, ip, range_start, range_end) + if not ok: + return ok, msg + + self.dhcp_servers[vlan_interface] = { + "ip": ip, + "range_start": range_start, + "range_end": range_end, + } + + if gateway: + self.set_gateway_for_network(vlan_interface, ip, gateway) + if dns: + self.gateways.setdefault(vlan_interface, {})["dns"] = dns + + return True, msg + + def stop_dhcp_server_for_vlan(self, vlan_interface: str) -> Tuple[bool, str]: + try: + subprocess.run(["pkill", "-f", f"dnsmasq.*{vlan_interface}"], check=False) + self.dhcp_servers.pop(vlan_interface, None) + return True, f"DHCP server on {vlan_interface} stopped" + except Exception as e: + return False, str(e) + + def list_active_dhcp_servers(self) -> Dict[str, Dict]: + return dict(self.dhcp_servers) + + def set_gateway_for_network(self, interface: str, ip: str, gateway: str) -> Tuple[bool, str]: + self.gateways[interface] = {"ip": ip, "gateway": gateway} + # A nivel de sistema, añadimos/actualizamos ruta por defecto para esa red + try: + self.add_route(dst="0.0.0.0/0", gateway=gateway, interface=interface) + return True, f"Gateway {gateway} set for {interface}" + except Exception as e: + return False, str(e) + + def remove_gateway_for_network(self, interface: str, ip: str) -> Tuple[bool, str]: + if interface in self.gateways: + gw = self.gateways[interface].get("gateway") + self.gateways.pop(interface, None) + if gw: + try: + self.delete_route(dst="0.0.0.0/0", gateway=gw, interface=interface) + except Exception: + pass + return True, f"Gateway removed for {interface}" + + def list_gateways(self) -> Dict[str, Dict]: + return dict(self.gateways) + + # ------------------------------------------------------------------ + # Routing y tablas + # ------------------------------------------------------------------ + + def add_route( + self, + dst: str, + gateway: str, + interface: str, + table: int = 254, + metric: Optional[int] = None, + ) -> Tuple[bool, str]: + try: + idx = self.ipr.link_lookup(ifname=interface)[0] + dst_ip, prefix = dst.split("/") + attrs = [("OIF", idx), ("GATEWAY", gateway)] + self.ipr.route( + "add", + dst=dst_ip, + mask=int(prefix), + table=table, + metrics={"metric": metric} if metric is not None else None, + attrs=attrs, + ) + self.routes.append( + { + "dst": dst, + "gateway": gateway, + "interface": interface, + "table": table, + "metric": metric, + } + ) + return True, "Route added" + except Exception as e: + return False, str(e) + + def delete_route( + self, + dst: str, + gateway: str, + interface: str, + table: int = 254, + ) -> Tuple[bool, str]: + try: + idx = self.ipr.link_lookup(ifname=interface)[0] + dst_ip, prefix = dst.split("/") + self.ipr.route( + "del", + dst=dst_ip, + mask=int(prefix), + table=table, + attrs=[("OIF", idx), ("GATEWAY", gateway)], + ) + self.routes = [ + r + for r in self.routes + if not ( + r["dst"] == dst + and r["gateway"] == gateway + and r["interface"] == interface + and r["table"] == table + ) + ] + return True, "Route deleted" + except Exception as e: + return False, str(e) + + def add_routing_rule( + self, + table: int, + priority: int, + src: Optional[str] = None, + dst: Optional[str] = None, + iifname: Optional[str] = None, + oifname: Optional[str] = None, + ) -> Tuple[bool, str]: + try: + kwargs = {"table": table, "priority": priority} + if src: + ip, plen = src.split("/") + kwargs["src"] = ip + kwargs["src_len"] = int(plen) + if dst: + ip, plen = dst.split("/") + kwargs["dst"] = ip + kwargs["dst_len"] = int(plen) + if iifname: + kwargs["iifname"] = iifname + if oifname: + kwargs["oifname"] = oifname + + self.ipr.rule("add", **kwargs) + self.routing_rules.append(kwargs) + return True, "Routing rule added" + except Exception as e: + return False, str(e) + + def list_routes(self, table: Optional[int] = None) -> List[Dict]: + if table is None: + return list(self.routes) + return [r for r in self.routes if r["table"] == table] + + def enable_ip_forwarding(self) -> Tuple[bool, str]: + try: + with open("/proc/sys/net/ipv4/ip_forward", "w") as f: + f.write("1") + return True, "IP forwarding enabled" + except Exception as e: + return False, str(e) + + # ------------------------------------------------------------------ + # NAT y Port Forwarding + # ------------------------------------------------------------------ + + def detect_firewall_backend(self) -> str: + """ + Devuelve 'nftables', 'iptables' o 'none'. + """ + from shutil import which + + if which("nft"): + return "nftables" + if which("iptables"): + return "iptables" + return "none" + + def _run_iptables(self, args: List[str]) -> None: + subprocess.run(["iptables"] + args, check=True) + + def _run_nft(self, args: List[str]) -> None: + subprocess.run(["nft"] + args, check=True) + + def enable_nat_for_interface(self, interface: str, external_iface: str) -> Tuple[bool, str]: + """ + Habilita SNAT/MASQUERADE para que las redes internas salgan por external_iface. + Implementación mínima usando iptables o nftables. + """ + try: + if self.firewall_backend == "iptables": + # Habilitar forwarding + self._run_iptables(["-A", "FORWARD", "-i", interface, "-o", external_iface, "-j", "ACCEPT"]) + self._run_iptables(["-A", "FORWARD", "-i", external_iface, "-o", interface, "-m", "state", "--state", "ESTABLISHED,RELATED", "-j", "ACCEPT"]) + # NAT + self._run_iptables(["-t", "nat", "-A", "POSTROUTING", "-o", external_iface, "-j", "MASQUERADE"]) + elif self.firewall_backend == "nftables": + # Asumimos tabla inet nat preexistente o intentamos crear algo simple + self._run_nft(["add", "table", "inet", "jnc_nat"]) + self._run_nft(["add", "chain", "inet", "jnc_nat", "postrouting", "{", "type", "nat", "hook", "postrouting", "priority", "100", ";", "}"]) + self._run_nft( + [ + "add", + "rule", + "inet", + "jnc_nat", + "postrouting", + "oifname", + external_iface, + "masquerade", + ] + ) + else: + return False, "No firewall backend (iptables/nftables) available" + + self.nat_rules.append({"interface": interface, "external": external_iface}) + return True, "NAT enabled" + except Exception as e: + return False, str(e) + + def disable_nat_for_interface(self, interface: str) -> Tuple[bool, str]: + # Implementación sencilla: no sabemos exactamente qué reglas borrar, + # así que por ahora dejamos registro lógico y se puede extender. + self.nat_rules = [r for r in self.nat_rules if r["interface"] != interface] + return True, "NAT rules removed from internal state (system rules may persist)" + + def add_port_forwarding( + self, + external_ip: str, + external_port: int, + internal_ip: str, + internal_port: int, + protocol: str = "tcp", + ) -> Tuple[bool, str]: + try: + if self.firewall_backend == "iptables": + self._run_iptables( + [ + "-t", + "nat", + "-A", + "PREROUTING", + "-p", + protocol, + "-d", + external_ip, + "--dport", + str(external_port), + "-j", + "DNAT", + "--to-destination", + f"{internal_ip}:{internal_port}", + ] + ) + elif self.firewall_backend == "nftables": + self._run_nft(["add", "table", "inet", "jnc_nat"]) + self._run_nft(["add", "chain", "inet", "jnc_nat", "prerouting", "{", "type", "nat", "hook", "prerouting", "priority", "0", ";", "}"]) + self._run_nft( + [ + "add", + "rule", + "inet", + "jnc_nat", + "prerouting", + "ip", + "daddr", + external_ip, + "tcp", + "dport", + str(external_port), + "dnat", + "to", + f"{internal_ip}:{internal_port}", + ] + ) + else: + return False, "No firewall backend (iptables/nftables) available" + + rule = { + "external_ip": external_ip, + "external_port": external_port, + "internal_ip": internal_ip, + "internal_port": internal_port, + "protocol": protocol, + } + self.nat_rules.append(rule) + return True, "Port forwarding rule added" + except Exception as e: + return False, str(e) + + # ------------------------------------------------------------------ + # Firewall / Visibilidad entre VLANs + # ------------------------------------------------------------------ + + def add_firewall_rule( + self, + chain: str, + action: str, + src_network: Optional[str] = None, + dst_network: Optional[str] = None, + protocol: Optional[str] = None, + src_port: Optional[int] = None, + dst_port: Optional[int] = None, + ) -> Tuple[bool, str]: + """ + Regla genérica simple sobre FORWARD/INPUT/OUTPUT. + """ + try: + if self.firewall_backend == "iptables": + args = ["-A", chain] + if src_network: + args += ["-s", src_network] + if dst_network: + args += ["-d", dst_network] + if protocol: + args += ["-p", protocol] + if src_port: + args += ["--sport", str(src_port)] + if dst_port: + args += ["--dport", str(dst_port)] + args += ["-j", action] + self._run_iptables(args) + elif self.firewall_backend == "nftables": + # Implementación básica centrada en cadena forward de tabla inet filter + self._run_nft(["add", "table", "inet", "jnc_filter"]) + self._run_nft(["add", "chain", "inet", "jnc_filter", "forward", "{", "type", "filter", "hook", "forward", "priority", "0", ";", "}"]) + rule_parts = ["add", "rule", "inet", "jnc_filter", "forward"] + if src_network: + rule_parts += ["ip", "saddr", src_network] + if dst_network: + rule_parts += ["ip", "daddr", dst_network] + if protocol: + rule_parts.append(protocol) + if dst_port: + rule_parts += ["dport", str(dst_port)] + rule_parts.append(action.lower()) + self._run_nft(rule_parts) + else: + return False, "No firewall backend (iptables/nftables) available" + + rule = { + "chain": chain, + "action": action, + "src": src_network, + "dst": dst_network, + "protocol": protocol, + "src_port": src_port, + "dst_port": dst_port, + } + self.firewall_rules.append(rule) + return True, "Firewall rule added" + except Exception as e: + return False, str(e) + + def block_vlan_to_vlan(self, src_vlan_id: int, dst_vlan_id: int) -> Tuple[bool, str]: + src_net = f"10.{src_vlan_id}.0.0/16" + dst_net = f"10.{dst_vlan_id}.0.0/16" + ok, msg = self.add_firewall_rule( + chain="FORWARD", + action="DROP", + src_network=src_net, + dst_network=dst_net, + ) + if ok: + self.vlan_visibility[(src_vlan_id, dst_vlan_id)] = False + return ok, msg + + def allow_vlan_to_vlan(self, src_vlan_id: int, dst_vlan_id: int) -> Tuple[bool, str]: + src_net = f"10.{src_vlan_id}.0.0/16" + dst_net = f"10.{dst_vlan_id}.0.0/16" + ok, msg = self.add_firewall_rule( + chain="FORWARD", + action="ACCEPT", + src_network=src_net, + dst_network=dst_net, + ) + if ok: + self.vlan_visibility[(src_vlan_id, dst_vlan_id)] = True + return ok, msg + + def set_vlan_isolation_mode(self, vlan_id: int, isolated: bool = True) -> Tuple[bool, str]: + """ + Implementación mínima: registra el estado lógico. Las reglas reales + se pueden derivar en la UI según las necesidades del despliegue. + """ + if isolated: + # Bloqueamos VLAN contra todas las demás de forma lógica + for other in range(1, 4095): + if other == vlan_id: + continue + self.vlan_visibility[(vlan_id, other)] = False + self.vlan_visibility[(other, vlan_id)] = False + return True, "VLAN marked as isolated (logical state)" + else: + # Permitimos de nuevo (sólo borra del estado interno) + self.vlan_visibility = { + k: v + for k, v in self.vlan_visibility.items() + if vlan_id not in k + } + return True, "VLAN isolation disabled in logical state" + + def list_firewall_rules(self) -> List[Dict]: + return list(self.firewall_rules) + + # ------------------------------------------------------------------ + # Gestión de estado + # ------------------------------------------------------------------ + + def get_routing_status(self) -> Dict: + return { + "vlans": self.vlans, + "vlan_redirections": self.vlan_redirections, + "dhcp_servers": self.dhcp_servers, + "gateways": self.gateways, + "routes": self.routes, + "routing_rules": self.routing_rules, + "nat_rules": self.nat_rules, + "firewall_rules": self.firewall_rules, + "vlan_visibility": {f"{a}-{b}": allowed for (a, b), allowed in self.vlan_visibility.items()}, + } + + def save_routing_config(self, filename: str) -> Tuple[bool, str]: + try: + data = self.get_routing_status() + with open(filename, "w") as f: + json.dump(data, f, indent=2) + return True, filename + except Exception as e: + return False, str(e) + + def load_routing_config(self, filename: str) -> Tuple[bool, str]: + try: + with open(filename, "r") as f: + data = json.load(f) + # Sólo restauramos estado en memoria; aplicar en sistema sería más complejo + self.vlans = data.get("vlans", {}) + self.vlan_redirections = data.get("vlan_redirections", []) + self.dhcp_servers = data.get("dhcp_servers", {}) + self.gateways = data.get("gateways", {}) + self.routes = data.get("routes", []) + self.routing_rules = data.get("routing_rules", []) + self.nat_rules = data.get("nat_rules", []) + self.firewall_rules = data.get("firewall_rules", []) + vis = data.get("vlan_visibility", {}) + self.vlan_visibility = {} + for key, allowed in vis.items(): + try: + a_str, b_str = key.split("-") + self.vlan_visibility[(int(a_str), int(b_str))] = bool(allowed) + except Exception: + continue + return True, "Routing configuration loaded into memory (no live apply)" + except Exception as e: + return False, str(e) + diff --git a/src/core/network_manager.py b/src/core/network_manager.py index bf4f726..55eebbe 100644 --- a/src/core/network_manager.py +++ b/src/core/network_manager.py @@ -4,6 +4,8 @@ import struct import psutil import subprocess import json +import time +from typing import Tuple from pyroute2 import IPRoute class NetworkManager: @@ -370,5 +372,30 @@ rsn_pairwise=CCMP except Exception as e: return False, str(e) + def reset_interface(self, iface_name: str) -> Tuple[bool, str]: + """ + Reinicia una interfaz de red haciendo down y luego up. + Útil cuando una interfaz se queda "tonta" y necesita reiniciarse. + """ + try: + iface_idx = self.ipr.link_lookup(ifname=iface_name) + if not iface_idx: + return False, f"Interfaz {iface_name} no encontrada" + + idx = iface_idx[0] + + # Bajar la interfaz + self.ipr.link("set", index=idx, state="down") + + # Pequeña pausa para asegurar que el down se procesa + time.sleep(0.5) + + # Subir la interfaz + self.ipr.link("set", index=idx, state="up") + + return True, f"Interfaz {iface_name} reiniciada correctamente (down → up)" + except Exception as e: + return False, f"Error al reiniciar interfaz {iface_name}: {str(e)}" + def close(self): self.ipr.close() diff --git a/src/core/vlan35_checker.py b/src/core/vlan35_checker.py new file mode 100644 index 0000000..cd0e8fa --- /dev/null +++ b/src/core/vlan35_checker.py @@ -0,0 +1,228 @@ +import socket +import subprocess +import time +from datetime import datetime +from typing import Dict, List, Tuple, Optional +from PyQt6.QtCore import QThread, pyqtSignal + + +class ConnectivityTestThread(QThread): + """Thread para ejecutar tests de conectividad en paralelo.""" + test_result = pyqtSignal(dict) # Emite resultado de cada test + all_finished = pyqtSignal() # Emite cuando todos terminan + + def __init__(self, tests: List[Dict]): + super().__init__() + self.tests = tests + self.results = [] + + def run(self): + for test in self.tests: + result = self._run_test(test) + self.results.append(result) + self.test_result.emit(result) + self.all_finished.emit() + + def _run_test(self, test: Dict) -> Dict: + """Ejecuta un test individual y retorna resultado.""" + host = test["host"] + port = test["port"] + protocol = test.get("protocol", "TCP") + test_type = test.get("type", "connectivity") + + result = { + "host": host, + "port": port, + "protocol": protocol, + "type": test_type, + "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), + "success": False, + "status": "Unknown", + "latency_ms": None, + "error": None, + "details": "", + } + + try: + if protocol == "TCP": + result.update(self._test_tcp(host, port)) + elif protocol == "UDP": + if test_type == "ntp": + result.update(self._test_ntp(host)) + else: + result.update(self._test_udp(host, port)) + else: + result["error"] = f"Protocolo desconocido: {protocol}" + except Exception as e: + result["error"] = str(e) + result["status"] = "Error" + + return result + + def _test_tcp(self, host: str, port: int) -> Dict: + """Test de conectividad TCP.""" + start_time = time.time() + try: + sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + sock.settimeout(5) + result = sock.connect_ex((host, port)) + latency = (time.time() - start_time) * 1000 + sock.close() + + if result == 0: + return { + "success": True, + "status": "Open", + "latency_ms": round(latency, 2), + "details": f"Conexión TCP exitosa en {latency:.2f}ms", + } + else: + return { + "success": False, + "status": "Closed/Filtered", + "latency_ms": None, + "details": f"No se pudo conectar (código: {result})", + } + except socket.timeout: + return { + "success": False, + "status": "Timeout", + "latency_ms": None, + "details": "Timeout esperando respuesta", + } + except Exception as e: + return { + "success": False, + "status": "Error", + "latency_ms": None, + "error": str(e), + "details": f"Error: {str(e)}", + } + + def _test_udp(self, host: str, port: int) -> Dict: + """Test básico UDP (envía paquete y espera respuesta).""" + try: + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.settimeout(3) + start_time = time.time() + + # Enviar paquete vacío + sock.sendto(b"", (host, port)) + + try: + data, addr = sock.recvfrom(1024) + latency = (time.time() - start_time) * 1000 + sock.close() + return { + "success": True, + "status": "Open", + "latency_ms": round(latency, 2), + "details": f"Respuesta UDP recibida en {latency:.2f}ms", + } + except socket.timeout: + sock.close() + return { + "success": False, + "status": "Open|Filtered", + "latency_ms": None, + "details": "No respuesta (puerto puede estar abierto pero filtrado)", + } + except Exception as e: + return { + "success": False, + "status": "Error", + "latency_ms": None, + "error": str(e), + "details": f"Error: {str(e)}", + } + + def _test_ntp(self, host: str) -> Dict: + """Test NTP específico (puerto UDP 123).""" + try: + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.settimeout(5) + start_time = time.time() + + # Paquete NTP básico (modo 3: cliente) + ntp_packet = bytearray(48) + ntp_packet[0] = 0x1B # LI=0, VN=3, Mode=3 + + sock.sendto(ntp_packet, (host, 123)) + + try: + data, addr = sock.recvfrom(1024) + latency = (time.time() - start_time) * 1000 + + if len(data) >= 48: + # Extraer timestamp NTP + try: + seconds = ( + (data[40] << 24) + | (data[41] << 16) + | (data[42] << 8) + | data[43] + ) + # NTP epoch: 1900-01-01 + ntp_epoch = datetime(1900, 1, 1) + ntp_time = ntp_epoch.timestamp() + seconds + ntp_dt = datetime.fromtimestamp(ntp_time) + time_str = ntp_dt.strftime('%Y-%m-%d %H:%M:%S') + except (ValueError, OverflowError, OSError): + time_str = "Timestamp no válido" + + sock.close() + return { + "success": True, + "status": "Open", + "latency_ms": round(latency, 2), + "details": f"NTP hora local: {time_str}", + } + else: + sock.close() + return { + "success": False, + "status": "Error", + "latency_ms": None, + "details": "Respuesta NTP incompleta", + } + except socket.timeout: + sock.close() + return { + "success": False, + "status": "Timeout", + "latency_ms": None, + "details": "Timeout esperando respuesta NTP", + } + except Exception as e: + return { + "success": False, + "status": "Error", + "latency_ms": None, + "error": str(e), + "details": f"Error NTP: {str(e)}", + } + + +class VLAN35Checker: + """Gestor de tests de conectividad para VLAN 35.""" + + def __init__(self): + # Configuración por defecto basada en el script .bat + self.default_tests = [ + {"host": "10.47.8.133", "port": 515, "protocol": "TCP", "type": "connectivity"}, + {"host": "10.47.8.133", "port": 9195, "protocol": "TCP", "type": "connectivity"}, + {"host": "10.47.8.135", "port": 9191, "protocol": "TCP", "type": "connectivity"}, + {"host": "10.47.8.135", "port": 9192, "protocol": "TCP", "type": "connectivity"}, + {"host": "10.240.225.171", "port": 465, "protocol": "TCP", "type": "connectivity"}, + {"host": "10.244.255.250", "port": 123, "protocol": "UDP", "type": "ntp"}, + ] + + def get_default_tests(self) -> List[Dict]: + """Retorna la lista de tests por defecto.""" + return self.default_tests.copy() + + def run_tests(self, tests: List[Dict]) -> ConnectivityTestThread: + """Inicia los tests en un thread.""" + thread = ConnectivityTestThread(tests) + thread.start() + return thread diff --git a/src/gui/__pycache__/main_window.cpython-312.pyc b/src/gui/__pycache__/main_window.cpython-312.pyc index 941de87..3ec6bef 100644 Binary files a/src/gui/__pycache__/main_window.cpython-312.pyc 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 index b99b5d9..4141218 100644 Binary files a/src/gui/__pycache__/styles.cpython-312.pyc and b/src/gui/__pycache__/styles.cpython-312.pyc differ diff --git a/src/gui/main_window.py b/src/gui/main_window.py index 01576c8..7c349a6 100644 --- a/src/gui/main_window.py +++ b/src/gui/main_window.py @@ -1,5 +1,13 @@ -from PyQt6.QtWidgets import (QMainWindow, QWidget, QVBoxLayout, QTabWidget, - QLabel, QHBoxLayout, QStatusBar) +from PyQt6.QtWidgets import ( + QMainWindow, + QWidget, + QVBoxLayout, + QTabWidget, + QLabel, + QHBoxLayout, + QStatusBar, + QFrame, +) from PyQt6.QtCore import Qt from src.core.network_manager import NetworkManager from src.gui.widgets.interface_list import InterfaceListWidget @@ -9,65 +17,118 @@ 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.gui.widgets.advanced_routing import AdvancedRoutingWidget +from src.gui.widgets.network_monitor import NetworkMonitorWidget +from src.gui.widgets.vlan35_check import VLAN35CheckWidget 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.setWindowTitle("JNC-NetTools · ADA - Agencia Digital de Andalucía") + self.setMinimumSize(1100, 720) # Tamaño mínimo algo más amplio self.nm = NetworkManager() - # Central Widget + # Central Widget y layout principal vertical central_widget = QWidget() self.setCentralWidget(central_widget) - main_layout = QHBoxLayout(central_widget) - - # Sidebar (Interface List) + outer_layout = QVBoxLayout(central_widget) + outer_layout.setContentsMargins(14, 10, 14, 10) + outer_layout.setSpacing(10) + + # Encabezado corporativo superior + header = QFrame() + header.setObjectName("HeaderBar") + header_layout = QHBoxLayout(header) + header_layout.setContentsMargins(12, 8, 12, 8) + + title_block = QVBoxLayout() + title_label = QLabel("JNC-NetTools") + title_label.setObjectName("HeaderTitle") + subtitle_label = QLabel("Suite de diagnóstico y exploración de red · ADA") + subtitle_label.setObjectName("HeaderSubtitle") + subtitle_label.setAlignment(Qt.AlignmentFlag.AlignVCenter | Qt.AlignmentFlag.AlignLeft) + title_block.addWidget(title_label) + title_block.addWidget(subtitle_label) + + header_layout.addLayout(title_block) + header_layout.addStretch() + + org_label = QLabel("Agencia Digital de Andalucía · Junta de Andalucía") + org_label.setObjectName("HeaderOrg") + header_layout.addWidget(org_label) + + outer_layout.addWidget(header) + + # Separador visual + separator = QFrame() + separator.setFrameShape(QFrame.Shape.HLine) + separator.setFrameShadow(QFrame.Shadow.Sunken) + outer_layout.addWidget(separator) + + # Área de contenido principal (sidebar + pestañas) + main_layout = QHBoxLayout() + main_layout.setSpacing(12) + outer_layout.addLayout(main_layout) + + # Sidebar (lista de interfaces) 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) + # Área principal de pestañas self.tabs = QTabWidget() main_layout.addWidget(self.tabs) - # IP Configuration Tab + # Pestaña Configuración IP self.ip_config_widget = IPConfigWidget(self.nm) - self.tabs.addTab(self.ip_config_widget, "IP Configuration") + self.tabs.addTab(self.ip_config_widget, "Configuración IP") - # VLAN & Bridge Tab + # Pestaña VLAN & Bridge self.vlan_bridge_widget = VLANBridgeWidget(self.nm) - self.tabs.addTab(self.vlan_bridge_widget, "VLAN / Bridge") + self.tabs.addTab(self.vlan_bridge_widget, "VLAN / Puente") - # Scanner Tab (Network Discovery) + # Pestaña de escaneo de red self.scanner_widget = ScannerWidget() - self.tabs.addTab(self.scanner_widget, "Net Scanner") + self.tabs.addTab(self.scanner_widget, "Exploración de Red") - # Port Scanner Tab (Dedicated) + # Pestaña de escáner de puertos self.port_scanner_widget = PortScannerWidget() - self.tabs.addTab(self.port_scanner_widget, "Port Scanner") + self.tabs.addTab(self.port_scanner_widget, "Escáner de Puertos") - # Sniffer Tab + # Pestaña Sniffer self.sniffer_widget = SnifferWidget(self.nm) - self.tabs.addTab(self.sniffer_widget, "Sniffer") + self.tabs.addTab(self.sniffer_widget, "Sniffer de Tráfico") - # Diagnostics Tab + # Pestaña Diagnóstico self.diagnostics_widget = DiagnosticsWidget() - self.tabs.addTab(self.diagnostics_widget, "Diagnostics") + self.tabs.addTab(self.diagnostics_widget, "Diagnóstico") + + # Pestaña Monitor de Red + self.network_monitor_widget = NetworkMonitorWidget(self.nm) + self.tabs.addTab(self.network_monitor_widget, "Monitor de Red") + + # Pestaña Routing Avanzado + self.advanced_routing_widget = AdvancedRoutingWidget(self.nm) + self.tabs.addTab(self.advanced_routing_widget, "Routing Avanzado") + + # Pestaña Comprobación VLAN35 + self.vlan35_check_widget = VLAN35CheckWidget() + self.tabs.addTab(self.vlan35_check_widget, "Comprobación VLAN35") - # 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) + # Barra de herramientas para acciones globales + toolbar = self.addToolBar("Principal") + toolbar.setMovable(False) + toolbar.setFloatable(False) + toolbar.addAction("Generar informe HTML", self.generate_report) - # Status Bar + # Barra de estado self.status_bar = QStatusBar() self.setStatusBar(self.status_bar) - self.status_bar.showMessage("Ready") + self.status_bar.showMessage("Listo") def on_interface_selected(self, iface_data): self.status_bar.showMessage(f"Selected Interface: {iface_data['name']}") @@ -78,6 +139,7 @@ class MainWindow(QMainWindow): self.scanner_widget.close() self.port_scanner_widget.close() self.vlan_bridge_widget.close() + # advanced_routing_widget no mantiene hilos propios, sólo estado lógico event.accept() def generate_report(self): diff --git a/src/gui/styles.py b/src/gui/styles.py index e44c519..f839b9e 100644 --- a/src/gui/styles.py +++ b/src/gui/styles.py @@ -1,20 +1,25 @@ from PyQt6.QtWidgets import QApplication from PyQt6.QtGui import QPalette, QColor +from PyQt6.QtCore import Qt + def apply_styles(app: QApplication): - """Applies a premium dark theme to the application.""" + """ + Aplica un tema oscuro corporativo inspirado en la identidad visual + de la Agencia Digital de Andalucía / Junta de Andalucía. + """ 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) - + + # Colores corporativos aproximados ADA / Junta de Andalucía + dark_bg = QColor(18, 18, 18) + darker_bg = QColor(10, 10, 10) + text_color = QColor(235, 235, 235) + accent_color = QColor(0, 133, 66) # Verde principal corporativo + accent_hover = QColor(0, 153, 80) + disabled_text = QColor(130, 130, 130) + palette.setColor(QPalette.ColorRole.Window, dark_bg) palette.setColor(QPalette.ColorRole.WindowText, text_color) palette.setColor(QPalette.ColorRole.Base, darker_bg) @@ -28,29 +33,28 @@ def apply_styles(app: QApplication): 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 + + # Stylesheet para dar un aspecto más “suite corporativa” app.setStyleSheet(""" QMainWindow { - background-color: #181818; + background-color: #101010; } QWidget { - font-family: 'Segoe UI', 'Roboto', sans-serif; + font-family: 'Segoe UI', 'Roboto', 'Noto Sans', sans-serif; font-size: 13px; - color: #e0e0e0; + color: #e5e5e5; } QTabWidget::pane { - border: 1px solid #333333; - background: #252526; + border: 1px solid #2b2b2b; + background: #181818; border-radius: 4px; top: -1px; } QTabBar::tab { - background: #1e1e1e; - color: #999999; + background: #151515; + color: #aaaaaa; padding: 10px 20px; border-top-left-radius: 4px; border-top-right-radius: 4px; @@ -59,17 +63,17 @@ def apply_styles(app: QApplication): border: 1px solid transparent; } QTabBar::tab:selected { - background: #252526; + background: #181818; color: #ffffff; - border-top: 2px solid #007acc; - border-bottom: 1px solid #252526; + border-top: 2px solid #008542; + border-bottom: 1px solid #181818; } QTabBar::tab:hover { - background: #2d2d2d; + background: #202020; color: #ffffff; } QPushButton { - background-color: #007acc; + background-color: #008542; color: white; border: none; padding: 8px 16px; @@ -77,89 +81,110 @@ def apply_styles(app: QApplication): font-weight: 600; } QPushButton:hover { - background-color: #0062a3; + background-color: #009b4f; } QPushButton:pressed { - background-color: #004c80; + background-color: #006f39; } QPushButton:disabled { - background-color: #333333; + background-color: #2a2a2a; color: #666666; } - QLineEdit, QComboBox, QSpinBox { - background-color: #333333; - border: 1px solid #444444; - color: #f0f0f0; - padding: 6px; + QLineEdit, QComboBox, QSpinBox, QDoubleSpinBox { + background-color: #232323; + border: 1px solid #3a3a3a; + color: #f5f5f5; + padding: 4px 6px; border-radius: 3px; + min-height: 22px; } - QLineEdit:focus, QComboBox:focus, QSpinBox:focus { - border: 1px solid #007acc; - background-color: #3c3c3c; + QLineEdit:focus, QComboBox:focus, QSpinBox:focus, QDoubleSpinBox:focus { + border: 1px solid #008542; + background-color: #272727; + } + /* Botones de los spinbox: que se vean siempre claros */ + QSpinBox::up-button, QDoubleSpinBox::up-button, + QSpinBox::down-button, QDoubleSpinBox::down-button { + background-color: #303030; + border: 1px solid #3a3a3a; + width: 16px; + } + QSpinBox::up-button:hover, QDoubleSpinBox::up-button:hover, + QSpinBox::down-button:hover, QDoubleSpinBox::down-button:hover { + background-color: #3a3a3a; + } + QSpinBox::up-arrow, QDoubleSpinBox::up-arrow, + QSpinBox::down-arrow, QDoubleSpinBox::down-arrow { + width: 8px; + height: 8px; + } + /* Formularios más compactos */ + QFormLayout QLabel { + min-width: 110px; + } + QGroupBox { + margin-top: 12px; + padding-top: 10px; } QTableWidget { - background-color: #1e1e1e; - border: 1px solid #333333; + background-color: #141414; + border: 1px solid #2b2b2b; border-radius: 4px; gridline-color: #333333; - selection-background-color: #264f78; + selection-background-color: #184b33; selection-color: white; - alternate-background-color: #252526; + alternate-background-color: #181818; } QHeaderView::section { - background-color: #252526; - color: #cccccc; + background-color: #181818; + color: #dddddd; padding: 6px; border: none; - border-right: 1px solid #333333; - border-bottom: 1px solid #333333; + border-right: 1px solid #2b2b2b; + border-bottom: 1px solid #2b2b2b; font-weight: 600; } QListWidget { - background-color: #1e1e1e; - border: 1px solid #333333; + background-color: #141414; + border: 1px solid #2b2b2b; border-radius: 4px; } QListWidget::item { padding: 8px; - border-bottom: 1px solid #252526; + border-bottom: 1px solid #202020; } QListWidget::item:selected { - background-color: #37373d; + background-color: #184b33; color: white; - border-left: 2px solid #007acc; + border-left: 2px solid #00a357; } QGroupBox { - border: 1px solid #444444; + border: 1px solid #3a3a3a; 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; + color: #00a357; left: 10px; } QStatusBar { - background-color: #007acc; + background-color: #101010; color: white; } QProgressBar { - border: 1px solid #444444; + border: 1px solid #3a3a3a; border-radius: 4px; text-align: center; - background-color: #333333; + background-color: #232323; } QProgressBar::chunk { - background-color: #007acc; + background-color: #00a357; width: 10px; } QSplitter::handle { - background-color: #333333; + background-color: #2b2b2b; } """) - -from PyQt6.QtCore import Qt diff --git a/src/gui/widgets/__pycache__/advanced_routing.cpython-312.pyc b/src/gui/widgets/__pycache__/advanced_routing.cpython-312.pyc new file mode 100644 index 0000000..fa6fb95 Binary files /dev/null and b/src/gui/widgets/__pycache__/advanced_routing.cpython-312.pyc differ diff --git a/src/gui/widgets/__pycache__/diagnostics_view.cpython-312.pyc b/src/gui/widgets/__pycache__/diagnostics_view.cpython-312.pyc index f494d62..d905f33 100644 Binary files a/src/gui/widgets/__pycache__/diagnostics_view.cpython-312.pyc 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 index 7e3cc37..c826dcb 100644 Binary files a/src/gui/widgets/__pycache__/interface_list.cpython-312.pyc 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 index da359fb..b8c3ad3 100644 Binary files a/src/gui/widgets/__pycache__/ip_config.cpython-312.pyc and b/src/gui/widgets/__pycache__/ip_config.cpython-312.pyc differ diff --git a/src/gui/widgets/__pycache__/network_monitor.cpython-312.pyc b/src/gui/widgets/__pycache__/network_monitor.cpython-312.pyc new file mode 100644 index 0000000..485b40e Binary files /dev/null and b/src/gui/widgets/__pycache__/network_monitor.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 index acdaa04..d3a8e94 100644 Binary files a/src/gui/widgets/__pycache__/port_scanner.cpython-312.pyc 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 index 3bf19de..391d99e 100644 Binary files a/src/gui/widgets/__pycache__/scanner_view.cpython-312.pyc 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 index 7eee041..88cff47 100644 Binary files a/src/gui/widgets/__pycache__/sniffer_view.cpython-312.pyc and b/src/gui/widgets/__pycache__/sniffer_view.cpython-312.pyc differ diff --git a/src/gui/widgets/__pycache__/vlan35_check.cpython-312.pyc b/src/gui/widgets/__pycache__/vlan35_check.cpython-312.pyc new file mode 100644 index 0000000..b296a3f Binary files /dev/null and b/src/gui/widgets/__pycache__/vlan35_check.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 index 320f527..196f0a0 100644 Binary files a/src/gui/widgets/__pycache__/vlan_bridge.cpython-312.pyc and b/src/gui/widgets/__pycache__/vlan_bridge.cpython-312.pyc differ diff --git a/src/gui/widgets/advanced_routing.py b/src/gui/widgets/advanced_routing.py new file mode 100644 index 0000000..6694f01 --- /dev/null +++ b/src/gui/widgets/advanced_routing.py @@ -0,0 +1,765 @@ +from PyQt6.QtWidgets import ( + QWidget, + QVBoxLayout, + QHBoxLayout, + QTabWidget, + QLabel, + QLineEdit, + QPushButton, + QTableWidget, + QTableWidgetItem, + QHeaderView, + QGroupBox, + QFormLayout, + QComboBox, + QCheckBox, + QSpinBox, + QMessageBox, +) +from PyQt6.QtCore import Qt + +from src.core.advanced_router import AdvancedRouter +from src.core.network_manager import NetworkManager + + +class AdvancedRoutingWidget(QWidget): + """ + Pestaña principal de \"Routing Avanzado\". + Organiza la funcionalidad en subpestañas internas: + - VLANs + - Redirección + - Redes (gateway/DHCP) + - Routing + - NAT + - Firewall + """ + + def __init__(self, network_manager: NetworkManager, parent=None): + super().__init__(parent) + self.nm = network_manager + self.router = AdvancedRouter(self.nm) + + main_layout = QVBoxLayout(self) + + title = QLabel("Routing Avanzado") + title.setAlignment(Qt.AlignmentFlag.AlignLeft | Qt.AlignmentFlag.AlignVCenter) + title.setStyleSheet("font-size: 18px; font-weight: 600;") + main_layout.addWidget(title) + + self.tabs = QTabWidget() + main_layout.addWidget(self.tabs) + + # Subpestañas + self._init_vlans_tab() + self._init_redirection_tab() + self._init_networks_tab() + self._init_routing_tab() + self._init_nat_tab() + self._init_firewall_tab() + + self.refresh_interfaces() + + # ------------------------------------------------------------------ + # VLANs + # ------------------------------------------------------------------ + + def _init_vlans_tab(self): + tab = QWidget() + layout = QVBoxLayout(tab) + + # Formulario creación VLANs + form_group = QGroupBox("Gestión de VLANs") + form_layout = QFormLayout(form_group) + + self.vlan_mode_combo = QComboBox() + self.vlan_mode_combo.addItems( + ["Trunk (Tagged)", "Access (Untagged)"] + ) + + self.vlan_iface_combo = QComboBox() + self.vlan_ids_input = QLineEdit() + self.vlan_ids_input.setPlaceholderText("Ej: 10,20,30 o 100-120") + + create_btn = QPushButton("Crear VLAN(es)") + create_btn.clicked.connect(self.on_create_vlans_clicked) + + form_layout.addRow("Modo:", self.vlan_mode_combo) + form_layout.addRow("Interfaz:", self.vlan_iface_combo) + form_layout.addRow("VLAN ID(s):", self.vlan_ids_input) + form_layout.addRow("", create_btn) + + layout.addWidget(form_group) + + # Tabla resumen VLANs + self.vlan_table = QTableWidget() + self.vlan_table.setColumnCount(5) + self.vlan_table.setHorizontalHeaderLabels( + ["Interface", "VLAN ID", "Modo", "Detalle", "Acción"] + ) + self.vlan_table.horizontalHeader().setSectionResizeMode( + QHeaderView.ResizeMode.Stretch + ) + layout.addWidget(self.vlan_table) + + self.tabs.addTab(tab, "VLANs") + + def parse_vlan_ids(self, text: str): + ids = set() + for part in text.split(","): + part = part.strip() + if not part: + continue + if "-" in part: + start, end = part.split("-", 1) + try: + s = int(start) + e = int(end) + except ValueError: + continue + for vid in range(s, e + 1): + ids.add(vid) + else: + try: + ids.add(int(part)) + except ValueError: + continue + return sorted(ids) + + def on_create_vlans_clicked(self): + iface = self.vlan_iface_combo.currentText() + vlan_text = self.vlan_ids_input.text().strip() + if not iface or not vlan_text: + QMessageBox.warning(self, "Error", "Selecciona interfaz y VLAN ID(s).") + return + + vlan_ids = self.parse_vlan_ids(vlan_text) + if not vlan_ids: + QMessageBox.warning(self, "Error", "Formato de VLAN ID(s) no válido.") + return + + mode = self.vlan_mode_combo.currentText() + if "Trunk" in mode: + ok, msg = self.router.create_vlan_trunk(iface, vlan_ids) + else: + if len(vlan_ids) != 1: + QMessageBox.warning( + self, + "Error", + "En modo Access sólo se permite un único VLAN ID.", + ) + return + ok, msg = self.router.create_vlan_access(iface, vlan_ids[0]) + + if ok: + QMessageBox.information(self, "VLAN", msg) + self.refresh_vlan_table() + else: + QMessageBox.critical(self, "Error VLAN", msg) + + def refresh_vlan_table(self): + status = self.router.get_routing_status() + vlans = status.get("vlans", {}) + self.vlan_table.setRowCount(len(vlans)) + for row, (ifname, data) in enumerate(vlans.items()): + self.vlan_table.setItem(row, 0, QTableWidgetItem(ifname)) + self.vlan_table.setItem( + row, 1, QTableWidgetItem(str(data.get("vlan_id", ""))) + ) + self.vlan_table.setItem(row, 2, QTableWidgetItem(data.get("mode", ""))) + self.vlan_table.setItem( + row, + 3, + QTableWidgetItem(f"parent={data.get('parent', '')}"), + ) + # Botón eliminar + delete_btn = QPushButton("Eliminar") + delete_btn.setStyleSheet("background-color: #800000; color: white; padding: 4px 8px;") + delete_btn.clicked.connect(lambda checked, vlan_name=ifname: self.on_delete_vlan_clicked(vlan_name)) + self.vlan_table.setCellWidget(row, 4, delete_btn) + + def on_delete_vlan_clicked(self, vlan_ifname: str): + """Elimina una VLAN tras confirmación del usuario.""" + reply = QMessageBox.question( + self, + "Confirmar eliminación", + f"¿Estás seguro de que quieres eliminar la VLAN '{vlan_ifname}'?\n\n" + "Esto eliminará la interfaz VLAN y limpiará todas las configuraciones asociadas.", + QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, + QMessageBox.StandardButton.No + ) + + if reply == QMessageBox.StandardButton.Yes: + ok, msg = self.router.delete_vlan(vlan_ifname) + if ok: + QMessageBox.information(self, "VLAN eliminada", msg) + self.refresh_vlan_table() + # También refrescamos otras tablas que puedan verse afectadas + self.refresh_redirection_table() + self.refresh_networks_table() + else: + QMessageBox.critical(self, "Error al eliminar VLAN", msg) + + # ------------------------------------------------------------------ + # Redirección de VLANs + # ------------------------------------------------------------------ + + def _init_redirection_tab(self): + tab = QWidget() + layout = QVBoxLayout(tab) + + form_group = QGroupBox("Redirección de VLANs") + form_layout = QFormLayout(form_group) + + self.src_iface_combo = QComboBox() + self.src_vlan_spin = QSpinBox() + self.src_vlan_spin.setRange(1, 4094) + self.src_vlan_spin.setSpecialValueText("Untagged") + + self.dst_iface_combo = QComboBox() + self.dst_vlan_spin = QSpinBox() + self.dst_vlan_spin.setRange(1, 4094) + self.dst_vlan_spin.setSpecialValueText("Untagged") + + self.tagged_to_untagged_cb = QCheckBox("Forzar Tagged → Untagged") + + redirect_btn = QPushButton("Crear redirección") + redirect_btn.clicked.connect(self.on_create_redirection_clicked) + + form_layout.addRow("Interfaz origen:", self.src_iface_combo) + form_layout.addRow("VLAN origen (0 = untagged):", self.src_vlan_spin) + form_layout.addRow("Interfaz destino:", self.dst_iface_combo) + form_layout.addRow("VLAN destino (0 = untagged):", self.dst_vlan_spin) + form_layout.addRow("", self.tagged_to_untagged_cb) + form_layout.addRow("", redirect_btn) + + layout.addWidget(form_group) + + self.redir_table = QTableWidget() + self.redir_table.setColumnCount(5) + self.redir_table.setHorizontalHeaderLabels( + ["Origen", "VLAN Origen", "Destino", "VLAN Destino", "Bridge"] + ) + self.redir_table.horizontalHeader().setSectionResizeMode( + QHeaderView.ResizeMode.Stretch + ) + layout.addWidget(self.redir_table) + + self.tabs.addTab(tab, "Redirección") + + def on_create_redirection_clicked(self): + src_iface = self.src_iface_combo.currentText() + dst_iface = self.dst_iface_combo.currentText() + src_vlan = self.src_vlan_spin.value() or None + dst_vlan = self.dst_vlan_spin.value() or None + + if not src_iface or not dst_iface: + QMessageBox.warning(self, "Error", "Selecciona interfaces de origen y destino.") + return + + if self.tagged_to_untagged_cb.isChecked(): + # Forzamos patrón trunk -> access + if src_vlan is None: + QMessageBox.warning( + self, + "Error", + "Para Tagged → Untagged la VLAN de origen no puede ser 0.", + ) + return + ok, msg = self.router.redirect_vlan_tagged_to_untagged( + trunk_iface=src_iface, vlan_id=src_vlan, access_iface=dst_iface + ) + else: + ok, msg = self.router.redirect_vlan_between_interfaces( + src_iface=src_iface, + src_vlan_id=src_vlan, + dst_iface=dst_iface, + dst_vlan_id=dst_vlan, + ) + + if ok: + QMessageBox.information(self, "Redirección", msg) + self.refresh_redirection_table() + else: + QMessageBox.critical(self, "Error redirección", msg) + + def refresh_redirection_table(self): + status = self.router.get_routing_status() + rules = status.get("vlan_redirections", []) + self.redir_table.setRowCount(len(rules)) + for row, r in enumerate(rules): + self.redir_table.setItem(row, 0, QTableWidgetItem(r.get("src_iface", ""))) + self.redir_table.setItem( + row, + 1, + QTableWidgetItem(str(r.get("src_vlan", "") or "untagged")), + ) + self.redir_table.setItem(row, 2, QTableWidgetItem(r.get("dst_iface", ""))) + self.redir_table.setItem( + row, + 3, + QTableWidgetItem(str(r.get("dst_vlan", "") or "untagged")), + ) + self.redir_table.setItem(row, 4, QTableWidgetItem(r.get("bridge", ""))) + + # ------------------------------------------------------------------ + # Redes (Gateway / DHCP) + # ------------------------------------------------------------------ + + def _init_networks_tab(self): + tab = QWidget() + layout = QVBoxLayout(tab) + + group = QGroupBox("Configuración de red por VLAN / interfaz") + form = QFormLayout(group) + + self.net_iface_combo = QComboBox() + self.net_ip_input = QLineEdit() + self.net_gw_input = QLineEdit() + self.net_dns_input = QLineEdit() + self.net_dhcp_cb = QCheckBox("Habilitar DHCP") + self.net_dhcp_start = QLineEdit() + self.net_dhcp_end = QLineEdit() + + apply_btn = QPushButton("Aplicar configuración") + apply_btn.clicked.connect(self.on_apply_network_config_clicked) + + form.addRow("Interfaz / VLAN:", self.net_iface_combo) + form.addRow("IP Gateway:", self.net_ip_input) + form.addRow("Gateway (next-hop):", self.net_gw_input) + form.addRow("DNS:", self.net_dns_input) + form.addRow(self.net_dhcp_cb) + form.addRow("DHCP Inicio:", self.net_dhcp_start) + form.addRow("DHCP Fin:", self.net_dhcp_end) + form.addRow("", apply_btn) + + layout.addWidget(group) + + self.net_table = QTableWidget() + self.net_table.setColumnCount(6) + self.net_table.setHorizontalHeaderLabels( + ["Interfaz", "IP", "Gateway", "DHCP", "Rango", "DNS"] + ) + self.net_table.horizontalHeader().setSectionResizeMode( + QHeaderView.ResizeMode.Stretch + ) + layout.addWidget(self.net_table) + + self.tabs.addTab(tab, "Redes") + + def on_apply_network_config_clicked(self): + iface = self.net_iface_combo.currentText() + ip = self.net_ip_input.text().strip() + gw = self.net_gw_input.text().strip() + dns = self.net_dns_input.text().strip() or None + dhcp = self.net_dhcp_cb.isChecked() + dhcp_start = self.net_dhcp_start.text().strip() + dhcp_end = self.net_dhcp_end.text().strip() + + if not iface or not ip: + QMessageBox.warning( + self, "Error", "Interfaz y IP son obligatorios para la red." + ) + return + + if gw: + ok, msg = self.router.set_gateway_for_network(iface, ip, gw) + if not ok: + QMessageBox.critical(self, "Error gateway", msg) + return + + if dhcp: + if not (dhcp_start and dhcp_end): + QMessageBox.warning( + self, + "Error", + "Para DHCP debes indicar rango de inicio y fin.", + ) + return + ok, msg = self.router.start_dhcp_server_for_vlan( + vlan_interface=iface, + ip=ip, + range_start=dhcp_start, + range_end=dhcp_end, + gateway=gw or None, + dns=dns, + ) + if not ok: + QMessageBox.critical(self, "Error DHCP", msg) + return + + QMessageBox.information(self, "Red", "Configuración aplicada.") + self.refresh_networks_table() + + def refresh_networks_table(self): + status = self.router.get_routing_status() + gws = status.get("gateways", {}) + dhcps = status.get("dhcp_servers", {}) + self.net_table.setRowCount(len(gws)) + for row, (iface, data) in enumerate(gws.items()): + ip = data.get("ip", "") + gw = data.get("gateway", "") + dns = data.get("dns", "") + dh = dhcps.get(iface) + dh_enabled = "Sí" if dh else "No" + dh_range = "" + if dh: + dh_range = f"{dh.get('range_start','')} - {dh.get('range_end','')}" + + self.net_table.setItem(row, 0, QTableWidgetItem(iface)) + self.net_table.setItem(row, 1, QTableWidgetItem(ip)) + self.net_table.setItem(row, 2, QTableWidgetItem(gw)) + self.net_table.setItem(row, 3, QTableWidgetItem(dh_enabled)) + self.net_table.setItem(row, 4, QTableWidgetItem(dh_range)) + self.net_table.setItem(row, 5, QTableWidgetItem(dns)) + + # ------------------------------------------------------------------ + # Routing (rutas + reglas) + # ------------------------------------------------------------------ + + def _init_routing_tab(self): + tab = QWidget() + layout = QVBoxLayout(tab) + + group = QGroupBox("Rutas estáticas") + form = QFormLayout(group) + + self.route_dst_input = QLineEdit() + self.route_dst_input.setPlaceholderText("Ej: 10.0.0.0/24") + self.route_gw_input = QLineEdit() + self.route_iface_combo = QComboBox() + self.route_table_spin = QSpinBox() + self.route_table_spin.setRange(1, 252) + self.route_table_spin.setValue(254) + self.route_metric_spin = QSpinBox() + self.route_metric_spin.setRange(1, 1000) + self.route_metric_spin.setValue(100) + + add_btn = QPushButton("Añadir ruta") + add_btn.clicked.connect(self.on_add_route_clicked) + + form.addRow("Destino:", self.route_dst_input) + form.addRow("Gateway:", self.route_gw_input) + form.addRow("Interfaz:", self.route_iface_combo) + form.addRow("Tabla:", self.route_table_spin) + form.addRow("Métrica:", self.route_metric_spin) + form.addRow("", add_btn) + + layout.addWidget(group) + + self.routes_table = QTableWidget() + self.routes_table.setColumnCount(5) + self.routes_table.setHorizontalHeaderLabels( + ["Destino", "Gateway", "Interfaz", "Tabla", "Métrica"] + ) + self.routes_table.horizontalHeader().setSectionResizeMode( + QHeaderView.ResizeMode.Stretch + ) + layout.addWidget(self.routes_table) + + fw_group = QGroupBox("Reglas de routing (policy-based)") + fw_form = QFormLayout(fw_group) + + self.rule_table_spin = QSpinBox() + self.rule_table_spin.setRange(1, 252) + self.rule_table_spin.setValue(100) + self.rule_prio_spin = QSpinBox() + self.rule_prio_spin.setRange(1, 32765) + self.rule_prio_spin.setValue(32000) + self.rule_src_input = QLineEdit() + self.rule_dst_input = QLineEdit() + self.rule_iif_input = QLineEdit() + self.rule_oif_input = QLineEdit() + add_rule_btn = QPushButton("Añadir regla") + add_rule_btn.clicked.connect(self.on_add_rule_clicked) + + fw_form.addRow("Tabla:", self.rule_table_spin) + fw_form.addRow("Prioridad:", self.rule_prio_spin) + fw_form.addRow("Origen (CIDR):", self.rule_src_input) + fw_form.addRow("Destino (CIDR):", self.rule_dst_input) + fw_form.addRow("Interfaz entrada:", self.rule_iif_input) + fw_form.addRow("Interfaz salida:", self.rule_oif_input) + fw_form.addRow("", add_rule_btn) + + layout.addWidget(fw_group) + + enable_fwd_btn = QPushButton("Habilitar IP forwarding") + enable_fwd_btn.clicked.connect(self.on_enable_forwarding_clicked) + layout.addWidget(enable_fwd_btn) + + self.tabs.addTab(tab, "Routing") + + def on_add_route_clicked(self): + dst = self.route_dst_input.text().strip() + gw = self.route_gw_input.text().strip() + iface = self.route_iface_combo.currentText() + table = self.route_table_spin.value() + metric = self.route_metric_spin.value() + + if not (dst and gw and iface): + QMessageBox.warning( + self, "Error", "Destino, gateway e interfaz son obligatorios." + ) + return + + ok, msg = self.router.add_route( + dst=dst, gateway=gw, interface=iface, table=table, metric=metric + ) + if ok: + QMessageBox.information(self, "Ruta", msg) + self.refresh_routes_table() + else: + QMessageBox.critical(self, "Error ruta", msg) + + def on_add_rule_clicked(self): + table = self.rule_table_spin.value() + prio = self.rule_prio_spin.value() + src = self.rule_src_input.text().strip() or None + dst = self.rule_dst_input.text().strip() or None + iif = self.rule_iif_input.text().strip() or None + oif = self.rule_oif_input.text().strip() or None + + ok, msg = self.router.add_routing_rule( + table=table, priority=prio, src=src, dst=dst, iifname=iif, oifname=oif + ) + if ok: + QMessageBox.information(self, "Regla", msg) + else: + QMessageBox.critical(self, "Error regla", msg) + + def on_enable_forwarding_clicked(self): + ok, msg = self.router.enable_ip_forwarding() + if ok: + QMessageBox.information(self, "IP forwarding", msg) + else: + QMessageBox.critical(self, "Error", msg) + + def refresh_routes_table(self): + routes = self.router.list_routes() + self.routes_table.setRowCount(len(routes)) + for row, r in enumerate(routes): + self.routes_table.setItem(row, 0, QTableWidgetItem(r.get("dst", ""))) + self.routes_table.setItem(row, 1, QTableWidgetItem(r.get("gateway", ""))) + self.routes_table.setItem(row, 2, QTableWidgetItem(r.get("interface", ""))) + self.routes_table.setItem( + row, 3, QTableWidgetItem(str(r.get("table", ""))) + ) + self.routes_table.setItem( + row, 4, QTableWidgetItem(str(r.get("metric", ""))) + ) + + # ------------------------------------------------------------------ + # NAT + # ------------------------------------------------------------------ + + def _init_nat_tab(self): + tab = QWidget() + layout = QVBoxLayout(tab) + + group = QGroupBox("NAT y Port Forwarding") + form = QFormLayout(group) + + self.nat_internal_iface_combo = QComboBox() + self.nat_external_iface_combo = QComboBox() + enable_nat_btn = QPushButton("Habilitar NAT") + enable_nat_btn.clicked.connect(self.on_enable_nat_clicked) + + self.pf_ext_ip = QLineEdit() + self.pf_ext_port = QSpinBox() + self.pf_ext_port.setRange(1, 65535) + self.pf_int_ip = QLineEdit() + self.pf_int_port = QSpinBox() + self.pf_int_port.setRange(1, 65535) + self.pf_proto_combo = QComboBox() + self.pf_proto_combo.addItems(["tcp", "udp"]) + add_pf_btn = QPushButton("Añadir port forwarding") + add_pf_btn.clicked.connect(self.on_add_pf_clicked) + + form.addRow("Interfaz interna:", self.nat_internal_iface_combo) + form.addRow("Interfaz externa:", self.nat_external_iface_combo) + form.addRow("", enable_nat_btn) + form.addRow(QLabel("— Port Forwarding —")) + form.addRow("IP externa:", self.pf_ext_ip) + form.addRow("Puerto externo:", self.pf_ext_port) + form.addRow("IP interna:", self.pf_int_ip) + form.addRow("Puerto interno:", self.pf_int_port) + form.addRow("Protocolo:", self.pf_proto_combo) + form.addRow("", add_pf_btn) + + layout.addWidget(group) + + self.pf_table = QTableWidget() + self.pf_table.setColumnCount(4) + self.pf_table.setHorizontalHeaderLabels( + ["Externo", "Interno", "Proto", "Estado"] + ) + self.pf_table.horizontalHeader().setSectionResizeMode( + QHeaderView.ResizeMode.Stretch + ) + layout.addWidget(self.pf_table) + + self.tabs.addTab(tab, "NAT") + + def on_enable_nat_clicked(self): + internal = self.nat_internal_iface_combo.currentText() + external = self.nat_external_iface_combo.currentText() + if not (internal and external): + QMessageBox.warning( + self, + "Error", + "Selecciona interfaz interna y externa para habilitar NAT.", + ) + return + ok, msg = self.router.enable_nat_for_interface( + interface=internal, external_iface=external + ) + if ok: + QMessageBox.information(self, "NAT", msg) + else: + QMessageBox.critical(self, "Error NAT", msg) + + def on_add_pf_clicked(self): + ext_ip = self.pf_ext_ip.text().strip() + ext_port = self.pf_ext_port.value() + int_ip = self.pf_int_ip.text().strip() + int_port = self.pf_int_port.value() + proto = self.pf_proto_combo.currentText() + + if not (ext_ip and int_ip): + QMessageBox.warning( + self, + "Error", + "Debes indicar IP externa e interna para el port forwarding.", + ) + return + + ok, msg = self.router.add_port_forwarding( + external_ip=ext_ip, + external_port=ext_port, + internal_ip=int_ip, + internal_port=int_port, + protocol=proto, + ) + if ok: + QMessageBox.information(self, "Port forwarding", msg) + self.refresh_pf_table() + else: + QMessageBox.critical(self, "Error port forwarding", msg) + + def refresh_pf_table(self): + status = self.router.get_routing_status() + rules = status.get("nat_rules", []) + # Sólo mostramos las que tienen campos de port forwarding + pf_rules = [r for r in rules if "external_ip" in r] + self.pf_table.setRowCount(len(pf_rules)) + for row, r in enumerate(pf_rules): + self.pf_table.setItem( + row, + 0, + QTableWidgetItem( + f"{r.get('external_ip','')}:{r.get('external_port','')}" + ), + ) + self.pf_table.setItem( + row, + 1, + QTableWidgetItem( + f"{r.get('internal_ip','')}:{r.get('internal_port','')}" + ), + ) + self.pf_table.setItem(row, 2, QTableWidgetItem(r.get("protocol", ""))) + self.pf_table.setItem(row, 3, QTableWidgetItem("Activo")) + + # ------------------------------------------------------------------ + # Firewall / Visibilidad + # ------------------------------------------------------------------ + + def _init_firewall_tab(self): + tab = QWidget() + layout = QVBoxLayout(tab) + + group = QGroupBox("Visibilidad entre VLANs (simple)") + form = QFormLayout(group) + + self.fw_src_vlan = QSpinBox() + self.fw_src_vlan.setRange(1, 4094) + self.fw_dst_vlan = QSpinBox() + self.fw_dst_vlan.setRange(1, 4094) + self.fw_allow_cb = QCheckBox("Permitir tráfico (si no, se bloquea)") + self.fw_allow_cb.setChecked(True) + apply_btn = QPushButton("Aplicar política") + apply_btn.clicked.connect(self.on_apply_vlan_visibility_clicked) + + form.addRow("VLAN origen:", self.fw_src_vlan) + form.addRow("VLAN destino:", self.fw_dst_vlan) + form.addRow(self.fw_allow_cb) + form.addRow("", apply_btn) + + layout.addWidget(group) + + self.fw_table = QTableWidget() + self.fw_table.setColumnCount(3) + self.fw_table.setHorizontalHeaderLabels(["VLAN origen", "VLAN destino", "Permitido"]) + self.fw_table.horizontalHeader().setSectionResizeMode( + QHeaderView.ResizeMode.Stretch + ) + layout.addWidget(self.fw_table) + + self.tabs.addTab(tab, "Firewall") + + def on_apply_vlan_visibility_clicked(self): + src = self.fw_src_vlan.value() + dst = self.fw_dst_vlan.value() + allow = self.fw_allow_cb.isChecked() + + if allow: + ok, msg = self.router.allow_vlan_to_vlan(src, dst) + else: + ok, msg = self.router.block_vlan_to_vlan(src, dst) + + if ok: + QMessageBox.information(self, "Firewall", msg) + self.refresh_fw_table() + else: + QMessageBox.critical(self, "Error firewall", msg) + + def refresh_fw_table(self): + status = self.router.get_routing_status() + vis = status.get("vlan_visibility", {}) + items = list(vis.items()) + self.fw_table.setRowCount(len(items)) + for row, (k, allowed) in enumerate(items): + src_str, dst_str = k.split("-") + self.fw_table.setItem(row, 0, QTableWidgetItem(src_str)) + self.fw_table.setItem(row, 1, QTableWidgetItem(dst_str)) + self.fw_table.setItem(row, 2, QTableWidgetItem("Sí" if allowed else "No")) + + # ------------------------------------------------------------------ + # Utilidades comunes + # ------------------------------------------------------------------ + + def refresh_interfaces(self): + interfaces = self.nm.list_interfaces() + names = [i["name"] for i in interfaces] + + combos = [ + self.vlan_iface_combo, + self.src_iface_combo, + self.dst_iface_combo, + self.net_iface_combo, + self.route_iface_combo, + self.nat_internal_iface_combo, + self.nat_external_iface_combo, + ] + + for combo in combos: + combo.clear() + combo.addItems(names) + + # Tras refrescar interfaces, actualizamos también tablas en blanco + self.refresh_vlan_table() + self.refresh_redirection_table() + self.refresh_networks_table() + self.refresh_routes_table() + self.refresh_pf_table() + self.refresh_fw_table() + diff --git a/src/gui/widgets/interface_list.py b/src/gui/widgets/interface_list.py index 1e9b2ea..9bfb585 100644 --- a/src/gui/widgets/interface_list.py +++ b/src/gui/widgets/interface_list.py @@ -1,5 +1,5 @@ from PyQt6.QtWidgets import (QWidget, QVBoxLayout, QListWidget, QListWidgetItem, - QLabel, QHBoxLayout, QFrame, QPushButton, QStyle) + QLabel, QHBoxLayout, QFrame, QPushButton, QStyle, QMessageBox) from PyQt6.QtCore import Qt, pyqtSignal from PyQt6.QtGui import QIcon, QColor, QFont @@ -18,6 +18,14 @@ class InterfaceListWidget(QWidget): header_layout.addWidget(title) header_layout.addStretch() + # Reset Button + self.reset_btn = QPushButton("Reset") + self.reset_btn.setToolTip("Reiniciar interfaz seleccionada (down → up)") + self.reset_btn.setStyleSheet("background-color: #ff9800; color: white; padding: 4px 12px; font-weight: 600;") + self.reset_btn.clicked.connect(self.on_reset_clicked) + self.reset_btn.setEnabled(False) # Deshabilitado hasta que se seleccione una interfaz + header_layout.addWidget(self.reset_btn) + # Refresh Button self.refresh_btn = QPushButton() self.refresh_btn.setIcon(self.style().standardIcon(QStyle.StandardPixmap.SP_BrowserReload)) @@ -28,6 +36,9 @@ class InterfaceListWidget(QWidget): self.layout.addLayout(header_layout) + # Guardar la interfaz seleccionada actual + self.current_selected_iface = None + # List self.list_widget = QListWidget() self.list_widget.itemClicked.connect(self.on_item_clicked) @@ -97,4 +108,33 @@ class InterfaceListWidget(QWidget): def on_item_clicked(self, item): iface_data = item.data(Qt.ItemDataRole.UserRole) + self.current_selected_iface = iface_data + self.reset_btn.setEnabled(True) # Habilitar botón reset cuando hay una interfaz seleccionada self.interface_selected.emit(iface_data) + + def on_reset_clicked(self): + """Reinicia la interfaz seleccionada.""" + if not self.current_selected_iface: + QMessageBox.warning(self, "Error", "No hay interfaz seleccionada.") + return + + iface_name = self.current_selected_iface['name'] + + # Confirmación + reply = QMessageBox.question( + self, + "Confirmar reinicio", + f"¿Estás seguro de que quieres reiniciar la interfaz '{iface_name}'?\n\n" + "Esto hará un down y luego up de la interfaz. La conexión se interrumpirá brevemente.", + QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No, + QMessageBox.StandardButton.No + ) + + if reply == QMessageBox.StandardButton.Yes: + success, msg = self.nm.reset_interface(iface_name) + if success: + QMessageBox.information(self, "Interfaz reiniciada", msg) + # Refrescar la lista para mostrar el nuevo estado + self.refresh_interfaces() + else: + QMessageBox.critical(self, "Error", msg) \ No newline at end of file diff --git a/src/gui/widgets/network_monitor.py b/src/gui/widgets/network_monitor.py new file mode 100644 index 0000000..973c5f2 --- /dev/null +++ b/src/gui/widgets/network_monitor.py @@ -0,0 +1,177 @@ +import subprocess +from typing import Dict + +import psutil +from PyQt6.QtCore import Qt, QTimer +from PyQt6.QtWidgets import ( + QWidget, + QVBoxLayout, + QHBoxLayout, + QLabel, + QTableWidget, + QTableWidgetItem, + QHeaderView, + QGroupBox, +) + +from src.core.network_manager import NetworkManager + + +class NetworkMonitorWidget(QWidget): + """ + Monitor de red en tiempo real: + - Tráfico por interfaz (bytes/seg, paquetes) + - Resumen básico de reglas de firewall (iptables FORWARD) + """ + + def __init__(self, network_manager: NetworkManager, parent=None): + super().__init__(parent) + self.nm = network_manager + + self.last_counters: Dict[str, psutil._common.snetio] = {} + + layout = QVBoxLayout(self) + + title = QLabel("Monitor de Red") + title.setAlignment(Qt.AlignmentFlag.AlignLeft | Qt.AlignmentFlag.AlignVCenter) + title.setStyleSheet("font-size: 18px; font-weight: 600;") + layout.addWidget(title) + + # Grupo de tráfico por interfaz + traffic_group = QGroupBox("Tráfico por interfaz") + traffic_layout = QVBoxLayout(traffic_group) + + self.iface_table = QTableWidget() + self.iface_table.setColumnCount(7) + self.iface_table.setHorizontalHeaderLabels( + [ + "Interfaz", + "RX total (MB)", + "TX total (MB)", + "RX (KB/s)", + "TX (KB/s)", + "Paquetes RX/s", + "Paquetes TX/s", + ] + ) + self.iface_table.horizontalHeader().setSectionResizeMode( + QHeaderView.ResizeMode.Stretch + ) + traffic_layout.addWidget(self.iface_table) + + layout.addWidget(traffic_group) + + # Grupo de resumen de firewall + fw_group = QGroupBox("Resumen de firewall (iptables FORWARD)") + fw_layout = QVBoxLayout(fw_group) + + self.fw_table = QTableWidget() + self.fw_table.setColumnCount(5) + self.fw_table.setHorizontalHeaderLabels( + ["Regla", "Política", "Paquetes", "Bytes", "Destino"] + ) + self.fw_table.horizontalHeader().setSectionResizeMode( + QHeaderView.ResizeMode.Stretch + ) + fw_layout.addWidget(self.fw_table) + + layout.addWidget(fw_group) + + # Temporizador de actualización + self.timer = QTimer(self) + self.timer.setInterval(1000) # 1 segundo + self.timer.timeout.connect(self.refresh) + self.timer.start() + + # Inicial + self.refresh(initial=True) + + # ------------------------------------------------------------------ + # Actualización + # ------------------------------------------------------------------ + + def refresh(self, initial: bool = False): + self.refresh_traffic(initial=initial) + self.refresh_firewall() + + def refresh_traffic(self, initial: bool = False): + counters = psutil.net_io_counters(pernic=True) + + # Aseguramos que la tabla tenga una fila por interfaz conocida por NetworkManager + interfaces = [iface["name"] for iface in self.nm.list_interfaces()] + + self.iface_table.setRowCount(len(interfaces)) + + for row, name in enumerate(interfaces): + data = counters.get(name) + if not data: + continue + + last = self.last_counters.get(name) + if last and not initial: + seconds = 1.0 # timer fijo a 1s + rx_rate = (data.bytes_recv - last.bytes_recv) / 1024.0 / seconds + tx_rate = (data.bytes_sent - last.bytes_sent) / 1024.0 / seconds + rx_pkts = (data.packets_recv - last.packets_recv) / seconds + tx_pkts = (data.packets_sent - last.packets_sent) / seconds + else: + rx_rate = tx_rate = rx_pkts = tx_pkts = 0.0 + + self.iface_table.setItem(row, 0, QTableWidgetItem(name)) + self.iface_table.setItem( + row, 1, QTableWidgetItem(f"{data.bytes_recv / (1024.0 * 1024.0):.2f}") + ) + self.iface_table.setItem( + row, 2, QTableWidgetItem(f"{data.bytes_sent / (1024.0 * 1024.0):.2f}") + ) + self.iface_table.setItem(row, 3, QTableWidgetItem(f"{rx_rate:.1f}")) + self.iface_table.setItem(row, 4, QTableWidgetItem(f"{tx_rate:.1f}")) + self.iface_table.setItem(row, 5, QTableWidgetItem(f"{rx_pkts:.1f}")) + self.iface_table.setItem(row, 6, QTableWidgetItem(f"{tx_pkts:.1f}")) + + self.last_counters = counters + + def refresh_firewall(self): + """ + Lee iptables FORWARD para mostrar paquetes/bytes aceptados y bloqueados. + Si iptables no está disponible, deja la tabla vacía. + """ + try: + result = subprocess.run( + ["iptables", "-L", "FORWARD", "-v", "-n"], + capture_output=True, + text=True, + check=False, + ) + except FileNotFoundError: + self.fw_table.setRowCount(0) + return + + lines = result.stdout.splitlines() + rules = [] + # Saltamos las dos primeras líneas de cabecera + for line in lines[2:]: + parts = line.split() + if len(parts) < 8: + continue + pkts = parts[0] + bytes_ = parts[1] + target = parts[2] + # parts[3] prot, parts[4] opt, parts[5] in, parts[6] out, parts[7] source, parts[8] dest (si existe) + dest = parts[8] if len(parts) > 8 else "" + rules.append( + { + "target": target, + "pkts": pkts, + "bytes": bytes_, + "dest": dest, + } + ) + + self.fw_table.setRowCount(len(rules)) + for row, r in enumerate(rules): + self.fw_table.setItem(row, 0, QTableWidgetItem(f"Regla {row + 1}")) + self.fw_table.setItem(row, 1, QTableWidgetItem(r["target"])) + self.fw_table.setItem(row, 2, QTableWidgetItem(r["pkts"])) + self.fw_table.setItem(row, 3, QTableWidgetItem(r["bytes"])) + self.fw_table.setItem(row, 4, QTableWidgetItem(r["dest"])) diff --git a/src/gui/widgets/vlan35_check.py b/src/gui/widgets/vlan35_check.py new file mode 100644 index 0000000..4ad4429 --- /dev/null +++ b/src/gui/widgets/vlan35_check.py @@ -0,0 +1,522 @@ +import datetime +import os +from PyQt6.QtWidgets import ( + QWidget, + QVBoxLayout, + QHBoxLayout, + QLabel, + QPushButton, + QTableWidget, + QTableWidgetItem, + QHeaderView, + QGroupBox, + QFormLayout, + QLineEdit, + QSpinBox, + QComboBox, + QMessageBox, + QFileDialog, + QTextEdit, + QCheckBox, +) +from PyQt6.QtCore import Qt +from PyQt6.QtGui import QColor + +from src.core.vlan35_checker import VLAN35Checker + + +class VLAN35CheckWidget(QWidget): + """ + Widget para comprobación de visibilidad en VLAN 35. + Ejecuta tests de conectividad TCP/UDP y genera informes HTML completos. + """ + + def __init__(self, parent=None): + super().__init__(parent) + self.checker = VLAN35Checker() + self.test_thread = None + self.test_results = [] + + layout = QVBoxLayout(self) + + title = QLabel("Comprobación VLAN35") + title.setAlignment(Qt.AlignmentFlag.AlignLeft | Qt.AlignmentFlag.AlignVCenter) + title.setStyleSheet("font-size: 18px; font-weight: 600;") + layout.addWidget(title) + + # Grupo de configuración de tests + config_group = QGroupBox("Configuración de Tests") + config_layout = QVBoxLayout(config_group) + + # Botones de acción rápida + action_layout = QHBoxLayout() + load_defaults_btn = QPushButton("Cargar Tests por Defecto") + load_defaults_btn.clicked.connect(self.load_default_tests) + action_layout.addWidget(load_defaults_btn) + + self.run_btn = QPushButton("Ejecutar Tests") + self.run_btn.setStyleSheet("background-color: #008542; color: white; padding: 6px 16px;") + self.run_btn.clicked.connect(self.run_tests) + action_layout.addWidget(self.run_btn) + + self.stop_btn = QPushButton("Detener") + self.stop_btn.setEnabled(False) + self.stop_btn.setStyleSheet("background-color: #800000; color: white;") + self.stop_btn.clicked.connect(self.stop_tests) + action_layout.addWidget(self.stop_btn) + + action_layout.addStretch() + + generate_report_btn = QPushButton("Generar Informe HTML") + generate_report_btn.clicked.connect(self.generate_report) + action_layout.addWidget(generate_report_btn) + + config_layout.addLayout(action_layout) + + # Tabla de tests configurados + self.tests_table = QTableWidget() + self.tests_table.setColumnCount(5) + self.tests_table.setHorizontalHeaderLabels( + ["Host", "Puerto", "Protocolo", "Tipo", "Acción"] + ) + self.tests_table.horizontalHeader().setSectionResizeMode( + QHeaderView.ResizeMode.Stretch + ) + config_layout.addWidget(self.tests_table) + + # Formulario para añadir nuevo test + form_group = QGroupBox("Añadir Nuevo Test") + form_layout = QFormLayout(form_group) + + self.new_host_input = QLineEdit() + self.new_host_input.setPlaceholderText("10.47.8.133") + self.new_port_spin = QSpinBox() + self.new_port_spin.setRange(1, 65535) + self.new_port_spin.setValue(80) + self.new_proto_combo = QComboBox() + self.new_proto_combo.addItems(["TCP", "UDP"]) + self.new_type_combo = QComboBox() + self.new_type_combo.addItems(["connectivity", "ntp"]) + + add_test_btn = QPushButton("Añadir Test") + add_test_btn.clicked.connect(self.add_test) + + form_layout.addRow("Host:", self.new_host_input) + form_layout.addRow("Puerto:", self.new_port_spin) + form_layout.addRow("Protocolo:", self.new_proto_combo) + form_layout.addRow("Tipo:", self.new_type_combo) + form_layout.addRow("", add_test_btn) + + config_layout.addWidget(form_group) + layout.addWidget(config_group) + + # Grupo de resultados + results_group = QGroupBox("Resultados de Tests") + results_layout = QVBoxLayout(results_group) + + self.results_table = QTableWidget() + self.results_table.setColumnCount(7) + self.results_table.setHorizontalHeaderLabels( + [ + "Host", + "Puerto", + "Protocolo", + "Estado", + "Latencia (ms)", + "Detalles", + "Timestamp", + ] + ) + self.results_table.horizontalHeader().setSectionResizeMode( + QHeaderView.ResizeMode.Stretch + ) + results_layout.addWidget(self.results_table) + + # Log de progreso + self.log_output = QTextEdit() + self.log_output.setReadOnly(True) + self.log_output.setMaximumHeight(120) + self.log_output.setStyleSheet( + "background-color: #1e1e1e; color: #00ff00; font-family: monospace; font-size: 11px;" + ) + results_layout.addWidget(self.log_output) + + layout.addWidget(results_group) + + # Cargar tests por defecto al inicio + self.load_default_tests() + + def load_default_tests(self): + """Carga los tests por defecto del checker.""" + tests = self.checker.get_default_tests() + self.tests_table.setRowCount(len(tests)) + for row, test in enumerate(tests): + self.tests_table.setItem(row, 0, QTableWidgetItem(test["host"])) + self.tests_table.setItem(row, 1, QTableWidgetItem(str(test["port"]))) + self.tests_table.setItem(row, 2, QTableWidgetItem(test["protocol"])) + self.tests_table.setItem(row, 3, QTableWidgetItem(test.get("type", "connectivity"))) + + delete_btn = QPushButton("Eliminar") + delete_btn.setStyleSheet("background-color: #800000; color: white; padding: 2px 6px;") + delete_btn.clicked.connect( + lambda checked, r=row: self.remove_test(r) + ) + self.tests_table.setCellWidget(row, 4, delete_btn) + + def add_test(self): + """Añade un nuevo test a la tabla.""" + host = self.new_host_input.text().strip() + port = self.new_port_spin.value() + protocol = self.new_proto_combo.currentText() + test_type = self.new_type_combo.currentText() + + if not host: + QMessageBox.warning(self, "Error", "Debes indicar un host.") + return + + row = self.tests_table.rowCount() + self.tests_table.insertRow(row) + self.tests_table.setItem(row, 0, QTableWidgetItem(host)) + self.tests_table.setItem(row, 1, QTableWidgetItem(str(port))) + self.tests_table.setItem(row, 2, QTableWidgetItem(protocol)) + self.tests_table.setItem(row, 3, QTableWidgetItem(test_type)) + + delete_btn = QPushButton("Eliminar") + delete_btn.setStyleSheet("background-color: #800000; color: white; padding: 2px 6px;") + delete_btn.clicked.connect(lambda checked, r=row: self.remove_test(r)) + self.tests_table.setCellWidget(row, 4, delete_btn) + + # Limpiar formulario + self.new_host_input.clear() + self.new_port_spin.setValue(80) + + def remove_test(self, row: int): + """Elimina un test de la tabla.""" + self.tests_table.removeRow(row) + + def get_tests_from_table(self) -> list: + """Extrae la lista de tests desde la tabla.""" + tests = [] + for row in range(self.tests_table.rowCount()): + host = self.tests_table.item(row, 0) + port = self.tests_table.item(row, 1) + protocol = self.tests_table.item(row, 2) + test_type = self.tests_table.item(row, 3) + + if host and port and protocol: + tests.append( + { + "host": host.text(), + "port": int(port.text()), + "protocol": protocol.text(), + "type": test_type.text() if test_type else "connectivity", + } + ) + return tests + + def run_tests(self): + """Ejecuta todos los tests configurados.""" + tests = self.get_tests_from_table() + if not tests: + QMessageBox.warning(self, "Error", "No hay tests configurados.") + return + + self.test_results = [] + self.results_table.setRowCount(0) + self.log_output.clear() + self.log_output.append(f"Iniciando {len(tests)} tests...\n") + + self.run_btn.setEnabled(False) + self.stop_btn.setEnabled(True) + + self.test_thread = self.checker.run_tests(tests) + self.test_thread.test_result.connect(self.on_test_result) + self.test_thread.all_finished.connect(self.on_all_finished) + + def stop_tests(self): + """Detiene los tests en ejecución.""" + if self.test_thread and self.test_thread.isRunning(): + self.test_thread.terminate() + self.test_thread.wait() + self.run_btn.setEnabled(True) + self.stop_btn.setEnabled(False) + self.log_output.append("\nTests detenidos por el usuario.") + + def on_test_result(self, result: dict): + """Procesa el resultado de un test individual.""" + self.test_results.append(result) + + row = self.results_table.rowCount() + self.results_table.insertRow(row) + + # Host + self.results_table.setItem(row, 0, QTableWidgetItem(result["host"])) + # Puerto + self.results_table.setItem(row, 1, QTableWidgetItem(str(result["port"]))) + # Protocolo + self.results_table.setItem(row, 2, QTableWidgetItem(result["protocol"])) + # Estado + status_item = QTableWidgetItem(result["status"]) + if result["success"]: + status_item.setForeground(QColor("#4caf50")) + else: + status_item.setForeground(QColor("#f44336")) + self.results_table.setItem(row, 3, status_item) + # Latencia + latency_text = ( + f"{result['latency_ms']:.2f}" if result["latency_ms"] is not None else "N/A" + ) + self.results_table.setItem(row, 4, QTableWidgetItem(latency_text)) + # Detalles + details = result.get("details", result.get("error", "")) + self.results_table.setItem(row, 5, QTableWidgetItem(details)) + # Timestamp + self.results_table.setItem(row, 6, QTableWidgetItem(result["timestamp"])) + + # Log + status_icon = "✓" if result["success"] else "✗" + self.log_output.append( + f"{status_icon} {result['host']}:{result['port']} ({result['protocol']}) - {result['status']}" + ) + scrollbar = self.log_output.verticalScrollBar() + scrollbar.setValue(scrollbar.maximum()) + + def on_all_finished(self): + """Se ejecuta cuando todos los tests terminan.""" + self.run_btn.setEnabled(True) + self.stop_btn.setEnabled(False) + + success_count = sum(1 for r in self.test_results if r["success"]) + total_count = len(self.test_results) + + self.log_output.append( + f"\n{'='*50}\nTests completados: {success_count}/{total_count} exitosos" + ) + + def generate_report(self): + """Genera un informe HTML completo con todos los resultados.""" + if not self.test_results: + QMessageBox.warning( + self, "Error", "No hay resultados para generar el informe. Ejecuta los tests primero." + ) + return + + filename, _ = QFileDialog.getSaveFileName( + self, "Guardar Informe", "vlan35_check_report.html", "HTML Files (*.html)" + ) + + if not filename: + return + + html = self._generate_html_report() + try: + with open(filename, "w", encoding="utf-8") as f: + f.write(html) + QMessageBox.information( + self, "Informe Generado", f"Informe guardado en:\n{os.path.abspath(filename)}" + ) + except Exception as e: + QMessageBox.critical(self, "Error", f"Error al guardar informe:\n{str(e)}") + + def _generate_html_report(self) -> str: + """Genera el contenido HTML del informe.""" + timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") + + # Estadísticas + total = len(self.test_results) + success = sum(1 for r in self.test_results if r["success"]) + failed = total - success + + html = f""" + + + + Informe de Comprobación VLAN35 + + + +
+

Informe de Comprobación VLAN35

+ +
+

Fecha y hora: {timestamp}

+

Total de tests ejecutados: {total}

+
+
+
{success}
+
Exitosos
+
+
+
{failed}
+
Fallidos
+
+
+
{total}
+
Total
+
+
+
+ +

Resultados Detallados

+ + + + + + + + + + +""" + + for result in self.test_results: + status_class = "status-success" if result["success"] else "status-failed" + latency = ( + f"{result['latency_ms']:.2f}" if result["latency_ms"] is not None else "N/A" + ) + details = result.get("details", result.get("error", "")) + + html += f""" + + + + + + + + + +""" + + html += """ +
HostPuertoProtocoloEstadoLatencia (ms)DetallesTimestamp
{result['host']}{result['port']}{result['protocol']}{result['status']}{latency}{details}{result['timestamp']}
+ +

Resumen por Host

+ + + + + + + +""" + + # Agrupar por host + host_stats = {} + for result in self.test_results: + host = result["host"] + if host not in host_stats: + host_stats[host] = {"success": 0, "failed": 0} + if result["success"]: + host_stats[host]["success"] += 1 + else: + host_stats[host]["failed"] += 1 + + for host, stats in host_stats.items(): + total_host = stats["success"] + stats["failed"] + html += f""" + + + + + + +""" + + html += """ +
HostTests ExitososTests FallidosTotal
{host}{stats['success']}{stats['failed']}{total_host}
+
+ + +""" + + return html diff --git a/src/utils/__pycache__/report_generator.cpython-312.pyc b/src/utils/__pycache__/report_generator.cpython-312.pyc index 9f2b044..d37bcf2 100644 Binary files a/src/utils/__pycache__/report_generator.cpython-312.pyc and b/src/utils/__pycache__/report_generator.cpython-312.pyc differ diff --git a/vlan35_check_report.html b/vlan35_check_report.html new file mode 100644 index 0000000..64cc6a5 --- /dev/null +++ b/vlan35_check_report.html @@ -0,0 +1,233 @@ + + + + + Informe de Comprobación VLAN35 + + + +
+

Informe de Comprobación VLAN35

+ +
+

Fecha y hora: 2026-02-16 11:10:26

+

Total de tests ejecutados: 6

+
+
+
6
+
Exitosos
+
+
+
0
+
Fallidos
+
+
+
6
+
Total
+
+
+
+ +

Resultados Detallados

+ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
HostPuertoProtocoloEstadoLatencia (ms)DetallesTimestamp
10.47.8.133515TCPOpen7.10Conexión TCP exitosa en 7.10ms2026-02-16 11:10:17
10.47.8.1339195TCPOpen7.88Conexión TCP exitosa en 7.88ms2026-02-16 11:10:17
10.47.8.1359191TCPOpen8.01Conexión TCP exitosa en 8.01ms2026-02-16 11:10:17
10.47.8.1359192TCPOpen4.41Conexión TCP exitosa en 4.41ms2026-02-16 11:10:17
10.240.225.171465TCPOpen3.96Conexión TCP exitosa en 3.96ms2026-02-16 11:10:17
10.244.255.250123UDPOpen5.11NTP hora local: 2026-02-16 11:25:012026-02-16 11:10:17
+ +

Resumen por Host

+ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
HostTests ExitososTests FallidosTotal
10.47.8.133202
10.47.8.135202
10.240.225.171101
10.244.255.250101
+
+ +