DDNS/ddns/util/http.py

# coding=utf-8
"""
HTTP请求工具模块

HTTP utilities module for DDNS project.
Provides common HTTP functionality including ssl, proxy, and basicauth.

@author: NewFuture
"""

from logging import getLogger
from re import compile
import ssl
import os

try:  # python 3
    from urllib.request import (  # noqa: F401
        HTTPBasicAuthHandler,
        HTTPSHandler,
        BaseHandler,
        Request,
        HTTPPasswordMgrWithDefaultRealm,
        HTTPDefaultErrorHandler,
        ProxyHandler,
        build_opener,
    )
    from urllib.parse import quote, urlencode, unquote
except ImportError:  # python 2
    from urllib2 import (  # type: ignore[no-redef]
        Request,
        HTTPSHandler,
        ProxyHandler,
        HTTPDefaultErrorHandler,
        HTTPPasswordMgrWithDefaultRealm,
        HTTPBasicAuthHandler,
        build_opener,
    )
    from urllib import urlencode, quote, unquote  # type: ignore[no-redef]


__all__ = ["send_http_request", "HttpResponse", "quote", "urlencode"]

logger = getLogger().getChild(__name__)
_AUTH_URL_RE = compile(r"^(https?://)([^:/?#]+):([^@]+)@(.+)$")


class NoHTTPErrorProcessor(HTTPDefaultErrorHandler):  # type: ignore[misc]
    """自定义HTTP错误处理器，处理所有HTTP错误状态码，返回响应而不抛出异常"""

    def http_error_default(self, req, fp, code, msg, hdrs):
        """处理所有HTTP错误状态码，返回响应而不抛出异常"""
        logger.warning("HTTP error %s: %s", code, msg)
        return fp


class SSLAutoFallbackHandler(HTTPSHandler):  # type: ignore[misc]
    """SSL自动降级处理器，处理 unable to get local issuer certificate 错误"""

    # 类级别的SSL上下文缓存
    _ssl_cache = {}  # type: dict[str, ssl.SSLContext]

    def __init__(self, verify_ssl):
        # type: (bool | str) -> None
        self._verify_ssl = verify_ssl
        ssl_context = self._get_ssl_context()
        # 兼容性：优先使用context参数，失败时降级
        try:  # python 3 / python 2.7.9+
            HTTPSHandler.__init__(self, context=ssl_context)
        except (TypeError, AttributeError):  # python 2.7.8-
            HTTPSHandler.__init__(self)

    def _get_ssl_context(self):
        # type: () -> ssl.SSLContext | None
        """创建或获取缓存的SSLContext"""
        # 缓存键
        cache_key = "default"
        if not self._verify_ssl:
            if not hasattr(ssl, "_create_unverified_context"):
                return None
            cache_key = "unverified"
            if cache_key not in self._ssl_cache:
                self._ssl_cache[cache_key] = ssl._create_unverified_context()
        elif hasattr(self._verify_ssl, "lower") and self._verify_ssl.lower() not in ("auto", "true"):  # type: ignore
            cache_key = str(self._verify_ssl)
            if cache_key not in self._ssl_cache:
                self._ssl_cache[cache_key] = ssl.create_default_context(ssl.Purpose.SERVER_AUTH, cafile=cache_key)
        elif cache_key not in self._ssl_cache:
            ssl_context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
            if not ssl_context.get_ca_certs():
                logger.warning("No system CA certificates found, loading default CA certificates")
                _load_system_ca_certs(ssl_context)
            self._ssl_cache[cache_key] = ssl_context

        return self._ssl_cache[cache_key]

    def https_open(self, req):
        """处理HTTPS请求，自动处理SSL错误"""
        try:
            return HTTPSHandler.https_open(self, req)
        except Exception as e:
            # SSL auto模式：只处理 unable to get local issuer certificate 错误
            if self._verify_ssl == "auto" and "unable to get local issuer certificate" in str(e).lower():
                msg = "unable to get local issuer certificate, switching to unverified connection for %s"
                logger.warning(msg, req.get_full_url())
                self._verify_ssl = False
                # 创建不验证SSL的临时处理器重试
                try:  # python 3 / python 2.7.9+
                    temp_handler = HTTPSHandler(context=self._get_ssl_context())
                except (TypeError, AttributeError):  # python 2.7.8-
                    temp_handler = HTTPSHandler()
                return temp_handler.https_open(req)
            else:
                raise


class HttpResponse(object):
    """HTTP响应封装类"""

    def __init__(self, status, reason, headers, body):
        # type: (int, str, Any, str) -> None
        """
        初始化HTTP响应对象

        Args:
            status (int): HTTP状态码
            reason (str): 状态原因短语
            headers (Any): 响应头对象，直接使用 response.info()
            body (str): 响应体内容
        """
        self.status = status
        self.reason = reason
        self.headers = headers
        self.body = body


