bgp: asrank, team cymru and disk-cached peeringdb parsing

src/netcheck/bgp.py

@@ -102,3 +102,72 @@ async def ripestat(
     response = await client.get(f"{providers.ripestat_base_url}/{call}/data.json", params=params)
     response.raise_for_status()
     return response.json()
+
+
+import json
+import os
+import time
+from pathlib import Path
+from typing import Awaitable, Callable
+
+from netcheck.models import IxpPresence
+
+
+def parse_asrank(payload: dict) -> tuple[int | None, int | None, int | None]:
+    asn = ((payload.get("data") or {}).get("asn")) or {}
+    cone = asn.get("cone") or {}
+    return asn.get("rank"), cone.get("numberAsns"), cone.get("numberPrefixes")
+
+
+def parse_cymru_origin(txt_record: str) -> dict[str, str]:
+    parts = [part.strip() for part in txt_record.split("|")]
+    if len(parts) < 5:
+        return {}
+    return {
+        "asn": _as_label(parts[0].split()[0]),
+        "prefix": parts[1],
+        "country": parts[2],
+        "registry": parts[3],
+        "allocated_at": parts[4],
+    }
+
+
+def parse_peeringdb_net(payload: dict) -> tuple[str | None, str | None, int | None]:
+    rows = payload.get("data") or []
+    if not rows:
+        return None, None, None
+    row = rows[0]
+    return row.get("info_type") or None, row.get("info_traffic") or None, row.get("id")
+
+
+def parse_peeringdb_netixlan(payload: dict) -> list[IxpPresence]:
+    return [
+        IxpPresence(
+            name=row.get("name", ""),
+            city=row.get("city"),
+            country=row.get("country"),
+            speed_mbps=row.get("speed"),
+        )
+        for row in payload.get("data") or []
+        if row.get("operational")
+    ]
+
+
+async def cached_json(
+    cache_dir: Path,
+    key: str,
+    ttl_hours: int,
+    fetch: Callable[[], Awaitable[dict]],
+) -> dict:
+    path = Path(cache_dir) / f"{key}.json"
+    if path.exists() and (time.time() - path.stat().st_mtime) < ttl_hours * 3600:
+        try:
+            return json.loads(path.read_text(encoding="utf-8"))
+        except (OSError, ValueError):
+            pass
+    payload = await fetch()
+    path.parent.mkdir(parents=True, exist_ok=True)
+    tmp = path.with_suffix(".json.tmp")
+    tmp.write_text(json.dumps(payload), encoding="utf-8")
+    os.replace(tmp, path)
+    return payload

tests/fixtures/api/asrank.json

@@ -0,0 +1,12 @@
+{
+  "data": {
+    "asn": {
+      "asn": "64500",
+      "rank": 1842,
+      "asnName": "EXAMPLE-AS",
+      "organization": { "orgName": "Example Telecom BV" },
+      "cone": { "numberAsns": 37, "numberPrefixes": 412, "numberAddresses": 1048576 },
+      "asnDegree": { "provider": 2, "peer": 1, "customer": 4, "total": 7 }
+    }
+  }
+}

tests/fixtures/api/peeringdb_net.json

@@ -0,0 +1,13 @@
+{
+  "data": [
+    {
+      "id": 4242,
+      "asn": 64500,
+      "name": "Example Telecom",
+      "info_type": "Cable/DSL/ISP",
+      "info_traffic": "100-200Gbps",
+      "info_scope": "Regional",
+      "policy_general": "Open"
+    }
+  ]
+}

tests/fixtures/api/peeringdb_netixlan.json

@@ -0,0 +1,7 @@
+{
+  "data": [
+    { "id": 1, "net_id": 4242, "name": "AMS-IX", "city": "Amsterdam", "country": "NL", "speed": 100000, "operational": true },
+    { "id": 2, "net_id": 4242, "name": "DE-CIX Frankfurt", "city": "Frankfurt", "country": "DE", "speed": 200000, "operational": true },
+    { "id": 3, "net_id": 4242, "name": "Dead-IX", "city": "Nowhere", "country": "NL", "speed": 1000, "operational": false }
+  ]
+}

tests/test_bgp.py

@@ -94,3 +94,101 @@ async def test_ripestat_does_not_bound_non_bulk_calls(httpx_mock):
     request = httpx_mock.get_request()
     assert "max_rows" not in request.url.params
     assert "starttime" not in request.url.params
+
+
+import json
+from pathlib import Path
+
+from netcheck.bgp import (
+    cached_json,
+    parse_asrank,
+    parse_cymru_origin,
+    parse_peeringdb_net,
+    parse_peeringdb_netixlan,
+)
+
+
+def test_asrank_gives_rank_and_customer_cone(api_fixture):
+    rank, cone_asns, cone_prefixes = parse_asrank(api_fixture("asrank.json"))
+    assert rank == 1842
+    assert cone_asns == 37
+    assert cone_prefixes == 412
+
+
+def test_asrank_of_an_unknown_asn_is_all_none():
+    assert parse_asrank({"data": {"asn": None}}) == (None, None, None)
+
+
+def test_cymru_origin_txt_is_split_into_fields():
+    record = "64500 | 203.0.113.0/24 | NL | ripencc | 2001-05-21"
+    assert parse_cymru_origin(record) == {
+        "asn": "AS64500",
+        "prefix": "203.0.113.0/24",
+        "country": "NL",
+        "registry": "ripencc",
+        "allocated_at": "2001-05-21",
+    }
+
+
+def test_cymru_origin_with_multiple_origin_asns_takes_the_first():
+    record = "64500 64501 | 203.0.113.0/24 | NL | ripencc | 2001-05-21"
+    assert parse_cymru_origin(record)["asn"] == "AS64500"
+
+
+def test_cymru_origin_of_garbage_is_empty():
+    assert parse_cymru_origin("no such name") == {}
+
+
+def test_peeringdb_net_gives_info_type_traffic_and_id(api_fixture):
+    info_type, traffic, net_id = parse_peeringdb_net(api_fixture("peeringdb_net.json"))
+    assert info_type == "Cable/DSL/ISP"
+    assert traffic == "100-200Gbps"
+    assert net_id == 4242
+
+
+def test_peeringdb_net_of_an_unlisted_asn_is_all_none():
+    assert parse_peeringdb_net({"data": []}) == (None, None, None)
+
+
+def test_peeringdb_netixlan_lists_only_operational_exchanges(api_fixture):
+    ixps = parse_peeringdb_netixlan(api_fixture("peeringdb_netixlan.json"))
+    assert [i.name for i in ixps] == ["AMS-IX", "DE-CIX Frankfurt"]
+    assert ixps[0].country == "NL"
+    assert ixps[1].speed_mbps == 200000
+
+
+async def test_cached_json_writes_then_reuses_the_cache(tmp_path: Path):
+    calls = []
+
+    async def fetch() -> dict:
+        calls.append(1)
+        return {"value": len(calls)}
+
+    first = await cached_json(tmp_path, "peeringdb-net-64500", ttl_hours=24, fetch=fetch)
+    second = await cached_json(tmp_path, "peeringdb-net-64500", ttl_hours=24, fetch=fetch)
+    assert first == {"value": 1}
+    assert second == {"value": 1}
+    assert len(calls) == 1
+    assert json.loads((tmp_path / "peeringdb-net-64500.json").read_text(encoding="utf-8"))["value"] == 1
+
+
+async def test_cached_json_refetches_once_the_entry_expires(tmp_path: Path):
+    calls = []
+
+    async def fetch() -> dict:
+        calls.append(1)
+        return {"value": len(calls)}
+
+    await cached_json(tmp_path, "k", ttl_hours=24, fetch=fetch)
+    result = await cached_json(tmp_path, "k", ttl_hours=0, fetch=fetch)
+    assert result == {"value": 2}
+    assert len(calls) == 2
+
+
+async def test_cached_json_survives_a_corrupt_cache_file(tmp_path: Path):
+    (tmp_path / "k.json").write_text("{not json", encoding="utf-8")
+
+    async def fetch() -> dict:
+        return {"value": "fresh"}
+
+    assert await cached_json(tmp_path, "k", ttl_hours=24, fetch=fetch) == {"value": "fresh"}