import json import requests from . import check_input_attribute, standard_error_message from collections import defaultdict from pymisp import MISPAttribute, MISPEvent, MISPObject misperrors = {'error': 'Error'} mispattributes = {'input': ['vulnerability'], 'format': 'misp_standard'} 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://cvepremium.circl.lu/api/' class VulnerabilityParser(): 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 = { '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', '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: self.__build_references() event = json.loads(self.misp_event.to_json()) results = {key: event[key] for key in ('Attribute', 'Object') if (key in event and event[key])} return {'results': results} def parse_vulnerability_information(self, vulnerability): vulnerability_object = MISPObject('vulnerability') 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, 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.misp_attribute.uuid, 'related-to') self.misp_event.add_object(vulnerability_object) 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(): for misp_object in self.misp_event.objects: if misp_object.uuid == object_uuid: for reference in references: misp_object.add_reference(**reference) break 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]) for related_weakness in capec['related_weakness']: capec_object.add_attribute('related-weakness', f"CWE-{related_weakness}") self.misp_event.add_object(capec_object) self.references[vulnerability_uuid].append( { 'referenced_uuid': capec_object.uuid, 'relationship_type': 'targeted-by' } ) 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 f"{url}/" if not url.endswith('/') else url def handler(q=False): if q is False: return False request = json.loads(q) if not request.get('attribute') or not check_input_attribute(request['attribute']): return {'error': f'{standard_error_message}, which should contain at least a type, a value and an uuid.'} attribute = request['attribute'] if attribute.get('type') != 'vulnerability': misperrors['error'] = 'Vulnerability id missing.' return misperrors 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: misperrors['error'] = 'Non existing CVE' return misperrors['error'] else: misperrors['error'] = 'API not accessible' return misperrors['error'] parser = VulnerabilityParser(attribute, api_url) parser.parse_vulnerability_information(vulnerability) return parser.get_result() def introspection(): return mispattributes def version(): moduleinfo['config'] = moduleconfig return moduleinfo