diff --git a/src/netcheck/reputation.py b/src/netcheck/reputation.py index 40456a4..d80a7b2 100644 --- a/src/netcheck/reputation.py +++ b/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 diff --git a/tests/test_dnsbl_decode.py b/tests/test_dnsbl_decode.py new file mode 100644 index 0000000..cb7d29d --- /dev/null +++ b/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