第336篇：BGP 路由黑洞导致业务中断分析

关键词：BGP路由黑洞、路由泄露、默认路由、iBGP全互联、路由反射、华为CE/NE系列

一、案例背景

某省级运营商的城域网在某天凌晨2:15发生大规模用户断网故障，影响用户数超过15万。故障持续约45分钟后恢复。故障现象表现为：大量家庭宽带用户无法访问互联网，但本地业务（如IPTV、局域网访问）正常。

网络架构： - 城域核心：4台NE8000（AS 65100），互联至国家骨干网 - 城域汇聚：16台NE40E（AS 65100），作为路由反射器Client - 城域接入：120台MA5800（AS 65100），PPPoE汇聚 - 互联网出口：通过2个运营商（电信、联通）互联

故障时间线： - 02:15：省公司网管告警，大量用户无法上网 - 02:18：运维人员收到告警通知 - 02:25：定位到核心层BGP路由异常 - 02:35：开始排查具体原因 - 03:00：找到根因并实施修复 - 03:15：业务逐步恢复

二、需求分析与方案设计

2.1 BGP路由黑洞原理

BGP路由黑洞（Black Hole）是指数据包到达某台路由器后，该路由器没有有效的转发路径，导致数据包被丢弃。在BGP网络中，路由黑洞通常由以下原因引起：

2.2 故障网络架构

2.3 故障根因

经深入排查，确定故障根因为：核心设备CE1上的一条BGP策略配置错误，导致0.0.0.0/0默认路由被错误地标记为"Blackhole"。

具体过程： 1. 运维人员在凌晨进行BGP策略优化，在CE1上新增了一条AS Path Filter。 2. 该Filter的配置语法有误，导致匹配范围扩大，意外匹配了默认路由0.0.0.0/0。 3. 被匹配的默认路由被标记为Blackhole（BGP Next-Hop为0.0.0.0）。 4. CE1通过iBGP将Blackhole默认路由通告给RR1，RR1再通告给所有RR Client。 5. 由于BGP选路规则中，Blackhole路由的优先级高于无路由，导致大量设备选用了这条无效路由。

三、技术细节与实施过程

3.1 故障排查过程

3.2 故障设备上的关键命令

# 查看默认路由状态（故障时）
<NE8000-Core-A> display bgp routing-table 0.0.0.0 0.0.0.0
BGP Local Router ID is 1.1.1.1
Status: * - valid, > - best, d - damped, h - history, i - internal,
        s - Stale, S - suppressed, r - reroute, b - backup, L - lbg

Network          NextHop         MED      LocPrf    PrefVal   Path/Ogn
*> 0.0.0.0/0      0.0.0.0         0        100       0        ?i  <-- Blackhole!

# 查看详细属性
<NE8000-Core-A> display bgp routing-table verbose 0.0.0.0 0.0.0.0
BGP routing table entry information of 0.0.0.0/0:
  Originator: 1.1.1.1
  NextHop: 0.0.0.0  <-- 无效的Next-Hop
  MED: 0
  Local-Pref: 100
  Community: noval
  Path: i
  Valid: Yes
  Preferenced Value: 0
  NextHop's Reachability: Inaccessible  <-- 不可达

# 查看当前BGP策略
<NE8000-Core-A> display current-configuration configuration bgp
# 发现错误配置
route-policy INTERNET_FILTER deny 10
 if-match as-path-filter 10
 apply next-hop 0.0.0.0  <-- 将匹配的路由标记为Blackhole

# AS Path Filter 10 配置
ip as-path-filter 10 _64512_  <-- 意图是过滤AS 64512的路由
# 但由于配置语法问题，该Filter意外匹配了默认路由

3.3 修复过程

# 步骤1：移除错误的BGP策略
<NE8000-Core-A> system-view
[NE8000-Core-A] bgp 65100
[NE8000-Core-A-bgp-65100] undo route-policy INTERNET_FILTER import
[NE8000-Core-A-bgp-65100] quit

# 步骤2：清除BGP路由表，强制重新收敛（建议使用精确重置，避免影响其他BGP业务）
[NE8000-Core-A] reset bgp ipvpn
# 或针对特定对等体重置：reset bgp peer x.x.x.x

