Source code for cobbler.cli

"""
Command line interface for Cobbler.
"""

# 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 optparse
import os
import sys
import time
import traceback
import xmlrpc.client
from typing import Optional

from cobbler import enums, power_manager, utils
from cobbler.utils import signatures

INVALID_TASK = "<<invalid>>"

OBJECT_ACTIONS_MAP = {
    "distro": ["add", "copy", "edit", "find", "list", "remove", "rename", "report"],
    "profile": [
        "add",
        "copy",
        "dumpvars",
        "edit",
        "find",
        "get-autoinstall",
        "list",
        "remove",
        "rename",
        "report",
    ],
    "system": [
        "add",
        "copy",
        "dumpvars",
        "edit",
        "find",
        "get-autoinstall",
        "list",
        "remove",
        "rename",
        "report",
        "poweron",
        "poweroff",
        "powerstatus",
        "reboot",
    ],
    "image": ["add", "copy", "edit", "find", "list", "remove", "rename", "report"],
    "repo": [
        "add",
        "copy",
        "edit",
        "find",
        "list",
        "remove",
        "rename",
        "report",
        "autoadd",
    ],
    "menu": ["add", "copy", "edit", "find", "list", "remove", "rename", "report"],
    "setting": ["edit", "report"],
    "signature": ["reload", "report", "update"],
}

OBJECT_TYPES = list(OBJECT_ACTIONS_MAP.keys())
# would like to use from_iterable here, but have to support python 2.4
OBJECT_ACTIONS = []
for actions in list(OBJECT_ACTIONS_MAP.values()):
    OBJECT_ACTIONS += actions
DIRECT_ACTIONS = [
    "aclsetup",
    "buildiso",
    "import",
    "list",
    "replicate",
    "report",
    "reposync",
    "sync",
    "validate-autoinstalls",
    "version",
    "signature",
    "hardlink",
    "mkloaders",
]

####################################################

# the fields has controls what data elements are part of each object.  To add a new field, just add a new
# entry to the list following some conventions to be described later.  You must also add a method called
# set_$fieldname.  Do not write a method called get_$fieldname, that will not be called.
#
# name | default | subobject default | display name | editable? | tooltip | values ? | type
#
# name -- what the filed should be called.   For the command line, underscores will be replaced with
#         a hyphen programatically, so use underscores to seperate things that are seperate words
#
# default value -- when a new object is created, what is the default value for this field?
#
# subobject default -- this applies ONLY to subprofiles, and is most always set to <<inherit>>.  If this
#                      is not item_profile.py it does not matter.
#
# display name -- how the field shows up in the web application and the "cobbler report" command
#
# editable -- should the field be editable in the CLI and web app?  Almost always yes unless
#                it is an internalism.  Fields that are not editable are "hidden"
#
# tooltip -- the caption to be shown in the web app or in "commandname --help" in the CLI
#
# values -- for fields that have a limited set of valid options and those options are always fixed
#           (such as architecture type), the list of valid options goes in this field.
#
# type -- the type of the field.  Used to determine which HTML form widget is used in the web interface
#
#
# the order in which the fields appear in the web application (for all non-hidden
# fields) is defined in field_ui_info.py. The CLI sorts fields alphabetically.
#
# field_ui_info.py also contains a set of "Groups" that describe what other fields
# are associated with what other fields.  This affects color coding and other
# display hints.  If you add a field, please edit field_ui_info.py carefully to match.
#
# additional:  see field_ui_info.py for some display hints.  By default, in the
# web app, all fields are text fields unless field_ui_info.py lists the field in
# one of those dictionaries.
#
# hidden fields should not be added without just cause, explanations about these are:
#
#   ctime, mtime -- times the object was modified, used internally by Cobbler for API purposes
#   uid -- also used for some external API purposes
#   source_repos -- an artifiact of import, this is too complicated to explain on IRC so we just hide it for RHEL split
#                   repos, this is a list of each of them in the install tree, used to generate repo lines in the
#                   automatic installation file to allow installation of x>=RHEL5. Otherwise unimportant.
#   depth -- used for "cobbler list" to print the tree, makes it easier to load objects from disk also
#   tree_build_time -- loaded from import, this is not useful to many folks so we just hide it.  Avail over API.
#
# so to add new fields
#   (A) understand the above
#   (B) add a field below
#   (C) add a set_fieldname method
#   (D) if field must be viewable/editable via web UI, add a entry in
#       corresponding *_UI_FIELDS_MAPPING dictionary in field_ui_info.py.
#       If field must not be displayed in a text field in web UI, also add
#       an entry in corresponding USES_* list in field_ui_info.py.
#
# in general the set_field_name method should raise exceptions on invalid fields, always.   There are adtl
# validation fields in is_valid to check to see that two seperate fields do not conflict, but in general
# design issues that require this should be avoided forever more, and there are few exceptions.  Cobbler
# must operate as normal with the default value for all fields and not choke on the default values.

DISTRO_FIELDS = [
    # non-editable in UI (internal)
    ["ctime", 0, 0, "", False, "", 0, "float"],
    ["depth", 0, 0, "Depth", False, "", 0, "int"],
    ["mtime", 0, 0, "", False, "", 0, "float"],
    ["source_repos", [], 0, "Source Repos", False, "", 0, "list"],
    ["tree_build_time", 0, 0, "Tree Build Time", False, "", 0, "str"],
    ["uid", "", 0, "", False, "", 0, "str"],
    # editable in UI
    [
        "arch",
        "x86_64",
        0,
        "Architecture",
        True,
        "",
        signatures.get_valid_archs(),
        "str",
    ],
    [
        "autoinstall_meta",
        {},
        0,
        "Automatic Installation Template Metadata",
        True,
        "Ex: dog=fang agent=86",
        0,
        "dict",
    ],
    [
        "boot_loaders",
        "<<inherit>>",
        "<<inherit>>",
        "Boot loaders",
        True,
        "Network installation boot loaders",
        0,
        "list",
    ],
    [
        "breed",
        "redhat",
        0,
        "Breed",
        True,
        "What is the type of distribution?",
        signatures.get_valid_breeds(),
        "str",
    ],
    ["comment", "", 0, "Comment", True, "Free form text description", 0, "str"],
    [
        "initrd",
        None,
        0,
        "Initrd",
        True,
        "Absolute path to kernel on filesystem",
        0,
        "str",
    ],
    [
        "kernel",
        None,
        0,
        "Kernel",
        True,
        "Absolute path to kernel on filesystem",
        0,
        "str",
    ],
    [
        "remote_boot_initrd",
        None,
        0,
        "Remote Boot Initrd",
        True,
        "URL the bootloader directly retrieves and boots from",
        0,
        "str",
    ],
    [
        "remote_boot_kernel",
        None,
        0,
        "Remote Boot Kernel",
        True,
        "URL the bootloader directly retrieves and boots from",
        0,
        "str",
    ],
    [
        "kernel_options",
        {},
        0,
        "Kernel Options",
        True,
        "Ex: selinux=permissive",
        0,
        "dict",
    ],
    [
        "kernel_options_post",
        {},
        0,
        "Kernel Options (Post Install)",
        True,
        "Ex: clocksource=pit noapic",
        0,
        "dict",
    ],
    ["name", "", 0, "Name", True, "Ex: Fedora-11-i386", 0, "str"],
    [
        "os_version",
        "virtio26",
        0,
        "OS Version",
        True,
        "Needed for some virtualization optimizations",
        signatures.get_valid_os_versions(),
        "str",
    ],
    [
        "owners",
        "SETTINGS:default_ownership",
        0,
        "Owners",
        True,
        "Owners list for authz_ownership (space delimited)",
        0,
        "list",
    ],
    [
        "redhat_management_key",
        "",
        "",
        "Redhat Management Key",
        True,
        "Registration key for RHN, Spacewalk, or Satellite",
        0,
        "str",
    ],
    [
        "template_files",
        {},
        0,
        "Template Files",
        True,
        "File mappings for built-in config management",
        0,
        "dict",
    ],
]