def _load_system_ca_certs(ssl_context):
    # type: (ssl.SSLContext) -> None
    """加载系统CA证书"""
    ca_paths = [
        # Linux/Unix常用路径
        "/etc/ssl/certs/ca-certificates.crt",  # Debian/Ubuntu
        "/etc/pki/tls/certs/ca-bundle.crt",  # RedHat/CentOS
        "/etc/ssl/ca-bundle.pem",  # OpenSUSE
        "/etc/ssl/cert.pem",  # OpenBSD
        "/usr/local/share/certs/ca-root-nss.crt",  # FreeBSD
        "/etc/pki/ca-trust/extracted/pem/tls-ca-bundle.pem",  # Fedora/RHEL
        # macOS路径
        "/usr/local/etc/openssl/cert.pem",  # macOS with Homebrew
        "/opt/local/etc/openssl/cert.pem",  # macOS with MacPorts
    ]

    for ca_path in ca_paths:
        if os.path.isfile(ca_path):
            try:
                ssl_context.load_verify_locations(ca_path)
                logger.info("Loaded CA certificates from: %s", ca_path)
                return  # 成功加载后立即返回
            except Exception as e:
                logger.warning("Failed to load CA certificates from %s: %s", ca_path, e)


def send_http_request(method, url, body=None, headers=None, proxy=None, verify_ssl=True, auth_handler=None):
    # type: (str, str, str | bytes | None, dict[str, str] | None, str | None, bool | str, BaseHandler | None) -> HttpResponse # noqa: E501
    """
    发送HTTP/HTTPS请求，支持重定向跟随和灵活的SSL验证

    Args:
        method (str): HTTP方法，如GET、POST等
        url (str): 请求的URL，支持嵌入式认证格式 https://user:[email protected]
        body (str | bytes | None): 请求体
        headers (dict[str, str] | None): 请求头
        proxy (str | None): 代理地址，格式为 http://proxy:port
        verify_ssl (bool | str): SSL验证配置
                                - True: 启用标准SSL验证
                                - False: 禁用SSL验证
                                - "auto": 启用验证，失败时自动回退到不验证
                                - str: 自定义CA证书文件路径
        auth_handler (BaseHandler | None): 自定义认证处理器

    Returns:
        HttpResponse: 响应对象

    Raises:
        URLError: 如果请求失败
        ssl.SSLError: 如果SSL验证失败
    """
    # 解析URL以检查是否包含嵌入式认证信息
    m = _AUTH_URL_RE.match(url)
    if m:
        protocol, username, password, rest = m.groups()
        clean_url = protocol + rest
        password_mgr = HTTPPasswordMgrWithDefaultRealm()  # 使用urllib的内置认证机制
        password_mgr.add_password(None, clean_url, unquote(username), unquote(password))
        auth_handler = HTTPBasicAuthHandler(password_mgr)
        url = clean_url

    # 准备请求
    if isinstance(body, str):
        body = body.encode("utf-8")

    req = Request(url, data=body)
    req.get_method = lambda: method  # type: ignore[attr-defined]

    if headers:
        for key, value in headers.items():
            req.add_header(key, value)

    # 创建opener并发送请求
    handlers = [NoHTTPErrorProcessor(), SSLAutoFallbackHandler(verify_ssl=verify_ssl)]  # type: list[BaseHandler]
    if proxy:
        handlers.append(ProxyHandler({"http": proxy, "https": proxy}))
    if auth_handler:
        handlers.append(auth_handler)
    opener = build_opener(*handlers)
    response = opener.open(req)

    # 处理响应
    response_headers = response.info()
    raw_body = response.read()
    decoded_body = _decode_response_body(raw_body, response_headers.get("Content-Type"))
    status_code = response.getcode()
    reason = getattr(response, "msg", "")

    return HttpResponse(status_code, reason, response_headers, decoded_body)


def _decode_response_body(raw_body, content_type):
    # type: (bytes, str | None) -> str
    """解码HTTP响应体，优先使用UTF-8"""
    if not raw_body:
        return ""

    # 从Content-Type提取charset
    charsets = ["utf-8", "gbk", "ascii", "latin-1"]
    if content_type and "charset=" in content_type.lower():
        start = content_type.lower().find("charset=") + 8
        end = content_type.find(";", start)
        if end == -1:
            end = len(content_type)
        charset = content_type[start:end].strip("'\" ").lower()

        # 处理常见别名映射
        charset_aliases = {"gb2312": "gbk", "iso-8859-1": "latin-1"}
        charset = charset_aliases.get(charset, charset)
        if charset in charsets:
            charsets.remove(charset)
        charsets.insert(0, charset)

    # 按优先级尝试解码
    for encoding in charsets:
        try:
            return raw_body.decode(encoding)
        except (UnicodeDecodeError, LookupError):
            continue

    # 最终后备：UTF-8替换错误字符
    return raw_body.decode("utf-8", errors="replace")