Generate JWT for Google IoT Core's Client

generate_jwt_for_gcp_iot_core.py

#!/usr/bin/env python

# fork from https://github.com/Nilhcem/esp32-cloud-iot-core-k8s/blob/master/04-generate-jwt/main.py

from optparse import OptionParser
import datetime
import jwt
import sys

def create_jwt(project_id, private_key_file, algorithm):
    """Creates a JWT (https://jwt.io) to establish an MQTT connection.
        Args:
         project_id: The cloud project ID this device belongs to
         private_key_file: A path to a file containing either an RSA256 or
                 ES256 private key.
         algorithm: The encryption algorithm to use. Either 'RS256' or 'ES256'
        Returns:
            An MQTT generated from the given project_id and private key, which
            expires in 60 minutes. After 60 minutes, your client will be
            disconnected, and a new JWT will have to be generated.
        Raises:
            ValueError: If the private_key_file does not contain a known key.
        """

    token = {
        # The time that the token was issued at
        'iat': datetime.datetime.utcnow(),
        # The time the token expires.
        'exp': datetime.datetime.utcnow() + datetime.timedelta(minutes=60),
        # The audience field should always be set to the GCP project id.
        'aud': project_id
    }

    # Read the private key file.
    with open(private_key_file, 'r') as f:
        private_key = f.read()

    print('Creating JWT using {} from private key file {}'.format(algorithm, private_key_file), file=sys.stderr)

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

if __name__ == "__main__":
    parser = OptionParser()
    parser.add_option("-i", "--project-id", dest="project_id", help="Project ID of GCP")
    parser.add_option("-k", "--key", dest="private_key_file",  help="TLS private key file", metavar="FILE")
    parser.add_option("-a", "--tls-algorithm", dest="tls_algorithm",  help="TLS Algorithm", default="RS256")
    (options, args) = parser.parse_args()

    jwt = create_jwt(options.project_id, options.private_key_file, options.tls_algorithm)
    print(jwt.decode('ascii'), file=sys.stdout, end="")