IMAGE_FIELDS = [
    # non-editable in UI (internal)
    ["ctime", 0, 0, "", False, "", 0, "float"],
    ["depth", 0, 0, "", False, "", 0, "int"],
    ["mtime", 0, 0, "", False, "", 0, "float"],
    ["parent", "", 0, "", False, "", 0, "str"],
    ["uid", "", 0, "", False, "", 0, "str"],
    # editable in UI
    [
        "arch",
        "x86_64",
        0,
        "Architecture",
        True,
        "",
        signatures.get_valid_archs(),
        "str",
    ],
    [
        "autoinstall",
        "",
        0,
        "Automatic installation file",
        True,
        "Path to autoinst/answer file template",
        0,
        "str",
    ],
    ["breed", "redhat", 0, "Breed", True, "", signatures.get_valid_breeds(), "str"],
    ["comment", "", 0, "Comment", True, "Free form text description", 0, "str"],
    [
        "file",
        "",
        0,
        "File",
        True,
        "Path to local file or nfs://user@host:path",
        0,
        "str",
    ],
    [
        "image_type",
        "iso",
        0,
        "Image Type",
        True,
        "",
        ["iso", "direct", "memdisk", "virt-image"],
        "str",
    ],
    ["name", "", 0, "Name", True, "", 0, "str"],
    ["network_count", 1, 0, "Virt NICs", True, "", 0, "int"],
    [
        "os_version",
        "",
        0,
        "OS Version",
        True,
        "ex: rhel4",
        signatures.get_valid_os_versions(),
        "str",
    ],
    [
        "owners",
        "SETTINGS:default_ownership",
        0,
        "Owners",
        True,
        "Owners list for authz_ownership (space delimited)",
        [],
        "list",
    ],
    ["menu", "", "", "Parent boot menu", True, "", [], "str"],
    ["display_name", "", "", "Display Name", True, "Ex: My Image", [], "str"],
    [
        "boot_loaders",
        "<<inherit>>",
        "<<inherit>>",
        "Boot loaders",
        True,
        "Network installation boot loaders",
        0,
        "list",
    ],
    [
        "virt_auto_boot",
        "SETTINGS:virt_auto_boot",
        0,
        "Virt Auto Boot",
        True,
        "Auto boot this VM?",
        0,
        "bool",
    ],
    [
        "virt_bridge",
        "SETTINGS:default_virt_bridge",
        0,
        "Virt Bridge",
        True,
        "",
        0,
        "str",
    ],
    ["virt_cpus", 1, 0, "Virt CPUs", True, "", 0, "int"],
    [
        "virt_disk_driver",
        "SETTINGS:default_virt_disk_driver",
        0,
        "Virt Disk Driver Type",
        True,
        "The on-disk format for the virtualization disk",
        "raw",
        "str",
    ],
    [
        "virt_file_size",
        "SETTINGS:default_virt_file_size",
        0.0,
        "Virt File Size (GB)",
        True,
        "",
        0.0,
        "float",
    ],
    ["virt_path", "", 0, "Virt Path", True, "Ex: /directory or VolGroup00", 0, "str"],
    ["virt_ram", "SETTINGS:default_virt_ram", 0, "Virt RAM (MB)", True, "", 0, "int"],
    [
        "virt_type",
        "SETTINGS:default_virt_type",
        0,
        "Virt Type",
        True,
        "",
        [e.value for e in enums.VirtType],
        "str",
    ],
]

MENU_FIELDS = [
    # non-editable in UI (internal)
    ["ctime", 0, 0, "", False, "", 0, "float"],
    ["depth", 1, 1, "", False, "", 0, "int"],
    ["mtime", 0, 0, "", False, "", 0, "int"],
    ["uid", "", "", "", False, "", 0, "str"],
    # editable in UI
    ["comment", "", "", "Comment", True, "Free form text description", 0, "str"],
    ["name", "", None, "Name", True, "Ex: Systems", 0, "str"],
    ["display_name", "", "", "Display Name", True, "Ex: Systems menu", [], "str"],
    ["parent", "", "", "Parent Menu", True, "", [], "str"],
]

