具有服务器端证书问题的Python Paho-mqtt客户端

Question

我的 python paho-mqtt client无法连接到用 java 编写的代理。代理已使用 jks 类型证书启用 SSL 连接。经纪人不在我的管理范围内。

我将 jks 证书转换为 pem 证书以在我的 python 代码中使用。但是当我运行代码时，出现错误:

Traceback (most recent call last):
  File "test.py", line 55, in <module>
    client.connect("192.168.110.2", 56785, 60)
  File "C:\Python\Python37\lib\site-packages\paho\mqtt\client.py", line 760, in
connect
    return self.reconnect()
  File "C:\Python\Python37\lib\site-packages\paho\mqtt\client.py", line 919, in
reconnect
    sock.do_handshake()
  File "C:\Python\Python37\lib\ssl.py", line 1117, in do_handshake
    self._sslobj.do_handshake()
ssl.SSLCertVerificationError: [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: CA signature digest algorithm too weak (_ssl.c:1056)

所以我认为证书有问题。如何绕过CA签名摘要算法检查？

我使用以下命令将 jks 证书转换为 pem 证书:

keytool -importkeystore -srckeystore server.jks -destkeystore server.p12 -srcstoretype jks -deststoretype pkcs12
openssl pkcs12 -in server.p12 -out server.pem

这是我的完整代码:

# -*- coding:utf-8 -*-

import json
import ssl
import time

import paho.mqtt.client as mqtt

# constants
token = 'token '
mqtt_username = 'name'
mqtt_passwd = 'pass'

test_payload = {"type": "a_type","data": "my data","tokens": [token]}


def on_connect(client, userdata, flags, rc):
    print("Connected with result code " + str(rc))
    if rc == 0:
        # subscribe
        client.subscribe("Client/%s/Biz/Down" % token, 1)
        time.sleep(3)
        client.publish('Client/%s/Biz/Up' % token,
                       json.dumps(test_payload))
    # time.sleep(5)
    else:
        client.disconnect()


def on_message(client, userdata, msg):
    print(msg.topic + " " + str(msg.payload))
    if ("Client/%s/Biz/Down" % token) == msg.topic:
        client.disconnect()


client = mqtt.Client('', True, None, mqtt.MQTTv31)
client.username_pw_set(mqtt_username, mqtt_passwd)
client.on_connect = on_connect
client.on_message = on_message
client.tls_set('./server.pem')

client.connect("192.168.110.2", 56785, 60)

client.loop_forever()

Answer 1

我想通了。在客户端，您不需要配置服务器的自签名证书。现在成功了!

# -*- coding:utf-8 -*-

import json
import ssl
import time

import paho.mqtt.client as mqtt

# constants
token = 'token '
mqtt_username = 'name'
mqtt_passwd = 'pass'

test_payload = {"type": "a_type","data": "my data","tokens": [token]}


def on_connect(client, userdata, flags, rc):
    print("Connected with result code " + str(rc))
    if rc == 0:
        # subscribe
        client.subscribe("Client/%s/Biz/Down" % token, 1)
        time.sleep(3)
        client.publish('Client/%s/Biz/Up' % token,
                       json.dumps(test_payload))
    # time.sleep(5)
    else:
        client.disconnect()


def on_message(client, userdata, msg):
    print(msg.topic + " " + str(msg.payload))
    if ("Client/%s/Biz/Down" % token) == msg.topic:
        client.disconnect()


client = mqtt.Client('', True, None, mqtt.MQTTv31)
client.username_pw_set(mqtt_username, mqtt_passwd)
client.on_connect = on_connect
client.on_message = on_message

# the key steps here
context = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
# if you do not want to check the cert hostname, skip it
# context.check_hostname = False
client.tls_set_context(context)

client.connect("192.168.110.2", 56785, 60)

client.loop_forever()