Python非Socket场景下TLS通信实现需求（USB嵌入式设备）

方案1：基于SSL模块的wrap_bio方法（推荐，Python原生支持） #

Python 3.7+的ssl.SSLContext.wrap_bio()方法可直接操作自定义输入输出流（BIO），完全脱离socket依赖。只需实现具备read()和write()方法的BIO类，将读写逻辑映射到USB操作即可。

示例代码： #

import ssl
import usb.core
import usb.util

class UsbBIO:
    def __init__(self, dev, ep_out, ep_in):
        self.dev = dev
        self.ep_out = ep_out  # USB输出端点（写设备）
        self.ep_in = ep_in    # USB输入端点（读设备）

    def write(self, data):
        # 向USB设备发送数据，返回写入长度
        return self.dev.write(self.ep_out, data, timeout=5000)

    def read(self, bufsize):
        # 从USB设备读取数据，返回字节串
        return self.dev.read(self.ep_in, bufsize, timeout=5000)

# 1. 初始化USB设备（替换为你的设备VID/PID）
dev = usb.core.find(idVendor=0x1234, idProduct=0x5678)
if not dev:
    raise RuntimeError("USB设备未找到")
dev.set_configuration()

# 2. 获取USB端点（根据你的设备描述符调整）
cfg = dev.get_active_configuration()
intf = cfg[(0, 0)]
ep_out = usb.util.find_descriptor(intf, custom_match=lambda e: usb.util.endpoint_direction(e.bEndpointAddress) == usb.util.ENDPOINT_OUT)
ep_in = usb.util.find_descriptor(intf, custom_match=lambda e: usb.util.endpoint_direction(e.bEndpointAddress) == usb.util.ENDPOINT_IN)

# 3. 创建SSL上下文
ctx = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
# 测试场景下可禁用证书验证（生产环境请加载对应CA证书）
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE

# 4. 绑定自定义BIO与SSL
usb_bio = UsbBIO(dev, ep_out, ep_in)
ssl_conn = ctx.wrap_bio(usb_bio, usb_bio, server_hostname=None)

# 5. 执行TLS握手并通信
try:
    ssl_conn.do_handshake()
    ssl_conn.write(b"Hello TLS over USB")
    resp = ssl_conn.read(1024)
    print(f"收到设备响应: {resp}")
finally:
    ssl_conn.close()
    usb.util.dispose_resources(dev)

方案2：模拟Socket类适配SSL.wrap_socket() #

如果需要兼容旧版Python，可自定义一个模拟socket接口的类，实现SSL模块依赖的send()、recv()、close()方法，部分场景需模拟fileno()（返回无效文件描述符即可）。

示例代码： #

import ssl
import usb.core
import usb.util

class UsbSocketSim:
    def __init__(self, dev, ep_out, ep_in):
        self.dev = dev
        self.ep_out = ep_out
        self.ep_in = ep_in

    def send(self, data):
        return self.dev.write(self.ep_out, data, timeout=5000)

    def recv(self, bufsize):
        return self.dev.read(self.ep_in, bufsize, timeout=5000)

    def close(self):
        usb.util.dispose_resources(self.dev)

    def fileno(self):
        # 模拟文件描述符，避免SSL模块报错
        return -1

# 初始化USB设备（同方案1）
dev = usb.core.find(idVendor=0x1234, idProduct=0x5678)
dev.set_configuration()
cfg = dev.get_active_configuration()
intf = cfg[(0, 0)]
ep_out = usb.util.find_descriptor(intf, custom_match=lambda e: usb.util.endpoint_direction(e.bEndpointAddress) == usb.util.ENDPOINT_OUT)
ep_in = usb.util.find_descriptor(intf, custom_match=lambda e: usb.util.endpoint_direction(e.bEndpointAddress) == usb.util.ENDPOINT_IN)

# 创建SSL上下文并模拟Socket
ctx = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE

usb_sim = UsbSocketSim(dev, ep_out, ep_in)
ssl_conn = ctx.wrap_socket(usb_sim, server_hostname=None)

# 通信逻辑
try:
    ssl_conn.send(b"Test via simulated socket")
    resp = ssl_conn.recv(1024)
    print(f"响应: {resp}")
finally:
    ssl_conn.close()

方案3：使用WolfSSL官方Python绑定 #

直接使用WolfSSL的Python库（wolfssl-python），可与嵌入式端的WolfSSL无缝适配，支持自定义IO回调函数，完全摆脱socket依赖。

示例代码： #

import wolfssl
import usb.core
import usb.util

# 初始化USB设备
dev = usb.core.find(idVendor=0x1234, idProduct=0x5678)
dev.set_configuration()
cfg = dev.get_active_configuration()
intf = cfg[(0, 0)]
ep_out = usb.util.find_descriptor(intf, custom_match=lambda e: usb.util.endpoint_direction(e.bEndpointAddress) == usb.util.ENDPOINT_OUT)
ep_in = usb.util.find_descriptor(intf, custom_match=lambda e: usb.util.endpoint_direction(e.bEndpointAddress) == usb.util.ENDPOINT_IN)

# 自定义USB读写回调
def usb_send(ctx, data, user_data):
    return dev.write(ep_out, data, timeout=5000)

def usb_recv(ctx, buf, buf_len, user_data):
    data = dev.read(ep_in, buf_len, timeout=5000)
    buf[:len(data)] = data
    return len(data)

# 创建WolfSSL上下文并设置回调
ctx = wolfssl.SSLContext(wolfssl.PROTOCOL_TLS_CLIENT)
ctx.check_hostname = False
ctx.verify_mode = wolfssl.CERT_NONE

ssl_conn = ctx.wrap_socket(None, server_hostname=None)
ssl_conn.set_send_callback(usb_send, None)
ssl_conn.set_recv_callback(usb_recv, None)

# 握手与通信
try:
    ssl_conn.do_handshake()
    ssl_conn.send(b"WolfSSL direct communication")
    resp = ssl_conn.recv(1024)
    print(f"设备响应: {resp}")
finally:
    ssl_conn.close()
    usb.util.dispose_resources(dev)

注意事项： #

• USB读写通常是阻塞式的，务必设置合理的超时时间，避免程序挂死。
• 生产环境中不要禁用证书验证，需加载嵌入式设备对应的CA证书或自签名证书。
• 确保USB端点的配置与设备实际一致，可通过usb.core.show_devices()查看设备描述符。

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