PROFILE_FIELDS = [
    # non-editable in UI (internal)
    ["ctime", 0, 0, "", False, "", 0, "float"],
    ["depth", 1, 1, "", False, "", 0, "int"],
    ["mtime", 0, 0, "", False, "", 0, "int"],
    ["uid", "", "", "", False, "", 0, "str"],
    # editable in UI
    [
        "autoinstall",
        "SETTINGS:autoinstall",
        "<<inherit>>",
        "Automatic Installation Template",
        True,
        "Path to automatic installation template",
        0,
        "str",
    ],
    [
        "autoinstall_meta",
        {},
        "<<inherit>>",
        "Automatic Installation Metadata",
        True,
        "Ex: dog=fang agent=86",
        0,
        "dict",
    ],
    [
        "boot_loaders",
        "<<inherit>>",
        "<<inherit>>",
        "Boot loaders",
        True,
        "Linux installation boot loaders",
        0,
        "list",
    ],
    ["comment", "", "", "Comment", True, "Free form text description", 0, "str"],
    [
        "dhcp_tag",
        "default",
        "<<inherit>>",
        "DHCP Tag",
        True,
        "See manpage or leave blank",
        0,
        "str",
    ],
    [
        "distro",
        None,
        "<<inherit>>",
        "Distribution",
        True,
        "Parent distribution",
        [],
        "str",
    ],
    [
        "enable_ipxe",
        "SETTINGS:enable_ipxe",
        0,
        "Enable iPXE?",
        True,
        "Use iPXE instead of PXELINUX for advanced booting options",
        0,
        "bool",
    ],
    [
        "enable_menu",
        "SETTINGS:enable_menu",
        "<<inherit>>",
        "Enable PXE Menu?",
        True,
        "Show this profile in the PXE menu?",
        0,
        "bool",
    ],
    [
        "kernel_options",
        {},
        "<<inherit>>",
        "Kernel Options",
        True,
        "Ex: selinux=permissive",
        0,
        "dict",
    ],
    [
        "kernel_options_post",
        {},
        "<<inherit>>",
        "Kernel Options (Post Install)",
        True,
        "Ex: clocksource=pit noapic",
        0,
        "dict",
    ],
    ["name", "", None, "Name", True, "Ex: F10-i386-webserver", 0, "str"],
    [
        "name_servers",
        "SETTINGS:default_name_servers",
        [],
        "Name Servers",
        True,
        "space delimited",
        0,
        "list",
    ],
    [
        "name_servers_search",
        "SETTINGS:default_name_servers_search",
        [],
        "Name Servers Search Path",
        True,
        "space delimited",
        0,
        "list",
    ],
    [
        "next_server_v4",
        "<<inherit>>",
        "<<inherit>>",
        "Next Server (IPv4) Override",
        True,
        "See manpage or leave blank",
        0,
        "str",
    ],
    [
        "next_server_v6",
        "<<inherit>>",
        "<<inherit>>",
        "Next Server (IPv6) Override",
        True,
        "See manpage or leave blank",
        0,
        "str",
    ],
    [
        "filename",
        "<<inherit>>",
        "<<inherit>>",
        "DHCP Filename Override",
        True,
        "Use to boot non-default bootloaders",
        0,
        "str",
    ],
    [
        "owners",
        "SETTINGS:default_ownership",
        "SETTINGS:default_ownership",
        "Owners",
        True,
        "Owners list for authz_ownership (space delimited)",
        0,
        "list",
    ],
    ["parent", "", "", "Parent Profile", True, "", [], "str"],
    [
        "proxy",
        "SETTINGS:proxy_url_int",
        "<<inherit>>",
        "Proxy",
        True,
        "Proxy URL",
        0,
        "str",
    ],
    [
        "redhat_management_key",
        "<<inherit>>",
        "<<inherit>>",
        "Red Hat Management Key",
        True,
        "Registration key for RHN, Spacewalk, or Satellite",
        0,
        "str",
    ],
    [
        "repos",
        [],
        "<<inherit>>",
        "Repos",
        True,
        "Repos to auto-assign to this profile",
        [],
        "list",
    ],
    [
        "server",
        "<<inherit>>",
        "<<inherit>>",
        "Server Override",
        True,
        "See manpage or leave blank",
        0,
        "str",
    ],
    [
        "template_files",
        {},
        "<<inherit>>",
        "Template Files",
        True,
        "File mappings for built-in config management",
        0,
        "dict",
    ],
    ["menu", "", "", "Parent boot menu", True, "", 0, "str"],
    ["display_name", "", "", "Display Name", True, "Ex: My Profile", [], "str"],
    [
        "virt_auto_boot",
        "SETTINGS:virt_auto_boot",
        "<<inherit>>",
        "Virt Auto Boot",
        True,
        "Auto boot this VM?",
        0,
        "bool",
    ],
    [
        "virt_bridge",
        "SETTINGS:default_virt_bridge",
        "<<inherit>>",
        "Virt Bridge",
        True,
        "",
        0,
        "str",
    ],
    ["virt_cpus", 1, "<<inherit>>", "Virt CPUs", True, "integer", 0, "int"],
    [
        "virt_disk_driver",
        "SETTINGS:default_virt_disk_driver",
        "<<inherit>>",
        "Virt Disk Driver Type",
        True,
        "The on-disk format for the virtualization disk",
        [e.value for e in enums.VirtDiskDrivers],
        "str",
    ],
    [
        "virt_file_size",
        "SETTINGS:default_virt_file_size",
        "<<inherit>>",
        "Virt File Size(GB)",
        True,
        "",
        0.0,
        "int",
    ],
    [
        "virt_path",
        "",
        "<<inherit>>",
        "Virt Path",
        True,
        "Ex: /directory OR VolGroup00",
        0,
        "str",
    ],
    [
        "virt_ram",
        "SETTINGS:default_virt_ram",
        "<<inherit>>",
        "Virt RAM (MB)",
        True,
        "",
        0,
        "int",
    ],
    [
        "virt_type",
        "SETTINGS:default_virt_type",
        "<<inherit>>",
        "Virt Type",
        True,
        "Virtualization technology to use",
        [e.value for e in enums.VirtType],
        "str",
    ],
]

REPO_FIELDS = [
    # non-editable in UI (internal)
    ["ctime", 0, 0, "", False, "", 0, "float"],
    ["depth", 2, 0, "", False, "", 0, "float"],
    ["mtime", 0, 0, "", False, "", 0, "float"],
    ["parent", None, 0, "", False, "", 0, "str"],
    ["uid", None, 0, "", False, "", 0, "str"],
    # editable in UI
    [
        "apt_components",
        "",
        0,
        "Apt Components (apt only)",
        True,
        "ex: main restricted universe",
        [],
        "list",
    ],
    [
        "apt_dists",
        "",
        0,
        "Apt Dist Names (apt only)",
        True,
        "ex: precise precise-updates",
        [],
        "list",
    ],
    [
        "arch",
        "x86_64",
        0,
        "Arch",
        True,
        "ex: i386, x86_64",
        [e.value for e in enums.RepoArchs],
        "str",
    ],
    [
        "breed",
        "rsync",
        0,
        "Breed",
        True,
        "",
        [e.value for e in enums.RepoBreeds],
        "str",
    ],
    ["comment", "", 0, "Comment", True, "Free form text description", 0, "str"],
    [
        "createrepo_flags",
        "<<inherit>>",
        0,
        "Createrepo Flags",
        True,
        "Flags to use with createrepo",
        0,
        "dict",
    ],
    [
        "environment",
        {},
        0,
        "Environment Variables",
        True,
        "Use these environment variables during commands (key=value, space delimited)",
        0,
        "dict",
    ],
    [
        "keep_updated",
        True,
        0,
        "Keep Updated",
        True,
        "Update this repo on next 'cobbler reposync'?",
        0,
        "bool",
    ],
    [
        "mirror",
        None,
        0,
        "Mirror",
        True,
        "Address of yum or rsync repo to mirror",
        0,
        "str",
    ],
    [
        "mirror_type",
        "baseurl",
        0,
        "Mirror Type",
        True,
        "",
        [e.value for e in enums.MirrorType],
        "str",
    ],
    [
        "mirror_locally",
        True,
        0,
        "Mirror locally",
        True,
        "Copy files or just reference the repo externally?",
        0,
        "bool",
    ],
    ["name", "", 0, "Name", True, "Ex: f10-i386-updates", 0, "str"],
    [
        "owners",
        "SETTINGS:default_ownership",
        0,
        "Owners",
        True,
        "Owners list for authz_ownership (space delimited)",
        [],
        "list",
    ],
    [
        "priority",
        99,
        0,
        "Priority",
        True,
        "Value for yum priorities plugin, if installed",
        0,
        "int",
    ],
    [
        "proxy",
        "SETTINGS:proxy_url_ext",
        "<<inherit>>",
        "Proxy information",
        True,
        "http://example.com:8080, or <<inherit>> to use proxy_url_ext from settings, blank or <<None>> for no proxy",
        0,
        "str",
    ],
    [
        "rpm_list",
        [],
        0,
        "RPM List",
        True,
        "Mirror just these RPMs (yum only)",
        0,
        "list",
    ],
    [
        "yumopts",
        {},
        0,
        "Yum Options",
        True,
        "Options to write to yum config file",
        0,
        "dict",
    ],
    [
        "rsyncopts",
        "",
        0,
        "Rsync Options",
        True,
        "Options to use with rsync repo",
        0,
        "dict",
    ],
]

