Source code for cobbler.autoinstallgen

Builds out filesystem trees/data based on the object tree. This is the code behind 'cobbler sync'.

# SPDX-License-Identifier: GPL-2.0-or-later
# SPDX-FileCopyrightText: Copyright 2006-2009, Red Hat, Inc and Others
# SPDX-FileCopyrightText: Michael DeHaan <michael.dehaan AT gmail>

import xml.dom.minidom
from typing import TYPE_CHECKING, Any, List, Optional, Union
from urllib import parse

from cobbler import templar, utils, validate
from cobbler.cexceptions import CX

    from cobbler.api import CobblerAPI
    from cobbler.items.distro import Distro
    from cobbler.items.profile import Profile
    from cobbler.items.system import System

[docs]class AutoInstallationGen: """ Handles conversion of internal state to the tftpboot tree layout """ def __init__(self, api: "CobblerAPI"): """ Constructor :param api: The API instance which is used for this object. Normally there is only one instance of the collection manager. """ self.api = api self.settings = api.settings() self.templar = templar.Templar(self.api)
[docs] def create_autoyast_script( self, document: xml.dom.minidom.Document, script: str, name: str ) -> xml.dom.minidom.Element: """ This method attaches a script with a given name to an existing AutoYaST XML file. :param document: The existing AutoYaST XML file. :param script: The script to attach. :param name: The name of the script. :return: The AutoYaST file with the attached script. """ new_script = document.createElement("script") new_script_source = document.createElement("source") new_script_source_text = document.createCDATASection(script) new_script.appendChild(new_script_source) # type: ignore new_script_file = document.createElement("filename") new_script_file_text = document.createTextNode(name) new_script.appendChild(new_script_file) # type: ignore new_script_source.appendChild(new_script_source_text) # type: ignore new_script_file.appendChild(new_script_file_text) # type: ignore return new_script
[docs] def add_autoyast_script( self, document: xml.dom.minidom.Document, script_type: str, source: str ): """ Add scripts to an existing AutoYaST XML. :param document: The existing AutoYaST XML object. :param script_type: The type of the script which should be added. :param source: The source of the script. This should be ideally a string. """ scripts = document.getElementsByTagName("scripts") # type: ignore if scripts.length == 0: # type: ignore new_scripts = document.createElement("scripts") document.documentElement.appendChild(new_scripts) scripts = document.getElementsByTagName("scripts") # type: ignore added = 0 for stype in scripts[0].childNodes: # type: ignore if stype.nodeType == stype.ELEMENT_NODE and stype.tagName == script_type: # type: ignore stype.appendChild( # type: ignore self.create_autoyast_script( document, source, script_type + "_cobbler" ) ) added = 1 if added == 0: new_chroot_scripts = document.createElement(script_type) new_chroot_scripts.setAttribute("config:type", "list") new_chroot_scripts.appendChild( # type: ignore self.create_autoyast_script(document, source, script_type + "_cobbler") ) scripts[0].appendChild(new_chroot_scripts) # type: ignore
[docs] def generate_autoyast( self, profile: Optional["Profile"] = None, system: Optional["System"] = None, raw_data: Optional[str] = None, ) -> str: """ Generate auto installation information for SUSE distribution (AutoYaST XML file) for a specific system or general profile. Only a system OR profile can be supplied, NOT both. :param profile: The profile to generate the AutoYaST file for. :param system: The system to generate the AutoYaST file for. :param raw_data: The raw data which should be included in the profile. :return: The generated AutoYaST XML file. """ "AutoYaST XML file found. Checkpoint: profile=%s system=%s", profile, system ) what = "none" blend_this = None if profile: what = "profile" blend_this = profile if system: what = "system" blend_this = system if blend_this is None: raise ValueError("Profile or System required for generating autoyast!") blended = utils.blender(self.api, False, blend_this) srv = blended["http_server"] document: xml.dom.minidom.Document = xml.dom.minidom.parseString(raw_data) # type: ignore[unkownMemberType] # Do we already have the #raw comment in the XML? (add_comment = 0 means, don't add #raw comment) add_comment = 1 for node in document.childNodes[1].childNodes: if node.nodeType == node.ELEMENT_NODE and node.tagName == "cobbler": add_comment = 0 break # Add some cobbler information to the XML file, maybe that should be configurable. if add_comment == 1: cobbler_element = document.createElement("cobbler") cobbler_element_system = xml.dom.minidom.Element("system_name") cobbler_element_profile = xml.dom.minidom.Element("profile_name") if system is not None: cobbler_text_system = document.createTextNode( cobbler_element_system.appendChild(cobbler_text_system) # type: ignore if profile is not None: cobbler_text_profile = document.createTextNode( cobbler_element_profile.appendChild(cobbler_text_profile) # type: ignore cobbler_element_server = document.createElement("server") cobbler_text_server = document.createTextNode(blended["http_server"]) cobbler_element_server.appendChild(cobbler_text_server) # type: ignore cobbler_element.appendChild(cobbler_element_server) # type: ignore cobbler_element.appendChild(cobbler_element_system) # type: ignore cobbler_element.appendChild(cobbler_element_profile) # type: ignore name = if self.settings.run_install_triggers: # notify cobblerd when we start/finished the installation protocol = self.api.settings().autoinstall_scheme self.add_autoyast_script( document, "pre-scripts", f'\ncurl "{protocol}://{srv}/cblr/svc/op/trig/mode/pre/{what}/{name}" > /dev/null', ) self.add_autoyast_script( document, "init-scripts", f'\ncurl "{protocol}://{srv}/cblr/svc/op/trig/mode/post/{what}/{name}" > /dev/null', ) return document.toxml() # type: ignore
[docs] def generate_repo_stanza( self, obj: Union["Profile", "System"], is_profile: bool = True ) -> str: """ Automatically attaches yum repos to profiles/systems in automatic installation files (template files) that contain the magic $yum_repo_stanza variable. This includes repo objects as well as the yum repos that are part of split tree installs, whose data is stored with the distro (example: RHEL5 imports) :param obj: The profile or system to generate the repo stanza for. :param is_profile: If True then obj is a profile, otherwise obj has to be a system. Otherwise this method will silently fail. :return: The string with the attached yum repos. """ buf = "" blended = utils.blender(self.api, False, obj) repos = blended["repos"] # keep track of URLs and be sure to not include any duplicates included = {} for repo in repos: # see if this is a source_repo or not; we know that this is a single match due to return_list=False repo_obj = self.api.find_repo(repo, return_list=False) # type: ignore if repo_obj is None or isinstance(repo_obj, list): # FIXME: what to do if we can't find the repo object that is listed? # This should be a warning at another point, probably not here so we'll just not list it so the # automatic installation file will still work as nothing will be here to read the output noise. # Logging might be useful. continue yumopts = "" for opt in repo_obj.yumopts: # filter invalid values to the repo statement in automatic installation files if opt in ["exclude", "include"]: value = repo_obj.yumopts[opt].replace(" ", ",") yumopts = yumopts + f" --{opt}pkgs={value}" elif not opt.lower() in validate.AUTOINSTALL_REPO_BLACKLIST: yumopts += f" {opt}={repo_obj.yumopts[opt]}" if "enabled" not in repo_obj.yumopts or repo_obj.yumopts["enabled"] == "1": if repo_obj.mirror_locally: baseurl = f"http://{blended['http_server']}/cobbler/repo_mirror/{}" if baseurl not in included: buf += f"repo --name={} --baseurl={baseurl}\n" included[baseurl] = 1 else: if repo_obj.mirror not in included: buf += f"repo --name={} --baseurl={repo_obj.mirror} {yumopts}\n" included[repo_obj.mirror] = 1 if is_profile: distro: Optional["Distro"] = obj.get_conceptual_parent() # type: ignore else: profile = obj.get_conceptual_parent() if profile is None: raise ValueError("Error finding distro!") distro = profile.get_conceptual_parent() # type: ignore if distro is None: raise ValueError("Error finding distro!") source_repos = distro.source_repos count = 0 for repo in source_repos: count += 1 if not repo[1] in included: buf += f"repo --name=source-{count} --baseurl={repo[1]}\n" included[repo[1]] = 1 return buf
[docs] def generate_config_stanza( self, obj: Union["Profile", "System"], is_profile: bool = True ) -> str: """ Add in automatic to configure /etc/yum.repos.d on the remote system if the automatic installation file (template file) contains the magic $yum_config_stanza. :param obj: The profile or system to generate a generate a config stanza for. :param is_profile: If the object is a profile. If False it is assumed that the object is a system. :return: The curl command to execute to get the configuration for a system or profile. """ if not self.settings.yum_post_install_mirror: return "" blended = utils.blender(self.api, False, obj) autoinstall_scheme = self.api.settings().autoinstall_scheme if is_profile: url = f"{autoinstall_scheme}://{blended['http_server']}/cblr/svc/op/yum/profile/{}" else: url = f"{autoinstall_scheme}://{blended['http_server']}/cblr/svc/op/yum/system/{}" return f'curl "{url}" --output /etc/yum.repos.d/cobbler-config.repo\n'
[docs] def generate_autoinstall_for_system(self, sys_name: str) -> str: """ Generate an autoinstall config or script for a system. :param sys_name: The system name to generate an autoinstall script for. :return: The generated output or an error message with a human readable description. :raises CX: Raised in case the system references a missing profile. """ system_obj = self.api.find_system(name=sys_name) if system_obj is None or isinstance(system_obj, list): return "# system not found" profile_obj: Optional["Profile"] = system_obj.get_conceptual_parent() # type: ignore if profile_obj is None: raise CX( "system %(system)s references missing profile %(profile)s" % {"system":, "profile": system_obj.profile} ) distro: Optional["Distro"] = profile_obj.get_conceptual_parent() # type: ignore if distro is None: # this is an image parented system, no automatic installation file available return "# image based systems do not have automatic installation files" return self.generate_autoinstall(profile=profile_obj, system=system_obj)
[docs] def generate_autoinstall( self, profile: Optional["Profile"] = None, system: Optional["System"] = None ) -> str: """ This is an internal method for generating an autoinstall config/script. Please use the ``generate_autoinstall_for_*`` methods. If you insist on using this mehtod please only supply a profile or a system, not both. :param profile: The profile to use for generating the autoinstall config/script. :param system: The system to use for generating the autoinstall config/script. If both arguments are given, this wins. :return: The autoinstall script or configuration file as a string. """ obj = None obj_type = "none" if profile: obj = system obj_type = "system" if system is None: obj = profile obj_type = "profile" if obj is None: raise ValueError("Neither profile nor system was given! One is required!") meta = utils.blender(self.api, False, obj) autoinstall_rel_path = meta["autoinstall"] if not autoinstall_rel_path: return f"# automatic installation file value missing or invalid at {obj_type} {}" # get parent distro if system is not None: profile = system.get_conceptual_parent() # type: ignore[reportGeneralTypeIssues] distro: Optional["Distro"] = profile.get_conceptual_parent() # type: ignore[reportGeneralTypeIssues] if distro is None: raise ValueError("Distro for object not found") # make autoinstall_meta metavariable available at top level autoinstall_meta = meta["autoinstall_meta"] del meta["autoinstall_meta"] meta.update(autoinstall_meta) # add package repositories metadata to autoinstall metavariables if distro.breed == "redhat": meta["yum_repo_stanza"] = self.generate_repo_stanza(obj, (system is None)) meta["yum_config_stanza"] = self.generate_config_stanza( obj, (system is None) ) # FIXME: implement something similar to zypper (SUSE based distros) and apt (Debian based distros) meta["kernel_options"] = utils.dict_to_string(meta["kernel_options"]) if "kernel_options_post" in meta: meta["kernel_options_post"] = utils.dict_to_string( meta["kernel_options_post"] ) # add install_source_directory metavariable to autoinstall metavariables if distro is based on Debian if distro.breed in ["debian", "ubuntu"] and "tree" in meta: urlparts = parse.urlsplit(meta["tree"]) # type: ignore meta["install_source_directory"] = urlparts[2] try: autoinstall_path = ( f"{self.settings.autoinstall_templates_dir}/{autoinstall_rel_path}" ) raw_data = utils.read_file_contents(autoinstall_path) if raw_data is None: raise FileNotFoundError("File to template could not be read!") data = self.templar.render(raw_data, meta, None) return data except FileNotFoundError: error_msg = ( f"automatic installation file {meta['autoinstall']} not found" f" at {self.settings.autoinstall_templates_dir}" ) self.api.logger.warning(error_msg) return f"# {error_msg}"
[docs] def generate_autoinstall_for_profile(self, profile: str) -> str: """ Generate an autoinstall config or script for a profile. :param profile: The Profile to generate the script/config for. :return: The generated output or an error message with a human readable description. :raises CX: Raised in case the profile references a missing distro. """ profile_obj: Optional["Profile"] = self.api.find_profile(name=profile) # type: ignore if profile_obj is None or isinstance(profile_obj, list): return "# profile not found" distro: Optional["Distro"] = profile_obj.get_conceptual_parent() # type: ignore if distro is None: raise CX(f'Profile "{}" references missing distro!') return self.generate_autoinstall(profile=profile_obj)
[docs] def get_last_errors(self) -> List[Any]: """ Returns the list of errors generated by the last template render action. :return: The list of error messages which are available. This may not only contain error messages related to generating autoinstallation configuration and scripts. """ return self.templar.last_errors