virt-manager/virtManager/createnet.py

# Copyright (C) 2006-2007, 2013 Red Hat, Inc.
# Copyright (C) 2006 Hugh O. Brock <hbrock@redhat.com>
#
# This work is licensed under the GNU GPLv2 or later.
# See the COPYING file in the top-level directory.

import ipaddress
import logging

from gi.repository import Gtk
from gi.repository import Gdk
from gi.repository import Pango

import libvirt

from virtinst import generatename
from virtinst import Network

from . import uiutil
from .asyncjob import vmmAsyncJob
from .baseclass import vmmGObjectUI
from .xmleditor import vmmXMLEditor

_green = Gdk.Color.parse("#c0ffc0")[1]
_red = Gdk.Color.parse("#ffc0c0")[1]
_black = Gdk.Color.parse("#000000")[1]
_white = Gdk.Color.parse("#f0f0f0")[1]


def _make_ipaddr(addrstr):
    if addrstr is None:
        return None
    try:
        return ipaddress.ip_network(str(addrstr), strict=False)
    except Exception:
        return None


class vmmCreateNetwork(vmmGObjectUI):
    def __init__(self, conn):
        vmmGObjectUI.__init__(self, "createnet.ui", "vmm-create-net")
        self.conn = conn

        self._xmleditor = vmmXMLEditor(self.builder, self.topwin,
                self.widget("net-details-align"),
                self.widget("net-details"))
        self._xmleditor.connect("xml-requested",
                self._xmleditor_xml_requested_cb)

        self.builder.connect_signals({
            "on_create_cancel_clicked": self.close,
            "on_vmm_create_delete_event": self.close,
            "on_create_finish_clicked": self.finish,

            "on_net_forward_mode_changed": self._net_forward_mode_changed_cb,
            "on_net_dns_use_toggled": self._net_dns_use_toggled_cb,

            "on_net-ipv4-enable_toggled":  self.change_ipv4_enable,
            "on_net-ipv4-network_changed":  self.change_ipv4_network,
            "on_net-dhcpv4-enable_toggled": self.change_dhcpv4_enable,
            "on_net-dhcpv4-start_changed":  self.change_dhcpv4_start,
            "on_net-dhcpv4-end_changed":    self.change_dhcpv4_end,

            "on_net-ipv6-enable_toggled":  self.change_ipv6_enable,
            "on_net-ipv6-network_changed":  self.change_ipv6_network,
            "on_net-dhcpv6-enable_toggled": self.change_dhcpv6_enable,
            "on_net-dhcpv6-start_changed":  self.change_dhcpv6_start,
            "on_net-dhcpv6-end_changed":    self.change_dhcpv6_end,
        })
        self.bind_escape_key_close()

        self.set_initial_state()


    ####################
    # Standard methods #
    ####################

    def show(self, parent):
        logging.debug("Showing new network wizard")
        self.reset_state()
        self.topwin.set_transient_for(parent)
        self.topwin.present()

    def close(self, ignore1=None, ignore2=None):
        logging.debug("Closing new network wizard")
        self.topwin.hide()
        return 1

    def _cleanup(self):
        self.conn = None
        self._xmleditor.cleanup()
        self._xmleditor = None


    ###########
    # UI init #
    ###########

    def set_initial_state(self):
        blue = Gdk.Color.parse("#0072A8")[1]
        self.widget("header").modify_bg(Gtk.StateType.NORMAL, blue)

        # [ label, dev name ]
        pf_list = self.widget("net-hostdevs")
        pf_model = Gtk.ListStore(str, str)
        pf_list.set_model(pf_model)
        text = uiutil.init_combo_text_column(pf_list, 1)
        text.set_property("ellipsize", Pango.EllipsizeMode.MIDDLE)

        # [ label, dev name ]
        fw_list = self.widget("net-forward-device")
        fw_model = Gtk.ListStore(str, str)
        fw_list.set_model(fw_model)
        uiutil.init_combo_text_column(fw_list, 1)

        # [ label, mode ]
        mode_list = self.widget("net-forward-mode")
        mode_model = Gtk.ListStore(str, str)
        mode_list.set_model(mode_model)
        uiutil.init_combo_text_column(mode_list, 1)

        mode_model.append(["nat", _("NAT")])
        mode_model.append(["route", _("Routed")])
        mode_model.append(["open", _("Open")])
        mode_model.append(["isolated", _("Isolated")])
        mode_model.append(["hostdev", _("SR-IOV pool")])

    def reset_state(self):
        self._xmleditor.reset_state()

        basename = "network"
        def cb(n):
            return generatename.check_libvirt_collision(
                self.conn.get_backend().networkLookupByName, n)
        default_name = generatename.generate_name(basename, cb)
        self.widget("net-name").set_text(default_name)

        self.widget("net-dns-use-netname").set_active(True)

        self.widget("net-ipv4-expander").set_visible(True)
        self.widget("net-ipv4-expander").set_expanded(False)
        self.widget("net-ipv6-expander").set_visible(True)
        self.widget("net-ipv6-expander").set_expanded(False)
        self.widget("net-dns-expander").set_visible(True)
        self.widget("net-dns-expander").set_expanded(False)

        self.widget("net-ipv4-enable").set_active(True)
        self.widget("net-ipv4-network").set_text("192.168.100.0/24")
        self.widget("net-dhcpv4-enable").set_active(True)
        self.widget("net-dhcpv4-start").set_text("192.168.100.128")
        self.widget("net-dhcpv4-end").set_text("192.168.100.254")

        self.widget("net-ipv6-enable").set_active(False)
        self.widget("net-ipv6-network").set_text("")
        self.widget("net-dhcpv6-enable").set_active(False)
        self.widget("net-dhcpv6-start").set_text("")
        self.widget("net-dhcpv6-end").set_text("")


        # Populate physical forward devices
        devnames = []
        for nodedev in self.conn.filter_nodedevs("net"):
            devnames.append(nodedev.xmlobj.interface)
        for iface in self.conn.list_interfaces():
            if iface.get_name() not in devnames:
                devnames.append(iface.get_name())

        fw_model = self.widget("net-forward-device").get_model()
        fw_model.clear()
        fw_model.append([None, _("Any physical device")])

        for name in devnames:
            fw_model.append([name, _("Physical device %s") % name])
        self.widget("net-forward-device").set_active(0)

        self.widget("net-forward-mode").set_active(0)


        # Populate hostdev forward devices
        devprettynames = []
        ifnames = []
        for pcidev in self.conn.filter_nodedevs("pci"):
            if not pcidev.xmlobj.is_pci_sriov():
                continue
            devdesc = pcidev.pretty_name()
            for netdev in self.conn.filter_nodedevs("net"):
                if pcidev.xmlobj.name != netdev.xmlobj.parent:
                    continue
                ifname = netdev.xmlobj.interface
                devprettyname = "%s (%s)" % (ifname, devdesc)
                devprettynames.append(devprettyname)
                ifnames.append(ifname)
                break

        pf_model = self.widget("net-hostdevs").get_model()
        pf_model.clear()
        for devprettyname, ifname in zip(devprettynames, ifnames):
            pf_model.append([ifname, devprettyname])
        if len(pf_model) == 0:
            pf_model.append([None, _("No available device")])
        self.widget("net-hostdevs").set_active(0)



    ##################
    # UI get helpers #
    ##################

    def get_config_ipv4_enable(self):
        return (self.widget("net-ipv4-expander").is_visible() and
                self.widget("net-ipv4-enable").get_active())
    def get_config_ipv6_enable(self):
        return (self.widget("net-ipv6-expander").is_visible() and
                self.widget("net-ipv6-enable").get_active())
    def get_config_dhcpv4_enable(self):
        return self.widget("net-dhcpv4-enable").get_active()
    def get_config_dhcpv6_enable(self):
        return self.widget("net-dhcpv6-enable").get_active()

    def get_config_domain_name(self):
        widget = self.widget("net-domain-name")
        if not widget.is_visible():
            return None

        if self.widget("net-dns-use-netname").get_active():
            return self.widget("net-name").get_text()
        return widget.get_text()

    def _get_network_helper(self, widgetname):
        widget = self.widget(widgetname)
        if not widget.is_visible() or not widget.is_sensitive():
            return None
        return _make_ipaddr(widget.get_text())

    def get_config_ip4(self):
        return self._get_network_helper("net-ipv4-network")
    def get_config_dhcpv4_start(self):
        return self._get_network_helper("net-dhcpv4-start")
    def get_config_dhcpv4_end(self):
        return self._get_network_helper("net-dhcpv4-end")
    def get_config_ip6(self):
        return self._get_network_helper("net-ipv6-network")
    def get_config_dhcpv6_start(self):
        return self._get_network_helper("net-dhcpv6-start")
    def get_config_dhcpv6_end(self):
        return self._get_network_helper("net-dhcpv6-end")

    def get_config_forwarding(self):
        mode = uiutil.get_list_selection(self.widget("net-forward-mode"))
        if mode == "isolated":
            return [None, None]

        if mode == "hostdev":
            dev = uiutil.get_list_selection(self.widget("net-hostdevs"))
        else:
            dev = uiutil.get_list_selection(self.widget("net-forward-device"))
        return [dev, mode]


    #############
    # Listeners #
    #############

    def _net_forward_mode_changed_cb(self, src):
        mode = uiutil.get_list_selection(self.widget("net-forward-mode"))

        fw_visible = mode not in ["open", "isolated", "hostdev"]
        is_hostdev = mode in ["hostdev"]

        uiutil.set_grid_row_visible(
            self.widget("net-forward-device"), fw_visible)
        uiutil.set_grid_row_visible(self.widget("net-hostdevs"), is_hostdev)

        self.widget("net-ipv4-expander").set_visible(not is_hostdev)
        self.widget("net-ipv6-expander").set_visible(not is_hostdev)
        self.widget("net-dns-expander").set_visible(not is_hostdev)

    def _net_dns_use_toggled_cb(self, src):
        custom = self.widget("net-dns-use-custom").get_active()
        self.widget("net-domain-name").set_sensitive(custom)

    def change_ipv4_enable(self, ignore):
        enabled = self.get_config_ipv4_enable()
        self.widget("net-ipv4-box").set_visible(enabled)
    def change_ipv6_enable(self, ignore):
        enabled = self.get_config_ipv6_enable()
        self.widget("net-ipv6-box").set_visible(enabled)

    def change_dhcpv4_enable(self, ignore):
        enabled = self.get_config_dhcpv4_enable()
        start = self.widget("net-dhcpv4-start")
        end = self.widget("net-dhcpv4-end")
        uiutil.set_grid_row_visible(start, enabled)
        uiutil.set_grid_row_visible(end, enabled)
    def change_dhcpv6_enable(self, ignore):
        enabled = self.get_config_dhcpv6_enable()
        start = self.widget("net-dhcpv6-start")
        end = self.widget("net-dhcpv6-end")
        uiutil.set_grid_row_visible(start, enabled)
        uiutil.set_grid_row_visible(end, enabled)

    def change_dhcpv4_start(self, src):
        start = self.get_config_dhcpv4_start()
        self.change_dhcpv4(src, start)
    def change_dhcpv4_end(self, src):
        end = self.get_config_dhcpv4_end()
        self.change_dhcpv4(src, end)
    def change_dhcpv4(self, src, addr):
        ip = self.get_config_ip4()
        if ip is None or addr is None:
            src.modify_bg(Gtk.StateType.NORMAL, _white)
            return

        if addr.version != 4 or not ip.overlaps(addr):
            src.modify_bg(Gtk.StateType.NORMAL, _red)
        else:
            src.modify_bg(Gtk.StateType.NORMAL, _green)

    def change_dhcpv6_start(self, src):
        start = self.get_config_dhcpv6_start()
        self.change_dhcpv6(src, start)
    def change_dhcpv6_end(self, src):
        end = self.get_config_dhcpv6_end()
        self.change_dhcpv6(src, end)
    def change_dhcpv6(self, src, addr):
        ip = self.get_config_ip6()
        if ip is None or addr is None:
            src.modify_bg(Gtk.StateType.NORMAL, _white)
            return

        if addr.version != 6 or not ip.overlaps(addr):
            src.modify_bg(Gtk.StateType.NORMAL, _red)
        else:
            src.modify_bg(Gtk.StateType.NORMAL, _green)


    def change_ipv4_network(self, src):
        ip = self.get_config_ip4()

        # No IP specified or invalid IP
        if ip is None or ip.version != 4:
            src.modify_bg(Gtk.StateType.NORMAL, _red)
            return

        valid_ip = (ip.num_addresses >= 8 and ip.is_private)
        start = int(ip.num_addresses // 2)
        end = int(ip.num_addresses - 2)

        src.modify_bg(Gtk.StateType.NORMAL, valid_ip and _green or _red)
        self.widget("net-dhcpv4-start").set_text(
            str(ip.network_address + start)
        )
        self.widget("net-dhcpv4-end").set_text(str(ip.network_address + end))

    def change_ipv6_network(self, src):
        ip = self.get_config_ip6()

        if ip is None or ip.version != 6:
            src.modify_bg(Gtk.StateType.NORMAL, _red)
            return

        valid_ip = (ip.num_addresses == 64 and ip.is_private)
        start = 256
        end = 512 - 1

        src.modify_bg(Gtk.StateType.NORMAL, valid_ip and _green or _red)
        self.widget("net-dhcpv6-start").set_text(
            str(ip.network_address + start)
        )
        self.widget("net-dhcpv6-end").set_text(str(ip.network_address + end))


    #########################
    # XML build and install #
    #########################

    def _validate(self, net):
        net.validate_generic_name(_("Network"), net.name)

        try:
            net.conn.networkLookupByName(net.name)
        except libvirt.libvirtError:
            return
        raise ValueError(_("Name '%s' already in use by another network." %
                         net.name))

    def _build_xmlobj_from_xmleditor(self):
        xml = self._xmleditor.get_xml()
        logging.debug("Using XML from xmleditor:\n%s", xml)
        return Network(self.conn.get_backend(), parsexml=xml)

    def _build_xmlobj_from_ui(self):
        net = Network(self.conn.get_backend())

        net.name = self.widget("net-name").get_text()
        net.domain_name = self.get_config_domain_name()

        dev, mode = self.get_config_forwarding()
        if mode:
            net.forward.mode = mode
            if mode == "open":
                net.forward.dev = None
            else:
                net.forward.dev = dev or None

        if net.forward.mode == "hostdev":
            net.forward.managed = "yes"
            pfobj = net.forward.pf.add_new()
            pfobj.dev = net.forward.dev
            net.forward.dev = None
            return net

        if self.get_config_ipv4_enable():
            ip = self.get_config_ip4()
            ipobj = net.ips.add_new()
            ipobj.address = str(ip.network_address + 1)
            ipobj.netmask = str(ip.netmask)

            if self.get_config_dhcpv4_enable():
                dhcpobj = ipobj.ranges.add_new()
                dhcpobj.start = str(
                    self.get_config_dhcpv4_start().network_address
                )
                dhcpobj.end = str(self.get_config_dhcpv4_end().network_address)

        if self.get_config_ipv6_enable():
            ip = self.get_config_ip6()
            ipobj = net.ips.add_new()
            ipobj.family = "ipv6"
            ipobj.address = str(ip.network_address + 1)
            ipobj.prefix = str(ip.prefixlen)

            if self.get_config_dhcpv6_enable():
                dhcpobj = ipobj.ranges.add_new()
                dhcpobj.start = str(
                    self.get_config_dhcpv6_start().network_address
                )
                dhcpobj.end = str(
                    self.get_config_dhcpv6_end().network_address
                )

        return net

    def _build_xmlobj(self, check_xmleditor):
        try:
            xmlobj = self._build_xmlobj_from_ui()
            if check_xmleditor and self._xmleditor.is_xml_selected():
                xmlobj = self._build_xmlobj_from_xmleditor()
            return xmlobj
        except Exception as e:
            self.err.show_err(_("Error building XML: %s") % str(e))

    def _finish_cb(self, error, details):
        self.reset_finish_cursor()

        if error:
            error = _("Error creating virtual network: %s") % str(error)
            self.err.show_err(error, details=details)
        else:
            self.conn.schedule_priority_tick(pollnet=True)
            self.close()

    def _async_net_create(self, asyncjob, net):
        ignore = asyncjob
        xml = net.get_xml()
        logging.debug("Creating virtual network '%s' with xml:\n%s",
                      net.name, xml)

        netobj = self.conn.get_backend().networkDefineXML(xml)
        try:
            netobj.create()
            netobj.setAutostart(True)
        except Exception:
            netobj.undefine()
            raise

    def finish(self, ignore):
        net = self._build_xmlobj(check_xmleditor=True)
        if not net:
            return

        try:
            self._validate(net)
        except Exception as e:
            return self.err.show_err(
                    _("Error validating network: %s") % e)

        self.set_finish_cursor()
        progWin = vmmAsyncJob(self._async_net_create, [net],
                              self._finish_cb, [],
                              _("Creating virtual network..."),
                              _("Creating the virtual network may take a "
                                "while..."),
                              self.topwin)
        progWin.run()



    ################
    # UI listeners #
    ################

    def _xmleditor_xml_requested_cb(self, src):
        xmlobj = self._build_xmlobj(check_xmleditor=False)
        self._xmleditor.set_xml(xmlobj and xmlobj.get_xml() or "")