How to Set Up BGP Monitoring and Route Alerts

# Install via npm globally
sudo npm install bgpalerter -g
 
# Verify installation
bgpalerter --version
 
# Create a working directory
sudo mkdir -p /opt/bgpalerter
cd /opt/bgpalerter
 
# Generate default configuration
bgpalerter generate

bgpalerter --version

# Production prefixes
- prefix: "192.0.2.0/24"
  description: "Primary production prefix"
  asn: [64500]
  ignoreMoreSpecifics: false
 
- prefix: "198.51.100.0/24"
  description: "Secondary services prefix"
  asn: [64500]
  ignoreMoreSpecifics: false
 
# IPv6 prefix
- prefix: "2001:db8::/32"
  description: "IPv6 allocation"
  asn: [64500]
  ignoreMoreSpecifics: false
 
# Upstream provider prefix (monitor for leaks)
- prefix: "203.0.113.0/24"
  description: "Upstream transit prefix"
  asn: [64501, 64502]
  ignoreMoreSpecifics: true

# Validate YAML syntax
python3 -c "import yaml; yaml.safe_load(open('prefixes.yml'))"

connectors:
  - file: connectorRIS
    name: ris
 
monitors:
  - file: monitorHijack
    channel: hijack
    name: basic-hijack-detection
 
  - file: monitorNewPrefix
    channel: newprefix
    name: prefix-detection
 
  - file: monitorVisibility
    channel: visibility
    name: visibility-loss-detection
 
  - file: monitorAS
    channel: misconfiguration
    name: as-monitor
 
reports:
  # Slack notifications
  - file: reportSlack
    channels:
      - webhook: "https://hooks.slack.com/services/T00/B00/xxxx"
        type: slack
 
  # Email notifications
  - file: reportEmail
    channels:
      - smtp:
          host: "smtp.example.com"
          port: 587
          secure: false
          auth:
            user: "alerts@example.com"
            pass: "${SMTP_PASSWORD}"
          from: "bgp-alerts@example.com"
          to:
            - "noc@example.com"
            - "security@example.com"
 
  # Syslog for SIEM integration
  - file: reportSyslog
    channels:
      - host: "siem.example.com"
        port: 514
        transport: "udp"
 
  # Local file logging (always recommended)
  - file: reportFile
    channels:
      - logdir: "/opt/bgpalerter/logs/"
        filenamePrefix: "bgp-alert"

# Test configuration syntax
bgpalerter --test

pip install websocket-client

import json
import websocket
import time
from datetime import datetime
 
MONITORED_PREFIXES = ["192.0.2.0/24", "198.51.100.0/24"]
EXPECTED_ORIGIN_ASN = "64500"
 
def on_message(ws, message):
    data = json.loads(message)
 
    if data.get("type") == "ris_message":
        msg = data.get("data", {})
        prefix = msg.get("prefix", "N/A")
        origin_asn = msg.get("origin", "N/A")
        peer = msg.get("peer", "N/A")
        path = msg.get("path", [])
        timestamp = datetime.fromtimestamp(msg.get("timestamp", 0))
 
        # Check for anomalies in monitored prefixes
        for monitored in MONITORED_PREFIXES:
            if prefix == monitored and str(origin_asn) != EXPECTED_ORIGIN_ASN:
                print(f"[ALERT] {timestamp} - Unexpected origin for {prefix}")
                print(f"  Origin AS: {origin_asn} (expected: {EXPECTED_ORIGIN_ASN})")
                print(f"  AS Path: {' -> '.join(map(str, path))}")
                print(f"  Peer: {peer}")
                # Add your alerting logic here (email, Slack, etc.)
 
def on_error(ws, error):
    print(f"[ERROR] WebSocket error: {error}")
 
def on_close(ws, close_status_code, close_msg):
    print("[INFO] Connection closed. Reconnecting in 10 seconds...")
    time.sleep(10)
    start_monitor()
 
def on_open(ws):
    # Subscribe to updates for monitored prefixes
    for prefix in MONITORED_PREFIXES:
        subscribe_msg = {
            "type": "ris_subscribe",
            "data": {
                "prefix": prefix,
                "type": "UPDATE",
                "socketOptions": {
                    "includeRaw": False
                }
            }
        }
        ws.send(json.dumps(subscribe_msg))
    print(f"[INFO] Subscribed to {len(MONITORED_PREFIXES)} prefixes")
 
def start_monitor():
    ws = websocket.WebSocketApp(
        "wss://ris-live.ripe.net/v1/ws/?client=bgp-monitor",
        on_open=on_open,
        on_message=on_message,
        on_error=on_error,
        on_close=on_close
    )
    ws.run_forever()
 
