msr.py 50.49 KiB
"""
MSR Support Endpoints
=========================
These endpoints are intended for use by MSR.
.. contents:: :local:
/msr/access-services
---------------------------------
.. autofunction::inventory_provider.routes.msr.get_access_services
/msr/bgp/logical-systems
-------------------------------------
.. autofunction:: inventory_provider.routes.msr.get_logical_systems
/msr/bgp/logical-system-peerings</name>
------------------------------------------
.. autofunction:: inventory_provider.routes.msr.logical_system_peerings
/msr/bgp/peering-services
-------------------------------------
.. autofunction:: inventory_provider.routes.msr.get_peering_services
/msr/bgp/groups
-------------------------------------
.. autofunction:: inventory_provider.routes.msr.get_peering_groups
/msr/bgp/group-peerings</name>
-------------------------------------
.. autofunction:: inventory_provider.routes.msr.bgp_group_peerings
/msr/bgp/routing-instances
-------------------------------------
.. autofunction:: inventory_provider.routes.msr.get_peering_routing_instances
/msr/bgp/routing-instance-peerings</name>
--------------------------------------------
.. autofunction:: inventory_provider.routes.msr.bgp_routing_instance_peerings
/msr/bgp
--------------------------------------------
.. autofunction:: inventory_provider.routes.msr.bgp_all_peerings
/msr/mdpvn
--------------------------------------------
.. autofunction:: inventory_provider.routes.msr.mdvpn
/msr/services
--------------------------------------------
.. autofunction:: inventory_provider.routes.msr.get_system_correlation_services
/msr/vpn-proxy
--------------------------------------------
.. autofunction:: inventory_provider.routes.msr.vpn_proxy
/msr/asn-peers
--------------------------------------------
.. autofunction:: inventory_provider.routes.msr._asn_peers
/msr/ip-services
--------------------------------------------
.. autofunction:: inventory_provider.routes.msr.ip_services
helpers
-------------------------------------
.. autofunction:: inventory_provider.routes.msr._handle_peering_group_list_request
.. autofunction:: inventory_provider.routes.msr._handle_peering_group_request
""" # noqa E501
import binascii
import functools
import hashlib
import itertools
import json
import ipaddress
import logging
import re
import threading
from collections import defaultdict
from typing import Dict
from flask import Blueprint, Response, request, current_app
import jsonschema
from inventory_provider.routes import common
from inventory_provider.routes.classifier import \
get_ims_equipment_name, get_ims_interface, get_interface_services_and_loc
from inventory_provider.routes.common import _ignore_cache_or_retrieve, \
ims_equipment_to_hostname
from inventory_provider.routes.poller import get_services
from inventory_provider.tasks import common as tasks_common
routes = Blueprint('msr-query-routes', __name__)
logger = logging.getLogger(__name__)
_subnet_lookup_semaphore = threading.Semaphore()
PEERING_GROUP_LIST_SCHEMA = {
'$schema': 'https://json-schema.org/draft-07/schema#',
'type': 'array',
'items': {'type': 'string'}
}
PEERING_LIST_SCHEMA = {
'$schema': 'https://json-schema.org/draft-07/schema#',
'definitions': {
'peering-instance': {
'type': 'object',
'properties': {
'address': {'type': 'string'},
'description': {'type': 'string'},
'logical-system': {'type': 'string'},
'group': {'type': 'string'},
'hostname': {'type': 'string', 'format': 'hostname'},
'remote-asn': {'type': 'integer'},
'local-asn': {'type': 'integer'},
'instance': {'type': 'string'}
},
# only vrr peerings have remote-asn
# only group peerings have local-asn or instance
# not all group peerings have 'description'
# and only vrr or vpn-proxy peerings are within a logical system
'required': [
'address',
'group',
'hostname'],
'additionalProperties': False
}
},
'type': 'array',
'items': {'$ref': '#/definitions/peering-instance'}
}
IP_ADDRESS_LIST_SCHEMA = {
'$schema': 'https://json-schema.org/draft-07/schema#',
'definitions': {
"ip-address": {
"type": "string",
"oneOf": [
{"pattern": r'^(\d+\.){3}\d+$'},
{"pattern": r'^[a-f0-9:]+$'}
]
}
},
'type': 'array',
'items': {'$ref': '#/definitions/ip-address'},
'minItems': 1
}
PEERING_ADDRESS_SERVICES_LIST = {
'$schema': 'https://json-schema.org/draft-07/schema#',
'definitions': {
'service': {
'properties': {
'id': {'type': 'integer'},
'name': {'type': 'string'},
'type': {'type': 'string'},
'status': {'type': 'string'}
},
'required': ['name', 'type', 'status'],
'additionalProperties': False
},
'address-service-info': {
'properties': {
'address': {'type': 'string'},
'hostname': {'type': 'string'},
'interface': {'type': 'string'},
'services': {
'type': 'array',
'items': {'$ref': '#/definitions/service'}
}
},
'required': ['address', 'hostname', 'interface', 'services'],
'additionalProperties': False
}
},
'type': 'array',
'items': {'$ref': '#/definitions/address-service-info'}
}
SYSTEM_CORRELATION_SERVICES_LIST_SCHEMA = {
'$schema': 'https://json-schema.org/draft-07/schema#',
'definitions': {
'v4-network': {'type': 'string'}, # TODO: can this be better?
'v6-network': {'type': 'string'}, # TODO: can this be better?
'ip-endpoint': {
'type': 'object',
'properties': {
'hostname': {'type': 'string'},
'interface': {'type': 'string'},
'addresses': {
'type': 'object',
'properties': {
'v4': {'$ref': '#/definitions/v4-network'},
'v6': {'$ref': '#/definitions/v6-network'}
},
# 'required': ['v4', 'v6'], # TODO: always require both?
'additionalProperties': False
}
},
'required': ['hostname', 'interface'],
'additionalProperties': False
},
'optical-endpoint': {
'type': 'object',
'properties': {
'equipment': {'type': 'string'},
'port': {'type': 'string'}
},
'required': ['equipment', 'port'],
'additionalProperties': False
},
'endpoints': {
'type': 'array',
'items': {
'oneOf': [
{'$ref': '#/definitions/ip-endpoint'},
{'$ref': '#/definitions/optical-endpoint'}
]
},
'minItems': 1
},
'service': {
'type': 'object',
'properties': {
'circuit_id': {'type': 'integer'},
'sid': {'type': 'string'},
'status': {'type': 'string'},
'monitored': {'type': 'boolean'},
'name': {'type': 'string'},
'speed': {'type': 'integer'},
'circuit_type': {'type': 'string'}, # TODO: remove this?
'service_type': {'type': 'string'}, # TODO: enum?
'project': {'type': 'string'}, # TODO: remove this?
'customer': {'type': 'string'},
'endpoints': {'$ref': '#/definitions/endpoints'},
'redundant_asn': {'type': 'integer'}
},
'required': [
'circuit_id', 'sid', 'name', 'speed', 'status', 'monitored',
# 'circuit_type', 'project', # TODO: keeping these?!?
'service_type', 'customer', 'endpoints'],
'additionalProperties': False
}
},
'type': 'array',
'items': {'$ref': '#/definitions/service'},
'minItems': 1 # otherwise the route should return 404
}
MDVPN_LIST_SCHEMA = {
'$schema': 'https://json-schema.org/draft-07/schema#',
'definitions': {
'mdvpn_group': {
'type': 'object',
'properties': {
'asn': {'type': 'integer'},
'AP': {'$ref': '#/definitions/ap_peerings'},
'VRR': {'$ref': '#/definitions/vrr_peerings'}
},
'required': [
'asn', 'AP', 'VRR'
],
'additionalProperties': False
},
'ap_peerings': {
'type': 'array',
'items': {
'$ref': '#/definitions/bgplu_peering'
}
},
'bgplu_peering': {
'type': 'object',
'properties': {
'name': {'type': 'string'},
'v4': {'type': 'string'},
'v6': {'type': 'string'},
'hostname': {'type': 'string'}
},
'required': [
'name', 'v4', 'v6', 'hostname'
],
'additionalProperties': False
},
'vrr_peerings': {
'type': 'array',
'items': {
'$ref': '#/definitions/vpn_peering'
}
},
'vpn_peering': {
'type': 'object',
'properties': {
'description': {'type': 'string'},
'v4': {'type': 'string'},
'hostname': {
'type': 'array',
'items': {
'type': 'string'
},
'minItems': 1
}
},
'additionalProperties': False
}
},
'type': 'array',
'items': {'$ref': '#/definitions/mdvpn_group'}
}
VPN_PROXY_LIST_SCHEMA = {
'$schema': 'https://json-schema.org/draft-07/schema#',
'definitions': {
'vpn_proxy_peering': {
'type': 'object',
'properties': {
'pop': {'type': 'string'},
'nren': {'type': 'string'},
'group': {'type': 'string'},
'v4': {'type': 'string'}
},
'additionalProperties': False
}
},
'type': 'array',
'items': {'$ref': '#/definitions/vpn_proxy_peering'}
}
DOMAIN_TO_POP_MAPPING = {
"mad.es": "Madrid",
"bra.sk": "Bratislava",
"vie.at": "Vienna",
"gen.ch": "Geneva",
"fra.de": "Frankfurt",
"pra.cz": "Prague",
"ams.nl": "Amsterdam"
}
IP_SERVICES_LIST_SCHEMA = {
'$schema': 'https://json-schema.org/draft-07/schema#',
'definitions': {
'minimal-peering-schema': {
# cf. PEERING_LIST_SCHEMA
'type': 'object',
'properties': {
'address': {'type': 'string'},
'group': {'type': 'string'},
'hostname': {'type': 'string', 'format': 'hostname'}
},
'required': ['address', 'group', 'hostname'],
'additionalProperties': True
},
'service': {
'type': 'object',
'properties': {
'name': {'type': 'string'},
'customer': {'type': 'string'},
'type': {'type': 'string'},
'status': {'type': 'string'},
'pop': {
'type': 'object',
'properties': {
'name': {'type': 'string'},
'abbrev': {'type': 'string'},
},
'required': ['name', 'abbrev'],
'additionalProperties': False
}
},
'required': ['name', 'customer', 'type', 'pop', 'status'],
'additionalProperties': False
},
'interface-address': {
'properties': {
'hostname': {'type': 'string', 'format': 'hostname'},
'port': {'type': 'string'},
'address': {'type': 'string'},
'services': {
'type': 'array',
'items': {'$ref': '#/definitions/service'}
},
'peerings': {
'type': 'array',
'items': {'$ref': '#/definitions/minimal-peering-schema'}
}
},
'required': [
'hostname', 'port', 'address', 'services', 'peerings'],
'additionalProperties': False
}
},
'type': 'array',
'items': {'$ref': '#/definitions/interface-address'}
}
# very similar to PEERING_LIST_SCHEMA but
# with a field for NREN, which is required
ASN_PEER_LIST_SCHEMA = {
'$schema': 'https://json-schema.org/draft-07/schema#',
'definitions': {
'peering-instance': {
'type': 'object',
'properties': {
'address': {'type': 'string'},
'description': {'type': 'string'},
'logical-system': {'type': 'string'},
'group': {'type': 'string'},
'hostname': {'type': 'string'},
'remote-asn': {'type': 'integer'},
'local-asn': {'type': 'integer'},
'instance': {'type': 'string'},
'nren': {'type': 'string'}
},
# only vrr peerings have remote-asn
# only group peerings have local-asn or instance
# not all group peerings have 'description'
# and only vrr or vpn-proxy peerings are within a logical system
'required': [
'address',
'group',
'hostname',
'nren'],
'additionalProperties': False
}
},
'type': 'array',
'items': {'$ref': '#/definitions/peering-instance'}
}
@routes.after_request
def after_request(resp):
return common.after_request(resp)
def _handle_peering_group_request(name, cache_key, group_key_base):
"""
Common method for used by
:meth:`inventory_provider.routes.msr.logical_system_peerings` and
:meth:`inventory_provider.routes.msr.bgp_group_peerings`.
This method will return a list of all peerings configured
for the specified group `name on any router,
or for all group names if `name` None.
The response will be formatted according to the following schema:
.. asjson::
inventory_provider.routes.msr.PEERING_LIST_SCHEMA
:param name: group/logical-system name, or None
:param cache_key: base cache key for this type of request
:param group_key_base: key above which the peerings are grouped
:return: a json list, formatted as above
"""
r = common.get_current_redis()
def _get_all_subkeys():
keys = []
for k in r.scan_iter(f'{group_key_base}:*', count=1000):
keys.append(k.decode('utf-8'))
return keys
def _load_list_items(key):
value = r.get(key)
if value:
yield from json.loads(value.decode('utf-8'))
if name:
cache_key = f'{cache_key}:{name}'
items = _ignore_cache_or_retrieve(request, cache_key, r)
if not items:
if name:
items = list(_load_list_items(f'{group_key_base}:{name}'))
else:
gen_list = list(map(_load_list_items, _get_all_subkeys()))
items = list(itertools.chain(*gen_list))
if not items:
return Response(
response='no peerings found',
status=404,
mimetype="text/html")
items = json.dumps(items)
r.set(cache_key, items.encode('utf-8'))
return Response(items, mimetype="application/json")
@routes.route('/bgp/logical-system-peerings', methods=['GET'])
@routes.route('/bgp/logical-system-peerings/<name>', methods=['GET'])
@common.require_accepts_json
def logical_system_peerings(name=None):
"""
Handler for `/msr/bgp/logical-system-peerings`
This method will return a list of all peerings configured
for the requested logical-system name on any router, or for any
logical system if no parameter is given.
:return: see :meth:`inventory_provider.routes.msr._handle_peering_group_request`
""" # noqa: E501
return _handle_peering_group_request(
name=name,
cache_key='classifier-cache:msr:logical-system-peerings',
group_key_base='juniper-peerings:logical-system')
@routes.route('/bgp/group-peerings', methods=['GET'])
@routes.route('/bgp/group-peerings/<name>', methods=['GET'])
@common.require_accepts_json
def bgp_group_peerings(name=None):
"""
Handler for `/msr/bgp/group-peerings`
This method will return a list of all peerings configured
for the requested group name on any router, or for any
group system if no parameter is given.
:return: see :meth:`inventory_provider.routes.msr._handle_peering_group_request`
""" # noqa: E501
return _handle_peering_group_request(
name=name,
cache_key='classifier-cache:msr:group-peerings',
group_key_base='juniper-peerings:group')
@routes.route('/bgp/routing-instance-peerings', methods=['GET'])
@routes.route('/bgp/routing-instance-peerings/<name>', methods=['GET'])
@common.require_accepts_json
def bgp_routing_instance_peerings(name=None):
"""
Handler for `/msr/bgp/routing-instance-peerings`
This method will return a list of all peerings configured
for the requested routing-instance name on any router, or for any
routing instance if no parameter is given.
:return: see :meth:`inventory_provider.routes.msr._handle_peering_group_request`
""" # noqa: E501
return _handle_peering_group_request(
name=name,
cache_key='classifier-cache:msr:routing-instance-peerings',
group_key_base='juniper-peerings:routing-instance')
def _handle_peering_group_list_request(cache_key, group_key_base):
"""
Common method for used by
:meth:`inventory_provider.routes.msr.get_logical_systems` and
:meth:`inventory_provider.routes.msr.get_peering_groups`.
This method will return a list of all immediate subkeys of
`group_key_base`.
The response will be formatted according to the following schema:
.. asjson::
inventory_provider.routes.msr.PEERING_GROUP_LIST_SCHEMA
:param cache_key: base cache key for this type of request
:param group_key_base: key above which the peerings are grouped
:return: a json list, formatted as above
"""
r = common.get_current_redis()
def _get_all_subkeys():
for k in r.scan_iter(f'{group_key_base}:*', count=1000):
k = k.decode('utf-8')
yield k[len(group_key_base) + 1:]
names = _ignore_cache_or_retrieve(request, cache_key, r)
if not names:
names = list(_get_all_subkeys())
if not names:
return Response(
response='no groups found',
status=404,
mimetype="text/html")
names = json.dumps(sorted(names))
r.set(cache_key, names.encode('utf-8'))
return Response(names, mimetype="application/json")
@routes.route('/bgp/logical-systems', methods=['GET'])
@common.require_accepts_json
def get_logical_systems():
"""
Handler for `/msr/bgp/logical-systems`
Returns a list of logical system names for which peering
information is available.
:return: see :meth:`inventory_provider.routes.msr._handle_peering_group_list_request`
""" # noqa: E501
return _handle_peering_group_list_request(
cache_key='classifier-cache:msr:logical-systems',
group_key_base='juniper-peerings:logical-system')
@routes.route('/bgp/groups', methods=['GET'])
@common.require_accepts_json
def get_peering_groups():
"""
Handler for `/msr/bgp/groups`
Returns a list of group names for which peering
information is available.
:return: see :meth:`inventory_provider.routes.msr._handle_peering_group_list_request`
""" # noqa: E501
return _handle_peering_group_list_request(
cache_key='classifier-cache:msr:peering-groups',
group_key_base='juniper-peerings:group')
@routes.route('/bgp/routing-instances', methods=['GET'])
@common.require_accepts_json
def get_peering_routing_instances():
"""
Handler for `/msr/bgp/routing-instances`
Returns a list of routing-instance names for which peering
information is available.
:return: see :meth:`inventory_provider.routes.msr._handle_peering_group_list_request`
""" # noqa: E501
return _handle_peering_group_list_request(
cache_key='classifier-cache:msr:routing-instances',
group_key_base='juniper-peerings:routing-instance')
@routes.route('/access-services', methods=['GET'])
@common.require_accepts_json
def get_access_services():
"""
Handler for `/msr/access-services`
Same as `/poller/services/geant_ip`
cf. :meth:`inventory_provider.routes.poller.get_services`
"""
return get_services(service_type='geant_ip')
def _find_subnet_keys(addresses):
"""
yields pairs like:
(redis key [str], address [str])
we search to the end of the list in case of network config
errors (same address in multiple subnets)
:param addresses: iterable of strings (like PEER_ADDRESS_LIST)
:return: as above
"""
# make a dict & remove duplicates
# will raise in case of invalid addresses
remaining_addresses = {
a: ipaddress.ip_address(a)
for a in set(addresses)
}
r = common.get_current_redis()
# scan with bigger batches, to mitigate network latency effects
for k in r.scan_iter('subnets:*', count=1000):
if not remaining_addresses:
break
k = k.decode('utf-8')
m = re.match(r'^subnets:(.*)$', k)
assert m, 'sanity failure: redis returned an invalid key name'
interface = ipaddress.ip_interface(m.group(1))
try:
matched_address = next(
a for a, v
in remaining_addresses.items()
if v == interface.ip)
del remaining_addresses[matched_address]
yield k, matched_address
except StopIteration:
# no match
continue
@functools.lru_cache(100)
def _get_subnets(r):
result = {}
for k in r.scan_iter('subnets:*', count=1000):
k = k.decode('utf-8')
m = re.match(r'^subnets:(.+)$', k)
assert m
result[k] = ipaddress.ip_interface(m.group(1)).network
return result
def _get_subnet_interfaces(address, r):
# synchronize calls to _get_subnets, so we don't
# call it many times together when running in
# multi-thread mode
_subnet_lookup_semaphore.acquire()
try:
all_subnets = _get_subnets(r)
except Exception:
logger.exception('error looking up subnets')
all_subnets = {}
finally:
_subnet_lookup_semaphore.release()
address = ipaddress.ip_address(address)
for key, network in all_subnets.items():
if address not in network:
continue
value = r.get(key)
if not value:
logger.error(f'no value for for redis key "{key}"')
continue
yield from json.loads(value.decode('utf-8'))
def _get_services_for_address(address: str, r):
"""
match this address against all interfaces, then look up
any known services for that port
address is assumed to be in a valid v4/v6 format (it's used to
construct a ipaddress.ip_address object without try/except)
:param address: ip address string
:param r: a Redis instance
:return: yields PEERING_ADDRESS_SERVICES_LIST elements
"""
def _formatted_service(s):
return {
'id': s['id'],
'name': s['name'],
'type': s['service_type'],
'status': s['status']
}
for ifc_info in _get_subnet_interfaces(address, r):
ims_source_equipment = get_ims_equipment_name(
ifc_info['router'], r)
ims_interface = get_ims_interface(ifc_info['interface name'])
service_info = get_interface_services_and_loc(
ims_source_equipment, ims_interface, r)
services = service_info.get('services', [])
services = map(_formatted_service, services)
services = sorted(services, key=lambda x: x['name'])
yield {
'address': address,
'hostname': ifc_info['router'],
'interface': ifc_info['interface name'],
'services': list(services)
}
def _load_address_services_proc(address_queue, results_queue, config_params):
"""
create a local redis connection with the current db index,
lookup the values of the keys that come from key_queue
and put them on value_queue
i/o contract:
None arriving on key_queue means no more keys are coming
put None in value_queue means we are finished
:param key_queue:
:param value_queue:
:param config_params: app config
:param doc_type: decoding type to do (xml or json)
:return: nothing
"""
try:
r = tasks_common.get_current_redis(config_params)
while True:
address = address_queue.get()
# contract is that None means no more addresses
if not address:
break
for service_info in _get_services_for_address(address, r):
results_queue.put(service_info)
except json.JSONDecodeError:
logger.exception(f'error decoding redis entry for {address}')
except Exception:
# just log info about this error (for debugging only)
# ... and quit (i.e. let finally cleanup)
logger.exception(f'error looking up service info for {address}')
finally:
# contract is to return None when finished
results_queue.put(None)
def _get_peering_services_multi_thread(addresses):
"""
normal handler for `/msr/bgp/peering-services`
this one does the lookups in multiple threads, each with its own
redis connection
(cf. _get_peering_services_single_thread)
:param addresses: iterable of address strings
:return: yields dicts returned from _get_services_for_address
"""
yield from common.distribute_jobs_across_workers(
worker_proc=_load_address_services_proc,
jobs=addresses,
input_ctx=current_app.config['INVENTORY_PROVIDER_CONFIG'],
num_threads=min(len(addresses), 10))
def _get_peering_services_single_thread(addresses):
"""
used by `/msr/bgp/peering-services`
this one does the lookups serially, in the current thread and a single
redis connection
(cf. _get_peering_services_multi_thread)
:param addresses: iterable of address strings
:return: yields dicts returned from _get_services_for_address
"""
r = common.get_current_redis()
for a in addresses:
yield from _get_services_for_address(a, r)
def _obj_key(o):
m = hashlib.sha256()
m.update(json.dumps(json.dumps(o)).encode('utf-8'))
digest = binascii.b2a_hex(m.digest()).decode('utf-8')
return digest.upper()[-4:]
@routes.route('/bgp/peering-services', methods=['POST'])
@common.require_accepts_json
def get_peering_services():
"""
Handler for `/msr/bgp/peering-services`
This method must be called with POST method, and the payload
should be a json-formatted list of addresses (strings), which will
be validated against the following schema:
.. asjson::
inventory_provider.routes.msr.IP_ADDRESS_LIST_SCHEMA
The response will be formatted as follows:
.. asjson::
inventory_provider.routes.msr.PEERING_ADDRESS_SERVICES_LIST
A `no-threads` can be also be given. If its truthiness
value evaluates to True, then the lookups are done in a single thread.
(This functionality is mainly for testing/debugging - it's not
expected to be used in production.)
:return:
"""
addresses = request.json
jsonschema.validate(addresses, IP_ADDRESS_LIST_SCHEMA)
addresses = set(addresses) # remove duplicates
input_data_key = _obj_key(sorted(list(addresses)))
cache_key = f'classifier-cache:msr:peering-services:{input_data_key}'
r = common.get_current_redis()
response = _ignore_cache_or_retrieve(request, cache_key, r)
if not response:
# validate addresses, to decrease chances of dying in a worker thread
for a in addresses:
assert ipaddress.ip_address(a)
no_threads = common.get_bool_request_arg('no-threads', False)
if no_threads:
response = _get_peering_services_single_thread(addresses)
else:
response = _get_peering_services_multi_thread(addresses)
response = list(response)
if response:
response = json.dumps(response)
r.set(cache_key, response.encode('utf-8'))
if not response:
return Response(
response='no interfaces found',
status=404,
mimetype="text/html")
return Response(response, mimetype="application/json")
# TODO: @functools.cache is only available in py3.9
@functools.lru_cache(maxsize=None)
def _load_all_interfaces():
"""
loads all ip interfaces in the network and returns as a dict
of dicts:
hostname -> interface name -> interface info
:return: dict of dicts
"""
# dict of dicts:
# peering_info[hostname][interface_name] = dict of ifc details
result = defaultdict(dict)
host_if_extraction_re = re.compile(r'^netconf-interfaces:(.+?):')
for doc in common.load_json_docs(
config_params=current_app.config['INVENTORY_PROVIDER_CONFIG'],
key_pattern='netconf-interfaces:*',
num_threads=20):
matches = host_if_extraction_re.match(doc['key'])
if matches:
hostname = matches[1]
interface_name = doc['value']['name']
result[hostname][interface_name] = doc['value']
return result
# TODO: @functools.cache is only available in py3.9
@functools.lru_cache(maxsize=None)
def _load_redundant_access_peers():
"""
load all peers that should be considered
redundant for access services
that is, all peers for services like NREN-APx
:return: dict of [peer address] -> [remote asn]
"""
r = common.get_current_redis()
result = {}
# cf. REPORTING-312: limited to eGEANT group,
# but this can be expanded here in future
redundant_access_groups = ['eGEANT']
for g in redundant_access_groups:
doc = r.get(f'juniper-peerings:group:{g}')
for peer in json.loads(doc.decode('utf-8')):
result[peer['address']] = peer['remote-asn']
return result
def _ip_endpoint_extractor(endpoint_details: dict):
"""
special-purpose method used only by _endpoint_extractor
operates on a dictionary formatted as in worker.transform_ims_data
(cf sid_services)
WARNING: assumes hostname is a router in _load_all_interfaces,
asserts if not
:param endpoint_details: dict formatted as in worker.transform_ims_data
:return: dict formatted
as SYSTEM_CORRELATION_SERVICES_LIST_SCHEMA.ip-endpoint
"""
hostname = ims_equipment_to_hostname(endpoint_details['equipment'])
interface = endpoint_details['port'].lower()
ip_endpoint = {
'hostname': hostname,
'interface': interface,
}
all_interfaces = _load_all_interfaces()
# sanity: should have already been checked
assert hostname in all_interfaces
host_info = all_interfaces[hostname]
interface_info = host_info.get(interface, {})
addresses = {}
ipv4 = interface_info.get('ipv4')
ipv6 = interface_info.get('ipv6')
if ipv4:
addresses['v4'] = ipv4[0]
if ipv6:
addresses['v6'] = ipv6[0]
if addresses:
ip_endpoint['addresses'] = addresses
return ip_endpoint
def _endpoint_extractor(endpoint_details: Dict):
"""
special-purpose method used only by get_system_correlation_services
operates on a dictionary formatted as in worker.transform_ims_data
(cf sid_services)
WARNING: assumes hostname is a router in _load_all_interfaces,
asserts if not
:param endpoint_details: dict formatted as in worker.transform_ims_data
:return: dict formatted as one element of the array
SYSTEM_CORRELATION_SERVICES_LIST_SCHEMA.endpoints
"""
if not endpoint_details['geant_equipment']:
return
potential_hostname = ims_equipment_to_hostname(
endpoint_details['equipment'])
all_routers = _load_all_interfaces().keys()
if potential_hostname in all_routers:
return _ip_endpoint_extractor(endpoint_details)
else:
return {
'equipment': endpoint_details['equipment'],
'port': endpoint_details['port']
}
@routes.route('/services', methods=['GET'])
@common.require_accepts_json
def get_system_correlation_services():
"""
Handler for `/msr/services`
This method returns all known services with with information required
by the reporting tool stack.
cf. https://jira.software.geant.org/browse/POL1-530
The response will be formatted as follows:
.. asjson::
inventory_provider.routes.msr.SYSTEM_CORRELATION_SERVICES_LIST_SCHEMA
:return:
"""
def _get_redundancy_asn(endpoints):
# endpoints should be a list formatted as
# SYSTEM_CORRELATION_SERVICES_LIST_SCHEMA.endpoints
redundant_peerings = _load_redundant_access_peers()
for ep in endpoints:
addresses = ep.get('addresses', {})
for ifc_address in addresses.values():
for p, asn in redundant_peerings.items():
peer = ipaddress.ip_address(p)
ifc = ipaddress.ip_interface(ifc_address)
if peer in ifc.network:
return asn
return None
cache_key = 'classifier-cache:msr:services'
r = common.get_current_redis()
response = _ignore_cache_or_retrieve(request, cache_key, r)
if not response:
sid_services = json.loads(r.get('ims:sid_services').decode('utf-8'))
response = []
for sid, details in sid_services.items():
service_info = {'endpoints': []}
for d in details:
if not service_info.get('sid'):
service_info['circuit_id'] = d['circuit_id']
service_info['sid'] = d['sid']
service_info['status'] = d['status']
service_info['monitored'] = d['monitored']
service_info['name'] = d['name']
service_info['speed'] = d['speed']
service_info['service_type'] = d['service_type']
service_info['customer'] = d['customer']
endpoint = _endpoint_extractor(d)
if endpoint:
service_info['endpoints'].append(endpoint)
if service_info.get('endpoints'):
asn = _get_redundancy_asn(service_info['endpoints'])
if asn:
service_info['redundant_asn'] = asn
response.append(service_info)
jsonschema.validate(
response, SYSTEM_CORRELATION_SERVICES_LIST_SCHEMA)
if response:
response = json.dumps(response, indent=2)
# r.set(cache_key, response.encode('utf-8'))
if not response:
return Response(
response='no services found',
status=404,
mimetype="text/html")
return Response(response, mimetype="application/json")
@routes.route('/bgp', methods=['GET'])
@common.require_accepts_json
def bgp_all_peerings():
"""
Handler for `/bgp`
This method returns a list of all BGP peerigns.
The response will be formatted according to the following schema:
.. asjson::
inventory_provider.routes.msr.PEERING_LIST_SCHEMA
:return:
"""
r = common.get_current_redis()
response = r.get('juniper-peerings:all')
return Response(response.decode('utf-8'), mimetype="application/json")
@routes.route('/mdvpn', methods=['GET'])
@common.require_accepts_json
def mdvpn():
"""
Handler for `/mdvpn`
This method returns a list of all BGP-LU peerings, and the VR peerings
for both Paris & Ljubljana.
The response will be formatted according to the following schema:
.. asjson::
inventory_provider.routes.msr.MDVPN_LIST_SCHEMA
:return:
"""
def _get_consistent_description(description):
"""
The same interface in VRR peerings can have multiple names.
These names are (currently) the same but with a different local prefix,
with no ordering guaranteed by the redis cache.
As only one description is returned by this endpoint for each
IPv4 address, this serves as a quick and dirty way of merging these
multiple descriptions into one an external user can use to identify
the peering reliably.
:param description: The raw description for a VRR peering
:return: The same description with location prefix removed
"""
# it is incredibly likely this will need revision later down the line
expected_prefixes = [
"MD-VPN-VRR-PARIS-",
"MD-VPN-VRR-LJUBLJANA-"
]
for prefix in expected_prefixes:
if description.startswith(prefix):
return description.replace(prefix, '')
return description
def _make_group_index(group, index_key):
"""
Utility function to take a list and make it a dict based off a given
key, for fast lookup of a specific key field.
:param group: A list of dicts which should all have `index_key` as a
field
:param index_key: Name of the key to index on
:return: Dict with `index_key` as the key field and a list of all
matching dicts as the value
"""
index = {}
for peering in group:
key = peering.get(index_key)
index.setdefault(key, []).append(peering)
return index
def _bgplu_peerings(asn, bgplu_index):
for peering in bgplu_index.get(asn, []):
formatted_peering = {
"name": peering['description'],
"v4": peering['address'],
"v6": '',
"hostname": peering['hostname']
}
yield formatted_peering
def _vpnrr_peerings(asn, vpnrr_index):
# rearrange into index using ipv4 as key
# this will collect related entries under the same ipv4
ip_index = _make_group_index(vpnrr_index.get(asn, []), 'address')
for ip, ip_details in ip_index.items():
hostnames = [item['hostname'] for item in ip_details]
description = ip_details[0]['description']
formatted_peering = {
"description": _get_consistent_description(description),
"v4": ip,
"hostname": hostnames
}
yield formatted_peering
def _peerings_for_nren(asn, bgplu_index, vpnrr_index):
return {
"asn": asn,
"AP": list(_bgplu_peerings(asn, bgplu_index)),
"VRR": list(_vpnrr_peerings(asn, vpnrr_index))
}
r = common.get_current_redis()
cache_key = 'classifier-cache:msr:mdvpn'
response = _ignore_cache_or_retrieve(request, cache_key, r)
if not response:
bgplu = json.loads(
r.get('juniper-peerings:group:BGPLU').decode('utf-8'))
vpnrr = json.loads(
r.get('juniper-peerings:group:VPN-RR').decode('utf-8'))
bgplu_index = _make_group_index(bgplu, 'remote-asn')
vpnrr_index = _make_group_index(vpnrr, 'remote-asn')
config = current_app.config['INVENTORY_PROVIDER_CONFIG']
nren_asn_map = config['nren-asn-map']
nren_details = [
_peerings_for_nren(pair['asn'],
bgplu_index,
vpnrr_index)
for pair in nren_asn_map]
response = json.dumps(nren_details)
return Response(response, mimetype='application/json')
@routes.route('/vpn-proxy', methods=['GET'])
@common.require_accepts_json
def vpn_proxy():
"""
Handler for `/vpn-proxy`
This method returns a list of all L3VPN related VPN proxy peerings.
The response will be formatted according to the following schema:
.. asjson::
inventory_provider.routes.msr.VPN_PROXY_LIST_SCHEMA
:return:
"""
def _is_relevant(item):
"""
Determine if a given peering in the VPN-PROXY logical system is
relevant to this endpoint (whether it's related to L3VPN)
:param item: peering dict
:return: True if the peering is L3VPN relevant, False otherwise
"""
desc = item.get("description")
if desc is None:
return False
return "L3VPN" in desc
def _look_up_city_from_hostname(hostname):
"""
Get the city name for a peering from a partial hostname match.
This uses a hardcoded lookup table.
:param hostname: hostname for the peering
:return: city name if found, "Unknown" otherwise
"""
for snippet in DOMAIN_TO_POP_MAPPING:
if snippet in hostname:
return DOMAIN_TO_POP_MAPPING[snippet]
return "Unknown"
def _extract_nren_from_description(desc, group):
"""
Retrieve the relevant NREN from the peering description and group.
This approach is, by its nature, very fragile to any changes to
descriptions, and should be revisited when that happens.
:param desc: description of a VPN-Proxy peering
:param group: group of the same VPN-Proxy peering
:return: name of the NREN
"""
if group == "PRACE":
# common trait: the NREN is the first word in the description
return desc.split(' ')[0]
else:
# only other group is XiFi, and only CESNet is relevant
return 'CESNet'
def _format_peerings(vpnproxy):
"""
Generator that iterates through a list of peering dicts, yielding
appropriately reformatted peerings if they are relevant to L3VPN.
:param vpnproxy: list of peering dicts taken from current redis
:return: generator of reformated peerings
"""
for peering in vpnproxy:
if _is_relevant(peering):
desc = peering["description"]
group = peering["group"]
hostname = peering["hostname"]
formatted_peering = {
"pop": _look_up_city_from_hostname(hostname),
"nren": _extract_nren_from_description(desc, group),
"group": group,
"v4": peering.get("address")
}
yield formatted_peering
r = common.get_current_redis()
cache_key = 'classifier-cache:msr:vpn-proxy'
response = _ignore_cache_or_retrieve(request, cache_key, r)
if not response:
vpnproxy = json.loads(
r.get('juniper-peerings:logical-system:VPN-PROXY').decode('utf-8'))
peerings = list(_format_peerings(vpnproxy))
response = json.dumps(peerings)
return Response(response, mimetype='application/json')
def _asn_peers(asn, group, instance):
"""
Handler for `/asn-peers`
This method returns a list of all peers filtered by `group` and `instance`,
which can be passed either as URL query parameters or as entries in a
POST request with a JSON body that matches this schema:
.. code-block:: json
{
"group": "group to filter by",
"instance": "instance to filter by"
}
Results are returned where all filters given are true, and exact string
matches.
An optional URL parameter can be used to also filter by a specific ASN.
The response will be formatted according to the following schema:
.. asjson::
inventory_provider.routes.msr.ASN_PEER_LIST_SCHEMA
:param asn: specific ASN to get peers for
:param group: group to filter by
:param instance: instance to filter by
:return: as above
"""
r = common.get_current_redis()
def _get_filtered_peers_for_asn(asn, nren, group, instance):
peers = json.loads(r.get(f'juniper-peerings:peer-asn:{asn}'))
def _attribute_filter(peer, name, value):
if value is None:
return True # no filter parameter given in request
if name not in peer:
return False # no value exists, cannot meet condition
return peer[name] == value
for peer in peers:
if _attribute_filter(peer, "group", group) and \
_attribute_filter(peer, "instance", instance):
peer['nren'] = nren
yield peer
def _get_filtered_peers(asn_nren_map, group, instance):
for asn, nren in asn_nren_map.items():
asn_peers = _get_filtered_peers_for_asn(asn, nren, group, instance)
for peer in asn_peers:
yield peer
cache_key = f'classifier-cache:msr:asn-peers:{asn}:{group}:{instance}'
response = _ignore_cache_or_retrieve(request, cache_key, r)
if not response:
config = current_app.config['INVENTORY_PROVIDER_CONFIG']
# set up quick lookup based on ASN
asn_nren_map = {
item['asn']: item['nren'] for item in config['nren-asn-map']
}
if asn is not None:
nren = asn_nren_map.get(asn, None)
peers = list(
_get_filtered_peers_for_asn(asn, nren, group, instance)
)
else:
peers = list(
_get_filtered_peers(asn_nren_map, group, instance)
)
response = json.dumps(peers)
r.set(cache_key, response.encode('utf-8'))
return Response(response, mimetype='application/json')
@routes.route('/asn-peers', methods=['GET'], defaults={'asn': None})
@routes.route('/asn-peers/<int:asn>', methods=['GET'])
@common.require_accepts_json
def asn_peers_get(asn):
"""
cf. doc for _asn_peers
"""
return _asn_peers(
asn=asn,
group=request.args.get('group'),
instance=request.args.get('instance'))
@routes.route('/asn-peers', methods=['POST'], defaults={'asn': None})
@routes.route('/asn-peers/<int:asn>', methods=['POST'])
@common.require_accepts_json
def asn_peers_post(asn):
"""
cf. doc for _asn_peers
"""
# TODO: some validation on this input
# TODO: verify content-type is application/json
params = json.loads(request.json)
return _asn_peers(
asn=asn,
group=params.get('group', None),
instance=params.get('instance', None))
def _dedupe(iter_of_objs):
"""
remove duplicates from the input iterable
the elements of the input iterable must be json-serializable
:param iter_of_objs: iterable
:return: a new iterable containing unique elements from the original
"""
iter_of_json = (json.dumps(_o, sort_keys=True) for _o in iter_of_objs)
return (json.loads(_s) for _s in set(iter_of_json))
def _load_ip_services():
"""
yields items that
:return:
"""
ims_interface_services = {}
equip_port_re = re.compile(r'^ims:interface_services:([^:]+):(.+)')
for doc in common.load_json_docs(
config_params=current_app.config['INVENTORY_PROVIDER_CONFIG'],
key_pattern='ims:interface_services:*',
num_threads=40):
m = equip_port_re.match(doc['key'])
assert m # sanity
equipment = m.group(1)
port = m.group(2)
assert all([_s['equipment'] == equipment
for _s in doc['value']]) # sanity
assert all([_s['port'] == port
for _s in doc['value']]) # sanity
hostname = ims_equipment_to_hostname(equipment)
ims_interface_services[f'{hostname}:{port.lower()}'] = doc['value']
netconf_interfaces = {}
host_if_extraction_re = re.compile(r'^netconf-interfaces:(.+?):')
for doc in common.load_json_docs(
config_params=current_app.config['INVENTORY_PROVIDER_CONFIG'],
key_pattern='netconf-interfaces:*',
num_threads=20):
matches = host_if_extraction_re.match(doc['key'])
if matches:
hostname = matches[1]
doc['value']['hostname'] = hostname
port = doc['value']['name']
netconf_interfaces[f'{hostname}:{port}'] = doc['value']
def _service_info(_s):
return {
'name': _s['name'],
'customer': _s['customer'],
'type': _s['service_type'],
'status': _s['status'],
'pop': {
'name': _s['pop_name'],
'abbrev': _s['pop_abbreviation']
}
}
# pre-compute remote ipaddress objects, and group to avoid duplicates
r = common.get_current_redis()
all_peerings = {}
for _remote in json.loads(r.get('juniper-peerings:all').decode('utf-8')):
remote_info = all_peerings.setdefault(_remote['address'], {
'address': ipaddress.ip_address(_remote['address']),
'info': []
})
remote_info['info'].append(_remote)
def _remotes_in_network(ifc_address):
network = ipaddress.ip_interface(ifc_address).network
for _r in all_peerings.values():
if _r['address'] in network:
yield from _r['info']
for key, ifc in netconf_interfaces.items():
_services = map(
_service_info,
ims_interface_services.get(key, []))
_services = list(_dedupe(_services))
for address in ifc['ipv4'] + ifc['ipv6']:
yield {
'hostname': ifc['hostname'],
'port': ifc['name'],
'address': address,
'services': _services,
'peerings': list(_remotes_in_network(address))
}
@routes.route('/ip-services', methods=['GET'])
@common.require_accepts_json
def ip_services():
"""
This method will return a list of all interface addresses
with any operational services defined for that interface.
The response will be formatted according to the following schema:
.. asjson::
inventory_provider.routes.msr.IP_SERVICES_LIST_SCHEMA
:return: a json list, formatted as above
"""
cache_key = 'classifier-cache:msr:ip-services'
r = common.get_current_redis()
result = _ignore_cache_or_retrieve(request, cache_key, r)
if not result:
result = list(_load_ip_services())
result = json.dumps(result)
r.set(cache_key, result.encode('utf-8'))
return Response(result.encode('utf-8'), mimetype='application/json')