Journal Petit exemple de plugin Ansible

Posté par  (site web personnel) . Licence CC By‑SA.
25
7
déc.
2022

Sommaire

Pour ceux qui utilisent Ansible, voici un petit exemple de filtre, que je viens de coder pour des besoins personnels. Je suis entrain d'intégrer un "Web Keys Directory (WKD)" à une solution d'auto-hébergement.

J'avais besoin d'obtenir le hash de l'utilisateur. Par exemple, pour une adresse email andre@homebox.space, le nom d'utilisateur est "andre", et le hash est z1cybqqife1c333kqxqifnz64w9tb3xh.

Petite parenthèse, sur le web keys directory, qui permet de publier votre clé, ou d'obtenir automatiquement les clés GPG de vos contacts, lors de l'envoi d'un email. Par exemple, Thunderbird recherche la clé publique, et vous propose de l'importer pour le chiffrage.

Si vous hébergez votre domaine, vous pouvez publier votre clé publique, en utilisant deux paradigmes possibles, la méthode avancée ou la méthode directe, la seule différence étant l'adresse:

Plus de détails dans les deux excellent journaux plus bas (du même auteur).

Problème: Il n'existe pas de filtre Ansible pour obtenir ce hash, voici donc le code, très simple, sans doute perfectible:

# WKD hash encoder ansibleplugin
# Thanks to
# - https://github.com/artisanofcode/python-zbase32
# - https://www.uriports.com/blog/setting-up-openpgp-web-key-directory/
# for the implementation
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)

from __future__ import (absolute_import, division, print_function)

import collections.abc
import math
import hashlib

__metaclass__ = type

DOCUMENTATION = r'''
  name: wkd_hash
  version_added: "1.0"
  short_description: Return a user ID encoded using GPG WKD
  description:
    - Return a hashed version of a user ID for a GPG Web keys directory
  positional: _input, query
  options:
    _input:
      description: String to encode
      type: str
      required: true
'''

EXAMPLES = r'''
    parts: '{{ "andre" | wkd_hash }}'
    # => "z1cybqqife1c333kqxqifnz64w9tb3xh"
'''

RETURN = r'''
  _value:
    description:
      - Compute a WKD hash from a user ID
    type: str
'''

from ansible.errors import AnsibleFilterError
from ansible.utils import helpers


def _chunks(buffer: bytearray, size: int) -> collections.abc.Generator[bytearray, None, None]:
    """
    chunks.
    :param buffer: the buffer to chunk
    :param size: the size of each chunk
    :return: an iterable of chunks
    """
    for i in range(0, len(buffer), size):
        yield buffer[i : i + size]


def wkd_hash_fn(input_str):

    _ALPHABET = b"ybndrfg8ejkmcpqxot1uwisza345h769"
    _INVERSE_ALPHABET = {key: value for value, key in enumerate(_ALPHABET)}

    # We should use string only on imput
    assert isinstance(input_str, str)

    # Encode using sha1sum
    hashed = hashlib.sha1(input_str.encode("utf-8"))

    # Convert to a byte array
    data = hashed.digest()

    result = bytearray()

    for chunk in _chunks(data, 5):

        buffer = bytearray(5)

        for index, byte in enumerate(chunk):
            buffer[index] = byte

        result.append(_ALPHABET[((buffer[0] & 0xF8) >> 3)])
        result.append(_ALPHABET[((buffer[0] & 0x07) << 2 | (buffer[1] & 0xC0) >> 6)])
        result.append(_ALPHABET[((buffer[1] & 0x3E) >> 1)])
        result.append(_ALPHABET[((buffer[1] & 0x01) << 4 | (buffer[2] & 0xF0) >> 4)])
        result.append(_ALPHABET[((buffer[2] & 0x0F) << 1 | (buffer[3] & 0x80) >> 7)])
        result.append(_ALPHABET[((buffer[3] & 0x7C) >> 2)])
        result.append(_ALPHABET[((buffer[3] & 0x03) << 3 | (buffer[4] & 0xE0) >> 5)])
        result.append(_ALPHABET[(buffer[4] & 0x1F)])

    length = math.ceil(len(data) * 8.0 / 5.0)

    return bytes(result[:length]).decode()


# ---- Ansible filters ----
class FilterModule(object):
    ''' WKD hash filter '''

    def filters(self):
        return {
            'wkd_hash': wkd_hash_fn
        }

on peut maintenant écrire avec Ansible, quelque chose commen ça:

- name: Loop over wkd hashes for the direct method
  set_fact:
    encoded_uid: '{{ key_info.uid | wkdhash }}'

Par défaut, les filtres doivent être stockés au même niveau que les "playbooks", mais je préfère rester maître de ma hiérarchie, donc j'utilise une option dans ansible.cfg:

[defaults]
retry_files_enabled = False
[...]
filter_plugins = {{ playbook_dir }}/../../common/filter-plugins/

Lorsque l'on utilise Ansible, certaines tâches à réaliser sont complexes, et on se retrouve vite à faire de la programmation en YAML… Écrire un filtre en Python permet de conserver la simplicité des tâches, en déplaçant la complexité vers un vrai langage de programmation.

Liens vers d'autres journaux

  • # extensions ansible

    Posté par  . Évalué à 3.

    Merci pour ton retour sur l'écriture d'un filtre pour ansible. C'est une force d'ansible, d'être simplement extensible par du code Python. Vas-tu faire une pull request vers les dépôts officiels ? J'avais eu la chance de voir mon module pam_limits accepté, il y a bien longtemps.

  • # no future

    Posté par  . Évalué à 1.

    from __future__ import (absolute_import, division, print_function)
    

    future is now, Python 2 is dead !

Suivre le flux des commentaires

Note : les commentaires appartiennent à celles et ceux qui les ont postés. Nous n’en sommes pas responsables.