RTBH prefixes validation on Junos || PoC setup || NANOG mailing list thread: https://lists.nanog.org/archives/list/nanog@lists.nanog.org/thread/O5XX6BHOSMINX4HKT2SMVOI66SMYXFOR/

Junos-RTBH-routes-validation.conf

root@vjr-17> show configuration routing-options bmp
/* Rotonda */
station BMP-feed-for-RTBH-1 {
    /* if the session breaks, then the router tries to reestablish it every 30 seconds */
    connection-mode active;
    /* Route Monitoring messages are sent only for peers defined under specific BGP groups */
    route-monitoring {
        none;
    }
    station-address 10.10.8.3;
    station-port 10179;
}
/* Rotonda */
station BMP-feed-for-RTBH-2 {
    connection-mode active;
    route-monitoring {
        none;
    }
    station-address 10.10.9.3;
    station-port 10179;
}
/* pmbmpd (pmacct BMP collector daemon) */
station BMP-full-feed {
    connection-mode active;
    station-address 10.5.5.13;
    station-port 10179;
}

root@vjr-17>

root@vjr-17> show configuration protocols bgp group IPT-fulltable-v4 bmp
monitor enable;
route-monitoring {
    pre-policy;
}

root@vjr-17>

root@vjr-17> show configuration protocols bgp group IPT-fulltable-v6 bmp
monitor enable;
route-monitoring {
    pre-policy;
}

root@vjr-17>

root@vjr-17> show configuration routing-options validation
group RPKI-validators {
    /* Routinator 3000 */
    session 10.10.8.2 {
        port 3323;
    }
    /* Routinator 3000 */
    session 10.10.9.2 {
        port 3323;
    }
}
group RTR-sessions-for-RTBH {
    /* RTRTR / aggressive timers to prevent stale records / 10.10.8.3 is same server as 10.10.8.2 */
    session 10.10.8.3 {
        /* send the Serial Query every 10 seconds */
        refresh-time 10;
        /* drop the session if 30 seconds have passed since the last received PDU */
        hold-time 30;
        /* flush the route validation records if 60 seconds have passed since the last received PDU */
        record-lifetime 60;
        port 3323;
    }
    /* RTRTR / aggressive timers to prevent stale records / 10.10.9.3 is same server as 10.10.9.2 */
    session 10.10.9.3 {
        refresh-time 10;
        hold-time 30;
        record-lifetime 60;
        port 3323;
    }
}

root@vjr-17>

root@vjr-17> show configuration policy-options policy-statement rpki-check
term reject-rpki-invalid {
    from {
        protocol bgp;
        validation-database invalid;
    }
    then {
        validation-state invalid;
        reject;
    }
}
term mark-rpki-valid {
    from {
        protocol bgp;
        validation-database valid;
    }
    then {
        validation-state valid;
        next policy;
    }
}
term skip-rpki-unknown {
    from {
        protocol bgp;
        validation-database unknown;
    }
    then {
        validation-state unknown;
        next policy;
    }
}
then next policy;

root@vjr-17>

root@vjr-17> show configuration policy-options policy-statement ipt-rtbh-v4
term accept-blackhole {
    from {
        family inet;
        community blackhole;
        route-filter 0.0.0.0/0 prefix-length-range /32-/32;
    }
    then {
        local-preference 170;
        origin igp;
        /* requires accept-remote-nexthop */
        next-hop 192.0.2.1;
        accept;
    }
}
term reject-invalid-length {
    from {
        family inet;
        community blackhole;
    }
    then reject;
}
then next policy;

root@vjr-17>

root@vjr-17> show configuration policy-options policy-statement ipt-rtbh-v6
term accept-blackhole {
    from {
        family inet6;
        community blackhole;
        route-filter ::/0 prefix-length-range /128-/128;
    }
    then {
        local-preference 170;
        origin igp;
        /* requires accept-remote-nexthop */
        next-hop 100::1;
        accept;
    }
}
term reject-invalid-length {
    from {
        family inet6;
        community blackhole;
    }
    then reject;
}
then next policy;

root@vjr-17>

Script which fetches blackhole routes from Rotonda RIB using its API, checks those routes against Routinator 3000 API and routes which have a covering VRP with ASN matching the origin ASN of the prefix, are written atomically to a JSON file:

martin@validator-1:~$ systemctl cat rtbh-generator.service
# /etc/systemd/system/rtbh-generator.service
[Unit]
Description=RTBH JSON Generator Daemon
After=network.target routinator.service rotonda.service

[Service]
Type=simple
User=rtbh-generator
ExecStart=/usr/local/bin/rtbh-daemon.py
Restart=always
RestartSec=10

[Install]
WantedBy=multi-user.target
martin@validator-1:~$

martin@validator-1:~$ cat /usr/local/bin/rtbh-daemon.py
#!/usr/bin/env python3

import os
import time
import json
import logging
import ipaddress
import requests

