diff --git a/pyxs/_internal.py b/pyxs/_internal.py index f8fa8d5..bfde2a9 100644 --- a/pyxs/_internal.py +++ b/pyxs/_internal.py @@ -13,6 +13,7 @@ __all__ = ["Event", "Op", "Packet"] +import os import struct from collections import namedtuple @@ -21,7 +22,7 @@ #: Operations supported by XenStore. Operations = Op = namedtuple("Operations", [ - "DEBUG", + "CONTROL", "DIRECTORY", "READ", "GET_PERMS", @@ -67,12 +68,17 @@ class Packet(namedtuple("_Packet", "op rq_id tx_id size payload")): def __new__(cls, op, payload, rq_id=None, tx_id=None): # Checking restrictions: - # a) payload is limited to 4096 bytes. - if len(payload) > 4096: - raise InvalidPayload(payload) - # b) operation requested is present in ``xsd_sockmsg_type``. + # a) operation requested is present in ``xsd_sockmsg_type``. if op not in Op: raise InvalidOperation(op) + # This is to code around http://lists.xenproject.org/archives/html/win-pv-devel/2016-02/msg00005.html + if os.name == "nt" and op == Op.READ and payload is None: + payload = "" + + # b) payload is limited to 4096 bytes. + if len(payload) > 4096: + raise InvalidPayload(payload) + return super(Packet, cls).__new__(cls, op, rq_id or 0, tx_id or 0, len(payload), payload) diff --git a/pyxs/client.py b/pyxs/client.py index 3ff1f40..929cb4d 100644 --- a/pyxs/client.py +++ b/pyxs/client.py @@ -24,10 +24,11 @@ import threading import time import posixpath +import os from collections import deque from ._internal import Event, Packet, Op -from .connection import UnixSocketConnection, XenBusConnection +from .connection import UnixSocketConnection, XenBusConnection, XenBusConnectionWinWINPV, XenBusConnectionWinGPLPV from .exceptions import UnexpectedPacket, PyXSError from .helpers import validate_path, validate_watch_path, validate_perms, \ dict_merge, force_unicode, error @@ -85,11 +86,23 @@ def __init__(self, unix_socket_path=None, socket_timeout=None, xen_bus_path=None, connection=None, transaction=None): if connection: self.connection = connection - elif unix_socket_path or not xen_bus_path: - self.connection = UnixSocketConnection( - unix_socket_path, socket_timeout=socket_timeout) + elif os.name in ["posix"]: + if xen_bus_path or not unix_socket_path: + self.connection = XenBusConnection(xen_bus_path) + else: + self.connection = UnixSocketConnection( + unix_socket_path, socket_timeout=socket_timeout) + elif os.name in ["nt"]: + # There are two windows pv driver projects, examine wmi + # classes to try and determine which one we want to use. + import wmi + try: + wmi.WMI(moniker="//./root/wmi", find_classes=False).XenProjectXenStoreBase() + self.connection = XenBusConnectionWinWINPV() + except AttributeError: + self.connection = XenBusConnectionWinGPLPV() else: - self.connection = XenBusConnection(xen_bus_path) + raise ConnectionError("no connection to xenstore available") self.tx_id = 0 self.tx_lock = threading.Lock() @@ -213,7 +226,7 @@ def ls(self, path): :param str path: path to list. """ payload = self.execute_command(Op.DIRECTORY, path) - return [] if payload is "" else payload.split("\x00") + return [] if payload is "" else map(lambda x : x.split('/')[-1], payload.split("\x00")) def get_permissions(self, path): """Returns a list of permissions for a given `path`, see @@ -441,7 +454,7 @@ def wait(self, sleep=None): # Executing a noop, hopefuly we'll get some events queued # in the meantime. Note: I know it sucks, but it seems like # there's no other way ... - self.client.execute_command(Op.DEBUG, "") + self.client.execute_command(Op.CONTROL, "noop") if sleep is not None: time.sleep(sleep) diff --git a/pyxs/connection.py b/pyxs/connection.py index a112409..d683a19 100644 --- a/pyxs/connection.py +++ b/pyxs/connection.py @@ -12,19 +12,38 @@ from __future__ import absolute_import, unicode_literals -__all__ = ["UnixSocketConnection", "XenBusConnection"] +__all__ = ["UnixSocketConnection", "XenBusConnection", "XenBusConnectionWinWINPV", "XenBusConnectionWinGPLPV"] +import logging import errno import os import platform import socket import sys +from time import sleep +from ._internal import Packet, Op +sys.coinit_flags = 0 + +if os.name in ["nt"]: + import wmi + import ctypes + from ctypes.wintypes import HANDLE + from ctypes.wintypes import BOOL + from ctypes.wintypes import HWND + from ctypes.wintypes import DWORD + from ctypes.wintypes import WORD + from ctypes.wintypes import LONG + from ctypes.wintypes import ULONG + from ctypes.wintypes import LPCSTR + from ctypes.wintypes import HKEY + from ctypes.wintypes import BYTE + sys.coinit_flags = 0 if sys.version_info[0] is not 3: bytes, str = str, unicode -from .exceptions import ConnectionError -from .helpers import writeall, readall +from .exceptions import ConnectionError, WindowsDriverError, PyXSError +from .helpers import writeall, readall, osnmopen, osnmclose, osnmread from ._internal import Packet @@ -51,7 +70,7 @@ def disconnect(self, silent=True): return try: - os.close(self.fd) + osnmclose(self.fd) except OSError as e: if not silent: raise ConnectionError(e.args) @@ -105,7 +124,7 @@ def recv(self): # blocking unless any new data appears, so we have to check # size value, before reading. payload = ("" if size is 0 else - os.read(self.fd, size).decode("utf-8")) + osnmread(self.fd, size).decode("utf-8")) return Packet(op, payload, rq_id, tx_id) @@ -148,6 +167,7 @@ def connect(self): else: self.fd = os.dup(sock.fileno()) + class XenBusConnection(FileDescriptorConnection): """XenStore connection through XenBus. @@ -164,7 +184,7 @@ def __init__(self, path=None): system = platform.system() if system == "Linux": - path = "/proc/xen/xenbus" + path = "/dev/xen/xenbus" if os.path.exists("/dev/xen/xenbus") else "/proc/xen/xenbus" elif system == "NetBSD": path = "/kern/xen/xenbus" else: @@ -180,7 +200,270 @@ def connect(self): return try: - self.fd = os.open(self.path, os.O_RDWR) + self.fd = osnmopen(self.path, os.O_RDWR) except OSError as e: raise ConnectionError("Error while opening {0!r}: {1}" .format(self.path, e.args)) + + +_wmiSession = None + +class XenBusConnectionWinWINPV(FileDescriptorConnection): + session = None + response_packet = None + + + def __init__(self): + pass + + + def __copy__(self): + return self.__class__(self.path) + + + def connect(self, retry=0): + global _wmiSession + + # Create a WMI Session + try: + if not _wmiSession or retry > 0: + _wmiSession = wmi.WMI(moniker="//./root/wmi", find_classes=False) + xenStoreBase = _wmiSession.XenProjectXenStoreBase()[0] + except Exception: # WMI can raise all sorts of exceptions + if retry < 20: + sleep(5) + self.connect(retry=(retry+1)) + return + else: + raise PyXSError, None, sys.exc_info()[2] + + try: + sessions = _wmiSession.query("select * from XenProjectXenStoreSession where InstanceName = 'Xen Interface\Session_PyxsSession_0'") + except Exception: + sessions = [] + + if len(sessions) <= 0: + session_name = "PyxsSession" + session_id = xenStoreBase.AddSession(Id=session_name)[0] + try: + sessions = _wmiSession.query("select * from XenProjectXenStoreSession where SessionId = {id}".format(id=session_id)) + except Exception: + sleep(0.5) + try: + sessions = _wmiSession.query("select * from XenProjectXenStoreSession where SessionId = {id}".format(id=session_id)) + except Exception: + raise PyXSError, None, sys.exc_info()[2] + + self.session = sessions.pop() + + + # Emulate sending the packet directly to the XenStore interface + # and store the result in response_packet + def send(self, packet): + global _wmiSession + + try: + if not _wmiSession or not self.session: + self.connect() + except wmi.x_wmi: + raise PyXSError, None, sys.exc_info()[2] + + remove_paths = lambda x : x.split('/')[-1] + + if packet.op == Op.READ: + #result = remove_paths(self.session.GetValue(packet.payload)[0]) + try: + result = self.session.GetValue(packet.payload)[0] + except wmi.x_wmi: + raise PyXSError, None, sys.exc_info()[2] + elif packet.op == Op.WRITE: + try: + payload = packet.payload.split('\x00', 1) + self.session.SetValue(payload[0], payload[1]) + except wmi.x_wmi: + raise PyXSError, None, sys.exc_info()[2] + result = "OK" + elif packet.op == Op.RM: + try: + self.session.RemoveValue(packet.payload)[0] + except wmi.x_wmi: + raise PyXSError, None, sys.exc_info()[2] + result = "OK" + elif packet.op == Op.DIRECTORY: + #result = map(remove_paths, self.session.GetChildren(packet.payload)[0].childNodes) + try: + result = self.session.GetChildren(packet.payload)[0].childNodes + result = "\x00".join(result) + except wmi.x_wmi: + raise PyXSError, None, sys.exc_info()[2] + else: + raise Exception("Unsupported XenStore Action ({x})".format(x=packet.op)) + self.response_packet = Packet(packet.op, result, packet.rq_id, packet.tx_id) + + + def recv(self): + return self.response_packet + + + def disconnect(self, silent=True): + self.session = None + + +_winDevicePath = None + +class XenBusConnectionWinGPLPV(FileDescriptorConnection): + def __init__(self): + global _winDevicePath + + # Once the windows device path is learned once reuse it otherwise + # ctypes.POINTER() for the same structure leaks memory. Although + # this can be reclaimed with ctypes._reset_cache() this is poking + # at the internals of ctypes which doesn't seem to be a good idea. + + if _winDevicePath: + self.path = _winDevicePath + + return + + # Determine self.path using some magic Windows code which is derived from + # http://pydoc.net/Python/pyserial/2.6/serial.tools.list_ports_windows/. + # The equivalent C from The GPLPV driver source can be found in get_xen_interface_path() of shutdownmon. + # - http://xenbits.xensource.com/ext/win-pvdrivers/file/896402519f15/shutdownmon/shutdownmon.c + + DIGCF_PRESENT = 2 + DIGCF_DEVICEINTERFACE = 16 + NULL = None + ERROR_SUCCESS = 0 + ERROR_INSUFFICIENT_BUFFER = 122 + ERROR_NO_MORE_ITEMS = 259 + + HDEVINFO = ctypes.c_void_p + PCTSTR = ctypes.c_char_p + CHAR = ctypes.c_char + PDWORD = ctypes.POINTER(DWORD) + LPDWORD = ctypes.POINTER(DWORD) + PULONG = ctypes.POINTER(ULONG) + + # Return code checkers + def ValidHandle(value, func, arguments): + if value == 0: + raise WindowsDriverError(str(ctypes.WinError())) + return value + + # Some structures used by the Windows API + class GUID(ctypes.Structure): + _fields_ = [ + ('Data1', DWORD), + ('Data2', WORD), + ('Data3', WORD), + ('Data4', BYTE*8), + ] + + def __str__(self): + return "{%08x-%04x-%04x-%s-%s}" % ( + self.Data1, + self.Data2, + self.Data3, + ''.join(["%02x" % d for d in self.Data4[:2]]), + ''.join(["%02x" % d for d in self.Data4[2:]]), + ) + + PGUID = ctypes.POINTER(GUID) + + class SP_DEVINFO_DATA(ctypes.Structure): + _fields_ = [ + ('cbSize', DWORD), + ('ClassGuid', GUID), + ('DevInst', DWORD), + ('Reserved', PULONG), + ] + + def __str__(self): + return "ClassGuid:%s DevInst:%s" % (self.ClassGuid, self.DevInst) + + PSP_DEVINFO_DATA = ctypes.POINTER(SP_DEVINFO_DATA) + + class SP_DEVICE_INTERFACE_DATA(ctypes.Structure): + _fields_ = [ + ('cbSize', DWORD), + ('InterfaceClassGuid', GUID), + ('Flags', DWORD), + ('Reserved', PULONG), + ] + + def __str__(self): + return "InterfaceClassGuid:%s Flags:%s" % (self.InterfaceClassGuid, self.Flags) + + PSP_DEVICE_INTERFACE_DATA = ctypes.POINTER(SP_DEVICE_INTERFACE_DATA) + PSP_DEVICE_INTERFACE_DETAIL_DATA = ctypes.c_void_p + + # Import the Windows APIs + setupapi = ctypes.windll.LoadLibrary("setupapi") + + SetupDiGetClassDevs = setupapi.SetupDiGetClassDevsA + SetupDiGetClassDevs.argtypes = [PGUID, PCTSTR, HWND, DWORD] + SetupDiGetClassDevs.restype = HDEVINFO + SetupDiGetClassDevs.errcheck = ValidHandle + + SetupDiEnumDeviceInterfaces = setupapi.SetupDiEnumDeviceInterfaces + SetupDiEnumDeviceInterfaces.argtypes = [HDEVINFO, PSP_DEVINFO_DATA, PGUID, DWORD, PSP_DEVICE_INTERFACE_DATA] + SetupDiEnumDeviceInterfaces.restype = BOOL + + SetupDiGetDeviceInterfaceDetail = setupapi.SetupDiGetDeviceInterfaceDetailA + SetupDiGetDeviceInterfaceDetail.argtypes = [HDEVINFO, PSP_DEVICE_INTERFACE_DATA, PSP_DEVICE_INTERFACE_DETAIL_DATA, DWORD, PDWORD, PSP_DEVINFO_DATA] + SetupDiGetDeviceInterfaceDetail.restype = BOOL + + SetupDiDestroyDeviceInfoList = setupapi.SetupDiDestroyDeviceInfoList + SetupDiDestroyDeviceInfoList.argtypes = [HDEVINFO] + SetupDiDestroyDeviceInfoList.restype = BOOL + + # Do stuff + GUID_XENBUS_IFACE = GUID(0x14ce175aL, 0x3ee2, 0x4fae, (BYTE*8)(0x92, 0x52, 0x0, 0xdb, 0xd8, 0x4f, 0x1, 0x8e)) + + handle = SetupDiGetClassDevs(ctypes.byref(GUID_XENBUS_IFACE), NULL, NULL, DIGCF_PRESENT | DIGCF_DEVICEINTERFACE); + + sdid = SP_DEVICE_INTERFACE_DATA() + sdid.cbSize = ctypes.sizeof(sdid) + if not SetupDiEnumDeviceInterfaces(handle, NULL, ctypes.byref(GUID_XENBUS_IFACE), 0, ctypes.byref(sdid)): + if ctypes.GetLastError() != ERROR_NO_MORE_ITEMS: + raise WindowsDriverError(str(ctypes.WinError())) + + buf_len = DWORD() + if not SetupDiGetDeviceInterfaceDetail(handle, ctypes.byref(sdid), NULL, 0, ctypes.byref(buf_len), NULL): + if ctypes.GetLastError() != ERROR_INSUFFICIENT_BUFFER: + raise WindowsDriverError(str(ctypes.WinError())) + + # We didn't know how big to make the structure until buf_len is assigned... + class SP_DEVICE_INTERFACE_DETAIL_DATA_A(ctypes.Structure): + _fields_ = [ + ('cbSize', DWORD), + ('DevicePath', CHAR*(buf_len.value - ctypes.sizeof(DWORD))), + ] + + def __str__(self): + return "DevicePath:%s" % (self.DevicePath,) + + sdidd = SP_DEVICE_INTERFACE_DETAIL_DATA_A() + sdidd.cbSize = ctypes.sizeof(ctypes.POINTER(SP_DEVICE_INTERFACE_DETAIL_DATA_A)) + if not SetupDiGetDeviceInterfaceDetail(handle, ctypes.byref(sdid), ctypes.byref(sdidd), buf_len, NULL, NULL): + raise WindowsDriverError(str(ctypes.WinError())) + self.path = ""+sdidd.DevicePath + + SetupDiDestroyDeviceInfoList(handle) + + _winDevicePath = self.path + + + def __copy__(self): + return self.__class__() + + + def connect(self): + if self.fd: + return + + try: + self.fd = osnmopen(self.path) + except Exception as e: + raise ConnectionError("Error while opening {0!r}: {1}" + .format(self.path, e.args)) diff --git a/pyxs/exceptions.py b/pyxs/exceptions.py index 1c1ad77..6c1afdc 100644 --- a/pyxs/exceptions.py +++ b/pyxs/exceptions.py @@ -75,3 +75,6 @@ class UnexpectedPacket(ConnectionError): ``op = Op.READ`` the incoming packet is expected to have ``op = Op.READ`` as well. """ + +class WindowsDriverError(PyXSError): + """Windows specific error when using native API to locate xenbus device""" diff --git a/pyxs/helpers.py b/pyxs/helpers.py index d2c7485..97df491 100644 --- a/pyxs/helpers.py +++ b/pyxs/helpers.py @@ -31,6 +31,52 @@ for code, message in errno.errorcode.items()) +if os.name in ["posix"]: + def osnmopen(path, *args): + return os.open(path, *args) + + def osnmclose(fd): + os.close(fd) + + def osnmwrite(fd, data): + return os.write(fd, data) + + def osnmread(fd, length): + return os.read(fd, length) + +elif os.name in ["nt"]: + from win32file import CreateFile, CloseHandle, ReadFile, WriteFile + from win32file import FILE_GENERIC_READ, FILE_GENERIC_WRITE, OPEN_EXISTING, FILE_ATTRIBUTE_NORMAL + + def osnmopen(path, *args): + # CreateFile(path, FILE_GENERIC_READ|FILE_GENERIC_WRITE, 0, NULL, OPEN_EXISTING, FILE_ATTRIBUTE_NORMAL, NULL); + # http://docs.activestate.com/activepython/2.7/pywin32/win32file__CreateFile_meth.html + # PyHANDLE = CreateFile(fileName, desiredAccess , shareMode , attributes , CreationDisposition , flagsAndAttributes , hTemplateFile ) + return CreateFile(path, FILE_GENERIC_READ|FILE_GENERIC_WRITE, 0, None, OPEN_EXISTING, FILE_ATTRIBUTE_NORMAL, None) + + def osnmclose(fd): + # http://docs.activestate.com/activepython/2.7/pywin32/win32file__CloseHandle_meth.html + # CloseHandle(handle) + CloseHandle(fd) + + def osnmread(fd, length): + # ReadFile(handle, buf, 1024, &bytes_read, NULL) + # http://docs.activestate.com/activepython/2.7/pywin32/win32file__ReadFile_meth.html + # (int, string) = ReadFile(hFile, buffer/bufSize , ol ) + (lread, data) = ReadFile(fd, length, None) + return data + + def osnmwrite(fd, data): + # WriteFile(handle, buf, sizeof(*msg) + msg->len, &bytes_written, NULL) + # http://docs.activestate.com/activepython/2.7/pywin32/win32file__WriteFile_meth.html + # int, int = WriteFile(hFile, data , ol ) + errCode, lwrite = WriteFile(fd, data, None) + return lwrite + +else: + raise NotImplemented("No operating system fd interface defined") + + def writeall(fd, data): """Writes a data string to the file descriptor. @@ -40,7 +86,7 @@ def writeall(fd, data): """ length = len(data) while length: - length -= os.write(fd, data[-length:]) + length -= osnmwrite(fd, data[-length:]) def readall(fd, length): @@ -51,7 +97,7 @@ def readall(fd, length): """ chunks = [] while length: - chunks.append(os.read(fd, length)) + chunks.append(osnmread(fd, length)) length -= len(chunks[-1]) else: return b"".join(chunks) diff --git a/tests.py b/tests.py index af8f462..a02acdf 100644 --- a/tests.py +++ b/tests.py @@ -32,7 +32,7 @@ def test_packet(): # b) invalid payload -- maximum size exceeded. with pytest.raises(InvalidPayload): - Packet(Op.DEBUG, "hello" * 4096, 0) + Packet(Op.CONTROL, "hello" * 4096, 0) # Helpers. @@ -169,13 +169,13 @@ def test_client_execute_command(): # a) arguments contain invalid characters. with pytest.raises(ValueError): - c.execute_command(Op.DEBUG, "\x07foo") + c.execute_command(Op.CONTROL, "\x07foo") # b) command validator fails. - c.COMMAND_VALIDATORS[Op.DEBUG] = lambda *args: False + c.COMMAND_VALIDATORS[Op.CONTROL] = lambda *args: False with pytest.raises(ValueError): - c.execute_command(Op.DEBUG, "foo") - c.COMMAND_VALIDATORS.pop(Op.DEBUG) + c.execute_command(Op.CONTROL, "foo") + c.COMMAND_VALIDATORS.pop(Op.CONTROL) # c) ``Packet`` constructor fails. with pytest.raises(InvalidPayload): @@ -187,7 +187,7 @@ def test_client_execute_command(): _old_recv = c.connection.recv # e) XenStore returns a packet with invalid operation in the header. - c.connection.recv = lambda *args: Packet(Op.DEBUG, "boo") + c.connection.recv = lambda *args: Packet(Op.CONTROL, "boo") with pytest.raises(UnexpectedPacket): c.execute_command(Op.READ, "/foo/bar") c.connection.recv = _old_recv