Подключение к точкам доступа Wi-Fi

SRV/api/v1/subnet.py

@@ -2,9 +2,10 @@ from fastapi import APIRouter, HTTPException
 from typing import List, Dict
 from models.subnet import SubnetScanResult
 from models.wifi import WifiScanResult
-from models.request import NoteRequest
+from models.request import NoteRequest, WifiConnectRequest, WifiDisconnectRequest
 from services.subnet_service import SubnetService
 from services.wifi_service import WifiService
+from services.connect_service import WifiConnectionService
 
 router = APIRouter()
 
@@ -128,4 +129,31 @@ def delete_wifi_note(ssid: str):
     except HTTPException as e:
         raise e
     except Exception as e:
-        raise HTTPException(status_code=500, detail=f"Ошибка удаления заметки: {str(e)}")
+        raise HTTPException(status_code=500, detail=f"Ошибка удаления заметки: {str(e)}")
+
+@router.post("/wifi/connect")
+def connect_to_wifi(request: WifiConnectRequest):
+    """Подключается к Wi-Fi сети по SSID и паролю."""
+    wifi_service = WifiConnectionService()
+    success, message = wifi_service.connect(request.ssid, request.password)
+    if not success:
+        raise HTTPException(status_code=400, detail=message)
+    return {"message": f"Успешно подключено к {request.ssid}"}
+
+@router.post("/wifi/disconnect")
+def disconnect_from_wifi(request: WifiDisconnectRequest):
+    """Отключается от Wi-Fi сети по указанному SSID."""
+    wifi_service = WifiConnectionService()
+    success, message = wifi_service.disconnect(request.ssid)
+    if not success:
+        raise HTTPException(status_code=400, detail=message)
+    return {"message": message or f"Успешно отключено от {request.ssid}"}
+
+@router.get("/wifi/status")
+def get_wifi_status():
+    """Проверяет статус текущего подключения к Wi-Fi."""
+    wifi_service = WifiConnectionService()
+    success, message = wifi_service.get_status()
+    if not success:
+        raise HTTPException(status_code=400, detail=message)
+    return {"status": "connected", "ssid": message}

SRV/main.py

@@ -18,5 +18,5 @@ app.include_router(router)
 network_service = NetworkService()
 
 if __name__ == "__main__":
-    # uvicorn.run("main:app", host='0.0.0.0', port=8001, reload=True, ssl_keyfile="/etc/gonets/certs/gonets.key", ssl_certfile="/etc/gonets/certs/gonets.crt")
+    #uvicorn.run("main:app", host='0.0.0.0', port=8001, reload=True, ssl_keyfile="/etc/gonets/certs/gonets.key", ssl_certfile="/etc/gonets/certs/gonets.crt")
     uvicorn.run("main:app", host="0.0.0.0", port=8001, reload=True)

SRV/models/request.py

@@ -10,3 +10,23 @@ class NoteRequest(BaseModel):
     """
 
     note: str
+
+class WifiConnectRequest(BaseModel):
+    """
+    Модель для подключения к точке доступа Wi-Fi
+
+    Атрибуты:
+        ssid (str): Название точки доступа Wi-Fi
+        password (str): Пароль точки доступа Wi-Fi
+    """
+    ssid: str
+    password: str
+
+class WifiDisconnectRequest(BaseModel):
+    """
+    Модель для отключения от точки доступа Wi-Fi
+
+    Атрибуты:
+        ssid (str): Название точки доступа Wi-Fi
+    """
+    ssid: str

SRV/services/connect_service.py

@@ -0,0 +1,96 @@
+import subprocess
+import logging
+from typing import Tuple, Optional
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
+logger = logging.getLogger(__name__)
+
+class WifiConnectionService:
+    def __init__(self):
+        self.interface = self._get_wifi_interface()
+
+    def _get_wifi_interface(self) -> str:
+        """Автоматически определяет Wi-Fi-интерфейс."""
+        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"Используется Wi-Fi-интерфейс: {iface}")
+                        return iface
+                logger.warning("Wi-Fi-интерфейс не найден или не подключён")
+                return "wlan0"  # Значение по умолчанию
+            logger.error("Wi-Fi радио отключено")
+            return ""
+        except Exception as e:
+            logger.error(f"Ошибка при определении Wi-Fi-интерфейса: {str(e)}")
+            return ""
+
+    def connect(self, ssid: str, password: str) -> Tuple[bool, Optional[str]]:
+        """Подключается к Wi-Fi сети по SSID и паролю."""
+        try:
+            if not self.interface:
+                return False, "Wi-Fi-интерфейс не доступен"
+            # Проверяем, доступна ли сеть
+            result = subprocess.run(["nmcli", "-f", "SSID", "device", "wifi", "list"], capture_output=True, text=True)
+            if ssid not in result.stdout:
+                return False, f"Сеть с SSID {ssid} не найдена"
+            # Подключаемся к сети
+            cmd = ["nmcli", "device", "wifi", "connect", ssid, "password", password]
+            subprocess.run(cmd, capture_output=True, text=True, check=True)
+            logger.info(f"Успешное подключение к сети {ssid}")
+            return True, None
+        except subprocess.CalledProcessError as e:
+            logger.error(f"Ошибка подключения к {ssid}: {e.stderr}")
+            return False, f"Ошибка подключения: {e.stderr}"
+        except Exception as e:
+            logger.error(f"Неизвестная ошибка при подключении к {ssid}: {str(e)}")
+            return False, f"Неизвестная ошибка: {str(e)}"
+
+    def disconnect(self, ssid: str) -> Tuple[bool, Optional[str]]:
+        """Отключается от Wi-Fi сети по указанному SSID."""
+        try:
+            if not self.interface:
+                return False, "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"Нет активных подключений для SSID {ssid}")
+                return True, f"Нет активного подключения для SSID {ssid}"
+            # Ищем соединение с указанным SSID
+            for conn in active_connections:
+                conn_details = conn.split()
+                if len(conn_details) > 1:
+                    conn_name = conn_details[0]
+                    # Получаем SSID для этого соединения
+                    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"Отключено соединение с SSID {ssid}")
+                        return True, None
+            logger.warning(f"Активное подключение с SSID {ssid} не найдено")
+            return True, f"Активное подключение с SSID {ssid} не найдено"
+        except subprocess.CalledProcessError as e:
+            logger.error(f"Ошибка при отключении от {ssid}: {e.stderr}")
+            return False, f"Ошибка отключения: {e.stderr}"
+        except Exception as e:
+            logger.error(f"Неизвестная ошибка при отключении от {ssid}: {str(e)}")
+            return False, f"Ошибка отключения: {str(e)}"
+
+    def get_status(self) -> Tuple[bool, Optional[str]]:
+        """Проверяет статус текущего подключения к Wi-Fi."""
+        try:
+            if not self.interface:
+                return False, "Wi-Fi-интерфейс не доступен"
+            result = subprocess.run(["nmcli", "connection", "show", "--active"], capture_output=True, text=True)
+            if result.stdout.strip():
+                ssid = result.stdout.splitlines()[0].split()[0]
+                logger.info(f"Текущая сеть: {ssid}")
+                return True, ssid
+            return False, "Не подключено к Wi-Fi"
+        except Exception as e:
+            logger.error(f"Ошибка при проверке статуса: {str(e)}")
+            return False, f"Ошибка проверки статуса: {str(e)}"