forked from CiscoDevNet/ncc
-
Notifications
You must be signed in to change notification settings - Fork 0
/
ncc.py
executable file
·394 lines (351 loc) · 13.8 KB
/
ncc.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
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
#!/usr/bin/env python
import sys
import os
from argparse import ArgumentParser
from ncclient import manager
from jinja2 import Environment
from jinja2.exceptions import UndefinedError
from jinja2 import meta
from jinja2 import FileSystemLoader
from jinja2 import StrictUndefined
from jinja2 import Template
from lxml import etree
import logging
import json
import re
#
# Add things people want logged here. Just various netconf things for
# now. SSH disabled as it is just too much right now.
#
LOGGING_TO_ENABLE = [
'ncclient.transport.ssh',
'ncclient.transport.session',
'ncclient.operations.rpc'
]
#
# Capability constants
#
NC_WRITABLE_RUNNING = 'urn:ietf:params:netconf:capability:writable-running:1.0'
NC_CANDIDATE = 'urn:ietf:params:netconf:capability:candidate:1.0'
#
# By default, don't support writeable-running or candidate configs
#
RUNNING = False
CANDIDATE = False
#
# Get where the script is; we will use this to find snippets for
# templates and filters unless overriden.
#
NCC_DIR, _ = os.path.split(os.path.realpath(__file__))
def display_capabilities(m):
"""Display the capabilities in a useful, categorized way.
"""
ietf_netconf_caps = []
ietf_models = []
openconfig_models = []
cisco_models = []
cisco_calvados_models = []
mib_models = []
other_models = []
# local function to pull out NS & module
def append_ns_and_module(c, module_list):
re_model = '^([^\?]+)\?module=([^&]+)&?'
m = re.search(re_model, c)
if m:
module_list.append('%s (%s)' % (m.group(2), m.group(1)))
else:
print('UNMATCHED model: %s' % c)
# pre-process capabilities, split into various categories
ns_to_list = [
('urn:ietf:params:xml:ns', ietf_models),
('http://openconfig.net/yang', openconfig_models),
('http://cisco.com/ns/yang', cisco_models,),
('http://cisco.com/calvados', cisco_calvados_models),
('http://cisco.com/panini/calvados', cisco_calvados_models),
('http://tail-f.com/ns/mibs', mib_models),
('http://tail-f.com/ns', cisco_calvados_models),
('http://tail-f.com/test', cisco_calvados_models),
('http://tail-f.com/yang', cisco_calvados_models),
('http://www.cisco.com/calvados', cisco_calvados_models),
('http://www.cisco.com/ns/calvados', cisco_calvados_models),
('http://www.cisco.com/panini/calvados', cisco_calvados_models),
('http://', other_models),
]
for c in m.server_capabilities:
matched = False
if c.startswith('urn:ietf:params:netconf'):
ietf_netconf_caps.append(c)
matched = True
else:
for ns, ns_list in ns_to_list:
if ns in c:
append_ns_and_module(c, ns_list)
matched = True
break
if matched==False:
print(c)
# now print them
list_to_heading = [
(ietf_netconf_caps, 'IETF NETCONF Capabilities:'),
(ietf_models, 'IETF Models:'),
(openconfig_models, 'OpenConfig Models:'),
(cisco_models, 'Cisco Models:'),
(cisco_calvados_models, 'Cisco Calvados Models:'),
(mib_models, 'MIB Models:'),
(other_models, 'Other Models:'),
]
for (l, h) in list_to_heading:
if len(l) > 0:
print(h)
for s in l:
print('\t%s' % s)
def query_model_support(m, re_module):
"""Search the capabilities for one or more models that match the provided
regex.
"""
matches = []
re_model = '^([^\?]+)\?module=([^&]+)&?'
for c in m.server_capabilities:
m = re.search(re_model, c)
if m:
model = m.group(2)
match = re.search(re_module, model)
if match:
matches.append(model)
return matches
def list_templates(header, source_env):
"""List out all the templates in the provided environment, parse them
and extract variables that should be provided.
UPDATED To present the VARS as JSON dict with enpty values
"""
print(header)
env = Environment()
for tname in sorted(source_env.list_templates()):
tfile = source_env.get_template(tname).filename
with open(tfile, 'r') as f:
vars = meta.find_undeclared_variables(env.parse(f.read()))
f.close()
print(" {}".format(tname.replace('.tmpl', ''))),
if vars:
print ":{",
#for v in sorted(vars):
# print('"%s" : ""' % v),
print ','.join(['"%s" : ""' %v for v in sorted(vars)]) ,
print "}"
else:
print
def do_templates(m, t_list, default_op='merge', **kwargs):
"""Execute a list of templates, using the kwargs passed in to
complete the rendering.
"""
for tmpl in t_list:
try:
data = tmpl.render(kwargs)
except UndefinedError as e:
print "Undefined variable %s. Use --params to specify json dict" % e.message
# assuming we should fail if a single template fails?
exit(1)
if CANDIDATE:
m.edit_config(data,
format='xml',
target='candidate',
default_operation=default_op)
elif RUNNING:
m.edit_config(data,
format='xml',
target='running',
default_operation=default_op)
if CANDIDATE:
m.commit()
def get_running_config(m, filter=None, xpath=None):
"""Get running config with a passed in filter. If both types of
filter are passed in for some reason, the subtree filter "wins".
"""
import time
if filter and len(filter) > 0:
c = m.get_config(source='running', filter=('subtree', filter))
elif xpath and len(xpath)>0:
c = m.get_config(source='running', filter=('xpath', xpath))
else:
c = m.get_config(source='running')
print(etree.tostring(c.data, pretty_print=True))
def get(m, filter=None, xpath=None):
"""Get state with a passed in filter. If both types of filter are
passed in for some reason, the subtree filter "wins".
"""
if filter and len(filter) > 0:
c = m.get(filter=('subtree', filter))
elif xpath and len(xpath)>0:
c = m.get(filter=('xpath', xpath))
else:
print("Need a filter for oper get!")
return
print(etree.tostring(c.data, pretty_print=True))
if __name__ == '__main__':
parser = ArgumentParser(description='Select your NETCONF operation and parameters:')
#
# NETCONF session parameters
#
parser.add_argument('--host', type=str, default='127.0.0.1',
help="The IP address for the device to connect to (default localhost)")
parser.add_argument('-u', '--username', type=str, default=os.environ.get('NCC_USERNAME', 'cisco'),
help="Username to use for SSH authentication (default 'cisco')")
parser.add_argument('-p', '--password', type=str, default=os.environ.get('NCC_PASSWORD', 'cisco'),
help="Password to use for SSH authentication (default 'cisco')")
parser.add_argument('--port', type=int, default=830,
help="Specify this if you want a non-default port (default 830)")
parser.add_argument('--timeout', type=int, default=60,
help="NETCONF operation timeout in seconds (default 60)")
parser.add_argument('-v', '--verbose', action='store_true',
help="Exceedingly verbose logging to the console")
parser.add_argument('--default-op', type=str, default='merge',
help="The NETCONF default operation to use (default 'merge')")
parser.add_argument('--device-type', type=str, default=None,
help='The device type to pass to ncclient (default: None)')
parser.add_argument('-w', '--where', action='store_true',
help="Print where script is and exit")
#
# Where we want to source snippets from
#
parser.add_argument('--snippets', type=str, default=os.environ.get('NCC_SNIPPETS', "%s/snippets" % NCC_DIR),
help="Directory where 'snippets' can be found; default is location of script")
#
# Various operation parameters. These will be put into a kwargs
# dictionary for use in template rendering.
#
parser.add_argument('--params', type=str,
help="JSON-encoded string of parameters dictionaryfor templates")
parser.add_argument('--params-file', type=str,
help="JSON-encoded file of parameters dictionary for templates")
#
# Only one type of filter allowed.
#
g = parser.add_mutually_exclusive_group()
g.add_argument('-f', '--filter', type=str,
help="NETCONF subtree filter")
g.add_argument('--named-filter', type=str,
help="Named NETCONF subtree filter")
g.add_argument('-x', '--xpath', type=str,
help="NETCONF XPath filter")
#
# Mutually exclusive operations.
#
g = parser.add_mutually_exclusive_group(required=True)
g.add_argument('-c', '--capabilities', action='store_true',
help="Display capabilities of the device.")
g.add_argument('--is-supported', type=str,
help="Query the server capabilities to determine whether the device claims to support YANG modules matching the provided regular expression. The regex provided is not automatically anchored to start or end. Note that the regex supplied must be in a format valid for Python and that it may be necessary to quote the argument.")
g.add_argument('--list-templates', action='store_true',
help="List out named edit-config templates")
g.add_argument('--list-filters', action='store_true',
help="List out named filters")
g.add_argument('-g', '--get-running', action='store_true',
help="Get the running config")
g.add_argument('--get-oper', action='store_true',
help="Get oper data")
g.add_argument('--do-edits', type=str, nargs='+',
help="Execute a sequence of named templates with an optional default operation and a single commit when candidate config supported. If only writable-running support, ALL operations will be attempted.")
#
# Finally, parse the arguments!
#
args = parser.parse_args()
#
# Setup the templates for use.
#
named_filters = Environment(loader=FileSystemLoader(
'%s/filters' % args.snippets),
undefined=StrictUndefined)
named_templates = Environment(loader=FileSystemLoader(
'%s/editconfigs' % args.snippets),
undefined = StrictUndefined)
#
# Do the named template/filter listing first, then exit.
#
if args.list_templates:
list_templates("Edit-config templates:", named_templates)
sys.exit(0)
elif args.list_filters:
list_templates("Named filters:", named_filters)
sys.exit(0)
#
# If the user specified verbose logging, set it up.
#
if args.verbose:
handler = logging.StreamHandler()
for l in LOGGING_TO_ENABLE:
logger = logging.getLogger(l)
logger.addHandler(handler)
logger.setLevel(logging.DEBUG)
#
# set up various keyword arguments that have specific arguments
#
kwargs = None
if args.params:
kwargs = json.loads(args.params)
elif args.params_file:
with open(args.params_file) as f:
kwargs = json.loads(f.read())
f.close()
else:
kwargs = {}
#
# This populates the filter if it's a canned filter.
#
if args.named_filter:
try:
args.filter = named_filters.get_template(
'%s.tmpl' % args.named_filter).render(**kwargs)
except UndefinedError as e:
print "Undefined variable %s. Use --params to specify json dict" % e.message
exit(1)
#
# Could use this extra param instead of the last four arguments
# specified below:
#
# device_params={'name': 'iosxr'}
#
def unknown_host_cb(host, fingerprint):
return True
device_params = {}
if args.device_type is not None:
device_params = {'name': args.device_type}
m = manager.connect(host=args.host,
port=args.port,
timeout=args.timeout,
username=args.username,
password=args.password,
allow_agent=False,
look_for_keys=False,
hostkey_verify=False,
device_params=device_params,
unknown_host_cb=unknown_host_cb)
#
# Extract the key capabilities that determine how we interact with
# the device. This script will prefer using candidate config.
#
if NC_WRITABLE_RUNNING in m.server_capabilities:
RUNNING = True
if NC_CANDIDATE in m.server_capabilities:
CANDIDATE = True
if args.get_running:
get_running_config(m, xpath=args.xpath, filter=args.filter)
elif args.get_oper:
get(m, filter=args.filter, xpath=args.xpath)
elif args.do_edits:
do_templates( m,
[named_templates.get_template('%s.tmpl' % t) for t in args.do_edits],
default_op=args.default_op,
**kwargs)
elif args.capabilities:
display_capabilities(m)
elif args.is_supported:
models = query_model_support(m, args.is_supported)
for model in models:
print(model)
#
# Orderly teardown of the netconf session.
# Ignore Value error sometimes returned in cleanup
try:
m.close_session()
except ValueError:
pass