264 lines
10 KiB
Python
264 lines
10 KiB
Python
#!/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() |