1
0
Fork 0
mirror of https://github.com/ansible-collections/community.general.git synced 2024-09-14 20:13:21 +02:00
community.general/plugins/module_utils/remote_management/dellemc/ome.py
Ansible Core Team aebc1b03fd Initial commit
2020-03-09 09:11:07 +00:00

181 lines
7.1 KiB
Python

# -*- coding: utf-8 -*-
# Dell EMC OpenManage Ansible Modules
# Version 1.3
# Copyright (C) 2019 Dell Inc. or its subsidiaries. All Rights Reserved.
# Redistribution and use in source and binary forms, with or without modification,
# are permitted provided that the following conditions are met:
# * Redistributions of source code must retain the above copyright notice,
# this list of conditions and the following disclaimer.
# * Redistributions in binary form must reproduce the above copyright notice,
# this list of conditions and the following disclaimer in the documentation
# and/or other materials provided with the distribution.
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
# IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
# INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
# PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
# USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import json
from ansible.module_utils.urls import open_url, ConnectionError, SSLValidationError
from ansible.module_utils.six.moves.urllib.error import URLError, HTTPError
from ansible.module_utils.six.moves.urllib.parse import urlencode
SESSION_RESOURCE_COLLECTION = {
"SESSION": "SessionService/Sessions",
"SESSION_ID": "SessionService/Sessions('{Id}')",
}
class OpenURLResponse(object):
"""Handles HTTPResponse"""
def __init__(self, resp):
self.body = None
self.resp = resp
if self.resp:
self.body = self.resp.read()
@property
def json_data(self):
try:
return json.loads(self.body)
except ValueError:
raise ValueError("Unable to parse json")
@property
def status_code(self):
return self.resp.getcode()
@property
def success(self):
return self.status_code in (200, 201, 202, 204)
@property
def token_header(self):
return self.resp.headers.get('X-Auth-Token')
class RestOME(object):
"""Handles OME API requests"""
def __init__(self, module_params=None, req_session=False):
self.module_params = module_params
self.hostname = self.module_params["hostname"]
self.username = self.module_params["username"]
self.password = self.module_params["password"]
self.port = self.module_params["port"]
self.req_session = req_session
self.session_id = None
self.protocol = 'https'
self._headers = {'Content-Type': 'application/json', 'Accept': 'application/json'}
def _get_base_url(self):
"""builds base url"""
return '{0}://{1}:{2}/api'.format(self.protocol, self.hostname, self.port)
def _build_url(self, path, query_param=None):
"""builds complete url"""
url = path
base_uri = self._get_base_url()
if path:
url = '{0}/{1}'.format(base_uri, path)
if query_param:
url += "?{0}".format(urlencode(query_param))
return url
def _url_common_args_spec(self, method, api_timeout, headers=None):
"""Creates an argument common spec"""
req_header = self._headers
if headers:
req_header.update(headers)
url_kwargs = {
"method": method,
"validate_certs": False,
"use_proxy": True,
"headers": req_header,
"timeout": api_timeout,
"follow_redirects": 'all',
}
return url_kwargs
def _args_without_session(self, method, api_timeout=30, headers=None):
"""Creates an argument spec in case of basic authentication"""
req_header = self._headers
if headers:
req_header.update(headers)
url_kwargs = self._url_common_args_spec(method, api_timeout, headers=headers)
url_kwargs["url_username"] = self.username
url_kwargs["url_password"] = self.password
url_kwargs["force_basic_auth"] = True
return url_kwargs
def _args_with_session(self, method, api_timeout=30, headers=None):
"""Creates an argument spec, in case of authentication with session"""
url_kwargs = self._url_common_args_spec(method, api_timeout, headers=headers)
url_kwargs["force_basic_auth"] = False
return url_kwargs
def invoke_request(self, method, path, data=None, query_param=None, headers=None,
api_timeout=30, dump=True):
"""
Sends a request via open_url
Returns :class:`OpenURLResponse` object.
:arg method: HTTP verb to use for the request
:arg path: path to request without query parameter
:arg data: (optional) Payload to send with the request
:arg query_param: (optional) Dictionary of query parameter to send with request
:arg headers: (optional) Dictionary of HTTP Headers to send with the
request
:arg api_timeout: (optional) How long to wait for the server to send
data before giving up
:arg dump: (Optional) boolean value for dumping payload data.
:returns: OpenURLResponse
"""
try:
if 'X-Auth-Token' in self._headers:
url_kwargs = self._args_with_session(method, api_timeout, headers=headers)
else:
url_kwargs = self._args_without_session(method, api_timeout, headers=headers)
if data and dump:
data = json.dumps(data)
url = self._build_url(path, query_param=query_param)
resp = open_url(url, data=data, **url_kwargs)
resp_data = OpenURLResponse(resp)
except (HTTPError, URLError, SSLValidationError, ConnectionError) as err:
raise err
return resp_data
def __enter__(self):
"""Creates sessions by passing it to header"""
if self.req_session:
payload = {'UserName': self.username,
'Password': self.password,
'SessionType': 'API', }
path = SESSION_RESOURCE_COLLECTION["SESSION"]
resp = self.invoke_request('POST', path, data=payload)
if resp and resp.success:
self.session_id = resp.json_data.get("Id")
self._headers["X-Auth-Token"] = resp.token_header
else:
msg = "Could not create the session"
raise ConnectionError(msg)
return self
def __exit__(self, exc_type, exc_value, traceback):
"""Deletes a session id, which is in use for request"""
if self.session_id:
path = SESSION_RESOURCE_COLLECTION["SESSION_ID"].format(Id=self.session_id)
self.invoke_request('DELETE', path)
return False