virt-manager/virtinst/install/urldetect.py

#
# Copyright 2006-2007, 2013 Red Hat, Inc.
#
# This work is licensed under the GNU GPLv2 or later.
# See the COPYING file in the top-level directory.

import configparser
import os
import re

from ..logger import log
from ..osdict import OSDB


###############################################
# Helpers for detecting distro from given URL #
###############################################

class _DistroCache(object):
    def __init__(self, fetcher):
        self._fetcher = fetcher
        self._filecache = {}

        self._treeinfo = None
        self.treeinfo_family = None
        self.treeinfo_version = None
        self.treeinfo_name = None
        self.treeinfo_matched = False

        self.suse_content = None
        self.checked_for_suse_content = False
        self.debian_media_type = None
        self.mageia_version = None

        self.libosinfo_os_variant = None
        self.libosinfo_mediaobj = None
        self.libosinfo_treeobj = None

    def acquire_file_content(self, path):
        if path not in self._filecache:
            try:
                content = self._fetcher.acquireFileContent(path)
            except ValueError as e:
                content = None
                log.debug("Failed to acquire file=%s: %s", path, e)
            self._filecache[path] = content
        return self._filecache[path]

    @property
    def treeinfo(self):
        if self._treeinfo:
            return self._treeinfo

        # Vast majority of trees here use .treeinfo. However, trees via
        # Red Hat satellite on akamai CDN will use treeinfo, because akamai
        # doesn't do dotfiles apparently:
        #
        #   https://bugzilla.redhat.com/show_bug.cgi?id=635065
        #
        # Anaconda is the canonical treeinfo consumer and they check for both
        # locations, so we need to do the same
        treeinfostr = (self.acquire_file_content(".treeinfo") or
            self.acquire_file_content("treeinfo"))
        if treeinfostr is None:
            return None

        # If the file doesn't parse or there's no 'family', this will
        # error, but that should be fine because we aren't going to
        # successfully detect the tree anyways
        treeinfo = configparser.ConfigParser()
        treeinfo.read_string(treeinfostr)
        self.treeinfo_family = treeinfo.get("general", "family")
        self._treeinfo = treeinfo
        log.debug("treeinfo family=%s", self.treeinfo_family)

        if self._treeinfo.has_option("general", "version"):
            self.treeinfo_version = self._treeinfo.get("general", "version")
            log.debug("Found treeinfo version=%s", self.treeinfo_version)

        if self._treeinfo.has_option("general", "name"):
            self.treeinfo_name = self._treeinfo.get("general", "name")
            log.debug("Found treeinfo name=%s", self.treeinfo_name)

        return self._treeinfo

    def treeinfo_family_regex(self, famregex):
        if not self.treeinfo:
            return False

        ret = bool(re.match(famregex, self.treeinfo_family))
        self.treeinfo_matched = ret
        if not ret:
            log.debug("Didn't match treeinfo family regex=%s", famregex)
        return ret

    def content_regex(self, filename, regex):
        """
        Fetch 'filename' and return True/False if it matches the regex
        """
        content = self.acquire_file_content(filename)
        if content is None:
            return False

        for line in content.splitlines():
            if re.match(regex, line):
                return True

        log.debug("found filename=%s but regex=%s didn't match",
                filename, regex)
        return False

    def get_treeinfo_media(self, typ):
        """
        Pull kernel/initrd/boot.iso paths out of the treeinfo for
        the passed data
        """
        def _get_treeinfo_path(media_name):
            image_type = self.treeinfo.get("general", "arch")
            if typ == "xen":
                image_type = "xen"
            return self.treeinfo.get("images-%s" % image_type, media_name)

        try:
            return [(_get_treeinfo_path("kernel"),
                     _get_treeinfo_path("initrd"))]
        except Exception:  # pragma: no cover
            log.debug("Failed to parse treeinfo kernel/initrd",
                    exc_info=True)
            return []

    def split_version(self):
        verstr = self.treeinfo_version
        def _safeint(c):
            try:
                return int(c)
            except Exception:
                return 0

        # Parse a string like 6.9 or 7.4 into its two parts
        # centos altarch's have just version=7
        update = 0
        version = _safeint(verstr)
        if verstr.count(".") >= 1:
            version = _safeint(verstr.split(".")[0])
            update = _safeint(verstr.split(".")[1])

        log.debug("converted verstr=%s to version=%s update=%s",
                verstr, version, update)
        return version, update

    def fetcher_is_iso(self):
        return self._fetcher.is_iso()

    def guess_os_from_iso(self):
        ret = OSDB.guess_os_by_iso(self._fetcher.location)
        if not ret:
            return False

        self.libosinfo_os_variant, self.libosinfo_mediaobj = ret
        if (not self.libosinfo_mediaobj.get_kernel_path() or
            not self.libosinfo_mediaobj.get_initrd_path()):  # pragma: no cover
            # This can happen if the media is live media, or just
            # with incomplete libosinfo data
            log.debug("libosinfo didn't report any media kernel/initrd "
                          "path for detected os_variant=%s",
                          self.libosinfo_mediaobj)
            return False
        return True

    def guess_os_from_tree(self):
        ret = OSDB.guess_os_by_tree(self._fetcher.location)
        if not ret:
            return False

        self.libosinfo_os_variant, self.libosinfo_treeobj = ret
        self.treeinfo_matched = True
        return True


