From 06a5a758a37bc0b63436b94ad8d026b486760fb2 Mon Sep 17 00:00:00 2001 From: "s.galanskiy" Date: Thu, 12 Jun 2025 16:32:05 +0500 Subject: [PATCH] =?UTF-8?q?=D0=A4=D0=BE=D1=80=D0=BC=D0=B0=D1=82=D0=B8?= =?UTF-8?q?=D1=80=D0=BE=D0=B2=D0=B0=D0=BD=D0=B8=D0=B5=20=D0=B2=D1=81=D0=B5?= =?UTF-8?q?=D0=B3=D0=BE=20=D1=81=D0=B5=D1=80=D0=B2=D0=B5=D1=80=D0=BD=D0=BE?= =?UTF-8?q?=D0=B3=D0=BE=20=D0=BA=D0=BE=D0=B4=D0=B0=20=D0=BF=D0=BE=20=D1=81?= =?UTF-8?q?=D1=82=D0=B0=D0=BD=D0=B4=D0=B0=D1=80=D1=82=D0=B0=D0=BC=20(black?= =?UTF-8?q?)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- SRV/api/router.py | 2 +- SRV/api/v1/auth.py | 30 ++-- SRV/api/v1/firewalld.py | 5 +- SRV/api/v1/ip.py | 31 ++-- SRV/api/v1/network.py | 73 +++++++--- SRV/api/v1/subnet.py | 21 ++- SRV/api/v1/ufw.py | 34 +++-- SRV/api/v1/websocket.py | 28 ++-- SRV/config/cors.py | 5 +- SRV/config/init_config.py | 16 ++- SRV/config/ports.py | 7 +- SRV/core/cli_executor.py | 6 +- SRV/main.py | 4 +- SRV/models/auth.py | 10 +- SRV/models/network.py | 9 ++ SRV/models/request.py | 3 +- SRV/models/subnet.py | 6 +- SRV/models/ufw.py | 116 ++++++++------- SRV/models/wifi.py | 2 + SRV/services/arp_service.py | 31 ++-- SRV/services/auth_service.py | 46 ++++-- SRV/services/firewalld_service.py | 5 +- SRV/services/ip_service.py | 25 ++-- SRV/services/network_service.py | 228 ++++++++++++++++++------------ SRV/services/subnet_service.py | 33 +++-- SRV/services/wifi_service.py | 64 ++++++--- SRV/storage/json_storage.py | 39 +++-- SRV/storage/session_storage.py | 37 +++-- 28 files changed, 611 insertions(+), 305 deletions(-) diff --git a/SRV/api/router.py b/SRV/api/router.py index e146840..167c19d 100644 --- a/SRV/api/router.py +++ b/SRV/api/router.py @@ -9,4 +9,4 @@ router.include_router(ip.router, prefix="/ip", tags=["IP"]) router.include_router(network.router, prefix="/network", tags=["Network"]) router.include_router(auth.router, prefix="/auth", tags=["Auth"]) -router.include_router(subnet.router, prefix="/subnet", tags=["Subnet"]) \ No newline at end of file +router.include_router(subnet.router, prefix="/subnet", tags=["Subnet"]) diff --git a/SRV/api/v1/auth.py b/SRV/api/v1/auth.py index b240d87..ad90dec 100644 --- a/SRV/api/v1/auth.py +++ b/SRV/api/v1/auth.py @@ -26,7 +26,9 @@ async def register(user: UserCreate): raise e except Exception as e: logger.error(f"Ошибка в register: {type(e).__name__} - {str(e)}") - raise HTTPException(status_code=500, detail="Не удалось зарегистрировать пользователя") + raise HTTPException( + status_code=500, detail="Не удалось зарегистрировать пользователя" + ) @router.post("/login", response_model=Token) @@ -36,13 +38,17 @@ async def login(user: UserLogin, request: Request): client_id = request.client.host logger.debug(f"Попытка авторизации для client_id {client_id}") result = auth_service.login(user, client_id) - logger.info(f"Успешная авторизация для client_id {client_id}, session_id: {result['session_id']}") + logger.info( + f"Успешная авторизация для client_id {client_id}, session_id: {result['session_id']}" + ) return result except HTTPException as e: raise e except Exception as e: logger.error(f"Ошибка в login: {type(e).__name__} - {str(e)}") - raise HTTPException(status_code=500, detail="Не удалось авторизовать пользователя") + raise HTTPException( + status_code=500, detail="Не удалось авторизовать пользователя" + ) @router.get("/me") @@ -55,13 +61,15 @@ async def get_current_user(request: Request): return { "user": result["user"], "access_token": result["access_token"], - "token_type": "bearer" + "token_type": "bearer", } except HTTPException as e: raise e except Exception as e: logger.error(f"Ошибка в get_current_user: {type(e).__name__} - {str(e)}") - raise HTTPException(status_code=500, detail="Не удалось получить данные пользователя") + raise HTTPException( + status_code=500, detail="Не удалось получить данные пользователя" + ) @router.get("/logout") @@ -90,7 +98,7 @@ async def check_users_file(): return { "exists": False, "has_users": False, - "message": "Файл users.json не существует" + "message": "Файл users.json не существует", } data = storage._read_data() @@ -100,15 +108,17 @@ async def check_users_file(): return { "exists": True, "has_users": True, - "message": "Файл users.json существует и содержит пользователей" + "message": "Файл users.json существует и содержит пользователей", } else: logger.info(f"Файл {file_path} существует, но не содержит пользователей") return { "exists": True, "has_users": False, - "message": "Файл users.json существует, но не содержит пользователей" + "message": "Файл users.json существует, но не содержит пользователей", } except Exception as e: - logger.error(f"Ошибка при проверке файла users.json: {type(e).__name__} - {str(e)}") - raise HTTPException(status_code=500, detail="Ошибка при проверке файла") \ No newline at end of file + logger.error( + f"Ошибка при проверке файла users.json: {type(e).__name__} - {str(e)}" + ) + raise HTTPException(status_code=500, detail="Ошибка при проверке файла") diff --git a/SRV/api/v1/firewalld.py b/SRV/api/v1/firewalld.py index 0c3dc42..4f2515f 100644 --- a/SRV/api/v1/firewalld.py +++ b/SRV/api/v1/firewalld.py @@ -123,10 +123,7 @@ async def get_zone(zone: str = Path(..., description="Zone name")) -> FirewallZo try: zone_info = firewalld.get_zone_settings(zone) return FirewallZone( - name=zone, - target=FirewallTarget.DEFAULT, - interfaces=[], - **zone_info + name=zone, target=FirewallTarget.DEFAULT, interfaces=[], **zone_info ) except ValueError: raise HTTPException( diff --git a/SRV/api/v1/ip.py b/SRV/api/v1/ip.py index 51402cc..bcd9589 100644 --- a/SRV/api/v1/ip.py +++ b/SRV/api/v1/ip.py @@ -5,33 +5,48 @@ from services.ip_service import IPService router = APIRouter() ip_service = IPService() + @router.post("/check_ip") async def check_ip_config(data: Dict[str, Any] = Body(...)): - """ Проверяет доступность IP-адреса """ + """Проверяет доступность IP-адреса""" try: result = await ip_service.check_ip_config(data) return result except Exception as e: - raise HTTPException(status_code=500, detail=f"Ошибка проверки IP конфигурации: {str(e)}") + raise HTTPException( + status_code=500, detail=f"Ошибка проверки IP конфигурации: {str(e)}" + ) + @router.post("/apply_ip") async def apply_ip_config(data: Dict[str, Any] = Body(...)): - """ Применяет проверенные IP-настройки """ + """Применяет проверенные IP-настройки""" try: await ip_service.apply_ip_config(data) - return {"success": True, "message": f"Настройки сетевого профиля для {data['connection_id']} применены"} + return { + "success": True, + "message": f"Настройки сетевого профиля для {data['connection_id']} применены", + } except ValueError as e: raise HTTPException(status_code=404, detail=str(e)) except Exception as e: - raise HTTPException(status_code=500, detail=f"Ошибка проверки IP конфигурации: {str(e)}") + raise HTTPException( + status_code=500, detail=f"Ошибка проверки IP конфигурации: {str(e)}" + ) + @router.post("/save_ip") async def save_ip_config(data: Dict[str, Any] = Body(...)): - """ Применяет проверенные IP-настройки (Нужен для того, чтобы откатить настройки сетевого профиля обратно) """ + """Применяет проверенные IP-настройки (Нужен для того, чтобы откатить настройки сетевого профиля обратно)""" try: await ip_service.apply_ip_config(data) - return {"success": True, "message": f"Настройки сетевого профиля для {data['connection_id']} применены"} + return { + "success": True, + "message": f"Настройки сетевого профиля для {data['connection_id']} применены", + } except ValueError as e: raise HTTPException(status_code=404, detail=str(e)) except Exception as e: - raise HTTPException(status_code=500, detail=f"Ошибка проверки IP конфигурации: {str(e)}") \ No newline at end of file + raise HTTPException( + status_code=500, detail=f"Ошибка проверки IP конфигурации: {str(e)}" + ) diff --git a/SRV/api/v1/network.py b/SRV/api/v1/network.py index cf8b647..cefbbc5 100644 --- a/SRV/api/v1/network.py +++ b/SRV/api/v1/network.py @@ -3,7 +3,13 @@ import logging from typing import Dict, Tuple, List, Union from services.network_service import NetworkService -from models.network import NetworkInterfacesContainer, Activation, Deactivation, NetworkInterface, WiFiDevice +from models.network import ( + NetworkInterfacesContainer, + Activation, + Deactivation, + NetworkInterface, + WiFiDevice, +) logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) @@ -19,19 +25,19 @@ class ContainerPositionManager: def __init__(self): self.positions: Dict[str, Dict[str, Union[int, bool]]] = { - 'activation': {'x': 0, 'y': 0, 'isOpen': False}, - 'deactivation': {'x': 0, 'y': 0, 'isOpen': False} + "activation": {"x": 0, "y": 0, "isOpen": False}, + "deactivation": {"x": 0, "y": 0, "isOpen": False}, } def get_position_and_state(self, container_name: str) -> Tuple[int, int, bool]: container = self.positions.get(container_name) if not container: raise ValueError(f"Неверное имя контейнера: {container_name}") - return container['x'], container['y'], container['isOpen'] + return container["x"], container["y"], container["isOpen"] def update_position(self, container_id: int, x: int, y: int, isOpen: bool) -> None: - container_name = 'activation' if container_id == 1 else 'deactivation' - self.positions[container_name] = {'x': x, 'y': y, 'isOpen': isOpen} + container_name = "activation" if container_id == 1 else "deactivation" + self.positions[container_name] = {"x": x, "y": y, "isOpen": isOpen} container_manager = ContainerPositionManager() @@ -45,44 +51,55 @@ async def get_network_data() -> Dict[str, Dict]: try: network_data = await network_service.get_network_interfaces_container() - activation_interfaces, deactivation_interfaces = _split_interfaces(network_data.interfaces) + activation_interfaces, deactivation_interfaces = _split_interfaces( + network_data.interfaces + ) - activation_x, activation_y, activation_is_open = container_manager.get_position_and_state('activation') - deactivation_x, deactivation_y, deactivation_is_open = container_manager.get_position_and_state('deactivation') + activation_x, activation_y, activation_is_open = ( + container_manager.get_position_and_state("activation") + ) + deactivation_x, deactivation_y, deactivation_is_open = ( + container_manager.get_position_and_state("deactivation") + ) activation = Activation( id=1, interfaces=activation_interfaces, container_x=activation_x, container_y=activation_y, - isOpen=activation_is_open + isOpen=activation_is_open, ) deactivation = Deactivation( id=2, interfaces=deactivation_interfaces, container_x=deactivation_x, container_y=deactivation_y, - isOpen=deactivation_is_open + isOpen=deactivation_is_open, ) return { "activation": activation.model_dump(), - "deactivation": deactivation.model_dump() + "deactivation": deactivation.model_dump(), } except Exception as e: logger.error(f"Ошибка в get_network_data: {e}") - raise HTTPException(status_code=500, detail="Не удалось получить сетевые данные") + raise HTTPException( + status_code=500, detail="Не удалось получить сетевые данные" + ) -def _split_interfaces(interfaces: List[Union[NetworkInterface, WiFiDevice]]) -> Tuple[ - List[Union[NetworkInterface, WiFiDevice]], List[Union[NetworkInterface, WiFiDevice]]]: +def _split_interfaces( + interfaces: List[Union[NetworkInterface, WiFiDevice]], +) -> Tuple[ + List[Union[NetworkInterface, WiFiDevice]], List[Union[NetworkInterface, WiFiDevice]] +]: """ Разделяет интерфейсы на активные и неактивные """ activation_interfaces = [] deactivation_interfaces = [] for interface in interfaces: - if interface.device_state == '100': + if interface.device_state == "100": activation_interfaces.append(interface) else: deactivation_interfaces.append(interface) @@ -98,7 +115,9 @@ async def update_container_position(data: Dict[str, Union[int, bool]] = Body(... required_keys = ["container_id", "container_x", "container_y", "isOpen"] if not all(key in data for key in required_keys): - raise ValueError("Отсутствуют обязательные поля: container_id, container_x, container_y, isOpen") + raise ValueError( + "Отсутствуют обязательные поля: container_id, container_x, container_y, isOpen" + ) container_id = data["container_id"] container_x = data["container_x"] @@ -106,16 +125,24 @@ async def update_container_position(data: Dict[str, Union[int, bool]] = Body(... is_open = data["isOpen"] # Проверка типов - if not isinstance(container_id, int) or not isinstance(container_x, int) or not isinstance(container_y, - int) or not isinstance( - is_open, bool): + if ( + not isinstance(container_id, int) + or not isinstance(container_x, int) + or not isinstance(container_y, int) + or not isinstance(is_open, bool) + ): raise ValueError("Неверный тип данных в запросе") - container_manager.update_position(container_id, container_x, container_y, is_open) + container_manager.update_position( + container_id, container_x, container_y, is_open + ) return {"success": True, "message": "Позиция и состояние контейнера обновлены"} except Exception as e: logger.error(f"Ошибка обновления позиции контейнера: {e}") - raise HTTPException(status_code=400, detail=f"Не удалось обновить положение контейнера: {str(e)}") + raise HTTPException( + status_code=400, + detail=f"Не удалось обновить положение контейнера: {str(e)}", + ) @router.get("/get_hostname") @@ -125,4 +152,4 @@ async def get_hostname(): return hostname_info except Exception as e: logger.error(f"Ошибка в get_hostname: {e}") - raise HTTPException(status_code=500, detail="Не удалось получить имя хоста") \ No newline at end of file + raise HTTPException(status_code=500, detail="Не удалось получить имя хоста") diff --git a/SRV/api/v1/subnet.py b/SRV/api/v1/subnet.py index 9dc4da0..a226e1c 100644 --- a/SRV/api/v1/subnet.py +++ b/SRV/api/v1/subnet.py @@ -8,6 +8,7 @@ from services.wifi_service import WifiService router = APIRouter() + @router.get("/scan", response_model=SubnetScanResult) def scan_subnet(subnet: str = "192.168.1.0/24", ports: List[int] = None): """Сканирует локальную подсеть и возвращает данные об устройствах""" @@ -18,6 +19,7 @@ def scan_subnet(subnet: str = "192.168.1.0/24", ports: List[int] = None): except Exception as e: raise HTTPException(status_code=500, detail=f"Ошибка сканирования: {str(e)}") + @router.get("/wifi", response_model=WifiScanResult) def scan_wifi(timeout: int = 30): """Сканирует Wi-Fi точки доступа и возвращает данные о них""" @@ -26,7 +28,10 @@ def scan_wifi(timeout: int = 30): result = wifi_service.scan_wifi(timeout=timeout) return result except Exception as e: - raise HTTPException(status_code=500, detail=f"Ошибка сканирования Wi-Fi: {str(e)}") + raise HTTPException( + status_code=500, detail=f"Ошибка сканирования Wi-Fi: {str(e)}" + ) + @router.post("/devices/{mac}/note") def add_note(mac: str, request: NoteRequest): @@ -43,7 +48,10 @@ def add_note(mac: str, request: NoteRequest): 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.put("/devices/{mac}/note") def update_note(mac: str, request: NoteRequest): @@ -60,7 +68,10 @@ def update_note(mac: str, request: NoteRequest): 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.delete("/devices/{mac}/note") def delete_note(mac: str): @@ -77,4 +88,6 @@ def delete_note(mac: 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)}" + ) diff --git a/SRV/api/v1/ufw.py b/SRV/api/v1/ufw.py index aeaa352..c2cac70 100644 --- a/SRV/api/v1/ufw.py +++ b/SRV/api/v1/ufw.py @@ -7,9 +7,11 @@ from services.ufw_service import UFWService router = APIRouter() + class RuleNumbers(BaseModel): numbers: List[int] = Field(..., description="Список номеров правил для удаления") + @router.get("/rules", response_model=UFWRulesResponse) async def get_ufw_rules(): """ @@ -21,6 +23,7 @@ async def get_ufw_rules(): except Exception as e: raise HTTPException(status_code=500, detail=f"Error: {str(e)}") + @router.get("/status", response_model=UFWStatus) async def get_ufw_status(): """ @@ -32,6 +35,7 @@ async def get_ufw_status(): except Exception as e: raise HTTPException(status_code=500, detail=f"Error: {str(e)}") + @router.post("/enable") async def enable_ufw(): """ @@ -47,6 +51,7 @@ async def enable_ufw(): except Exception as e: raise HTTPException(status_code=500, detail=f"Error: {str(e)}") + @router.post("/disable") async def disable_ufw(): """ @@ -62,6 +67,7 @@ async def disable_ufw(): except Exception as e: raise HTTPException(status_code=500, detail=f"Error: {str(e)}") + @router.post("/rules/in") async def add_ufw_in_rules(rules: List[UFWRule]): """ @@ -74,9 +80,9 @@ async def add_ufw_in_rules(rules: List[UFWRule]): if not status.enabled: raise HTTPException( status_code=400, - detail="UFW отключен. Включите UFW перед добавлением правил." + detail="UFW отключен. Включите UFW перед добавлением правил.", ) - + result = ufw_service.add_rules(rules) if result["success"]: successful_rules = len([r for r in result["results"] if r["success"]]) @@ -84,13 +90,17 @@ async def add_ufw_in_rules(rules: List[UFWRule]): return { "status": "success", "message": f"Добавлено {successful_rules} из {total_rules} правил", - "details": result + "details": result, } else: - raise HTTPException(status_code=500, detail=result.get("error", "Не удалось добавить правила")) + raise HTTPException( + status_code=500, + detail=result.get("error", "Не удалось добавить правила"), + ) except Exception as e: raise HTTPException(status_code=500, detail=f"Error: {str(e)}") + @router.post("/rules/out") async def add_ufw_out_rules(rules: List[UFWRule]): """ @@ -103,17 +113,23 @@ async def add_ufw_out_rules(rules: List[UFWRule]): if not status.enabled: raise HTTPException( status_code=400, - detail="UFW отключен. Включите UFW перед добавлением правил." + detail="UFW отключен. Включите UFW перед добавлением правил.", ) - + success = ufw_service.add_ufw_out_rules(rules) if success: - return {"status": "success", "message": "Исходящие правила UFW успешно добавлены"} + return { + "status": "success", + "message": "Исходящие правила UFW успешно добавлены", + } else: - raise HTTPException(status_code=500, detail="Не удалось добавить исходящие правила UFW") + raise HTTPException( + status_code=500, detail="Не удалось добавить исходящие правила UFW" + ) except Exception as e: raise HTTPException(status_code=500, detail=f"Error: {str(e)}") + @router.delete("/rules") async def delete_ufw_rules(rule_numbers: RuleNumbers): """ @@ -126,7 +142,7 @@ async def delete_ufw_rules(rule_numbers: RuleNumbers): return { "status": "success", "message": f"Удалено {result['successful']} из {result['total']} правил", - "details": result + "details": result, } else: raise HTTPException(status_code=500, detail=result["error"]) diff --git a/SRV/api/v1/websocket.py b/SRV/api/v1/websocket.py index 4637ebb..58d2c5b 100644 --- a/SRV/api/v1/websocket.py +++ b/SRV/api/v1/websocket.py @@ -12,25 +12,27 @@ router = APIRouter() logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) + class WebSocketManager: """ Класс для управления подключениями к WebSocket-серверу """ + def __init__(self): - """ Инициализация менеджера подключений к WebSocket-серверу """ + """Инициализация менеджера подключений к WebSocket-серверу""" self.active_connections: Set[WebSocket] = set() async def connect(self, websocket: WebSocket): - """ Подключение клиента к WebSocket-серверу """ + """Подключение клиента к WebSocket-серверу""" await websocket.accept() self.active_connections.add(websocket) def disconnect(self, websocket: WebSocket): - """ Отключение клиента от WebSocket-серверу """ + """Отключение клиента от WebSocket-серверу""" self.active_connections.remove(websocket) async def broadcast(self, message: dict): - """ Трансляция сообщений всем подключенным клиентам """ + """Трансляция сообщений всем подключенным клиентам""" for connection in list(self.active_connections): try: if connection.client_state == WebSocketState.CONNECTED: @@ -41,18 +43,22 @@ class WebSocketManager: logger.error(f"Сообщение об ошибке трансляции: {e}") self.disconnect(connection) + manager = WebSocketManager() + @router.websocket("/api/v1") async def websocket_endpoint(websocket: WebSocket): - """ Обработчик WebSocket-соединения """ + """Обработчик WebSocket-соединения""" await manager.connect(websocket) try: while True: data = await websocket.receive_json() logger.info(f"Полученные данные: {data}") - if all(key in data for key in ["container_id", "container_x", "container_y"]): + if all( + key in data for key in ["container_id", "container_x", "container_y"] + ): await update_container_position(data) network_data = await get_network_data() @@ -69,18 +75,22 @@ async def websocket_endpoint(websocket: WebSocket): if websocket.client_state != WebSocketState.DISCONNECTED: await websocket.close() + @router.on_event("startup") async def startup_event(): - """ Запуск таска для трансляции сетевых данных """ + """Запуск таска для трансляции сетевых данных""" asyncio.create_task(broadcast_data()) + async def broadcast_data(): - """ Таск для трансляции сетевых данных """ + """Таск для трансляции сетевых данных""" while True: try: network_data = await get_network_data() if "activation" not in network_data: - logger.error("Ошибка: Ключ 'activation' отсутствует в полученных данных сети") + logger.error( + "Ошибка: Ключ 'activation' отсутствует в полученных данных сети" + ) await asyncio.sleep(2) continue diff --git a/SRV/config/cors.py b/SRV/config/cors.py index 8702cfc..46a11a7 100644 --- a/SRV/config/cors.py +++ b/SRV/config/cors.py @@ -6,8 +6,9 @@ class CORSMiddlewareConfig: """ Класс для настройки CORS политики в FastAPI """ + def __init__(self, app: FastAPI) -> None: - """ Инициализируем приложение FastAPI и список допустимых CORS-оригинов """ + """Инициализируем приложение FastAPI и список допустимых CORS-оригинов""" self.app = app self.origins = [ "http://localhost.tiangolo.com", @@ -19,7 +20,7 @@ class CORSMiddlewareConfig: ] def add_cors_middleware(self) -> None: - """ Добавляем CORS в приложение FastAPI """ + """Добавляем CORS в приложение FastAPI""" self.app.add_middleware( CORSMiddleware, allow_origins=self.origins, diff --git a/SRV/config/init_config.py b/SRV/config/init_config.py index 0683fd9..3a54adb 100644 --- a/SRV/config/init_config.py +++ b/SRV/config/init_config.py @@ -7,29 +7,33 @@ import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) + def init_config(): """Создаёт конфигурационный файл с SECRET_KEY""" config_path = "/etc/gonets/config.ini" config_dir = os.path.dirname(config_path) if os.path.exists(config_path): - logger.info(f"Конфигурационный файл {config_path} уже существует, пропускаем инициализацию") + logger.info( + f"Конфигурационный файл {config_path} уже существует, пропускаем инициализацию" + ) return try: secret_key = secrets.token_urlsafe(32) os.makedirs(config_dir, exist_ok=True) config = configparser.ConfigParser() - config["JWT"] = { - "secret_key": secret_key - } + config["JWT"] = {"secret_key": secret_key} with open(config_path, "w") as config_file: config.write(config_file) os.chmod(config_path, 0o600) logger.info(f"Создан конфигурационный файл {config_path}") except Exception as e: - logger.error(f"Ошибка при создании конфигурационного файла: {type(e).__name__} - {str(e)}") + logger.error( + f"Ошибка при создании конфигурационного файла: {type(e).__name__} - {str(e)}" + ) raise + if __name__ == "__main__": - init_config() \ No newline at end of file + init_config() diff --git a/SRV/config/ports.py b/SRV/config/ports.py index ccecd3e..1e41272 100644 --- a/SRV/config/ports.py +++ b/SRV/config/ports.py @@ -33,7 +33,6 @@ class PortConfig: 631: "IPP (Printing)", 993: "IMAPS", 995: "POP3S", - # --- Базы данных --- 1433: "Microsoft SQL Server", 1521: "Oracle DB", @@ -41,7 +40,6 @@ class PortConfig: 5432: "PostgreSQL", 6379: "Redis", 27017: "MongoDB", - # --- VPN и удалённый доступ --- 1723: "PPTP VPN", 1701: "L2TP", @@ -50,20 +48,17 @@ class PortConfig: 1194: "OpenVPN", 3389: "RDP (Remote Desktop Protocol)", 5900: "VNC", - # --- Прокси / Веб --- 1080: "SOCKS Proxy", 8000: "HTTP-Alt", 8080: "HTTP Proxy", 8443: "HTTPS-Alt", - # --- DevOps / Monitoring / Tools --- 9000: "SonarQube / PHP-FPM", 9100: "Prometheus Node Exporter", 9200: "Elasticsearch HTTP", 9300: "Elasticsearch Cluster", 11211: "Memcached", - # --- Прочее --- 6667: "IRC", 4000: "ICQ / Custom apps", @@ -104,4 +99,4 @@ class PortConfig: return list(self._ports.keys()) -port_config = PortConfig() \ No newline at end of file +port_config = PortConfig() diff --git a/SRV/core/cli_executor.py b/SRV/core/cli_executor.py index 4214ec3..a6f9332 100644 --- a/SRV/core/cli_executor.py +++ b/SRV/core/cli_executor.py @@ -1,10 +1,12 @@ import subprocess from typing import Tuple + class CLIExecutor: """ Класс для выполнения CLI-команд """ + @staticmethod def run(command: list) -> Tuple[int, str]: """ @@ -20,6 +22,6 @@ class CLIExecutor: command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, - check=False # Не вызываем исключение при ошибке + check=False, # Не вызываем исключение при ошибке ) - return result.returncode, result.stdout.decode('utf-8') + return result.returncode, result.stdout.decode("utf-8") diff --git a/SRV/main.py b/SRV/main.py index 2533bae..06bf591 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) \ No newline at end of file + # 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/auth.py b/SRV/models/auth.py index 805a5b2..194c067 100644 --- a/SRV/models/auth.py +++ b/SRV/models/auth.py @@ -1,6 +1,7 @@ from pydantic import BaseModel from typing import Optional + class UserCreate(BaseModel): """ Модель созданного пользователя @@ -10,10 +11,12 @@ class UserCreate(BaseModel): password (str): Пароль пользователя is_admin (bool): Административный доступ """ + username: str password: str is_admin: bool = False + class UserLogin(BaseModel): """ Модель авторизации пользователя @@ -22,9 +25,11 @@ class UserLogin(BaseModel): username (str): Имя пользователя password (str): Пароль пользователя """ + username: str password: str + class UserResponse(BaseModel): """ Модель проверки наличия пользователя @@ -34,10 +39,12 @@ class UserResponse(BaseModel): username (str): Имя пользователя is_admin (bool): Административный доступ """ + id: int username: str is_admin: bool + class Token(BaseModel): """ Токен пользователя @@ -48,7 +55,8 @@ class Token(BaseModel): session_id (Optional[str]): Идентификатор сессии user (Optional[UserResponse]): Пользователь на основе проверки наличия пользователя """ + access_token: str token_type: str session_id: Optional[str] = None - user: Optional[UserResponse] = None \ No newline at end of file + user: Optional[UserResponse] = None diff --git a/SRV/models/network.py b/SRV/models/network.py index 9589215..b9314d1 100644 --- a/SRV/models/network.py +++ b/SRV/models/network.py @@ -12,6 +12,7 @@ class IPv4Route(BaseModel): next_hop (str): Адрес следующего узла (маршрутизатор) metric (int): Метрика маршрута """ + destination: str prefix: int next_hop: str @@ -28,6 +29,7 @@ class IPv6Route(BaseModel): next_hop (str): Адрес следующего узла (маршрутизатор) metric (int): Метрика маршрута """ + destination: str prefix: int next_hop: str @@ -45,6 +47,7 @@ class IndividualIPConfig(BaseModel): gateway (str): Основной шлюз по умолчанию dns (List[str]): Список DNS-серверов """ + address: str routes: List[Union[IPv4Route, IPv6Route]] method: Optional[str] @@ -59,6 +62,7 @@ class IPConfig(BaseModel): Атрибуты: configurations (List[IndividualIPConfig]): Список индивидуальных IP-конфигураций """ + configurations: List[IndividualIPConfig] @@ -78,6 +82,7 @@ class NetworkInterface(BaseModel): container_x (int): Положение по X на холсте container_y (int): Положение по Y на холсте """ + name: str type: str mac_address: str @@ -102,6 +107,7 @@ class WiFiDevice(NetworkInterface): frequency (Optional[float]): Частота mode (Union[str, int]): Режим работы """ + ssid: Optional[str] = None bssid: Optional[str] = None bitrate: Optional[int] = None @@ -119,6 +125,7 @@ class NetworkInterfacesContainer(BaseModel): container_x (int): Положение по оси X container_y (int): Положение по оси Y """ + interfaces: List[Union[NetworkInterface, WiFiDevice]] container_x: int container_y: int @@ -135,6 +142,7 @@ class Activation(BaseModel): container_y (int): Положение по Y на холсте isOpen (bool): Флаг скрытия ноды на холсте """ + id: int interfaces: List[Union[NetworkInterface, WiFiDevice]] container_x: int @@ -153,6 +161,7 @@ class Deactivation(BaseModel): container_y (int): Положение по Y на холсте isOpen (bool): Флаг скрытия ноды на холсте """ + id: int interfaces: List[Union[NetworkInterface, WiFiDevice]] container_x: int diff --git a/SRV/models/request.py b/SRV/models/request.py index 4f98e7f..238c11c 100644 --- a/SRV/models/request.py +++ b/SRV/models/request.py @@ -8,4 +8,5 @@ class NoteRequest(BaseModel): Атрибуты: note (str): Заметка устройства """ - note: str \ No newline at end of file + + note: str diff --git a/SRV/models/subnet.py b/SRV/models/subnet.py index 1e22ac4..b6534fc 100644 --- a/SRV/models/subnet.py +++ b/SRV/models/subnet.py @@ -1,6 +1,7 @@ from pydantic import BaseModel from typing import Optional, List + class Device(BaseModel): """ Сканируемое устройство в подсети @@ -15,6 +16,7 @@ class Device(BaseModel): last_seen (Optional[str]): Последняя активность устройства note (Optional[str]): Заметка устройства """ + ip: str mac: Optional[str] = None hostname: Optional[str] = None @@ -24,6 +26,7 @@ class Device(BaseModel): last_seen: Optional[str] = None note: Optional[str] = None + class SubnetScanResult(BaseModel): """ Результат сканирования устройств в подсети @@ -31,4 +34,5 @@ class SubnetScanResult(BaseModel): Атрибуты: devices (List[Device]): Список просканированных устройств """ - devices: List[Device] = [] \ No newline at end of file + + devices: List[Device] = [] diff --git a/SRV/models/ufw.py b/SRV/models/ufw.py index eef34ec..c0fec4e 100644 --- a/SRV/models/ufw.py +++ b/SRV/models/ufw.py @@ -8,6 +8,7 @@ class Protocol(str, Enum): """ Enum для сетевых протоколов UFW. """ + TCP = "tcp" UDP = "udp" BOTH = "both" @@ -21,12 +22,13 @@ class Protocol(str, Enum): class Action(str, Enum): """ Enum для действий UFW. - + Attributes: ALLOW: Разрешить трафик DENY: Запретить трафик без уведомления REJECT: Запретить трафик с уведомлением отправителю """ + ALLOW = "allow" DENY = "deny" REJECT = "reject" @@ -35,17 +37,19 @@ class Action(str, Enum): class Direction(str, Enum): """ Enum для направления трафика UFW. - + Attributes: IN: Входящий трафик OUT: Исходящий трафик """ + IN = "in" OUT = "out" class UFWRule(BaseModel): """Модель правила UFW""" + action: Action direction: Direction from_ip: str @@ -58,13 +62,13 @@ class UFWRule(BaseModel): def validate_port(cls, v): """ Проверяет корректность порта или диапазона портов. - + Args: v: Порт (число) или диапазон портов (строка вида "3000:3999") - + Returns: Union[int, str]: Валидированное значение порта - + Raises: ValueError: Если порт или диапазон портов некорректен """ @@ -74,14 +78,18 @@ class UFWRule(BaseModel): return v elif isinstance(v, str): try: - start, end = map(int, v.split(':')) + start, end = map(int, v.split(":")) if not (1 <= start <= end <= 65535): raise ValueError return v except (ValueError, TypeError): - raise ValueError(f"Некорректный диапазон портов: {v}. Формат: start:end, где 1 <= start <= end <= 65535") + raise ValueError( + f"Некорректный диапазон портов: {v}. Формат: start:end, где 1 <= start <= end <= 65535" + ) else: - raise ValueError(f"Порт должен быть числом или строкой с диапазоном, получено: {type(v)}") + raise ValueError( + f"Порт должен быть числом или строкой с диапазоном, получено: {type(v)}" + ) @validator("from_ip", "to_ip") def validate_ip(cls, v): @@ -98,14 +106,18 @@ class UFWRule(BaseModel): """Проверяет совместимость версий IP адресов""" if "from_ip" not in values: return v - + # Определяем версии IP from_ip = ipaddress.ip_network(values["from_ip"], strict=False) to_ip = ipaddress.ip_network(v, strict=False) - - if isinstance(from_ip, ipaddress.IPv6Network) != isinstance(to_ip, ipaddress.IPv6Network): - raise ValueError("Несовместимые версии IP: нельзя смешивать IPv4 и IPv6 в одном правиле") - + + if isinstance(from_ip, ipaddress.IPv6Network) != isinstance( + to_ip, ipaddress.IPv6Network + ): + raise ValueError( + "Несовместимые версии IP: нельзя смешивать IPv4 и IPv6 в одном правиле" + ) + return v @validator("action", pre=True) @@ -125,44 +137,44 @@ class UFWRule(BaseModel): def to_ufw_command(self) -> str: """Преобразует правило в команду UFW.""" cmd_parts = ["ufw"] - + # Добавляем действие cmd_parts.append(self.action.value) - + # Добавляем направление cmd_parts.append(self.direction.value) - + # Добавляем from с IP cmd_parts.extend(["from", self.from_ip]) - + # Добавляем to с IP cmd_parts.extend(["to", self.to_ip]) - + # Добавляем порт и протокол port_str = str(self.port) # Работает как с числами, так и с диапазонами cmd_parts.extend(["port", port_str]) - + # Добавляем протокол proto = self.protocol.value if proto == "both": cmd_parts.append("proto tcp/udp") else: cmd_parts.extend(["proto", proto]) - + # Добавляем комментарий если есть if self.comment: cmd_parts.extend(["comment", f"'{self.comment}'"]) - + return " ".join(cmd_parts) @classmethod def from_ufw_output(cls, line: str) -> "UFWRule": """ Создает экземпляр UFWRule из строки вывода команды 'ufw status numbered'. - + Args: line: Строка вывода UFW (например, "[1] 80/tcp ALLOW IN 192.168.1.0/24 # Web Server") - + Returns: UFWRule: Новый экземпляр правила """ @@ -180,47 +192,53 @@ class UFWRule(BaseModel): (?:\#\s*(?P.*?))? \s*$ """ - + match = re.match(pattern, line, re.VERBOSE | re.IGNORECASE) if not match: raise ValueError(f"Невозможно разобрать строку правила UFW: {line}") - + groups = match.groupdict() - + # Преобразуем протокол в enum - proto_str = groups['protocol'].lower() - protocol = Protocol.BOTH if proto_str == "tcp/udp" else \ - Protocol.UDP if proto_str == "udp" else Protocol.TCP - + proto_str = groups["protocol"].lower() + protocol = ( + Protocol.BOTH + if proto_str == "tcp/udp" + else Protocol.UDP if proto_str == "udp" else Protocol.TCP + ) + # Преобразуем действие в enum (игнорируем регистр) - action_str = groups['action'].lower() - action = Action.ALLOW if action_str == "allow" else \ - Action.DENY if action_str == "deny" else Action.REJECT - + action_str = groups["action"].lower() + action = ( + Action.ALLOW + if action_str == "allow" + else Action.DENY if action_str == "deny" else Action.REJECT + ) + # Преобразуем направление в enum - direction_str = groups['direction'].lower() + direction_str = groups["direction"].lower() direction = Direction.IN if direction_str == "in" else Direction.OUT - + # Обрабатываем IP адреса - from_ip = groups['from_ip'].strip() + from_ip = groups["from_ip"].strip() if from_ip.lower() == "anywhere": from_ip = "0.0.0.0/0" - + # Для исходящих правил меняем местами from и to to_ip = "0.0.0.0/0" # По умолчанию anywhere - + # Обрабатываем порт - port = groups['port'] - if ':' in port: + port = groups["port"] + if ":" in port: # Оставляем как строку для диапазона port = port.strip() else: # Преобразуем в число для одиночного порта port = int(port) - + # Обрабатываем комментарий - comment = groups.get('comment', '').strip() if groups.get('comment') else None - + comment = groups.get("comment", "").strip() if groups.get("comment") else None + return cls( action=action, direction=direction, @@ -228,22 +246,23 @@ class UFWRule(BaseModel): to_ip=to_ip, port=port, protocol=protocol, - comment=comment + comment=comment, ) class UFWRulesResponse(BaseModel): """ Модель для представления списка правил UFW. - + Attributes: enabled: Включен ли UFW rules: Список правил UFW total_rules: Общее количество правил """ + enabled: bool = Field(..., description="Включен ли UFW") rules: List[UFWRule] = Field(default_factory=list, description="Список правил UFW") - + @property def total_rules(self) -> int: """Возвращает общее количество правил.""" @@ -253,11 +272,12 @@ class UFWRulesResponse(BaseModel): class UFWStatus(BaseModel): """ Модель для представления статуса UFW. - + Attributes: enabled: Включен ли UFW message: Сообщение о статусе last_checked: Время последней проверки статуса """ + enabled: bool = Field(..., description="Включен ли UFW") message: str = Field(..., description="Сообщение о статусе") diff --git a/SRV/models/wifi.py b/SRV/models/wifi.py index ee1b7ae..bfbb731 100644 --- a/SRV/models/wifi.py +++ b/SRV/models/wifi.py @@ -16,6 +16,7 @@ class WifiAccessPoint(BaseModel): vendor (Optional[str]): Производитель точки доступа (определяется по MAC-адресу) hidden (bool): Флаг скрытия названия точки доступа (ssid) """ + ssid: Optional[str] = None bssid: str channel: Optional[int] = None @@ -33,4 +34,5 @@ class WifiScanResult(BaseModel): Атрибуты: access_points (List[WifiAccessPoint]): Список найденных Wi-Fi точек доступа """ + access_points: List[WifiAccessPoint] diff --git a/SRV/services/arp_service.py b/SRV/services/arp_service.py index d635134..df7d115 100644 --- a/SRV/services/arp_service.py +++ b/SRV/services/arp_service.py @@ -3,20 +3,28 @@ from scapy.sendrecv import srp import netifaces from netaddr import IPNetwork + class NetworkScanner: """Иницилазицая сетевого сканера""" + def __init__(self, interface=None): - """Инициализация параметров интерфейсов """ - self.interface = interface or netifaces.gateways()['default'][netifaces.AF_INET][1] + """Инициализация параметров интерфейсов""" + self.interface = ( + interface or netifaces.gateways()["default"][netifaces.AF_INET][1] + ) 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'] + 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" + ] def _create_arp_packet(self, target_ip): """Создание пакета ARP для отправки""" arp = ARP(pdst=target_ip) ether = Ether(dst="ff:ff:ff:ff:ff:ff") - return ether/arp + return ether / arp def scan_network(self, attempts=3, timeout=2): """Подготовка к сканированию сети""" @@ -28,9 +36,11 @@ class NetworkScanner: clients = [] for _ in range(attempts): - result = srp(packet, timeout=timeout, iface=self.interface, verbose=0)[0] + result = srp(packet, timeout=timeout, iface=self.interface, verbose=0)[ + 0 + ] for _, received in result: - client = {'ip': received.psrc, 'mac': received.hwsrc} + client = {"ip": received.psrc, "mac": received.hwsrc} if client not in clients: # Избегаем дублирования клиентов clients.append(client) @@ -38,7 +48,7 @@ class NetworkScanner: # Добавляет все локальные IP-адреса интерфейса как клиенты for ip_info in self.ip_info: - local_client = {'ip': ip_info['addr'], 'mac': self.local_mac} + local_client = {"ip": ip_info["addr"], "mac": self.local_mac} if local_client not in all_clients: all_clients.append(local_client) @@ -49,10 +59,11 @@ class NetworkScanner: def print_clients(self, clients): """Лист адресов клиентов""" print("Доступные устройства в сети:") - print("IP" + " "*18+"MAC") - for client in sorted(clients, key=lambda x: x['ip']): + print("IP" + " " * 18 + "MAC") + for client in sorted(clients, key=lambda x: x["ip"]): print(f"{client['ip']:<16} {client['mac']}") + if __name__ == "__main__": scanner = NetworkScanner() clients = scanner.scan_network() diff --git a/SRV/services/auth_service.py b/SRV/services/auth_service.py index e6f198a..dc39d25 100644 --- a/SRV/services/auth_service.py +++ b/SRV/services/auth_service.py @@ -26,8 +26,10 @@ ACCESS_TOKEN_EXPIRE_MINUTES = 30 oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login") + class AuthService: """Сервис для регистрации, авторизации и управления сессиями""" + def __init__(self): self.storage = JsonStorage() self.session_storage = SessionStorage() @@ -57,10 +59,14 @@ class AuthService: db_user = self.storage.create_user(user) return db_user except ValueError as e: - logger.error(f"Ошибка при регистрации пользователя {user.username}: {str(e)}") + logger.error( + f"Ошибка при регистрации пользователя {user.username}: {str(e)}" + ) raise HTTPException(status_code=400, detail=str(e)) except Exception as e: - logger.error(f"Ошибка при регистрации пользователя {user.username}: {type(e).__name__} - {str(e)}") + logger.error( + f"Ошибка при регистрации пользователя {user.username}: {type(e).__name__} - {str(e)}" + ) raise HTTPException(status_code=500, detail="Ошибка регистрации") def login(self, user: UserLogin, client_id: str) -> dict: @@ -70,20 +76,30 @@ class AuthService: if not db_user: logger.error(f"Пользователь {user.username} не найден") raise HTTPException(status_code=401, detail="Неверный логин или пароль") - if not self.storage.verify_password(user.password, db_user["password_hash"]): + if not self.storage.verify_password( + user.password, db_user["password_hash"] + ): logger.error(f"Неверный пароль для пользователя {user.username}") raise HTTPException(status_code=401, detail="Неверный логин или пароль") access_token = self.create_access_token(data={"sub": user.username}) - session_id = self.session_storage.create_session(user.username, access_token, client_id) + session_id = self.session_storage.create_session( + user.username, access_token, client_id + ) logger.info(f"Пользователь {user.username} успешно авторизован") return { "access_token": access_token, "token_type": "bearer", "session_id": session_id, - "user": UserResponse(id=db_user["id"], username=db_user["username"], is_admin=db_user["is_admin"]) + "user": UserResponse( + id=db_user["id"], + username=db_user["username"], + is_admin=db_user["is_admin"], + ), } except Exception as e: - logger.error(f"Ошибка при авторизации пользователя {user.username}: {type(e).__name__} - {str(e)}") + logger.error( + f"Ошибка при авторизации пользователя {user.username}: {type(e).__name__} - {str(e)}" + ) raise HTTPException(status_code=500, detail="Ошибка авторизации") def get_current_user(self, token: str = Depends(oauth2_scheme)) -> UserResponse: @@ -97,7 +113,9 @@ class AuthService: if not db_user: logger.error(f"Пользователь {username} не найден") raise HTTPException(status_code=401, detail="Пользователь не найден") - return UserResponse(id=db_user["id"], username=db_user["username"], is_admin=db_user["is_admin"]) + return UserResponse( + id=db_user["id"], username=db_user["username"], is_admin=db_user["is_admin"] + ) def get_user_by_session(self, client_id: str) -> dict: """Получает пользователя и токен по client_id""" @@ -112,8 +130,12 @@ class AuthService: logger.error(f"Пользователь {username} не найден") raise HTTPException(status_code=401, detail="Пользователь не найден") return { - "user": UserResponse(id=db_user["id"], username=db_user["username"], is_admin=db_user["is_admin"]), - "access_token": token + "user": UserResponse( + id=db_user["id"], + username=db_user["username"], + is_admin=db_user["is_admin"], + ), + "access_token": token, } def logout(self, client_id: str): @@ -122,8 +144,10 @@ class AuthService: session = self.session_storage.get_latest_session(client_id) if session: self.session_storage.delete_session(session["session_id"]) - logger.info(f"Пользователь успешно вышел из системы для client_id {client_id}") + logger.info( + f"Пользователь успешно вышел из системы для client_id {client_id}" + ) return {"message": "Успешный выход из системы"} except Exception as e: logger.error(f"Ошибка при выходе из системы: {type(e).__name__} - {str(e)}") - raise HTTPException(status_code=500, detail="Ошибка при выходе из системы") \ No newline at end of file + raise HTTPException(status_code=500, detail="Ошибка при выходе из системы") diff --git a/SRV/services/firewalld_service.py b/SRV/services/firewalld_service.py index 1ee2fba..bb5e859 100644 --- a/SRV/services/firewalld_service.py +++ b/SRV/services/firewalld_service.py @@ -67,10 +67,9 @@ class FirewalldService: print(f"Timeout: {timeout}") # Get the zone interface zone_proxy = self.bus.get( - "org.fedoraproject.FirewallD1", - "/org/fedoraproject/FirewallD1" + "org.fedoraproject.FirewallD1", "/org/fedoraproject/FirewallD1" ) - zone_iface = zone_proxy['org.fedoraproject.FirewallD1.zone'] + zone_iface = zone_proxy["org.fedoraproject.FirewallD1.zone"] # Convert port to string port_str = str(port) # Convert timeout to integer diff --git a/SRV/services/ip_service.py b/SRV/services/ip_service.py index 84c3cca..89922a9 100644 --- a/SRV/services/ip_service.py +++ b/SRV/services/ip_service.py @@ -10,34 +10,39 @@ import sdbus from sdbus_async.networkmanager.settings import ConnectionProfile from services.arp_service import NetworkScanner + class IPService: """ Сервис проверки доступности IP-адреса и применения настроек сети """ + def __init__(self): self.arp_scanner = NetworkScanner() self.bus = sdbus.sd_bus_open_system() sdbus.set_default_bus(self.bus) def _reset_bus(self): - """ Перезапускает соединение с системной шиной D-Bus """ + """Перезапускает соединение с системной шиной D-Bus""" new_bus = sdbus.sd_bus_open_system() sdbus.set_default_bus(new_bus) - if hasattr(self, 'bus') and self.bus is not None: + if hasattr(self, "bus") and self.bus is not None: self.bus = None self.bus = new_bus async def check_ip_config(self, args: Dict[str, Any]) -> Dict[str, Any]: - """ Проверяет доступность IP-адреса перед его применением """ + """Проверяет доступность IP-адреса перед его применением""" self._reset_bus() clients = self.arp_scanner.scan_network() new_ip = args["ipv4_updates"]["ipv4.address-data"][0]["address"] - if any(client['ip'] == new_ip for client in clients): + if any(client["ip"] == new_ip for client in clients): return {"available": False, "message": f"IP address {new_ip} уже занят."} - return {"available": True, "message": "Network profile settings are available for use."} + return { + "available": True, + "message": "Network profile settings are available for use.", + } async def apply_ip_config(self, args: Dict[str, Any]) -> None: - """ Применяет проверенные настройки IP """ + """Применяет проверенные настройки IP""" self._reset_bus() nm = NetworkManager() settings = NetworkManagerSettings() @@ -51,7 +56,7 @@ class IPService: settings_dict = profile.to_settings_dict() for key_path, value in args["ipv4_updates"].items(): - section, *props = key_path.split('.') + section, *props = key_path.split(".") current = settings_dict[section] for prop in props[:-1]: current = current.setdefault(prop, {}) @@ -61,14 +66,16 @@ class IPService: await connection.update(new_profile.to_dbus()) print(f"Обновлены настройки для {args['connection_id']}:") - interface_name = settings_dict['connection']['interface-name'] + interface_name = settings_dict["connection"]["interface-name"] devices = await nm.get_devices() for device_path in devices: device = NetworkDeviceGeneric(device_path) if await device.interface == interface_name: await device.disconnect() print(f"Отключён от {interface_name}") - await nm.activate_connection(connection=connection_paths[0], device=device_path) + await nm.activate_connection( + connection=connection_paths[0], device=device_path + ) print(f"Переподключён к {args['connection_id']}") return diff --git a/SRV/services/network_service.py b/SRV/services/network_service.py index 1d4dbc5..7c77b8b 100644 --- a/SRV/services/network_service.py +++ b/SRV/services/network_service.py @@ -1,26 +1,27 @@ from typing import List, Dict, Tuple, Union, Optional from sdbus_async.networkmanager import ( - NetworkManager, - NetworkDeviceGeneric, - IPv4Config, - IPv6Config, - NetworkManagerSettings, - NetworkDeviceWireless, - AccessPoint, - DeviceType + NetworkManager, + NetworkDeviceGeneric, + IPv4Config, + IPv6Config, + NetworkManagerSettings, + NetworkDeviceWireless, + AccessPoint, + DeviceType, ) from sdbus import sd_bus_open_system from models.network import ( - NetworkInterface, - IPv4Route, - IPv6Route, - IPConfig, - IndividualIPConfig, - NetworkInterfacesContainer, - WiFiDevice + NetworkInterface, + IPv4Route, + IPv6Route, + IPConfig, + IndividualIPConfig, + NetworkInterfacesContainer, + WiFiDevice, ) import subprocess + class NetworkService: """Сервис для взаимодействия с NetworkManager через D-Bus""" @@ -34,71 +35,91 @@ class NetworkService: try: config = await device.get_applied_connection() settings = config[0].get(ip_type, {}) - return str(settings.get("method", ('s', 'unknown'))[1]) + return str(settings.get("method", ("s", "unknown"))[1]) except Exception: - return 'unknown' + return "unknown" async def get_connection_name(self, device: NetworkDeviceGeneric) -> str: """Извлекает имя соединения (GENERAL.CONNECTION) для устройства""" try: applied_connection = await device.get_applied_connection() connection_settings = applied_connection[0] - connection_info = connection_settings.get('connection', {}) - return str(connection_info.get('id', ('s', 'unknown'))[1]) + connection_info = connection_settings.get("connection", {}) + return str(connection_info.get("id", ("s", "unknown"))[1]) except Exception as e: # print(f"Ошибка при получении соединения для устройства {await device.interface}: {e}") # срёт ошибки дл отключенных интерфейсов # добавить потом условия для отключенных интерфейсов - return 'unknown' + return "unknown" - async def get_ip_config(self, device: NetworkDeviceGeneric, ip_version: str) -> List[IndividualIPConfig]: + async def get_ip_config( + self, device: NetworkDeviceGeneric, ip_version: str + ) -> List[IndividualIPConfig]: """Получает конфигурацию IP для устройства""" configs = [] - device_ip = await getattr(device, f'ip{ip_version}_config') - if device_ip != '/': - ip_config = globals()[f'IPv{ip_version}Config'](device_ip, self.system_bus) + device_ip = await getattr(device, f"ip{ip_version}_config") + if device_ip != "/": + ip_config = globals()[f"IPv{ip_version}Config"](device_ip, self.system_bus) address_data = await ip_config.address_data for address in address_data: config = IndividualIPConfig( - address=str(address['address'][1]), + address=str(address["address"][1]), routes=[ - globals()[f'IPv{ip_version}Route']( - destination=route.get('dest', ('s', '0.0.0.0' if ip_version == '4' else '::'))[1], - prefix=route.get('prefix', ('u', 0))[1], - next_hop=route.get('next-hop', ('s', '0.0.0.0' if ip_version == '4' else '::'))[1], - metric=route.get('metric', ('u', 0))[1] - ) for route in await ip_config.route_data + globals()[f"IPv{ip_version}Route"]( + destination=route.get( + "dest", ("s", "0.0.0.0" if ip_version == "4" else "::") + )[1], + prefix=route.get("prefix", ("u", 0))[1], + next_hop=route.get( + "next-hop", + ("s", "0.0.0.0" if ip_version == "4" else "::"), + )[1], + metric=route.get("metric", ("u", 0))[1], + ) + for route in await ip_config.route_data ], - method=await self.get_ip_method(device, f'ipv{ip_version}'), + method=await self.get_ip_method(device, f"ipv{ip_version}"), gateway=str(await ip_config.gateway), - dns=[str(ns['address'][1]) if ip_version == '4' else str(ns) for ns in await getattr(ip_config, 'nameserver_data' if ip_version == '4' else 'nameservers')] + dns=[ + str(ns["address"][1]) if ip_version == "4" else str(ns) + for ns in await getattr( + ip_config, + "nameserver_data" if ip_version == "4" else "nameservers", + ) + ], ) configs.append(config) return configs - async def get_wifi_info(self, wifi_device: NetworkDeviceWireless) -> Dict[str, Optional[Union[str, int]]]: + async def get_wifi_info( + self, wifi_device: NetworkDeviceWireless + ) -> Dict[str, Optional[Union[str, int]]]: """Получает информацию о Wi-Fi устройстве""" active_ap = await wifi_device.active_access_point wifi_info = { - 'ssid': None, - 'bssid': None, - 'bitrate': await wifi_device.bitrate, - 'strength': None, - 'frequency': None, - 'mode': None + "ssid": None, + "bssid": None, + "bitrate": await wifi_device.bitrate, + "strength": None, + "frequency": None, + "mode": None, } - if active_ap != '/': + if active_ap != "/": ap = AccessPoint(active_ap, self.system_bus) - wifi_info.update({ - 'ssid': await ap.ssid, - 'bssid': await ap.hw_address, - 'strength': await ap.strength, - 'frequency': await ap.frequency, - 'mode': await ap.mode - }) + wifi_info.update( + { + "ssid": await ap.ssid, + "bssid": await ap.hw_address, + "strength": await ap.strength, + "frequency": await ap.frequency, + "mode": await ap.mode, + } + ) return wifi_info - async def get_network_interfaces(self, initial_positions: Dict[str, Tuple[int, int]] = None) -> List[Union[NetworkInterface, WiFiDevice]]: + async def get_network_interfaces( + self, initial_positions: Dict[str, Tuple[int, int]] = None + ) -> List[Union[NetworkInterface, WiFiDevice]]: """Получает список сетевых интерфейсов""" devices_paths = await self.nm.get_devices() network_interfaces = [] @@ -109,24 +130,28 @@ class NetworkService: interface_name = str(await generic_device.interface) device_type = DeviceType(await generic_device.device_type).name - if device_type == 'LOOPBACK': + if device_type == "LOOPBACK": continue device_state = str(await generic_device.state) - if device_state == '10': + if device_state == "10": continue common_data = { - 'name': interface_name, - 'type': device_type, - 'device_state': device_state, - 'active_connection': await self.get_connection_name(generic_device), - 'mac_address': str(await generic_device.hw_address), - 'device_mtu': str(await generic_device.mtu), - 'ipv4': IPConfig(configurations=await self.get_ip_config(generic_device, '4')), - 'ipv6': IPConfig(configurations=await self.get_ip_config(generic_device, '6')), - 'x': initial_positions.get(interface_name, (100, 100))[0], - 'y': initial_positions.get(interface_name, (100, 100))[1], + "name": interface_name, + "type": device_type, + "device_state": device_state, + "active_connection": await self.get_connection_name(generic_device), + "mac_address": str(await generic_device.hw_address), + "device_mtu": str(await generic_device.mtu), + "ipv4": IPConfig( + configurations=await self.get_ip_config(generic_device, "4") + ), + "ipv6": IPConfig( + configurations=await self.get_ip_config(generic_device, "6") + ), + "x": initial_positions.get(interface_name, (100, 100))[0], + "y": initial_positions.get(interface_name, (100, 100))[1], } if device_type == "WIFI": @@ -141,7 +166,9 @@ class NetworkService: async def get_hostname(self) -> Dict[str, str]: """Возвращает имя хоста из команды hostname в формате JSON""" try: - result = subprocess.run(['hostname'], capture_output=True, text=True, check=True) + result = subprocess.run( + ["hostname"], capture_output=True, text=True, check=True + ) hostname = result.stdout.strip() return {"hostname": hostname} except subprocess.CalledProcessError as e: @@ -149,52 +176,77 @@ class NetworkService: except Exception as e: raise RuntimeError(f"Error processing hostname output: {e}") - async def get_interfaces_by_state(self, state: str, initial_positions: Dict[str, Tuple[int, int]] = None) -> List[NetworkInterface]: + async def get_interfaces_by_state( + self, state: str, initial_positions: Dict[str, Tuple[int, int]] = None + ) -> List[NetworkInterface]: """ Получает список интерфейсов по состоянию """ all_interfaces = await self.get_network_interfaces(initial_positions) - return [interface for interface in all_interfaces if interface.device_state == state] + return [ + interface for interface in all_interfaces if interface.device_state == state + ] - async def get_interfaces_container(self, state: Optional[str] = None, initial_positions: Dict[str, Tuple[int, int]] = None, container_x: int = 0, container_y: int = 0) -> NetworkInterfacesContainer: - """ Получает объект, содержащий список интерфейсов и координаты контейнера """ + async def get_interfaces_container( + self, + state: Optional[str] = None, + initial_positions: Dict[str, Tuple[int, int]] = None, + container_x: int = 0, + container_y: int = 0, + ) -> NetworkInterfacesContainer: + """Получает объект, содержащий список интерфейсов и координаты контейнера""" if state: interfaces = await self.get_interfaces_by_state(state, initial_positions) else: interfaces = await self.get_network_interfaces(initial_positions) return NetworkInterfacesContainer( - interfaces=interfaces, - container_x=container_x, - container_y=container_y + interfaces=interfaces, container_x=container_x, container_y=container_y ) - async def get_network_interfaces_container(self, initial_positions: Optional[Dict[str, Tuple[int, int]]] = None, ) -> NetworkInterfacesContainer: - """ Получает контейнер с сетевыми интерфейсами и координатами контейнера """ + async def get_network_interfaces_container( + self, + initial_positions: Optional[Dict[str, Tuple[int, int]]] = None, + ) -> NetworkInterfacesContainer: + """Получает контейнер с сетевыми интерфейсами и координатами контейнера""" interfaces = await self.get_network_interfaces(initial_positions) return NetworkInterfacesContainer( - interfaces=interfaces, - container_x=0, - container_y=0 + interfaces=interfaces, container_x=0, container_y=0 ) - async def get_activated_interfaces(self, initial_positions: Dict[str, Tuple[int, int]] = None) -> List[NetworkInterface]: - """ Получает список активных сетевых интерфейсов """ - return await self.get_interfaces_by_state('100', initial_positions) + async def get_activated_interfaces( + self, initial_positions: Dict[str, Tuple[int, int]] = None + ) -> List[NetworkInterface]: + """Получает список активных сетевых интерфейсов""" + return await self.get_interfaces_by_state("100", initial_positions) - async def get_deactivated_interfaces(self, initial_positions: Dict[str, Tuple[int, int]] = None) -> List[NetworkInterface]: - """ Получает список неактивных сетевых интерфейсов """ + async def get_deactivated_interfaces( + self, initial_positions: Dict[str, Tuple[int, int]] = None + ) -> List[NetworkInterface]: + """Получает список неактивных сетевых интерфейсов""" all_interfaces = await self.get_network_interfaces(initial_positions) - return [interface for interface in all_interfaces if interface.device_state != '100'] + return [ + interface for interface in all_interfaces if interface.device_state != "100" + ] - async def get_activated_interfaces_container(self, initial_positions: Dict[str, Tuple[int, int]] = None, container_x: int = 0, container_y: int = 0) -> NetworkInterfacesContainer: - """ Получает объект, содержащий список активных интерфейсов и координаты контейнера """ - return await self.get_interfaces_container('100', initial_positions, container_x, container_y) + async def get_activated_interfaces_container( + self, + initial_positions: Dict[str, Tuple[int, int]] = None, + container_x: int = 0, + container_y: int = 0, + ) -> NetworkInterfacesContainer: + """Получает объект, содержащий список активных интерфейсов и координаты контейнера""" + return await self.get_interfaces_container( + "100", initial_positions, container_x, container_y + ) - async def get_deactivated_interfaces_container(self, initial_positions: Dict[str, Tuple[int, int]] = None, container_x: int = 0, container_y: int = 0) -> NetworkInterfacesContainer: - """ Получает объект, содержащий список неактивных интерфейсов и координаты контейнера """ + async def get_deactivated_interfaces_container( + self, + initial_positions: Dict[str, Tuple[int, int]] = None, + container_x: int = 0, + container_y: int = 0, + ) -> NetworkInterfacesContainer: + """Получает объект, содержащий список неактивных интерфейсов и координаты контейнера""" interfaces = await self.get_deactivated_interfaces(initial_positions) return NetworkInterfacesContainer( - interfaces=interfaces, - container_x=container_x, - container_y=container_y + interfaces=interfaces, container_x=container_x, container_y=container_y ) diff --git a/SRV/services/subnet_service.py b/SRV/services/subnet_service.py index ab395c8..954c6bd 100644 --- a/SRV/services/subnet_service.py +++ b/SRV/services/subnet_service.py @@ -19,9 +19,12 @@ from scapy.layers.inet import IP, TCP from scapy.layers.l2 import ARP, Ether from scapy.sendrecv import srp, sr1 -logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s") +logging.basicConfig( + level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" +) logger = logging.getLogger(__name__) + class NetworkScanner: def __init__(self, interface: str): self.interface = interface @@ -38,6 +41,7 @@ class NetworkScanner: logger.error(f"Ошибка ARP-сканирования подсети {subnet}: {str(e)}") return [] + class PortScanner: def __init__(self, interface: str): self.interface = interface @@ -51,7 +55,7 @@ class PortScanner: IP(dst=ip) / TCP(dport=port, flags="S"), timeout=2, verbose=False, - iface=self.interface + iface=self.interface, ) if pkt and pkt.haslayer(TCP) and pkt[TCP].flags == 0x12: return port_config.get_port_name(port) @@ -71,6 +75,7 @@ class PortScanner: logger.warning(f"Ошибка при сканировании портов для IP {ip}: {str(e)}") return [] + class DeviceInfoCollector: def __init__(self): self.mac_lookup = MacLookup() @@ -80,7 +85,9 @@ class DeviceInfoCollector: try: return self.mac_lookup.lookup(mac) except Exception as e: - logger.warning(f"Не удалось определить производителя для MAC {mac}: {str(e)}") + logger.warning( + f"Не удалось определить производителя для MAC {mac}: {str(e)}" + ) return "Unknown" def get_hostname(self, ip: str) -> Optional[str]: @@ -89,6 +96,7 @@ class DeviceInfoCollector: except: return None + class DeviceStorage: def __init__(self, storage_path: str = "storage/devices/devices.json"): self.storage_file = Path(storage_path) @@ -137,6 +145,7 @@ class DeviceStorage: except Exception as e: logger.error(f"Ошибка при сохранении устройств: {str(e)}") + class SubnetService: def __init__(self): self.interface = self._get_default_interface() @@ -187,12 +196,14 @@ class SubnetService: return Device( ip=ip, mac=mac, - hostname=self.device_info.get_hostname(ip) or existing_device.hostname, + hostname=self.device_info.get_hostname(ip) + or existing_device.hostname, vendor=self.device_info.get_vendor(mac), - open_ports=self.port_scanner.scan(ip) or existing_device.open_ports, + open_ports=self.port_scanner.scan(ip) + or existing_device.open_ports, status="online", last_seen=now, - note=existing_device.note + note=existing_device.note, ) vendor = self.device_info.get_vendor(mac) hostname = self.device_info.get_hostname(ip) @@ -205,7 +216,7 @@ class SubnetService: open_ports=open_ports, status="online", last_seen=now, - note=None + note=None, ) with ThreadPoolExecutor(max_workers=10) as executor: @@ -251,7 +262,9 @@ class SubnetService: vendor = self.device_info.get_vendor(mac) hostname = self.device_info.get_hostname(ip) open_ports = self.port_scanner.scan(ip) - logger.debug(f"Локальное устройство: IP={ip}, MAC={mac}, Ports={open_ports}") + logger.debug( + f"Локальное устройство: IP={ip}, MAC={mac}, Ports={open_ports}" + ) return Device( ip=ip, mac=mac, @@ -260,7 +273,7 @@ class SubnetService: open_ports=open_ports, status="online", last_seen=now, - note=None + note=None, ) logger.warning(f"Локальное устройство не найдено в подсети {subnet}") return None @@ -358,4 +371,4 @@ class SubnetService: return True, None except Exception as e: logger.error(f"Ошибка при удалении заметки для MAC {mac}: {str(e)}") - return False, f"Внутренняя ошибка сервера: {str(e)}" \ No newline at end of file + return False, f"Внутренняя ошибка сервера: {str(e)}" diff --git a/SRV/services/wifi_service.py b/SRV/services/wifi_service.py index ffd2b5e..e14c254 100644 --- a/SRV/services/wifi_service.py +++ b/SRV/services/wifi_service.py @@ -9,9 +9,12 @@ import pywifi from pywifi import Profile from models.wifi import WifiAccessPoint, WifiScanResult -logging.basicConfig(level=logging.DEBUG, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s") +logging.basicConfig( + level=logging.DEBUG, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" +) logger = logging.getLogger(__name__) + class WifiService: """Сервис для сканирования текущих Wi-Fi точек доступа""" @@ -28,11 +31,15 @@ class WifiService: logger.error("Не найдено сетевых интерфейсов") return None - logger.debug(f"Доступные интерфейсы: {[iface.name() for iface in interfaces]}") + logger.debug( + f"Доступные интерфейсы: {[iface.name() for iface in interfaces]}" + ) for iface in interfaces: iface_name = iface.name() - if 'p2p' in iface_name.lower(): # Исключаю сетевые интерфейсы типа p2p из выбора, возможно сделать решение получше нежели исключать интерфейсы + if ( + "p2p" in iface_name.lower() + ): # Исключаю сетевые интерфейсы типа p2p из выбора, возможно сделать решение получше нежели исключать интерфейсы logger.debug(f"Пропущен интерфейс {iface_name} (P2P)") continue try: @@ -42,7 +49,9 @@ class WifiService: if results: logger.info(f"Выбран Wi-Fi-интерфейс: {iface_name}") return iface - logger.debug(f"Интерфейс {iface_name} не вернул результатов сканирования") + logger.debug( + f"Интерфейс {iface_name} не вернул результатов сканирования" + ) except Exception as e: logger.debug(f"Интерфейс {iface_name} неактивен: {str(e)}") @@ -57,7 +66,9 @@ class WifiService: try: return self.mac_lookup.lookup(mac) except Exception as e: - logger.warning(f"Не удалось определить производителя для MAC {mac}: {str(e)}") + logger.warning( + f"Не удалось определить производителя для MAC {mac}: {str(e)}" + ) return "Unknown" def _get_encryption_from_iw(self, ap_info: Dict) -> str: @@ -86,7 +97,9 @@ class WifiService: logger.debug(f"Частота {frequency} вне диапазона 2.4/5 GHz") return None except Exception as e: - logger.debug(f"Ошибка при определении канала для частоты {frequency}: {str(e)}") + logger.debug( + f"Ошибка при определении канала для частоты {frequency}: {str(e)}" + ) return None def _get_frequency(self, frequency: Optional[int]) -> Optional[str]: @@ -101,7 +114,9 @@ class WifiService: logger.debug(f"Частота {frequency} вне диапазона 2.4/5 GHz") return None except Exception as e: - logger.debug(f"Ошибка при определении диапазона для частоты {frequency}: {str(e)}") + logger.debug( + f"Ошибка при определении диапазона для частоты {frequency}: {str(e)}" + ) return None def _get_iw_scan_data(self) -> List[Dict]: @@ -137,7 +152,9 @@ class WifiService: current_ap = {"bssid": bssid_match.group(1).lower(), "security": ""} elif current_ap: if ssid_match := ssid_pattern.match(line): - current_ap["ssid"] = ssid_match.group(1) if ssid_match.group(1).strip() else "" + current_ap["ssid"] = ( + ssid_match.group(1) if ssid_match.group(1).strip() else "" + ) elif freq_match := freq_pattern.match(line): current_ap["frequency"] = int(freq_match.group(1)) elif signal_match := signal_pattern.match(line): @@ -152,7 +169,9 @@ class WifiService: if current_ap: aps.append(current_ap) - logger.debug(f"Найдено {len(aps)} точек через iw: {[ap['bssid'] for ap in aps]}") + logger.debug( + f"Найдено {len(aps)} точек через iw: {[ap['bssid'] for ap in aps]}" + ) for ap in aps: logger.debug(f"Данные точки {ap['bssid']}: {ap}") return aps @@ -176,16 +195,22 @@ class WifiService: return WifiScanResult(access_points=aps) try: - logger.info(f"Начало сканирования Wi-Fi на интерфейсе {self.interface.name()}") + logger.info( + f"Начало сканирования Wi-Fi на интерфейсе {self.interface.name()}" + ) self.interface.scan() - time.sleep(10) # Нативное значение для wifi интерфейса предназначенное на сканирование и сбор результата. Оптимальное решение от 5 до 15 секунд + time.sleep( + 10 + ) # Нативное значение для wifi интерфейса предназначенное на сканирование и сбор результата. Оптимальное решение от 5 до 15 секунд py_results = self.interface.scan_results() logger.debug(f"Найдено {len(py_results)} точек через pywifi") iw_results = self._get_iw_scan_data() - iw_by_bssid = {ap['bssid']: ap for ap in iw_results if ap.get('bssid')} + iw_by_bssid = {ap["bssid"]: ap for ap in iw_results if ap.get("bssid")} - best_aps_by_ssid: Dict[str, WifiAccessPoint] = {} # Сделал словарь, чтобы агрегировать точки доступа у сетей для выявления лучшего + best_aps_by_ssid: Dict[str, WifiAccessPoint] = ( + {} + ) # Сделал словарь, чтобы агрегировать точки доступа у сетей для выявления лучшего for ap in py_results: bssid = ap.bssid.lower() if ap.bssid else None if not bssid: @@ -203,7 +228,8 @@ class WifiService: vendor = self.get_vendor(bssid) logger.debug( - f"Точка {bssid}: freq={frequency}, channel={channel}, encryption={encryption}, signal={signal}") + f"Точка {bssid}: freq={frequency}, channel={channel}, encryption={encryption}, signal={signal}" + ) current_ap = WifiAccessPoint( ssid=ssid, @@ -213,10 +239,14 @@ class WifiService: encryption=encryption, frequency=freq_band, vendor=vendor, - hidden=hidden + hidden=hidden, ) - if ssid not in best_aps_by_ssid or (current_ap.signal and best_aps_by_ssid[ssid].signal and current_ap.signal > best_aps_by_ssid[ssid].signal): + if ssid not in best_aps_by_ssid or ( + current_ap.signal + and best_aps_by_ssid[ssid].signal + and current_ap.signal > best_aps_by_ssid[ssid].signal + ): best_aps_by_ssid[ssid] = current_ap aps = list(best_aps_by_ssid.values()) @@ -229,4 +259,4 @@ class WifiService: except Exception as e: logger.error(f"Ошибка при сканировании Wi-Fi: {str(e)}") - return WifiScanResult(access_points=aps) \ No newline at end of file + return WifiScanResult(access_points=aps) diff --git a/SRV/storage/json_storage.py b/SRV/storage/json_storage.py index 6562f24..f86d630 100644 --- a/SRV/storage/json_storage.py +++ b/SRV/storage/json_storage.py @@ -9,6 +9,7 @@ from models.auth import UserCreate, UserResponse logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) + class JsonStorage: """Хранилище пользователей в JSON-файле""" @@ -32,7 +33,9 @@ class JsonStorage: else: logger.info(f"Файл {self.file_path} уже существует") except Exception as e: - logger.error(f"Ошибка при создании файла {self.file_path}: {type(e).__name__} - {str(e)}") + logger.error( + f"Ошибка при создании файла {self.file_path}: {type(e).__name__} - {str(e)}" + ) raise def _read_data(self) -> dict: @@ -41,10 +44,14 @@ class JsonStorage: with open(self.file_path, "r") as f: return json.load(f) except FileNotFoundError: - logger.info(f"Файл {self.file_path} не существует, возвращаем пустую структуру") + logger.info( + f"Файл {self.file_path} не существует, возвращаем пустую структуру" + ) return {"users": [], "next_id": 1} except Exception as e: - logger.error(f"Ошибка при чтении файла {self.file_path}: {type(e).__name__} - {str(e)}") + logger.error( + f"Ошибка при чтении файла {self.file_path}: {type(e).__name__} - {str(e)}" + ) raise def _write_data(self, data: dict): @@ -53,7 +60,9 @@ class JsonStorage: with open(self.file_path, "w") as f: json.dump(data, f, indent=2) except Exception as e: - logger.error(f"Ошибка при записи файла {self.file_path}: {type(e).__name__} - {str(e)}") + logger.error( + f"Ошибка при записи файла {self.file_path}: {type(e).__name__} - {str(e)}" + ) raise def create_user(self, user: UserCreate) -> UserResponse: @@ -70,22 +79,30 @@ class JsonStorage: "id": next_id, "username": user.username, "password_hash": password_hash, - "is_admin": user.is_admin + "is_admin": user.is_admin, } users.append(new_user) data["next_id"] = next_id + 1 self._write_data(data) logger.info(f"Пользователь {user.username} успешно зарегистрирован") - return UserResponse(id=new_user["id"], username=new_user["username"], is_admin=new_user["is_admin"]) + return UserResponse( + id=new_user["id"], + username=new_user["username"], + is_admin=new_user["is_admin"], + ) except Exception as e: - logger.error(f"Ошибка при создании пользователя {user.username}: {type(e).__name__} - {str(e)}") + logger.error( + f"Ошибка при создании пользователя {user.username}: {type(e).__name__} - {str(e)}" + ) raise def get_user_by_username(self, username: str) -> Optional[dict]: """Получает пользователя по имени""" try: if not os.path.exists(self.file_path): - logger.info(f"Файл {self.file_path} не существует, пользователь {username} не найден") + logger.info( + f"Файл {self.file_path} не существует, пользователь {username} не найден" + ) return None data = self._read_data() for user in data["users"]: @@ -93,9 +110,11 @@ class JsonStorage: return user return None except Exception as e: - logger.error(f"Ошибка при получении пользователя {username}: {type(e).__name__} - {str(e)}") + logger.error( + f"Ошибка при получении пользователя {username}: {type(e).__name__} - {str(e)}" + ) raise def verify_password(self, plain_password: str, hashed_password: str) -> bool: """Проверяет пароль""" - return self.pwd_context.verify(plain_password, hashed_password) \ No newline at end of file + return self.pwd_context.verify(plain_password, hashed_password) diff --git a/SRV/storage/session_storage.py b/SRV/storage/session_storage.py index c34568e..8854af1 100644 --- a/SRV/storage/session_storage.py +++ b/SRV/storage/session_storage.py @@ -9,6 +9,7 @@ from datetime import datetime, timedelta logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) + class SessionStorage: def __init__(self, file_path: str = "/etc/gonets/sessions.json"): self.file_path = file_path @@ -28,7 +29,9 @@ class SessionStorage: else: logger.info(f"Файл {self.file_path} уже существует") except Exception as e: - logger.error(f"Ошибка при создании файла {self.file_path}: {type(e).__name__} - {str(e)}") + logger.error( + f"Ошибка при создании файла {self.file_path}: {type(e).__name__} - {str(e)}" + ) raise def _read_data(self) -> Dict: @@ -36,7 +39,9 @@ class SessionStorage: with open(self.file_path, "r") as f: return json.load(f) except Exception as e: - logger.error(f"Ошибка при чтении файла {self.file_path}: {type(e).__name__} - {str(e)}") + logger.error( + f"Ошибка при чтении файла {self.file_path}: {type(e).__name__} - {str(e)}" + ) raise def _write_data(self, data: Dict): @@ -44,7 +49,9 @@ class SessionStorage: with open(self.file_path, "w") as f: json.dump(data, f, indent=2) except Exception as e: - logger.error(f"Ошибка при записи файла {self.file_path}: {type(e).__name__} - {str(e)}") + logger.error( + f"Ошибка при записи файла {self.file_path}: {type(e).__name__} - {str(e)}" + ) raise def create_session(self, username: str, token: str, client_id: str) -> str: @@ -60,13 +67,17 @@ class SessionStorage: "token": token, "client_id": client_id, "created_at": created_at, - "expiry": expiry + "expiry": expiry, } self._write_data(data) - logger.info(f"Сессия создана для пользователя {username} с session_id {session_id}") + logger.info( + f"Сессия создана для пользователя {username} с session_id {session_id}" + ) return session_id except Exception as e: - logger.error(f"Ошибка при создании сессии для {username}: {type(e).__name__} - {str(e)}") + logger.error( + f"Ошибка при создании сессии для {username}: {type(e).__name__} - {str(e)}" + ) raise def get_latest_session(self, client_id: str) -> Optional[Dict]: @@ -95,7 +106,9 @@ class SessionStorage: self._write_data(data) return None except Exception as e: - logger.error(f"Ошибка при получении сессии для client_id {client_id}: {type(e).__name__} - {str(e)}") + logger.error( + f"Ошибка при получении сессии для client_id {client_id}: {type(e).__name__} - {str(e)}" + ) raise def delete_session(self, session_id: str): @@ -108,7 +121,9 @@ class SessionStorage: self._write_data(data) logger.info(f"Сессия с session_id {session_id} удалена") except Exception as e: - logger.error(f"Ошибка при удалении сессии {session_id}: {type(e).__name__} - {str(e)}") + logger.error( + f"Ошибка при удалении сессии {session_id}: {type(e).__name__} - {str(e)}" + ) raise def clean_expired_sessions(self): @@ -137,5 +152,7 @@ class SessionStorage: logger.debug("Истекшие сессии не найдены") except Exception as e: - logger.error(f"Ошибка при очистке истекших сессий: {type(e).__name__} - {str(e)}") - raise \ No newline at end of file + logger.error( + f"Ошибка при очистке истекших сессий: {type(e).__name__} - {str(e)}" + ) + raise