diff --git a/SRV/api/v1/subnet.py b/SRV/api/v1/subnet.py index 3c98846..1588537 100644 --- a/SRV/api/v1/subnet.py +++ b/SRV/api/v1/subnet.py @@ -7,18 +7,22 @@ from services.wifi_service import WifiService from services.connect_service import WifiConnectionService from utils.i18n import get_translations, _ from babel.support import Translations +import logging router = APIRouter() +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + + @router.post("/devices", response_model=SubnetScanResult) def scan_subnet(request: SubnetScanRequest = Body(...), translations: Translations = Depends(get_translations("api_v1"))): """Сканирует локальную подсеть и возвращает данные об устройствах. - Выполняет сканирование указанной подсети для обнаружения активных устройств - с возможностью проверки указанных портов. + Выполняет сканирование указанной подсети для обнаружения активных устройств. Args: - request (SubnetScanRequest): Объект с полями subnet и ports. + request (SubnetScanRequest): Объект с полем subnet. translations: Объект переводов для текущего языка. Returns: @@ -29,10 +33,15 @@ def scan_subnet(request: SubnetScanRequest = Body(...), translations: Translatio """ try: subnet_service = SubnetService() - subnet = request.subnet or "192.168.1.0/24" + # Если subnet пустая строка, определяем автоматически + subnet = request.subnet.strip() if request.subnet else None + logger.info(f"Requested subnet scan for: {subnet or 'auto-detected'}") result = subnet_service.scan_subnet(subnet) + if not result.devices: + logger.warning(f"No devices found in subnet {subnet or 'auto-detected'}") return result except Exception as e: + logger.error(f"Error in scan_subnet: {str(e)}") raise HTTPException(status_code=500, detail=translations.gettext("Ошибка сканирования: {error}").format(error=str(e))) @router.get("/wifi", response_model=WifiScanResult) diff --git a/SRV/models/subnet.py b/SRV/models/subnet.py index f46c9b3..8352e9c 100644 --- a/SRV/models/subnet.py +++ b/SRV/models/subnet.py @@ -37,11 +37,12 @@ class SubnetScanResult(BaseModel): devices: List[Device] = [] + class SubnetScanRequest(BaseModel): """ Параметры для сканирования подсети Атрибуты: - subnet (str): Подсеть для сканирования (например 192.168.1.0/24) + subnet (Optional[str]): Подсеть для сканирования (например 192.168.1.0/24), опционально """ - subnet: Optional[str] = "192.168.1.0/24" + subnet: Optional[str] = None \ No newline at end of file diff --git a/SRV/services/subnet_service.py b/SRV/services/subnet_service.py index 46ce179..0dd85e4 100644 --- a/SRV/services/subnet_service.py +++ b/SRV/services/subnet_service.py @@ -12,172 +12,69 @@ import json from pathlib import Path from datetime import datetime, timedelta import subprocess -from utils.i18n import get_translations, _ + +from scapy.layers.l2 import ARP, Ether +from scapy.sendrecv import srp +from utils.i18n import get_translations from babel.support import Translations from fastapi import Depends from models.subnet import SubnetScanResult, Device -from config.ports import port_config -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") + format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" +) logger = logging.getLogger(__name__) class NetworkScanner: - """Класс для сканирования подсети с использованием ARP-запросов. - - Выполняет обнаружение устройств в указанной подсети через ARP-пакеты. - """ - + """Класс для сканирования подсети с использованием ARP-запросов.""" def __init__(self, interface: str, translations: Translations = Depends(get_translations("services"))): - """Инициализирует сканер сети с указанным сетевым интерфейсом. - - Args: - interface (str): Название сетевого интерфейса (например, "eth0"). - """ self.interface = interface conf.iface = interface self.translations = translations def scan(self, subnet: str) -> List[Tuple[str, str]]: - """Сканирует подсеть с использованием ARP-запросов. - - Args: - subnet (str): Подсеть в формате CIDR (например, "192.168.1.0/24"). - - Returns: - List[Tuple[str, str]]: Список кортежей (IP-адрес, MAC-адрес) обнаруженных - устройств. - - Raises: - Exception: Если произошла ошибка при отправке ARP-пакетов. - """ try: 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"Error scanning subnet {subnet}: {str(e)}") return [] class PortScanner: - """Класс для сканирования открытых портов на устройствах. - - Выполняет сканирование портов с использованием TCP SYN-запросов. - """ - + """Класс для сканирования открытых портов на устройствах.""" def __init__(self, interface: str, translations: Translations = Depends(get_translations("services"))): - """Инициализирует сканер портов с указанным сетевым интерфейсом. - - Args: - interface (str): Название сетевого интерфейса (например, "eth0"). - """ self.interface = interface self.translations = translations def scan(self, ip: str) -> List[str]: - """Сканирует открытые порты на указанном IP-адресе. - - Args: - ip (str): IP-адрес устройства для сканирования. - - Returns: - List[str]: Список имён открытых портов (например, "HTTP", "SSH"). - - Raises: - Exception: Если произошла ошибка при сканировании портов. - """ - open_ports = [] - - def check_port(port: int) -> Optional[str]: - """Проверяет доступность конкретного порта. - - Args: - port (int): Номер порта для проверки. - - Returns: - Optional[str]: Имя порта, если открыт, иначе None. - """ - try: - pkt = sr1( - IP(dst=ip) / TCP(dport=port, flags="S"), - timeout=2, - verbose=False, - ) - if pkt and pkt.haslayer(TCP) and pkt[TCP].flags == 0x12: - return port_config.get_port_name(port) - return None - except Exception as e: - logger.debug( - f"Error checking port {port} for IP {ip}: {str(e)}") - return None - - try: - max_workers = min(len(port_config.get_ports()), 15) - with ThreadPoolExecutor(max_workers=max_workers) as executor: - results = executor.map(check_port, port_config.get_ports()) - open_ports = [port for port in results if port is not None] - logger.debug(f"Open ports for IP {ip}: {open_ports}") - return open_ports - except Exception as e: - logger.warning( - f"Error scanning ports for IP {ip}: {str(e)}") - return [] + return [] class DeviceInfoCollector: - """Класс для сбора информации об устройствах, включая вендоров и хостнеймы. - - Управляет кэшированием данных NetBIOS и DNS. - """ - + """Класс для сбора информации об устройствах, включая вендоров и хостнеймы.""" def __init__(self, translations: Translations = Depends(get_translations("services"))): - """Инициализирует коллектор информации об устройствах. - - Создаёт объект MacLookup для определения вендоров и инициализирует кэши. - """ self.mac_lookup = MacLookup() self.nbtscan_cache = {} self.nbtscan_results = {} - self.translations = translations + if not hasattr(translations, 'gettext'): + logger.warning("Translations object does not support gettext, using fallback") + self.translations = Translations.load("utils/locales/en/LC_MESSAGES/services.mo") + else: + self.translations = translations @lru_cache(maxsize=1000) def get_vendor(self, mac: str) -> str: - """Определяет вендора по MAC-адресу с использованием кэширования. - - Args: - mac (str): MAC-адрес устройства в формате XX:XX:XX:XX:XX:XX. - - Returns: - str: Название вендора или "Unknown", если вендор не определён. - - Raises: - Exception: Если произошла ошибка при запросе к базе вендоров. - """ try: return self.mac_lookup.lookup(mac) except Exception as e: - logger.warning( - f"Failed to determine vendor for MAC {mac}: {str(e)}") + logger.warning(f"Failed to determine vendor for MAC {mac}: {str(e)}") return self.translations.gettext("Неизвестный") def scan_nbtscan(self, subnet: str) -> None: - """Выполняет сканирование NetBIOS для указанной подсети. - - Args: - subnet (str): Подсеть в формате CIDR (например, "192.168.1.0/24"). - - Сохраняет результаты в кэш, игнорируя хостнеймы "". - """ if subnet not in self.nbtscan_results: try: logger.debug(f"Starting nbtscan for subnet {subnet}") @@ -204,16 +101,6 @@ class DeviceInfoCollector: logger.error(f"NetBIOS: Error for subnet {subnet}: {str(e)}") def get_hostname(self, ip: str) -> Optional[str]: - """Определяет хостнейм для указанного IP-адреса. - - Args: - ip (str): IP-адрес устройства. - - Returns: - Optional[str]: Хостнейм, если найден, иначе None. - - Сначала пытается использовать DNS, затем NetBIOS через nbtscan. - """ try: socket.setdefaulttimeout(2) return socket.gethostbyaddr(ip)[0] @@ -231,38 +118,25 @@ class DeviceInfoCollector: return None class DeviceStorage: - """Класс для хранения и управления данными об устройствах. - - Хранит устройства в JSON-файле и удаляет устаревшие записи. - """ - + """Класс для хранения и управления данными об устройствах.""" def __init__(self, storage_path: str = "storage/devices/devices.json", translations: Translations = Depends(get_translations("services"))): - """Инициализирует хранилище устройств с указанным путём. - - Args: - storage_path (str, optional): Путь к файлу хранения. - По умолчанию "storage/devices/devices.json". - - Создаёт родительские директории, если они не существуют. - """ self.storage_file = Path(storage_path) self.storage_file.parent.mkdir(parents=True, exist_ok=True) self.offline_timeout = timedelta(days=7) - self.translations = translations + if not hasattr(translations, 'gettext'): + logger.warning("Translations object does not support gettext, using fallback") + self.translations = Translations.load("utils/locales/en/LC_MESSAGES/services.mo") + else: + self.translations = translations def load_devices(self) -> Dict[str, Device]: - """Загружает устройства из файла хранения. - - Returns: - Dict[str, Device]: Словарь устройств, где ключ — MAC-адрес. - - Удаляет устройства с неверным форматом last_seen и возвращает пустой словарь - при ошибке. - """ try: if self.storage_file.exists(): with open(self.storage_file, "r", encoding="utf-8") as f: data = json.load(f) + if not isinstance(data, dict) or "devices" not in data: + logger.warning(f"Invalid JSON structure in {self.storage_file}, expected {'devices': [...]}, got {data}") + return {} devices = {} for device_data in data.get("devices", []): device = Device(**device_data) @@ -270,20 +144,14 @@ class DeviceStorage: devices[device.mac] = device return self._clean_old_devices(devices) return {} + except json.JSONDecodeError as e: + logger.error(f"JSON decode error in {self.storage_file}: {str(e)}") + return {} except Exception as e: logger.error(f"Error loading devices: {str(e)}") return {} def _clean_old_devices(self, devices: Dict[str, Device]) -> Dict[str, Device]: - """Очищает устаревшие устройства из словаря. - - Args: - devices (Dict[str, Device]): Словарь устройств с MAC-адресами как ключами. - - Returns: - Dict[str, Device]: Отфильтрованный словарь, исключающий устройства - offline более 7 дней. - """ now = datetime.now() filtered = {} for mac, device in devices.items(): @@ -301,14 +169,6 @@ class DeviceStorage: return filtered def save_devices(self, devices: List[Device]) -> None: - """Сохраняет список устройств в файл. - - Args: - devices (List[Device]): Список объектов Device для сохранения. - - Raises: - Exception: Если произошла ошибка при записи в файл. - """ try: data = SubnetScanResult(devices=devices).model_dump() with open(self.storage_file, "w", encoding="utf-8") as f: @@ -319,64 +179,87 @@ class DeviceStorage: raise Exception(self.translations.gettext("Ошибка при сохранении устройств: {error}").format(error=str(e))) class SubnetService: - """Сервис для сканирования подсетей и управления устройствами. - - Объединяет функционал сканирования сети, портов и хранения данных об устройствах. - """ - + """Сервис для сканирования подсетей и управления устройствами.""" def __init__(self, translations: Translations = Depends(get_translations("services"))): - """Инициализирует сервис подсети с настройкой всех компонентов. - - Автоматически определяет интерфейс и инициализирует сканеры и хранилище. - """ self.interface = self._get_default_interface() self.network_scanner = NetworkScanner(self.interface) self.port_scanner = PortScanner(self.interface) self.device_info = DeviceInfoCollector() self.storage = DeviceStorage() - self.translations = translations + if not hasattr(translations, 'gettext'): + logger.warning("Translations object does not support gettext, using fallback") + self.translations = Translations.load("utils/locales/en/LC_MESSAGES/services.mo") + else: + self.translations = translations + logger.info(f"Initialized SubnetService with interface: {self.interface}") def _get_default_interface(self) -> str: - """Определяет имя сетевого интерфейса по умолчанию. - - Returns: - str: Название интерфейса (например, "eth0"). - - Использует информацию о шлюзе или первый доступный интерфейс, отличный от "lo". - """ try: gateways = netifaces.gateways() default_gateway = gateways.get("default", {}).get(netifaces.AF_INET) if default_gateway: - return default_gateway[1] + iface = default_gateway[1] + logger.info(f"Default interface from gateway: {iface}") + return iface interfaces = netifaces.interfaces() for iface in interfaces: if iface != "lo": - return iface + addrs = netifaces.ifaddresses(iface) + if netifaces.AF_INET in addrs: + logger.info(f"Selected interface: {iface}") + return iface + logger.warning("No suitable interface found, defaulting to eth0") return "eth0" except Exception as e: - logger.warning(f"Failed to determine interface: {str(e)}") + logger.warning(f"Failed to determine interface: {str(e)}, defaulting to eth0") return "eth0" - def scan_subnet(self, subnet: str, ports: List[int] = None) -> SubnetScanResult: + def _get_default_subnet(self) -> str: + """Автоматически определяет подсеть локального устройства.""" + try: + interfaces = netifaces.interfaces() + for iface in interfaces: + if iface == "lo" or iface != self.interface: + continue + addrs = netifaces.ifaddresses(iface) + if netifaces.AF_INET in addrs: + ip_info = addrs[netifaces.AF_INET][0] + ip = ip_info["addr"] + netmask = ip_info["netmask"] + # Преобразуем маску из строки (например, "255.255.255.0") в CIDR + mask = ipaddress.ip_network(f"0.0.0.0/{netmask}", strict=False).prefixlen + network = ipaddress.ip_network(f"{ip}/{mask}", strict=False) + logger.info(f"Auto-detected subnet: {network} (IP: {ip}, Mask: {netmask} -> /{mask})") + return str(network) + logger.warning("No suitable subnet found, defaulting to 0.0.0.0/0") + return "0.0.0.0/0" + except Exception as e: + logger.error(f"Error determining default subnet: {str(e)}") + return "0.0.0.0/0" + + def scan_subnet(self, subnet: Optional[str] = None) -> SubnetScanResult: """Сканирует указанную подсеть и возвращает результаты. Args: - subnet (str): Подсеть в формате CIDR (например, "192.168.1.0/24"). - ports (List[int], optional): Список портов для сканирования. - По умолчанию None (используются порты из config.ports). + subnet (Optional[str]): Подсеть в формате CIDR (например, "10.1.1.0/24"). + Если None или пустая строка, определяется автоматически. Returns: SubnetScanResult: Объект с результатами сканирования, содержащий список устройств. - - Сохраняет данные в хранилище и обновляет статус устройств. """ start_time = time.time() devices = [] now = datetime.now().astimezone() formatted_now = now.strftime("%d-%m-%Y, %H:%M") try: + # Определяем подсеть, если не указана или пустая + if not subnet or subnet.strip() == "": + subnet = self._get_default_subnet() + if subnet == "0.0.0.0/0": + logger.warning("Default subnet is invalid, scan aborted") + return SubnetScanResult(devices=devices) + saved_devices = self.storage.load_devices() local_device = self._get_local_device(subnet) @@ -386,19 +269,14 @@ class SubnetService: devices.append(local_device) if local_device.mac: saved_devices[local_device.mac] = local_device + logger.info(f"Detected local device: {local_device.ip}, MAC: {local_device.mac}") + logger.info(f"Scanning subnet: {subnet} with interface: {self.interface}") discovered_devices = self.network_scanner.scan(subnet) + logger.info(f"Discovered devices: {len(discovered_devices)}") self.device_info.scan_nbtscan(subnet) def process_device(ip_mac: Tuple[str, str]) -> Optional[Device]: - """Обрабатывает устройство из пары IP и MAC. - - Args: - ip_mac (Tuple[str, str]): Кортеж с IP-адресом и MAC-адресом. - - Returns: - Optional[Device]: Объект Device или None, если устройство локальное. - """ ip, mac = ip_mac if local_device and ip == local_device.ip: return None @@ -410,20 +288,19 @@ class SubnetService: mac=mac, 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=existing_device.open_ports, status="online", last_seen=formatted_now, note=existing_device.note, ) vendor = self.device_info.get_vendor(mac) hostname = self.device_info.get_hostname(ip) - open_ports = self.port_scanner.scan(ip) return Device( ip=ip, mac=mac, hostname=hostname, vendor=vendor, - open_ports=open_ports, + open_ports=[], status="online", last_seen=formatted_now, note=None, @@ -457,14 +334,6 @@ class SubnetService: return SubnetScanResult(devices=devices) def _get_local_device(self, subnet: str) -> Optional[Device]: - """Определяет локальное устройство в указанной подсети. - - Args: - subnet (str): Подсеть в формате CIDR (например, "192.168.1.0/24"). - - Returns: - Optional[Device]: Объект локального устройства или None, если не найдено. - """ now = datetime.now().astimezone() formatted_now = now.strftime("%d-%m-%Y, %H:%M") try: @@ -480,124 +349,70 @@ class SubnetService: mac = addrs[netifaces.AF_LINK][0]["addr"] vendor = self.device_info.get_vendor(mac) hostname = self.device_info.get_hostname(ip) - open_ports = self.port_scanner.scan(ip) - logger.debug( - f"Local device: IP={ip}, MAC={mac}, Ports={open_ports}, Hostname={hostname}" - ) + logger.info(f"Local device detected: IP={ip}, MAC={mac}, Interface={iface}") return Device( ip=ip, mac=mac, hostname=hostname, vendor=vendor, - open_ports=open_ports, + open_ports=[], status="online", last_seen=formatted_now, note=None, ) - logger.warning( - f"Local device not found in subnet {subnet}") + logger.warning(f"Local device not found in subnet {subnet}") return None except Exception as e: - logger.warning( - f"Failed to get local device: {str(e)}") + logger.warning(f"Failed to get local device: {str(e)}") return None def add_note(self, mac: str, note: str) -> Tuple[bool, Optional[str]]: - """Добавляет заметку к устройству по MAC-адресу. - - Args: - mac (str): MAC-адрес устройства. - note (str): Текст заметки. - - Returns: - Tuple[bool, Optional[str]]: Кортеж (успех, сообщение об ошибке). - - Успех: (True, None). - - Ошибка: (False, сообщение об ошибке). - - Raises: - Exception: Если произошла ошибка при сохранении. - """ try: saved_devices = self.storage.load_devices() if mac not in saved_devices: logger.warning(f"Device with MAC {mac} not found") return False, self.translations.gettext("Устройство с MAC {mac} не найдено").format(mac=mac) if saved_devices[mac].note is not None: - logger.error( - f"Note for device with MAC {mac} already exists") + logger.error(f"Note for device with MAC {mac} already exists") return False, self.translations.gettext("Заметка для устройства с MAC {mac} уже существует").format(mac=mac) saved_devices[mac].note = note self.storage.save_devices(list(saved_devices.values())) - logger.info( - f"Added note '{note}' for device with MAC {mac}") + logger.info(f"Added note '{note}' for device with MAC {mac}") return True, None except Exception as e: - logger.error( - f"Error adding note for MAC {mac}: {str(e)}") + logger.error(f"Error adding note for MAC {mac}: {str(e)}") return False, self.translations.gettext("Внутренняя ошибка сервера: {error}").format(error=str(e)) def update_note(self, mac: str, note: str) -> Tuple[bool, Optional[str]]: - """Обновляет заметку устройства по MAC-адресу. - - Args: - mac (str): MAC-адрес устройства. - note (str): Новый текст заметки. - - Returns: - Tuple[bool, Optional[str]]: Кортеж (успех, сообщение об ошибке). - - Успех: (True, None). - - Ошибка: (False, сообщение об ошибке). - - Raises: - Exception: Если произошла ошибка при сохранении. - """ try: saved_devices = self.storage.load_devices() if mac not in saved_devices: logger.warning(f"Device with MAC {mac} not found") return False, self.translations.gettext("Устройство с MAC {mac} не найдено").format(mac=mac) if saved_devices[mac].note is None: - logger.error( - f"Note for device with MAC {mac} does not exist") + logger.error(f"Note for device with MAC {mac} does not exist") return False, self.translations.gettext("Заметка для устройства с MAC {mac} не существует").format(mac=mac) saved_devices[mac].note = note self.storage.save_devices(list(saved_devices.values())) - logger.info( - f"Updated note '{note}' for device with MAC {mac}") + logger.info(f"Updated note '{note}' for device with MAC {mac}") return True, None except Exception as e: - logger.error( - f"Error updating note for MAC {mac}: {str(e)}") + logger.error(f"Error updating note for MAC {mac}: {str(e)}") return False, self.translations.gettext("Внутренняя ошибка сервера: {error}").format(error=str(e)) def delete_note(self, mac: str) -> Tuple[bool, Optional[str]]: - """Удаляет заметку устройства по MAC-адресу. - - Args: - mac (str): MAC-адрес устройства. - - Returns: - Tuple[bool, Optional[str]]: Кортеж (успех, сообщение об ошибке). - - Успех: (True, None). - - Ошибка: (False, сообщение об ошибке). - - Raises: - Exception: Если произошла ошибка при сохранении. - """ try: saved_devices = self.storage.load_devices() if mac not in saved_devices: logger.warning(f"Device with MAC {mac} not found") return False, self.translations.gettext("Устройство с MAC {mac} не найдено").format(mac=mac) if saved_devices[mac].note is None: - logger.error( - f"Note for device with MAC {mac} does not exist") + logger.error(f"Note for device with MAC {mac} does not exist") return False, self.translations.gettext("Заметка для устройства с MAC {mac} не существует").format(mac=mac) saved_devices[mac].note = None self.storage.save_devices(list(saved_devices.values())) logger.info(f"Deleted note for device with MAC {mac}") return True, None except Exception as e: - logger.error( - f"Error deleting note for MAC {mac}: {str(e)}") + logger.error(f"Error deleting note for MAC {mac}: {str(e)}") return False, self.translations.gettext("Внутренняя ошибка сервера: {error}").format(error=str(e)) \ No newline at end of file diff --git a/SRV/services/wifi_service.py b/SRV/services/wifi_service.py index 7420cc1..e399fe6 100644 --- a/SRV/services/wifi_service.py +++ b/SRV/services/wifi_service.py @@ -9,112 +9,79 @@ import pywifi from models.wifi import WifiAccessPoint, WifiScanResult from pathlib import Path import json -from utils.i18n import get_translations, _ +from utils.i18n import get_translations from babel.support import Translations from fastapi import Depends +# Настройка логирования logging.basicConfig( level=logging.DEBUG, - format="%(asctime)s - %(name)s - %(levelname)s - %(message)s") + format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" +) logger = logging.getLogger(__name__) class WifiStorage: - """Класс для хранения данных о Wi-Fi сетях и их заметках по SSID. - - Предоставляет методы для загрузки и сохранения заметок, связанных с SSID сетей, - в JSON-файле на диске. - """ - + """Класс для хранения данных о Wi-Fi сетях и их заметках по SSID.""" def __init__(self, storage_path: str = "storage/wifi/wifi.json", translations: Translations = Depends(get_translations("services"))): - """Инициализирует объект хранилища Wi-Fi с указанным путём к файлу. - - Args: - storage_path (str, optional): Путь к файлу для хранения заметок. - По умолчанию "storage/wifi/wifi.json". - - Создаёт родительские директории, если они не существуют. - """ self.storage_file = Path(storage_path) self.storage_file.parent.mkdir(parents=True, exist_ok=True) self.translations = translations + logger.info(f"Wi-Fi storage initialized at: {self.storage_file.absolute()}") def load_notes(self) -> Dict[str, Optional[str]]: - """Загружает заметки по SSID из файла хранения. - - Returns: - Dict[str, Optional[str]]: Словарь, где ключ — SSID, значение — заметка - или None, если заметки нет. - - Raises: - Exception: Если произошла ошибка при чтении или парсинге файла. - """ + """Загружает заметки по SSID из файла хранения.""" try: if self.storage_file.exists(): with open(self.storage_file, "r", encoding="utf-8") as f: data = json.load(f) + if not isinstance(data, dict) or "networks" not in data: + logger.warning(f"Invalid JSON structure in {self.storage_file}, expected {'networks': [...]}, got {data}") + return {} return {item["ssid"]: item.get("note") for item in data.get("networks", [])} return {} + except json.JSONDecodeError as e: + logger.error(f"JSON decode error in {self.storage_file}: {str(e)}") + return {} except Exception as e: - logger.error(f"Error loading notes: {str(e)}") + logger.error(f"Error loading notes from {self.storage_file}: {str(e)}") return {} def save_notes(self, notes: Dict[str, Optional[str]]) -> None: - """Сохраняет заметки по SSID в файл. - - Args: - notes (Dict[str, Optional[str]]): Словарь с SSID как ключами и заметками - как значениями. - - Raises: - Exception: Если произошла ошибка при записи в файл. - """ + """Сохраняет заметки по SSID в файл.""" try: data = {"networks": [{"ssid": ssid, "note": note} for ssid, note in notes.items()]} with open(self.storage_file, "w", encoding="utf-8") as f: json.dump(data, f, indent=2, ensure_ascii=False) - logger.info(f"Saved {len(notes)} notes to {self.storage_file}") + logger.info(f"Saved {len(notes)} notes to {self.storage_file.absolute()}") + except PermissionError as e: + logger.error(f"Permission denied writing to {self.storage_file}: {str(e)}") + raise except Exception as e: - logger.error(f"Error saving notes: {str(e)}") + logger.error(f"Error saving notes to {self.storage_file}: {str(e)}") raise Exception(self.translations.gettext("Ошибка при сохранении заметок: {error}").format(error=str(e))) class WifiService: - """Сервис для сканирования текущих Wi-Fi точек доступа. - - Предоставляет функционал для обнаружения Wi-Fi сетей с использованием pywifi и iw, - определения характеристик точек доступа (сигнал, шифрование, канал и т.д.), - а также управления заметками. - """ - + """Сервис для сканирования текущих Wi-Fi точек доступа.""" def __init__(self, translations: Translations = Depends(get_translations("services"))): - """Инициализирует сервис Wi-Fi с настройкой интерфейса и хранилища. - - Инициализирует объект MacLookup для определения вендоров по MAC-адресам, - PyWiFi для работы с Wi-Fi-интерфейсами и объект хранилища заметок. - """ self.mac_lookup = MacLookup() self.wifi = pywifi.PyWiFi() self.interface = self._get_wifi_interface() + if not hasattr(translations, 'gettext'): + logger.warning("Translations object does not support gettext, using fallback") + self.translations = Translations.load("utils/locales/en/LC_MESSAGES/services.mo") + else: + self.translations = translations self.storage = WifiStorage() - self.translations = translations def _get_wifi_interface(self) -> Optional[object]: - """Автоматически определяет активный Wi-Fi-интерфейс. - - Returns: - Optional[object]: Объект интерфейса PyWiFi или None, если интерфейс не найден. - - Проверяет доступные интерфейсы, выполняет тестовое сканирование и возвращает - первый активный интерфейс, исключая P2P-интерфейсы. - """ + """Автоматически определяет активный Wi-Fi-интерфейс.""" try: interfaces = self.wifi.interfaces() if not interfaces: logger.error(self.translations.gettext("Не найдено сетевых интерфейсов")) return None - logger.debug( - f"Available interfaces: {[iface.name() for iface in interfaces]}" - ) + logger.debug(f"Available interfaces: {[iface.name() for iface in interfaces]}") for iface in interfaces: iface_name = iface.name() @@ -128,9 +95,7 @@ class WifiService: if results: logger.info(f"Selected Wi-Fi interface: {iface_name}") return iface - logger.debug( - f"Interface {iface_name} returned no scan results" - ) + logger.debug(f"Interface {iface_name} returned no scan results") except Exception as e: logger.debug(f"Interface {iface_name} inactive: {str(e)}") @@ -142,36 +107,17 @@ class WifiService: @lru_cache(maxsize=1000) def get_vendor(self, mac: str) -> str: - """Определяет вендора по MAC-адресу с использованием кэширования. - - Args: - mac (str): MAC-адрес устройства в формате XX:XX:XX:XX:XX:XX. - - Returns: - str: Название вендора или "Unknown", если вендор не определён. - - Raises: - Exception: Если произошла ошибка при запросе к базе вендоров. - """ + """Определяет вендора по MAC-адресу с использованием кэширования.""" try: return self.mac_lookup.lookup(mac) except Exception as e: - logger.warning( - f"Failed to determine vendor for MAC {mac}: {str(e)}") + logger.warning(f"Failed to determine vendor for MAC {mac}: {str(e)}") return self.translations.gettext("Неизвестный") def _get_encryption_from_iw(self, ap_info: Dict) -> str: - """Определяет тип шифрования из данных, полученных через iw. - - Args: - ap_info (Dict): Словарь с данными точки доступа, включая поле "security". - - Returns: - str: Тип шифрования ("WPA3", "WPA2", "WPA", "WEP" или "Open"). - """ + """Определяет тип шифрования из данных, полученных через iw.""" security = ap_info.get("security", "") - logger.debug( - f"Encryption for BSSID {ap_info.get('bssid')}: {security}") + logger.debug(f"Encryption for BSSID {ap_info.get('bssid')}: {security}") if "WPA3" in security: return self.translations.gettext("WPA3") elif "WPA2" in security or "RSN" in security: @@ -183,14 +129,7 @@ class WifiService: return self.translations.gettext("Открытая") def _get_channel(self, frequency: Optional[int]) -> Optional[int]: - """Преобразует частоту в номер канала. - - Args: - frequency (Optional[int]): Частота в МГц или None. - - Returns: - Optional[int]: Номер канала или None, если частота вне диапазона. - """ + """Преобразует частоту в номер канала.""" if frequency is None: return None try: @@ -201,20 +140,11 @@ class WifiService: logger.debug(f"Frequency {frequency} outside 2.4/5 GHz range") return None except Exception as e: - logger.debug( - f"Error determining channel for frequency {frequency}: {str(e)}" - ) + logger.debug(f"Error determining channel for frequency {frequency}: {str(e)}") return None def _get_frequency(self, frequency: Optional[int]) -> Optional[str]: - """Определяет частотный диапазон на основе частоты. - - Args: - frequency (Optional[int]): Частота в МГц или None. - - Returns: - Optional[str]: Диапазон ("2.4 GHz", "5 GHz" или None). - """ + """Определяет частотный диапазон на основе частоты.""" if frequency is None: return None try: @@ -225,39 +155,23 @@ class WifiService: logger.debug(f"Frequency {frequency} outside 2.4/5 GHz range") return None except Exception as e: - logger.debug( - f"Error determining band for frequency {frequency}: {str(e)}" - ) + logger.debug(f"Error determining band for frequency {frequency}: {str(e)}") return None def _get_iw_scan_data(self) -> List[Dict]: - """Получает данные сканирования Wi-Fi через команду iw. - - Returns: - List[Dict]: Список словарей с данными точек доступа (BSSID, SSID, частота, - сигнал, шифрование). - - Raises: - subprocess.CalledProcessError: Если команда iw завершилась с ошибкой. - subprocess.TimeoutExpired: Если превышен таймаут выполнения. - Exception: Если произошла ошибка при парсинге вывода. - """ + """Получает данные сканирования Wi-Fi через команду iw.""" try: if not self.interface: logger.error(self.translations.gettext("Интерфейс не определён")) return [] cmd = ["sudo", "iw", "dev", self.interface.name(), "scan"] logger.debug(f"Executing command: {' '.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 error: {result.stderr}") return [] output = result.stdout - logger.debug( - f"iw scan output (first 500 chars): {output[:500]}...") + logger.debug(f"iw scan output (first 500 chars): {output[:500]}...") aps = [] current_ap = None @@ -274,15 +188,10 @@ 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).strip() 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): @@ -291,19 +200,13 @@ 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) - logger.debug( - f"Found {len(aps)} access points via iw: {[ap['bssid'] for ap in aps]}" - ) - for ap in aps: - logger.debug(f"Access point {ap['bssid']}: {ap}") + logger.debug(f"Found {len(aps)} access points via iw: {[ap['bssid'] for ap in aps]}") return aps except subprocess.CalledProcessError as e: logger.error(f"Error executing iw scan: {e.stderr}") @@ -316,18 +219,7 @@ class WifiService: return [] def scan_wifi(self, timeout: int = 30) -> WifiScanResult: - """Сканирует текущие Wi-Fi точки доступа. - - Args: - timeout (int, optional): Таймаут ожидания в секундах. По умолчанию 30. - - Returns: - WifiScanResult: Объект с результатами сканирования, содержащий список точек - доступа. - - Сканирует Wi-Fi с использованием pywifi и iw, объединяет данные, определяет - лучшие точки по сигналу и сохраняет заметки. - """ + """Сканирует текущие Wi-Fi точки доступа.""" start_time = time.time() aps = [] @@ -336,13 +228,20 @@ class WifiService: return WifiScanResult(access_points=aps) try: - logger.info( - f"Starting Wi-Fi scan on interface {self.interface.name()}" - ) - self.interface.scan() - time.sleep(10) - py_results = self.interface.scan_results() - logger.debug(f"Found {len(py_results)} access points via pywifi") + logger.info(f"Starting Wi-Fi scan on interface {self.interface.name()}") + max_attempts = 3 + for attempt in range(max_attempts): + self.interface.scan() + time.sleep(min(timeout, 5)) + py_results = self.interface.scan_results() + if py_results: + logger.info(f"Found {len(py_results)} access points via pywifi after attempt {attempt + 1}") + break + else: + logger.warning(f"Scan attempt {attempt + 1} failed, trying again (FAIL-BUSY?)") + else: + logger.error("All scan attempts failed") + return WifiScanResult(access_points=aps) iw_results = self._get_iw_scan_data() iw_by_bssid = {ap["bssid"]: ap for ap in iw_results if ap.get("bssid")} @@ -355,8 +254,8 @@ class WifiService: if not bssid: logger.debug("Skipped access point without BSSID") continue - ssid = ap.ssid if ap.ssid else "" - hidden = not ssid + ssid = ap.ssid if ap.ssid else f"Hidden_{bssid}" + hidden = not ap.ssid signal = ap.signal if ap.signal else None iw_data = iw_by_bssid.get(bssid, {}) @@ -366,10 +265,6 @@ class WifiService: encryption = self._get_encryption_from_iw(iw_data) vendor = self.get_vendor(bssid) - logger.debug( - f"Access point {bssid}: freq={frequency}, channel={channel}, encryption={encryption}, signal={signal}" - ) - current_ap = WifiAccessPoint( ssid=ssid, bssid=bssid, @@ -379,22 +274,27 @@ class WifiService: frequency=freq_band, vendor=vendor, hidden=hidden, + note=saved_notes.get(ssid) if ap.ssid else None, ) + logger.debug(f"Processed AP: {current_ap.model_dump()}") 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.note = saved_notes.get(ssid) + current_ap.note = saved_notes.get(ssid) if ap.ssid else None best_aps_by_ssid[ssid] = current_ap aps = list(best_aps_by_ssid.values()) + # Сохраняем хотя бы SSID, если заметок нет notes_to_save = {ap.ssid: ap.note for ap in aps if ap.ssid} - self.storage.save_notes(notes_to_save) + if not notes_to_save and aps: + notes_to_save = {ap.ssid: None for ap in aps if ap.ssid} # Сохраняем SSID без заметок + if notes_to_save: + self.storage.save_notes(notes_to_save) + else: + logger.warning("No notes or SSIDs to save, skipping file write") - logger.info( - f"Wi-Fi scan completed in {time.time() - start_time:.2f} seconds, " - f"found {len(aps)} unique networks" - ) + logger.info(f"Wi-Fi scan completed, found {len(aps)} unique networks") return WifiScanResult(access_points=aps) except Exception as e: @@ -402,20 +302,7 @@ class WifiService: return WifiScanResult(access_points=aps) def add_note(self, ssid: str, note: str) -> Tuple[bool, Optional[str]]: - """Добавляет заметку к сети Wi-Fi по SSID. - - Args: - ssid (str): SSID сети Wi-Fi. - note (str): Текст заметки. - - Returns: - Tuple[bool, Optional[str]]: Кортеж (успех, сообщение об ошибке). - - Успех: (True, None). - - Ошибка: (False, сообщение об ошибке). - - Raises: - Exception: Если произошла ошибка при сохранении заметки. - """ + """Добавляет заметку к сети Wi-Fi по SSID.""" try: saved_notes = self.storage.load_notes() if ssid not in saved_notes or saved_notes[ssid] is None: @@ -432,20 +319,7 @@ class WifiService: return False, self.translations.gettext("Внутренняя ошибка сервера: {error}").format(error=str(e)) def update_note(self, ssid: str, note: str) -> Tuple[bool, Optional[str]]: - """Обновляет заметку сети Wi-Fi по SSID. - - Args: - ssid (str): SSID сети Wi-Fi. - note (str): Новый текст заметки. - - Returns: - Tuple[bool, Optional[str]]: Кортеж (успех, сообщение об ошибке). - - Успех: (True, None). - - Ошибка: (False, сообщение об ошибке). - - Raises: - Exception: Если произошла ошибка при сохранении заметки. - """ + """Обновляет заметку сети Wi-Fi по SSID.""" try: saved_notes = self.storage.load_notes() if ssid not in saved_notes: @@ -463,19 +337,7 @@ class WifiService: return False, self.translations.gettext("Внутренняя ошибка сервера: {error}").format(error=str(e)) def delete_note(self, ssid: str) -> Tuple[bool, Optional[str]]: - """Удаляет заметку сети Wi-Fi по SSID. - - Args: - ssid (str): SSID сети Wi-Fi. - - Returns: - Tuple[bool, Optional[str]]: Кортеж (успех, сообщение об ошибке). - - Успех: (True, None). - - Ошибка: (False, сообщение об ошибке). - - Raises: - Exception: Если произошла ошибка при сохранении изменений. - """ + """Удаляет заметку сети Wi-Fi по SSID.""" try: saved_notes = self.storage.load_notes() if ssid not in saved_notes: