JNC Linux

This commit is contained in:
joaquin.navajas
2026-05-11 08:09:47 +02:00
parent 373493ca2f
commit 0d10ab8110
31 changed files with 3165 additions and 90 deletions
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
+996
View File
@@ -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)
+27
View File
@@ -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()
+228
View File
@@ -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
Binary file not shown.
Binary file not shown.
+89 -27
View File
@@ -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):
+87 -62
View File
@@ -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
+765
View File
@@ -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()
+41 -1
View File
@@ -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)
+177
View File
@@ -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"]))
+522
View File
@@ -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"""<!DOCTYPE html>
<html lang="es">
<head>
<meta charset="UTF-8">
<title>Informe de Comprobación VLAN35</title>
<style>
body {{
font-family: 'Segoe UI', 'Roboto', sans-serif;
background-color: #f5f5f5;
color: #333;
margin: 0;
padding: 20px;
}}
.container {{
max-width: 1200px;
margin: 0 auto;
background: white;
padding: 30px;
box-shadow: 0 2px 10px rgba(0,0,0,0.1);
border-radius: 8px;
}}
h1 {{
color: #008542;
border-bottom: 3px solid #008542;
padding-bottom: 10px;
margin-bottom: 20px;
}}
h2 {{
color: #444;
margin-top: 30px;
border-left: 5px solid #008542;
padding-left: 10px;
}}
.header-info {{
background-color: #f8f9fa;
padding: 15px;
border-radius: 5px;
margin-bottom: 20px;
}}
.stats {{
display: flex;
gap: 20px;
margin-top: 15px;
}}
.stat-box {{
flex: 1;
padding: 15px;
border-radius: 5px;
text-align: center;
}}
.stat-success {{
background-color: #e8f5e9;
color: #2e7d32;
}}
.stat-failed {{
background-color: #ffebee;
color: #c62828;
}}
.stat-total {{
background-color: #e3f2fd;
color: #1565c0;
}}
table {{
width: 100%;
border-collapse: collapse;
margin-top: 15px;
font-size: 0.9em;
}}
th, td {{
border: 1px solid #e0e0e0;
padding: 10px;
text-align: left;
}}
th {{
background-color: #008542;
color: white;
font-weight: 600;
}}
tr:nth-child(even) {{
background-color: #f9f9f9;
}}
.status-success {{
color: #2e7d32;
font-weight: bold;
}}
.status-failed {{
color: #c62828;
font-weight: bold;
}}
.timestamp {{
color: #666;
font-size: 0.9em;
}}
</style>
</head>
<body>
<div class="container">
<h1>Informe de Comprobación VLAN35</h1>
<div class="header-info">
<p><strong>Fecha y hora:</strong> {timestamp}</p>
<p><strong>Total de tests ejecutados:</strong> {total}</p>
<div class="stats">
<div class="stat-box stat-success">
<div style="font-size: 24px; font-weight: bold;">{success}</div>
<div>Exitosos</div>
</div>
<div class="stat-box stat-failed">
<div style="font-size: 24px; font-weight: bold;">{failed}</div>
<div>Fallidos</div>
</div>
<div class="stat-box stat-total">
<div style="font-size: 24px; font-weight: bold;">{total}</div>
<div>Total</div>
</div>
</div>
</div>
<h2>Resultados Detallados</h2>
<table>
<tr>
<th>Host</th>
<th>Puerto</th>
<th>Protocolo</th>
<th>Estado</th>
<th>Latencia (ms)</th>
<th>Detalles</th>
<th>Timestamp</th>
</tr>
"""
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"""
<tr>
<td><strong>{result['host']}</strong></td>
<td>{result['port']}</td>
<td>{result['protocol']}</td>
<td class="{status_class}">{result['status']}</td>
<td>{latency}</td>
<td>{details}</td>
<td class="timestamp">{result['timestamp']}</td>
</tr>
"""
html += """
</table>
<h2>Resumen por Host</h2>
<table>
<tr>
<th>Host</th>
<th>Tests Exitosos</th>
<th>Tests Fallidos</th>
<th>Total</th>
</tr>
"""
# 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"""
<tr>
<td><strong>{host}</strong></td>
<td class="status-success">{stats['success']}</td>
<td class="status-failed">{stats['failed']}</td>
<td>{total_host}</td>
</tr>
"""
html += """
</table>
</div>
</body>
</html>
"""
return html
+233
View File
@@ -0,0 +1,233 @@
<!DOCTYPE html>
<html lang="es">
<head>
<meta charset="UTF-8">
<title>Informe de Comprobación VLAN35</title>
<style>
body {
font-family: 'Segoe UI', 'Roboto', sans-serif;
background-color: #f5f5f5;
color: #333;
margin: 0;
padding: 20px;
}
.container {
max-width: 1200px;
margin: 0 auto;
background: white;
padding: 30px;
box-shadow: 0 2px 10px rgba(0,0,0,0.1);
border-radius: 8px;
}
h1 {
color: #008542;
border-bottom: 3px solid #008542;
padding-bottom: 10px;
margin-bottom: 20px;
}
h2 {
color: #444;
margin-top: 30px;
border-left: 5px solid #008542;
padding-left: 10px;
}
.header-info {
background-color: #f8f9fa;
padding: 15px;
border-radius: 5px;
margin-bottom: 20px;
}
.stats {
display: flex;
gap: 20px;
margin-top: 15px;
}
.stat-box {
flex: 1;
padding: 15px;
border-radius: 5px;
text-align: center;
}
.stat-success {
background-color: #e8f5e9;
color: #2e7d32;
}
.stat-failed {
background-color: #ffebee;
color: #c62828;
}
.stat-total {
background-color: #e3f2fd;
color: #1565c0;
}
table {
width: 100%;
border-collapse: collapse;
margin-top: 15px;
font-size: 0.9em;
}
th, td {
border: 1px solid #e0e0e0;
padding: 10px;
text-align: left;
}
th {
background-color: #008542;
color: white;
font-weight: 600;
}
tr:nth-child(even) {
background-color: #f9f9f9;
}
.status-success {
color: #2e7d32;
font-weight: bold;
}
.status-failed {
color: #c62828;
font-weight: bold;
}
.timestamp {
color: #666;
font-size: 0.9em;
}
</style>
</head>
<body>
<div class="container">
<h1>Informe de Comprobación VLAN35</h1>
<div class="header-info">
<p><strong>Fecha y hora:</strong> 2026-02-16 11:10:26</p>
<p><strong>Total de tests ejecutados:</strong> 6</p>
<div class="stats">
<div class="stat-box stat-success">
<div style="font-size: 24px; font-weight: bold;">6</div>
<div>Exitosos</div>
</div>
<div class="stat-box stat-failed">
<div style="font-size: 24px; font-weight: bold;">0</div>
<div>Fallidos</div>
</div>
<div class="stat-box stat-total">
<div style="font-size: 24px; font-weight: bold;">6</div>
<div>Total</div>
</div>
</div>
</div>
<h2>Resultados Detallados</h2>
<table>
<tr>
<th>Host</th>
<th>Puerto</th>
<th>Protocolo</th>
<th>Estado</th>
<th>Latencia (ms)</th>
<th>Detalles</th>
<th>Timestamp</th>
</tr>
<tr>
<td><strong>10.47.8.133</strong></td>
<td>515</td>
<td>TCP</td>
<td class="status-success">Open</td>
<td>7.10</td>
<td>Conexión TCP exitosa en 7.10ms</td>
<td class="timestamp">2026-02-16 11:10:17</td>
</tr>
<tr>
<td><strong>10.47.8.133</strong></td>
<td>9195</td>
<td>TCP</td>
<td class="status-success">Open</td>
<td>7.88</td>
<td>Conexión TCP exitosa en 7.88ms</td>
<td class="timestamp">2026-02-16 11:10:17</td>
</tr>
<tr>
<td><strong>10.47.8.135</strong></td>
<td>9191</td>
<td>TCP</td>
<td class="status-success">Open</td>
<td>8.01</td>
<td>Conexión TCP exitosa en 8.01ms</td>
<td class="timestamp">2026-02-16 11:10:17</td>
</tr>
<tr>
<td><strong>10.47.8.135</strong></td>
<td>9192</td>
<td>TCP</td>
<td class="status-success">Open</td>
<td>4.41</td>
<td>Conexión TCP exitosa en 4.41ms</td>
<td class="timestamp">2026-02-16 11:10:17</td>
</tr>
<tr>
<td><strong>10.240.225.171</strong></td>
<td>465</td>
<td>TCP</td>
<td class="status-success">Open</td>
<td>3.96</td>
<td>Conexión TCP exitosa en 3.96ms</td>
<td class="timestamp">2026-02-16 11:10:17</td>
</tr>
<tr>
<td><strong>10.244.255.250</strong></td>
<td>123</td>
<td>UDP</td>
<td class="status-success">Open</td>
<td>5.11</td>
<td>NTP hora local: 2026-02-16 11:25:01</td>
<td class="timestamp">2026-02-16 11:10:17</td>
</tr>
</table>
<h2>Resumen por Host</h2>
<table>
<tr>
<th>Host</th>
<th>Tests Exitosos</th>
<th>Tests Fallidos</th>
<th>Total</th>
</tr>
<tr>
<td><strong>10.47.8.133</strong></td>
<td class="status-success">2</td>
<td class="status-failed">0</td>
<td>2</td>
</tr>
<tr>
<td><strong>10.47.8.135</strong></td>
<td class="status-success">2</td>
<td class="status-failed">0</td>
<td>2</td>
</tr>
<tr>
<td><strong>10.240.225.171</strong></td>
<td class="status-success">1</td>
<td class="status-failed">0</td>
<td>1</td>
</tr>
<tr>
<td><strong>10.244.255.250</strong></td>
<td class="status-success">1</td>
<td class="status-failed">0</td>
<td>1</td>
</tr>
</table>
</div>
</body>
</html>