Python

To authenticate to the akenza MQTT broker using device credentials, a device must send a JSON Web Token (JWT, RFC 7519).

Each JWT is composed of three components: a header, a payload (containing a claim set), and a signature.

The following python modules need to be installed in order for the scripts to work.

• paho-mqtt for sending mqtt messages

• pycryptodome for reading public key PEM files and creating the fingerprint

Using an RSA Private Key

import paho.mqtt.publish as publish
import jwt
import datetime
import ssl
import hashlib
from Crypto.PublicKey import RSA

device_id = "<deviceId>"
topic = f"/up/device/id/{device_id}"
host = "mqtt.akenza.io"
port = 8883
payload = '{"temperature":22}'

public_key_file_path = "./rsa_public.pem"
private_key_file_path = "./rsa_private.pem"


def create_jwt(device_id, private_key_file, public_key_file, algorithm="RS256"):
    token = {
        "iat": datetime.datetime.now(tz=datetime.timezone.utc),
        "exp": datetime.datetime.now(tz=datetime.timezone.utc) + datetime.timedelta(minutes=20),
        "aud": f"https://akenza.io/devices/{device_id}"
    }
    headers = {
        "kid": get_public_key_fingerprint(public_key_file)
    }
    with open(private_key_file, "r") as f:
        private_key = f.read()

    return jwt.encode(token, private_key, algorithm=algorithm, headers=headers)


def get_public_key_fingerprint(public_key_file):
    with open(public_key_file, "r") as f:
        public_key = RSA.import_key(f.read())

    return hashlib.sha256(public_key.export_key(format="DER")).hexdigest()


tls_config = {
    "cert_reqs": ssl.CERT_REQUIRED,
    "tls_version": ssl.PROTOCOL_TLSv1_2
}

mqtt_password = create_jwt(device_id, private_key_file_path, public_key_file_path)
auth = {
    "username": "unused",
    "password": mqtt_password
}

publish.single(topic, payload, hostname=host, port=port, tls=tls_config, auth=auth, client_id=device_id)

Using an EC Private Key

import paho.mqtt.publish as publish
import jwt
import datetime
import ssl
import hashlib
from Crypto.PublicKey import ECC

device_id = "<deviceId>"
topic = f"/up/device/id/{device_id}"
host = "mqtt.akenza.io"
port = 8883
payload = '{"temperature":22}'

public_key_file_path = "./ec_public.pem"
private_key_file_path = "./ec_private.pem"


def create_jwt(device_id, private_key_file, public_key_file, algorithm="ES256"):
    token = {
        "iat": datetime.datetime.now(tz=datetime.timezone.utc),
        "exp": datetime.datetime.now(tz=datetime.timezone.utc) + datetime.timedelta(minutes=20),
        "aud": f"https://akenza.io/devices/{device_id}"
    }
    headers = {
        "kid": get_public_key_fingerprint(public_key_file)
    }
    with open(private_key_file, "r") as f:
        private_key = f.read()

    return jwt.encode(token, private_key, algorithm=algorithm, headers=headers)


def get_public_key_fingerprint(public_key_file):
    with open(public_key_file, "r") as f:
        public_key = ECC.import_key(f.read())

    return hashlib.sha256(public_key.export_key(format="DER")).hexdigest()


tls_config = {
    "cert_reqs": ssl.CERT_REQUIRED,
    "tls_version": ssl.PROTOCOL_TLSv1_2
}

mqtt_password = create_jwt(device_id, private_key_file_path, public_key_file_path)
auth = {
    "username": "unused",
    "password": mqtt_password
}

publish.single(topic, payload, hostname=host, port=port, tls=tls_config, auth=auth, client_id=device_id)