diff options
| -rwxr-xr-x | cc_cli.py | 3 | ||||
| -rw-r--r-- | examples/cc_cpe_labeling.py | 2 | ||||
| -rw-r--r-- | examples/cc_oop_demo.py | 4 | ||||
| -rw-r--r-- | examples/readme.md | 2 | ||||
| -rw-r--r-- | sec_certs/certificate/common_criteria.py | 34 | ||||
| -rw-r--r-- | sec_certs/dataset/common_criteria.py | 194 | ||||
| -rw-r--r-- | tests/data/test_cc_oop/fictional_cert.json | 4 | ||||
| -rw-r--r-- | tests/data/test_cc_oop/toy_dataset.json | 9 | ||||
| -rw-r--r-- | tests/data/test_cpe_cve/vulnerable_dataset.json | 5 | ||||
| -rw-r--r-- | tests/test_cc_heuristics.py | 4 |
10 files changed, 113 insertions, 148 deletions
@@ -89,8 +89,7 @@ def main(configpath: Optional[str], actions: List[str], inputpath: Optional[Path if not dset.state.pdfs_converted: print('Error: You want to process txt documents of certificates, but pdfs were not converted. You must use \'convert\' action first.') sys.exit(1) - dset.extract_data() - dset.compute_heuristics() + dset.analyze_certificates() if 'maintenances' in actions: if not dset.state.meta_sources_parsed: diff --git a/examples/cc_cpe_labeling.py b/examples/cc_cpe_labeling.py index b9104d92..e467cfa9 100644 --- a/examples/cc_cpe_labeling.py +++ b/examples/cc_cpe_labeling.py @@ -19,7 +19,7 @@ def main(): dset = CCDataset({}, Path('./my_debug_datset'), 'cc_full_dataset', 'Full CC dataset') dset.get_certs_from_web(to_download=True) - dset.compute_heuristics() + dset._compute_heuristics() dset.manually_verify_cpe_matches() logger.info(f'{dset.json_path} should now contain fully labeled dataset.') diff --git a/examples/cc_oop_demo.py b/examples/cc_oop_demo.py index 63be8a32..c89832fe 100644 --- a/examples/cc_oop_demo.py +++ b/examples/cc_oop_demo.py @@ -40,13 +40,13 @@ def main(): dset.convert_all_pdfs() # Extract data from txt files and update json - dset.extract_data() + dset._extract_data() # transform to pandas DataFrame df = dset.to_pandas() # Compute heuristics on the dataset - dset.compute_heuristics() + dset._compute_heuristics() # Manually verify CPE findings and compute related cves # dset.manually_verify_cpe_matches(update_json=True) diff --git a/examples/readme.md b/examples/readme.md index 0e6fd2a9..e92de44f 100644 --- a/examples/readme.md +++ b/examples/readme.md @@ -12,7 +12,7 @@ The tool contains a fuzzy procedure that attempts to map [CPE names](https://nvd ```python dset = CCDataset({}, Path('./my_debug_datset'), 'cc_full_dataset', 'Full CC dataset') dset.get_certs_from_web(to_download=True, update_json=True) -dset.compute_heuristics() +dset._compute_heuristics() dset.manually_verify_cpe_matches() ``` diff --git a/sec_certs/certificate/common_criteria.py b/sec_certs/certificate/common_criteria.py index 6ebde681..b8539c41 100644 --- a/sec_certs/certificate/common_criteria.py +++ b/sec_certs/certificate/common_criteria.py @@ -47,8 +47,8 @@ class CommonCriteriaCert(Certificate, ComplexSerializableType): @dataclass(init=False) class InternalState(ComplexSerializableType): - st_link_ok: bool - report_link_ok: bool + st_download_ok: bool + report_download_ok: bool st_convert_ok: bool report_convert_ok: bool st_extract_ok: bool @@ -59,12 +59,12 @@ class CommonCriteriaCert(Certificate, ComplexSerializableType): report_txt_path: Path errors: Optional[List[str]] - def __init__(self, st_link_ok: bool = True, report_link_ok: bool = True, + def __init__(self, st_download_ok: bool = True, report_download_ok: bool = True, st_convert_ok: bool = True, report_convert_ok: bool = True, st_extract_ok: bool = True, report_extract_ok: bool = True, errors: Optional[List[str]] = None): - self.st_link_ok = st_link_ok - self.report_link_ok = report_link_ok + self.st_download_ok = st_download_ok + self.report_download_ok = report_download_ok self.st_convert_ok = st_convert_ok self.report_convert_ok = report_convert_ok self.st_extract_ok = st_extract_ok @@ -75,6 +75,26 @@ class CommonCriteriaCert(Certificate, ComplexSerializableType): else: self.errors = errors + def report_is_ok_to_download(self, fresh: bool = True): + return True if fresh else not self.report_download_ok + + def st_is_ok_to_download(self, fresh: bool = True): + return True if fresh else not self.st_download_ok + + def report_is_ok_to_convert(self, fresh: bool = True): + return self.report_download_ok if fresh else self.report_download_ok and not self.report_convert_ok + + def st_is_ok_to_convert(self, fresh: bool = True): + return self.st_download_ok if fresh else self.st_download_ok and not self.st_convert_ok + + def report_is_ok_to_analyze(self, fresh: bool = True): + # Currently extract_ok watches two things at once: extraction of stuff from txt and heuristics + return self.report_convert_ok and self.report_extract_ok if fresh else self.report_convert_ok and not self.report_extract_ok + + def st_is_ok_to_analyze(self, fresh: bool = True): + # Currently extract_ok watches two things at once: extraction of stuff from txt and heuristics + return self.st_convert_ok and self.st_extract_ok if fresh else self.st_convert_ok and not self.st_extract_ok + @dataclass(init=False) class PdfData(ComplexSerializableType): report_metadata: Dict[str, Any] @@ -405,7 +425,7 @@ class CommonCriteriaCert(Certificate, ComplexSerializableType): if exit_code != requests.codes.ok: error_msg = f'failed to download report from {cert.report_link}, code: {exit_code}' logger.error(f'Cert dgst: {cert.dgst} ' + error_msg) - cert.state.report_link_ok = False + cert.state.report_download_ok = False cert.state.errors.append(error_msg) return cert @@ -415,7 +435,7 @@ class CommonCriteriaCert(Certificate, ComplexSerializableType): if exit_code != requests.codes.ok: error_msg = f'failed to download ST from {cert.report_link}, code: {exit_code}' logger.error(f'Cert dgst: {cert.dgst}' + error_msg) - cert.state.st_link_ok = False + cert.state.st_download_ok = False cert.state.errors.append(error_msg) return cert diff --git a/sec_certs/dataset/common_criteria.py b/sec_certs/dataset/common_criteria.py index 3add738e..5f492058 100644 --- a/sec_certs/dataset/common_criteria.py +++ b/sec_certs/dataset/common_criteria.py @@ -31,7 +31,6 @@ class CCDataset(Dataset, ComplexSerializableType): meta_sources_parsed: bool = False pdfs_downloaded: bool = False pdfs_converted: bool = False - txt_data_extracted: bool = False certs_analyzed: bool = False def __bool__(self): @@ -442,22 +441,12 @@ class CCDataset(Dataset, ComplexSerializableType): def _download_reports(self, fresh=True): self.reports_pdf_dir.mkdir(parents=True, exist_ok=True) - - if fresh is True: - certs_to_process = self.certs.values() - else: - certs_to_process = [x for x in self.certs.values() if not x.state.report_link_ok] - + certs_to_process = [x for x in self if x.state.report_is_ok_to_download(fresh)] cert_processing.process_parallel(CommonCriteriaCert.download_pdf_report, certs_to_process, config.n_threads) def _download_targets(self, fresh=True): self.targets_pdf_dir.mkdir(parents=True, exist_ok=True) - - if fresh is True: - certs_to_process = self.certs.values() - else: - certs_to_process = [x for x in self.certs.values() if not x.state.st_link_ok] - + certs_to_process = [x for x in self if x.state.report_is_ok_to_download(fresh)] cert_processing.process_parallel(CommonCriteriaCert.download_pdf_target, certs_to_process, config.n_threads) def download_all_pdfs(self, fresh: bool = True, update_json: bool = True): @@ -472,14 +461,11 @@ class CCDataset(Dataset, ComplexSerializableType): self._download_targets(fresh) if fresh is True: - # Attempt to re-download once if some files are missing - if any(filter(lambda x: not x.state.report_link_ok, self.certs.values())): - logger.info('Attempting to re-download failed report links.') - self._download_reports(False) + logger.info('Attempting to re-download failed report links.') + self._download_reports(False) - if any(filter(lambda x: not x.state.st_link_ok, self.certs.values())): - logger.info('Attempting to re-download failed security target links.') - self._download_targets(False) + logger.info('Attempting to re-download failed security target links.') + self._download_targets(False) self.state.pdfs_downloaded = True @@ -488,21 +474,12 @@ class CCDataset(Dataset, ComplexSerializableType): def _convert_reports_to_txt(self, fresh: bool = True): self.reports_txt_dir.mkdir(parents=True, exist_ok=True) - - 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 if x.state.report_is_ok_to_convert(fresh)] cert_processing.process_parallel(CommonCriteriaCert.convert_report_pdf, certs_to_process, config.n_threads) def _convert_targets_to_txt(self, fresh: bool = True): self.targets_txt_dir.mkdir(parents=True, exist_ok=True) - - if fresh is True: - certs_to_process = [x for x in self.certs.values() if x.state.st_link_ok] - else: - certs_to_process = [x for x in self.certs.values() if x.state.st_link_ok and not x.state.st_convert_ok] + certs_to_process = [x for x in self if x.state.st_is_ok_to_convert(fresh)] cert_processing.process_parallel(CommonCriteriaCert.convert_target_pdf, certs_to_process, config.n_threads) def convert_all_pdfs(self, fresh: bool = True, update_json: bool = True): @@ -517,13 +494,11 @@ class CCDataset(Dataset, ComplexSerializableType): self._convert_targets_to_txt(fresh) if fresh is True: - # Attempt to re-convert once if some files failed but downloads are ok - if any(filter(lambda x: x.state.report_link_ok and not x.state.report_convert_ok, self.certs.values())): - logger.info('Attempting to re-convert failed report pdfs') - self._convert_reports_to_txt(False) - if any(filter(lambda x: x.state.st_link_ok and not x.state.st_convert_ok, self.certs.values())): - logger.info('Attempting to re-convert failed target pdfs') - self._convert_targets_to_txt(False) + logger.info('Attempting to re-convert failed report pdfs') + self._convert_reports_to_txt(False) + + logger.info('Attempting to re-convert failed target pdfs') + self._convert_targets_to_txt(False) self.state.pdfs_converted = True @@ -531,43 +506,25 @@ class CCDataset(Dataset, ComplexSerializableType): self.to_json(self.json_path) def _extract_report_metadata(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_metadata, certs_to_process, - config.n_threads) + certs_to_process = [x for x in self if x.state.report_is_ok_to_analyze(fresh)] + cert_processing.process_parallel(CommonCriteriaCert.extract_report_pdf_metadata, certs_to_process, config.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, - config.n_threads) + certs_to_process = [x for x in self if x.state.st_is_ok_to_analyze(fresh)] + cert_processing.process_parallel(CommonCriteriaCert.extract_st_pdf_metadata, certs_to_process, config.n_threads) def extract_pdf_metadata(self, fresh: bool = True): logger.info('Extracting pdf metadata from CC dataset') self._extract_report_metadata(fresh) self._extract_targets_metadata(fresh) - def _extract_targets_frontpage(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_frontpage, certs_to_process, - config.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, - config.n_threads) + certs_to_process = [x for x in self if x.state.report_is_ok_to_analyze(fresh)] + cert_processing.process_parallel(CommonCriteriaCert.extract_report_pdf_frontpage, certs_to_process, config.n_threads) + + def _extract_targets_frontpage(self, fresh: bool = True): + certs_to_process = [x for x in self if x.state.st_is_ok_to_analyze(fresh)] + cert_processing.process_parallel(CommonCriteriaCert.extract_st_pdf_frontpage, certs_to_process, config.n_threads) def extract_pdf_frontpage(self, fresh: bool = True): logger.info('Extracting pdf frontpages from CC dataset.') @@ -575,53 +532,34 @@ class CCDataset(Dataset, ComplexSerializableType): self._extract_targets_frontpage(fresh) def _extract_report_keywords(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_keywords, certs_to_process, - config.n_threads) + certs_to_process = [x for x in self if x.state.report_is_ok_to_analyze(fresh)] + cert_processing.process_parallel(CommonCriteriaCert.extract_report_pdf_keywords, certs_to_process, config.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, - config.n_threads) + certs_to_process = [x for x in self if x.state.st_is_ok_to_analyze(fresh)] + cert_processing.process_parallel(CommonCriteriaCert.extract_st_pdf_keywords, certs_to_process, config.n_threads) def extract_pdf_keywords(self, fresh: bool = True): logger.info('Extracting pdf keywords from CC dataset.') self._extract_report_keywords(fresh) self._extract_targets_keywords(fresh) - def extract_data(self, fresh: bool = True, update_json: bool = True): - if self.state.pdfs_converted is False: - logger.info('Attempting to extract data from txt while not having the pdf->txt conversion done. Returning.') - return - + def _extract_data(self, fresh: bool = True): logger.info('Extracting various stuff from converted txt filed from CC dataset.') self.extract_pdf_metadata(fresh) self.extract_pdf_frontpage(fresh) self.extract_pdf_keywords(fresh) if fresh is True: - if any(filter(lambda x: x.state.report_convert_ok and not x.state.report_extract_ok, self.certs.values())): - logger.info('Attempting to re-extract failed data from report txts') - self._extract_report_metadata(False) - self._extract_report_frontpage(False) - self._extract_report_keywords(False) - if any(filter(lambda x: x.state.st_convert_ok and not x.state.st_extract_ok, self.certs.values())): - logger.info('Attempting to re-extract failed data from ST txts') - self._extract_targets_metadata(False) - self._extract_targets_frontpage(False) - self._extract_targets_keywords(False) + logger.info('Attempting to re-extract failed data from report txts') + self._extract_report_metadata(False) + self._extract_report_frontpage(False) + self._extract_report_keywords(False) - self.state.txt_data_extracted = True - - if update_json is True: - self.to_json(self.json_path) + logger.info('Attempting to re-extract failed data from ST txts') + self._extract_targets_metadata(False) + self._extract_targets_frontpage(False) + self._extract_targets_keywords(False) def prepare_cpe_dataset(self, download_fresh_cpes: bool = False) -> CPEDataset: logger.info('Preparing CPE dataset.') @@ -649,37 +587,47 @@ class CCDataset(Dataset, ComplexSerializableType): return cve_dataset - def compute_heuristics(self, update_json=True, download_fresh_cpes: bool = False): - def compute_candidate_versions(): - logger.info('Computing heuristics: possible product versions in certificate name') - for cert in self: - cert.compute_heuristics_version() + def _compute_candidate_versions(self): + logger.info('Computing heuristics: possible product versions in certificate name') + for cert in self: + cert.compute_heuristics_version() - def compute_cpe_matches(cpe_dataset: CPEDataset): - logger.info('Computing heuristics: Finding CPE matches for certificates') - for cert in self: - cert.compute_heuristics_cpe_match(cpe_dataset) + def _compute_cpe_matches(self, download_fresh_cpes: bool = False): + logger.info('Computing heuristics: Finding CPE matches for certificates') + cpe_dset = self.prepare_cpe_dataset(download_fresh_cpes) + for cert in self: + cert.compute_heuristics_cpe_match(cpe_dset) - def compute_cert_labs(): - logger.info('Deriving information about laboratories involved in certification.') - for cert in self: - cert.compute_heuristics_cert_lab() + def _compute_cert_labs(self): + logger.info('Deriving information about laboratories involved in certification.') + certs_to_process = [x for x in self if x.state.report_is_ok_to_analyze()] + for cert in certs_to_process: + cert.compute_heuristics_cert_lab() - def compute_cert_ids(): - logger.info('Deriving information about certificate ids from pdf scan.') - for cert in self: - cert.compute_heuristics_cert_id() + def _compute_cert_ids(self): + logger.info('Deriving information about certificate ids from pdf scan.') + certs_to_process = [x for x in self if x.state.report_is_ok_to_analyze()] + for cert in certs_to_process: + cert.compute_heuristics_cert_id() - compute_candidate_versions() - cpe_dset = self.prepare_cpe_dataset(download_fresh_cpes) - compute_cpe_matches(cpe_dset) - compute_cert_labs() - compute_cert_ids() + def _compute_heuristics(self): + self._compute_candidate_versions() + self._compute_cpe_matches() + self._compute_cert_labs() + self._compute_cert_ids() + + def analyze_certificates(self, update_json: bool = True, fresh: bool = True): + if self.state.pdfs_converted is False: + logger.info('Attempting run analysis of txt files while not having the pdf->txt conversion done. Returning.') + return + + self._extract_data(fresh) + self._compute_heuristics() self.state.certs_analyzed = True if update_json is True: - self.to_json(self.json_path) + self.to_json() def manually_verify_cpe_matches(self, update_json=True): def verify_certs(certificates_to_verify: List[CommonCriteriaCert]): @@ -777,7 +725,7 @@ class CCDataset(Dataset, ComplexSerializableType): update_dset.set_local_paths() update_dset.download_all_pdfs() update_dset.convert_all_pdfs() - update_dset.extract_data() + update_dset._extract_data() class CCDatasetMaintenanceUpdates(CCDataset): @@ -796,7 +744,7 @@ class CCDatasetMaintenanceUpdates(CCDataset): def __iter__(self) -> CommonCriteriaMaintenanceUpdate: yield from self.certs.values() - def compute_heuristics(self, update_json=True, download_fresh_cpes: bool = False): + def _compute_heuristics(self, update_json=True, download_fresh_cpes: bool = False): raise NotImplementedError def compute_related_cves(self, download_fresh_cves: bool = False): diff --git a/tests/data/test_cc_oop/fictional_cert.json b/tests/data/test_cc_oop/fictional_cert.json index 0cf17df8..87871037 100644 --- a/tests/data/test_cc_oop/fictional_cert.json +++ b/tests/data/test_cc_oop/fictional_cert.json @@ -34,8 +34,8 @@ ], "state": { "_type": "InternalState", - "st_link_ok": true, - "report_link_ok": true, + "st_download_ok": true, + "report_download_ok": true, "st_convert_ok": true, "report_convert_ok": true, "st_extract_ok": true, diff --git a/tests/data/test_cc_oop/toy_dataset.json b/tests/data/test_cc_oop/toy_dataset.json index 71d715ee..c682fc65 100644 --- a/tests/data/test_cc_oop/toy_dataset.json +++ b/tests/data/test_cc_oop/toy_dataset.json @@ -5,7 +5,6 @@ "meta_sources_parsed": true, "pdfs_downloaded": false, "pdfs_converted": false, - "txt_data_extracted": false, "certs_analyzed": false }, "timestamp": "2020-11-16 17:04:14.770153", @@ -36,8 +35,8 @@ "maintainance_updates": [], "state": { "_type": "InternalState", - "st_link_ok": true, - "report_link_ok": true, + "st_download_ok": true, + "report_download_ok": true, "st_convert_ok": true, "report_convert_ok": true, "st_extract_ok": true, @@ -90,8 +89,8 @@ "maintainance_updates": [], "state": { "_type": "InternalState", - "st_link_ok": true, - "report_link_ok": true, + "st_download_ok": true, + "report_download_ok": true, "st_convert_ok": true, "report_convert_ok": true, "st_extract_ok": true, diff --git a/tests/data/test_cpe_cve/vulnerable_dataset.json b/tests/data/test_cpe_cve/vulnerable_dataset.json index 18d0306e..0d01e60f 100644 --- a/tests/data/test_cpe_cve/vulnerable_dataset.json +++ b/tests/data/test_cpe_cve/vulnerable_dataset.json @@ -5,7 +5,6 @@ "meta_sources_parsed": true, "pdfs_downloaded": false, "pdfs_converted": false, - "txt_data_extracted": false, "certs_analyzed": false }, "timestamp": "2021-04-16 15:05:18.386794", @@ -36,8 +35,8 @@ "maintainance_updates": [], "state": { "_type": "InternalState", - "st_link_ok": true, - "report_link_ok": true, + "st_download_ok": true, + "report_download_ok": true, "st_convert_ok": true, "report_convert_ok": true, "st_extract_ok": true, diff --git a/tests/test_cc_heuristics.py b/tests/test_cc_heuristics.py index cb1c5835..84f96776 100644 --- a/tests/test_cc_heuristics.py +++ b/tests/test_cc_heuristics.py @@ -25,8 +25,8 @@ class TestCommonCriteriaHeuristics(TestCase): cls.cc_dset.process_protection_profiles() cls.cc_dset.download_all_pdfs() cls.cc_dset.convert_all_pdfs() - cls.cc_dset.extract_data() - cls.cc_dset.compute_heuristics() + cls.cc_dset._extract_data() + cls.cc_dset._compute_heuristics() cls.cpes = [CPE("cpe:2.3:a:ibm:security_access_manager_for_enterprise_single_sign-on:8.2.2:*:*:*:*:*:*:*", "IBM Security Access Manager For Enterprise Single Sign-On 8.2.2"), |