-
Notifications
You must be signed in to change notification settings - Fork 0
/
base_connection.py
1570 lines (1306 loc) · 61.6 KB
/
base_connection.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
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
'''
Base connection class for netmiko
Handles SSH connection and methods that are generically applicable to different
platforms (Cisco and non-Cisco).
Also defines methods that should generally be supported by child classes
'''
from __future__ import print_function
from __future__ import unicode_literals
import io
import re
import socket
import telnetlib
import time
from os import path
from threading import Lock
import paramiko
import serial
from netmiko import log
from netmiko.netmiko_globals import MAX_BUFFER, BACKSPACE_CHAR
from netmiko.py23_compat import string_types, bufferedio_types, text_type
from netmiko.ssh_exception import NetMikoTimeoutException, NetMikoAuthenticationException
from netmiko.utilities import write_bytes, check_serial_port, get_structured_data
class BaseConnection(object):
"""
Defines vendor independent methods.
Otherwise method left as a stub method.
"""
def __init__(self, ip='', host='', username='', password='', secret='', port=None,
device_type='', verbose=False, global_delay_factor=1, use_keys=False,
key_file=None, pkey=None, passphrase=None, allow_agent=False,
ssh_strict=False, system_host_keys=False,
alt_host_keys=False, alt_key_file='', ssh_config_file=None, timeout=100,
session_timeout=60, auth_timeout=None, blocking_timeout=8, keepalive=0,
default_enter=None, response_return=None, serial_settings=None, fast_cli=False,
session_log=None, session_log_record_writes=False, session_log_file_mode='write',
allow_auto_change=False, encoding='ascii'):
"""
Initialize attributes for establishing connection to target device.
:param ip: IP address of target device. Not required if `host` is
provided.
:type ip: str
:param host: Hostname of target device. Not required if `ip` is
provided.
:type host: str
:param username: Username to authenticate against target device if
required.
:type username: str
:param password: Password to authenticate against target device if
required.
:type password: str
:param secret: The enable password if target device requires one.
:type secret: str
:param port: The destination port used to connect to the target
device.
:type port: int or None
:param device_type: Class selection based on device type.
:type device_type: str
:param verbose: Enable additional messages to standard output.
:type verbose: bool
:param global_delay_factor: Multiplication factor affecting Netmiko delays (default: 1).
:type global_delay_factor: int
:param use_keys: Connect to target device using SSH keys.
:type use_keys: bool
:param key_file: Filename path of the SSH key file to use.
:type key_file: str
:param pkey: SSH key object to use.
:type pkey: paramiko.PKey
:param passphrase: Passphrase to use for encrypted key; password will be used for key
decryption if not specified.
:type passphrase: str
:param allow_agent: Enable use of SSH key-agent.
:type allow_agent: bool
:param ssh_strict: Automatically reject unknown SSH host keys (default: False, which
means unknown SSH host keys will be accepted).
:type ssh_strict: bool
:param system_host_keys: Load host keys from the user's 'known_hosts' file.
:type system_host_keys: bool
:param alt_host_keys: If `True` host keys will be loaded from the file specified in
'alt_key_file'.
:type alt_host_keys: bool
:param alt_key_file: SSH host key file to use (if alt_host_keys=True).
:type alt_key_file: str
:param ssh_config_file: File name of OpenSSH configuration file.
:type ssh_config_file: str
:param timeout: Connection timeout.
:type timeout: float
:param session_timeout: Set a timeout for parallel requests.
:type session_timeout: float
:param auth_timeout: Set a timeout (in seconds) to wait for an authentication response.
:type auth_timeout: float
:param keepalive: Send SSH keepalive packets at a specific interval, in seconds.
Currently defaults to 0, for backwards compatibility (it will not attempt
to keep the connection alive).
:type keepalive: int
:param default_enter: Character(s) to send to correspond to enter key (default: '\n').
:type default_enter: str
:param response_return: Character(s) to use in normalized return data to represent
enter key (default: '\n')
:type response_return: str
:param fast_cli: Provide a way to optimize for performance. Converts select_delay_factor
to select smallest of global and specific. Sets default global_delay_factor to .1
(default: False)
:type fast_cli: boolean
:param session_log: File path or BufferedIOBase subclass object to write the session log to.
:type session_log: str
:param session_log_record_writes: The session log generally only records channel reads due
to eliminate command duplication due to command echo. You can enable this if you
want to record both channel reads and channel writes in the log (default: False).
:type session_log_record_writes: boolean
:param session_log_file_mode: "write" or "append" for session_log file mode
(default: "write")
:type session_log_file_mode: str
:param allow_auto_change: Allow automatic configuration changes for terminal settings.
(default: False)
:type allow_auto_change: bool
:param encoding: Encoding to be used when writing bytes to the output channel.
(default: 'ascii')
:type encoding: str
"""
self.remote_conn = None
self.RETURN = '\n' if default_enter is None else default_enter
self.TELNET_RETURN = '\r\n'
# Line Separator in response lines
self.RESPONSE_RETURN = '\n' if response_return is None else response_return
if ip:
self.host = ip
self.ip = ip
elif host:
self.host = host
if not ip and not host and 'serial' not in device_type:
raise ValueError("Either ip or host must be set")
if port is None:
if 'telnet' in device_type:
port = 23
else:
port = 22
self.port = int(port)
self.username = username
self.password = password
self.secret = secret
self.device_type = device_type
self.ansi_escape_codes = False
self.verbose = verbose
self.timeout = timeout
self.auth_timeout = auth_timeout
self.session_timeout = session_timeout
self.blocking_timeout = blocking_timeout
self.keepalive = keepalive
self.allow_auto_change = allow_auto_change
self.encoding = encoding
# Netmiko will close the session_log if we open the file
self.session_log = None
self.session_log_record_writes = session_log_record_writes
self._session_log_close = False
# Ensures last write operations prior to disconnect are recorded.
self._session_log_fin = False
if session_log is not None:
if isinstance(session_log, string_types):
# If session_log is a string, open a file corresponding to string name.
self.open_session_log(filename=session_log, mode=session_log_file_mode)
elif isinstance(session_log, bufferedio_types):
# In-memory buffer or an already open file handle
self.session_log = session_log
else:
raise ValueError("session_log must be a path to a file, a file handle, "
"or a BufferedIOBase subclass.")
# Default values
self.serial_settings = {
'port': 'COM1',
'baudrate': 9600,
'bytesize': serial.EIGHTBITS,
'parity': serial.PARITY_NONE,
'stopbits': serial.STOPBITS_ONE
}
if serial_settings is None:
serial_settings = {}
self.serial_settings.update(serial_settings)
if 'serial' in device_type:
self.host = 'serial'
comm_port = self.serial_settings.pop('port')
# Get the proper comm port reference if a name was enterred
comm_port = check_serial_port(comm_port)
self.serial_settings.update({'port': comm_port})
self.fast_cli = fast_cli
self.global_delay_factor = global_delay_factor
if self.fast_cli and self.global_delay_factor == 1:
self.global_delay_factor = .1
# set in set_base_prompt method
self.base_prompt = ''
self._session_locker = Lock()
# determine if telnet or SSH
if '_telnet' in device_type:
self.protocol = 'telnet'
self._modify_connection_params()
self.establish_connection()
self._try_session_preparation()
elif '_serial' in device_type:
self.protocol = 'serial'
self._modify_connection_params()
self.establish_connection()
self._try_session_preparation()
else:
self.protocol = 'ssh'
if not ssh_strict:
self.key_policy = paramiko.AutoAddPolicy()
else:
self.key_policy = paramiko.RejectPolicy()
# Options for SSH host_keys
self.use_keys = use_keys
self.key_file = key_file
self.pkey = pkey
self.passphrase = passphrase
self.allow_agent = allow_agent
self.system_host_keys = system_host_keys
self.alt_host_keys = alt_host_keys
self.alt_key_file = alt_key_file
# For SSH proxy support
self.ssh_config_file = ssh_config_file
self._modify_connection_params()
self.establish_connection()
self._try_session_preparation()
def __enter__(self):
"""Establish a session using a Context Manager."""
return self
def __exit__(self, exc_type, exc_value, traceback):
"""Gracefully close connection on Context Manager exit."""
self.disconnect()
def _modify_connection_params(self):
"""Modify connection parameters prior to SSH connection."""
pass
def _timeout_exceeded(self, start, msg='Timeout exceeded!'):
"""Raise NetMikoTimeoutException if waiting too much in the serving queue.
:param start: Initial start time to see if session lock timeout has been exceeded
:type start: float (from time.time() call i.e. epoch time)
:param msg: Exception message if timeout was exceeded
:type msg: str
"""
if not start:
# Must provide a comparison time
return False
if time.time() - start > self.session_timeout:
# session_timeout exceeded
raise NetMikoTimeoutException(msg)
return False
def _lock_netmiko_session(self, start=None):
"""Try to acquire the Netmiko session lock. If not available, wait in the queue until
the channel is available again.
:param start: Initial start time to measure the session timeout
:type start: float (from time.time() call i.e. epoch time)
"""
if not start:
start = time.time()
# Wait here until the SSH channel lock is acquired or until session_timeout exceeded
while (not self._session_locker.acquire(False) and
not self._timeout_exceeded(start, 'The netmiko channel is not available!')):
time.sleep(.1)
return True
def _unlock_netmiko_session(self):
"""
Release the channel at the end of the task.
"""
if self._session_locker.locked():
self._session_locker.release()
def _write_channel(self, out_data):
"""Generic handler that will write to both SSH and telnet channel.
:param out_data: data to be written to the channel
:type out_data: str (can be either unicode/byte string)
"""
if self.protocol == 'ssh':
self.remote_conn.sendall(write_bytes(out_data, encoding=self.encoding))
elif self.protocol == 'telnet':
self.remote_conn.write(write_bytes(out_data, encoding=self.encoding))
elif self.protocol == 'serial':
self.remote_conn.write(write_bytes(out_data, encoding=self.encoding))
self.remote_conn.flush()
else:
raise ValueError("Invalid protocol specified")
try:
log.debug("write_channel: {}".format(write_bytes(out_data, encoding=self.encoding)))
if self._session_log_fin or self.session_log_record_writes:
self._write_session_log(out_data)
except UnicodeDecodeError:
# Don't log non-ASCII characters; this is null characters and telnet IAC (PY2)
pass
def _write_session_log(self, data):
if self.session_log is not None and len(data) > 0:
self.session_log.write(write_bytes(data, encoding=self.encoding))
self.session_log.flush()
def write_channel(self, out_data):
"""Generic handler that will write to both SSH and telnet channel.
:param out_data: data to be written to the channel
:type out_data: str (can be either unicode/byte string)
"""
self._lock_netmiko_session()
try:
self._write_channel(out_data)
finally:
# Always unlock the SSH channel, even on exception.
self._unlock_netmiko_session()
def is_alive(self):
"""Returns a boolean flag with the state of the connection."""
null = chr(0)
if self.remote_conn is None:
log.error("Connection is not initialised, is_alive returns False")
return False
if self.protocol == 'telnet':
try:
# Try sending IAC + NOP (IAC is telnet way of sending command)
# IAC = Interpret as Command; it comes before the NOP.
log.debug("Sending IAC + NOP")
# Need to send multiple times to test connection
self.remote_conn.sock.sendall(telnetlib.IAC + telnetlib.NOP)
self.remote_conn.sock.sendall(telnetlib.IAC + telnetlib.NOP)
self.remote_conn.sock.sendall(telnetlib.IAC + telnetlib.NOP)
return True
except AttributeError:
return False
else:
# SSH
try:
# Try sending ASCII null byte to maintain the connection alive
log.debug("Sending the NULL byte")
self.write_channel(null)
return self.remote_conn.transport.is_active()
except (socket.error, EOFError):
log.error("Unable to send", exc_info=True)
# If unable to send, we can tell for sure that the connection is unusable
return False
return False
def _read_channel(self):
"""Generic handler that will read all the data from an SSH or telnet channel."""
if self.protocol == 'ssh':
output = ""
while True:
if self.remote_conn.recv_ready():
outbuf = self.remote_conn.recv(MAX_BUFFER)
if len(outbuf) == 0:
raise EOFError("Channel stream closed by remote device.")
output += outbuf.decode('utf-8', 'ignore')
else:
break
elif self.protocol == 'telnet':
output = self.remote_conn.read_very_eager().decode('utf-8', 'ignore')
elif self.protocol == 'serial':
output = ""
while (self.remote_conn.in_waiting > 0):
output += self.remote_conn.read(self.remote_conn.in_waiting)
log.debug("read_channel: {}".format(output))
self._write_session_log(output)
return output
def read_channel(self):
"""Generic handler that will read all the data from an SSH or telnet channel."""
output = ""
self._lock_netmiko_session()
try:
output = self._read_channel()
finally:
# Always unlock the SSH channel, even on exception.
self._unlock_netmiko_session()
return output
def _read_channel_expect(self, pattern='', re_flags=0, max_loops=150):
"""Function that reads channel until pattern is detected.
pattern takes a regular expression.
By default pattern will be self.base_prompt
Note: this currently reads beyond pattern. In the case of SSH it reads MAX_BUFFER.
In the case of telnet it reads all non-blocking data.
There are dependencies here like determining whether in config_mode that are actually
depending on reading beyond pattern.
:param pattern: Regular expression pattern used to identify the command is done \
(defaults to self.base_prompt)
:type pattern: str (regular expression)
:param re_flags: regex flags used in conjunction with pattern to search for prompt \
(defaults to no flags)
:type re_flags: int
:param max_loops: max number of iterations to read the channel before raising exception.
Will default to be based upon self.timeout.
:type max_loops: int
"""
output = ''
if not pattern:
pattern = re.escape(self.base_prompt)
log.debug("Pattern is: {}".format(pattern))
i = 1
loop_delay = .1
# Default to making loop time be roughly equivalent to self.timeout (support old max_loops
# argument for backwards compatibility).
if max_loops == 150:
max_loops = self.timeout / loop_delay
while i < max_loops:
if self.protocol == 'ssh':
try:
# If no data available will wait timeout seconds trying to read
self._lock_netmiko_session()
new_data = self.remote_conn.recv(MAX_BUFFER)
if len(new_data) == 0:
raise EOFError("Channel stream closed by remote device.")
new_data = new_data.decode('utf-8', 'ignore')
log.debug("_read_channel_expect read_data: {}".format(new_data))
output += new_data
self._write_session_log(new_data)
except socket.timeout:
raise NetMikoTimeoutException("Timed-out reading channel, data not available.")
finally:
self._unlock_netmiko_session()
elif self.protocol == 'telnet' or 'serial':
output += self.read_channel()
if re.search(pattern, output, flags=re_flags):
log.debug("Pattern found: {} {}".format(pattern, output))
return output
time.sleep(loop_delay * self.global_delay_factor)
i += 1
raise NetMikoTimeoutException("Timed-out reading channel, pattern not found in output: {}"
.format(pattern))
def _read_channel_timing(self, delay_factor=1, max_loops=150):
"""Read data on the channel based on timing delays.
Attempt to read channel max_loops number of times. If no data this will cause a 15 second
delay.
Once data is encountered read channel for another two seconds (2 * delay_factor) to make
sure reading of channel is complete.
:param delay_factor: multiplicative factor to adjust delay when reading channel (delays
get multiplied by this factor)
:type delay_factor: int or float
:param max_loops: maximum number of loops to iterate through before returning channel data.
Will default to be based upon self.timeout.
:type max_loops: int
"""
# Time to delay in each read loop
loop_delay = .1
final_delay = 2
# Default to making loop time be roughly equivalent to self.timeout (support old max_loops
# and delay_factor arguments for backwards compatibility).
delay_factor = self.select_delay_factor(delay_factor)
if delay_factor == 1 and max_loops == 150:
max_loops = int(self.timeout / loop_delay)
channel_data = ""
i = 0
while i <= max_loops:
time.sleep(loop_delay * delay_factor)
new_data = self.read_channel()
if new_data:
channel_data += new_data
else:
# Safeguard to make sure really done
time.sleep(final_delay * delay_factor)
new_data = self.read_channel()
if not new_data:
break
else:
channel_data += new_data
i += 1
return channel_data
def read_until_prompt(self, *args, **kwargs):
"""Read channel until self.base_prompt detected. Return ALL data available."""
return self._read_channel_expect(*args, **kwargs)
def read_until_pattern(self, *args, **kwargs):
"""Read channel until pattern detected. Return ALL data available."""
return self._read_channel_expect(*args, **kwargs)
def read_until_prompt_or_pattern(self, pattern='', re_flags=0):
"""Read until either self.base_prompt or pattern is detected.
:param pattern: the pattern used to identify that the output is complete (i.e. stop \
reading when pattern is detected). pattern will be combined with self.base_prompt to \
terminate output reading when the first of self.base_prompt or pattern is detected.
:type pattern: regular expression string
:param re_flags: regex flags used in conjunction with pattern to search for prompt \
(defaults to no flags)
:type re_flags: int
"""
combined_pattern = re.escape(self.base_prompt)
if pattern:
combined_pattern = r"({}|{})".format(combined_pattern, pattern)
return self._read_channel_expect(combined_pattern, re_flags=re_flags)
def serial_login(self, pri_prompt_terminator=r'#\s*$', alt_prompt_terminator=r'>\s*$',
username_pattern=r"(?:[Uu]ser:|sername|ogin)", pwd_pattern=r"assword",
delay_factor=1, max_loops=20):
self.telnet_login(pri_prompt_terminator, alt_prompt_terminator, username_pattern,
pwd_pattern, delay_factor, max_loops)
def telnet_login(self, pri_prompt_terminator=r'#\s*$', alt_prompt_terminator=r'>\s*$',
username_pattern=r"(?:user:|username|login|user name)", pwd_pattern=r"assword",
delay_factor=1, max_loops=20):
"""Telnet login. Can be username/password or just password.
:param pri_prompt_terminator: Primary trailing delimiter for identifying a device prompt
:type pri_prompt_terminator: str
:param alt_prompt_terminator: Alternate trailing delimiter for identifying a device prompt
:type alt_prompt_terminator: str
:param username_pattern: Pattern used to identify the username prompt
:type username_pattern: str
:param delay_factor: See __init__: global_delay_factor
:type delay_factor: int
:param max_loops: Controls the wait time in conjunction with the delay_factor
(default: 20)
"""
delay_factor = self.select_delay_factor(delay_factor)
time.sleep(1 * delay_factor)
output = ''
return_msg = ''
i = 1
while i <= max_loops:
try:
output = self.read_channel()
return_msg += output
# Search for username pattern / send username
if re.search(username_pattern, output, flags=re.I):
self.write_channel(self.username + self.TELNET_RETURN)
time.sleep(1 * delay_factor)
output = self.read_channel()
return_msg += output
# Search for password pattern / send password
if re.search(pwd_pattern, output, flags=re.I):
self.write_channel(self.password + self.TELNET_RETURN)
time.sleep(.5 * delay_factor)
output = self.read_channel()
return_msg += output
if (re.search(pri_prompt_terminator, output, flags=re.M)
or re.search(alt_prompt_terminator, output, flags=re.M)):
return return_msg
# Check if proper data received
if (re.search(pri_prompt_terminator, output, flags=re.M)
or re.search(alt_prompt_terminator, output, flags=re.M)):
return return_msg
self.write_channel(self.TELNET_RETURN)
time.sleep(.5 * delay_factor)
i += 1
except EOFError:
self.remote_conn.close()
msg = "Login failed: {}".format(self.host)
raise NetMikoAuthenticationException(msg)
# Last try to see if we already logged in
self.write_channel(self.TELNET_RETURN)
time.sleep(.5 * delay_factor)
output = self.read_channel()
return_msg += output
if (re.search(pri_prompt_terminator, output, flags=re.M)
or re.search(alt_prompt_terminator, output, flags=re.M)):
return return_msg
msg = "Login failed: {}".format(self.host)
self.remote_conn.close()
raise NetMikoAuthenticationException(msg)
def _try_session_preparation(self):
"""
In case of an exception happening during `session_preparation()` Netmiko should
gracefully clean-up after itself. This might be challenging for library users
to do since they don't have a reference to the object. This is possibly related
to threads used in Paramiko.
"""
try:
self.session_preparation()
except Exception:
self.disconnect()
raise
def session_preparation(self):
"""
Prepare the session after the connection has been established
This method handles some differences that occur between various devices
early on in the session.
In general, it should include:
self._test_channel_read()
self.set_base_prompt()
self.disable_paging()
self.set_terminal_width()
self.clear_buffer()
"""
self._test_channel_read()
self.set_base_prompt()
self.disable_paging()
self.set_terminal_width()
# Clear the read buffer
time.sleep(.3 * self.global_delay_factor)
self.clear_buffer()
def _use_ssh_config(self, dict_arg):
"""Update SSH connection parameters based on contents of SSH 'config' file.
:param dict_arg: Dictionary of SSH connection parameters
:type dict_arg: dict
"""
connect_dict = dict_arg.copy()
# Use SSHConfig to generate source content.
full_path = path.abspath(path.expanduser(self.ssh_config_file))
if path.exists(full_path):
ssh_config_instance = paramiko.SSHConfig()
with io.open(full_path, "rt", encoding='utf-8') as f:
ssh_config_instance.parse(f)
source = ssh_config_instance.lookup(self.host)
else:
source = {}
if "proxycommand" in source:
proxy = paramiko.ProxyCommand(source['proxycommand'])
elif "ProxyCommand" in source:
proxy = paramiko.ProxyCommand(source['ProxyCommand'])
else:
proxy = None
# Only update 'hostname', 'sock', 'port', and 'username'
# For 'port' and 'username' only update if using object defaults
if connect_dict['port'] == 22:
connect_dict['port'] = int(source.get('port', self.port))
if connect_dict['username'] == '':
connect_dict['username'] = source.get('username', self.username)
if proxy:
connect_dict['sock'] = proxy
connect_dict['hostname'] = source.get('hostname', self.host)
return connect_dict
def _connect_params_dict(self):
"""Generate dictionary of Paramiko connection parameters."""
conn_dict = {
'hostname': self.host,
'port': self.port,
'username': self.username,
'password': self.password,
'look_for_keys': self.use_keys,
'allow_agent': self.allow_agent,
'key_filename': self.key_file,
'pkey': self.pkey,
'passphrase': self.passphrase,
'timeout': self.timeout,
'auth_timeout': self.auth_timeout
}
# Check if using SSH 'config' file mainly for SSH proxy support
if self.ssh_config_file:
conn_dict = self._use_ssh_config(conn_dict)
return conn_dict
def _sanitize_output(self, output, strip_command=False, command_string=None,
strip_prompt=False):
"""Strip out command echo, trailing router prompt and ANSI escape codes.
:param output: Output from a remote network device
:type output: unicode string
:param strip_command:
:type strip_command:
"""
if self.ansi_escape_codes:
output = self.strip_ansi_escape_codes(output)
output = self.normalize_linefeeds(output)
if strip_command and command_string:
command_string = self.normalize_linefeeds(command_string)
output = self.strip_command(command_string, output)
if strip_prompt:
output = self.strip_prompt(output)
return output
def establish_connection(self, width=None, height=None):
"""Establish SSH connection to the network device
Timeout will generate a NetMikoTimeoutException
Authentication failure will generate a NetMikoAuthenticationException
width and height are needed for Fortinet paging setting.
:param width: Specified width of the VT100 terminal window
:type width: int
:param height: Specified height of the VT100 terminal window
:type height: int
"""
if self.protocol == 'telnet':
self.remote_conn = telnetlib.Telnet(self.host, port=self.port, timeout=self.timeout)
self.telnet_login()
elif self.protocol == 'serial':
self.remote_conn = serial.Serial(**self.serial_settings)
self.serial_login()
elif self.protocol == 'ssh':
ssh_connect_params = self._connect_params_dict()
self.remote_conn_pre = self._build_ssh_client()
# initiate SSH connection
try:
self.remote_conn_pre.connect(**ssh_connect_params)
except socket.error:
self.paramiko_cleanup()
msg = "Connection to device timed-out: {device_type} {ip}:{port}".format(
device_type=self.device_type, ip=self.host, port=self.port)
raise NetMikoTimeoutException(msg)
except paramiko.ssh_exception.AuthenticationException as auth_err:
self.paramiko_cleanup()
msg = "Authentication failure: unable to connect {device_type} {ip}:{port}".format(
device_type=self.device_type, ip=self.host, port=self.port)
msg += self.RETURN + text_type(auth_err)
raise NetMikoAuthenticationException(msg)
if self.verbose:
print("SSH connection established to {}:{}".format(self.host, self.port))
# Use invoke_shell to establish an 'interactive session'
if width and height:
self.remote_conn = self.remote_conn_pre.invoke_shell(term='vt100', width=width,
height=height)
else:
self.remote_conn = self.remote_conn_pre.invoke_shell()
self.remote_conn.settimeout(self.blocking_timeout)
if self.keepalive:
self.remote_conn.transport.set_keepalive(self.keepalive)
self.special_login_handler()
if self.verbose:
print("Interactive SSH session established")
return ""
def _test_channel_read(self, count=40, pattern=""):
"""Try to read the channel (generally post login) verify you receive data back.
:param count: the number of times to check the channel for data
:type count: int
:param pattern: Regular expression pattern used to determine end of channel read
:type pattern: str
"""
def _increment_delay(main_delay, increment=1.1, maximum=8):
"""Increment sleep time to a maximum value."""
main_delay = main_delay * increment
if main_delay >= maximum:
main_delay = maximum
return main_delay
i = 0
delay_factor = self.select_delay_factor(delay_factor=0)
main_delay = delay_factor * .1
time.sleep(main_delay * 10)
new_data = ""
while i <= count:
new_data += self._read_channel_timing()
if new_data and pattern:
if re.search(pattern, new_data):
break
elif new_data:
break
else:
self.write_channel(self.RETURN)
main_delay = _increment_delay(main_delay)
time.sleep(main_delay)
i += 1
# check if data was ever present
if new_data:
return ""
else:
raise NetMikoTimeoutException("Timed out waiting for data")
def _build_ssh_client(self):
"""Prepare for Paramiko SSH connection."""
# Create instance of SSHClient object
remote_conn_pre = paramiko.SSHClient()
# Load host_keys for better SSH security
if self.system_host_keys:
remote_conn_pre.load_system_host_keys()
if self.alt_host_keys and path.isfile(self.alt_key_file):
remote_conn_pre.load_host_keys(self.alt_key_file)
# Default is to automatically add untrusted hosts (make sure appropriate for your env)
remote_conn_pre.set_missing_host_key_policy(self.key_policy)
return remote_conn_pre
def select_delay_factor(self, delay_factor):
"""
Choose the greater of delay_factor or self.global_delay_factor (default).
In fast_cli choose the lesser of delay_factor of self.global_delay_factor.
:param delay_factor: See __init__: global_delay_factor
:type delay_factor: int
"""
if self.fast_cli:
if delay_factor <= self.global_delay_factor:
return delay_factor
else:
return self.global_delay_factor
else:
if delay_factor >= self.global_delay_factor:
return delay_factor
else:
return self.global_delay_factor
def special_login_handler(self, delay_factor=1):
"""Handler for devices like WLC, Extreme ERS that throw up characters prior to login."""
pass
def disable_paging(self, command="terminal length 0", delay_factor=1):
"""Disable paging default to a Cisco CLI method.
:param command: Device command to disable pagination of output
:type command: str
:param delay_factor: See __init__: global_delay_factor
:type delay_factor: int
"""
delay_factor = self.select_delay_factor(delay_factor)
time.sleep(delay_factor * .1)
self.clear_buffer()
command = self.normalize_cmd(command)
log.debug("In disable_paging")
log.debug("Command: {0}".format(command))
self.write_channel(command)
output = self.read_until_prompt()
if self.ansi_escape_codes:
output = self.strip_ansi_escape_codes(output)
log.debug("{0}".format(output))
log.debug("Exiting disable_paging")
return output
def set_terminal_width(self, command="", delay_factor=1):
"""CLI terminals try to automatically adjust the line based on the width of the terminal.
This causes the output to get distorted when accessed programmatically.
Set terminal width to 511 which works on a broad set of devices.
:param command: Command string to send to the device
:type command: str
:param delay_factor: See __init__: global_delay_factor
:type delay_factor: int
"""
if not command:
return ""
delay_factor = self.select_delay_factor(delay_factor)
command = self.normalize_cmd(command)
self.write_channel(command)
output = self.read_until_prompt()
if self.ansi_escape_codes:
output = self.strip_ansi_escape_codes(output)
return output
def set_base_prompt(self, pri_prompt_terminator='#',
alt_prompt_terminator='>', delay_factor=1):
"""Sets self.base_prompt
Used as delimiter for stripping of trailing prompt in output.
Should be set to something that is general and applies in multiple contexts. For Cisco
devices this will be set to router hostname (i.e. prompt without '>' or '#').
This will be set on entering user exec or privileged exec on Cisco, but not when
entering/exiting config mode.
:param pri_prompt_terminator: Primary trailing delimiter for identifying a device prompt
:type pri_prompt_terminator: str
:param alt_prompt_terminator: Alternate trailing delimiter for identifying a device prompt
:type alt_prompt_terminator: str
:param delay_factor: See __init__: global_delay_factor
:type delay_factor: int
"""
prompt = self.find_prompt(delay_factor=delay_factor)
if not prompt[-1] in (pri_prompt_terminator, alt_prompt_terminator):
raise ValueError("Router prompt not found: {0}".format(repr(prompt)))
# Strip off trailing terminator
self.base_prompt = prompt[:-1]
return self.base_prompt
def find_prompt(self, delay_factor=1):
"""Finds the current network device prompt, last line only.
:param delay_factor: See __init__: global_delay_factor
:type delay_factor: int
"""
delay_factor = self.select_delay_factor(delay_factor)
self.clear_buffer()
self.write_channel(self.RETURN)
time.sleep(delay_factor * .1)
# Initial attempt to get prompt
prompt = self.read_channel()
if self.ansi_escape_codes:
prompt = self.strip_ansi_escape_codes(prompt)
# Check if the only thing you received was a newline
count = 0
prompt = prompt.strip()
while count <= 10 and not prompt:
prompt = self.read_channel().strip()
if prompt:
if self.ansi_escape_codes:
prompt = self.strip_ansi_escape_codes(prompt).strip()
else:
self.write_channel(self.RETURN)
time.sleep(delay_factor * .1)
count += 1
# If multiple lines in the output take the last line
prompt = self.normalize_linefeeds(prompt)
prompt = prompt.split(self.RESPONSE_RETURN)[-1]
prompt = prompt.strip()
if not prompt:
raise ValueError("Unable to find prompt: {}".format(prompt))
time.sleep(delay_factor * .1)
self.clear_buffer()