import contextlib
import logging
import re
import ipaddress
import ncclient
import ncclient.operations
import ncclient.manager
from jnpr.junos import Device
from jnpr.junos import exception as EzErrors
from lxml import etree
import netifaces
from inventory_provider.tasks.common import asn_to_int
CONFIG_SCHEMA = """<?xml version="1.1" encoding="UTF-8" ?>
<xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema">
<xs:complexType name="generic-sequence">
<xs:sequence>
<xs:any processContents="lax" minOccurs="0" maxOccurs="unbounded"/>
</xs:sequence>
<xs:anyAttribute processContents="skip" />
</xs:complexType>
<!-- NOTE: 'unit' content isn't validated -->
<xs:complexType name="juniper-interface">
<xs:sequence>
<xs:choice minOccurs="1" maxOccurs="unbounded">
<xs:element name="name" minOccurs="1" maxOccurs="1" type="xs:string" />
<xs:element name="description" minOccurs="0" maxOccurs="1">
<xs:complexType>
<xs:simpleContent>
<xs:extension base="xs:string">
<xs:attribute name="inactive" type="xs:string" />
</xs:extension>
</xs:simpleContent>
</xs:complexType>
</xs:element>
<xs:any processContents="lax" minOccurs="0" maxOccurs="unbounded" />
</xs:choice>
</xs:sequence>
<xs:attribute name="inactive" type="xs:string" />
</xs:complexType>
<xs:element name="configuration">
<xs:complexType>
<xs:sequence>
<xs:choice minOccurs="1" maxOccurs="unbounded">
<xs:element name="transfer-on-commit" minOccurs="0" type="xs:string" />
<xs:element name="archive-sites" minOccurs="0" type="generic-sequence" />
<xs:element name="version" minOccurs="0" type="xs:string" />
<xs:element name="groups" minOccurs="0" type="generic-sequence" />
<xs:element name="apply-groups" minOccurs="0" type="xs:string" />
<xs:element name="system" minOccurs="0" type="generic-sequence" />
<xs:element name="logical-systems" minOccurs="0" type="generic-sequence" />
<xs:element name="chassis" minOccurs="0" type="generic-sequence" />
<xs:element name="services" minOccurs="0" type="generic-sequence" />
<xs:element name="interfaces" minOccurs="0">
<xs:complexType>
<xs:sequence>
<xs:choice minOccurs="1" maxOccurs="unbounded">
<xs:element name="apply-groups" minOccurs="0" type="xs:string" />
<xs:element name="interface-range" minOccurs="0" type="generic-sequence" />
<xs:element name="interface" minOccurs="1" maxOccurs="unbounded" type="juniper-interface" />
</xs:choice>
</xs:sequence>
</xs:complexType>
</xs:element>
<xs:element name="snmp" minOccurs="0" type="generic-sequence" />
<xs:element name="forwarding-options" minOccurs="0" type="generic-sequence" />
<xs:element name="routing-options" minOccurs="0" type="generic-sequence" />
<xs:element name="protocols" minOccurs="0" type="generic-sequence" />
<xs:element name="policy-options" minOccurs="0" type="generic-sequence" />
<xs:element name="class-of-service" minOccurs="0" type="generic-sequence" />
<xs:element name="firewall" minOccurs="0" type="generic-sequence" />
<xs:element name="routing-instances" minOccurs="0" type="generic-sequence" />
<xs:element name="bridge-domains" minOccurs="0" type="generic-sequence" />
<xs:element name="virtual-chassis" minOccurs="0" type="generic-sequence" />
<xs:element name="vlans" minOccurs="0" type="generic-sequence" />
<xs:element name="comment" minOccurs="0" type="xs:string" />
</xs:choice>
</xs:sequence>
<xs:attribute name="changed-seconds" type="xs:string" />
<xs:attribute name="changed-localtime" type="xs:string" />
<xs:attribute name="inactive" type="xs:string" />
</xs:complexType>
</xs:element>
</xs:schema>
""" # noqa: E501
UNIT_SCHEMA = """<?xml version="1.1" encoding="UTF-8" ?>
<xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema">
<xs:complexType name="generic-sequence">
<xs:sequence>
<xs:any processContents="lax" minOccurs="0" maxOccurs="unbounded"/>
</xs:sequence>
<xs:anyAttribute processContents="skip" />
</xs:complexType>
<xs:element name="unit">
<xs:complexType>
<xs:sequence>
<xs:element name="name" minOccurs="1" maxOccurs="1" type="xs:int" />
<xs:element name="description" minOccurs="0" maxOccurs="1" type="xs:string" />
<xs:any processContents="lax" minOccurs="0" maxOccurs="unbounded"/>
</xs:sequence>
<xs:attribute name="inactive" type="xs:string" />
</xs:complexType>
</xs:element>
</xs:schema>
""" # noqa: E501
class NetconfHandlingError(Exception):
pass
class TimeoutError(Exception):
pass
TIMEOUT = 10.0
@contextlib.contextmanager
def _nc_connection(host_params, ssh_params):
conn = ncclient.manager.connect(
host=host_params['hostname'],
port=host_params['port'],
username=ssh_params['username'],
key_filename=ssh_params['private-key'],
hostkey_verify=False,
manager_params={'timeout': TIMEOUT},
device_params={'name': 'junos'})
try:
yield conn # wait here until caller context ends
except (EzErrors.ConnectTimeoutError, ncclient.operations.errors.TimeoutExpiredError):
raise TimeoutError
finally:
conn.close_session()
def _raw_rpc(router, command):
# _rpc uses the easier to deal with library but can only
# query for router configs
# this is needed for querying for other things eg. interface speeds
obj = router.rpc(command)
xml = obj.tostring
return xml
@contextlib.contextmanager
def _rpc(hostname, ssh):
dev = Device(
host=hostname,
user=ssh['username'],
ssh_private_key_file=ssh['private-key'])
try:
dev.open()
yield dev.rpc
finally:
dev.close()
def validate_netconf_config(config_doc):
"""
:param config_doc:
:return:
:raises: NetconfHandlingError in case of validation errors
"""
logger = logging.getLogger(__name__)
def _validate(schema, doc):
if schema.validate(doc):
return
messages = []
for e in schema.error_log:
msg = f'{e.line}.{e.column}: {e.message}'
messages.append(msg)
logger.error(msg)
raise NetconfHandlingError('\n'.join(messages))
schema_doc = etree.XML(CONFIG_SCHEMA.encode('utf-8'))
config_schema = etree.XMLSchema(schema_doc)
_validate(config_schema, config_doc)
# validate interfaces/interface/unit elements ...
schema_doc = etree.XML(UNIT_SCHEMA.encode('utf-8'))
unit_schema = etree.XMLSchema(schema_doc)
for i in config_doc.xpath('//configuration/interfaces/interface'):
for u in i.xpath('./unit'):
_validate(unit_schema, u)
def load_config(hostname, ssh_params, validate=True):
"""
loads netconf data from the router, validates (by default) and
returns as an lxml etree doc
:param hostname: router hostname
:param ssh_params: 'ssh' config element(cf. config.py:CONFIG_SCHEMA)
:param validate: whether or not to validate netconf data (default True)
:return:
:raises: NetconfHandlingError or ConnectionError
"""
logger = logging.getLogger(__name__)
logger.info("capturing netconf data for '%s'" % hostname)
try:
with _rpc(hostname, ssh_params) as router:
config = router.get_config()
if validate:
validate_netconf_config(config)
return config
except (EzErrors.ConnectError, EzErrors.RpcError) as e:
raise ConnectionError(str(e))
def list_interfaces(netconf_config):
"""
generator that parses netconf output and
yields a list of interfaces
:param netconf_config: xml doc that was generated by load_config
:return:
"""
def _ifc_info(e):
# warning: this structure should match the default
# returned from routes.classifier.juniper_link_info
_name = e.find('name')
assert _name is not None, "expected interface 'name' child element"
ifc = {
'name': _name.text,
'description': '',
'bundle': [],
'speed': '' # set elsewhere but needs to be included to maintain default structure
}
description = e.find('description')
if description is not None:
ifc['description'] = description.text
ifc['bundle'] = e.xpath(
"./gigether-options[not(@inactive='inactive')]"
"/ieee-802.3ad[not(@inactive='inactive')]/bundle/text()")
ifc['ipv4'] = e.xpath('./family/inet/address/name/text()')
ifc['ipv6'] = e.xpath('./family/inet6/address/name/text()')
return ifc
def _inactive(interface_node):
return interface_node.get('inactive', None) == 'inactive'
def _units(base_name, interface_node):
for u in interface_node.xpath('./unit'):
if _inactive(u):
continue
unit_info = _ifc_info(u)
unit_info['name'] = f'{base_name}.{unit_info["name"]}'
yield unit_info
for i in netconf_config.xpath('//configuration/interfaces/interface'):
if _inactive(i):
continue
info = _ifc_info(i)
yield info
yield from _units(info['name'], i)
for ls_node in netconf_config.xpath('//configuration/logical-systems'):
logical_system = ls_node.xpath('./name/text()')
assert logical_system, 'no logical-system name found'
for i in ls_node.xpath('.//interfaces/interface'):
name = i.xpath('./name/text()')
assert name, "expected interface 'name' child element"
for u in _units(name[0], i):
u['logical-system'] = logical_system[0]
yield u
def _system_bgp_peers(system_node):
def _peering_params(neighbor_node):
address = neighbor_node.find('name').text
info = {'address': ipaddress.ip_address(address).exploded}
peer_as = neighbor_node.find('peer-as')
if peer_as is not None:
# lxml usage warning: can't just test `if peer_as:`
info['remote-asn'] = asn_to_int(peer_as.text)
local_as = neighbor_node.find('local-as')
if local_as is not None:
asn_value_node = local_as.find('as-number')
info['local-asn'] = asn_to_int(asn_value_node.text)
description = neighbor_node.find('description')
if description is not None:
# lxml usage warning: can't just test `if description:`
info['description'] = description.text
return info
def _neighbors(group_node):
for neighbor in group_node.xpath('./neighbor'):
inactive = neighbor.get('inactive')
if inactive == 'inactive':
continue
yield _peering_params(neighbor)
for group in system_node.xpath('./protocols/bgp/group'):
group_name = group.find('name').text
for peer in _neighbors(group):
peer['group'] = group_name
yield peer
for instance in system_node.xpath(
'./routing-instances/instance'):
instance_name = instance.find('name').text
for peer in _system_bgp_peers(instance):
peer['instance'] = instance_name
yield peer
def all_bgp_peers(netconf_config):
"""
Return all active bgp peering sessions defined for this router.
The response will be a generator, which renders a list
formatted according to the following schema:
.. asjson::
inventory_provider.routes.msr.PEERING_LIST_SCHEMA
EXCEPT: the 'hostname' parameter is not present
:param netconf_config:
:return: yields active peering sessions
"""
for base_system in netconf_config.xpath('//configuration'):
# there should only be one
yield from _system_bgp_peers(base_system)
for logical_system in netconf_config.xpath(
'//configuration/logical-systems'):
logical_system_name = logical_system.find('name').text
for peer in _system_bgp_peers(logical_system):
peer['logical-system'] = logical_system_name
yield peer
def interface_addresses(netconf_config):
"""
Yields a list of all distinct interface addresses.
:param netconf_config:
:return:
"""
for ifc in list_interfaces(netconf_config):
for address in ifc['ipv4'] + ifc['ipv6']:
yield {
"name": ipaddress.ip_interface(address).ip.exploded,
"interface address": address,
"interface name": ifc['name']
}
def local_interfaces(
type=netifaces.AF_INET,
omit_link_local=True,
omit_loopback=True):
"""
Generator yielding IPv4Interface or IPv6Interface objects for
the interfaces present on the local system,
depending on the value of type.
:param type: hopefully AF_INET or AF_INET6
:param omit_link_local: skip v6 fe80* addresses if true
:param omit_loopback: skip lo* interfaces if true
:return:
"""
for n in netifaces.interfaces():
if omit_loopback and re.match(r'^lo\d+', n):
continue
am = netifaces.ifaddresses(n)
for a in am.get(type, []):
if omit_link_local and a['addr'].startswith('fe80:'):
continue
m = re.match(r'^(.+?)(%.*)?$', a['addr'])
assert m
addr = m.group(1)
m = re.match(r'.*/(\d+)$', a['netmask'])
if m:
mask = m.group(1)
else:
mask = a['netmask']
yield ipaddress.ip_interface('%s/%s' % (addr, mask))
def snmp_community_string(netconf_config):
my_addressess = list([i.ip for i in local_interfaces()])
for community in netconf_config.xpath('//configuration/snmp/community'):
for subnet in community.xpath('./clients/name/text()'):
allowed_network = ipaddress.ip_network(subnet, strict=False)
for me in my_addressess:
if me in allowed_network:
return community.xpath('./name/text()')[0]
return None
def netconf_changed_timestamp(netconf_config):
"""
Return the last change timestamp published by the config document.
:param netconf_config: netconf lxml etree document
:return: an epoch timestamp (integer number of seconds) or None
"""
for ts in netconf_config.xpath('/configuration/@changed-seconds'):
if re.match(r'^\d+$', ts):
return int(ts)
logger = logging.getLogger(__name__)
logger.warning('no valid timestamp found in netconf configuration')
return None
def logical_systems(netconf_config):
"""
Return a list of logical system names for the router.
It's not an error if a router has no defined logical systems.
:param netconf_config: netconf lxml etree document
:return: a list of strings
"""
return netconf_config.xpath('//configuration/logical-systems/name/text()')
def get_interface_info_for_router(hostname, ssh_config):
"""
Return an XML document specifying all interfaces for the router at hostname.
:param hostname: hostname for router
:param ssh_config: config SSH dict
:return: string containing XML document giving interface information (including speeds)
"""
host_params = {
"hostname": hostname,
"port": 830
}
with _nc_connection(
host_params=host_params,
ssh_params=ssh_config) as router:
reply = _raw_rpc(router, etree.Element('get-interface-information'))
return reply
def get_interface_speeds(interface_info):
"""
Using the output from load_interfaces_for_router, retrieve the speeds matched to each interface.
:param interface_info: interface info lxml etree document
:return: dictionary specifying interface name and its associated speed
"""
def _get_speed_for_interface(physical_interface):
name = physical_interface.xpath("./*[local-name() = 'name']")
if len(name) > 0:
name = name[0].text.strip()
else:
logger = logging.getLogger(__name__)
logger.warning('no name entries for interface')
speed = physical_interface.xpath("./*[local-name() = 'speed']")
if len(speed) > 0:
speed = speed[0].text.strip()
else:
speed = ""
return {
"interface": name,
"speed": speed
}
all_speeds = [_get_speed_for_interface(interface) for interface in
interface_info.xpath("//*[local-name() = 'physical-interface']")]
all_speeds_index = {interface["interface"]: interface["speed"] for interface in all_speeds}
return all_speeds_index