Using adafruit_minimqtt with default networking

[wz2b]

I am trying to use adafruit_minimqtt with the default circuitpython socket implementation, like this:

pool = socketpool.SocketPool(wifi.radio)
ssl_context = ssl.create_default_context()
ssl_context.check_hostname = False

mqtt_client = MQTT.MQTT(
    broker="192.168.1.2",
    port=8883,
    username="abcdefgh",
    password="abcdefgh",
    socket_pool=pool,
    ssl_context=ssl_context,
    is_ssl=True
)

On connect, my MQTT broker sees this:

1716384906: New connection from 172.28.0.1:49242 on port 8883.
1716384906: OpenSSL Error[0]: error:14037412:SSL routines:ACCEPT_SR_KEY_EXCH:sslv3 alert bad certificate
1716384906: OpenSSL Error[1]: error:140370E5:SSL routines:ACCEPT_SR_KEY_EXCH:ssl handshake failure

I'm not quite sure why. It works okay with a version of umqtt that I modified to work with socketpool. For what I'm doing here I really don't want it checking the certificate chain (I'm not sure how to disable that) but even so, the error message above looks like something else, like it's trying to use certificate (rather than username/password) client authentication.

Am I missing some constructor parameters?

[justmobilize]

As far as I know, there isn't a way to not check the certificate chain. Can you run it off of the non-ssl port instead? check_hostname just flags whether or not to match the peer cert’s hostname in SSLSocket.do_handshake()

[wz2b]

I would definitely prefer to not do that, but it will be my fallback position.

I wonder if there's a way for me to shim 'ssl' to make it skip the check. Actually, I don't even know that verifying the chain is the problem. The error is "sslv3 alert bad certificate" - I'm not sure what certificate it means when that error is coming from the server side. The server is the thing providing the certificate, and the certificate is acceptable to normal paho mqtt clients. It's as if minimqtt is trying to send a certificate to the broker even though I didn't tell it to do so.

[justmobilize]

@wz2b what device are you using, and what version of CP?

And if you remove the ssl_context.check_hostname = False do you get the same error?

[dhalbert]

Are you using a self-signed certificate? If so, are you doing something like this?

pool = adafruit_connection_manager.get_radio_socketpool(wifi.radio)
ssl_context = adafruit_connection_manager.get_radio_ssl_context(wifi.radio)
cert = open("certificate.pem").read()
ssl_context.load_verify_locations(cadata=cert)

[dhalbert]

@wz2b Could you open an issue on https://github.com/adafruit/circuitpython saying what you would like?

[wz2b]

Are you using a self-signed certificate? If so, are you doing something like this?
I'm actually using a legit certificate on the server, signed by incommon. Still, what I'm observing doesn't seem to be that CP doesn't trust the server. It almost seems like it's a subsequent key negotiation problem. I'm digging in but haven't figured it out yet.

I'm doing this on an m5stacks dial, with CircuitPython 9.1.0-beta.2. The dial is really a stamp s3 though.

I will try loading an incommon intermediate/root and see what happens, but my sense is that's not really the problem.

[wz2b]

@wz2b Could you open an issue on https://github.com/adafruit/circuitpython saying what you would like?

Sure. I'm not 100% sure where the problem is, but it seems like it's one of these two projects. I'm leaning toward it being minimqtt rather than circuitpython, because I was able to modify umqtt.simple to work with circuitpython's idea of socketpools and ssl. It doesn't work out of the box because the socket api is different (send vs write, recv vs. recv_into) but with just a few tweaks it connects to the exact same server as above.

import struct
import socketpool
import wifi
import ssl

class MQTTException(Exception):
    pass

