运行aiosmtpd遇SSLV3_ALERT_CERTIFICATE_EXPIRED，如何获取过期证书名？

1. sys.excepthook的局限性 #

默认的sys.excepthook无法直接获取过期证书的名称，因为该错误是在asyncio的SSLProtocol内部抛出的，栈跟踪仅包含SSL协议和传输对象，没有直接暴露证书元数据。而且asyncio事件循环内的异常通常不会触发sys.excepthook，需要专门处理asyncio层面的异常。

2. 可行的解决方案 #

方案一：自定义SSL上下文的验证回调 #

在创建SSL上下文时，通过set_verify方法设置自定义验证函数，在握手过程中提前检查证书有效期，同时提取证书名称：

import ssl
import datetime
from aiosmtpd.controller import Controller
from aiosmtpd.handlers import Sink

def validate_certificate(cert, hostname):
    # 解析证书有效期
    not_after_str = cert["notAfter"]
    not_after = datetime.datetime.strptime(not_after_str, "%b %d %H:%M:%S %Y %Z").replace(tzinfo=datetime.timezone.utc)
    now = datetime.datetime.now(datetime.timezone.utc)
    
    # 检查证书是否过期
    if not_after < now:
        # 提取证书通用名称（CN）
        cert_subject = dict(item[0] for item in cert["subject"])
        cert_name = cert_subject.get("commonName", "未知证书")
        print(f"检测到过期证书: {cert_name}")
        # 抛出包含证书名称的异常，方便后续捕获处理
        raise ssl.SSLError(f"证书 {cert_name} 已过期")
    
    # 返回True表示验证通过
    return True

# 构建SSL上下文
ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
ssl_context.load_cert_chain(certfile="your_server_cert.pem", keyfile="your_server_key.pem")
# 设置验证模式和自定义回调函数
ssl_context.set_verify(ssl.CERT_REQUIRED, validate_certificate)

# 启动aiosmtpd服务
handler = Sink()
controller = Controller(handler, hostname="0.0.0.0", port=465, ssl_context=ssl_context)
controller.start()
print("SMTP服务已启动")

方案二：处理asyncio事件循环的异常 #

如果需要在错误发生后捕获并提取信息，可以为asyncio事件循环设置自定义异常处理器，尝试从异常关联的protocol对象中获取SSL信息（注意：握手失败后SSL对象可能已不可用，此方法可靠性低于方案一）：

import asyncio
import ssl
from aiosmtpd.controller import Controller
from aiosmtpd.handlers import Sink

def handle_async_exception(loop, context):
    exc = context.get("exception")
    if isinstance(exc, ssl.SSLError) and "CERTIFICATE_EXPIRED" in str(exc):
        protocol = context.get("protocol")
        if protocol and hasattr(protocol, "_sslobj"):
            try:
                # 尝试获取已协商的证书（握手失败时可能返回None）
                peer_cert = protocol._sslobj.getpeercert()
                if peer_cert:
                    cert_subject = dict(item[0] for item in peer_cert["subject"])
                    cert_name = cert_subject.get("commonName", "未知证书")
                    print(f"过期证书名称: {cert_name}")
            except Exception:
                print("无法提取证书信息")
    # 调用默认异常处理逻辑
    loop.default_exception_handler(context)

# 配置事件循环
loop = asyncio.get_event_loop()
loop.set_exception_handler(handle_async_exception)

# 构建SSL上下文并启动服务
ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
ssl_context.load_cert_chain(certfile="your_server_cert.pem", keyfile="your_server_key.pem")

handler = Sink()
controller = Controller(handler, hostname="0.0.0.0", port=465, ssl_context=ssl_context, loop=loop)
controller.start()
print("SMTP服务已启动")
loop.run_forever()

总结 #

优先推荐方案一，因为它在握手过程中主动检查证书，能可靠提取证书名称并提前终止无效连接；方案二作为补充，适用于需要事后捕获异常的场景，但可靠性较低。

内容的提问来源于stack exchange，提问作者Cyril N.