class _SUSEContent(object):
    """
    Helper class tracking the SUSE 'content' files
    """
    def __init__(self, content_str):
        self.content_str = content_str
        self.content_dict = {}

        for line in self.content_str.splitlines():
            for prefix in ["LABEL", "DISTRO", "VERSION",
                           "BASEARCHS", "DEFAULTBASE", "REPOID"]:
                if line.startswith(prefix + " "):
                    self.content_dict[prefix] = line.split(" ", 1)[1]

        log.debug("SUSE content dict: %s", self.content_dict)
        self.tree_arch = self._get_tree_arch()
        self.product_name = self._get_product_name()
        self.product_version = self._get_product_version()
        log.debug("SUSE content product_name=%s product_version=%s "
            "tree_arch=%s", self.product_name, self.product_version,
            self.tree_arch)

    def _get_tree_arch(self):
        # Examples:
        # opensuse 11.4: BASEARCHS i586 x86_64
        # opensuse 12.3: BASEARCHS i586 x86_64
        # opensuse 10.3: DEFAULTBASE i586
        distro_arch = (self.content_dict.get("BASEARCHS") or
                       self.content_dict.get("DEFAULTBASE"))
        if not distro_arch and "REPOID" in self.content_dict:
            distro_arch = self.content_dict["REPOID"].rsplit('/', 1)[1]
        if not distro_arch:
            return None  # pragma: no cover

        tree_arch = distro_arch.strip()
        # Fix for 13.2 official oss repo
        if tree_arch.find("i586-x86_64") != -1:
            tree_arch = "x86_64"
        return tree_arch

    def _get_product_name(self):
        """
        Parse the SUSE product name. Examples:
        SUSE Linux Enterprise Server 11 SP4
        openSUSE 11.4
        """
        # Some field examples in the wild
        #
        # opensuse 10.3: LABEL openSUSE 10.3
        # opensuse 11.4: LABEL openSUSE 11.4
        # opensuse 12.3: LABEL openSUSE
        # sles11sp4 DVD: LABEL SUSE Linux Enterprise Server 11 SP4
        #
        #
        # DISTRO cpe:/o:opensuse:opensuse:13.2,openSUSE
        # DISTRO cpe:/o:suse:sled:12:sp3,SUSE Linux Enterprise Desktop 12 SP3
        #
        # As of 2018 all latest distros match only DISTRO and REPOID.
        product_name = None
        if "LABEL" in self.content_dict:
            product_name = self.content_dict["LABEL"]
        elif "," in self.content_dict.get("DISTRO", ""):
            product_name = self.content_dict["DISTRO"].rsplit(",", 1)[1]

        log.debug("SUSE content product_name=%s", product_name)
        return product_name

    def _get_product_version(self):
        # Some example fields:
        #
        # opensuse 10.3: VERSION 10.3
        # opensuse 12.3: VERSION 12.3
        # SLES-10-SP4-DVD-x86_64-GM-DVD1.iso: VERSION 10.4-0
        #
        # REPOID obsproduct://build.suse.de/SUSE:SLE-11-SP4:GA/SUSE_SLES/11.4/DVD/x86_64
        # REPOID obsproduct://build.suse.de/SUSE:SLE-12-SP3:GA/SLES/12.3/DVD/aarch64
        #
        # As of 2018 all latest distros match only DISTRO and REPOID.
        if not self.product_name:
            return None  # pragma: no cover

        distro_version = self.content_dict.get("VERSION", "")
        if "-" in distro_version:
            distro_version = distro_version.split('-', 1)[0]

        # Special case, parse version out of a line like this
        # cpe:/o:opensuse:opensuse:13.2,openSUSE
        if (not distro_version and
            re.match("^.*:.*,openSUSE*", self.content_dict["DISTRO"])):
            distro_version = self.content_dict["DISTRO"].rsplit(
                    ",", 1)[0].strip().rsplit(":")[4]
        distro_version = distro_version.strip()

        if "Enterprise" in self.product_name or "SLES" in self.product_name:
            sle_version = self.product_name.strip().rsplit(' ')[4]
            if len(self.product_name.strip().rsplit(' ')) > 5:
                sle_version = (sle_version + '.' +
                        self.product_name.strip().rsplit(' ')[5][2])
            distro_version = sle_version

        return distro_version


