BGP Prefix Origin and RPKI Watch with RIPEstat

id: bgp-prefix-origin-and-rpki-watch
namespace: company.team
description: |
  Watch how the internet routes the IP prefixes your company announces. Using public RIPE RIS data,
  alert in Slack when another network originates your prefix or a more-specific of it (a possible BGP
  hijack), when the prefix loses visibility, or when its RPKI status for your ASN is no longer valid.

inputs:
  - id: prefixes
    type: JSON
    displayName: Prefixes to watch
    description: |
      List of objects with the announced `prefix`, the `origin` ASN that should announce it, optional
      `allowed_origins` (extra ASNs allowed to announce it or a more-specific, such as a DDoS scrubbing
      provider), and an `owner` to mention.
    defaults: |
      [
        {"prefix": "1.1.1.0/24", "origin": 13335, "allowed_origins": [], "owner": "@network"},
        {"prefix": "2606:4700:4700::/48", "origin": 13335, "allowed_origins": [], "owner": "@dns-team"},
        {"prefix": "140.82.112.0/20", "origin": 36459, "allowed_origins": [], "owner": "@network"}
      ]

  - id: min_visibility_percent
    type: INT
    displayName: Minimum visibility (%)
    description: Alert when fewer than this share of RIPE RIS full-table peers see
      the prefix.
    defaults: 80

tasks:
  - id: for_each_prefix
    type: io.kestra.plugin.core.flow.Loop
    description: Check every prefix independently, four at a time.
    values: "{{ inputs.prefixes }}"
    concurrencyLimit: 4
    tasks:
      - id: fetch_routing_status
        type: io.kestra.plugin.core.http.Download
        description: Origins, more-specific announcements and peer visibility seen by
          RIPE RIS (public, no key).
        uri: "https://stat.ripe.net/data/routing-status/data.json?resource={{
          fromJson(item.value).prefix }}&sourceapp=kestra-blueprint"

      - id: fetch_rpki
        type: io.kestra.plugin.core.http.Download
        description: RPKI route origin validation of the prefix for the expected origin ASN.
        uri: "https://stat.ripe.net/data/rpki-validation/data.json?resource=AS{{
          fromJson(item.value).origin }}&prefix={{ fromJson(item.value).prefix
          }}&sourceapp=kestra-blueprint"

      - id: load_previous
        type: io.kestra.plugin.core.kv.Get
        description: Findings reported at the last run. Missing on the first run.
        key: "bgp_watch_{{ fromJson(item.value).prefix | replace({':': '-'}) | slugify
          }}"
        errorOnMissing: false

      - id: evaluate
        type: io.kestra.plugin.scripts.python.Script
        description: Compare what the internet sees with what you expect, and list new
          and cleared findings.
        containerImage: python:3.12-slim
        inputFiles:
          routing.json: "{{ outputs.fetch_routing_status.uri }}"
          rpki.json: "{{ outputs.fetch_rpki.uri }}"
          expected.json: "{{ item.value }}"
          previous.json: "{{ (outputs.load_previous.value ?? {}) | toJson }}"
        script: |
          import json

          def load(name):
              with open(name, encoding="utf-8") as f:
                  data = json.load(f)
              return json.loads(data) if isinstance(data, str) else data

          routing = load("routing.json")["data"]
          rpki = load("rpki.json")["data"]
          expected = load("expected.json")
          previous = load("previous.json") or {}

          prefix = expected["prefix"]
          allowed = {int(expected["origin"])} | {int(a) for a in expected.get("allowed_origins", [])}
          family = "v6" if ":" in prefix else "v4"
          seen = routing["visibility"][family]["ris_peers_seeing"]
          total = routing["visibility"][family]["total_ris_peers"] or 1
          visibility = round(100 * seen / total)
          origins = sorted(int(o["origin"]) for o in routing.get("origins", []))
          rpki_status = rpki.get("status", "unknown")

          findings = {}
          for asn in origins:
              if asn not in allowed:
                  findings[f"origin:{asn}"] = f"AS{asn} originates {prefix}, which only {', '.join(f'AS{a}' for a in sorted(allowed))} should announce."
          for route in routing.get("more_specifics", []):
              asn = int(route["origin"])
              if asn not in allowed:
                  findings[f"more:{route['prefix']}:{asn}"] = f"AS{asn} announces the more-specific {route['prefix']} (more-specific routes win, so this attracts the traffic)."
          if visibility < {{ inputs.min_visibility_percent }}:
              findings["visibility"] = f"Only {seen} of {total} RIS peers ({visibility}%) see {prefix}. The prefix may be withdrawn or filtered."
          if rpki_status != "valid":
              roas = ", ".join(f"AS{r['origin']} {r['prefix']} max /{r['max_length']}" for r in rpki.get("validating_roas", [])) or "none"
              effect = ("No ROA covers it, so networks that validate routes cannot reject a hijack of it."
                        if rpki_status == "unknown" else
                        "Networks that drop RPKI-invalid routes reject your own announcement.")
              findings[f"rpki:{rpki_status}"] = f"RPKI status for AS{expected['origin']} is `{rpki_status}` (ROAs covering it: {roas}). {effect}"

          new = [findings[k] for k in findings if k not in previous]
          cleared = [previous[k] for k in previous if k not in findings]
          print(f"{prefix}: origins {origins or 'none'}, visibility {visibility}%, RPKI {rpki_status}, "
                f"{len(findings)} findings ({len(new)} new, {len(cleared)} cleared)")
          print("::" + json.dumps({"outputs": {
              "new_findings": new,
              "cleared": cleared,
              "notify": bool(new or cleared),
              "origins": ", ".join(f"AS{a}" for a in origins) or "none",
              "visibility": visibility,
              "rpki_status": rpki_status,
              "findings": findings,
          }}) + "::")

      - id: if_changed
        type: io.kestra.plugin.core.flow.If
        description: Only post when a finding appears or clears.
        condition: "{{ outputs.evaluate.vars.notify }}"
        then:
          - id: alert_owner
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: Post the new and cleared findings with the current routing facts
              from RIPE RIS.
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            messageText: |
              {{ outputs.evaluate.vars.new_findings | length > 0 ? ':rotating_light:' : ':white_check_mark:' }} *BGP routing of {{ fromJson(item.value).prefix }} changed* (owner: {{ fromJson(item.value).owner }})
              {% for f in outputs.evaluate.vars.new_findings %}- {{ f }}
              {% endfor %}{% for c in outputs.evaluate.vars.cleared %}- Resolved: {{ c }}
              {% endfor %}Now: origins {{ outputs.evaluate.vars.origins }} | visibility {{ outputs.evaluate.vars.visibility }}% | RPKI {{ outputs.evaluate.vars.rpki_status }}
              <https://stat.ripe.net/{{ fromJson(item.value).prefix }}|Open in RIPEstat>

      - id: remember_findings
        type: io.kestra.plugin.core.kv.Set
        description: Store the current findings so each one is reported once. Runs after
          the alert.
        key: "bgp_watch_{{ fromJson(item.value).prefix | replace({':': '-'}) | slugify
          }}"
        kvType: JSON
        overwrite: true
        value: "{{ outputs.evaluate.vars.findings | toJson }}"

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: A prefix that cannot be checked must not look healthy.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    messageText: ":warning: BGP prefix watch failed in execution {{ execution.id }}
      ({{ flow.namespace }}.{{ flow.id }}). Open the failed loop iteration: the
      prefix may be malformed, or RIPEstat may be unavailable."

