背景: 这是基于此 docker 映像的 python 3.11.6:python:3.11.6-slim-bullseye
这是一个最新的 openssl 库
print(ssl.OPENSSL_VERSION)
PyDev console: starting.
OpenSSL 1.1.1w 11 Sep 2023
我有这个测试脚本:
def test_connection(db_no_rollback):
HTTPConnection.debuglevel = 1
# Initialize a logger
logging.basicConfig()
logging.getLogger().setLevel(logging.DEBUG)
requests_log = logging.getLogger("requests.packages.urllib3")
requests_log.setLevel(logging.DEBUG)
requests_log.propagate = True
import ssl, socket
hostname = 'www.handlingandfulfilment.co.uk'
# hostname = "www.growthpath.com.au"
# context = ssl.create_default_context()
context = ssl.SSLContext(ssl.PROTOCOL_TLS) # create default context
context.options |= ssl.OP_NO_TLSv1 | ssl.OP_NO_TLSv1_1 # disable TLS 1.0 and 1.1
with socket.create_connection((hostname, 443)) as sock:
with context.wrap_socket(sock, server_hostname=hostname) as ssock:
print(ssock.getpeercert())
assert True
对于大多数主机名,wireshark 显示如下行:
6412 410.231952331 192.168.41.11 188.166.227.141 TLSv1.3 583 Client Hello
这对我来说很有意义,TLSv1.3
但是对于有问题的服务器,我得到了这个:
421 9.896938114 192.168.41.11 193.109.12.6 TLSv1 571 Client Hello
源地址(192.168.41.11)是我的开发机器。 所以我的 python 代码使用 TLSv1 发送 hello
这会导致异常:
self = <ssl.SSLSocket [closed] fd=-1, family=2, type=1, proto=6>, block = False
@_sslcopydoc
def do_handshake(self, block=False):
self._check_connected()
timeout = self.gettimeout()
try:
if timeout == 0.0 and block:
self.settimeout(None)
> self._sslobj.do_handshake()
E ConnectionResetError: [Errno 104] Connection reset by peer
我确信这是因为服务器拒绝 TLSv1 另外,服务器管理员指责我尝试连接此版本。
为什么此代码会尝试协商 TLSv1?
生产是该映像的 kubernetes 部署,也有同样的问题,所以这对我的机器或办公室网络来说没有什么特殊之处。
我找到了一种通过猜测使其发挥作用的方法,chatgpt 提供了一些帮助,但我不知道为什么它会起作用。而且很复杂。
执行此操作以确保“纯”openssl 正常工作后:
openssl s_client -connect www.handlingandfulfilment.co.uk:8079
我获得了 v1.2 连接和密码
New, TLSv1.2, Cipher is AES256-GCM-SHA384
这匹配
context.set_ciphers('ECDHE-RSA-AES256-GCM-SHA384:AES256-GCM-SHA384:!aNULL:!MD5')
但是由于证书未知而失败。 certifi 可以解决这个问题。
我明白了,断言是真的
# try a different way
# Create a socket
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect((hostname, 443))
context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
import certifi
context.load_verify_locations(certifi.where())
context.set_ciphers('ECDHE-RSA-AES256-GCM-SHA384:AES256-GCM-SHA384:!aNULL:!MD5')
wrappedSocket = context.wrap_socket(sock, server_hostname=hostname)
assert wrappedSocket
另一端的服务器是微软服务器。我在使用 python 3.11 时找不到任何异常之处,最终他们使用的是主流服务器证书。也许他们有非正统的安全设置。
修复:
第1步: 我这样做是为了测试连接:
openssl s_client -connect www.handlingandfulfilment.co.uk:8079
这有效,输出提到正在使用密码。
新,TLSv1.2,密码为 AES256-GCM-SHA384
我使用chatgpt给了我匹配的正确密码字符串,它做出了回应
context.set_ciphers('ECDHE-RSA-AES256-GCM-SHA384:AES256-GCM-SHA384:!aNULL:!MD5')
这产生了一个涉及证书的新错误,所以我使用了 certifi。
用于建立 zeep 连接的代码看起来是这样的(我可能没有获得所有导入):
import certifi
# debugging aid
from django.core.wsgi import get_wsgi_application
from urllib3 import PoolManager
from zeep import Client
from zeep.transports import Transport
import requests
from requests.adapters import HTTPAdapter, Retry
from urllib3.util.ssl_ import create_urllib3_context
CIPHERS = 'ECDHE-RSA-AES256-GCM-SHA384:AES256-GCM-SHA384:!aNULL:!MD5'
class TLSAdapter(HTTPAdapter):
def init_poolmanager(self, *args, **kwargs):
context = create_urllib3_context(ciphers=CIPHERS)
# Load the certifi certificates
context.load_verify_locations(certifi.where())
context.set_ciphers(CIPHERS)
context.options |= 0x80000 # SSL_OP_NO_TLSv1
context.options |= 0x1000000 # SSL_OP_NO_TLSv1_1
self.poolmanager = PoolManager(*args, ssl_context=context, **kwargs)
def requests_retry_session(
retries=8,
backoff_factor=0.3,
status_forcelist=(500, 502, 503, 504),
session=None,
) -> requests.Session:
session = session or requests.Session()
retry = Retry(
total=retries,
read=retries,
connect=retries,
backoff_factor=backoff_factor,
status_forcelist=status_forcelist,
)
adapter = TLSAdapter(max_retries=retry)
session.mount('http://', adapter)
session.mount('https://', adapter)
return session
class API_error(Exception):
pass
@dataclass(order=True)
class ArkH:
wsdl_url: str
consumerName: str
passCode: str
helixClientName: str
helixUsername: str
userPassword: str
client: Client = field(init=False)
dummyCustomer: str
dummy_customer_mapping = {'CTS':'CTS'} #map a Dear customer, fall back to dummyCustomer
dear_warehouse:str
dear_ship_after_3PL_shipment_date:bool
def __post_init__(self):
session = requests_retry_session()
transport = Transport(session=session,timeout=40,operation_timeout=40)
self.client = Client(self.wsdl_url,transport=transport)