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

chg: [cve_advanced] Updated the module to use cvepremium & a few improvements

pull/590/head
Christian Studer 2022-10-25 22:20:30 +02:00
parent 38a6dc810e
commit 8e97bf9938
No known key found for this signature in database
GPG Key ID: 6BBED1B63A6D639F
1 changed files with 89 additions and 57 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

122
misp_modules/modules/expansion/cve_advanced.py
View File

@ -2,34 +2,70 @@ import json
import requests
from . import check_input_attribute, standard_error_message
from collections import defaultdict
from pymisp import MISPEvent, MISPObject
from pymisp import MISPAttribute, MISPEvent, MISPObject
misperrors = {'error': 'Error'}
mispattributes = {'input': ['vulnerability'], 'format': 'misp_standard'}
moduleinfo = {'version': '1', 'author': 'Christian Studer',
moduleinfo = {'version': '2', 'author': 'Christian Studer',
'description': 'An expansion module to enrich a CVE attribute with the vulnerability information.',
'module-type': ['expansion', 'hover']}
moduleconfig = ["custom_API"]
cveapi_url = 'https://cve.circl.lu/api/cve/'
cveapi_url = 'https://cvepremium.circl.lu/api/'
class VulnerabilityParser():
def __init__(self, attribute, vulnerability, api_url):
self.attribute = attribute
self.vulnerability = vulnerability
self.api_url = api_url
self.misp_event = MISPEvent()
self.misp_event.add_attribute(**attribute)
def __init__(self, attribute, api_url):
misp_attribute = MISPAttribute()
misp_attribute.from_dict(**attribute)
misp_event = MISPEvent()
misp_event.add_attribute(**misp_attribute)
self.__misp_attribute = misp_attribute
self.__misp_event = misp_event
self.__api_url = api_url
self.references = defaultdict(list)
self.capec_features = ('id', 'name', 'summary', 'prerequisites', 'solutions')
self.vulnerability_mapping = {
self.__capec_features = ('id', 'name', 'summary', 'prerequisites', 'solutions')
self.__vulnerability_mapping = {
'id': 'id', 'summary': 'summary',
'Modified': 'modified', 'cvss3': 'cvss-score',
'cvss3-vector': 'cvss-string'
}
self.__vulnerability_multiple_mapping = {
'vulnerable_configuration': 'vulnerable-configuration',
'vulnerable_configuration_cpe_2_2': 'vulnerable-configuration',
'Modified': 'modified', 'Published': 'published',
'references': 'references', 'cvss': 'cvss-score'}
self.weakness_mapping = {'name': 'name', 'description_summary': 'description',
'status': 'status', 'weaknessabs': 'weakness-abs'}
'references': 'references'
}
self.__weakness_mapping = {
'name': 'name', 'description_summary': 'description',
'status': 'status', 'weaknessabs': 'weakness-abs'
}
@property
def api_url(self) -> str:
return self.__api_url
@property
def capec_features(self) -> tuple:
return self.__capec_features
@property
def misp_attribute(self) -> MISPAttribute:
return self.__misp_attribute
@property
def misp_event(self) -> MISPEvent:
return self.__misp_event
@property
def vulnerability_mapping(self) -> dict:
return self.__vulnerability_mapping
@property
def vulnerability_multiple_mapping(self) -> dict:
return self.__vulnerability_multiple_mapping
@property
def weakness_mapping(self) -> dict:
return self.__weakness_mapping
def get_result(self):
if self.references:
@ -38,28 +74,26 @@ class VulnerabilityParser():
results = {key: event[key] for key in ('Attribute', 'Object') if (key in event and event[key])}
return {'results': results}
def parse_vulnerability_information(self):
def parse_vulnerability_information(self, vulnerability):
vulnerability_object = MISPObject('vulnerability')
for feature in ('id', 'summary', 'Modified', 'cvss'):
value = self.vulnerability.get(feature)
if value:
vulnerability_object.add_attribute(self.vulnerability_mapping[feature], value)
if 'Published' in self.vulnerability:
vulnerability_object.add_attribute('published', self.vulnerability['Published'])
for feature, relation in self.vulnerability_mapping.items():
if vulnerability.get(feature):
vulnerability_object.add_attribute(relation, vulnerability[feature])
if 'Published' in vulnerability:
vulnerability_object.add_attribute('published', vulnerability['Published'])
vulnerability_object.add_attribute('state', 'Published')
for feature in ('references', 'vulnerable_configuration', 'vulnerable_configuration_cpe_2_2'):
if feature in self.vulnerability:
relation = self.vulnerability_mapping[feature]
for value in self.vulnerability[feature]:
for feature, relation in self.vulnerability_multiple_mapping.items():
if feature in vulnerability:
for value in vulnerability[feature]:
if isinstance(value, dict):
value = value['title']
vulnerability_object.add_attribute(relation, value)
vulnerability_object.add_reference(self.attribute['uuid'], 'related-to')
vulnerability_object.add_reference(self.misp_attribute.uuid, 'related-to')
self.misp_event.add_object(vulnerability_object)
if 'cwe' in self.vulnerability and self.vulnerability['cwe'] not in ('Unknown', 'NVD-CWE-noinfo'):
self.__parse_weakness(vulnerability_object.uuid)
if 'capec' in self.vulnerability:
self.__parse_capec(vulnerability_object.uuid)
if 'cwe' in vulnerability and vulnerability['cwe'] not in ('Unknown', 'NVD-CWE-noinfo'):
self.__parse_weakness(vulnerability['cwe'], vulnerability_object.uuid)
if 'capec' in vulnerability:
self.__parse_capec(vulnerability['capec'], vulnerability_object.uuid)
def __build_references(self):
for object_uuid, references in self.references.items():
@ -69,8 +103,8 @@ class VulnerabilityParser():
misp_object.add_reference(**reference)
break
def __parse_capec(self, vulnerability_uuid):
for capec in self.vulnerability['capec']:
def __parse_capec(self, capec_values, vulnerability_uuid):
for capec in capec_values:
capec_object = MISPObject('attack-pattern')
for feature in self.capec_features:
capec_object.add_attribute(feature, capec[feature])
@ -84,12 +118,11 @@ class VulnerabilityParser():
}
)
def __parse_weakness(self, vulnerability_uuid):
cwe_string, cwe_id = self.vulnerability['cwe'].split('-')[:2]
cwes = requests.get(self.api_url.replace('/cve/', '/cwe'))
if cwes.status_code == 200:
for cwe in cwes.json():
if cwe['id'] == cwe_id:
def __parse_weakness(self, cwe_value, vulnerability_uuid):
cwe_string, cwe_id = cwe_value.split('-')[:2]
cwe = requests.get(f'{self.api_url}cwe/{cwe_id}')
if cwe.status_code == 200:
cwe = cwe.json()
weakness_object = MISPObject('weakness')
weakness_object.add_attribute('id', f'{cwe_string}-{cwe_id}')
for feature, relation in self.weakness_mapping.items():
@ -102,11 +135,10 @@ class VulnerabilityParser():
'relationship_type': 'weakened-by'
}
)
break
def check_url(url):
return "{}/".format(url) if not url.endswith('/') else url
return f"{url}/" if not url.endswith('/') else url
def handler(q=False):
@ -119,8 +151,8 @@ def handler(q=False):
if attribute.get('type') != 'vulnerability':
misperrors['error'] = 'Vulnerability id missing.'
return misperrors
api_url = check_url(request['config']['custom_API']) if request['config'].get('custom_API') else cveapi_url
r = requests.get("{}{}".format(api_url, attribute['value']))
api_url = check_url(request['config']['custom_API']) if request.get('config', {}).get('custom_API') else cveapi_url
r = requests.get(f"{api_url}cve/{attribute['value']}")
if r.status_code == 200:
vulnerability = r.json()
if not vulnerability:
@ -129,8 +161,8 @@ def handler(q=False):
else:
misperrors['error'] = 'API not accessible'
return misperrors['error']
parser = VulnerabilityParser(attribute, vulnerability, api_url)
parser.parse_vulnerability_information()
parser = VulnerabilityParser(attribute, api_url)
parser.parse_vulnerability_information(vulnerability)
return parser.get_result()