MISP
/
misp-modules
mirror of https://github.com/MISP/misp-modules
2
0
Fork 0
Code Issues Releases Wiki Activity

updated

pull/618/head
Usama015 2023-06-20 13:50:31 +05:00
parent aeb65b1328
commit 4021b97254
1 changed files with 19 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

65
misp_modules/modules/expansion/ipgeolocation.py
View File

@ -32,64 +32,37 @@ def handler(q=False):
attribute = request['attribute'] attribute = request['attribute']
ip = request['attribute']['value'] ip = request['attribute']['value']
apiKey = request['config']['apiKey'] apiKey = request['config']['apiKey']
# Correct query = requests.get(f"https://api.ipgeolocation.io/ipgeo?apiKey={apiKey}&ip={ip}")
response = handle_ip(apiKey, ip, attribute) if query.status_code != 200:
return {'error' : f'Completed Response - {response}'} return {'error': f'Error while querying ipGeolocation.io - {query.status_code}: {query.reason}'}
def handle_ip(apiKey, ip, attribute):
try:
results = query_ipgeolocation(apiKey, ip)
except Exception:
return {'error' : 'Error during querying IPGeolocation API.'}
# Check if the IP address is not reserved for special use # Check if the IP address is not reserved for special use
if results.get('message'): if query.get('message'):
if 'bogon' in results['message']: if 'bogon' in query['message']:
return {'error': 'The IP address(bogon IP) is reserved for special use'} return {'error': 'The IP address(bogon IP) is reserved for special use'}
else: else:
return {'error': 'Error Occurred during IP data Extraction from Message'} return {'error': 'Error Occurred during IP data Extraction from Message'}
try: misp_event = MISPEvent()
misp_event = MISPEvent() input_attribute = MISPAttribute()
input_attribute = MISPAttribute() input_attribute.from_dict(**attribute)
input_attribute.from_dict(**attribute) misp_event.add_attribute(**input_attribute)
misp_event.add_attribute(**input_attribute)
except Exception:
return {'error': f'Error on line 58 - {traceback.print_exc()}'}
ipObject = MISPObject('ip-api-address') ipObject = MISPObject('ip-api-address')
# Correct # Correct
try: mapping = get_mapping()
mapping = get_mapping() for field, relation in mapping.items():
except Exception: ipObject.add_attribute(relation, query[field])
return {'error': f'Error on line 66 - {traceback.print_exc()}'} misp_event.add_object(ipObject)
try:
for field, relation in mapping.items():
ipObject.add_attribute(relation, results[field])
except Exception:
return {'error': f'Error on line 71 - {traceback.print_exc()}'}
try:
misp_event.add_object(ipObject)
except Exception:
return {'error': f'Error on line 75 - {traceback.print_exc()}'}
# Return the results in MISP format # Return the results in MISP format
try: event = json.loads(misp_event.to_json())
event = json.loads(misp_event.to_json()) return {
return { 'results': {key: event[key] for key in ('Attribute', 'Object')}
'results': {key: event[key] for key in ('Attribute', 'Object')} }
} # return {'error' : 'Completed Response'}
except Exception:
return {'error': f'Error on line 83 - {traceback.print_exc()}'}
def query_ipgeolocation(apiKey, ip):
query = requests.get(
f"https://api.ipgeolocation.io/ipgeo?apiKey={apiKey}&ip={ip}"
)
if query.status_code != 200:
return {'error': f'Error while querying ipGeolocation.io - {query.status_code}: {query.reason}'}
return query.json()
def get_mapping(): def get_mapping():
return { return {