From c8517eb2f14e86bbf207bd664fe0bbe21dc9175a Mon Sep 17 00:00:00 2001 From: Cat <86354789+MiCat-S@users.noreply.github.com> Date: Thu, 22 Oct 2026 16:43:20 +0800 Subject: [PATCH] fix(doh): skip CNAME answers and cool down failed resolvers (#7135) --- app/adapters/network/doh.py | 188 ++++++++++--- tests/test_doh_helper.py | 337 ++++++++++++++++++++++++ tests/test_typing_runtime_boundaries.py | 7 +- 3 files changed, 493 insertions(+), 39 deletions(-) diff --git a/app/adapters/network/doh.py b/app/adapters/network/doh.py index f46978bb1..2848c806c 100644 --- a/app/adapters/network/doh.py +++ b/app/adapters/network/doh.py @@ -9,8 +9,9 @@ import socket import struct import urllib import urllib.request -from concurrent.futures import as_completed +from concurrent.futures import Future, as_completed from threading import Lock +from time import monotonic from typing import Any, Callable, Dict, Optional, cast from app.foundation.singleton import Singleton @@ -27,8 +28,21 @@ _DOH_EXECUTOR_STOP_TIMEOUT_SECONDS = 10.0 # 定义默认的DoH配置 _doh_timeout = 5 +# 所有解析器都失败后的冷却秒数；冷却期内该域名直接使用系统 DNS，避免每次请求都等待解析超时并重复告警 +_DOH_FAILURE_COOLDOWN_SECONDS = 300.0 _doh_cache: Dict[str, str] = {} +# 域名到解析失败冷却截止时间（monotonic 秒）的映射 +_doh_failure_until: Dict[str, float] = {} +# 解析状态代次；每次清空缓存时递增，查询结果只能写回发起查询时的代次，避免旧配置结果回填 +_doh_generation = 0 _doh_lock = Lock() +# DNS 报文固定头部长度（RFC 1035） +_DNS_HEADER_LENGTH = 12 +# 资源记录在域名之后的固定字段长度：type:2 + class:2 + ttl:4 + rdlength:2 +_DNS_RECORD_FIXED_LENGTH = 10 +# A 记录类型与互联网类别 +_DNS_TYPE_A = 1 +_DNS_CLASS_IN = 1 # 保存原始的 socket.getaddrinfo 方法 _orig_getaddrinfo: Callable[..., Any] = socket.getaddrinfo @@ -66,35 +80,34 @@ def enable_doh(enable: bool) -> bool: """ if host not in _doh_setting("DOH_DOMAINS").split(","): return _orig_getaddrinfo(host, *args, **kwargs) - # 检查主机是否已解析 + # 检查主机是否已解析，或仍处于全部解析器失败后的冷却期 with _doh_lock: ip = _doh_cache.get(host, None) + cooling_down = _doh_failure_until.get(host, 0.0) > monotonic() + generation = _doh_generation if ip is not None: logger.info(f"已解析 [{host}] 为 [{ip}] (缓存)") return _orig_getaddrinfo(ip, *args, **kwargs) # 使用DoH解析主机 + futures: list[Future[Optional[str]]] = [] with _executor_lock: - if not _doh_enabled: - return _orig_getaddrinfo(host, *args, **kwargs) - executor = _get_executor_locked() - # 一次解析的任务必须在同一临界区提交完，避免关闭过程中部分任务落入新线程池 - futures = [ - submit_with_context(executor, _doh_query, resolver, host) - for resolver in _doh_setting("DOH_RESOLVERS").split(",") - ] + use_system_dns = not _doh_enabled or cooling_down + if not use_system_dns: + executor = _get_executor_locked() + # 一次解析的任务必须在同一临界区提交完，避免关闭过程中部分任务落入新线程池 + futures = [ + submit_with_context(executor, _doh_query, resolver, host) + for resolver in _doh_setting("DOH_RESOLVERS").split(",") + ] + # 系统 DNS 可能阻塞，必须在释放执行器锁后调用，避免拖住其他查询和有界关闭 + if use_system_dns: + return _orig_getaddrinfo(host, *args, **kwargs) for future in as_completed(futures): ip = future.result() if ip is not None: - logger.info(f"已解析 [{host}] 为 [{ip}]") - # 关闭可能在查询等待期间恢复系统 DNS；关闭后的结果不得回填到下一轮配置。 - with _executor_lock: - cache_allowed = _doh_enabled - if cache_allowed: - with _doh_lock: - _doh_cache[host] = ip - host = ip break - return _orig_getaddrinfo(host, *args, **kwargs) + _remember_doh_result(host, ip, generation) + return _orig_getaddrinfo(ip or host, *args, **kwargs) with _executor_lock: if enable and _executor is not None and not _executor.accepting: @@ -111,6 +124,51 @@ def enable_doh(enable: bool) -> bool: return True +def _remember_doh_result(host: str, ip: Optional[str], generation: int) -> None: + """ + 记录一次 DoH 并行查询的结果。 + + 成功时缓存解析地址并解除该域名的失败冷却；所有解析器都失败时进入冷却期， + 冷却期内该域名直接使用系统 DNS，并发查询只告警一次。 + DoH 已关闭或配置已变更时丢弃结果，避免旧配置的结果回填到下一轮配置。 + + :param host: 查询的域名 + :param ip: 第一个成功解析器返回的 IPv4 地址；全部失败时为 None + :param generation: 发起查询时的解析状态代次 + """ + if ip is not None: + logger.info(f"已解析 [{host}] 为 [{ip}]") + with _executor_lock: + cache_allowed = _doh_enabled + with _doh_lock: + if not cache_allowed or generation != _doh_generation: + return + if ip is not None: + _doh_cache[host] = ip + _doh_failure_until.pop(host, None) + return + now = monotonic() + if _doh_failure_until.get(host, 0.0) > now: + # 并发查询已记录冷却，不再重复告警 + return + _doh_failure_until[host] = now + _DOH_FAILURE_COOLDOWN_SECONDS + logger.warning( + f"所有 DoH 解析器均未能解析 [{host}]，" + f"{_DOH_FAILURE_COOLDOWN_SECONDS:.0f} 秒内改用系统 DNS，请检查 DOH_RESOLVERS 是否可达" + ) + + +def _reset_doh_state() -> None: + """ + 清空解析缓存与失败冷却，并递增代次让仍在进行的旧查询结果失效。 + """ + global _doh_generation + with _doh_lock: + _doh_cache.clear() + _doh_failure_until.clear() + _doh_generation += 1 + + class DohHelper(ConfigReloadMixin, metaclass=Singleton): """ DoH帮助类，用于处理DNS over HTTPS解析。 @@ -131,9 +189,8 @@ class DohHelper(ConfigReloadMixin, metaclass=Singleton): if not self.shutdown(): logger.error("DoH配置关闭后查询线程池未在预算内收敛") return - with _doh_lock: - # DOH配置有变动的情况下，清空缓存 - _doh_cache.clear() + # DOH配置有变动的情况下，清空缓存和失败冷却，让新解析器立即生效 + _reset_doh_state() if not enable_doh(True): logger.error("DoH查询线程池尚未收敛，暂不重新启用") @@ -161,8 +218,7 @@ class DohHelper(ConfigReloadMixin, metaclass=Singleton): converged = executor is None or executor.shutdown_bounded(timeout=timeout) if converged and _executor is executor: _executor = None - with _doh_lock: - _doh_cache.clear() + _reset_doh_state() return converged @@ -220,18 +276,72 @@ def _doh_query(resolver: str, host: str) -> Optional[str]: return None resp_body = response.read() - # 解析DNS响应消息（RFC 1035） - # name（压缩）:2 + type:2 + class:2 + ttl:4 + rdlength:2 = 12字节 - first_rdata_start = len(header) + len(question) + 12 - # rdata（A记录）= 4字节 - first_rdata_end = first_rdata_start + 4 - # 将rdata转换为IP地址 - return socket.inet_ntoa(resp_body[first_rdata_start:first_rdata_end]) + ip = _parse_dns_a_record(resp_body) + if ip is None: + logger.debug(f"解析器({resolver})未返回 [{host}] 的 A 记录") + return ip except Exception as e: - logger.error(f"解析器({resolver})请求错误: {e}") + # 单个解析器失败很常见，由调用方在全部失败时统一告警 + logger.debug(f"解析器({resolver})请求错误: {e}") return None +def _skip_dns_name(message: bytes, offset: int) -> Optional[int]: + """ + 跳过 DNS 报文中的一个域名字段。 + + 域名由若干长度前缀标签组成，以零字节或压缩指针结尾；保留的标签类型按无效报文处理。 + + :param message: 完整 DNS 报文 + :param offset: 域名字段的起始位置 + :return: 域名之后第一个字节的位置；报文截断或标签无效时返回 None + """ + while offset < len(message): + length = message[offset] + if length & 0xC0 == 0xC0: + return offset + 2 + if length & 0xC0: + return None + if length == 0: + return offset + 1 + offset += 1 + length + return None + + +def _parse_dns_a_record(message: bytes) -> Optional[str]: + """ + 从 DNS 应答报文中取第一条 A 记录的 IPv4 地址（RFC 1035）。 + + 应答区可能先给出 CNAME 等其他类型的记录，必须逐条跳过， + 不能按固定偏移读取第一条记录，否则会把别名的域名字节误当成 IP。 + + :param message: 解析器返回的完整 DNS 报文 + :return: 第一条 A 记录的点分十进制地址；响应码非成功、没有 A 记录或报文截断时返回 None + """ + if len(message) < _DNS_HEADER_LENGTH or message[3] & 0x0F: + return None + question_count, answer_count = struct.unpack_from(">HH", message, 4) + offset = _DNS_HEADER_LENGTH + for _ in range(question_count): + name_end = _skip_dns_name(message, offset) + if name_end is None: + return None + # 跳过问题的 type 与 class + offset = name_end + 4 + for _ in range(answer_count): + name_end = _skip_dns_name(message, offset) + if name_end is None or name_end + _DNS_RECORD_FIXED_LENGTH > len(message): + return None + record_type, record_class, _, data_length = struct.unpack_from(">HHIH", message, name_end) + offset = name_end + _DNS_RECORD_FIXED_LENGTH + if offset + data_length > len(message): + return None + if record_type == _DNS_TYPE_A and record_class == _DNS_CLASS_IN and data_length == 4: + return socket.inet_ntoa(message[offset:offset + 4]) + offset += data_length + return None + + def doh_query_json(resolver: str, host: str) -> Optional[str]: """ 使用给定的DoH解析器查询给定主机的IP地址。 @@ -248,10 +358,16 @@ def doh_query_json(resolver: str, host: str) -> Optional[str]: response_body = response.read().decode("utf-8") logger.debug(f"<== body: {response_body}") answer = json.loads(response_body).get("Answer") - if not isinstance(answer, list) or not answer: + if not isinstance(answer, list): return None - value = answer[0].get("data") if isinstance(answer[0], dict) else None - return value if isinstance(value, str) else None + # 应答可能先给出 CNAME 链，只取 A 记录的地址 + for record in answer: + if isinstance(record, dict) and record.get("type") == _DNS_TYPE_A: + value = record.get("data") + if isinstance(value, str): + return value + return None except Exception as e: - logger.error(f"解析器({resolver})请求错误: {e}") + # 单个解析器失败很常见，由调用方在全部失败时统一告警 + logger.debug(f"解析器({resolver})请求错误: {e}") return None diff --git a/tests/test_doh_helper.py b/tests/test_doh_helper.py index fab8f821d..1217100f6 100644 --- a/tests/test_doh_helper.py +++ b/tests/test_doh_helper.py @@ -2,9 +2,14 @@ 验证加密 DNS 执行器的惰性创建、关闭和套接字恢复。 """ +import json import socket +import struct import threading import time +import urllib.request + +import pytest from app.adapters.network import doh from app.runtime.correlation import correlation_scope, get_correlation_id @@ -190,3 +195,335 @@ def test_doh_queries_use_each_request_context(monkeypatch): socket.getaddrinfo = original_getaddrinfo assert observed == ["doh-first", "doh-second"] + + +def _dns_name(name: str) -> bytes: + """ + 按 RFC 1035 编码未压缩的域名。 + """ + return b"".join(bytes([len(label)]) + label.encode() for label in name.split(".")) + b"\x00" + + +def _dns_response(host: str, answers: list[tuple[bytes, int, bytes]], rcode: int = 0) -> bytes: + """ + 构造单问题 DNS 应答报文。 + + :param host: 问题区的域名 + :param answers: 依次排列的应答记录，每项为 (编码后的记录名, 记录类型, rdata) + :param rcode: 响应码，0 表示成功 + :return: 完整报文 + """ + header = struct.pack(">HHHHHH", 0, 0x8180 | rcode, 1, len(answers), 0, 0) + question = _dns_name(host) + struct.pack(">HH", 1, 1) + records = b"".join( + name + struct.pack(">HHIH", record_type, 1, 60, len(rdata)) + rdata + for name, record_type, rdata in answers + ) + return header + question + records + + +def _cname_then_a_response() -> bytes: + """ + 构造 image.tmdb.org 先返回 CNAME、再返回 CDN A 记录的真实形态应答。 + + 第一条记录名压缩指向问题区，第二条记录名压缩指向 CNAME 的目标域名。 + """ + host = "image.tmdb.org" + question_end = 12 + len(_dns_name(host)) + 4 + cname_rdata_offset = question_end + 2 + 10 + return _dns_response( + host, + [ + (b"\xc0\x0c", 5, _dns_name("tmdb-image-prod.b-cdn.net")), + (struct.pack(">H", 0xC000 | cname_rdata_offset), 1, socket.inet_aton("169.150.249.164")), + ], + ) + + +class _FakeDohResponse: + """ + 模拟 urllib 返回的 DoH 响应上下文。 + """ + + def __init__(self, body: bytes, status: int = 200) -> None: + """ + :param body: 响应正文 + :param status: HTTP 状态码 + """ + self.status = status + self._body = body + + def __enter__(self) -> "_FakeDohResponse": + return self + + def __exit__(self, *_exc: object) -> None: + return None + + def read(self) -> bytes: + """ + 返回完整响应正文。 + """ + return self._body + + +def test_doh_query_follows_cname_to_a_record(monkeypatch): + """ + 应答先给 CNAME 时应跳过别名记录取 A 记录，而不是把别名字节当成 IP。 + """ + monkeypatch.setattr( + urllib.request, + "urlopen", + lambda _request, timeout: _FakeDohResponse(_cname_then_a_response()), + ) + + # 旧实现按固定偏移读取，会把 "\x0ftmd" 误解析为 15.116.109.100 + assert doh._doh_query("resolver.test", "image.tmdb.org") == "169.150.249.164" + + +@pytest.mark.parametrize( + "message", + [ + _dns_response("example.com", [], rcode=3), + _dns_response("example.com", [(b"\xc0\x0c", 5, _dns_name("alias.example.net"))]), + _dns_response("example.com", [(b"\xc0\x0c", 1, socket.inet_aton("203.0.113.7"))])[:-2], + _dns_response("example.com", [(b"\x40", 1, socket.inet_aton("203.0.113.7"))]), + b"\x00\x00\x81\x80", + ], + ids=["nxdomain", "cname-only", "truncated-rdata", "reserved-label", "short-header"], +) +def test_parse_dns_a_record_rejects_unusable_answers(message): + """ + 失败响应码、没有 A 记录、截断报文和保留标签类型都不应产生地址。 + """ + assert doh._parse_dns_a_record(message) is None + + +def test_parse_dns_a_record_reads_plain_a_record(): + """ + 直接返回 A 记录的应答应取第一条地址。 + """ + message = _dns_response( + "example.com", + [ + (b"\xc0\x0c", 1, socket.inet_aton("203.0.113.7")), + (b"\xc0\x0c", 1, socket.inet_aton("203.0.113.8")), + ], + ) + + assert doh._parse_dns_a_record(message) == "203.0.113.7" + + +def test_doh_query_json_skips_cname(monkeypatch): + """ + JSON 应答同样应跳过 CNAME 链，只返回 A 记录地址。 + """ + body = json.dumps({ + "Status": 0, + "Answer": [ + {"name": "image.tmdb.org.", "type": 5, "data": "tmdb-image-prod.b-cdn.net."}, + {"name": "tmdb-image-prod.b-cdn.net.", "type": 1, "data": "169.150.249.164"}, + ], + }).encode() + monkeypatch.setattr( + urllib.request, + "urlopen", + lambda _request, timeout: _FakeDohResponse(body), + ) + + assert doh.doh_query_json("resolver.test", "image.tmdb.org") == "169.150.249.164" + + +def test_doh_query_logs_resolver_failure_at_debug(monkeypatch): + """ + 单个解析器超时只记调试日志，不再逐条输出错误。 + """ + errors = [] + + def timed_out(_request, timeout): + raise OSError("timed out") + + monkeypatch.setattr(urllib.request, "urlopen", timed_out) + monkeypatch.setattr(doh.logger, "error", errors.append) + + assert doh._doh_query("resolver.test", "example.com") is None + assert doh.doh_query_json("resolver.test", "example.com") is None + assert errors == [] + + +def test_enable_doh_cools_down_after_all_resolvers_fail(monkeypatch): + """ + 所有解析器都失败后，冷却期内直接走系统 DNS 且只告警一次；冷却结束后重新查询。 + """ + clock = [1000.0] + query_calls = [] + query_result = [None] + resolved_hosts = [] + warnings = [] + + def fake_query(resolver: str, host: str): + query_calls.append((resolver, host)) + return query_result[0] + + def fake_getaddrinfo(host: str, *args, **kwargs): + resolved_hosts.append(host) + return [] + + original_getaddrinfo = socket.getaddrinfo + helper = object.__new__(doh.DohHelper) + monkeypatch.setattr("app.runtime.config.settings.DOH_DOMAINS", "example.com") + monkeypatch.setattr("app.runtime.config.settings.DOH_RESOLVERS", "resolver.test") + monkeypatch.setattr(doh, "_doh_query", fake_query) + monkeypatch.setattr(doh, "_orig_getaddrinfo", fake_getaddrinfo) + monkeypatch.setattr(doh, "monotonic", lambda: clock[0]) + monkeypatch.setattr(doh.logger, "warning", warnings.append) + + try: + assert helper.shutdown() is True + assert doh.enable_doh(True) is True + + socket.getaddrinfo("example.com", None) + assert query_calls == [("resolver.test", "example.com")] + assert resolved_hosts == ["example.com"] + assert len(warnings) == 1 + + clock[0] += doh._DOH_FAILURE_COOLDOWN_SECONDS - 1 + socket.getaddrinfo("example.com", None) + assert len(query_calls) == 1 + assert resolved_hosts == ["example.com", "example.com"] + assert len(warnings) == 1 + + clock[0] += 2 + query_result[0] = "203.0.113.7" + socket.getaddrinfo("example.com", None) + assert len(query_calls) == 2 + assert resolved_hosts[-1] == "203.0.113.7" + assert "example.com" not in doh._doh_failure_until + assert doh._doh_cache["example.com"] == "203.0.113.7" + finally: + helper.shutdown() + socket.getaddrinfo = original_getaddrinfo + + assert doh._doh_cache == {} + assert doh._doh_failure_until == {} + + +def test_doh_config_reload_clears_failure_cooldown(monkeypatch): + """ + 修改 DoH 配置后应清除失败冷却，让新的解析器立即生效。 + """ + query_calls = [] + original_getaddrinfo = socket.getaddrinfo + helper = object.__new__(doh.DohHelper) + monkeypatch.setattr("app.runtime.config.settings.DOH_ENABLE", True) + monkeypatch.setattr("app.runtime.config.settings.DOH_DOMAINS", "example.com") + monkeypatch.setattr("app.runtime.config.settings.DOH_RESOLVERS", "resolver.test") + monkeypatch.setattr(doh, "_doh_query", lambda resolver, host: query_calls.append(resolver)) + monkeypatch.setattr(doh, "_orig_getaddrinfo", lambda host, *args, **kwargs: []) + monkeypatch.setattr(doh.logger, "warning", lambda _message: None) + + try: + assert helper.shutdown() is True + assert doh.enable_doh(True) is True + socket.getaddrinfo("example.com", None) + assert "example.com" in doh._doh_failure_until + + monkeypatch.setattr("app.runtime.config.settings.DOH_RESOLVERS", "fixed.test") + helper.on_config_changed() + assert doh._doh_failure_until == {} + + socket.getaddrinfo("example.com", None) + assert query_calls == ["resolver.test", "fixed.test"] + finally: + helper.shutdown() + socket.getaddrinfo = original_getaddrinfo + + +def test_doh_cooldown_falls_back_without_holding_executor_lock(monkeypatch): + """ + 冷却期回退系统 DNS 时不得持有执行器锁，否则系统 DNS 阻塞会拖住有界关闭。 + """ + lock_held = [] + original_getaddrinfo = socket.getaddrinfo + helper = object.__new__(doh.DohHelper) + monkeypatch.setattr("app.runtime.config.settings.DOH_DOMAINS", "example.com") + monkeypatch.setattr("app.runtime.config.settings.DOH_RESOLVERS", "resolver.test") + monkeypatch.setattr(doh, "_doh_query", lambda resolver, host: None) + monkeypatch.setattr( + doh, + "_orig_getaddrinfo", + lambda host, *args, **kwargs: lock_held.append(doh._executor_lock.locked()) or [], + ) + monkeypatch.setattr(doh.logger, "warning", lambda _message: None) + + try: + assert helper.shutdown() is True + assert doh.enable_doh(True) is True + socket.getaddrinfo("example.com", None) + socket.getaddrinfo("example.com", None) + finally: + helper.shutdown() + socket.getaddrinfo = original_getaddrinfo + + assert lock_held == [False, False] + + +@pytest.mark.parametrize("stale_result", [None, "203.0.113.7"]) +def test_doh_reload_discards_in_flight_query_result(stale_result, monkeypatch): + """ + 配置重载期间仍在进行的旧查询结果不得写回缓存或失败冷却。 + """ + helper = object.__new__(doh.DohHelper) + query_calls = [] + warnings = [] + + def reload_during_query(resolver: str, host: str): + query_calls.append(resolver) + if len(query_calls) == 1: + monkeypatch.setattr("app.runtime.config.settings.DOH_RESOLVERS", "fixed.test") + helper.on_config_changed() + return stale_result + return None + + original_getaddrinfo = socket.getaddrinfo + monkeypatch.setattr("app.runtime.config.settings.DOH_ENABLE", True) + monkeypatch.setattr("app.runtime.config.settings.DOH_DOMAINS", "example.com") + monkeypatch.setattr("app.runtime.config.settings.DOH_RESOLVERS", "resolver.test") + monkeypatch.setattr(doh, "_doh_query", reload_during_query) + monkeypatch.setattr(doh, "_orig_getaddrinfo", lambda host, *args, **kwargs: []) + monkeypatch.setattr(doh.logger, "warning", warnings.append) + + try: + assert helper.shutdown() is True + assert doh.enable_doh(True) is True + socket.getaddrinfo("example.com", None) + assert doh._doh_cache == {} + assert doh._doh_failure_until == {} + assert warnings == [] + + socket.getaddrinfo("example.com", None) + assert query_calls == ["resolver.test", "fixed.test"] + finally: + helper.shutdown() + socket.getaddrinfo = original_getaddrinfo + + +def test_doh_concurrent_failures_warn_once(monkeypatch): + """ + 同一代次的并发查询全部失败时，只有首个记录冷却的查询输出告警。 + """ + warnings = [] + original_getaddrinfo = socket.getaddrinfo + helper = object.__new__(doh.DohHelper) + monkeypatch.setattr(doh.logger, "warning", warnings.append) + + try: + assert helper.shutdown() is True + assert doh.enable_doh(True) is True + generation = doh._doh_generation + doh._remember_doh_result("example.com", None, generation) + doh._remember_doh_result("example.com", None, generation) + finally: + helper.shutdown() + socket.getaddrinfo = original_getaddrinfo + + assert len(warnings) == 1 diff --git a/tests/test_typing_runtime_boundaries.py b/tests/test_typing_runtime_boundaries.py index 28cd9a4cf..e3a4cfa50 100644 --- a/tests/test_typing_runtime_boundaries.py +++ b/tests/test_typing_runtime_boundaries.py @@ -23,12 +23,13 @@ from app.schemas.agent import AgentMcpServerConfig @pytest.mark.parametrize(("answer", "expected"), [ - (None, None), ([], None), ([None], None), ([{"data": 123}], None), - ([{"data": "203.0.113.7"}], "203.0.113.7"), + (None, None), ([], None), ([None], None), ([{"type": 1, "data": 123}], None), + ([{"data": "203.0.113.7"}], None), + ([{"type": 1, "data": "203.0.113.7"}], "203.0.113.7"), ]) def test_doh_json_only_accepts_text_address(answer, expected, monkeypatch): """ - DNS JSON 缺少答案或返回错误类型时视为未解析，合法地址原样返回。 + DNS JSON 缺少答案、没有 A 记录或返回错误类型时视为未解析，合法 A 记录地址原样返回。 """ response = MagicMock(status=200) response.read.return_value = json.dumps({"Answer": answer}).encode()