mirror of
https://github.com/vmware/vsphere-automation-sdk-python.git
synced 2024-11-23 09:59:59 -05:00
b25c66ee46
Remove old sso samples and created new samples that match Java SDK. Minor changes to some helper methods. Update README
346 lines
15 KiB
Python
346 lines
15 KiB
Python
"""
|
|
* *******************************************************
|
|
* Copyright (c) VMware, Inc. 2013. All Rights Reserved.
|
|
* *******************************************************
|
|
*
|
|
* DISCLAIMER. THIS PROGRAM IS PROVIDED TO YOU "AS IS" WITHOUT
|
|
* WARRANTIES OR CONDITIONS OF ANY KIND, WHETHER ORAL OR WRITTEN,
|
|
* EXPRESS OR IMPLIED. THE AUTHOR SPECIFICALLY DISCLAIMS ANY IMPLIED
|
|
* WARRANTIES OR CONDITIONS OF MERCHANTABILITY, SATISFACTORY QUALITY,
|
|
* NON-INFRINGEMENT AND FITNESS FOR A PARTICULAR PURPOSE.
|
|
"""
|
|
|
|
__author__ = 'VMware, Inc.'
|
|
__copyright__ = 'Copyright 2013 VMware, Inc. All rights reserved.'
|
|
|
|
import os
|
|
from suds.client import Client
|
|
|
|
class LookupServiceHelper(object):
|
|
def __init__(self, wsdl_url, soap_url, skip_verification):
|
|
self.wsdl_url = wsdl_url
|
|
self.soap_url = soap_url
|
|
self.skip_verification = skip_verification
|
|
self.client = None
|
|
self.managedObjectReference = None
|
|
self.serviceRegistration = None
|
|
|
|
def connect(self):
|
|
if self.client is None:
|
|
# Suds library doesn't support passing unverified context to disable
|
|
# server certificate verification. Thus disable checking globally in
|
|
# order to skip verification. This is not recommended in production
|
|
# code. see https://www.python.org/dev/peps/pep-0476/
|
|
if self.skip_verification:
|
|
import ssl
|
|
try:
|
|
_create_unverified_https_context = \
|
|
ssl._create_unverified_context
|
|
except AttributeError:
|
|
# Legacy Python that doesn't verify HTTPS certificates by
|
|
# default
|
|
pass
|
|
else:
|
|
# Handle target environment that doesn't support HTTPS
|
|
# verification
|
|
ssl._create_default_https_context = \
|
|
_create_unverified_https_context
|
|
|
|
self.client = Client(url=self.wsdl_url, location=self.soap_url)
|
|
assert self.client is not None
|
|
# print(self.client)
|
|
self.client.set_options(service='LsService', port='LsPort')
|
|
|
|
self.managedObjectReference = self.client.factory.create('ns0:ManagedObjectReference')
|
|
self.managedObjectReference._type = 'LookupServiceInstance'
|
|
self.managedObjectReference.value = 'ServiceInstance'
|
|
# print(self.managedObjectReference)
|
|
|
|
lookupServiceContent = self.client.service.RetrieveServiceContent(self.managedObjectReference)
|
|
# print(lookupServiceContent)
|
|
|
|
self.serviceRegistration = lookupServiceContent.serviceRegistration
|
|
# print(self.serviceRegistration)
|
|
|
|
def find_sso_urls(self):
|
|
"""
|
|
Finds all the SSO service URLs.
|
|
In an MxN setup where there are more than one infrastructure node; This method returns more than one URL.
|
|
|
|
:rtype: list
|
|
:return: list of SSO Service endpoint URLs
|
|
"""
|
|
return self.__find_platform_service_urls(product='com.vmware.cis',
|
|
service='cs.identity',
|
|
endpoint='com.vmware.cis.cs.identity.sso',
|
|
protocol='wsTrust')
|
|
|
|
def find_sso_url(self):
|
|
"""
|
|
Finds the SSO service URL.
|
|
In an MxN setup where there are more than one infrastructure node; This method returns the first SSO service endpoint URL
|
|
as returned by the lookup service.
|
|
|
|
:rtype: :class:`str`
|
|
:return: SSO Service endpoint URL
|
|
"""
|
|
result = self.__find_platform_service_urls(product='com.vmware.cis',
|
|
service='cs.identity',
|
|
endpoint='com.vmware.cis.cs.identity.sso',
|
|
protocol='wsTrust')
|
|
return result[0]
|
|
|
|
def find_vapi_urls(self):
|
|
"""
|
|
Finds all the vAPI service endpoint URLs.
|
|
In an MxN setup where there are more than one management node; this method returns more than one URL.
|
|
|
|
:rtype: dictionary
|
|
:return: vapi service endpoint URLs in a dictionary where the key is the node_id and the value is the service URL
|
|
"""
|
|
return self.__find_service_urls(product='com.vmware.cis',
|
|
service='cs.vapi',
|
|
endpoint='com.vmware.vapi.endpoint',
|
|
protocol='vapi.json.https.public')
|
|
|
|
def find_vapi_url(self, node_id):
|
|
"""
|
|
Finds the vapi service endpoint URL of a management node
|
|
|
|
:type: :class:`str`
|
|
:param node_id: The UUID of the management node
|
|
:rtype: :class:`str`
|
|
:return: vapi service endpoint URL of a management node or, None if no vapi endpoint is found
|
|
"""
|
|
assert node_id is not None
|
|
result = self.__find_service_urls(product='com.vmware.cis',
|
|
service='cs.vapi',
|
|
endpoint='com.vmware.vapi.endpoint',
|
|
protocol='vapi.json.https.public')
|
|
assert result is not None
|
|
return result.get(node_id)
|
|
|
|
def find_vim_urls(self):
|
|
"""
|
|
Finds all the vim service endpoint URLs.
|
|
In an MxN setup where there are more than one management node; this method returns more than one URL.
|
|
|
|
:rtype: dictionary
|
|
:return: vim service endpoint URLs in a dictionary where the key is the node_id and the value is the service URL
|
|
"""
|
|
return self.__find_service_urls(product='com.vmware.cis',
|
|
service='vcenterserver',
|
|
endpoint='com.vmware.vim',
|
|
protocol='vmomi')
|
|
|
|
def find_vim_url(self, node_id):
|
|
"""
|
|
Finds the vim service endpoint URL of a management node
|
|
|
|
:type: :class:`str`
|
|
:param node_id: The UUID of the management node
|
|
:rtype: :class:`str`
|
|
:return: vim service endpoint URL of a management node or, None if no vim endpoint is found
|
|
"""
|
|
assert node_id is not None
|
|
result = self.__find_service_urls(product='com.vmware.cis',
|
|
service='vcenterserver',
|
|
endpoint='com.vmware.vim',
|
|
protocol='vmomi')
|
|
assert result is not None
|
|
return result.get(node_id)
|
|
|
|
def find_vim_pbm_urls(self):
|
|
"""
|
|
Finds all the spbm service endpoint URLs.
|
|
In an MxN setup where there are more than one management node; this method returns more than one URL.
|
|
|
|
:rtype: dictionary
|
|
:return: spbm service endpoint URLs in a dictionary where the key is the node_id and the value is the service URL
|
|
"""
|
|
return self.__find_service_urls(product='com.vmware.vim.sms',
|
|
service='sms',
|
|
endpoint='com.vmware.vim.pbm',
|
|
protocol='https')
|
|
|
|
def find_vim_pbm_url(self, node_id):
|
|
"""
|
|
Finds the spbm service endpoint URL of a management node
|
|
|
|
:type: :class:`str`
|
|
:param node_id: The UUID of the management node
|
|
:rtype: :class:`str`
|
|
:return: spbm service endpoint URL of a management node or, None if no spbm endpoint is found
|
|
"""
|
|
assert node_id is not None
|
|
result = self.__find_service_urls(product='com.vmware.vim.sms',
|
|
service='sms',
|
|
endpoint='com.vmware.vim.pbm',
|
|
protocol='https')
|
|
assert result is not None
|
|
return result.get(node_id)
|
|
|
|
def __find_service_urls(self, product, service, endpoint, protocol):
|
|
"""
|
|
Finds the endpoint URLs of a service running on management nodes.
|
|
Returns a dictionary where the key is the management node id.
|
|
"""
|
|
assert self.client is not None
|
|
assert self.serviceRegistration is not None
|
|
|
|
lookupServiceRegistrationFilter = self.__create_filter_spec(product, service, endpoint, protocol)
|
|
# print(lookupServiceRegistrationFilter)
|
|
|
|
result = self.client.service.List(self.serviceRegistration, lookupServiceRegistrationFilter)
|
|
# print(result)
|
|
assert len(result) > 0
|
|
# Support for MxN
|
|
# return the results in a dictionary where key is NodeId and Value is Service URL
|
|
results_dict = {}
|
|
for lookupServiceRegistrationInfo in result:
|
|
lookupServiceRegistrationEndpoint = lookupServiceRegistrationInfo.serviceEndpoints[0]
|
|
assert lookupServiceRegistrationEndpoint is not None
|
|
results_dict[lookupServiceRegistrationInfo.nodeId] = lookupServiceRegistrationEndpoint.url
|
|
return results_dict
|
|
|
|
def __find_platform_service_urls(self, product, service, endpoint, protocol):
|
|
"""
|
|
Finds the endpoint URLs of a service running on PSCs (Platform Service Controller).
|
|
Returns a list of service URLs since there is no node id associated with the PSC.
|
|
"""
|
|
assert self.client is not None
|
|
assert self.serviceRegistration is not None
|
|
|
|
lookupServiceRegistrationFilter = self.__create_filter_spec(product, service, endpoint, protocol)
|
|
# print(lookupServiceRegistrationFilter)
|
|
|
|
result = self.client.service.List(self.serviceRegistration, lookupServiceRegistrationFilter)
|
|
# print(result)
|
|
assert len(result) > 0
|
|
|
|
urls = []
|
|
for lookupServiceRegistrationInfo in result:
|
|
lookupServiceRegistrationEndpoint = lookupServiceRegistrationInfo.serviceEndpoints[0]
|
|
assert lookupServiceRegistrationEndpoint is not None
|
|
urls.append(lookupServiceRegistrationEndpoint.url)
|
|
return urls
|
|
|
|
def __create_filter_spec(self, product, service, endpoint, protocol):
|
|
assert self.client is not None
|
|
|
|
lookupServiceRegistrationServiceType = self.client.factory.create('ns0:LookupServiceRegistrationServiceType')
|
|
lookupServiceRegistrationServiceType.product = product
|
|
lookupServiceRegistrationServiceType.type = service
|
|
|
|
lookupServiceRegistrationEndpointType = self.client.factory.create('ns0:LookupServiceRegistrationEndpointType')
|
|
lookupServiceRegistrationEndpointType.type = endpoint
|
|
lookupServiceRegistrationEndpointType.protocol = protocol
|
|
|
|
lookupServiceRegistrationFilter = self.client.factory.create('ns0:LookupServiceRegistrationFilter')
|
|
lookupServiceRegistrationFilter.serviceType = lookupServiceRegistrationServiceType
|
|
lookupServiceRegistrationFilter.endpointType = lookupServiceRegistrationEndpointType
|
|
return lookupServiceRegistrationFilter
|
|
|
|
def find_mgmt_nodes(self):
|
|
"""
|
|
Finds all the management nodes
|
|
|
|
:rtype: dictionary
|
|
:return: management node instance name and node id (UUID) in a dictionary
|
|
"""
|
|
assert self.client is not None
|
|
assert self.serviceRegistration is not None
|
|
|
|
lookupServiceRegistrationServiceType = self.client.factory.create('ns0:LookupServiceRegistrationServiceType')
|
|
lookupServiceRegistrationServiceType.product = 'com.vmware.cis'
|
|
lookupServiceRegistrationServiceType.type = 'vcenterserver'
|
|
# print(lookupServiceRegistrationServiceType)
|
|
|
|
lookupServiceRegistrationEndpointType = self.client.factory.create('ns0:LookupServiceRegistrationEndpointType')
|
|
lookupServiceRegistrationEndpointType.type = 'com.vmware.vim'
|
|
lookupServiceRegistrationEndpointType.protocol = 'vmomi'
|
|
# print(lookupServiceRegistrationEndpointType)
|
|
|
|
lookupServiceRegistrationFilter = self.client.factory.create('ns0:LookupServiceRegistrationFilter')
|
|
lookupServiceRegistrationFilter.serviceType = lookupServiceRegistrationServiceType
|
|
lookupServiceRegistrationFilter.endpointType = lookupServiceRegistrationEndpointType
|
|
# print(lookupServiceRegistrationFilter)
|
|
|
|
result = self.client.service.List(self.serviceRegistration, lookupServiceRegistrationFilter)
|
|
# print(result)
|
|
assert len(result) > 0
|
|
|
|
results_dict = {}
|
|
for lookupServiceRegistrationInfo in result:
|
|
for lookupServiceRegistrationAttribute in lookupServiceRegistrationInfo.serviceAttributes:
|
|
if lookupServiceRegistrationAttribute.key == 'com.vmware.vim.vcenter.instanceName':
|
|
results_dict[lookupServiceRegistrationAttribute.value] = lookupServiceRegistrationInfo.nodeId
|
|
return results_dict
|
|
|
|
def get_mgmt_node_id(self, instance_name):
|
|
"""
|
|
Get the management node id from the instance name
|
|
|
|
:type: :class:`str`
|
|
:param instance_name: The instance name of the management node
|
|
:rtype: :class:`str`
|
|
:return: The UUID of the management node or, None is no management node is found by the given instance name
|
|
"""
|
|
result = self.find_mgmt_nodes()
|
|
assert result is not None
|
|
return result.get(instance_name)
|
|
|
|
def get_mgmt_node_instance_name(self, node_id):
|
|
result = self.find_mgmt_nodes()
|
|
assert result is not None
|
|
for k, v in result.items():
|
|
if v == node_id:
|
|
return k
|
|
|
|
def get_default_mgmt_node(self):
|
|
"""
|
|
Finds the instance name and UUID of the management node for M1xN or, when the PSC and
|
|
management services all reside on a single node (embedded).
|
|
"""
|
|
result = self.find_mgmt_nodes()
|
|
assert result is not None
|
|
if len(result) < 1:
|
|
raise Exception('No management node found')
|
|
if len(result) > 1:
|
|
raise MultipleManagementNodeException(MultipleManagementNodeException.format(result))
|
|
return list(result.keys())[0], list(result.values())[0] # python 3.x dict.keys() returns a view rather than a list
|
|
|
|
|
|
class MultipleManagementNodeException(Exception):
|
|
def __init__(self, message):
|
|
super(MultipleManagementNodeException, self).__init__(message)
|
|
|
|
@staticmethod
|
|
def format(nodes):
|
|
"""
|
|
Formats the multiple management node exception message
|
|
|
|
:type: :class:`dict`
|
|
:param nodes: The dictionary containing management nodes
|
|
:rtype: :class:`str`
|
|
:return: Formatted string output
|
|
"""
|
|
message = 'Multiple Management Node Found on server'
|
|
for k, v in nodes.items():
|
|
message = message + os.linesep + 'Node name: {0} uuid: {1}'.format(k, v)
|
|
return message
|
|
|
|
|
|
def main():
|
|
lookup_service_helper = LookupServiceHelper(wsdl_url='file:///path/to/lookupservice.wsdl',
|
|
soap_url='https://server_ip/lookupservice/sdk')
|
|
lookup_service_helper.connect()
|
|
print(lookup_service_helper.find_sso_url())
|
|
print(lookup_service_helper.find_vapi_urls())
|
|
print(lookup_service_helper.find_vim_urls())
|
|
print(lookup_service_helper.find_vim_pbm_urls())
|
|
print(lookup_service_helper.find_mgmt_nodes())
|
|
|
|
# Start program
|
|
if __name__ == "__main__":
|
|
main() |