class MQTTClient:

    def __init__(self, client_id, server, port=0, user=None, password=None, keepalive=0,
                 ssl=False, ssl_params={}):
        if port == 0:
            port = 8883 if ssl else 1883
        self.client_id = client_id
        self.sock = None
        self.server = server
        self.port = port
        self.ssl = ssl
        self.ssl_params = ssl_params
        self.pid = 0
        self.cb = None
        self.user = user
        self.pswd = password
        self.keepalive = keepalive
        self.lw_topic = None
        self.lw_msg = None
        self.lw_qos = 0
        self.lw_retain = False


    def _read(self, size):
        buf = bytearray(size)
        self.sock.recv_into(buf, size)
        return buf

    def _write(self, data):
        return self.sock.send(data)
        
    def _send_str(self, string):
        self._write(len(string).to_bytes(2, 'big'))
        self._write(string.encode('utf-8'))

    def _recv_len(self):
        n = 0
        sh = 0
        while True:
            b = self.sock._read(1)[0]
            n |= (b & 0x7f) << sh
            if not b & 0x80:
                return n
            sh += 7

    def set_callback(self, f):
        self.cb = f

    def set_last_will(self, topic, msg, retain=False, qos=0):
        assert 0 <= qos <= 2
        assert topic
        self.lw_topic = topic
        self.lw_msg = msg
        self.lw_qos = qos
        self.lw_retain = retain

    def connect(self, clean_session=True):
        pool = socketpool.SocketPool(wifi.radio)
        sock = pool.socket(pool.AF_INET, pool.SOCK_STREAM)
        self.sock = sock # LoggingSocket(sock)

        
        # Resolve address and connect
        addr_info = pool.getaddrinfo(self.server, self.port)
        addr = addr_info[0][-1]
        self.sock.connect(addr)

        if self.ssl:
            context = ssl.create_default_context()
            self.sock = context.wrap_socket(self.sock, server_hostname=self.server, **self.ssl_params)

        premsg = bytearray(b"\x10\0\0\0\0\0")
        msg = bytearray(b"\x04MQTT\x04\x02\0\0")

        sz = 10 + 2 + len(self.client_id)
        msg[6] = clean_session << 1
        if self.user is not None:
            sz += 2 + len(self.user) + 2 + len(self.pswd)
            msg[6] |= 0xC0
        if self.keepalive:
            assert self.keepalive < 65536
            msg[7] |= self.keepalive >> 8
            msg[8] |= self.keepalive & 0x00FF
        if self.lw_topic:
            sz += 2 + len(self.lw_topic) + 2 + len(self.lw_msg)
            msg[6] |= 0x4 | (self.lw_qos & 0x1) << 3 | (self.lw_qos & 0x2) << 3
            msg[6] |= self.lw_retain << 5

        i = 1
        while sz > 0x7f:
            premsg[i] = (sz & 0x7f) | 0x80
            sz >>= 7
            i += 1
        premsg[i] = sz

        self._write(premsg[:i + 2])
        self._write(msg)
        self._send_str(self.client_id)
        if self.lw_topic:
            self._send_str(self.lw_topic)
            self._send_str(self.lw_msg)
        if self.user is not None:
            self._send_str(self.user)
            self._send_str(self.pswd)
        resp = self._read(4)
        assert resp[0] == 0x20 and resp[1] == 0x02
        if resp[3] != 0:
            raise MQTTException(resp[3])
        return resp[2] & 1
    
    def disconnect(self):
        self._write(b"\xe0\0")
        self.sock.close()

    def ping(self):
        self._write(b"\xc0\0")

    def publish(self, topic, msg, retain=False, qos=0):
        pkt = bytearray(b"\x30\0\0\0")
        pkt[0] |= qos << 1 | retain
        sz = 2 + len(topic) + len(msg)
        if qos > 0:
            sz += 2
        assert sz < 2097152
        i = 1
        while sz > 0x7f:
            pkt[i] = (sz & 0x7f) | 0x80
            sz >>= 7
            i += 1
        pkt[i] = sz
        #print(hex(len(pkt)), hexlify(pkt, ":"))
        self._write(pkt[:i + 1])
        self._send_str(topic)
        if qos > 0:
            self.pid += 1
            pid = self.pid
            struct.pack_into("!H", pkt, 0, pid)
            self._write(pkt[:2])
        self._write(msg)
        if qos == 1:
            while True:
                op = self.wait_msg()
                if op == 0x40:
                    sz = self._read(1)
                    assert sz == b"\x02"
                    rcv_pid = self._read(2)
                    rcv_pid = rcv_pid[0] << 8 | rcv_pid[1]
                    if pid == rcv_pid:
                        return
        elif qos == 2:
            assert 0

    def subscribe(self, topic, qos=0):
        assert self.cb is not None, "Subscribe callback is not set"
        pkt = bytearray(b"\x82\0\0\0")
        self.pid += 1
        struct.pack_into("!BH", pkt, 1, 2 + 2 + len(topic) + 1, self.pid)
        #print(hex(len(pkt)), hexlify(pkt, ":"))
        self._write(pkt)
        self._send_str(topic)
        self._write(qos.to_bytes(1, "little"))
        while True:
            op = self.wait_msg()
            if op == 0x90:
                resp = self.sock._read(4)
                #print(resp)
                assert resp[1] == pkt[2] and resp[2] == pkt[3]
                if resp[3] == 0x80:
                    raise MQTTException(resp[3])
                return

    # Wait for a single incoming MQTT message and process it.
    # Subscribed messages are delivered to a callback previously
    # set by .set_callback() method. Other (internal) MQTT
    # messages processed internally.
    def wait_msg(self):
        res = self._read(1)
        self.sock.setblocking(True)
        if res is None:
            return None
        if res == b"":
            raise OSError(-1)
        if res == b"\xd0":  # PINGRESP
            sz = self._read(1)[0]
            assert sz == 0
            return None
        op = res[0]
        if op & 0xf0 != 0x30:
            return op
        sz = self._recv_len()
        topic_len = self._read(2)
        topic_len = (topic_len[0] << 8) | topic_len[1]
        topic = self._read(topic_len)
        sz -= topic_len + 2
        if op & 6:
            pid = self._read(2)
            pid = pid[0] << 8 | pid[1]
            sz -= 2
        msg = self._read(sz)
        self.cb(topic, msg)
        if op & 6 == 2:
            pkt = bytearray(b"\x40\x02\0\0")
            struct.pack_into("!H", pkt, 2, pid)
            self.sock._write(pkt)
        elif op & 6 == 4:
            assert 0

    # Checks whether a pending message from server is available.
    # If not, returns immediately with None. Otherwise, does
    # the same processing as wait_msg.
    def check_msg(self):
        self.sock.setblocking(False)
        return self.wait_msg()

[dhalbert]

@wz2b

My udnerstanding was that you were running a local mosquitto server that is providing https. But did you set up that server with its own self-signed cert? Or do you have some root or root-based intermediate cert that is the server cert?

Re issue: sorry, I meant opening an issue not about the problem but about new networking features you would like, such as suppressing cert authentication.

[justmobilize]

Does it work in CPython? You can pip install this library, and do:

radio = adafruit_connection_manager.CPythonNetwork()
pool = adafruit_connection_manager.get_radio_socketpool(radio)
ssl_context = adafruit_connection_manager.get_radio_ssl_context(radio)
...