Algorithm matching + saner usage (#43)

* fixed continuous adding certs

* little push

* date

* added switches

* json file + formatting + match algs

* matching algs

* moved downloading of algorithms, boolean switch and logging

* parsing rules to match random matches

* Description match fix

* removed unnecessary check for web redo

* match_algs sane + web redo

* processing always from scratch, redo wipes web

* nist pages seem a bit... slower

* download htmls twice + small cleanup

* added weirdly shaped "-"

* added matching web algs to pdf

* redo quick fix

* security functions included in table finding

* saner logging

* repeated downloading now just works :tm:

* redo in demo + matching algs demo

* moved plotting and parsing to helpers

* removed dependency

Co-authored-by: Stanislav Boboň <xbobon@fi.muni.cz>

1 files changed, 118 insertions, 81 deletions

diff --git a/sec_certs/dataset.py b/sec_certs/dataset.py
index 3b983fcc..a9854edd 100644
--- a/sec_certs/dataset.py
+++ b/sec_certs/dataset.py
@@ -19,7 +19,6 @@ from bs4 import BeautifulSoup, Tag
 from rapidfuzz import process, fuzz
 import xml.etree.ElementTree as ET
 
-
 import sec_certs.helpers as helpers
 import sec_certs.constants as constants
 import sec_certs.cert_processing as cert_processing
@@ -145,7 +144,8 @@ class CCDataset(Dataset, ComplexSerializableType):
         def from_dict(cls, dct: Dict[str, bool]):
             return cls(*tuple(dct.values()))
 
-    def __init__(self, certs: Dict[str, 'Certificate'], root_dir: Path, name: str = 'dataset name', description: str = 'dataset_description', state: Optional[DatasetInternalState] = None):
+    def __init__(self, certs: Dict[str, 'Certificate'], root_dir: Path, name: str = 'dataset name',
+                 description: str = 'dataset_description', state: Optional[DatasetInternalState] = None):
         super().__init__(certs, root_dir, name, description)
         if state is None:
             state = self.DatasetInternalState()
@@ -368,11 +368,13 @@ class CCDataset(Dataset, ComplexSerializableType):
                                                                       x.maintainance_report_link,
                                                                       x.maintainance_st_link))
 
-        certs = {x.dgst: CommonCriteriaCert(cert_status, x.category, x.cert_name, x.manufacturer, x.scheme, x.security_level,
-                                            x.not_valid_before, x.not_valid_after, x.report_link, x.st_link, 'csv',
-                                            None, None, profiles.get(x.dgst, None), updates.get(x.dgst, None), None, None, None) for
-                 x in
-                 df_base.itertuples()}
+        certs = {
+            x.dgst: CommonCriteriaCert(cert_status, x.category, x.cert_name, x.manufacturer, x.scheme, x.security_level,
+                                       x.not_valid_before, x.not_valid_after, x.report_link, x.st_link, 'csv',
+                                       None, None, profiles.get(x.dgst, None), updates.get(x.dgst, None), None, None,
+                                       None) for
+            x in
+            df_base.itertuples()}
         return certs
 
     def _get_all_certs_from_html(self, get_active: bool, get_archived: bool) -> Dict[str, 'CommonCriteriaCert']:
@@ -408,7 +410,8 @@ class CCDataset(Dataset, ComplexSerializableType):
                                  date_string[1] + ' ' + time_string
             return datetime.strptime(formatted_datetime, ' %B %d %Y %I:%M %p')
 
-        def _parse_table(soup: BeautifulSoup, cert_status: str, table_id: str, category_string: str) -> Dict[str, 'CommonCriteriaCert']:
+        def _parse_table(soup: BeautifulSoup, cert_status: str, table_id: str, category_string: str) -> Dict[
+            str, 'CommonCriteriaCert']:
             tables = soup.find_all('table', id=table_id)
             assert len(tables) <= 1
 
@@ -516,7 +519,8 @@ class CCDataset(Dataset, ComplexSerializableType):
         if fresh is True:
             certs_to_process = [x for x in self.certs.values() if x.state.report_link_ok]
         else:
