#!/usr/bin/env python3 """ Скрипт для загрузки префиксов из текстовых файлов в FRR """ import sys import os import time import subprocess import ipaddress from pathlib import Path from typing import List, Set # Настройки DATA_DIR = "/app/data" LOG_FILE = "/app/logs/prefixes.log" PREFIX_FILES = ["prefixes.txt", "additional_prefixes.txt"] def log_message(message: str): """Запись сообщения в лог""" timestamp = time.strftime("%Y-%m-%d %H:%M:%S") log_entry = f"[{timestamp}] {message}" # Вывод в stdout print(log_entry, flush=True) # Запись в файл (с обработкой ошибок) try: with open(LOG_FILE, "a") as f: f.write(log_entry + "\n") except (PermissionError, OSError) as e: # Если не можем записать в файл, просто выводим в stderr print(f"Warning: Could not write to log file: {e}", file=sys.stderr, flush=True) def log_error(message: str): """Запись ошибки в лог""" log_message(f"ERROR: {message}") def validate_prefix(prefix: str) -> bool: """Валидация IP префикса""" try: ipaddress.ip_network(prefix, strict=False) return True except ValueError: return False def load_prefixes_from_file(file_path: Path) -> Set[str]: """Загрузка префиксов из файла""" prefixes = set() if not file_path.exists(): log_message(f"File not found: {file_path}") return prefixes try: with open(file_path, 'r') as f: for line_num, line in enumerate(f, 1): line = line.strip() # Пропуск пустых строк и комментариев if not line or line.startswith('#'): continue # Валидация префикса if validate_prefix(line): prefixes.add(line) else: log_message(f"Invalid prefix in {file_path}:{line_num}: {line}") except Exception as e: log_message(f"Error reading {file_path}: {e}") return prefixes def load_all_prefixes() -> Set[str]: """Загрузка всех префиксов из всех файлов""" all_prefixes = set() for filename in PREFIX_FILES: file_path = Path(DATA_DIR) / filename prefixes = load_prefixes_from_file(file_path) all_prefixes.update(prefixes) log_message(f"Loaded {len(prefixes)} prefixes from {filename}") return all_prefixes def run_vtysh_command(command: str) -> bool: """Выполнение команды через vtysh""" try: result = subprocess.run( ['vtysh', '-c', command], capture_output=True, text=True, timeout=30 ) if result.returncode != 0: log_error(f"vtysh command failed: {command}") log_error(f"stderr: {result.stderr}") return False return True except subprocess.TimeoutExpired: log_error(f"vtysh command timeout: {command}") return False except Exception as e: log_error(f"Error running vtysh command '{command}': {e}") return False def announce_prefixes(prefixes: Set[str]): """Анонсирование префиксов через FRR""" log_message(f"Announcing {len(prefixes)} prefixes...") for prefix in sorted(prefixes): # Добавление статического маршрута в FRR command = f"ip route {prefix} Null0" if run_vtysh_command(command): log_message(f"Announced prefix: {prefix}") else: log_error(f"Failed to announce prefix: {prefix}") def withdraw_prefixes(prefixes: Set[str]): """Отзыв префиксов через FRR""" log_message(f"Withdrawing {len(prefixes)} prefixes...") for prefix in sorted(prefixes): # Удаление статического маршрута из FRR command = f"no ip route {prefix} Null0" if run_vtysh_command(command): log_message(f"Withdrew prefix: {prefix}") else: log_error(f"Failed to withdraw prefix: {prefix}") def check_frr_status() -> bool: """Проверка статуса FRR""" try: result = subprocess.run( ['vtysh', '-c', 'show version'], capture_output=True, text=True, timeout=10 ) return result.returncode == 0 except Exception: return False def wait_for_frr(): """Ожидание запуска FRR""" log_message("Waiting for FRR to start...") max_attempts = 30 attempt = 0 while attempt < max_attempts: if check_frr_status(): log_message("FRR is ready") return True attempt += 1 time.sleep(2) if attempt % 5 == 0: log_message(f"Still waiting for FRR... (attempt {attempt}/{max_attempts})") log_error("FRR did not start within expected time") return False def main(): """Основная функция""" # Создание директории для логов в начале try: os.makedirs(os.path.dirname(LOG_FILE), exist_ok=True) except (PermissionError, OSError): pass # Игнорируем ошибки создания директории # Проверка тестового режима if len(sys.argv) > 1 and sys.argv[1] == '--test': print("Testing prefix loader...") prefixes = load_all_prefixes() print(f"Found {len(prefixes)} prefixes: {list(prefixes)}") return # Проверка режима отзыва if len(sys.argv) > 1 and sys.argv[1] == '--withdraw': log_message("Withdraw mode: loading and withdrawing all prefixes...") prefixes = load_all_prefixes() if wait_for_frr(): withdraw_prefixes(prefixes) return log_message("Starting prefix loader...") # Ожидание запуска FRR if not wait_for_frr(): log_error("Cannot proceed without FRR") return # Загрузка префиксов prefixes = load_all_prefixes() log_message(f"Total prefixes loaded: {len(prefixes)}") if not prefixes: log_message("No valid prefixes found. Exiting.") return # Анонсирование префиксов announce_prefixes(prefixes) log_message("All prefixes announced successfully") # Мониторинг изменений в файлах log_message("Starting file monitoring...") try: while True: time.sleep(30) # Проверка каждые 30 секунд # Перезагрузка префиксов new_prefixes = load_all_prefixes() # Определение изменений added = new_prefixes - prefixes removed = prefixes - new_prefixes if added: log_message(f"Adding {len(added)} new prefixes: {list(added)}") announce_prefixes(added) if removed: log_message(f"Removing {len(removed)} prefixes: {list(removed)}") withdraw_prefixes(removed) if added or removed: prefixes = new_prefixes log_message(f"Updated prefix count: {len(prefixes)}") except KeyboardInterrupt: log_message("Received interrupt signal") except Exception as e: log_message(f"Error in main loop: {e}") finally: # Отзыв всех префиксов при завершении log_message("Withdrawing all prefixes...") withdraw_prefixes(prefixes) log_message("Prefix loader stopped") if __name__ == "__main__": main()