#!/usr/bin/env python3 """ Скрипт для анализа VLAN на множестве роутеров OpenWrt. Для каждого роутера: 1. Находит VLAN с наибольшим количеством портов 2. Отключает все порты в нем (кроме 6t) 3. Запускает iperf3 -c 172.20.100.203 --bidir на роутере 4. Сохраняет результат в отдельный файл 5. Включает порты обратно """ import paramiko import re import sys import time import datetime import os import threading from concurrent.futures import ThreadPoolExecutor, as_completed class RouterTester: def __init__(self, host, username, password, port=22): self.host = host self.username = username self.password = password self.port = port self.ssh = None self.vlan_info = None self.ports_toggled = [] def connect(self): """Подключается к роутеру по SSH.""" try: self.ssh = paramiko.SSHClient() self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) self.ssh.connect( self.host, port=self.port, username=self.username, password=self.password, timeout=10 ) return True except Exception as e: print(f" ❌ {self.host}: Ошибка подключения: {e}") return False def close(self): """Закрывает SSH-соединение.""" if self.ssh: self.ssh.close() def exec_command(self, command): """Выполняет команду на роутере и возвращает вывод.""" stdin, stdout, stderr = self.ssh.exec_command(command) output = stdout.read().decode('utf-8') error = stderr.read().decode('utf-8') return output, error def get_switch_config(self): """Получает конфигурацию коммутатора.""" output, error = self.exec_command('swconfig dev switch0 show') if error: print(f" ❌ {self.host}: Ошибка swconfig: {error}") return None return output def parse_vlans(self, config_text): """Парсит VLAN из вывода swconfig.""" vlans = [] current_vlan = None lines = config_text.split('\n') for line in lines: line = line.strip() vlan_match = re.match(r'^VLAN\s+(\d+):$', line) if vlan_match: if current_vlan: vlans.append(current_vlan) current_vlan = {'vid': None, 'ports': [], 'vlan_num': int(vlan_match.group(1))} continue if current_vlan is not None: vid_match = re.match(r'^vid:\s+(\d+)$', line) if vid_match: current_vlan['vid'] = int(vid_match.group(1)) continue ports_match = re.match(r'^ports:\s+(.+)$', line) if ports_match: ports_str = ports_match.group(1) current_vlan['ports'] = [p.strip() for p in ports_str.split() if p.strip()] continue if current_vlan and current_vlan.get('vid') is not None: vlans.append(current_vlan) return vlans def find_vlan_with_most_ports(self, vlans): """Находит VLAN с максимальным количеством портов.""" if not vlans: return None return sorted(vlans, key=lambda v: len(v['ports']), reverse=True)[0] def get_ports_to_toggle(self, ports): """Возвращает порты для отключения (исключая 6t).""" return [p for p in ports if p != '6t'] def disable_ports(self, ports): """Отключает указанные порты.""" print(f" {self.host}: Отключаю порты: {' '.join(ports)}") for port in ports: port_num = port.replace('t', '') command = f'swconfig dev switch0 port {port_num} set port_disabled 1' self.exec_command(command) self.exec_command('swconfig dev switch0 set apply') self.ports_toggled = ports time.sleep(1) def enable_ports(self): """Включает порты обратно.""" if not self.ports_toggled: return print(f" {self.host}: Включаю порты: {' '.join(self.ports_toggled)}") for port in self.ports_toggled: port_num = port.replace('t', '') command = f'swconfig dev switch0 port {port_num} set port_disabled 0' self.exec_command(command) self.exec_command('swconfig dev switch0 set apply') time.sleep(1) def run_iperf_test(self, server_ip, output_file, duration=10, port=5201): """ Запускает iperf3 -c на роутере с флагом --bidir. """ print(f" {self.host}: Запуск iperf3 -c {server_ip} --bidir (duration={duration}с)") # Временный файл на роутере temp_file = '/tmp/iperf3_result.log' # Команда с флагом --bidir command = f'iperf3 -c {server_ip} -t {duration} -p {port} --bidir > {temp_file} 2>&1' # Запускаем iperf3 в фоновом режиме self.exec_command(command) # Ждем завершения теста time.sleep(duration + 3) # Проверяем, завершился ли процесс output, _ = self.exec_command('ps aux | grep "iperf3 -c" | grep -v grep') if output.strip(): print(f" {self.host}: ⚠️ iperf3 все еще работает, принудительно останавливаю...") self.exec_command('pkill -f "iperf3 -c"') time.sleep(1) # Читаем результат output, error = self.exec_command(f'cat {temp_file}') self.exec_command(f'rm -f {temp_file}') if error: print(f" {self.host}: ❌ Ошибка при чтении результата: {error}") return False # Сохраняем результат try: with open(output_file, 'w') as f: f.write(f"=== iperf3 тест (bidir) ===\n") f.write(f"Роутер: {self.host}\n") f.write(f"Сервер: {server_ip}:{port}\n") f.write(f"Режим: bidirectional (--bidir)\n") f.write(f"Длительность: {duration} сек\n") f.write(f"Время запуска: {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n") f.write("=" * 50 + "\n\n") f.write(output) # Проверяем наличие ошибок в выводе if "error" in output.lower() or "failed" in output.lower(): print(f" {self.host}: ⚠️ iperf3 завершился с ошибкой") return False print(f" {self.host}: ✅ iperf3 тест завершен успешно") return True except Exception as e: print(f" {self.host}: ❌ Ошибка при сохранении результата: {e}") return False def check_iperf3(self): """Проверяет наличие iperf3 на роутере.""" output, _ = self.exec_command('which iperf3') return bool(output.strip()) def process(self, server_ip, output_dir, duration=10, port=5201): """ Полный процесс обработки одного роутера. """ print(f"\n{'=' * 50}") print(f"🔍 Обработка роутера: {self.host}") print(f"{'=' * 50}") # Подключаемся if not self.connect(): return False try: # Проверяем iperf3 if not self.check_iperf3(): print(f" {self.host}: ❌ iperf3 не установлен") print(f" {self.host}: Установите: opkg update && opkg install iperf3") return False # Получаем конфигурацию VLAN config = self.get_switch_config() if config is None: return False # Парсим VLAN vlans = self.parse_vlans(config) if not vlans: print(f" {self.host}: ❌ VLAN не найдены") return False # Находим VLAN с наибольшим количеством портов best_vlan = self.find_vlan_with_most_ports(vlans) print(f" {self.host}: Выбран VLAN {best_vlan['vlan_num']} (vid: {best_vlan['vid']})") print(f" {self.host}: Порты: {' '.join(best_vlan['ports'])}") # Определяем порты для отключения ports_to_toggle = self.get_ports_to_toggle(best_vlan['ports']) if not ports_to_toggle: print(f" {self.host}: ⚠️ Нет портов для отключения (только 6t)") return False print(f" {self.host}: Будут отключены порты: {' '.join(ports_to_toggle)}") # Генерируем имя файла timestamp = datetime.datetime.now().strftime('%Y%m%d_%H%M%S') output_file = os.path.join(output_dir, f'iperf3_{self.host}_{timestamp}.log') # Шаг 1: Отключаем порты print(f" {self.host}: --- Шаг 1: Отключение портов ---") self.disable_ports(ports_to_toggle) # Шаг 2: Запускаем iperf3 с --bidir print(f" {self.host}: --- Шаг 2: iperf3 тест ---") success = self.run_iperf_test(server_ip, output_file, duration, port) # Шаг 3: Включаем порты обратно print(f" {self.host}: --- Шаг 3: Включение портов ---") self.enable_ports() return success except Exception as e: print(f" {self.host}: ❌ Критическая ошибка: {e}") # Пытаемся включить порты в случае ошибки try: self.enable_ports() except: pass return False finally: self.close() def main(): # Настройки iperf3 IPERF_SERVER = '172.20.100.203' IPERF_PORT = 5201 IPERF_DURATION = 10 # Директория для результатов OUTPUT_DIR = './iperf_results' if not os.path.exists(OUTPUT_DIR): os.makedirs(OUTPUT_DIR) # Список роутеров routers = [ #{'host': '192.168.16.4', 'username': 'root', 'password': 'root'}, #{'host': '192.168.16.8', 'username': 'root', 'password': 'root'}, #{'host': '192.168.16.9', 'username': 'root', 'password': 'root'}, #{'host': '192.168.16.10', 'username': 'root', 'password': 'root'}, #{'host': '192.168.16.11', 'username': 'root', 'password': 'root'}, #{'host': '192.168.16.27', 'username': 'root', 'password': 'root'}, #{'host': '192.168.16.28', 'username': 'root', 'password': 'root'}, #{'host': '192.168.16.29', 'username': 'root', 'password': 'root'}, #{'host': '192.168.16.30', 'username': 'root', 'password': 'root'}, #{'host': '192.168.16.32', 'username': 'root', 'password': 'root'}, #{'host': '192.168.16.34', 'username': 'root', 'password': 'root'}, #{'host': '192.168.16.35', 'username': 'root', 'password': 'root'}, #{'host': '192.168.16.38', 'username': 'root', 'password': 'root'}, #{'host': '192.168.16.39', 'username': 'root', 'password': 'root'}, #{'host': '192.168.16.40', 'username': 'root', 'password': 'root'}, #{'host': '192.168.16.41', 'username': 'root', 'password': 'root'}, #{'host': '192.168.16.42', 'username': 'root', 'password': 'root'}, #{'host': '192.168.16.43', 'username': 'root', 'password': 'root'}, #{'host': '192.168.16.44', 'username': 'root', 'password': 'root'}, #{'host': '192.168.16.45', 'username': 'root', 'password': 'root'}, #{'host': '192.168.16.49', 'username': 'root', 'password': 'root'}, #{'host': '192.168.16.50', 'username': 'root', 'password': 'root'}, #{'host': '192.168.16.51', 'username': 'root', 'password': 'root'}, #{'host': '192.168.16.55', 'username': 'root', 'password': 'root'}, #{'host': '192.168.16.57', 'username': 'root', 'password': 'root'}, #{'host': '192.168.16.58', 'username': 'root', 'password': 'root'}, #{'host': '192.168.16.62', 'username': 'root', 'password': 'root'}, #{'host': '192.168.16.65', 'username': 'root', 'password': 'root'}, #{'host': '192.168.16.66', 'username': 'root', 'password': 'root'}, #{'host': '192.168.16.67', 'username': 'root', 'password': 'root'}, #{'host': '192.168.16.68', 'username': 'root', 'password': 'root'}, #{'host': '192.168.16.69', 'username': 'root', 'password': 'root'}, #{'host': '192.168.16.70', 'username': 'root', 'password': 'root'}, #{'host': '192.168.16.76', 'username': 'root', 'password': 'root'}, #{'host': '192.168.16.81', 'username': 'root', 'password': 'root'}, {'host': '192.168.16.82', 'username': 'root', 'password': 'root'}, {'host': '192.168.16.84', 'username': 'root', 'password': 'root'}, ] print(f"🚀 Запуск тестирования {len(routers)} роутеров") print(f"📡 iperf3 сервер: {IPERF_SERVER}:{IPERF_PORT}") print(f"⏱️ Длительность теста: {IPERF_DURATION} сек") print(f"📁 Результаты сохраняются в: {OUTPUT_DIR}") print("=" * 60) # Статистика success_count = 0 fail_count = 0 results = [] # Обрабатываем роутеры последовательно for i, router_config in enumerate(routers, 1): print(f"\n📌 [{i}/{len(routers)}] Обработка роутера {router_config['host']}") tester = RouterTester( router_config['host'], router_config['username'], router_config['password'] ) success = tester.process( IPERF_SERVER, OUTPUT_DIR, duration=IPERF_DURATION, port=IPERF_PORT ) if success: success_count += 1 results.append(f"✅ {router_config['host']}: УСПЕШНО") else: fail_count += 1 results.append(f"❌ {router_config['host']}: ОШИБКА") # Небольшая пауза между роутерами if i < len(routers): print(f"\n⏳ Пауза 3 секунды перед следующим роутером...") time.sleep(3) # Итоговая статистика print("\n" + "=" * 60) print("📊 ИТОГОВАЯ СТАТИСТИКА") print("=" * 60) for result in results: print(result) print("-" * 60) print(f"✅ Успешно: {success_count}") print(f"❌ Ошибок: {fail_count}") print(f"📁 Все результаты сохранены в: {OUTPUT_DIR}") print("=" * 60) if __name__ == '__main__': main()