def getDistroStore(guest, fetcher, skip_error):
    log.debug("Finding distro store for location=%s", fetcher.location)

    arch = guest.os.arch
    _type = guest.os.os_type
    osobj = guest.osinfo
    stores = _build_distro_list(osobj)
    cache = _DistroCache(fetcher)

    for sclass in stores:
        if not sclass.is_valid(cache):
            continue

        store = sclass(fetcher.location, arch, _type, cache)
        log.debug("Detected class=%s osvariant=%s",
                      store.__class__.__name__, store.get_osdict_info())
        return store

    if skip_error:
        return None

    # No distro was detected. See if the URL even resolves, and if not
    # give the user a hint that maybe they mistyped. This won't always
    # be true since some webservers don't allow directory listing.
    # https://www.redhat.com/archives/virt-tools-list/2014-December/msg00048.html
    extramsg = ""
    if not fetcher.can_access():
        extramsg = (": " +
            _("The URL could not be accessed, maybe you mistyped?"))

    msg = (_("Could not find an installable distribution at URL '%s'") %
            fetcher.location)
    msg += extramsg
    msg += "\n\n"
    msg += _("The location must be the root directory of an install tree.\n"
          "See virt-install man page for various distro examples.")
    raise ValueError(msg)


##################
# Distro classes #
##################

class _DistroTree(object):
    """
    Class for determining the kernel/initrd path for an install
    tree (URL, ISO, or local directory)
    """
    PRETTY_NAME = None
    matching_distros = []

    def __init__(self, location, arch, vmtype, cache):
        self.type = vmtype
        self.arch = arch
        self.uri = location
        self.cache = cache

        if self.cache.libosinfo_os_variant:
            self._os_variant = self.cache.libosinfo_os_variant
        else:
            self._os_variant = self._detect_version()

        if (self._os_variant and
            not OSDB.lookup_os(self._os_variant)):
            log.debug("Detected os_variant as %s, which is not in osdict.",
                    self._os_variant)
            self._os_variant = None

        self._kernel_paths = []
        if self.cache.treeinfo_matched:
            self._kernel_paths = self.cache.get_treeinfo_media(self.type)
        else:
            self._set_manual_kernel_paths()


    def _set_manual_kernel_paths(self):
        """
        If kernel/initrd path could not be determined from a source
        like treeinfo, subclasses can override this to set a list
        of manual paths
        """

    def _detect_version(self):
        """
        Hook for subclasses to detect media os variant.
        """
        log.debug("%s does not implement any osdict detection", self)
        return None


    ##############
    # Public API #
    ##############

    @classmethod
    def is_valid(cls, cache):
        raise NotImplementedError

    def get_kernel_paths(self):
        return self._kernel_paths

    def get_osdict_info(self):
        """
        Return detected osdict value
        """
        return self._os_variant

    def get_os_media(self):
        """
        Return an OsMedia wrapper around the detected libosinfo media object
        """
        return self.cache.libosinfo_mediaobj

    def get_os_tree(self):
        """
        Return an OsTree wrapper around the detected libosinfo media object
        """
        return self.cache.libosinfo_treeobj