SYSTEM_FIELDS = [
    # non-editable in UI (internal)
    ["ctime", 0, 0, "", False, "", 0, "float"],
    ["depth", 2, 0, "", False, "", 0, "int"],
    ["ipv6_autoconfiguration", False, 0, "IPv6 Autoconfiguration", True, "", 0, "bool"],
    ["mtime", 0, 0, "", False, "", 0, "float"],
    [
        "repos_enabled",
        False,
        0,
        "Repos Enabled",
        True,
        "(re)configure local repos on this machine at next config update?",
        0,
        "bool",
    ],
    ["uid", "", 0, "", False, "", 0, "str"],
    # editable in UI
    [
        "autoinstall",
        "<<inherit>>",
        0,
        "Automatic Installation Template",
        True,
        "Path to automatic installation template",
        0,
        "str",
    ],
    [
        "autoinstall_meta",
        {},
        0,
        "Automatic Installation Template Metadata",
        True,
        "Ex: dog=fang agent=86",
        0,
        "dict",
    ],
    [
        "boot_loaders",
        "<<inherit>>",
        "<<inherit>>",
        "Boot loaders",
        True,
        "Linux installation boot loaders",
        0,
        "list",
    ],
    ["comment", "", 0, "Comment", True, "Free form text description", 0, "str"],
    [
        "enable_ipxe",
        "<<inherit>>",
        0,
        "Enable iPXE?",
        True,
        "Use iPXE instead of PXELINUX for advanced booting options",
        0,
        "bool",
    ],
    ["gateway", "", 0, "Gateway", True, "", 0, "str"],
    ["hostname", "", 0, "Hostname", True, "", 0, "str"],
    ["image", None, 0, "Image", True, "Parent image (if not a profile)", 0, "str"],
    ["ipv6_default_device", "", 0, "IPv6 Default Device", True, "", 0, "str"],
    [
        "kernel_options",
        {},
        0,
        "Kernel Options",
        True,
        "Ex: selinux=permissive",
        0,
        "dict",
    ],
    [
        "kernel_options_post",
        {},
        0,
        "Kernel Options (Post Install)",
        True,
        "Ex: clocksource=pit noapic",
        0,
        "dict",
    ],
    ["name", "", 0, "Name", True, "Ex: vanhalen.example.org", 0, "str"],
    ["name_servers", [], 0, "Name Servers", True, "space delimited", 0, "list"],
    [
        "name_servers_search",
        [],
        0,
        "Name Servers Search Path",
        True,
        "space delimited",
        0,
        "list",
    ],
    [
        "netboot_enabled",
        True,
        0,
        "Netboot Enabled",
        True,
        "PXE (re)install this machine at next boot?",
        0,
        "bool",
    ],
    [
        "next_server_v4",
        "<<inherit>>",
        0,
        "Next Server (IPv4) Override",
        True,
        "See manpage or leave blank",
        0,
        "str",
    ],
    [
        "next_server_v6",
        "<<inherit>>",
        0,
        "Next Server (IPv6) Override",
        True,
        "See manpage or leave blank",
        0,
        "str",
    ],
    [
        "filename",
        "<<inherit>>",
        "<<inherit>>",
        "DHCP Filename Override",
        True,
        "Use to boot non-default bootloaders",
        0,
        "str",
    ],
    [
        "owners",
        "<<inherit>>",
        0,
        "Owners",
        True,
        "Owners list for authz_ownership (space delimited)",
        0,
        "list",
    ],
    [
        "power_address",
        "",
        0,
        "Power Management Address",
        True,
        "Ex: power-device.example.org",
        0,
        "str",
    ],
    [
        "power_id",
        "",
        0,
        "Power Management ID",
        True,
        "Usually a plug number or blade name, if power type requires it",
        0,
        "str",
    ],
    ["power_pass", "", 0, "Power Management Password", True, "", 0, "str"],
    [
        "power_type",
        "SETTINGS:power_management_default_type",
        0,
        "Power Management Type",
        True,
        "Power management script to use",
        power_manager.get_power_types(),
        "str",
    ],
    ["power_user", "", 0, "Power Management Username", True, "", 0, "str"],
    [
        "power_options",
        "",
        0,
        "Power Management Options",
        True,
        "Additional options, to be passed to the fencing agent",
        0,
        "str",
    ],
    [
        "power_identity_file",
        "",
        0,
        "Power Identity File",
        True,
        "Identity file to be passed to the fencing agent (ssh key)",
        0,
        "str",
    ],
    ["profile", None, 0, "Profile", True, "Parent profile", [], "str"],
    ["proxy", "<<inherit>>", 0, "Internal Proxy", True, "Internal proxy URL", 0, "str"],
    [
        "redhat_management_key",
        "<<inherit>>",
        0,
        "Redhat Management Key",
        True,
        "Registration key for RHN, Spacewalk, or Satellite",
        0,
        "str",
    ],
    [
        "server",
        "<<inherit>>",
        0,
        "Server Override",
        True,
        "See manpage or leave blank",
        0,
        "str",
    ],
    [
        "status",
        "production",
        0,
        "Status",
        True,
        "System status",
        ["", "development", "testing", "acceptance", "production"],
        "str",
    ],
    [
        "template_files",
        {},
        0,
        "Template Files",
        True,
        "File mappings for built-in configuration management",
        0,
        "dict",
    ],
    [
        "virt_auto_boot",
        "<<inherit>>",
        0,
        "Virt Auto Boot",
        True,
        "Auto boot this VM?",
        0,
        "bool",
    ],
    ["virt_cpus", "<<inherit>>", 0, "Virt CPUs", True, "", 0, "int"],
    [
        "virt_disk_driver",
        "<<inherit>>",
        0,
        "Virt Disk Driver Type",
        True,
        "The on-disk format for the virtualization disk",
        [e.value for e in enums.VirtDiskDrivers],
        "str",
    ],
    [
        "virt_file_size",
        "<<inherit>>",
        0.0,
        "Virt File Size(GB)",
        True,
        "",
        0.0,
        "float",
    ],
    [
        "virt_path",
        "<<inherit>>",
        0,
        "Virt Path",
        True,
        "Ex: /directory or VolGroup00",
        0,
        "str",
    ],
    [
        "virt_pxe_boot",
        0,
        0,
        "Virt PXE Boot",
        True,
        "Use PXE to build this VM?",
        0,
        "bool",
    ],
    ["virt_ram", "<<inherit>>", 0, "Virt RAM (MB)", True, "", 0, "int"],
    [
        "virt_type",
        "<<inherit>>",
        0,
        "Virt Type",
        True,
        "Virtualization technology to use",
        [e.value for e in enums.VirtType],
        "str",
    ],
    ["serial_device", "", 0, "Serial Device #", True, "Serial Device Number", 0, "int"],
    [
        "serial_baud_rate",
        "",
        0,
        "Serial Baud Rate",
        True,
        "Serial Baud Rate",
        ["", "2400", "4800", "9600", "19200", "38400", "57600", "115200"],
        "int",
    ],
    ["display_name", "", "", "Display Name", True, "Ex: My System", [], "str"],
]

