From 437f68961b984208d299fed9f55a4fddde0d84f9 Mon Sep 17 00:00:00 2001 From: grossmj Date: Tue, 20 Oct 2026 11:51:29 +0700 Subject: [PATCH] chore: clean up Windows specific code --- gns3server/compute/dynamips/__init__.py | 19 ++--- gns3server/compute/qemu/__init__.py | 32 -------- gns3server/compute/qemu/qemu_vm.py | 11 +-- gns3server/compute/vpcs/vpcs_vm.py | 3 - gns3server/controller/__init__.py | 24 +++--- gns3server/controller/export_project.py | 4 - gns3server/utils/interfaces.py | 100 ------------------------ 7 files changed, 16 insertions(+), 177 deletions(-) diff --git a/gns3server/compute/dynamips/__init__.py b/gns3server/compute/dynamips/__init__.py index 0b9ad2c91..938834b48 100644 --- a/gns3server/compute/dynamips/__init__.py +++ b/gns3server/compute/dynamips/__init__.py @@ -26,7 +26,6 @@ import re import shutil import socket import subprocess -import sys import tempfile import time @@ -253,11 +252,6 @@ class Dynamips(BaseManager): # look for Dynamips dynamips_path = self.config.settings.Dynamips.dynamips_path if not os.path.isabs(dynamips_path): - if sys.platform.startswith("win") and hasattr(sys, "frozen"): - dynamips_dir = os.path.normpath( - os.path.join(os.path.dirname(os.path.abspath(sys.executable)), "dynamips") - ) - os.environ["PATH"] = os.pathsep.join(dynamips_dir) + os.pathsep + os.environ.get("PATH", "") dynamips_path = shutil.which(dynamips_path) if not dynamips_path: @@ -311,13 +305,12 @@ class Dynamips(BaseManager): if parse_version(dynamips_version) < parse_version("0.2.11"): raise DynamipsError(f"Dynamips version must be >= 0.2.11, detected version is {dynamips_version}") - if not sys.platform.startswith("win"): - # Hypervisor should always listen to 127.0.0.1 - # See https://github.com/GNS3/dynamips/issues/62 - # This was fixed in Dynamips v0.2.23 which hasn't been built for Windows - if parse_version(dynamips_version) >= parse_version("0.2.23"): - server_host = "127.0.0.1" - bind_console_host = True + # Hypervisor should always listen to 127.0.0.1 + # See https://github.com/GNS3/dynamips/issues/62 + # This was fixed in Dynamips v0.2.23 + if parse_version(dynamips_version) >= parse_version("0.2.23"): + server_host = "127.0.0.1" + bind_console_host = True try: info = socket.getaddrinfo(server_host, 0, socket.AF_UNSPEC, socket.SOCK_STREAM, 0, socket.AI_PASSIVE) diff --git a/gns3server/compute/qemu/__init__.py b/gns3server/compute/qemu/__init__.py index 351e318e5..92781790d 100644 --- a/gns3server/compute/qemu/__init__.py +++ b/gns3server/compute/qemu/__init__.py @@ -242,38 +242,6 @@ class Qemu(BaseManager): except (OSError, subprocess.SubprocessError) as e: raise QemuError(f"Error while looking for the swtpm version: {e}") - @staticmethod - def get_haxm_windows_version(): - """ - Gets the HAXM version number (Windows). - - :returns: HAXM version number. Returns None if HAXM is not installed. - """ - - assert sys.platform.startswith("win") - import winreg - - hkey = winreg.OpenKey( - winreg.HKEY_LOCAL_MACHINE, r"SOFTWARE\Microsoft\Windows\CurrentVersion\Installer\UserData\S-1-5-18\Products" - ) - version = None - for index in range(winreg.QueryInfoKey(hkey)[0]): - product_id = winreg.EnumKey(hkey, index) - try: - product_key = winreg.OpenKey(hkey, rf"{product_id}\InstallProperties") - try: - if winreg.QueryValueEx(product_key, "DisplayName")[0].endswith( - "Hardware Accelerated Execution Manager" - ): - version = winreg.QueryValueEx(product_key, "DisplayVersion")[0] - break - finally: - winreg.CloseKey(product_key) - except OSError: - continue - winreg.CloseKey(hkey) - return version - @staticmethod def get_legacy_vm_workdir(legacy_vm_id, name): """ diff --git a/gns3server/compute/qemu/qemu_vm.py b/gns3server/compute/qemu/qemu_vm.py index 8b97b0c14..f321fc21b 100644 --- a/gns3server/compute/qemu/qemu_vm.py +++ b/gns3server/compute/qemu/qemu_vm.py @@ -1128,15 +1128,8 @@ class QemuVM(KernelDatapathMixin, BaseNode): return try: - if sys.platform.startswith("win") and hasattr(sys, "frozen"): - cpulimit_exec = os.path.join( - os.path.dirname(os.path.abspath(sys.executable)), "cpulimit", "cpulimit.exe" - ) - else: - cpulimit_exec = "cpulimit" - command = [ - cpulimit_exec, + "cpulimit", "--lazy", f"--pid={self._process.pid}", f"--limit={self._cpu_throttling}", @@ -2599,8 +2592,6 @@ class QemuVM(KernelDatapathMixin, BaseNode): Start swtpm (TPM emulator) """ - if sys.platform.startswith("win"): - raise QemuError("swtpm (TPM emulator) is not supported on Windows") tpm_dir = os.path.join(self.working_dir, "tpm") os.makedirs(tpm_dir, exist_ok=True) tpm_sock = os.path.join(self.temporary_directory, "swtpm.sock") diff --git a/gns3server/compute/vpcs/vpcs_vm.py b/gns3server/compute/vpcs/vpcs_vm.py index 536aef70e..f94c73676 100644 --- a/gns3server/compute/vpcs/vpcs_vm.py +++ b/gns3server/compute/vpcs/vpcs_vm.py @@ -162,9 +162,6 @@ class VPCSVM(KernelDatapathMixin, BaseNode): vpcs_path = self._manager.config.settings.VPCS.vpcs_path if not os.path.isabs(vpcs_path): - if sys.platform.startswith("win") and hasattr(sys, "frozen"): - vpcs_dir = os.path.normpath(os.path.join(os.path.dirname(os.path.abspath(sys.executable)), "vpcs")) - os.environ["PATH"] = os.pathsep.join(vpcs_dir) + os.pathsep + os.environ.get("PATH", "") vpcs_path = shutil.which(vpcs_path) return vpcs_path diff --git a/gns3server/controller/__init__.py b/gns3server/controller/__init__.py index b7fe5b4c9..456d7eaf7 100644 --- a/gns3server/controller/__init__.py +++ b/gns3server/controller/__init__.py @@ -511,21 +511,15 @@ class Controller: dst_md5 = await wait_run_in_executor(md5sum, dst) return src_md5 != dst_md5 - if hasattr(sys, "frozen") and sys.platform.startswith("win"): - resource_path = os.path.normpath(os.path.join(os.path.dirname(sys.executable), resource_name)) - for filename in os.listdir(resource_path): - if not os.path.exists(os.path.join(dst_path, filename)): - shutil.copy(os.path.join(resource_path, filename), os.path.join(dst_path, filename)) - else: - for entry in importlib_resources.files("gns3server").joinpath(resource_name).iterdir(): - full_path = os.path.join(dst_path, entry.name) - if entry.is_file() and await should_copy(str(entry), full_path, upgrade_resources): - log.debug(f'Installing {resource_name} resource file "{entry.name}" to "{full_path}"') - shutil.copy(str(entry), os.path.join(full_path)) - installed_resources.append(full_path) - elif entry.is_dir(): - os.makedirs(full_path, exist_ok=True) - await Controller.install_resource_files(full_path, os.path.join(resource_name, entry.name)) + for entry in importlib_resources.files("gns3server").joinpath(resource_name).iterdir(): + full_path = os.path.join(dst_path, entry.name) + if entry.is_file() and await should_copy(str(entry), full_path, upgrade_resources): + log.debug(f'Installing {resource_name} resource file "{entry.name}" to "{full_path}"') + shutil.copy(str(entry), os.path.join(full_path)) + installed_resources.append(full_path) + elif entry.is_dir(): + os.makedirs(full_path, exist_ok=True) + await Controller.install_resource_files(full_path, os.path.join(resource_name, entry.name)) return installed_resources async def _install_base_configs(self): diff --git a/gns3server/controller/export_project.py b/gns3server/controller/export_project.py index cb4c4f2fd..081b3b84c 100644 --- a/gns3server/controller/export_project.py +++ b/gns3server/controller/export_project.py @@ -18,7 +18,6 @@ import json import logging import os -import sys import tempfile import zipfile from datetime import datetime @@ -154,9 +153,6 @@ def _patch_mtime(path): :param path: file path """ - if sys.platform.startswith("win"): - # only UNIX type platforms - return st = os.stat(path, follow_symlinks=False) file_date = datetime.fromtimestamp(st.st_mtime) if file_date.year < 1980: diff --git a/gns3server/utils/interfaces.py b/gns3server/utils/interfaces.py index 34f9422b6..719d340c0 100644 --- a/gns3server/utils/interfaces.py +++ b/gns3server/utils/interfaces.py @@ -35,106 +35,6 @@ import logging log = logging.getLogger(__name__) -def _get_windows_interfaces_from_registry(): - - import winreg - - # HKLM\SYSTEM\CurrentControlSet\Services\Tcpip\Parameters\Interfaces - interfaces = [] - try: - hkey = winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, r"SOFTWARE\Microsoft\Windows NT\CurrentVersion\NetworkCards") - for index in range(winreg.QueryInfoKey(hkey)[0]): - network_card_id = winreg.EnumKey(hkey, index) - hkeycard = winreg.OpenKey(hkey, network_card_id) - guid, _ = winreg.QueryValueEx(hkeycard, "ServiceName") - netcard, _ = winreg.QueryValueEx(hkeycard, "Description") - connection = ( - r"SYSTEM\CurrentControlSet\Control\Network\{4D36E972-E325-11CE-BFC1-08002BE10318}" - + rf"\{guid}\Connection" - ) - hkeycon = winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, connection) - name, _ = winreg.QueryValueEx(hkeycon, "Name") - interface = rf"SYSTEM\CurrentControlSet\Services\Tcpip\Parameters\Interfaces\{guid}" - hkeyinterface = winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, interface) - is_dhcp_enabled, _ = winreg.QueryValueEx(hkeyinterface, "EnableDHCP") - if is_dhcp_enabled: - ip_address, _ = winreg.QueryValueEx(hkeyinterface, "DhcpIPAddress") - netmask, _ = winreg.QueryValueEx(hkeyinterface, "DhcpSubnetMask") - else: - ip_address, _ = winreg.QueryValueEx(hkeyinterface, "IPAddress") - netmask, _ = winreg.QueryValueEx(hkeyinterface, "SubnetMask") - if ip_address: - # get the first IPv4 address only - ip_address = ip_address[0] - npf_interface = f"\\Device\\NPF_{guid}" - interfaces.append( - { - "id": npf_interface, - "name": name, - "ip_address": ip_address, - "mac_address": "", # TODO: find MAC address in registry - "netcard": netcard, - "netmask": netmask, - "type": "ethernet", - } - ) - winreg.CloseKey(hkeyinterface) - winreg.CloseKey(hkeycon) - winreg.CloseKey(hkeycard) - winreg.CloseKey(hkey) - except OSError as e: - log.error(f"could not read registry information: {e}") - - return interfaces - - -def get_windows_interfaces(): - """ - Get Windows interfaces. - - :returns: list of windows interfaces - """ - - import pywintypes - import win32com.client - - interfaces = [] - try: - locator = win32com.client.Dispatch("WbemScripting.SWbemLocator") - service = locator.ConnectServer(".", r"root\cimv2") - network_configs = service.InstancesOf("Win32_NetworkAdapterConfiguration") - # more info on Win32_NetworkAdapter: http://msdn.microsoft.com/en-us/library/aa394216%28v=vs.85%29.aspx - for adapter in service.InstancesOf("Win32_NetworkAdapter"): - if adapter.NetConnectionStatus == 2 or adapter.NetConnectionStatus == 7: - # adapter is connected or media disconnected - ip_address = "" - netmask = "" - for network_config in network_configs: - if network_config.InterfaceIndex == adapter.InterfaceIndex: - if network_config.IPAddress: - # get the first IPv4 address only - ip_address = network_config.IPAddress[0] - netmask = network_config.IPSubnet[0] - break - npf_interface = f"\\Device\\NPF_{adapter.GUID}" - interfaces.append( - { - "id": npf_interface, - "name": adapter.NetConnectionID, - "ip_address": ip_address, - "mac_address": adapter.MACAddress, - "netcard": adapter.name, - "netmask": netmask, - "type": "ethernet", - } - ) - except (AttributeError, pywintypes.com_error): - log.warning("Could not use the COM service to retrieve interface info, trying using the registry...") - return _get_windows_interfaces_from_registry() - - return interfaces - - def has_netmask(interface_name): """ Checks if an interface has a netmask.