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
|
# (c) 2018 Red Hat Inc.
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
DOCUMENTATION = """
author: Ansible Networking Team (@ansible-network)
name: eos
short_description: Use eAPI to run command on eos platform
description:
- This eos plugin provides low level abstraction api's for sending and receiving CLI
commands with eos network devices.
version_added: 1.0.0
options:
eos_use_sessions:
type: bool
default: yes
description:
- Specifies if sessions should be used on remote host or not
env:
- name: ANSIBLE_EOS_USE_SESSIONS
vars:
- name: ansible_eos_use_sessions
"""
import json
from ansible.errors import AnsibleConnectionFailure
from ansible.module_utils._text import to_text
from ansible.module_utils.connection import ConnectionError
from ansible_collections.ansible.netcommon.plugins.module_utils.network.common.utils import to_list
from ansible_collections.ansible.netcommon.plugins.plugin_utils.httpapi_base import HttpApiBase
from ansible_collections.arista.eos.plugins.module_utils.network.eos.eos import session_name
OPTIONS = {
"format": ["text", "json"],
"diff_match": ["line", "strict", "exact", "none"],
"diff_replace": ["line", "block", "config"],
"output": ["text", "json"],
}
class HttpApi(HttpApiBase):
def __init__(self, *args, **kwargs):
super(HttpApi, self).__init__(*args, **kwargs)
self._device_info = None
self._session_support = None
def supports_sessions(self):
if not self.get_option("eos_use_sessions"):
self._session_support = False
else:
if self._session_support:
return self._session_support
try:
response = self.send_request("show configuration sessions")
self._session_support = "error" not in response
except AnsibleConnectionFailure:
self._session_support = False
return self._session_support
def send_request(self, data, **message_kwargs):
data = to_list(data)
become = self._become
if become:
self.connection.queue_message("vvvv", "firing event: on_become")
data.insert(0, {"cmd": "enable", "input": self._become_pass})
output = message_kwargs.get("output") or "text"
version = message_kwargs.get("version") or "latest"
request = request_builder(data, output, version)
headers = {"Content-Type": "application/json-rpc"}
_response, response_data = self.connection.send(
"/command-api",
request,
headers=headers,
method="POST",
)
try:
response_data = json.loads(to_text(response_data.getvalue()))
except ValueError:
raise ConnectionError(
"Response was not valid JSON, got {0}".format(
to_text(response_data.getvalue()),
),
)
results = handle_response(response_data)
if become:
results = results[1:]
if len(results) == 1:
results = results[0]
return results
def get_device_info(self):
if self._device_info:
return self._device_info
device_info = {}
device_info["network_os"] = "eos"
reply = self.send_request("show version", output="json")
data = json.loads(reply)
device_info["network_os_version"] = data["version"]
device_info["network_os_model"] = data["modelName"]
reply = self.send_request("show hostname", output="json")
data = json.loads(reply)
device_info["network_os_hostname"] = data["hostname"]
self._device_info = device_info
return self._device_info
def get_device_operations(self):
return {
"supports_diff_replace": True,
"supports_commit": bool(self.supports_sessions()),
"supports_rollback": False,
"supports_defaults": False,
"supports_onbox_diff": bool(self.supports_sessions()),
"supports_commit_comment": False,
"supports_multiline_delimiter": False,
"supports_diff_match": True,
"supports_diff_ignore_lines": True,
"supports_generate_diff": not bool(self.supports_sessions()),
"supports_replace": bool(self.supports_sessions()),
}
def get_capabilities(self):
result = {}
result["rpc"] = []
result["device_info"] = self.get_device_info()
result["device_operations"] = self.get_device_operations()
result.update(OPTIONS)
result["network_api"] = "eapi"
return json.dumps(result)
# Shims for resource module support
def get(self, command, output=None):
# This method is ONLY here to support resource modules. Therefore most
# arguments are unsupported and not present.
return self.send_request(data=command, output=output)
def edit_config(self, candidate):
# This method is ONLY here to support resource modules. Therefore most
# arguments are unsupported and not present.
session = None
if self.supports_sessions():
session = session_name()
candidate = ["configure session %s" % session] + candidate
else:
candidate = ["configure"] + candidate
candidate.append("commit")
try:
responses = self.send_request(candidate)
except ConnectionError:
if session:
self.send_request(["configure session %s" % session, "abort"])
raise
return [resp for resp in to_list(responses) if resp != "{}"]
def handle_response(response):
if "error" in response:
error = response["error"]
error_text = []
for data in error.get("data", []):
error_text.extend(data.get("errors", []))
error_text = "\n".join(error_text) or error["message"]
raise ConnectionError(error_text, code=error["code"])
results = []
for result in response["result"]:
if "messages" in result:
results.append(result["messages"][0])
elif "output" in result:
results.append(result["output"].strip())
else:
results.append(json.dumps(result))
return results
def request_builder(commands, output, version, reqid=None):
if version != "latest":
version = int(version)
params = dict(version=version, cmds=commands, format=output)
return json.dumps(
dict(jsonrpc="2.0", id=reqid, method="runCmds", params=params),
)
|
