From 1874f3f3c2b1bad00f15e12579a40e764c30545e Mon Sep 17 00:00:00 2001 From: Andrew Pimenov Date: Sun, 9 Aug 2026 17:45:37 +0300 Subject: [PATCH] add Cloudflare CF-RAY edge/anycast path diversity probe --- src/netsleuth/probes/bgp_path.py | 169 +++++++++++++++++++++++++++++++ tests/test_bgp_path.py | 116 +++++++++++++++++++++ 2 files changed, 285 insertions(+) create mode 100644 src/netsleuth/probes/bgp_path.py create mode 100644 tests/test_bgp_path.py diff --git a/src/netsleuth/probes/bgp_path.py b/src/netsleuth/probes/bgp_path.py new file mode 100644 index 0000000..d9c6ad4 --- /dev/null +++ b/src/netsleuth/probes/bgp_path.py @@ -0,0 +1,169 @@ +from __future__ import annotations + +import re + +import httpx + +from netsleuth.ip_geo import parse_cf_trace +from netsleuth.models import AnycastHop, PathDiversity +from netsleuth.probes.latency import tcp_connect_rtt +from netsleuth.speed import parse_server_timing_cfl4 + +_CF_RAY_RE = re.compile(r"-([A-Za-z]{2,4})$") + +COLO_LOCATIONS: dict[str, tuple[str, str]] = { + "DME": ("Moscow", "RU"), + "SVO": ("Moscow", "RU"), + "LED": ("Saint Petersburg", "RU"), + "KJA": ("Krasnoyarsk", "RU"), + "HEL": ("Helsinki", "FI"), + "ARN": ("Stockholm", "SE"), + "RIX": ("Riga", "LV"), + "TLL": ("Tallinn", "EE"), + "VNO": ("Vilnius", "LT"), + "WAW": ("Warsaw", "PL"), + "FRA": ("Frankfurt", "DE"), + "AMS": ("Amsterdam", "NL"), + "CDG": ("Paris", "FR"), + "LHR": ("London", "GB"), + "VIE": ("Vienna", "AT"), + "PRG": ("Prague", "CZ"), + "BUD": ("Budapest", "HU"), + "OTP": ("Bucharest", "RO"), + "SOF": ("Sofia", "BG"), + "IST": ("Istanbul", "TR"), + "KBP": ("Kyiv", "UA"), + "ALA": ("Almaty", "KZ"), + "TAS": ("Tashkent", "UZ"), + "TBS": ("Tbilisi", "GE"), + "EVN": ("Yerevan", "AM"), + "DXB": ("Dubai", "AE"), + "SIN": ("Singapore", "SG"), + "NRT": ("Tokyo", "JP"), + "LAX": ("Los Angeles", "US"), + "IAD": ("Ashburn", "US"), + "ORD": ("Chicago", "US"), + "SEA": ("Seattle", "US"), +} + + +def parse_cf_ray(header: str | None) -> str | None: + if not header: + return None + match = _CF_RAY_RE.search(header) + if not match: + return None + return match.group(1).upper() + + +def colo_location(code: str | None) -> tuple[str | None, str | None]: + if not code: + return None, None + return COLO_LOCATIONS.get(code.upper(), (None, None)) + + +def detect_international_loop( + client_country: str | None, ip_country: str | None, edge_country: str | None +) -> tuple[bool, list[str]]: + if not client_country or not ip_country or client_country != ip_country: + return False, [] + if not edge_country or edge_country == client_country: + return False, [] + return True, [edge_country] + + +def build_path_diversity(client_country: str | None, hops: list[AnycastHop]) -> PathDiversity: + if not hops: + return PathDiversity(client_country=client_country, hops=[]) + + international_loop = False + detour_countries: list[str] = [] + for hop in hops: + is_loop, countries = detect_international_loop(client_country, hop.ip_country, hop.edge_country) + if is_loop: + international_loop = True + for country in countries: + if country not in detour_countries: + detour_countries.append(country) + + if international_loop: + note = ( + f"Anycast routed traffic through {', '.join(detour_countries)} " + f"even though the client is in {client_country}." + ) + note_ru = ( + f"Anycast завернул трафик через {', '.join(detour_countries)}, " + f"хотя клиент находится в {client_country}." + ) + else: + note = "No international routing loop detected." + note_ru = "Международной петли в маршруте не обнаружено." + + return PathDiversity( + client_country=client_country, + hops=hops, + international_loop=international_loop, + detour_countries=detour_countries, + note=note, + note_ru=note_ru, + ) + + +async def probe_edge( + client: httpx.AsyncClient, target: str, *, timeout: float, source_ip: str | None = None +) -> AnycastHop: + try: + response = await client.get(f"https://{target}/cdn-cgi/trace", timeout=timeout) + except httpx.HTTPError: + return AnycastHop(target=target, source="none") + + client_rtt_ms = await tcp_connect_rtt(target, timeout=timeout, source_ip=source_ip) + + edge_colo = parse_cf_ray(response.headers.get("cf-ray")) + edge_rtt_ms = None + source = "none" + + if edge_colo: + source = "cf_ray" + else: + cfl4 = parse_server_timing_cfl4(response.headers.get("server-timing", "")) + if cfl4: + edge_rtt_ms = cfl4.rtt_ms + source = "server_timing" + + resolved_ip = None + cf_trace = parse_cf_trace(response.text) if response.text else None + if cf_trace: + resolved_ip = cf_trace.ip + if not edge_colo and cf_trace.colo: + edge_colo = cf_trace.colo + source = "cf_trace" + + edge_city, edge_country = colo_location(edge_colo) + + return AnycastHop( + target=target, + resolved_ip=resolved_ip, + edge_colo=edge_colo, + edge_city=edge_city, + edge_country=edge_country, + edge_rtt_ms=edge_rtt_ms, + client_rtt_ms=client_rtt_ms, + source=source, + ) + + +async def collect_path_diversity( + client: httpx.AsyncClient, + targets: list[tuple[str, str]], + client_country: str | None, + *, + max_targets: int, + timeout: float, + source_ip: str | None = None, +) -> PathDiversity: + hops = [ + await probe_edge(client, host, timeout=timeout, source_ip=source_ip) + for _label, host in targets[:max_targets] + ] + return build_path_diversity(client_country, hops) diff --git a/tests/test_bgp_path.py b/tests/test_bgp_path.py new file mode 100644 index 0000000..01866b1 --- /dev/null +++ b/tests/test_bgp_path.py @@ -0,0 +1,116 @@ +from __future__ import annotations + +import httpx +import pytest +from pytest_httpx import HTTPXMock + +from netsleuth.models import AnycastHop, PathDiversity +from netsleuth.probes.bgp_path import ( + build_path_diversity, + colo_location, + detect_international_loop, + parse_cf_ray, + probe_edge, +) + +_CYRILLIC = set("абвгдеёжзийклмнопрстуфхцчшщъыьэюя") + + +def has_cyrillic(text: str) -> bool: + return any(ch in _CYRILLIC for ch in (text or "").lower()) + + +def test_parse_cf_ray_extracts_colo_code(): + assert parse_cf_ray("9a1b2c3d4e5f6789-DME") == "DME" + + +def test_parse_cf_ray_normalizes_lowercase(): + assert parse_cf_ray("9a1b2c3d4e5f6789-dme") == "DME" + + +def test_parse_cf_ray_none_input(): + assert parse_cf_ray(None) is None + + +def test_parse_cf_ray_empty_string(): + assert parse_cf_ray("") is None + + +def test_parse_cf_ray_no_dash(): + assert parse_cf_ray("nodash") is None + + +def test_parse_cf_ray_trailing_dash_empty_suffix(): + assert parse_cf_ray("trailing-") is None + + +def test_colo_location_known_code(): + city, country = colo_location("DME") + assert country == "RU" + assert city + + +def test_colo_location_unknown_code(): + assert colo_location("ZZZ") == (None, None) + + +def test_colo_location_none(): + assert colo_location(None) == (None, None) + + +def test_detect_international_loop_true_when_edge_diverges(): + assert detect_international_loop("RU", "RU", "DE") == (True, ["DE"]) + + +def test_detect_international_loop_false_when_edge_matches_client(): + assert detect_international_loop("RU", "RU", "RU") == (False, []) + + +def test_detect_international_loop_false_when_edge_unknown(): + assert detect_international_loop("RU", "RU", None) == (False, []) + + +def test_detect_international_loop_false_when_client_country_missing(): + assert detect_international_loop(None, "RU", "DE") == (False, []) + + +def test_build_path_diversity_with_looping_hop(): + hop = AnycastHop(target="example.com", ip_country="RU", edge_country="DE", source="cf_ray") + diversity = build_path_diversity("RU", [hop]) + assert diversity.international_loop is True + assert diversity.detour_countries + assert has_cyrillic(diversity.note_ru) + + +def test_build_path_diversity_with_empty_hops_is_default(): + assert build_path_diversity("RU", []) == PathDiversity(client_country="RU", hops=[]) + + +def test_anycast_hop_rejects_unknown_source(): + with pytest.raises(ValueError): + AnycastHop(target="x", source="bogus") + + +@pytest.mark.asyncio +async def test_probe_edge_uses_cf_ray_header(httpx_mock: HTTPXMock): + httpx_mock.add_response( + url="https://example.com/cdn-cgi/trace", + headers={"cf-ray": "abc123-FRA"}, + text="ip=1.2.3.4\ncolo=FRA\n", + ) + + async with httpx.AsyncClient() as client: + hop = await probe_edge(client, "example.com", timeout=2.0) + + assert hop.edge_colo == "FRA" + assert hop.source == "cf_ray" + + +@pytest.mark.asyncio +async def test_probe_edge_swallows_connection_error(httpx_mock: HTTPXMock): + httpx_mock.add_exception(httpx.ConnectError("boom")) + + async with httpx.AsyncClient() as client: + hop = await probe_edge(client, "example.com", timeout=2.0) + + assert hop == AnycastHop(target="example.com", source="none")