diff --git a/SRV/api/v1/subnet.py b/SRV/api/v1/subnet.py index 9601f11..e500010 100644 --- a/SRV/api/v1/subnet.py +++ b/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)}") \ No newline at end of file + 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} \ No newline at end of file diff --git a/SRV/main.py b/SRV/main.py index 06bf591..9a9925e 100644 --- a/SRV/main.py +++ b/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) diff --git a/SRV/models/request.py b/SRV/models/request.py index 238c11c..cb6a3b4 100644 --- a/SRV/models/request.py +++ b/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 \ No newline at end of file diff --git a/SRV/services/connect_service.py b/SRV/services/connect_service.py new file mode 100644 index 0000000..e8d02e9 --- /dev/null +++ b/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)}" \ No newline at end of file