diff options
Diffstat (limited to 'sec_certs')
| -rw-r--r-- | sec_certs/config/configuration.py | 8 | ||||
| -rw-r--r-- | sec_certs/dataset/common_criteria.py | 158 | ||||
| -rw-r--r-- | sec_certs/dataset/cpe.py | 47 | ||||
| -rw-r--r-- | sec_certs/dataset/dataset.py | 8 | ||||
| -rw-r--r-- | sec_certs/dataset/fips.py | 176 | ||||
| -rw-r--r-- | sec_certs/model/cpe_matching.py | 95 | ||||
| -rw-r--r-- | sec_certs/model/dependency_finder.py | 26 | ||||
| -rw-r--r-- | sec_certs/model/dependency_vulnerability_finder.py | 23 | ||||
| -rw-r--r-- | sec_certs/model/sar_transformer.py | 39 | ||||
| -rw-r--r-- | sec_certs/pandas_helpers.py | 46 | ||||
| -rw-r--r-- | sec_certs/sample/common_criteria.py | 186 | ||||
| -rw-r--r-- | sec_certs/sample/fips.py | 161 | ||||
| -rw-r--r-- | sec_certs/serialization/json.py | 2 |
13 files changed, 728 insertions, 247 deletions
diff --git a/sec_certs/config/configuration.py b/sec_certs/config/configuration.py index 8560529c..6a9a5880 100644 --- a/sec_certs/config/configuration.py +++ b/sec_certs/config/configuration.py @@ -1,6 +1,6 @@ import json from pathlib import Path -from typing import Any, Union +from typing import Any, Optional, Union import jsonschema import yaml @@ -30,6 +30,12 @@ class Configuration(object): return res["value"] return object.__getattribute__(self, key) + def get_desription(self, key: str) -> Optional[str]: + res = object.__getattribute__(self, key) + if isinstance(res, dict) and "description" in res: + return res["description"] + return None + DEFAULT_CONFIG_PATH = Path(__file__).parent / "settings.yaml" config = Configuration() diff --git a/sec_certs/dataset/common_criteria.py b/sec_certs/dataset/common_criteria.py index 7bdb5ae1..8ea091db 100644 --- a/sec_certs/dataset/common_criteria.py +++ b/sec_certs/dataset/common_criteria.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import copy import itertools import json @@ -28,6 +30,11 @@ from sec_certs.serialization.json import ComplexSerializableType, CustomJSONDeco class CCDataset(Dataset[CommonCriteriaCert], ComplexSerializableType): + """ + Class that holds CommonCriteriaCert. Serializable into json, pandas, dictionary. Conveys basic certificate manipulations + and dataset transformations. Many private methods that perform internal operations, feel free to exploit them. + """ + @dataclass class DatasetInternalState(ComplexSerializableType): meta_sources_parsed: bool = False @@ -56,9 +63,15 @@ class CCDataset(Dataset[CommonCriteriaCert], ComplexSerializableType): return copy.deepcopy(self) def to_dict(self) -> Dict[str, Any]: + """ + Returns self serialized into dictionary + """ return {**{"state": self.state}, **super().to_dict()} def to_pandas(self) -> pd.DataFrame: + """ + Return self serialized into pandas DataFrame + """ df = pd.DataFrame([x.pandas_tuple for x in self.certs.values()], columns=CommonCriteriaCert.pandas_columns) df = df.set_index("dgst") @@ -79,7 +92,10 @@ class CCDataset(Dataset[CommonCriteriaCert], ComplexSerializableType): return df @classmethod - def from_dict(cls: Type["CCDataset"], dct: Dict[str, Any]) -> "CCDataset": + def from_dict(cls: Type[CCDataset], dct: Dict[str, Any]) -> "CCDataset": + """ + Reconstructs the CCDataset object from a dictionary + """ dset = super().from_dict(dct) dset.state = copy.deepcopy(dct["state"]) return dset @@ -89,14 +105,14 @@ class CCDataset(Dataset[CommonCriteriaCert], ComplexSerializableType): def root_dir(self, new_dir: Union[str, Path]) -> None: old_dset = copy.deepcopy(self) Dataset.root_dir.fset(self, new_dir) # type: ignore - self.set_local_paths() + self._set_local_paths() if self.state and old_dset.root_dir != Path(".."): logger.info(f"Changing root dir of partially processed dataset. All contents will get copied to {new_dir}") - self.copy_dataset_contents(old_dset) + self._copy_dataset_contents(old_dset) self.to_json() - def copy_dataset_contents(self, old_dset: "CCDataset") -> None: + def _copy_dataset_contents(self, old_dset: "CCDataset") -> None: if old_dset.state.meta_sources_parsed: try: shutil.copytree(old_dset.web_dir, self.web_dir) @@ -115,42 +131,72 @@ class CCDataset(Dataset[CommonCriteriaCert], ComplexSerializableType): @property def certs_dir(self) -> Path: + """ + Returns directory that holds files associated with certificates + """ return self.root_dir / "certs" @property def reports_dir(self) -> Path: + """ + Returns directory that holds files associated with certification reports + """ return self.certs_dir / "reports" @property def reports_pdf_dir(self) -> Path: + """ + Returns directory that holds PDFs associated with certification reports + """ return self.reports_dir / "pdf" @property def reports_txt_dir(self) -> Path: + """ + Returns directory that holds TXTs associated with certification reports + """ return self.reports_dir / "txt" @property def targets_dir(self) -> Path: + """ + Returns directory that holds files associated with security targets + """ return self.certs_dir / "targets" @property def targets_pdf_dir(self) -> Path: + """ + Returns directory that holds PDFs associated with security targets + """ return self.targets_dir / "pdf" @property def targets_txt_dir(self) -> Path: + """ + Returns directory that holds TXTs associated with security targets + """ return self.targets_dir / "txt" @property def pp_dataset_path(self) -> Path: + """ + Returns directory that holds files associated with Protection profiles + """ return self.auxillary_datasets_dir / "pp_dataset.json" @property def mu_dataset_path(self) -> Path: + """ + Returns directory that holds dataset of maintenance updates + """ return self.certs_dir / "maintenances" @property def mu_dataset(self) -> "CCDatasetMaintenanceUpdates": + """ + Returns object of dataset of maintenance updates if it exists, raises otherwise + """ if not self.mu_dataset_path.exists(): raise ValueError("The dataset with maintenance updates does not exist") @@ -176,22 +222,42 @@ class CCDataset(Dataset[CommonCriteriaCert], ComplexSerializableType): @property def active_html_tuples(self) -> List[Tuple[str, Path]]: + """ + Returns List Tuple[str, Path] where first element is name of html file and second element is its Path. + The files correspond to html files parsed from CC website that list all *active* certificates. + """ return [(x, self.web_dir / y) for y, x in self.HTML_PRODUCTS_URL.items() if "active" in y] @property def archived_html_tuples(self) -> List[Tuple[str, Path]]: + """ + Returns List Tuple[str, Path] where first element is name of html file and second element is its Path. + The files correspond to html files parsed from CC website that list all *archived* certificates. + """ return [(x, self.web_dir / y) for y, x in self.HTML_PRODUCTS_URL.items() if "archived" in y] @property def active_csv_tuples(self) -> List[Tuple[str, Path]]: + """ + Returns List Tuple[str, Path] where first element is name of csv file and second element is its Path. + The files correspond to csv files downloaded from CC website that list all *active* certificates. + """ return [(x, self.web_dir / y) for y, x in self.CSV_PRODUCTS_URL.items() if "active" in y] @property def archived_csv_tuples(self) -> List[Tuple[str, Path]]: + """ + Returns List Tuple[str, Path] where first element is name of csv file and second element is its Path. + The files correspond to csv files downloaded from CC website that list all *archived* certificates. + """ return [(x, self.web_dir / y) for y, x in self.CSV_PRODUCTS_URL.items() if "archived" in y] @classmethod def from_web_latest(cls) -> "CCDataset": + """ + Fetches the fresh snapshot of CCDataset from seccerts.org + :return CCDataset: returns the CCDataset object downloaded from seccerts.org + """ with tempfile.TemporaryDirectory() as tmp_dir: dset_path = Path(tmp_dir) / "cc_latest_dataset.json" helpers.download_file( @@ -200,7 +266,7 @@ class CCDataset(Dataset[CommonCriteriaCert], ComplexSerializableType): return cls.from_json(dset_path) @property - def all_cert_ids(self): + def _all_cert_ids(self): all_cert_ids = set() for cert_obj in self: @@ -220,7 +286,7 @@ class CCDataset(Dataset[CommonCriteriaCert], ComplexSerializableType): return all_cert_ids - def set_local_paths(self): + def _set_local_paths(self): for cert in self: cert.set_local_paths(self.reports_pdf_dir, self.targets_pdf_dir, self.reports_txt_dir, self.targets_txt_dir) @@ -237,7 +303,7 @@ class CCDataset(Dataset[CommonCriteriaCert], ComplexSerializableType): logger.info(f"Added {len(new_certs)} new and merged further {len(certs_to_merge)} certificates to the dataset.") - def download_csv_html_resources(self, get_active: bool = True, get_archived: bool = True) -> None: + def _download_csv_html_resources(self, get_active: bool = True, get_archived: bool = True) -> None: self.web_dir.mkdir(parents=True, exist_ok=True) html_items = [] @@ -258,6 +324,14 @@ class CCDataset(Dataset[CommonCriteriaCert], ComplexSerializableType): @serialize def process_protection_profiles(self, to_download: bool = True, keep_metadata: bool = True) -> None: + """ + Downloads new snapshot of dataset with processed protection profiles (if it doesn't exist) and links PPs + with certificates within self. Assigns PPs to all certificates + + :param bool to_download: If dataset should be downloaded or fetched from json, defaults to True + :param bool keep_metadata: If json related to the PP dataset should be kept on drive, defaults to True + :raises RuntimeError: When building of PPDataset fails + """ logger.info("Processing protection profiles.") constructor: Dict[bool, Callable[..., ProtectionProfileDataset]] = { True: ProtectionProfileDataset.from_web, @@ -280,10 +354,16 @@ class CCDataset(Dataset[CommonCriteriaCert], ComplexSerializableType): self, to_download: bool = True, keep_metadata: bool = True, get_active: bool = True, get_archived: bool = True ) -> None: """ - Parses all metadata about certificates + Downloads CSV and HTML files that hold lists of certificates from common criteria website. Parses these files + and constructs CommonCriteriaCert objects, fills the dataset with those. + + :param bool to_download: If CSV and HTML files shall be downloaded (or existing files utilized), defaults to True + :param bool keep_metadata: If CSV and HTML files shall be kept on disk after download, defaults to True + :param bool get_active: If active certificates shall be parsed, defaults to True + :param bool get_archived: If archived certificates shall be parsed, defaults to True """ if to_download is True: - self.download_csv_html_resources(get_active, get_archived) + self._download_csv_html_resources(get_active, get_archived) logger.info("Adding CSV certificates to CommonCriteria dataset.") csv_certs = self._get_all_certs_from_csv(get_active, get_archived) @@ -299,7 +379,7 @@ class CCDataset(Dataset[CommonCriteriaCert], ComplexSerializableType): if not keep_metadata: shutil.rmtree(self.web_dir) - self.set_local_paths() + self._set_local_paths() self.state.meta_sources_parsed = True def _get_all_certs_from_csv(self, get_active: bool, get_archived: bool) -> Dict[str, "CommonCriteriaCert"]: @@ -538,7 +618,7 @@ class CCDataset(Dataset[CommonCriteriaCert], ComplexSerializableType): self.targets_pdf_dir.mkdir(parents=True, exist_ok=True) 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, + CommonCriteriaCert.download_pdf_st, certs_to_process, config.n_threads, progress_bar_desc="Downloading targets", @@ -546,6 +626,11 @@ class CCDataset(Dataset[CommonCriteriaCert], ComplexSerializableType): @serialize def download_all_pdfs(self, fresh: bool = True) -> None: + """ + Downloads all pdf files associated with certificates of the datset. + + :param bool fresh: whether all (true) or only failed (false) pdfs shall be downloaded, defaults to True + """ if self.state.meta_sources_parsed is False: logger.error("Attempting to download pdfs while not having csv/html meta-sources parsed. Returning.") return @@ -579,7 +664,7 @@ class CCDataset(Dataset[CommonCriteriaCert], ComplexSerializableType): self.targets_txt_dir.mkdir(parents=True, exist_ok=True) 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, + CommonCriteriaCert.convert_st_pdf, certs_to_process, config.n_threads, progress_bar_desc="Converting targets to txt", @@ -587,6 +672,11 @@ class CCDataset(Dataset[CommonCriteriaCert], ComplexSerializableType): @serialize def convert_all_pdfs(self, fresh: bool = True) -> None: + """ + Converts all pdfs associated to certificates of the dataset into txt files. + + :param bool fresh: whether all (true) or only failed (false) pdfs shall be converted, defaults to True + """ if self.state.pdfs_downloaded is False: logger.info("Attempting to convert pdf while not having them downloaded. Returning.") return @@ -607,6 +697,11 @@ class CCDataset(Dataset[CommonCriteriaCert], ComplexSerializableType): self.state.pdfs_converted = True def update_with_certs(self, certs: List[CommonCriteriaCert]) -> None: + """ + Enriches the dataset with `certs` + + :param List[CommonCriteriaCert] certs: new certs to include into the dataset. + """ if any([x not in self for x in certs]): logger.warning("Updating dataset with certificates outside of the dataset!") self.certs.update({x.dgst: x for x in certs}) @@ -633,7 +728,7 @@ class CCDataset(Dataset[CommonCriteriaCert], ComplexSerializableType): ) self.update_with_certs(processed_certs) - def extract_pdf_metadata(self, fresh: bool = True) -> None: + def _extract_pdf_metadata(self, fresh: bool = True) -> None: logger.info("Extracting pdf metadata from CC dataset") self._extract_report_metadata(fresh) self._extract_targets_metadata(fresh) @@ -660,7 +755,7 @@ class CCDataset(Dataset[CommonCriteriaCert], ComplexSerializableType): ) self.update_with_certs(processed_certs) - def extract_pdf_frontpage(self, fresh: bool = True) -> None: + def _extract_pdf_frontpage(self, fresh: bool = True) -> None: logger.info("Extracting pdf frontpages from CC dataset.") self._extract_report_frontpage(fresh) self._extract_targets_frontpage(fresh) @@ -687,16 +782,16 @@ class CCDataset(Dataset[CommonCriteriaCert], ComplexSerializableType): ) self.update_with_certs(processed_certs) - def extract_pdf_keywords(self, fresh: bool = True) -> None: + def _extract_pdf_keywords(self, fresh: bool = True) -> None: 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) -> None: 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) + self._extract_pdf_metadata(fresh) + self._extract_pdf_frontpage(fresh) + self._extract_pdf_keywords(fresh) if fresh is True: logger.info("Attempting to re-extract failed data from report txts") @@ -718,7 +813,7 @@ class CCDataset(Dataset[CommonCriteriaCert], ComplexSerializableType): def _compute_normalized_cert_ids(self): logger.info("Deriving information about sample ids from pdf scan.") for cert in self: - cert.compute_heuristics_cert_id(self.all_cert_ids) + cert.compute_heuristics_cert_id(self._all_cert_ids) def _compute_dependency_vulnerabilities(self): cve_dependency_finder = DependencyVulnerabilityFinder() @@ -767,6 +862,11 @@ class CCDataset(Dataset[CommonCriteriaCert], ComplexSerializableType): @serialize def analyze_certificates(self, fresh: bool = True) -> None: + """ + Searches for all RE in the txt files of the dataset. These are further processed during heuristics computation. + + :param bool fresh: Whether to run this phase on all certificates (true) or only on those that failed (false), defaults to 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." @@ -778,10 +878,16 @@ class CCDataset(Dataset[CommonCriteriaCert], ComplexSerializableType): self.state.certs_analyzed = True - def get_certs_from_name(self, cert_name: str) -> List[CommonCriteriaCert]: + def _get_certs_from_name(self, cert_name: str) -> List[CommonCriteriaCert]: return [crt for crt in self if crt.name == cert_name] - def process_maintenance_updates(self) -> "CCDatasetMaintenanceUpdates": + def process_maintenance_updates(self) -> CCDatasetMaintenanceUpdates: + """ + Creates and fully processes (download, convert, analyze) a dataset of maintenance updates that are related + to certificates of self. + + :return CCDatasetMaintenanceUpdates: the resulting dataset of maintenance updates + """ maintained_certs: List[CommonCriteriaCert] = [x for x in self if x.maintenance_updates] updates = list( itertools.chain.from_iterable( @@ -791,23 +897,16 @@ class CCDataset(Dataset[CommonCriteriaCert], ComplexSerializableType): update_dset: CCDatasetMaintenanceUpdates = CCDatasetMaintenanceUpdates( {x.dgst: x for x in updates}, root_dir=self.mu_dataset_path, name="Maintenance updates" ) - update_dset.set_local_paths() update_dset.download_all_pdfs() update_dset.convert_all_pdfs() update_dset._extract_data() return update_dset - def generate_cert_name_keywords(self) -> Set[str]: - df = self.to_pandas() - certificate_names = set(df["name"]) - keywords = set(itertools.chain.from_iterable([x.lower().split(" ") for x in certificate_names])) - keywords.add("1.02.013") - return {x for x in keywords if len(x) > config.minimal_token_length} - class CCDatasetMaintenanceUpdates(CCDataset, ComplexSerializableType): """ + Dataset of maintenance updates related to certificates of CCDataset dataset. Should be used merely for actions related to Maintenance updates: download pdfs, convert pdfs, extract data from pdfs """ @@ -822,6 +921,7 @@ class CCDatasetMaintenanceUpdates(CCDataset, ComplexSerializableType): ): super().__init__(certs, root_dir, name, description, state) # type: ignore self.state.meta_sources_parsed = True + self._set_local_paths() @property def certs_dir(self) -> Path: diff --git a/sec_certs/dataset/cpe.py b/sec_certs/dataset/cpe.py index 43ae3365..0ae35081 100644 --- a/sec_certs/dataset/cpe.py +++ b/sec_certs/dataset/cpe.py @@ -20,6 +20,10 @@ logger = logging.getLogger(__name__) @dataclass class CPEDataset(ComplexSerializableType): + """ + Dataset of CPE records. Includes look-up dictionaries for fast search. + """ + was_enhanced_with_vuln_cpes: bool json_path: Path cpes: Dict[str, CPE] @@ -68,7 +72,7 @@ class CPEDataset(ComplexSerializableType): def build_lookup_dicts(self) -> None: """ - Will build look-up dictionaries that are used for fast matching + Will build look-up dictionaries that are used for fast matching. """ logger.info("CPE dataset: building lookup dictionaries.") self.vendor_to_versions = {x.vendor: set() for x in self} @@ -90,6 +94,13 @@ class CPEDataset(ComplexSerializableType): @classmethod def from_web(cls, json_path: Union[str, Path], init_lookup_dicts: bool = True) -> "CPEDataset": + """ + Creates CPEDataset from NIST resources published on-line + + :param Union[str, Path] json_path: Path to store the dataset to + :param bool init_lookup_dicts: If dictionaries for fast matching should be computed, defaults to True + :return CPEDataset: The resulting dataset + """ with tempfile.TemporaryDirectory() as tmp_dir: xml_path = Path(tmp_dir) / cls.cpe_xml_basename zip_path = Path(tmp_dir) / (cls.cpe_xml_basename + ".zip") @@ -98,10 +109,10 @@ class CPEDataset(ComplexSerializableType): with zipfile.ZipFile(zip_path, "r") as zip_ref: zip_ref.extractall(tmp_dir) - return cls.from_xml(xml_path, json_path, init_lookup_dicts) + return cls._from_xml(xml_path, json_path, init_lookup_dicts) @classmethod - def from_xml( + def _from_xml( cls, xml_path: Union[str, Path], json_path: Union[str, Path], init_lookup_dicts: bool = True ) -> "CPEDataset": logger.info("Loading CPE dataset from XML.") @@ -128,22 +139,48 @@ class CPEDataset(ComplexSerializableType): @classmethod def from_json(cls, input_path: Union[str, Path]) -> "CPEDataset": + """ + Loads dataset from json + + :param Union[str, Path] input_path: Path to the serialized json dataset + :return CPEDataset: the resulting dataset. + """ dset = cast("CPEDataset", ComplexSerializableType.from_json(input_path)) dset.json_path = Path(input_path) return dset @classmethod def from_dict(cls, dct: Dict[str, Any], init_lookup_dicts: bool = True) -> "CPEDataset": + """ + Loads dataset from dictionary. + + :param Dict[str, Any] dct: Dictionary that holds the dataset + :param bool init_lookup_dicts: Whether look-up dicts should be computed as a part of initialization, defaults to True + :return CPEDataset: the resulting dataset. + """ return cls(dct["was_enhanced_with_vuln_cpes"], Path("../"), dct["cpes"], init_lookup_dicts) def to_pandas(self) -> pd.DataFrame: + """ + Turns the dataset into pandas DataFrame. Each CPE record forms a row. + + :return pd.DataFrame: the resulting DataFrame + """ df = pd.DataFrame([x.pandas_tuple for x in self], columns=CPE.pandas_columns) df = df.set_index("uri") return df @serialize def enhance_with_cpes_from_cve_dataset(self, cve_dset: Union[CVEDataset, str, Path]) -> None: - def adding_condition( + """ + Some CPEs are present only in the CVEDataset and are missing from the CPE Dataset. + This method goes through the provided CVEDataset and enriches self with CPEs from + the CVEDataset. + + :param Union[CVEDataset, str, Path] cve_dset: CVEDataset of a path to it. + """ + + def _adding_condition( considered_cpe: CPE, vndr_item_lookup: Set[Tuple[str, str]], vndr_item_version_lookup: Set[Tuple[str, str, str]], @@ -174,7 +211,7 @@ class CPEDataset(ComplexSerializableType): vendor_item_lookup = {(cpe.vendor, cpe.item_name) for cpe in self} vendor_item_version_lookup = {(cpe.vendor, cpe.item_name, cpe.version) for cpe in self} for cpe in helpers.tqdm(all_cpes_in_cve_dset, desc="Enriching CPE dataset with new CPEs"): - if adding_condition(cpe, vendor_item_lookup, vendor_item_version_lookup): + if _adding_condition(cpe, vendor_item_lookup, vendor_item_version_lookup): new_cpe = copy.deepcopy(cpe) new_cpe.start_version = None new_cpe.end_version = None diff --git a/sec_certs/dataset/dataset.py b/sec_certs/dataset/dataset.py index e63aea9f..191c6759 100644 --- a/sec_certs/dataset/dataset.py +++ b/sec_certs/dataset/dataset.py @@ -128,10 +128,10 @@ class Dataset(Generic[CertSubType], ABC): def from_json(cls: Type[DatasetSubType], input_path: Union[str, Path]) -> DatasetSubType: dset = cast("DatasetSubType", ComplexSerializableType.from_json(input_path)) dset.root_dir = Path(input_path).parent.absolute() - dset.set_local_paths() + dset._set_local_paths() return dset - def set_local_paths(self) -> None: + def _set_local_paths(self) -> None: raise NotImplementedError("Not meant to be implemented by the base class.") @abstractmethod @@ -289,12 +289,12 @@ class Dataset(Generic[CertSubType], ABC): else: cert_name = annotation["text"] - certs = self.get_certs_from_name(cert_name) + certs = self._get_certs_from_name(cert_name) for c in certs: c.heuristics.verified_cpe_matches = {x.uri for x in cpes if x is not None} if cpes else None - def get_certs_from_name(self, name: str) -> List[CertSubType]: + def _get_certs_from_name(self, name: str) -> List[CertSubType]: raise NotImplementedError("Not meant to be implemented by the base class.") def enrich_automated_cpes_with_manual_labels(self) -> None: diff --git a/sec_certs/dataset/fips.py b/sec_certs/dataset/fips.py index 2defd9f1..d70cf57f 100644 --- a/sec_certs/dataset/fips.py +++ b/sec_certs/dataset/fips.py @@ -1,14 +1,11 @@ import logging -import os import tempfile -from itertools import groupby from pathlib import Path -from typing import Any, Dict, List, Optional, Set, Tuple +from typing import Any, Dict, List, Optional, Set from bs4 import BeautifulSoup, NavigableString from graphviz import Digraph -from sec_certs import constants as constants from sec_certs import helpers as helpers from sec_certs import parallel_processing as cert_processing from sec_certs.config.configuration import config @@ -23,6 +20,10 @@ logger = logging.getLogger(__name__) class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): + """ + Class for processing of FIPSCertificate samples. Inherits from `ComplexSerializableType` and base abstract `Dataset` class. + """ + def __init__( self, certs: Dict[str, FIPSCertificate] = dict(), @@ -36,48 +37,37 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): self.new_files = 0 @property - def results_dir(self) -> Path: - return self.root_dir / "results" - - @property - def policies_dir(self) -> Path: + def _policies_dir(self) -> Path: return self.root_dir / "security_policies" @property - def fragments_dir(self) -> Path: + def _fragments_dir(self) -> Path: return self.root_dir / "fragments" @property - def algs_dir(self) -> Path: + def _algs_dir(self) -> Path: return self.web_dir / "algorithms" - # After web scan, there should be a FIPSCertificate object created for every entry - @property - def successful_web_scan(self) -> bool: - return all(self.certs) and all(cert.web_scan for cert in self.certs.values() if cert is not None) - - @property - def successful_pdf_scan(self) -> bool: - return all(cert.pdf_scan for cert in self.certs.values() if cert is not None) + def _get_certs_from_name(self, module_name: str) -> List[FIPSCertificate]: + """ + Returns list of certificates that match given name. - def get_certs_from_name(self, module_name: str) -> List[FIPSCertificate]: + :param str module_name: name to search for + :return List[FIPSCertificate]: List of certificates with web_scan.module_name == module_name + """ return [crt for crt in self if crt.web_scan.module_name == module_name] - def find_empty_pdfs(self) -> Tuple[List, List]: - missing = [] - not_available = [] - for i in self.certs: - if not (self.policies_dir / f"{i}.pdf").exists(): - missing.append(i) - elif os.path.getsize(self.policies_dir / f"{i}.pdf") < constants.FIPS_NOT_AVAILABLE_CERT_SIZE: - not_available.append(i) - return missing, not_available - @serialize def pdf_scan(self, redo: bool = False) -> None: + """ + pdf_scan() + Extracts data from pdf files + + :param bool redo: Whether to try again with failed files, defaults to False + """ logger.info("Entering PDF scan.") - self.fragments_dir.mkdir(parents=True, exist_ok=True) + self._fragments_dir.mkdir(parents=True, exist_ok=True) keywords = cert_processing.process_parallel( FIPSCertificate.find_keywords, @@ -89,7 +79,7 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): for keyword, cert in keywords: self.certs[cert.dgst].pdf_scan.keywords = keyword - def match_algs(self) -> Dict[str, int]: + def _match_algs(self) -> Dict[str, int]: output = {} for cert in self.certs.values(): # if the pdf has not been processed, no matching can be done @@ -103,18 +93,24 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): return output def download_all_pdfs(self, cert_ids: Optional[Set[str]] = None) -> None: + """ + Downloads all pdf files related to the certificates specified with cert_ids. + + :param Optional[Set[str]] cert_ids: cert_ids to download the pdfs foor, defaults to None + :raises RuntimeError: If no cert_ids are specified, raises. + """ sp_paths, sp_urls = [], [] - self.policies_dir.mkdir(exist_ok=True) + self._policies_dir.mkdir(exist_ok=True) if cert_ids is None: 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 ( + if not (self._policies_dir / f"{cert_id}.pdf").exists() or ( 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" ) - sp_paths.append(self.policies_dir / f"{cert_id}.pdf") + sp_paths.append(self._policies_dir / f"{cert_id}.pdf") logger.info(f"downloading {len(sp_urls)} module pdf files") cert_processing.process_parallel( FIPSCertificate.download_security_policy, @@ -124,7 +120,7 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): ) self.new_files += len(sp_urls) - def download_all_htmls(self, cert_ids: Set[str]) -> List[str]: + def _download_all_htmls(self, cert_ids: Set[str]) -> List[str]: html_paths, html_urls = [], [] new_files = [] self.web_dir.mkdir(exist_ok=True) @@ -158,17 +154,20 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): @serialize def convert_all_pdfs(self) -> None: + """ + Converts all pdfs to text files + """ logger.info("Converting FIPS sample reports to .txt") tuples = [ - (cert, self.policies_dir / f"{cert.cert_id}.pdf", self.policies_dir / f"{cert.cert_id}.pdf.txt") + (cert, self._policies_dir / f"{cert.cert_id}.pdf", self._policies_dir / f"{cert.cert_id}.pdf.txt") for cert in self.certs.values() - if not cert.state.txt_state and (self.policies_dir / f"{cert.cert_id}.pdf").exists() + if not cert.state.txt_state and (self._policies_dir / f"{cert.cert_id}.pdf").exists() ] cert_processing.process_parallel( FIPSCertificate.convert_pdf_file, tuples, config.n_threads, progress_bar_desc="Converting to txt" ) - def prepare_dataset(self, test: Optional[Path] = None, update: bool = False) -> Set[str]: + def _prepare_dataset(self, test: Optional[Path] = None, update: bool = False) -> Set[str]: if test: html_files = [test] else: @@ -197,8 +196,8 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): return cert_ids - def download_neccessary_files(self, cert_ids: Set[str]) -> None: - self.download_all_htmls(cert_ids) + def _download_neccessary_files(self, cert_ids: Set[str]) -> None: + self._download_all_htmls(cert_ids) self.download_all_pdfs(cert_ids) def _get_certificates_from_html(self, html_file: Path, update: bool = False) -> Set[str]: @@ -220,15 +219,21 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): @serialize def web_scan(self, cert_ids: Set[int], redo: bool = False) -> None: + """ + Creates FIPSCertificate object from the relevant html file that must be downlaoded. + + :param Set[int] cert_ids: Cert ids to create FIPSCertificate objects for. + :param bool redo: whether to re-attempt with failed certificates, defaults to False + """ logger.info("Entering web scan.") for cert_id in cert_ids: dgst = fips_dgst(cert_id) - self.certs[dgst] = FIPSCertificate.html_from_file( + self.certs[dgst] = FIPSCertificate.from_html_file( self.web_dir / f"{cert_id}.html", FIPSCertificate.State( - (self.policies_dir / str(cert_id)).with_suffix(".pdf"), + (self._policies_dir / str(cert_id)).with_suffix(".pdf"), (self.web_dir / str(cert_id)).with_suffix(".html"), - (self.fragments_dir / str(cert_id)).with_suffix(".txt"), + (self._fragments_dir / str(cert_id)).with_suffix(".txt"), False, None, False, @@ -239,6 +244,9 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): @classmethod def from_web_latest(cls) -> "FIPSDataset": + """ + Fetches the fresh snapshot of FIPSDataset from mirror. + """ with tempfile.TemporaryDirectory() as tmp_dir: dset_path = Path(tmp_dir) / "fips_latest_dataset.json" logger.info("Downloading the latest FIPS dataset.") @@ -254,14 +262,15 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): len(dset), len(dset.algorithms) if dset.algorithms is not None else 0, ) - logger.info("The dataset does not contain the results of the dependency analysis - calculating them now...") - dset.finalize_results() + # TODO: Fixme, this is really costly + # logger.info("The dataset does not contain the results of the dependency analysis - calculating them now...") + # dset.finalize_results() return dset - def set_local_paths(self) -> None: + def _set_local_paths(self) -> None: cert: FIPSCertificate for cert in self.certs.values(): - cert.set_local_paths(self.policies_dir, self.web_dir, self.fragments_dir) + cert.set_local_paths(self._policies_dir, self.web_dir, self._fragments_dir) @serialize def get_certs_from_web( @@ -284,11 +293,11 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): logger.info("Downloading required html files") self.web_dir.mkdir(parents=True, exist_ok=True) - self.policies_dir.mkdir(exist_ok=True) - self.algs_dir.mkdir(exist_ok=True) + self._policies_dir.mkdir(exist_ok=True) + self._algs_dir.mkdir(exist_ok=True) # Download files containing all available module certs (always) - cert_ids = self.prepare_dataset(test, update) + cert_ids = self._prepare_dataset(test, update) if not no_download_algorithms: aset = FIPSAlgorithmDataset({}, Path(self.root_dir / "web" / "algorithms"), "algorithms", "sample algs") @@ -298,23 +307,11 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): self.algorithms = aset logger.info("Downloading certificate html and security policies") - self.download_neccessary_files(cert_ids) + self._download_neccessary_files(cert_ids) self.web_scan(cert_ids, redo=redo_web_scan) @serialize - def deprocess(self) -> None: - # TODO think of a better way to make resulting json smaller and easier to share - logger.info( - "Removing 'heuristics' field. This dataset can be used to be uploaded and later downloaded using latest_snapshot() or something" - ) - cert: FIPSCertificate - for cert in self.certs.values(): - cert.heuristics = FIPSCertificate.FIPSHeuristics(dict(), [], 0) - - self.match_algs() - - @serialize def extract_certs_from_tables(self, high_precision: bool) -> List[Path]: """ Function that extracts algorithm IDs from tables in security policies files. @@ -341,11 +338,11 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): certificate.pdf_scan.algorithms += algorithms return not_decoded - def remove_algorithms_from_extracted_data(self) -> None: + def _remove_algorithms_from_extracted_data(self) -> None: for cert in self.certs.values(): cert.remove_algorithms() - def unify_algorithms(self) -> None: + def _unify_algorithms(self) -> None: for certificate in self.certs.values(): new_algorithms: List[Dict] = [] united_algorithms = [ @@ -439,7 +436,7 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): return False return True - def validate_results(self) -> None: + def _validate_results(self) -> None: """ Function that validates results and finds the final connection output """ @@ -481,10 +478,16 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): @serialize def finalize_results(self, use_nist_cpe_matching_dict: bool = True, perform_cpe_heuristics: bool = True): + """ + Performs processing of extracted data. Computes all heuristics. + + :param bool use_nist_cpe_matching_dict: If NIST CPE matching dictionary shall be used to drive computing related CVEs, defaults to True + :param bool perform_cpe_heuristics: If CPE heuristics shall be computed, defaults to True + """ logger.info("Entering 'analysis' and building connections between certificates.") - self.unify_algorithms() - self.remove_algorithms_from_extracted_data() - self.validate_results() + self._unify_algorithms() + self._remove_algorithms_from_extracted_data() + self._validate_results() if perform_cpe_heuristics: _, _, cve_dset = self.compute_cpe_heuristics() self.compute_related_cves(use_nist_cpe_matching_dict=use_nist_cpe_matching_dict, cve_dset=cve_dset) @@ -520,7 +523,7 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): attr = {"pdf": "pdf_scan", "web": "web_scan", "heuristics": "heuristics"}[connection_list] return getattr(self.certs[dgst], attr).connections - def create_dot_graph( + def _create_dot_graph( self, output_file_name: str, connection_list: str = "heuristics", @@ -585,6 +588,11 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): single_dot.render(self.root_dir / (str(output_file_name) + "_single"), view=show) def to_dict(self) -> Dict[str, Any]: + """ + Serializes dataset into a dictionary + + :return Dict[str, Any]: Dictionary that holds the whole dataset. + """ return { "timestamp": self.timestamp, "sha256_digest": self.sha256_digest, @@ -597,6 +605,12 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): @classmethod def from_dict(cls, dct: Dict[str, Any]) -> "FIPSDataset": + """ + Reconstructs the original dataset from a dictionary + + :param Dict[str, Any] dct: Dictionary that holds the serialized dataset + :return FIPSDataset: Deserialized FIPSDataset that corresponds to `dct` contents + """ certs = dct["certs"] dset = cls(certs, Path("./"), dct["name"], dct["description"]) dset.algorithms = dct["algs"] @@ -606,16 +620,12 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): ) return dset - def group_vendors(self) -> Dict[str, List[str]]: - vendors: Dict[str, List[str]] = {} - v = {x.web_scan.vendor.lower() for x in self.certs.values() if x is not None and x.web_scan.vendor is not None} - v_sorted = sorted(v, key=FIPSCertificate.get_compare) - for prefix, a in groupby(v_sorted, key=FIPSCertificate.get_compare): - vendors[prefix] = list(a) - - return vendors - def plot_graphs(self, show: bool = False) -> None: - self.create_dot_graph("full_graph", show=show) - self.create_dot_graph("web_only_graph", "web", show=show) - self.create_dot_graph("pdf_only_graph", "pdf", show=show) + """ + Plots FIPS graphs. + # TODO: Currently broken, see https://github.com/crocs-muni/sec-certs/issues/211 + :param bool show: If plots should be showed with .show() method, defaults to False + """ + self._create_dot_graph("full_graph", show=show) + self._create_dot_graph("web_only_graph", "web", show=show) + self._create_dot_graph("pdf_only_graph", "pdf", show=show) diff --git a/sec_certs/model/cpe_matching.py b/sec_certs/model/cpe_matching.py index 1dcf8801..7b25b2a2 100644 --- a/sec_certs/model/cpe_matching.py +++ b/sec_certs/model/cpe_matching.py @@ -34,30 +34,34 @@ class CPEClassifier(BaseEstimator): def fit(self, X: List[CPE], y: Optional[List[str]] = None) -> "CPEClassifier": """ Just creates look-up structures from provided list of CPEs - @param X: List of CPEs that can be matched with predict() - @param y: will be ignored, specified to adhere to sklearn BaseEstimator interface + + :param List[CPE] X: List of CPEs that can be matched with predict() + :param Optional[List[str]] y: will be ignored, specified to adhere to sklearn BaseEstimator interface, defaults to None + :return CPEClassifier: return self to allow method chaining """ - self.build_lookup_structures(X) + self._build_lookup_structures(X) return self @staticmethod - def filter_short_cpes(cpes: List[CPE]) -> List[CPE]: + def _filter_short_cpes(cpes: List[CPE]) -> List[CPE]: """ Short CPE items are super easy to match with 100% rank, but they are hardly informative. This method discards them. - @param cpes: List of CPEs to filtered - @return All CPEs in cpes variable which item name has at least 4 characters. + + :param List[CPE] cpes: List of CPEs to filtered + :return List[CPE]: All CPEs in cpes variable which item name has at least 4 characters. """ return list(filter(lambda x: x.item_name is not None and len(x.item_name) > 3, cpes)) - def build_lookup_structures(self, X: List[CPE]) -> None: + def _build_lookup_structures(self, X: List[CPE]) -> None: """ Builds several look-up dictionaries for fast matching. - vendor_to_version_: each vendor is mapped to set of versions that appear in combination with vendor in CPE dataset - vendor_version_to_cpe_: Each (vendor, version) tuple is mapped to a set of CPE items that appear in combination with this tuple in CPE dataset - vendors_: Just aggregates set of vendors, used for prunning later on. - @param X: List of CPEs that will be used to build the dictionaries + + :param List[CPE] X: List of CPEs that will be used to build the dictionaries """ - sufficiently_long_cpes = self.filter_short_cpes(X) + sufficiently_long_cpes = self._filter_short_cpes(X) self.vendor_to_versions_ = {x.vendor: set() for x in sufficiently_long_cpes} self.vendors_ = set(self.vendor_to_versions_.keys()) self.vendor_version_to_cpe_ = dict() @@ -72,8 +76,9 @@ class CPEClassifier(BaseEstimator): def predict(self, X: List[Tuple[str, str, str]]) -> List[Optional[Set[str]]]: """ Will predict CPE uris for List of Tuples (vendor, product name, identified versions in product name) - @param X: tuples (vendor, product name, identified versions in product name) - @return: List of CPE uris that correspond to given input, None if nothing was found. + + :param List[Tuple[str, str, str]] X: tuples (vendor, product name, identified versions in product name) + :return List[Optional[Set[str]]]: List of CPE uris that correspond to given input, None if nothing was found. """ return [self.predict_single_cert(x[0], x[1], x[2]) for x in helpers.tqdm(X, desc="Predicting")] @@ -93,23 +98,23 @@ class CPEClassifier(BaseEstimator): 4. Compute string similarity of the candidate CPE matches and certificate name 5. Evaluate best string similarity, if above threshold, declare it a match. 6. If no CPE item is matched, try again but relax version and check CPEs that don't have their version specified. - Also, search for 100% CPE matches on item name instead of title. - @param vendor: manufacturer of the certificate - @param product_name: name of the certificate - @param versions: List of versions that appear in the certificate name - @param relax_version: bool, see step 6 above. - @param relax_title: bool - @return: - """ + 7. (Also, search for 100% CPE matches on item name instead of title.) + :param Optional[str] vendor: manufacturer of the certificate + :param str product_name: name of the certificate + :param Set[str] versions: List of versions that appear in the certificate name + :param bool relax_version: See step 6 above., defaults to False + :param bool relax_title: See step 7 above, defaults to False + :return Optional[Set[str]]: Set of matching CPE uris, None if no matches found + """ lemmatized_product_name = self._lemmatize_product_name(product_name) - candidate_vendors = self.get_candidate_list_of_vendors( + candidate_vendors = self._get_candidate_list_of_vendors( CPEClassifier._discard_trademark_symbols(vendor).lower() if vendor else vendor ) - candidates = self.get_candidate_cpe_matches(candidate_vendors, versions) + candidates = self._get_candidate_cpe_matches(candidate_vendors, versions) ratings = [ - self.compute_best_match(cpe, lemmatized_product_name, candidate_vendors, versions, relax_title=relax_title) + self._compute_best_match(cpe, lemmatized_product_name, candidate_vendors, versions, relax_title=relax_title) for cpe in candidates ] threshold = self.match_threshold if not relax_version else 100 @@ -131,7 +136,7 @@ class CPEClassifier(BaseEstimator): return final_matches if final_matches else None - def compute_best_match( + def _compute_best_match( self, cpe: CPE, product_name: str, @@ -143,11 +148,13 @@ class CPEClassifier(BaseEstimator): Tries several different settings in which string similarity between CPE and certificate name is tested. For definition of string similarity, see rapidfuzz package on GitHub. Both token set ratio and partial ratio are tested, always both on CPE title and CPE item name. - @param cpe: CPE to test - @param product_name: name of the certificate - @param candidate_vendors: vendors that appear in the certificate - @param versions: versions that appear in the certificate - @return: Maximal value of the four string similarities discussed above. + + :param CPE cpe: CPE to test + :param str product_name: name of the certificate + :param Optional[Set[str]] candidate_vendors: vendors that appear in the certificate + :param Set[str] versions: versions that appear in the certificate + :param bool relax_title: if to relax title or not, defaults to False + :return float: Maximal value of the four string similarities discussed above. """ if relax_title: sanitized_title = ( @@ -225,16 +232,17 @@ class CPEClassifier(BaseEstimator): if "athena" in tokenized and "smartcard" in tokenized: result.add("athena-scs") if tokenized[0] == "the" and not result: - candidate_result = self.get_candidate_list_of_vendors(" ".join(tokenized[1:])) + candidate_result = self._get_candidate_list_of_vendors(" ".join(tokenized[1:])) return set(candidate_result) if candidate_result else set() return set(result) if result else set() - def get_candidate_list_of_vendors(self, manufacturer: Optional[str]) -> Set[str]: + def _get_candidate_list_of_vendors(self, manufacturer: Optional[str]) -> Set[str]: """ Given manufacturer name, this method will find list of plausible vendors from CPE dataset that are likely related. - @param manufacturer: manufacturer - @return: List of related manufacturers, None if nothing relevant is found. + + :param Optional[str] manufacturer: manufacturer + :return Set[str]: List of related manufacturers, None if nothing relevant is found. """ result: Set[str] = set() if not manufacturer: @@ -246,7 +254,7 @@ class CPEClassifier(BaseEstimator): vendor_tokens = set( itertools.chain.from_iterable([[x.strip() for x in manufacturer.split(s)] for s in splits]) ) - result_aux = [self.get_candidate_list_of_vendors(x) for x in vendor_tokens] + result_aux = [self._get_candidate_list_of_vendors(x) for x in vendor_tokens] result_used = set(set(itertools.chain.from_iterable([x for x in result_aux if x]))) return result_used if result_used else set() @@ -255,14 +263,16 @@ class CPEClassifier(BaseEstimator): return self._process_manufacturer(manufacturer, result) - def get_candidate_vendor_version_pairs( + def _get_candidate_vendor_version_pairs( self, cert_candidate_cpe_vendors: Set[str], cert_candidate_versions: Set[str] ) -> Optional[List[Tuple[str, str]]]: """ Given parameters, will return Pairs (cpe_vendor, cpe_version) that are relevant to a given sample - @param cert_candidate_cpe_vendors: list of CPE vendors relevant to a sample - @param cert_candidate_versions: List of versions heuristically extracted from the sample name - @return: List of tuples (cpe_vendor, cpe_version) that can be used in the lookup table to search the CPE dataset. + + + :param Set[str] cert_candidate_cpe_vendors: list of CPE vendors relevant to a sample + :param Set[str] cert_candidate_versions: List of versions heuristically extracted from the sample name + :return Optional[List[Tuple[str, str]]]: List of tuples (cpe_vendor, cpe_version) that can be used in the lookup table to search the CPE dataset. """ def is_cpe_version_among_cert_versions(cpe_version: Optional[str], cert_versions: Set[str]) -> bool: @@ -296,14 +306,15 @@ class CPEClassifier(BaseEstimator): candidate_vendor_version_pairs.extend([(vendor, x) for x in matched_cpe_versions]) return candidate_vendor_version_pairs - def get_candidate_cpe_matches(self, candidate_vendors: Set[str], candidate_versions: Set[str]) -> List[CPE]: + def _get_candidate_cpe_matches(self, candidate_vendors: Set[str], candidate_versions: Set[str]) -> List[CPE]: """ Given List of candidate vendors and candidate versions found in certificate, candidate CPE matches are found - @param candidate_vendors: List of version strings that were found in the certificate - @param candidate_versions: List of vendor strings that were found in the certificate - @return: + + :param Set[str] candidate_vendors: List of version strings that were found in the certificate + :param Set[str] candidate_versions: List of vendor strings that were found in the certificate + :return List[CPE]: List of CPE records that could match, to be refined later """ - candidate_vendor_version_pairs = self.get_candidate_vendor_version_pairs(candidate_vendors, candidate_versions) + candidate_vendor_version_pairs = self._get_candidate_vendor_version_pairs(candidate_vendors, candidate_versions) return ( list( itertools.chain.from_iterable([self.vendor_version_to_cpe_[x] for x in candidate_vendor_version_pairs]) diff --git a/sec_certs/model/dependency_finder.py b/sec_certs/model/dependency_finder.py index afc4db59..8410a9c5 100644 --- a/sec_certs/model/dependency_finder.py +++ b/sec_certs/model/dependency_finder.py @@ -11,6 +11,12 @@ ReferenceLookupFunc = Callable[[Certificate], Set[str]] class DependencyFinder: + """ + The class assigns references of other certificate instances for each instance. + Adheres to sklearn BaseEstimator interface. + The fit is called on a dictionary of certificates, builds a hashmap of references, and assigns references for each certificate in the dictionary. + """ + def __init__(self): self.dependencies: Dependencies = {} @@ -96,6 +102,14 @@ class DependencyFinder: return referenced_by_indirect.get(cert, None) def fit(self, certificates: Certificates, id_func: IDLookupFunc, ref_lookup_func: ReferenceLookupFunc) -> None: + """ + Builds a list of references and assigns references for each certificate instance. + + :param Certificates certificates: dictionary of certificates with hashes as key + :param IDLookupFunc id_func: lookup function for cert id + :param ReferenceLookupFunc ref_lookup_func: lookup for references + :return None: None + """ referenced_by_direct, referenced_by_indirect = DependencyFinder._build_cert_references( certificates, id_func, ref_lookup_func ) @@ -140,6 +154,12 @@ class DependencyFinder: return set(res) if res else None def predict_single_cert(self, dgst: str) -> References: + """ + Method returns references object for specified certificate digest + + :param str dgst: certificate digest + :return References: References object + """ return References( self._get_directly_referenced_by(dgst), self._get_indirectly_referenced_by(dgst), @@ -148,6 +168,12 @@ class DependencyFinder: ) def predict(self, dgst_list: List[str]) -> Dict[str, References]: + """ + Method returns references for a list of certificate digests + + :param List[str] dgst_list: List of certificate hashes + :return Dict[str, References]: Dict with certificate hash and References object. + """ cert_references = {} for dgst in dgst_list: diff --git a/sec_certs/model/dependency_vulnerability_finder.py b/sec_certs/model/dependency_vulnerability_finder.py index 2f66f30c..dd25774d 100644 --- a/sec_certs/model/dependency_vulnerability_finder.py +++ b/sec_certs/model/dependency_vulnerability_finder.py @@ -23,6 +23,11 @@ Vulnerabilities = Dict[str, Dict[str, Optional[Set[str]]]] class DependencyVulnerabilityFinder: + """ + The class assigns vulnerabilities to each certificate instance caused by dependencies among certificate instances. + Adheres to sklearn BaseEstimator interface. + """ + def __init__(self): self.vulnerabilities: Vulnerabilities = {} self.certificates: Certificates = {} @@ -77,6 +82,12 @@ class DependencyVulnerabilityFinder: return vulnerabilities if vulnerabilities else None def fit(self, certificates: Certificates) -> Vulnerabilities: + """ + Method assigns each certificate vulnerabilities caused by dependencies among certificates + + :param Certificates certificates: Dictionary of certificates with digests + :return Vulnerabilities: Dictionary of vulnerabilities of certificate instances + """ self._overwrite_previous_state(certificates) cert_id_occurrences = self._get_dataset_cert_ids_occurrences() @@ -106,6 +117,12 @@ class DependencyVulnerabilityFinder: return self.vulnerabilities def predict_single_cert(self, dgst: str) -> DependencyCVE: + """ + Method returns vulnerabilities for certificate digest + + :param str dgst: Digest of certificate + :return DependencyCVE: DependencyCVE object of certificate + """ if not self.vulnerabilities.get(dgst): return DependencyCVE(direct_dependency_cves=None, indirect_dependency_cves=None) @@ -115,6 +132,12 @@ class DependencyVulnerabilityFinder: ) def predict(self, dgst_list: List[str]) -> Dict[str, DependencyCVE]: + """ + Method returns vulnerabilities for a list of certificate digests + + :param List[str] dgst_list: list of certificate digests + :return Dict[str, DependencyCVE]: Dictionary of DependencyCVE objects for specified certificate digests + """ cert_vulnerabilities = {} for dgst in dgst_list: diff --git a/sec_certs/model/sar_transformer.py b/sec_certs/model/sar_transformer.py index ee99b279..58f96c0b 100644 --- a/sec_certs/model/sar_transformer.py +++ b/sec_certs/model/sar_transformer.py @@ -19,17 +19,38 @@ class SARTransformer(BaseEstimator, TransformerMixin): """ def fit(self, certificates: Iterable[CommonCriteriaCert]) -> SARTransformer: + """ + Just returns self, no fitting needed + + :param Iterable[CommonCriteriaCert] certificates: Unused parameter + :return SARTransformer: return self + """ return self def transform(self, certificates: Iterable[CommonCriteriaCert]) -> List[Optional[Set[SAR]]]: + """ + Just a wrapper around transform_single_cert() called on an iterable of CommonCriteriaCert. + + :param Iterable[CommonCriteriaCert] certificates: Iterable of CommonCriteriaCert objects to perform the extraction on. + :return List[Optional[Set[SAR]]]: Returns List of results from transform_single_cert(). + """ return [self.transform_single_cert(cert) for cert in certificates] def transform_single_cert(self, cert: CommonCriteriaCert) -> Optional[Set[SAR]]: - sec_level_candidates, st_candidates, report_candidates = self.collect_sar_candidates_from_all_sources(cert) - return self.resolve_candidate_conflicts(sec_level_candidates, st_candidates, report_candidates, cert.dgst) + """ + Given CommonCriteriaCert, will transform SAR keywords extracted from txt files + into a set of SAR objects. Also handles extractin of correct SAR levels, duplicities and filtering. + Uses three sources: CSV scan, security target, and certification report. + The caller should assure that the certificates have the keywords extracted. + + :param CommonCriteriaCert cert: Certificate to extract SARs from + :return Optional[Set[SAR]]: Set of SARs, None if none were identified. + """ + sec_level_candidates, st_candidates, report_candidates = self._collect_sar_candidates_from_all_sources(cert) + return self._resolve_candidate_conflicts(sec_level_candidates, st_candidates, report_candidates, cert.dgst) @staticmethod - def collect_sar_candidates_from_all_sources(cert: CommonCriteriaCert) -> Tuple[Set[SAR], Set[SAR], Set[SAR]]: + def _collect_sar_candidates_from_all_sources(cert: CommonCriteriaCert) -> Tuple[Set[SAR], Set[SAR], Set[SAR]]: """ Parses SARs from three distinct sources and returns the results as a three tuple: - Security level from CSV scan @@ -43,24 +64,24 @@ class SARTransformer(BaseEstimator, TransformerMixin): def report_keywords_may_have_sars(sample: CommonCriteriaCert): return sample.pdf_data.report_keywords and SAR_DICT_KEY in sample.pdf_data.report_keywords - sec_level_sars = SARTransformer.parse_sars_from_security_level_list(cert.security_level) + sec_level_sars = SARTransformer._parse_sars_from_security_level_list(cert.security_level) if st_keywords_may_have_sars(cert): st_dict: Dict = cast(Dict, cert.pdf_data.st_keywords) - st_sars = SARTransformer.parse_sar_dict(st_dict[SAR_DICT_KEY], cert.dgst) + st_sars = SARTransformer._parse_sar_dict(st_dict[SAR_DICT_KEY], cert.dgst) else: st_sars = set() if report_keywords_may_have_sars(cert): report_dict: Dict = cast(Dict, cert.pdf_data.report_keywords) - report_sars = SARTransformer.parse_sar_dict(report_dict[SAR_DICT_KEY], cert.dgst) + report_sars = SARTransformer._parse_sar_dict(report_dict[SAR_DICT_KEY], cert.dgst) else: report_sars = set() return sec_level_sars, st_sars, report_sars @staticmethod - def resolve_candidate_conflicts( + def _resolve_candidate_conflicts( sec_level_candidates: Set[SAR], st_candidates: Set[SAR], report_candidates: Set[SAR], cert_dgst: str ) -> Optional[Set[SAR]]: final_candidates: Dict[str, SAR] = {x.family: x for x in sec_level_candidates} @@ -94,7 +115,7 @@ class SARTransformer(BaseEstimator, TransformerMixin): return set(final_candidates.values()) if final_candidates else None @staticmethod - def parse_sar_dict(dct: Dict[str, int], dgst: str) -> Set[SAR]: + def _parse_sar_dict(dct: Dict[str, int], dgst: str) -> Set[SAR]: """ Accepts st_keywords or report_keywords dictionary. Will reconstruct SAR objects from it. Each SAR family can appear multiple times in the dictionary (due to conflicts) with different levels. Iterated item will replace @@ -126,7 +147,7 @@ class SARTransformer(BaseEstimator, TransformerMixin): return {x[0] for x in sars.values()} if sars else set() @staticmethod - def parse_sars_from_security_level_list(lst: Iterable[str]) -> Set[SAR]: + def _parse_sars_from_security_level_list(lst: Iterable[str]) -> Set[SAR]: sars = set() for element in lst: try: diff --git a/sec_certs/pandas_helpers.py b/sec_certs/pandas_helpers.py index 2847a167..dca1feb4 100644 --- a/sec_certs/pandas_helpers.py +++ b/sec_certs/pandas_helpers.py @@ -38,38 +38,42 @@ def compute_cve_correlations( filter_nans: bool = True, ) -> pd.DataFrame: """ - Computes correlations of EAL and different SARs and two columns: (n_cves, worst_cve). Few assumptions about the passed dataframe: + Computes correlations of EAL and different SARs and two columns: (n_cves, worst_cve_score, avg_cve_score). Few assumptions about the passed dataframe: - EAL column must be categorical data type - SAR column must be a set of SARs - - `n_cves` and `worst_cve` columns must be present in the dataframe + - `n_cves` and `worst_cve_score`, `avg_cve_score` columns must be present in the dataframe Possibly, it can filter columns will both values NaN (due to division by zero or super low supports.) """ - df_sar = df.loc[:, ["highest_security_level", "sars", "worst_cve", "n_cves"]] - df_sar = df_sar.rename(columns={"highest_security_level": "EAL"}) - families = discover_sar_families(df_sar.sars) - df_sar.EAL = df_sar.EAL.cat.codes + df_sar = df.loc[:, ["eal", "extracted_sars", "worst_cve_score", "avg_cve_score", "n_cves"]] + families = discover_sar_families(df_sar.extracted_sars) + df_sar.eal = df_sar.eal.cat.codes if exclude_vuln_free_certs: df_sar = df_sar.loc[df_sar.n_cves > 0] - n_cves_corrs = [df_sar["EAL"].corr(df_sar.n_cves)] - worst_cve_corrs = [df_sar["EAL"].corr(df_sar.worst_cve)] - supports = [df_sar.loc[~df_sar["EAL"].isnull()].shape[0]] + n_cves_corrs = [df_sar["eal"].corr(df_sar.n_cves)] + worst_cve_corrs = [df_sar["eal"].corr(df_sar.worst_cve_score)] + avg_cve_corrs = [df_sar["eal"].corr(df_sar.avg_cve_score)] + supports = [df_sar.loc[~df_sar["eal"].isnull()].shape[0]] for family in tqdm(families): - df_sar[family] = df_sar.sars.map(lambda x: get_sar_level_from_set(x, family)) + df_sar[family] = df_sar.extracted_sars.map(lambda x: get_sar_level_from_set(x, family)) n_cves_corrs.append(df_sar[family].corr(df_sar.n_cves)) - worst_cve_corrs.append(df_sar[family].corr(df_sar.worst_cve)) + worst_cve_corrs.append(df_sar[family].corr(df_sar.worst_cve_score)) + avg_cve_corrs.append(df_sar[family].corr(df_sar.avg_cve_score)) supports.append(df_sar.loc[~df_sar[family].isnull()].shape[0]) df_sar = df_sar.copy() - tuples = list(zip(n_cves_corrs, worst_cve_corrs, supports)) - dct = {family: correlations for family, correlations in zip(["EAL"] + families, tuples)} - df_corr = pd.DataFrame.from_dict(dct, orient="index", columns=["n_cves_corr", "worst_cve_corr", "support"]) + tuples = list(zip(n_cves_corrs, worst_cve_corrs, avg_cve_corrs, supports)) + dct = {family: correlations for family, correlations in zip(["eal"] + families, tuples)} + df_corr = pd.DataFrame.from_dict( + dct, orient="index", columns=["n_cves_corr", "worst_cve_score_corr", "avg_cve_score_corr", "support"] + ) df_corr.style.set_caption("Correlations between EAL, SARs and CVEs") + df_corr = df_corr.sort_values(by="support", ascending=False) if filter_nans: - df_corr = df_corr.dropna(how="all", subset=["n_cves_corr", "worst_cve_corr"]) + df_corr = df_corr.dropna(how="all", subset=["n_cves_corr", "worst_cve_score_corr", "avg_cve_score_corr"]) if output_path: df_corr.to_csv(output_path) @@ -92,9 +96,19 @@ def expand_cc_df_with_cve_cols(cc_df: pd.DataFrame, cve_dset: CVEDataset) -> pd. lambda x: [cve_dset[y].published_date.date() for y in x] if x is not np.nan else np.nan # type: ignore ) df["earliest_cve"] = df.cve_published_dates.map(lambda x: min(x) if isinstance(x, list) else np.nan) - df["worst_cve"] = df.related_cves.map( + df["worst_cve_score"] = df.related_cves.map( lambda x: max([cve_dset[cve].impact.base_score for cve in x]) if x is not np.nan else np.nan ) + + """ + Note: Technically, CVE can have 0 base score. This happens when the CVE is discarded from the database. + This could skew the results. During May 2022 analysis, we encountered a single CVE with such score. + Therefore, we do not treat this case. + To properly treat this, the average should be taken across CVEs with >0 base_socre. + """ + df["avg_cve_score"] = df.related_cves.map( + lambda x: np.mean([cve_dset[cve].impact.base_score for cve in x]) if x is not np.nan else np.nan + ) return df diff --git a/sec_certs/sample/common_criteria.py b/sec_certs/sample/common_criteria.py index 47ea3190..a9721672 100644 --- a/sec_certs/sample/common_criteria.py +++ b/sec_certs/sample/common_criteria.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import copy import operator import re @@ -39,6 +41,13 @@ class CommonCriteriaCert( PandasSerializableType, ComplexSerializableType, ): + """ + Data structure for common criteria certificate. Contains several inner classes that layer the data logic. + Can be serialized into/from json (`ComplexSerializableType`) or pandas (`PandasSerializableType)`. + Is basic element of `CCDataset`. The functionality is mostly related to holding data and transformations that + the certificate can handle itself. `CCDataset` class then instrument this functionality. + """ + cc_url = "http://www.commoncriteriaportal.org" empty_st_url = "http://www.commoncriteriaportal.org/files/epfiles/" @@ -60,7 +69,7 @@ class CommonCriteriaCert( super().__setattr__("maintenance_date", helpers.sanitize_date(self.maintenance_date)) @classmethod - def from_dict(cls, dct: Dict) -> "CommonCriteriaCert.MaintenanceReport": + def from_dict(cls, dct: Dict) -> CommonCriteriaCert.MaintenanceReport: new_dct = dct.copy() new_dct["maintenance_date"] = ( date.fromisoformat(dct["maintenance_date"]) @@ -74,6 +83,11 @@ class CommonCriteriaCert( @dataclass(init=False) class InternalState(ComplexSerializableType): + """ + Holds internal state of the dataset, whether downloads and converts of individual components succeeded. Also + holds information about errors and paths to the files. + """ + st_download_ok: bool report_download_ok: bool st_convert_ok: bool @@ -143,6 +157,10 @@ class CommonCriteriaCert( @dataclass class PdfData(ComplexSerializableType): + """ + Class that holds data extracted from pdf files. + """ + report_metadata: Optional[Dict[str, Any]] = field(default=None) st_metadata: Optional[Dict[str, Any]] = field(default=None) report_frontpage: Optional[Dict[str, Dict[str, Any]]] = field(default=None) @@ -155,26 +173,44 @@ class CommonCriteriaCert( @property def bsi_data(self) -> Optional[Dict[str, Any]]: + """ + Returns frontpage data related to BSI-provided information + """ return self.report_frontpage.get("bsi", None) if self.report_frontpage else None @property def niap_data(self) -> Optional[Dict[str, Any]]: + """ + Returns frontpage data related to niap-provided information + """ return self.report_frontpage.get("niap", None) if self.report_frontpage else None @property def nscib_data(self) -> Optional[Dict[str, Any]]: + """ + Returns frontpage data related to nscib-provided information + """ return self.report_frontpage.get("nscib", None) if self.report_frontpage else None @property def canada_data(self) -> Optional[Dict[str, Any]]: + """ + Returns frontpage data related to canada-provided information + """ return self.report_frontpage.get("canada", None) if self.report_frontpage else None @property def anssi_data(self) -> Optional[Dict[str, Any]]: + """ + Returns frontpage data related to ANSSI-provided information + """ return self.report_frontpage.get("anssi", None) if self.report_frontpage else None @property def cert_lab(self) -> Optional[List[str]]: + """ + Returns labs for which certificate data was parsed. + """ labs = [ data["cert_lab"].split(" ")[0].upper() for data in [self.bsi_data, self.anssi_data, self.niap_data, self.nscib_data, self.canada_data] @@ -204,6 +240,9 @@ class CommonCriteriaCert( @property def processed_cert_id(self) -> Optional[str]: + """ + Returns processed cert id extracted from the pdf data. + """ cert_ids = set( filter( lambda x: x, @@ -225,7 +264,8 @@ class CommonCriteriaCert( @property def keywords_cert_id(self) -> Optional[str]: """ - :return: the most occurring among cert ids captured in keywords scan + Returns the most frequently appearing cert id. If you don't know why to use this, you should probably use + `cert_id` property. """ if not self.keywords_rules_cert_id: return None @@ -236,10 +276,17 @@ class CommonCriteriaCert( @property def cert_id(self) -> Optional[str]: - return processed if (processed := self.processed_cert_id) else self.keywords_cert_id + """ + Returns `processed_cert_id` if it exists, else return `keyword_cert_id` + """ + return self.processed_cert_id if self.processed_cert_id else self.keywords_cert_id @dataclass class CCHeuristics(Heuristics, ComplexSerializableType): + """ + Class for various heuristics related to CommonCriteriaCert + """ + extracted_versions: Optional[Set[str]] = field(default=None) cpe_matches: Optional[Set[str]] = field(default=None) verified_cpe_matches: Optional[Set[str]] = field(default=None) @@ -324,7 +371,7 @@ class CommonCriteriaCert( self.maintenance_updates = maintenance_updates self.state = self.InternalState() if not state else state self.pdf_data = self.PdfData() if not pdf_data else pdf_data - self.heuristics: "CommonCriteriaCert.CCHeuristics" = self.CCHeuristics() if not heuristics else heuristics + self.heuristics: CommonCriteriaCert.CCHeuristics = self.CCHeuristics() if not heuristics else heuristics @property def dgst(self) -> str: @@ -337,6 +384,9 @@ class CommonCriteriaCert( @property def eal(self) -> Optional[str]: + """ + Returns EAL of certificate if it was extracted, None otherwise. + """ res = [x for x in self.security_level if re.match(security_level_csv_scan, x)] if not res: return None @@ -368,6 +418,9 @@ class CommonCriteriaCert( @property def pandas_tuple(self) -> Tuple: + """ + Returns tuple of attributes meant for pandas serialization + """ return ( self.dgst, self.heuristics.cert_id, @@ -398,7 +451,7 @@ class CommonCriteriaCert( printed_manufacturer = self.manufacturer if self.manufacturer else "Unknown manufacturer" return str(printed_manufacturer) + " " + str(self.name) + " dgst: " + self.dgst - def merge(self, other: "CommonCriteriaCert", other_source: Optional[str] = None) -> None: + def merge(self, other: CommonCriteriaCert, other_source: Optional[str] = None) -> None: """ Merges with other CC sample. Assuming they come from different sources, e.g., csv and html. Assuming that html source has better protection profiles, they overwrite CSV info @@ -425,7 +478,10 @@ class CommonCriteriaCert( ) @classmethod - def from_dict(cls, dct: Dict) -> "CommonCriteriaCert": + def from_dict(cls, dct: Dict) -> CommonCriteriaCert: + """ + Deserializes dictionary into `CommonCriteriaCert` + """ new_dct = dct.copy() new_dct["maintenance_updates"] = set(dct["maintenance_updates"]) new_dct["protection_profiles"] = set(dct["protection_profiles"]) @@ -508,7 +564,7 @@ class CommonCriteriaCert( return None @staticmethod - def _html_row_get_maintenance_updates(main_div: Tag) -> Set["CommonCriteriaCert.MaintenanceReport"]: + def _html_row_get_maintenance_updates(main_div: Tag) -> Set[CommonCriteriaCert.MaintenanceReport]: possible_updates = list(main_div.find_all("li")) maintenance_updates = set() for u in possible_updates: @@ -531,9 +587,9 @@ class CommonCriteriaCert( return maintenance_updates @classmethod - def from_html_row(cls, row: Tag, status: str, category: str) -> "CommonCriteriaCert": + def from_html_row(cls, row: Tag, status: str, category: str) -> CommonCriteriaCert: """ - Creates a CC sample from html row + Creates a CC sample from html row of commoncriteria.org webpage. """ cells = list(row.find_all("td")) @@ -582,6 +638,14 @@ class CommonCriteriaCert( report_txt_dir: Optional[Union[str, Path]], st_txt_dir: Optional[Union[str, Path]], ) -> None: + """ + Sets paths to files given the requested directories + + :param Optional[Union[str, Path]] report_pdf_dir: Directory where pdf reports shall be stored + :param Optional[Union[str, Path]] st_pdf_dir: Directory where pdf security targets shall be stored + :param Optional[Union[str, Path]] report_txt_dir: Directory where txt reports shall be stored + :param Optional[Union[str, Path]] st_txt_dir: Directory where txt security targets shall be stored + """ if report_pdf_dir is not None: self.state.report_pdf_path = Path(report_pdf_dir) / (self.dgst + ".pdf") if st_pdf_dir is not None: @@ -592,7 +656,13 @@ class CommonCriteriaCert( self.state.st_txt_path = Path(st_txt_dir) / (self.dgst + ".txt") @staticmethod - def download_pdf_report(cert: "CommonCriteriaCert") -> "CommonCriteriaCert": + def download_pdf_report(cert: CommonCriteriaCert) -> CommonCriteriaCert: + """ + Downloads pdf of certification report given the certificate. Staticmethod to allow for parallelization. + + :param CommonCriteriaCert cert: cert to download the pdf report for + :return CommonCriteriaCert: returns the modified certificate with updated state + """ exit_code: Union[str, int] if not cert.report_link: exit_code = "No link" @@ -606,7 +676,13 @@ class CommonCriteriaCert( return cert @staticmethod - def download_pdf_target(cert: "CommonCriteriaCert") -> "CommonCriteriaCert": + def download_pdf_st(cert: CommonCriteriaCert) -> CommonCriteriaCert: + """ + Downloads pdf of security target given the certificate. Staticmethod to allow for parallelization. + + :param CommonCriteriaCert cert: cert to download the pdf security target for + :return CommonCriteriaCert: returns the modified certificate with updated state + """ exit_code: Union[str, int] if not cert.st_link: exit_code = "No link" @@ -620,7 +696,13 @@ class CommonCriteriaCert( return cert @staticmethod - def convert_report_pdf(cert: "CommonCriteriaCert") -> "CommonCriteriaCert": + def convert_report_pdf(cert: CommonCriteriaCert) -> CommonCriteriaCert: + """ + Converts the pdf certification report to txt, given the certificate. Staticmethod to allow for parallelization. + + :param CommonCriteriaCert cert: cert to download the pdf report for + :return CommonCriteriaCert: the modified certificate with updated state + """ exit_code = helpers.convert_pdf_file(cert.state.report_pdf_path, cert.state.report_txt_path) if exit_code != constants.RETURNCODE_OK: error_msg = "failed to convert report pdf->txt" @@ -630,7 +712,13 @@ class CommonCriteriaCert( return cert @staticmethod - def convert_target_pdf(cert: "CommonCriteriaCert") -> "CommonCriteriaCert": + def convert_st_pdf(cert: CommonCriteriaCert) -> CommonCriteriaCert: + """ + Converts the pdf security target to txt, given the certificate. Staticmethod to allow for parallelization. + + :param CommonCriteriaCert cert: cert to download the pdf security target for + :return CommonCriteriaCert: the modified certificate with updated state + """ exit_code = helpers.convert_pdf_file(cert.state.st_pdf_path, cert.state.st_txt_path) if exit_code != constants.RETURNCODE_OK: error_msg = "failed to convert security target pdf->txt" @@ -640,7 +728,13 @@ class CommonCriteriaCert( return cert @staticmethod - def extract_st_pdf_metadata(cert: "CommonCriteriaCert") -> "CommonCriteriaCert": + def extract_st_pdf_metadata(cert: CommonCriteriaCert) -> CommonCriteriaCert: + """ + Extracts metadata from security target pdf given the certificate. Staticmethod to allow for parallelization. + + :param CommonCriteriaCert cert: cert to extract the metadata for. + :return CommonCriteriaCert: the modified certificate with updated state + """ response, cert.pdf_data.st_metadata = helpers.extract_pdf_metadata(cert.state.st_pdf_path) if response != constants.RETURNCODE_OK: cert.state.st_extract_ok = False @@ -648,7 +742,13 @@ class CommonCriteriaCert( return cert @staticmethod - def extract_report_pdf_metadata(cert: "CommonCriteriaCert") -> "CommonCriteriaCert": + def extract_report_pdf_metadata(cert: CommonCriteriaCert) -> CommonCriteriaCert: + """ + Extracts metadata from certification report pdf given the certificate. Staticmethod to allow for parallelization. + + :param CommonCriteriaCert cert: cert to extract the metadata for. + :return CommonCriteriaCert: the modified certificate with updated state + """ response, cert.pdf_data.report_metadata = helpers.extract_pdf_metadata(cert.state.report_pdf_path) if response != constants.RETURNCODE_OK: cert.state.report_extract_ok = False @@ -656,7 +756,13 @@ class CommonCriteriaCert( return cert @staticmethod - def extract_st_pdf_frontpage(cert: "CommonCriteriaCert") -> "CommonCriteriaCert": + def extract_st_pdf_frontpage(cert: CommonCriteriaCert) -> CommonCriteriaCert: + """ + Extracts data from security target pdf frontpage given the certificate. Staticmethod to allow for parallelization. + + :param CommonCriteriaCert cert: cert to extract the frontpage data for. + :return CommonCriteriaCert: the modified certificate with updated state + """ cert.pdf_data.st_frontpage = {} for header_type, associated_header_func in HEADERS.items(): @@ -671,7 +777,13 @@ class CommonCriteriaCert( return cert @staticmethod - def extract_report_pdf_frontpage(cert: "CommonCriteriaCert") -> "CommonCriteriaCert": + def extract_report_pdf_frontpage(cert: CommonCriteriaCert) -> CommonCriteriaCert: + """ + Extracts data from certification report pdf frontpage given the certificate. Staticmethod to allow for parallelization. + + :param CommonCriteriaCert cert: cert to extract the frontpage data for. + :return CommonCriteriaCert: the modified certificate with updated state + """ cert.pdf_data.report_frontpage = {} for header_type, associated_header_func in HEADERS.items(): @@ -686,14 +798,28 @@ class CommonCriteriaCert( return cert @staticmethod - def extract_report_pdf_keywords(cert: "CommonCriteriaCert") -> "CommonCriteriaCert": + def extract_report_pdf_keywords(cert: CommonCriteriaCert) -> CommonCriteriaCert: + """ + Matches regular expresions in txt obtained from certification report and extracts the matches into attribute. + Static method to allow for parallelization + + :param CommonCriteriaCert cert: certificate to extract the keywords for. + :return CommonCriteriaCert: the modified certificate with extracted keywords. + """ response, cert.pdf_data.report_keywords = helpers.extract_keywords(cert.state.report_txt_path) if response != constants.RETURNCODE_OK: cert.state.report_extract_ok = False return cert @staticmethod - def extract_st_pdf_keywords(cert: "CommonCriteriaCert") -> "CommonCriteriaCert": + def extract_st_pdf_keywords(cert: CommonCriteriaCert) -> CommonCriteriaCert: + """ + Matches regular expresions in txt obtained from security target and extracts the matches into attribute. + Static method to allow for parallelization + + :param CommonCriteriaCert cert: certificate to extract the keywords for. + :return CommonCriteriaCert: the modified certificate with extracted keywords. + """ response, cert.pdf_data.st_keywords = helpers.extract_keywords(cert.state.st_txt_path) if response != constants.RETURNCODE_OK: cert.state.st_extract_ok = False @@ -701,15 +827,26 @@ class CommonCriteriaCert( return cert def compute_heuristics_version(self) -> None: + """ + Fills in the heuristically obtained version of certified product into attribute in heuristics class. + """ self.heuristics.extracted_versions = helpers.compute_heuristics_version(self.name) if self.name else set() def compute_heuristics_cert_lab(self) -> None: + """ + Fills in the heuristically obtained evaluation laboratory into attribute in heuristics class. + """ if not self.pdf_data: logger.error("Cannot compute sample lab when pdf files were not processed.") return self.heuristics.cert_lab = self.pdf_data.cert_lab def compute_heuristics_cert_id(self, all_cert_ids: Set[str]): + """ + Given list of cert ids from the whole dataset, will normalize own cert id into canonical form + + :param Set[str] all_cert_ids: cert ids from the whole dataset. + """ if not self.pdf_data: logger.warning("Cannot compute sample id when pdf files were not processed.") return @@ -729,6 +866,7 @@ class CommonCriteriaCert( # Bug - getting out of index - ANSSI-2009/30 # TMP solution + # TODO: Fix me, @georgefi if len(new_cert_id) >= len("ANSSI-CC-0000") + 1: if ( new_cert_id[len("ANSSI-CC-0000")] == "_" @@ -821,7 +959,7 @@ class CommonCriteriaCert( return new_cert_id - def get_cert_laboratory(self) -> str: + def _get_cert_laboratory(self) -> str: if not self.heuristics.cert_id: raise ValueError("Cert ID was None but cert laboratory was to be computed based on its value.") cert_id = self.heuristics.cert_id.strip() @@ -841,6 +979,12 @@ class CommonCriteriaCert( return "unknown" def normalize_cert_id(self, all_cert_ids: Set[str]) -> None: + """ + Attempts to find certification laboratory and transform certificate id into canonical form. This is achieved + also by comparisons to all other cert ids in the dataset. + + :param Set[str] all_cert_ids: set of all cert ids in the dataset. + """ fix_methods: Dict[str, Callable] = { "anssi": CommonCriteriaCert._fix_anssi_cert_id, "bsi": partial(CommonCriteriaCert._fix_bsi_cert_id, all_cert_ids=all_cert_ids), @@ -849,7 +993,7 @@ class CommonCriteriaCert( } try: - cert_lab = self.get_cert_laboratory() + cert_lab = self._get_cert_laboratory() except ValueError: return None diff --git a/sec_certs/sample/fips.py b/sec_certs/sample/fips.py index 0b769dec..b2680f0c 100644 --- a/sec_certs/sample/fips.py +++ b/sec_certs/sample/fips.py @@ -1,3 +1,5 @@ +from __future__ import annotations + import copy import re from dataclasses import dataclass, field @@ -22,6 +24,13 @@ from sec_certs.serialization.json import ComplexSerializableType class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuristics"], ComplexSerializableType): + """ + Data structure for common FIPS 140 certificate. Contains several inner classes that layer the data logic. + Can be serialized into/from json (`ComplexSerializableType`). + Is basic element of `FIPSDataset`. The functionality is mostly related to holding data and transformations that + the certificate can handle itself. `FIPSDataset` class then instrument this functionality. + """ + FIPS_BASE_URL: ClassVar[str] = "https://csrc.nist.gov" FIPS_MODULE_URL: ClassVar[ str @@ -29,6 +38,10 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris @dataclass(eq=True) class State(ComplexSerializableType): + """ + Holds state of the `FIPSCertificate` + """ + sp_path: Path html_path: Path fragment_path: Path @@ -67,6 +80,10 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris @dataclass(eq=True) class Algorithm(ComplexSerializableType): + """ + Data structure for algorithm of `FIPSCertificate` + """ + cert_id: str vendor: str implementation: str @@ -87,6 +104,10 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris @dataclass(eq=True) class WebScan(ComplexSerializableType): + """ + Data structure for data obtained from scanning certificate webpage at NIST.gov + """ + module_name: Optional[str] standard: Optional[str] status: Optional[str] @@ -147,6 +168,10 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris @dataclass(eq=True) class PdfScan(ComplexSerializableType): + """ + Data structure that holds data obtained from scanning pdf files (or their converted txt documents). + """ + cert_id: int keywords: Dict algorithms: List @@ -165,6 +190,10 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris @dataclass(eq=True) class FIPSHeuristics(Heuristics, ComplexSerializableType): + """ + Data structure that holds data obtained by processing the certificate and applying various heuristics. + """ + keywords: Dict[str, Dict] algorithms: List[Dict[str, Dict]] unmatched_algs: int @@ -192,6 +221,9 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris @property def dgst(self) -> str: + """ + Returns primary key of the certificate, its id. + """ return fips_dgst(self.cert_id) # TODO: Fix type errors, they exist because FIPS uses this as property to change variable names, while CC and abstract class have variables @@ -221,6 +253,9 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris @staticmethod def download_security_policy(cert: Tuple[str, Path]) -> None: + """ + Downloads security policy file from web. Staticmethod to allow for parametrization. + """ exit_code = helpers.download_file(*cert, delay=1) if exit_code != requests.codes.ok: logger.error(f"Failed to download security policy from {cert[0]}, code: {exit_code}") @@ -228,20 +263,26 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris def __init__( self, cert_id: int, - web_scan: "FIPSCertificate.WebScan", - pdf_scan: "FIPSCertificate.PdfScan", - heuristics: "FIPSCertificate.FIPSHeuristics", + web_scan: FIPSCertificate.WebScan, + pdf_scan: FIPSCertificate.PdfScan, + heuristics: FIPSCertificate.FIPSHeuristics, state: State, ): super().__init__() self.cert_id = cert_id self.web_scan = web_scan self.pdf_scan = pdf_scan - self.heuristics: "FIPSCertificate.FIPSHeuristics" = heuristics + self.heuristics: FIPSCertificate.FIPSHeuristics = heuristics self.state = state @classmethod - def from_dict(cls, dct: Dict) -> "FIPSCertificate": + def from_dict(cls, dct: Dict) -> FIPSCertificate: + """ + Deserializes dictionary into FIPSCertificate + + :param Dict dct: dictionary that holds the FIPSCertificate data + :return FIPSCertificate: object reconstructed from dct + """ new_dct = dct.copy() if new_dct["web_scan"].date_validation: @@ -253,6 +294,12 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris @staticmethod def download_html_page(cert: Tuple[str, Path]) -> Optional[Tuple[str, Path]]: + """ + Wrapper for downloading a file. `delay=1` introduced to avoid problems with requests at NIST.gov + + :param Tuple[str, Path] cert: tuple url, output_path + :return Optional[Tuple[str, Path]]: None on success, `cert` on failure. + """ exit_code = helpers.download_file(*cert, delay=1) if exit_code != requests.codes.ok: logger.error(f"Failed to download html page from {cert[0]}, code: {exit_code}") @@ -260,7 +307,7 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris return None @staticmethod - def initialize_dictionary() -> Dict[str, Any]: + def _initialize_dictionary() -> Dict[str, Any]: return { "module_name": None, "standard": None, @@ -296,8 +343,9 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris def parse_caveat(current_text: str) -> Dict[str, Dict[str, int]]: """ Parses content of "Caveat" of FIPS CMVP .html file - :param current_text: text of "Caveat" - :return: dictionary of all found algorithm IDs + + :param str current_text: text of "Caveat" + :return Dict[str, Dict[str, int]]: dictionary of all found algorithm IDs """ ids_found: Dict[str, Dict[str, int]] = {} r_key = r"(?P<word>\w+)?\s?(?:#\s?|Cert\.?(?!.\s)\s?|Certificate\s?)+(?P<id>\d+)" @@ -315,9 +363,10 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris def extract_algorithm_certificates(current_text: str, in_pdf: bool = False) -> List[Optional[Dict[str, List[str]]]]: """ Parses table of FIPS (non) allowed algorithms - :param current_text: Contents of the table - :param in_pdf: Specifies whether the table was found in a PDF security policies file - :return: List containing one element - dictionary with all parsed algorithm cert ids + + :param str current_text: Contents of the table + :param bool in_pdf: Specifies whether the table was found in a PDF security policies file, defaults to False + :return List[Optional[Dict[str, List[str]]]]: List containing one element - dictionary with all parsed algorithm cert ids """ set_items = set() if in_pdf: @@ -333,9 +382,11 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris def parse_table(element: Union[Tag, NavigableString]) -> List[Dict[str, Any]]: """ Parses content of <table> tags in FIPS .html CMVP page - :param element: text in <table> tags - :return: list of all found algorithm IDs + + :param Union[Tag, NavigableString] element: text in <table> tags + :return List[Dict[str, Any]]: list of all found algorithm IDs """ + found_items = [] trs = element.find_all("tr") for tr in trs: @@ -355,7 +406,7 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris return found_items @staticmethod - def parse_html_main(current_div: Tag, html_items_found: Dict, pairs: Dict[str, str]) -> None: + def _parse_html_main(current_div: Tag, html_items_found: Dict, pairs: Dict[str, str]) -> None: title = current_div.find("div", class_="col-md-3").text.strip() content = ( current_div.find("div", class_="col-md-9") @@ -391,7 +442,7 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris html_items_found[pairs[title]] = content @staticmethod - def parse_vendor(current_div: Tag, html_items_found: Dict, current_file: Path) -> None: + def _parse_vendor(current_div: Tag, html_items_found: Dict, current_file: Path) -> None: vendor_string = current_div.find("div", "panel-body").find("a") if not vendor_string: @@ -406,7 +457,7 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris logger.warning(f"NO VENDOR FOUND {current_file}") @staticmethod - def parse_lab(current_div: Tag, html_items_found: Dict, current_file: Path) -> None: + def _parse_lab(current_div: Tag, html_items_found: Dict, current_file: Path) -> None: html_items_found["lab"] = list(current_div.find("div", "panel-body").children)[0].strip() html_items_found["nvlap_code"] = ( list(current_div.find("div", "panel-body").children)[2].strip().split("\n")[1].strip() @@ -427,20 +478,29 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris html_items_found["certificate_www"] = constants.FIPS_BASE_URL + links[1].get("href") @staticmethod - def normalize(items: Dict) -> None: + def _normalize(items: Dict) -> None: 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) -> None: + def _parse_validation_dates(current_div: Tag, html_items_found: Dict) -> None: 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 - ) -> "FIPSCertificate": + def from_html_file( + cls, file: Path, state: State, initialized: FIPSCertificate = None, redo: bool = False + ) -> FIPSCertificate: + """ + Constructs FIPSCertificate object from html file. + + :param Path file: path to the html file to use for initialization + :param State state: state of the certificate + :param FIPSCertificate initialized: possibly partially initialized FIPSCertificate, defaults to None + :param bool redo: if the method should be reattempted in case of failure, defaults to False + :return FIPSCertificate: resulting `FIPSCertificate` object. + """ pairs = { "Module Name": "module_name", "Standard": "standard", @@ -466,7 +526,7 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris "Product URL": "product_url", } if not initialized: - items_found = FIPSCertificate.initialize_dictionary() + items_found = FIPSCertificate._initialize_dictionary() items_found["cert_id"] = int(file.stem) else: items_found = initialized.web_scan.__dict__ @@ -479,28 +539,28 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris state.txt_state = initialized.state.txt_state if redo: - items_found = FIPSCertificate.initialize_dictionary() + items_found = FIPSCertificate._initialize_dictionary() items_found["cert_id"] = int(file.stem) text = helpers.load_cert_html_file(str(file)) soup = BeautifulSoup(text, "html.parser") for div in soup.find_all("div", class_="row padrow"): - FIPSCertificate.parse_html_main(div, items_found, pairs) + FIPSCertificate._parse_html_main(div, items_found, pairs) for div in soup.find_all("div", class_="panel panel-default")[1:]: if div.find("h4", class_="panel-title").text == "Vendor": - FIPSCertificate.parse_vendor(div, items_found, file) + FIPSCertificate._parse_vendor(div, items_found, file) if div.find("h4", class_="panel-title").text == "Lab": - FIPSCertificate.parse_lab(div, items_found, file) + FIPSCertificate._parse_lab(div, items_found, file) 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._parse_validation_dates(div, items_found) - FIPSCertificate.normalize(items_found) + FIPSCertificate._normalize(items_found) return FIPSCertificate( items_found["cert_id"], @@ -543,7 +603,13 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris ) @staticmethod - def convert_pdf_file(tup: Tuple["FIPSCertificate", Path, Path]) -> "FIPSCertificate": + def convert_pdf_file(tup: Tuple[FIPSCertificate, Path, Path]) -> FIPSCertificate: + """ + Converts pdf file of FIPSCertificate. Staticmethod to allow for paralelization. + + :param Tuple[FIPSCertificate, Path, Path] tup: object which file will be converted, path to pdf, path to txt. + :return FIPSCertificate: the modified FIPSCertificate. + """ cert, pdf_path, txt_path = tup if not cert.state.txt_state: exit_code = helpers.convert_pdf_file(pdf_path, txt_path) @@ -565,7 +631,7 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris return len(text) * 0.5 <= len("".join(filter(str.isalpha, text))) @staticmethod - def find_keywords(cert: "FIPSCertificate") -> Tuple[Optional[Dict], "FIPSCertificate"]: + def find_keywords(cert: FIPSCertificate) -> Tuple[Optional[Dict], FIPSCertificate]: if not cert.state.txt_state: return None, cert @@ -580,11 +646,11 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris if config.ignore_first_page: text_to_parse = text_to_parse[text_to_parse.index(" ") :] - items_found, fips_text = FIPSCertificate.parse_cert_file(FIPSCertificate.remove_platforms(text_to_parse)) + items_found, fips_text = FIPSCertificate._parse_cert_file(FIPSCertificate._remove_platforms(text_to_parse)) save_modified_cert_file(cert.state.fragment_path.with_suffix(".fips.txt"), fips_text, unicode_error) - common_items_found, common_text = FIPSCertificate.parse_cert_file_common( + common_items_found, common_text = FIPSCertificate._parse_cert_file_common( text_to_parse, text_with_newlines, fips_common_rules ) @@ -594,7 +660,13 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris return items_found, cert @staticmethod - def match_web_algs_to_pdf(cert: "FIPSCertificate") -> int: + def match_web_algs_to_pdf(cert: FIPSCertificate) -> int: + """ + Finds algorithms in FIPSCertificate. Staticmethod to allow for parallelization. + + :param FIPSCertificate cert: cert to search for algorithms. + :return int: number of identified algorithms. + """ 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] @@ -619,7 +691,7 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris return len(not_found) @staticmethod - def remove_platforms(text_to_parse: str) -> str: + def _remove_platforms(text_to_parse: str) -> str: pat = re.compile(r"(?:(?:modification|revision|change) history|version control)\n[\s\S]*? ", re.IGNORECASE) for match in pat.finditer(text_to_parse): text_to_parse = text_to_parse.replace(match.group(), "x" * len(match.group())) @@ -669,7 +741,7 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris items_found[rule_str][match][constants.TAG_MATCH_MATCHES].append([match_span[0], match_span[1]]) @staticmethod - def parse_cert_file_common( + def _parse_cert_file_common( text_to_parse: str, whole_text_with_newlines: str, search_rules: Dict ) -> Tuple[Dict[Pattern, Dict], str]: # apply all rules @@ -699,7 +771,7 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris return items_found_all, whole_text_with_newlines @staticmethod - def parse_cert_file(text_to_parse: str) -> Tuple[Dict[Pattern, Dict], str]: + def _parse_cert_file(text_to_parse: str) -> Tuple[Dict[Pattern, Dict], str]: # apply all rules items_found_all: Dict = {} @@ -733,7 +805,13 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris return items_found_all, text_to_parse @staticmethod - def analyze_tables(tup: Tuple["FIPSCertificate", bool]) -> Tuple[bool, "FIPSCertificate", List]: + def analyze_tables(tup: Tuple[FIPSCertificate, bool]) -> Tuple[bool, FIPSCertificate, List]: + """ + Searches for tables in pdf documents of the instance. + + :param Tuple[FIPSCertificate, bool] tup: certificate object, whether to use high precision results or approx. results + :return Tuple[bool, FIPSCertificate, List]: True on success / False otherwise, modified cert object, List of processed tables. + """ cert, precision = tup if (not precision and cert.state.tables_done) or ( precision and cert.heuristics.unmatched_algs < config.cert_threshold @@ -801,6 +879,9 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris to_pop.add(cert) def remove_algorithms(self) -> None: + """ + Removes algorithms from the certificate. + """ self.state.file_status = True if not self.pdf_scan.keywords: return @@ -829,12 +910,18 @@ class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuris @staticmethod def get_compare(vendor: str) -> str: + """ + Tokenizes vendor name of the certificate. + """ vendor_split = ( vendor.replace(",", "").replace("-", " ").replace("+", " ").replace("®", "").replace("(R)", "").split() ) return vendor_split[0][:4] if len(vendor_split) > 0 else vendor def compute_heuristics_version(self) -> None: + """ + Heuristically computes the version of the product. + """ versions_for_extraction = "" if self.web_scan.module_name: versions_for_extraction += f" {self.web_scan.module_name}" diff --git a/sec_certs/serialization/json.py b/sec_certs/serialization/json.py index 4799e259..1664f799 100644 --- a/sec_certs/serialization/json.py +++ b/sec_certs/serialization/json.py @@ -1,6 +1,7 @@ import copy import json from datetime import date +from functools import wraps from pathlib import Path from typing import Any, Callable, Dict, List, Optional, Type, TypeVar, Union @@ -55,6 +56,7 @@ class ComplexSerializableType: # Decorator for serialization def serialize(func: Callable): + @wraps(func) def inner_func(*args, **kwargs): if not args or not issubclass(type(args[0]), ComplexSerializableType): raise ValueError( |