如何使用Twisted连接带自签名过期SSL证书的设备

问题原因 #

你的代码已尝试信任指定证书，但报错核心原因是证书已过期——Twisted默认SSL验证逻辑会检查证书有效期，即便信任该证书，过期状态仍会触发验证失败。

以下提供两种可行方案：

方案1：信任指定的过期自签名证书（推荐，仅信任目标证书） #

修改CustomPolicyForHTTPS类，自定义证书验证回调，跳过过期检查的同时，仅信任你提供的目标证书：

#!/usr/bin/python3

import sys
from twisted.internet import ssl, reactor
from twisted.web.client import Agent
from twisted.internet.ssl import Certificate
from twisted.web.http_headers import Headers
from twisted.web.client import readBody
from twisted.web.iweb import IPolicyForHTTPS
from zope.interface import implementer
import argparse
from OpenSSL.SSL import VERIFY_NONE, Context

@implementer(IPolicyForHTTPS)
class CustomPolicyForHTTPS:
    """
    自定义HTTPS策略：信任指定的过期自签名证书
    """
    def __init__(self, public_cert_path):
        with open(public_cert_path, 'rb') as f:
            cert_data = f.read()
        # 加载信任的证书并保存原始OpenSSL对象
        self.trusted_cert = Certificate.loadPEM(cert_data).original

    def creatorForNetloc(self, hostname, port):
        def verify_callback(connection, x509, errnum, errdepth, ok):
            # 对比服务器证书与本地证书的SHA256摘要，确保只信任目标证书
            if x509.digest('sha256') == self.trusted_cert.digest('sha256'):
                # 跳过过期检查，直接返回验证通过
                return True
            return ok

        ctx = Context(ssl.SSLv23_METHOD)
        # 设置自定义验证回调，禁用默认验证逻辑
        ctx.set_verify(VERIFY_NONE, verify_callback)
        return ssl.ClientContextFactory.forContext(ctx)

def make_request(url, cert_path):
    policy = CustomPolicyForHTTPS(cert_path)
    agent = Agent(reactor, contextFactory=policy)
    
    d = agent.request(
        b'GET',
        url.encode('utf-8'),
        Headers({'User-Agent': ['Twisted Web Client Example']}),
        None
    )
    
    def cb_response(response):
        print('响应状态码:', response.code)
        return readBody(response)
    
    def cb_body(body):
        print('响应内容:', body.decode('utf-8'))
        reactor.stop()
    
    def handle_error(err):
        print(f"错误: {err.value}")
        print(f"错误详情: {err.getErrorMessage()}")
        if reactor.running:
            reactor.stop()
    
    d.addCallback(cb_response)
    d.addCallback(cb_body)
    d.addErrback(handle_error)
    
    return d

def parse_arguments():
    parser = argparse.ArgumentParser(description='使用信任证书发起HTTPS请求')
    parser.add_argument('url', help='目标URL')
    parser.add_argument('cert_path', help='PEM格式的公钥证书路径')
    parser.add_argument('--verbose', '-v', action='store_true', help='启用详细输出')
    
    return parser.parse_args()

if __name__ == "__main__":
    args = parse_arguments()

    if args.verbose:
        print(f"连接到: {args.url}")
        print(f"使用证书: {args.cert_path}")

    make_request(args.url, args.cert_path)
    reactor.run()

关键修改点： #

• 自定义verify_callback，通过SHA256摘要匹配服务器证书与本地证书，确保仅信任目标设备
• 匹配成功时直接跳过过期检查，返回验证通过
• 使用VERIFY_NONE配合自定义回调，替代默认的严格验证逻辑

方案2：完全忽略所有SSL验证（适合封闭网络） #

若完全不担心安全问题，可直接禁用所有SSL验证，这是最简单的实现方式：

#!/usr/bin/python3

import sys
from twisted.internet import ssl, reactor
from twisted.web.client import Agent
from twisted.web.http_headers import Headers
from twisted.web.client import readBody
from twisted.web.iweb import IPolicyForHTTPS
from zope.interface import implementer
import argparse
from OpenSSL.SSL import VERIFY_NONE, Context

@implementer(IPolicyForHTTPS)
class NoVerifyPolicyForHTTPS:
    """
    完全忽略SSL验证的HTTPS策略
    """
    def creatorForNetloc(self, hostname, port):
        ctx = Context(ssl.SSLv23_METHOD)
        # 禁用所有SSL验证步骤
        ctx.set_verify(VERIFY_NONE, lambda *args: True)
        return ssl.ClientContextFactory.forContext(ctx)

def make_request(url):
    policy = NoVerifyPolicyForHTTPS()
    agent = Agent(reactor, contextFactory=policy)
    
    d = agent.request(
        b'GET',
        url.encode('utf-8'),
        Headers({'User-Agent': ['Twisted Web Client Example']}),
        None
    )
    
    def cb_response(response):
        print('响应状态码:', response.code)
        return readBody(response)
    
    def cb_body(body):
        print('响应内容:', body.decode('utf-8'))
        reactor.stop()
    
    def handle_error(err):
        print(f"错误: {err.value}")
        print(f"错误详情: {err.getErrorMessage()}")
        if reactor.running:
            reactor.stop()
    
    d.addCallback(cb_response)
    d.addCallback(cb_body)
    d.addErrback(handle_error)
    
    return d

def parse_arguments():
    parser = argparse.ArgumentParser(description='忽略SSL验证发起HTTPS请求')
    parser.add_argument('url', help='目标URL')
    parser.add_argument('--verbose', '-v', action='store_true', help='启用详细输出')
    
    return parser.parse_args()

if __name__ == "__main__":
    args = parse_arguments()

    if args.verbose:
        print(f"连接到: {args.url}")

    make_request(args.url)
    reactor.run()

说明： #

• 该方案会跳过所有SSL验证，包括证书过期、自签名、主机名不匹配等所有问题
• 仅推荐在完全可控的封闭网络环境中使用

关于ReactorNotRestartable报错 #

日志中的ReactorNotRestartable是因为错误处理中重复调用reactor.stop()导致的。优化后的代码已添加if reactor.running:判断，确保反应器仅停止一次，避免该报错。

内容的提问来源于stack exchange，提问作者Maxwell Bottiger