Dumps IRRd database IRR route objects to JSON file || NANOG mailing list thread: https://lists.nanog.org/archives/list/nanog@lists.nanog.org/thread/O5XX6BHOSMINX4HKT2SMVOI66SMYXFOR/

dump-irr-db.py

#!/usr/bin/env python3
"""
Fetches valid route and route6 objects from the IRRd PostgreSQL
database and atomically generates a JSON file formatted for RTRTR.

Expected to be run by systemd-timer under the 'irrd' user.
"""

import ipaddress
import json
import logging
import os
import re
import sys
from logging.handlers import SysLogHandler
from pathlib import Path

import psycopg
from psycopg.rows import tuple_row


def fetch_route_objects() -> list[tuple[str, str]]:
    """
    Connects to IRRd's PostgreSQL database using
    Unix domain socket and fetches the route objects.
    """

    # Query below will contain duplicate prefix and ASN
    # pairs. For example, in case when the prefix and
    # ASN pair is present both in RPKI and one of the
    # IRR sources or in multiple IRR sources with equal
    # route object preference.
    query = """
        SELECT rpsl_pk, object_class
        FROM rpsl_objects
        WHERE object_class IN ('route', 'route6')
          AND scopefilter_status IN ('in_scope')
          AND rpki_status IN ('valid', 'not_found')
          AND route_preference_status IN ('visible');
    """

    logging.info("Connecting to IRRd database...")

    try:
        with psycopg.connect(dbname="irrd", row_factory=tuple_row) as conn:
            with conn.cursor() as cur:

                logging.info("Executing database query...")

                cur.execute(query)
                results = cur.fetchall()

                logging.info(
                    f"Successfully retrieved {len(results)} rows "
                    "from database."
                )
                return results

    except psycopg.Error as err:
        logging.error(f"Database connection or query failed: {err}")
        sys.exit(1)


def parse_route_objects(
    db_rows: list[tuple[str, str]],
) -> list[dict[str, str | int]]:
    """
    Parses database rows, extracts prefix and ASN, validates the
    prefix and constructs the list of route objects dictionaries.
    """

    # Set data type is used in order to get rid of
    # possible duplicate prefix and ASN pairs.
    unique_route_objects = {
        ("AS0", "0.0.0.0/0", 32),
        ("AS0", "::/0", 128),
    }

    skipped = 0
    rpsl_pk_pattern = re.compile(r"^(.*?)AS(\d+)")

    for rpsl_pk, object_class in db_rows:

        match = rpsl_pk_pattern.match(rpsl_pk)

        if not match:
            logging.warning(f"Skipping malformed rpsl_pk: {rpsl_pk}")
            skipped += 1
            continue

        prefix_str, asn_str = match.groups()

        # Sanity check. Additionally, the ip_network() guarantees
        # standard CIDR format. /32 v4 and /128 v6 RPKI ROAs are
        # stored in IRRd database without the CIDR mask, e.g
        # 199.7.48.7AS7342/ML32 or 2a03:ef42:c000::1AS57218/ML128.
        try:
            network = ipaddress.ip_network(prefix_str, strict=True)

        except ValueError:
            logging.warning(
                f"Skipping invalid IP prefix in rpsl_pk: {rpsl_pk}"
            )
            skipped += 1
            continue

        max_length = 32 if object_class == "route" else 128

        unique_route_objects.add((f"AS{asn_str}", str(network), max_length))

    if skipped:
        logging.warning(
            f"Skipped {skipped} rows due to malformed data or "
            "invalid prefixes."
        )

    route_objects: list[dict[str, str | int]] = [
        {
            "asn": asn,
            "prefix": prefix,
            "maxLength": max_length,
            "ta": "local-irr",
        }
        for asn, prefix, max_length in unique_route_objects
    ]

    return route_objects


def write_route_objects(
    route_objects: list[dict[str, str | int]], output_path: Path
) -> None:
    """
    Writes the route objects to the JSON file for RTRTR atomically.
    """

    output_data = {"roas": route_objects}
    temp_file = output_path.with_suffix(".tmp")

    try:
        with open(temp_file, "w", encoding="utf-8") as file_object:
            json.dump(output_data, file_object, indent=2)
            file_object.write("\n")

        # Atomic replace.
        os.replace(temp_file, output_path)
        logging.info(
            f"Successfully wrote {len(route_objects)} route "
            f"objects to {output_path}"
        )

    except OSError as err:

        logging.error(f"Failed to write or replace JSON file: {err}")

        # Cleanup temp file on failure if it exists.
        temp_file.unlink(missing_ok=True)
        sys.exit(1)


def main() -> None:
    """
    Main function of the script. Configures logging and calls functions.
    """

    # Script logs to syslog in order to preserve the journalctl
    # capability to filter and color the messages based on the
    # severity level.
    syslog = SysLogHandler(address="/dev/log", facility=SysLogHandler.LOG_USER)

    logging.basicConfig(
        level=logging.INFO,
        format="[{funcName}] - {levelname} - {message}",
        style="{",
        handlers=[syslog],
    )

    output_file = Path("/var/lib/irrd/irr-db-dump.json")

    db_rows = fetch_route_objects()

    # Sanity check.
    if len(db_rows) <= 4_000_000:
        logging.error(
            f"{len(db_rows)} route objects is below threshold. Do not "
            f"overwrite the {output_file}. Abort."
        )
        sys.exit(1)

    route_objects = parse_route_objects(db_rows)

    write_route_objects(route_objects, output_file)


if __name__ == "__main__":
    main()