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

Merge pull request #626 from GeekWeekSteph/abuseipdb2 

Fixed object reference issue for the AbuseIPDB expansion module
pull/630/head
Alexandre Dulaunoy 2023-07-11 06:35:05 +02:00 committed by GitHub
parent ea0c6f9ac2 513d292994
commit 8401470359
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 45 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

70
misp_modules/modules/expansion/abuseipdb.py
View File

@ -1,11 +1,11 @@
import requests import requests
import json import json
from pymisp import MISPObject from pymisp import MISPObject, MISPAttribute, MISPEvent
from . import check_input_attribute, checking_error, standard_error_message from . import check_input_attribute, checking_error, standard_error_message
import dns.resolver import dns.resolver
misperrors = {'error': 'Error'} misperrors = {'error': 'Error'}
mispattributes = {'input': ['hostname', 'domain', 'domain|ip'], 'output': ['boolean', 'counter'], 'format': 'misp_standard'} mispattributes = {'input': ['hostname', 'domain', 'domain|ip'], 'format': 'misp_standard'}
moduleinfo = {'version': '0.1', 'author': 'Stephanie S', moduleinfo = {'version': '0.1', 'author': 'Stephanie S',
'description': 'AbuseIPDB MISP expansion module', 'description': 'AbuseIPDB MISP expansion module',
'module-type': ['expansion', 'hover']} 'module-type': ['expansion', 'hover']}
@ -57,33 +57,53 @@ def handler(q=False):
r = {"results": []} r = {"results": []}
response = requests.request(method='GET', url=api_endpoint, headers=headers, params=querystring) response = requests.request(method='GET', url=api_endpoint, headers=headers, params=querystring)
response_json = json.loads(response.text)
is_whitelisted = response_json['data']['isWhitelisted'] if (response.status_code == 200):
is_tor = response_json['data']['isTor'] response_json = json.loads(response.text)
is_public = response_json['data']['isPublic'] is_whitelisted = response_json['data']['isWhitelisted']
abuse_confidence_score = response_json['data']['abuseConfidenceScore'] is_tor = response_json['data']['isTor']
is_public = response_json['data']['isPublic']
abuse_confidence_score = response_json['data']['abuseConfidenceScore']
if (is_whitelisted == False): if (is_whitelisted == False):
is_whitelisted = 0 is_whitelisted = 0
if (is_tor == False): if (is_tor == False):
is_tor = 0 is_tor = 0
if (is_public == False): if (is_public == False):
is_public = 0 is_public = 0
if (abuse_confidence_score == None): if (abuse_confidence_score == None):
abuse_confidence_score = 0 abuse_confidence_score = 0
if (response_json.get("errors")):
return {'error': 'AbuseIPDB error, check logs'}
else:
event = MISPEvent()
obj = MISPObject('abuseipdb')
attribute = MISPAttribute()
event.add_attribute(**request['attribute'])
if is_whitelisted is not None:
obj.add_attribute('is-whitelisted', **{'type': 'boolean', 'value': is_whitelisted})
obj.add_attribute('is-tor', **{'type': 'boolean', 'value': is_tor})
obj.add_attribute('is-public', **{'type': 'boolean', 'value': is_public})
obj.add_attribute('abuse-confidence-score', **{'type': 'counter', 'value': abuse_confidence_score})
obj.add_reference(request['attribute']['uuid'], "describes")
event.add_object(obj)
# Avoid serialization issue
event = json.loads(event.to_json())
r['results'] = {'Object': event['Object'], 'Attribute': event['Attribute']}
return r
if (response_json.get("errors")):
return {'error': 'AbuseIPDB error, check logs'}
else: else:
obj = MISPObject('abuseipdb') try:
obj.add_attribute('is-whitelisted', **{'type': 'boolean', 'value': is_whitelisted}) response_json = json.loads(response.text)
obj.add_attribute('is-tor', **{'type': 'boolean', 'value': is_tor}) if (response_json['errors']):
obj.add_attribute('is-public', **{'type': 'boolean', 'value': is_public}) return {"error": "API not reachable, status code: " + str(response.status_code) + " " + str(response_json['errors'][0]['detail'])}
obj.add_attribute('abuse-confidence-score', **{'type': 'counter', 'value': abuse_confidence_score}) except:
pass
r['results'] = {'Object': [json.loads(obj.to_json())]} return {"error": "API not reachable, status code: " + str(response.status_code)}
return r
def introspection(): def introspection():
return mispattributes return mispattributes