The pyvix2 module wraps the VIX API with Python code. The VIX documentation breaks down the API into several sections, based on the type of handle.
"""PyVIX 2 -- Implements VIX API via :mod:`ctypes`.
"""
from __future__ import print_function
import sys
import os.path
import time
from ctypes import *
import logging
import pprint
logger= logging.getLogger( 'pyvix2' )
The C-language headers provide the following
from pyvix2.enum import *
def VIX_ERROR_CODE(err):
return ((err) & 0xFFFF)
def VIX_SUCCEEDED(err):
return (VIX_OK == (err))
def VIX_FAILED(err):
return (VIX_OK != (err))
Since we’re using ctypes to wrap a .dll or .so, we need to locate the appropriate module.
We’ll define a function and then simply execute that function to set a global, _vix, that is the external C-language module we’ll be using.
windows = sys.platform.lower().startswith('win')
def findVix():
"""Locate the VIX installation, return the relevant module."""
if windows:
import _winreg
reg = _winreg.ConnectRegistry(None, _winreg.HKEY_LOCAL_MACHINE)
with _winreg.OpenKey(reg,
r'SOFTWARE\VMware, Inc.\VMware VIX'
#r'SOFTWARE\VMware, Inc.\VMware Server'
) as key:
vix_dir = _winreg.QueryValueEx(
key, 'InstallPath'
)[0]
reg.Close()
logger.debug( "VIX Directory %r", vix_dir )
# Read vixwrapper-product-config.txt to get proper "variant"
# Workstation 8 variant, also true for player 4
variant= "Workstation-8.0.0-and-vSphere-5.0.0"
path= os.path.join( os.path.normpath(vix_dir), variant, "32bit" )
lib= "vix"
# Change directory to assure DLL's all found
here= os.getcwd()
os.chdir( path )
logger.info( os.path.join(path, lib) )
vix= cdll.LoadLibrary( os.path.join(path, lib) )
os.chdir( here )
return vix
else:
raise NotImplemented( "Sorry, Windows-only for now." )
path= "/usr/lib/vmware/libvixAllProducts.so"
return cdll.LoadLibrary( path )
_vix = findVix()
logger.info( "VMWare: %s", _vix._name )
We’ll define a customized exception.
class Error(Exception):
"""Error encountered performing a VIX request."""
def __init__(self,errorCode):
self.vix = _vix
self.errorCode = errorCode
_vix.Vix_GetErrorText.restype = c_char_p
self.args = [ _vix.Vix_GetErrorText(self.errorCode, None) ]
These are generic features of all handles.
class Handle( object ):
"""General features of all VIX Handles."""
log= logging.getLogger( 'pyvix2.handle' )
def __init__( self, handle=VIX_INVALID_HANDLE ):
self._vix= _vix
self.handle= handle
self.log.debug( "%s %s", self.__class__.__name__, "acquire" )
def get_error_text( self ):
_vix.Vix_GetErrorText.restype = c_char_p
return _vix.Vix_GetErrorText(self.errorCode, None)
def get_property( self, property_id=VIX_PROPERTY_VM_TOOLS_STATE, prop_type=c_int ):
value = prop_type()
err = self._vix.Vix_GetProperties( self.handle,
property_id,
byref(value),
VIX_PROPERTY_NONE )
if err != VIX_OK:
raise Error(err)
return value
def __del__( self ):
self.log.debug( "%s %s", self.__class__.__name__, "release" )
self._vix.Vix_ReleaseHandle(self.handle)
A Host is a handle that references the VMware service provider.
Sometimes, it’s necessary to override the default service provider setting.
python
host= Host()
self.service_provider=pyvix2.enum.VIX_SERVICEPROVIDER_VMWARE_WORKSTATION
The options are
class Host( Handle ):
"""VixHost API: Connect, Disconnect, FindItems and OpenVM."""
def __init__( self ):
super( Host, self ).__init__()
self.service_provider=VIX_SERVICEPROVIDER_DEFAULT
def connect(self, hostname = None, hostport = 0, username = None, password = None):
self.handle = VIX_INVALID_HANDLE
_vix.VixHost_Connect.argtypes = [c_int, c_int, c_char_p, c_int, c_char_p, c_char_p, c_int, c_int, c_void_p, c_void_p ]
jobHandle = Job( _vix.VixHost_Connect(VIX_API_VERSION,
self.service_provider,
hostname,
hostport,
username,
password,
0,
VIX_INVALID_HANDLE,
None, None,
) )
result= jobHandle.wait( )
logger.debug( "%s %s return %d", self.__class__.__name__, "connect", result.value )
self.handle= jobHandle.get_property(VIX_PROPERTY_JOB_RESULT_HANDLE, c_long )
def disconnect( self ):
_vix.VixHost_Disconnect(self.handle)
self.handle = VIX_INVALID_HANDLE
def find_items( self ):
running = []
FINDFUNC = CFUNCTYPE(None, c_void_p, c_int, c_void_p, c_void_p)
def py_finder( jobHandle, eventType, moreEventInfo, clientData ):
if eventType == VIX_EVENTTYPE_JOB_COMPLETED:
#logger.debug( "VIX_EVENTTYPE_JOB_COMPLETED" )
pass
elif eventType == VIX_EVENTTYPE_FIND_ITEM:
name= Handle( moreEventInfo ).get_property( VIX_PROPERTY_FOUND_ITEM_LOCATION, c_char_p ).value
running.append( name )
elif eventType == VIX_EVENTTYPE_JOB_PROGRESS:
#logger.debug( "VIX_EVENTTYPE_JOB_PROGRESS" )
pass
else:
logger.warning( "eventType %r?", eventType )
finder_func = FINDFUNC(py_finder)
jobHandle = Job( _vix.VixHost_FindItems(self.handle,
VIX_FIND_RUNNING_VMS,
VIX_INVALID_HANDLE, # searchCriteria
-1, # timeout
finder_func, None, # Callback and Arg
) )
result= jobHandle.wait( )
logger.debug( "%s %s return %d", self.__class__.__name__, "find_items", result.value )
return running
def openVM( self, vmxFilePathName ):
jobHandle= Job( _vix.VixHost_OpenVM( self.handle,
vmxFilePathName,
VIX_VMOPEN_NORMAL,
VIX_INVALID_HANDLE,
None, None,
) )
result= jobHandle.wait( )
logger.debug( "%s %s return %d", self.__class__.__name__, "openVM", result.value )
vmHandle= VM( jobHandle.get_property( VIX_PROPERTY_JOB_RESULT_HANDLE ) )
return vmHandle
Many VMware methods are asynchronous. They return a Job handle that reflects the work being done by the VMware service provider.
A Job handle provides status information while the job is running. It also provides results when the job completes.
class Job( Handle ):
"""VixJob API: check completion, get error, get nth property,
get num properties, wait."""
def check_completion( self ):
status= c_int()
err= _vix.VixJob_CheckCompletion(self.handle, byref(status))
if err != VIX_OK:
raise Error(err)
return status
def get_error( self ):
err= _vix.VixJob_GetError(self.handle)
return err
def get_nth_properties( self, n, property_id, prop_type=c_int ):
result= prop_type()
err = _vix.VixJob_GetNthProperties(self.handle,
n,
property_id,
byref( result ),
VIX_PROPERTY_NONE)
if err != VIX_OK:
raise Error(err)
return result
def get_num_properties( self, property_id ):
num= _vix.VixJob_GetNumProperties( self.handle, property_id )
return num
def wait( self, property_id=VIX_PROPERTY_JOB_RESULT_ERROR_CODE, prop_type=c_int ):
property = prop_type()
err = _vix.VixJob_Wait(self.handle,
property_id, byref(property),
VIX_PROPERTY_NONE)
if err != VIX_OK:
raise Error(err)
return property
A PropertyList is a handle.
class PropertyList( Handle ):
"""VixPropertList API.
Not Implemented.
"""
A Snapshot is a handle to a snapshot made of a VM.
class Snapshot( Handle ):
"""VixSnapshot API."""
def get_child( self ):
"""Given a snapshot handle, return the Nth child snapshot."""
def get_num_children( self ):
"""Given a snapshot handle, return the number of children."""
def get_parent( self ):
"""Given a snapshot handle, return the parent snapshot."""
A VM is a handle to the virtual machine. It’s the .vmx file that contains the image.
The VM has a rather complex API. We’ll decompose it into sections. The first part is power-on, power-off, and the like.
Tools_State = {
VIX_TOOLSSTATE_RUNNING: "running",
VIX_TOOLSSTATE_UNKNOWN: "unknown",
VIX_TOOLSSTATE_NOT_INSTALLED: "not installed",
}
class VM( Handle ):
"""VixVM API."""
def power_on( self ):
jobHandle = Job( _vix.VixVM_PowerOn(self.handle,
VIX_VMPOWEROP_LAUNCH_GUI, # VIX_VMPOWEROP_NORMAL,
VIX_INVALID_HANDLE, None, None,
) )
result= jobHandle.wait( )
logger.debug( "%s %s return %d", self.__class__.__name__, "power_on", result.value )
return result
def wait_for_tools_in_guest( self, timeout=120 ):
jobHandle = Job( _vix.VixVM_WaitForToolsInGuest(self.handle,
timeout, # timeoutInSeconds
None, # callbackProc
None, # clientData
) )
result= jobHandle.wait( )
logger.debug( "%s %s return %d", self.__class__.__name__, "wait_for_tools_in_guest", result.value )
return result
@property
def tools_state( self ):
self._ts= self.get_property( property_id=VIX_PROPERTY_VM_TOOLS_STATE )
logger.info( "VIX_PROPERTY_VM_TOOLS_STATE: %s", Tools_State[self._ts.value] )
return self._ts
@property
def tools_running( self ):
return self.tools_state.value == VIX_TOOLSSTATE_RUNNING
@property
def tools_unknown( self ):
return self.tools_state.value == VIX_TOOLSSTATE_UNKNOWN
@property
def tools_unknown( self ):
return self.tools_state.value == VIX_TOOLSSTATE_NOT_INSTALLED
def power_off( self ):
if self.tools_running:
option= VIX_VMPOWEROP_FROM_GUEST
else:
option= VIX_VMPOWEROP_NORMAL
jobHandle = Job( _vix.VixVM_PowerOff(self.handle,
option,
VIX_INVALID_HANDLE, None, None,
) )
result= jobHandle.wait( )
logger.debug( "%s %s return %d", self.__class__.__name__, "power_off", result.value )
return result
These VM methods are related to login and logout. They require tools. They’re a precondition for many other methods.
def login( self, username, password ):
#The option VIX_LOGIN_IN_GUEST_REQUIRE_INTERACTIVE_ENVIRONMENT
#should be used to ensure that the functions
#VixVM_RunProgramInGuest work correctly.
jobHandle = Job( _vix.VixVM_LoginInGuest(self.handle,
username, password,
0,
VIX_INVALID_HANDLE,
None, None ) )
result= jobHandle.wait( )
logger.debug( "%s %s return %d", self.__class__.__name__, "login", result.value )
return result
def logout( self ):
jobHandle = Job( _vix.VixVM_LogoutFromGuest(self.handle,
None, None ) )
result= jobHandle.wait( )
logger.debug( "%s %s return %d", self.__class__.__name__, "logout", result.value )
return result
These VM methods are related to files and directories. They require tools as well as a login.
def file_copy_to_guest( self, hostPathName, guestPathName ):
jobHandle = Job( _vix.VixVM_CopyFileFromHostToGuest(self.handle,
hostPathName,
guestPathName,
0,
VIX_INVALID_HANDLE,
None, None ) )
result= jobHandle.wait( )
logger.debug( "%s %s return %d", self.__class__.__name__, "file_copy_to_guest", result.value )
return result
def file_copy_from_guest( self, guestPathName, hostPathName ):
jobHandle = Job( _vix.VixVM_CopyFileFromGuestToHost(self.handle,
guestPathName,
hostPathName,
0,
VIX_INVALID_HANDLE,
None, None ) )
result= jobHandle.wait( )
logger.debug( "%s %s return %d", self.__class__.__name__, "file_copy_from_guest", result.value )
return result
def directory_create( self, guestPathName ):
jobHandle = Job( _vix.VixVM_CreateDirectoryInGuest(self.handle,
guestPathName,
VIX_INVALID_HANDLE,
None, None ) )
result= jobHandle.wait( )
return result
def tempfile_create( self ):
jobHandle = Job( _vix.VixVM_CreateTempFileInGuest(self.handle,
0,
VIX_INVALID_HANDLE,
None, None ) )
result= jobHandle.wait( )
return result
def directory_delete( self, guestPathName ):
jobHandle = Job( _vix.VixVM_DeleteDirectoryInGuest(self.handle,
guestPathName,
VIX_INVALID_HANDLE,
None, None ) )
result= jobHandle.wait( )
return result
def file_delete( self, guestPathName ):
jobHandle = Job( _vix.VixVM_DeleteFileInGuest(self.handle,
guestPathName,
VIX_INVALID_HANDLE,
None, None ) )
result= jobHandle.wait( )
return result
def directory_exists( self, guestPathName ):
jobHandle = Job( _vix.VixVM_DirectoryExistsInGuest(self.handle,
guestPathName,
VIX_INVALID_HANDLE,
None, None ) )
result= jobHandle.wait( )
return result
def file_exists( self, guestPathName ):
jobHandle = Job( _vix.VixVM_FileExistsInGuest(self.handle,
guestPathName,
VIX_INVALID_HANDLE,
None, None ) )
result= jobHandle.wait( )
return result
def directory_list( self, guestPathName ):
jobHandle = Job( _vix.VixVM_ListDirectoryInGuest(self.handle,
guestPathName,
VIX_INVALID_HANDLE,
None, None ) )
result= jobHandle.wait( )
logger.debug( "%s %s return %d", self.__class__.__name__, "directory_list", result.value )
num= jobHandle.get_num_properties(VIX_PROPERTY_JOB_RESULT_ITEM_NAME)
logger.debug( "Files: %d", num )
files= []
for i in xrange(num):
name= jobHandle.get_nth_properties( i, VIX_PROPERTY_JOB_RESULT_ITEM_NAME, c_char_p ).value
flags= jobHandle.get_nth_properties( i, VIX_PROPERTY_JOB_RESULT_FILE_FLAGS, c_int ).value
mod_time= jobHandle.get_nth_properties( i,VIX_PROPERTY_JOB_RESULT_FILE_MOD_TIME, c_long ).value
#start_string= time.asctime( time.localtime(start_time) )
files.append( (name, flags, mod_time) )
return files
def file_rename( self, oldname, newname ):
jobHandle = Job( _vix.VixVM_RenameFileInGuest(self.handle,
oldname,
newname,
0,
VIX_INVALID_HANDLE,
None, None ) )
result= jobHandle.wait( )
return result
These VM methods are related to processes. They require tools as well as a login.
def process_kill( self, pid ):
jobHandle = Job( _vix.VixVM_KillProcessInGuest(self.handle,
pid,
VIX_INVALID_HANDLE,
None, None ) )
result= jobHandle.wait( )
return result
def process_list( self ):
jobHandle = Job( _vix.VixVM_ListProcessesInGuest(self.handle,
0,
None, None ) )
result= jobHandle.wait( )
logger.debug( "%s %s return %d", self.__class__.__name__, "process_list", result.value )
num= jobHandle.get_num_properties(VIX_PROPERTY_JOB_RESULT_ITEM_NAME)
logger.debug( "Processes: %d", num )
procs= []
for i in xrange(num):
name= jobHandle.get_nth_properties( i, VIX_PROPERTY_JOB_RESULT_ITEM_NAME, c_char_p ).value
pid= jobHandle.get_nth_properties( i, VIX_PROPERTY_JOB_RESULT_PROCESS_ID, c_int ).value
owner= jobHandle.get_nth_properties( i, VIX_PROPERTY_JOB_RESULT_PROCESS_OWNER, c_char_p ).value
command= jobHandle.get_nth_properties( i, VIX_PROPERTY_JOB_RESULT_PROCESS_COMMAND, c_char_p ).value
debugged= jobHandle.get_nth_properties( i, VIX_PROPERTY_JOB_RESULT_PROCESS_BEING_DEBUGGED, c_int ).value
start_time= jobHandle.get_nth_properties( i,VIX_PROPERTY_JOB_RESULT_PROCESS_START_TIME, c_long ).value
#start_string= time.asctime( time.localtime(start_time) )
procs.append( (pid, owner, name, start_time, command) )
return procs
def program_run( self, guestProgramName, commandLineArgs, nowait=False ):
jobHandle = Job( _vix.VixVM_RunProgramInGuest(self.handle,
guestProgramName, commandLineArgs,
VIX_RUNPROGRAM_RETURN_IMMEDIATELY if nowait else 0,
VIX_INVALID_HANDLE,
None, None ) )
result= jobHandle.wait( )
logger.debug( "%s %s return %d", self.__class__.__name__, "program_run", result.value )
pid= jobHandle.get_property( VIX_PROPERTY_JOB_RESULT_PROCESS_ID, c_long ).value
if nowait:
et= None
exit= None
else:
et= jobHandle.get_property( VIX_PROPERTY_JOB_RESULT_GUEST_PROGRAM_ELAPSED_TIME, c_long ).value
exit= jobHandle.get_property( VIX_PROPERTY_JOB_RESULT_GUEST_PROGRAM_EXIT_CODE, c_int ).value
return pid, et, exit
def script_run( self, interpreter, scriptText, nowait=False):
jobHandle = Job( _vix.VixVM_RunScriptInGuest(self.handle,
interpreter, scriptText,
VIX_RUNPROGRAM_RETURN_IMMEDIATELY if nowait else 0,
VIX_INVALID_HANDLE,
None, None ) )
result= jobHandle.wait( )
logger.debug( "%s %s return %d", self.__class__.__name__, "script_run", result.value )
pid= jobHandle.get_property( VIX_PROPERTY_JOB_RESULT_PROCESS_ID, c_long ).value
if nowait:
et= None
exit= None
else:
et= jobHandle.get_property( VIX_PROPERTY_JOB_RESULT_GUEST_PROGRAM_ELAPSED_TIME, c_long ).value
exit= jobHandle.get_property( VIX_PROPERTY_JOB_RESULT_GUEST_PROGRAM_EXIT_CODE, c_int ).value
return pid, et, exit
These VM methods are related to snapshots.
# Snapshot-related VM methods
def snapshot( self, name, description ):
"""Save virtual machine state as a snapshot object and return a handle to it."""
jobHandle= Job( _vix.VixVM_CreateSnapshot(
self.handle,
name, description,
VIX_SNAPSHOT_INCLUDE_MEMORY,
VIX_INVALID_HANDLE,
None, None
) )
result= jobHandle.wait( )
logger.debug( "%s %s return %d", self.__class__.__name__, "snapshot", result.value )
snapshotHandle= jobHandle.get_property( VIX_PROPERTY_JOB_RESULT_HANDLE, c_long )
return Snapshot( snapshotHandle )
def snapshot_get( self ):
"""Return a handle to the current active snapshot."""
snapshotHandle= c_long()
err= _vix.VixVM_GetCurrentSnapshot(
self.handle,
byref(snapshotHandle)
)
if err != VIX_OK:
raise Error(err)
logger.debug( "%s %s return %d", self.__class__.__name__, "snapshot_get", err )
return Snapshot( snapshotHandle )
def snapshot_get_byname( self, name ):
"""Return a handle to the snapshot of a given unique name."""
snapshotHandle= c_long()
err= _vix.VixVM_GetNamedSnapshot(
self.handle,
name,
byref(snapshotHandle)
)
if err != VIX_OK:
raise Error(err)
logger.debug( "%s %s return %d", self.__class__.__name__, "snapshot_get_byname", err )
return Snapshot( snapshotHandle )
def snapshot_num_roots( self ):
"""Return the number of root-based snapshots (on VMware Server always 0 or 1)."""
count= c_int()
err= _vix.VixVM_GetNumRootSnapshots(
self.handle,
byref(count)
)
if err != VIX_OK:
raise Error(err)
logger.debug( "%s %s return %d", self.__class__.__name__, "snapshot_num_roots", err )
return count
def snapshot_get_root( self, index ):
"""Return the Nth root-based snapshot (on VMware Server always 0)."""
snapshotHandle= c_long()
err= _vix.VixVM_GetRootSnapshot(
self.handle,
index,
byref(snapshotHandle)
)
if err != VIX_OK:
raise Error(err)
logger.debug( "%s %s return %d", self.__class__.__name__, "snapshot_get_root", err )
return Snapshot( snapshotHandle )
def snapshot_remove( self, snapshot ):
"""Given a handle, delete that snapshot, and optionally its children."""
jobHandle = Job( _vix.VixVM_RemoveSnapshot(self.handle,
snapshot.handle,
0, # options
None, None ) )
result= jobHandle.wait( )
logger.debug( "%s %s return %d", self.__class__.__name__, "snapshot_remove", result.value )
return result
def snapshot_revert( self, snapshot ):
"""Restore a virtual machine to its state as of the specified snapshot."""
_vix.VixVM_RevertToSnapshot
jobHandle = Job( _vix.VixVM_RevertToSnapshot(self.handle,
shapshot.handle,
0, # options
VIX_INVALID_HANDLE,
None, None ) )
result= jobHandle.wait( )
logger.debug( "%s %s return %d", self.__class__.__name__, "snapshot_revert", result.value )
return result