filters.py

"""
Custom filters for Ansible.

Can be used by saving it (for example) as ~/.ansible/plugins/filter/filters.py

"""

def depem(string):
  import re
  return re.sub(r'\s+|(-----(BEGIN|END).*-----)', '', string)

def pubkey2hash(string, algo='sha256'):
    from base64 import b64encode as b64encode
    from base64 import b64decode as b64decode
    from binascii import unhexlify as hex2bin
    import hashlib
    hash = hashlib.new(algo, b64decode(string))
    return b64encode(hex2bin(hash.hexdigest())).decode("utf-8")

def sshkey_is_valid(string):
    from sshpubkeys import SSHKey
    ssh = SSHKey(string)
    try:
        ssh.parse()
    except Exception as e:
        return False
    return True

# https://help.ubuntu.com/community/Grub2/Passwords
# https://github.com/rhboot/grub2/blob/fedora-39/util/grub-mkpasswd-pbkdf2.c
def grub_mkpasswd_pbkdf2(
        passwd,
        rounds=10000,
        salt=None,
        digest='sha512'
        ):
    from hashlib import pbkdf2_hmac
    from binascii import hexlify
    import secrets

    if salt is None:
        salt = secrets.token_bytes(64)
    else:
        salt = salt.encode()

    dk = pbkdf2_hmac(digest,
                     bytes(passwd, 'utf-8'),
                     salt,
                     rounds
                     )
    return (
            "grub.pbkdf2." + digest + "." + str(rounds) + "." +
            hexlify(salt).decode().upper() + "." +
            hexlify(dk).decode().upper()
            )

def mosquitto_passwd(passwd):
    from hashlib import pbkdf2_hmac
    from base64 import b64encode
    import secrets

    # See https://github.com/eclipse/mosquitto/blob/master/src/password_mosq.h
    iterations = 101

    salt = secrets.token_bytes(12)
    dk = pbkdf2_hmac('sha512',
                     bytes(passwd, 'utf-8'),
                     salt,
                     iterations
                     )
    return (
            "$7$" + str(iterations) + "$" +
            b64encode(salt).decode() + "$" +
            b64encode(dk).decode()
            )

class FilterModule(object):
    def filters(self):
        return {
            'depem': depem,
            'sshkey_is_valid': sshkey_is_valid,
            'pubkey2hash': pubkey2hash,
            'mosquitto_passwd': mosquitto_passwd,
            'grub_mkpasswd_pbkdf2': grub_mkpasswd_pbkdf2
        }