main resolver

bin/resolve.py

@@ -0,0 +1,171 @@
+#!/usr/bin/env python3
+import os
+import re
+import subprocess
+import ipaddress
+from pathlib import Path
+
+# Конфигурация путей
+SRC_DIR = Path("src")
+RESOLVE_DIR = Path("resolve")
+
+# Регулярное выражение для имени файла: {community}.{name}.list
+FILE_PATTERN = re.compile(r'^(\d+)\.(\w+)\.list$')
+
+def ensure_dirs():
+    """Создает директорию resolve, если её нет."""
+    RESOLVE_DIR.mkdir(parents=True, exist_ok=True)
+
+def determine_type(line: str):
+    """Определяет тип записи в строке."""
+    line = line.strip()
+    if not line or line.startswith('#'):
+        return None, None
+
+    # 1. Ссылка (URL)
+    if line.startswith('http://') or line.startswith('https://'):
+        return 'url', line
+
+    # 2. Автономная система (AS)
+    if re.match(r'^AS[\d\w-]+$', line, re.IGNORECASE):
+        return 'as', line
+
+    # 3. CIDR (проверка через встроенную библиотеку ipaddress)
+    if '/' in line:
+        try:
+            # strict=False позволяет принимать 1.2.3.4/24 и нормализовать до 1.2.3.0/24
+            ipaddress.ip_network(line, strict=False)
+            return 'cidr', line
+        except ValueError:
+            pass
+
+    # 4. IP-адрес
+    try:
+        ipaddress.ip_address(line)
+        return 'ip', line
+    except ValueError:
+        pass
+
+    # 5. Домен (остальное, что похоже на доменное имя)
+    if re.match(r'^[a-zA-Z0-9]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?(\.[a-zA-Z0-9]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?)*\.[a-zA-Z]{2,}$', line):
+        return 'domain', line
+
+    return 'unknown', line
+
+def process_file(filepath: Path, community: str, name: str):
+    """Обрабатывает один файл списка."""
+    output_lines = []
+    
+    # Шаблон для итоговой записи (f-string, поэтому {{ превращается в {)
+    # target_format = f"route {{cidr}} blackhole {{ bgp_community.add((64888, {community})); }};"
+    target_format = "route {cidr} blackhole {{ bgp_community.add((64888, {community})); }};"
+    
+    with open(filepath, 'r', encoding='utf-8') as f:
+        for raw_line in f:
+            entry_type, value = determine_type(raw_line)
+            
+            if entry_type == 'url':
+                try:
+                    result = subprocess.run(
+                        ['curl', '-s', '-X', 'GET', value],
+                        capture_output=True, text=True, timeout=15
+                    )
+                    for res_line in result.stdout.splitlines():
+                        res_line = res_line.strip()
+                        if not res_line or res_line.startswith('#'):
+                            continue
+                        try:
+                            # Пытаемся распарсить строку как CIDR
+                            normalized_cidr = str(ipaddress.ip_network(res_line, strict=False))
+                            output_lines.append(target_format.format(cidr=normalized_cidr, community=community))
+                        except ValueError:
+                            print(f"[WARN] Не-CIDR строка в ответе от {value}: '{res_line}'")
+                except Exception as e:
+                    print(f"[WARN] Ошибка curl для {value}: {e}")
+
+            elif entry_type == 'as':
+                # f-string, {{ превращается в {
+                bgpq4_format = f"route %n/%l via blackhole {{ bgp_community.add((64888, {community})); }};\n"
+                try:
+                    result = subprocess.run(
+                        ['bgpq4', '-A', '-F', bgpq4_format, value],
+                        capture_output=True, text=True, timeout=30
+                    )
+                    if result.returncode == 0:
+                        for res_line in result.stdout.splitlines():
+                            if res_line.strip():
+                                output_lines.append(res_line.strip())
+                    else:
+                        print(f"[WARN] bgpq4 вернул ошибку для {value}: {result.stderr.strip()}")
+                except FileNotFoundError:
+                    print("[ERROR] Утилита bgpq4 не найдена. Установите её (например, apt install bgpq4).")
+                except Exception as e:
+                    print(f"[WARN] Ошибка выполнения bgpq4 для {value}: {e}")
+
+            elif entry_type == 'domain':
+                try:
+                    result = subprocess.run(
+                        ['dig', '+short', 'A', value],
+                        capture_output=True, text=True, timeout=10
+                    )
+                    # dig может вернуть несколько IP через пробел или перенос строки
+                    ips = result.stdout.split()
+                    for ip in ips:
+                        try:
+                            # Проверяем, что это действительно IP (а не CNAME или ошибка)
+                            ipaddress.ip_address(ip)
+                            cidr = f"{ip}/32"
+                            output_lines.append(target_format.format(cidr=cidr, community=community))
+                        except ValueError:
+                            pass  # Игнорируем не-IP записи (например, CNAME)
+                except Exception as e:
+                    print(f"[WARN] Ошибка dig для {value}: {e}")
+
+            elif entry_type == 'ip':
+                cidr = f"{value}/32"
+                output_lines.append(target_format.format(cidr=cidr, community=community))
+
+            elif entry_type == 'cidr':
+                # Нормализуем CIDR (например, 192.168.1.5/24 -> 192.168.1.0/24)
+                normalized_cidr = str(ipaddress.ip_network(value, strict=False))
+                # output_lines.append(target_format.format(cidr=normalized_cidr))
+                output_lines.append(target_format.format(cidr=normalized_cidr, community=community))
+
+            elif entry_type == 'unknown':
+                print(f"[WARN] Неизвестный или некорректный формат строки: '{value}' в файле {filepath.name}")
+
+    # Удаляем дубликаты, сохраняя порядок строк
+    unique_lines = list(dict.fromkeys(output_lines))
+
+    # Записываем результат
+    out_filepath = RESOLVE_DIR / f"{community}.{name}.routes"
+    with open(out_filepath, 'w', encoding='utf-8') as f:
+        for out_line in unique_lines:
+            f.write(out_line + "\n")
+    
+    print(f"[OK] Обработан: {filepath.name} -> {out_filepath} ({len(unique_lines)} записей)")
+
+def main():
+    ensure_dirs()
+    
+    if not SRC_DIR.exists():
+        print(f"[ERROR] Директория '{SRC_DIR}' не найдена.")
+        return
+
+    processed_count = 0
+    for filename in os.listdir(SRC_DIR):
+        match = FILE_PATTERN.match(filename)
+        if match:
+            community = match.group(1)
+            name = match.group(2)
+            filepath = SRC_DIR / filename
+            process_file(filepath, community, name)
+            processed_count += 1
+
+    if processed_count == 0:
+        print(f"[INFO] В директории '{SRC_DIR}' не найдено файлов, соответствующих шаблону {{цифры}}.{{слово}}.list")
+    else:
+        print(f"\n[DONE] Успешно обработано файлов: {processed_count}")
+
+if __name__ == '__main__':
+    main()