-            certs_to_process = [x for x in self.certs.values() if x.state.report_link_ok and not x.state.report_convert_ok]
+            certs_to_process = [x for x in self.certs.values() if
+                                x.state.report_link_ok and not x.state.report_convert_ok]
         cert_processing.process_parallel(CommonCriteriaCert.convert_report_pdf, certs_to_process, constants.N_THREADS)
 
     def _convert_targets_to_txt(self, fresh: bool = True):
@@ -557,15 +561,18 @@ class CCDataset(Dataset, ComplexSerializableType):
         if fresh is True:
             certs_to_process = [x for x in self.certs.values() if x.state.report_convert_ok]
         else:
-            certs_to_process = [x for x in self.certs.values() if x.state.report_convert_ok and not x.state.report_extract_ok]
-        cert_processing.process_parallel(CommonCriteriaCert.extract_report_pdf_metadata, certs_to_process, constants.N_THREADS)
+            certs_to_process = [x for x in self.certs.values() if
+                                x.state.report_convert_ok and not x.state.report_extract_ok]
+        cert_processing.process_parallel(CommonCriteriaCert.extract_report_pdf_metadata, certs_to_process,
+                                         constants.N_THREADS)
 
     def _extract_targets_metadata(self, fresh: bool = True):
         if fresh is True:
             certs_to_process = [x for x in self.certs.values() if x.state.st_convert_ok]
         else:
             certs_to_process = [x for x in self.certs.values() if x.state.st_convert_ok and not x.state.st_extract_ok]
-        cert_processing.process_parallel(CommonCriteriaCert.extract_st_pdf_metadata, certs_to_process, constants.N_THREADS)
+        cert_processing.process_parallel(CommonCriteriaCert.extract_st_pdf_metadata, certs_to_process,
+                                         constants.N_THREADS)
 
     def extract_pdf_metadata(self, fresh: bool = True):
         logger.info('Extracting pdf metadata from CC dataset')
@@ -577,14 +584,17 @@ class CCDataset(Dataset, ComplexSerializableType):
             certs_to_process = [x for x in self.certs.values() if x.state.st_convert_ok]
         else:
             certs_to_process = [x for x in self.certs.values() if x.state.st_convert_ok and not x.state.st_extract_ok]
-        cert_processing.process_parallel(CommonCriteriaCert.extract_st_pdf_frontpage, certs_to_process, constants.N_THREADS)
+        cert_processing.process_parallel(CommonCriteriaCert.extract_st_pdf_frontpage, certs_to_process,
+                                         constants.N_THREADS)
 
     def _extract_report_frontpage(self, fresh: bool = True):
         if fresh is True:
             certs_to_process = [x for x in self.certs.values() if x.state.report_convert_ok]
         else:
-            certs_to_process = [x for x in self.certs.values() if x.state.report_convert_ok and not x.state.report_extract_ok]
-        cert_processing.process_parallel(CommonCriteriaCert.extract_report_pdf_frontpage, certs_to_process, constants.N_THREADS)
+            certs_to_process = [x for x in self.certs.values() if
+                                x.state.report_convert_ok and not x.state.report_extract_ok]
+        cert_processing.process_parallel(CommonCriteriaCert.extract_report_pdf_frontpage, certs_to_process,
+                                         constants.N_THREADS)
 
     def extract_pdf_frontpage(self, fresh: bool = True):
         logger.info('Extracting pdf frontpages from CC dataset.')
@@ -595,15 +605,18 @@ class CCDataset(Dataset, ComplexSerializableType):
         if fresh is True:
             certs_to_process = [x for x in self.certs.values() if x.state.report_convert_ok]
         else:
-            certs_to_process = [x for x in self.certs.values() if x.state.report_convert_ok and not x.state.report_extract_ok]
-        cert_processing.process_parallel(CommonCriteriaCert.extract_report_pdf_keywords, certs_to_process, constants.N_THREADS)
+            certs_to_process = [x for x in self.certs.values() if
+                                x.state.report_convert_ok and not x.state.report_extract_ok]
+        cert_processing.process_parallel(CommonCriteriaCert.extract_report_pdf_keywords, certs_to_process,
+                                         constants.N_THREADS)
 
     def _extract_targets_keywords(self, fresh: bool = True):
         if fresh is True:
             certs_to_process = [x for x in self.certs.values() if x.state.st_convert_ok]
         else:
             certs_to_process = [x for x in self.certs.values() if x.state.st_convert_ok and not x.state.st_extract_ok]
