# HG changeset patch
# User john.levon@xxxxxxx
# Date 1228922588 28800
# Node ID 5101328492a7092d4a3f376914b5ffd67b9fdf47
# Parent 1590180314624d0d919b61652b0474fd653b5078
Fossilize public virtinst.util API
Make virtinst.util API frozen, and move the utility functions into a
private namespace away from the prying eyes of API clients.
As a consequence, create a User class for checking privileges.
Signed-off-by: John Levon <john.levon@xxxxxxx>
diff --git a/virt-clone b/virt-clone
--- a/virt-clone
+++ b/virt-clone
@@ -29,6 +29,7 @@ import locale
import locale
import virtinst.cli as cli
from virtinst.cli import fail
+from virtinst.User import User
locale.setlocale(locale.LC_ALL, '')
gettext.bindtextdomain(virtinst.gettext_app, virtinst.gettext_dir)
@@ -184,9 +185,8 @@ def main():
logging.debug("start clone with HV " + options.connect)
- if options.connect is None or options.connect.lower()[0:3] == "xen":
- if not virtinst.util.privileged_user():
- fail(_("Must be root to clone Xen guests"))
+ if not User.current().has_priv(User.PRIV_CLONE, options.connect):
+ fail(_("Must be privileged to clone Xen guests"))
conn = cli.getConnection(options.connect)
design = clmgr.CloneDesign(connection=conn)
diff --git a/virtinst/CloneManager.py b/virtinst/CloneManager.py
--- a/virtinst/CloneManager.py
+++ b/virtinst/CloneManager.py
@@ -23,7 +23,7 @@ import libxml2
import libxml2
import logging
import urlgrabber.progress as progress
-import util
+import _util
import libvirt
import Guest
from VirtualDisk import VirtualDisk
@@ -263,7 +263,7 @@ class CloneDesign(object):
node[0].setContent(self._clone_uuid)
else:
while 1:
- uuid = util.uuidToString(util.randomUUID())
+ uuid = _util.uuidToString(_util.randomUUID())
if self._check_uuid(uuid) == True:
continue
else:
@@ -278,7 +278,7 @@ class CloneDesign(object):
node[0].setContent(self._clone_mac[i-1])
except Exception:
while 1:
- mac = util.randomMAC(typ)
+ mac = _util.randomMAC(typ)
dummy, msg = self._check_mac(mac)
if msg is not None:
continue
@@ -396,7 +396,7 @@ class CloneDesign(object):
for i in lst:
mode = os.stat(i)[stat.ST_MODE]
if stat.S_ISBLK(mode):
- size.append(util.blkdev_size(i))
+ size.append(_util.blkdev_size(i))
typ.append(False)
elif stat.S_ISREG(mode):
size.append(os.path.getsize(i))
@@ -450,7 +450,7 @@ class CloneDesign(object):
continue
mode = os.stat(i)[stat.ST_MODE]
if stat.S_ISBLK(mode):
- size.append(util.blkdev_size(i))
+ size.append(_util.blkdev_size(i))
typ.append(False)
elif stat.S_ISREG(mode):
size.append(os.path.getsize(i))
diff --git a/virtinst/DistroManager.py b/virtinst/DistroManager.py
--- a/virtinst/DistroManager.py
+++ b/virtinst/DistroManager.py
@@ -22,9 +22,10 @@
import logging
import os
-import util
+import _util
import Guest
from VirtualDisk import VirtualDisk
+from User import User
from virtinst import _virtinst as _
import virtinst
@@ -158,7 +159,7 @@ class DistroInstaller(Guest.Installer):
logging.debug("DistroInstaller location is a (poolname, volname)"
" tuple")
elif os.path.exists(os.path.abspath(val)) \
- and (not self.conn or not util.is_uri_remote(self.conn.getURI())):
+ and (not self.conn or not _util.is_uri_remote(self.conn.getURI())):
val = os.path.abspath(val)
logging.debug("DistroInstaller location is a local "
"file/path: %s" % val)
@@ -178,8 +179,8 @@ class DistroInstaller(Guest.Installer):
elif (val.startswith("http://";) or val.startswith("ftp://";) or
val.startswith("nfs:")):
logging.debug("DistroInstaller location is a network source.")
- elif self.conn and util.is_storage_capable(self.conn) and \
- util.is_uri_remote(self.conn.getURI()):
+ elif self.conn and _util.is_storage_capable(self.conn) and \
+ _util.is_uri_remote(self.conn.getURI()):
# If conn is specified, pass the path to a VirtualDisk object
# and see what comes back
try:
@@ -193,8 +194,9 @@ class DistroInstaller(Guest.Installer):
"or FTP network install source, or an existing "
"local file/device"))
- if val.startswith("nfs:") and not util.privileged_user():
- raise ValueError(_("NFS installations are only supported as root"))
+ if (val.startswith("nfs:") and not
+ User.current().has_priv(User.PRIV_NFS_MOUNT, self.conn.getURI())):
+ raise ValueError(_('Privilege is required for NFS installations'))
self._location = val
location = property(get_location, set_location)
diff --git a/virtinst/FullVirtGuest.py b/virtinst/FullVirtGuest.py
--- a/virtinst/FullVirtGuest.py
+++ b/virtinst/FullVirtGuest.py
@@ -20,7 +20,7 @@
# MA 02110-1301 USA.
import os
-import util
+import _util
import DistroManager
import logging
import time
@@ -38,7 +38,7 @@ class FullVirtGuest(Guest):
installer = DistroManager.DistroInstaller(type = type, os_type = "hvm")
Guest.__init__(self, type, connection, hypervisorURI, installer)
self.disknode = "hd"
- self.features = { "acpi": None, "pae": util.is_pae_capable(), "apic": None }
+ self.features = { "acpi": None, "pae": _util.is_pae_capable(), "apic": None }
if arch is None:
arch = platform.machine()
self.arch = arch
diff --git a/virtinst/Guest.py b/virtinst/Guest.py
--- a/virtinst/Guest.py
+++ b/virtinst/Guest.py
@@ -26,7 +26,7 @@ import re
import re
import libxml2
import urlgrabber.progress as progress
-import util
+import _util
import libvirt
import platform
import __builtin__
@@ -131,7 +131,7 @@ class VirtualNetworkInterface(VirtualDev
logging.warn("conflict_net: Failed to lookup domain %d" % name)
# get the Host's NIC MACaddress
- hostdevs = util.get_host_network_devices()
+ hostdevs = _util.get_host_network_devices()
if self.countMACaddr(vms) > 0:
return (True, _("The MAC address you entered is already in use by another active virtual machine."))
@@ -145,7 +145,7 @@ class VirtualNetworkInterface(VirtualDev
def setup(self, conn):
if self.macaddr is None:
while 1:
- self.macaddr = util.randomMAC(type=conn.getType().lower())
+ self.macaddr = _util.randomMAC(type=conn.getType().lower())
if self.is_conflict_net(conn)[1] is not None:
continue
else:
@@ -159,7 +159,7 @@ class VirtualNetworkInterface(VirtualDev
raise RuntimeError(msg)
if not self.bridge and self.type == "bridge":
- self.bridge = util.default_bridge()
+ self.bridge = _util.default_bridge()
def get_xml_config(self):
src_xml = ""
@@ -192,7 +192,7 @@ class VirtualNetworkInterface(VirtualDev
try:
for mac in ctx.xpathEval("/domain/devices/interface/mac"):
macaddr = mac.xpathEval("attribute::address")[0].content
- if macaddr and util.compareMAC(self.macaddr, macaddr) == 0:
+ if macaddr and _util.compareMAC(self.macaddr, macaddr) == 0:
count += 1
finally:
if ctx is not None:
@@ -250,7 +250,7 @@ class VirtualGraphics(object):
return self._keymap
def set_keymap(self, val):
if not val:
- val = util.default_keymap()
+ val = _util.default_keymap()
if not val or type(val) != type("string"):
raise ValueError, _("Keymap must be a string")
if len(val) > 16:
@@ -352,7 +352,7 @@ class Installer(object):
return '/var/tmp'
if self.type == "xen" and os.path.exists(XEN_SCRATCH):
return XEN_SCRATCH
- if util.privileged_user() and os.path.exists(LIBVIRT_SCRATCH):
+ if os.geteuid() == 0 and os.path.exists(LIBVIRT_SCRATCH):
return LIBVIRT_SCRATCH
else:
return os.path.expanduser("~/.virtinst/boot")
@@ -407,7 +407,7 @@ class Installer(object):
conn=None, kernel=None, bootdev=None):
osblob = ""
if not isinstall and not ishvm:
- return "<bootloader>%s</bootloader>" % util.pygrub_path(conn)
+ return "<bootloader>%s</bootloader>" % _util.pygrub_path(conn)
osblob = "<os>\n"
@@ -425,9 +425,9 @@ class Installer(object):
osblob += " <loader>%s</loader>\n" % loader
if isinstall and kernel and kernel["kernel"]:
- osblob += " <kernel>%s</kernel>\n" % util.xml_escape(kernel["kernel"])
- osblob += " <initrd>%s</initrd>\n" % util.xml_escape(kernel["initrd"])
- osblob += " <cmdline>%s</cmdline>\n" % util.xml_escape(kernel["extraargs"])
+ osblob += " <kernel>%s</kernel>\n" % _util.xml_escape(kernel["kernel"])
+ osblob += " <initrd>%s</initrd>\n" % _util.xml_escape(kernel["initrd"])
+ osblob += " <cmdline>%s</cmdline>\n" % _util.xml_escape(kernel["extraargs"])
elif bootdev is not None:
osblob += " <boot dev='%s'/>\n" % bootdev
@@ -466,7 +466,7 @@ class Installer(object):
@type L{Guest}
"""
- if util.is_uri_remote(guest.conn.getURI()):
+ if _util.is_uri_remote(guest.conn.getURI()):
# XXX: Use block peek for this?
return True
@@ -479,7 +479,7 @@ class Installer(object):
fd = os.open(guest.disks[0].path, os.O_RDONLY)
except OSError, (err, msg):
logging.debug("Failed to open guest disk: %s" % msg)
- if err == errno.EACCES and not util.privileged_user():
+ if err == errno.EACCES and os.geteuid() != 0:
return True # non root might not have access to block devices
else:
raise
@@ -624,7 +624,7 @@ class Guest(object):
def get_vcpus(self):
return self._vcpus
def set_vcpus(self, val):
- maxvcpus = util.get_max_vcpus(self.conn, self.type)
+ maxvcpus = _util.get_max_vcpus(self.conn, self.type)
if type(val) is not int or val < 1:
raise ValueError, _("Number of vcpus must be a postive integer.")
if val > maxvcpus:
@@ -642,7 +642,7 @@ class Guest(object):
if re.match("^[0-9,-]*$", val) is None:
raise ValueError, _("cpuset can only contain numeric, ',', or '-' characters")
- pcpus = util.get_phy_cpus(self.conn)
+ pcpus = _util.get_phy_cpus(self.conn)
for c in val.split(','):
if c.find('-') != -1:
(x, y) = c.split('-')
@@ -1061,7 +1061,7 @@ class Guest(object):
def _set_defaults(self):
if self.uuid is None:
while 1:
- self.uuid = util.uuidToString(util.randomUUID())
+ self.uuid = _util.uuidToString(_util.randomUUID())
try:
if self.conn.lookupByUUIDString(self.uuid) is not None:
continue
diff --git a/virtinst/ImageManager.py b/virtinst/ImageManager.py
--- a/virtinst/ImageManager.py
+++ b/virtinst/ImageManager.py
@@ -23,7 +23,7 @@ import CapabilitiesParser as Cap
import CapabilitiesParser as Cap
from VirtualDisk import VirtualDisk
import os
-import util
+import _util
from virtinst import _virtinst as _
class ImageInstallerException(Exception):
@@ -101,7 +101,7 @@ class ImageInstaller(Guest.Installer):
d = VirtualDisk(p, s,
device = device,
type = VirtualDisk.TYPE_FILE)
- if self.boot_caps.type == "xen" and util.is_blktap_capable():
+ if self.boot_caps.type == "xen" and _util.is_blktap_capable():
d.driver_name = VirtualDisk.DRIVER_TAP
d.target = m.target
@@ -127,9 +127,9 @@ class ImageInstaller(Guest.Installer):
if loader:
osblob += " <loader>%s</loader>\n" % loader
if self.boot_caps.kernel:
- osblob += " <kernel>%s</kernel>\n" % util.xml_escape(self._abspath(self.boot_caps.kernel))
- osblob += " <initrd>%s</initrd>\n" % util.xml_escape(self._abspath(self.boot_caps.initrd))
- osblob += " <cmdline>%s</cmdline>\n" % util.xml_escape(self.boot_caps.cmdline)
+ osblob += " <kernel>%s</kernel>\n" % _util.xml_escape(self._abspath(self.boot_caps.kernel))
+ osblob += " <initrd>%s</initrd>\n" % _util.xml_escape(self._abspath(self.boot_caps.initrd))
+ osblob += " <cmdline>%s</cmdline>\n" % _util.xml_escape(self.boot_caps.cmdline)
osblob += " </os>"
elif hvm:
if self.boot_caps.bootdev:
@@ -137,7 +137,7 @@ class ImageInstaller(Guest.Installer):
osblob += " </os>"
elif self.boot_caps.loader == "pygrub" or (self.boot_caps.loader is None and self.boot_caps.type == "xen"):
osblob += " </os>\n"
- osblob += " <bootloader>%s</bootloader>" % util.pygrub_path(conn)
+ osblob += " <bootloader>%s</bootloader>" % _util.pygrub_path(conn)
return osblob
diff --git a/virtinst/Storage.py b/virtinst/Storage.py
--- a/virtinst/Storage.py
+++ b/virtinst/Storage.py
@@ -52,7 +52,7 @@ import logging
import logging
from xml.sax.saxutils import escape
-import util
+import _util
from virtinst import _virtinst as _
DEFAULT_DEV_TARGET = "/dev"
@@ -112,7 +112,7 @@ class StorageObject(object):
def set_conn(self, val):
if not isinstance(val, libvirt.virConnect):
raise ValueError(_("'conn' must be a libvirt connection object."))
- if not util.is_storage_capable(val):
+ if not _util.is_storage_capable(val):
raise ValueError(_("Passed connection is not libvirt storage "
"capable"))
self._conn = val
@@ -278,7 +278,7 @@ class StoragePool(StorageObject):
self._source_path = None
if not uuid:
self._uuid = None
- self._random_uuid = util.uuidToString(util.randomUUID())
+ self._random_uuid = _util.uuidToString(_util.randomUUID())
# Properties used by all pools
def get_type(self):
@@ -705,7 +705,7 @@ class StorageVolume(StorageObject):
pool_object = StorageVolume.lookup_pool_by_name(pool_object=pool_object,
pool_name=pool_name,
conn=conn)
- return StoragePool.get_volume_for_pool(util.get_xml_path(pool_object.XMLDesc(0), "/pool/@type"))
+ return StoragePool.get_volume_for_pool(_util.get_xml_path(pool_object.XMLDesc(0), "/pool/@type"))
get_volume_for_pool = staticmethod(get_volume_for_pool)
def find_free_name(name, pool_object=None, pool_name=None, conn=None,
@@ -756,7 +756,7 @@ class StorageVolume(StorageObject):
if pool_name is not None and pool_object is None:
if conn is None:
raise ValueError(_("'conn' must be specified with 'pool_name'"))
- if not util.is_storage_capable(conn):
+ if not _util.is_storage_capable(conn):
raise ValueError(_("Connection does not support storage "
"management."))
try:
diff --git a/virtinst/VirtualDevice.py b/virtinst/VirtualDevice.py
--- a/virtinst/VirtualDevice.py
+++ b/virtinst/VirtualDevice.py
@@ -22,7 +22,7 @@ import libvirt
import libvirt
import CapabilitiesParser
-import util
+import _util
from virtinst import _virtinst as _
class VirtualDevice(object):
@@ -45,7 +45,7 @@ class VirtualDevice(object):
self.__remote = None
if self.conn:
- self.__remote = util.is_uri_remote(self.conn.getURI())
+ self.__remote = _util.is_uri_remote(self.conn.getURI())
self._caps = None
if self.conn:
diff --git a/virtinst/VirtualDisk.py b/virtinst/VirtualDisk.py
--- a/virtinst/VirtualDisk.py
+++ b/virtinst/VirtualDisk.py
@@ -24,7 +24,7 @@ import logging
import logging
import libvirt
-import util
+import _util
import Storage
from VirtualDevice import VirtualDevice
from virtinst import _virtinst as _
@@ -302,7 +302,7 @@ class VirtualDisk(VirtualDevice):
"('poolname', 'volname')"))
if not self.conn:
raise ValueError(_("'volName' requires a passed connection."))
- if not util.is_storage_capable(self.conn):
+ if not _util.is_storage_capable(self.conn):
raise ValueError(_("Connection does not support storage lookup."))
try:
pool = self.conn.storagePoolLookupByName(name_tuple[0])
@@ -321,7 +321,7 @@ class VirtualDisk(VirtualDevice):
def __check_if_path_managed(self):
vol = None
verr = None
- pool = util.lookup_pool_by_path(self.conn,
+ pool = _util.lookup_pool_by_path(self.conn,
os.path.dirname(self.path))
if pool:
try:
@@ -378,7 +378,7 @@ class VirtualDisk(VirtualDevice):
# if no obj: if remote, error
storage_capable = False
if self.conn:
- storage_capable = util.is_storage_capable(self.conn)
+ storage_capable = _util.is_storage_capable(self.conn)
if not storage_capable and self._is_remote():
raise ValueError, _("Connection doesn't support remote storage.")
@@ -528,7 +528,7 @@ class VirtualDisk(VirtualDevice):
elif self.path:
path = self.path
if path:
- path = util.xml_escape(path)
+ path = _util.xml_escape(path)
ret = " <disk type='%(type)s' device='%(device)s'>\n" % { "type": self.type, "device": self.device }
if not(self.driver_name is None):
diff --git a/virtinst/__init__.py b/virtinst/__init__.py
--- a/virtinst/__init__.py
+++ b/virtinst/__init__.py
@@ -34,3 +34,4 @@ from LiveCDInstaller import LiveCDInstal
from LiveCDInstaller import LiveCDInstaller
from ImageManager import ImageInstaller
from CloneManager import CloneDesign
+from User import User
diff --git a/virtinst/util.py b/virtinst/_util.py
copy from virtinst/util.py
copy to virtinst/_util.py
--- a/virtinst/_util.py
+++ b/virtinst/_util.py
@@ -1,5 +1,3 @@
-#
-# Utility functions used for guest installation
#
# Copyright 2006 Red Hat, Inc.
# Jeremy Katz <katzj@xxxxxxxxxx>
@@ -18,246 +16,17 @@
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
# MA 02110-1301 USA.
+#
-import platform
-import random
-import os.path
-import re
-import libxml2
-import logging
-from sys import stderr
+#
+# Internal utility functions. These do NOT form part of the API and must
+# not be used by clients.
+#
-import libvirt
-from virtinst import _virtinst as _
-from virtinst import CapabilitiesParser
+import os
+from virtinst import util
-KEYBOARD_DIR = "/etc/sysconfig/keyboard"
-XORG_CONF = "/etc/X11/xorg.conf"
-
-def default_route():
- route_file = "/proc/net/route"
- d = file(route_file)
-
- defn = 0
- for line in d.xreadlines():
- info = line.split()
- if (len(info) != 11): # 11 = typical num of fields in the file
- print >> stderr, _("Invalid line length while parsing %s.") %(route_file)
- print >> stderr, _("Defaulting bridge to xenbr%d") % (defn)
- break
- try:
- route = int(info[1],16)
- if route == 0:
- return info[0]
- except ValueError:
- continue
- return None
-
-# Legacy for compat only.
-def default_bridge():
- rt = default_route()
- if rt is None:
- defn = None
- else:
- defn = int(rt[-1])
-
- if defn is None:
- return "xenbr0"
- else:
- return "xenbr%d"%(defn)
-
-def default_network(conn):
- dev = default_route()
-
- if dev is not None and not is_uri_remote(conn.getURI()):
- # New style peth0 == phys dev, eth0 == bridge, eth0 == default route
- if os.path.exists("/sys/class/net/%s/bridge" % dev):
- return ["bridge", dev]
-
- # Old style, peth0 == phys dev, eth0 == netloop, xenbr0 == bridge,
- # vif0.0 == netloop enslaved, eth0 == default route
- defn = int(dev[-1])
- if os.path.exists("/sys/class/net/peth%d/brport" % defn) and \
- os.path.exists("/sys/class/net/xenbr%d/bridge" % defn):
- return ["bridge", "xenbr%d" % defn]
-
- return ["network", "default"]
-
-def default_connection():
- if os.path.exists("/var/lib/xend") and os.path.exists("/proc/xen"):
- return "xen"
- elif os.path.exists("/usr/bin/qemu") or \
- os.path.exists("/usr/bin/qemu-kvm") or \
- os.path.exists("/usr/bin/kvm") or \
- os.path.exists("/usr/bin/xenner"):
- if privileged_user():
- return "qemu:///system"
- else:
- return "qemu:///session"
- return None
-
-def get_cpu_flags():
- f = open("/proc/cpuinfo")
- lines = f.readlines()
- f.close()
- for line in lines:
- if not line.startswith("flags"):
- continue
- # get the actual flags
- flags = line[:-1].split(":", 1)[1]
- # and split them
- flst = flags.split(" ")
- return flst
- return []
-
-def is_pae_capable():
- """Determine if a machine is PAE capable or not."""
- flags = get_cpu_flags()
- if "pae" in flags:
- return True
- return False
-
-def is_hvm_capable():
- """Determine if a machine is HVM capable or not."""
-
- caps = ""
- if os.path.exists("/sys/hypervisor/properties/capabilities"):
- caps = open("/sys/hypervisor/properties/capabilities").read()
- if caps.find("hvm") != -1:
- return True
- return False
-
-def is_kqemu_capable():
- return os.path.exists("/dev/kqemu")
-
-def is_kvm_capable():
- return os.path.exists("/dev/kvm")
-
-def is_blktap_capable():
- #return os.path.exists("/dev/xen/blktapctrl")
- f = open("/proc/modules")
- lines = f.readlines()
- f.close()
- for line in lines:
- if line.startswith("blktap ") or line.startswith("xenblktap "):
- return True
- return False
-
-def get_default_arch():
- arch = os.uname()[4]
- if arch == "x86_64":
- return "x86_64"
- return "i686"
-
-# this function is directly from xend/server/netif.py and is thus
-# available under the LGPL,
-# Copyright 2004, 2005 Mike Wray <mike.wray@xxxxxx>
-# Copyright 2005 XenSource Ltd
-def randomMAC(type = "xen"):
- """Generate a random MAC address.
-
- 00-16-3E allocated to xensource
- 54-52-00 used by qemu/kvm
-
- The OUI list is available at http://standards.ieee.org/regauth/oui/oui.txt.
-
- The remaining 3 fields are random, with the first bit of the first
- random field set 0.
-
- >>> randomMAC().startswith("00:16:36")
- True
- >>> randomMAC("foobar").startswith("00:16:36")
- True
- >>> randomMAC("xen").startswith("00:16:36")
- True
- >>> randomMAC("qemu").startswith("54:52:00")
- True
-
- @return: MAC address string
- """
- ouis = { 'xen': [ 0x00, 0x16, 0x36 ], 'qemu': [ 0x54, 0x52, 0x00 ] }
-
- try:
- oui = ouis[type]
- except KeyError:
- oui = ouis['xen']
-
- mac = oui + [
- random.randint(0x00, 0x7f),
- random.randint(0x00, 0xff),
- random.randint(0x00, 0xff) ]
- return ':'.join(map(lambda x: "%02x" % x, mac))
-
-# the following three functions are from xend/uuid.py and are thus
-# available under the LGPL,
-# Copyright 2005 Mike Wray <mike.wray@xxxxxx>
-# Copyright 2005 XenSource Ltd
-def randomUUID():
- """Generate a random UUID."""
-
- return [ random.randint(0, 255) for dummy in range(0, 16) ]
-
-def uuidToString(u):
- return "-".join(["%02x" * 4, "%02x" * 2, "%02x" * 2, "%02x" * 2,
- "%02x" * 6]) % tuple(u)
-
-def uuidFromString(s):
- s = s.replace('-', '')
- return [ int(s[i : i + 2], 16) for i in range(0, 32, 2) ]
-
-# the following function quotes from python2.5/uuid.py
-def get_host_network_devices():
- device = []
- for dirname in ['', '/sbin/', '/usr/sbin']:
- executable = os.path.join(dirname, "ifconfig")
- if not os.path.exists(executable):
- continue
- try:
- cmd = 'LC_ALL=C %s -a 2>/dev/null' % (executable)
- pipe = os.popen(cmd)
- except IOError:
- continue
- for line in pipe:
- if line.find("encap:Ethernet") > 0:
- words = line.lower().split()
- for i in range(len(words)):
- if words[i] == "hwaddr":
- device.append(words)
- return device
-
-def get_max_vcpus(conn, type=None):
- """@conn libvirt connection to poll for max possible vcpus
- @type optional guest type (kvm, etc.)"""
- if type is None:
- type = conn.getType()
- try:
- m = conn.getMaxVcpus(type.lower())
- except libvirt.libvirtError:
- m = 32
- return m
-
-def get_phy_cpus(conn):
- """Get number of physical CPUs."""
- hostinfo = conn.getInfo()
- pcpus = hostinfo[4] * hostinfo[5] * hostinfo[6] * hostinfo[7]
- return pcpus
-
-def system(cmd):
- st = os.system(cmd)
- if os.WIFEXITED(st) and os.WEXITSTATUS(st) != 0:
- raise OSError("Failed to run %s, exited with %d" %
- (cmd, os.WEXITSTATUS(st)))
-
-def xml_escape(str):
- """Replaces chars ' " < > & with xml safe counterparts"""
- str = str.replace("&", "&")
- str = str.replace("'", "'")
- str = str.replace("\"", """)
- str = str.replace("<", "<")
- str = str.replace(">", ">")
- return str
-
def blkdev_size(path):
"""Return the size of the block device. We can't use os.stat() as
that returns zero on many platforms."""
@@ -267,255 +36,33 @@ def blkdev_size(path):
os.close(fd)
return size
-def compareMAC(p, q):
- """Compare two MAC addresses"""
- pa = p.split(":")
- qa = q.split(":")
-
- if len(pa) != len(qa):
- if p > q:
- return 1
- else:
- return -1
-
- for i in xrange(len(pa)):
- n = int(pa[i], 0x10) - int(qa[i], 0x10)
- if n > 0:
- return 1
- elif n < 0:
- return -1
- return 0
-
-def _xorg_keymap():
- """Look in /etc/X11/xorg.conf for the host machine's keymap, and attempt to
- map it to a keymap supported by qemu"""
-
- kt = None
- try:
- f = open(XORG_CONF, "r")
- except IOError, e:
- logging.debug('Could not open "%s": %s ' % (XORG_CONF, str(e)))
- else:
- keymap_re = re.compile(r'\s*Option\s+"XkbLayout"\s+"(?P<kt>[a-z-]+)"')
- for line in f:
- m = keymap_re.match(line)
- if m:
- kt = m.group('kt')
- break
- else:
- logging.debug("Didn't find keymap in '%s'!" % XORG_CONF)
- f.close()
- return kt
-
-def default_keymap():
- """Look in /etc/sysconfig for the host machine's keymap, and attempt to
- map it to a keymap supported by qemu"""
-
- # Set keymap to same as hosts
- import keytable
- keymap = "en-us"
- kt = None
- try:
- f = open(KEYBOARD_DIR, "r")
- except IOError, e:
- logging.debug('Could not open "/etc/sysconfig/keyboard" ' + str(e))
- kt = _xorg_keymap()
- else:
- while 1:
- s = f.readline()
- if s == "":
- break
- if re.search("KEYTABLE", s) != None or \
- (re.search("KEYBOARD", s) != None and
- re.search("KEYBOARDTYPE", s) == None):
- if s.count('"'):
- delim = '"'
- elif s.count('='):
- delim = '='
- else:
- continue
- kt = s.split(delim)[1].strip()
- f.close()
-
- if kt and keytable.keytable.has_key(kt.lower()):
- keymap = keytable.keytable[kt]
- else:
- logging.debug("Didn't find keymap '%s' in keytable!" % kt)
- return keymap
-
-def pygrub_path(conn=None):
- """
- Return the pygrub path for the current host, or connection if
- available.
- """
- # FIXME: This should be removed/deprecated when capabilities are
- # fixed to provide bootloader info
- if conn:
- cap = CapabilitiesParser.parse(conn.getCapabilities())
- if (cap.host.arch == "i86pc"):
- return "/usr/lib/xen/bin/pygrub"
- else:
- return "/usr/bin/pygrub"
-
- if platform.system() == "SunOS":
- return "/usr/lib/xen/bin/pygrub"
- return "/usr/bin/pygrub"
-
-def uri_split(uri):
- """
- Parse a libvirt hypervisor uri into it's individual parts
- @returns: tuple of the form (scheme (ex. 'qemu', 'xen+ssh'), username,
- hostname, path (ex. '/system'), query,
- fragment)
- """
- def splitnetloc(url, start=0):
- for c in '/?#': # the order is important!
- delim = url.find(c, start)
- if delim >= 0:
- break
- else:
- delim = len(url)
- return url[start:delim], url[delim:]
-
- username = netloc = query = fragment = ''
- i = uri.find(":")
- if i > 0:
- scheme, uri = uri[:i].lower(), uri[i+1:]
- if uri[:2] == '//':
- netloc, uri = splitnetloc(uri, 2)
- offset = netloc.find("@")
- if offset > 0:
- username = netloc[0:offset]
- netloc = netloc[offset+1:]
- if '#' in uri:
- uri, fragment = uri.split('#', 1)
- if '?' in uri:
- uri, query = uri.split('?', 1)
- else:
- scheme = uri.lower()
- return scheme, username, netloc, uri, query, fragment
-
-
-def is_uri_remote(uri):
- try:
- split_uri = uri_split(uri)
- netloc = split_uri[2]
-
- if netloc == "":
- return False
- return True
- except Exception, e:
- logging.exception("Error parsing URI in is_remote: %s" % e)
- return True
-
-def get_uri_hostname(uri):
- try:
- split_uri = uri_split(uri)
- netloc = split_uri[2]
-
- if netloc != "":
- return netloc
- except Exception, e:
- logging.warning("Cannot parse URI %s: %s" % (uri, str(e)))
- return "localhost"
-
-def get_uri_transport(uri):
- try:
- split_uri = uri_split(uri)
- scheme = split_uri[0]
- username = split_uri[1]
-
- if scheme:
- offset = scheme.index("+")
- if offset > 0:
- return [scheme[offset+1:], username]
- except:
- pass
- return [None, None]
-
-def get_uri_driver(uri):
- try:
- split_uri = uri_split(uri)
- scheme = split_uri[0]
-
- if scheme:
- offset = scheme.find("+")
- if offset > 0:
- return scheme[:offset]
- return scheme
- except Exception:
- pass
- return "xen"
-
-def is_storage_capable(conn):
- """check if virConnectPtr passed has storage API support"""
- if not conn:
- return False
- if not isinstance(conn, libvirt.virConnect):
- raise ValueError(_("'conn' must be a virConnect instance."))
- try:
- if not dir(conn).count("listStoragePools"):
- return False
- conn.listStoragePools()
- except libvirt.libvirtError, e:
- if e.get_error_code() == libvirt.VIR_ERR_RPC or \
- e.get_error_code() == libvirt.VIR_ERR_NO_SUPPORT:
- return False
- return True
-
-def get_xml_path(xml, path):
- """return the xpath from the passed xml"""
- doc = None
- ctx = None
- result = None
- try:
- doc = libxml2.parseDoc(xml)
- ctx = doc.xpathNewContext()
- ret = ctx.xpathEval(path)
- val = None
- if ret != None:
- if type(ret) == list:
- if len(ret) == 1:
- val = ret[0].content
- else:
- val = ret
- result = val
- finally:
- if doc:
- doc.freeDoc()
- if ctx:
- ctx.xpathFreeContext()
- return result
-
-def lookup_pool_by_path(conn, path):
- """
- Return the first pool with matching matching target path.
- return the first we find, active or inactive. This iterates over
- all pools and dumps their xml, so it is NOT quick.
- @return virStoragePool object if found, None otherwise
- """
- if not is_storage_capable(conn):
- return None
-
- pool_list = conn.listStoragePools() + conn.listDefinedStoragePools()
- for name in pool_list:
- pool = conn.storagePoolLookupByName(name)
- xml_path = get_xml_path(pool.XMLDesc(0), "/pool/target/path")
- if os.path.abspath(xml_path) == path:
- return pool
- return None
-
-def privileged_user():
- """
- Return true if the user is privileged enough. On Linux, this
- equates to being root. On Solaris, it's more complicated, so we
- just assume we're OK.
- """
- return os.uname()[0] == 'SunOS' or os.geteuid() == 0
-
-def _test():
- import doctest
- doctest.testmod()
-
-if __name__ == "__main__":
- _test()
+#
+# These functions accidentally ended up in the API under virtinst.util
+#
+default_route = util.default_route
+default_bridge = util.default_bridge
+default_network = util.default_network
+default_connection = util.default_connection
+get_cpu_flags = util.get_cpu_flags
+is_pae_capable = util.is_pae_capable
+is_blktap_capable = util.is_blktap_capable
+get_default_arch = util.get_default_arch
+randomMAC = util.randomMAC
+randomUUID = util.randomUUID
+uuidToString = util.uuidToString
+uuidFromString = util.uuidFromString
+get_host_network_devices = util.get_host_network_devices
+get_max_vcpus = util.get_max_vcpus
+get_phy_cpus = util.get_phy_cpus
+xml_escape = util.xml_escape
+compareMAC = util.compareMAC
+default_keymap = util.default_keymap
+pygrub_path = util.pygrub_path
+uri_split = util.uri_split
+is_uri_remote = util.is_uri_remote
+get_uri_hostname = util.get_uri_hostname
+get_uri_transport = util.get_uri_transport
+get_uri_driver = util.get_uri_driver
+is_storage_capable = util.is_storage_capable
+get_xml_path = util.get_xml_path
+lookup_pool_by_path = util.lookup_pool_by_path
diff --git a/virtinst/cli.py b/virtinst/cli.py
--- a/virtinst/cli.py
+++ b/virtinst/cli.py
@@ -26,9 +26,9 @@ from optparse import OptionValueError, O
from optparse import OptionValueError, OptionParser
import libvirt
-import util
+import _util
from virtinst import CapabilitiesParser, VirtualNetworkInterface, \
- VirtualGraphics, VirtualAudio
+ VirtualGraphics, VirtualAudio, User
from virtinst import _virtinst as _
MIN_RAM = 64
@@ -117,9 +117,8 @@ def nice_exit():
sys.exit(0)
def getConnection(connect):
- if connect and connect.lower()[0:3] == "xen":
- if not util.privileged_user():
- fail(_("Must be root to create Xen guests"))
+ if not User.current().has_priv(User.PRIV_CREATE_DOMAIN, connect):
+ fail(_("Must be root to create Xen guests"))
if connect is None:
fail(_("Could not find usable default libvirt connection."))
@@ -298,8 +297,8 @@ def digest_networks(conn, macs, bridges,
# With just one mac, create a default network if one is not
# specified.
if len(macs) == 1 and len(networks) == 0:
- if util.privileged_user():
- net = util.default_network(conn)
+ if User.current().has_priv(User.PRIV_CREATE_NETWORK, conn.getURI()):
+ net = _util.default_network(conn)
networks.append(net[0] + ":" + net[1])
else:
networks.append("user")
@@ -316,8 +315,8 @@ def digest_networks(conn, macs, bridges,
# Create extra networks up to the number of nics requested
if len(macs) < nics:
for dummy in range(len(macs),nics):
- if util.privileged_user():
- net = util.default_network(conn)
+ if User.current().has_priv(User.PRIV_CREATE_NETWORK, conn.getURI()):
+ net = _util.default_network(conn)
networks.append(net[0] + ":" + net[1])
else:
networks.append("user")
diff --git a/virtinst/util.py b/virtinst/util.py
--- a/virtinst/util.py
+++ b/virtinst/util.py
@@ -18,6 +18,14 @@
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
# MA 02110-1301 USA.
+
+#
+# WARNING: the contents of this file, somewhat unfortunately, are legacy
+# API. No incompatible changes are allowed to this file, and no new
+# code should be added here (utility functions live in _util.py).
+# Clients of virtinst shouldn't use these functions: if you think you
+# need to, tell us why.
+#
import platform
import random
@@ -30,6 +38,7 @@ import libvirt
import libvirt
from virtinst import _virtinst as _
from virtinst import CapabilitiesParser
+from virtinst import User
KEYBOARD_DIR = "/etc/sysconfig/keyboard"
@@ -54,7 +63,6 @@ def default_route():
continue
return None
-# Legacy for compat only.
def default_bridge():
rt = default_route()
if rt is None:
@@ -91,7 +99,7 @@ def default_connection():
os.path.exists("/usr/bin/qemu-kvm") or \
os.path.exists("/usr/bin/kvm") or \
os.path.exists("/usr/bin/xenner"):
- if privileged_user():
+ if User.current().has_priv(User.PRIV_QEMU_SYSTEM):
return "qemu:///system"
else:
return "qemu:///session"
@@ -258,15 +266,6 @@ def xml_escape(str):
str = str.replace(">", ">")
return str
-def blkdev_size(path):
- """Return the size of the block device. We can't use os.stat() as
- that returns zero on many platforms."""
- fd = os.open(path, os.O_RDONLY)
- # os.SEEK_END is not present on all systems
- size = os.lseek(fd, 0, 2)
- os.close(fd)
- return size
-
def compareMAC(p, q):
"""Compare two MAC addresses"""
pa = p.split(":")
@@ -505,14 +504,6 @@ def lookup_pool_by_path(conn, path):
return pool
return None
-def privileged_user():
- """
- Return true if the user is privileged enough. On Linux, this
- equates to being root. On Solaris, it's more complicated, so we
- just assume we're OK.
- """
- return os.uname()[0] == 'SunOS' or os.geteuid() == 0
-
def _test():
import doctest
doctest.testmod()
_______________________________________________
et-mgmt-tools mailing list
et-mgmt-tools@xxxxxxxxxx
https://www.redhat.com/mailman/listinfo/et-mgmt-tools