# 步骤3：修正AS Path Filter配置
[NE8000-Core-A] ip as-path-filter 10 _64512_
[NE8000-Core-A] route-policy CORRECT_FILTER deny 10
[NE8000-Core-A-route-policy-CORRECT_FILTER] if-match as-path-filter 10
[NE8000-Core-A-route-policy-CORRECT_FILTER] quit
[NE8000-Core-A] route-policy CORRECT_FILTER permit 20
[NE8000-Core-A-route-policy-CORRECT_FILTER] quit

# 步骤4：验证默认路由恢复
<NE8000-Core-A> display bgp routing-table 0.0.0.0 0.0.0.0
*> 0.0.0.0/0      202.97.0.1      0        100       0        4134?i  <-- 正确路由
*> 0.0.0.0/0      202.98.0.1      0        100       0        4837?i  <-- 正确路由

3.4 BGP路由黑洞的预防机制

四、实战配置/案例分析

4.1 正确的BGP默认路由发布配置

# NE8000核心设备正确配置模板
<Huawei> system-view
[Huawei] sysname NE8000-Core

# 引入静态默认路由到BGP
[NE8000-Core] ip route-static 0.0.0.0 0.0.0.0 202.97.0.1 preference 5
[NE8000-Core] ip route-static 0.0.0.0 0.0.0.0 202.98.0.1 preference 5

# BGP配置
[NE8000-Core] bgp 65100
[NE8000-Core-bgp-65100] router-id 1.1.1.1

# eBGP对等体（互联网出口）
[NE8000-Core-bgp-65100] peer 202.97.0.1 as-number 4134
[NE8000-Core-bgp-65100] peer 202.97.0.1 connect-interface GigabitEthernet 0/0/1
[NE8000-Core-bgp-65100] peer 202.98.0.1 as-number 4837
[NE8000-Core-bgp-65100] peer 202.98.0.1 connect-interface GigabitEthernet 0/0/2

# iBGP对等体（RR）
[NE8000-Core-bgp-65100] peer 10.0.0.1 as-number 65100
[NE8000-Core-bgp-65100] peer 10.0.0.1 connect-interface LoopBack 0
[NE8000-Core-bgp-65100] peer 10.0.0.1 route-reflector-client

# 引入静态路由到BGP（默认路由）
[NE8000-Core-bgp-65100] import-route static route-policy DEFAULT_ROUTE
[NE8000-Core-bgp-65100] quit

# 路由策略：仅允许默认路由被引入
[NE8000-Core] route-policy DEFAULT_ROUTE permit node 10
[NE8000-Core-route-policy-DEFAULT_ROUTE] if-match ip-address prefix-list DEFAULT_ONLY
[NE8000-Core-route-policy-DEFAULT_ROUTE] quit

# Prefix-List：仅匹配0.0.0.0/0
[NE8000-Core] ip prefix-list DEFAULT_ONLY permit 0.0.0.0 0 le 0

4.2 路由反射器配置与最佳实践

# 路由反射器配置
<NE40E-RR-01> system-view
[NE40E-RR-01] bgp 65100
[NE40E-RR-01-bgp-65100] router-id 10.0.0.1

# 与核心的iBGP会话（非Client）
[NE40E-RR-01-bgp-65100] peer 1.1.1.1 as-number 65100
[NE40E-RR-01-bgp-65100] peer 1.1.1.1 connect-interface LoopBack 0
[NE40E-RR-01-bgp-65100] peer 2.2.2.2 as-number 65100
[NE40E-RR-01-bgp-65100] peer 2.2.2.2 connect-interface LoopBack 0

# RR Client配置
[NE40E-RR-01-bgp-65100] peer 10.0.1.0 as-number 65100
[NE40E-RR-01-bgp-65100] peer 10.0.1.0 connect-interface LoopBack 0
[NE40E-RR-01-bgp-65100] peer 10.0.1.0 route-reflector-client
# ... 更多Client

# 最佳实践：配置Route-Map过滤不必要的路由
[NE40E-RR-01-bgp-65100] peer 10.0.1.0 route-policy CLIENT_FILTER in
[NE40E-RR-01-bgp-65100] peer 10.0.1.0 route-policy CLIENT_FILTER out

4.3 案例分析：如何避免类似故障

根本原因回顾： - 直接原因：AS Path Filter配置错误，意外匹配默认路由 - 深层原因：缺乏配置变更前验证机制、缺乏变更灰度发布流程

改进措施：

1. 配置变更Pre-Check机制

# 变更前使用dry-run模拟影响范围
<NE8000-Core-A> display bgp routing-table all | include "0.0.0.0"
# 记录变更前默认路由状态

