1130 lines
43 KiB
Python
1130 lines
43 KiB
Python
import tkinter as tk
|
|
from tkinter import ttk, messagebox
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import re
|
|
import ctypes
|
|
import sys
|
|
import ipaddress
|
|
from typing import List, Optional, Dict
|
|
from dataclasses import dataclass
|
|
import threading
|
|
import time
|
|
import socket
|
|
|
|
# Add libraries for ping and SNMP functionality
|
|
try:
|
|
from pythonping import ping as pythonping
|
|
|
|
PING_LIBRARY_AVAILABLE = True
|
|
except ImportError:
|
|
PING_LIBRARY_AVAILABLE = False
|
|
print(
|
|
"PythonPing module not found. Install with 'pip install pythonping' for better ping functionality."
|
|
)
|
|
|
|
try:
|
|
from pysnmp.hlapi import *
|
|
|
|
SNMP_AVAILABLE = True
|
|
except ImportError:
|
|
SNMP_AVAILABLE = False
|
|
print(
|
|
"pysnmp module not found. Install with 'pip install pysnmp' for SNMP functionality."
|
|
)
|
|
|
|
try:
|
|
import wmi
|
|
|
|
WMI_AVAILABLE = True
|
|
except ImportError:
|
|
WMI_AVAILABLE = False
|
|
print(
|
|
"WMI module not found. Using alternative method for network interface detection."
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class NetworkInterface:
|
|
name: str
|
|
description: str
|
|
ip_address: str = ""
|
|
subnet_mask: str = ""
|
|
gateway: str = ""
|
|
dhcp_enabled: bool = True
|
|
active: bool = True
|
|
adapter_id: str = ""
|
|
|
|
|
|
class IPChangeConfig:
|
|
def __init__(self):
|
|
self.config_file = "ip_config.json"
|
|
self.history_file = "ip_history.json"
|
|
self.ping_targets_file = "ping_targets.json" # New file for ping targets
|
|
self.load_config()
|
|
self.load_ping_targets() # Load ping targets
|
|
|
|
def load_config(self):
|
|
try:
|
|
if os.path.exists(self.config_file):
|
|
with open(self.config_file, "r") as f:
|
|
config = json.load(f)
|
|
self.last_interface = config.get("last_interface", "")
|
|
self.last_ip_prefix = config.get("last_ip_prefix", "")
|
|
self.previous_config = config.get("previous_config", {})
|
|
else:
|
|
self.last_interface = ""
|
|
self.last_ip_prefix = ""
|
|
self.previous_config = {}
|
|
except Exception as e:
|
|
print(f"Error loading config: {e}")
|
|
self.last_interface = ""
|
|
self.last_ip_prefix = ""
|
|
self.previous_config = {}
|
|
|
|
def save_config(self):
|
|
try:
|
|
config = {
|
|
"last_interface": self.last_interface,
|
|
"last_ip_prefix": self.last_ip_prefix,
|
|
"previous_config": self.previous_config,
|
|
}
|
|
with open(self.config_file, "w") as f:
|
|
json.dump(config, f)
|
|
except Exception as e:
|
|
print(f"Error saving config: {e}")
|
|
|
|
def load_ping_targets(self):
|
|
"""Load saved ping targets for each IP prefix"""
|
|
try:
|
|
if os.path.exists(self.ping_targets_file):
|
|
with open(self.ping_targets_file, "r") as f:
|
|
self.ping_targets = json.load(f)
|
|
else:
|
|
self.ping_targets = {} # Dictionary to store ping targets by IP prefix
|
|
except Exception as e:
|
|
print(f"Error loading ping targets: {e}")
|
|
self.ping_targets = {}
|
|
|
|
def save_ping_targets(self):
|
|
"""Save ping targets for each IP prefix"""
|
|
try:
|
|
with open(self.ping_targets_file, "w") as f:
|
|
json.dump(self.ping_targets, f)
|
|
except Exception as e:
|
|
print(f"Error saving ping targets: {e}")
|
|
|
|
def get_ping_target(self, ip_prefix):
|
|
"""Get the saved ping target for the given IP prefix"""
|
|
if not ip_prefix:
|
|
return ""
|
|
|
|
# Try to find exact match first
|
|
if ip_prefix in self.ping_targets:
|
|
return self.ping_targets[ip_prefix]
|
|
|
|
# If no exact match, try to find a matching IP prefix
|
|
for prefix, target in self.ping_targets.items():
|
|
if ip_prefix.startswith(prefix) or prefix.startswith(ip_prefix):
|
|
return target
|
|
|
|
# Default target: IP prefix + ".1" (usually the gateway)
|
|
return f"{ip_prefix}.1"
|
|
|
|
def set_ping_target(self, ip_prefix, target):
|
|
"""Save a ping target for the given IP prefix"""
|
|
if ip_prefix and target:
|
|
self.ping_targets[ip_prefix] = target
|
|
self.save_ping_targets()
|
|
|
|
|
|
class IPChangerApp:
|
|
def __init__(self, master):
|
|
self.master = master
|
|
self.master.title("Network Interface IP Configuration")
|
|
self.master.geometry("900x700")
|
|
|
|
# Inicializar configuración antes que nada
|
|
self.config = IPChangeConfig()
|
|
|
|
# Inicializar estructuras de datos
|
|
self.interfaces: List[NetworkInterface] = []
|
|
self.ip_history: List[str] = []
|
|
|
|
# Inicializar WMI si está disponible
|
|
if WMI_AVAILABLE:
|
|
self.wmi_client = wmi.WMI()
|
|
|
|
# Cargar datos guardados antes de crear widgets
|
|
self.load_ip_history()
|
|
|
|
# Crear la interfaz
|
|
self.create_widgets()
|
|
|
|
# Initialize trace variables
|
|
self.subnet_trace_id = None
|
|
self.cidr_trace_id = None
|
|
|
|
# Set up initial traces safely
|
|
self.setup_mask_traces()
|
|
|
|
# Actualizar la lista de IPs en el combo
|
|
self.update_history_display()
|
|
|
|
# Refrescar interfaces
|
|
self.refresh_interfaces()
|
|
|
|
# Configurar pesos de la cuadrícula
|
|
self.master.grid_columnconfigure(0, weight=1)
|
|
self.master.grid_rowconfigure(1, weight=1) # El log expandible
|
|
|
|
def setup_mask_traces(self):
|
|
"""Set up traces for subnet mask and CIDR fields safely"""
|
|
# First initialize variables
|
|
self.subnet_mask.set(self.subnet_mask.get()) # Ensure current value is valid
|
|
self.cidr_bits.set(self.cidr_bits.get()) # Ensure current value is valid
|
|
|
|
# Add traces using a different approach - using add not trace_add
|
|
self.subnet_mask.trace("w", self.on_subnet_mask_changed)
|
|
self.cidr_bits.trace("w", self.on_cidr_bits_changed)
|
|
|
|
self.log_message("Subnet mask traces initialized")
|
|
|
|
def remove_mask_traces(self):
|
|
"""Not needed - we'll use a different approach"""
|
|
pass # Intentionally empty
|
|
|
|
def get_ip_prefix(self, ip: str) -> str:
|
|
"""Extrae los primeros 3 octetos de una dirección IP"""
|
|
parts = ip.split(".")
|
|
if len(parts) >= 3:
|
|
return ".".join(parts[:3])
|
|
return ip
|
|
|
|
def create_widgets(self):
|
|
# Panel superior para controles
|
|
self.control_frame = ttk.Frame(self.master, padding="10")
|
|
self.control_frame.grid(row=0, column=0, sticky="new")
|
|
|
|
# Sección de interfaces
|
|
self.interfaces_frame = ttk.LabelFrame(
|
|
self.control_frame, text="Network Interfaces", padding="5"
|
|
)
|
|
self.interfaces_frame.grid(row=0, column=0, sticky="ew", pady=(0, 10))
|
|
|
|
ttk.Label(self.interfaces_frame, text="Select Interface:").grid(
|
|
row=0, column=0, sticky="w", padx=5
|
|
)
|
|
self.if_var = tk.StringVar()
|
|
self.if_combo = ttk.Combobox(
|
|
self.interfaces_frame, textvariable=self.if_var, state="readonly", width=50
|
|
)
|
|
self.if_combo.grid(row=0, column=1, sticky="ew", padx=5)
|
|
self.if_combo.bind("<<ComboboxSelected>>", self.on_interface_selected)
|
|
|
|
# Info de configuración actual
|
|
self.current_frame = ttk.LabelFrame(
|
|
self.control_frame, text="Current Configuration", padding="5"
|
|
)
|
|
self.current_frame.grid(row=1, column=0, sticky="ew", pady=(0, 10))
|
|
|
|
self.current_ip_var = tk.StringVar()
|
|
self.current_mask_var = tk.StringVar()
|
|
self.current_gateway_var = tk.StringVar()
|
|
self.dhcp_status_var = tk.StringVar()
|
|
|
|
ttk.Label(self.current_frame, text="Current IP:").grid(
|
|
row=0, column=0, sticky="w", padx=5
|
|
)
|
|
ttk.Label(self.current_frame, textvariable=self.current_ip_var).grid(
|
|
row=0, column=1, sticky="w"
|
|
)
|
|
ttk.Label(self.current_frame, text="Subnet Mask:").grid(
|
|
row=1, column=0, sticky="w", padx=5
|
|
)
|
|
ttk.Label(self.current_frame, textvariable=self.current_mask_var).grid(
|
|
row=1, column=1, sticky="w"
|
|
)
|
|
ttk.Label(self.current_frame, text="Gateway:").grid(
|
|
row=2, column=0, sticky="w", padx=5
|
|
)
|
|
ttk.Label(self.current_frame, textvariable=self.current_gateway_var).grid(
|
|
row=2, column=1, sticky="w"
|
|
)
|
|
ttk.Label(self.current_frame, text="DHCP Status:").grid(
|
|
row=3, column=0, sticky="w", padx=5
|
|
)
|
|
ttk.Label(self.current_frame, textvariable=self.dhcp_status_var).grid(
|
|
row=3, column=1, sticky="w"
|
|
)
|
|
|
|
# Sección de configuración IP
|
|
self.ip_frame = ttk.LabelFrame(
|
|
self.control_frame, text="IP Configuration", padding="5"
|
|
)
|
|
self.ip_frame.grid(row=2, column=0, sticky="ew", pady=(0, 10))
|
|
|
|
ttk.Label(self.ip_frame, text="IP Prefix (first 3 octets):").grid(
|
|
row=0, column=0, sticky="w", padx=5
|
|
)
|
|
self.ip_prefix = tk.StringVar(value=self.config.last_ip_prefix)
|
|
self.ip_entry = ttk.Entry(self.ip_frame, textvariable=self.ip_prefix, width=30)
|
|
self.ip_entry.grid(row=0, column=1, sticky="w", padx=5)
|
|
|
|
ttk.Label(self.ip_frame, text="Last Octet:").grid(
|
|
row=0, column=2, sticky="w", padx=5
|
|
)
|
|
self.last_octet = tk.StringVar(value="249")
|
|
self.last_octet_entry = ttk.Entry(
|
|
self.ip_frame, textvariable=self.last_octet, width=5
|
|
)
|
|
self.last_octet_entry.grid(row=0, column=3, sticky="w", padx=5)
|
|
|
|
# Add subnet mask configuration
|
|
ttk.Label(self.ip_frame, text="Subnet Mask:").grid(
|
|
row=1, column=0, sticky="w", padx=5
|
|
)
|
|
self.subnet_mask = tk.StringVar(value="255.255.255.0")
|
|
self.subnet_entry = ttk.Entry(
|
|
self.ip_frame, textvariable=self.subnet_mask, width=15
|
|
)
|
|
self.subnet_entry.grid(row=1, column=1, sticky="w", padx=5)
|
|
|
|
ttk.Label(self.ip_frame, text="CIDR (/bits):").grid(
|
|
row=1, column=2, sticky="w", padx=5
|
|
)
|
|
self.cidr_bits = tk.StringVar(value="24")
|
|
self.cidr_entry = ttk.Entry(self.ip_frame, textvariable=self.cidr_bits, width=5)
|
|
self.cidr_entry.grid(row=1, column=3, sticky="w", padx=5)
|
|
|
|
# Don't add traces here, we'll do it after widget creation is complete
|
|
|
|
# Sección de historial
|
|
self.history_frame = ttk.LabelFrame(
|
|
self.control_frame, text="IP History", padding="5"
|
|
)
|
|
self.history_frame.grid(row=3, column=0, sticky="ew", pady=(0, 10))
|
|
|
|
ttk.Label(self.history_frame, text="Previous IPs:").grid(
|
|
row=0, column=0, sticky="w", padx=5
|
|
)
|
|
self.history_var = tk.StringVar()
|
|
self.history_combo = ttk.Combobox(
|
|
self.history_frame,
|
|
textvariable=self.history_var,
|
|
state="readonly",
|
|
width=50,
|
|
)
|
|
self.history_combo.grid(row=0, column=1, sticky="ew", padx=5)
|
|
self.history_combo.bind("<<ComboboxSelected>>", self.on_history_selected)
|
|
|
|
# Botones de acción
|
|
self.button_frame = ttk.Frame(self.control_frame)
|
|
self.button_frame.grid(row=4, column=0, sticky="ew", pady=(0, 10))
|
|
|
|
ttk.Button(
|
|
self.button_frame, text="Set Static IP", command=self.set_static_ip
|
|
).grid(row=0, column=0, padx=5)
|
|
ttk.Button(self.button_frame, text="Enable DHCP", command=self.set_dhcp).grid(
|
|
row=0, column=1, padx=5
|
|
)
|
|
ttk.Button(
|
|
self.button_frame, text="Restore Previous", command=self.restore_previous
|
|
).grid(row=0, column=2, padx=5)
|
|
ttk.Button(
|
|
self.button_frame,
|
|
text="Refresh Interfaces",
|
|
command=self.refresh_interfaces,
|
|
).grid(row=0, column=3, padx=5)
|
|
|
|
# Sección de Ping - add initial value and bind event
|
|
self.ping_frame = ttk.LabelFrame(
|
|
self.control_frame, text="Network Tools", padding="5"
|
|
)
|
|
self.ping_frame.grid(row=5, column=0, sticky="ew", pady=(0, 10))
|
|
|
|
ttk.Label(self.ping_frame, text="Target IP/Host:").grid(
|
|
row=0, column=0, sticky="w", padx=5
|
|
)
|
|
self.ping_target = tk.StringVar()
|
|
self.ping_entry = ttk.Entry(
|
|
self.ping_frame, textvariable=self.ping_target, width=40
|
|
)
|
|
self.ping_entry.grid(row=0, column=1, sticky="w", padx=5)
|
|
# Bind the FocusOut event to save the ping target
|
|
self.ping_entry.bind("<FocusOut>", self.save_current_ping_target)
|
|
ttk.Button(self.ping_frame, text="Ping", command=self.do_ping).grid(
|
|
row=0, column=2, padx=5
|
|
)
|
|
|
|
# Log en la parte inferior
|
|
self.log_frame = ttk.LabelFrame(self.master, text="Operation Log", padding="5")
|
|
self.log_frame.grid(row=1, column=0, sticky="nsew", padx=10, pady=(0, 10))
|
|
|
|
self.log_text = tk.Text(self.log_frame, wrap="none", height=10)
|
|
self.log_scrollbar_y = ttk.Scrollbar(
|
|
self.log_frame, orient="vertical", command=self.log_text.yview
|
|
)
|
|
self.log_scrollbar_x = ttk.Scrollbar(
|
|
self.log_frame, orient="horizontal", command=self.log_text.xview
|
|
)
|
|
|
|
self.log_text.configure(
|
|
yscrollcommand=self.log_scrollbar_y.set,
|
|
xscrollcommand=self.log_scrollbar_x.set,
|
|
)
|
|
|
|
self.log_text.grid(row=0, column=0, sticky="nsew")
|
|
self.log_scrollbar_y.grid(row=0, column=1, sticky="ns")
|
|
self.log_scrollbar_x.grid(row=1, column=0, sticky="ew")
|
|
|
|
ttk.Button(self.log_frame, text="Clear Log", command=self.clear_log).grid(
|
|
row=2, column=0, columnspan=2, pady=5
|
|
)
|
|
|
|
self.log_frame.grid_columnconfigure(0, weight=1)
|
|
self.log_frame.grid_rowconfigure(0, weight=1)
|
|
|
|
def clear_log(self):
|
|
self.log_text.delete("1.0", tk.END)
|
|
|
|
def log_message(self, message: str):
|
|
self.log_text.insert(tk.END, f"{message}\n")
|
|
self.log_text.see(tk.END)
|
|
self.log_text.update_idletasks()
|
|
|
|
def load_ip_history(self):
|
|
try:
|
|
if os.path.exists(self.config.history_file):
|
|
with open(self.config.history_file, "r") as f:
|
|
self.ip_history = json.load(f)
|
|
else:
|
|
self.ip_history = []
|
|
except Exception as e:
|
|
self.log_message(f"Error loading IP history: {e}")
|
|
self.ip_history = []
|
|
|
|
def save_ip_history(self):
|
|
try:
|
|
with open(self.config.history_file, "w") as f:
|
|
json.dump(self.ip_history, f)
|
|
except Exception as e:
|
|
self.log_message(f"Error saving IP history: {e}")
|
|
|
|
def add_to_history(self, ip_prefix: str):
|
|
if ip_prefix and ip_prefix not in self.ip_history:
|
|
self.ip_history.insert(0, ip_prefix)
|
|
self.ip_history = self.ip_history[:15] # Mantener solo las últimas 15 IPs
|
|
self.save_ip_history()
|
|
self.update_history_display()
|
|
|
|
def update_history_display(self):
|
|
self.history_combo["values"] = self.ip_history
|
|
if self.ip_history:
|
|
self.history_combo.set(self.ip_history[0])
|
|
|
|
def on_history_selected(self, event):
|
|
selected_ip = self.history_var.get()
|
|
if selected_ip:
|
|
# Ensure we only get the first three octets
|
|
ip_parts = selected_ip.split(".")
|
|
if len(ip_parts) >= 3:
|
|
ip_prefix = ".".join(ip_parts[:3])
|
|
self.ip_prefix.set(ip_prefix)
|
|
self.config.last_ip_prefix = ip_prefix
|
|
self.config.save_config()
|
|
self.log_message(f"Selected IP prefix from history: {ip_prefix}")
|
|
|
|
# Update ping target for the new IP prefix
|
|
self.update_ping_target(ip_prefix)
|
|
|
|
def refresh_interfaces(self):
|
|
self.interfaces.clear()
|
|
try:
|
|
if WMI_AVAILABLE:
|
|
self._refresh_interfaces_wmi()
|
|
else:
|
|
self._refresh_interfaces_netsh()
|
|
|
|
# Actualizar combobox
|
|
self.if_combo["values"] = [
|
|
inf.name for inf in self.interfaces if inf.active
|
|
]
|
|
|
|
# Seleccionar última interfaz usada si está disponible
|
|
if self.config.last_interface in self.if_combo["values"]:
|
|
self.if_combo.set(self.config.last_interface)
|
|
self.on_interface_selected()
|
|
|
|
self.log_message("Interfaces refreshed successfully")
|
|
|
|
except Exception as e:
|
|
self.log_message(f"Error refreshing interfaces: {str(e)}")
|
|
self.show_error(f"Error refreshing interfaces: {str(e)}")
|
|
|
|
def _refresh_interfaces_wmi(self):
|
|
try:
|
|
adapters = self.wmi_client.Win32_NetworkAdapter(PhysicalAdapter=True)
|
|
configs = self.wmi_client.Win32_NetworkAdapterConfiguration()
|
|
|
|
for adapter in adapters:
|
|
config = next(
|
|
(
|
|
cfg
|
|
for cfg in configs
|
|
if cfg.InterfaceIndex == adapter.InterfaceIndex
|
|
),
|
|
None,
|
|
)
|
|
|
|
if config and config.IPEnabled:
|
|
interface = NetworkInterface(
|
|
name=adapter.NetConnectionID,
|
|
description=adapter.Description,
|
|
ip_address=config.IPAddress[0] if config.IPAddress else "",
|
|
subnet_mask=config.IPSubnet[0] if config.IPSubnet else "",
|
|
gateway=(
|
|
config.DefaultIPGateway[0]
|
|
if config.DefaultIPGateway
|
|
else ""
|
|
),
|
|
dhcp_enabled=config.DHCPEnabled,
|
|
active=adapter.NetEnabled,
|
|
adapter_id=adapter.GUID,
|
|
)
|
|
self.interfaces.append(interface)
|
|
except Exception as e:
|
|
raise Exception(f"WMI refresh error: {str(e)}")
|
|
|
|
def _refresh_interfaces_netsh(self):
|
|
try:
|
|
output = subprocess.check_output(
|
|
"netsh interface ipv4 show config", shell=True, text=True
|
|
)
|
|
configs = output.split("\n\n")
|
|
|
|
for config in configs:
|
|
if not config.strip():
|
|
continue
|
|
|
|
name_match = re.search(r"Configuración de\s+\"(.+?)\"", config)
|
|
if not name_match:
|
|
continue
|
|
|
|
name = name_match.group(1)
|
|
|
|
ip_match = re.search(r"Dirección IP:\s+(\d+\.\d+\.\d+\.\d+)", config)
|
|
mask_match = re.search(
|
|
r"Máscara de subred:\s+(\d+\.\d+\.\d+\.\d+)", config
|
|
)
|
|
gateway_match = re.search(
|
|
r"Puerta de enlace predeterminada:\s+(\d+\.\d+\.\d+\.\d+)", config
|
|
)
|
|
dhcp_match = re.search(r"DHCP habilitado:\s+(\w+)", config)
|
|
|
|
interface = NetworkInterface(
|
|
name=name,
|
|
description=name,
|
|
ip_address=ip_match.group(1) if ip_match else "",
|
|
subnet_mask=mask_match.group(1) if mask_match else "",
|
|
gateway=gateway_match.group(1) if gateway_match else "",
|
|
dhcp_enabled=(
|
|
dhcp_match.group(1).lower() == "sí" if dhcp_match else True
|
|
),
|
|
active=True,
|
|
)
|
|
self.interfaces.append(interface)
|
|
|
|
except subprocess.CalledProcessError as e:
|
|
raise Exception(f"Error executing netsh: {str(e)}")
|
|
except Exception as e:
|
|
raise Exception(f"Error parsing netsh output: {str(e)}")
|
|
|
|
def on_interface_selected(self, event=None):
|
|
selected = self.if_var.get()
|
|
interface = next((inf for inf in self.interfaces if inf.name == selected), None)
|
|
|
|
if interface:
|
|
self.current_ip_var.set(interface.ip_address)
|
|
self.current_mask_var.set(interface.subnet_mask)
|
|
self.current_gateway_var.set(interface.gateway)
|
|
self.dhcp_status_var.set(
|
|
"Enabled" if interface.dhcp_enabled else "Disabled"
|
|
)
|
|
|
|
# Actualizar IP prefix si hay una IP válida
|
|
if interface.ip_address:
|
|
prefix = ".".join(interface.ip_address.split(".")[:3])
|
|
self.ip_prefix.set(prefix)
|
|
|
|
# Update ping target for the new IP prefix
|
|
self.update_ping_target(prefix)
|
|
|
|
# Guardar interfaz seleccionada
|
|
self.config.last_interface = selected
|
|
self.config.save_config()
|
|
|
|
self.log_message(f"Selected interface: {selected}")
|
|
|
|
def validate_ip_input(self) -> tuple[bool, str]:
|
|
try:
|
|
prefix = self.ip_prefix.get().strip()
|
|
last_octet = self.last_octet.get().strip()
|
|
subnet_mask = self.subnet_mask.get().strip()
|
|
|
|
# Validar formato del prefijo
|
|
if not re.match(r"^\d{1,3}\.\d{1,3}\.\d{1,3}", prefix):
|
|
return False, "IP prefix must be in format: xxx.xxx.xxx"
|
|
|
|
# Validar último octeto
|
|
if not last_octet.isdigit() or not 0 <= int(last_octet) <= 255:
|
|
return False, "Last octet must be between 0 and 255"
|
|
|
|
# Validar máscara de subred
|
|
if not self.is_valid_subnet_mask(subnet_mask):
|
|
return False, "Invalid subnet mask format"
|
|
|
|
# Construir y validar IP completa
|
|
ip = f"{prefix}.{last_octet}"
|
|
ipaddress.IPv4Address(ip)
|
|
|
|
return True, ip
|
|
|
|
except ValueError as e:
|
|
return False, f"Invalid IP address: {str(e)}"
|
|
|
|
def execute_admin_script(
|
|
self,
|
|
interface: str,
|
|
mode: str,
|
|
debug: bool = True,
|
|
subnet_mask: str = "255.255.255.0",
|
|
) -> bool:
|
|
"""
|
|
Ejecuta el script administrativo con los argumentos proporcionados
|
|
mode puede ser una IP o 'dhcp'
|
|
"""
|
|
# Obtener rutas absolutas
|
|
current_dir = os.path.dirname(os.path.abspath(__file__))
|
|
script_path = os.path.join(current_dir, "ip-changer-admin.py")
|
|
python_exe = sys.executable
|
|
|
|
# Verificar que el script existe
|
|
if not os.path.exists(script_path):
|
|
error_msg = f"Admin script not found at: {script_path}"
|
|
self.log_message(error_msg)
|
|
return False
|
|
|
|
# Construir la línea de argumentos - pass both IP and subnet as a single parameter
|
|
if mode != "dhcp":
|
|
# For static IP, create the netsh command with subnet mask
|
|
gateway = (
|
|
f"{self.get_ip_prefix(mode)}.1" # Use the first three octets of the IP
|
|
)
|
|
netsh_cmd = f'netsh interface ip set address "{interface}" static {mode} {subnet_mask} {gateway}'
|
|
args = f'"{script_path}" "{interface}" "{netsh_cmd}"'
|
|
else:
|
|
# For DHCP, keep it simple
|
|
args = f'"{script_path}" "{interface}" "dhcp"'
|
|
|
|
if debug:
|
|
self.log_message("Debug Information:")
|
|
self.log_message(f"Current Directory: {current_dir}")
|
|
self.log_message(f"Script Path: {script_path}")
|
|
self.log_message(f"Python Executable: {python_exe}")
|
|
self.log_message(f"Full Command: {python_exe} {args}")
|
|
|
|
try:
|
|
self.log_message(
|
|
f"Attempting to execute admin script with elevated privileges"
|
|
)
|
|
|
|
ret = ctypes.windll.shell32.ShellExecuteW(
|
|
None, # hwnd
|
|
"runas", # operation
|
|
python_exe, # file
|
|
args, # parameters
|
|
current_dir, # directory
|
|
1, # show command
|
|
)
|
|
|
|
result_code = int(ret)
|
|
if result_code > 32:
|
|
self.log_message(
|
|
f"ShellExecuteW successful (return code: {result_code})"
|
|
)
|
|
return True
|
|
else:
|
|
error_codes = {
|
|
0: "The operating system is out of memory or resources",
|
|
2: "The specified file was not found",
|
|
3: "The specified path was not found",
|
|
5: "Access denied",
|
|
8: "Insufficient memory to complete the operation",
|
|
11: "Bad format",
|
|
26: "A sharing violation occurred",
|
|
27: "The file name association is incomplete or invalid",
|
|
28: "The DDE operation timed out",
|
|
29: "The DDE operation failed",
|
|
30: "The DDE operation is busy",
|
|
31: "The file name association is unavailable",
|
|
32: "No application is associated with the file",
|
|
}
|
|
|
|
self.log_message(
|
|
f"Failed to execute admin script. Error code {result_code}: {error_msg}"
|
|
)
|
|
return False
|
|
|
|
except Exception as e:
|
|
self.log_message(f"Exception while executing admin script: {str(e)}")
|
|
return False
|
|
|
|
def save_previous_config(self, interface_name: str):
|
|
"""Guarda la configuración actual de la interfaz antes de cambiarla"""
|
|
interface = next(
|
|
(inf for inf in self.interfaces if inf.name == interface_name), None
|
|
)
|
|
if interface:
|
|
self.config.previous_config = {
|
|
"name": interface.name,
|
|
"ip_address": interface.ip_address,
|
|
"subnet_mask": interface.subnet_mask,
|
|
"gateway": interface.gateway,
|
|
"dhcp_enabled": interface.dhcp_enabled,
|
|
}
|
|
self.config.save_config()
|
|
self.log_message(f"Previous configuration saved for {interface.name}")
|
|
|
|
def restore_previous(self):
|
|
"""Restaura la configuración IP anterior"""
|
|
prev_config = self.config.previous_config
|
|
if not prev_config:
|
|
self.show_error("No previous configuration available")
|
|
return
|
|
|
|
interface_name = prev_config.get("name")
|
|
if not interface_name:
|
|
self.show_error("Invalid previous configuration")
|
|
return
|
|
|
|
# Verificar que la interfaz existe
|
|
if interface_name not in [inf.name for inf in self.interfaces]:
|
|
self.show_error(f"Interface {interface_name} not found")
|
|
return
|
|
|
|
self.log_message(f"Restoring previous configuration for {interface_name}")
|
|
|
|
if prev_config.get("dhcp_enabled", True):
|
|
# Si la configuración anterior era DHCP
|
|
if self.execute_admin_script(interface_name, "dhcp"):
|
|
self.log_message(f"Successfully restored DHCP on {interface_name}")
|
|
time.sleep(2)
|
|
self.refresh_interfaces()
|
|
else:
|
|
self.show_error("Failed to restore DHCP configuration")
|
|
else:
|
|
# Si la configuración anterior era IP estática
|
|
ip = prev_config.get("ip_address")
|
|
if not ip:
|
|
self.show_error("Previous IP address not available")
|
|
return
|
|
|
|
if self.execute_admin_script(interface_name, ip):
|
|
self.log_message(f"Successfully restored IP {ip} on {interface_name}")
|
|
time.sleep(2)
|
|
self.refresh_interfaces()
|
|
else:
|
|
self.show_error("Failed to restore static IP configuration")
|
|
|
|
def do_ping(self):
|
|
"""Realiza un ping a la dirección especificada usando una biblioteca de Python"""
|
|
target = self.ping_target.get().strip()
|
|
if not target:
|
|
self.show_error("Please enter a target IP or hostname")
|
|
return
|
|
|
|
# Save the ping target when used
|
|
self.save_current_ping_target()
|
|
|
|
self.log_message(f"Pinging {target}...")
|
|
|
|
# Crear un hilo para el ping para no bloquear la interfaz
|
|
threading.Thread(target=self._execute_ping, args=(target,), daemon=True).start()
|
|
|
|
def _execute_ping(self, target: str):
|
|
"""Ejecuta el ping usando bibliotecas de Python en lugar del comando del sistema"""
|
|
try:
|
|
if PING_LIBRARY_AVAILABLE:
|
|
# Usar PythonPing que es multiplataforma y no requiere privilegios
|
|
self.log_message(
|
|
f"Sending 4 ICMP echo requests to {target} using PythonPing..."
|
|
)
|
|
|
|
# Ejecutar el ping
|
|
response = pythonping(target, count=4, timeout=1)
|
|
|
|
# Comprobar si el ping tuvo éxito
|
|
if response.success():
|
|
min_rtt = min(r.time_elapsed_ms for r in response)
|
|
max_rtt = max(r.time_elapsed_ms for r in response)
|
|
avg_rtt = sum(r.time_elapsed_ms for r in response) / len(response)
|
|
|
|
# Mostrar resultados individuales
|
|
for i, reply in enumerate(response):
|
|
self.log_message(
|
|
f"Reply from {target}: time={reply.time_elapsed_ms:.2f}ms"
|
|
)
|
|
|
|
# Resumen
|
|
success_count = len([r for r in response if r.success])
|
|
self.log_message(f"Ping statistics for {target}:")
|
|
self.log_message(
|
|
f" Packets: Sent = 4, Received = {success_count}, Lost = {4 - success_count}"
|
|
)
|
|
self.log_message(f"Approximate round trip times in milliseconds:")
|
|
self.log_message(
|
|
f" Minimum = {min_rtt:.2f}ms, Maximum = {max_rtt:.2f}ms, Average = {avg_rtt:.2f}ms"
|
|
)
|
|
else:
|
|
self.log_message(f"Could not reach host {target}")
|
|
else:
|
|
# Usar socket para hacer un ping básico (solo comprueba si el host está activo)
|
|
self.log_message(
|
|
f"PythonPing not available, checking if host {target} is reachable..."
|
|
)
|
|
self.log_message(
|
|
"Note: This is not a true ICMP ping, just a TCP connection test"
|
|
)
|
|
|
|
for port in [80, 443, 22, 21]:
|
|
try:
|
|
start_time = time.time()
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
s.settimeout(1)
|
|
result = s.connect_ex((target, port))
|
|
elapsed_time = (time.time() - start_time) * 1000 # ms
|
|
s.close()
|
|
|
|
if result == 0:
|
|
self.log_message(
|
|
f"Port {port} on {target} is open (TCP connect time: {elapsed_time:.2f}ms)"
|
|
)
|
|
else:
|
|
self.log_message(
|
|
f"Port {port} on {target} is not responding"
|
|
)
|
|
except socket.gaierror:
|
|
self.log_message(f"Could not resolve hostname {target}")
|
|
break
|
|
except socket.error as e:
|
|
self.log_message(f"Error connecting to {target}: {str(e)}")
|
|
except AttributeError as e:
|
|
# Handle specifically the attribute error
|
|
self.log_message(f"Error with PythonPing API: {str(e)}")
|
|
|
|
# Fallback to alternative ping implementation
|
|
self.log_message("Falling back to socket-based connectivity test...")
|
|
|
|
# Implement a simple socket-based ping
|
|
success = False
|
|
try:
|
|
# Try to connect to the host - this just checks if it's reachable
|
|
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
s.settimeout(2)
|
|
start_time = time.time()
|
|
s.connect((target, 80)) # Try to connect to port 80
|
|
response_time = (time.time() - start_time) * 1000
|
|
s.close()
|
|
self.log_message(
|
|
f"Host {target} is reachable. Response time: {response_time:.2f}ms"
|
|
)
|
|
success = True
|
|
except socket.error as se:
|
|
self.log_message(f"Could not connect to {target}: {str(se)}")
|
|
|
|
# Try to at least resolve the hostname
|
|
try:
|
|
ip = socket.gethostbyname(target)
|
|
self.log_message(
|
|
f"Hostname {target} resolves to {ip}, but connection failed"
|
|
)
|
|
except socket.gaierror:
|
|
self.log_message(f"Could not resolve hostname {target}")
|
|
|
|
# Alternative implementation using subprocess directly as last resort
|
|
try:
|
|
self.log_message("Attempting to ping using system command...")
|
|
# Use subprocess to run ping directly (won't work if command not available)
|
|
use_count = "-n" if sys.platform.lower() == "win32" else "-c"
|
|
cmd = ["ping", use_count, "4", target]
|
|
process = subprocess.Popen(
|
|
cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE
|
|
)
|
|
stdout, stderr = process.communicate()
|
|
|
|
if process.returncode == 0:
|
|
# Process and display output
|
|
output = stdout.decode("utf-8", errors="replace")
|
|
for line in output.splitlines():
|
|
if line.strip():
|
|
self.log_message(line)
|
|
else:
|
|
self.log_message(
|
|
f"System ping command failed: {stderr.decode('utf-8', errors='replace')}"
|
|
)
|
|
except Exception as subproc_e:
|
|
self.log_message(f"System ping command error: {str(subproc_e)}")
|
|
|
|
except Exception as e:
|
|
self.log_message(f"Error executing ping: {str(e)}")
|
|
self.log_message(
|
|
"PythonPing may need to be reinstalled or updated. Run: pip install -U pythonping"
|
|
)
|
|
|
|
# Try to give more information
|
|
try:
|
|
import pkg_resources
|
|
|
|
version = pkg_resources.get_distribution("pythonping").version
|
|
self.log_message(f"Installed PythonPing version: {version}")
|
|
except Exception:
|
|
self.log_message("Could not determine PythonPing version")
|
|
|
|
def set_static_ip(self):
|
|
interface = self.if_var.get()
|
|
if not interface:
|
|
self.show_error("Please select a network interface")
|
|
return
|
|
|
|
# Guardar configuración actual antes de cambiarla
|
|
self.save_previous_config(interface)
|
|
|
|
# Validar IP
|
|
is_valid, result = self.validate_ip_input()
|
|
if not is_valid:
|
|
self.show_error(result)
|
|
return
|
|
|
|
ip = result
|
|
subnet_mask = self.subnet_mask.get()
|
|
|
|
# Guardar IP actual en el historial
|
|
self.add_to_history(self.ip_prefix.get())
|
|
|
|
# Log the actual command we'll be executing
|
|
gateway = f"{self.ip_prefix.get()}.1"
|
|
command = f'netsh interface ip set address "{interface}" static {ip} {subnet_mask} {gateway}'
|
|
self.log_message(f"Executing network command: {command}")
|
|
|
|
# Ejecutar script con privilegios - pass subnet mask
|
|
if self.execute_admin_script(
|
|
interface, ip, debug=True, subnet_mask=subnet_mask
|
|
):
|
|
time.sleep(2) # Esperar a que se apliquen los cambios
|
|
self.refresh_interfaces()
|
|
self.log_message(
|
|
f"Successfully set static IP {ip} with mask {subnet_mask} on {interface}"
|
|
)
|
|
else:
|
|
self.show_error("Failed to set static IP. Check the log for details.")
|
|
|
|
def set_dhcp(self):
|
|
interface = self.if_var.get()
|
|
if not interface:
|
|
self.show_error("Please select a network interface")
|
|
return
|
|
|
|
# Guardar configuración actual antes de cambiarla
|
|
self.save_previous_config(interface)
|
|
|
|
# Ejecutar script con privilegios
|
|
if self.execute_admin_script(interface, "dhcp", debug=True):
|
|
time.sleep(2) # Esperar a que se apliquen los cambios
|
|
self.refresh_interfaces()
|
|
self.log_message(f"Successfully enabled DHCP on {interface}")
|
|
else:
|
|
self.show_error("Failed to enable DHCP. Check the log for details.")
|
|
|
|
def run_command(self, cmd: str) -> bool:
|
|
self.log_message(f"Executing command: {cmd}")
|
|
try:
|
|
process = subprocess.Popen(
|
|
cmd,
|
|
shell=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
universal_newlines=True,
|
|
)
|
|
|
|
while True:
|
|
output = process.stdout.readline()
|
|
if output == "" and process.poll() is not None:
|
|
break
|
|
if output:
|
|
self.log_message(output.strip())
|
|
|
|
retcode = process.poll()
|
|
|
|
error_output = process.stderr.read()
|
|
if error_output:
|
|
self.log_message(f"Error output:\n{error_output}")
|
|
|
|
if retcode == 0:
|
|
self.log_message("Command executed successfully")
|
|
return True
|
|
else:
|
|
self.log_message(f"Command failed with return code: {retcode}")
|
|
return False
|
|
|
|
except Exception as e:
|
|
self.log_message(f"Error executing command: {str(e)}")
|
|
return False
|
|
|
|
def is_admin(self):
|
|
try:
|
|
return ctypes.windll.shell32.IsUserAnAdmin()
|
|
except Exception as e:
|
|
self.log_message(f"Error checking admin privileges: {str(e)}")
|
|
return False
|
|
|
|
def elevate_privileges(self):
|
|
if sys.platform == "win32":
|
|
self.show_info(
|
|
"The application needs administrator privileges to change network settings. "
|
|
"Please confirm the UAC prompt."
|
|
)
|
|
script_path = os.path.abspath(sys.argv[0])
|
|
try:
|
|
ctypes.windll.shell32.ShellExecuteW(
|
|
None, "runas", sys.executable, f'"{script_path}"', None, 1
|
|
)
|
|
self.master.quit()
|
|
except Exception as e:
|
|
self.log_message(f"Error elevating privileges: {str(e)}")
|
|
self.show_error("Failed to elevate privileges")
|
|
|
|
def show_error(self, message: str):
|
|
self.log_message(f"ERROR: {message}")
|
|
messagebox.showerror("Error", message)
|
|
|
|
def show_info(self, message: str):
|
|
self.log_message(f"INFO: {message}")
|
|
messagebox.showinfo("Information", message)
|
|
|
|
# Add new methods for subnet mask conversion
|
|
def on_subnet_mask_changed(self, *args):
|
|
"""Update CIDR bits when subnet mask changes"""
|
|
# Use a static variable to prevent recursion
|
|
if hasattr(self, "_updating_mask") and self._updating_mask:
|
|
return
|
|
|
|
try:
|
|
self._updating_mask = True
|
|
|
|
mask = self.subnet_mask.get()
|
|
if self.is_valid_subnet_mask(mask):
|
|
# Convert mask to bits
|
|
bits = self.subnet_mask_to_cidr(mask)
|
|
self.cidr_bits.set(str(bits))
|
|
except Exception as e:
|
|
self.log_message(f"Error updating CIDR bits: {str(e)}")
|
|
finally:
|
|
self._updating_mask = False
|
|
|
|
def on_cidr_bits_changed(self, *args):
|
|
"""Update subnet mask when CIDR bits change"""
|
|
# Use a static variable to prevent recursion
|
|
if hasattr(self, "_updating_bits") and self._updating_bits:
|
|
return
|
|
|
|
try:
|
|
self._updating_bits = True
|
|
|
|
bits_str = self.cidr_bits.get()
|
|
if bits_str.isdigit():
|
|
bits = int(bits_str)
|
|
if 0 <= bits <= 32:
|
|
# Convert bits to mask
|
|
mask = self.cidr_to_subnet_mask(bits)
|
|
self.subnet_mask.set(mask)
|
|
except Exception as e:
|
|
self.log_message(f"Error updating subnet mask: {str(e)}")
|
|
finally:
|
|
self._updating_bits = False
|
|
|
|
def is_valid_subnet_mask(self, mask: str) -> bool:
|
|
"""Validate if the string is a valid subnet mask"""
|
|
try:
|
|
# Check if it has the format x.x.x.x
|
|
parts = mask.split(".")
|
|
if len(parts) != 4:
|
|
return False
|
|
|
|
# Each part should be a number between 0-255
|
|
for part in parts:
|
|
if not part.isdigit() or not 0 <= int(part) <= 255:
|
|
return False
|
|
|
|
# Check if it's a valid subnet mask pattern
|
|
# Convert to binary and ensure it's a continuous sequence of 1s followed by 0s
|
|
binary = "".join([bin(int(octet))[2:].zfill(8) for octet in parts])
|
|
if "01" in binary: # If there's a 0 followed by 1, it's not valid
|
|
return False
|
|
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
def subnet_mask_to_cidr(self, mask: str) -> int:
|
|
"""Convert subnet mask to CIDR notation bits"""
|
|
try:
|
|
# Convert each octet to binary and count the number of 1s
|
|
parts = mask.split(".")
|
|
binary = "".join([bin(int(octet))[2:].zfill(8) for octet in parts])
|
|
return binary.count("1")
|
|
except Exception as e:
|
|
self.log_message(f"Error converting subnet mask to CIDR: {str(e)}")
|
|
return 24 # Default to /24
|
|
|
|
def cidr_to_subnet_mask(self, bits: int) -> str:
|
|
"""Convert CIDR bits to subnet mask in dotted decimal notation"""
|
|
try:
|
|
# Create a binary string with the specified number of 1s followed by 0s
|
|
binary = "1" * bits + "0" * (32 - bits)
|
|
|
|
# Split the binary string into 4 octets and convert each to decimal
|
|
octets = [binary[i : i + 8] for i in range(0, 32, 8)]
|
|
decimals = [str(int(octet, 2)) for octet in octets]
|
|
|
|
return ".".join(decimals)
|
|
except Exception as e:
|
|
self.log_message(f"Error converting CIDR to subnet mask: {str(e)}")
|
|
return "255.255.255.0" # Default to 255.255.255.0
|
|
|
|
def update_ping_target(self, ip_prefix):
|
|
"""Update the ping target field based on the selected IP prefix"""
|
|
if ip_prefix:
|
|
target = self.config.get_ping_target(ip_prefix)
|
|
self.ping_target.set(target)
|
|
|
|
def save_current_ping_target(self, event=None):
|
|
"""Save the current ping target for the current IP prefix"""
|
|
ip_prefix = self.ip_prefix.get()
|
|
target = self.ping_target.get()
|
|
|
|
if ip_prefix and target:
|
|
self.config.set_ping_target(ip_prefix, target)
|
|
self.log_message(f"Saved ping target {target} for IP prefix {ip_prefix}")
|
|
|
|
|
|
def main():
|
|
root = tk.Tk()
|
|
app = IPChangerApp(root)
|
|
root.mainloop()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|