Source code for pywbem._server

#
# This library is free software; you can redistribute it and/or
# modify it under the terms of the GNU Lesser General Public
# License as published by the Free Software Foundation; either
# version 2.1 of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public
# License along with this program; if not, write to the Free Software
# Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
#

"""
*New in pywbem 0.9 as experimental and finalized in 0.10.*

The WBEM server library API of pywbem encapsulates selected functionality of a
WBEM server for use by a WBEM client application, such as determining the
Interop namespace and other basic information about the server, or the
management profiles advertised by the server.

This chapter has the following sections:

* :ref:`Example <Server Example>` - An example on how to use the WBEM server
  library API.

* :ref:`WBEMServer` - The :class:`~pywbem.WBEMServer` class serves as a general
  access point for clients to WBEM servers. It allows determining the Interop
  namespace of the server and other basic information about the server, or the
  advertised management profiles.

.. _`Server Example`:

Example
-------

The following example code displays some information about a WBEM server:

::

    from pywbem import WBEMConnection, WBEMServer, ValueMapping

    def explore_server(server_url, username, password):

        print(f"WBEM server URL:\\n  {server_url}")

        conn = WBEMConnection(server_url, (username, password))
        server = WBEMServer(conn)

        print(f"Brand:\\n  {server.brand}")
        print(f"Version:\\n  {server.version}")
        print(f"Interop namespace:\\n  {server.interop_ns}")

        print("All namespaces:")
        for ns in server.namespaces:
            print(f"  {ns}")

        print("Advertised management profiles:")
        org_vm = ValueMapping.for_property(server, server.interop_ns,
            'CIM_RegisteredProfile', 'RegisteredOrganization')
        for inst in server.profiles:
            org = org_vm.tovalues(inst['RegisteredOrganization'])
            name = inst['RegisteredName']
            vers = inst['RegisteredVersion']
            print(f"  {org} {name} Profile {vers}")

Example output:

::

    WBEM server URL:
      http://0.0.0.0
    Brand:
      OpenPegasus
    Version:
      2.12.0
    Interop namespace:
      root/PG_Interop
    All namespaces:
      root/PG_InterOp
      root/PG_Internal
      root/cimv2
      root
    Advertised management profiles:
      SNIA Indication Profile 1.1.0
      SNIA Indication Profile 1.2.0
      SNIA Software Profile 1.1.0
      SNIA Software Profile 1.2.0
      SNIA Profile Registration Profile 1.0.0
      SNIA SMI-S Profile 1.2.0
      SNIA Server Profile 1.1.0
      SNIA Server Profile 1.2.0
      DMTF Profile Registration Profile 1.0.0
      DMTF Indications Profile 1.1.0
"""

import re
import warnings

from ._cim_constants import CIM_ERR_INVALID_NAMESPACE, CIM_ERR_INVALID_CLASS, \
    CIM_ERR_METHOD_NOT_FOUND, CIM_ERR_METHOD_NOT_AVAILABLE, \
    CIM_ERR_NOT_SUPPORTED, CIM_ERR_NOT_FOUND, CIM_ERR_FAILED, \
    CIM_ERR_NAMESPACE_NOT_EMPTY
from ._exceptions import CIMError, CIMXMLParseError, XMLParseError, ModelError
from ._warnings import ToleratedServerIssueWarning
from ._nocasedict import NocaseDict
from ._cim_obj import CIMInstanceName, CIMInstance
from ._cim_operations import WBEMConnection
from ._valuemapping import ValueMapping
from ._utils import _ensure_unicode, _format

__all__ = ['WBEMServer']