# 应用新策略前预览匹配结果
<NE8000-Core-A> display bgp routing-table all | filter as-path-filter 10
# 检查哪些路由会被该Filter匹配

1. 变更灰度发布流程

1. Blackhole路由检测脚本

# 定期检测Blackhole路由的监控脚本
display bgp routing-table all | include "0.0.0.0.*0.0.0.0"
# 如果发现默认路由的Next-Hop为0.0.0.0，立即告警

4.4 BGP Blackhole路由产生的常见场景

4.5 BGP路由健康检查脚本

#!/usr/bin/env python3
# bgp_health_check.py - BGP路由健康检查脚本
import paramiko
import re
import json
from datetime import datetime

def check_bgp_blackhole(hostname, username, password):
    """检测BGP路由表中的Blackhole路由"""
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    ssh.connect(hostname, username=username, password=password)

    # 获取BGP路由表
    stdin, stdout, stderr = ssh.exec_command(
        "display bgp routing-table all"
    )
    bgp_table = stdout.read().decode()

    # 获取BGP Peer状态
    stdin, stdout, stderr = ssh.exec_command(
        "display bgp peer all"
    )
    peer_status = stdout.read().decode()

    # 检测Blackhole路由（Next-Hop为0.0.0.0）
    blackhole_routes = []
    for line in bgp_table.split('\n'):
        if '0.0.0.0' in line and '*>' in line:
            # 提取路由前缀
            parts = line.split()
            if len(parts) >= 2:
                blackhole_routes.append({
                    "prefix": parts[1],
                    "nexthop": "0.0.0.0",
                    "status": "BLACKHOLE"
                })

    # 统计Peer状态
    established_count = len(re.findall(r'Established', peer_status))
    total_peers = len(re.findall(r'Peer', peer_status))

    ssh.close()

    result = {
        "timestamp": datetime.now().isoformat(),
        "device": hostname,
        "blackhole_routes": blackhole_routes,
        "peer_status": {
            "total": total_peers,
            "established": established_count,
            "down": total_peers - established_count
        },
        "alert_level": "CRITICAL" if blackhole_routes else "OK"
    }

    return result

# 检查所有核心设备
devices = [
    ("10.0.1.1", "admin", "password"),
    ("10.0.1.2", "admin", "password"),
    ("10.0.1.3", "admin", "password"),
    ("10.0.1.4", "admin", "password"),
]

alerts = []
for host, user, pwd in devices:
    result = check_bgp_blackhole(host, user, pwd)
    if result["alert_level"] == "CRITICAL":
        alerts.append(result)
        print(f"🚨 {result['device']}: 发现{len(result['blackhole_routes'])}条Blackhole路由")
        for route in result['blackhole_routes']:
            print(f"   - {route['prefix']} (Next-Hop: {route['nexthop']})")

if alerts:
    print(f"\n⚠ 共发现{len(alerts)}台设备存在Blackhole路由，请立即处理！")
else:
    print("✓ 所有设备BGP路由表正常")

4.6 BGP变更安全规范

4.7 Python实战：BGP路由黑洞自动化检测与修复

4.7.1 BlackholeDetector：路由黑洞自动检测

#!/usr/bin/env python3
"""
BGP路由黑洞自动检测器
扫描BGP路由表，识别黑洞路由并生成检测报告
"""
import paramiko
import re
import json
from dataclasses import dataclass, field
from typing import Dict, List, Optional
from datetime import datetime


@dataclass
class BlackholeRoute:
    """黑洞路由"""
    prefix: str              # 路由前缀
    nexthop: str             # 下一跳（0.0.0.0表示黑洞）
    origin_as: str           # 来源AS
    peer: str                # 对等体
    med: int = 0
    local_pref: int = 100
    path: str = ""
    is_best: bool = False


@dataclass
class DeviceScanResult:
    """设备扫描结果"""
    device_ip: str
    device_name: str
    scan_time: str
    blackhole_routes: List[BlackholeRoute] = field(default_factory=list)
    total_bgp_routes: int = 0
    invalid_routes: int = 0
    peer_status: Dict = field(default_factory=dict)


