#!/usr/bin/env python3 """Summarize an existing monocle JSONL file for BGP communities in 65535:*. Usage: python3 summarize_existing_well_known.py INPUT.jsonl OUT_PREFIX Writes: -summary.json -by-community.csv -blackhole.csv -nopeer.csv The script normalizes symbolic well-known community names that monocle may emit (e.g. "no-export") back to their numeric 65535:* values before grouping. """ import csv import ipaddress import json import sys from collections import Counter, defaultdict from pathlib import Path NAME_TO_NUMERIC = { "no-export": "65535:65281", "no_advertise": "65535:65282", "no-advertise": "65535:65282", "no-export-subconfed": "65535:65283", "no_export_subconfed": "65535:65283", "nopeer": "65535:65284", "blackhole": "65535:666", "graceful-shutdown": "65535:0", "accept-own": "65535:1", "llgr-stale": "65535:6", "no-llgr": "65535:7", } FIELDNAMES_BY_COMMUNITY = [ "community", "rows", "distinct_prefixes", "distinct_collectors", "distinct_peer_asns", "distinct_origin_asns", "afi", "top_collectors", "top_peer_asns", "top_origin_asns", "top_prefix_lengths", ] SAMPLE_FIELDNAMES = [ "timestamp", "collector", "peer_asn", "origin_asns", "prefix", "afi", "prefix_len", "longest_prefix", "has_no_export", "has_no_advertise", "as_path", "communities", ] def usage() -> None: print("usage: summarize_existing_well_known.py INPUT.jsonl OUT_PREFIX", file=sys.stderr) def normalize_communities(communities: list[str]) -> list[str]: """Return communities with symbolic well-known names converted to numeric form.""" return [NAME_TO_NUMERIC.get(community.lower(), community) for community in communities] def matching_well_known_values(communities: list[str]) -> list[str]: """Return unique numeric 65535:* communities from a normalized community list.""" return sorted( { community for community in communities if community.startswith("65535:") and community.count(":") == 1 } ) def prefix_details(prefix: str) -> tuple[str, int | None, int | None]: """Return (afi, prefix_length, max_prefix_length) for a prefix string.""" try: network = ipaddress.ip_network(prefix, strict=False) except ValueError: return "unknown", None, None afi = "ipv6" if network.version == 6 else "ipv4" max_length = 128 if network.version == 6 else 32 return afi, network.prefixlen, max_length def top(counter: Counter, limit: int = 10) -> list[dict[str, int | str]]: """Return a JSON-friendly top-N list from a Counter.""" return [{"key": key, "count": count} for key, count in counter.most_common(limit)] def new_summary_bucket() -> dict[str, int | Counter]: return { "rows": 0, "prefixes": Counter(), "collectors": Counter(), "peer_asns": Counter(), "origin_asns": Counter(), "prefix_lengths": Counter(), "afi": Counter(), } def write_by_community_csv(path: Path, communities: dict) -> None: with path.open("w", newline="") as output: writer = csv.writer(output) writer.writerow(FIELDNAMES_BY_COMMUNITY) for community, details in communities.items(): writer.writerow( [ community, details["rows"], details["distinct_prefixes"], details["distinct_collectors"], details["distinct_peer_asns"], details["distinct_origin_asns"], json.dumps(details["afi"]), json.dumps(details["top_collectors"][:5]), json.dumps(details["top_peer_asns"][:5]), json.dumps(details["top_origin_asns"][:5]), json.dumps(details["top_prefix_lengths"][:8]), ] ) def write_sample_csv(path: Path, rows: list[dict]) -> None: if not rows: return with path.open("w", newline="") as output: writer = csv.DictWriter(output, fieldnames=SAMPLE_FIELDNAMES) writer.writeheader() writer.writerows(rows) def main() -> int: if len(sys.argv) != 3: usage() return 2 input_path = Path(sys.argv[1]) output_prefix = Path(sys.argv[2]) well_known = defaultdict(new_summary_bucket) total_rows = 0 blackhole_rows = [] nopeer_rows = [] with input_path.open() as input_file: for line_number, line in enumerate(input_file, start=1): line = line.strip() if not line: continue total_rows += 1 try: record = json.loads(line) except json.JSONDecodeError as error: print(f"bad json line {line_number}: {error}", file=sys.stderr) continue original_communities = [str(c) for c in record.get("communities") or []] normalized_communities = normalize_communities(original_communities) values = matching_well_known_values(normalized_communities) if not values: continue prefix = record.get("prefix") or "" afi, prefix_length, max_prefix_length = prefix_details(prefix) origin_asns = [str(asn) for asn in (record.get("origin_asns") or [])] for value in values: bucket = well_known[value] bucket["rows"] += 1 if prefix: bucket["prefixes"][prefix] += 1 if record.get("collector"): bucket["collectors"][str(record["collector"])] += 1 if record.get("peer_asn") is not None: bucket["peer_asns"][str(record["peer_asn"])] += 1 for origin_asn in origin_asns: bucket["origin_asns"][origin_asn] += 1 if prefix_length is not None: bucket["prefix_lengths"][str(prefix_length)] += 1 bucket["afi"][afi] += 1 sample = { "timestamp": record.get("timestamp"), "collector": record.get("collector"), "peer_asn": record.get("peer_asn"), "origin_asns": " ".join(origin_asns), "prefix": prefix, "afi": afi, "prefix_len": prefix_length, "as_path": record.get("as_path"), "communities": " ".join(original_communities), } if "65535:666" in values: blackhole_rows.append( { **sample, "longest_prefix": ( prefix_length == max_prefix_length if prefix_length is not None else None ), "has_no_export": "65535:65281" in normalized_communities, "has_no_advertise": "65535:65282" in normalized_communities, } ) if "65535:65284" in values: nopeer_rows.append(sample) summary = {"input": str(input_path), "rows": total_rows, "communities": {}} for value, bucket in sorted(well_known.items(), key=lambda item: (-item[1]["rows"], item[0])): summary["communities"][value] = { "rows": bucket["rows"], "distinct_prefixes": len(bucket["prefixes"]), "distinct_collectors": len(bucket["collectors"]), "distinct_peer_asns": len(bucket["peer_asns"]), "distinct_origin_asns": len(bucket["origin_asns"]), "top_collectors": top(bucket["collectors"]), "top_peer_asns": top(bucket["peer_asns"]), "top_origin_asns": top(bucket["origin_asns"]), "top_prefix_lengths": top(bucket["prefix_lengths"]), "afi": dict(bucket["afi"]), } summary_path = output_prefix.with_name(output_prefix.name + "-summary.json") by_community_path = output_prefix.with_name(output_prefix.name + "-by-community.csv") blackhole_path = output_prefix.with_name(output_prefix.name + "-blackhole.csv") nopeer_path = output_prefix.with_name(output_prefix.name + "-nopeer.csv") with summary_path.open("w") as output: json.dump(summary, output, indent=2, sort_keys=True) write_by_community_csv(by_community_path, summary["communities"]) write_sample_csv(blackhole_path, blackhole_rows) write_sample_csv(nopeer_path, nopeer_rows) print( json.dumps( { "rows": total_rows, "communities": len(well_known), "blackhole_rows": len(blackhole_rows), "nopeer_rows": len(nopeer_rows), }, indent=2, ) ) return 0 if __name__ == "__main__": raise SystemExit(main())