From db055d8a85f78c3f24b5f4e2c32907dcea7acf9a Mon Sep 17 00:00:00 2001 From: Andrew Pimenov Date: Sun, 9 Aug 2026 02:14:03 +0300 Subject: [PATCH] ip_geo: field-wise merge, dual-stack comparison, provider fan-out --- src/netcheck/ip_geo.py | 108 +++++++++++++++++++++++++++++++++ tests/test_ip_geo_normalize.py | 70 +++++++++++++++++++++ 2 files changed, 178 insertions(+) diff --git a/src/netcheck/ip_geo.py b/src/netcheck/ip_geo.py index 1038e60..363cdf6 100644 --- a/src/netcheck/ip_geo.py +++ b/src/netcheck/ip_geo.py @@ -166,3 +166,111 @@ def parse_cf_trace(text: str) -> CfTrace: rbi=raw.get("rbi"), raw=raw, ) + + +from dataclasses import fields as dataclass_fields + +_MERGE_SKIP = {"sources", "ip_type"} + + +def merge_geo(candidates: list[tuple[str, IpGeo]]) -> IpGeo: + merged = IpGeo() + sources: dict[str, str] = {} + for name, geo in candidates: + for f in dataclass_fields(IpGeo): + if f.name in _MERGE_SKIP: + continue + if getattr(merged, f.name) is not None: + continue + value = getattr(geo, f.name) + if value in (None, ""): + continue + setattr(merged, f.name, value) + sources[f.name] = name + if merged.ip_type == "unknown" and geo.ip_type != "unknown": + merged.ip_type = geo.ip_type + sources["ip_type"] = name + merged.sources = sources + return merged + + +def dual_stack_mismatch(v4: IpGeo | None, v6: IpGeo | None) -> str | None: + if v4 is None or v6 is None or not v4.asn or not v6.asn: + return None + if v4.asn == v6.asn and (v4.country_code or "") == (v6.country_code or ""): + return None + return ( + f"IPv4 egress {v4.ip} is {v4.asn} ({v4.country_code}) but IPv6 egress {v6.ip} " + f"is {v6.asn} ({v6.country_code}); the two stacks leave through different networks." + ) + + +import asyncio + +import httpx + +from netcheck.config import Providers + + +async def _json(client: httpx.AsyncClient, url: str, **kwargs) -> dict: + response = await client.get(url, **kwargs) + response.raise_for_status() + return response.json() + + +async def _text(client: httpx.AsyncClient, url: str) -> str: + response = await client.get(url) + response.raise_for_status() + return response.text + + +async def gather_identity( + client: httpx.AsyncClient, + providers: Providers, + ip: str | None = None, + ipinfo_token: str | None = None, +) -> tuple[IpGeo, CfTrace | None, dict[str, bool], dict[str, object]]: + suffix = ip or "" + headers = {"Authorization": f"Bearer {ipinfo_token}"} if ipinfo_token else {} + calls = { + "cf-trace": _text(client, providers.cf_trace_url), + "ip-api": _json(client, f"{providers.ip_api_url}{suffix}"), + "freeipapi": _json(client, f"{providers.freeipapi_url}{suffix}"), + "ipinfo": _json(client, f"{providers.ipinfo_url}{suffix}json", headers=headers), + "ipwho.is": _json(client, f"{providers.ipwhois_url}{suffix}"), + } + settled = await asyncio.gather(*calls.values(), return_exceptions=True) + payloads = dict(zip(calls.keys(), settled)) + + raw: dict[str, object] = {k: v for k, v in payloads.items() if not isinstance(v, BaseException)} + cf = parse_cf_trace(payloads["cf-trace"]) if isinstance(payloads["cf-trace"], str) else None + flags = provider_flags(payloads["ip-api"]) if isinstance(payloads["ip-api"], dict) else {} + + candidates: list[tuple[str, IpGeo]] = [] + if cf: + candidates.append(("cf-trace", IpGeo(ip=cf.ip, country_code=cf.loc, sources={}))) + for name, normalizer in ( + ("ip-api", normalize_ip_api), + ("freeipapi", normalize_freeipapi), + ("ipinfo", normalize_ipinfo), + ("ipwho.is", normalize_ipwhois), + ): + payload = payloads[name] + if isinstance(payload, dict): + candidates.append((name, normalizer(payload))) + + merged = merge_geo(candidates) + if merged.ip: + try: + network_info = await _json( + client, f"{providers.ripestat_base_url}/network-info/data.json", params={"resource": merged.ip} + ) + except (httpx.HTTPError, ValueError): + network_info = None + if network_info: + raw["ripestat-network-info"] = network_info + authoritative = normalize_ripestat_network_info(network_info) + if authoritative.asn: + merged.asn = authoritative.asn + merged.sources["asn"] = "ripestat" + return merged, cf, flags, raw diff --git a/tests/test_ip_geo_normalize.py b/tests/test_ip_geo_normalize.py index 5202909..b51eb5c 100644 --- a/tests/test_ip_geo_normalize.py +++ b/tests/test_ip_geo_normalize.py @@ -125,3 +125,73 @@ def test_provider_flags_of_a_failed_payload_is_empty(api_fixture): ) def test_ip_type_classification(mobile, proxy, hosting, known, expected): assert classify_ip_type(mobile, proxy, hosting, known) == expected + + +from netcheck.models import IpGeo +from netcheck.ip_geo import dual_stack_mismatch, merge_geo + + +def test_merge_takes_the_first_non_empty_value_in_priority_order(): + merged = merge_geo( + [ + ("cf-trace", IpGeo(ip="203.0.113.44")), + ("ip-api", IpGeo(asn="AS64500", city="Amsterdam", country_code="NL")), + ("ipwho.is", IpGeo(asn="AS64999", city="Rotterdam", timezone="Europe/Amsterdam")), + ] + ) + assert merged.ip == "203.0.113.44" + assert merged.asn == "AS64500" + assert merged.city == "Amsterdam" + assert merged.timezone == "Europe/Amsterdam" + + +def test_merge_records_which_provider_supplied_each_field(): + merged = merge_geo( + [ + ("cf-trace", IpGeo(ip="203.0.113.44")), + ("ip-api", IpGeo(asn="AS64500")), + ("ipwho.is", IpGeo(timezone="Europe/Amsterdam")), + ] + ) + assert merged.sources["ip"] == "cf-trace" + assert merged.sources["asn"] == "ip-api" + assert merged.sources["timezone"] == "ipwho.is" + + +def test_merge_prefers_a_known_ip_type_over_unknown(): + merged = merge_geo([("a", IpGeo(ip_type="unknown")), ("b", IpGeo(ip_type="mobile"))]) + assert merged.ip_type == "mobile" + assert merged.sources["ip_type"] == "b" + + +def test_merge_of_nothing_is_an_empty_geo(): + merged = merge_geo([]) + assert merged.ip is None + assert merged.ip_type == "unknown" + assert merged.sources == {} + + +def test_merge_ignores_a_provider_that_returned_nothing(): + merged = merge_geo([("dead", IpGeo()), ("live", IpGeo(asn="AS64500"))]) + assert merged.asn == "AS64500" + assert merged.sources["asn"] == "live" + + +def test_dual_stack_mismatch_is_reported_not_resolved(): + v4 = IpGeo(ip="203.0.113.44", asn="AS64500", country_code="NL") + v6 = IpGeo(ip="2001:db8::1", asn="AS64777", country_code="DE") + note = dual_stack_mismatch(v4, v6) + assert note is not None + assert "AS64500" in note + assert "AS64777" in note + + +def test_dual_stack_agreement_produces_no_note(): + v4 = IpGeo(ip="203.0.113.44", asn="AS64500", country_code="NL") + v6 = IpGeo(ip="2001:db8::1", asn="AS64500", country_code="NL") + assert dual_stack_mismatch(v4, v6) is None + + +def test_dual_stack_comparison_needs_both_sides(): + assert dual_stack_mismatch(IpGeo(asn="AS64500"), None) is None + assert dual_stack_mismatch(None, IpGeo(asn="AS64500")) is None