One Hat Cyber Team

View File Name : connectauth.py

# Copyright (C) 2012-2013 Red Hat, Inc.
# Copyright (C) 2012 Cole Robinson <crobinso@redhat.com>
#
# This work is licensed under the GNU GPLv2 or later.
# See the COPYING file in the top-level directory.

import collections
import os
import re
import shutil
import time

from gi.repository import GLib
from gi.repository import Gio
from gi.repository import Gtk

import libvirt

from virtinst import log

from ..baseclass import vmmGObjectUI
from . import uiutil


def do_we_have_session():
    pid = os.getpid()

    ret = False
    try:  # pragma: no cover
        bus = Gio.bus_get_sync(Gio.BusType.SYSTEM, None)
        manager = Gio.DBusProxy.new_sync(bus, 0, None,
                        "org.freedesktop.login1",
                        "/org/freedesktop/login1",
                        "org.freedesktop.login1.Manager", None)

        # This raises an error exception
        out = manager.GetSessionByPID("(u)", pid)
        log.debug("Found login1 session=%s", out)
        ret = True
    except Exception:  # pragma: no cover
        log.exception("Failure talking to logind")

    return ret


class _vmmConnectAuth(vmmGObjectUI):
    def __init__(self, creds):
        vmmGObjectUI.__init__(self, "connectauth.ui", "connectauth")
        self.creds = creds
        self.topwin.set_title(_("Authentication required"))

        self.builder.connect_signals({
            "on_connectauth_cancel_clicked": self._cancel_cb,
            "on_connectauth_ok_clicked": self._ok_cb,
            "on_entry1_activate": self._entry_cb,
            "on_entry2_activate": self._entry_cb,
        })

        self.entry1 = self.widget("entry1")
        self.entry2 = self.widget("entry2")
        self._init_ui()

    def _cleanup(self):
        pass

    def _init_ui(self):
        uiutil.set_grid_row_visible(self.entry1, False)
        uiutil.set_grid_row_visible(self.entry2, False)

        for idx, cred in enumerate(self.creds):
            # Libvirt virConnectCredential
            credtype, prompt, _challenge, _defresult, _result = cred
            noecho = credtype in [
                    libvirt.VIR_CRED_PASSPHRASE,
                    libvirt.VIR_CRED_NOECHOPROMPT]
            if not prompt:  # pragma: no cover
                raise RuntimeError("No prompt for auth credtype=%s" % credtype)

            prompt += ": "
            label = self.widget("label%s" % (idx + 1))
            entry = self.widget("entry%s" % (idx + 1))
            uiutil.set_grid_row_visible(label, True)
            label.set_text(prompt)
            entry.set_visibility(not noecho)
            entry.get_accessible().set_name(prompt + " entry")

    def run(self):
        self.topwin.show()
        res = self.topwin.run()
        self.topwin.hide()

        if res != Gtk.ResponseType.OK:
            return -1

        self.creds[0][4] = self.entry1.get_text()
        if self.entry2.get_visible():
            self.creds[1][4] = self.entry2.get_text()
        return 0

    def _ok_cb(self, src):
        self.topwin.response(Gtk.ResponseType.OK)
    def _cancel_cb(self, src):
        self.topwin.response(Gtk.ResponseType.CANCEL)

    def _entry_cb(self, src):
        """
        If entry 1 activated and entry2 visible, jump to entry 2.
        Otherwise, click OK
        """
        if src == self.entry1 and self.entry2.is_visible():
            self.entry2.grab_focus()
            return
        self.topwin.response(Gtk.ResponseType.OK)


def creds_dialog(creds, cbdata):
    """
    Thread safe wrapper for libvirt openAuth user/pass callback
    """
    retipc = []

    def wrapper(creds, cbdata):
        try:
            _conn = cbdata
            dialogobj = _vmmConnectAuth(creds)
            ret = dialogobj.run()
            dialogobj.cleanup()
        except Exception:  # pragma: no cover
            log.exception("Error from creds dialog")
            ret = -1
        retipc.append(ret)

    GLib.idle_add(wrapper, creds, cbdata)

    while not retipc:
        time.sleep(.1)

    return retipc[0]


def connect_error(conn, errmsg, tb, warnconsole):
    """
    Format connection error message
    """
    errmsg = errmsg.strip(" \n")
    tb = tb.strip(" \n")
    hint = ""
    show_errmsg = True

    if conn.is_remote():
        log.debug("connect_error: conn transport=%s",
            conn.get_uri_transport())
        if re.search(r"nc: .* -- 'U'", tb):  # pragma: no cover
            hint += _("The remote host requires a version of netcat/nc "
                      "which supports the -U option.")
            show_errmsg = False
        elif (conn.get_uri_transport() == "ssh" and
                re.search(r"askpass", tb)):  # pragma: no cover

            hint += _("Configure SSH key access for the remote host, "
                      "or install an SSH askpass package locally.")
            show_errmsg = False
        else:
            hint += _("Verify that the 'libvirtd' daemon is running "
                      "on the remote host.")

    elif conn.is_xen():  # pragma: no cover
        hint += _("Verify that:\n"
                  " - A Xen host kernel was booted\n"
                  " - The Xen service has been started")

    else:
        if warnconsole:
            hint += _("Could not detect a local session: if you are "
                      "running virt-manager over ssh -X or VNC, you "
                      "may not be able to connect to libvirt as a "
                      "regular user. Try running as root.")
            show_errmsg = False
        elif re.search(r"libvirt-sock", tb):  # pragma: no cover
            hint += _("Verify that the 'libvirtd' daemon is running.")
            show_errmsg = False

    msg = _("Unable to connect to libvirt %s." % conn.get_uri())
    if show_errmsg:
        msg += "\n\n%s" % errmsg
    if hint:
        msg += "\n\n%s" % hint

    msg = msg.strip("\n")
    details = msg
    details += "\n\n"
    details += "Libvirt URI is: %s\n\n" % conn.get_uri()
    details += tb

    title = _("Virtual Machine Manager Connection Failure")

    ConnectError = collections.namedtuple("ConnectError",
            ["msg", "details", "title"])
    return ConnectError(msg, details, title)


##################################
# App first run connection setup #
##################################

def setup_first_uri(config, tryuri):
    # Add /usr/sbin to the path in case non-root user launches virt-manager
    libvirtd_installed = bool(shutil.which("libvirtd", path=os.environ['PATH'] + os.pathsep + "/usr/sbin"))
    if config.CLITestOptions.fake_no_libvirtd:
        libvirtd_installed = False

    if tryuri and libvirtd_installed:
        return

    # Manager fail message
    msg = ""
    if not libvirtd_installed:  # pragma: no cover
        msg += _("The libvirtd service does not appear to be installed. "
                 "Install and run the libvirtd service to manage "
                 "virtualization on this host.")

    if not tryuri or "qemu" not in tryuri:
        if msg:
            msg += "\n\n"  # pragma: no cover
        msg += _("Could not detect a default hypervisor. Make "
                "sure the appropriate QEMU/KVM virtualization "
                "packages are installed to manage virtualization "
                "on this host.")

    if msg:
        msg += "\n\n"
        msg += _("A virtualization connection can be manually "
                 "added via File->Add Connection")

    return msg or None