diff options
| -rw-r--r-- | fips_oop_demo.py | 22 | ||||
| -rw-r--r-- | sec_certs/cert_rules.py | 69 | ||||
| -rw-r--r-- | sec_certs/certificate.py | 147 | ||||
| -rw-r--r-- | sec_certs/constants.py | 2 | ||||
| -rw-r--r-- | sec_certs/dataset.py | 199 | ||||
| -rw-r--r-- | sec_certs/helpers.py | 14 | ||||
| -rw-r--r-- | sec_certs/settings.yaml | 2 |
7 files changed, 259 insertions, 196 deletions
diff --git a/fips_oop_demo.py b/fips_oop_demo.py index 561549b6..7015574e 100644 --- a/fips_oop_demo.py +++ b/fips_oop_demo.py @@ -4,15 +4,18 @@ import logging import click from sec_certs.dataset import FIPSDataset, FIPSAlgorithmDataset from sec_certs.configuration import config +from sec_certs.helpers import analyze_matched_algs @click.command() @click.option('--config-file', help='Path to config file') -def main(config_file): +@click.option('--json-file', help='Path to dataset json file') +@click.option('--no-download-algs', help='don\'t download algs', is_flag=True) +def main(config_file, json_file, no_download_algs): logging.basicConfig(level=logging.INFO) start = datetime.now() # Load config - config.load(config_file) + config.load(config_file if config_file else 'sec_certs/settings.yaml') # Create empty dataset dset = FIPSDataset({}, Path('./fips_dataset'), 'sample_dataset', 'sample dataset description') @@ -21,7 +24,7 @@ def main(config_file): # dset = FIPSDataset({}, Path('./fips_test_dataset'), 'small dataset', 'small dataset for keyword testing') # Load metadata for certificates from CSV and HTML sources - dset.get_certs_from_web() + dset.get_certs_from_web(json_file=json_file, redo=True) logging.info(f'Finished parsing. Have dataset with {len(dset)} certificates.') # Dump dataset into JSON @@ -33,7 +36,7 @@ def main(config_file): dset.to_json(dset.root_dir / 'fips_full_dataset.json') logging.info("Extracting keywords now.") - dset.extract_keywords() + dset.extract_keywords(redo=True) logging.info(f'Finished extracting certificates for {len(dset.certs)} items.') logging.info("Dumping dataset again...") @@ -46,10 +49,11 @@ def main(config_file): logging.info(f"Done. Files not decoded: {not_decoded_files}") dset.to_json(dset.root_dir / 'fips_mentioned.json') logging.info("Parsing algorithms") - aset = FIPSAlgorithmDataset({}, Path('fips_dataset/web/algorithms'), 'algorithms', 'sample algs') - aset.parse_html() + if not no_download_algs: + aset = FIPSAlgorithmDataset({}, Path('fips_dataset/web/algorithms'), 'algorithms', 'sample algs') + aset.get_certs_from_web() - dset.algorithms = aset + dset.algorithms = aset logging.info("finalizing results.") @@ -59,6 +63,10 @@ def main(config_file): dset.to_json(dset.root_dir / 'fips_full_dataset.json') dset.get_dot_graph('different_new') + + data = dset.match_algs() + analyze_matched_algs(data) + end = datetime.now() logging.info(f'The computation took {(end - start)} seconds.') diff --git a/sec_certs/cert_rules.py b/sec_certs/cert_rules.py index d8749495..034dd0b9 100644 --- a/sec_certs/cert_rules.py +++ b/sec_certs/cert_rules.py @@ -332,52 +332,52 @@ rules_fips_remove_algorithm_ids = [ r"HMAC(?:-SHA)?(?:-1)?[ -]*((?:160|224|256|384|512)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[, ]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{1})", # --- SHS/A - (bits) (method) ((cert #) numbers) --- - r"SH[SA][-]*(?:160|224|256|384|512)?(?:[\s(\[]*?(?:KAT)?[\s,]*?[\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)(\d{4})(?:\)?\[#?\d+\])?", - r"SH[SA][-]*(?:160|224|256|384|512)?(?:[\s(\[]*?(?:KAT)?[\s,]*?[\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)(\d{3})(?:\)?\[#?\d+\])?", - r"SH[SA][-]*(?:160|224|256|384|512)?(?:[\s(\[]*?(?:KAT)?[\s,]*?[\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)(\d{2})(?:\)?\[#?\d+\])?", - r"SH[SA][-]*(?:160|224|256|384|512)?(?:[\s(\[]*?(?:KAT)?[\s,]*?[\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)(\d{1})(?:\)?\[#?\d+\])?", + r"SH[SA][-– ]*(?:160|224|256|384|512)?(?:[\s(\[]*?(?:KAT)?[\s,]*?[\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)(\d{4})(?:\)?\[#?\d+\])?", + r"SH[SA][-– ]*(?:160|224|256|384|512)?(?:[\s(\[]*?(?:KAT)?[\s,]*?[\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)(\d{3})(?:\)?\[#?\d+\])?", + r"SH[SA][-– ]*(?:160|224|256|384|512)?(?:[\s(\[]*?(?:KAT)?[\s,]*?[\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)(\d{2})(?:\)?\[#?\d+\])?", + r"SH[SA][-– ]*(?:160|224|256|384|512)?(?:[\s(\[]*?(?:KAT)?[\s,]*?[\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)(\d{1})(?:\)?\[#?\d+\])?", # --- RSA (bits) (method) ((cert #)) --- - r"RSA(?:[- ]*(?:512|768|1024|1280|1536|2048|3072|4096|8192)\s\(\[]*?(?:(?:KAT|Verify|PSS|\s)*?)?[\s,]*?[\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{4})", - r"RSA(?:[- ]*(?:512|768|1024|1280|1536|2048|3072|4096|8192)\s\(\[]*?(?:(?:KAT|Verify|PSS|\s)*?)?[\s,]*?[\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{3})", - r"RSA(?:[- ]*(?:512|768|1024|1280|1536|2048|3072|4096|8192)\s\(\[]*?(?:(?:KAT|Verify|PSS|\s)*?)?[\s,]*?[\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{2})", - r"RSA(?:[- ]*(?:512|768|1024|1280|1536|2048|3072|4096|8192)\s\(\[]*?(?:(?:KAT|Verify|PSS|\s)*?)?[\s,]*?[\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{1})", + r"RSA(?:[-– ]*(?:512|768|1024|1280|1536|2048|3072|4096|8192)\s\(\[]*?(?:(?:KAT|Verify|PSS|\s)*?)?[\s,]*?[\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{4})", + r"RSA(?:[-– ]*(?:512|768|1024|1280|1536|2048|3072|4096|8192)\s\(\[]*?(?:(?:KAT|Verify|PSS|\s)*?)?[\s,]*?[\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{3})", + r"RSA(?:[-– ]*(?:512|768|1024|1280|1536|2048|3072|4096|8192)\s\(\[]*?(?:(?:KAT|Verify|PSS|\s)*?)?[\s,]*?[\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{2})", + r"RSA(?:[-– ]*(?:512|768|1024|1280|1536|2048|3072|4096|8192)\s\(\[]*?(?:(?:KAT|Verify|PSS|\s)*?)?[\s,]*?[\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{1})", # --- RSA (SSA) (PKCS) (version) (#) --- - r"(?:RSA)?[- ]?(?:SSA)?[- ]?PKCS\s?#?\d(?:-[Vv]1_5| [Vv]1[-_]5)?\s?(\d{4})?", - r"(?:RSA)?[- ]?(?:SSA)?[- ]?PKCS\s?#?\d(?:-[Vv]1_5| [Vv]1[-_]5)?\s?(\d{3})?", - r"(?:RSA)?[- ]?(?:SSA)?[- ]?PKCS\s?#?\d(?:-[Vv]1_5| [Vv]1[-_]5)?\s?(\d{2})?", - r"(?:RSA)?[- ]?(?:SSA)?[- ]?PKCS\s?#?\d(?:-[Vv]1_5| [Vv]1[-_]5)?\s?(\d{1})?", + r"(?:RSA)?[-– ]?(?:SSA)?[- ]?PKCS\s?#?\d(?:-[Vv]1_5| [Vv]1[-_]5)?\s?(\d{4})?", + r"(?:RSA)?[-– ]?(?:SSA)?[- ]?PKCS\s?#?\d(?:-[Vv]1_5| [Vv]1[-_]5)?\s?(\d{3})?", + r"(?:RSA)?[-– ]?(?:SSA)?[- ]?PKCS\s?#?\d(?:-[Vv]1_5| [Vv]1[-_]5)?\s?(\d{2})?", + r"(?:RSA)?[-– ]?(?:SSA)?[- ]?PKCS\s?#?\d(?:-[Vv]1_5| [Vv]1[-_]5)?\s?(\d{1})?", # --- AES (bits) (method) ((cert #)) --- - r"AES[- ]*((?:128|192|256|)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT|CMAC|CTR|GCM|IV)*?[,\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{4})(?:\)?\[#?\d+\])?", - r"AES[- ]*((?:128|192|256|)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT|CMAC|CTR|GCM|IV)*?[,\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{3})(?:\)?\[#?\d+\])?", - r"AES[- ]*((?:128|192|256|)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT|CMAC|CTR|GCM|IV)*?[,\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{2})(?:\)?\[#?\d+\])?", - r"AES[- ]*((?:128|192|256|)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT|CMAC|CTR|GCM|IV)*?[,\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{1})(?:\)?\[#?\d+\])?", + r"AES[-– ]*((?:128|192|256|)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT|CMAC|CTR|GCM|IV)*?[,\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{4})(?:\)?\[#?\d+\])?", + r"AES[-– ]*((?:128|192|256|)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT|CMAC|CTR|GCM|IV)*?[,\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{3})(?:\)?\[#?\d+\])?", + r"AES[-– ]*((?:128|192|256|)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT|CMAC|CTR|GCM|IV)*?[,\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{2})(?:\)?\[#?\d+\])?", + r"AES[-– ]*((?:128|192|256|)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT|CMAC|CTR|GCM|IV)*?[,\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{1})(?:\)?\[#?\d+\])?", # --- Diffie Helman (CVL) ((cert #)) --- - r"Diffie[- ]*Hellman[,\s(\[]*?(?:CVL|\s)*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?\s?(\d{4})", - r"Diffie[- ]*Hellman[,\s(\[]*?(?:CVL|\s)*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?\s?(\d{3})", - r"Diffie[- ]*Hellman[,\s(\[]*?(?:CVL|\s)*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?\s?(\d{2})", - r"Diffie[- ]*Hellman[,\s(\[]*?(?:CVL|\s)*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?\s?(\d{1})", + r"Diffie[-– ]*Hellman[,\s(\[]*?(?:CVL|\s)*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?\s?(\d{4})", + r"Diffie[-– ]*Hellman[,\s(\[]*?(?:CVL|\s)*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?\s?(\d{3})", + r"Diffie[-– ]*Hellman[,\s(\[]*?(?:CVL|\s)*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?\s?(\d{2})", + r"Diffie[-– ]*Hellman[,\s(\[]*?(?:CVL|\s)*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?\s?(\d{1})", # --- DRBG (bits) (method) (cert #) --- - r"DRBG[ -]*((?:160|224|256|384|512)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[,\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{4})", - r"DRBG[ -]*((?:160|224|256|384|512)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[,\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{3})", - r"DRBG[ -]*((?:160|224|256|384|512)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[,\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{2})", - r"DRBG[ -]*((?:160|224|256|384|512)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[,\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{1})", + r"DRBG[ –-]*((?:160|224|256|384|512)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[,\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{4})", + r"DRBG[ –-]*((?:160|224|256|384|512)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[,\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{3})", + r"DRBG[ –-]*((?:160|224|256|384|512)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[,\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{2})", + r"DRBG[ –-]*((?:160|224|256|384|512)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[,\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{1})", # --- DES (bits) (method) (cert #) - r"DES[ -]*((?:160|224|256|384|512)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[,\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{4})", - r"DES[ -]*((?:160|224|256|384|512)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[,\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{3})", - r"DES[ -]*((?:160|224|256|384|512)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[,\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{2})", - r"DES[ -]*((?:160|224|256|384|512)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[,\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{1})", + r"DES[ –-]*((?:160|224|256|384|512)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[,\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{4})", + r"DES[ –-]*((?:160|224|256|384|512)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[,\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{3})", + r"DES[ –-]*((?:160|224|256|384|512)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[,\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{2})", + r"DES[ –-]*((?:160|224|256|384|512)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[,\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{1})", # --- DSA (bits) (method) (cert #) - r"DSA[ -]*((?:160|224|256|384|512)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[,\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{4})", - r"DSA[ -]*((?:160|224|256|384|512)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[,\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{3})", - r"DSA[ -]*((?:160|224|256|384|512)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[,\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{2})", - r"DSA[ -]*((?:160|224|256|384|512)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[,\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{1})", + r"DSA[ –-]*((?:160|224|256|384|512)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[,\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{4})", + r"DSA[ –-]*((?:160|224|256|384|512)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[,\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{3})", + r"DSA[ –-]*((?:160|224|256|384|512)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[,\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{2})", + r"DSA[ –-]*((?:160|224|256|384|512)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[,\s(\[]*?(?:#|cert\.?|certificate|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{1})", # --- platforms (#)+ - this is used in modification history --- r"[Pp]latforms? #\d+(?:#\d+|,| |-|and)*[^\n]*", @@ -399,8 +399,10 @@ rules_fips_remove_algorithm_ids = [ # --- PKCS (#) --- r"PKCS[\s]?#?\d+", - r"PKSC[\s]?#?\d+", # typo, #625 + r"PKSC[\s]?#?\d+" # typo, #625 +] +rules_fips_to_remove = [ # --- random words found --- r"Survey #192", # why would they get an address like this /o\ cert 2079 r"[Ss]lot #\d", # a card slot, #2069 @@ -501,6 +503,7 @@ rules.update(common_rules) # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ fips_rules = {} fips_rules['rules_fips_algorithms'] = rules_fips_remove_algorithm_ids +fips_rules['rules_to_remove'] = rules_fips_to_remove fips_rules['rules_security_level'] = rules_fips_security_level fips_rules['rules_cert_id'] = rules_fips_cert fips_common_rules = copy.deepcopy(common_rules) # make separate copy not to process cc rules by fips's re.compile diff --git a/sec_certs/certificate.py b/sec_certs/certificate.py index 61e0003a..7a85facc 100644 --- a/sec_certs/certificate.py +++ b/sec_certs/certificate.py @@ -11,7 +11,7 @@ from dateutil import parser from abc import ABC, abstractmethod from bs4 import Tag, BeautifulSoup, NavigableString -from typing import Union, Optional, List, Dict, ClassVar, TypeVar, Type, Tuple, Pattern +from typing import Union, Optional, List, Dict, ClassVar, TypeVar, Type, Tuple, Pattern, Set from tabula import read_pdf @@ -87,7 +87,7 @@ class FIPSCertificate(Certificate, ComplexSerializableType): file_status: Optional[bool] txt_state: bool - @dataclass(eq=True, frozen=True) + @dataclass(eq=True) class Algorithm(ComplexSerializableType): cert_id: str vendor: str @@ -146,8 +146,9 @@ class FIPSCertificate(Certificate, ComplexSerializableType): product_url: Optional[str] def __post_init__(self): - self.date_validation = [parser.parse(x) for x in self.date_validation] if self.date_validation else None - self.date_sunset = parser.parse(self.date_sunset) if self.date_sunset else None + 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): @@ -228,6 +229,10 @@ class FIPSCertificate(Certificate, ComplexSerializableType): if exit_code != requests.codes.ok: logger.error(f'Failed to download security policy from {cert[0]}, code: {exit_code}') + @classmethod + def from_dict(cls, dct: dict): + return cls(*tuple(dct.values())) + def __init__(self, cert_id: str, web_scan: 'FIPSCertificate.WebScan', pdf_scan: 'FIPSCertificate.PdfScan', @@ -241,19 +246,12 @@ class FIPSCertificate(Certificate, ComplexSerializableType): self.state = state @staticmethod - def download_html_page(cert: Tuple[str, Path]) -> None: + def download_html_page(cert: Tuple[str, Path]) -> Optional[Tuple[str, Path]]: exit_code = helpers.download_file(*cert) if exit_code != requests.codes.ok: logger.error(f'Failed to download html page from {cert[0]}, code: {exit_code}') - - @staticmethod - def extract_filename(file: str) -> str: - """ - Extracts filename from path - @param file: UN*X path - :return: filename without last extension - """ - return os.path.splitext(os.path.basename(file))[0] + return cert + return None @staticmethod def initialize_dictionary() -> Dict: @@ -294,9 +292,11 @@ class FIPSCertificate(Certificate, ComplexSerializableType): :return: List containing one element - dictionary with all parsed algorithm cert ids """ set_items = set() - for m in re.finditer( - rf"(?:#{'?' if in_pdf else '[CcAa]?'}\s?|(?:Cert{'' if in_pdf else '?'})\.?[^. ]*?\s?)(?:[CcAa]\s)?(?P<id>\d+)", - current_text): + if in_pdf: + reg = r"(?:#?\s?|(?:Cert)\.?[^. ]*?\s?)(?:[CcAa]\s)?(?P<id>[CcAa]? ?\d+)" + else: + reg = r"(?:#[CcAa]?\s?|(?:Cert)\.?[^. ]*?\s?)(?:[CcAa]\s)?(?P<id>\d+)" + for m in re.finditer(reg, current_text): set_items.add(m.group()) return [{"Certificate": list(set_items)}] @@ -394,7 +394,8 @@ class FIPSCertificate(Certificate, ComplexSerializableType): items['embodiment'] = items['embodiment'].lower().replace('-', ' ').replace('stand alone', 'standalone').title() @classmethod - def html_from_file(cls, file: Path, state: State, initialized: 'FIPSCertificate' = None) -> 'FIPSCertificate': + def html_from_file(cls, file: Path, state: State, initialized: 'FIPSCertificate' = None, + redo: bool = False) -> 'FIPSCertificate': pairs = { 'Module Name': 'module_name', 'Standard': 'standard', @@ -422,7 +423,6 @@ class FIPSCertificate(Certificate, ComplexSerializableType): if not initialized: items_found = FIPSCertificate.initialize_dictionary() items_found['cert_id'] = file.stem - else: items_found = initialized.web_scan.__dict__ items_found['cert_id'] = initialized.cert_id @@ -434,6 +434,10 @@ class FIPSCertificate(Certificate, ComplexSerializableType): state.txt_state = initialized.state.txt_state initialized.processed.connections = [] + if redo: + items_found = FIPSCertificate.initialize_dictionary() + items_found['cert_id'] = file.stem + text = extract_certificates.load_cert_html_file(file) soup = BeautifulSoup(text, 'html.parser') for div in soup.find_all('div', class_='row padrow'): @@ -449,37 +453,6 @@ class FIPSCertificate(Certificate, ComplexSerializableType): if div.find('h4', class_='panel-title').text == 'Related Files': FIPSCertificate.parse_related_files(div, items_found) - if initialized: - new_algs = [] - not_defined = set() - for i, alg in enumerate(items_found['algorithms']): - if 'Name' not in alg: - for cert_id in alg['Certificate']: - not_defined.add(cert_id) - continue - - for pair in range(i + 1, len(items_found['algorithms'])): - if 'Name' in items_found['algorithms'][pair] \ - and alg['Name'] == items_found['algorithms'][pair]['Name']: - entry = {'Name': alg['Name'], 'Certificate': - list(set([x for x in alg['Certificate']]) - | set(items_found['algorithms'][pair]['Certificate'])), - 'Raw': items_found['algorithms'][pair]['Raw'], - 'Links': items_found['algorithms'][pair]['Links']} - if entry not in new_algs: - new_algs.append(entry) - for entry in new_algs: - if entry['Name'] == 'Not Defined': - entry['Certificate'] = list(set(entry['Certificate'] | not_defined)) - break - else: - new_algs.append({'Name': 'Not Defined', 'Certificate': list(not_defined)}) - - - new_algs = [x for x in new_algs if x != {'Certificate': []}] - - items_found['algorithms'] = new_algs - FIPSCertificate.normalize(items_found) return FIPSCertificate(items_found['cert_id'], @@ -507,7 +480,7 @@ class FIPSCertificate(Certificate, ComplexSerializableType): items_found['certificate_www'] if 'certificate_www' in items_found else None, items_found['hw_versions'] if 'hw_versions' in items_found else None, items_found['fw_versions'] if 'fw_versions' in items_found else None, - items_found['revoked_reason'] if 'revoked_reason' in items_found else None, + items_found['revoked_reason'] if 'revoked_reason' in items_found else None, items_found['revoked_link'] if 'revoked_link' in items_found else None, items_found['sw_versions'] if 'sw_versions' in items_found else None, items_found['product_url']) if 'product_url' in items_found else None, @@ -516,7 +489,7 @@ class FIPSCertificate(Certificate, ComplexSerializableType): {} if not initialized else initialized.pdf_scan.keywords, [] if not initialized else initialized.pdf_scan.algorithms ), - FIPSCertificate.Processed(None, {}, []) if not initialized else initialized.processed, + FIPSCertificate.Processed(None, {}, []), state ) @@ -556,6 +529,31 @@ class FIPSCertificate(Certificate, ComplexSerializableType): return items_found, cert @staticmethod + def match_web_algs_to_pdf(cert: 'FIPSCertificate') -> int: + algs_vals = list(cert.pdf_scan.keywords['rules_fips_algorithms'].values()) + table_vals = [x['Certificate'] for x in cert.pdf_scan.algorithms] + tables = [x.strip() for y in table_vals for x in y] + iterable = [l for x in algs_vals for l in list(x.keys())] + iterable += tables + all_algorithms = set() + for x in iterable: + if '#' in x: + # erase everything until "#" included and take digits + all_algorithms.add(''.join(filter(str.isdigit, x[x.index('#') + 1:]))) + else: + all_algorithms.add(''.join(filter(str.isdigit, x))) + not_found = [] + for alg_list in (a['Certificate'] for a in cert.web_scan.algorithms): + for web_alg in alg_list: + if ''.join(filter(str.isdigit, web_alg)) not in all_algorithms: + not_found.append(web_alg) + logger.error( + f"For cert {cert.dgst}:\n\tNOT FOUND: {len(not_found)}\n" + f"\tFOUND: {sum([len(a['Certificate']) for a in cert.web_scan.algorithms]) - len(not_found)}") + logger.error(f"Not found: {not_found}") + return len(not_found) + + @staticmethod def remove_platforms(text_to_parse: str): pat = re.compile(r"(?:modification|revision|change) history\n[\s\S]*? ", re.IGNORECASE) for match in pat.finditer(text_to_parse): @@ -654,15 +652,6 @@ class FIPSCertificate(Certificate, ComplexSerializableType): if match == '': continue - certs = [x['Certificate'] for x in algorithms] - - match_cert_id = ''.join(filter(str.isdigit, match)) - - for fips_cert in certs: - for actual_cert in fips_cert: - if actual_cert != '' and match_cert_id == ''.join(filter(str.isdigit, actual_cert)): - continue - if match not in items_found[rule.pattern]: items_found[rule.pattern][match] = {} items_found[rule.pattern][match][constants.TAG_MATCH_COUNTER] = 0 @@ -706,6 +695,12 @@ class FIPSCertificate(Certificate, ComplexSerializableType): lst += FIPSCertificate.extract_algorithm_certificates(df.to_string(index=False)) return True, cert, lst + def _create_alg_set(self) -> Set: + result = set() + for alg in self.web_scan.algorithms: + result.update(cert for cert in alg['Certificate']) + return result + def remove_algorithms(self): self.state.file_status = True if not self.pdf_scan.keywords: @@ -716,10 +711,15 @@ class FIPSCertificate(Certificate, ComplexSerializableType): for item in self.web_scan.mentioned_certs: self.processed.keywords['rules_cert_id'].update(item) + alg_set = self._create_alg_set() + for rule in self.processed.keywords['rules_cert_id']: to_pop = set() rr = re.compile(rule) for cert in self.processed.keywords['rules_cert_id'][rule]: + if cert in alg_set: + to_pop.add(cert) + continue for alg in self.processed.keywords['rules_fips_algorithms']: for found in self.processed.keywords['rules_fips_algorithms'][alg]: if rr.search(found) \ @@ -847,9 +847,10 @@ class CommonCriteriaCert(Certificate, ComplexSerializableType): report_keywords: Dict[str, str] st_keywords: Dict[str, str] - def __init__(self, report_metadata: Optional[Dict[str, str]] = None, st_metadata: Optional[Dict[str, str]] = None, + def __init__(self, report_metadata: Optional[Dict[str, str]] = None, + st_metadata: Optional[Dict[str, str]] = None, report_frontpage: Optional[Dict[str, str]] = None, st_frontpage: Optional[Dict[str, str]] = None, - report_keywords: Optional[Dict[str,str]] = None, st_keywords: Optional[Dict[str, str]] = None): + report_keywords: Optional[Dict[str, str]] = None, st_keywords: Optional[Dict[str, str]] = None): self.report_metadata = report_metadata self.st_metadata = st_metadata self.report_frontpage = report_frontpage @@ -858,8 +859,10 @@ class CommonCriteriaCert(Certificate, ComplexSerializableType): self.st_keywords = st_keywords def to_dict(self): - return {'report_metadata': self.report_metadata, 'st_metadata': self.st_metadata, 'report_frontpage': self.report_frontpage, - 'st_frontpage': self.st_frontpage, 'report_keywords': self.report_keywords, 'st_keywords': self.st_keywords} + return {'report_metadata': self.report_metadata, 'st_metadata': self.st_metadata, + 'report_frontpage': self.report_frontpage, + 'st_frontpage': self.st_frontpage, 'report_keywords': self.report_keywords, + 'st_keywords': self.st_keywords} @classmethod def from_dict(cls, dct: Dict[str, bool]): @@ -868,7 +871,7 @@ class CommonCriteriaCert(Certificate, ComplexSerializableType): pandas_serialization_vars = ['dgst', 'name', 'manufacturer', 'scheme', 'security_level', 'not_valid_before', 'not_valid_after', 'report_link', 'st_link', 'src', 'manufacturer_web'] - def __init__(self, status:str, category: str, name: str, manufacturer: str, scheme: str, + def __init__(self, status: str, category: str, name: str, manufacturer: str, scheme: str, security_level: Union[str, set], not_valid_before: date, not_valid_after: date, report_link: str, st_link: str, src: str, cert_link: Optional[str], manufacturer_web: Optional[str], @@ -898,7 +901,7 @@ class CommonCriteriaCert(Certificate, ComplexSerializableType): if state is None: state = self.InternalState() self.state = state - + if pdf_data is None: pdf_data = self.PdfData() self.pdf_data = pdf_data @@ -1062,7 +1065,8 @@ class CommonCriteriaCert(Certificate, ComplexSerializableType): maintainances = _get_maintainance_updates( maintainance_div) if maintainance_div else set() - return cls(status, category, name, manufacturer, scheme, security_level, not_valid_before, not_valid_after, report_link, + return cls(status, category, name, manufacturer, scheme, security_level, not_valid_before, not_valid_after, + report_link, st_link, 'html', cert_link, manufacturer_web, protection_profiles, maintainances, None, None, None) def set_local_paths(self, @@ -1157,8 +1161,10 @@ class CommonCriteriaCert(Certificate, ComplexSerializableType): @staticmethod def extract_report_pdf_frontpage(cert: 'CommonCriteriaCert') -> 'CommonCriteriaCert': cert.pdf_data.report_frontpage = dict() - response_bsi, cert.pdf_data.report_frontpage['bsi'] = helpers.search_only_headers_bsi(cert.state.report_txt_path) - response_anssi, cert.pdf_data.report_frontpage['anssi'] = helpers.search_only_headers_anssi(cert.state.report_txt_path) + response_bsi, cert.pdf_data.report_frontpage['bsi'] = helpers.search_only_headers_bsi( + cert.state.report_txt_path) + response_anssi, cert.pdf_data.report_frontpage['anssi'] = helpers.search_only_headers_anssi( + cert.state.report_txt_path) if response_anssi != constants.RETURNCODE_OK: cert.state.report_extract_ok = False @@ -1183,4 +1189,3 @@ class CommonCriteriaCert(Certificate, ComplexSerializableType): cert.state.st_extract_ok = False cert.state.errors.append(response) return cert - diff --git a/sec_certs/constants.py b/sec_certs/constants.py index 18c9998c..adf30869 100644 --- a/sec_certs/constants.py +++ b/sec_certs/constants.py @@ -4,7 +4,7 @@ N_THREADS = 8 RESPONSE_OK = 200 RETURNCODE_OK = 'ok' RETURNCODE_NOK = 'nok' -REQUEST_TIMEOUT = 5 +REQUEST_TIMEOUT = 10 MIN_CORRECT_CERT_SIZE = 5000 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 + - ' ' + - self.certs[current_key].web_scan.vendor + - ' ' + - (self.certs[current_key].web_scan.module_name - if self.certs[current_key].web_scan.module_name else '')) + label=current_key + + ' ' + + self.certs[current_key].web_scan.vendor + + ' ' + + (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): diff --git a/sec_certs/helpers.py b/sec_certs/helpers.py index 01729891..a8711f73 100644 --- a/sec_certs/helpers.py +++ b/sec_certs/helpers.py @@ -16,6 +16,7 @@ import subprocess import sec_certs.constants as constants from enum import Enum from rapidfuzz import process, fuzz +import matplotlib.pyplot as plt from PyPDF2 import PdfFileReader @@ -118,7 +119,7 @@ def parse_list_of_tables(txt: str) -> Set[str]: :param txt: chunk of text :return: set of all pages mentioning algorithm table """ - rr = re.compile(r"^.+?(?:[Ff]unction|[Aa]lgorithm).+?(?P<page_num>\d+)$", re.MULTILINE) + rr = re.compile(r"^.+?(?:[Ff]unction|[Aa]lgorithm|[Ss]ecurity [Ff]unctions?).+?(?P<page_num>\d+)$", re.MULTILINE) pages = set() for m in rr.finditer(txt): pages.add(m.group('page_num')) @@ -494,4 +495,13 @@ def match_certs(certs: Dict[str, str], cpes: List[str]): results = {} for dgst, cert_name in certs.items(): results[dgst] = process.extract(cert_name, cpes, scorer=fuzz.token_set_ratio, limit=10) - return results
\ No newline at end of file + return results + +def analyze_matched_algs(data: Dict): + pd_data = pd.Series(data) + pd_data.hist(bins=50) + plt.show() + + sorted_data = pd_data.value_counts(ascending=True) + + logging.info(sorted_data.where(sorted_data > 1).dropna())
\ No newline at end of file diff --git a/sec_certs/settings.yaml b/sec_certs/settings.yaml index a35a18de..0c8b130f 100644 --- a/sec_certs/settings.yaml +++ b/sec_certs/settings.yaml @@ -2,7 +2,7 @@ smallest_certificate_id_to_connect: description: During validation we don't connect certificates with number lower than _this_ to connections - value: 60 + value: 40 year_difference_between_validations: description: During validation we don't connect certificates with validation dates difference higher than _this_ |