diff --git a/backend/package/yuxi/utils/crypto.py b/backend/package/yuxi/utils/crypto.py new file mode 100644 index 00000000..6c895b1f --- /dev/null +++ b/backend/package/yuxi/utils/crypto.py @@ -0,0 +1,348 @@ +"""敏感字段加密与脱敏通用工具。 + +基于 Fernet 对称加密，支持递归处理配置字典中的敏感字段。 +所有加密、解密、脱敏、密钥校验逻辑均封装在 ``CryptoHelper`` 类中， +模块级便捷函数仅作为默认实例的薄封装，便于全局调用与测试注入。 +""" + +from __future__ import annotations + +import os +import re +from typing import Any, Literal + +from cryptography.fernet import Fernet, InvalidToken + +from yuxi.utils import logger + + +class CryptoHelper: + """敏感字段加密与脱敏工具（完整闭环）。 + + 基于 Fernet 对称加密，支持递归处理配置字典中的敏感字段。 + + 密钥来源优先级： + 1. 构造函数注入的 ``fernet`` 实例 + 2. 构造函数注入的 ``key`` 字符串 + 3. 环境变量 ``ENCRYPTION_KEY`` + + 未提供密钥时，加密/解密操作将透传原值（开发环境兼容），并在首次 + 访问时记录一次警告日志；生产环境可通过 ``validate_for_environment`` + 强制校验密钥存在性。 + + 敏感字段判定：字段名（小写归一化后）命中 ``_SENSITIVE_KEYWORDS`` 中 + 任一关键词即视为敏感。设计原则是宁可漏判不可误判，因此避免使用过于 + 宽泛的词（如 "key" / "cert" / "auth" / "id"），防止误伤 + ``auth_type`` / ``auth_config`` / ``content-type`` / ``client_id`` + 等非敏感字段。 + """ + + # 加密密钥环境变量名 + _ENV_KEY_NAME = "ENCRYPTION_KEY" + + # 用于识别 Fernet 加密值的统一前缀 + _ENCRYPTION_PREFIX = "enc:" + + # 敏感字段关键词：字段名（小写归一化后）命中任一关键词即视为敏感。 + # 设计原则：宁可漏判（漏判可由调用方补充），不可误判（误判会破坏正常字段）。 + # 因此避免使用过于宽泛的词（如 "key" / "cert" / "auth" / "id"）， + # 防止误伤 auth_type / auth_config / content-type / client_id 等非敏感字段。 + _SENSITIVE_KEYWORDS = ( + # 密码类 + "password", + "passwd", + "pwd", + # 令牌类 + "token", + "access_token", + "refresh_token", + "id_token", + "bearer", + "authorization", + # 密钥类 + "secret", + "api_key", + "apikey", + "api_secret", + "access_key", + "secret_key", + "private_key", + "client_secret", + "client_key", + # 证书类 + "ca_cert", + "client_cert", + "certificate", + # 凭证类 + "credential", + "credentials", + ) + + def __init__(self, key: str | None = None, fernet: Fernet | None = None) -> None: + """初始化加密助手。 + + Args: + key: Fernet 密钥字符串。提供时优先使用，忽略环境变量。 + fernet: 已构造的 Fernet 实例。提供时优先于 ``key``。 + """ + if fernet is not None: + self._fernet: Fernet | None = fernet + self._injected_key: str | None = None + elif key is not None: + self._fernet = Fernet(key.encode()) + self._injected_key = key + else: + self._fernet = None + self._injected_key = None + self._missing_logged = False + + # ------------------------------------------------------------------ + # 密钥管理 + # ------------------------------------------------------------------ + + @classmethod + def from_env(cls) -> CryptoHelper: + """从环境变量 ``ENCRYPTION_KEY`` 创建实例。 + + 环境变量未设置时返回未配置密钥的实例（开发环境兼容）。 + """ + key = os.getenv(cls._ENV_KEY_NAME) + return cls(key=key) if key else cls() + + def has_key(self) -> bool: + """检查是否存在有效密钥（不触发日志，不抛异常）。""" + if self._fernet is not None: + return True + if self._injected_key is not None: + return True + return bool(os.getenv(self._ENV_KEY_NAME)) + + def _get_fernet(self) -> Fernet | None: + """获取 Fernet 实例，按密钥来源优先级解析。 + + 密钥未配置时记录一次警告并返回 None；密钥格式无效时抛出 ValueError。 + """ + if self._fernet is not None: + return self._fernet + + key = self._injected_key if self._injected_key is not None else os.getenv(self._ENV_KEY_NAME) + if not key: + if not self._missing_logged: + logger.warning( + f"环境变量 {self._ENV_KEY_NAME} 未设置，配置中的敏感字段将以明文存储。" + "生产环境请使用 Fernet.generate_key() 生成密钥并配置。" + ) + self._missing_logged = True + return None + + try: + self._fernet = Fernet(key.encode()) + except Exception as exc: + raise ValueError(f"{self._ENV_KEY_NAME} 格式无效，无法初始化 Fernet") from exc + return self._fernet + + def validate_for_environment(self, environment: str) -> None: + """校验生产环境是否正确配置加密密钥。 + + 非生产环境跳过校验；生产环境要求密钥必须存在且能初始化 Fernet， + 否则抛出 ValueError。 + """ + if environment != "production": + return + + if not self.has_key(): + raise ValueError(f"生产环境必须设置 {self._ENV_KEY_NAME}") + + try: + self._get_fernet() + except ValueError as exc: + raise ValueError(f"{self._ENV_KEY_NAME} 格式无效") from exc + + # ------------------------------------------------------------------ + # 单值加密/解密 + # ------------------------------------------------------------------ + + def encrypt(self, value: str) -> str: + """加密字符串，返回带 ``enc:`` 前缀的密文。 + + 密钥未配置时透传原值。 + """ + fernet = self._get_fernet() + if fernet is None: + return value + encrypted = fernet.encrypt(value.encode()) + return f"{self._ENCRYPTION_PREFIX}{encrypted.decode()}" + + def decrypt(self, value: str) -> str: + """解密带 ``enc:`` 前缀的密文，返回明文。 + + 非 ``enc:`` 前缀的值视为未加密，透传返回；密钥未配置时记录警告 + 并透传；解密失败（密钥变更或数据损坏）抛出 ValueError。 + """ + if not isinstance(value, str) or not value.startswith(self._ENCRYPTION_PREFIX): + return value + + fernet = self._get_fernet() + if fernet is None: + logger.warning("缺少加密密钥，无法解密配置中的敏感字段") + return value + + try: + decrypted = fernet.decrypt(value[len(self._ENCRYPTION_PREFIX) :].encode()) + except InvalidToken as exc: + raise ValueError("配置中的敏感字段解密失败，密钥可能已变更") from exc + return decrypted.decode() + + # ------------------------------------------------------------------ + # 递归加密/解密/脱敏 + # ------------------------------------------------------------------ + + def encrypt_sensitive_fields(self, config: dict[str, Any] | None) -> dict[str, Any] | None: + """递归加密配置字典中的敏感字符串字段。仅加密敏感 key 对应的字符串。""" + return self._transform_config(config, "encrypt") + + def decrypt_sensitive_fields(self, config: dict[str, Any] | None) -> dict[str, Any] | None: + """递归解密配置字典中的敏感字符串字段。""" + return self._transform_config(config, "decrypt") + + def mask_sensitive_fields(self, config: dict[str, Any] | None) -> dict[str, Any]: + """返回配置字典的脱敏副本，用于 API 响应。""" + result = self._transform_config(config, "mask") + return result if result is not None else {} + + def mask_all_string_values(self, obj: Any) -> Any: + """递归将对象中的所有字符串值替换为脱敏占位符。""" + if isinstance(obj, dict): + return {k: self.mask_all_string_values(v) for k, v in obj.items()} + if isinstance(obj, list): + return [self.mask_all_string_values(item) for item in obj] + if isinstance(obj, str): + return "***" + return obj + + # ------------------------------------------------------------------ + # 内部实现 + # ------------------------------------------------------------------ + + @classmethod + def _is_sensitive_key(cls, key: str) -> bool: + """判断字段名是否属于敏感字段。""" + normalized = re.sub(r"[^a-z0-9_]+", "_", key.lower()) + return any(keyword in normalized for keyword in cls._SENSITIVE_KEYWORDS) + + def _transform_config( + self, + config: dict[str, Any] | None, + mode: Literal["encrypt", "decrypt", "mask"], + ) -> dict[str, Any] | None: + """通用配置转换：递归加密/解密/脱敏敏感字段。""" + if config is None: + return None + + if mode == "encrypt": + return self._encrypt_nested(config) + if mode == "decrypt": + return self._decrypt_nested(config) + # mask + return self._mask_nested(config) + + def _encrypt_nested(self, obj: Any) -> Any: + if isinstance(obj, dict): + return { + k: self.encrypt(v) if self._is_sensitive_key(k) and isinstance(v, str) else self._encrypt_nested(v) + for k, v in obj.items() + } + if isinstance(obj, list): + return [self._encrypt_nested(item) for item in obj] + return obj + + def _decrypt_nested(self, obj: Any) -> Any: + if isinstance(obj, dict): + return { + k: self.decrypt(v) if self._is_sensitive_key(k) and isinstance(v, str) else self._decrypt_nested(v) + for k, v in obj.items() + } + if isinstance(obj, list): + return [self._decrypt_nested(item) for item in obj] + return obj + + def _mask_nested(self, obj: Any) -> Any: + if isinstance(obj, dict): + return { + k: self._mask_string(v) if self._is_sensitive_key(k) and isinstance(v, str) else self._mask_nested(v) + for k, v in obj.items() + } + if isinstance(obj, list): + return [self._mask_nested(item) for item in obj] + return obj + + @staticmethod + def _mask_string(value: str) -> str: + """对字符串进行脱敏，返回统一占位符。""" + return "***" + + +# ---------------------------------------------------------------------- +# 模块级默认实例与便捷函数（向后兼容现有调用） +# ---------------------------------------------------------------------- + +_default_helper = CryptoHelper() + + +def set_crypto_helper(helper: CryptoHelper) -> None: + """替换全局默认 crypto helper，主要用于测试注入。""" + global _default_helper + _default_helper = helper + + +def reset_crypto_helper() -> None: + """重置全局默认 crypto helper 为未配置密钥的实例，主要用于测试隔离。""" + global _default_helper + _default_helper = CryptoHelper() + + +def get_crypto_helper() -> CryptoHelper: + """获取全局默认 crypto helper 实例。""" + return _default_helper + + +def validate_encryption_key(environment: str) -> None: + """校验生产环境是否正确配置加密密钥（委托给默认实例）。""" + _default_helper.validate_for_environment(environment) + + +def encrypt_sensitive_fields(config: dict[str, Any] | None) -> dict[str, Any] | None: + """递归加密配置字典中的敏感字符串字段（委托给默认实例）。""" + return _default_helper.encrypt_sensitive_fields(config) + + +def decrypt_sensitive_fields(config: dict[str, Any] | None) -> dict[str, Any] | None: + """递归解密配置字典中的敏感字符串字段（委托给默认实例）。""" + return _default_helper.decrypt_sensitive_fields(config) + + +def encrypt(value: str) -> str: + """加密单个字符串，返回带 ``enc:`` 前缀的密文（委托给默认实例）。 + + 密钥未配置时透传原值。注意：加密不幂等，调用方必须保证传入的是明文。 + """ + return _default_helper.encrypt(value) + + +def decrypt(value: str) -> str: + """解密单个字符串（委托给默认实例）。 + + 非 ``enc:`` 前缀的值视为未加密，透传返回；密钥未配置时记录警告并透传； + 解密失败（密钥变更或数据损坏）抛出 ``ValueError``。 + """ + return _default_helper.decrypt(value) + + +def mask_sensitive_fields(config: dict[str, Any] | None) -> dict[str, Any]: + """返回配置字典的脱敏副本（委托给默认实例）。""" + return _default_helper.mask_sensitive_fields(config) + + +def mask_all_string_values(obj: Any) -> Any: + """递归将对象中的所有字符串值替换为脱敏占位符（委托给默认实例）。""" + return _default_helper.mask_all_string_values(obj) diff --git a/backend/package/yuxi/utils/net_security.py b/backend/package/yuxi/utils/net_security.py index 084f07f7..ba37a75e 100644 --- a/backend/package/yuxi/utils/net_security.py +++ b/backend/package/yuxi/utils/net_security.py @@ -1,13 +1,14 @@ """通用 SSRF 防护模块。 提供纵深防御的 URL 安全校验能力，防止 Agent 通过工具访问内网服务。 -校验顺序：协议校验 → 主机名阻止(Pre-DNS) → 域名白/黑名单(Pre-DNS) → IP 字面量校验(Pre-DNS) → DNS 解析后 IP 校验(Post-DNS) +校验顺序：协议校验 → 主机名阻止(Pre-DNS) → 域名白/黑名单(Pre-DNS) → +IP 字面量校验(Pre-DNS) → DNS 解析后 IP 校验(Post-DNS) """ from __future__ import annotations +import asyncio import ipaddress -import re import socket import time from urllib.parse import urlparse @@ -82,18 +83,10 @@ class SSRFPolicy(BaseModel): model_config = {"arbitrary_types_allowed": True} def model_post_init(self, __context) -> None: - self._blocked_networks = [ - ipaddress.ip_network(cidr) for cidr in self.blocked_ip_ranges - ] - self._blocked_domains_normalized = { - _normalize_hostname(d) for d in self.blocked_domains - } - self._blocked_hostnames_normalized = { - _normalize_hostname(h) for h in self.blocked_hostnames - } - self._blocked_hostname_suffixes_normalized = [ - _normalize_hostname(s) for s in self.blocked_hostname_suffixes - ] + self._blocked_networks = [ipaddress.ip_network(cidr) for cidr in self.blocked_ip_ranges] + self._blocked_domains_normalized = {_normalize_hostname(d) for d in self.blocked_domains} + self._blocked_hostnames_normalized = {_normalize_hostname(h) for h in self.blocked_hostnames} + self._blocked_hostname_suffixes_normalized = [_normalize_hostname(s) for s in self.blocked_hostname_suffixes] class SSRFBlockedError(Exception): @@ -155,16 +148,52 @@ def _looks_like_ipv4_literal(hostname: str) -> bool: if any(not p for p in parts): return True # 空段，异常格式 return all( - p.isdigit() or (p.lower().startswith("0x") and all(c in "0123456789abcdef" for c in p[2:])) - for p in parts + p.isdigit() or (p.lower().startswith("0x") and all(c in "0123456789abcdef" for c in p[2:])) for p in parts ) +# 已知会在低 32 位嵌入 IPv4 地址的 IPv6 前缀。 +# 仅在这些前缀内才提取尾部 IPv4，避免对普通公网 IPv6 产生误拦。 +_IPV4_EMBEDDED_PREFIXES: tuple[ipaddress.IPv4Network | ipaddress.IPv6Network, ...] = ( + ipaddress.ip_network("::/96"), # IPv4-compatible（已废弃，但部分系统仍解析） + ipaddress.ip_network("::ffff:0:0/96"), # IPv4-mapped + ipaddress.ip_network("64:ff9b::/96"), # NAT64 well-known prefix + ipaddress.ip_network("64:ff9b:1::/48"), # NAT64 local-use prefix +) + + +def _extract_embedded_ipv4(addr: ipaddress.IPv6Address) -> ipaddress.IPv4Address | None: + """从 IPv6 地址中提取嵌入的 IPv4 地址（如果属于已知嵌入前缀）。""" + if addr.ipv4_mapped is not None: + return addr.ipv4_mapped + for net in _IPV4_EMBEDDED_PREFIXES: + if addr in net: + return ipaddress.IPv4Address(int(addr) & 0xFFFFFFFF) + return None + + def _is_blocked_ip(addr: ipaddress.IPv4Address | ipaddress.IPv6Address, policy: SSRFPolicy) -> bool: """检查 IP 是否在阻止范围内。""" return any(addr in net for net in policy._blocked_networks) +def _is_blocked_ip_or_embedded( + addr: ipaddress.IPv4Address | ipaddress.IPv6Address, + policy: SSRFPolicy, +) -> tuple[bool, str]: + """检查 IP 本身或其嵌入的 IPv4 是否被阻止。 + + 返回 (blocked, reason)。 + """ + if _is_blocked_ip(addr, policy): + return True, f"目标 IP 在阻止范围内: {addr}" + if isinstance(addr, ipaddress.IPv6Address): + embedded = _extract_embedded_ipv4(addr) + if embedded is not None and _is_blocked_ip(embedded, policy): + return True, f"IPv6 地址嵌入被阻止的 IPv4: {addr} -> {embedded}" + return False, "" + + def _check_ip_literal(hostname: str, policy: SSRFPolicy) -> tuple[bool, str]: """检查主机名是否为 IP 字面量（含非标准格式和嵌入式 IPv4-in-IPv6）。 @@ -175,13 +204,13 @@ def _check_ip_literal(hostname: str, policy: SSRFPolicy) -> tuple[bool, str]: # 1. 标准 IPv4/IPv6 地址 try: ip = ipaddress.ip_address(normalized) - if _is_blocked_ip(ip, policy): - return True, f"目标 IP 在阻止范围内: {normalized}" - # IPv4-mapped IPv6: 检查嵌入式 IPv4（如 ::ffff:127.0.0.1） - if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped: - if _is_blocked_ip(ip.ipv4_mapped, policy): - return True, f"IPv4-mapped 地址解析到被阻止的 IP: {normalized} -> {ip.ipv4_mapped}" - return True, "" # 合法公网 IP，通过 + blocked, reason = _is_blocked_ip_or_embedded(ip, policy) + if blocked: + return True, f"{reason}" + # 包含 IPv4 点分写法但不是已知的 IPv4 嵌入前缀，按安全策略阻止（fail-closed） + if isinstance(ip, ipaddress.IPv6Address) and "." in normalized and _extract_embedded_ipv4(ip) is None: + return True, f"IPv6 地址包含非标准 IPv4 嵌入，按安全策略阻止: {normalized}" + return True, "" # 合法公网 IP 字面量，通过 except ValueError: pass @@ -199,6 +228,14 @@ def _check_ip_literal(hostname: str, policy: SSRFPolicy) -> tuple[bool, str]: return False, "" # 不是 IP 字面量，是域名 +def _is_ambiguous_netloc(netloc: str) -> bool: + """检测 netloc 是否包含不同 URL 解析器可能解释不一致的特征。 + + 多个 @ 或反斜杠常被用于 SSRF 解析绕过，按安全策略直接拒绝。 + """ + return netloc.count("@") > 1 or "\\" in netloc + + def log_ssrf_block(hostname: str, reason: str, context: str = "url-fetch") -> None: """记录 SSRF 拦截事件（安全审计日志）。 @@ -238,6 +275,9 @@ class SSRFValidator: if parsed.scheme not in self.policy.allowed_schemes: return False, f"不允许的协议: {parsed.scheme}，仅允许 {self.policy.allowed_schemes}" + if _is_ambiguous_netloc(parsed.netloc): + return False, "URL 的 netloc 存在歧义，按安全策略阻止" + hostname = parsed.hostname if not hostname: return False, "URL 缺少主机名" @@ -270,18 +310,38 @@ class SSRFValidator: for ip_str in resolved_ips: try: ip = ipaddress.ip_address(ip_str) - if _is_blocked_ip(ip, self.policy): - return False, f"DNS 解析到被阻止的 IP: {hostname} -> {ip_str}" + blocked, reason = _is_blocked_ip_or_embedded(ip, self.policy) + if blocked: + return False, f"DNS 解析到被阻止的 IP: {hostname} -> {ip_str} ({reason})" except ValueError: # 解析出非标准 IP — fail-closed return False, f"DNS 解析到非标准地址，按安全策略阻止: {hostname} -> {ip_str}" return True, "通过" + async def validate_url_and_resolve(self, url: str) -> tuple[bool, str, list[str]]: + """校验 URL 并返回缓存的解析 IP 列表。 + + 调用方可以使用返回的 IP 替换主机名发起请求（配合 Host 头）， + 从而把 DNS 解析结果固定到真实连接上，降低 DNS Rebinding 风险。 + + Returns: + (是否通过, 原因描述, 解析到的 IP 字符串列表)。未通过时 IP 列表为空。 + """ + allowed, reason = await self.validate_url(url) + if not allowed: + return False, reason, [] + + parsed = urlparse(url) + hostname = parsed.hostname + if not hostname: + return False, "URL 缺少主机名", [] + + resolved_ips = await self._resolve_with_pin(hostname) + return True, "通过", resolved_ips + def validate_url_sync(self, url: str) -> tuple[bool, str]: """同步版本的 validate_url，用于非 async 上下文。DNS 解析会阻塞。""" - import asyncio - try: loop = asyncio.get_running_loop() except RuntimeError: @@ -289,15 +349,97 @@ class SSRFValidator: if loop and loop.is_running(): raise RuntimeError("validate_url_sync 不能在已运行的事件循环中调用，请使用 validate_url") - return asyncio.run(self.validate_url(url)) + return self._validate_url_core_sync(url) + + def validate_url_and_resolve_sync(self, url: str) -> tuple[bool, str, list[str]]: + """同步版本的 validate_url_and_resolve。DNS 解析会阻塞。""" + try: + loop = asyncio.get_running_loop() + except RuntimeError: + loop = None + + if loop and loop.is_running(): + raise RuntimeError( + "validate_url_and_resolve_sync 不能在已运行的事件循环中调用，请使用 validate_url_and_resolve" + ) + + allowed, reason = self._validate_url_core_sync(url) + if not allowed: + return False, reason, [] + + parsed = urlparse(url) + hostname = parsed.hostname + if not hostname: + return False, "URL 缺少主机名", [] + + resolved_ips = self._resolve_with_pin_sync(hostname) + return True, "通过", resolved_ips + + def _validate_url_core_sync(self, url: str) -> tuple[bool, str]: + """URL 安全校验核心逻辑（同步版本，供 validate_url_sync 使用）。""" + # 1. 解析 URL + try: + parsed = urlparse(url) + except Exception: + return False, "URL 格式无效" + + # 2. 协议校验 + if parsed.scheme not in self.policy.allowed_schemes: + return False, f"不允许的协议: {parsed.scheme}，仅允许 {self.policy.allowed_schemes}" + + if _is_ambiguous_netloc(parsed.netloc): + return False, "URL 的 netloc 存在歧义，按安全策略阻止" + + hostname = parsed.hostname + if not hostname: + return False, "URL 缺少主机名" + + # 3. 主机名阻止列表（Pre-DNS） + if _is_blocked_hostname(hostname, self.policy): + return False, f"主机名在阻止列表中: {hostname}" + + # 4. 域名白名单/黑名单（Pre-DNS） + if not _is_allowed_by_domain_whitelist(hostname, self.policy.allowed_domains): + return False, f"域名不在白名单中: {hostname}" + + if _normalize_hostname(hostname) in self.policy._blocked_domains_normalized: + return False, f"域名在黑名单中: {hostname}" + + # 5. IP 字面量校验（Pre-DNS） + is_handled, reason = _check_ip_literal(hostname, self.policy) + if is_handled: + if reason: + return False, reason + # 合法公网 IP 字面量，通过 + return True, "通过" + + # 6. DNS 解析后 IP 校验（Post-DNS，防重绑定） + resolved_ips = self._resolve_with_pin_sync(hostname) + if not resolved_ips: + # DNS 解析失败 — fail-closed + return False, f"DNS 解析失败，按安全策略阻止: {hostname}" + + for ip_str in resolved_ips: + try: + ip = ipaddress.ip_address(ip_str) + blocked, reason = _is_blocked_ip_or_embedded(ip, self.policy) + if blocked: + return False, f"DNS 解析到被阻止的 IP: {hostname} -> {ip_str} ({reason})" + except ValueError: + # 解析出非标准 IP — fail-closed + return False, f"DNS 解析到非标准地址，按安全策略阻止: {hostname} -> {ip_str}" + + return True, "通过" async def _resolve_with_pin(self, hostname: str) -> list[str]: """DNS 解析并缓存，防止重绑定攻击。 在 dns_pinned_ttl 秒内，同一主机名返回缓存的 IP 列表。 """ - import asyncio + return await asyncio.to_thread(self._resolve_with_pin_sync, hostname) + def _resolve_with_pin_sync(self, hostname: str) -> list[str]: + """同步版本 DNS 解析与缓存。""" now = time.time() if hostname in self._dns_cache: cached_ips, cached_at = self._dns_cache[hostname] @@ -305,9 +447,7 @@ class SSRFValidator: return cached_ips try: - addr_infos = await asyncio.to_thread( - socket.getaddrinfo, hostname, None, socket.AF_UNSPEC, socket.SOCK_STREAM - ) + addr_infos = socket.getaddrinfo(hostname, None, socket.AF_UNSPEC, socket.SOCK_STREAM) resolved = list({addr[4][0] for addr in addr_infos}) except (socket.gaierror, OSError): resolved = [] # fail-closed: 调用方应检查空列表