rpki/scripts/ta_constraints/generate_afrinic_current_holdings.py

#!/usr/bin/env python3
"""Generate a current-only AFRINIC IPv4 holdings constraints fixture.

The AFRINIC delegated extended format contains one IPv4 allocation/assignment
per line.  This tool keeps only active registry holdings (``allocated`` and
``assigned``), computes their exact union, and emits an allow-all policy with
the union as IPv4 deny entries.  ``reserved`` and ``available`` records are
intentionally excluded.
"""

from __future__ import annotations

import argparse
import datetime as dt
import hashlib
import ipaddress
import pathlib
import sys
import urllib.error
import urllib.request
from collections import Counter
from typing import Iterable


DEFAULT_SOURCE_URL = (
    "https://ftp.afrinic.net/pub/stats/afrinic/"
    "delegated-afrinic-extended-latest"
)
ACTIVE_STATUSES = {"allocated", "assigned"}


def merge_intervals(
    intervals: Iterable[tuple[int, int]],
) -> list[tuple[int, int]]:
    merged: list[tuple[int, int]] = []
    for start, end in sorted(intervals):
        if merged and start <= merged[-1][1] + 1:
            merged[-1] = (merged[-1][0], max(merged[-1][1], end))
        else:
            merged.append((start, end))
    return merged


def parse_source(raw: bytes) -> tuple[list[tuple[int, int]], Counter[str], Counter[str]]:
    status_counts: Counter[str] = Counter()
    ipv4_status_counts: Counter[str] = Counter()
    active: list[tuple[int, int]] = []
    for line_number, raw_line in enumerate(raw.decode("utf-8", "replace").splitlines(), 1):
        if not raw_line or raw_line.startswith("#"):
            continue
        fields = raw_line.split("|")
        if len(fields) < 7 or fields[2] != "ipv4":
            continue
        status = fields[6]
        status_counts[status] += 1
        ipv4_status_counts[status] += 1
        if status not in ACTIVE_STATUSES:
            continue
        try:
            start = int(ipaddress.IPv4Address(fields[3]))
            count = int(fields[4])
        except (ValueError, TypeError) as error:
            raise ValueError(f"line {line_number}: invalid IPv4 record: {raw_line}") from error
        if count <= 0 or start + count > 1 << 32:
            raise ValueError(f"line {line_number}: invalid IPv4 range: {raw_line}")
        active.append((start, start + count - 1))
    return active, status_counts, ipv4_status_counts


def render(
    source_url: str,
    retrieved: str,
    source_sha256: str,
    source_ipv4_count: int,
    status_counts: Counter[str],
    active_source: list[tuple[int, int]],
) -> tuple[str, dict[str, int]]:
    merged = merge_intervals(active_source)
    networks = [
        network
        for start, end in merged
        for network in ipaddress.summarize_address_range(
            ipaddress.IPv4Address(start), ipaddress.IPv4Address(end)
        )
    ]
    networks.sort(key=lambda network: int(network.network_address))
    active_addresses = sum(end - start + 1 for start, end in merged)
    counts = {
        "source_ipv4_records": source_ipv4_count,
        "allocated": status_counts["allocated"],
        "assigned": status_counts["assigned"],
        "reserved": status_counts["reserved"],
        "available": status_counts["available"],
        "selected_source_records": len(active_source),
        "selected_addresses": active_addresses,
        "merged_intervals": len(merged),
        "cidr_rules": len(networks),
    }
    lines = [
        "# Feature #151 M7 current AFRINIC IPv4 holdings deny fixture.",
        "#",
        "# Source: AFRINIC delegated extended latest.",
        f"# URL: {source_url}",
        f"# Retrieved: {retrieved}",
        f"# Source SHA-256: {source_sha256}",
        "#",
        "# Selection: IPv4 records with status allocated or assigned only.",
        "# Excluded statuses: reserved and available (not current registry holdings).",
        (
            "# Source IPv4 records: {source_ipv4_records}; allocated={allocated}; "
            "assigned={assigned}; reserved={reserved}; available={available}."
        ).format(**counts),
        "# Selected source records: {selected_source_records}; selected addresses: {selected_addresses}.".format(
            **counts
        ),
        "# Overlap/adjacency union intervals: {merged_intervals}; exact CIDR deny rules: {cidr_rules}.".format(
            **counts
        ),
        "# IPv6 and ASN remain allow-all to isolate the current AFRINIC IPv4 policy.",
        "allow 0.0.0.0/0",
        "allow ::/0",
        "allow 0 - 4294967295",
        "",
    ]
    lines.extend(f"deny {network}" for network in networks)
    return "\n".join(lines) + "\n", counts


def fetch_source(url: str, timeout: float) -> bytes:
    request = urllib.request.Request(
        url,
        headers={"User-Agent": "rpki-dev-feature151-afrinic-holdings/1.0"},
    )
    with urllib.request.urlopen(request, timeout=timeout) as response:
        return response.read()


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    source = parser.add_mutually_exclusive_group()
    source.add_argument("--url", default=DEFAULT_SOURCE_URL, help="AFRINIC source URL")
    source.add_argument("--input", type=pathlib.Path, help="local delegated file for replay")
    parser.add_argument("--output", type=pathlib.Path, required=True)
    parser.add_argument("--retrieved", help="UTC date recorded in the fixture header")
    parser.add_argument("--timeout", type=float, default=60.0)
    args = parser.parse_args()

    source_url = args.url if args.input is None else f"file://{args.input.resolve()}"
    raw = args.input.read_bytes() if args.input is not None else fetch_source(args.url, args.timeout)
    retrieved = args.retrieved or dt.datetime.now(dt.timezone.utc).date().isoformat()
    active, status_counts, ipv4_status_counts = parse_source(raw)
    rendered, counts = render(
        source_url,
        retrieved,
        hashlib.sha256(raw).hexdigest(),
        sum(ipv4_status_counts.values()),
        status_counts,
        active,
    )
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(rendered, encoding="utf-8")
    print(
        "generated {output}: {cidr_rules} CIDRs from {selected_source_records} active "
        "records ({selected_addresses} addresses); source_sha256={sha256}".format(
            output=args.output,
            sha256=hashlib.sha256(raw).hexdigest(),
            **counts,
        )
    )
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except (OSError, ValueError, urllib.error.URLError) as error:
        print(f"error: {error}", file=sys.stderr)
        raise SystemExit(2)