triggers:
  - id: every_15_minutes
    type: io.kestra.plugin.core.trigger.Schedule
    description: RIPE RIS refreshes routing state continuously. Disabled until the
      Slack secret is set.
    cron: "*/15 * * * *"
    disabled: true

If another network announces your IP prefix, or a more-specific piece of it, part of the internet sends your traffic there: that is a BGP hijack or a route leak, and it is usually noticed only when customers complain. A prefix can also silently lose visibility after a router change, or become RPKI-invalid when a ROA expires or a new origin is added without one, and then networks that validate routes drop it.

This blueprint checks each prefix you announce against RIPE RIS, the public route collectors that peer with hundreds of networks, through the free RIPEstat Data API. It reports, once each, and again when they clear:

an origin ASN other than yours (or an allowed one) announcing the prefix;
a more-specific of the prefix announced by an ASN you did not allow;
visibility below min_visibility_percent of RIS peers;
an RPKI status other than valid for your origin ASN (unknown means no ROA, invalid_asn or invalid_length means a ROA exists but does not match).

How it works

For each entry in prefixes (Loop, four at a time):

fetch_routing_status downloads the origins, more-specifics and peer visibility.
fetch_rpki downloads the RPKI validation for your origin ASN and the ROAs that cover the prefix.
load_previous reads the findings reported at the last run from the Kestra KV store.
evaluate compares what the internet sees with what you expect and lists new and cleared findings.
if_changed skips Slack when nothing changed.
alert_owner posts the findings, the current origins, visibility and RPKI status, and a RIPEstat link.
remember_findings stores the findings, after the alert, so each one is posted once.

Prerequisites

A Slack Incoming Webhook. RIPEstat is public and needs no key; the flow identifies itself with the sourceapp parameter, as RIPE asks.
The prefixes exactly as you announce them. A covering aggregate you do not announce has no origin and zero visibility, so list the announced prefixes instead.

Secrets

SLACK_WEBHOOK_URL: Slack Incoming Webhook URL for alerts and failures.

In the open-source edition, secrets are environment variables on the Kestra container, prefixed with SECRET_ and base64-encoded. See how to manage secrets.

Inputs

prefixes (JSON): list of {"prefix", "origin", "allowed_origins", "owner"} objects. Add your DDoS protection or transit provider's ASN to allowed_origins if it announces your prefix or more-specifics during mitigation.
min_visibility_percent (INT, default 80): the visibility alert threshold.

Quick start

Add the Slack secret and replace the default prefixes (Cloudflare's and GitHub's) with yours.
Execute the flow. Existing problems, such as a missing ROA, are posted once.
Enable every_15_minutes. From then on, Slack hears when routing changes and when it recovers.

To see a prefix the way the flow does, open https://stat.ripe.net/<prefix>.

Outputs

{{ outputs.evaluate.vars }} per iteration: origins, visibility, rpki_status, new_findings, cleared.
KV entries bgp_watch_<prefix>: the findings already reported.

Detect BGP Hijacks and RPKI Problems on Your IP Prefixes

How it works

Prerequisites

Secrets

Inputs

Quick start

Outputs

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Detect BGP Hijacks and RPKI Problems on Your IP Prefixes

How it works

Prerequisites

Secrets

Inputs

Quick start

Outputs

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Infrastructure Blueprints

Faktory Job Enqueue Path Reachability Sentinel

Falco Cloud Native Runtime Threat and Security Priority Event Sentinel

Fastly Month-to-Date Cost Guard

Fastly Daily Traffic Digest to Slack

Fastly Surrogate Key Purge for CMS Content Updates

Purge Fastly Cache on Deploy via Webhook

New to Kestra?