path: root/virtinst/domcapabilities.py

#
# Support for parsing libvirt's domcapabilities XML
#
# Copyright 2014 Red Hat, Inc.
#
# This work is licensed under the GNU GPLv2 or later.
# See the COPYING file in the top-level directory.

import re
import xml.etree.ElementTree as ET

import libvirt

from .domain import DomainCpu
from .logger import log
from .xmlbuilder import XMLBuilder, XMLChildProperty, XMLProperty


########################################
# Genering <enum> and <value> handling #
########################################

class _Value(XMLBuilder):
    XML_NAME = "value"
    value = XMLProperty(".")


class _HasValues(XMLBuilder):
    values = XMLChildProperty(_Value)

    def get_values(self):
        return [v.value for v in self.values]

    def has_value(self, val):
        return val in self.get_values()


class _Enum(_HasValues):
    XML_NAME = "enum"
    name = XMLProperty("./@name")


class _CapsBlock(_HasValues):
    supported = XMLProperty("./@supported", is_yesno=True)
    _supported_present = XMLProperty("./@supported")
    enums = XMLChildProperty(_Enum)

    @property
    def present(self):
        return self._supported_present is not None

    def enum_names(self):
        return [e.name for e in self.enums]

    def has_enum(self, name):
        return name in self.enum_names()

    def get_enum(self, name):
        for enum in self.enums:
            if enum.name == name:
                return enum
        # Didn't find a match. Could be talking to older libvirt, or
        # driver with incomplete info. Return a stub enum
        return _Enum(self.conn)


def _make_capsblock(xml_root_name):
    """
    Build a class object representing a list of <enum> in the XML. For
    example, domcapabilities may have a block like:

    <graphics supported='yes'>
      <enum name='type'>
        <value>sdl</value>
        <value>vnc</value>
        <value>spice</value>
      </enum>
    </graphics>

    To build a class that tracks that whole <graphics> block, call this
    like _make_capsblock("graphics")
    """
    class TmpClass(_CapsBlock):
        pass
    setattr(TmpClass, "XML_NAME", xml_root_name)
    return TmpClass


################################
# SEV launch security handling #
################################

class _SEV(XMLBuilder):
    XML_NAME = "sev"
    supported = XMLProperty("./@supported", is_yesno=True)
    maxESGuests = XMLProperty("./maxESGuests")


#############################
# Misc toplevel XML classes #
#############################

class _OS(_CapsBlock):
    XML_NAME = "os"
    loader = XMLChildProperty(_make_capsblock("loader"), is_single=True)


class _Devices(_CapsBlock):
    XML_NAME = "devices"
    hostdev = XMLChildProperty(_make_capsblock("hostdev"), is_single=True)
    disk = XMLChildProperty(_make_capsblock("disk"), is_single=True)
    video = XMLChildProperty(_make_capsblock("video"), is_single=True)
    graphics = XMLChildProperty(_make_capsblock("graphics"), is_single=True)
    tpm = XMLChildProperty(_make_capsblock("tpm"), is_single=True)
    filesystem = XMLChildProperty(_make_capsblock("filesystem"), is_single=True)
    redirdev = XMLChildProperty(_make_capsblock("redirdev"), is_single=True)
    channel = XMLChildProperty(_make_capsblock("channel"), is_single=True)


class _Features(_CapsBlock):
    XML_NAME = "features"
    gic = XMLChildProperty(_make_capsblock("gic"), is_single=True)
    sev = XMLChildProperty(_SEV, is_single=True)


class _MemoryBacking(_CapsBlock):
    XML_NAME = "memoryBacking"


###############
# CPU classes #
###############

class _CPUModel(XMLBuilder):
    XML_NAME = "model"
    model = XMLProperty(".")
    usable = XMLProperty("./@usable")
    fallback = XMLProperty("./@fallback")


class _CPUMode(_CapsBlock):
    XML_NAME = "mode"
    name = XMLProperty("./@name")

    models = XMLChildProperty(_CPUModel)
    def get_model(self, name):
        for model in self.models:
            if model.model == name:
                return model


class _CPU(XMLBuilder):
    XML_NAME = "cpu"
    modes = XMLChildProperty(_CPUMode)

    def get_mode(self, name):
        for mode in self.modes:
            if mode.name == name:
                return mode


#############################
# CPU flags/baseline helpers#
#############################

def _convert_mode_to_cpu(xml, arch):
    root = ET.fromstring(xml)
    root.tag = "cpu"
    root.attrib = {}
    aelement = ET.SubElement(root, "arch")
    aelement.text = arch
    return ET.tostring(root, encoding="unicode")


