FIPS - datetime, validation dates and dgsts (#154)

* Fixes of everything that failed.

* Fixed formatting

* Removed unused imports

3 files changed, 43 insertions, 39 deletions

diff --git a/requirements.txt b/requirements.txt
index 35a1cae8..85bd6f74 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -12,7 +12,7 @@ kiwisolver==1.3.1
 lxml==4.6.5
 matplotlib==3.3.2
 numpy==1.21.4
-pandas==1.1.4
+pandas==1.3.5
 pikepdf==2.10.0
 Pillow==8.3.2
 pyparsing==2.4.7
@@ -26,7 +26,7 @@ tabula-py==2.2.0
 tabulate==0.8.7
 tqdm==4.51.0
 urllib3
-rapidfuzz==1.1.1
+rapidfuzz==1.9.1
 pyyaml
 importlib-resources==5.1.2
 scikit-learn
diff --git a/sec_certs/dataset/fips.py b/sec_certs/dataset/fips.py
index 6a664544..0fa9da01 100644
--- a/sec_certs/dataset/fips.py
+++ b/sec_certs/dataset/fips.py
@@ -1,4 +1,3 @@
-import datetime
 import logging
 import os
 import tempfile
@@ -112,7 +111,7 @@ class FIPSDataset(Dataset, ComplexSerializableType):
             raise RuntimeError("You need to provide cert ids to FIPS download PDFs functionality.")
         for cert_id in cert_ids:
             if not (self.policies_dir / f"{cert_id}.pdf").exists() or (
-                cert_id in self.certs and not self.certs[fips_dgst(cert_id)].state.txt_state
+                fips_dgst(cert_id) in self.certs and not self.certs[fips_dgst(cert_id)].state.txt_state
             ):
                 sp_urls.append(
                     f"https://csrc.nist.gov/CSRC/media/projects/cryptographic-module-validation-program/documents/security-policies/140sp{cert_id}.pdf"
@@ -360,6 +359,7 @@ class FIPSDataset(Dataset, ComplexSerializableType):
     def _compare_certs(self, current_certificate: "FIPSCertificate", other_id: str):
         other_dgst = fips_dgst(other_id)
         other_cert = self.certs[other_dgst]
+
         if (
             current_certificate.web_scan.date_validation is None
             or other_cert is None
@@ -372,14 +372,6 @@ class FIPSDataset(Dataset, ComplexSerializableType):
         conn_first = other_cert.web_scan.date_validation[0]
         conn_last = other_cert.web_scan.date_validation[-1]
 
-        if (
-            not isinstance(cert_first, datetime.date)
-            or not isinstance(cert_last, datetime.date)
-            or not isinstance(conn_first, datetime.date)
-            or not isinstance(conn_last, datetime.date)
-        ):
-            raise RuntimeError("Dataset was probably not built correctly - this should not be happening.")
-
         return (
             cert_first.year - conn_first.year > config.year_difference_between_validations
             and cert_last.year - conn_last.year > config.year_difference_between_validations
@@ -421,7 +413,7 @@ class FIPSDataset(Dataset, ComplexSerializableType):
                 if curr_id == cert_candidate_id:
                     return False
 
-        algs = self.algorithms.certs[candidate_dgst]
+        algs = self.algorithms.certs[cert_candidate_id]
         for current_alg in algs:
             if current_alg.vendor is None or processed_cert.web_scan.vendor is None:
                 raise RuntimeError("Dataset was probably not built correctly - this should not be happening.")
@@ -541,8 +533,7 @@ class FIPSDataset(Dataset, ComplexSerializableType):
         for key in self.certs:
             cert = self.certs[key]
 
-            # TODO: What? How can a key from self.certs be "Not Found"?
-            if key == "Not found" or not cert.state.file_status:
+            if not cert.state.file_status:
                 continue
 
             processed = self._get_processed_list(connection_list, key)
@@ -555,7 +546,7 @@ class FIPSDataset(Dataset, ComplexSerializableType):
                 self._highlight_vendor_in_dot(dot, key, highlighted_vendor)
                 single_dot.node(
                     key,
-                    label=key + "\r\n" + cert.web_scan.vendor
+                    label=str(cert.cert_id) + "\r\n" + cert.web_scan.vendor
                     if cert.web_scan.vendor is not None
                     else "" + ("\r\n" + cert.web_scan.module_name if cert.web_scan.module_name else ""),
                 )
@@ -563,12 +554,11 @@ class FIPSDataset(Dataset, ComplexSerializableType):
         for key in self.certs:
             cert = self.certs[key]
 
-            # TODO: What? How can a key from self.certs be "Not Found"?
-            if key == "Not found" or not cert.state.file_status:
+            if not cert.state.file_status:
                 continue
             processed = self._get_processed_list(connection_list, key)
             for conn in processed:
-                self._add_colored_node(dot, conn, highlighted_vendor)
+                self._add_colored_node(dot, fips_dgst(conn), highlighted_vendor)
                 dot.edge(key, conn)
                 edges += 1
 
diff --git a/sec_certs/sample/fips.py b/sec_certs/sample/fips.py
index 5c47c4d7..3deee713 100644
--- a/sec_certs/sample/fips.py
+++ b/sec_certs/sample/fips.py
@@ -61,11 +61,11 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
         fragment_dir: Optional[Union[str, Path]],
     ):
         if sp_dir is not None:
-            self.state.sp_path = (Path(sp_dir) / (self.dgst)).with_suffix(".pdf")
+            self.state.sp_path = (Path(sp_dir) / (str(self.cert_id))).with_suffix(".pdf")
         if html_dir is not None:
-            self.state.html_path = (Path(html_dir) / (self.dgst)).with_suffix(".html")
+            self.state.html_path = (Path(html_dir) / (str(self.cert_id))).with_suffix(".html")
         if fragment_dir is not None:
-            self.state.fragment_path = (Path(fragment_dir) / (self.dgst)).with_suffix(".txt")
+            self.state.fragment_path = (Path(fragment_dir) / (str(self.cert_id))).with_suffix(".txt")
 
     @dataclass(eq=True)
     class Algorithm(ComplexSerializableType):
@@ -92,8 +92,8 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
         module_name: Optional[str]
         standard: Optional[str]
         status: Optional[str]
-        date_sunset: Optional[Union[str, datetime]]
-        date_validation: Optional[List[Union[str, datetime]]]
+        date_sunset: Optional[datetime]
+        date_validation: Optional[List[datetime]]
         level: Optional[str]
         caveat: Optional[str]
         exceptions: Optional[List[str]]
@@ -118,12 +118,6 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
         product_url: Optional[str]
         connections: List[str]
 
-        def __post_init__(self):
-            self.date_validation = (
-                [parser.parse(x).date() for x in self.date_validation] if self.date_validation else None
-            )
-            self.date_sunset = parser.parse(self.date_sunset).date() if self.date_sunset else None
-
         @property
         def dgst(self):
             # certs in dataset are in format { id: [FIPSAlgorithm] }, there is only one type of algorithm
@@ -232,6 +226,17 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
         self.heuristics = heuristics
         self.state = state
 
+    @classmethod
+    def from_dict(cls, dct: Dict) -> "FIPSCertificate":
+        new_dct = dct.copy()
+
+        if new_dct["web_scan"].date_validation:
+            new_dct["web_scan"].date_validation = [parser.parse(x).date() for x in new_dct["web_scan"].date_validation]
+
+        if new_dct["web_scan"].date_sunset:
+            new_dct["web_scan"].date_sunset = parser.parse(new_dct["web_scan"].date_sunset).date()
+        return super(cls, FIPSCertificate).from_dict(new_dct)
+
     @staticmethod
     def download_html_page(cert: Tuple[str, Path]) -> Optional[Tuple[str, Path]]:
         exit_code = helpers.download_file(*cert, delay=1)
@@ -251,7 +256,7 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
             "level": None,
             "caveat": None,
             "exceptions": None,
-            "type": None,
+            "module_type": None,
             "embodiment": None,
             "tested_conf": None,
             "description": None,
@@ -345,8 +350,8 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
         )
 
         if title in pairs:
-            if "date_validation" == pairs[title]:
-                html_items_found[pairs[title]] = [x for x in content.split(";")]
+            if "date_sunset" == pairs[title]:
+                html_items_found[pairs[title]] = parser.parse(content).date()
 
             elif "caveat" in pairs[title]:
                 html_items_found[pairs[title]] = content
@@ -407,9 +412,15 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
 
     @staticmethod
     def normalize(items: Dict):
-        items["type"] = items["type"].lower().replace("-", " ").title()
+        items["module_type"] = items["module_type"].lower().replace("-", " ").title()
         items["embodiment"] = items["embodiment"].lower().replace("-", " ").replace("stand alone", "standalone").title()
 
+    @staticmethod
+    def parse_validation_dates(current_div: Tag, html_items_found: Dict):
+        table = current_div.find("table")
+        rows = table.find("tbody").findAll("tr")
+        html_items_found["date_validation"] = [parser.parse(td.text).date() for td in [row.find("td") for row in rows]]
+
     @classmethod
     def html_from_file(
         cls, file: Path, state: State, initialized: "FIPSCertificate" = None, redo: bool = False
@@ -423,7 +434,7 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
             "Overall Level": "level",
             "Caveat": "caveat",
             "Security Level Exceptions": "exceptions",
-            "Module Type": "type",
+            "Module Type": "module_type",
             "Embodiment": "embodiment",
             "FIPS Algorithms": "algorithms",
             "Allowed Algorithms": "algorithms",
@@ -471,6 +482,9 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
             if div.find("h4", class_="panel-title").text == "Related Files":
                 FIPSCertificate.parse_related_files(div, items_found)
 
+            if div.find("h4", class_="panel-title").text == "Validation History":
+                FIPSCertificate.parse_validation_dates(div, items_found)
+
         FIPSCertificate.normalize(items_found)
 
         return FIPSCertificate(
@@ -521,7 +535,7 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
         if not cert.state.txt_state:
             exit_code = helpers.convert_pdf_file(pdf_path, txt_path, ["-raw"])
             if exit_code != constants.RETURNCODE_OK:
-                logger.error(f"Cert dgst: {cert.dgst} failed to convert security policy pdf->txt")
+                logger.error(f"Cert dgst: {cert.cert_id} failed to convert security policy pdf->txt")
                 cert.state.txt_state = False
             else:
                 cert.state.txt_state = True
@@ -583,7 +597,7 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
         not_found = []
 
         if cert.web_scan.algorithms is None:
-            raise RuntimeError(f"Algorithms were not found for cert {cert.dgst} - this should not be happening.")
+            raise RuntimeError(f"Algorithms were not found for cert {cert.cert_id} - this should not be happening.")
 
         for alg_list in [a["Certificate"] for a in cert.web_scan.algorithms]:
             for web_alg in alg_list:
@@ -744,7 +758,7 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
         result: Set[str] = set()
 
         if self.web_scan.algorithms is None:
-            raise RuntimeError(f"Algorithms were not found for cert {self.dgst} - this should not be happening.")
+            raise RuntimeError(f"Algorithms were not found for cert {self.cert_id} - this should not be happening.")
 
         for alg in self.web_scan.algorithms:
             result.update(cert for cert in alg["Certificate"])
@@ -808,7 +822,7 @@ class FIPSCertificate(Certificate, ComplexSerializableType):
     # TODO: This function is probably safe to delete // I'll not type it then - older API probably?
     def compute_heuristics_cpe_vendors(self, cpe_dataset: CPEDataset):
         if self.web_scan.vendor is None:
-            raise RuntimeError(f"Vendor for cert {self.dgst} not found - this should not be happening.")
+            raise RuntimeError(f"Vendor for cert {self.cert_id} not found - this should not be happening.")
         self.heuristics.cpe_candidate_vendors = cpe_dataset.get_candidate_list_of_vendors(self.web_scan.vendor)  # type: ignore
 
     def compute_heuristics_cpe_match(self, cpe_classifier: CPEClassifier):