JNC Windows
This commit is contained in:
joaquin.navajas
@@ -0,0 +1,173 @@
|
||||
"""
|
||||
Implementación de NetworkManager para Windows.
|
||||
Usa psutil + netsh; no requiere pyroute2 ni nmcli.
|
||||
"""
|
||||
import socket
|
||||
import subprocess
|
||||
import re
|
||||
from typing import Tuple, List, Dict, Any
|
||||
|
||||
import psutil
|
||||
|
||||
|
||||
def _parse_netsh_config() -> Dict[str, Dict[str, str]]:
|
||||
"""Ejecuta netsh interface ipv4 show config y devuelve dict por interfaz."""
|
||||
out: Dict[str, Dict[str, str]] = {}
|
||||
try:
|
||||
r = subprocess.run(
|
||||
["netsh", "interface", "ipv4", "show", "config"],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=15,
|
||||
creationflags=subprocess.CREATE_NO_WINDOW if hasattr(subprocess, "CREATE_NO_WINDOW") else 0,
|
||||
)
|
||||
if r.returncode != 0:
|
||||
return out
|
||||
current = None
|
||||
for line in r.stdout.splitlines():
|
||||
# "Configuration for interface "Ethernet""
|
||||
m = re.match(r'Configuration for interface "([^"]+)"', line)
|
||||
if m:
|
||||
current = m.group(1)
|
||||
out[current] = {"method": "unknown", "ip": "", "prefix": "", "gateway": "", "dns": ""}
|
||||
continue
|
||||
if current is None:
|
||||
continue
|
||||
line = line.strip()
|
||||
if line.startswith("DHCP enabled:"):
|
||||
val = line.split(":", 1)[1].strip().lower()
|
||||
out[current]["method"] = "auto" if val == "yes" else "manual"
|
||||
elif line.startswith("IP Address:"):
|
||||
out[current]["ip"] = line.split(":", 1)[1].strip()
|
||||
elif line.startswith("Subnet Prefix:"):
|
||||
val = line.split(":", 1)[1].strip()
|
||||
if "/" in val:
|
||||
ip_part, prefix_part = val.split("/", 1)
|
||||
out[current]["ip"] = out[current]["ip"] or ip_part.strip()
|
||||
out[current]["prefix"] = prefix_part.strip()
|
||||
elif line.startswith("Default Gateway:"):
|
||||
out[current]["gateway"] = line.split(":", 1)[1].strip()
|
||||
elif line.startswith("DNS Servers:"):
|
||||
out[current]["dns"] = line.split(":", 1)[1].strip()
|
||||
except Exception:
|
||||
pass
|
||||
return out
|
||||
|
||||
|
||||
def _is_mac_address(addr: str) -> bool:
|
||||
if not addr or len(addr) != 17:
|
||||
return False
|
||||
parts = addr.split(":")
|
||||
return len(parts) == 6 and all(len(p) == 2 and p.isalnum() for p in parts)
|
||||
|
||||
|
||||
class NetworkManager:
|
||||
"""NetworkManager para Windows: list_interfaces y get_interface_config con psutil + netsh."""
|
||||
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
def list_interfaces(self) -> List[Dict[str, Any]]:
|
||||
interfaces = []
|
||||
stats = psutil.net_if_stats()
|
||||
addrs = psutil.net_if_addrs()
|
||||
netsh_config = _parse_netsh_config()
|
||||
|
||||
for if_name, if_addrs in addrs.items():
|
||||
if if_name == "lo" or "Loopback" in if_name:
|
||||
continue
|
||||
is_up = False
|
||||
if if_name in stats:
|
||||
is_up = stats[if_name].isup
|
||||
ipv4 = []
|
||||
ipv6 = []
|
||||
mac = ""
|
||||
for addr in if_addrs:
|
||||
try:
|
||||
af = addr.family
|
||||
addr_str = addr.address
|
||||
except AttributeError:
|
||||
af = addr[0]
|
||||
addr_str = addr[1]
|
||||
if af == socket.AF_INET:
|
||||
ipv4.append(addr_str)
|
||||
elif af == socket.AF_INET6 and not addr_str.startswith("fe80") and addr_str != "::1":
|
||||
ipv6.append(addr_str)
|
||||
elif _is_mac_address(addr_str):
|
||||
mac = addr_str
|
||||
if not mac:
|
||||
for addr in if_addrs:
|
||||
try:
|
||||
a = getattr(addr, "address", addr[1])
|
||||
except (IndexError, TypeError):
|
||||
continue
|
||||
if _is_mac_address(a):
|
||||
mac = a
|
||||
break
|
||||
config = netsh_config.get(if_name, {
|
||||
"method": "unknown", "ip": "", "prefix": "", "gateway": "", "dns": ""
|
||||
})
|
||||
if ipv4 and not config.get("ip"):
|
||||
config = dict(config)
|
||||
config["ip"] = ipv4[0].split("/")[0] if "/" in str(ipv4[0]) else str(ipv4[0])
|
||||
# Formato ip/prefix como en Linux cuando hay prefix en netsh
|
||||
if config.get("prefix") and config.get("ip"):
|
||||
ipv4_formatted = [f"{config['ip']}/{config['prefix']}"]
|
||||
elif ipv4:
|
||||
ipv4_formatted = [a if "/" in str(a) else str(a) for a in ipv4]
|
||||
else:
|
||||
ipv4_formatted = []
|
||||
interfaces.append({
|
||||
"name": if_name,
|
||||
"mac": mac or "N/A",
|
||||
"status": "UP" if is_up else "DOWN",
|
||||
"ipv4": ipv4_formatted if ipv4 else [],
|
||||
"ipv6": ipv6,
|
||||
"type": "Wired",
|
||||
"is_up": is_up,
|
||||
"config": config,
|
||||
})
|
||||
return interfaces
|
||||
|
||||
def get_interface_config(self, iface_name: str) -> Dict[str, str]:
|
||||
configs = _parse_netsh_config()
|
||||
return configs.get(iface_name, {
|
||||
"method": "unknown", "ip": "", "prefix": "", "gateway": "", "dns": ""
|
||||
})
|
||||
|
||||
def apply_interface_config(
|
||||
self, iface_name: str, method: str,
|
||||
ip: str = None, prefix: str = None, gateway: str = None, dns: str = None
|
||||
) -> Tuple[bool, str]:
|
||||
return False, "Configuración IP no disponible en Windows desde esta aplicación."
|
||||
|
||||
def set_ip_config(self, iface_name: str, ip_address: str, prefix_len: str, gateway: str = None) -> Tuple[bool, str]:
|
||||
return False, "No disponible en Windows."
|
||||
|
||||
def set_dhcp(self, iface_name: str) -> Tuple[bool, str]:
|
||||
return False, "No disponible en Windows."
|
||||
|
||||
def create_vlan_bridge(self, trunk_iface: str, vlan_id: int, access_iface: str) -> Tuple[bool, str]:
|
||||
return False, "VLAN/Bridge no disponible en Windows."
|
||||
|
||||
def create_simple_bridge(self, iface1: str, iface2: str) -> Tuple[bool, str]:
|
||||
return False, "Bridge no disponible en Windows."
|
||||
|
||||
def start_dhcp_server(
|
||||
self, interface: str, ip: str, range_start: str, range_end: str
|
||||
) -> Tuple[bool, str]:
|
||||
return False, "DHCP no disponible en Windows."
|
||||
|
||||
def delete_bridge(self, bridge_name: str) -> Tuple[bool, str]:
|
||||
return False, "No disponible en Windows."
|
||||
|
||||
def setup_wifi_ap(
|
||||
self, wifi_iface: str, ssid: str, password: str, bridge_name: str
|
||||
) -> Tuple[bool, str]:
|
||||
return False, "Wi-Fi AP no disponible en Windows."
|
||||
|
||||
def reset_interface(self, iface_name: str) -> Tuple[bool, str]:
|
||||
return False, "Reinicio de interfaz no disponible en Windows."
|
||||
|
||||
def close(self) -> None:
|
||||
pass
|
||||
Reference in New Issue
Block a user