class _FedoraDistro(_DistroTree):
    PRETTY_NAME = "Fedora"
    matching_distros = ["fedora"]

    @classmethod
    def is_valid(cls, cache):
        famregex = ".*Fedora.*"
        return cache.treeinfo_family_regex(famregex)

    def _detect_version(self):
        latest_variant = "fedora-unknown"

        verstr = self.cache.treeinfo_version
        if not verstr:  # pragma: no cover
            log.debug("No treeinfo version? Assume latest_variant=%s",
                    latest_variant)
            return latest_variant

        # rawhide trees changed to use version=Rawhide in Apr 2016
        if verstr in ["development", "rawhide", "Rawhide"]:
            log.debug("treeinfo version=%s, using latest_variant=%s",
                    verstr, latest_variant)
            return latest_variant

        # treeinfo version is just an integer
        variant = "fedora" + verstr
        if OSDB.lookup_os(variant):
            return variant

        log.debug(
                "variant=%s from treeinfo version=%s not found, "
                "using latest_variant=%s", variant, verstr, latest_variant)
        return latest_variant


class _RHELDistro(_DistroTree):
    PRETTY_NAME = "Red Hat Enterprise Linux"
    matching_distros = ["rhel"]
    _variant_prefix = "rhel"

    @classmethod
    def is_valid(cls, cache):
        # Matches:
        #   Red Hat Enterprise Linux
        #   RHEL Atomic Host
        famregex = ".*(Red Hat Enterprise Linux|RHEL).*"
        if cache.treeinfo_family_regex(famregex):
            return True

    def _detect_version(self):
        if not self.cache.treeinfo_version:  # pragma: no cover
            log.debug("No treeinfo version? Not setting an os_variant")
            return

        version, update = self.cache.split_version()

        # start with example base=rhel7, then walk backwards
        # through the OS list to find the latest os name that matches
        # this way we handle rhel7.6 from treeinfo when osdict only
        # knows about rhel7.5
        base = self._variant_prefix + str(version)
        while update >= 0:
            tryvar = base + ".%s" % update
            if OSDB.lookup_os(tryvar):
                return tryvar
            update -= 1


class _CentOSDistro(_RHELDistro):
    PRETTY_NAME = "CentOS"
    matching_distros = ["centos"]
    _variant_prefix = "centos"

    @classmethod
    def is_valid(cls, cache):
        if cache.treeinfo_family_regex(".*CentOS.*"):
            return True
        if cache.treeinfo_family_regex(".*Scientific.*"):
            return True



