Test EMQX v5.5 JWT authentication and authorization (ACL)

connect-with-jwt.sh

#!/bin/bash -e

#password='eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJ1c2VybmFtZSI6InB1YjEiLCJhY2wiOlt7InBlcm1pc3Npb24iOiJhbGxvdyIsImFjdGlvbiI6ImFsbCIsInRvcGljIjoiZXEgYS8jIn1dfQ.skzXjZOPU9jQLGqwvF2wh-Tr0oNN1GBc55NbSylHaQ0'
password="$(python3 ./jwt-gen-token.py)"
json="$(echo $password | cut -d '.' -f 2)"
echo "${json}" | base64 -d 2>/dev/null || true
echo

mqttx sub -t 'a/#' -h localhost -p 1883 -i aaa -P "$password" -u pub1

jwt-auth-cluster.hocon

# in single node test, add 'include jwt-auth-cluster.hocon' to emqx.conf
authentication = [
  {
    algorithm = hmac-based
    from = password
    mechanism = jwt
    secret = ppp
    secret_base64_encoded = false
    use_jwks = false
    verify_claims {}
  }
]
authorization {
  cache {
    enable = true
    excludes = []
    max_size = 32
    ttl = 1m
  }
  deny_action = disconnect
  no_match = deny
  sources = []
}

jwt-gen-token.py

#!/usr/bin/env python3

import jwt
from datetime import datetime, timedelta

# Your secret key
secret = "ppp"

# Current time
now = datetime.utcnow()

# JWT Payload with dynamic expiration
payload = {
    "username": "pub1",
    "acl": [
        {"permission": "allow", "action": "sub", "topic": "eq a/#"},
        {"permission": "allow", "action": "pub", "topic": "a/1"}
    ],
    # Set expiration to a specific time in the future (e.g., 1 minute from now)
    "exp": now + timedelta(minutes=1)
}

# Generate JWT
encoded_jwt = jwt.encode(payload, secret, algorithm="HS256")

decoded_jwt = encoded_jwt.decode('utf-8')

print(f"{decoded_jwt}")