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
|
# (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
httpapi: 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: "2.6"
options:
eos_use_sessions:
type: int
default: 1
description:
- Specifies if sessions should be used on remote host or not
env:
- name: ANSIBLE_EOS_USE_SESSIONS
vars:
- name: ansible_eos_use_sessions
version_added: '2.8'
"""
import json
from ansible.module_utils._text import to_text
from ansible.module_utils.connection import ConnectionError
from ansible.module_utils.network.common.utils import to_list
from ansible.plugins.httpapi import HttpApiBase
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):
use_session = self.get_option('eos_use_sessions')
try:
use_session = int(use_session)
except ValueError:
pass
if not bool(use_session):
self._session_support = False
else:
if self._session_support:
return self._session_support
response = self.send_request('show configuration sessions')
self._session_support = 'error' not in response
return self._session_support
def send_request(self, data, **message_kwargs):
data = to_list(data)
if self._become:
self.connection.queue_message('vvvv', 'firing event: on_become')
data.insert(0, {"cmd": "enable", "input": self._become_pass})
output = message_kwargs.get('output', 'text')
request = request_builder(data, output)
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 self._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 | 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 | 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)
def handle_response(response):
if 'error' in response:
error = response['error']
error_text = []
for data in error['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, reqid=None):
params = dict(version=1, cmds=commands, format=output)
return json.dumps(dict(jsonrpc='2.0', id=reqid, method='runCmds', params=params))
|