class _SuseDistro(_RHELDistro):
    PRETTY_NAME = None
    _suse_regex = []
    matching_distros = []
    _variant_prefix = NotImplementedError
    famregex = NotImplementedError

    @classmethod
    def is_valid(cls, cache):
        if cache.treeinfo_family_regex(cls.famregex):
            return True

        if not cache.checked_for_suse_content:
            cache.checked_for_suse_content = True
            content_str = cache.acquire_file_content("content")
            if content_str is None:
                return False

            try:
                cache.suse_content = _SUSEContent(content_str)
            except Exception as e:  # pragma: no cover
                log.debug("Error parsing SUSE content file: %s", str(e))
                return False

        if not cache.suse_content:
            return False
        for regex in cls._suse_regex:
            if re.match(regex, cache.suse_content.product_name or ""):
                return True
        return False

    def _set_manual_kernel_paths(self):
        # We only reach here if no treeinfo was matched
        tree_arch = self.cache.suse_content.tree_arch

        if re.match(r'i[4-9]86', tree_arch):
            tree_arch = 'i386'

        oldkern = "linux"
        oldinit = "initrd"
        if tree_arch == "x86_64":
            oldkern += "64"
            oldinit += "64"

        if self.type == "xen":
            # Matches Opensuse > 10.2 and sles 10
            self._kernel_paths.append(
                ("boot/%s/vmlinuz-xen" % tree_arch,
                 "boot/%s/initrd-xen" % tree_arch))

        if str(self._os_variant).startswith(("sles11", "sled11")):
            if tree_arch == "s390x":
                self._kernel_paths.append(
                    ("boot/s390x/vmrdr.ikr", "boot/s390x/initrd"))
            if tree_arch == "ppc64":
                self._kernel_paths.append(
                    ("suseboot/linux64", "suseboot/initrd64"))

        # Tested with SLES 12 for ppc64le, all s390x
        self._kernel_paths.append(
            ("boot/%s/linux" % tree_arch,
             "boot/%s/initrd" % tree_arch))
        # Tested with Opensuse 10.0
        self._kernel_paths.append(
            ("boot/loader/%s" % oldkern,
             "boot/loader/%s" % oldinit))
        # Tested with Opensuse >= 10.2, 11, and sles 10
        self._kernel_paths.append(
            ("boot/%s/loader/linux" % tree_arch,
             "boot/%s/loader/initrd" % tree_arch))

    def _detect_osdict_from_suse_content(self):
        if not self.cache.suse_content:
            return  # pragma: no cover

        distro_version = self.cache.suse_content.product_version
        if not distro_version:
            return  # pragma: no cover

        version = distro_version.split('.', 1)[0].strip()

        if str(self._variant_prefix).startswith(("sles", "sled")):
            sp_version = ""
            if len(distro_version.split('.', 1)) == 2:
                sp_version = 'sp' + distro_version.split('.', 1)[1].strip()

            return self._variant_prefix + version + sp_version

        return self._variant_prefix + distro_version

    def _detect_osdict_from_url(self):
        root = "opensuse"
        oses = [n for n in OSDB.list_os() if n.name.startswith(root)]

        for osobj in oses:
            codename = osobj.name[len(root):]
            if re.search("/%s/" % codename, self.uri):
                return osobj.name

    def _detect_from_treeinfo(self):
        if not self.cache.treeinfo_name:
            return
        if re.search("openSUSE Tumbleweed", self.cache.treeinfo_name):
            return "opensusetumbleweed"

        version, update = self.cache.split_version()
        base = self._variant_prefix + str(version)
        while update >= 0:
            tryvar = base
            # SLE doesn't use '.0' for initial releases in
            # osinfo-db (sles11, sles12, etc)
            if update > 0 or not base.startswith('sle'):
                tryvar += ".%s" % update
            if OSDB.lookup_os(tryvar):
                return tryvar
            update -= 1

    def _detect_version(self):
        var = self._detect_from_treeinfo()
        if not var:
            var = self._detect_osdict_from_url()
        if not var:
            var = self._detect_osdict_from_suse_content()
        return var


class _SLESDistro(_SuseDistro):
    PRETTY_NAME = "SLES"
    matching_distros = ["sles"]
    _variant_prefix = "sles"
    _suse_regex = [".*SUSE Linux Enterprise Server*", ".*SUSE SLES*"]
    famregex = ".*SUSE Linux Enterprise.*"


class _SLEDDistro(_SuseDistro):
    PRETTY_NAME = "SLED"
    matching_distros = ["sled"]
    _variant_prefix = "sled"
    _suse_regex = [".*SUSE Linux Enterprise Desktop*"]
    famregex = ".*SUSE Linux Enterprise.*"


class _OpensuseDistro(_SuseDistro):
    PRETTY_NAME = "openSUSE"
    matching_distros = ["opensuse"]
    _variant_prefix = "opensuse"
    _suse_regex = [".*openSUSE.*"]
    famregex = ".*openSUSE.*"


