akenza.io
Search…
⌃K

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)