diff --git a/docs/source/sample.rst b/docs/source/sample.rst index 9b7d7bd..645ffe8 100644 --- a/docs/source/sample.rst +++ b/docs/source/sample.rst @@ -320,10 +320,13 @@ Ouput:: I AM LOADING -.. _wmi_request: +.. _wmi_samples: + +WMI +""" WMI requests -"""""""""""" +'''''''''''' .. literalinclude:: ..\..\samples\wmi\wmi_request.py @@ -332,6 +335,15 @@ Output .. literalinclude:: samples_output\wmi_wmi_request.txt +WMI Create Process +'''''''''''''''''' + +.. literalinclude:: ..\..\samples\wmi\create_process.py + +Output + +.. literalinclude:: samples_output\wmi_create_process.txt + .. _sample_com: diff --git a/docs/source/samples_output/wmi_create_process.txt b/docs/source/samples_output/wmi_create_process.txt new file mode 100644 index 0000000..58f02bc --- /dev/null +++ b/docs/source/samples_output/wmi_create_process.txt @@ -0,0 +1,14 @@ +(cmd) python wmi\create_process.py +WMI namespace is <> +Process class is +Method Create InParams is <> +Method Create InParams properties are <[u'CommandLine', u'CurrentDirectory', u'ProcessStartupInformation']> +Creating instance of inparam +InParam instance is <> +Setting +Executing method +OutParams is +Out params values are: [u'ProcessId', u'ReturnValue'] +Created process is +Waiting 1s +Killing the process diff --git a/docs/source/samples_output/wmi_wmi_request.txt b/docs/source/samples_output/wmi_wmi_request.txt index 53d055d..bfc741b 100644 --- a/docs/source/samples_output/wmi_wmi_request.txt +++ b/docs/source/samples_output/wmi_wmi_request.txt @@ -1,28 +1,30 @@ (cmd) python wmi\wmi_request.py -WMI requester is +WMI requester is Selecting * from 'Win32_Process' -They are <188> processes +They are <308> processes Looking for ourself via pid Some info about our process: * Name -> python.exe - * ProcessId -> 8144 - * OSName -> Microsoft Windows 10 Home|C:\WINDOWS|\Device\Harddisk0\Partition2 - * UserModeTime -> 1875000 - * WindowsVersion -> 10.0.16299 + * ProcessId -> 27144 + * OSName -> Microsoft Windows 10 Pro|C:\WINDOWS|\Device\Harddisk0\Partition2 + * UserModeTime -> 2812500 + * WindowsVersion -> 10.0.17134 * CommandLine -> C:\Python27\python.exe wmi\wmi_request.py :") -for vol in windows.system.wmi.select("Win32_LogicalDisk", ["Caption", "FileSystem", "FreeSpace"]): - print(" * " + str(vol)) +for vol in windows.system.wmi.query("select Caption,FileSystem,FreeSpace from Win32_LogicalDisk"): + # Filter out system-properties for the sample + print(" * " + str({k:v for k,v in vol.items() if not k.startswith("_")})) print("\n ==== Advanced use ====") print("Listing some namespaces:") @@ -33,7 +34,7 @@ for namespace in [ns for ns in windows.system.wmi.namespaces if "2" in ns]: security2 = windows.system.wmi["root\\SecurityCenter2"] print("Querying non-default namespace: {0}".format(security2)) print("Listing some available classes:") -for clsname in [x for x in security2.classes if x.endswith("Product")]: +for clsname in [x for x in security2.classes if x["__CLASS"].endswith("Product")]: print(" * {0}".format(clsname)) print("Listing :") diff --git a/windows/winobject/wmi.py b/windows/winobject/wmi.py index c32b60b..2da0b81 100644 --- a/windows/winobject/wmi.py +++ b/windows/winobject/wmi.py @@ -9,10 +9,12 @@ from ctypes.wintypes import * import windows.com import windows.generated_def as gdef from windows.generated_def.winstructs import * +from functools import partial # Common error check for all WMI COM interfaces # This 'just' add the corresponding 'WBEMSTATUS' to the hresult error code class WmiComInterface(object): + """Base class used for COM call error checking for WMI interfaces""" def errcheck(self, result, func, args): if result < 0: wmitag = gdef.WBEMSTATUS.mapper[result & 0xffffffff] @@ -25,9 +27,15 @@ WmiMethod = namedtuple("WmiMethod", ["inparam", "outparam"]) # https://docs.microsoft.com/en-us/windows/desktop/WmiSdk/calling-a-method class WmiObject(gdef.IWbemClassObject, WmiComInterface): - ## low level API + """The WmiObject (which wrap ``IWbemClassObject``) contains and manipulates both class definitions and class object instances. + Can be used as a mapping to access properties. + """ def get_variant(self, name): + """Retrieve the value of property ``name`` as a :class:`~windows.com.Variant` + + :return: :class:`~windows.com.Variant` + """ if not isinstance(name, basestring): nametype = type(name).__name__ raise TypeError("WmiObject attributes name must be str, not <{0}>".format(nametype)) @@ -36,9 +44,14 @@ class WmiObject(gdef.IWbemClassObject, WmiComInterface): return variant_res def get(self, name): + """Return the value of the property ``name``. The return value depends of the type of the property and can vary""" return self.get_variant(name).value def get_method(self, name): + """Return the information about the method ``name`` + + :returns: :class:`WmiMethod` + """ inpararm = type(self)() outpararm = type(self)() variant_res = windows.com.Variant() @@ -53,32 +66,54 @@ class WmiObject(gdef.IWbemClassObject, WmiComInterface): return self.Put(name, 0, variant, 0) def put(self, name, value): + """Set the property ``name`` to ``value``""" variant_value = windows.com.Variant(value) return self.put_variant(name, variant_value) def spawn_instance(self): + """Create a new object of the class represented by the current :class:`WmiObject` + + :returns: :class:`WmiObject` + """ instance = type(self)() self.SpawnInstance(0, instance) return instance @property def genus(self): + """The genus of the object. + + :returns: ``WBEM_GENUS_CLASS(0x1L)`` if the :class:`WmiObject` is a Class and + ``WBEM_GENUS_INSTANCE(0x2L)`` for instances and events + """ return gdef.tag_WBEM_GENUS_TYPE.mapper[self.get("__GENUS")] ## Higher level API - def get_properties(self): - # res = POINTER(SAFEARRAY)() + def get_properties(self, system_properties=False): + """Return the list of properties' names available for the current object. + If ``system_properties`` is ``False`` property names begining with ``_` are ignored. + + :returns: [:class:`str`] -- A list of string + .. note: + + About system properties: https://docs.microsoft.com/en-us/windows/desktop/wmisdk/wmi-system-properties + """ res = POINTER(windows.com.SafeArray)() x = ctypes.pointer(res) self.GetNames(None, 0, None, cast(x, POINTER(POINTER(gdef.SAFEARRAY)))) # need to free the safearray / unlock ? - return res[0].to_list(BSTR) + properties = [p for p in res[0].to_list(BSTR) if system_properties or (not p.startswith("_"))] + return properties - properties = property(get_properties) + properties = property(get_properties) #: The properties of the object (exclude system properties) # Make WmiObject a mapping object - keys = get_properties + + def keys(self): + """The properties of the object (include system properties)""" + return self.get_properties(system_properties=True) + __getitem__ = get __setitem__ = put @@ -100,9 +135,15 @@ class WmiObject(gdef.IWbemClassObject, WmiComInterface): class WmiEnumeration(gdef.IEnumWbemClassObject, WmiComInterface): - DEFAULT_TIMEOUT = gdef.WBEM_INFINITE + """Represent an enumeration of object that can be itered""" + DEFAULT_TIMEOUT = gdef.WBEM_INFINITE #: The default timeout def next(self, timeout=None): + """Return the next object in the enumeration with `timeout`. + + :raises: ``WindowsError(WBEM_S_TIMEDOUT)`` if timeout expire + :returns: :class:`WmiObject` + """ timeout = self.DEFAULT_TIMEOUT if timeout is None else timeout # For now the count is hardcoded to 1 obj = WmiObject() @@ -116,9 +157,11 @@ class WmiEnumeration(gdef.IEnumWbemClassObject, WmiComInterface): return obj def __iter__(self): + """Return an iterator with ``DEFAULT_TIMEOUT``""" return self.iter_timeout(self.DEFAULT_TIMEOUT) def iter_timeout(self, timeout=None): + """Return an iterator with a custom ``timeout``""" while True: obj = self.next(timeout) if obj is None: @@ -126,36 +169,46 @@ class WmiEnumeration(gdef.IEnumWbemClassObject, WmiComInterface): yield obj def all(self): + """Return all elements in the enumeration as a list + + :returns: [:class:`WmiObject`] - A list of :class:`WmiObject` + """ return list(self) # SqlAlchemy like :) class WmiCallResult(gdef.IWbemCallResult, WmiComInterface): + """The result of a WMI call/query. Real result value type depends of the context""" def __init__(self, result_type=None, namespace_name=None): self.result_type = result_type self.namespace_name = namespace_name def get_call_status(self, timeout=gdef.WBEM_INFINITE): + """The status of the call""" status = gdef.LONG() self.GetCallStatus(timeout, status) return WBEMSTATUS.mapper[status.value & 0xffffffff] def get_result_object(self, timeout=gdef.WBEM_INFINITE): + """The result as a :class:`WmiObject` (returned by :func:`WmiNamespace.exec_method`)""" result = WmiObject() self.GetResultObject(timeout, result) return result def get_result_string(self, timeout=gdef.WBEM_INFINITE): + """The result as a :class:`WmiObject` (returned by :func:`WmiNamespace.put_instance`)""" result = gdef.BSTR() self.GetResultString(timeout, result) return result def get_result_service(self, timeout=gdef.WBEM_INFINITE): + """The result as a :class:`WmiNamespace` (not used yet)""" result = WmiNamespace() self.GetResultServices(timeout, result) return result @property def result(self): + """The result of the correct type based on ``self.result_type``""" if self.result_type is None: raise ValueError("Cannot call with no result_type") return getattr(self, "get_result_" + self.result_type)() @@ -167,19 +220,22 @@ class WmiLocator(gdef.IWbemLocator, WmiComInterface): # !TEST CODE class WmiNamespace(gdef.IWbemServices, WmiComInterface): - r"""An object to perform wmi request to ``a given namespace``""" + r"""An object to perform wmi request to a given ``namespace``""" #CLSID_WbemAdministrativeLocator_IID = windows.com.IID.from_string('CB8555CC-9128-11D1-AD9B-00C04FD8FDFF') WbemLocator_CLSID = windows.com.IID.from_string('4590F811-1D3A-11D0-891F-00AA004B2E24') DEFAULT_ENUM_FLAGS = (gdef.WBEM_FLAG_RETURN_IMMEDIATELY | - WBEM_FLAG_FORWARD_ONLY) + WBEM_FLAG_FORWARD_ONLY) #: The defauls flags used for enumeration. ``(WBEM_FLAG_RETURN_IMMEDIATELY | WBEM_FLAG_FORWARD_ONLY)`` - def __init__(self, namespace, *args, **kwargs): + def __init__(self, namespace): self.name = namespace @classmethod def connect(cls, namespace, user=None, password=None): + """Connect to ``namespace`` using ``user`` and ``password`` for authentification if given + + :return: :class:`WmiNamespace` - The connected :class:`WmiNamespace`""" # this method assert com is initialised self = cls(namespace) # IWbemServices subclass locator = WmiLocator() @@ -189,54 +245,112 @@ class WmiNamespace(gdef.IWbemServices, WmiComInterface): return self def query(self, query): - """TODO: doc""" + """Return the list of :class:`WmiObject` matching ``query``. + + This API is the `simple one`, if you need timeout or complexe feature see :func:`exec_query` + + :return: [:class:`WmiObject`] - A list of :class:`WmiObject` + """ return list(self.exec_query(query)) + def select(self, clsname, deep=True): + """Return the list of :class:`WmiObject` that are instance of ``clsname``. Deep has the same meaning as in :func:`create_instance_enum`. + + This API is the `simple one`, if you need timeout or complexe feature see :func:`create_instance_enum` + + :return: [:class:`WmiObject`] - A list of :class:`WmiObject` + """ + return list(self.create_instance_enum(clsname, deep=deep)) + + def exec_query(self, query, flags=DEFAULT_ENUM_FLAGS, ctx=None): - """TODO:DOC: Default flags are: WBEM_FLAG_FORWARD_ONLY | WBEM_FLAG_RETURN_IMMEDIATELY""" + """Execute a WQL query with custom flags and returns a ::class:`WmiEnumeration` that can be used to + iter the result with timeouts + + :returns: :class:`WmiEnumeration` + """ enumerator = WmiEnumeration() - # import pdb;pdb.set_trace() - execq = self.ExecQuery - # execq.func.errcheck = self.errck - # execq.func.restype = YoloCheck - # import pdb;pdb.set_trace() - # self.ExecQuery("WQL", query, flags, ctx, enumerator) - execq("WQL", query, flags, ctx, enumerator) + self.ExecQuery("WQL", query, flags, ctx, enumerator) return enumerator # Create friendly name for create_class_enum & create_instance_enum ? def create_class_enum(self, superclass, flags=DEFAULT_ENUM_FLAGS, deep=True): - flags |= gdef.WBEM_FLAG_DEEP if deep else gdef.WBEM_FLAG_SHALLOW + """Enumerate the classes in the ``namespace`` that match ``superclass``. + if ``superclass`` is None will enumerate all top-level class. ``deep`` allow to returns all subclasses + :returns: :class:`WmiEnumeration` + + .. note:: + + See https://docs.microsoft.com/en-us/windows/desktop/api/wbemcli/nf-wbemcli-iwbemservices-createclassenum + """ + + flags |= gdef.WBEM_FLAG_DEEP if deep else gdef.WBEM_FLAG_SHALLOW enumerator = WmiEnumeration() self.CreateClassEnum(superclass, flags, None, enumerator) return enumerator - # subclasses ? + @property + def classes(self): + """The list of classes in the namespace. This a a wrapper arround :func:`create_class_enum`. - def create_instance_enum(self, filter, flags=DEFAULT_ENUM_FLAGS, deep=True): - # ??? marche pas :( + :return: [:class:`WmiObject`] - A list of :class:`WmiObject` + """ + return self.create_class_enum(None, deep=True) + + def create_instance_enum(self, clsname, flags=DEFAULT_ENUM_FLAGS, deep=True): + """Enumerate the instances of ``clsname``. Deep allows to enumerate the instance of subclasses as well + + :returns: :class:`WmiEnumeration` + + Example: + >>> windows.system.wmi["root\\subscription"].create_instance_enum("__EventConsumer", deep=False).all() + [] + >>> windows.system.wmi["root\\subscription"].create_instance_enum("__EventConsumer", deep=True).all() + [] + + .. note:: + + See https://docs.microsoft.com/en-us/windows/desktop/api/wbemcli/nf-wbemcli-iwbemservices-createinstanceenum + """ flags |= gdef.WBEM_FLAG_DEEP if deep else gdef.WBEM_FLAG_SHALLOW - enumerator = WmiEnumeration() - self.CreateInstanceEnum(filter, flags, None, enumerator) + self.CreateInstanceEnum(clsname, flags, None, enumerator) return enumerator - select = create_instance_enum - def get_object(self, path): + """Return the object matching ``path``. If ``path`` is a class name return the class object`` + + :return: :class:`WmiObject` + """ result = WmiObject() self.GetObject(path, gdef.WBEM_FLAG_RETURN_WBEM_COMPLETE, None, result, None) return result - def put_instance(self, instance): - # TODO: change flag + def put_instance(self, instance, flags=gdef.WBEM_FLAG_CREATE_ONLY): + """Creates or updates an instance of an existing class in the namespace + + :return: :class:`WmiCallResult` ``(string)`` - Used to retrieve the string representing the path of the object created/updated + """ res = WmiCallResult(result_type="string") - self.PutInstance(instance, gdef.WBEM_FLAG_CREATE_ONLY, None, res) + self.PutInstance(instance, flags, None, res) return res def exec_method(self, obj, method, inparam, flags=0): + """Exec method named on ``object`` with ``inparam``. + + :params obj: The :class:`WmiObject` or path of the object the call apply to + :params method: The name of the method to call on the object + :params inparam: The :class:`WmiObject` representing the input parameters and retrieve using :func:`WmiObject.get_method` + + :returns: :class:`WmiCallResult` ``(object)`` if flag `WBEM_FLAG_RETURN_IMMEDIATELY` was passed + :returns: :class:`WmiObject` the outparam object if flag `WBEM_FLAG_RETURN_IMMEDIATELY` was NOT passed + + .. note:: + + This API will lakely change to better wrap with WmiObject/inparam/Dict & co + """ if flags & gdef.WBEM_FLAG_RETURN_IMMEDIATELY: # semisynchronous call -> WmiCallResult result = WmiCallResult(result_type="object")