new: Modules for greynoise, haveibeenpwned and macvendors

Source: https://github.com/src7/misp-modules

Author: Rafiot

misp_modules/modules/expansion/greynoise.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 <[email protected]>', '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

misp_modules/modules/expansion/hibp.py

@@ -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

misp_modules/modules/expansion/macvendors.py

@@ -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

tests/test_expansions.py

@@ -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": "[email protected]"}
+        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')