class _DebianDistro(_DistroTree):
    # ex. http://ftp.egr.msu.edu/debian/dists/sarge/main/installer-i386/
    # daily builds: https://d-i.debian.org/daily-images/amd64/
    PRETTY_NAME = "Debian"
    matching_distros = ["debian"]
    _debname = "debian"

    @classmethod
    def is_valid(cls, cache):
        def check_manifest(mfile):
            is_ubuntu = cls._debname == "ubuntu"
            if cache.content_regex(mfile, ".*[Uu]buntu.*"):
                return is_ubuntu
            return cache.content_regex(mfile, ".*[Dd]ebian.*")

        media_type = None
        if check_manifest("current/images/MANIFEST"):
            media_type = "url"
        elif check_manifest("current/legacy-images/MANIFEST"):
            media_type = "legacy_url"
        elif check_manifest("daily/MANIFEST"):
            media_type = "daily"
        elif cache.content_regex(".disk/info",
                "%s.*" % cls._debname.capitalize()):
            # There's two cases here:
            # 1) Direct access ISO, attached as CDROM afterwards. We
            #    use one set of kernels in that case which seem to
            #    assume the prescence of CDROM media
            # 2) ISO mounted and exported over URL. We use a different
            #    set of kernels that expect to boot from the network
            if cache.fetcher_is_iso():
                media_type = "disk"
            else:
                media_type = "mounted_iso_url"

        if media_type:
            cache.debian_media_type = media_type
        return bool(media_type)


    def _set_manual_kernel_paths(self):
        if self.cache.debian_media_type == "disk":
            self._set_installcd_paths()
        else:
            self._set_url_paths()


    def _find_treearch(self):
        for pattern in [r"^.*/installer-(\w+)/?$",
                        r"^.*/daily-images/(\w+)/?$"]:
            arch = re.findall(pattern, self.uri)
            if not arch:
                continue
            log.debug("Found pattern=%s treearch=%s in uri",
                pattern, arch[0])
            return arch[0]

        # Check for standard arch strings which will be
        # in the URI name for --location $ISO mounts
        for arch in ["i386", "amd64", "x86_64", "arm64"]:
            if arch in self.uri:
                log.debug("Found treearch=%s in uri", arch)
                if arch == "x86_64":
                    arch = "amd64"  # pragma: no cover
                return arch

        # Otherwise default to i386
        arch = "i386"
        log.debug("No treearch found in uri, defaulting to arch=%s", arch)
        return arch

    def _set_url_paths(self):
        url_prefix = "current/images"
        if self.cache.debian_media_type == "daily":
            url_prefix = "daily"
        elif self.cache.debian_media_type == "mounted_iso_url":
            url_prefix = "install"
        elif self.cache.debian_media_type == "legacy_url":
            url_prefix = "current/legacy-images"

        tree_arch = self._find_treearch()
        hvmroot = "%s/netboot/%s-installer/%s/" % (url_prefix,
                self._debname, tree_arch)
        initrd_basename = "initrd.gz"
        kernel_basename = "linux"
        if tree_arch in ["ppc64el"]:
            kernel_basename = "vmlinux"

        if tree_arch == "s390x":
            hvmroot = "%s/generic/" % url_prefix
            kernel_basename = "kernel.%s" % self._debname
            initrd_basename = "initrd.%s" % self._debname


        if self.type == "xen":
            xenroot = "%s/netboot/xen/" % url_prefix
            self._kernel_paths.append(
                    (xenroot + "vmlinuz", xenroot + "initrd.gz"))
        self._kernel_paths.append(
                (hvmroot + kernel_basename, hvmroot + initrd_basename))

    def _set_installcd_paths(self):
        if self._debname == "ubuntu":
            if not self.arch == "s390x":
                kpair = ("install/vmlinuz", "install/initrd.gz")
            else:
                kpair = ("boot/kernel.ubuntu", "boot/initrd.ubuntu")
        elif self.arch == "x86_64":
            kpair = ("install.amd/vmlinuz", "install.amd/initrd.gz")
        elif self.arch == "i686":
            kpair = ("install.386/vmlinuz", "install.386/initrd.gz")
        elif self.arch == "aarch64":
            kpair = ("install.a64/vmlinuz", "install.a64/initrd.gz")
        elif self.arch == "ppc64le":
            kpair = ("install/vmlinux", "install/initrd.gz")
        elif self.arch == "s390x":
            kpair = ("boot/linux_vm", "boot/root.bin")
        else:
            kpair = ("install/vmlinuz", "install/initrd.gz")
        self._kernel_paths += [kpair]
        return True

    def _detect_version(self):
        oses = [n for n in OSDB.list_os() if n.name.startswith(self._debname)]

        if self.cache.debian_media_type == "daily":
            log.debug("Appears to be debian 'daily' URL, using latest "
                "debian OS")
            return oses[0].name

        for osobj in oses:
            if osobj.codename:
                # Ubuntu codenames look like 'Warty Warthog'
                codename = osobj.codename.split()[0].lower()
            else:
                if " " not in osobj.label:
                    continue  # pragma: no cover
                # Debian labels look like 'Debian Sarge'
                codename = osobj.label.split()[1].lower()

            if ("/%s/" % codename) in self.uri:
                log.debug("Found codename=%s in the URL string", codename)
                return osobj.name


