bgp-server/scripts/reload_prefixes.py

#!/usr/bin/env python3
"""
Скрипт для перезагрузки префиксов в FRR
"""

import sys
import os
import time
import subprocess
import ipaddress
from pathlib import Path
from typing import Set

# Настройки
DATA_DIR = "/app/data"
LOG_FILE = "/app/logs/reload.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}"
    
    print(log_entry, flush=True)
    
    try:
        with open(LOG_FILE, "a") as f:
            f.write(log_entry + "\n")
    except (PermissionError, OSError):
        pass

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 get_current_routes() -> Set[str]:
    """Получение текущих статических маршрутов из FRR"""
    try:
        result = subprocess.run(
            ['vtysh', '-c', 'show ip route static'],
            capture_output=True,
            text=True,
            timeout=30
        )
        
        if result.returncode != 0:
            log_message("Could not get current routes from FRR")
            return set()
        
        routes = set()
        for line in result.stdout.split('\n'):
            line = line.strip()
            if line and 'via Null0' in line:
                # Парсинг строки маршрута
                parts = line.split()
                if len(parts) >= 2:
                    route = parts[1]
                    if validate_prefix(route):
                        routes.add(route)
        
        return routes
    except Exception as e:
        log_message(f"Error getting current routes: {e}")
        return set()

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_message(f"vtysh command failed: {command}")
            return False
        
        return True
    except Exception as e:
        log_message(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):
        command = f"ip route {prefix} Null0"
        if run_vtysh_command(command):
            log_message(f"Announced prefix: {prefix}")
        else:
            log_message(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):
        command = f"no ip route {prefix} Null0"
        if run_vtysh_command(command):
            log_message(f"Withdrew prefix: {prefix}")
        else:
            log_message(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 main():
    """Основная функция"""
    log_message("Starting prefix reload...")
    
    # Проверка статуса FRR
    if not check_frr_status():
        log_message("ERROR: FRR is not running")
        return 1
    
    # Загрузка новых префиксов
    new_prefixes = load_all_prefixes()
    log_message(f"Loaded {len(new_prefixes)} prefixes from files")
    
    # Получение текущих маршрутов
    current_routes = get_current_routes()
    log_message(f"Found {len(current_routes)} current routes in FRR")
    
    # Определение изменений
    to_add = new_prefixes - current_routes
    to_remove = current_routes - new_prefixes
    
    log_message(f"Changes: {len(to_add)} to add, {len(to_remove)} to remove")
    
    # Применение изменений
    if to_remove:
        withdraw_prefixes(to_remove)
    
    if to_add:
        announce_prefixes(to_add)
    
    # Финальная проверка
    final_routes = get_current_routes()
    log_message(f"Final route count: {len(final_routes)}")
    
    if len(final_routes) == len(new_prefixes):
        log_message("Reload completed successfully!")
        return 0
    else:
        log_message("WARNING: Route count mismatch after reload")
        return 1

if __name__ == "__main__":
    sys.exit(main())