# network interface fields are in a separate list because a system may contain
# several network interfaces and thus several values for each one of those fields
# (1-N cardinality), while it may contain only one value for other fields
# (1-1 cardinality). This difference requires special handling.
NETWORK_INTERFACE_FIELDS = [
    [
        "bonding_opts",
        "",
        0,
        "Bonding Opts",
        True,
        "Should be used with --interface",
        0,
        "str",
    ],
    [
        "bridge_opts",
        "",
        0,
        "Bridge Opts",
        True,
        "Should be used with --interface",
        0,
        "str",
    ],
    [
        "cnames",
        [],
        0,
        "CNAMES",
        True,
        "Cannonical Name Records, should be used with --interface, In quotes, space delimited",
        0,
        "list",
    ],
    [
        "connected_mode",
        False,
        0,
        "InfiniBand Connected Mode",
        True,
        "Should be used with --interface",
        0,
        "bool",
    ],
    ["dhcp_tag", "", 0, "DHCP Tag", True, "Should be used with --interface", 0, "str"],
    ["dns_name", "", 0, "DNS Name", True, "Should be used with --interface", 0, "str"],
    [
        "if_gateway",
        "",
        0,
        "Per-Interface Gateway",
        True,
        "Should be used with --interface",
        0,
        "str",
    ],
    [
        "interface_master",
        "",
        0,
        "Master Interface",
        True,
        "Should be used with --interface",
        0,
        "str",
    ],
    [
        "interface_type",
        "na",
        0,
        "Interface Type",
        True,
        "Should be used with --interface",
        [
            "na",
            "bond",
            "bond_slave",
            "bridge",
            "bridge_slave",
            "bonded_bridge_slave",
            "bmc",
            "infiniband",
        ],
        "str",
    ],
    [
        "ip_address",
        "",
        0,
        "IP Address",
        True,
        "Should be used with --interface",
        0,
        "str",
    ],
    [
        "ipv6_address",
        "",
        0,
        "IPv6 Address",
        True,
        "Should be used with --interface",
        0,
        "str",
    ],
    [
        "ipv6_default_gateway",
        "",
        0,
        "IPv6 Default Gateway",
        True,
        "Should be used with --interface",
        0,
        "str",
    ],
    ["ipv6_mtu", "", 0, "IPv6 MTU", True, "Should be used with --interface", 0, "str"],
    [
        "ipv6_prefix",
        "",
        0,
        "IPv6 Prefix",
        True,
        "Should be used with --interface",
        0,
        "str",
    ],
    [
        "ipv6_secondaries",
        [],
        0,
        "IPv6 Secondaries",
        True,
        "Space delimited. Should be used with --interface",
        0,
        "list",
    ],
    [
        "ipv6_static_routes",
        [],
        0,
        "IPv6 Static Routes",
        True,
        "Should be used with --interface",
        0,
        "list",
    ],
    [
        "mac_address",
        "",
        0,
        "MAC Address",
        True,
        '(Place "random" in this field for a random MAC Address.)',
        0,
        "str",
    ],
    [
        "management",
        False,
        0,
        "Management Interface",
        True,
        "Is this the management interface? Should be used with --interface",
        0,
        "bool",
    ],
    ["mtu", "", 0, "MTU", True, "", 0, "str"],
    [
        "netmask",
        "",
        0,
        "Subnet Mask",
        True,
        "Should be used with --interface",
        0,
        "str",
    ],
    [
        "static",
        False,
        0,
        "Static",
        True,
        "Is this interface static? Should be used with --interface",
        0,
        "bool",
    ],
    [
        "static_routes",
        [],
        0,
        "Static Routes",
        True,
        "Should be used with --interface",
        0,
        "list",
    ],
    [
        "virt_bridge",
        "",
        0,
        "Virt Bridge",
        True,
        "Should be used with --interface",
        0,
        "str",
    ],
]

SETTINGS_FIELDS = [
    ["name", "", "", "Name", True, "Ex: server", 0, "str"],
    ["value", "", "", "Value", True, "Ex: 127.0.0.1", 0, "str"],
]


####################################################


