From ab38a0db5063a1e831849171e081fd98ef2bb77a Mon Sep 17 00:00:00 2001 From: Merelden Date: Tue, 21 Oct 2025 12:27:01 +0500 Subject: [PATCH] =?UTF-8?q?=D0=94=D0=BE=D0=BF=D0=BE=D0=BB=D0=BD=D0=B5?= =?UTF-8?q?=D0=BD=D0=B8=D0=B5=20=D0=BB=D0=BE=D0=B3=D0=B8=D0=BA=D0=B8=20?= =?UTF-8?q?=D0=BF=D0=BE=D0=B4=D0=BA=D0=BB=D1=8E=D1=87=D0=B5=D0=BD=D0=B8?= =?UTF-8?q?=D1=8F=20=D0=BA=20WiFi,=20=D0=B5=D1=81=D0=BB=D0=B8=20=D0=BF?= =?UTF-8?q?=D1=80=D0=BE=D1=84=D0=B8=D0=BB=D1=8C=20=D0=BE=D0=BA=D0=B0=D0=B7?= =?UTF-8?q?=D1=8B=D0=B2=D0=B0=D0=B5=D1=82=D1=81=D1=8F=20=D0=B1=D0=B8=D1=82?= =?UTF-8?q?=D1=8B=D0=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- SRV/services/arp_service.py | 81 +++++++++++++++++++++++-------- SRV/services/connect_service.py | 86 +++++++++++++++++++++++++++------ 2 files changed, 130 insertions(+), 37 deletions(-) diff --git a/SRV/services/arp_service.py b/SRV/services/arp_service.py index 5b09966..0766372 100644 --- a/SRV/services/arp_service.py +++ b/SRV/services/arp_service.py @@ -24,6 +24,8 @@ from netaddr import IPNetwork from utils.i18n import get_translations, _ from babel.support import Translations +logger = logging.getLogger(__name__) + try: from scapy.layers.l2 import ARP, Ether from scapy.sendrecv import srp @@ -53,35 +55,57 @@ class NetworkScanner: Инициализирует интерфейс, информацию об IP-адресах и локальный MAC-адрес. """ - try: - if interface: - self.interface = interface - else: - default_gateway = netifaces.gateways().get("default", {}).get( - netifaces.AF_INET) - if default_gateway: - self.interface = default_gateway[1] - else: - raise KeyError("Default gateway not found") + self.translations = translations + self.interface = None + self.ip_info = [] + self.networks = [] + self.local_mac = None - self.ip_info = netifaces.ifaddresses(self.interface).get( - netifaces.AF_INET, []) - self.networks = [ - IPNetwork(f"{ip['addr']}/{ip['netmask']}") for ip in self.ip_info - ] - self.local_mac = netifaces.ifaddresses( - self.interface)[netifaces.AF_LINK][0]["addr"] + try: + candidate_interface = interface + if not candidate_interface: + gateways = netifaces.gateways() + default_gateway = gateways.get("default", {}).get(netifaces.AF_INET) + if default_gateway: + candidate_interface = default_gateway[1] + else: + for iface in netifaces.interfaces(): + if iface == "lo": + continue + addrs = netifaces.ifaddresses(iface) + if netifaces.AF_INET in addrs and addrs[netifaces.AF_INET]: + candidate_interface = iface + break + + if candidate_interface: + self.interface = candidate_interface + addresses = netifaces.ifaddresses(self.interface) + self.ip_info = addresses.get(netifaces.AF_INET, []) + self.networks = [ + IPNetwork( + f"{ip['addr']}/{ip['netmask']}" + ) for ip in self.ip_info if ip.get("addr") and ip.get("netmask") + ] + link_info = addresses.get(netifaces.AF_LINK, []) + if link_info: + self.local_mac = link_info[0].get("addr") + else: + logger.warning( + "NetworkScanner: IPv4 интерфейс не найден, сканирование будет недоступно" + ) except Exception as exc: if os.getenv("SPHINX_BUILD"): self.interface = interface or "eth0" self.ip_info = [] self.networks = [] self.local_mac = "00:00:00:00:00:00" - logger = logging.getLogger(__name__) - logger.warning("NetworkScanner falling back to mocked interface during documentation build: %s", exc) + logger.warning( + "NetworkScanner falling back to mocked interface during documentation build: %s", + exc, + ) else: + logger.error("NetworkScanner инициализация завершилась ошибкой: %s", exc) raise - self.translations = translations def _create_arp_packet(self, target_ip): """Создаёт ARP-пакет для отправки на указанный IP-адрес. @@ -112,6 +136,16 @@ class NetworkScanner: Note: Учитывает локальные IP-адреса интерфейса и удаляет дубликаты. """ + if not self.interface or not self.networks or ARP is None or Ether is None: + fallback_clients = [] + if self.ip_info: + for ip_info in self.ip_info: + fallback_clients.append({ + "ip": ip_info.get("addr"), + "mac": self.local_mac, + }) + return [client for client in fallback_clients if client["ip"]] + all_clients = [] for network in self.networks: @@ -132,7 +166,12 @@ class NetworkScanner: all_clients.extend(clients) for ip_info in self.ip_info: - local_client = {"ip": ip_info["addr"], "mac": self.local_mac} + if not ip_info.get("addr"): + continue + local_client = { + "ip": ip_info["addr"], + "mac": self.local_mac, + } if local_client not in all_clients: all_clients.append(local_client) diff --git a/SRV/services/connect_service.py b/SRV/services/connect_service.py index 74f4434..04fee67 100644 --- a/SRV/services/connect_service.py +++ b/SRV/services/connect_service.py @@ -56,17 +56,27 @@ class WifiConnectionService: Exception: Если произошла ошибка при выполнении команд nmcli. """ try: - result = subprocess.run(["nmcli", "radio", "wifi"], capture_output=True, text=True) - if "enabled" in result.stdout.lower(): - interfaces = subprocess.run(["nmcli", "device", "status"], capture_output=True, text=True).stdout - for line in interfaces.splitlines(): - if "wifi" in line.lower() and "connected" in line.lower(): - iface = line.split()[0] - logger.info(f"Using Wi-Fi interface: {iface}") - return iface - logger.warning("Wi-Fi interface not found or not connected") - return "wlan0" - logger.error("Wi-Fi radio is disabled") + result = subprocess.run( + ["nmcli", "radio", "wifi"], capture_output=True, text=True + ) + if "enabled" not in result.stdout.lower(): + logger.error("Wi-Fi radio is disabled") + return "" + + interfaces_output = subprocess.run( + ["nmcli", "device", "status"], capture_output=True, text=True + ).stdout + + for line in interfaces_output.splitlines(): + parts = line.split() + if len(parts) < 2: + continue + device_name, device_type = parts[0], parts[1] + if device_type.lower() == "wifi": + logger.info(f"Using Wi-Fi interface: {device_name}") + return device_name + + logger.warning("Wi-Fi interface not found in `nmcli device status` output") return "" except Exception as e: logger.error(f"Error detecting Wi-Fi interface: {str(e)}") @@ -92,20 +102,64 @@ class WifiConnectionService: if not self.interface: return False, self.translations.gettext("Wi-Fi-интерфейс не доступен") - result = subprocess.run(["nmcli", "-f", "SSID", "device", "wifi", "list"], capture_output=True, text=True) + result = subprocess.run( + ["nmcli", "-f", "SSID", "device", "wifi", "list"], + capture_output=True, + text=True, + ) if ssid not in result.stdout: return False, self.translations.gettext("Сеть с SSID {ssid} не найдена в списке доступных сетей").format(ssid=ssid) - cmd = ["nmcli", "device", "wifi", "connect", ssid, "password", password] - process = subprocess.run(cmd, capture_output=True, text=True, check=False) + def run_connect() -> subprocess.CompletedProcess[str]: + cmd = [ + "nmcli", + "device", + "wifi", + "connect", + ssid, + "ifname", + self.interface, + ] + if password: + cmd.extend(["password", password]) + return subprocess.run(cmd, capture_output=True, text=True, check=False) + + process = run_connect() if process.returncode != 0: error_msg = process.stderr.lower() - if "secrets were required" in error_msg or "authentication failed" in error_msg or "802-11-wireless-security.psk: недействительное свойство" in process.stderr: - return False, self.translations.gettext("Ошибка подключения к {ssid}: Неверный пароль").format(ssid=ssid) + key_mgmt_error = ( + "secrets were required" in error_msg + or "authentication failed" in error_msg + or "802-11-wireless-security.psk" in error_msg + or "key-mgmt" in error_msg + ) + if key_mgmt_error: + subprocess.run( + ["nmcli", "connection", "delete", "id", ssid], + capture_output=True, + text=True, + ) + retry_process = run_connect() + if retry_process.returncode == 0: + logger.info(f"Successfully connected to network {ssid} after profile reset") + return True, None + error_msg = retry_process.stderr.lower() + if ( + "secrets were required" in error_msg + or "authentication failed" in error_msg + or "802-11-wireless-security.psk" in error_msg + or "key-mgmt" in error_msg + ): + return False, self.translations.gettext("Ошибка подключения к {ssid}: Неверный пароль или тип безопасности").format(ssid=ssid) + elif "no network with ssid" in error_msg: + return False, self.translations.gettext("Ошибка подключения к {ssid}: Сеть с SSID {ssid} недоступна или неверно указана").format(ssid=ssid) + else: + return False, self.translations.gettext("Ошибка подключения к {ssid}: {error}").format(ssid=ssid, error=retry_process.stderr) elif "no network with ssid" in error_msg: return False, self.translations.gettext("Ошибка подключения к {ssid}: Сеть с SSID {ssid} недоступна или неверно указана").format(ssid=ssid) else: return False, self.translations.gettext("Ошибка подключения к {ssid}: {error}").format(ssid=ssid, error=process.stderr) + logger.info(f"Successfully connected to network {ssid}") return True, None except subprocess.CalledProcessError as e: