1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
|
# Copyright: (c) 2020, Lionel Hercot (@lhercot) <lhercot@cisco.com>
# Copyright: (c) 2020, Cindy Zhao (@cizhao) <cizhao@cisco.com>
# Copyright: (c) 2023, Akini Ross (@akinross) <akinross@cisco.com>
#
# GNU General Public License v3.0+ (see LICENSE or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
DOCUMENTATION = """
---
name: mso
short_description: MSO Ansible HTTPAPI Plugin.
description:
- This MSO plugin provides the HTTPAPI transport methods needed to initiate
a connection to MSO, send API requests and process the
response.
version_added: "1.2.0"
options:
login_domain:
description:
- The login domain name to use for authentication.
- The default value is Local.
type: string
env:
- name: ANSIBLE_HTTPAPI_LOGIN_DOMAIN
vars:
- name: ansible_httpapi_login_domain
"""
import json
import re
import traceback
from ansible.module_utils.six import PY3
from ansible.module_utils._text import to_text
from ansible.module_utils.connection import ConnectionError
from ansible.plugins.httpapi import HttpApiBase
from copy import copy
CONNECTION_MAP = {"username": "remote_user", "timeout": "persistent_command_timeout"}
RESET_KEYS = ["username", "password", "login_domain", "host", "port"]
CONNECTION_KEYS = RESET_KEYS + ["use_proxy", "use_ssl", "timeout", "validate_certs"]
class HttpApi(HttpApiBase):
def __init__(self, *args, **kwargs):
super(HttpApi, self).__init__(*args, **kwargs)
self.platform = "cisco.mso"
self.headers = {"Content-Type": "application/json"}
self.params = {}
self.auth = None
self.backup_hosts = None
self.host_counter = 0
self.error = None
self.method = "GET"
self.path = ""
self.status = -1
self.info = {}
self.connection_parameters = {}
def get_platform(self):
return self.platform
def set_params(self, params):
self.params = params
def set_backup_hosts(self):
try:
list_of_hosts = re.sub(r"[[\]]", "", self.connection.get_option("host")).split(",")
# ipaddress.ip_address(list_of_hosts[0])
return list_of_hosts
except Exception:
return []
def login(self, username, password):
"""Log in to MSO"""
# Perform login request
self.connection.queue_message("vvvv", "Starting Login to {0}".format(self.connection.get_option("host")))
method = "POST"
path = "/mso/api/v1/auth/login"
full_path = self.connection.get_option("host") + path
payload = {"username": username, "password": password}
if self.connection_parameters["login_domain"] is not None and self.connection_parameters["login_domain"] != "Local":
payload["domainId"] = self._get_login_domain_id(self.connection_parameters["login_domain"])
data = json.dumps(payload)
try:
payload.pop("password")
self.connection.queue_message("vvvv", "login() - connection.send({0}, {1}, {2}, {3})".format(path, payload, method, self.headers))
response, response_data = self.connection.send(path, data, method=method, headers=self.headers)
# Handle MSO response
self.status = response.getcode()
if self.status != 201:
self.connection.queue_message("vvvv", "login status incorrect status={0}".format(self.status))
json_response = self._response_to_json(response_data)
self.error = dict(code=self.status, message="Authentication failed: {0}".format(json_response))
raise ConnectionError(json.dumps(self._verify_response(response, method, full_path, response_data)))
self.connection._auth = {"Authorization": "Bearer {0}".format(self._response_to_json(response_data).get("token"))}
except ConnectionError:
self.connection.queue_message("vvvv", "login() - ConnectionError Exception")
raise
except Exception as e:
self.connection.queue_message("vvvv", "login() - Generic Exception")
self.error = dict(code=self.status, message="Authentication failed: Request failed: {0}".format(e))
raise ConnectionError(json.dumps(self._verify_response(None, method, full_path, None)))
def logout(self):
method = "DELETE"
path = "/mso/api/v1/auth/logout"
try:
self.connection.send(path, {}, method=method, headers=self.headers)
except Exception as e:
self.error = dict(code=self.status, message="Error on attempt to logout from MSO. {0}".format(e))
raise ConnectionError(json.dumps(self._verify_response(None, method, self.connection.get_option("host") + path, None)))
self.connection._auth = None
def send_request(self, method, path, data=None):
"""This method handles all MSO REST API requests other than login"""
self.error = None
self.path = ""
self.status = -1
self.info = {}
self.method = "GET"
if data is None:
data = {}
self.connection.queue_message("vvvv", "send_request method called")
self.set_connection_parameters()
# Perform some very basic path input validation.
path = str(path)
if path[0] != "/":
self.error = dict(code=self.status, message="Value of <path> does not appear to be formated properly")
raise ConnectionError(json.dumps(self._verify_response(None, method, path, None)))
full_path = self.connection.get_option("host") + path
try:
self.connection.queue_message("vvvv", "send_request() - connection.send({0}, {1}, {2}, {3})".format(path, data, method, self.headers))
response, rdata = self.connection.send(path, data, method=method, headers=self.headers)
except ConnectionError:
self.connection.queue_message("vvvv", "login() - ConnectionError Exception")
raise
except Exception as e:
self.connection.queue_message("vvvv", "send_request() - Generic Exception")
if self.error is None:
self.error = dict(code=self.status, message="MSO HTTPAPI send_request() Exception: {0} - {1}".format(e, traceback.format_exc()))
raise ConnectionError(json.dumps(self._verify_response(None, method, full_path, None)))
return self._verify_response(response, method, full_path, rdata)
def set_connection_parameters(self):
connection_parameters = {}
for key in CONNECTION_KEYS:
if key == "login_domain":
value = self.params.get(key) if self.params.get(key) is not None else self.get_option(CONNECTION_MAP.get(key, key))
self.set_option(key, value)
else:
value = self.params.get(key) if self.params.get(key) is not None else self.connection.get_option(CONNECTION_MAP.get(key, key))
self.connection.set_option(CONNECTION_MAP.get(key, key), value)
connection_parameters[key] = value
if value != self.connection_parameters.get(key) and key in RESET_KEYS:
self.connection._connected = False
self.connection.queue_message("vvvv", "set_connection_parameters() - resetting connection due to '{0}' change".format(key))
if self.connection_parameters != connection_parameters:
self.connection_parameters = copy(connection_parameters)
connection_parameters.pop("password")
msg = "set_connection_parameters() - changed connection parameters {0}".format(connection_parameters)
self.connection.queue_message("vvvv", msg)
def _verify_response(self, response, method, path, data):
"""Process the return code and response object from MSO"""
response_data = None
response_code = -1
self.info.update(dict(url=path))
if data is not None:
response_data = self._response_to_json(data)
if response is not None:
response_code = response.getcode()
path = response.geturl()
self.info.update(self._get_formated_info(response))
# Handle possible MSO error information
if response_code not in [200, 201, 202, 204]:
self.error = dict(code=self.status, message=response_data)
self.info["method"] = method
if self.error is not None:
self.info["error"] = self.error
self.info["body"] = response_data
return self.info
def _response_to_json(self, response_data):
"""Convert response_data to json format"""
try:
response_value = response_data.getvalue()
except Exception:
response_value = response_data
response_text = to_text(response_value)
try:
return json.loads(response_text) if response_text else {}
# JSONDecodeError only available on Python 3.5+
except Exception as e:
# Expose RAW output for troubleshooting
self.error = dict(code=-1, message="Unable to parse output as JSON, see 'raw' output. {0}".format(e))
self.info["raw"] = response_text
return
def _get_login_domain_id(self, domain_name):
"""Get a domain and return its id"""
if domain_name is None:
return None
method = "GET"
path = "/mso/api/v1/auth/login-domains"
full_path = self.connection.get_option("host") + path
# TODO: Replace response by -
response, data = self.connection.send(path, None, method=method, headers=self.headers)
if data is not None:
response_data = self._response_to_json(data)
domains = response_data.get("domains")
if domains is not None:
for domain in domains:
if domain.get("name") == domain_name:
if "id" in domain:
return domain.get("id")
else:
self.error = dict(code=-1, message="Login domain lookup failed for domain '{0}': {1}".format(domain_name, domain))
raise ConnectionError(json.dumps(self._verify_response(None, method, full_path, None)))
self.error = dict(code=-1, message="Login domain '{0}' is not a valid domain name.".format(domain_name))
raise ConnectionError(json.dumps(self._verify_response(None, method, full_path, None)))
else:
self.error = dict(code=-1, message="Key 'domains' missing from data")
raise ConnectionError(json.dumps(self._verify_response(None, method, full_path, None)))
def _get_formated_info(self, response):
"""The code in this function is based out of Ansible fetch_url code
at https://github.com/ansible/ansible/blob/devel/lib/ansible/module_utils/urls.py"""
info = dict(msg="OK (%s bytes)" % response.headers.get("Content-Length", "unknown"), url=response.geturl(), status=response.getcode())
# Lowercase keys, to conform to py2 behavior, so that py3 and py2 are predictable
info.update(dict((k.lower(), v) for k, v in response.info().items()))
# Don't be lossy, append header values for duplicate headers
# In Py2 there is nothing that needs done, py2 does this for us
if PY3:
temp_headers = {}
for name, value in response.headers.items():
# The same as above, lower case keys to match py2 behavior, and create more consistent results
name = name.lower()
if name in temp_headers:
temp_headers[name] = ", ".join((temp_headers[name], value))
else:
temp_headers[name] = value
info.update(temp_headers)
return info
|
