streams-ops/docs/setup_cosmos_modules.example

#!/usr/bin/env python3
""" Write out a puppet cosmos-modules.conf """

import hashlib
import os
import os.path
import sys

try:
    from configobj import ConfigObj

    OS_INFO = ConfigObj("/etc/os-release")
except (IOError, ModuleNotFoundError):
    OS_INFO = None


def get_file_hash(modulesfile):
    """
    Based on https://github.com/python/cpython/pull/31930: should use
    hashlib.file_digest() but it is only available in python 3.11
    """
    try:
        with open(modulesfile, "rb") as fileobj:
            digestobj = hashlib.sha256()
            _bufsize = 2**18
            buf = bytearray(_bufsize)  # Reusable buffer to reduce allocations.
            view = memoryview(buf)
            while True:
                size = fileobj.readinto(buf)
                if size == 0:
                    break  # EOF
                digestobj.update(view[:size])
    except FileNotFoundError:
        return ""

    return digestobj.hexdigest()


def get_list_hash(file_lines):
    """Get hash of list contents"""

    file_lines_hash = hashlib.sha256()
    for line in file_lines:
        file_lines_hash.update(line)

    return file_lines_hash.hexdigest()


def create_file_content(modules):
    """
    Write out the expected file contents to a list so we can check the
    expected checksum before writing anything
    """
    file_lines = []
    file_lines.append(
        "# Generated by {}\n".format(  # pylint: disable=consider-using-f-string
            os.path.basename(sys.argv[0])
        ).encode("utf-8")
    )
    for key in modules:
        file_lines.append(
            "{0:11} {1} {2} {3}\n".format(  # pylint: disable=consider-using-f-string
                key,
                modules[key]["repo"],
                modules[key]["upgrade"],
                modules[key]["tag"],
            ).encode("utf-8")
        )

    return file_lines


def main():
    """Starting point of the program"""

    modulesfile: str = "/etc/puppet/cosmos-modules.conf"
    modulesfile_tmp: str = modulesfile + ".tmp"

    modules: dict = {
        "concat": {
            "repo": "https://github.com/SUNET/puppetlabs-concat.git",
            "upgrade": "yes",
            "tag": "sunet-2*",
        },
        "stdlib": {
            "repo": "https://github.com/SUNET/puppetlabs-stdlib.git",
            "upgrade": "yes",
            "tag": "sunet-2*",
        },
        "cosmos": {
            "repo": "https://github.com/SUNET/puppet-cosmos.git",
            "upgrade": "yes",
            "tag": "sunet-2*",
        },
        "ufw": {
            "repo": "https://github.com/SUNET/puppet-module-ufw.git",
            "upgrade": "yes",
            "tag": "sunet-2*",
        },
        "apt": {
            "repo": "https://github.com/SUNET/puppetlabs-apt.git",
            "upgrade": "yes",
            "tag": "sunet-2*",
        },
        "vcsrepo": {
            "repo": "https://github.com/SUNET/puppetlabs-vcsrepo.git",
            "upgrade": "yes",
            "tag": "sunet-2*",
        },
        "xinetd": {
            "repo": "https://github.com/SUNET/puppetlabs-xinetd.git",
            "upgrade": "yes",
            "tag": "sunet-2*",
        },
        "python": {
            "repo": "https://github.com/SUNET/puppet-python.git",
            "upgrade": "yes",
            "tag": "sunet-2*",
        },
        "hiera-gpg": {
            "repo": "https://github.com/SUNET/hiera-gpg.git",
            "upgrade": "yes",
            "tag": "sunet-2*",
        },
        "pound": {
            "repo": "https://github.com/SUNET/puppet-pound.git",
            "upgrade": "yes",
            "tag": "sunet-2*",
        },
        "augeas": {
            "repo": "https://github.com/SUNET/puppet-augeas.git",
            "upgrade": "yes",
            "tag": "sunet-2*",
        },
        "bastion": {
            "repo": "https://github.com/SUNET/puppet-bastion.git",
            "upgrade": "yes",
            "tag": "sunet-2*",
        },
        "pyff": {
            "repo": "https://github.com/samlbits/puppet-pyff.git",
            "upgrade": "yes",
            "tag": "puppet-pyff-*",
        },
        "dhcp": {
            "repo": "https://github.com/SUNET/puppetlabs-dhcp.git",
            "upgrade": "yes",
            "tag": "sunet_dev-2*",
        },
        "varnish": {
            "repo": "https://github.com/samlbits/puppet-varnish.git",
            "upgrade": "yes",
            "tag": "puppet-varnish-*",
        },
        "apparmor": {
            "repo": "https://github.com/SUNET/puppet-apparmor.git",
            "upgrade": "yes",
            "tag": "sunet-2*",
        },
        "docker": {
            "repo": "https://github.com/SUNET/garethr-docker.git",
            "upgrade": "yes",
            "tag": "sunet-2*",
        },
        "network": {
            "repo": "https://github.com/SUNET/attachmentgenie-network.git",
            "upgrade": "yes",
            "tag": "sunet-2*",
        },
        "sunet": {
            "repo": "https://github.com/SUNET/puppet-sunet.git",
            "upgrade": "yes",
            "tag": "sunet-2*",
        },
        "sysctl": {
            "repo": "https://github.com/SUNET/puppet-sysctl.git",
            "upgrade": "yes",
            "tag": "sunet-2*",
        },
        "nagioscfg": {
            "repo": "https://github.com/SUNET/puppet-nagioscfg.git",
            "upgrade": "yes",
            "tag": "sunet-2*",
        },
    }

    # When/if we want we can do stuff to modules here
    if OS_INFO:
        if OS_INFO["VERSION_CODENAME"] == "bullseye":
            pass

    # Build list of expected file content
    file_lines = create_file_content(modules)

    # Get hash of the list
    list_hash = get_list_hash(file_lines)

    # Get hash of the existing file on disk
    file_hash = get_file_hash(modulesfile)

    # Update the file if necessary
    if list_hash != file_hash:
        # Since we are reading the file with 'rb' when computing our hash use 'wb' when
        # writing so we dont end up creating a file that does not match the
        # expected hash
        with open(modulesfile_tmp, "wb") as fileobj:
            for line in file_lines:
                fileobj.write(line)

        # Rename it in place so the update is atomic for anything else trying to
        # read the file
        os.rename(modulesfile_tmp, modulesfile)


if __name__ == "__main__":
    main()