diff --git a/misp_modules/modules/expansion/cve_advanced.py b/misp_modules/modules/expansion/cve_advanced.py index 4be69c6..32f86d1 100644 --- a/misp_modules/modules/expansion/cve_advanced.py +++ b/misp_modules/modules/expansion/cve_advanced.py @@ -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,29 +118,27 @@ 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: - weakness_object = MISPObject('weakness') - weakness_object.add_attribute('id', f'{cwe_string}-{cwe_id}') - for feature, relation in self.weakness_mapping.items(): - if cwe.get(feature): - weakness_object.add_attribute(relation, cwe[feature]) - self.misp_event.add_object(weakness_object) - self.references[vulnerability_uuid].append( - { - 'referenced_uuid': weakness_object.uuid, - 'relationship_type': 'weakened-by' - } - ) - break + 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(): + if cwe.get(feature): + weakness_object.add_attribute(relation, cwe[feature]) + self.misp_event.add_object(weakness_object) + self.references[vulnerability_uuid].append( + { + 'referenced_uuid': weakness_object.uuid, + 'relationship_type': 'weakened-by' + } + ) 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()