class _UbuntuDistro(_DebianDistro):
    # https://archive.ubuntu.com/ubuntu/dists/natty/main/installer-amd64/
    PRETTY_NAME = "Ubuntu"
    matching_distros = ["ubuntu"]
    _debname = "ubuntu"


class _MageiaDistro(_DistroTree):
    # https://distro.ibiblio.org/mageia/distrib/cauldron/x86_64/
    PRETTY_NAME = "Mageia"
    matching_distros = ["mageia"]

    @classmethod
    def is_valid(cls, cache):
        if not cache.mageia_version:
            content = cache.acquire_file_content("VERSION")
            if not content:
                return False

            m = re.match(r"^Mageia (\d+) .*", content)
            if not m:
                return False  # pragma: no cover

            cache.mageia_version = m.group(1)

        return bool(cache.mageia_version)

    def _set_manual_kernel_paths(self):
        self._kernel_paths += [
            ("isolinux/%s/vmlinuz" % self.arch,
             "isolinux/%s/all.rdz" % self.arch)]

    def _detect_version(self):
        # version is just an integer
        variant = "mageia" + self.cache.mageia_version
        if OSDB.lookup_os(variant):
            return variant


class _GenericTreeinfoDistro(_DistroTree):
    """
    Generic catchall class for .treeinfo using distros
    """
    PRETTY_NAME = "Generic Treeinfo"
    matching_distros = []

    @classmethod
    def is_valid(cls, cache):
        if cache.treeinfo:
            cache.treeinfo_matched = True
            return True
        return False


class _LibosinfoDistro(_DistroTree):
    """
    For ISO media detection that was fully handled by libosinfo
    """
    PRETTY_NAME = "Libosinfo detected"
    matching_distros = []

    @classmethod
    def is_valid(cls, cache):
        if cache.fetcher_is_iso():
            return cache.guess_os_from_iso()
        return cache.guess_os_from_tree()

    def _set_manual_kernel_paths(self):
        self._kernel_paths += [
                (self.cache.libosinfo_mediaobj.get_kernel_path(),
                 self.cache.libosinfo_mediaobj.get_initrd_path())
        ]


def _build_distro_list(osobj):
    allstores = [
        # Libosinfo takes priority
        _LibosinfoDistro,
        _FedoraDistro,
        _RHELDistro,
        _CentOSDistro,
        _SLESDistro,
        _SLEDDistro,
        _OpensuseDistro,
        _DebianDistro,
        _UbuntuDistro,
        _MageiaDistro,
        # Always stick GenericDistro at the end, since it's a catchall
        _GenericTreeinfoDistro,
    ]

    # If user manually specified an os_distro, bump its URL class
    # to the top of the list
    if osobj.distro:
        log.debug("variant=%s has distro=%s, looking for matching "
                      "distro store to prioritize",
                      osobj.name, osobj.distro)
        found_store = None
        for store in allstores:
            if osobj.distro in store.matching_distros:
                found_store = store

        if found_store:
            log.debug("Prioritizing distro store=%s", found_store)
            allstores.remove(found_store)
            allstores.insert(0, found_store)
        else:
            log.debug("No matching store found, not prioritizing anything")

    force_libosinfo = os.environ.get("VIRTINST_TEST_SUITE_FORCE_LIBOSINFO")
    if force_libosinfo:  # pragma: no cover
        if bool(int(force_libosinfo)):
            allstores = [_LibosinfoDistro]
        else:
            allstores.remove(_LibosinfoDistro)

    return allstores