akenza.io
WebsiteAPI DocumentationWhat's newLogin
  • Overview
  • Getting Started
    • Connect a Device
  • Changelog
  • General
    • Applications
    • Organization context
    • Workspace Context
    • Users
      • User Roles
  • Device Management
    • Managing an organization
      • API keys
      • Users (Organization)
        • Add & removing users from your organization
    • Managing a workspace
      • General Settings
        • Device Lifecycle Reports
        • Workspace properties
        • Occupancy settings
        • Device Setup Assistant
      • Custom fields
      • Tags
      • Users (Workspace)
    • Devices
      • Device
      • Device Simulator
      • Bulk actions
      • Bulk import CSV templates
    • Rules
      • Input
      • Logic blocks
        • Comparison
        • Custom Logic
          • Logic Block Scripting
      • Timed Rules
      • Rule Actions
        • Downlink
    • Data Flows
      • Device Connectors
        • Device Security
          • Using Device Credentials
            • Creating Public/Private Key Pairs
            • Using JSON Web Tokens (JWTs)
              • Python
              • Java
        • HTTP
        • MQTT
        • CoAP
        • LoRaWAN
          • Connectivity Management
          • Swisscom LoRaWAN
          • The Things Network
          • Loriot
          • Actility’s ThingPark Wireless
          • EWZ
          • Cibicom
          • Helium
          • ChirpStack
        • NB-IoT
        • mioty
        • Disruptive Technologies
        • VergeSense
        • Spaceti
        • Haltian
      • Device Types
        • Custom Device Types
          • Uplink
          • Downlink
          • Scripting
        • Library
      • Output Connectors
        • Databases
          • akenza DB
          • InfluxDB
          • SQL Databases
        • Streaming
          • Webhook
          • Azure IoT Hub
          • AWS Kinesis
          • Google Cloud Pub/Sub
          • Apache Kafka
        • Notifications
          • E-Mail
          • SMS
          • Microsoft Teams
          • Slack
    • Custom Components
    • Integrations
    • Reference
      • REST API
        • Filtering
        • Querying Device Data
      • WebSocket API
      • Scripting
        • Stateful Operations
        • Utility Functions
      • Payload Templating
  • Reference
  • Dashboard Builder
    • Managing Dashboards
      • Embedding dashboards
    • Components
      • Map
      • Floorplan
  • Device Setup Assistant
    • Device Setup Assistant - Overview
  • Tutorials
    • BI Tools
      • Grafana Data Source Plugin
      • How to build a dashboard with Retool
      • How to analyze data with AWS QuickSight
    • Devices
      • How to integrate the XDK device from Legic via MQTT on akenza
      • How to connect the Disruptive Technologies-cloud on akenza
      • How to send Downlinks to the Kuando Busylight device
      • How to integrate an Arduino device via MQTT on akenza
      • Integrate a MClimate Vicki LoRaWAN Radiator Thermostat on akenza
      • How to integrate an ERS Elsys device with Loriot on akenza
      • How to integrate the IAM Decentlab device with TTN on akenza
      • How to integrate the Seeed SenseCAP T1000 tracker on akenza
      • How to integrate a Swisscom Multisense device on akenza
    • Notifications
      • How to send SMS notifications
      • How to send notifications to Slack
      • How to send notifications to Microsoft Teams
    • Enterprise solutions
      • How to send data to Azure IoT Hub
      • How to send data to the Google Cloud Pub/Sub
      • How to send data to InfluxDB
      • How to send data to AWS Kinesis
      • How to send data to Azure Event Hubs with Apache Kafka
    • IoT Starter Kits
      • How to integrate the IAQ Kit with Actility on akenza
      • How to integrate the CoWork Kit with Actility on akenza
      • How to integrate the Smart Building Kit with Actility on akenza
      • How to integrate the Pepperl+Fuchs Kit with Actility on akenza
  • Support Center
    • FAQ
    • Status Page
    • Service Desk
    • Request a feature
  • Deprecated
    • SIM-Cards
    • Everynet
    • Sigfox
    • How to connect the Yanzi Lifecycle cloud on akenza
Powered by GitBook
On this page
  • Using an RSA Private Key
  • Using an EC Private Key

Was this helpful?

  1. Device Management
  2. Data Flows
  3. Device Connectors
  4. Device Security
  5. Using Device Credentials
  6. Using JSON Web Tokens (JWTs)

Python

PreviousUsing JSON Web Tokens (JWTs)NextJava

Last updated 1 month ago

Was this helpful?

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

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.

  • for sending mqtt messages

  • for reading public key PEM files and creating the fingerprint

  • s a Python library which allows you to encode and decode JSON Web Tokens (JWT).

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)
RFC 7519
paho-mqtt
pycryptodome
PyJWT