392 lines
18 KiB
Python
392 lines
18 KiB
Python
|
import time
|
|||
|
|
import logging
|
|||
|
|
import subprocess
|
|||
|
|
import re
|
|||
|
|
from functools import lru_cache
|
|||
|
|
from typing import List, Optional, Dict, Tuple
|
|||
|
|
from mac_vendor_lookup import MacLookup
|
|||
|
|
import pywifi
|
|||
|
|
from pywifi import Profile
|
|||
|
|
from models.wifi import WifiAccessPoint, WifiScanResult
|
|||
|
|
from pathlib import Path
|
|||
|
|
import json
|
|||
|
|
from datetime import datetime
|
|||
|
|
|
|||
|
|
logging.basicConfig(
|
|||
|
|
level=logging.DEBUG,
|
|||
|
|
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
|
|||
|
|
logger = logging.getLogger(__name__)
|
|||
|
|
|
|||
|
|
class WifiStorage:
|
|||
|
|
"""Класс для хранения данных о Wi-Fi сетях и их заметках по SSID."""
|
|||
|
|
|
|||
|
|
def __init__(self, storage_path: str = "etc/gonets/SRV/storage/wifi/wifi.json"):
|
|||
|
|
self.storage_file = Path(storage_path)
|
|||
|
|
self.storage_file.parent.mkdir(parents=True, exist_ok=True)
|
|||
|
|
|
|||
|
|
def load_notes(self) -> Dict[str, Optional[str]]:
|
|||
|
|
"""Загружает заметки по SSID из файла хранения."""
|
|||
|
|
try:
|
|||
|
|
if self.storage_file.exists():
|
|||
|
|
with open(self.storage_file, "r", encoding="utf-8") as f:
|
|||
|
|
data = json.load(f)
|
|||
|
|
return {item["ssid"]: item.get("note") for item in data.get("networks", [])}
|
|||
|
|
return {}
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"Ошибка при загрузке заметок: {str(e)}")
|
|||
|
|
return {}
|
|||
|
|
|
|||
|
|
def save_notes(self, notes: Dict[str, Optional[str]]) -> None:
|
|||
|
|
"""Сохраняет заметки по 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"Сохранено {len(notes)} заметок в {self.storage_file}")
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"Ошибка при сохранении заметок: {str(e)}")
|
|||
|
|
|
|||
|
|
class WifiService:
|
|||
|
|
"""Сервис для сканирования текущих Wi-Fi точек доступа"""
|
|||
|
|
|
|||
|
|
def __init__(self):
|
|||
|
|
self.mac_lookup = MacLookup()
|
|||
|
|
self.wifi = pywifi.PyWiFi()
|
|||
|
|
self.interface = self._get_wifi_interface()
|
|||
|
|
self.storage = WifiStorage()
|
|||
|
|
|
|||
|
|
def _get_wifi_interface(self) -> Optional[object]:
|
|||
|
|
"""Автоматически определяет активный Wi-Fi-интерфейс"""
|
|||
|
|
try:
|
|||
|
|
interfaces = self.wifi.interfaces()
|
|||
|
|
if not interfaces:
|
|||
|
|
logger.error("Не найдено сетевых интерфейсов")
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
logger.debug(
|
|||
|
|
f"Доступные интерфейсы: {[iface.name() for iface in interfaces]}"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
for iface in interfaces:
|
|||
|
|
iface_name = iface.name()
|
|||
|
|
if "p2p" in iface_name.lower():
|
|||
|
|
logger.debug(f"Пропущен интерфейс {iface_name} (P2P)")
|
|||
|
|
continue
|
|||
|
|
try:
|
|||
|
|
iface.scan()
|
|||
|
|
time.sleep(1)
|
|||
|
|
results = iface.scan_results()
|
|||
|
|
if results:
|
|||
|
|
logger.info(f"Выбран Wi-Fi-интерфейс: {iface_name}")
|
|||
|
|
return iface
|
|||
|
|
logger.debug(
|
|||
|
|
f"Интерфейс {iface_name} не вернул результатов сканирования"
|
|||
|
|
)
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.debug(f"Интерфейс {iface_name} неактивен: {str(e)}")
|
|||
|
|
|
|||
|
|
logger.error("Не найден активный Wi-Fi-интерфейс")
|
|||
|
|
return None
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"Ошибка при определении Wi-Fi-интерфейса: {str(e)}")
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
@lru_cache(maxsize=1000)
|
|||
|
|
def get_vendor(self, mac: str) -> str:
|
|||
|
|
try:
|
|||
|
|
return self.mac_lookup.lookup(mac)
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.warning(
|
|||
|
|
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}")
|
|||
|
|
if "WPA3" in security:
|
|||
|
|
return "WPA3"
|
|||
|
|
elif "WPA2" in security or "RSN" in security:
|
|||
|
|
return "WPA2"
|
|||
|
|
elif "WPA" in security:
|
|||
|
|
return "WPA"
|
|||
|
|
elif "WEP" in security:
|
|||
|
|
return "WEP"
|
|||
|
|
return "Open"
|
|||
|
|
|
|||
|
|
def _get_channel(self, frequency: Optional[int]) -> Optional[int]:
|
|||
|
|
"""Преобразует частоту в канал."""
|
|||
|
|
if frequency is None:
|
|||
|
|
return None
|
|||
|
|
try:
|
|||
|
|
if 2412 <= frequency <= 2472:
|
|||
|
|
return (frequency - 2412) // 5 + 1
|
|||
|
|
elif 5180 <= frequency <= 5825:
|
|||
|
|
return (frequency - 5180) // 5 + 36
|
|||
|
|
logger.debug(f"Частота {frequency} вне диапазона 2.4/5 GHz")
|
|||
|
|
return None
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.debug(
|
|||
|
|
f"Ошибка при определении канала для частоты {frequency}: {str(e)}"
|
|||
|
|
)
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
def _get_frequency(self, frequency: Optional[int]) -> Optional[str]:
|
|||
|
|
"""Определяет частотный диапазон."""
|
|||
|
|
if frequency is None:
|
|||
|
|
return None
|
|||
|
|
try:
|
|||
|
|
if 2412 <= frequency <= 2472:
|
|||
|
|
return "2.4 GHz"
|
|||
|
|
elif 5180 <= frequency <= 5825:
|
|||
|
|
return "5 GHz"
|
|||
|
|
logger.debug(f"Частота {frequency} вне диапазона 2.4/5 GHz")
|
|||
|
|
return None
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.debug(
|
|||
|
|
f"Ошибка при определении диапазона для частоты {frequency}: {str(e)}"
|
|||
|
|
)
|
|||
|
|
return None
|
|||
|
|
|
|||
|
|
def _get_iw_scan_data(self) -> List[Dict]:
|
|||
|
|
"""Получает данные сканирования через iw."""
|
|||
|
|
try:
|
|||
|
|
if not self.interface:
|
|||
|
|
logger.error("Интерфейс не определён")
|
|||
|
|
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)
|
|||
|
|
if result.returncode != 0:
|
|||
|
|
logger.error(f"Ошибка iw scan: {result.stderr}")
|
|||
|
|
return []
|
|||
|
|
output = result.stdout
|
|||
|
|
logger.debug(
|
|||
|
|
f"Вывод iw scan (первые 500 символов): {output[:500]}...")
|
|||
|
|
|
|||
|
|
aps = []
|
|||
|
|
current_ap = None
|
|||
|
|
bssid_pattern = re.compile(r"BSS ([0-9a-f:]{17})")
|
|||
|
|
ssid_pattern = re.compile(r"SSID: ?(.*)")
|
|||
|
|
freq_pattern = re.compile(r"freq: (\d+)")
|
|||
|
|
signal_pattern = re.compile(r"signal: ([-\d.]+) dBm")
|
|||
|
|
rsn_pattern = re.compile(r"RSN:")
|
|||
|
|
wpa_pattern = re.compile(r"WPA:")
|
|||
|
|
security_type_pattern = re.compile(r"\* (WPA3|WPA2|WPA|WEP)")
|
|||
|
|
|
|||
|
|
for line in output.splitlines():
|
|||
|
|
line = line.strip()
|
|||
|
|
if bssid_match := bssid_pattern.match(line):
|
|||
|
|
if current_ap:
|
|||
|
|
aps.append(current_ap)
|
|||
|
|
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 "")
|
|||
|
|
elif freq_match := freq_pattern.match(line):
|
|||
|
|
current_ap["frequency"] = int(freq_match.group(1))
|
|||
|
|
elif signal_match := signal_pattern.match(line):
|
|||
|
|
current_ap["signal"] = float(signal_match.group(1))
|
|||
|
|
elif rsn_pattern.match(line):
|
|||
|
|
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) + ";"
|
|||
|
|
|
|||
|
|
if current_ap:
|
|||
|
|
aps.append(current_ap)
|
|||
|
|
|
|||
|
|
logger.debug(
|
|||
|
|
f"Найдено {len(aps)} точек через iw: {[ap['bssid'] for ap in aps]}"
|
|||
|
|
)
|
|||
|
|
for ap in aps:
|
|||
|
|
logger.debug(f"Данные точки {ap['bssid']}: {ap}")
|
|||
|
|
return aps
|
|||
|
|
except subprocess.CalledProcessError as e:
|
|||
|
|
logger.error(f"Ошибка при выполнении iw scan: {e.stderr}")
|
|||
|
|
return []
|
|||
|
|
except subprocess.TimeoutExpired:
|
|||
|
|
logger.error("Таймаут при выполнении iw scan")
|
|||
|
|
return []
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"Ошибка при парсинге iw scan: {str(e)}")
|
|||
|
|
return []
|
|||
|
|
|
|||
|
|
def scan_wifi(self, timeout: int = 30) -> WifiScanResult:
|
|||
|
|
"""Сканирует текущие Wi-Fi точки доступа"""
|
|||
|
|
start_time = time.time()
|
|||
|
|
aps = []
|
|||
|
|
|
|||
|
|
if not self.interface:
|
|||
|
|
logger.error("Wi-Fi-интерфейс не доступен")
|
|||
|
|
return WifiScanResult(access_points=aps)
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
logger.info(
|
|||
|
|
f"Начало сканирования Wi-Fi на интерфейсе {self.interface.name()}"
|
|||
|
|
)
|
|||
|
|
self.interface.scan()
|
|||
|
|
time.sleep(10)
|
|||
|
|
py_results = self.interface.scan_results()
|
|||
|
|
logger.debug(f"Найдено {len(py_results)} точек через pywifi")
|
|||
|
|
|
|||
|
|
iw_results = self._get_iw_scan_data()
|
|||
|
|
iw_by_bssid = {ap["bssid"]: ap for ap in iw_results if ap.get("bssid")}
|
|||
|
|
|
|||
|
|
best_aps_by_ssid: Dict[str, WifiAccessPoint] = {}
|
|||
|
|
saved_notes = self.storage.load_notes()
|
|||
|
|
|
|||
|
|
for ap in py_results:
|
|||
|
|
bssid = ap.bssid.lower() if ap.bssid else None
|
|||
|
|
if not bssid:
|
|||
|
|
logger.debug("Пропущена точка без BSSID")
|
|||
|
|
continue
|
|||
|
|
ssid = ap.ssid if ap.ssid else ""
|
|||
|
|
hidden = not ssid
|
|||
|
|
signal = ap.signal if ap.signal else None
|
|||
|
|
iw_data = iw_by_bssid.get(bssid, {})
|
|||
|
|
|
|||
|
|
frequency = iw_data.get("frequency")
|
|||
|
|
channel = self._get_channel(frequency)
|
|||
|
|
freq_band = self._get_frequency(frequency)
|
|||
|
|
encryption = self._get_encryption_from_iw(iw_data)
|
|||
|
|
vendor = self.get_vendor(bssid)
|
|||
|
|
|
|||
|
|
logger.debug(
|
|||
|
|
f"Точка {bssid}: freq={frequency}, channel={channel}, encryption={encryption}, signal={signal}"
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
current_ap = WifiAccessPoint(
|
|||
|
|
ssid=ssid,
|
|||
|
|
bssid=bssid,
|
|||
|
|
channel=channel,
|
|||
|
|
signal=signal if signal else iw_data.get("signal"),
|
|||
|
|
encryption=encryption,
|
|||
|
|
frequency=freq_band,
|
|||
|
|
vendor=vendor,
|
|||
|
|
hidden=hidden,
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
if ssid not in best_aps_by_ssid or (
|
|||
|
|
current_ap.signal and best_aps_by_ssid[ssid].signal
|
|||
|
|
and current_ap.signal > best_aps_by_ssid[ssid].signal):
|
|||
|
|
# При обновлении или добавлении точки сохраняем заметку из saved_notes
|
|||
|
|
current_ap.note = saved_notes.get(ssid)
|
|||
|
|
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)
|
|||
|
|
|
|||
|
|
logger.info(
|
|||
|
|
f"Сканирование Wi-Fi завершено за {time.time() - start_time:.2f} секунд, "
|
|||
|
|
f"найдено {len(aps)} уникальных сетей")
|
|||
|
|
return WifiScanResult(access_points=aps)
|
|||
|
|
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"Ошибка при сканировании Wi-Fi: {str(e)}")
|
|||
|
|
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: Если произошла ошибка при сохранении.
|
|||
|
|
"""
|
|||
|
|
try:
|
|||
|
|
saved_notes = self.storage.load_notes()
|
|||
|
|
if ssid not in saved_notes or saved_notes[ssid] is None:
|
|||
|
|
logger.info(f"Добавлена новая заметка для сети с SSID {ssid}")
|
|||
|
|
elif saved_notes[ssid] is not None:
|
|||
|
|
logger.error(f"Заметка для сети с SSID {ssid} уже существует")
|
|||
|
|
return False, f"Заметка для сети с SSID {ssid} уже существует"
|
|||
|
|
saved_notes[ssid] = note
|
|||
|
|
self.storage.save_notes(saved_notes)
|
|||
|
|
logger.info(f"Добавлена заметка '{note}' для сети с SSID {ssid}")
|
|||
|
|
return True, None
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"Ошибка при добавлении заметки для SSID {ssid}: {str(e)}")
|
|||
|
|
return False, f"Внутренняя ошибка сервера: {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: Если произошла ошибка при сохранении.
|
|||
|
|
"""
|
|||
|
|
try:
|
|||
|
|
saved_notes = self.storage.load_notes()
|
|||
|
|
if ssid not in saved_notes:
|
|||
|
|
logger.warning(f"Сеть с SSID {ssid} не найдена")
|
|||
|
|
return False, f"Сеть с SSID {ssid} не найдена"
|
|||
|
|
if saved_notes[ssid] is None:
|
|||
|
|
logger.error(f"Заметка для сети с SSID {ssid} не существует")
|
|||
|
|
return False, f"Заметка для сети с SSID {ssid} не существует"
|
|||
|
|
saved_notes[ssid] = note
|
|||
|
|
self.storage.save_notes(saved_notes)
|
|||
|
|
logger.info(f"Обновлена заметка '{note}' для сети с SSID {ssid}")
|
|||
|
|
return True, None
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"Ошибка при обновлении заметки для SSID {ssid}: {str(e)}")
|
|||
|
|
return False, f"Внутренняя ошибка сервера: {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: Если произошла ошибка при сохранении.
|
|||
|
|
"""
|
|||
|
|
try:
|
|||
|
|
saved_notes = self.storage.load_notes()
|
|||
|
|
if ssid not in saved_notes:
|
|||
|
|
logger.warning(f"Сеть с SSID {ssid} не найдена")
|
|||
|
|
return False, f"Сеть с SSID {ssid} не найдена"
|
|||
|
|
if saved_notes[ssid] is None:
|
|||
|
|
logger.error(f"Заметка для сети с SSID {ssid} не существует")
|
|||
|
|
return False, f"Заметка для сети с SSID {ssid} не существует"
|
|||
|
|
saved_notes[ssid] = None
|
|||
|
|
self.storage.save_notes(saved_notes)
|
|||
|
|
logger.info(f"Удалена заметка для сети с SSID {ssid}")
|
|||
|
|
return True, None
|
|||
|
|
except Exception as e:
|
|||
|
|
logger.error(f"Ошибка при удалении заметки для SSID {ssid}: {str(e)}")
|
|||
|
|
return False, f"Внутренняя ошибка сервера: {str(e)}"
|