如何使用自签名 SSL 将 mqtt 与 Kafka 安全连接 [重复]

Question

我正在使用在 Ubuntu 服务器上运行的 mosquitto 代理。我想将 MQTT 与 Kafka 连接起来，所以我使用了这个连接器 https://github.com/evokly/kafka-connect-mqtt/ 。当服务器证书被签名或验证时，它运行良好。但是，我无法使用自签名证书连接到本地服务器。当我以独立模式运行连接器时，配置如下

connector.class=com.evokly.kafka.connect.mqtt.MqttSourceConnector
tasks.max=1
kafka.topic=kafkaSSL
mqtt.client_id=mqttSSLClient-15
mqtt.clean_session=true
mqtt.connection_timeout=30
mqtt.keep_alive_interval=60
mqtt.server_uris=ssl://192.168.0.2:8883
mqtt.topic=mqttSSL
mqtt.ssl.ca_cert=/home/ca.crt
mqtt.ssl.cert=/home/client.crt
mqtt.ssl.key=/home/client.key

它会抛出以下错误：

MqttException (0) - javax.net.ssl.SSLException
        at org.eclipse.paho.client.mqttv3.internal.ExceptionHelper.createMqttException(ExceptionHelper.java:38)
        at org.eclipse.paho.client.mqttv3.internal.ClientComms$ConnectBG.run(ClientComms.java:604)
        at java.lang.Thread.run(Thread.java:748)
Caused by: javax.net.ssl.SSLException
        at sun.security.ssl.Alert.createSSLException(Alert.java:133)
        at sun.security.ssl.TransportContext.fatal(TransportContext.java:324)
        at sun.security.ssl.TransportContext.fatal(TransportContext.java:267)
        at sun.security.ssl.TransportContext.fatal(TransportContext.java:262)
        at sun.security.ssl.SSLSocketImpl.handleException(SSLSocketImpl.java:1554)
        at sun.security.ssl.SSLSocketImpl.startHandshake(SSLSocketImpl.java:441)
        at org.eclipse.paho.client.mqttv3.internal.SSLNetworkModule.start(SSLNetworkModule.java:89)
        at org.eclipse.paho.client.mqttv3.internal.ClientComms$ConnectBG.run(ClientComms.java:590)
Caused by: java.lang.NullPointerException
        at org.bouncycastle.crypto.signers.PSSSigner.generateSignature(Unknown Source)
        at org.bouncycastle.jcajce.provider.asymmetric.rsa.PSSSignatureSpi.engineSign(Unknown Source)
        at java.security.Signature$Delegate.engineSign(Signature.java:1382)
        at java.security.Signature.sign(Signature.java:698)
        at sun.security.ssl.CertificateVerify$T12CertificateVerifyMessage.<init>(CertificateVerify.java:608)
        at sun.security.ssl.CertificateVerify$T12CertificateVerifyProducer.produce(CertificateVerify.java:760)
        at sun.security.ssl.SSLHandshake.produce(SSLHandshake.java:421)
        at sun.security.ssl.ServerHelloDone$ServerHelloDoneConsumer.consume(ServerHelloDone.java:182)
        at sun.security.ssl.SSLHandshake.consume(SSLHandshake.java:377)
        at sun.security.ssl.HandshakeContext.dispatch(HandshakeContext.java:444)
        at sun.security.ssl.HandshakeContext.dispatch(HandshakeContext.java:422)
        at sun.security.ssl.TransportContext.dispatch(TransportContext.java:182)
        at sun.security.ssl.SSLTransport.decode(SSLTransport.java:152)
        at sun.security.ssl.SSLSocketImpl.decode(SSLSocketImpl.java:1383)
        at sun.security.ssl.SSLSocketImpl.readHandshakeRecord(SSLSocketImpl.java:1291)
        at sun.security.ssl.SSLSocketImpl.startHandshake(SSLSocketImpl.java:435)
[2021-11-11 15:34:08,922] ERROR [mqttSSLClient-15] Subscribe failed!  (com.evokly.kafka.connect.mqtt.MqttSourceConnector:132)
Client is not connected (32104)
        at org.eclipse.paho.client.mqttv3.internal.ExceptionHelper.createMqttException(ExceptionHelper.java:31)
        at org.eclipse.paho.client.mqttv3.internal.ClientComms.sendNoWait(ClientComms.java:143)
        at org.eclipse.paho.client.mqttv3.MqttAsyncClient.subscribe(MqttAsyncClient.java:721)
        at org.eclipse.paho.client.mqttv3.MqttClient.subscribe(MqttClient.java:320)
        at org.eclipse.paho.client.mqttv3.MqttClient.subscribe(MqttClient.java:313)
        at com.evokly.kafka.connect.mqtt.MqttSourceTask.start(MqttSourceTask.java:127)
        at org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:224)
        at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:182)
        at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:231)
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
        at java.util.concurrent.FutureTask.run(FutureTask.java:266)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)

MQTT 代理显示：

1636626848: OpenSSL Error[0]: error:14094438:SSL routines:ssl3_read_bytes:tlsv1 alert internal error
1636626848: Client <unknown> disconnected: Protocol error.

我在 Ubuntu 上设置了 apache 网络服务器，可以通过 https://192.168.xxx.xxx 访问服务器 IP

请帮助我如何验证连接器是否可以连接到 8883 端口上的服务器 IP。这是mosquitto的配置文件：

listener 8883
cafile home/ca.crt
certfile home/server.crt
keyfile home/server.key
require certificate true
tls_version tls1.2

我在配置文件中缺少什么？

Answer 1

来自连接器的示例 config 文件：

# CA cert to use to connect to mqtt broker, can be used when connecting to TLS secured brokers - default `null`
#mqtt.ssl.ca_cert=null

您需要向连接器传递一份用于签署代理证书 (ca.crt) 的 CA 证书副本