[docs]def to_string_from_fields(item_dict, fields, interface_fields=None) -> str: """ item_dict is a dictionary, fields is something like item_distro.FIELDS :param item_dict: The dictionary representation of a Cobbler item. :param fields: This is the list of fields a Cobbler item has. :param interface_fields: This is the list of fields from a network interface of a system. This is optional. :return: The string representation of a Cobbler item with all its values. """ buf = "" keys = [] for elem in fields: keys.append((elem[0], elem[3], elem[4])) keys.sort() buf += f"{'Name':<30} : {item_dict['name']}\n" for (k, nicename, editable) in keys: # FIXME: supress fields users don't need to see? # FIXME: interfaces should be sorted # FIXME: print ctime, mtime nicely if not editable: continue if k != "name": # FIXME: move examples one field over, use description here. buf += f"{nicename:<30} : {item_dict[k]}\n" # somewhat brain-melting special handling to print the dicts # inside of the interfaces more neatly. if "interfaces" in item_dict and interface_fields is not None: keys = [] for elem in interface_fields: keys.append((elem[0], elem[3], elem[4])) keys.sort() for iname in list(item_dict["interfaces"].keys()): # FIXME: inames possibly not sorted buf += f"{'Interface ===== ':<30} : {iname}\n" for (k, nicename, editable) in keys: if editable: buf += f"{nicename:<30} : {item_dict['interfaces'][iname].get(k, '')}\n" return buf
[docs]def report_items(remote, otype: str): """ Return all items for a given collection. :param remote: The remote to use as the query-source. The remote to use as the query-source. :param otype: The object type to query. """ if otype == "setting": items = remote.get_settings() keys = list(items.keys()) keys.sort() for key in keys: item = {"name": key, "value": items[key]} report_item(remote, otype, item=item) elif otype == "signature": items = remote.get_signatures() total_breeds = 0 total_sigs = 0 if "breeds" in items: print("Currently loaded signatures:") bkeys = list(items["breeds"].keys()) bkeys.sort() total_breeds = len(bkeys) for breed in bkeys: total_sigs += report_single_breed(breed, items) print( f"\n{total_breeds:d} breeds with {total_sigs:d} total signatures loaded" ) else: print("No breeds found in the signature, a signature update is recommended") return 1 else: items = remote.get_items(otype) for item in items: report_item(remote, otype, item=item)
[docs]def report_single_breed(name: str, items: dict) -> int: """ Helper function which prints a single signature breed list to the terminal. """ new_sigs = 0 print(f"{name}:") oskeys = list(items["breeds"][name].keys()) oskeys.sort() if len(oskeys) > 0: new_sigs = len(oskeys) for osversion in oskeys: print(f"\t{osversion}") else: print("\t(none)") return new_sigs
[docs]def report_item(remote, otype: str, item=None, name=None): """ Return a single item in a given collection. Either this is an item object or this method searches for a name. :param remote: The remote to use as the query-source. :param otype: The object type to query. :param item: The item to display :param name: The name to search for and display. """ if item is None: if otype == "setting": cur_settings = remote.get_settings() try: item = {"name": name, "value": cur_settings[name]} except Exception: print(f"Setting not found: {name}") return 1 elif otype == "signature": items = remote.get_signatures() total_sigs = 0 if "breeds" in items: print("Currently loaded signatures:") if name in items["breeds"]: total_sigs += report_single_breed(name, items) print(f"\nBreed '{name}' has {total_sigs:d} total signatures") else: print(f"No breed named '{name}' found") return 1 else: print( "No breeds found in the signature, a signature update is recommended" ) return 1 return else: item = remote.get_item(otype, name) if item == "~": print(f"No {otype} found: {name}") return 1 if otype == "distro": data = to_string_from_fields(item, DISTRO_FIELDS) elif otype == "profile": data = to_string_from_fields(item, PROFILE_FIELDS) elif otype == "system": data = to_string_from_fields(item, SYSTEM_FIELDS, NETWORK_INTERFACE_FIELDS) elif otype == "repo": data = to_string_from_fields(item, REPO_FIELDS) elif otype == "image": data = to_string_from_fields(item, IMAGE_FIELDS) elif otype == "menu": data = to_string_from_fields(item, MENU_FIELDS) elif otype == "setting": data = f"{item['name']:<40}: {item['value']}" else: data = "Unknown item type selected!" print(data)
[docs]def list_items(remote, otype): """ List all items of a given object type and print it to stdout. :param remote: The remote to use as the query-source. :param otype: The object type to query. """ items = remote.get_item_names(otype) items.sort() for item in items: print(f" {item}")
[docs]def n2s(data): """ Return spaces for None :param data: The data to check for. :return: The data itself or an empty string. """ if data is None: return "" return data
[docs]def opt(options, k, defval=""): """ Returns an option from an Optparse values instance :param options: The options object to search in. :param k: The key which is in the optparse values instance. :param defval: The default value to return. :return: The value for the specified key. """ try: data = getattr(options, k) except Exception: # FIXME: debug only # traceback.print_exc() return defval return n2s(data)
def _add_parser_option_from_field(parser, field, settings): """ Add options from a field dynamically to an optparse instance. :param parser: The optparse instance to add the options to. :param field: The field to parse. :param settings: Global cobbler settings as returned from ``CollectionManager.settings()`` """ # extract data from field dictionary name = field[0] default = field[1] if isinstance(default, str) and default.startswith("SETTINGS:"): setting_name = default.replace("SETTINGS:", "", 1) default = settings[setting_name] description = field[3] tooltip = field[5] choices = field[6] if choices and default not in choices: raise Exception( f"field {name} default value ({default}) is not listed in choices ({str(choices)})" ) if tooltip != "": description += f" ({tooltip})" # generate option string option_string = f"--{name.replace('_', '-')}" # add option to parser if isinstance(choices, list) and len(choices) != 0: description += f" (valid options: {','.join(choices)})" parser.add_option(option_string, dest=name, help=description, choices=choices) else: parser.add_option(option_string, dest=name, help=description)
[docs]def add_options_from_fields( object_type, parser, fields, network_interface_fields, settings, object_action ): """ Add options to the command line from the fields queried from the Cobbler server. :param object_type: The object type to add options for. :param parser: The optparse instance to add options to. :param fields: The list of fields to add options for. :param network_interface_fields: The list of network interface fields if the object type is a system. :param settings: Global cobbler settings as returned from ``CollectionManager.settings()`` :param object_action: The object action to add options for. May be "add", "edit", "find", "copy", "rename", "remove". If none of these options is given then this method does nothing. """ if object_action in ["add", "edit", "find", "copy", "rename"]: for field in fields: _add_parser_option_from_field(parser, field, settings) # system object if object_type == "system": for field in network_interface_fields: _add_parser_option_from_field(parser, field, settings) parser.add_option( "--interface", dest="interface", help="the interface to operate on (can only be " "specified once per command line)", ) if object_action in ["add", "edit"]: parser.add_option( "--delete-interface", dest="delete_interface", action="store_true" ) parser.add_option("--rename-interface", dest="rename_interface") if object_action in ["copy", "rename"]: parser.add_option("--newname", help="new object name") if object_action not in ["find"] and object_type != "setting": parser.add_option( "--in-place", action="store_true", dest="in_place", help="edit items in kopts or autoinstall without clearing the other items", ) elif object_action == "remove": parser.add_option("--name", help=f"{object_type} name to remove") parser.add_option( "--recursive", action="store_true", dest="recursive", help="also delete child objects", )
[docs]def get_comma_separated_args( option: optparse.Option, opt_str, value: str, parser: optparse.OptionParser ): """ Simple callback function to achieve option split with comma. Reference for the method signature can be found at: https://docs.python.org/3/library/optparse.html#defining-a-callback-option :param option: The option the callback is executed for :param opt_str: Unused for this callback function. Would be the extended option if the user used the short version. :param value: The value which should be split by comma. :param parser: The optparse instance which the callback should be added to. """ # TODO: Migrate to argparse if not isinstance(option, optparse.Option): raise optparse.OptionValueError("Option is not an optparse.Option object!") if not isinstance(value, str): raise optparse.OptionValueError("Value is not a string!") if not isinstance(parser, optparse.OptionParser): raise optparse.OptionValueError( "Parser is not an optparse.OptionParser object!" ) setattr(parser.values, str(option.dest), value.split(","))
[docs]class CobblerCLI: """ Main CLI Class which contains the logic to communicate with the Cobbler Server. """ def __init__(self, cliargs): """ The constructor to create a Cobbler CLI. """ # Load server ip and ports from local config self.url_cobbler_api = utils.local_get_cobbler_api_url() self.url_cobbler_xmlrpc = utils.local_get_cobbler_xmlrpc_url() # FIXME: allow specifying other endpoints, and user+pass self.parser = optparse.OptionParser() self.remote = xmlrpc.client.Server(self.url_cobbler_api) self.shared_secret = utils.get_shared_secret() self.args = cliargs
[docs] def start_task(self, name: str, options: dict) -> str: r""" Start an asynchronous task in the background. :param name: "background\_" % name function must exist in remote.py. This function will be called in a subthread. :param options: Dictionary of options passed to the newly started thread :return: Id of the newly started task """ options = utils.strip_none(vars(options), omit_none=True) background_fn = getattr(self.remote, f"background_{name}") return background_fn(options, self.token)
[docs] def get_object_type(self, args) -> Optional[str]: """ If this is a CLI command about an object type, e.g. "cobbler distro add", return the type, like "distro" :param args: The args from the CLI. :return: The object type or None """ if len(args) < 2: return None if args[1] in OBJECT_TYPES: return args[1] return None
[docs] def get_object_action(self, object_type, args) -> Optional[str]: """ If this is a CLI command about an object type, e.g. "cobbler distro add", return the action, like "add" :param object_type: The object type. :param args: The args from the CLI. :return: The action or None. """ if object_type is None or len(args) < 3: return None if args[2] in OBJECT_ACTIONS_MAP[object_type]: return args[2] return None
[docs] def get_direct_action(self, object_type, args) -> Optional[str]: """ If this is a general command, e.g. "cobbler hardlink", return the action, like "hardlink" :param object_type: Must be None or None is returned. :param args: The arg from the CLI. :return: The action key, "version" or None. """ if object_type is not None: return None if len(args) < 2: return None if args[1] == "--help": return None if args[1] == "--version": return "version" return args[1]
[docs] def check_setup(self) -> int: """ Detect permissions and service accessibility problems and provide nicer error messages for them. """ with xmlrpc.client.ServerProxy(self.url_cobbler_xmlrpc) as xmlrpc_server: try: xmlrpc_server.ping() except Exception as exception: print( f"cobblerd does not appear to be running/accessible: {repr(exception)}", file=sys.stderr, ) return 411 with xmlrpc.client.ServerProxy(self.url_cobbler_api) as xmlrpc_server: try: xmlrpc_server.ping() except Exception: print( "httpd does not appear to be running and proxying Cobbler, or SELinux is in the way. Original " "traceback:", file=sys.stderr, ) traceback.print_exc() return 411 if not os.path.exists("/var/lib/cobbler/web.ss"): print( "Missing login credentials file. Has cobblerd failed to start?", file=sys.stderr, ) return 411 if not os.access("/var/lib/cobbler/web.ss", os.R_OK): print( "User cannot run command line, need read access to /var/lib/cobbler/web.ss", file=sys.stderr, ) return 411 return 0
[docs] def run(self, args) -> int: """ Process the command line and do what the user asks. :param args: The args of the CLI """ self.token = self.remote.login("", self.shared_secret) object_type = self.get_object_type(args) object_action = self.get_object_action(object_type, args) direct_action = self.get_direct_action(object_type, args) try: if object_type is not None: if object_action is not None: return self.object_command(object_type, object_action) return self.print_object_help(object_type) if direct_action is not None: return self.direct_command(direct_action) return self.print_help() except xmlrpc.client.Fault as err: if err.faultString.find("cobbler.cexceptions.CX") != -1: print(self.cleanup_fault_string(err.faultString)) else: print("### ERROR ###") print( "Unexpected remote error, check the server side logs for further info" ) print(err.faultString) return 1
[docs] def cleanup_fault_string(self, fault_str: str) -> str: """ Make a remote exception nicely readable by humans so it's not evident that is a remote fault. Users should not have to understand tracebacks. :param fault_str: The stacktrace to niceify. :return: A nicer error messsage. """ if fault_str.find(">:") != -1: (_, rest) = fault_str.split(">:", 1) if rest.startswith('"') or rest.startswith("'"): rest = rest[1:] if rest.endswith('"') or rest.endswith("'"): rest = rest[:-1] return rest return fault_str
[docs] def get_fields(self, object_type: str) -> list: """ For a given name of an object type, return the FIELDS data structure. :param object_type: The object to return the fields of. :return: The fields or None """ if object_type == "distro": return DISTRO_FIELDS if object_type == "profile": return PROFILE_FIELDS if object_type == "system": return SYSTEM_FIELDS if object_type == "repo": return REPO_FIELDS if object_type == "image": return IMAGE_FIELDS if object_type == "menu": return MENU_FIELDS if object_type == "setting": return SETTINGS_FIELDS return []
[docs] def object_command(self, object_type: str, object_action: str) -> int: """ Process object-based commands such as "distro add" or "profile rename" :param object_type: The object type to execute an action for. :param object_action: The action to execute. :return: Depending on the object and action. :raises NotImplementedError: :raises RuntimeError: """ # if assigned, we must tail the logfile task_id = INVALID_TASK settings = self.remote.get_settings() fields = self.get_fields(object_type) network_interface_fields = None if object_type == "system": network_interface_fields = NETWORK_INTERFACE_FIELDS if object_action in ["add", "edit", "copy", "rename", "find", "remove"]: add_options_from_fields( object_type, self.parser, fields, network_interface_fields, settings, object_action, ) elif object_action in ["list", "autoadd"]: pass elif object_action not in ("reload", "update"): self.parser.add_option("--name", dest="name", help="name of object") elif object_action == "reload": self.parser.add_option( "--filename", dest="filename", help="filename to load data from" ) (options, _) = self.parser.parse_args(self.args) # the first three don't require a name if object_action == "report": if options.name is not None: report_item(self.remote, object_type, None, options.name) else: report_items(self.remote, object_type) elif object_action == "list": list_items(self.remote, object_type) elif object_action == "find": items = self.remote.find_items( object_type, utils.strip_none(vars(options), omit_none=True), "name", False, ) for item in items: print(item) elif object_action == "autoadd" and object_type == "repo": try: self.remote.auto_add_repos(self.token) except xmlrpc.client.Fault as err: (_, emsg) = err.faultString.split(":", 1) print(f"exception on server: {emsg}") return 1 elif object_action in OBJECT_ACTIONS: if opt(options, "name") == "" and object_action not in ("reload", "update"): print("--name is required") return 1 if object_action in ["add", "edit", "copy", "rename", "remove"]: try: if object_type == "setting": settings = self.remote.get_settings() if options.value is None: raise RuntimeError( "You must specify a --value when editing a setting" ) if not settings.get("allow_dynamic_settings", False): raise RuntimeError( "Dynamic settings changes are not enabled. Change the " "allow_dynamic_settings to True and restart cobblerd to enable dynamic " "settings changes" ) if options.name == "allow_dynamic_settings": raise RuntimeError("Cannot modify that setting live") if self.remote.modify_setting( options.name, options.value, self.token ): raise RuntimeError("Changing the setting failed") else: self.remote.xapi_object_edit( object_type, options.name, object_action, utils.strip_none(vars(options), omit_none=True), self.token, ) except xmlrpc.client.Fault as error: (_, emsg) = error.faultString.split(":", 1) print(f"exception on server: {emsg}") return 1 except RuntimeError as error: print(error.args[0]) return 1 elif object_action == "get-autoinstall": if object_type == "profile": data = self.remote.generate_profile_autoinstall(options.name) elif object_type == "system": data = self.remote.generate_system_autoinstall(options.name) else: print( 'Invalid object type selected! Allowed are "profile" and "system".' ) return 1 print(data) elif object_action == "dumpvars": if object_type == "profile": data = self.remote.get_blended_data(options.name, "") elif object_type == "system": data = self.remote.get_blended_data("", options.name) else: print( 'Invalid object type selected! Allowed are "profile" and "system".' ) return 1 # FIXME: pretty-printing and sorting here keys = list(data.keys()) keys.sort() for key in keys: print(f"{key}: {data[key]}") elif object_action in ["poweron", "poweroff", "powerstatus", "reboot"]: power = { "power": object_action.replace("power", ""), "systems": [options.name], } task_id = self.remote.background_power_system(power, self.token) elif object_action == "update": task_id = self.remote.background_signature_update( utils.strip_none(vars(options), omit_none=True), self.token ) elif object_action == "reload": filename = opt( options, "filename", "/var/lib/cobbler/distro_signatures.json" ) try: signatures.load_signatures(filename, cache=True) except Exception: print( f"There was an error loading the signature data in {filename}." ) print( "Please check the JSON file or run 'cobbler signature update'." ) return 1 else: print("Signatures were successfully loaded") else: raise NotImplementedError() else: raise NotImplementedError() # FIXME: add tail/polling code here if task_id != INVALID_TASK: self.print_task(task_id) return self.follow_task(task_id) return 0
[docs] def direct_command(self, action_name: str): """ Process non-object based commands like "sync" and "hardlink". :param action_name: The action to execute. :return: Depending on the action. """ task_id = INVALID_TASK self.parser.set_usage(f"Usage: %prog {action_name} [options]") if action_name == "buildiso": defaultiso = os.path.join(os.getcwd(), "generated.iso") self.parser.add_option( "--iso", dest="iso", default=defaultiso, help="(OPTIONAL) output ISO to this file", ) self.parser.add_option( "--profiles", dest="profiles", help="(OPTIONAL) use these profiles only" ) self.parser.add_option( "--systems", dest="systems", help="(OPTIONAL) use these systems only" ) self.parser.add_option( "--tempdir", dest="buildisodir", help="(OPTIONAL) working directory" ) self.parser.add_option( "--distro", dest="distro", help="Must be specified to choose the Kernel and Initrd for the ISO being built.", ) self.parser.add_option( "--standalone", dest="standalone", action="store_true", help="(OPTIONAL) creates a standalone ISO with all required distro files, " "but without any added repos", ) self.parser.add_option( "--airgapped", dest="airgapped", action="store_true", help="(OPTIONAL) implies --standalone but additionally includes the repository files into ISO", ) self.parser.add_option( "--source", dest="source", help="(OPTIONAL) used with --standalone/--airgapped to specify a source for the distribution files", ) self.parser.add_option( "--exclude-dns", dest="exclude_dns", action="store_true", help="(OPTIONAL) prevents addition of name server addresses to the kernel boot options", ) self.parser.add_option( "--exclude-systems", dest="exclude_systems", action="store_true", help="(OPTIONAL) prevents writing system records", ) self.parser.add_option( "--mkisofs-opts", dest="mkisofs_opts", help="(OPTIONAL) extra options for xorrisofs", ) self.parser.add_option( "--esp", dest="esp", help="(OPTIONAL) location of ESP partition, e.g. for secure boot", ) (options, _) = self.parser.parse_args(self.args) task_id = self.start_task("buildiso", options) elif action_name == "replicate": self.parser.add_option( "--master", dest="master", help="Cobbler server to replicate from." ) self.parser.add_option("--port", dest="port", help="Remote port.") self.parser.add_option( "--distros", dest="distro_patterns", help="patterns of distros to replicate", ) self.parser.add_option( "--profiles", dest="profile_patterns", help="patterns of profiles to replicate", ) self.parser.add_option( "--systems", dest="system_patterns", help="patterns of systems to replicate", ) self.parser.add_option( "--repos", dest="repo_patterns", help="patterns of repos to replicate" ) self.parser.add_option( "--image", dest="image_patterns", help="patterns of images to replicate" ) self.parser.add_option( "--omit-data", dest="omit_data", action="store_true", help="do not rsync data", ) self.parser.add_option( "--sync-all", dest="sync_all", action="store_true", help="sync all data" ) self.parser.add_option( "--prune", dest="prune", action="store_true", help="remove objects (of all types) not found on the master", ) self.parser.add_option( "--use-ssl", dest="use_ssl", action="store_true", help="use ssl to access the Cobbler master server api", ) (options, _) = self.parser.parse_args(self.args) task_id = self.start_task("replicate", options) elif action_name == "aclsetup": self.parser.add_option( "--adduser", dest="adduser", help="give acls to this user" ) self.parser.add_option( "--addgroup", dest="addgroup", help="give acls to this group" ) self.parser.add_option( "--removeuser", dest="removeuser", help="remove acls from this user" ) self.parser.add_option( "--removegroup", dest="removegroup", help="remove acls from this group" ) (options, _) = self.parser.parse_args(self.args) task_id = self.start_task("aclsetup", options) elif action_name == "version": version = self.remote.extended_version() print(f"Cobbler {version['version']}") print(f" source: {version['gitstamp']}, {version['gitdate']}") print(f" build time: {version['builddate']}") elif action_name == "hardlink": (options, _) = self.parser.parse_args(self.args) task_id = self.start_task("hardlink", options) elif action_name == "status": self.parser.parse_args(self.args) print(self.remote.get_status("text", self.token)) elif action_name == "validate-autoinstalls": (options, _) = self.parser.parse_args(self.args) task_id = self.start_task("validate_autoinstall_files", options) elif action_name == "import": self.parser.add_option( "--arch", dest="arch", help="OS architecture being imported" ) self.parser.add_option( "--breed", dest="breed", help="the breed being imported" ) self.parser.add_option( "--os-version", dest="os_version", help="the version being imported" ) self.parser.add_option( "--path", dest="path", help="local path or rsync location" ) self.parser.add_option("--name", dest="name", help="name, ex 'RHEL-5'") self.parser.add_option( "--available-as", dest="available_as", help="tree is here, don't mirror" ) self.parser.add_option( "--autoinstall", dest="autoinstall_file", help="assign this autoinstall file", ) self.parser.add_option( "--rsync-flags", dest="rsync_flags", help="pass additional flags to rsync", ) (options, _) = self.parser.parse_args(self.args) if options.path and "rsync://" not in options.path: # convert relative path to absolute path options.path = os.path.abspath(options.path) task_id = self.start_task("import", options) elif action_name == "reposync": self.parser.add_option( "--only", dest="only", help="update only this repository name" ) self.parser.add_option( "--tries", dest="tries", help="try each repo this many times", default=1, type="int", ) self.parser.add_option( "--no-fail", dest="nofail", help="don't stop reposyncing if a failure occurs", action="store_true", ) (options, args) = self.parser.parse_args(self.args) task_id = self.start_task("reposync", options) elif action_name == "check": results = self.remote.check(self.token) counter = 0 if len(results) > 0: print( "The following are potential configuration items that you may want to fix:\n" ) for result in results: counter += 1 print(f"{counter}: {result}") print( "\nRestart cobblerd and then run 'cobbler sync' to apply changes." ) else: print("No configuration problems found. All systems go.") elif action_name == "sync": self.parser.add_option( "--verbose", dest="verbose", action="store_true", help="run sync with more output", ) self.parser.add_option( "--dhcp", dest="dhcp", action="store_true", help="write DHCP config files and restart service", ) self.parser.add_option( "--dns", dest="dns", action="store_true", help="write DNS config files and restart service", ) self.parser.add_option( "--systems", dest="systems", type="string", action="callback", callback=get_comma_separated_args, help="run a sync only on specified systems", ) # ToDo: Add tftp syncing when it's cleaned up (options, args) = self.parser.parse_args(self.args) if options.systems is not None: task_id = self.start_task("syncsystems", options) else: task_id = self.start_task("sync", options) elif action_name == "report": (options, args) = self.parser.parse_args(self.args) print("distros:\n==========") report_items(self.remote, "distro") print("\nprofiles:\n==========") report_items(self.remote, "profile") print("\nsystems:\n==========") report_items(self.remote, "system") print("\nrepos:\n==========") report_items(self.remote, "repo") print("\nimages:\n==========") report_items(self.remote, "image") print("\nmenus:\n==========") report_items(self.remote, "menu") elif action_name == "list": # no tree view like 1.6? This is more efficient remotely # for large configs and prevents xfering the whole config # though we could consider that... (options, args) = self.parser.parse_args(self.args) print("distros:") list_items(self.remote, "distro") print("\nprofiles:") list_items(self.remote, "profile") print("\nsystems:") list_items(self.remote, "system") print("\nrepos:") list_items(self.remote, "repo") print("\nimages:") list_items(self.remote, "image") print("\nmenus:") list_items(self.remote, "menu") elif action_name == "mkloaders": (options, _) = self.parser.parse_args(self.args) task_id = self.start_task("mkloaders", options) else: print(f"No such command: {action_name}") return 1 # FIXME: add tail/polling code here if task_id != INVALID_TASK: self.print_task(task_id) return self.follow_task(task_id) return 0
[docs] def print_task(self, task_id): """ Pretty print a task executed on the server. This prints to stdout. :param task_id: The id of the task to be pretty printed. """ print(f"task started: {task_id}") events = self.remote.get_events() (etime, name, _, _) = events[task_id] atime = time.asctime(time.localtime(etime)) print(f"task started (id={name}, time={atime})")
[docs] def follow_task(self, task_id): """ Parse out this task's specific messages from the global log :param task_id: The id of the task to follow. """ logfile = "/var/log/cobbler/cobbler.log" # adapted from: https://code.activestate.com/recipes/157035/ with open(logfile, "r", encoding="UTF-8") as file: # Find the size of the file and move to the end # st_results = os.stat(filename) # st_size = st_results[6] # file.seek(st_size) while 1: where = file.tell() line = file.readline() if not line.startswith("[" + task_id + "]"): continue if line.find("### TASK COMPLETE ###") != -1: print("*** TASK COMPLETE ***") return 0 if line.find("### TASK FAILED ###") != -1: print("!!! TASK FAILED !!!") return 1 if not line: time.sleep(1) file.seek(where) else: if line.find(" | "): line = line.split(" | ")[-1] print(line, end="")
[docs] def print_object_help(self, object_type) -> int: """ Prints the subcommands for a given object, e.g. "cobbler distro --help" :param object_type: The object type to print the help for. """ commands = OBJECT_ACTIONS_MAP[object_type] commands.sort() print("usage\n=====") for command in commands: print(f"cobbler {object_type} {command}") return 2
[docs] def print_help(self) -> int: """ Prints general-top level help, e.g. "cobbler --help" or "cobbler" or "cobbler command-does-not-exist" """ print("usage\n=====") print("cobbler <distro|profile|system|repo|image|menu> ... ") print( " [add|edit|copy|get-autoinstall*|list|remove|rename|report] [options|--help]" ) print("cobbler setting [edit|report]") print(f"cobbler <{'|'.join(DIRECT_ACTIONS)}> [options|--help]") return 2
[docs]def main() -> int: """ CLI entry point """ cli = CobblerCLI(sys.argv) return_code = cli.check_setup() if return_code != 0: return return_code return_code = cli.run(sys.argv) if return_code is None: return 0 return return_code
if __name__ == "__main__": sys.exit(main())