ROTONDA_V4_URL = "http://127.0.0.1:8008/api/v1/ribs/ipv4unicast/routes"
ROTONDA_V6_URL = "http://127.0.0.1:8008/api/v1/ribs/ipv6unicast/routes"
ROUTINATOR_URL = "http://127.0.0.1:8323/validity"
TARGET_COMMUNITY = "BLACKHOLE"
OUTPUT_FILE = "/var/lib/rtbh-generator/rtbh.json"
POLL_INTERVAL = 1

# Setup logging.
logging.basicConfig(level=logging.INFO, format="%(levelname)s - %(message)s")


def get_rotonda_candidates(url, expected_length):
    """Fetches routes from Rotonda and filters for /32 or /128 with
    the target community.
    """

    candidates = {}
    try:
        resp = requests.get(url, timeout=10)
        resp.raise_for_status()
        data = resp.json()
    except Exception as e:
        logging.error(f"Failed to fetch data from {url}: {e}")
        return candidates

    more_specifics = data.get("included", {}).get("moreSpecifics", [])

    for item in more_specifics:
        nlri = item.get("nlri")
        if not nlri:
            continue

        # Sanity check the prefix length.
        try:
            network = ipaddress.ip_network(nlri)
        except ValueError:
            continue

        if network.prefixlen != expected_length:
            continue

        # Check path attributes of active routes.
        for route in item.get("routes", []):
            if route.get("status") != "active":
                continue

            path_attrs = route.get("pathAttributes", [])
            communities = []
            as_path = []

            for attr in path_attrs:
                if "standardCommunities" in attr:
                    communities = attr["standardCommunities"]
                if "asPath" in attr:
                    as_path = attr["asPath"]

            # Validate target community presence and extract origin ASN.
            if TARGET_COMMUNITY in communities and as_path:

                # The last ASN in the path is the origin.
                origin_asn = as_path[-1]
                candidates[nlri] = origin_asn
                # Found a valid active path for this prefix,
                # skip to next prefix.
                break

    return candidates


def check_routinator(prefix, origin_asn):
    """Checks Routinator to ensure a covering VRP matches
    the origin ASN.
    """

    # Rotonda returns ASN as "AS12345", Routinator
    # query expects "12345".
    asn_num = origin_asn.lstrip("AS")

    try:
        params = {"asn": asn_num, "prefix": prefix}
        resp = requests.get(ROUTINATOR_URL, params=params, timeout=10)
        resp.raise_for_status()
        data = resp.json()
    except Exception as e:
        logging.error(
            f"Failed to check Routinator for {prefix} (AS{asn_num}): {e}"
        )
        return False

    vrps = data.get("validated_route", {}).get("validity", {}).get("VRPs", {})

    # We need at least one covering ROA where the ASN explicitly matches
    # our origin_asn.
    for category in ["matched", "unmatched_length"]:
        for vrp in vrps.get(category, []):
            if vrp.get("asn") == origin_asn:
                return True

    return False


def write_rtbh_file(valid_routes):
    """Writes the validated prefixes to the JSON file for
    rtrtr atomically.
    """

    roas = []
    for prefix, origin_asn in valid_routes.items():
        try:
            network = ipaddress.ip_network(prefix)
            roas.append(
                {
                    "asn": origin_asn,
                    "prefix": prefix,
                    "maxLength": network.prefixlen,
                    "ta": "local-rtbh",
                }
            )
        except ValueError:
            continue

    output_data = {"roas": roas}
    temp_file = f"{OUTPUT_FILE}.tmp"

    try:
        with open(temp_file, "w") as f:
            json.dump(output_data, f, indent=2)
            f.write("\n")

        # Atomic replace guarantees rtrtr never reads a partial file.
        os.replace(temp_file, OUTPUT_FILE)
        logging.info(
            f"Successfully wrote {len(roas)} valid RTBH ROAs to {OUTPUT_FILE}"
        )
    except Exception as e:
        logging.error(f"Failed to write RTBH JSON file: {e}")


def main():

    logging.info("Starting RTBH Generator Daemon...")

    while True:
        valid_routes = {}

        # Fetch and filter from Rotonda.
        v4_candidates = get_rotonda_candidates(ROTONDA_V4_URL, 32)
        v6_candidates = get_rotonda_candidates(ROTONDA_V6_URL, 128)

        all_candidates = {**v4_candidates, **v6_candidates}

        for prefix, origin_asn in all_candidates.items():

            if check_routinator(prefix, origin_asn):
                valid_routes[prefix] = origin_asn
            else:
                logging.info(
                    f"Prefix {prefix} (Origin: {origin_asn}) rejected "
                    "by Routinator check."
                )

        write_rtbh_file(valid_routes)
        time.sleep(POLL_INTERVAL)


if __name__ == "__main__":
    main()
martin@validator-1:~$

RTRTR configuration file /etc/rtrtr.conf:

log-level = "debug"
log = "syslog"
syslog-facility = "daemon"

http-listen = [ "0.0.0.0:8080" ]

[units.rtbh]
type = "json"
uri = "file:///var/lib/rtbh-generator/rtbh.json"
refresh = 1

[targets.local-3323]
type = "rtr"
unit = "rtbh"
listen = [ "10.10.8.3:3323" ]