new: Modules for greynoise, haveibeenpwned and macvendors

Source: https://github.com/src7/misp-modules
pull/290/head
Raphaël Vinot 4 years ago
parent 3e34f38cac
commit b89d068c04
  1. 40
      misp_modules/modules/expansion/greynoise.py
  2. 40
      misp_modules/modules/expansion/hibp.py
  3. 42
      misp_modules/modules/expansion/macvendors.py
  4. 57
      tests/test_expansions.py

@ -0,0 +1,40 @@
import requests
import json
misperrors = {'error': 'Error'}
mispattributes = {'input': ['ip-dst', 'ip-src'], 'output': ['text']}
moduleinfo = {'version': '0.1', 'author': 'Aurélien Schwab <aurelien.schwab+dev@gmail.com>', 'description': 'Module to access GreyNoise.io API.', 'module-type': ['hover']}
moduleconfig = ['user-agent']#TODO take this into account in the code
greynoise_api_url = 'http://api.greynoise.io:8888/v1/query/ip'
default_user_agent = 'MISP-Module'
def handler(q=False):
if q is False:
return False
request = json.loads(q)
for input_type in mispattributes['input']:
if input_type in request:
ip = request[input_type]
break
else:
misperrors['error'] = "Unsupported attributes type"
return misperrors
data = {'ip':ip}
r = requests.post(greynoise_api_url, data=data, headers={'user-agent': default_user_agent})#Real request
if r.status_code == 200:#OK (record found)
response = json.loads(r.text)
if response:
return {'results': [{'types': mispattributes['output'], 'values': response}]}
elif r.status_code == 404:#Not found (not an error)
return {'results': [{'types': mispattributes['output'], 'values': 'No data'}]}
else:#Real error
misperrors['error'] = 'GreyNoise API not accessible (HTTP ' + str(r.status_code) + ')'
return misperrors['error']
def introspection():
return mispattributes
def version():
moduleinfo['config'] = moduleconfig
return moduleinf

@ -0,0 +1,40 @@
import requests
import json
misperrors = {'error': 'Error'}
mispattributes = {'input': ['email-dst', 'email-src'], 'output': ['text']}#All mails as input
moduleinfo = {'version': '0.1', 'author': 'Aurélien Schwab', 'description': 'Module to access haveibeenpwned.com API.', 'module-type': ['hover']}
moduleconfig = ['user-agent']#TODO take this into account in the code
haveibeenpwned_api_url = 'https://api.haveibeenpwned.com/api/v2/breachedaccount/'
default_user_agent = 'MISP-Module'#User agent (must be set, requiered by API))
def handler(q=False):
if q is False:
return False
request = json.loads(q)
for input_type in mispattributes['input']:
if input_type in request:
email = request[input_type]
break
else:
misperrors['error'] = "Unsupported attributes type"
return misperrors
r = requests.get(haveibeenpwned_api_url + email, headers={'user-agent': default_user_agent})#Real request
if r.status_code == 200:##OK (record found)
breaches = json.loads(r.text)
if breaches:
return {'results': [{'types': mispattributes['output'], 'values': breaches}]}
elif r.status_code == 404:#Not found (not an error)
return {'results': [{'types': mispattributes['output'], 'values': 'OK (Not Found)'}]}
else:#Real error
misperrors['error'] = 'haveibeenpwned.com API not accessible (HTTP ' + str(r.status_code) + ')'
return misperrors['error']
def introspection():
return mispattributes
def version():
moduleinfo['config'] = moduleconfig
return moduleinf

@ -0,0 +1,42 @@
import requests
import json
misperrors = {'error': 'Error'}
mispattributes = {'input': ['mac-address'], 'output': ['text']}
moduleinfo = {'version': '0.1', 'author': 'Aurélien Schwab', 'description': 'Module to access Macvendors API.', 'module-type': ['hover']}
moduleconfig = ['user-agent'] # TODO take this into account in the code
macvendors_api_url = 'https://api.macvendors.com/'
default_user_agent = 'MISP-Module'
def handler(q=False):
if q is False:
return False
request = json.loads(q)
for input_type in mispattributes['input']:
if input_type in request:
mac = request[input_type]
break
else:
misperrors['error'] = "Unsupported attributes type"
return misperrors
r = requests.get(macvendors_api_url + mac, headers={'user-agent': default_user_agent}) # Real request
if r.status_code == 200: # OK (record found)
response = r.text
if response:
return {'results': [{'types': mispattributes['output'], 'values': response}]}
elif r.status_code == 404: # Not found (not an error)
return {'results': [{'types': mispattributes['output'], 'values': 'Not found'}]}
else: # Real error
misperrors['error'] = 'MacVendors API not accessible (HTTP ' + str(r.status_code) + ')'
return misperrors['error']
def introspection():
return mispattributes
def version():
moduleinfo['config'] = moduleconfig
return moduleinfo

@ -0,0 +1,57 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import unittest
import requests
from urllib.parse import urljoin
class TestExpansions(unittest.TestCase):
def setUp(self):
self.maxDiff = None
self.headers = {'Content-Type': 'application/json'}
self.url = "http://127.0.0.1:6666/"
def misp_modules_post(self, query):
return requests.post(urljoin(self.url, "query"), json=query)
def get_values(self, response):
return response.json()['results'][0]['values']
def test_cve(self):
query = {"module": "cve", "vulnerability": "CVE-2010-3333"}
response = self.misp_modules_post(query)
self.assertTrue(self.get_values(response).startswith("Stack-based buffer overflow in Microsoft Office XP SP3, Office 2003 SP3"))
def test_dns(self):
query = {"module": "dns", "hostname": "www.circl.lu", "config": {"nameserver": "8.8.8.8"}}
response = self.misp_modules_post(query)
self.assertEqual(self.get_values(response), ['149.13.33.14'])
def test_macvendors(self):
query = {"module": "macvendors", "mac-address": "FC-A1-3E-2A-1C-33"}
response = self.misp_modules_post(query)
self.assertEqual(self.get_values(response), 'Samsung Electronics Co.,Ltd')
def test_haveibeenpwned(self):
query = {"module": "hibp", "email-src": "info@circl.lu"}
response = self.misp_modules_post(query)
self.assertEqual(self.get_values(response), 'OK (Not Found)')
def test_greynoise(self):
query = {"module": "greynoise", "ip-dst": "1.1.1.1"}
response = self.misp_modules_post(query)
self.assertEqual(self.get_values(response)['status'], 'ok')
def test_ipasn(self):
query = {"module": "ipasn", "ip-dst": "1.1.1.1"}
response = self.misp_modules_post(query)
key = list(self.get_values(response)['response'].keys())[0]
entry = self.get_values(response)['response'][key]['asn']
self.assertEqual(entry, '13335')
def test_bgpranking(self):
query = {"module": "bgpranking", "AS": "13335"}
response = self.misp_modules_post(query)
self.assertEqual(self.get_values(response)['response']['asn_description'], 'CLOUDFLARENET - Cloudflare, Inc., US')
Loading…
Cancel
Save