使用XTB Python wrapper登录时报ssl.wrap_socket属性不存在错误

问题背景 #

使用XTB官方Python wrapper编写登录测试及tick数据流获取代码，凭证存储在config文件中，代码如下：

#import libraries
from xAPIConnector import *
import config
import time  # 补充：原代码缺少time模块导入，需添加

#connect to API
user_id = config.user_id
pwd= config.pwd

client=APIClient()
resp=client.execute(loginCommand(user_id, pwd))
ssid=resp['streamSessionId']

sclient=APIStreamClient(ssId=ssid, tickFun=procTickExample)
sclient.subscribePrice('EURUSD')

time.sleep(60)

sclient.unsubscribePrice('EURUSD')
time.sleep(5)
sclient.disconnect()
client.disconnect()

运行错误 #

执行代码时触发如下错误：

Traceback (most recent call last):
  File "c:\Python\robot.py", line 9, in <module>
    client=APIClient()
           ^^^^^^^^^^^
  File "c:\Python\xAPIConnector.py", line 160, in __init__ 
    super(APIClient, self).__init__(address, port, encrypt)
  File "c:\Python\xAPIConnector.py", line 58, in __init__  
    self.socket = ssl.wrap_socket(sock)
                  ^^^^^^^^^^^^^^^
AttributeError: module 'ssl' has no attribute 'wrap_socket'

环境为Python 3.12，无同名ssl.py文件，需修改wrapper代码解决。

原Wrapper相关代码 #

wrapper中的初始化方法代码片段：

def __init__(self, address, port, encrypt = False):
    self._ssl = encrypt 
    if self._ssl != True:
        self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    else:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.socket = ssl.wrap_socket(sock)
    self.conn = self.socket
    self._timeout = None
    self._address = address
    self._port = port
    self._decoder = json.JSONDecoder()
    self._receivedData = ''

修改方案 #

ssl.wrap_socket在Python 3.7起被标记为废弃，Python 3.12已完全移除该方法，需改用官方推荐的ssl.create_default_context()方式创建SSL套接字，修改后的代码如下：

def __init__(self, address, port, encrypt = False):
    self._ssl = encrypt 
    if self._ssl != True:
        self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    else:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        # 替换废弃的ssl.wrap_socket为现代SSL上下文实现
        context = ssl.create_default_context()
        self.socket = context.wrap_socket(sock, server_hostname=address)
    self.conn = self.socket
    self._timeout = None
    self._address = address
    self._port = port
    self._decoder = json.JSONDecoder()
    self._receivedData = ''

修改说明 #

• ssl.create_default_context()会创建包含安全默认配置的SSL上下文，比旧方法更安全
• 添加server_hostname=address参数，确保SNI（服务器名称指示）功能正常，避免部分服务器的SSL握手失败问题

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