misp-modules/misp_modules/modules/expansion/bgpranking.py

87 lines
3.3 KiB
Python
Raw Normal View History

2016-05-08 13:07:09 +02:00
# -*- coding: utf-8 -*-
import json
from . import check_input_attribute, standard_error_message
from datetime import date, datetime, timedelta
from pybgpranking import BGPRanking
from pymisp import MISPAttribute, MISPEvent, MISPObject
2016-05-08 13:07:09 +02:00
misperrors = {'error': 'Error'}
mispattributes = {'input': ['AS'], 'format': 'misp_standard'}
2016-05-08 13:07:09 +02:00
moduleinfo = {'version': '0.1', 'author': 'Raphaël Vinot',
'description': 'Query BGP Ranking to get the ranking of an Autonomous System number.',
2016-05-08 13:07:09 +02:00
'module-type': ['expansion', 'hover']}
def handler(q=False):
if q is False:
return False
request = json.loads(q)
if not request.get('attribute') or not check_input_attribute(request['attribute']):
return {'error': f'{standard_error_message}, which should contain at least a type, a value and an uuid.'}
toquery = request['attribute']
if toquery['type'] not in mispattributes['input']:
return {'error': 'Unsupported attribute type.'}
2016-05-08 13:07:09 +02:00
bgpranking = BGPRanking()
value_toquery = int(toquery['value'][2:]) if toquery['value'].startswith('AS') else int(toquery['value'])
values = bgpranking.query(value_toquery, date=(date.today() - timedelta(1)).isoformat())
2016-05-08 13:07:09 +02:00
if not values['response'] or not values['response']['asn_description']:
misperrors['error'] = 'There is no result about this ASN in BGP Ranking'
2016-05-08 13:07:09 +02:00
return misperrors
event = MISPEvent()
attribute = MISPAttribute()
attribute.from_dict(**toquery)
event.add_attribute(**attribute)
asn_object = MISPObject('asn')
asn_object.add_attribute(**{
'type': 'AS',
'object_relation': 'asn',
'value': values['meta']['asn']
})
description, country = values['response']['asn_description'].split(', ')
for relation, value in zip(('description', 'country'), (description, country)):
asn_object.add_attribute(**{
'type': 'text',
'object_relation': relation,
'value': value
})
mapping = {
'address_family': {'type': 'text', 'object_relation': 'address-family'},
'date': {'type': 'datetime', 'object_relation': 'date'},
'position': {'type': 'float', 'object_relation': 'position'},
'rank': {'type': 'float', 'object_relation': 'ranking'}
}
bgp_object = MISPObject('bgp-ranking')
for feature in ('rank', 'position'):
bgp_attribute = {'value': values['response']['ranking'][feature]}
bgp_attribute.update(mapping[feature])
bgp_object.add_attribute(**bgp_attribute)
date_attribute = {'value': datetime.strptime(values['meta']['date'], '%Y-%m-%d')}
date_attribute.update(mapping['date'])
bgp_object.add_attribute(**date_attribute)
address_attribute = {'value': values['meta']['address_family']}
address_attribute.update(mapping['address_family'])
bgp_object.add_attribute(**address_attribute)
asn_object.add_reference(attribute.uuid, 'describes')
asn_object.add_reference(bgp_object.uuid, 'ranked-with')
event.add_object(asn_object)
event.add_object(bgp_object)
event = json.loads(event.to_json())
results = {key: event[key] for key in ('Attribute', 'Object')}
return {'results': results}
2016-05-08 13:07:09 +02:00
def introspection():
return mispattributes
def version():
return moduleinfo