-        cert_processing.process_parallel(CommonCriteriaCert.extract_st_pdf_keywords, certs_to_process, constants.N_THREADS)
+        cert_processing.process_parallel(CommonCriteriaCert.extract_st_pdf_keywords, certs_to_process,
+                                         constants.N_THREADS)
 
     def extract_pdf_keywords(self, fresh: bool = True):
         logger.info('Extracting pdf keywords from CC dataset.')
@@ -641,7 +654,8 @@ class CCDataset(Dataset, ComplexSerializableType):
     def fuzzy_match_cpe(self, cpe_path: Path, update_json: bool = False):
         def get_cpe_titles(cpe_path: Path):
             root = ET.parse(str(cpe_path)).getroot()
-            return [child.text for child in root.findall('{http://cpe.mitre.org/dictionary/2.0}cpe-item/{http://cpe.mitre.org/dictionary/2.0}title')]
+            return [child.text for child in root.findall(
+                '{http://cpe.mitre.org/dictionary/2.0}cpe-item/{http://cpe.mitre.org/dictionary/2.0}title')]
 
         digests = [x for x in self.certs.keys()]
         cpe_titles = get_cpe_titles(cpe_path)
@@ -653,7 +667,9 @@ class CCDataset(Dataset, ComplexSerializableType):
         chunks = chunk_list(digests, constants.N_THREADS)
         chunks_dicts = [{x: self[x].name for x in y} for y in chunks]
 
-        results = cert_processing.process_parallel(helpers.match_certs, list(zip(chunks_dicts, [cpe_titles for _ in range(constants.N_THREADS)])), constants.N_THREADS, use_threading=False, unpack=True)
+        results = cert_processing.process_parallel(helpers.match_certs, list(
+            zip(chunks_dicts, [cpe_titles for _ in range(constants.N_THREADS)])), constants.N_THREADS,
+                                                   use_threading=False, unpack=True)
 
         for chunk in results:
             for digest, matches in chunk.items():
@@ -663,7 +679,6 @@ class CCDataset(Dataset, ComplexSerializableType):
             self.to_json(self.json_path)
 
 
