mirror of
https://github.com/MayersScott/rkn-block-checker.git
synced 2026-07-09 17:29:01 +00:00
fix: compare full DNS address sets instead of single IPs to avoid false rewriting flags
This commit is contained in:
MayersScott
parent
0591e4f4bb
commit
78e9809efe
5 changed files with 250 additions and 62 deletions
|
|
@ -36,38 +36,43 @@ def check_url(
|
|||
host = urlparse(url).hostname or url
|
||||
res = CheckResult(name=name, url=url)
|
||||
|
||||
res.sys_ip = dns_mod.resolve_system(host)
|
||||
res.doh_ip = dns_mod.resolve_doh(host, timeout=timeout)
|
||||
sys_ips = dns_mod.resolve_system_all(host)
|
||||
doh_ips = dns_mod.resolve_doh_all(host, timeout=timeout)
|
||||
|
||||
if res.sys_ip is None and res.doh_ip is not None:
|
||||
res.sys_ips = sorted(sys_ips)
|
||||
res.doh_ips = sorted(doh_ips)
|
||||
res.sys_ip = res.sys_ips[0] if res.sys_ips else None
|
||||
res.doh_ip = res.doh_ips[0] if res.doh_ips else None
|
||||
|
||||
if not sys_ips and doh_ips:
|
||||
res.verdict = Verdict.DNS_BLOCK
|
||||
res.confidence = Confidence.HIGH
|
||||
res.dns_error = "system resolver failed, DoH succeeded"
|
||||
res.notes.append(
|
||||
"system DNS doesn't resolve, DoH does — consistent with DNS poisoning"
|
||||
"system DNS doesn't resolve, DoH does - consistent with DNS poisoning"
|
||||
)
|
||||
return res
|
||||
|
||||
if res.sys_ip is None and res.doh_ip is None:
|
||||
if not sys_ips and not doh_ips:
|
||||
res.verdict = Verdict.DOWN
|
||||
res.confidence = Confidence.LOW
|
||||
res.dns_error = "domain not resolved anywhere"
|
||||
res.notes.append(
|
||||
"domain doesn't resolve via system DNS or DoH — could be NXDOMAIN, "
|
||||
"domain doesn't resolve via system DNS or DoH - could be NXDOMAIN, "
|
||||
"downed authoritative server, or DNS-level block"
|
||||
)
|
||||
return res
|
||||
|
||||
if res.sys_ip and res.doh_ip and res.sys_ip != res.doh_ip:
|
||||
if sys_ips and doh_ips and sys_ips.isdisjoint(doh_ips):
|
||||
res.dns_mismatch = True
|
||||
res.notes.append(
|
||||
f"DNS mismatch: sys={res.sys_ip} vs doh={res.doh_ip} "
|
||||
"(may indicate transparent DNS rewriting)"
|
||||
f"DNS mismatch: sys={sorted(sys_ips)} vs doh={sorted(doh_ips)} "
|
||||
"(disjoint address sets - may indicate transparent DNS rewriting)"
|
||||
)
|
||||
|
||||
if res.sys_ip is not None and res.doh_ip is None:
|
||||
if sys_ips and not doh_ips:
|
||||
res.notes.append(
|
||||
"DoH lookup failed — control comparison unavailable, "
|
||||
"DoH lookup failed - control comparison unavailable, "
|
||||
"DNS poisoning cannot be ruled out"
|
||||
)
|
||||
|
||||
|
|
@ -79,14 +84,14 @@ def check_url(
|
|||
res.verdict = Verdict.TIMEOUT
|
||||
res.confidence = Confidence.LOW
|
||||
res.notes.append(
|
||||
"TCP timeout on port 443 — could be IP block, route loss, "
|
||||
"TCP timeout on port 443 - could be IP block, route loss, "
|
||||
"or upstream congestion"
|
||||
)
|
||||
elif "reset" in (res.tcp_error or ""):
|
||||
res.verdict = Verdict.TCP_RESET
|
||||
res.confidence = Confidence.MEDIUM
|
||||
res.notes.append(
|
||||
"TCP RST received — pattern matches RST injection by a "
|
||||
"TCP RST received - pattern matches RST injection by a "
|
||||
"middlebox, but a busy server can also send RST"
|
||||
)
|
||||
else:
|
||||
|
|
@ -107,14 +112,14 @@ def check_url(
|
|||
res.verdict = Verdict.TLS_BLOCK
|
||||
res.confidence = Confidence.MEDIUM
|
||||
res.notes.append(
|
||||
"TLS reset right after ClientHello — consistent with SNI-based "
|
||||
"TLS reset right after ClientHello - consistent with SNI-based "
|
||||
"DPI filtering (typical TSPU/RKN signature), not proof"
|
||||
)
|
||||
elif "timeout" in err:
|
||||
res.verdict = Verdict.TLS_BLOCK
|
||||
res.confidence = Confidence.MEDIUM
|
||||
res.notes.append(
|
||||
"TLS handshake silently dropped — consistent with DPI filtering "
|
||||
"TLS handshake silently dropped - consistent with DPI filtering "
|
||||
"by ClientHello, but could be a flaky path"
|
||||
)
|
||||
else:
|
||||
|
|
@ -140,7 +145,7 @@ def check_url(
|
|||
if res.status_code == 451:
|
||||
res.verdict = Verdict.HTTP_STUB
|
||||
res.confidence = Confidence.HIGH
|
||||
res.notes.append("HTTP 451 — Unavailable For Legal Reasons (explicit)")
|
||||
res.notes.append("HTTP 451 - Unavailable For Legal Reasons (explicit)")
|
||||
return res
|
||||
|
||||
if http_mod.looks_like_stub(probe.body_snippet):
|
||||
|
|
@ -164,7 +169,7 @@ def iter_check_urls(
|
|||
) -> Iterator[CheckResult]:
|
||||
"""Yield CheckResult objects as soon as each probe finishes.
|
||||
|
||||
Order is *not* the input order — it's the completion order. Callers that
|
||||
Order is *not* the input order - it's the completion order. Callers that
|
||||
need the original order should sort by name (or look it up) after
|
||||
consuming the iterator. Callers that just want to print results live as
|
||||
they arrive can iterate directly.
|
||||
|
|
@ -196,9 +201,9 @@ def check_urls_parallel(
|
|||
) -> list[CheckResult]:
|
||||
"""Run all probes in parallel and return results in the original input order.
|
||||
|
||||
Backwards-compatible wrapper around iter_check_urls — used by the JSON
|
||||
Backwards-compatible wrapper around iter_check_urls - used by the JSON
|
||||
output path where streaming gives no benefit (the document has to be
|
||||
emitted as a whole anyway).
|
||||
emitted as a whole anyway)
|
||||
"""
|
||||
name_order = list(urls.keys())
|
||||
by_name = {
|
||||
|
|
|
|||
|
|
@ -13,15 +13,22 @@ DOH_ENDPOINT = "https://cloudflare-dns.com/dns-query"
|
|||
DOH_TIMEOUT = 5.0
|
||||
|
||||
|
||||
def resolve_system(host: str) -> Optional[str]:
|
||||
def resolve_system_all(host: str) -> frozenset[str]:
|
||||
try:
|
||||
return socket.gethostbyname(host)
|
||||
infos = socket.getaddrinfo(host, None, family=socket.AF_INET)
|
||||
except socket.gaierror as e:
|
||||
logger.debug("system DNS failed for %s: %s", host, e)
|
||||
return None
|
||||
return frozenset()
|
||||
return frozenset(info[4][0] for info in infos)
|
||||
|
||||
|
||||
def resolve_doh(host: str, timeout: float = DOH_TIMEOUT) -> Optional[str]:
|
||||
def resolve_doh_all(host: str, timeout: float = DOH_TIMEOUT) -> frozenset[str]:
|
||||
"""Return every IPv4 address Cloudflare's DoH endpoint returns for `host`.
|
||||
|
||||
Mirror of resolve_system_all but going over HTTPS to a public resolver,
|
||||
so callers can spot poisoning by comparing the two sets. Empty set on
|
||||
failure
|
||||
"""
|
||||
try:
|
||||
r = requests.get(
|
||||
DOH_ENDPOINT,
|
||||
|
|
@ -30,10 +37,28 @@ def resolve_doh(host: str, timeout: float = DOH_TIMEOUT) -> Optional[str]:
|
|||
timeout=timeout,
|
||||
)
|
||||
if not r.ok:
|
||||
return None
|
||||
for ans in r.json().get("Answer", []):
|
||||
if ans.get("type") == 1: # A record
|
||||
return ans.get("data")
|
||||
return frozenset()
|
||||
ips: set[str] = set()
|
||||
for ans in r.json().get("Answer", ()):
|
||||
if ans.get("type") == 1:
|
||||
data = ans.get("data")
|
||||
if data:
|
||||
ips.add(data)
|
||||
return frozenset(ips)
|
||||
except (requests.RequestException, json.JSONDecodeError) as e:
|
||||
logger.debug("DoH failed for %s: %s", host, e)
|
||||
return None
|
||||
return frozenset()
|
||||
|
||||
|
||||
def resolve_system(host: str) -> Optional[str]:
|
||||
ips = resolve_system_all(host)
|
||||
if not ips:
|
||||
return None
|
||||
return sorted(ips)[0]
|
||||
|
||||
|
||||
def resolve_doh(host: str, timeout: float = DOH_TIMEOUT) -> Optional[str]:
|
||||
ips = resolve_doh_all(host, timeout=timeout)
|
||||
if not ips:
|
||||
return None
|
||||
return sorted(ips)[0]
|
||||
|
|
|
|||
|
|
@ -53,6 +53,8 @@ class CheckResult:
|
|||
|
||||
sys_ip: Optional[str] = None
|
||||
doh_ip: Optional[str] = None
|
||||
sys_ips: list[str] = field(default_factory=list)
|
||||
doh_ips: list[str] = field(default_factory=list)
|
||||
dns_mismatch: bool = False
|
||||
dns_error: Optional[str] = None
|
||||
|
||||
|
|
|
|||
|
|
@ -5,19 +5,31 @@ from rkn_checker.http import HttpProbe
|
|||
from rkn_checker.models import Confidence, Verdict
|
||||
|
||||
|
||||
def _ipset(*ips: str) -> frozenset[str]:
|
||||
return frozenset(ips)
|
||||
|
||||
|
||||
def _patches(
|
||||
*,
|
||||
sys_ip="1.2.3.4",
|
||||
doh_ip="1.2.3.4",
|
||||
sys_ips=("1.2.3.4",),
|
||||
doh_ips=("1.2.3.4",),
|
||||
tcp=(True, 10.0, None),
|
||||
tls=(True, 20.0, "example.com", None),
|
||||
http=None,
|
||||
):
|
||||
"""Build the patch list for a check_url call.
|
||||
|
||||
`sys_ips` / `doh_ips` accept any iterable. Pass an empty tuple to
|
||||
simulate a resolution failure (NXDOMAIN, network error, etc.) — that
|
||||
used to be `sys_ip=None` in the single-IP era.
|
||||
"""
|
||||
if http is None:
|
||||
http = HttpProbe(status_code=200, elapsed_ms=100.0, body_snippet="<html>ok</html>")
|
||||
return [
|
||||
patch("rkn_checker.core.dns_mod.resolve_system", return_value=sys_ip),
|
||||
patch("rkn_checker.core.dns_mod.resolve_doh", return_value=doh_ip),
|
||||
patch("rkn_checker.core.dns_mod.resolve_system_all",
|
||||
return_value=_ipset(*sys_ips)),
|
||||
patch("rkn_checker.core.dns_mod.resolve_doh_all",
|
||||
return_value=_ipset(*doh_ips)),
|
||||
patch("rkn_checker.core.network.check_tcp", return_value=tcp),
|
||||
patch("rkn_checker.core.network.check_tls", return_value=tls),
|
||||
patch("rkn_checker.core.http_mod.fetch", return_value=http),
|
||||
|
|
@ -41,17 +53,18 @@ class TestVerdictPath:
|
|||
assert r.confidence == Confidence.HIGH
|
||||
|
||||
def test_dns_block_when_system_fails_but_doh_works(self):
|
||||
r = _run_with(_patches(sys_ip=None, doh_ip="1.2.3.4"))
|
||||
r = _run_with(_patches(sys_ips=(), doh_ips=("1.2.3.4",)))
|
||||
assert r.verdict == Verdict.DNS_BLOCK
|
||||
assert r.confidence == Confidence.HIGH
|
||||
|
||||
def test_down_when_neither_resolver_finds_domain(self):
|
||||
r = _run_with(_patches(sys_ip=None, doh_ip=None))
|
||||
r = _run_with(_patches(sys_ips=(), doh_ips=()))
|
||||
assert r.verdict == Verdict.DOWN
|
||||
assert r.confidence == Confidence.LOW
|
||||
|
||||
def test_dns_mismatch_is_flagged_but_not_fatal(self):
|
||||
r = _run_with(_patches(sys_ip="1.1.1.1", doh_ip="2.2.2.2"))
|
||||
def test_disjoint_dns_sets_flag_mismatch(self):
|
||||
# Real rewriting case: not a single overlap between the two answers.
|
||||
r = _run_with(_patches(sys_ips=("1.1.1.1",), doh_ips=("2.2.2.2",)))
|
||||
assert r.dns_mismatch is True
|
||||
assert r.verdict == Verdict.OK
|
||||
assert r.confidence == Confidence.MEDIUM
|
||||
|
|
@ -106,14 +119,68 @@ class TestVerdictPath:
|
|||
assert r.confidence == Confidence.HIGH
|
||||
|
||||
def test_doh_failure_is_noted(self):
|
||||
r = _run_with(_patches(sys_ip="1.2.3.4", doh_ip=None))
|
||||
r = _run_with(_patches(sys_ips=("1.2.3.4",), doh_ips=()))
|
||||
assert any("DoH lookup failed" in n for n in r.notes)
|
||||
|
||||
def test_doh_failure_continues_probing(self):
|
||||
r = _run_with(_patches(sys_ip="1.2.3.4", doh_ip=None))
|
||||
r = _run_with(_patches(sys_ips=("1.2.3.4",), doh_ips=()))
|
||||
assert r.verdict == Verdict.OK
|
||||
|
||||
def test_dns_mismatch_ok_gets_medium_confidence(self):
|
||||
r = _run_with(_patches(sys_ip="1.1.1.1", doh_ip="2.2.2.2"))
|
||||
|
||||
class TestDnsSetComparison:
|
||||
"""Multi-A-record sites (vk.ru, lenta.ru, anything CDN-fronted) used to
|
||||
misfire as 'transparent DNS rewriting' on every other run because the
|
||||
OS resolver rotates the answer order and we compared only the first IP
|
||||
from each side. The fix is to compare *sets*: a real rewrite means the
|
||||
two sets don't overlap at all; any shared IP means it's load balancing,
|
||||
not poisoning."""
|
||||
|
||||
def test_overlapping_pools_do_not_flag_mismatch(self):
|
||||
# Real-world example shape: sys returns three IPs in one rotation,
|
||||
# DoH returns the same three in a different order. Old code would
|
||||
# see sys_ip=A, doh_ip=B and flag mismatch every time.
|
||||
r = _run_with(_patches(
|
||||
sys_ips=("81.19.72.32", "81.19.72.33", "81.19.72.34"),
|
||||
doh_ips=("81.19.72.34", "81.19.72.32", "81.19.72.33"),
|
||||
))
|
||||
assert r.dns_mismatch is False
|
||||
assert r.verdict == Verdict.OK
|
||||
assert r.confidence == Confidence.MEDIUM
|
||||
assert r.confidence == Confidence.HIGH
|
||||
|
||||
def test_partial_overlap_does_not_flag_mismatch(self):
|
||||
# CDN edge nodes can return slightly different subsets to different
|
||||
# resolvers (especially with EDNS Client Subnet). One shared IP is
|
||||
# enough to confirm we're talking to the same service.
|
||||
r = _run_with(_patches(
|
||||
sys_ips=("1.1.1.1", "2.2.2.2"),
|
||||
doh_ips=("2.2.2.2", "3.3.3.3"),
|
||||
))
|
||||
assert r.dns_mismatch is False
|
||||
assert r.verdict == Verdict.OK
|
||||
|
||||
def test_completely_disjoint_pools_flag_mismatch(self):
|
||||
r = _run_with(_patches(
|
||||
sys_ips=("1.1.1.1", "2.2.2.2"),
|
||||
doh_ips=("9.9.9.9", "8.8.8.8"),
|
||||
))
|
||||
assert r.dns_mismatch is True
|
||||
# Note text should hint at the disjointness so users can debug.
|
||||
assert any("disjoint" in n for n in r.notes)
|
||||
|
||||
def test_single_ip_match_does_not_flag_mismatch(self):
|
||||
# The original simple case still works: one IP each side, identical.
|
||||
r = _run_with(_patches(sys_ips=("1.2.3.4",), doh_ips=("1.2.3.4",)))
|
||||
assert r.dns_mismatch is False
|
||||
|
||||
def test_full_address_pools_recorded_in_result(self):
|
||||
# Both sets must be on the result so JSON consumers can audit the
|
||||
# decision. Sorted for determinism in tests and in tool output.
|
||||
r = _run_with(_patches(
|
||||
sys_ips=("3.3.3.3", "1.1.1.1"),
|
||||
doh_ips=("2.2.2.2",),
|
||||
))
|
||||
assert r.sys_ips == ["1.1.1.1", "3.3.3.3"]
|
||||
assert r.doh_ips == ["2.2.2.2"]
|
||||
# And the legacy single-IP fields stay populated for backward compat.
|
||||
assert r.sys_ip == "1.1.1.1"
|
||||
assert r.doh_ip == "2.2.2.2"
|
||||
|
|
|
|||
|
|
@ -1,43 +1,132 @@
|
|||
from unittest.mock import patch, MagicMock
|
||||
|
||||
from rkn_checker.dns import resolve_system, resolve_doh
|
||||
from rkn_checker.dns import (
|
||||
resolve_doh,
|
||||
resolve_doh_all,
|
||||
resolve_system,
|
||||
resolve_system_all,
|
||||
)
|
||||
|
||||
|
||||
class TestResolveSystem:
|
||||
@patch("rkn_checker.dns.socket.gethostbyname", return_value="1.2.3.4")
|
||||
def test_returns_ip_on_success(self, mock_gethost):
|
||||
assert resolve_system("example.com") == "1.2.3.4"
|
||||
def _addrinfo(*ips: str):
|
||||
"""Build a getaddrinfo-shaped return value with the given IPv4 addresses.
|
||||
|
||||
@patch("rkn_checker.dns.socket.gethostbyname", side_effect=__import__("socket").gaierror("fail"))
|
||||
def test_returns_none_on_gaierror(self, mock_gethost):
|
||||
assert resolve_system("example.com") is None
|
||||
Each entry is (family, type, proto, canonname, sockaddr) where sockaddr
|
||||
is (ip, port) for IPv4. We only care about the IP, so the rest is filler.
|
||||
"""
|
||||
import socket
|
||||
return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", (ip, 0)) for ip in ips]
|
||||
|
||||
|
||||
class TestResolveDoh:
|
||||
class TestResolveSystemAll:
|
||||
"""The all-addresses helper is the new source of truth — single-IP
|
||||
helpers are derived from it. These tests pin down its behaviour
|
||||
independently."""
|
||||
|
||||
@patch("rkn_checker.dns.socket.getaddrinfo",
|
||||
return_value=[__import__("socket").AF_INET]) # placeholder, patched below
|
||||
def test_returns_every_ip_in_response(self, mock_getaddr):
|
||||
mock_getaddr.return_value = _addrinfo("1.2.3.4", "5.6.7.8", "9.10.11.12")
|
||||
ips = resolve_system_all("example.com")
|
||||
assert ips == frozenset({"1.2.3.4", "5.6.7.8", "9.10.11.12"})
|
||||
|
||||
@patch("rkn_checker.dns.socket.getaddrinfo")
|
||||
def test_dedupes_repeated_ips(self, mock_getaddr):
|
||||
# The OS resolver can hand back duplicates across SOCK_STREAM/SOCK_DGRAM
|
||||
# entries. Caller compares sets, so we must not leak that detail.
|
||||
mock_getaddr.return_value = _addrinfo("1.2.3.4", "1.2.3.4", "5.6.7.8")
|
||||
assert resolve_system_all("example.com") == frozenset({"1.2.3.4", "5.6.7.8"})
|
||||
|
||||
@patch("rkn_checker.dns.socket.getaddrinfo",
|
||||
side_effect=__import__("socket").gaierror("fail"))
|
||||
def test_returns_empty_on_gaierror(self, mock_getaddr):
|
||||
assert resolve_system_all("example.com") == frozenset()
|
||||
|
||||
@patch("rkn_checker.dns.socket.getaddrinfo")
|
||||
def test_requests_only_ipv4(self, mock_getaddr):
|
||||
# Mixing in IPv6 results would muddy mismatch comparison; verify
|
||||
# that the underlying call is constrained to AF_INET.
|
||||
import socket
|
||||
mock_getaddr.return_value = _addrinfo("1.2.3.4")
|
||||
resolve_system_all("example.com")
|
||||
assert mock_getaddr.call_args[1]["family"] == socket.AF_INET
|
||||
|
||||
|
||||
class TestResolveDohAll:
|
||||
@patch("rkn_checker.dns.requests.get")
|
||||
def test_returns_ip_from_answer(self, mock_get):
|
||||
def test_returns_every_a_record(self, mock_get):
|
||||
resp = MagicMock()
|
||||
resp.ok = True
|
||||
resp.json.return_value = {"Answer": [{"type": 1, "data": "9.9.9.9"}]}
|
||||
resp.json.return_value = {"Answer": [
|
||||
{"type": 1, "data": "9.9.9.9"},
|
||||
{"type": 1, "data": "8.8.8.8"},
|
||||
{"type": 1, "data": "1.1.1.1"},
|
||||
]}
|
||||
mock_get.return_value = resp
|
||||
assert resolve_doh("example.com") == "9.9.9.9"
|
||||
assert resolve_doh_all("example.com") == frozenset({
|
||||
"9.9.9.9", "8.8.8.8", "1.1.1.1",
|
||||
})
|
||||
|
||||
@patch("rkn_checker.dns.requests.get")
|
||||
def test_returns_none_on_http_error(self, mock_get):
|
||||
def test_ignores_non_a_record_types(self, mock_get):
|
||||
# CNAME (type 5) gets followed by upstream resolvers — for our
|
||||
# comparison only the final A records matter.
|
||||
resp = MagicMock()
|
||||
resp.ok = True
|
||||
resp.json.return_value = {"Answer": [
|
||||
{"type": 5, "data": "alias.example.com."},
|
||||
{"type": 1, "data": "9.9.9.9"},
|
||||
{"type": 28, "data": "::1"}, # AAAA
|
||||
]}
|
||||
mock_get.return_value = resp
|
||||
assert resolve_doh_all("example.com") == frozenset({"9.9.9.9"})
|
||||
|
||||
@patch("rkn_checker.dns.requests.get")
|
||||
def test_returns_empty_on_http_error(self, mock_get):
|
||||
resp = MagicMock()
|
||||
resp.ok = False
|
||||
mock_get.return_value = resp
|
||||
assert resolve_doh("example.com") is None
|
||||
assert resolve_doh_all("example.com") == frozenset()
|
||||
|
||||
@patch("rkn_checker.dns.requests.get", side_effect=__import__("requests").exceptions.RequestException("network error"))
|
||||
def test_returns_none_on_request_exception(self, mock_get):
|
||||
assert resolve_doh("example.com") is None
|
||||
@patch("rkn_checker.dns.requests.get",
|
||||
side_effect=__import__("requests").exceptions.RequestException("net"))
|
||||
def test_returns_empty_on_request_exception(self, mock_get):
|
||||
assert resolve_doh_all("example.com") == frozenset()
|
||||
|
||||
@patch("rkn_checker.dns.requests.get")
|
||||
def test_passes_timeout_param(self, mock_get):
|
||||
def test_returns_empty_on_no_answer_field(self, mock_get):
|
||||
resp = MagicMock()
|
||||
resp.ok = False
|
||||
resp.ok = True
|
||||
resp.json.return_value = {} # NXDOMAIN-shape
|
||||
mock_get.return_value = resp
|
||||
assert resolve_doh_all("example.com") == frozenset()
|
||||
|
||||
|
||||
class TestSingleIpHelpers:
|
||||
"""The legacy single-IP helpers are now thin wrappers. We just need to
|
||||
confirm they pull from the *_all variants and pick deterministically."""
|
||||
|
||||
@patch("rkn_checker.dns.resolve_system_all",
|
||||
return_value=frozenset({"5.5.5.5", "1.1.1.1", "9.9.9.9"}))
|
||||
def test_resolve_system_picks_lowest_for_stability(self, _):
|
||||
# Sorted-first means tests don't flake based on set hash order, and
|
||||
# rkn-check output stays the same between runs against the same host.
|
||||
assert resolve_system("example.com") == "1.1.1.1"
|
||||
|
||||
@patch("rkn_checker.dns.resolve_system_all", return_value=frozenset())
|
||||
def test_resolve_system_returns_none_on_empty(self, _):
|
||||
assert resolve_system("example.com") is None
|
||||
|
||||
@patch("rkn_checker.dns.resolve_doh_all",
|
||||
return_value=frozenset({"9.9.9.9", "8.8.8.8"}))
|
||||
def test_resolve_doh_picks_lowest(self, _):
|
||||
assert resolve_doh("example.com") == "8.8.8.8"
|
||||
|
||||
@patch("rkn_checker.dns.resolve_doh_all", return_value=frozenset())
|
||||
def test_resolve_doh_returns_none_on_empty(self, _):
|
||||
assert resolve_doh("example.com") is None
|
||||
|
||||
@patch("rkn_checker.dns.resolve_doh_all", return_value=frozenset({"1.1.1.1"}))
|
||||
def test_resolve_doh_passes_timeout(self, mock_all):
|
||||
resolve_doh("example.com", timeout=2.5)
|
||||
mock_get.assert_called_once()
|
||||
assert mock_get.call_args[1]["timeout"] == 2.5
|
||||
assert mock_all.call_args[1]["timeout"] == 2.5
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue