# coding=utf-8 """ ## SimpleProvider 简单DNS抽象基类 * set_record() ## BaseProvider 标准DNS抽象基类 定义所有 DNS 服务商 API 类应继承的抽象基类，统一接口，便于扩展适配多服务商。 Abstract base class for DNS provider APIs. Defines a unified interface to support extension and adaptation across providers. * _query_zone_id * _query_record * _update_record * _create_record ┌──────────────────────────────────────────────────┐ │ 用户调用 set_record(domain, value...) │ └──────────────────────────────────────────────────┘ │ ▼ ┌──────────────────────────────────────┐ │ 快速解析 是否包含 ~ 或 + 分隔符？ │ └──────────────────────────────────────┘ │ │ [是，拆解成功] [否，无法拆解] sub 和 main│ │ domain ▼ ▼ ┌────────────────────────┐ ┌──────────────────────────┐ │ 查询 zone_id │ │ 自动循环解析 while: │ │ _query_zone_id(main) │ │ _query_zone_id(...) │ └────────────────────────┘ └──────────────────────────┘ │ │ ▼ ▼ zone_id ←──────────────┬─── sub ▼ ┌─────────────────────────────────────┐ │ 查询 record: │ │ _query_record(zone_id, sub, ...) │ └─────────────────────────────────────┘ │ ┌─────────────┴────────────────┐ │ record_id 是否存在？ │ └────────────┬─────────────────┘ │ ┌──────────────┴─────────────┐ │ │ ▼ ▼ ┌─────────────────────┐ ┌─────────────────────┐ │ 更新记录 │ │ 创建记录 │ │ _update_record(...) │ │ _create_record(...) │ └─────────────────────┘ └─────────────────────┘ │ │ ▼ ▼ ┌───────────────────────────────┐ │ 返回操作结果 │ └───────────────────────────────┘ @author: NewFuture """ from os import environ from abc import ABCMeta, abstractmethod from json import loads as jsondecode, dumps as jsonencode from logging import Logger, getLogger # noqa:F401 # type: ignore[no-redef] from ..util.http import send_http_request try: # python 3 from urllib.parse import quote, urlencode except ImportError: # python 2 from urllib import urlencode, quote # type: ignore[no-redef,import-untyped] TYPE_FORM = "application/x-www-form-urlencoded" TYPE_JSON = "application/json" class SimpleProvider(object): """ 简单DNS服务商接口的抽象基类, 必须实现 `set_record` 方法。 Abstract base class for all simple DNS provider APIs. Subclasses must implement `set_record`. * set_record(domain, value, record_type="A", ttl=None, line=None, **extra) """ __metaclass__ = ABCMeta # API endpoint domain (to be defined in subclass) API = "" # type: str # https://exampledns.com # Content-Type for requests (to be defined in subclass) content_type = TYPE_FORM # type: Literal["application/x-www-form-urlencoded"] | Literal["application/json"] # 默认 accept 头部, 空则不设置 accept = TYPE_JSON # type: str | None # Decode Response as JSON by default decode_response = True # 是否验证 SSL 证书，默认为 True verify_ssl = "auto" # type: bool | str # 版本 version = environ.get("DDNS_VERSION", "0.0.0") # Description remark = "Managed by [DDNS v{}](https://ddns.newfuture.cc)".format(version) def __init__(self, auth_id, auth_token, logger=None, verify_ssl=None, **options): # type: (str, str, Logger | None, bool|str| None, **object) -> None """ 初始化服务商对象 Initialize provider instance. Args: auth_id (str): 身份认证 ID / Authentication ID auth_token (str): 密钥 / Authentication Token options (dict): 其它参数，如代理、调试等 / Additional options """ self.auth_id = auth_id # type: str self.auth_token = auth_token # type: str self.options = options name = self.__class__.__name__ self.logger = (logger or getLogger()).getChild(name) self.proxy = None # type: str | None if verify_ssl is not None: self.verify_ssl = verify_ssl self._zone_map = {} # type: dict[str, str] self.logger.debug("%s initialized with: %s", self.__class__.__name__, auth_id) self._validate() # 验证身份认证信息 @abstractmethod def set_record(self, domain, value, record_type="A", ttl=None, line=None, **extra): # type: (str, str, str, str | int | None, str | None, **object) -> bool """ 设置 DNS 记录（创建或更新） Set or update DNS record. Args: domain (str): 完整域名 value (str): 新记录值 record_type (str): 记录类型 ttl (int | None): TTL 值，可选 line (str | None): 线路信息 extra (dict): 额外参数 Returns: Any: 执行结果 """ raise NotImplementedError("This set_record should be implemented by subclasses") def set_proxy(self, proxy_str): # type: (str | None) -> SimpleProvider """ 设置代理服务器 Set HTTPS proxy string. Args: proxy_str (str): 代理地址 Returns: Self: 自身 """ self.proxy = proxy_str return self def _validate(self): # type: () -> None """ 验证身份认证信息是否填写 Validate authentication credentials. """ if not self.auth_id: raise ValueError("id must be configured") if not self.auth_token: raise ValueError("token must be configured") if not self.API: raise ValueError("API endpoint must be defined in {}".format(self.__class__.__name__)) def _http(self, method, url, params=None, body=None, queries=None, headers=None): # noqa: C901 # type: (str, str, dict[str,Any]|str|None, dict[str,Any]|str|None, dict[str,Any]|None, dict|None) -> Any """ 发送 HTTP/HTTPS 请求，自动根据 API/url 选择协议。 Args: method (str): 请求方法，如 GET、POST url (str): 请求路径 params (dict[str, Any] | None): 请求参数,自动处理 query string 或者body body (dict[str, Any] | str | None): 请求体内容 queries (dict[str, Any] | None): 查询参数，自动处理为 URL 查询字符串 headers (dict): 头部，可选 Returns: Any: 解析后的响应内容 Raises: RuntimeError: 当响应状态码为400/401或5xx(服务器错误)时抛出异常 """ method = method.upper() # 简化参数处理逻辑 query_params = queries or {} if params: if method in ("GET", "DELETE"): if isinstance(params, dict): query_params.update(params) else: # params是字符串，直接作为查询字符串 url += ("&" if "?" in url else "?") + str(params) params = None elif body is None: body = params # 构建查询字符串 if len(query_params) > 0: url += ("&" if "?" in url else "?") + self._encode(query_params) # 构建完整URL if not url.startswith("http://") and not url.startswith("https://"): if not url.startswith("/") and self.API.endswith("/"): url = "/" + url url = self.API + url # 记录请求日志 self.logger.info("%s %s", method, self._mask_sensitive_data(url)) # 处理请求体 body_data, headers = None, headers or {} if body: if "content-type" not in headers: headers["content-type"] = self.content_type if isinstance(body, (str, bytes)): body_data = body elif self.content_type == TYPE_FORM: body_data = self._encode(body) else: body_data = jsonencode(body) self.logger.debug("body:\n%s", self._mask_sensitive_data(body_data)) # 处理headers if self.accept and "accept" not in headers and "Accept" not in headers: headers["accept"] = self.accept if len(headers) > 2: self.logger.debug("headers:\n%s", {k: self._mask_sensitive_data(v) for k, v in headers.items()}) response = send_http_request( url=url, method=method, body=body_data, headers=headers, proxy=self.proxy, max_redirects=5, verify_ssl=self.verify_ssl, ) # 处理响应 status_code = response.status if not (200 <= status_code < 300): self.logger.warning("response status: %s %s", status_code, response.reason) res = response.body # 针对客户端错误、认证/授权错误和服务器错误直接抛出异常 if status_code >= 500 or status_code in (400, 401, 403): self.logger.error("HTTP error:\n%s", res) if status_code == 400: raise RuntimeError("请求参数错误 [400]: " + response.reason) elif status_code == 401: raise RuntimeError("认证失败 [401]: " + response.reason) elif status_code == 403: raise RuntimeError("权限不足 [403]: " + response.reason) else: raise RuntimeError("服务器错误 [{}]: {}".format(status_code, response.reason)) self.logger.debug("response:\n%s", res) if not self.decode_response: return res try: return jsondecode(res) except Exception as e: self.logger.error("fail to decode response: %s", e) return res @staticmethod def _encode(params): # type: (dict|list|str|bytes|None) -> str """ 编码参数为 URL 查询字符串 Args: params (dict|list|str|bytes|None): 参数字典、列表或字符串 Returns: str: 编码后的查询字符串 """ if not params: return "" elif isinstance(params, (str, bytes)): return params # type: ignore[return-value] return urlencode(params, doseq=True) @staticmethod def _quote(data, safe="/"): # type: (str, str) -> str """ 对字符串进行 URL 编码 Args: data (str): 待编码字符串 Returns: str: 编码后的字符串 """ return quote(data, safe=safe) def _mask_sensitive_data(self, data): # type: (str | bytes | None) -> str | bytes | None """ 对敏感数据进行打码处理，用于日志输出，支持URL编码的敏感信息 Args: data (str | bytes | None): 需要处理的数据 Returns: str | bytes | None: 打码后的字符串 """ if not data or not self.auth_token: return data # 生成打码后的token token_masked = self.auth_token[:2] + "***" + self.auth_token[-2:] if len(self.auth_token) > 4 else "***" token_encoded = quote(self.auth_token, safe="") if isinstance(data, bytes): # 处理字节数据 return data.replace(self.auth_token.encode(), token_masked.encode()).replace( token_encoded.encode(), token_masked.encode() ) if hasattr(data, "replace"): # 处理字符串数据 return data.replace(self.auth_token, token_masked).replace(token_encoded, token_masked) return data class BaseProvider(SimpleProvider): """ 标准DNS服务商接口的抽象基类 Abstract base class for all standard DNS provider APIs. Subclasses must implement the abstract methods to support various providers. * _query_zone_id() * _query_record_id() * _update_record() * _create_record() """ def set_record(self, domain, value, record_type="A", ttl=None, line=None, **extra): # type: (str, str, str, str | int | None, str | None, **Any) -> bool """ 设置 DNS 记录（创建或更新） Set or update DNS record. Args: domain (str): 完整域名 value (str): 新记录值 record_type (str): 记录类型 ttl (int | None): TTL 值，可选 line (str | None): 线路信息 extra (dict): 额外参数 Returns: bool: 执行结果 """ domain = domain.lower() self.logger.info("%s => %s(%s)", domain, value, record_type) # 优化域名解析逻辑 sub, main = self._split_custom_domain(domain) try: if sub is not None: # 使用自定义分隔符格式 zone_id = self.get_zone_id(main) else: # 自动分析域名 zone_id, sub, main = self._split_zone_and_sub(domain) self.logger.info("sub: %s, main: %s(id=%s)", sub, main, zone_id) if not zone_id or sub is None: self.logger.critical("找不到 zone_id 或 subdomain: %s", domain) return False # 查询现有记录 record = self._query_record(zone_id, sub, main, record_type=record_type, line=line, extra=extra) # 更新或创建记录 if record: self.logger.info("Found existing record: %s", record) return self._update_record(zone_id, record, value, record_type, ttl=ttl, line=line, extra=extra) else: self.logger.warning("No existing record found, creating new one") return self._create_record(zone_id, sub, main, value, record_type, ttl=ttl, line=line, extra=extra) except Exception as e: self.logger.exception("Error setting record for %s: %s", domain, e) return False def get_zone_id(self, domain): # type: (str) -> str | None """ 查询指定域名对应的 zone_id Get zone_id for the domain. Args: domain (str): 主域名 / main name Returns: str | None: 区域 ID / Zone identifier """ if domain in self._zone_map: return self._zone_map[domain] zone_id = self._query_zone_id(domain) if zone_id: self._zone_map[domain] = zone_id return zone_id @abstractmethod def _query_zone_id(self, domain): # type: (str) -> str | None """ 查询主域名的 zone ID Args: domain (str): 主域名 Returns: str | None: Zone ID """ return domain @abstractmethod def _query_record(self, zone_id, subdomain, main_domain, record_type, line, extra): # type: (str, str, str, str, str | None, dict) -> Any """ 查询 DNS 记录 ID Args: zone_id (str): 区域 ID subdomain (str): 子域名 main_domain (str): 主域名 record_type (str): 记录类型，例如 A、AAAA line (str | None): 线路选项，可选 extra (dict): 额外参数 Returns: Any | None: 记录 """ raise NotImplementedError("This _query_record should be implemented by subclasses") @abstractmethod def _create_record(self, zone_id, subdomain, main_domain, value, record_type, ttl, line, extra): # type: (str, str, str, str, str, int | str | None, str | None, dict) -> bool """ 创建新 DNS 记录 Args: zone_id (str): 区域 ID subdomain (str): 子域名 main_domain (str): 主域名 value (str): 记录值 record_type (str): 类型，如 A ttl (int | None): TTL 可选 line (str | None): 线路选项 extra (dict | None): 额外字段 Returns: Any: 操作结果 """ raise NotImplementedError("This _create_record should be implemented by subclasses") @abstractmethod def _update_record(self, zone_id, old_record, value, record_type, ttl, line, extra): # type: (str, dict, str, str, int | str | None, str | None, dict) -> bool """ 更新已有 DNS 记录 Args: zone_id (str): 区域 ID old_record (dict): 旧记录信息 value (str): 新的记录值 record_type (str): 类型 ttl (int | None): TTL line (str | None): 线路 extra (dict | None): 额外参数 Returns: bool: 操作结果 """ raise NotImplementedError("This _update_record should be implemented by subclasses") def _split_zone_and_sub(self, domain): # type: (str) -> tuple[str | None, str | None, str ] """ 从完整域名拆分主域名和子域名 Args: domain (str): 完整域名 Returns: (zone_id, sub): 元组 """ domain_split = domain.split(".") zone_id = None index = 2 main = "" while not zone_id and index <= len(domain_split): main = ".".join(domain_split[-index:]) zone_id = self.get_zone_id(main) index += 1 if zone_id: sub = ".".join(domain_split[: -index + 1]) or "@" self.logger.debug("zone_id: %s, sub: %s", zone_id, sub) return zone_id, sub, main return None, None, main @staticmethod def _split_custom_domain(domain): # type: (str) -> tuple[str | None, str] """ 拆分支持 ~ 或 + 的自定义格式域名为 (子域, 主域) 如 sub~example.com => ('sub', 'example.com') Returns: (sub, main): 子域 + 主域 """ for sep in ("~", "+"): if sep in domain: sub, main = domain.split(sep, 1) return sub, main return None, domain @staticmethod def _join_domain(sub, main): # type: (str | None, str) -> str """ 合并子域名和主域名为完整域名 Args: sub (str | None): 子域名 main (str): 主域名 Returns: str: 完整域名 """ sub = sub and sub.strip(".").strip().lower() main = main and main.strip(".").strip().lower() if not sub or sub == "@": if not main: raise ValueError("Both sub and main cannot be empty") return main if not main: return sub return "{}.{}".format(sub, main)