chg: [vulnerability_parser] Supporting CNVD vulnerability IDs

Author: chrisr3d

misp_modules/modules/expansion/_vulnerability_parser/vulnerability_parser.py

@@ -98,6 +98,14 @@ def _parse_variot_description(self, query_results):
 
 
 class VulnerabilityLookupMapping(VulnerabilityMapping):
+    __cnvd_mapping = {
+        "number": "id",
+        "title": "summary",
+        "description": "description",
+        "referenceLink": "references",
+        "submitTime": "published",
+        "openTime": "modified"
+    }
     __csaf_mapping = {
         "id": "id",
         "initial_release_date": "published",
@@ -137,6 +145,7 @@ class VulnerabilityLookupMapping(VulnerabilityMapping):
         "discovery_date": "published"
     }
     __source_mapping = {
+        "cnvd": "_parse_cnvd_description",
         "cve": "_parse_cve_description",
         'fkie_cve': '_parse_fkie_description',
         "ghsa": "_parse_standard_description",
@@ -170,6 +179,10 @@ class VulnerabilityLookupMapping(VulnerabilityMapping):
         "published": "published"
     }
 
+    @classmethod
+    def cnvd_mapping(cls) -> dict:
+        return cls.__cnvd_mapping
+
     @classmethod
     def csaf_mapping(cls) -> dict:
         return cls.__csaf_mapping
@@ -249,18 +262,36 @@ def _create_vulnerability_object(self, vuln_id: str) -> MISPObject:
         )
         return misp_object
 
+    def _parse_alias(self, alias: str) -> str:
+        query = requests.get(f"{self.api_url}/api/vulnerability/{alias}")
+        if query.status_code != 200:
+            self.errors.append(
+                f"Unable to query related vulnerability id {alias}"
+            )
+            return
+        vulnerability = query.json()
+        if not vulnerability:
+            self.errors.append(
+                f"No results for related vulnerability id{alias}"
+            )
+            return
+        feature = self.mapping.source_mapping(alias.split("-")[0].lower())
+        return getattr(self, feature)(vulnerability)
+
     def _parse_aliases(self, *aliases: tuple) -> Iterator[str]:
         for alias in aliases:
-            query = requests.get(f"{self.api_url}/api/vulnerability/{alias}")
-            if query.status_code != 200:
-                self.errors.append(f"Unable to query related vulnerability id {alias}")
-                continue
-            vulnerability = query.json()
-            if not vulnerability:
-                self.errors.append(f"No results for related vulnerability id{alias}")
-                continue
-            feature = self.mapping.source_mapping(alias.split("-")[0].lower())
-            yield getattr(self, feature)(vulnerability)
+            yield self._parse_alias(alias)
+
+    def _parse_cnvd_description(self, lookup_result: dict) -> str:
+        misp_object = self._create_vulnerability_object(lookup_result['number'])
+        for field, relation in self.mapping.cnvd_mapping().items():
+            misp_object.add_attribute(relation, lookup_result[field])
+        vulnerability_object = self.misp_event.add_object(misp_object)
+        cve = lookup_result.get('cves', {}).get('cve', {}).get('cveNumber')
+        if cve is not None:
+            vulnerability_object.add_reference(
+                self._parse_alias(cve), 'related-to'
+            )
 
     def _parse_csaf_branch(self, branch: list) -> Iterator[str]:
         for sub_branch in branch:
@@ -434,7 +465,7 @@ def _parse_jvn_description(self, lookup_result: dict) -> str:
                 else:
                     misp_object.add_reference(
                         self.misp_event.add_attribute(type="weakness", value=reference["@id"]).uuid,
-                        "weakened-by",
+                        "weakened-by"
                     )
                 continue
             if source == "JVN":