Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
#!/usr/bin/env python
import time
from os.path import dirname, join
from threading import Lock
from netmiko import NetMikoTimeoutException
from netmiko.base_connection import BaseConnection
RESOURCE_FOLDER = join(dirname(dirname(__file__)), "etc")
class FakeBaseConnection(BaseConnection):
def __init__(self, **kwargs):
for key, value in kwargs.items():
setattr(self, key, value)
self._session_locker = Lock()
def test_timeout_exceeded():
"""Raise NetMikoTimeoutException if waiting too much"""
connection = FakeBaseConnection(session_timeout=10)
start = time.time() - 11
try:
connection._timeout_exceeded(start)
except NetMikoTimeoutException as exc:
assert isinstance(exc, NetMikoTimeoutException)
return
"""Generic Terminal Server driver."""
from __future__ import unicode_literals
from netmiko.base_connection import BaseConnection
class TerminalServer(BaseConnection):
"""Generic Terminal Server driver.
Allow direct write_channel / read_channel operations without session_preparation causing
an exception.
"""
def session_preparation(self):
"""Do nothing here; base_prompt is not set; paging is not disabled."""
pass
class TerminalServerSSH(TerminalServer):
"""Generic Terminal Server driver SSH."""
pass
from __future__ import unicode_literals
import time
from netmiko.base_connection import BaseConnection
class F5TmshSSH(BaseConnection):
def session_preparation(self):
"""Prepare the session after the connection has been established."""
self._test_channel_read()
self.set_base_prompt()
self.tmsh_mode()
self.set_base_prompt()
self.disable_paging(
command="modify cli preference pager disabled display-threshold 0"
)
self.clear_buffer()
cmd = 'run /util bash -c "stty cols 255"'
self.set_terminal_width(command=cmd)
def tmsh_mode(self, delay_factor=1):
"""tmsh command is equivalent to config command on F5."""
delay_factor = self.select_delay_factor(delay_factor)
from __future__ import unicode_literals
from netmiko.base_connection import BaseConnection
class NetAppcDotSSH(BaseConnection):
def session_preparation(self):
"""Prepare the session after the connection has been established."""
self.set_base_prompt()
cmd = self.RETURN + "rows 0" + self.RETURN
self.disable_paging(command=cmd)
def send_command_with_y(self, *args, **kwargs):
output = self.send_command_timing(*args, **kwargs)
if "{y|n}" in output:
output += self.send_command_timing(
"y", strip_prompt=False, strip_command=False
)
return output
def check_config_mode(self, check_string="*>"):
return super(NetAppcDotSSH, self).check_config_mode(check_string=check_string)
import time
from netmiko.base_connection import BaseConnection
class JuniperScreenOsSSH(BaseConnection):
"""
Implement methods for interacting with Juniper ScreenOS devices.
"""
def session_preparation(self):
"""
Prepare the session after the connection has been established.
Disable paging (the '--more--' prompts).
Set the base prompt for interaction ('>').
"""
self._test_channel_read()
self.set_base_prompt()
self.disable_paging(command="set console page 0")
# Clear the read buffer
time.sleep(0.3 * self.global_delay_factor)
from __future__ import unicode_literals
import time
from netmiko.base_connection import BaseConnection
class PluribusSSH(BaseConnection):
"""Common methods for Pluribus."""
def __init__(self, *args, **kwargs):
super(PluribusSSH, self).__init__(*args, **kwargs)
self._config_mode = False
def disable_paging(self, command="pager off", delay_factor=1):
"""Make sure paging is disabled."""
return super(PluribusSSH, self).disable_paging(
command=command, delay_factor=delay_factor
)
def session_preparation(self):
"""Prepare the netmiko session."""
self._test_channel_read()
self.set_base_prompt()
from __future__ import unicode_literals
import time
import re
from netmiko.base_connection import BaseConnection
class PaloAltoPanosBase(BaseConnection):
"""
Implement methods for interacting with PaloAlto devices.
Disables `enable()` and `check_enable_mode()`
methods. Overrides several methods for PaloAlto-specific compatibility.
"""
def session_preparation(self):
"""
Prepare the session after the connection has been established.
Disable paging (the '--more--' prompts).
Set the base prompt for interaction ('>').
"""
self._test_channel_read()
self.set_base_prompt(delay_factor=20)
import time
from netmiko.base_connection import BaseConnection
class RadETXBase(BaseConnection):
"""RAD ETX Support, Tested on RAD 203AX, 205A and 220A."""
def session_preparation(self):
self._test_channel_read()
self.set_base_prompt()
self.disable_paging(command="config term length 0")
# Clear the read buffer
time.sleep(0.3 * self.global_delay_factor)
self.clear_buffer()
def save_config(self, cmd="admin save", confirm=False, confirm_response=""):
"""Saves Config Using admin save."""
if confirm:
output = self.send_command_timing(command_string=cmd)
if confirm_response:
output += self.send_command_timing(confirm_response)
import time
import re
from netmiko.base_connection import BaseConnection
class DellIsilonSSH(BaseConnection):
def set_base_prompt(
self, pri_prompt_terminator="$", alt_prompt_terminator="#", delay_factor=1
):
"""Determine base prompt."""
return super(DellIsilonSSH, self).set_base_prompt(
pri_prompt_terminator=pri_prompt_terminator,
alt_prompt_terminator=alt_prompt_terminator,
delay_factor=delay_factor,
)
def strip_ansi_escape_codes(self, string_buffer):
"""Remove Null code"""
output = re.sub(r"\x00", "", string_buffer)
return super(DellIsilonSSH, self).strip_ansi_escape_codes(output)
def session_preparation(self):
from __future__ import unicode_literals
import time
from netmiko.base_connection import BaseConnection
class CheckPointGaiaSSH(BaseConnection):
"""
Implements methods for communicating with Check Point Gaia
firewalls.
"""
def session_preparation(self):
"""
Prepare the session after the connection has been established.
Set the base prompt for interaction ('>').
"""
self._test_channel_read()
self.set_base_prompt()
self.disable_paging(command="set clienv rows 0")
# Clear the read buffer
time.sleep(0.3 * self.global_delay_factor)