#!/usr/bin/env python3 """ BGP Prefix Loader Script Загружает префиксы из файла prefixes.txt в BIRD с настройкой community """ import os import sys import subprocess import ipaddress import logging from datetime import datetime from typing import List, Tuple, Optional # Настройка логирования logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('/var/log/bird/prefix_loader.log'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) class BGPPrefixLoader: def __init__(self, prefixes_file: str = '/opt/bgp-server/data/prefixes.txt'): self.prefixes_file = prefixes_file self.birdc_path = '/usr/bin/birdc' self.loaded_prefixes = set() def validate_ip_prefix(self, prefix: str) -> bool: """Проверяет корректность IP префикса""" try: ipaddress.ip_network(prefix, strict=False) return True except ValueError: return False def validate_community(self, community: str) -> bool: """Проверяет корректность community (формат ASN:value)""" try: if ':' not in community: return False asn, value = community.split(':', 1) int(asn) int(value) return True except (ValueError, AttributeError): return False def parse_prefixes_file(self) -> List[Tuple[str, str, str]]: """Парсит файл prefixes.txt и возвращает список префиксов с community""" prefixes = [] if not os.path.exists(self.prefixes_file): logger.error(f"Файл {self.prefixes_file} не найден") return prefixes try: with open(self.prefixes_file, 'r', encoding='utf-8') as f: for line_num, line in enumerate(f, 1): line = line.strip() # Пропускаем пустые строки и комментарии if not line or line.startswith('#'): continue # Парсим строку: prefix community description parts = line.split(None, 2) if len(parts) < 2: logger.warning(f"Строка {line_num}: неверный формат - {line}") continue prefix, community = parts[0], parts[1] description = parts[2] if len(parts) > 2 else "" # Валидация if not self.validate_ip_prefix(prefix): logger.warning(f"Строка {line_num}: неверный префикс - {prefix}") continue if not self.validate_community(community): logger.warning(f"Строка {line_num}: неверный community - {community}") continue prefixes.append((prefix, community, description)) logger.info(f"Добавлен префикс: {prefix} с community {community} - {description}") except Exception as e: logger.error(f"Ошибка при чтении файла: {e}") return prefixes def execute_birdc_command(self, command: str) -> Tuple[bool, str]: """Выполняет команду через birdc""" try: result = subprocess.run( [self.birdc_path] + command.split(), capture_output=True, text=True, timeout=30 ) if result.returncode == 0: return True, result.stdout else: return False, result.stderr except subprocess.TimeoutExpired: return False, "Timeout при выполнении команды" except Exception as e: return False, str(e) def add_static_route(self, prefix: str, community: str) -> bool: """Добавляет статический маршрут в BIRD""" # Команда для добавления статического маршрута command = f"configure 'route {prefix} via 127.0.0.1;'" success, output = self.execute_birdc_command(command) if success: logger.info(f"Добавлен маршрут: {prefix}") self.loaded_prefixes.add(prefix) else: logger.error(f"Ошибка добавления маршрута {prefix}: {output}") return success def remove_static_route(self, prefix: str) -> bool: """Удаляет статический маршрут из BIRD""" command = f"configure 'route {prefix} via 127.0.0.1;'" success, output = self.execute_birdc_command(command) if success: logger.info(f"Удален маршрут: {prefix}") self.loaded_prefixes.discard(prefix) else: logger.error(f"Ошибка удаления маршрута {prefix}: {output}") return success def get_current_routes(self) -> set: """Получает текущие статические маршруты из BIRD""" success, output = self.execute_birdc_command("show route table bgp_table") if not success: logger.error(f"Ошибка получения маршрутов: {output}") return set() routes = set() for line in output.split('\n'): if 'via' in line and 'static' in line: # Парсим префикс из строки parts = line.split() if len(parts) > 0: prefix = parts[0] routes.add(prefix) return routes def reload_configuration(self) -> bool: """Перезагружает конфигурацию BIRD""" success, output = self.execute_birdc_command("configure") if success: logger.info("Конфигурация BIRD перезагружена") else: logger.error(f"Ошибка перезагрузки конфигурации: {output}") return success def load_prefixes(self, force_reload: bool = False) -> bool: """Основная функция загрузки префиксов""" logger.info("Начинаем загрузку префиксов...") # Получаем текущие маршруты current_routes = self.get_current_routes() logger.info(f"Текущих маршрутов: {len(current_routes)}") # Парсим файл с префиксами prefixes = self.parse_prefixes_file() logger.info(f"Найдено префиксов в файле: {len(prefixes)}") if not prefixes: logger.warning("Нет префиксов для загрузки") return False # Создаем временный конфигурационный файл temp_config = "/tmp/bird_static_routes.conf" try: with open(temp_config, 'w') as f: f.write("# Автоматически сгенерированные статические маршруты\n") f.write("# Время создания: {}\n\n".format(datetime.now())) for prefix, community, description in prefixes: f.write(f"route {prefix} via 127.0.0.1; # {community} - {description}\n") # Применяем конфигурацию if force_reload: success = self.reload_configuration() else: success = True if success: logger.info(f"Успешно загружено {len(prefixes)} префиксов") return True else: logger.error("Ошибка при применении конфигурации") return False except Exception as e: logger.error(f"Ошибка при создании конфигурации: {e}") return False finally: # Удаляем временный файл if os.path.exists(temp_config): os.remove(temp_config) def show_statistics(self): """Показывает статистику загруженных префиксов""" current_routes = self.get_current_routes() print(f"\n=== Статистика BGP сервера ===") print(f"Всего загружено префиксов: {len(current_routes)}") print(f"Время: {datetime.now()}") if current_routes: print("\nЗагруженные префиксы:") for prefix in sorted(current_routes): print(f" - {prefix}") def main(): """Основная функция""" import argparse parser = argparse.ArgumentParser(description='BGP Prefix Loader') parser.add_argument('--file', '-f', default='/opt/bgp-server/data/prefixes.txt', help='Путь к файлу с префиксами') parser.add_argument('--force', action='store_true', help='Принудительная перезагрузка конфигурации') parser.add_argument('--stats', action='store_true', help='Показать статистику') args = parser.parse_args() loader = BGPPrefixLoader(args.file) if args.stats: loader.show_statistics() else: success = loader.load_prefixes(args.force) if success: loader.show_statistics() sys.exit(0) else: sys.exit(1) if __name__ == "__main__": main()