Инициализация в новом репозитории
This commit is contained in:
commit
eb1f18f53e
374 changed files with 226678 additions and 0 deletions
200
SRV/services/network_service.py
Normal file
200
SRV/services/network_service.py
Normal file
|
|
@ -0,0 +1,200 @@
|
|||
from typing import List, Dict, Tuple, Union, Optional
|
||||
from sdbus_async.networkmanager import (
|
||||
NetworkManager,
|
||||
NetworkDeviceGeneric,
|
||||
IPv4Config,
|
||||
IPv6Config,
|
||||
NetworkManagerSettings,
|
||||
NetworkDeviceWireless,
|
||||
AccessPoint,
|
||||
DeviceType
|
||||
)
|
||||
from sdbus import sd_bus_open_system
|
||||
from models.network import (
|
||||
NetworkInterface,
|
||||
IPv4Route,
|
||||
IPv6Route,
|
||||
IPConfig,
|
||||
IndividualIPConfig,
|
||||
NetworkInterfacesContainer,
|
||||
WiFiDevice
|
||||
)
|
||||
import subprocess
|
||||
|
||||
class NetworkService:
|
||||
"""Сервис для взаимодействия с NetworkManager через D-Bus"""
|
||||
|
||||
def __init__(self):
|
||||
self.system_bus = sd_bus_open_system()
|
||||
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:
|
||||
"""Получает метод IP для устройства"""
|
||||
try:
|
||||
config = await device.get_applied_connection()
|
||||
settings = config[0].get(ip_type, {})
|
||||
return str(settings.get("method", ('s', 'unknown'))[1])
|
||||
except Exception:
|
||||
return 'unknown'
|
||||
|
||||
async def get_connection_name(self, device: NetworkDeviceGeneric) -> str:
|
||||
"""Извлекает имя соединения (GENERAL.CONNECTION) для устройства"""
|
||||
try:
|
||||
applied_connection = await device.get_applied_connection()
|
||||
connection_settings = applied_connection[0]
|
||||
connection_info = connection_settings.get('connection', {})
|
||||
return str(connection_info.get('id', ('s', 'unknown'))[1])
|
||||
except Exception as e:
|
||||
# print(f"Ошибка при получении соединения для устройства {await device.interface}: {e}")
|
||||
# срёт ошибки дл отключенных интерфейсов
|
||||
# добавить потом условия для отключенных интерфейсов
|
||||
return 'unknown'
|
||||
|
||||
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)
|
||||
address_data = await ip_config.address_data
|
||||
for address in address_data:
|
||||
config = IndividualIPConfig(
|
||||
address=str(address['address'][1]),
|
||||
routes=[
|
||||
globals()[f'IPv{ip_version}Route'](
|
||||
destination=route.get('dest', ('s', '0.0.0.0' if ip_version == '4' else '::'))[1],
|
||||
prefix=route.get('prefix', ('u', 0))[1],
|
||||
next_hop=route.get('next-hop', ('s', '0.0.0.0' if ip_version == '4' else '::'))[1],
|
||||
metric=route.get('metric', ('u', 0))[1]
|
||||
) for route in await ip_config.route_data
|
||||
],
|
||||
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')]
|
||||
)
|
||||
configs.append(config)
|
||||
return configs
|
||||
|
||||
async def get_wifi_info(self, wifi_device: NetworkDeviceWireless) -> Dict[str, Optional[Union[str, int]]]:
|
||||
"""Получает информацию о Wi-Fi устройстве"""
|
||||
active_ap = await wifi_device.active_access_point
|
||||
wifi_info = {
|
||||
'ssid': None,
|
||||
'bssid': None,
|
||||
'bitrate': await wifi_device.bitrate,
|
||||
'strength': None,
|
||||
'frequency': None,
|
||||
'mode': None
|
||||
}
|
||||
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
|
||||
})
|
||||
return wifi_info
|
||||
|
||||
async def get_network_interfaces(self, initial_positions: Dict[str, Tuple[int, int]] = None) -> List[Union[NetworkInterface, WiFiDevice]]:
|
||||
"""Получает список сетевых интерфейсов"""
|
||||
devices_paths = await self.nm.get_devices()
|
||||
network_interfaces = []
|
||||
initial_positions = initial_positions or {}
|
||||
|
||||
for device_path in devices_paths:
|
||||
generic_device = NetworkDeviceGeneric(device_path, self.system_bus)
|
||||
interface_name = str(await generic_device.interface)
|
||||
device_type = DeviceType(await generic_device.device_type).name
|
||||
|
||||
if device_type == 'LOOPBACK':
|
||||
continue
|
||||
|
||||
device_state = str(await generic_device.state)
|
||||
if device_state == '10':
|
||||
continue
|
||||
|
||||
common_data = {
|
||||
'name': interface_name,
|
||||
'type': device_type,
|
||||
'device_state': device_state,
|
||||
'active_connection': await self.get_connection_name(generic_device),
|
||||
'mac_address': str(await generic_device.hw_address),
|
||||
'device_mtu': str(await generic_device.mtu),
|
||||
'ipv4': IPConfig(configurations=await self.get_ip_config(generic_device, '4')),
|
||||
'ipv6': IPConfig(configurations=await self.get_ip_config(generic_device, '6')),
|
||||
'x': initial_positions.get(interface_name, (100, 100))[0],
|
||||
'y': initial_positions.get(interface_name, (100, 100))[1],
|
||||
}
|
||||
|
||||
if device_type == "WIFI":
|
||||
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))
|
||||
else:
|
||||
network_interfaces.append(NetworkInterface(**common_data))
|
||||
|
||||
return network_interfaces
|
||||
|
||||
async def get_hostname(self) -> Dict[str, str]:
|
||||
"""Возвращает имя хоста из команды hostname в формате JSON"""
|
||||
try:
|
||||
result = subprocess.run(['hostname'], capture_output=True, text=True, check=True)
|
||||
hostname = result.stdout.strip()
|
||||
return {"hostname": hostname}
|
||||
except subprocess.CalledProcessError as e:
|
||||
raise RuntimeError(f"Failed to execute hostname: {e}")
|
||||
except Exception as e:
|
||||
raise RuntimeError(f"Error processing hostname output: {e}")
|
||||
|
||||
async def get_interfaces_by_state(self, state: str, initial_positions: Dict[str, Tuple[int, int]] = None) -> List[NetworkInterface]:
|
||||
"""
|
||||
Получает список интерфейсов по состоянию
|
||||
"""
|
||||
all_interfaces = await self.get_network_interfaces(initial_positions)
|
||||
return [interface for interface in all_interfaces if interface.device_state == state]
|
||||
|
||||
async def get_interfaces_container(self, state: Optional[str] = None, initial_positions: Dict[str, Tuple[int, int]] = None, container_x: int = 0, container_y: int = 0) -> NetworkInterfacesContainer:
|
||||
""" Получает объект, содержащий список интерфейсов и координаты контейнера """
|
||||
if state:
|
||||
interfaces = await self.get_interfaces_by_state(state, initial_positions)
|
||||
else:
|
||||
interfaces = await self.get_network_interfaces(initial_positions)
|
||||
return NetworkInterfacesContainer(
|
||||
interfaces=interfaces,
|
||||
container_x=container_x,
|
||||
container_y=container_y
|
||||
)
|
||||
|
||||
async def get_network_interfaces_container(self, initial_positions: Optional[Dict[str, Tuple[int, int]]] = None, ) -> NetworkInterfacesContainer:
|
||||
""" Получает контейнер с сетевыми интерфейсами и координатами контейнера """
|
||||
interfaces = await self.get_network_interfaces(initial_positions)
|
||||
return NetworkInterfacesContainer(
|
||||
interfaces=interfaces,
|
||||
container_x=0,
|
||||
container_y=0
|
||||
)
|
||||
|
||||
async def get_activated_interfaces(self, initial_positions: Dict[str, Tuple[int, int]] = None) -> List[NetworkInterface]:
|
||||
""" Получает список активных сетевых интерфейсов """
|
||||
return await self.get_interfaces_by_state('100', initial_positions)
|
||||
|
||||
async def get_deactivated_interfaces(self, initial_positions: Dict[str, Tuple[int, int]] = None) -> List[NetworkInterface]:
|
||||
""" Получает список неактивных сетевых интерфейсов """
|
||||
all_interfaces = await self.get_network_interfaces(initial_positions)
|
||||
return [interface for interface in all_interfaces if interface.device_state != '100']
|
||||
|
||||
async def get_activated_interfaces_container(self, initial_positions: Dict[str, Tuple[int, int]] = None, container_x: int = 0, container_y: int = 0) -> NetworkInterfacesContainer:
|
||||
""" Получает объект, содержащий список активных интерфейсов и координаты контейнера """
|
||||
return await self.get_interfaces_container('100', initial_positions, container_x, container_y)
|
||||
|
||||
async def get_deactivated_interfaces_container(self, initial_positions: Dict[str, Tuple[int, int]] = None, container_x: int = 0, container_y: int = 0) -> NetworkInterfacesContainer:
|
||||
""" Получает объект, содержащий список неактивных интерфейсов и координаты контейнера """
|
||||
interfaces = await self.get_deactivated_interfaces(initial_positions)
|
||||
return NetworkInterfacesContainer(
|
||||
interfaces=interfaces,
|
||||
container_x=container_x,
|
||||
container_y=container_y
|
||||
)
|
||||
Loading…
Add table
Add a link
Reference in a new issue