mirror of https://github.com/MISP/misp-modules
96 lines
4.3 KiB
Python
Executable File
96 lines
4.3 KiB
Python
Executable File
import json
|
|
import requests
|
|
from . import check_input_attribute, standard_error_message
|
|
from pymisp import MISPAttribute, MISPEvent, MISPObject
|
|
|
|
misperrors = {'error': 'Error'}
|
|
mispattributes = {'input': ['domain', 'hostname'], 'format': 'misp_standard'}
|
|
moduleinfo = {'version': '0.1', 'author': 'Christian Studer',
|
|
'description': 'On demand query API for APIVoid.',
|
|
'module-type': ['expansion', 'hover']}
|
|
moduleconfig = ['apikey']
|
|
|
|
|
|
class APIVoidParser():
|
|
def __init__(self, attribute):
|
|
self.misp_event = MISPEvent()
|
|
self.attribute = MISPAttribute()
|
|
self.attribute.from_dict(**attribute)
|
|
self.misp_event.add_attribute(**self.attribute)
|
|
self.url = 'https://endpoint.apivoid.com/{}/v1/pay-as-you-go/?key={}&'
|
|
|
|
def get_results(self):
|
|
if hasattr(self, 'result'):
|
|
return self.result
|
|
event = json.loads(self.misp_event.to_json())
|
|
results = {key: event[key] for key in ('Attribute', 'Object')}
|
|
return {'results': results}
|
|
|
|
def parse_domain(self, apikey):
|
|
feature = 'dnslookup'
|
|
if requests.get(f'{self.url.format(feature, apikey)}stats').json()['credits_remained'] < 0.13:
|
|
self.result = {'error': 'You do not have enough APIVoid credits to proceed your request.'}
|
|
return
|
|
mapping = {'A': 'resolution-of', 'MX': 'mail-server-of', 'NS': 'server-name-of'}
|
|
dnslookup = requests.get(f'{self.url.format(feature, apikey)}action=dns-any&host={self.attribute.value}').json()
|
|
for item in dnslookup['data']['records']['items']:
|
|
record_type = item['type']
|
|
try:
|
|
relationship = mapping[record_type]
|
|
except KeyError:
|
|
continue
|
|
self._handle_dns_record(item, record_type, relationship)
|
|
ssl = requests.get(f'{self.url.format("sslinfo", apikey)}host={self.attribute.value}').json()
|
|
self._parse_ssl_certificate(ssl['data']['certificate'])
|
|
|
|
def _handle_dns_record(self, item, record_type, relationship):
|
|
dns_record = MISPObject('dns-record')
|
|
dns_record.add_attribute('queried-domain', type='domain', value=item['host'])
|
|
attribute_type, feature = ('ip-dst', 'ip') if record_type == 'A' else ('domain', 'target')
|
|
dns_record.add_attribute(f'{record_type.lower()}-record', type=attribute_type, value=item[feature])
|
|
dns_record.add_reference(self.attribute.uuid, relationship)
|
|
self.misp_event.add_object(**dns_record)
|
|
|
|
def _parse_ssl_certificate(self, certificate):
|
|
x509 = MISPObject('x509')
|
|
fingerprint = 'x509-fingerprint-sha1'
|
|
x509.add_attribute(fingerprint, type=fingerprint, value=certificate['fingerprint'])
|
|
x509_mapping = {'subject': {'name': ('text', 'subject')},
|
|
'issuer': {'common_name': ('text', 'issuer')},
|
|
'signature': {'serial': ('text', 'serial-number')},
|
|
'validity': {'valid_from': ('datetime', 'validity-not-before'),
|
|
'valid_to': ('datetime', 'validity-not-after')}}
|
|
certificate = certificate['details']
|
|
for feature, subfeatures in x509_mapping.items():
|
|
for subfeature, mapping in subfeatures.items():
|
|
attribute_type, relation = mapping
|
|
x509.add_attribute(relation, type=attribute_type, value=certificate[feature][subfeature])
|
|
x509.add_reference(self.attribute.uuid, 'seen-by')
|
|
self.misp_event.add_object(**x509)
|
|
|
|
|
|
def handler(q=False):
|
|
if q is False:
|
|
return False
|
|
request = json.loads(q)
|
|
if not request.get('config', {}).get('apikey'):
|
|
return {'error': 'An API key for APIVoid is required.'}
|
|
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.'}
|
|
attribute = request['attribute']
|
|
if attribute['type'] not in mispattributes['input']:
|
|
return {'error': 'Unsupported attribute type.'}
|
|
apikey = request['config']['apikey']
|
|
apivoid_parser = APIVoidParser(attribute)
|
|
apivoid_parser.parse_domain(apikey)
|
|
return apivoid_parser.get_results()
|
|
|
|
|
|
def introspection():
|
|
return mispattributes
|
|
|
|
|
|
def version():
|
|
moduleinfo['config'] = moduleconfig
|
|
return moduleinfo
|