-
Notifications
You must be signed in to change notification settings - Fork 389
/
manager.py
349 lines (275 loc) · 11.9 KB
/
manager.py
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
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
# Copyright 2009 Shikhar Bhushan
# Copyright 2011 Leonidas Poulopoulos
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
This module is a thin layer of abstraction around the library.
It exposes all core functionality.
"""
from ncclient import capabilities
from ncclient import operations
from ncclient import transport
import six
import logging
import functools
from ncclient.xml_ import *
logger = logging.getLogger('ncclient.manager')
OPERATIONS = {
"get": operations.Get,
"get_config": operations.GetConfig,
"get_schema": operations.GetSchema,
"dispatch": operations.Dispatch,
"edit_config": operations.EditConfig,
"copy_config": operations.CopyConfig,
"validate": operations.Validate,
"commit": operations.Commit,
"discard_changes": operations.DiscardChanges,
"cancel_commit": operations.CancelCommit,
"delete_config": operations.DeleteConfig,
"lock": operations.Lock,
"unlock": operations.Unlock,
"create_subscription": operations.CreateSubscription,
"close_session": operations.CloseSession,
"kill_session": operations.KillSession,
"poweroff_machine": operations.PoweroffMachine,
"reboot_machine": operations.RebootMachine,
}
"""
Dictionary of base method names and corresponding :class:`~ncclient.operations.RPC`
subclasses. It is used to lookup operations, e.g. `get_config` is mapped to
:class:`~ncclient.operations.GetConfig`. It is thus possible to add additional
operations to the :class:`Manager` API.
"""
VENDOR_OPERATIONS = {}
def make_device_handler(device_params):
"""
Create a device handler object that provides device specific parameters and
functions, which are called in various places throughout our code.
If no device_params are defined or the "name" in the parameter dict is not
known then a default handler will be returned.
"""
if device_params is None:
device_params = {}
handler = device_params.get('handler', None)
if handler:
return handler(device_params)
device_name = device_params.get("name", "default")
# Attempt to import device handler class. All device handlers are
# in a module called "ncclient.devices.<devicename>" and in a class named
# "<devicename>DeviceHandler", with the first letter capitalized.
class_name = "%sDeviceHandler" % device_name.capitalize()
devices_module_name = "ncclient.devices.%s" % device_name
dev_module_obj = __import__(devices_module_name)
handler_module_obj = getattr(getattr(dev_module_obj, "devices"), device_name)
class_obj = getattr(handler_module_obj, class_name)
handler_obj = class_obj(device_params)
return handler_obj
def _extract_device_params(kwds):
device_params = kwds.pop("device_params", None)
return device_params
def _extract_manager_params(kwds):
manager_params = kwds.pop("manager_params", {})
# To maintain backward compatibility
if 'timeout' not in manager_params and 'timeout' in kwds:
manager_params['timeout'] = kwds['timeout']
return manager_params
def connect_ssh(*args, **kwds):
"""
Initialize a :class:`Manager` over the SSH transport.
For documentation of arguments see :meth:`ncclient.transport.SSHSession.connect`.
The underlying :class:`ncclient.transport.SSHSession` is created with
:data:`CAPABILITIES`. It is first instructed to
:meth:`~ncclient.transport.SSHSession.load_known_hosts` and then
all the provided arguments are passed directly to its implementation
of :meth:`~ncclient.transport.SSHSession.connect`.
To customize the :class:`Manager`, add a `manager_params` dictionary in connection
parameters (e.g. `manager_params={'timeout': 60}` for a bigger RPC timeout parameter)
To invoke advanced vendor related operation add
`device_params={'name': '<vendor_alias>'}` in connection parameters. For the time,
'junos' and 'nexus' are supported for Juniper and Cisco Nexus respectively.
A custom device handler can be provided with
`device_params={'handler':<handler class>}` in connection parameters.
"""
# Extract device parameter and manager parameter dictionaries, if they were passed into this function.
# Remove them from kwds (which should keep only session.connect() parameters).
device_params = _extract_device_params(kwds)
manager_params = _extract_manager_params(kwds)
device_handler = make_device_handler(device_params)
device_handler.add_additional_ssh_connect_params(kwds)
global VENDOR_OPERATIONS
VENDOR_OPERATIONS.update(device_handler.add_additional_operations())
session = transport.SSHSession(device_handler)
if "hostkey_verify" not in kwds or kwds["hostkey_verify"]:
session.load_known_hosts()
try:
session.connect(*args, **kwds)
except Exception as ex:
if session.transport:
session.close()
raise
return Manager(session, device_handler, **manager_params)
def connect_ioproc(*args, **kwds):
device_params = _extract_device_params(kwds)
manager_params = _extract_manager_params(kwds)
if device_params:
import_string = 'ncclient.transport.third_party.'
import_string += device_params['name'] + '.ioproc'
third_party_import = __import__(import_string, fromlist=['IOProc'])
device_handler = make_device_handler(device_params)
global VENDOR_OPERATIONS
VENDOR_OPERATIONS.update(device_handler.add_additional_operations())
session = third_party_import.IOProc(device_handler)
session.connect()
return Manager(session, device_handler, **manager_params)
def connect(*args, **kwds):
if "host" in kwds:
host = kwds["host"]
device_params = kwds.get('device_params', {})
if host == 'localhost' and device_params.get('name') == 'junos' \
and device_params.get('local'):
return connect_ioproc(*args, **kwds)
else:
return connect_ssh(*args, **kwds)
class Manager(object):
"""
For details on the expected behavior of the operations and their
parameters refer to :rfc:`4741`.
Manager instances are also context managers so you can use it like this::
with manager.connect("host") as m:
# do your stuff
... or like this::
m = manager.connect("host")
try:
# do your stuff
finally:
m.close_session()
"""
# __metaclass__ = OpExecutor
HUGE_TREE_DEFAULT = False
"""Default for `huge_tree` support for XML parsing of RPC replies (defaults to False)"""
def __init__(self, session, device_handler, timeout=30):
self._session = session
self._async_mode = False
self._timeout = timeout
self._raise_mode = operations.RaiseMode.ALL
self._huge_tree = self.HUGE_TREE_DEFAULT
self._device_handler = device_handler
def __enter__(self):
return self
def __exit__(self, *args):
self.close_session()
return False
def __set_timeout(self, timeout):
self._timeout = timeout
def __set_async_mode(self, mode):
self._async_mode = mode
def __set_raise_mode(self, mode):
assert(mode in (operations.RaiseMode.NONE, operations.RaiseMode.ERRORS, operations.RaiseMode.ALL))
self._raise_mode = mode
def execute(self, cls, *args, **kwds):
return cls(self._session,
device_handler=self._device_handler,
async_mode=self._async_mode,
timeout=self._timeout,
raise_mode=self._raise_mode,
huge_tree=self._huge_tree).request(*args, **kwds)
def locked(self, target):
"""Returns a context manager for a lock on a datastore, where
*target* is the name of the configuration datastore to lock, e.g.::
with m.locked("running"):
# do your stuff
... instead of::
m.lock("running")
try:
# do your stuff
finally:
m.unlock("running")
"""
return operations.LockContext(self._session, self._device_handler, target)
def scp(self):
return self._session.scp()
def session(self):
raise NotImplementedError
def __getattr__(self, method):
if method in VENDOR_OPERATIONS:
return functools.partial(self.execute, VENDOR_OPERATIONS[method])
elif method in OPERATIONS:
return functools.partial(self.execute, OPERATIONS[method])
else:
"""Parse args/kwargs correctly in order to build XML element"""
def _missing(*args, **kwargs):
m = method.replace('_', '-')
root = new_ele(m)
if args:
for arg in args:
sub_ele(root, arg)
r = self.rpc(root)
return r
return _missing
def take_notification(self, block=True, timeout=None):
"""Attempt to retrieve one notification from the queue of received
notifications.
If block is True, the call will wait until a notification is
received.
If timeout is a number greater than 0, the call will wait that
many seconds to receive a notification before timing out.
If there is no notification available when block is False or
when the timeout has elapse, None will be returned.
Otherwise a :class:`~ncclient.operations.notify.Notification`
object will be returned.
"""
return self._session.take_notification(block, timeout)
@property
def client_capabilities(self):
""":class:`~ncclient.capabilities.Capabilities` object representing
the client's capabilities."""
return self._session._client_capabilities
@property
def server_capabilities(self):
""":class:`~ncclient.capabilities.Capabilities` object representing
the server's capabilities."""
return self._session._server_capabilities
@property
def channel_id(self):
return self._session._channel_id
@property
def channel_name(self):
return self._session._channel_name
@property
def session_id(self):
"""`session-id` assigned by the NETCONF server."""
return self._session.id
@property
def connected(self):
"""Whether currently connected to the NETCONF server."""
return self._session.connected
async_mode = property(fget=lambda self: self._async_mode,
fset=__set_async_mode)
"""Specify whether operations are executed asynchronously (`True`) or
synchronously (`False`) (the default)."""
timeout = property(fget=lambda self: self._timeout, fset=__set_timeout)
"""Specify the timeout for synchronous RPC requests."""
raise_mode = property(fget=lambda self: self._raise_mode,
fset=__set_raise_mode)
"""Specify which errors are raised as :exc:`~ncclient.operations.RPCError`
exceptions. Valid values are the constants defined in
:class:`~ncclient.operations.RaiseMode`.
The default value is :attr:`~ncclient.operations.RaiseMode.ALL`."""
@property
def huge_tree(self):
"""Whether `huge_tree` support for XML parsing of RPC replies is enabled (default=False)
The default value is configurable through :attr:`~ncclient.manager.Manager.HUGE_TREE_DEFAULT`"""
return self._huge_tree
@huge_tree.setter
def huge_tree(self, x):
self._huge_tree = x