webvirtcloud/vrtManager/util.py

from django.utils.translation import gettext_noop
import ipaddress
import secrets
import re
import string

import libvirt
import lxml.etree as etree

from django.utils.translation import gettext



def message(text, **values):
    """Our text, marked with gettext_noop, with %(name)s for each of values.
    A value may be a list of messages: they are put in its place one after
    another."""
    return {"text": str(text), "values": values}


def render(text, values, translate):
    """text with its values, through translate (gettext, or str for English)."""
    def shown(value):
        if isinstance(value, list):
            return "; ".join(render(m["text"], m["values"], translate) for m in value)
        if isinstance(value, OperationError):
            return render(value.text, value.values, translate)
        return value
    text = translate(text)
    return text % {key: shown(value) for key, value in values.items()} if values else text


class OperationError(libvirt.libvirtError):
    """An error with WebVirtCloud's own message (see message()). Its args are
    the English message, for logs and the audit log; str() is the message in
    the active language, English outside a request. A job keeps the text and
    values (operations.utils.said) and shows them in its reader's language.

    libvirtError(msg) reports the thread's last libvirt error instead of msg
    when there is one, so an earlier, unrelated failure would be shown.
    """

    def __init__(self, text, **values):
        self.text, self.values = str(text), values
        Exception.__init__(self, render(self.text, values, str))
        self.err = None

    def __str__(self):
        return render(self.text, self.values, gettext)


class InUse(OperationError):
    """An object something still uses cannot be changed or deleted."""


class ConnectionFailed(OperationError):
    """The host could not be connected to; the message may name it and how it
    is reached."""


class HostBusy(ConnectionFailed):
    """The host is connected, but this process's web requests to it
    (HOST_REQUESTS) still wait for its answers."""


def randomUUID():
    """Generate a random UUID."""
    u = [secrets.randbelow(256) for _ in range(16)]
    u[6] = u[6] & 15 | 4 << 4
    u[8] = u[8] & 63 | 2 << 6
    return "-".join(["%02x" * 4, "%02x" * 2, "%02x" * 2, "%02x" * 2, "%02x" * 6]) % tuple(u)


def randomPasswd(length=12, alphabet=string.ascii_letters + string.digits):
    """Generate a random password"""
    return "".join([secrets.choice(alphabet) for _ in range(length)])


def xml_escape(xml_str):
    """Replaces chars ' " < > & with xml safe counterparts"""
    if xml_str is None:
        return None

    xml_str = xml_str.replace("&", "&amp;")
    xml_str = xml_str.replace("'", "&apos;")
    xml_str = xml_str.replace('"', "&quot;")
    xml_str = xml_str.replace("<", "&lt;")
    xml_str = xml_str.replace(">", "&gt;")
    return xml_str


def parse_xml(xml):
    """XML a user entered, read for the panel's checks: its entities not
    expanded, nothing fetched and no DTD; libvirt gets the text itself."""
    parser = etree.XMLParser(resolve_entities=False, no_network=True)
    try:
        tree = etree.fromstring(xml.encode(), parser)  # bytes: the XML may declare its encoding
    except etree.XMLSyntaxError as err:
        raise OperationError(gettext_noop("The XML is not valid: %(err)s"), err=err) from None
    if tree.getroottree().docinfo.internalDTD is not None:
        raise OperationError(gettext_noop("The XML may not declare a DTD"))
    return tree


def get_xml_path(xml, path=None, func=None):
    """
    Return the content from the passed xml xpath, or return the result
    of a passed function (receives xpathContext as its only arg)
    """
    doc = None
    result = None

    doc = etree.fromstring(xml)
    if path:
        result = get_xpath(doc, path)
    elif func:
        result = func(doc)

    else:
        raise ValueError("'path' or 'func' is required.")
    return result


def get_xpath(doc, path):
    result = None

    ret = doc.xpath(path)
    if ret is not None:
        if isinstance(ret, list):
            if len(ret) >= 1:
                result = ret[0].text if hasattr(ret[0], "text") else ret[0]
        else:
            result = ret

    return result


def validate_macaddr(val):
    if val is None:
        return

    if not (isinstance(val, str) or isinstance(val, str)):
        raise ValueError("MAC address must be a string.")

    form = re.match("^([0-9a-fA-F]{1,2}:){5}[0-9a-fA-F]{1,2}$", val)
    if form is None:
        raise ValueError(gettext("MAC address must be of the format AA:BB:CC:DD:EE:FF, was %(mac)s") % {"mac": val})


def vol_dev_type(vol_bus):
    if vol_bus == "ide":
        return "hd"
    elif vol_bus == "fdc":
        return "fd"
    elif vol_bus == "virtio":
        return "vd"
    elif vol_bus in ["scsi", "sata", "usb"]:
        return "sd"


def vol_driver_type(vol_format, pool_type=None):
    """The disk driver type for a volume of vol_format: raw for a volume
    whose pool knows no image format of it (none, unknown, an ISO, a disk
    pool's partition), as libvirt uses a disk without a driver type."""
    if pool_type == "disk" or vol_format in (None, "", "none", "unknown", "iso"):
        return "raw"
    return vol_format


def is_loopback(address):
    """127.0.0.0/8, ::1 or localhost: reachable only on that host itself."""
    try:
        return ipaddress.ip_address(address).is_loopback
    except ValueError:
        return address == "localhost"