import base64 import shutil import subprocess from pathlib import Path from typing import Optional, Iterable, Tuple def gen_cloud_config(name: str, cloud_init_path: Path, wireguard_config: str, ssh_authorized_keys: Optional[Iterable[str]] = None ) -> Tuple[Path, Path]: if ssh_authorized_keys is not None: ssh_keys = 'ssh_authorized_keys\n' + \ '\n'.join((f'- {key}' for key in ssh_authorized_keys)) else: ssh_keys = '' user_data = f""" #cloud-config chpasswd: {{ expire: False }} password: root {ssh_keys} ssh_pwauth: False timezone: Europe/Berlin users: - default package_update: true packages: - wireguard write_files: - encoding: b64 content: {base64.b64encode(wireguard_config.encode())} owner: root:root path: /etc/wireguard/wg0.conf permissions: 0600 """ meta_data = f""" local-hostname: {name} """ user_data_path = cloud_init_path / 'user_data' meta_data_path = cloud_init_path / 'meta_data' with open(user_data_path, 'w+') as handle: handle.write(user_data) with open(meta_data_path, 'w+') as handle: handle.write(meta_data) return user_data_path, meta_data_path def _gen_cloudinit_iso_image(path: Path, user_data_path: Path, meta_data_path: Path, volume_label: str = 'cloud_init'): genisoimage_executable = shutil.which('genisoimage') if genisoimage_executable is None: raise FileNotFoundError('could not locate genisoimage executable!') command = ( genisoimage_executable, '-output', str(path), '-V', volume_label, '-r', '-J', user_data_path, meta_data_path ) try: subprocess.run( command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=True ) except subprocess.CalledProcessError as ex: print(f'genisoimage error, stderr was: {ex.stderr}, stdout was: {ex.stdout}') raise ex