mirror of https://github.com/MISP/misp-modules
293 lines
14 KiB
Python
293 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
|
|
'''
|
|
Import VulnDB
|
|
https://vulndb.cyberriskanalytics.com/
|
|
https://www.riskbasedsecurity.com/
|
|
'''
|
|
|
|
import oauth2 as oauth
|
|
import json
|
|
|
|
import logging
|
|
import sys
|
|
|
|
|
|
log = logging.getLogger('vulndb')
|
|
log.setLevel(logging.DEBUG)
|
|
ch = logging.StreamHandler(sys.stdout)
|
|
ch.setLevel(logging.DEBUG)
|
|
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
|
|
ch.setFormatter(formatter)
|
|
log.addHandler(ch)
|
|
|
|
|
|
misperrors = {'error': 'Error'}
|
|
mispattributes = {
|
|
'input': ['vulnerability'],
|
|
'output': ['text', 'link', 'cpe']}
|
|
moduleinfo = {
|
|
'version': '0.1',
|
|
'author': 'Koen Van Impe',
|
|
'description': 'Module to query VulnDB (RiskBasedSecurity.com).',
|
|
'module-type': ['expansion', 'hover'],
|
|
'name': 'VulnDB Lookup',
|
|
'logo': 'vulndb.png',
|
|
'requirements': ['An access to the VulnDB API (apikey, apisecret)'],
|
|
'features': 'This module takes a vulnerability attribute as input and queries VulnDB in order to get some additional data about it.\n\nThe API gives the result of the query which can be displayed in the screen, and/or mapped into MISP attributes to add in the event.',
|
|
'references': ['https://vulndb.cyberriskanalytics.com/'],
|
|
'input': 'A vulnerability attribute.',
|
|
'output': 'Additional data enriching the CVE input, fetched from VulnDB.',
|
|
}
|
|
|
|
moduleconfig = ['apikey', 'apisecret', 'discard_dates', 'discard_external_references', 'discard_cvss', 'discard_productinformation', 'discard_classification', 'discard_cpe']
|
|
|
|
|
|
def handler(q=False):
|
|
# Base URL for VulnDB
|
|
VULNDB_URL = "https://vulndb.cyberriskanalytics.com"
|
|
|
|
if q is False:
|
|
return False
|
|
request = json.loads(q)
|
|
|
|
# Only continue if we have a vulnerability attribute
|
|
if not request.get('vulnerability'):
|
|
misperrors['error'] = 'Vulnerability ID missing for VulnDB.'
|
|
return misperrors
|
|
vulnerability = request.get('vulnerability')
|
|
|
|
if request["config"].get("apikey") is None or request["config"].get("apisecret") is None:
|
|
misperrors["error"] = "Missing API key or secret value for VulnDB."
|
|
return misperrors
|
|
apikey = request["config"].get("apikey")
|
|
apisecret = request["config"].get("apisecret")
|
|
|
|
# This has to be done the 'inverse' way, MISP-server settings are set to False by default
|
|
add_cvss = True
|
|
add_products = True
|
|
add_classifications = True
|
|
add_cpe = True
|
|
add_dates = True
|
|
add_ext_references = True
|
|
|
|
if request["config"].get("discard_dates") is not None and request["config"].get("discard_dates").lower() == "true":
|
|
add_dates = False
|
|
if request["config"].get("discard_external_references") is not None and request["config"].get("discard_external_references").lower() == "true":
|
|
add_ext_references = False
|
|
if request["config"].get("discard_cvss") is not None and request["config"].get("discard_cvss").lower() == "true":
|
|
add_cvss = False
|
|
if request["config"].get("discard_productinformation") is not None and request["config"].get("discard_productinformation").lower() == "true":
|
|
add_products = False
|
|
if request["config"].get("discard_classification") is not None and request["config"].get("discard_classification").lower() == "true":
|
|
add_classifications = False
|
|
if request["config"].get("discard_cpe") is not None and request["config"].get("discard_cpe").lower() == "true":
|
|
add_cpe = False
|
|
|
|
cpu_vulndb = ""
|
|
if add_cpe:
|
|
cpu_vulndb = "?show_cpe=true"
|
|
|
|
find_by_cve_url = "%s/api/v1/vulnerabilities/%s/find_by_cve_id%s" % (VULNDB_URL, vulnerability, cpu_vulndb)
|
|
log.debug(find_by_cve_url)
|
|
|
|
try:
|
|
|
|
consumer = oauth.Consumer(key=apikey, secret=apisecret)
|
|
client = oauth.Client(consumer)
|
|
resp, content = client.request(find_by_cve_url, "GET")
|
|
content_json = json.loads(content.decode())
|
|
|
|
if content_json:
|
|
if 'error' in content_json:
|
|
misperrors["error"] = "No CVE information found."
|
|
return misperrors
|
|
else:
|
|
output = {'results': list()}
|
|
values_text = list()
|
|
values_links = list()
|
|
values_cpe = list()
|
|
|
|
results = content_json["results"][0]
|
|
|
|
# Include the VulnDB title and ID
|
|
values_text.append(results["title"])
|
|
vulndb_id_link = "%s/vulnerabilities/%s" % (VULNDB_URL, results["vulndb_id"])
|
|
values_links.append(vulndb_id_link)
|
|
|
|
# Descriptive part of the VulnDB item
|
|
description = results.get('description', '') or ''
|
|
keywords = results.get('keywords', '') or ''
|
|
solution = results.get('solution', '') or ''
|
|
manual_notes = results.get('manual_notes', '') or ''
|
|
t_description = results.get('t_description', '') or ''
|
|
if description:
|
|
values_text.append(description)
|
|
if t_description:
|
|
values_text.append(t_description)
|
|
if manual_notes:
|
|
values_text.append("Notes: " + manual_notes)
|
|
if keywords:
|
|
values_text.append("Keywords: " + keywords)
|
|
if solution:
|
|
values_text.append("Solution: " + solution)
|
|
|
|
# VulnDB items contain a number of dates, do we include them?
|
|
if add_dates:
|
|
log.debug("Include dates")
|
|
solution_date = results.get('solution_date', '') or ''
|
|
if solution_date:
|
|
values_text.append("Solution date: " + solution_date)
|
|
disclosure_date = results.get('disclosure_date', '') or ''
|
|
if disclosure_date:
|
|
values_text.append("Disclosure date: " + disclosure_date)
|
|
discovery_date = results.get('discovery_date', '') or ''
|
|
if discovery_date:
|
|
values_text.append("Discovery date: " + discovery_date)
|
|
exploit_publish_date = results.get('exploit_publish_date', '') or ''
|
|
if exploit_publish_date:
|
|
values_text.append("Exploit published date: " + exploit_publish_date)
|
|
vendor_informed_date = results.get('vendor_informed_date', '') or ''
|
|
if vendor_informed_date:
|
|
values_text.append("Vendor informed date: " + vendor_informed_date)
|
|
vendor_ack_date = results.get('vendor_ack_date', '') or ''
|
|
if vendor_ack_date:
|
|
values_text.append("Vendor acknowledgement date: " + vendor_ack_date)
|
|
third_party_solution_date = results.get('third_party_solution_date', '') or ''
|
|
if third_party_solution_date:
|
|
values_text.append("Third party solution date: " + third_party_solution_date)
|
|
|
|
# External references
|
|
if add_ext_references:
|
|
ext_references = results.get('ext_references')
|
|
if ext_references:
|
|
log.debug("Include external references")
|
|
for reference in ext_references:
|
|
reference_type = reference["type"]
|
|
if reference_type == "Other Advisory URL":
|
|
values_links.append(reference["value"])
|
|
elif reference_type == "News Article":
|
|
values_links.append(reference["value"])
|
|
elif reference_type == "Generic Informational URL":
|
|
values_links.append(reference["value"])
|
|
elif reference_type == "Vendor Specific Advisory URL":
|
|
values_links.append(reference["value"])
|
|
elif reference_type == "Vendor URL":
|
|
values_links.append(reference["value"])
|
|
elif reference_type == "Mail List Post":
|
|
values_links.append(reference["value"])
|
|
elif reference_type == "Metasploit URL":
|
|
values_links.append(reference["value"])
|
|
elif reference_type == "Packet Storm":
|
|
values_links.append(reference["value"])
|
|
elif reference_type == "Generic Exploit URL":
|
|
values_links.append(reference["value"])
|
|
elif reference_type == "CERT VU":
|
|
reference_link = "http://www.kb.cert.org/vuls/id/%s" % reference["value"]
|
|
values_links.append(reference_link)
|
|
elif reference_type == "CVE ID":
|
|
reference_link = "https://nvd.nist.gov/vuln/detail/%s" % reference["value"]
|
|
values_links.append(reference_link)
|
|
elif reference_type == "Microsoft Knowledge Base Article":
|
|
reference_link = "https://support.microsoft.com/en-us/help/%s" % reference["value"]
|
|
values_links.append(reference_link)
|
|
elif reference_type == "Exploit Database":
|
|
reference_link = "https://www.exploit-db.com/exploits/%s" % reference["value"]
|
|
values_links.append(reference_link)
|
|
elif reference_type == "Generic Informational URL":
|
|
values_links.append(reference["value"])
|
|
elif reference_type == "Generic Informational URL":
|
|
values_links.append(reference["value"])
|
|
elif reference_type == "Generic Informational URL":
|
|
values_links.append(reference["value"])
|
|
elif reference_type == "Generic Informational URL":
|
|
values_links.append(reference["value"])
|
|
elif reference_type == "Generic Informational URL":
|
|
values_links.append(reference["value"])
|
|
|
|
# CVSS Scoring
|
|
if add_cvss:
|
|
cvss = results.get('cvss_metrics')
|
|
if cvss:
|
|
log.debug("Include CVSS")
|
|
for cvss_metric in cvss:
|
|
score = cvss_metric.get("score")
|
|
if score:
|
|
values_text.append("CVSS %s (base: %s) (source: %s)" % (score, cvss_metric.get("calculated_cvss_base_score"), cvss_metric.get("source")))
|
|
|
|
# Add products
|
|
if add_products:
|
|
products = results.get('products')
|
|
if products and len(products) > 0:
|
|
|
|
# Get the vendors
|
|
vendors = results.get('vendors')
|
|
vendors_name = ""
|
|
log.debug("Include product information")
|
|
if vendors:
|
|
for vendor in vendors:
|
|
vendor_detail = vendor.get("vendor")
|
|
if vendor_detail:
|
|
vendor_name = vendor_detail.get("name")
|
|
if vendor_name:
|
|
vendors_name += vendor_name + " "
|
|
|
|
# Walk through all vendors
|
|
for product in products:
|
|
vulnerable_product = vendors_name
|
|
name = product.get("name")
|
|
if name:
|
|
vulnerable_product += "%s " % name
|
|
versions = product.get("versions")
|
|
if versions:
|
|
vulnerable_product += "("
|
|
for version in versions:
|
|
affected = version.get("affected")
|
|
|
|
if affected and affected == "true":
|
|
vulnerable_product += " %s " % version.get("name")
|
|
if add_cpe:
|
|
version_cpe = version.get("cpe")
|
|
if version_cpe:
|
|
cpe = version_cpe[0].get("cpe")
|
|
if cpe:
|
|
values_cpe.append(cpe)
|
|
|
|
vulnerable_product += ")"
|
|
# Add vulnerable products
|
|
values_text.append(vulnerable_product)
|
|
|
|
# Add vulnerability classifications
|
|
if add_classifications:
|
|
classifications = results.get("classifications")
|
|
if classifications and len(classifications) > 0:
|
|
vulnerability_classification = ""
|
|
log.debug("Include classifications")
|
|
for classification in classifications:
|
|
longname = classification.get("longname")
|
|
description = classification.get("description")
|
|
vulnerability_classification += " \"%s\" " % longname
|
|
values_text.append(vulnerability_classification)
|
|
|
|
# Finished processing the VulnDB reply; set the result for MISP
|
|
output['results'] += [{'types': 'text', 'values': values_text}]
|
|
output['results'] += [{'types': 'link', 'values': values_links}]
|
|
if add_cpe:
|
|
output['results'] += [{'types': 'cpe', 'values': values_cpe}]
|
|
return output
|
|
else:
|
|
misperrors["error"] = "No information retrieved from VulnDB."
|
|
return misperrors
|
|
except Exception:
|
|
misperrors["error"] = "Error while fetching information from VulnDB, wrong API keys?"
|
|
return misperrors
|
|
|
|
|
|
def introspection():
|
|
return mispattributes
|
|
|
|
|
|
def version():
|
|
moduleinfo['config'] = moduleconfig
|
|
return moduleinfo
|