reputation: decode dnsbl 127.255.255.x error range as unavailable

src/netcheck/reputation.py

@@ -4,11 +4,13 @@ import ipaddress
 import os
 import time
 from bisect import bisect_right
+from dataclasses import dataclass, field
 from pathlib import Path
 
 import httpx
 
 from netcheck.config import Providers
+from netcheck.models import DnsblHit
 
 
 def parse_netset(text: str) -> list[str]:
@@ -91,3 +93,55 @@ async def refresh_netsets(
                     continue
         index.add(name, parse_netset(path.read_text(encoding="utf-8")))
     return index
+
+
+# Spamhaus (and its mirrors) return codes in 127.255.255.0/24 to signal a
+# problem with the QUERY, not a property of the queried IP. .254 means the
+# question arrived via a public resolver, which is the normal case for anyone
+# on 1.1.1.1 or 8.8.8.8; .255 means rate limited. Reading either as "listed"
+# red-flags most users, which is exactly the false positive this decoder exists
+# to prevent.
+DNSBL_ERROR_PREFIX = "127.255.255."
+_ERROR_REASONS = {
+    "127.255.255.254": "query_via_public_resolver",
+    "127.255.255.255": "rate_limited",
+}
+
+
+@dataclass
+class DnsblOutcome:
+    zone: str
+    listed: bool = False
+    codes: list[str] = field(default_factory=list)
+    unavailable_reason: str | None = None
+
+
+def reverse_ip(ip: str) -> str:
+    address = ipaddress.ip_address(ip)
+    if address.version != 4:
+        raise ValueError("classic DNSBL zones accept IPv4 only")
+    return ".".join(reversed(str(address).split(".")))
+
+
+def decode_dnsbl(zone: str, answers: list[str]) -> DnsblOutcome:
+    codes = list(answers)
+    errors = [code for code in codes if code.startswith(DNSBL_ERROR_PREFIX)]
+    if errors:
+        return DnsblOutcome(
+            zone=zone,
+            listed=False,
+            codes=codes,
+            unavailable_reason=_ERROR_REASONS.get(errors[0], "provider_error"),
+        )
+    listed = any(code.startswith("127.") for code in codes)
+    return DnsblOutcome(zone=zone, listed=listed, codes=codes, unavailable_reason=None)
+
+
+def summarize_dnsbl(outcomes: list[DnsblOutcome]) -> tuple[list[DnsblHit], bool]:
+    hits = [
+        DnsblHit(zone=o.zone, codes=o.codes, meaning="listed")
+        for o in outcomes
+        if o.listed
+    ]
+    blocked = any(o.unavailable_reason is not None for o in outcomes)
+    return hits, blocked

tests/test_dnsbl_decode.py

@@ -0,0 +1,87 @@
+from __future__ import annotations
+
+import pytest
+
+from netcheck.reputation import decode_dnsbl, reverse_ip, summarize_dnsbl
+
+
+def test_reverse_ip_builds_the_dnsbl_query_label():
+    assert reverse_ip("203.0.113.44") == "44.113.0.203"
+    assert reverse_ip("8.8.8.8") == "8.8.8.8"
+
+
+def test_reverse_ip_rejects_ipv6_because_classic_dnsbls_are_v4_only():
+    with pytest.raises(ValueError):
+        reverse_ip("2001:db8::1")
+
+
+def test_a_real_spamhaus_listing_is_reported_as_listed():
+    outcome = decode_dnsbl("zen.spamhaus.org", ["127.0.0.2", "127.0.0.4"])
+    assert outcome.listed is True
+    assert outcome.unavailable_reason is None
+    assert outcome.codes == ["127.0.0.2", "127.0.0.4"]
+
+
+def test_public_resolver_error_code_is_not_a_listing():
+    outcome = decode_dnsbl("zen.spamhaus.org", ["127.255.255.254"])
+    assert outcome.listed is False
+    assert outcome.unavailable_reason == "query_via_public_resolver"
+    assert outcome.codes == ["127.255.255.254"]
+
+
+def test_rate_limit_error_code_is_not_a_listing():
+    outcome = decode_dnsbl("zen.spamhaus.org", ["127.255.255.255"])
+    assert outcome.listed is False
+    assert outcome.unavailable_reason == "rate_limited"
+
+
+@pytest.mark.parametrize(
+    "code",
+    ["127.255.255.252", "127.255.255.253", "127.255.255.0", "127.255.255.1"],
+)
+def test_every_other_code_in_the_error_range_is_a_provider_error_not_a_listing(code):
+    outcome = decode_dnsbl("zen.spamhaus.org", [code])
+    assert outcome.listed is False
+    assert outcome.unavailable_reason == "provider_error"
+
+
+def test_an_error_code_mixed_with_a_listing_code_still_means_unavailable():
+    # A response carrying an error code cannot be trusted for the listing bits.
+    outcome = decode_dnsbl("zen.spamhaus.org", ["127.0.0.2", "127.255.255.254"])
+    assert outcome.listed is False
+    assert outcome.unavailable_reason == "query_via_public_resolver"
+
+
+def test_no_answers_means_not_listed_and_no_error():
+    outcome = decode_dnsbl("bl.spamcop.net", [])
+    assert outcome.listed is False
+    assert outcome.unavailable_reason is None
+    assert outcome.codes == []
+
+
+def test_a_non_loopback_answer_is_ignored_as_a_listing():
+    outcome = decode_dnsbl("dnsbl.dronebl.org", ["10.0.0.1"])
+    assert outcome.listed is False
+    assert outcome.unavailable_reason is None
+
+
+def test_summarize_returns_only_real_listings():
+    outcomes = [
+        decode_dnsbl("zen.spamhaus.org", ["127.255.255.254"]),
+        decode_dnsbl("bl.spamcop.net", ["127.0.0.2"]),
+        decode_dnsbl("b.barracudacentral.org", []),
+    ]
+    hits, blocked = summarize_dnsbl(outcomes)
+    assert [h.zone for h in hits] == ["bl.spamcop.net"]
+    assert hits[0].codes == ["127.0.0.2"]
+    assert blocked is True
+
+
+def test_summarize_reports_not_blocked_when_every_zone_answered_cleanly():
+    outcomes = [
+        decode_dnsbl("zen.spamhaus.org", []),
+        decode_dnsbl("bl.spamcop.net", []),
+    ]
+    hits, blocked = summarize_dnsbl(outcomes)
+    assert hits == []
+    assert blocked is False