def _get_expanded_cpu(domcaps, mode):
    cpuXML = _convert_mode_to_cpu(mode.get_xml(), domcaps.arch)
    log.debug("Generated CPU XML for security flag baseline:\n%s", cpuXML)

    try:
        expandedXML = domcaps.conn.baselineHypervisorCPU(
                domcaps.path, domcaps.arch,
                domcaps.machine, domcaps.domain, [cpuXML],
                libvirt.VIR_CONNECT_BASELINE_CPU_EXPAND_FEATURES)
    except (libvirt.libvirtError, AttributeError):
        expandedXML = domcaps.conn.baselineCPU([cpuXML],
                libvirt.VIR_CONNECT_BASELINE_CPU_EXPAND_FEATURES)

    return DomainCpu(domcaps.conn, expandedXML)


def _lookup_cpu_security_features(domcaps):
    ret = []
    sec_features = [
            'spec-ctrl',
            'ssbd',
            'ibpb',
            'virt-ssbd',
            'md-clear']

    for m in domcaps.cpu.modes:
        if m.name != "host-model" or not m.supported:
            continue  # pragma: no cover

        try:
            cpu = _get_expanded_cpu(domcaps, m)
        except libvirt.libvirtError as e:  # pragma: no cover
            log.warning(_("Failed to get expanded CPU XML: %s"), e)
            break

        for feature in cpu.features:
            if feature.name in sec_features:
                ret.append(feature.name)

    log.debug("Found host-model security features: %s", ret)
    return ret


#################################
# DomainCapabilities main class #
#################################

