Using JSON Web Tokens (JWTs)

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
1
import paho.mqtt.publish as publish
2
import jwt
3
import datetime
4
import ssl
5
import hashlib
6
from Crypto.PublicKey import RSA
7
​
8
device_id = "<deviceId>"
9
topic = f"/up/device/id/{device_id}"
10
host = "mqtt.akenza.io"
11
port = 8883
12
payload = '{"temperature":22}'
13
​
14
public_key_file_path = "./rsa_public.pem"
15
private_key_file_path = "./rsa_private.pem"
16
​
17
​
18
def create_jwt(device_id, private_key_file, public_key_file, algorithm="RS256"):
19
    token = {
20
        "iat": datetime.datetime.now(tz=datetime.timezone.utc),
21
        "exp": datetime.datetime.now(tz=datetime.timezone.utc) + datetime.timedelta(minutes=20),
22
        "aud": f"https://akenza.io/devices/{device_id}"
23
    }
24
    headers = {
25
        "kid": get_public_key_fingerprint(public_key_file)
26
    }
27
    with open(private_key_file, "r") as f:
28
        private_key = f.read()
29
​
30
    return jwt.encode(token, private_key, algorithm=algorithm, headers=headers)
31
​
32
​
33
def get_public_key_fingerprint(public_key_file):
34
    with open(public_key_file, "r") as f:
35
        public_key = RSA.import_key(f.read())
36
​
37
    return hashlib.sha256(public_key.export_key(format="DER")).hexdigest()
38
​
39
​
40
tls_config = {
41
    "cert_reqs": ssl.CERT_REQUIRED,
42
    "tls_version": ssl.PROTOCOL_TLSv1_2
43
}
44
​
45
mqtt_password = create_jwt(device_id, private_key_file_path, public_key_file_path)
46
auth = {
47
    "username": "unused",
48
    "password": mqtt_password
49
}
50
​
51
publish.single(topic, payload, hostname=host, port=port, tls=tls_config, auth=auth, client_id=device_id)
52
​
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)
​
Last modified 1mo ago