[docs] class WBEMServer: """ *New in pywbem 0.9 as experimental and finalized in 0.10.* A representation of a WBEM server that serves as a general access point to a client. It supports determining the Interop namespace of the server, all namespaces, its brand and version, the advertised management profiles and finally allows retrieving the central instances of an implementation of a management profile with one method invocation regardless of whether the profile implementation chose to implement the central or scoping class profile advertisement methodology (see section :ref:`Profile advertisement methodologies`). It also provides functions to subscribe for indications. """ #: A class variable with the possible names of Interop namespaces that #: should be tried when determining the Interop namespace on the WBEM #: server. INTEROP_NAMESPACES = [ 'interop', 'root/interop', 'root/PG_Interop', # OpenPegasus before version 2.12.0 defined only PG_Interop as the # interop namespace. Using other namespaces was a manual # modification of at least the pegasus/mak/configschema.mak file. # Starting in version 2.12.0 a configuration variable was defined # so that OpenPegasus could be build with any of the specified names # for interop namespace (PEGASUS_INTEROP_NAMESPACE) ] #: A class variable with the possible names of CIM classes for #: representing CIM namespaces, that should be tried when determining the #: namespaces on the WBEM server. NAMESPACE_CLASSNAMES = [ 'CIM_Namespace', 'CIM_WBEMServerNamespace', '__Namespace', ] def __init__(self, conn): """ Parameters: conn (:class:`~pywbem.WBEMConnection`): Connection to the WBEM server. """ if not isinstance(conn, WBEMConnection): raise TypeError( _format("conn argument of WBEMServer must be a WBEMConnection " "object, but has type: {0}", type(conn))) self._conn = conn self._interop_ns = None self._namespaces = None self._namespace_paths = None self._namespace_classname = None self._brand = None self._version = None self._cimom_inst = None self._profiles = None
[docs] def __str__(self): """ Return a representation of the :class:`~pywbem.WBEMServer` object with a subset of its attributes. """ return _format( "WBEMServer(" "conn.url={s._conn.url!A}, " "brand={s._brand!A}, " "version={s._version!A}, " "... )", s=self)
[docs] def __repr__(self): """ Return a representation of the :class:`~pywbem.WBEMServer` object with all attributes, that is suitable for debugging. """ return _format( "WBEMServer(" "conn.url={s._conn.url!A}, " "interop_ns={s._interop_ns!A}, " "namespaces={s._namespaces!A}, " "namespace_paths={s._namespace_paths!A}, " "namespace_classname={s._namespace_classname!A}, " "brand={s._brand!A}, " "version={s._version!A}, " "profiles=[{pnum} instances])", s=self, pnum=len(self._profiles) if self._profiles else 0)
@property def url(self): """ :term:`string`: URL of the WBEM server. """ return self._conn.url @property def conn(self): """ :class:`~pywbem.WBEMConnection`: Connection to the WBEM server. """ return self._conn @property def interop_ns(self): """ :term:`string`: Name of the Interop namespace of the WBEM server. Raises: Exceptions raised by :class:`~pywbem.WBEMConnection`. CIMError: CIM_ERR_NOT_FOUND, Interop namespace could not be determined. """ if self._interop_ns is None: self._determine_interop_ns() return self._interop_ns @property def namespace_classname(self): """ :term:`string`: Name of the CIM class that was found to represent the CIM namespaces of the WBEM server. Raises: Exceptions raised by :class:`~pywbem.WBEMConnection`. ModelError: An error with the model implemented by the WBEM server. """ if self._namespace_classname is None: self._determine_namespaces() return self._namespace_classname @property def namespaces(self): """ list of :term:`string`: Names of all namespaces of the WBEM server. Raises: Exceptions raised by :class:`~pywbem.WBEMConnection`. ModelError: An error with the model implemented by the WBEM server. """ if self._namespaces is None: self._determine_namespaces() return self._namespaces @property def namespace_paths(self): """ list of :class:`~pywbem.CIMInstanceName`: Instance paths of the CIM instances in the Interop namespace that represent the namespaces of the WBEM server. Note: One WBEM server has been found to support an Interop namespace without representing it as a CIM instance. In that case, this property will not have an instance path for the Interop namespace, but the :attr:`namespaces` property will have the name of the Interop namespace. Raises: Exceptions raised by :class:`~pywbem.WBEMConnection`. ModelError: An error with the model implemented by the WBEM server. """ if self._namespace_paths is None: self._determine_namespaces() return self._namespace_paths @property def brand(self): """ :term:`string`: Brand of the WBEM server. The brand is determined from the `CIM_ObjectManager` instance in the Interop namespace, by looking at its `ElementName` property. For known WBEM servers, the brand is then normalized in order to make it identifiable: * ``"OpenPegasus"`` * ``"SFCB"`` (Small Footprint CIM Broker) * ``"WBEM Solutions J WBEM Server"`` * ``"EMC CIM Server"`` * ``"FUJITSU CIM Object Manager"`` For all other WBEM servers, the brand is the value of the `ElementName` property, or the string ``"unknown"``, if that property is not set or the empty string. Raises: Exceptions raised by :class:`~pywbem.WBEMConnection`. CIMError: CIM_ERR_NOT_FOUND, Interop namespace could not be determined. CIMError: CIM_ERR_NOT_FOUND, Unexpected number of `CIM_ObjectManager` instances. """ if self._brand is None: self._determine_brand() return self._brand @property def version(self): """ :term:`string`: Version of the WBEM server. `None`, if the version cannot be determined. The version is determined from the `CIM_ObjectManager` instance in the Interop namespace, by looking at its `ElementName` property, or if that is not set, at its `Description` property, and by taking the string after ``"version"`` or ``"release"`` (case insensitively). Raises: Exceptions raised by :class:`~pywbem.WBEMConnection`. CIMError: CIM_ERR_NOT_FOUND, Interop namespace could not be determined. CIMError: CIM_ERR_NOT_FOUND, Unexpected number of `CIM_ObjectManager` instances. """ if self._version is None: self._determine_brand() return self._version @property def cimom_inst(self): """ :class:`~pywbem.CIMInstance`: CIM instance of class `CIM_ObjectManager` that represents the WBEM server. Raises: Exceptions raised by :class:`~pywbem.WBEMConnection`. CIMError: CIM_ERR_NOT_FOUND, Interop namespace could not be determined. CIMError: CIM_ERR_NOT_FOUND, Unexpected number of `CIM_ObjectManager` instances. """ if self._cimom_inst is None: self._determine_brand() return self._cimom_inst @property def profiles(self): """ list of :class:`~pywbem.CIMInstance`: The `CIM_RegisteredProfile` instances representing all management profiles advertised by the WBEM server. Raises: Exceptions raised by :class:`~pywbem.WBEMConnection`. CIMError: CIM_ERR_NOT_FOUND, Interop namespace could not be determined. """ if self._profiles is None: self._determine_profiles() return self._profiles
[docs] def create_namespace(self, namespace, verbose=False): """ Create the specified CIM namespace in the WBEM server and update this WBEMServer object to reflect the new namespace there. This method cannot create an Interop namespace because creating the an Interop namespace with client operations depends on the prior existence of an Interop namespace, and creating additional Interop namespaces if one already exists is usually prevented by servers. This method attempts the following approaches for creating the namespace, in order, until an approach succeeds: 1. Namespace creation as described in the WBEM Server profile (:term:`DSP1092`) via CIM method `CIM_WBEMServer.CreateWBEMServerNamespace()`. This is a new standard approach that is not likely to be widely implemented yet. 2. Issuing the `CreateInstance` operation using the CIM class representing namespaces in the WBEM server, against the Interop namespace. An exception is OpenPegasus, where 'PG_Namespace' is used instead of 'CIM_Namespace'. This approach is typically supported in WBEM servers that support the creation of CIM namespaces. This approach is similar to the approach described in :term:`DSP0200`. Creating namespaces using the `__Namespace` pseudo-class has been deprecated already in DSP0200 1.1.0 (released in 01/2003), and pywbem does not implement that approach. Parameters: namespace (:term:`string`): CIM namespace name. Must not be `None`. The namespace may contain leading and a trailing slash, both of which will be ignored. verbose (:class:`py:bool`): Verbose mode: Print a message about the namespace creation. Returns: :term:`unicode string`: The specified CIM namespace name in its standard format (i.e. without leading or trailing slash characters). Raises: Exceptions raised by :class:`~pywbem.WBEMConnection`. ModelError: An error with the model implemented by the WBEM server. """ std_namespace = _ensure_unicode(namespace.strip('/')) if verbose: print(f"Creating namespace {std_namespace} (in WBEMServer)") try: ws_profiles = self.get_selected_profiles('DMTF', 'WBEM Server') except (CIMError, ModelError): ws_profiles = None if ws_profiles: # Use approach 1: Method defined in WBEM Server profile ws_profiles_sorted = sorted( ws_profiles, key=lambda prof: prof['RegisteredVersion']) ws_profile_inst = ws_profiles_sorted[-1] # latest version ws_insts = self.get_central_instances(ws_profile_inst.path) if len(ws_insts) != 1: raise ModelError( _format("Unexpected number of central instances of WBEM " "Server profile: {0!A}", [i.path for i in ws_insts])) ws_inst = ws_insts[0] ns_inst = CIMInstance('CIM_WBEMServerNamespace') ns_inst['Name'] = std_namespace try: (ret_val, out_params) = self._conn.InvokeMethod( MethodName="CreateWBEMServerNamespace", ObjectName=ws_inst.path, Params=[('NamespaceTemplate', ns_inst)]) except CIMError as exc: if exc.status_code in (CIM_ERR_METHOD_NOT_FOUND, CIM_ERR_METHOD_NOT_AVAILABLE, CIM_ERR_NOT_SUPPORTED): # Method is not implemented. # CIM_ERR_NOT_SUPPORTED is not an official status code for # this situation, but is used by some implementations. pass # try next approach else: raise else: if ret_val != 0: raise ModelError( _format("The CreateWBEMServerNamespace() method is " "implemented but failed: {0}", out_params['Errors'])) else: # Use approach 2: CreateInstance of CIM class for namespaces # For the new namespace, we use the same class name that is used for # already existing namespaces in this particular WBEM server. ns_classname = self.namespace_classname # Determines if needed # For OpenPegasus, use 'PG_Namespace' class to account for issue # when using 'CIM_Namespace'. See OpenPegasus bug 10112: # https://bugzilla.openpegasus.org/show_bug.cgi?id=10112 if self.brand == "OpenPegasus" and ns_classname == 'CIM_Namespace': ns_classname = 'PG_Namespace' ns_inst = CIMInstance(ns_classname) # OpenPegasus requires this property to be True, in order to # allow schema updates in the namespace. if self.brand == "OpenPegasus": ns_inst['SchemaUpdatesAllowed'] = True ns_inst['Name'] = std_namespace # DSP0200 is not clear as to whether just "Name" or all key # properties need to be provided. For now, we provide all key # properties. # OpenPegasus requires all key properties, and it re-creates the # 5 key properties besides "Name" so that the returned instance # path may differ from the key properties provided. ns_inst['CreationClassName'] = ns_classname ns_inst['ObjectManagerName'] = self.cimom_inst['Name'] ns_inst['ObjectManagerCreationClassName'] = \ self.cimom_inst['CreationClassName'] ns_inst['SystemName'] = self.cimom_inst['SystemName'] ns_inst['SystemCreationClassName'] = \ self.cimom_inst['SystemCreationClassName'] interop_ns = self.interop_ns # Determines the Interop namespace self.conn.CreateInstance(ns_inst, namespace=interop_ns) # Refresh the list of namespaces in this object to include the one # we just created. # Namespace creation is such a rare operation that we can afford # the extra namespace determination operations, to make sure we # really have the new namespace. self._determine_namespaces() return std_namespace
[docs] def delete_namespace(self, namespace, verbose=False): """ Delete the specified CIM namespace in the WBEM server and update this WBEMServer object to reflect the removed namespace there. The specified namespace must be empty (i.e. must not contain any classes, instances, or qualifier types. This method cannot delete the Interop namespace because servers will usually prevent their deletion. This method attempts the following approaches for deleting the namespace, in order, until an approach succeeds: 1. Issuing the `DeleteInstance` operation using the CIM class representing namespaces ('PG_Namespace' for OpenPegasus, and 'CIM_Namespace' otherwise), against the Interop namespace. This approach is typically supported in WBEM servers that support the creation of CIM namespaces. This approach is similar to the approach described in :term:`DSP0200`. The approach described in the WBEM Server profile (:term:`DSP1092`) via deleting the `CIM_WBEMServerNamespace` instance is not implemented because that would also delete any classes, instances, and qualifier types in the namespace. Deleting namespaces using the `__Namespace` pseudo-class has been deprecated already in DSP0200 1.1.0 (released in 01/2003), and pywbem does not implement that approach. Parameters: namespace (:term:`string`): CIM namespace name. Must not be `None`. The namespace may contain leading and a trailing slash, both of which will be ignored. verbose (:class:`py:bool`): Verbose mode: Print a message about the namespace creation. Returns: :term:`unicode string`: The specified CIM namespace name in its standard format (i.e. without leading or trailing slash characters). Raises: Exceptions raised by :class:`~pywbem.WBEMConnection`. ModelError: An error with the model implemented by the WBEM server. CIMError: CIM_ERR_NOT_FOUND, Specified namespace does not exist. CIMError: CIM_ERR_NAMESPACE_NOT_EMPTY, Specified namespace is not empty. """ std_namespace = _ensure_unicode(namespace.strip('/')) if verbose: print(f"Deleting namespace {std_namespace} (in WBEMServer)") # Use approach 1: DeleteInstance of CIM class for namespaces # Refresh the list of namespaces in this object to make sure # it is up to date. self._determine_namespaces() if std_namespace not in self.namespaces: raise CIMError( CIM_ERR_NOT_FOUND, _format("pywbem detected that the specified namespace does not " "exist: {0!A}", std_namespace), conn_id=self.conn.conn_id) ns_path = None for p in self.namespace_paths: if p.keybindings['Name'] == std_namespace: ns_path = p assert ns_path is not None # Ensure the namespace is empty. We do not check for instances, because # classes are a prerequisite for instances, so if no classes exist, # no instances will exist. # WBEM servers that do not support class operations (e.g. SFCB) will # raise a CIMError with status CIM_ERR_NOT_SUPPORTED. class_paths = self.conn.EnumerateClassNames( namespace=std_namespace, ClassName=None, DeepInheritance=False) quals = self.conn.EnumerateQualifiers(namespace=std_namespace) if class_paths or quals: raise CIMError( CIM_ERR_NAMESPACE_NOT_EMPTY, _format("pywbem detected that the specified namespace {0!A} " "is not empty; it contains " "{1} top-level classes and {2} qualifier types", std_namespace, len(class_paths), len(quals)), conn_id=self.conn.conn_id) self.conn.DeleteInstance(ns_path) # Refresh the list of namespaces in this object to remove the one # we just deleted. self._determine_namespaces() return std_namespace
[docs] def get_selected_profiles(self, registered_org=None, registered_name=None, registered_version=None): """ Return the `CIM_RegisteredProfile` instances representing a filtered subset of the management profiles advertised by the WBEM server, that can be filtered by registered organization, registered name, and/or registered version. Parameters: registered_org (:term:`string`): A filter for the registered organization of the profile, matching (case insensitively) the `RegisteredOrganization` property of the `CIM_RegisteredProfile` instance, via its `Values` qualifier. If `None`, this parameter is ignored for filtering. registered_name (:term:`string`): A filter for the registered name of the profile, matching (case insensitively) the `RegisteredName` property of the `CIM_RegisteredProfile` instance. If `None`, this parameter is ignored for filtering. registered_version (:term:`string`): A filter for the registered version of the profile, matching (case insensitively) the `RegisteredVersion` property of the `CIM_RegisteredProfile` instance. Note that version strings may contain aplhabetic characters to indicate the draft level. If `None`, this parameter is ignored for filtering. Returns: list of :class:`~pywbem.CIMInstance`: The `CIM_RegisteredProfile` instances representing the filtered subset of the management profiles advertised by the WBEM server. Raises: Exceptions raised by :class:`~pywbem.WBEMConnection`. CIMError: CIM_ERR_NOT_FOUND, Interop namespace could not be determined. ModelError: If an instance in the list of profiles is incomplete and does not include the required properties. """ interop_ns = self.interop_ns # Determines the Interop namespace org_vm = ValueMapping.for_property(self, interop_ns, 'CIM_RegisteredProfile', 'RegisteredOrganization') org_lower = registered_org.lower() \ if registered_org is not None else None name_lower = registered_name.lower() \ if registered_name is not None else None version_lower = registered_version.lower() \ if registered_version is not None else None rtn = [] for inst in self.profiles: try: inst_org_value = inst['RegisteredOrganization'] except KeyError: raise ModelError( _format("CIM_RegisteredProfile instance in namespace " "{0!A} does not have a property " "'RegisteredOrganization'", interop_ns)) inst_org = org_vm.tovalues(inst_org_value) try: inst_name = inst['RegisteredName'] except KeyError: raise ModelError( _format("CIM_RegisteredProfile instance in namespace " "{0!A} does not have a property " "'RegisteredName'", interop_ns)) try: inst_version = inst['RegisteredVersion'] except KeyError: raise ModelError( _format("CIM_RegisteredProfile instance in namespace " "{0!A} does not have a property " "'RegisteredVersion'", interop_ns)) inst_org_lower = inst_org.lower() \ if inst_org is not None else None inst_name_lower = inst_name.lower() \ if inst_name is not None else None inst_version_lower = inst_version.lower() \ if inst_version is not None else None # pylint: disable=too-many-boolean-expressions if (org_lower is None or org_lower == inst_org_lower) and \ (name_lower is None or name_lower == inst_name_lower) and \ (version_lower is None or version_lower == inst_version_lower): rtn.append(inst) return rtn
[docs] def get_central_instances(self, profile_path, central_class=None, scoping_class=None, scoping_path=None, reference_direction='dmtf', try_gci_method=False): """ Return the instance paths of the central instances of a management profile. DMTF defines the following profile advertisement methodologies in :term:`DSP1033`: * GetCentralInstances methodology (new in :term:`DSP1033` 1.1, only when explicitly requested by the caller) * Central class methodology * Scoping class methodology A brief explanation of these methodologies can be found in section :ref:`Profile advertisement methodologies`. Pywbem attempts all three profile advertisement methodologies in the order listed above. All three methodologies start from the CIM_RegisteredProfile instance referenced by the `profile_path` parameter. That instance represents a management profile. In case of multiple uses of a component profile in a WBEM server, one such instance is supposed to represent one such profile use. If the profile is a component profile and its implementation does not support the GetCentralInstances or central class methodologies, the `central_class`, `scoping_class`, and `scoping_path` parameters are required in order for the method to attempt the scoping class methodology. The method will not fail if these parameters are not provided, as long as the profile implementation supports the GetCentralInstances or central class methodology. Example parameters for a 1-hop scoping path: * ``central_class = "CIM_Fan"`` * ``scoping_path = ["CIM_SystemDevice"]`` * ``scoping_class = "CIM_ComputerSystem"`` Example parameters for a 2-hop scoping path: * ``central_class = "CIM_Sensor"`` * ``scoping_path = ["CIM_AssociatedSensor", "CIM_Fan", "CIM_SystemDevice"]`` * ``scoping_class = "CIM_ComputerSystem"`` Parameters: profile_path (:class:`~pywbem.CIMInstanceName`): Instance path of the `CIM_RegisteredProfile` instance representing the management profile (or its use, if there are multiple uses in a WBEM server). central_class (:term:`string`): Class name of central class defined by the management profile. Will be ignored, unless the profile is a component profile and its implementation supports only the scoping class methodology. `None` will cause the scoping class methodology not to be attempted. scoping_class (:term:`string`): Class name of scoping class defined by the management profile. Will be ignored, unless the profile is a component profile and its implementation supports only the scoping class methodology. `None` will cause the scoping class methodology not to be attempted. scoping_path (list of :term:`string`): Scoping path defined by the management profile. Will be ignored, unless the profile is a component profile and its implementation supports only the scoping class methodology. `None` will cause the scoping class methodology not to be attempted. reference_direction (:term:`string`): Defines the navigation direction across the CIM_ReferencedProfile association when navigating from the current profile to its scoping (= referencing, autonomous) profile when using the scoping class methodology, as follows: * 'dmtf' (default): Assume DMTF conformance, i.e. the 'Dependent' end is followed. * 'snia': Assume SNIA SMI-S conformance, i.e. the 'Antecedent' end is followed. This parameter supports the different definitions between DMTF and SNIA SMI-S standards regarding the use of the two ends of the CIM_ReferencedProfile association: * The DMTF standards define in DSP1033 and DSP1001: - Antecedent = referenced profile = component profile - Dependent = referencing profile = autonomous profile * The SNIA SMI-S standard defines in the "Profile Registration Profile" (in the SMI-S "Common Profiles" book): - Antecedent = autonomous profile - Dependent = component (= sub) profile It should be assumed that all profiles that are directly or indirectly scoped by a particular top-level (= wrapper) specification implement the reference direction that matches the registered organisation of the top-level specification. Examples: * All profiles scoped by the SNIA SMI-S top-level specification should be assumed to implement the 'snia' reference direction. * All profiles scoped by the DMTF SMASH wrapper specification should be assumed to implement the 'dmtf' reference direction. try_gci_method (:class:`py:bool`): Flag indicating that the GetCentralInstances methodology should be attempted. This methodology is not expected to be implemented by WBEM servers at this point, and causes undesirable behavior with some WBEM servers, so it is not attempted by default. Note that WBEM servers are required to support the scoping class methodology. Returns: list of :class:`~pywbem.CIMInstanceName`: The instance paths of the central instances of the implementation of the management profile. Raises: Exceptions raised by :class:`~pywbem.WBEMConnection`. ModelError: An error with the model implemented by the WBEM server. ValueError: User errors regarding input parameter values. TypeError: User errors regarding input parameter types. """ if reference_direction not in ('dmtf', 'snia'): raise ValueError( _format("The reference_direction parameter must be 'dmtf' or " "'snia', but is: {0!A}", reference_direction)) scoping_result_role = "Dependent" if reference_direction == 'dmtf' \ else "Antecedent" if not isinstance(profile_path, CIMInstanceName): raise TypeError( _format("The profile_path argument must be a CIMInstanceName, " "but has type: {0}", type(profile_path))) if try_gci_method: # Try GetCentralInstances() method: try: (ret_val, out_params) = self._conn.InvokeMethod( MethodName="GetCentralInstances", ObjectName=profile_path) except CIMError as exc: if exc.status_code in (CIM_ERR_FAILED, CIM_ERR_METHOD_NOT_FOUND, CIM_ERR_METHOD_NOT_AVAILABLE, CIM_ERR_NOT_SUPPORTED): # Method is not implemented. # CIM_ERR_NOT_SUPPORTED is not an official status code for # this situation, but is used by some implementations. pass # try next approach else: raise except (CIMXMLParseError, XMLParseError) as exc: # The EMC server returns XMLParseError due to ill-formed XML. # The Dell server returns CIMXMLParseError due to INSTANCE # element with missing CLASSNAME attribute. # In both cases the ERROR element is parseable in the CIM-XML # string. Tolerate that behavior if the error is that the # method is not implemented, and try next approach. reply_oneline = self._conn.last_raw_reply.replace(b'\n', b'') m = re.search(b'<ERROR CODE="([0-9]+)"', reply_oneline) if m: try: status_code = int(m.group(1)) except ValueError: status_code = None if status_code in (CIM_ERR_FAILED, CIM_ERR_METHOD_NOT_FOUND, CIM_ERR_METHOD_NOT_AVAILABLE, CIM_ERR_NOT_SUPPORTED): warnings.warn( _format("Tolerating {0} raised when parsing " "CIM-XML response of invoking CIM method " "GetCentralInstances, with a CIM status " "code {1}: {2}", exc.__class__.__name__, status_code, exc), ToleratedServerIssueWarning, 1) # try next approach else: # It would be nice to extend the exception message to # indicate that a CIM status code was detectable, but # it is not possible to do that if we want to maintain # standard exception semantics w.r.t. the message. # So we just re-raise the original CIMXMLParseError or # XMLParseError. raise else: # In this case, the ERROR element is not recognizable, so # we just re-raise the original CIMXMLParseError or # XMLParseError. raise else: if ret_val != 0: raise ModelError( _format("The GetCentralInstances() method is " "implemented but failed with rc={0} for " "profile {1!A}", ret_val, profile_path.to_wbem_uri())) central_inst_paths = out_params['CentralInstances'] return central_inst_paths # Try central methodology try: central_inst_paths = self._conn.AssociatorNames( ObjectName=profile_path, AssocClass="CIM_ElementConformsToProfile", ResultRole="ManagedElement") except CIMError as exc: if exc.status_code == CIM_ERR_NOT_SUPPORTED: # This association traversal is not implemented, so we can # conclude that the central methodology is not implemented. pass # Try next methodology else: raise else: # The central methodology is implemented. # Note: It is possible (and valid) that there are no central # instances. return central_inst_paths # Try scoping methodology if central_class is None or \ scoping_class is None or \ scoping_path is None: raise ValueError( _format("Parameters required for scoping class methodology not " "specified and other methodologies not implemented " "for profile {0!A}", profile_path.to_wbem_uri())) # Go up one level on the profile side, to the scoping profile referencing_profile_paths = self._conn.AssociatorNames( ObjectName=profile_path, AssocClass="CIM_ReferencedProfile", ResultRole=scoping_result_role) if not referencing_profile_paths: raise ModelError( _format("No referencing profile found for profile {0!A} when " "attempting the scoping class methodology (traversing " "CIM_ReferencedProfile to its {1!A} end using the {2} " "reference direction)", profile_path.to_wbem_uri(), scoping_result_role, reference_direction)) if len(referencing_profile_paths) > 1: raise ModelError( _format("More than one referencing profile found for profile " "{0!A} when attempting the scoping class methodology " "(traversing CIM_ReferencedProfile to its {1!A} end " "using the {2} reference direction). " "Found referencing profiles {3!A}", profile_path.to_wbem_uri(), scoping_result_role, reference_direction, [p.to_wbem_uri() for p in referencing_profile_paths])) # Traverse to the resource side. # Note: Because the scoping path of the scoping profile is not known, # all we can do is to rely on the central methodology to be implemented # by the scoping profile. Therefore, we pass only the central class of # the scoping profile (which is the scoping class of the original # profile). # Note: It is assumed that the scoping profile has the same reference # direction as the original profile. scoping_profile_path = referencing_profile_paths[0] scoping_inst_paths = self.get_central_instances( scoping_profile_path, scoping_class, None, None, reference_direction) if not scoping_inst_paths: # In order to be able to traverse down to the original profile's # central instances, we need the scoping profile to have # at least one central instance. raise ModelError( _format("Profile {0!A} does not have any central instances, " "but that is required in its role of the scoping " "profile of profile {1!A} when attempting the " "scoping class methodology", scoping_profile_path.to_wbem_uri(), profile_path.to_wbem_uri())) # On the resource side, traverse down the reversed scoping path, # to the central instances of the original profile. traversal_path = list(reversed(scoping_path)) traversal_path.append(central_class) central_inst_paths = self._traverse(scoping_inst_paths, traversal_path) return central_inst_paths
def _traverse(self, start_paths, traversal_path): """ Traverse a multi-hop traversal path from a list of start instance paths, and return the resulting list of instance paths. Parameters: start_paths (list of CIMInstanceName): Instance paths to start traversal from. traversal_path (list of string): Traversal hops, where the list contains pairs of items: association class name, far end class name. Example: a 2-hop traversal is represented as `['A1', 'C1', 'A2', 'C2']`. Returns: List of CIMInstanceName: Instances at the far end of the traversal. """ assert len(traversal_path) >= 2 assoc_class = traversal_path[0] far_class = traversal_path[1] total_next_paths = [] for path in start_paths: next_paths = self._conn.AssociatorNames( ObjectName=path, AssocClass=assoc_class, ResultClass=far_class) total_next_paths.extend(next_paths) traversal_path = traversal_path[2:] if traversal_path: total_next_paths = self._traverse(total_next_paths, traversal_path) return total_next_paths def _determine_interop_ns(self): """ Determine the name of the Interop namespace of the WBEM server, by trying to communicate with it on a number of possible Interop namespace names, that are defined in the :attr:`INTEROP_NAMESPACES` class variable. If the Interop namespace could be determined, this method sets the :attr:`interop_ns` property of this object to that namespace and returns. Otherwise, it raises an exception. Raises: Exceptions raised by :class:`~pywbem.WBEMConnection`. ModelError: An error with the model implemented by the WBEM server. """ test_classname = 'CIM_Namespace' interop_ns = None for ns in self.INTEROP_NAMESPACES: try: inst_paths = self._conn.EnumerateInstanceNames(test_classname, namespace=ns) except CIMError as exc: if exc.status_code == CIM_ERR_INVALID_NAMESPACE: # Current namespace does not exist. continue if exc.status_code in (CIM_ERR_INVALID_CLASS, CIM_ERR_NOT_FOUND): # Class is not implemented, but current namespace exists. interop_ns = ns break # Some other error happened. raise # Namespace class is implemented in the current namespace. # Use the returned namespace name, if possible. ns_names = [p.keybindings['name'] for p in inst_paths] ns_dict = NocaseDict(list(zip(ns_names, ns_names))) try: interop_ns = ns_dict[ns] except KeyError: interop_ns = ns break if interop_ns is None: # Exhausted the possible namespaces raise ModelError( _format("Interop namespace does not exist (tried {0!A})", self.INTEROP_NAMESPACES), conn_id=self.conn.conn_id) self._interop_ns = interop_ns def _validate_interop_ns(self, interop_ns): """ Validate whether the specified Interop namespace exists in the WBEM server, by communicating with it. If the specified Interop namespace exists, this method sets the :attr:`interop_ns` property of this object to that namespace and returns. Otherwise, it raises an exception. Parameters: interop_ns (:term:`string`): Name of the Interop namespace to be validated. Raises: Exceptions raised by :class:`~pywbem.WBEMConnection`. """ test_classname = 'CIM_Namespace' try: self._conn.EnumerateInstanceNames(test_classname, namespace=interop_ns) except CIMError as exc: # We tolerate it if the WBEM server does not implement this class, # as long as it does not return CIM_ERR_INVALID_NAMESPACE. if exc.status_code in (CIM_ERR_INVALID_CLASS, CIM_ERR_NOT_FOUND): pass else: raise self._interop_ns = interop_ns def _determine_namespaces(self): """ Determine the names of all namespaces of the WBEM server, by communicating with it and enumerating the instances of a number of possible CIM classes that typically represent CIM namespaces. Their class names are defined in the :attr:`NAMESPACE_CLASSNAMES` class variable. If the namespaces could be determined, this method sets the following properties of this object: * :attr:`namespace_classname` * :attr:`namespaces` * :attr:`namespace_paths` Otherwise, it raises an exception. Note that there is at least one WBEM server that implements an Interop namespace but does not represent that with a CIM instance. In that case, the :attr:`namespaces` property will include the Interop namespace, but the :attr:`namespace_paths` property will not. Raises: Exceptions raised by :class:`~pywbem.WBEMConnection`. ModelError: An error with the model implemented by the WBEM server. """ ns_insts = None ns_classname = None interop_ns = self.interop_ns # Determines the Interop namespace for classname in self.NAMESPACE_CLASSNAMES: try: ns_insts = self._conn.EnumerateInstances( classname, namespace=interop_ns) except CIMError as exc: if exc.status_code in (CIM_ERR_INVALID_CLASS, CIM_ERR_NOT_FOUND): # Class is not implemented, try next one. continue # Some other error. raise # Found a namespace class that is implemented. ns_classname = classname break if ns_insts is None: # Exhausted the possible class names raise ModelError( _format("Namespace class could not be determined " "(tried {0!A})", self.NAMESPACE_CLASSNAMES), conn_id=self.conn.conn_id) self._namespace_classname = ns_classname self._namespaces = [inst['Name'] for inst in ns_insts] self._namespace_paths = [inst.path for inst in ns_insts] # An old version of a Hitachi server supports an Interop namespace # named 'interop' but does not represent it with a CIM instance. namespaces_lower = [ns.lower() for ns in self._namespaces] if interop_ns.lower() not in namespaces_lower: warnings.warn( _format("Server at {0} has an Interop namespace {1!A}, but " "does not return it when enumerating class {2!A} " "- adding it to the 'namespaces' property", self.conn.url, interop_ns, ns_classname), ToleratedServerIssueWarning, stacklevel=2) self._namespaces.append(interop_ns) def _determine_brand(self): """ Determine the brand of the WBEM server (e.g. OpenPegasus, SFCB, ...) and its version, by communicating with it and retrieving the `CIM_ObjectManager` instance. On success, this method sets the :attr:`brand`, :attr:`version`, and :attr:`cimom_inst` attributes of this object and returns. Otherwise, it raises an exception. Raises: Exceptions raised by :class:`~pywbem.WBEMConnection`. ModelError: An error with the model implemented by the WBEM server. """ interop_ns = self.interop_ns # Determines the Interop namespace cimom_insts = self._conn.EnumerateInstances( "CIM_ObjectManager", namespace=interop_ns) if len(cimom_insts) != 1: raise ModelError( _format("Unexpected number of CIM_ObjectManager instances: " "{0!A} ", [i['ElementName'] for i in cimom_insts]), conn_id=self.conn.conn_id) cimom_inst = cimom_insts[0] elementname = cimom_inst.get('ElementName', None) description = cimom_inst.get('Description', None) elementname_lower = elementname.lower() if "pegasus" in elementname_lower: # ElementName = "Pegasus" # Description = "Pegasus CIM Server Version 2.13.0" # | "Pegasus CIM Server Version 2.15.0 Released" brand = "OpenPegasus" m = re.match(r'^.*(?:version) *([0-9]+\.[0-9]+[^ ]*)(?: .+)?$', description, re.IGNORECASE) version = m.group(1) if m else None elif "sfcb" in elementname_lower: # ElementName = "sfcb" # Description = "Small Footprint CIM Broker 1.3.11" brand = "SFCB" m = re.match(r'^.* ([0-9]+\.[0-9]+.*)$', description, re.IGNORECASE) version = m.group(1) if m else None elif " j wbem server" in elementname_lower: # ElementName = "WS J WBEM Server" | "WBEM Solutions J WBEM Server" # Description = "WS J WBEM Server" | "WBEM Solutions J WBEM Server" # Version = "4.5.1" # Build = "02/25/2016 13:09" brand = "WBEM Solutions J WBEM Server" version = cimom_inst.get('Version', None) elif "emc cim server" in elementname_lower: # ElementName = "EMC CIM Server" # Description = "EMC CIM Server Version 2.7.3.6.0.11D" brand = "EMC CIM Server" m = re.match(r'^.* version *(.*)$', description, re.IGNORECASE) version = m.group(1) if m else None elif "CIM Object Manager for FUJITSU" in elementname_lower: # ElementName = "CIM Object Manager for FUJITSU storage system" # Description = "CIM Object Manager for FUJITSU storage system" brand = "FUJITSU CIM Object Manager" version = None else: brand = elementname or 'unknown' version = None if description is not None: m = re.match(r'^.* (?:version|release) *(.+)?$', description, re.IGNORECASE) version = m.group(1) if m else None self._brand = brand self._version = version self._cimom_inst = cimom_inst def _determine_profiles(self): """ Determine the WBEM management profiles advertised by the WBEM server, by communicating with it and enumerating the instances of `CIM_RegisteredProfile`. If the profiles could be determined, this method sets the :attr:`profiles` property of this object to the list of `CIM_RegisteredProfile` instances (as :class:`~pywbem.CIMInstance` objects), and returns. Otherwise, it raises an exception. Raises: Exceptions raised by :class:`~pywbem.WBEMConnection`. ModelError: An error with the model implemented by the WBEM server. """ interop_ns = self.interop_ns # Determines the Interop namespace mp_insts = self._conn.EnumerateInstances("CIM_RegisteredProfile", namespace=interop_ns) self._profiles = mp_insts