if __name__ == "__main__":
    print("[INFO] Starting RIPE RIS Live BGP monitor...")
    start_monitor()

# Test the monitor (press Ctrl+C after a few updates)
python3 /opt/bgpalerter/ris-monitor.py

# Query BGP routes for a specific prefix
curl -s "https://api.cloudflare.com/client/v4/radar/bgp/routes?prefix=192.0.2.0/24" \
  -H "Authorization: Bearer $CF_API_TOKEN" \
  -H "Content-Type: application/json" | python3 -m json.tool
 
# Query BGP route origin changes over the last 24 hours
curl -s "https://api.cloudflare.com/client/v4/radar/bgp/routes/stats?dateRange=1d" \
  -H "Authorization: Bearer $CF_API_TOKEN" \
  -H "Content-Type: application/json" | python3 -m json.tool

#!/bin/bash
# /opt/bgpalerter/check-cf-radar.sh
# Run via cron every 15 minutes
 
PREFIX="192.0.2.0/24"
EXPECTED_ORIGIN="64500"
 
RESULT=$(curl -s "https://api.cloudflare.com/client/v4/radar/bgp/routes?prefix=$PREFIX" \
  -H "Authorization: Bearer $CF_API_TOKEN")
 
ORIGIN=$(echo "$RESULT" | python3 -c "
import sys, json
data = json.load(sys.stdin)
routes = data.get('result', {}).get('routes', [])
for r in routes:
    print(r.get('origin_asn', 'unknown'))
")
 
if echo "$ORIGIN" | grep -qv "$EXPECTED_ORIGIN"; then
  echo "[ALERT] Unexpected origin ASN detected for $PREFIX: $ORIGIN"
  # Send alert via your preferred method
fi

chmod +x /opt/bgpalerter/check-cf-radar.sh
/opt/bgpalerter/check-cf-radar.sh

[Unit]
Description=BGPalerter BGP Monitoring Service
Documentation=https://github.com/nttgin/BGPalerter
After=network.target
Wants=network-online.target
 
[Service]
Type=simple
User=bgpalerter
Group=bgpalerter
WorkingDirectory=/opt/bgpalerter
ExecStart=/usr/bin/bgpalerter
Restart=always
RestartSec=10
StandardOutput=journal
StandardError=journal
SyslogIdentifier=bgpalerter
 
# Security hardening
NoNewPrivileges=true
ProtectSystem=strict
ReadWritePaths=/opt/bgpalerter/logs
 
[Install]
WantedBy=multi-user.target

# Create dedicated service user
sudo useradd -r -s /usr/sbin/nologin -d /opt/bgpalerter bgpalerter
sudo chown -R bgpalerter:bgpalerter /opt/bgpalerter
 
# Reload systemd and enable service
sudo systemctl daemon-reload
sudo systemctl enable bgpalerter
sudo systemctl start bgpalerter

[Unit]
Description=RIPE RIS Live BGP Monitor
After=network.target
Wants=network-online.target
 
[Service]
Type=simple
User=bgpalerter
Group=bgpalerter
WorkingDirectory=/opt/bgpalerter
ExecStart=/usr/bin/python3 /opt/bgpalerter/ris-monitor.py
Restart=always
RestartSec=15
 
[Install]
WantedBy=multi-user.target

sudo systemctl daemon-reload
sudo systemctl enable ris-monitor
sudo systemctl start ris-monitor

sudo systemctl status bgpalerter
sudo systemctl status ris-monitor
journalctl -u bgpalerter --no-pager -n 20

# BGPalerter includes a test mode that simulates events
cd /opt/bgpalerter
bgpalerter --test-alert

pip install pybgpstream

import pybgpstream
 
# Query historical BGP data for your prefix
stream = pybgpstream.BGPStream(
    from_time="2026-02-16 00:00:00",
    until_time="2026-02-16 06:00:00",
    collectors=["rrc00", "route-views2"],
    record_type="updates",
    filter="prefix more 192.0.2.0/24"
)
 
for rec in stream.records():
    for elem in rec:
        print(f"{elem.time} | {elem.type} | {elem.fields.get('prefix', 'N/A')} | "
              f"Origin: {elem.fields.get('as-path', '').split()[-1] if elem.fields.get('as-path') else 'N/A'}")

# Verify Slack webhook is reachable
curl -s -o /dev/null -w "%{http_code}" \
  -X POST "https://hooks.slack.com/services/T00/B00/xxxx" \
  -H "Content-Type: application/json" \
  -d '{"text":"BGP Monitor test alert"}'
 
# Verify email delivery
echo "BGP Monitor test" | mail -s "BGP Alert Test" noc@example.com
 
# Check log files are being written
ls -la /opt/bgpalerter/logs/