executing-plans
executing-plans
Use when executing an implementation plan in the current session as the implementer yourself — your human partner chose inline execution, or no subagent tool is available
implementing-bgp-security-with-rpki使用路由来源授权(ROA)、RPKI-to-Router 协议和 Cisco、Juniper 路由器上的 ROV 策略,实施 BGP 路由来源验证(RPKI),以防止路由劫持。
Independent GitHub snapshot · Automated classification · Original documentation governs
资源公钥基础设施(RPKI,Resource Public Key Infrastructure)为 BGP 路由来源提供密码学验证,以防止路由劫持(route hijacking)和意外路由泄漏。RPKI 使网络运营商能够创建路由来源授权(ROA,Route Origin Authorization),声明哪些自治系统(AS,Autonomous System)被授权通告特定 IP 前缀。BGP 路由器通过路由来源验证(ROV,Route Origin Validation)将收到的路由通告与 RPKI 数据进行验证,拒绝来源无效的路由。本技能涵盖通过地区互联网注册机构(RIR,Regional Internet Registry)创建 ROA、部署 RPKI 验证器软件、在 Cisco IOS-XE 和 Juniper Junos 路由器上配置 ROV,以及基于 RPKI 验证状态实施 BGP 过滤策略。
┌──────────────────────────────────────────────┐
│ 地区互联网注册机构 │
│ (ARIN, RIPE, APNIC, AFRINIC, LACNIC) │
│ │
│ ┌─────────────────────────────────────────┐ │
│ │ 信任锚点(根 CA 证书) │ │
│ │ ├── CA 证书(ISP/组织) │ │
│ │ │ ├── ROA: AS64512 → 198.51.100.0/24 │ │
│ │ │ └── ROA: AS64512 → 2001:db8::/32 │ │
│ │ └── CA 证书(另一组织) │ │
│ │ └── ROA: AS64513 → 203.0.113.0/24 │ │
│ └─────────────────────────────────────────┘ │
└──────────────────────────────────────────────┘
│ rsync/RRDP
▼
┌──────────────────────┐
│ RPKI 验证器/缓存 │ (Routinator, FORT, OctoRPKI)
│ 验证 ROA │
│ 通过 RTR 提供 VRP │
└──────────────────────┘
│ RTR 协议(TCP 8323)
▼
┌──────────────────────┐
│ BGP 路由器 │
│ 执行 ROV │
│ 应用策略: │
│ Valid → Accept │
│ Invalid → Reject │
│ NotFound → Accept │
└──────────────────────┘
| 状态 | 含义 | 建议操作 |
|---|---|---|
| Valid(有效) | ROA 存在,来源 AS 和前缀匹配 | 接受路由(优先) |
| Invalid(无效) | ROA 存在,但来源 AS 或前缀长度不匹配 | 拒绝路由 |
| NotFound(未找到) | 没有 ROA 覆盖此前缀 | 接受(但降低优先级) |
ROA 是一个签名对象,声明:
ARIN(北美):
RIPE NCC(欧洲):
# 在 Ubuntu 上安装 Routinator
sudo apt install -y routinator
# 或通过 Cargo(Rust)安装
curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh
cargo install routinator
# 初始化 Routinator(接受 TAL)
routinator init --accept-arin-rpa
# 以 RTR 服务器模式启动 Routinator
routinator server \
--rtr 0.0.0.0:8323 \
--http 0.0.0.0:8080 \
--refresh 600 \
--retry 60 \
--expire 7200
# 以 systemd 服务运行
cat > /etc/systemd/system/routinator.service << 'SYSTEMD'
[Unit]
Description=Routinator RPKI Validator
After=network.target
[Service]
Type=simple
User=routinator
ExecStart=/usr/bin/routinator server --rtr 0.0.0.0:8323 --http 0.0.0.0:8080
Restart=always
RestartSec=10
[Install]
WantedBy=multi-user.target
SYSTEMD
sudo systemctl enable routinator
sudo systemctl start routinator
# 验证 Routinator 正在提供数据
curl http://localhost:8080/api/v1/status
curl http://localhost:8080/api/v1/validity/AS64512/198.51.100.0/24
# 查看验证后的 ROA 载荷(VRP)
routinator vrps --format json | head -50
! 配置 RPKI 缓存服务器连接
router bgp 64512
bgp rpki server tcp 10.0.5.50 port 8323 refresh 600
! 验证 RPKI 会话
show bgp rpki server
show bgp rpki table
! 创建基于 RPKI 的过滤路由映射
route-map RPKI-FILTER permit 10
match rpki valid
set local-preference 200
route-map RPKI-FILTER permit 20
match rpki not-found
set local-preference 100
route-map RPKI-FILTER deny 30
match rpki invalid
! 应用到 BGP 邻居
router bgp 64512
address-family ipv4 unicast
neighbor 198.51.100.1 route-map RPKI-FILTER in
neighbor 203.0.113.1 route-map RPKI-FILTER in
address-family ipv6 unicast
neighbor 2001:db8::1 route-map RPKI-FILTER in
! 验证 ROV 操作
show bgp ipv4 unicast rpki validation
show bgp ipv4 unicast 198.51.100.0/24
show ip bgp rpki table
show ip bgp neighbors 198.51.100.1 rpki state
# 配置 RPKI 缓存连接
set routing-options validation group RPKI-VALIDATORS session 10.0.5.50 port 8323
set routing-options validation group RPKI-VALIDATORS session 10.0.5.50 refresh-time 600
set routing-options validation group RPKI-VALIDATORS session 10.0.5.50 hold-time 7200
set routing-options validation group RPKI-VALIDATORS session 10.0.5.50 record-lifetime 7200
# 创建验证策略
set policy-options policy-statement RPKI-POLICY term valid from validation-database valid
set policy-options policy-statement RPKI-POLICY term valid then validation-state valid
set policy-options policy-statement RPKI-POLICY term valid then local-preference 200
set policy-options policy-statement RPKI-POLICY term valid then accept
set policy-options policy-statement RPKI-POLICY term invalid from validation-database invalid
set policy-options policy-statement RPKI-POLICY term invalid then validation-state invalid
set policy-options policy-statement RPKI-POLICY term invalid then reject
set policy-options policy-statement RPKI-POLICY term unknown from validation-database unknown
set policy-options policy-statement RPKI-POLICY term unknown then validation-state unknown
set policy-options policy-statement RPKI-POLICY term unknown then local-preference 100
set policy-options policy-statement RPKI-POLICY term unknown then accept
# 应用到 BGP 对等体
set protocols bgp group TRANSIT import RPKI-POLICY
set protocols bgp group PEERS import RPKI-POLICY
# 验证
show validation session
show validation database
show validation statistics
show route validation-state invalid
#!/usr/bin/env python3
"""监控 RPKI ROV 部署健康状况和覆盖率统计。"""
import json
import sys
import urllib.request
class RPKIMonitor:
def __init__(self, routinator_url: str = "http://localhost:8080"):
self.routinator_url = routinator_url
def get_status(self) -> dict:
"""获取 Routinator 服务器状态。"""
url = f"{self.routinator_url}/api/v1/status"
try:
with urllib.request.urlopen(url) as resp:
return json.loads(resp.read())
except Exception as e:
print(f"连接 Routinator 时出错:{e}")
return {}
def check_validity(self, asn: int, prefix: str) -> dict:
"""检查前缀/来源对的 RPKI 有效性。"""
url = f"{self.routinator_url}/api/v1/validity/AS{asn}/{prefix}"
try:
with urllib.request.urlopen(url) as resp:
return json.loads(resp.read())
except Exception as e:
return {"error": str(e)}
def get_vrp_count(self) -> int:
"""获取经验证的 ROA 载荷(VRP)总数。"""
status = self.get_status()
return status.get("vrpsCount", 0)
def report(self, prefixes_to_check: list):
"""生成 RPKI 监控报告。"""
status = self.get_status()
print(f"\n{'='*60}")
print("RPKI 监控报告")
print(f"{'='*60}")
print(f"\nRoutinator 状态:")
print(f" 版本:{status.get('version', '未知')}")
print(f" VRP 总数:{status.get('vrpsCount', 'N/A')}")
print(f" 上次更新:{status.get('lastUpdateDone', 'N/A')}")
if prefixes_to_check:
print(f"\n前缀有效性检查:")
for asn, prefix in prefixes_to_check:
result = self.check_validity(asn, prefix)
validity = result.get("validated_route", {}).get(
"validity", {}).get("state", "error")
print(f" AS{asn} -> {prefix}: {validity.upper()}")
if __name__ == "__main__":
monitor = RPKIMonitor()
# 检查自己的前缀
own_prefixes = [
(64512, "198.51.100.0/24"),
]
monitor.report(own_prefixes)