rut-test-ops/global/overlay/etc/puppet/setup_cosmos_modules

191 lines
6.2 KiB
Python
Executable file

#!/usr/bin/env python3
"""Write out a puppet cosmos-modules.conf"""
import hashlib
import os
import os.path
import socket
import sys
try:
from configobj import ConfigObj
os_info = ConfigObj("/etc/os-release")
except (IOError, ModuleNotFoundError):
os_info = None
try:
fqdn = socket.getfqdn()
hostname = socket.gethostname()
except OSError:
host_info = None
else:
domainname = ".".join([x for x in fqdn.split(".")[:1]])
hostname = fqdn.split(".")[0]
instance, location, environment, function, number = hostname.split("-")
service = fqdn.split(".")[1]
host_info = {
"domainname": domainname,
"environment": environment,
"fqdn": fqdn,
"function": function,
"hostname": hostname,
"instance": instance,
"location": location,
"number": number,
}
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 = {
"apparmor": {
"repo": "https://github.com/SUNET/puppet-apparmor.git",
"upgrade": "yes",
"tag": "sunet-2*",
},
"bastion": {
"repo": "https://github.com/SUNET/puppet-bastion.git",
"upgrade": "yes",
"tag": "sunet-2*",
},
"nagioscfg": {
"repo": "https://github.com/SUNET/puppet-nagioscfg.git",
"upgrade": "yes",
"tag": "sunet-2*",
},
"sunet": {
"repo": "https://github.com/SUNET/puppet-sunet.git",
"upgrade": "yes",
"tag": "stable-2*",
},
"ufw": {
"repo": "https://github.com/SUNET/puppet-module-ufw.git",
"upgrade": "yes",
"tag": "sunet-2*",
},
}
# When/if we want we can do stuff to modules here
if host_info:
if host_info["fqdn"] == "internal-sto4-test-monitor-1.rut.sunet.se":
modules["sunet"]["tag"] = "testing*"
if "internal-sto4-test-k8sc" in host_info["fqdn"]:
modules["sunet"]["tag"] = "testing*"
if "internal-sto4-test-k8sw" in host_info["fqdn"]:
modules["sunet"]["tag"] = "testing*"
# if host_info["fqdn"] == "internal-sto1-test-forgejo-2.platform.sunet.se":
# modules["sunet"]["tag"] = "testing-2*"
# if host_info["fqdn"] == "internal-sto3-test-monitor-1.platform.sunet.se":
# modules["sunet"]["tag"] = "jocar-monitor-handle-legacy-scriptherder-2024-02-14-v10"
# if host_info["fqdn"] == "internal-sto1-test-actrunner-1.platform.sunet.se":
# modules["sunet"]["tag"] = "testing-2*"
## modules["nagioscfg"]["tag"] = "*mifr-2*"
# # modules["stdlib"] = {
# # "tag": "sunet-2*",
# # "upgrade": "yes",
# "repo": "https://github.com/SUNET/puppetlabs-stdlib.git",
# }
# modules["apt"] = {
# "tag": "sunet-2*",
# "upgrade": "yes",
# "repo": "https://github.com/SUNET/puppetlabs-apt.git",
# }
# modules["concat"] = {
# "tag": "sunet-2*",
# "upgrade": "yes",
# "repo": "https://github.com/SUNET/puppetlabs-concat.git",
# }
# if host_info:
# if host_info["environment"] == "test":
# modules["sunet"]["tag"] = "kano-mail-2*"
# if os_info:
# if os_info["ID"] == "debian" and os_info["VERSION_ID"] == "12":
# modules["augeas"]["repo"] = "https://github.com/SUNET/puppetlabs-augeas_core"
# 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()