#!/usr/bin/env python3 """Summarize monocle JSONL output for BGP communities in 65535:*. Usage: monocle search ... --format json-line | python3 summarize_well_known.py OUT_PREFIX Writes: .jsonl -summary.json -by-community.csv -blackhole.csv -nopeer.csv This is a stream-oriented variant of summarize_existing_well_known.py. It reads JSONL from stdin and saves a copy to .jsonl while summarizing. """ 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", } BY_COMMUNITY_FIELDS = [ "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_FIELDS = [ "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_well_known.py OUT_PREFIX", file=sys.stderr) def new_bucket() -> dict[str, int | Counter]: return { "rows": 0, "prefixes": Counter(), "collectors": Counter(), "peer_asns": Counter(), "origin_asns": Counter(), "prefix_lengths": Counter(), "afi": Counter(), } def normalize_communities(communities: list[str]) -> list[str]: return [NAME_TO_NUMERIC.get(community.lower(), community) for community in communities] def matching_values(communities: list[str]) -> list[str]: 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]: 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 [{"key": key, "count": count} for key, count in counter.most_common(limit)] def write_by_community(path: Path, communities: dict) -> None: with path.open("w", newline="") as output: writer = csv.writer(output) writer.writerow(BY_COMMUNITY_FIELDS) 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_rows(path: Path, rows: list[dict]) -> None: if not rows: return with path.open("w", newline="") as output: writer = csv.DictWriter(output, fieldnames=SAMPLE_FIELDS) writer.writeheader() writer.writerows(rows) def process_record(record: dict, well_known: defaultdict, blackhole_rows: list, nopeer_rows: list) -> None: original_communities = [str(community) for community in record.get("communities") or []] normalized_communities = normalize_communities(original_communities) values = matching_values(normalized_communities) if not values: return 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) def build_summary(total_rows: int, well_known: defaultdict) -> dict: summary = {"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"]), } return summary def main() -> int: if len(sys.argv) != 2: usage() return 2 output_prefix = Path(sys.argv[1]) raw_path = output_prefix.with_suffix(".jsonl") well_known = defaultdict(new_bucket) blackhole_rows = [] nopeer_rows = [] total_rows = 0 with raw_path.open("w") as raw_output: for line_number, line in enumerate(sys.stdin, start=1): line = line.strip() if not line: continue total_rows += 1 raw_output.write(line + "\n") try: record = json.loads(line) except json.JSONDecodeError as error: print(f"bad json line {line_number}: {error}", file=sys.stderr) continue process_record(record, well_known, blackhole_rows, nopeer_rows) summary = build_summary(total_rows, well_known) 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(by_community_path, summary["communities"]) write_sample_rows(blackhole_path, blackhole_rows) write_sample_rows(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())