Create SCRAM user directly on Zookeeper for Apache Kafka to use it without AdminClient API

create-scram-user.py

#!/usr/bin/env python3

# This is free and unencumbered software released into the public domain.
# 
# Anyone is free to copy, modify, publish, use, compile, sell, or
# distribute this software, either in source code form or as a compiled
# binary, for any purpose, commercial or non-commercial, and by any
# means.
# 
# In jurisdictions that recognize copyright laws, the author or authors
# of this software dedicate any and all copyright interest in the
# software to the public domain. We make this dedication for the benefit
# of the public at large and to the detriment of our heirs and
# successors. We intend this dedication to be an overt act of
# relinquishment in perpetuity of all present and future rights to this
# software under copyright law.
# 
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
# IN NO EVENT SHALL THE AUTHORS BE LIABLE FOR ANY CLAIM, DAMAGES OR
# OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,
# ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
# OTHER DEALINGS IN THE SOFTWARE.
# 
# For more information, please refer to <https://unlicense.org>

import base64
import logging
import json
from scramp import ScramMechanism
from passlib.hash import scram
from kazoo.client import KazooClient
from secrets import token_urlsafe

logger = '[%(asctime)s] %(levelname)-5s %(message)s'
logging.basicConfig(format=logger,
                    level=logging.INFO,
                    datefmt='%d-%m-%Y %H:%M:%S')
log = logging.getLogger(__name__)

def base64EncodeToStr(input):
    return base64.b64encode(input).decode("ASCII")


def scramhasher(username, password):
    m = ScramMechanism(mechanism='SCRAM-SHA-512')
    hash = scram.using(rounds=4096, algs="sha-1,sha-512").hash(password)
    salt, iteration_count, digest = scram.extract_digest_info(hash, 'sha-512')
    stored_key, server_key = m.make_stored_server_keys(digest)

    zkuserconfig = {
        "version": 1,
        "config": {
            "SCRAM-SHA-512": f"salt={base64EncodeToStr(salt)},stored_key={base64EncodeToStr(stored_key)},server_key={base64EncodeToStr(server_key)},iterations={iteration_count}"
        }
    }

    zkchangeconfig = {
        "version": 2,
        "entity_path": f"users/{username}"
    }

    zkuserencoded = json.dumps(
        zkuserconfig, separators=(',', ':')).encode('ASCII')
    zkchangeconfigencoded = json.dumps(
        zkchangeconfig, separators=(',', ':')).encode('ASCII')

    return zkuserencoded, zkchangeconfigencoded


def zkwriter(username, password, host, ca, clientKey, clientCert):
    zkuserencoded, zkchangeconfigencoded = scramhasher(username, password)
    zk = KazooClient(
        hosts=host,
        ca=ca,
        keyfile=clientKey,
        certfile=clientCert,
        use_ssl=True
    )
    
    zk.start()
    zk.create(f"/config/users/{username}", zkuserencoded)
    zk.create("/config/changes/config_change_",
              zkchangeconfigencoded, sequence=True)
    zk.stop()

requirements.txt

kazoo==2.8.0
passlib==1.7.4
scramp==1.4.1