gonets/SRV/services/connect_service.py

240 lines
13 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""
Сервис управления подключениями Wi-Fi через `nmcli`.
Назначение:
- Предоставляет операции подключения/отключения к точкам доступа и проверки текущего статуса, ориентируясь на системный NetworkManager.
Поведение:
- Автоматически определяет активный Wi-Fi-интерфейс.
- Валидирует наличие SSID в листинге перед подключением и интерпретирует частые ошибки `nmcli` (неверный пароль, отсутствующая сеть и т.п.).
Ограничения:
- Требуются права для управления сетевыми соединениями в системе.
- В окружениях без NetworkManager команды `nmcli` будут недоступны.
"""
import subprocess
import logging
from typing import Tuple, Optional
from utils.i18n import get_translations, _
from babel.support import Translations
from fastapi import Depends
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)
class WifiConnectionService:
"""Сервис для управления подключением к Wi-Fi сетям с использованием nmcli.
Предоставляет методы для подключения, отключения и проверки статуса Wi-Fi соединений.
"""
def __init__(self, translations: Optional[Translations] = None):
"""Инициализирует сервис Wi-Fi.
Автоматически определяет доступный Wi-Fi-интерфейс и сохраняет его имя.
"""
self.interface = self._get_wifi_interface()
if translations is None:
try:
self.translations = Translations.load("locales", ["ru"], "messages")
except Exception:
self.translations = Translations()
else:
self.translations = translations
def _get_wifi_interface(self) -> str:
"""Автоматически определяет Wi-Fi-интерфейс.
Returns:
str: Название Wi-Fi-интерфейса (например, "wlan0") или пустая строка,
если интерфейс не найден или Wi-Fi отключён.
Raises:
Exception: Если произошла ошибка при выполнении команд nmcli.
"""
try:
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)}")
return ""
def connect(self, ssid: str, password: str) -> Tuple[bool, Optional[str]]:
"""Подключается к Wi-Fi сети по SSID и паролю.
Args:
ssid (str): Идентификатор сети Wi-Fi.
password (str): Пароль для подключения.
Returns:
Tuple[bool, Optional[str]]: Кортеж (успех, сообщение об ошибке).
- Успех: (True, None).
- Ошибка: (False, сообщение об ошибке).
Raises:
subprocess.CalledProcessError: Если команда nmcli завершилась с ошибкой.
Exception: Если произошла неизвестная ошибка при выполнении.
"""
try:
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,
)
if ssid not in result.stdout:
return False, self.translations.gettext("Сеть с SSID {ssid} не найдена в списке доступных сетей").format(ssid=ssid)
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()
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:
logger.error(f"Error connecting to {ssid}: {e.stderr}")
error_msg = e.stderr.lower()
if "secrets were required" in error_msg or "authentication failed" in error_msg or "802-11-wireless-security.psk: недействительное свойство" in e.stderr:
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)
return False, self.translations.gettext("Ошибка подключения к {ssid}: {error}").format(ssid=ssid, error=e.stderr)
except Exception as e:
logger.error(f"Unknown error connecting to {ssid}: {str(e)}")
return False, self.translations.gettext("Ошибка подключения к {ssid}: Неизвестная ошибка - {error}").format(ssid=ssid, error=str(e))
def disconnect(self, ssid: str) -> Tuple[bool, Optional[str]]:
"""Отключается от Wi-Fi сети по указанному SSID.
Args:
ssid (str): Идентификатор сети Wi-Fi для отключения.
Returns:
Tuple[bool, Optional[str]]: Кортеж (успех, сообщение об ошибке).
- Успех: (True, None или сообщение, если подключение не найдено).
- Ошибка: (False, сообщение об ошибке).
Raises:
subprocess.CalledProcessError: Если команда nmcli завершилась с ошибкой.
Exception: Если произошла неизвестная ошибка при выполнении.
"""
try:
if not self.interface:
return False, self.translations.gettext("Wi-Fi-интерфейс не доступен")
result = subprocess.run(["nmcli", "connection", "show", "--active"], capture_output=True, text=True)
active_connections = result.stdout.strip().splitlines()
if not active_connections:
logger.info(f"No active connections for SSID {ssid}")
return True, self.translations.gettext("Нет активного подключения для SSID {ssid}").format(ssid=ssid)
for conn in active_connections:
conn_details = conn.split()
if len(conn_details) > 1:
conn_name = conn_details[0]
conn_info = subprocess.run(["nmcli", "connection", "show", conn_name], capture_output=True, text=True).stdout
if "802-11-wireless.ssid:" in conn_info and ssid in conn_info:
subprocess.run(["nmcli", "connection", "down", conn_name], capture_output=True, text=True, check=True)
logger.info(f"Disconnected from SSID {ssid}")
return True, None
logger.warning(f"Active connection with SSID {ssid} not found")
return True, self.translations.gettext("Активное подключение с SSID {ssid} не найдено").format(ssid=ssid)
except subprocess.CalledProcessError as e:
logger.error(f"Error disconnecting from {ssid}: {e.stderr}")
return False, self.translations.gettext("Ошибка отключения от {ssid}: {error}").format(ssid=ssid, error=e.stderr)
except Exception as e:
logger.error(f"Unknown error disconnecting from {ssid}: {str(e)}")
return False, self.translations.gettext("Ошибка отключения от {ssid}: {error}").format(ssid=ssid, error=str(e))
def get_status(self) -> Tuple[bool, Optional[str]]:
"""Проверяет статус текущего подключения к Wi-Fi.
Returns:
Tuple[bool, Optional[str]]: Кортеж (успех, SSID или сообщение об ошибке).
- Успех: (True, SSID текущей сети).
- Ошибка или отсутствие подключения: (False, сообщение об ошибке).
Raises:
Exception: Если произошла ошибка при выполнении команды nmcli.
"""
try:
if not self.interface:
return False, self.translations.gettext("Wi-Fi-интерфейс не доступен")
result = subprocess.run(["nmcli", "connection", "show", "--active"], capture_output=True, text=True)
if result.stdout.strip():
ssid = result.stdout.splitlines()[1].split()[0]
logger.info(f"Current network: {ssid}")
return True, ssid
return False, self.translations.gettext("Не подключено к Wi-Fi")
except Exception as e:
logger.error(f"Error checking status: {str(e)}")
return False, self.translations.gettext("Ошибка проверки статуса: {error}").format(error=str(e))