diff --git a/SRV/api/v1/auth.py b/SRV/api/v1/auth.py index ad90dec..ffb5d6d 100644 --- a/SRV/api/v1/auth.py +++ b/SRV/api/v1/auth.py @@ -26,9 +26,8 @@ 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) @@ -46,9 +45,8 @@ async def login(user: UserLogin, request: Request): 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") @@ -66,10 +64,10 @@ async def get_current_user(request: Request): 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="Не удалось получить данные пользователя" - ) + logger.error( + f"Ошибка в get_current_user: {type(e).__name__} - {str(e)}") + raise HTTPException(status_code=500, + detail="Не удалось получить данные пользователя") @router.get("/logout") @@ -83,7 +81,8 @@ async def logout(request: Request): raise e except Exception as e: logger.error(f"Ошибка в logout: {type(e).__name__} - {str(e)}") - raise HTTPException(status_code=500, detail="Не удалось выйти из системы") + raise HTTPException(status_code=500, + detail="Не удалось выйти из системы") @router.get("/check_json") @@ -104,21 +103,28 @@ async def check_users_file(): data = storage._read_data() has_users = len(data.get("users", [])) > 0 if has_users: - logger.info(f"Файл {file_path} существует и содержит пользователей") + logger.info( + f"Файл {file_path} существует и содержит пользователей") return { "exists": True, "has_users": True, - "message": "Файл users.json существует и содержит пользователей", + "message": + "Файл users.json существует и содержит пользователей", } else: - logger.info(f"Файл {file_path} существует, но не содержит пользователей") + logger.info( + f"Файл {file_path} существует, но не содержит пользователей") return { - "exists": True, - "has_users": False, - "message": "Файл users.json существует, но не содержит пользователей", + "exists": + True, + "has_users": + False, + "message": + "Файл users.json существует, но не содержит пользователей", } except Exception as e: logger.error( f"Ошибка при проверке файла users.json: {type(e).__name__} - {str(e)}" ) - raise HTTPException(status_code=500, detail="Ошибка при проверке файла") + raise HTTPException(status_code=500, + detail="Ошибка при проверке файла") diff --git a/SRV/api/v1/firewalld.py b/SRV/api/v1/firewalld.py index 4f2515f..c1799a6 100644 --- a/SRV/api/v1/firewalld.py +++ b/SRV/api/v1/firewalld.py @@ -31,7 +31,9 @@ class SuccessResponse(BaseModel): @router.get( "/zones", response_model=List[str], - responses={500: {"model": ErrorResponse}}, + responses={500: { + "model": ErrorResponse + }}, summary="Get all available zones", ) async def get_zones() -> List[str]: @@ -53,7 +55,9 @@ async def get_zones() -> List[str]: @router.get( "/zones/default", response_model=str, - responses={500: {"model": ErrorResponse}}, + responses={500: { + "model": ErrorResponse + }}, summary="Get default zone", ) async def get_default_zone() -> str: @@ -75,7 +79,14 @@ async def get_default_zone() -> str: @router.put( "/zones/default/{zone}", response_model=SuccessResponse, - responses={404: {"model": ErrorResponse}, 500: {"model": ErrorResponse}}, + responses={ + 404: { + "model": ErrorResponse + }, + 500: { + "model": ErrorResponse + } + }, summary="Set default zone", ) async def set_default_zone( @@ -94,9 +105,8 @@ async def set_default_zone( firewalld.set_default_zone(zone) return SuccessResponse(message=f"Default zone set to {zone}") except ValueError: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, detail=f"Zone '{zone}' not found" - ) + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, + detail=f"Zone '{zone}' not found") except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, @@ -107,10 +117,18 @@ async def set_default_zone( @router.get( "/zones/{zone}", response_model=FirewallZone, - responses={404: {"model": ErrorResponse}, 500: {"model": ErrorResponse}}, + responses={ + 404: { + "model": ErrorResponse + }, + 500: { + "model": ErrorResponse + } + }, summary="Get zone configuration", ) -async def get_zone(zone: str = Path(..., description="Zone name")) -> FirewallZone: +async def get_zone(zone: str = Path(..., + description="Zone name")) -> FirewallZone: """ Get detailed configuration for a specific zone. @@ -122,13 +140,13 @@ 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 - ) + return FirewallZone(name=zone, + target=FirewallTarget.DEFAULT, + interfaces=[], + **zone_info) except ValueError: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, detail=f"Zone '{zone}' not found" - ) + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, + detail=f"Zone '{zone}' not found") except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, @@ -139,10 +157,13 @@ async def get_zone(zone: str = Path(..., description="Zone name")) -> FirewallZo @router.get( "/zones/{zone}/services", response_model=List[str], - responses={500: {"model": ErrorResponse}}, + responses={500: { + "model": ErrorResponse + }}, summary="Get services in zone", ) -async def get_services(zone: str = Path(..., description="Zone name")) -> List[str]: +async def get_services(zone: str = Path(..., + description="Zone name")) -> List[str]: """ Get all services in a zone. @@ -164,7 +185,14 @@ async def get_services(zone: str = Path(..., description="Zone name")) -> List[s @router.post( "/zones/{zone}/services", response_model=SuccessResponse, - responses={404: {"model": ErrorResponse}, 500: {"model": ErrorResponse}}, + responses={ + 404: { + "model": ErrorResponse + }, + 500: { + "model": ErrorResponse + } + }, summary="Add service to zone", ) async def add_service( @@ -183,11 +211,11 @@ async def add_service( """ try: firewalld.add_service(zone, rule.name, rule.timeout) - return SuccessResponse(message=f"Service {rule.name} added to zone {zone}") + return SuccessResponse( + message=f"Service {rule.name} added to zone {zone}") except ValueError: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, detail=f"Zone '{zone}' not found" - ) + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, + detail=f"Zone '{zone}' not found") except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, @@ -198,12 +226,19 @@ async def add_service( @router.delete( "/zones/{zone}/services/{service}", response_model=SuccessResponse, - responses={404: {"model": ErrorResponse}, 500: {"model": ErrorResponse}}, + responses={ + 404: { + "model": ErrorResponse + }, + 500: { + "model": ErrorResponse + } + }, summary="Remove service from zone", ) async def remove_service( - zone: str = Path(..., description="Zone name"), - service: str = Path(..., description="Service name"), + zone: str = Path(..., description="Zone name"), + service: str = Path(..., description="Service name"), ) -> SuccessResponse: """ Remove a service from a zone. @@ -217,7 +252,8 @@ async def remove_service( """ try: firewalld.remove_service(zone, service) - return SuccessResponse(message=f"Service {service} removed from zone {zone}") + return SuccessResponse( + message=f"Service {service} removed from zone {zone}") except ValueError: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, @@ -233,10 +269,13 @@ async def remove_service( @router.get( "/zones/{zone}/ports", response_model=List[str], - responses={500: {"model": ErrorResponse}}, + responses={500: { + "model": ErrorResponse + }}, summary="Get ports in zone", ) -async def get_ports(zone: str = Path(..., description="Zone name")) -> List[str]: +async def get_ports(zone: str = Path(..., + description="Zone name")) -> List[str]: """ Get all ports in a zone. @@ -259,9 +298,15 @@ async def get_ports(zone: str = Path(..., description="Zone name")) -> List[str] "/zones/{zone}/ports", response_model=SuccessResponse, responses={ - 400: {"model": ErrorResponse}, - 404: {"model": ErrorResponse}, - 500: {"model": ErrorResponse}, + 400: { + "model": ErrorResponse + }, + 404: { + "model": ErrorResponse + }, + 500: { + "model": ErrorResponse + }, }, summary="Add port to zone", ) @@ -282,10 +327,11 @@ async def add_port( try: firewalld.add_port(zone, rule.port, rule.protocol.value, rule.timeout) return SuccessResponse( - message=f"Port {rule.port}/{rule.protocol.value} added to zone {zone}" - ) + message= + f"Port {rule.port}/{rule.protocol.value} added to zone {zone}") except ValueError as e: - raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, + detail=str(e)) except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, @@ -296,7 +342,14 @@ async def add_port( @router.delete( "/zones/{zone}/ports", response_model=SuccessResponse, - responses={404: {"model": ErrorResponse}, 500: {"model": ErrorResponse}}, + responses={ + 404: { + "model": ErrorResponse + }, + 500: { + "model": ErrorResponse + } + }, summary="Remove port from zone", ) async def remove_port( @@ -316,12 +369,11 @@ async def remove_port( try: firewalld.remove_port(zone, rule.port, rule.protocol.value) return SuccessResponse( - message=f"Port {rule.port}/{rule.protocol.value} removed from zone {zone}" - ) + message= + f"Port {rule.port}/{rule.protocol.value} removed from zone {zone}") except ValueError: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, detail=f"Zone '{zone}' not found" - ) + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, + detail=f"Zone '{zone}' not found") except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, @@ -332,10 +384,13 @@ async def remove_port( @router.get( "/zones/{zone}/sources", response_model=List[str], - responses={500: {"model": ErrorResponse}}, + responses={500: { + "model": ErrorResponse + }}, summary="Get sources in zone", ) -async def get_sources(zone: str = Path(..., description="Zone name")) -> List[str]: +async def get_sources(zone: str = Path(..., + description="Zone name")) -> List[str]: """ Get all sources in a zone. @@ -357,7 +412,14 @@ async def get_sources(zone: str = Path(..., description="Zone name")) -> List[st @router.post( "/zones/{zone}/sources", response_model=SuccessResponse, - responses={404: {"model": ErrorResponse}, 500: {"model": ErrorResponse}}, + responses={ + 404: { + "model": ErrorResponse + }, + 500: { + "model": ErrorResponse + } + }, summary="Add source to zone", ) async def add_source( @@ -376,11 +438,11 @@ async def add_source( """ try: firewalld.add_source(zone, rule.address) - return SuccessResponse(message=f"Source {rule.address} added to zone {zone}") + return SuccessResponse( + message=f"Source {rule.address} added to zone {zone}") except ValueError: - raise HTTPException( - status_code=status.HTTP_404_NOT_FOUND, detail=f"Zone '{zone}' not found" - ) + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, + detail=f"Zone '{zone}' not found") except Exception as e: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, @@ -391,7 +453,14 @@ async def add_source( @router.delete( "/zones/{zone}/sources/{source}", response_model=SuccessResponse, - responses={404: {"model": ErrorResponse}, 500: {"model": ErrorResponse}}, + responses={ + 404: { + "model": ErrorResponse + }, + 500: { + "model": ErrorResponse + } + }, summary="Remove source from zone", ) async def remove_source( @@ -410,7 +479,8 @@ async def remove_source( """ try: firewalld.remove_source(zone, source) - return SuccessResponse(message=f"Source {source} removed from zone {zone}") + return SuccessResponse( + message=f"Source {source} removed from zone {zone}") except ValueError: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, @@ -426,7 +496,9 @@ async def remove_source( @router.post( "/reload", response_model=SuccessResponse, - responses={500: {"model": ErrorResponse}}, + responses={500: { + "model": ErrorResponse + }}, summary="Reload firewall rules", ) async def reload_firewall() -> SuccessResponse: diff --git a/SRV/api/v1/ip.py b/SRV/api/v1/ip.py index bcd9589..b5702d3 100644 --- a/SRV/api/v1/ip.py +++ b/SRV/api/v1/ip.py @@ -14,8 +14,8 @@ async def check_ip_config(data: Dict[str, Any] = Body(...)): return result except Exception as e: raise HTTPException( - status_code=500, detail=f"Ошибка проверки IP конфигурации: {str(e)}" - ) + status_code=500, + detail=f"Ошибка проверки IP конфигурации: {str(e)}") @router.post("/apply_ip") @@ -24,15 +24,17 @@ async def apply_ip_config(data: Dict[str, Any] = Body(...)): try: await ip_service.apply_ip_config(data) return { - "success": True, - "message": f"Настройки сетевого профиля для {data['connection_id']} применены", + "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)}" - ) + status_code=500, + detail=f"Ошибка проверки IP конфигурации: {str(e)}") @router.post("/save_ip") @@ -41,12 +43,14 @@ async def save_ip_config(data: Dict[str, Any] = Body(...)): try: await ip_service.apply_ip_config(data) return { - "success": True, - "message": f"Настройки сетевого профиля для {data['connection_id']} применены", + "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)}" - ) + status_code=500, + detail=f"Ошибка проверки IP конфигурации: {str(e)}") diff --git a/SRV/api/v1/network.py b/SRV/api/v1/network.py index cefbbc5..78c0cae 100644 --- a/SRV/api/v1/network.py +++ b/SRV/api/v1/network.py @@ -25,17 +25,27 @@ 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]: + 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"] - def update_position(self, container_id: int, x: int, y: int, isOpen: bool) -> None: + 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} @@ -52,15 +62,12 @@ async def get_network_data() -> Dict[str, Dict]: network_data = await network_service.get_network_interfaces_container() activation_interfaces, deactivation_interfaces = _split_interfaces( - network_data.interfaces - ) + network_data.interfaces) activation_x, activation_y, activation_is_open = ( - container_manager.get_position_and_state("activation") - ) + container_manager.get_position_and_state("activation")) deactivation_x, deactivation_y, deactivation_is_open = ( - container_manager.get_position_and_state("deactivation") - ) + container_manager.get_position_and_state("deactivation")) activation = Activation( id=1, @@ -83,16 +90,14 @@ async def get_network_data() -> Dict[str, Dict]: } 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]] -]: +) -> Tuple[List[Union[NetworkInterface, WiFiDevice]], List[Union[ + NetworkInterface, WiFiDevice]]]: """ Разделяет интерфейсы на активные и неактивные """ @@ -107,13 +112,16 @@ def _split_interfaces( @router.post("/update_container_position") -async def update_container_position(data: Dict[str, Union[int, bool]] = Body(...)): +async def update_container_position(data: Dict[str, Union[int, + bool]] = Body(...)): """ Обновляет позицию и состояние контейнера """ try: - required_keys = ["container_id", "container_x", "container_y", "isOpen"] + 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" @@ -125,18 +133,18 @@ 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 - ) - return {"success": True, "message": "Позиция и состояние контейнера обновлены"} + 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( @@ -152,4 +160,5 @@ async def get_hostname(): return hostname_info except Exception as e: logger.error(f"Ошибка в get_hostname: {e}") - raise HTTPException(status_code=500, detail="Не удалось получить имя хоста") + raise HTTPException(status_code=500, + detail="Не удалось получить имя хоста") diff --git a/SRV/api/v1/subnet.py b/SRV/api/v1/subnet.py index a226e1c..bad8885 100644 --- a/SRV/api/v1/subnet.py +++ b/SRV/api/v1/subnet.py @@ -17,7 +17,8 @@ def scan_subnet(subnet: str = "192.168.1.0/24", ports: List[int] = None): result = subnet_service.scan_subnet(subnet, ports) return result except Exception as e: - raise HTTPException(status_code=500, detail=f"Ошибка сканирования: {str(e)}") + raise HTTPException(status_code=500, + detail=f"Ошибка сканирования: {str(e)}") @router.get("/wifi", response_model=WifiScanResult) @@ -28,9 +29,8 @@ 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") @@ -44,13 +44,15 @@ def add_note(mac: str, request: NoteRequest): raise HTTPException(status_code=404, detail=error_message) elif "уже существует" in error_message: raise HTTPException(status_code=400, detail=error_message) - return {"message": f"Заметка '{request.note}' добавлена для устройства {mac}"} + return { + "message": + f"Заметка '{request.note}' добавлена для устройства {mac}" + } 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") @@ -64,13 +66,15 @@ def update_note(mac: str, request: NoteRequest): raise HTTPException(status_code=404, detail=error_message) elif "не существует" in error_message: raise HTTPException(status_code=400, detail=error_message) - return {"message": f"Заметка '{request.note}' обновлена для устройства {mac}"} + return { + "message": + f"Заметка '{request.note}' обновлена для устройства {mac}" + } 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") @@ -88,6 +92,5 @@ def delete_note(mac: 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)}") diff --git a/SRV/api/v1/ufw.py b/SRV/api/v1/ufw.py index c2cac70..067ca9e 100644 --- a/SRV/api/v1/ufw.py +++ b/SRV/api/v1/ufw.py @@ -9,7 +9,8 @@ router = APIRouter() class RuleNumbers(BaseModel): - numbers: List[int] = Field(..., description="Список номеров правил для удаления") + numbers: List[int] = Field( + ..., description="Список номеров правил для удаления") @router.get("/rules", response_model=UFWRulesResponse) @@ -47,7 +48,8 @@ async def enable_ufw(): if success: 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)}") @@ -63,7 +65,8 @@ async def disable_ufw(): if success: 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)}") @@ -85,11 +88,13 @@ async def add_ufw_in_rules(rules: List[UFWRule]): result = ufw_service.add_rules(rules) if result["success"]: - successful_rules = len([r for r in result["results"] if r["success"]]) + successful_rules = len( + [r for r in result["results"] if r["success"]]) total_rules = len(result["results"]) return { "status": "success", - "message": f"Добавлено {successful_rules} из {total_rules} правил", + "message": + f"Добавлено {successful_rules} из {total_rules} правил", "details": result, } else: @@ -124,8 +129,8 @@ async def add_ufw_out_rules(rules: List[UFWRule]): } else: raise HTTPException( - status_code=500, detail="Не удалось добавить исходящие правила UFW" - ) + status_code=500, + detail="Не удалось добавить исходящие правила UFW") except Exception as e: raise HTTPException(status_code=500, detail=f"Error: {str(e)}") @@ -141,7 +146,8 @@ async def delete_ufw_rules(rule_numbers: RuleNumbers): if result["success"]: return { "status": "success", - "message": f"Удалено {result['successful']} из {result['total']} правил", + "message": + f"Удалено {result['successful']} из {result['total']} правил", "details": result, } else: diff --git a/SRV/api/v1/websocket.py b/SRV/api/v1/websocket.py index 58d2c5b..8001103 100644 --- a/SRV/api/v1/websocket.py +++ b/SRV/api/v1/websocket.py @@ -56,9 +56,8 @@ async def websocket_endpoint(websocket: WebSocket): 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() diff --git a/SRV/models/firewalld.py b/SRV/models/firewalld.py index b192c47..4afad5c 100644 --- a/SRV/models/firewalld.py +++ b/SRV/models/firewalld.py @@ -25,42 +25,35 @@ class FirewallZone(BaseModel): """Represents a firewalld zone configuration""" name: Annotated[str, StringConstraints(min_length=1)] = Field( - ..., description="Name of the zone" - ) - target: Optional[FirewallTarget] = Field( - default=FirewallTarget.DEFAULT, description="Zone target type" - ) - services: List[str] = Field( - default_factory=list, description="List of enabled services" - ) - ports: List[str] = Field(default_factory=list, description="List of enabled ports") - sources: List[str] = Field( - default_factory=list, description="List of source addresses" - ) - interfaces: List[str] = Field( - default_factory=list, description="List of network interfaces" - ) + ..., description="Name of the zone") + target: Optional[FirewallTarget] = Field(default=FirewallTarget.DEFAULT, + description="Zone target type") + services: List[str] = Field(default_factory=list, + description="List of enabled services") + ports: List[str] = Field(default_factory=list, + description="List of enabled ports") + sources: List[str] = Field(default_factory=list, + description="List of source addresses") + interfaces: List[str] = Field(default_factory=list, + description="List of network interfaces") class FirewallService(BaseModel): """Represents a firewalld service configuration""" name: Annotated[str, StringConstraints(min_length=1)] = Field( - ..., description="Name of the service" - ) - ports: List[str] = Field( - default_factory=list, description="List of ports used by the service" - ) + ..., description="Name of the service") + ports: List[str] = Field(default_factory=list, + description="List of ports used by the service") protocols: List[str] = Field( - default_factory=list, description="List of protocols used by the service" - ) + default_factory=list, + description="List of protocols used by the service") modules: List[str] = Field( default_factory=list, description="List of kernel modules required by the service", ) timeout: Optional[int] = Field( - default=0, ge=0, description="Timeout in seconds (0 for permanent)" - ) + default=0, ge=0, description="Timeout in seconds (0 for permanent)") class FirewallPort(BaseModel): @@ -69,8 +62,7 @@ class FirewallPort(BaseModel): port: int = Field(..., ge=1, le=65535, description="Port number (1-65535)") protocol: Protocol = Field(..., description="Network protocol (tcp/udp)") timeout: Optional[int] = Field( - default=0, ge=0, description="Timeout in seconds (0 for permanent)" - ) + default=0, ge=0, description="Timeout in seconds (0 for permanent)") def __str__(self) -> str: return f"{self.port}/{self.protocol.value}" @@ -81,9 +73,11 @@ class FirewallSource(BaseModel): address: Annotated[ str, - StringConstraints(pattern=r"^(?:[0-9]{1,3}\.){3}[0-9]{1,3}(?:/[0-9]{1,2})?$"), + StringConstraints( + pattern=r"^(?:[0-9]{1,3}\.){3}[0-9]{1,3}(?:/[0-9]{1,2})?$"), ] = Field(..., description="IP address or network in CIDR notation") - zone: Optional[str] = Field(default=None, description="Zone name for the source") + zone: Optional[str] = Field(default=None, + description="Zone name for the source") @field_validator("address") def validate_ip_address(cls, v): @@ -106,10 +100,10 @@ class FirewallRule(BaseModel): zone: str = Field(..., description="Zone name") source: Optional[FirewallSource] = Field( - default=None, description="Source configuration" - ) - port: Optional[FirewallPort] = Field(default=None, description="Port configuration") + default=None, description="Source configuration") + port: Optional[FirewallPort] = Field(default=None, + description="Port configuration") service: Optional[str] = Field(default=None, description="Service name") - timeout: Optional[int] = Field( - default=0, ge=0, description="Rule timeout in seconds" - ) + timeout: Optional[int] = Field(default=0, + ge=0, + description="Rule timeout in seconds") diff --git a/SRV/models/ufw.py b/SRV/models/ufw.py index c0fec4e..bf6b667 100644 --- a/SRV/models/ufw.py +++ b/SRV/models/ufw.py @@ -74,7 +74,8 @@ class UFWRule(BaseModel): """ if isinstance(v, int): if not (1 <= v <= 65535): - raise ValueError(f"Порт должен быть в диапазоне 1-65535, получено: {v}") + raise ValueError( + f"Порт должен быть в диапазоне 1-65535, получено: {v}") return v elif isinstance(v, str): try: @@ -112,8 +113,7 @@ class UFWRule(BaseModel): to_ip = ipaddress.ip_network(v, strict=False) if isinstance(from_ip, ipaddress.IPv6Network) != isinstance( - to_ip, ipaddress.IPv6Network - ): + to_ip, ipaddress.IPv6Network): raise ValueError( "Несовместимые версии IP: нельзя смешивать IPv4 и IPv6 в одном правиле" ) @@ -151,7 +151,8 @@ class UFWRule(BaseModel): cmd_parts.extend(["to", self.to_ip]) # Добавляем порт и протокол - port_str = str(self.port) # Работает как с числами, так и с диапазонами + port_str = str( + self.port) # Работает как с числами, так и с диапазонами cmd_parts.extend(["port", port_str]) # Добавляем протокол @@ -195,25 +196,20 @@ class UFWRule(BaseModel): match = re.match(pattern, line, re.VERBOSE | re.IGNORECASE) if not match: - raise ValueError(f"Невозможно разобрать строку правила UFW: {line}") + 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 - ) + 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 = (Action.ALLOW if action_str == "allow" else + Action.DENY if action_str == "deny" else Action.REJECT) # Преобразуем направление в enum direction_str = groups["direction"].lower() @@ -237,7 +233,8 @@ class UFWRule(BaseModel): 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, @@ -261,7 +258,8 @@ class UFWRulesResponse(BaseModel): """ enabled: bool = Field(..., description="Включен ли UFW") - rules: List[UFWRule] = Field(default_factory=list, description="Список правил UFW") + rules: List[UFWRule] = Field(default_factory=list, + description="Список правил UFW") @property def total_rules(self) -> int: diff --git a/SRV/services/arp_service.py b/SRV/services/arp_service.py index df7d115..97ef2eb 100644 --- a/SRV/services/arp_service.py +++ b/SRV/services/arp_service.py @@ -10,15 +10,14 @@ class NetworkScanner: def __init__(self, interface=None): """Инициализация параметров интерфейсов""" self.interface = ( - interface or netifaces.gateways()["default"][netifaces.AF_INET][1] - ) - self.ip_info = netifaces.ifaddresses(self.interface).get(netifaces.AF_INET, []) + 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.local_mac = netifaces.ifaddresses( + self.interface)[netifaces.AF_LINK][0]["addr"] def _create_arp_packet(self, target_ip): """Создание пакета ARP для отправки""" @@ -36,9 +35,10 @@ 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} if client not in clients: # Избегаем дублирования клиентов diff --git a/SRV/services/auth_service.py b/SRV/services/auth_service.py index dc39d25..8ed1435 100644 --- a/SRV/services/auth_service.py +++ b/SRV/services/auth_service.py @@ -37,7 +37,8 @@ class AuthService: def create_access_token(self, data: dict) -> str: """Создаёт токен""" to_encode = data.copy() - expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) + expire = datetime.utcnow() + timedelta( + minutes=ACCESS_TOKEN_EXPIRE_MINUTES) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) return encoded_jwt @@ -75,22 +76,28 @@ class AuthService: db_user = self.storage.get_user_by_username(user.username) 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"] - ): - logger.error(f"Неверный пароль для пользователя {user.username}") - raise HTTPException(status_code=401, detail="Неверный логин или пароль") - access_token = self.create_access_token(data={"sub": user.username}) + raise HTTPException(status_code=401, + detail="Неверный логин или пароль") + 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 - ) + 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( + "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"], @@ -102,7 +109,8 @@ class AuthService: ) raise HTTPException(status_code=500, detail="Ошибка авторизации") - def get_current_user(self, token: str = Depends(oauth2_scheme)) -> UserResponse: + def get_current_user( + self, token: str = Depends(oauth2_scheme)) -> UserResponse: """Получает текущего пользователя по токену""" credentials = self.verify_token(token) if not credentials: @@ -112,30 +120,36 @@ class AuthService: db_user = self.storage.get_user_by_username(username) 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"] - ) + raise HTTPException(status_code=401, + detail="Пользователь не найден") + 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""" session = self.session_storage.get_latest_session(client_id) if not session: - logger.error(f"Сессия для client_id {client_id} не найдена или истекла") - raise HTTPException(status_code=401, detail="Сессия недействительна") + logger.error( + f"Сессия для client_id {client_id} не найдена или истекла") + raise HTTPException(status_code=401, + detail="Сессия недействительна") username = session["username"] token = session["token"] db_user = self.storage.get_user_by_username(username) if not db_user: logger.error(f"Пользователь {username} не найден") - raise HTTPException(status_code=401, detail="Пользователь не найден") + raise HTTPException(status_code=401, + detail="Пользователь не найден") return { - "user": UserResponse( + "user": + UserResponse( id=db_user["id"], username=db_user["username"], is_admin=db_user["is_admin"], ), - "access_token": token, + "access_token": + token, } def logout(self, client_id: str): @@ -149,5 +163,7 @@ class AuthService: ) return {"message": "Успешный выход из системы"} except Exception as e: - logger.error(f"Ошибка при выходе из системы: {type(e).__name__} - {str(e)}") - raise HTTPException(status_code=500, detail="Ошибка при выходе из системы") + logger.error( + f"Ошибка при выходе из системы: {type(e).__name__} - {str(e)}") + raise HTTPException(status_code=500, + detail="Ошибка при выходе из системы") diff --git a/SRV/services/firewalld_service.py b/SRV/services/firewalld_service.py index bb5e859..ef0c666 100644 --- a/SRV/services/firewalld_service.py +++ b/SRV/services/firewalld_service.py @@ -7,11 +7,11 @@ DEBUG = True class FirewalldService: + def __init__(self): self.bus = SystemBus() - self.proxy = self.bus.get( - "org.fedoraproject.FirewallD1", "/org/fedoraproject/FirewallD1" - ) + self.proxy = self.bus.get("org.fedoraproject.FirewallD1", + "/org/fedoraproject/FirewallD1") def get_zones(self) -> List[str]: """Get list of all zones.""" @@ -34,12 +34,16 @@ class FirewalldService: except Exception as e: raise Exception(f"Failed to set default zone to {zone}: {str(e)}") - def add_service(self, zone: str, service: str, timeout: Optional[int] = 0) -> None: + def add_service(self, + zone: str, + service: str, + timeout: Optional[int] = 0) -> None: """Add a service to a zone.""" try: self.proxy.addService(zone, service, timeout) except Exception as e: - raise Exception(f"Failed to add service {service} to zone {zone}: {str(e)}") + raise Exception( + f"Failed to add service {service} to zone {zone}: {str(e)}") def remove_service(self, zone: str, service: str) -> None: """Remove a service from a zone.""" @@ -55,20 +59,22 @@ class FirewalldService: try: return self.proxy.getServices(zone) except Exception as e: - raise Exception(f"Failed to get services for zone {zone}: {str(e)}") + raise Exception( + f"Failed to get services for zone {zone}: {str(e)}") - def add_port( - self, zone: str, port: int, protocol: str, timeout: Optional[int] = 0 - ) -> None: + def add_port(self, + zone: str, + port: int, + protocol: str, + timeout: Optional[int] = 0) -> None: """Add a port to a zone.""" try: if DEBUG: print(f"Adding port {port}/{protocol} to zone {zone}") print(f"Timeout: {timeout}") # Get the zone interface - zone_proxy = self.bus.get( - "org.fedoraproject.FirewallD1", "/org/fedoraproject/FirewallD1" - ) + zone_proxy = self.bus.get("org.fedoraproject.FirewallD1", + "/org/fedoraproject/FirewallD1") zone_iface = zone_proxy["org.fedoraproject.FirewallD1.zone"] # Convert port to string port_str = str(port) @@ -105,7 +111,8 @@ class FirewalldService: try: self.proxy.addSource(zone, source) except Exception as e: - raise Exception(f"Failed to add source {source} to zone {zone}: {str(e)}") + raise Exception( + f"Failed to add source {source} to zone {zone}: {str(e)}") def remove_source(self, zone: str, source: str) -> None: """Remove a source from a zone.""" @@ -113,8 +120,7 @@ class FirewalldService: self.proxy.removeSource(zone, source) except Exception as e: raise Exception( - f"Failed to remove source {source} from zone {zone}: {str(e)}" - ) + f"Failed to remove source {source} from zone {zone}: {str(e)}") def get_sources(self, zone: str) -> List[str]: """Get list of sources in a zone.""" @@ -139,4 +145,5 @@ class FirewalldService: "sources": self.get_sources(zone), } except Exception as e: - raise Exception(f"Failed to get zone settings for {zone}: {str(e)}") + raise Exception( + f"Failed to get zone settings for {zone}: {str(e)}") diff --git a/SRV/services/ip_service.py b/SRV/services/ip_service.py index 89922a9..99e3bea 100644 --- a/SRV/services/ip_service.py +++ b/SRV/services/ip_service.py @@ -35,7 +35,10 @@ class IPService: 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): - return {"available": False, "message": f"IP address {new_ip} уже занят."} + return { + "available": False, + "message": f"IP address {new_ip} уже занят." + } return { "available": True, "message": "Network profile settings are available for use.", @@ -46,7 +49,8 @@ class IPService: self._reset_bus() nm = NetworkManager() settings = NetworkManagerSettings() - connection_paths = await settings.get_connections_by_id(args["connection_id"]) + connection_paths = await settings.get_connections_by_id( + args["connection_id"]) if not connection_paths: raise ValueError(f"Подключение {args['connection_id']} не найдено") @@ -73,10 +77,10 @@ class IPService: 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 - raise RuntimeError(f"Устройство для интерфейса {interface_name} не найдено") + raise RuntimeError( + f"Устройство для интерфейса {interface_name} не найдено") diff --git a/SRV/services/network_service.py b/SRV/services/network_service.py index 7c77b8b..3d60ff9 100644 --- a/SRV/services/network_service.py +++ b/SRV/services/network_service.py @@ -30,7 +30,8 @@ class NetworkService: self.nm = NetworkManager(self.system_bus) self.nm_settings = NetworkManagerSettings(self.system_bus) - async def get_ip_method(self, device: NetworkDeviceGeneric, ip_type: str) -> str: + async def get_ip_method(self, device: NetworkDeviceGeneric, + ip_type: str) -> str: """Получает метод IP для устройства""" try: config = await device.get_applied_connection() @@ -52,14 +53,14 @@ class NetworkService: # добавить потом условия для отключенных интерфейсов 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) + 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( @@ -67,24 +68,27 @@ class NetworkService: routes=[ globals()[f"IPv{ip_version}Route"]( destination=route.get( - "dest", ("s", "0.0.0.0" if ip_version == "4" else "::") - )[1], + "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 "::"), + ("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 + ) 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", + "nameserver_data" if ip_version == + "4" else "nameservers", ) ], ) @@ -106,19 +110,18 @@ class NetworkService: } 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 + self, + initial_positions: Dict[str, Tuple[int, int]] = None ) -> List[Union[NetworkInterface, WiFiDevice]]: """Получает список сетевых интерфейсов""" devices_paths = await self.nm.get_devices() @@ -138,26 +141,36 @@ class NetworkService: 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": - wifi_device = NetworkDeviceWireless(device_path, self.system_bus) + wifi_device = NetworkDeviceWireless(device_path, + self.system_bus) wifi_info = await self.get_wifi_info(wifi_device) - network_interfaces.append(WiFiDevice(**common_data, **wifi_info)) + network_interfaces.append( + WiFiDevice(**common_data, **wifi_info)) else: network_interfaces.append(NetworkInterface(**common_data)) @@ -166,9 +179,10 @@ 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: @@ -177,14 +191,17 @@ class NetworkService: 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 + 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 + interface for interface in all_interfaces + if interface.device_state == state ] async def get_interfaces_container( @@ -196,12 +213,13 @@ class NetworkService: ) -> NetworkInterfacesContainer: """Получает объект, содержащий список интерфейсов и координаты контейнера""" if state: - interfaces = await self.get_interfaces_by_state(state, initial_positions) + 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 - ) + return NetworkInterfacesContainer(interfaces=interfaces, + container_x=container_x, + container_y=container_y) async def get_network_interfaces_container( self, @@ -209,23 +227,26 @@ class NetworkService: ) -> NetworkInterfacesContainer: """Получает контейнер с сетевыми интерфейсами и координатами контейнера""" interfaces = await self.get_network_interfaces(initial_positions) - return NetworkInterfacesContainer( - interfaces=interfaces, container_x=0, container_y=0 - ) + return NetworkInterfacesContainer(interfaces=interfaces, + container_x=0, + container_y=0) async def get_activated_interfaces( - self, initial_positions: Dict[str, Tuple[int, int]] = None + 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 + 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" + interface for interface in all_interfaces + if interface.device_state != "100" ] async def get_activated_interfaces_container( @@ -235,9 +256,8 @@ class NetworkService: container_y: int = 0, ) -> NetworkInterfacesContainer: """Получает объект, содержащий список активных интерфейсов и координаты контейнера""" - return await self.get_interfaces_container( - "100", initial_positions, container_x, container_y - ) + return await self.get_interfaces_container("100", initial_positions, + container_x, container_y) async def get_deactivated_interfaces_container( self, @@ -247,6 +267,6 @@ class NetworkService: ) -> NetworkInterfacesContainer: """Получает объект, содержащий список неактивных интерфейсов и координаты контейнера""" interfaces = await self.get_deactivated_interfaces(initial_positions) - return NetworkInterfacesContainer( - interfaces=interfaces, container_x=container_x, container_y=container_y - ) + return NetworkInterfacesContainer(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 954c6bd..b2a008a 100644 --- a/SRV/services/subnet_service.py +++ b/SRV/services/subnet_service.py @@ -20,12 +20,13 @@ 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" -) + 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 conf.iface = interface @@ -35,14 +36,19 @@ class NetworkScanner: arp = ARP(pdst=subnet) ether = Ether(dst="ff:ff:ff:ff:ff:ff") packet = ether / arp - result = srp(packet, timeout=2, verbose=False, iface=self.interface)[0] - return [(received.psrc, received.hwsrc) for sent, received in result] + result = srp(packet, + timeout=2, + verbose=False, + iface=self.interface)[0] + return [(received.psrc, received.hwsrc) + for sent, received in result] except Exception as e: logger.error(f"Ошибка ARP-сканирования подсети {subnet}: {str(e)}") return [] class PortScanner: + def __init__(self, interface: str): self.interface = interface @@ -61,7 +67,8 @@ class PortScanner: return port_config.get_port_name(port) return None except Exception as e: - logger.debug(f"Ошибка проверки порта {port} для IP {ip}: {str(e)}") + logger.debug( + f"Ошибка проверки порта {port} для IP {ip}: {str(e)}") return None try: @@ -72,11 +79,13 @@ class PortScanner: logger.debug(f"Открытые порты для IP {ip}: {open_ports}") return open_ports except Exception as e: - logger.warning(f"Ошибка при сканировании портов для IP {ip}: {str(e)}") + logger.warning( + f"Ошибка при сканировании портов для IP {ip}: {str(e)}") return [] class DeviceInfoCollector: + def __init__(self): self.mac_lookup = MacLookup() @@ -86,8 +95,7 @@ class DeviceInfoCollector: return self.mac_lookup.lookup(mac) except Exception as e: logger.warning( - f"Не удалось определить производителя для MAC {mac}: {str(e)}" - ) + f"Не удалось определить производителя для MAC {mac}: {str(e)}") return "Unknown" def get_hostname(self, ip: str) -> Optional[str]: @@ -98,6 +106,7 @@ class DeviceInfoCollector: class DeviceStorage: + def __init__(self, storage_path: str = "storage/devices/devices.json"): self.storage_file = Path(storage_path) self.storage_file.parent.mkdir(parents=True, exist_ok=True) @@ -119,7 +128,8 @@ class DeviceStorage: logger.error(f"Ошибка при загрузке устройств: {str(e)}") return {} - def _clean_old_devices(self, devices: Dict[str, Device]) -> Dict[str, Device]: + def _clean_old_devices(self, devices: Dict[str, + Device]) -> Dict[str, Device]: now = datetime.now(ZoneInfo("UTC")) filtered = {} for mac, device in devices.items(): @@ -129,9 +139,11 @@ class DeviceStorage: if now - last_seen <= self.offline_timeout: filtered[mac] = device else: - logger.info(f"Удалено устройство {mac} (offline > 7 дней)") + logger.info( + f"Удалено устройство {mac} (offline > 7 дней)") except ValueError: - logger.warning(f"Неверный формат last_seen для устройства {mac}") + logger.warning( + f"Неверный формат last_seen для устройства {mac}") else: filtered[mac] = device return filtered @@ -141,12 +153,14 @@ class DeviceStorage: data = SubnetScanResult(devices=devices).model_dump() with open(self.storage_file, "w", encoding="utf-8") as f: json.dump(data, f, indent=2, ensure_ascii=False) - logger.info(f"Сохранено {len(devices)} устройств в {self.storage_file}") + logger.info( + f"Сохранено {len(devices)} устройств в {self.storage_file}") except Exception as e: logger.error(f"Ошибка при сохранении устройств: {str(e)}") class SubnetService: + def __init__(self): self.interface = self._get_default_interface() self.network_scanner = NetworkScanner(self.interface) @@ -157,7 +171,8 @@ class SubnetService: def _get_default_interface(self) -> str: try: gateways = netifaces.gateways() - default_gateway = gateways.get("default", {}).get(netifaces.AF_INET) + default_gateway = gateways.get("default", + {}).get(netifaces.AF_INET) if default_gateway: return default_gateway[1] interfaces = netifaces.interfaces() @@ -169,7 +184,9 @@ class SubnetService: logger.warning(f"Не удалось определить интерфейс: {str(e)}") return "eth0" - def scan_subnet(self, subnet: str, ports: List[int] = None) -> SubnetScanResult: + def scan_subnet(self, + subnet: str, + ports: List[int] = None) -> SubnetScanResult: start_time = time.time() devices = [] now = datetime.now(ZoneInfo("UTC")).isoformat() @@ -220,7 +237,8 @@ class SubnetService: ) with ThreadPoolExecutor(max_workers=10) as executor: - device_results = executor.map(process_device, discovered_devices) + device_results = executor.map(process_device, + discovered_devices) devices.extend([d for d in device_results if d is not None]) active_macs = {d.mac for d in devices if d.mac} @@ -275,10 +293,12 @@ class SubnetService: last_seen=now, note=None, ) - logger.warning(f"Локальное устройство не найдено в подсети {subnet}") + logger.warning( + f"Локальное устройство не найдено в подсети {subnet}") return None except Exception as e: - logger.warning(f"Не удалось получить локальное устройство: {str(e)}") + logger.warning( + f"Не удалось получить локальное устройство: {str(e)}") return None def add_note(self, mac: str, note: str) -> Tuple[bool, Optional[str]]: @@ -302,14 +322,17 @@ class SubnetService: logger.warning(f"Устройство с MAC {mac} не найдено") return False, f"Устройство с MAC {mac} не найдено" if saved_devices[mac].note is not None: - logger.error(f"Заметка для устройства с MAC {mac} уже существует") + logger.error( + f"Заметка для устройства с MAC {mac} уже существует") return False, f"Заметка для устройства с MAC {mac} уже существует" saved_devices[mac].note = note self.storage.save_devices(list(saved_devices.values())) - logger.info(f"Добавлена заметка '{note}' для устройства с MAC {mac}") + logger.info( + f"Добавлена заметка '{note}' для устройства с MAC {mac}") return True, None except Exception as e: - logger.error(f"Ошибка при добавлении заметки для MAC {mac}: {str(e)}") + logger.error( + f"Ошибка при добавлении заметки для MAC {mac}: {str(e)}") return False, f"Внутренняя ошибка сервера: {str(e)}" def update_note(self, mac: str, note: str) -> Tuple[bool, Optional[str]]: @@ -333,14 +356,17 @@ class SubnetService: logger.warning(f"Устройство с MAC {mac} не найдено") return False, f"Устройство с MAC {mac} не найдено" if saved_devices[mac].note is None: - logger.error(f"Заметка для устройства с MAC {mac} не существует") + logger.error( + f"Заметка для устройства с MAC {mac} не существует") return False, f"Заметка для устройства с MAC {mac} не существует" saved_devices[mac].note = note self.storage.save_devices(list(saved_devices.values())) - logger.info(f"Обновлена заметка '{note}' для устройства с MAC {mac}") + logger.info( + f"Обновлена заметка '{note}' для устройства с MAC {mac}") return True, None except Exception as e: - logger.error(f"Ошибка при обновлении заметки для MAC {mac}: {str(e)}") + logger.error( + f"Ошибка при обновлении заметки для MAC {mac}: {str(e)}") return False, f"Внутренняя ошибка сервера: {str(e)}" def delete_note(self, mac: str) -> Tuple[bool, Optional[str]]: @@ -363,12 +389,14 @@ class SubnetService: logger.warning(f"Устройство с MAC {mac} не найдено") return False, f"Устройство с MAC {mac} не найдено" if saved_devices[mac].note is None: - logger.error(f"Заметка для устройства с MAC {mac} не существует") + logger.error( + f"Заметка для устройства с MAC {mac} не существует") return False, f"Заметка для устройства с MAC {mac} не существует" saved_devices[mac].note = None self.storage.save_devices(list(saved_devices.values())) logger.info(f"Удалена заметка для устройства с MAC {mac}") return True, None except Exception as e: - logger.error(f"Ошибка при удалении заметки для MAC {mac}: {str(e)}") + logger.error( + f"Ошибка при удалении заметки для MAC {mac}: {str(e)}") return False, f"Внутренняя ошибка сервера: {str(e)}" diff --git a/SRV/services/ufw_service.py b/SRV/services/ufw_service.py index 1e7b936..f851ab3 100644 --- a/SRV/services/ufw_service.py +++ b/SRV/services/ufw_service.py @@ -41,9 +41,9 @@ class UFWService: print(f"Command error: {result.stderr}") if result.returncode != 0: - raise subprocess.CalledProcessError( - result.returncode, command, result.stdout, result.stderr - ) + raise subprocess.CalledProcessError(result.returncode, command, + result.stdout, + result.stderr) return result.stdout @@ -116,9 +116,8 @@ class UFWService: """ try: status_result = self._run_command("ufw status") - first_line = ( - status_result.splitlines()[0] if status_result.splitlines() else "" - ) + first_line = (status_result.splitlines()[0] + if status_result.splitlines() else "") is_enabled = "Status: active" in first_line message = "UFW активен" if is_enabled else "UFW неактивен" @@ -166,7 +165,9 @@ class UFWService: print(f"Ошибка при выключении UFW: {e}") return False - def add_rules_base(self, rules: List[UFWRule], direction: str = "") -> bool: + def add_rules_base(self, + rules: List[UFWRule], + direction: str = "") -> bool: """ Базовый метод для добавления правил UFW. @@ -296,7 +297,10 @@ class UFWService: except Exception as e: error_msg = f"Ошибка выполнения команды: {str(e)}" print(error_msg) - results.append({"success": False, "message": error_msg}) + results.append({ + "success": False, + "message": error_msg + }) except Exception as e: error_msg = f"Ошибка при добавлении правила: {str(e)}" @@ -334,7 +338,7 @@ class UFWService: try: # Извлекаем номер правила из строки [N] - rule_num = int(line[1 : line.index("]")]) + rule_num = int(line[1:line.index("]")]) rules.append(rule_num) except (ValueError, IndexError): continue @@ -349,32 +353,37 @@ class UFWService: try: # Проверяем существует ли правило с таким номером if rule_number not in rules: - results.append( - { - "number": rule_number, - "success": False, - "error": f"Правило с номером {rule_number} не найдено", - } - ) + results.append({ + "number": + rule_number, + "success": + False, + "error": + f"Правило с номером {rule_number} не найдено", + }) continue # Удаляем правило self._run_command(f"ufw --force delete {rule_number}") - results.append( - { - "number": rule_number, - "success": True, - "message": f"Правило {rule_number} успешно удалено", - } - ) + results.append({ + "number": + rule_number, + "success": + True, + "message": + f"Правило {rule_number} успешно удалено", + }) except Exception as e: if DEBUG: - print(f"Ошибка при удалении правила {rule_number}: {e}") - results.append( - {"number": rule_number, "success": False, "error": str(e)} - ) + print( + f"Ошибка при удалении правила {rule_number}: {e}") + results.append({ + "number": rule_number, + "success": False, + "error": str(e) + }) # Подсчитываем статистику total = len(rule_numbers) diff --git a/SRV/services/wifi_service.py b/SRV/services/wifi_service.py index e14c254..3e41ae0 100644 --- a/SRV/services/wifi_service.py +++ b/SRV/services/wifi_service.py @@ -10,8 +10,8 @@ from pywifi import Profile from models.wifi import WifiAccessPoint, WifiScanResult logging.basicConfig( - level=logging.DEBUG, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" -) + level=logging.DEBUG, + format="%(asctime)s - %(name)s - %(levelname)s - %(message)s") logger = logging.getLogger(__name__) @@ -38,7 +38,7 @@ class WifiService: for iface in interfaces: iface_name = iface.name() if ( - "p2p" in iface_name.lower() + "p2p" in iface_name.lower() ): # Исключаю сетевые интерфейсы типа p2p из выбора, возможно сделать решение получше нежели исключать интерфейсы logger.debug(f"Пропущен интерфейс {iface_name} (P2P)") continue @@ -67,14 +67,14 @@ class WifiService: return self.mac_lookup.lookup(mac) except Exception as e: logger.warning( - f"Не удалось определить производителя для MAC {mac}: {str(e)}" - ) + f"Не удалось определить производителя для MAC {mac}: {str(e)}") return "Unknown" def _get_encryption_from_iw(self, ap_info: Dict) -> str: """Определяет шифрование из данных iw""" security = ap_info.get("security", "") - logger.debug(f"Шифрование для BSSID {ap_info.get('bssid')}: {security}") + logger.debug( + f"Шифрование для BSSID {ap_info.get('bssid')}: {security}") if "WPA3" in security: return "WPA3" elif "WPA2" in security or "RSN" in security: @@ -127,12 +127,16 @@ class WifiService: return [] cmd = ["sudo", "iw", "dev", self.interface.name(), "scan"] logger.debug(f"Выполняем команду: {' '.join(cmd)}") - result = subprocess.run(cmd, capture_output=True, text=True, timeout=10) + result = subprocess.run(cmd, + capture_output=True, + text=True, + timeout=10) if result.returncode != 0: logger.error(f"Ошибка iw scan: {result.stderr}") return [] output = result.stdout - logger.debug(f"Вывод iw scan (первые 500 символов): {output[:500]}...") + logger.debug( + f"Вывод iw scan (первые 500 символов): {output[:500]}...") aps = [] current_ap = None @@ -149,12 +153,15 @@ class WifiService: if bssid_match := bssid_pattern.match(line): if current_ap: aps.append(current_ap) - current_ap = {"bssid": bssid_match.group(1).lower(), "security": ""} + 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): @@ -163,8 +170,10 @@ class WifiService: current_ap["security"] += "RSN;" elif wpa_pattern.match(line): current_ap["security"] += "WPA;" - elif security_type_match := security_type_pattern.search(line): - current_ap["security"] += security_type_match.group(1) + ";" + elif security_type_match := security_type_pattern.search( + line): + current_ap["security"] += security_type_match.group( + 1) + ";" if current_ap: aps.append(current_ap) @@ -206,7 +215,10 @@ class WifiService: 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] = ( {} @@ -243,18 +255,15 @@ class WifiService: ) 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 - ): + 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()) logger.info( f"Сканирование Wi-Fi завершено за {time.time() - start_time:.2f} секунд, " - f"найдено {len(aps)} уникальных сетей" - ) + f"найдено {len(aps)} уникальных сетей") return WifiScanResult(access_points=aps) except Exception as e: diff --git a/SRV/storage/json_storage.py b/SRV/storage/json_storage.py index f86d630..8a13d89 100644 --- a/SRV/storage/json_storage.py +++ b/SRV/storage/json_storage.py @@ -23,7 +23,8 @@ class JsonStorage: try: logger.info(f"Проверка существования файла {self.file_path}") if not os.path.exists(self.file_path): - logger.info(f"Создание директории {os.path.dirname(self.file_path)}") + logger.info( + f"Создание директории {os.path.dirname(self.file_path)}") os.makedirs(os.path.dirname(self.file_path), exist_ok=True) logger.info(f"Создание файла {self.file_path}") with open(self.file_path, "w") as f: @@ -84,7 +85,8 @@ class JsonStorage: users.append(new_user) data["next_id"] = next_id + 1 self._write_data(data) - logger.info(f"Пользователь {user.username} успешно зарегистрирован") + logger.info( + f"Пользователь {user.username} успешно зарегистрирован") return UserResponse( id=new_user["id"], username=new_user["username"], @@ -115,6 +117,7 @@ class JsonStorage: ) raise - def verify_password(self, plain_password: str, hashed_password: str) -> bool: + def verify_password(self, plain_password: str, + hashed_password: str) -> bool: """Проверяет пароль""" return self.pwd_context.verify(plain_password, hashed_password) diff --git a/SRV/storage/session_storage.py b/SRV/storage/session_storage.py index 8854af1..83aee33 100644 --- a/SRV/storage/session_storage.py +++ b/SRV/storage/session_storage.py @@ -11,6 +11,7 @@ logger = logging.getLogger(__name__) class SessionStorage: + def __init__(self, file_path: str = "/etc/gonets/sessions.json"): self.file_path = file_path self._ensure_file_exists() @@ -19,7 +20,8 @@ class SessionStorage: try: logger.info(f"Проверка существования файла {self.file_path}") if not os.path.exists(self.file_path): - logger.info(f"Создание директории {os.path.dirname(self.file_path)}") + logger.info( + f"Создание директории {os.path.dirname(self.file_path)}") os.makedirs(os.path.dirname(self.file_path), exist_ok=True) logger.info(f"Создание файла {self.file_path}") with open(self.file_path, "w") as f: @@ -142,7 +144,8 @@ class SessionStorage: # Удаляем истекшие сессии for session_id in expired_sessions: del data["sessions"][session_id] - logger.info(f"Удалена истекшая сессия с session_id {session_id}") + logger.info( + f"Удалена истекшая сессия с session_id {session_id}") # Записываем обновлённые данные if expired_sessions: