random fixes

4 files changed, 50 insertions, 11 deletions

diff --git a/src/sec_certs/dataset/dataset.py b/src/sec_certs/dataset/dataset.py
index a623fa0f..939f8488 100644
--- a/src/sec_certs/dataset/dataset.py
+++ b/src/sec_certs/dataset/dataset.py
@@ -423,6 +423,8 @@ class Dataset(Generic[CertSubType, AuxiliaryDatasetsSubType], ComplexSerializabl
                     with gzip.open(str(dset_path)) as handle:
                         json_str = handle.read().decode("utf-8")
                         cpe_match_dict = json.loads(json_str)
+            with self.cpe_match_json_path.open("w") as handle:
+                json.dump(cpe_match_dict, handle, indent=4)
 
         return cpe_match_dict
 
diff --git a/src/sec_certs/sample/cpe.py b/src/sec_certs/sample/cpe.py
index 400c5a39..a94f5ee1 100644
--- a/src/sec_certs/sample/cpe.py
+++ b/src/sec_certs/sample/cpe.py
@@ -99,7 +99,8 @@ class CPE(PandasSerializableType, ComplexSerializableType):
     __slots__ = ["cpe_id", "uri", "version", "vendor", "item_name", "title"]
 
     pandas_columns: ClassVar[list[str]] = [
-        "cpe_id" "uri",
+        "cpe_id",
+        "uri",
         "vendor",
         "item_name",
         "version",
diff --git a/src/sec_certs/sample/cve.py b/src/sec_certs/sample/cve.py
index 3e4c3603..333f1899 100644
--- a/src/sec_certs/sample/cve.py
+++ b/src/sec_certs/sample/cve.py
@@ -183,8 +183,8 @@ class CVE(PandasSerializableType, ComplexSerializableType):
     ) -> tuple[list[CPEMatchCriteria], list[CPEMatchCriteriaConfiguration]]:
         criteria = []
         criteria_configurations = []
-
         configurations = dct.get("configurations", [])
+
         for conf in configurations:
             new_criteria, new_criteria_configuration = CVE.parse_single_configuration(conf)
             criteria.extend(new_criteria)
@@ -196,17 +196,43 @@ class CVE(PandasSerializableType, ComplexSerializableType):
     def parse_single_configuration(
         configuration: dict[str, Any]
     ) -> tuple[list[CPEMatchCriteria], CPEMatchCriteriaConfiguration | None]:
-        if "operator" not in configuration or configuration["operator"] == "OR":
-            assert len(configuration["nodes"]) == 1 and "cpeMatch" in configuration["nodes"][0]
-            return CVE.get_criteria_from_node(configuration["nodes"][0]["cpeMatch"]), None
+        if CVE.configuration_is_simple(configuration):
+            return CVE.get_simple_criteria_from_cpe_matches(configuration["nodes"][0]["cpeMatch"]), None
+        else:
+            return [], CVE.get_configuration_criteria_from_configuration_nodes(configuration["nodes"])
 
-        return [], CVE.get_configuration_criteria_from_nodes(configuration["nodes"])
+    @staticmethod
+    def configuration_is_simple(configuration: dict) -> bool:
+        return (
+            len(configuration["nodes"]) == 1
+            and "cpeMatch" in configuration["nodes"][0]
+            and (configuration.get("operator", "OR") == "OR" or len(configuration["nodes"][0]["cpeMatch"]) == 1)
+        )
 
     @staticmethod
-    def get_configuration_criteria_from_nodes(nodes) -> CPEMatchCriteriaConfiguration:
-        assert all("cpeMatch" in x for x in nodes)  # the next layer are matches
-        return CPEMatchCriteriaConfiguration([CVE.get_criteria_from_node(x["cpeMatch"]) for x in nodes])
+    def get_configuration_criteria_from_configuration_nodes(
+        configuration_nodes: dict,
+    ) -> CPEMatchCriteriaConfiguration | None:
+        """
+        Retrieves complex configuration criteria from a dictionary of configuration nodes.
+        It is aasserted that the dictionary has two layers at most, that the top-level children are in AND relationship,
+        and that the individual elements are in OR relationship (otherwise, they would be parsed by different method.)
+
+        We cannot process configuration when elements of a single component are in AND relationship.
+        Out of all configurations in dataset as of April 2023, only 3 were detected in the dataset.
+        We ignore those on purpose.
+
+        :param dict configuration_nodes: _description_
+        :return CPEMatchCriteriaConfiguration | None: _description_
+        """
+        assert all("cpeMatch" in x for x in configuration_nodes)  # the next layer are matches
+        nodes = [x for x in configuration_nodes if "operator" not in x or x["operator"] == "OR"]
+        if nodes:
+            return CPEMatchCriteriaConfiguration(
+                [CVE.get_simple_criteria_from_cpe_matches(x["cpeMatch"]) for x in nodes]
+            )
+        return None
 
     @staticmethod
-    def get_criteria_from_node(cpe_matches: list[dict[str, Any]]) -> list[CPEMatchCriteria]:
+    def get_simple_criteria_from_cpe_matches(cpe_matches: list[dict[str, Any]]) -> list[CPEMatchCriteria]:
         return [CPEMatchCriteria.from_nist_dict(x) for x in cpe_matches]
diff --git a/src/sec_certs/utils/nvd_dataset_builder.py b/src/sec_certs/utils/nvd_dataset_builder.py
index 5a991f9d..b6d3497a 100644
--- a/src/sec_certs/utils/nvd_dataset_builder.py
+++ b/src/sec_certs/utils/nvd_dataset_builder.py
@@ -324,8 +324,13 @@ class CpeMatchNvdDatasetBuilder(NvdDatasetBuilder[dict]):
     ENDPOINT: ClassVar[str] = "CPEMatch"
     ENDPOINT_URL: ClassVar[str] = "https://services.nvd.nist.gov/rest/json/cpematch/2.0"
     RESULTS_PER_PAGE: ClassVar[int] = 5000
+    VERSION_KEYS: ClassVar[list[str]] = [
+        "versionStartIncluding",
+        "versionStartExcluding",
+        "versionEndIncluding",
+        "versionEndExcluding",
+    ]
 
-    # TODO: I'm actually forgetting to process start_version and end_version
     def _process_responses(self, responses: list[Response], dataset_to_fill: dict) -> dict:
         timestamp = self._end_mod_date.isoformat() if self._end_mod_date else responses[-1].json()["timestamp"]
         match_strings = list(itertools.chain.from_iterable(response.json()["matchStrings"] for response in responses))
@@ -341,6 +346,11 @@ class CpeMatchNvdDatasetBuilder(NvdDatasetBuilder[dict]):
                         "criteria": m["matchString"]["criteria"],
                         "matches": m["matchString"]["matches"],
                     }
+                    for version_key in self.VERSION_KEYS:
+                        if version_key in m["matchString"]:
+                            dataset_to_fill["match_strings"][m["matchString"]["matchCriteriaId"]][version_key] = m[
+                                "matchString"
+                            ][version_key]
 
         for inactive in inactive_criteria:
             dataset_to_fill["match_strings"].pop(inactive, None)