class DomainCapabilities(XMLBuilder):
    XML_NAME = "domainCapabilities"
    os = XMLChildProperty(_OS, is_single=True)
    cpu = XMLChildProperty(_CPU, is_single=True)
    devices = XMLChildProperty(_Devices, is_single=True)
    features = XMLChildProperty(_Features, is_single=True)
    memorybacking = XMLChildProperty(_MemoryBacking, is_single=True)

    arch = XMLProperty("./arch")
    domain = XMLProperty("./domain")
    machine = XMLProperty("./machine")
    path = XMLProperty("./path")


    ################
    # Init helpers #
    ################

    @staticmethod
    def build_from_params(conn, emulator, arch, machine, hvtype):
        xml = None
        if conn.support.conn_domain_capabilities():
            try:
                xml = conn.getDomainCapabilities(emulator, arch,
                    machine, hvtype)
                log.debug("Fetched domain capabilities for (%s,%s,%s,%s): %s",
                          emulator, arch, machine, hvtype, xml)
            except Exception:  # pragma: no cover
                log.debug("Error fetching domcapabilities XML",
                    exc_info=True)

        if not xml:
            # If not supported, just use a stub object
            return DomainCapabilities(conn)
        return DomainCapabilities(conn, parsexml=xml)

    @staticmethod
    def build_from_guest(guest):
        return DomainCapabilities.build_from_params(guest.conn,
            guest.emulator, guest.os.arch, guest.os.machine, guest.type)


    #########################
    # UEFI/firmware methods #
    #########################

    # Mapping of UEFI binary names to their associated architectures. We
    # only use this info to do things automagically for the user, it shouldn't
    # validate anything the user explicitly enters.
    _uefi_arch_patterns = {
        "i686": [
            r".*edk2-i386-.*\.fd",  # upstream qemu
            r".*ovmf-ia32.*",  # fedora, gerd's firmware repo
        ],
        "x86_64": [
            r".*edk2-x86_64-.*\.fd",  # upstream qemu
            r".*OVMF_CODE\.fd",  # RHEL
            r".*ovmf-x64/OVMF.*\.fd",  # gerd's firmware repo
            r".*ovmf-x86_64-.*",  # SUSE
            r".*ovmf.*", ".*OVMF.*",  # generic attempt at a catchall
        ],
        "aarch64": [
            r".*AAVMF_CODE\.fd",  # RHEL
            r".*aarch64/QEMU_EFI.*",  # gerd's firmware repo
            r".*aarch64.*",  # generic attempt at a catchall
            r".*edk2-aarch64-code\.fd",  # upstream qemu
        ],
        "armv7l": [
            r".*AAVMF32_CODE\.fd",  # Debian qemu-efi-arm package
            r".*arm/QEMU_EFI.*",  # fedora, gerd's firmware repo
            r".*edk2-arm-code\.fd"  # upstream qemu
        ],
    }

    def find_uefi_path_for_arch(self):
        """
        Search the loader paths for one that matches the passed arch
        """
        if not self.arch_can_uefi():
            return  # pragma: no cover

        firmware_files = [f.value for f in self.os.loader.values]
        if self.conn.is_bhyve():
            for firmware_file in firmware_files:
                if 'BHYVE_UEFI.fd' in firmware_file:
                    return firmware_file
            return (firmware_files and
                    firmware_files[0] or None)  # pragma: no cover

        patterns = self._uefi_arch_patterns.get(self.arch)
        for pattern in patterns:
            for path in firmware_files:
                if re.match(pattern, path):
                    return path

    def label_for_firmware_path(self, path):
        """
        Return a pretty label for passed path, based on if we know
        about it or not
        """
        if not path:
            if self.arch in ["i686", "x86_64"]:
                return _("BIOS")
            return _("Default")

        for arch, patterns in self._uefi_arch_patterns.items():
            for pattern in patterns:
                if re.match(pattern, path):
                    return (_("UEFI %(arch)s: %(path)s") %
                        {"arch": arch, "path": path})

        return _("Custom: %(path)s" % {"path": path})

    def arch_can_uefi(self):
        """
        Return True if we know how to setup UEFI for the passed arch
        """
        return self.arch in self._uefi_arch_patterns

    def supports_uefi_loader(self):
        """
        Return True if libvirt advertises support for UEFI loader
        """
        return self.os.loader.get_enum("readonly").has_value("yes")

    def supports_firmware_efi(self):
        return self.os.get_enum("firmware").has_value("efi")


    #######################
    # CPU support methods #
    #######################

    def supports_safe_host_model(self):
        """
        Return True if domcaps reports support for cpu mode=host-model.
        host-model in fact predates this support, however it wasn't
        general purpose safe prior to domcaps advertisement.
        """
        m = self.cpu.get_mode("host-model")
        return (m and m.supported and
                m.models[0].fallback == "forbid")

    def supports_safe_host_passthrough(self):
        """
        Return True if host-passthrough is safe enough to use by default.
        We limit this to domcaps new enough to report whether host-passthrough
        is migratable or not, which also means libvirt is about new enough
        to not taint the VM for using host-passthrough
        """
        m = self.cpu.get_mode("host-passthrough")
        return (m and m.supported and
                "on" in m.get_enum("hostPassthroughMigratable").get_values())

    def get_cpu_models(self):
        models = []

        for m in self.cpu.modes:
            if m.name == "custom" and m.supported:
                for model in m.models:
                    if model.usable != "no":
                        models.append(model.model)

        return models

    _features = None
    def get_cpu_security_features(self):
        if self._features is None:
            self._features = _lookup_cpu_security_features(self) or []
        return self._features


    ########################
    # Misc support methods #
    ########################

    def supports_sev_launch_security(self, check_es=False):
        """
        Returns False if either libvirt doesn't advertise support for SEV at
        all (< libvirt-4.5.0) or if it explicitly advertises it as unsupported
        on the platform
        """
        if check_es:
            return bool(self.features.sev.supported and
                        self.features.sev.maxESGuests)
        return bool(self.features.sev.supported)

    def supports_video_bochs(self):
        """
        Returns False if either libvirt or qemu do not have support to bochs
        video type.
        """
        return self.devices.video.get_enum("modelType").has_value("bochs")

    def supports_video_qxl(self):
        if not self.devices.video.has_enum("modelType"):
            # qxl long predates modelType in domcaps, so if it is missing,
            # use spice support as a rough value
            return self.supports_graphics_spice()
        return self.devices.video.get_enum("modelType").has_value("qxl")

    def supports_video_virtio(self):
        return self.devices.video.get_enum("modelType").has_value("virtio")

    def supports_tpm_emulator(self):
        """
        Returns False if either libvirt or qemu do not have support for
        emulating a TPM.
        """
        models = self.devices.tpm.get_enum("model").get_values()
        backends = self.devices.tpm.get_enum("backendModel").get_values()

        if self.arch == "armv7l" and models == ["tpm-tis"]:
            # libvirt as of 8.4.0 can advertise armv7l tpm-tis support,
            # but then explicitly rejects that config. If we see it,
            # assume TPM is not supported
            # https://gitlab.com/libvirt/libvirt/-/issues/329
            return False

        return len(models) > 0 and bool("emulator" in backends)

    def supports_graphics_spice(self):
        if not self.devices.graphics.supported:
            # domcaps is too old, or the driver doesn't advertise graphics
            # support. Use our pre-existing logic
            if not self.conn.is_qemu() and not self.conn.is_test():
                return False
            return self.conn.caps.host.cpu.arch in ["i686", "x86_64"]

        return self.devices.graphics.get_enum("type").has_value("spice")

    def supports_channel_spicevmc(self):
        """
        Return False if libvirt explicitly advertises no support for
        spice channel
        """
        if self.devices.channel.supported is None:
            # Follow the original behavior in case of talking to older
            # libvirt.
            return True

        return self.devices.channel.get_enum("type").has_value("spicevmc")

    def supports_redirdev_usb(self):
        """
        Return False if libvirt explicitly advertises no support for
        USB redirect
        """
        if self.devices.redirdev.supported is None:
            # Follow the original behavior in case of talking to older
            # libvirt.
            return True

        return self.devices.redirdev.get_enum("bus").has_value("usb")

    def supports_filesystem_virtiofs(self):
        """
        Return True if libvirt advertises support for virtiofs
        """
        return self.devices.filesystem.get_enum(
                "driverType").has_value("virtiofs")

    def supports_memorybacking_memfd(self):
        """
        Return True if libvirt advertises support for memfd memory backend
        """
        return self.memorybacking.get_enum("sourceType").has_value("memfd")