-
 class FIPSDataset(Dataset, ComplexSerializableType):
     FIPS_BASE_URL: ClassVar[str] = 'https://csrc.nist.gov'
     FIPS_MODULE_URL: ClassVar[
@@ -717,12 +732,21 @@ class FIPSDataset(Dataset, ComplexSerializableType):
         for keyword, cert in keywords:
             self.certs[cert.dgst].pdf_scan.keywords = keyword
 
+    def match_algs(self, show_graph=False) -> Dict:
+        output = {}
+        for cert in self.certs.values():
+            output[cert.dgst] = FIPSCertificate.match_web_algs_to_pdf(cert)
+
+        return output
+
+
+
     def download_all_pdfs(self):
         sp_paths, sp_urls = [], []
         self.policies_dir.mkdir(exist_ok=True)
 
         for cert_id in list(self.certs.keys()):
-            if not (self.policies_dir / f'{cert_id}.pdf').exists():
+            if not (self.policies_dir / f'{cert_id}.pdf').exists() or not self.certs[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")
                 sp_paths.append(self.policies_dir / f"{cert_id}.pdf")
@@ -731,44 +755,27 @@ class FIPSDataset(Dataset, ComplexSerializableType):
                                          constants.N_THREADS)
         self.new_files += len(sp_urls)
 
-    def download_all_htmls(self):
+    def download_all_htmls(self) -> List[str]:
         html_paths, html_urls = [], []
-
+        new_files = []
         self.web_dir.mkdir(exist_ok=True)
-        for cert_id in list(self.certs.keys()):
+        for cert_id in self.certs.keys():
             if not (self.web_dir / f'{cert_id}.html').exists():
                 html_urls.append(
                     f"https://csrc.nist.gov/projects/cryptographic-module-validation-program/certificate/{cert_id}")
                 html_paths.append(self.web_dir / f"{cert_id}.html")
+                new_files.append(cert_id)
 
         logging.info(f"downloading {len(html_urls)} module html files")
-        cert_processing.process_parallel(FIPSCertificate.download_html_page, list(zip(html_urls, html_paths)),
+        failed = cert_processing.process_parallel(FIPSCertificate.download_html_page, list(zip(html_urls, html_paths)),
                                          constants.N_THREADS)
-        self.new_files += len(html_urls)
-
-    def download_all_algs(self):
-        algs_paths, algs_urls = [], []
-
-        # get first page to find out how many pages there are
-        helpers.download_file(
-            constants.FIPS_ALG_URL + '1',
-            self.algs_dir / "page1.html")
-
-        with open(self.algs_dir / "page1.html", "r") as alg_file:
-            soup = BeautifulSoup(alg_file.read(), 'html.parser')
-            num_pages = soup.select('span[data-total-pages]')[0].attrs
-
-        self.algs_dir.mkdir(exist_ok=True)
-        for i in range(1, int(num_pages['data-total-pages'])):
-            if not (self.algs_dir / f'page{i}.html').exists():
-                algs_urls.append(
-                    constants.FIPS_ALG_URL + str(i))
-                algs_paths.append(self.algs_dir / f"page{i}.html")
+        failed = [c for c in failed if c]
 
-        logging.info(f"downloading {len(algs_urls)} algs html files")
-        cert_processing.process_parallel(FIPSCertificate.download_html_page, list(zip(algs_urls, algs_paths)),
+        self.new_files += len(html_urls)
+        logging.info(f"Download failed for {len(failed)} files. Retrying...")
+        cert_processing.process_parallel(FIPSCertificate.download_html_page, failed,
                                          constants.N_THREADS)
-        self.new_files += len(algs_urls)
+        return new_files
 
     def convert_all_pdfs(self):
         logger.info('Converting FIPS certificate reports to .txt')
@@ -779,11 +786,11 @@ class FIPSDataset(Dataset, ComplexSerializableType):
         ]
         cert_processing.process_parallel(FIPSCertificate.convert_pdf_file, tuples, constants.N_THREADS)
 
-    def get_certs_from_web(self, redo=False):
-        def download_html_pages():
+    def get_certs_from_web(self, redo: bool = False, json_file: Optional[Path] = None):
+        def download_html_pages() -> List[str]:
+            new_files = self.download_all_htmls()
             self.download_all_pdfs()
-            self.download_all_htmls()
-            self.download_all_algs()
+            return new_files
 
         def get_certificates_from_html(html_file: Path) -> None:
             logger.info(f'Getting certificate ids from {html_file}')
@@ -819,29 +826,36 @@ class FIPSDataset(Dataset, ComplexSerializableType):
             get_certificates_from_html(self.web_dir / f)
 
         logger.info('Downloading certificate html and security policies')
-        download_html_pages()
+
+        if not json_file:
+            json_file = self.root_dir / 'fips_full_dataset.json'
+
+        if json_file.exists():
+            logger.info("Certs loaded from previous scanning")
+            dataset = self.from_json(json_file)
+            self.certs = dataset.certs
+            self.algorithms = dataset.algorithms
+
+        new_certs = download_html_pages()
 
         logger.info(f"{self.new_files} needed to be downloaded")
 
-        if not (self.root_dir / 'fips_full_dataset.json').exists():
-            for cert_id in self.certs:
-                self.certs[cert_id] = FIPSCertificate.html_from_file(
-                    self.web_dir / f'{cert_id}.html',
-                    FIPSCertificate.State((self.policies_dir / cert_id).with_suffix('.pdf'),
-                                          (self.web_dir / cert_id).with_suffix('.html'),
-                                          (self.fragments_dir / cert_id).with_suffix('.txt'), False, None, False))
-            return
+        for cert_id in new_certs:
+            self.certs[cert_id] = None
 
-        logger.info("Certs loaded from previous scanning")
-        dataset = self.from_json(self.root_dir / 'fips_full_dataset.json')
-        self.certs = dataset.certs
-        if redo or self.new_files > 0:
-            for cert_id, cert in self.certs.items():
-                self.certs[cert_id] = FIPSCertificate.html_from_file(
-                    self.web_dir / f'{cert_id}.html',
-                    FIPSCertificate.State((self.policies_dir / cert_id).with_suffix('.pdf'),
-                                          (self.web_dir / cert_id).with_suffix('.html'),
-                                          (self.fragments_dir / cert_id).with_suffix('.txt'), False, None, False), cert)
+        if not redo and self.new_files == 0:
+            logger.info('No new changes to web_scan are going to be made')
+            return
+        # now we want to do redo, because we want to avoid duplicites
+        redo = True
+        logger.info(f'Parsing web pages{" from scratch" if redo else ""}...')
+        for cert_id, cert in self.certs.items():
+            self.certs[cert_id] = FIPSCertificate.html_from_file(
+                self.web_dir / f'{cert_id}.html',
+                FIPSCertificate.State((self.policies_dir / cert_id).with_suffix('.pdf'),
+                                      (self.web_dir / cert_id).with_suffix('.html'),
+                                      (self.fragments_dir / cert_id).with_suffix('.txt'), False, None, False),
+                cert, redo=redo)
 
     def extract_certs_from_tables(self) -> List[Path]:
         """
@@ -894,9 +908,10 @@ class FIPSDataset(Dataset, ComplexSerializableType):
                 conn_last = self.certs[other_id].web_scan.date_validation[-1].year
 
                 return cert_first - conn_first > config.year_difference_between_validations['value'] \
-                       and cert_last - conn_last > config.year_difference_between_validations['value']
+                       and cert_last - conn_last > config.year_difference_between_validations['value'] \
+                       or cert_first < conn_first
 
-            # "< 105" still needs to be used, because of some old certs being revalidated
+            # "< number" still needs to be used, because of some old certs being revalidated
             if cert_candidate.isdecimal() \
                     and int(cert_candidate) < config.smallest_certificate_id_to_connect['value'] or \
                     compare_certs(processed_cert, cert_candidate):
@@ -988,12 +1003,12 @@ class FIPSDataset(Dataset, ComplexSerializableType):
                 dot.attr('node', color='gold')
             found_interesting_cert(current_key)
             dot.node(current_key,
-                     label= current_key +
-                     '&#10;' +
-                     self.certs[current_key].web_scan.vendor +
-                     '&#10;' +
-                     (self.certs[current_key].web_scan.module_name
-                      if self.certs[current_key].web_scan.module_name else ''))
+                     label=current_key +
+                           '&#10;' +
+                           self.certs[current_key].web_scan.vendor +
+                           '&#10;' +
+                           (self.certs[current_key].web_scan.module_name
+                            if self.certs[current_key].web_scan.module_name else ''))
 
         keys = 0
         edges = 0
@@ -1062,7 +1077,29 @@ class FIPSDataset(Dataset, ComplexSerializableType):
 class FIPSAlgorithmDataset(Dataset, ComplexSerializableType):
 
     def get_certs_from_web(self):
-        pass
+        self.root_dir.mkdir(exist_ok=True)
+        algs_paths, algs_urls = [], []
+
+        # get first page to find out how many pages there are
+        helpers.download_file(
+            constants.FIPS_ALG_URL + '1',
+            self.root_dir / "page1.html")
+
+        with open(self.root_dir / "page1.html", "r") as alg_file:
+            soup = BeautifulSoup(alg_file.read(), 'html.parser')
+            num_pages = soup.select('span[data-total-pages]')[0].attrs
+
+        for i in range(1, int(num_pages['data-total-pages'])):
+            if not (self.root_dir / f'page{i}.html').exists():
+                algs_urls.append(
+                    constants.FIPS_ALG_URL + str(i))
+                algs_paths.append(self.root_dir / f"page{i}.html")
+
+        logging.info(f"downloading {len(algs_urls)} algs html files")
+        cert_processing.process_parallel(FIPSCertificate.download_html_page, list(zip(algs_urls, algs_paths)),
+                                         constants.N_THREADS)
+
+        self.parse_html()
 
     def parse_html(self):
         def split_alg(alg_string):