mirror of
https://github.com/netbox-community/netbox.git
synced 2024-05-10 07:54:54 +00:00
273 lines
8.6 KiB
Python
273 lines
8.6 KiB
Python
from ncclient import manager
|
|
import paramiko
|
|
import re
|
|
import xmltodict
|
|
import time
|
|
|
|
|
|
CONNECT_TIMEOUT = 5 # seconds
|
|
|
|
|
|
class RPCClient(object):
|
|
|
|
def __init__(self, device, username='', password=''):
|
|
self.username = username
|
|
self.password = password
|
|
try:
|
|
self.host = str(device.primary_ip.address.ip)
|
|
except AttributeError:
|
|
raise Exception("Specified device ({}) does not have a primary IP defined.".format(device))
|
|
|
|
def get_lldp_neighbors(self):
|
|
"""
|
|
Returns a list of dictionaries, each representing an LLDP neighbor adjacency.
|
|
|
|
{
|
|
'local-interface': <str>,
|
|
'name': <str>,
|
|
'remote-interface': <str>,
|
|
'chassis-id': <str>,
|
|
}
|
|
"""
|
|
raise NotImplementedError("Feature not implemented for this platform.")
|
|
|
|
def get_inventory(self):
|
|
"""
|
|
Returns a dictionary representing the device chassis and installed inventory items.
|
|
|
|
{
|
|
'chassis': {
|
|
'serial': <str>,
|
|
'description': <str>,
|
|
}
|
|
'items': [
|
|
{
|
|
'name': <str>,
|
|
'part_id': <str>,
|
|
'serial': <str>,
|
|
},
|
|
...
|
|
]
|
|
}
|
|
"""
|
|
raise NotImplementedError("Feature not implemented for this platform.")
|
|
|
|
|
|
class SSHClient(RPCClient):
|
|
def __enter__(self):
|
|
|
|
self.ssh = paramiko.SSHClient()
|
|
self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
|
|
try:
|
|
self.ssh.connect(
|
|
self.host,
|
|
username=self.username,
|
|
password=self.password,
|
|
timeout=CONNECT_TIMEOUT,
|
|
allow_agent=False,
|
|
look_for_keys=False,
|
|
)
|
|
except paramiko.AuthenticationException:
|
|
# Try default credentials if the configured creds don't work
|
|
try:
|
|
default_creds = self.default_credentials
|
|
if default_creds.get('username') and default_creds.get('password'):
|
|
self.ssh.connect(
|
|
self.host,
|
|
username=default_creds['username'],
|
|
password=default_creds['password'],
|
|
timeout=CONNECT_TIMEOUT,
|
|
allow_agent=False,
|
|
look_for_keys=False,
|
|
)
|
|
else:
|
|
raise ValueError('default_credentials are incomplete.')
|
|
except AttributeError:
|
|
raise paramiko.AuthenticationException
|
|
|
|
self.session = self.ssh.invoke_shell()
|
|
self.session.recv(1000)
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
|
self.ssh.close()
|
|
|
|
def _send(self, cmd, pause=1):
|
|
self.session.send('{}\n'.format(cmd))
|
|
data = ''
|
|
time.sleep(pause)
|
|
while self.session.recv_ready():
|
|
data += self.session.recv(4096).decode()
|
|
if not data:
|
|
break
|
|
return data
|
|
|
|
|
|
class JunosNC(RPCClient):
|
|
"""
|
|
NETCONF client for Juniper Junos devices
|
|
"""
|
|
|
|
def __enter__(self):
|
|
|
|
# Initiate a connection to the device
|
|
self.manager = manager.connect(host=self.host, username=self.username, password=self.password,
|
|
hostkey_verify=False, timeout=CONNECT_TIMEOUT)
|
|
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
|
|
|
# Close the connection to the device
|
|
self.manager.close_session()
|
|
|
|
def get_lldp_neighbors(self):
|
|
|
|
rpc_reply = self.manager.dispatch('get-lldp-neighbors-information')
|
|
lldp_neighbors_raw = xmltodict.parse(rpc_reply.xml)['rpc-reply']['lldp-neighbors-information']['lldp-neighbor-information']
|
|
|
|
result = []
|
|
for neighbor_raw in lldp_neighbors_raw:
|
|
neighbor = dict()
|
|
neighbor['local-interface'] = neighbor_raw.get('lldp-local-port-id')
|
|
name = neighbor_raw.get('lldp-remote-system-name')
|
|
if name:
|
|
neighbor['name'] = name.split('.')[0] # Split hostname from domain if one is present
|
|
else:
|
|
neighbor['name'] = ''
|
|
try:
|
|
neighbor['remote-interface'] = neighbor_raw['lldp-remote-port-description']
|
|
except KeyError:
|
|
# Older versions of Junos report on interface ID instead of description
|
|
neighbor['remote-interface'] = neighbor_raw.get('lldp-remote-port-id')
|
|
neighbor['chassis-id'] = neighbor_raw.get('lldp-remote-chassis-id')
|
|
result.append(neighbor)
|
|
|
|
return result
|
|
|
|
def get_inventory(self):
|
|
|
|
def glean_items(node, depth=0):
|
|
items = []
|
|
items_list = node.get('chassis{}-module'.format('-sub' * depth), [])
|
|
# Junos like to return single children directly instead of as a single-item list
|
|
if hasattr(items_list, 'items'):
|
|
items_list = [items_list]
|
|
for item in items_list:
|
|
m = {
|
|
'name': item['name'],
|
|
'part_id': item.get('model-number') or item.get('part-number', ''),
|
|
'serial': item.get('serial-number', ''),
|
|
}
|
|
child_items = glean_items(item, depth + 1)
|
|
if child_items:
|
|
m['items'] = child_items
|
|
items.append(m)
|
|
return items
|
|
|
|
rpc_reply = self.manager.dispatch('get-chassis-inventory')
|
|
inventory_raw = xmltodict.parse(rpc_reply.xml)['rpc-reply']['chassis-inventory']['chassis']
|
|
|
|
result = dict()
|
|
|
|
# Gather chassis data
|
|
result['chassis'] = {
|
|
'serial': inventory_raw['serial-number'],
|
|
'description': inventory_raw['description'],
|
|
}
|
|
|
|
# Gather inventory items
|
|
result['items'] = glean_items(inventory_raw)
|
|
|
|
return result
|
|
|
|
|
|
class IOSSSH(SSHClient):
|
|
"""
|
|
SSH client for Cisco IOS devices
|
|
"""
|
|
|
|
def get_inventory(self):
|
|
def version():
|
|
|
|
def parse(cmd_out, rex):
|
|
for i in cmd_out:
|
|
match = re.search(rex, i)
|
|
if match:
|
|
return match.groups()[0]
|
|
|
|
sh_ver = self._send('show version').split('\r\n')
|
|
return {
|
|
'serial': parse(sh_ver, 'Processor board ID ([^\s]+)'),
|
|
'description': parse(sh_ver, 'cisco ([^\s]+)')
|
|
}
|
|
|
|
def items(chassis_serial=None):
|
|
cmd = self._send('show inventory').split('\r\n\r\n')
|
|
for i in cmd:
|
|
i_fmt = i.replace('\r\n', ' ')
|
|
try:
|
|
m_name = re.search('NAME: "([^"]+)"', i_fmt).group(1)
|
|
m_pid = re.search('PID: ([^\s]+)', i_fmt).group(1)
|
|
m_serial = re.search('SN: ([^\s]+)', i_fmt).group(1)
|
|
# Omit built-in items and those with no PID
|
|
if m_serial != chassis_serial and m_pid.lower() != 'unspecified':
|
|
yield {
|
|
'name': m_name,
|
|
'part_id': m_pid,
|
|
'serial': m_serial,
|
|
}
|
|
except AttributeError:
|
|
continue
|
|
|
|
self._send('term length 0')
|
|
sh_version = version()
|
|
|
|
return {
|
|
'chassis': sh_version,
|
|
'items': list(items(chassis_serial=sh_version.get('serial')))
|
|
}
|
|
|
|
|
|
class OpengearSSH(SSHClient):
|
|
"""
|
|
SSH client for Opengear devices
|
|
"""
|
|
default_credentials = {
|
|
'username': 'root',
|
|
'password': 'default',
|
|
}
|
|
|
|
def get_inventory(self):
|
|
|
|
try:
|
|
stdin, stdout, stderr = self.ssh.exec_command("showserial")
|
|
serial = stdout.readlines()[0].strip()
|
|
except:
|
|
raise RuntimeError("Failed to glean chassis serial from device.")
|
|
# Older models don't provide serial info
|
|
if serial == "No serial number information available":
|
|
serial = ''
|
|
|
|
try:
|
|
stdin, stdout, stderr = self.ssh.exec_command("config -g config.system.model")
|
|
description = stdout.readlines()[0].split(' ', 1)[1].strip()
|
|
except:
|
|
raise RuntimeError("Failed to glean chassis description from device.")
|
|
|
|
return {
|
|
'chassis': {
|
|
'serial': serial,
|
|
'description': description,
|
|
},
|
|
'items': [],
|
|
}
|
|
|
|
|
|
# For mapping platform -> NC client
|
|
RPC_CLIENTS = {
|
|
'juniper-junos': JunosNC,
|
|
'cisco-ios': IOSSSH,
|
|
'opengear': OpengearSSH,
|
|
}
|