class BlackholeDetector:
    """BGP路由黑洞检测器"""

    def __init__(self, devices: List[Dict]):
        """
        :param devices: 设备列表，每项含host, username, password, name
        """
        self.devices = devices
        self.results: List[DeviceScanResult] = []

    def scan_device(self, device: Dict) -> DeviceScanResult:
        """扫描单台设备"""
        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        ssh.connect(device["host"], username=device["username"],
                   password=device["password"], timeout=15)

        result = DeviceScanResult(
            device_ip=device["host"],
            device_name=device.get("name", device["host"]),
            scan_time=datetime.now().isoformat()
        )

        # 1. 获取BGP路由表
        stdout = self._exec(ssh, "display bgp routing-table all")
        result.total_bgp_routes = self._count_routes(stdout)

        # 2. 检测黑洞路由（Next-Hop为0.0.0.0的最优路由）
        result.blackhole_routes = self._find_blackhole_routes(stdout)

        # 3. 统计Invalid路由
        result.invalid_routes = stdout.count("Inaccessible")

        # 4. 获取Peer状态
        peer_output = self._exec(ssh, "display bgp peer all")
        result.peer_status = self._parse_peer_status(peer_output)

        ssh.close()
        self.results.append(result)
        return result

    def scan_all(self) -> List[DeviceScanResult]:
        """扫描所有设备"""
        for device in self.devices:
            print(f"[*] 扫描设备: {device.get('name', device['host'])}...")
            self.scan_device(device)
        return self.results

    def generate_report(self) -> str:
        """生成检测报告"""
        lines = []
        lines.append("=" * 65)
        lines.append("BGP路由黑洞检测报告")
        lines.append(f"扫描时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        lines.append("=" * 65)

        total_bh = sum(len(r.blackhole_routes) for r in self.results)
        lines.append(f"扫描设备数: {len(self.results)}")
        lines.append(f"发现黑洞路由总数: {total_bh}")
        lines.append(f"告警级别: {'🔴 CRITICAL' if total_bh > 0 else '🟢 OK'}")

        for r in self.results:
            bh_count = len(r.blackhole_routes)
            if bh_count > 0:
                lines.append(f"\n{'!' * 65}")
                lines.append(f"⚠ 设备: {r.device_name} ({r.device_ip})")
                lines.append(f"  黑洞路由数: {bh_count}")
                lines.append(f"  总BGP路由: {r.total_bgp_routes}")
                lines.append(f"  Invalid路由: {r.invalid_routes}")

                # Peer状态
                if r.peer_status:
                    lines.append(f"  Peer状态: {r.peer_status['established']}/{r.peer_status['total']} Established")

                lines.append(f"\n  黑洞路由明细:")
                for i, route in enumerate(r.blackhole_routes, 1):
                    lines.append(f"    {i}. {route.prefix}  NH={route.nexthop}  "
                               f"AS_PATH={route.path}  Peer={route.peer}")

        if total_bh == 0:
            lines.append("\n✓ 所有设备BGP路由表正常，未发现黑洞路由")

        return "\n".join(lines)

    def _exec(self, ssh, command: str) -> str:
        stdin, stdout, stderr = ssh.exec_command(command, timeout=30)
        return stdout.read().decode().strip()

    def _count_routes(self, output: str) -> int:
        match = re.search(r"Routes\s*:\s*(\d+)", output)
        return int(match.group(1)) if match else 0

    def _find_blackhole_routes(self, output: str) -> List[BlackholeRoute]:
        routes = []
        for line in output.split("\n"):
            # 匹配格式: *> prefix  nexthop  MED  LocPrf  PrefVal  Path
            if "*>" in line:
                parts = line.split()
                if len(parts) >= 6:
                    try:
                        prefix = parts[1]
                        nexthop = parts[2]
                        if nexthop == "0.0.0.0":
                            route = BlackholeRoute(
                                prefix=prefix,
                                nexthop=nexthop,
                                med=int(parts[3]) if parts[3].isdigit() else 0,
                                local_pref=int(parts[4]) if parts[4].isdigit() else 100,
                                path=parts[-1] if parts else "",
                                is_best=True
                            )
                            routes.append(route)
                    except (ValueError, IndexError):
                        continue
        return routes

    def _parse_peer_status(self, output: str) -> Dict:
        total = len(re.findall(r"\d+\.\d+\.\d+\.\d+", output)) // 2  # 粗略估计
        established = output.count("Established")
        idle = output.count("Idle")
        active = output.count("Active")
        return {
            "total": max(total, established + idle + active),
            "established": established,
            "idle": idle,
            "active": active,
            "down": max(0, total - established)
        }


if __name__ == "__main__":
    devices = [
        {"host": "10.0.1.1", "username": "admin", "password": "pass", "name": "NE8000-Core-A"},
        {"host": "10.0.1.2", "username": "admin", "password": "pass", "name": "NE8000-Core-B"},
    ]
    detector = BlackholeDetector(devices)
    detector.scan_all()
    print(detector.generate_report())

4.7.2 BGPHealthMonitor：BGP健康监控

#!/usr/bin/env python3
"""
BGP健康监控器
采集基线、检测异常、检查RR客户端一致性
"""
import paramiko
import re
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Set
from datetime import datetime
from collections import defaultdict


@dataclass
class PeerBaseline:
    """Peer基线"""
    peer_ip: str
    remote_as: int
    state: str
    msg_rcvd: int
    msg_sent: int
    uptime: str
    prefix_received: int


@dataclass
class HealthCheckResult:
    """健康检查结果"""
    device: str
    timestamp: str
    anomalies: List[Dict] = field(default_factory=list)
    peer_changes: List[Dict] = field(default_factory=list)
    rr_inconsistencies: List[Dict] = field(default_factory=list)
    score: float = 100.0  # 健康评分


class BGPHealthMonitor:
    """BGP健康监控器"""

    def __init__(self, devices: List[Dict], rr_peers: Dict[str, List[str]] = None):
        """
        :param devices: 设备列表
        :param rr_peers: RR与其Client的映射 {rr_ip: [client_ip1, client_ip2, ...]}
        """
        self.devices = devices
        self.rr_peers = rr_peers or {}
        self.baseline: Dict[str, Dict[str, PeerBaseline]] = {}  # {device: {peer: baseline}}

    def collect_baseline(self, device: Dict) -> Dict[str, PeerBaseline]:
        """
        采集单台设备的BGP Peer基线
        :return: {peer_ip: PeerBaseline}
        """
        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        ssh.connect(device["host"], username=device["username"],
                   password=device["password"], timeout=15)

        baselines = {}
        output = self._exec(ssh, "display bgp peer all")

        for line in output.split("\n"):
            parts = line.split()
            if len(parts) >= 6 and parts[0] not in ("Peer", "----"):
                try:
                    baseline = PeerBaseline(
                        peer_ip=parts[0],
                        remote_as=int(parts[1]) if parts[1].isdigit() else 0,
                        state=parts[2],
                        msg_rcvd=int(parts[3]) if parts[3].isdigit() else 0,
                        msg_sent=int(parts[4]) if parts[4].isdigit() else 0,
                        uptime=parts[5] if len(parts) > 5 else "",
                        prefix_received=0
                    )
                    baselines[parts[0]] = baseline
                except (ValueError, IndexError):
                    continue

        ssh.close()
        self.baseline[device.get("name", device["host"])] = baselines
        return baselines

    def check_device_health(self, device: Dict) -> HealthCheckResult:
        """
        检查单台设备BGP健康状态
        :return: 检查结果
        """
        result = HealthCheckResult(
            device=device.get("name", device["host"]),
            timestamp=datetime.now().isoformat()
        )

        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        ssh.connect(device["host"], username=device["username"],
                   password=device["password"], timeout=15)

        # 1. 检测Peer状态变化
        current_peers = {}
        output = self._exec(ssh, "display bgp peer all")
        for line in output.split("\n"):
            parts = line.split()
            if len(parts) >= 6 and parts[0] not in ("Peer", "----"):
                try:
                    current_peers[parts[0]] = {
                        "state": parts[2],
                        "msg_rcvd": int(parts[3]) if parts[3].isdigit() else 0
                    }
                except (ValueError, IndexError):
                    continue

        # 对比基线
        dev_baseline = self.baseline.get(result.device, {})
        for peer_ip, current in current_peers.items():
            if peer_ip in dev_baseline:
                old = dev_baseline[peer_ip]
                if old.state != current["state"]:
                    result.anomalies.append({
                        "type": "peer_state_change",
                        "peer": peer_ip,
                        "old_state": old.state,
                        "new_state": current["state"],
                        "severity": "HIGH" if current["state"] != "Established" else "LOW"
                    })
                    result.score -= 15 if current["state"] != "Established" else 0
            else:
                result.peer_changes.append({
                    "type": "new_peer",
                    "peer": peer_ip,
                    "state": current["state"]
                })

        # 检测缺失的Peer
        for peer_ip in dev_baseline:
            if peer_ip not in current_peers:
                result.anomalies.append({
                    "type": "peer_missing",
                    "peer": peer_ip,
                    "severity": "CRITICAL"
                })
                result.score -= 20

        # 2. 检测黑洞路由
        bgp_output = self._exec(ssh, "display bgp routing-table all")
        bh_count = bgp_output.count("0.0.0.0") - bgp_output.count("0.0.0.0/0")
        if bh_count > 0:
            result.anomalies.append({
                "type": "blackhole_routes",
                "count": bh_count,
                "severity": "CRITICAL"
            })
            result.score -= bh_count * 5

        result.score = max(0, result.score)

        ssh.close()
        return result

    def check_rr_client_consistency(self,
                                     rr_device: Dict,
                                     client_devices: List[Dict]
                                     ) -> HealthCheckResult:
        """
        检查RR与Client间的路由一致性
        :param rr_device: RR设备
        :param client_devices: Client设备列表
        :return: 一致性检查结果
        """
        result = HealthCheckResult(
            device=f"RR-{rr_device.get('name', '')}",
            timestamp=datetime.now().isoformat()
        )

        # 1. 获取RR上的路由前缀集合
        rr_prefixes = self._get_prefix_set(rr_device)

        # 2. 对比每个Client
        for client in client_devices:
            client_prefixes = self._get_prefix_set(client)

            # 检测Client缺失RR上的路由
            missing = rr_prefixes - client_prefixes
            if missing:
                # 只报告前缀较长（更具体）的缺失路由
                significant_missing = [p for p in missing if "/" in p and int(p.split("/")[1]) >= 24]
                if significant_missing:
                    result.rr_inconsistencies.append({
                        "client": client.get("name", client["host"]),
                        "missing_prefixes": len(significant_missing),
                        "sample": list(significant_missing)[:5],
                        "severity": "HIGH" if len(significant_missing) > 10 else "MEDIUM"
                    })
                    result.score -= min(30, len(significant_missing) // 10)

        result.score = max(0, result.score)
        return result

    def _get_prefix_set(self, device: Dict) -> Set[str]:
        """获取设备上的BGP路由前缀集合"""
        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        ssh.connect(device["host"], username=device["username"],
                   password=device["password"], timeout=30)

        output = self._exec(ssh, "display bgp routing-table all")
        prefixes = set()
        for line in output.split("\n"):
            parts = line.split()
            if len(parts) >= 2 and "/" in parts[1]:
                prefixes.add(parts[1])

        ssh.close()
        return prefixes

    def _exec(self, ssh, command: str) -> str:
        stdin, stdout, stderr = ssh.exec_command(command, timeout=30)
        return stdout.read().decode().strip()

    def generate_health_summary(self, checks: List[HealthCheckResult]) -> str:
        """生成健康汇总"""
        lines = []
        lines.append("=" * 60)
        lines.append("BGP健康检查汇总")
        lines.append(f"检查时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        lines.append("=" * 60)

        avg_score = sum(c.score for c in checks) / max(1, len(checks))
        lines.append(f"平均健康评分: {avg_score:.0f}/100")
        lines.append(f"告警级别: {'🔴 CRITICAL' if avg_score < 60 else '🟡 WARNING' if avg_score < 80 else '🟢 OK'}")

        for check in checks:
            status = "🟢" if check.score >= 80 else "🟡" if check.score >= 60 else "🔴"
            lines.append(f"\n{status} {check.device} (评分: {check.score:.0f})")
            if check.anomalies:
                for anomaly in check.anomalies:
                    lines.append(f"  ⚠ [{anomaly['severity']}] {anomaly['type']}: "
                               f"{anomaly.get('peer', '')} {anomaly.get('count', '')}")
            if check.rr_inconsistencies:
                for inc in check.rr_inconsistencies:
                    lines.append(f"  ⚠ RR一致性: {inc['client']} 缺失{inc['missing_prefixes']}条路由")

        return "\n".join(lines)


if __name__ == "__main__":
    devices = [
        {"host": "10.0.1.1", "username": "admin", "password": "pass", "name": "NE8000-RR"},
        {"host": "10.0.1.2", "username": "admin", "password": "pass", "name": "NE40E-Client-1"},
    ]
    monitor = BGPHealthMonitor(devices)
    # 先采集基线
    for d in devices:
        monitor.collect_baseline(d)
    # 后执行健康检查
    results = [monitor.check_device_health(d) for d in devices]
    print(monitor.generate_health_summary(results))

4.7.3 BGPBlackholeFixer：黑洞路由自动修复

#!/usr/bin/env python3
"""
BGP黑洞路由自动修复器
检查RR客户端状态、修复下一跳、验证修复效果
"""
import paramiko
import re
from dataclasses import dataclass, field
from typing import Dict, List, Optional
from datetime import datetime


@dataclass
class FixAction:
    """修复动作"""
    device: str
    action_type: str       # fix_nexthop / clear_peer / fix_policy
    command: str
    rollback_command: str
    applied: bool = False
    verified: bool = False


class BGPBlackholeFixer:
    """BGP黑洞路由自动修复器"""

    def __init__(self, devices: List[Dict]):
        self.devices = devices
        self.fix_log: List[FixAction] = []

    def check_rr_clients(self, rr_device: Dict) -> List[Dict]:
        """
        检查RR客户端状态
        :return: 客户端状态列表
        """
        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        ssh.connect(rr_device["host"], username=rr_device["username"],
                   password=rr_device["password"], timeout=15)

        output = self._exec(ssh, "display bgp peer all")
        clients = []

        for line in output.split("\n"):
            parts = line.split()
            if len(parts) >= 7 and "RR_Client" in " ".join(parts):
                try:
                    clients.append({
                        "peer_ip": parts[0],
                        "remote_as": parts[1],
                        "state": parts[2],
                        "msg_rcvd": parts[3],
                        "msg_sent": parts[4],
                        "uptime": parts[5],
                        "is_rr_client": True
                    })
                except IndexError:
                    continue

        ssh.close()
        return clients

    def fix_nexthop(self,
                    device: Dict,
                    peer_ip: str,
                    new_nexthop: str) -> FixAction:
        """
        修复下一跳
        :param device: 设备信息
        :param peer_ip: Peer IP
        :param new_nexthop: 新的下一跳地址
        :return: 修复动作
        """
        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        ssh.connect(device["host"], username=device["username"],
                   password=device["password"], timeout=15)

        action = FixAction(
            device=device.get("name", device["host"]),
            action_type="fix_nexthop",
            command=f"peer {peer_ip} nexthop-{new_nexthop}",
            rollback_command=f"undo peer {peer_ip} nexthop-{new_nexthop}"
        )

        try:
            # 进入BGP配置
            self._exec(ssh, "system-view")
            # 查找BGP进程号
            bgp_output = self._exec(ssh, "display current-configuration | include ^bgp ")
            bgp_match = re.search(r"bgp\s+(\d+)", bgp_output)
            if bgp_match:
                bgp_as = bgp_match.group(1)
                self._exec(ssh, f"bgp {bgp_as}")
                self._exec(ssh, action.command)
                self._exec(ssh, "quit")
                action.applied = True
            else:
                print(f"[-] 未在{device['name']}上找到BGP配置")

            # 验证
            verify_output = self._exec(ssh, f"display bgp peer {peer_ip} verbose")
            if new_nexthop in verify_output:
                action.verified = True

        except Exception as e:
            print(f"[!] 修复失败: {e}")
        finally:
            ssh.close()

        self.fix_log.append(action)
        return action

    def verify_fix(self, device: Dict, prefix: str) -> Dict:
        """
        验证修复效果
        :return: 验证结果
        """
        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        ssh.connect(device["host"], username=device["username"],
                   password=device["password"], timeout=15)

        result = {
            "device": device.get("name", device["host"]),
            "prefix": prefix,
            "is_blackhole": False,
            "nexthop": "",
            "is_valid": False,
            "is_best": False
        }

        output = self._exec(ssh, f"display bgp routing-table {prefix}")
        for line in output.split("\n"):
            parts = line.split()
            if len(parts) >= 4 and prefix.split("/")[0] in parts:
                result["nexthop"] = parts[2] if len(parts) > 2 else ""
                result["is_blackhole"] = result["nexthop"] == "0.0.0.0"
                result["is_best"] = "*>" in line
                result["is_valid"] = "valid" in line.lower() or "*>" in line

        ssh.close()
        return result

    def clear_bgp_peer(self, device: Dict, peer_ip: str = None) -> FixAction:
        """
        重置BGP Peer会话
        :param peer_ip: 特定Peer，None=重置所有
        :return: 修复动作
        """
        action = FixAction(
            device=device.get("name", device["host"]),
            action_type="clear_peer",
            command=f"reset bgp peer {peer_ip}" if peer_ip else "reset bgp all",
            rollback_command="(无需回滚，BGP会自动重建立)"
        )

        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        ssh.connect(device["host"], username=device["username"],
                   password=device["password"], timeout=15)

        try:
            self._exec(ssh, action.command)
            action.applied = True
        except Exception as e:
            print(f"[!] Peer重置失败: {e}")
        finally:
            ssh.close()

        self.fix_log.append(action)
        return action

    def generate_fix_report(self) -> str:
        """生成修复报告"""
        lines = []
        lines.append("=" * 60)
        lines.append("BGP黑洞路由修复报告")
        lines.append(f"修复时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        lines.append("=" * 60)
        lines.append(f"执行修复动作数: {len(self.fix_log)}")

        applied = sum(1 for f in self.fix_log if f.applied)
        verified = sum(1 for f in self.fix_log if f.verified)
        lines.append(f"已应用: {applied}  已验证: {verified}  失败: {len(self.fix_log) - applied}")

        for i, fix in enumerate(self.fix_log, 1):
            status = "✓ 已验证" if fix.verified else "✓ 已应用" if fix.applied else "✗ 失败"
            lines.append(f"\n{i}. [{status}] {fix.device} - {fix.action_type}")
            lines.append(f"   命令: {fix.command}")
            lines.append(f"   回滚: {fix.rollback_command}")

        return "\n".join(lines)

    def _exec(self, ssh, command: str) -> str:
        stdin, stdout, stderr = ssh.exec_command(command, timeout=30)
        return stdout.read().decode().strip()


# ========== 自动化修复流程 ==========
def automated_blackhole_fix(detector_results: List, devices: List[Dict]):
    """
    自动化修复流程入口
    检测→诊断→修复→验证
    """
    print("[1/4] 检测黑洞路由...")
    blackhole_devices = [r for r in detector_results if r.blackhole_routes]

    if not blackhole_devices:
        print("✓ 未发现黑洞路由，无需修复")
        return

    print(f"[2/4] 发现{len(blackhole_devices)}台设备存在黑洞路由")

    fixer = BGPBlackholeFixer(devices)

    print("[3/4] 执行修复...")
    for result in blackhole_devices:
        for bh_route in result.blackhole_routes:
            # 诊断修复策略
            if bh_route.prefix == "0.0.0.0/0":
                # 默认路由黑洞：检查并修复策略
                print(f"  修复默认路由黑洞: {result.device_name}")
                # 先清除错误策略，再重置BGP
                fixer.clear_bgp_peer(
                    next((d for d in devices if d.get("name") == result.device_name), devices[0])
                )
            else:
                # 普通路由黑洞：修复下一跳
                print(f"  修复路由黑洞: {bh_route.prefix}")

    print("[4/4] 验证修复效果...")
    report = fixer.generate_fix_report()
    print(report)


if __name__ == "__main__":
    devices = [
        {"host": "10.0.1.1", "username": "admin", "password": "pass", "name": "NE8000-Core-A"},
    ]
    fixer = BGPBlackholeFixer(devices)
    # 示例：修复下一跳
    # fixer.fix_nexthop(devices[0], "10.0.0.2", "127.0.0.1")
    # 示例：验证修复
    # print(fixer.verify_fix(devices[0], "0.0.0.0/0"))

4.7.4 自动化修复流程时序图

自动化修复效果：

安全控制： | 控制项 | 实现方式 | |--------|---------| | 操作权限 | 修复器仅连接白名单内的设备 | | 操作审计 | 每条命令及回滚命令均记录审计日志 | | 人工确认 | CRITICAL级别操作需人工审批后执行 | | 回滚保护 | 所有修改操作自动记录回滚命令 | | 频率限制 | 同一设备30分钟内最多修复2次 |

思考

1. BGP路由反射器架构下，如果RR本身产生了一条Blackhole路由，如何快速隔离影响范围？
2. 使用BGP communities（团体属性）是否可以实现更精细化的默认路由控制？如何实现？
3. 如果城域网有多个互联网出口（如5个运营商），如何设计BGP策略确保默认路由的可靠性和负载分担？

下篇预告：第337篇：OSPF 与 BGP 交互引起的次优路径