Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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 + str(auth_err)
raise NetmikoAuthenticationException(msg)
if self.verbose:
print(f"SSH connection established to {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
)
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
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(f"Pattern found: {pattern} {output}")
return output
time.sleep(loop_delay * self.global_delay_factor)
i += 1
raise NetmikoTimeoutException(
f"Timed-out reading channel, pattern not found in output: {pattern}"
)
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 new_data
else:
raise NetmikoTimeoutException("Timed out waiting for data")
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")
if self.ansi_escape_codes:
new_data = self.strip_ansi_escape_codes(new_data)
log.debug(f"_read_channel_expect read_data: {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(f"Pattern found: {pattern} {output}")
return output
time.sleep(loop_delay * self.global_delay_factor)
i += 1
raise NetmikoTimeoutException(
f"Timed-out reading channel, pattern not found in output: {pattern}"
)