diff options
| author | adamjanovsky | 2022-02-17 21:10:29 +0100 |
|---|---|---|
| committer | GitHub | 2022-02-17 21:10:29 +0100 |
| commit | 47863129815d4d283ae2cdc4212492953a6dde4b (patch) | |
| tree | 11110343d8cf9766574f4110f89f5860388a2595 | |
| parent | a7431d9c458d873c3d39e4e43dca7ffbd575b1ff (diff) | |
| parent | cbbb57f7097b9a61154d59bdd965811e0650befe (diff) | |
| download | sec-certs-47863129815d4d283ae2cdc4212492953a6dde4b.tar.gz sec-certs-47863129815d4d283ae2cdc4212492953a6dde4b.tar.zst sec-certs-47863129815d4d283ae2cdc4212492953a6dde4b.zip | |
Merge pull request #171 from crocs-muni/fix-cpe-matching-inconsistencies
Fix cpe matching inconsistencies
| -rw-r--r-- | examples/cc_cpe_labeling.py | 4 | ||||
| -rw-r--r-- | sec_certs/constants.py | 2 | ||||
| -rw-r--r-- | sec_certs/dataset/cpe.py | 73 | ||||
| -rw-r--r-- | sec_certs/dataset/cve.py | 22 | ||||
| -rw-r--r-- | sec_certs/dataset/dataset.py | 20 | ||||
| -rw-r--r-- | sec_certs/dataset/fips.py | 2 | ||||
| -rw-r--r-- | sec_certs/helpers.py | 44 | ||||
| -rw-r--r-- | sec_certs/model/cpe_matching.py | 49 | ||||
| -rw-r--r-- | sec_certs/model/dependency_finder.py | 16 | ||||
| -rw-r--r-- | sec_certs/model/evaluation.py | 21 | ||||
| -rw-r--r-- | sec_certs/sample/certificate.py | 14 | ||||
| -rw-r--r-- | sec_certs/sample/common_criteria.py | 51 | ||||
| -rw-r--r-- | sec_certs/sample/cpe.py | 29 | ||||
| -rw-r--r-- | sec_certs/sample/fips.py | 29 | ||||
| -rw-r--r-- | tests/test_cc_heuristics.py | 41 |
15 files changed, 272 insertions, 145 deletions
diff --git a/examples/cc_cpe_labeling.py b/examples/cc_cpe_labeling.py index 89527bbe..752b8052 100644 --- a/examples/cc_cpe_labeling.py +++ b/examples/cc_cpe_labeling.py @@ -23,7 +23,7 @@ def main(): dset.get_certs_from_web(to_download=True) # Automatically match CPEs and CVEs - dset.compute_cpe_heuristics() + _, cpe_dset, _ = dset.compute_cpe_heuristics() dset.compute_related_cves() # Load dataset of ground truth CPE labels @@ -37,7 +37,7 @@ def main(): # Evaluate CPE matching performance metrics (on validation set) and dump classification report into json y_valid = [(x.heuristics.verified_cpe_matches) for x in validation_certs] - evaluate(validation_certs, y_valid, "./my_debug_dataset/classification_report.json") + evaluate(validation_certs, y_valid, "./my_debug_dataset/classification_report.json", cpe_dset) logger.info(f"{dset.json_path} should now contain fully labeled dataset.") diff --git a/sec_certs/constants.py b/sec_certs/constants.py index 2ea0622e..13eb1aba 100644 --- a/sec_certs/constants.py +++ b/sec_certs/constants.py @@ -17,6 +17,8 @@ class CertFramework(Enum): FIPS = "FIPS" +CPE_VERSION_NA = "-" + FIPS_BASE_URL = "https://csrc.nist.gov" FIPS_MODULE_URL = "https://csrc.nist.gov/projects/cryptographic-module-validation-program/certificate/" diff --git a/sec_certs/dataset/cpe.py b/sec_certs/dataset/cpe.py index b9c04cad..43ae3365 100644 --- a/sec_certs/dataset/cpe.py +++ b/sec_certs/dataset/cpe.py @@ -1,15 +1,16 @@ +import copy import itertools import logging import tempfile import xml.etree.ElementTree as ET import zipfile -from dataclasses import dataclass, field +from dataclasses import InitVar, dataclass, field from pathlib import Path from typing import Any, ClassVar, Dict, Iterator, List, Set, Tuple, Union, cast import pandas as pd -import sec_certs.helpers as helpers +from sec_certs import constants, helpers from sec_certs.dataset.cve import CVEDataset from sec_certs.sample.cpe import CPE, cached_cpe from sec_certs.serialization.json import ComplexSerializableType, serialize @@ -22,13 +23,18 @@ class CPEDataset(ComplexSerializableType): was_enhanced_with_vuln_cpes: bool json_path: Path cpes: Dict[str, CPE] - vendor_to_versions: Dict[str, Set[str]] = field(init=False) # Look-up dict cpe_vendor: list of viable versions + vendor_to_versions: Dict[str, Set[str]] = field( + init=False, default_factory=dict + ) # Look-up dict cpe_vendor: list of viable versions vendor_version_to_cpe: Dict[Tuple[str, str], Set[CPE]] = field( - init=False + init=False, default_factory=dict ) # Look-up dict (cpe_vendor, cpe_version): List of viable cpe items - title_to_cpes: Dict[str, Set[CPE]] = field(init=False) # Look-up dict title: List of cert items - vendors: Set[str] = field(init=False) + title_to_cpes: Dict[str, Set[CPE]] = field( + init=False, default_factory=dict + ) # Look-up dict title: List of cert items + vendors: Set[str] = field(init=False, default_factory=set) + init_lookup_dicts: InitVar[bool] = True cpe_xml_basename: ClassVar[str] = "official-cpe-dictionary_v2.3.xml" cpe_url: ClassVar[str] = "https://nvd.nist.gov/feeds/xml/cpe/dictionary/" + cpe_xml_basename + ".zip" @@ -47,13 +53,20 @@ class CPEDataset(ComplexSerializableType): def __contains__(self, item: CPE) -> bool: if not isinstance(item, CPE): raise ValueError(f"{item} is not of CPE class") - return item.uri in self.cpes.keys() + return item.uri in self.cpes.keys() and self.cpes[item.uri] == item + + def __eq__(self, other: object) -> bool: + return isinstance(other, CPEDataset) and self.cpes == other.cpes @property def serialized_attributes(self) -> List[str]: return ["was_enhanced_with_vuln_cpes", "json_path", "cpes"] - def __post_init__(self) -> None: + def __post_init__(self, init_lookup_dicts: bool): + if init_lookup_dicts: + self.build_lookup_dicts() + + def build_lookup_dicts(self) -> None: """ Will build look-up dictionaries that are used for fast matching """ @@ -76,7 +89,7 @@ class CPEDataset(ComplexSerializableType): self.title_to_cpes[cpe.title].add(cpe) @classmethod - def from_web(cls, json_path: Union[str, Path]) -> "CPEDataset": + def from_web(cls, json_path: Union[str, Path], init_lookup_dicts: bool = True) -> "CPEDataset": 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") @@ -85,10 +98,12 @@ 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) + return cls.from_xml(xml_path, json_path, init_lookup_dicts) @classmethod - def from_xml(cls, xml_path: Union[str, Path], json_path: Union[str, Path]) -> "CPEDataset": + 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.") root = ET.parse(xml_path).getroot() dct = {} @@ -108,7 +123,8 @@ class CPEDataset(ComplexSerializableType): cpe_uri = found_cpe_uri.attrib["name"] dct[cpe_uri] = cached_cpe(cpe_uri, title) - return cls(False, Path(json_path), dct) + + return cls(False, Path(json_path), dct, init_lookup_dicts) @classmethod def from_json(cls, input_path: Union[str, Path]) -> "CPEDataset": @@ -117,8 +133,8 @@ class CPEDataset(ComplexSerializableType): return dset @classmethod - def from_dict(cls, dct: Dict[str, Any]) -> "CPEDataset": - return cls(dct["was_enhanced_with_vuln_cpes"], Path("../"), dct["cpes"]) + def from_dict(cls, dct: Dict[str, Any], init_lookup_dicts: bool = True) -> "CPEDataset": + return cls(dct["was_enhanced_with_vuln_cpes"], Path("../"), dct["cpes"], init_lookup_dicts) def to_pandas(self) -> pd.DataFrame: df = pd.DataFrame([x.pandas_tuple for x in self], columns=CPE.pandas_columns) @@ -127,6 +143,24 @@ class CPEDataset(ComplexSerializableType): @serialize def enhance_with_cpes_from_cve_dataset(self, cve_dset: Union[CVEDataset, str, Path]) -> None: + def adding_condition( + considered_cpe: CPE, + vndr_item_lookup: Set[Tuple[str, str]], + vndr_item_version_lookup: Set[Tuple[str, str, str]], + ) -> bool: + if ( + considered_cpe.version == constants.CPE_VERSION_NA + and (considered_cpe.vendor, considered_cpe.item_name) not in vndr_item_lookup + ): + return True + elif ( + considered_cpe.version != constants.CPE_VERSION_NA + and (considered_cpe.vendor, considered_cpe.item_name, considered_cpe.version) + not in vndr_item_version_lookup + ): + return True + return False + if isinstance(cve_dset, (str, Path)): cve_dset = CVEDataset.from_json(cve_dset) @@ -136,9 +170,16 @@ class CPEDataset(ComplexSerializableType): old_len = len(self.cpes) + # We only enrich if tuple (vendor, item_name) is not already in the dataset + 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 cpe not in self: - self[cpe.uri] = cpe + 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 + self[new_cpe.uri] = new_cpe + self.build_lookup_dicts() logger.info(f"Enriched the CPE dataset with {len(self.cpes) - old_len} new CPE records.") self.was_enhanced_with_vuln_cpes = True diff --git a/sec_certs/dataset/cve.py b/sec_certs/dataset/cve.py index 4dc1fadc..a18364d7 100644 --- a/sec_certs/dataset/cve.py +++ b/sec_certs/dataset/cve.py @@ -26,7 +26,7 @@ logger = logging.getLogger(__name__) @dataclass class CVEDataset(ComplexSerializableType): cves: Dict[str, CVE] - cpe_to_cve_ids_lookup: Dict[str, List[str]] = field(init=False) + cpe_to_cve_ids_lookup: Dict[str, Set[str]] = field(init=False) cve_url: Final[str] = "https://nvd.nist.gov/feeds/json/cve/1.1/nvdcve-1.1-" cpe_match_feed_url: Final[str] = "https://nvd.nist.gov/feeds/json/cpematch/1.0/nvdcpematch-1.0.json.zip" @@ -51,6 +51,7 @@ class CVEDataset(ComplexSerializableType): def build_lookup_dict(self, use_nist_mapping: bool = True, nist_matching_filepath: Optional[Path] = None): """ + Builds look-up dictionary CPE -> Set[CVE] Developer's note: There are 3 CPEs that are present in the cpe matching feed, but are badly processed by CVE feed, in which case they won't be found as a key in the dictionary. We intentionally ignore those. Feel free to add corner cases and manual fixes. According to our investigation, the suffereing CPEs are: @@ -66,19 +67,20 @@ class CVEDataset(ComplexSerializableType): if use_nist_mapping: matching_dict = self.get_nist_cpe_matching_dict(nist_matching_filepath) + cve: CVE for cve in helpers.tqdm(self, desc="Building-up lookup dictionaries for fast CVE matching"): # See note above, we use matching_dict.get(cpe, []) instead of matching_dict[cpe] as would be expected if use_nist_mapping: - vulnerable_configurations = itertools.chain.from_iterable( - [matching_dict.get(cpe, []) for cpe in cve.vulnerable_cpes] + vulnerable_configurations = list( + itertools.chain.from_iterable([matching_dict.get(cpe, []) for cpe in cve.vulnerable_cpes]) ) else: vulnerable_configurations = cve.vulnerable_cpes for cpe in vulnerable_configurations: if cpe.uri not in self.cpe_to_cve_ids_lookup: - self.cpe_to_cve_ids_lookup[cpe.uri] = [cve.cve_id] + self.cpe_to_cve_ids_lookup[cpe.uri] = {cve.cve_id} else: - self.cpe_to_cve_ids_lookup[cpe.uri].append(cve.cve_id) + self.cpe_to_cve_ids_lookup[cpe.uri].add(cve.cve_id) @classmethod def download_cves(cls, output_path_str: str, start_year: int, end_year: int): @@ -144,9 +146,7 @@ class CVEDataset(ComplexSerializableType): dset = json.load(handle, cls=CustomJSONDecoder) return dset - def get_cve_ids_for_cpe_uri(self, cpe_uri: str) -> Optional[List[str]]: - if not isinstance(cpe_uri, str): - return None + def get_cve_ids_for_cpe_uri(self, cpe_uri: str) -> Optional[Set[str]]: return self.cpe_to_cve_ids_lookup.get(cpe_uri, None) def filter_related_cpes(self, relevant_cpes: Set[CPE]): @@ -175,7 +175,11 @@ class CVEDataset(ComplexSerializableType): df = pd.DataFrame([x.pandas_tuple for x in self], columns=CVE.pandas_columns) return df.set_index("cve_id") - def get_nist_cpe_matching_dict(self, input_filepath: Optional[Path]): + def get_nist_cpe_matching_dict(self, input_filepath: Optional[Path]) -> Dict[CPE, List[CPE]]: + """ + Computes dictionary that maps complex CPEs to list of simple CPEs. + """ + def parse_key_cpe(field: Dict) -> CPE: start_version = None if "versionStartIncluding" in field: diff --git a/sec_certs/dataset/dataset.py b/sec_certs/dataset/dataset.py index 78fef0b6..2745f35f 100644 --- a/sec_certs/dataset/dataset.py +++ b/sec_certs/dataset/dataset.py @@ -162,13 +162,13 @@ class Dataset(Generic[CertSubType], ABC): logger.error(f"Corrupted file at: {p}") p.unlink() - def _prepare_cpe_dataset(self, download_fresh_cpes: bool = False) -> CPEDataset: + def _prepare_cpe_dataset(self, download_fresh_cpes: bool = False, init_lookup_dicts: bool = True) -> CPEDataset: logger.info("Preparing CPE dataset.") if not self.auxillary_datasets_dir.exists(): self.auxillary_datasets_dir.mkdir(parents=True) if not self.cpe_dataset_path.exists() or download_fresh_cpes is True: - cpe_dataset = CPEDataset.from_web(self.cpe_dataset_path) + cpe_dataset = CPEDataset.from_web(self.cpe_dataset_path, init_lookup_dicts) cpe_dataset.to_json(str(self.cpe_dataset_path)) else: cpe_dataset = CPEDataset.from_json(str(self.cpe_dataset_path)) @@ -191,11 +191,6 @@ class Dataset(Generic[CertSubType], ABC): cve_dataset.build_lookup_dict(use_nist_cpe_matching_dict, self.nist_cve_cpe_matching_dset_path) return cve_dataset - def _compute_candidate_versions(self) -> None: - logger.info("Computing heuristics: possible product versions in sample name") - for cert in cast(Iterator[Certificate], self): - cert.compute_heuristics_version() - def _compute_cpe_matches( self, download_fresh_cpes: bool = False ) -> Tuple[CPEClassifier, CPEDataset, Optional[CVEDataset]]: @@ -219,15 +214,18 @@ class Dataset(Generic[CertSubType], ABC): return True logger.info("Computing heuristics: Finding CPE matches for certificates") - cpe_dset = self._prepare_cpe_dataset(download_fresh_cpes) + cpe_dset = self._prepare_cpe_dataset(download_fresh_cpes, init_lookup_dicts=False) cve_dset = None if not cpe_dset.was_enhanced_with_vuln_cpes: - cve_dset = self._prepare_cve_dataset(False) - cpe_dset.enhance_with_cpes_from_cve_dataset(cve_dset) + cve_dset = self._prepare_cve_dataset(download_fresh_cves=False) + cpe_dset.enhance_with_cpes_from_cve_dataset(cve_dset) # this also calls build_lookup_dicts() on cpe_dset + else: + cpe_dset.build_lookup_dicts() clf = CPEClassifier(config.cpe_matching_threshold, config.cpe_n_max_matches) clf.fit([x for x in cpe_dset if filter_condition(x)]) + cert: CertSubType for cert in helpers.tqdm(self, desc="Predicting CPE matches with the classifier"): cert.compute_heuristics_cpe_match(clf) @@ -235,7 +233,6 @@ class Dataset(Generic[CertSubType], ABC): @serialize def compute_cpe_heuristics(self) -> Tuple[CPEClassifier, CPEDataset, Optional[CVEDataset]]: - self._compute_candidate_versions() return self._compute_cpe_matches() def to_label_studio_json(self, output_path: Union[str, Path]) -> None: @@ -327,6 +324,7 @@ class Dataset(Generic[CertSubType], ABC): relevant_cpes = set(itertools.chain.from_iterable([x.heuristics.cpe_matches for x in cpe_rich_certs])) cve_dset.filter_related_cpes(relevant_cpes) + cert: Certificate for cert in helpers.tqdm(cpe_rich_certs, desc="Computing related CVES"): cert.compute_heuristics_related_cves(cve_dset) diff --git a/sec_certs/dataset/fips.py b/sec_certs/dataset/fips.py index 2148038f..91926211 100644 --- a/sec_certs/dataset/fips.py +++ b/sec_certs/dataset/fips.py @@ -304,7 +304,7 @@ class FIPSDataset(Dataset[FIPSCertificate], ComplexSerializableType): ) cert: FIPSCertificate for cert in self.certs.values(): - cert.heuristics = FIPSCertificate.FIPSHeuristics(None, [], [], 0) + cert.heuristics = FIPSCertificate.FIPSHeuristics(dict(), [], [], 0) self.match_algs() diff --git a/sec_certs/helpers.py b/sec_certs/helpers.py index 8224011e..159e42e0 100644 --- a/sec_certs/helpers.py +++ b/sec_certs/helpers.py @@ -35,6 +35,7 @@ from sec_certs.constants import ( logger = logging.getLogger(__name__) +# TODO: Once typehints in tqdm are implemented, we should use them: https://github.com/tqdm/tqdm/issues/260 def tqdm(*args, **kwargs): if "disable" in kwargs: return tqdm_original(*args, **kwargs) @@ -1051,7 +1052,7 @@ def gen_dict_extract(dct: Dict, searched_key: Hashable = "count") -> Generator[A yield key, result -def compute_heuristics_version(cert_name: str) -> List[str]: +def compute_heuristics_version(cert_name: str) -> Set[str]: """ Will extract possible versions from the name of sample """ @@ -1076,10 +1077,10 @@ def compute_heuristics_version(cert_name: str) -> List[str]: # return identified_versions if identified_versions else ['-'] if not matches: - return ["-"] + return {constants.CPE_VERSION_NA} matched = [re.search(normalizer, x) for x in matches] - return [x.group() for x in matched if x is not None] + return {x.group() for x in matched if x is not None} def tokenize_dataset(dset: List[str], keywords: Set[str]) -> np.ndarray: @@ -1088,3 +1089,40 @@ def tokenize_dataset(dset: List[str], keywords: Set[str]) -> np.ndarray: def tokenize(string: str, keywords: Set[str]) -> str: return " ".join([x for x in string.split() if x.lower() in keywords]) + + +# Credit: https://stackoverflow.com/questions/18092354/ +def split_unescape(s: str, delim: str, escape: str = "\\", unescape: bool = True) -> List[str]: + """ + >>> split_unescape('foo,bar', ',') + ['foo', 'bar'] + >>> split_unescape('foo$,bar', ',', '$') + ['foo,bar'] + >>> split_unescape('foo$$,bar', ',', '$', unescape=True) + ['foo$', 'bar'] + >>> split_unescape('foo$$,bar', ',', '$', unescape=False) + ['foo$$', 'bar'] + >>> split_unescape('foo$', ',', '$', unescape=True) + ['foo$'] + """ + ret = [] + current = [] + itr = iter(s) + for ch in itr: + if ch == escape: + try: + # skip the next character; it has been escaped! + if not unescape: + current.append(escape) + current.append(next(itr)) + except StopIteration: + if unescape: + current.append(escape) + elif ch == delim: + # split! (add current to the list and reset it) + ret.append("".join(current)) + current = [] + else: + current.append(ch) + ret.append("".join(current)) + return ret diff --git a/sec_certs/model/cpe_matching.py b/sec_certs/model/cpe_matching.py index 79cba635..60d3c158 100644 --- a/sec_certs/model/cpe_matching.py +++ b/sec_certs/model/cpe_matching.py @@ -1,5 +1,6 @@ import itertools import logging +import operator import re from typing import Dict, List, Optional, Set, Tuple @@ -7,6 +8,7 @@ from rapidfuzz import fuzz from sklearn.base import BaseEstimator import sec_certs.helpers as helpers +from sec_certs import constants from sec_certs.sample.cpe import CPE logger = logging.getLogger(__name__) @@ -67,7 +69,7 @@ class CPEClassifier(BaseEstimator): else: self.vendor_version_to_cpe_[(cpe.vendor, cpe.version)].add(cpe) - def predict(self, X: List[Tuple[str, str, str]]) -> List[Optional[List[str]]]: + 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) @@ -79,10 +81,10 @@ class CPEClassifier(BaseEstimator): self, vendor: Optional[str], product_name: str, - versions: List[str], + versions: Set[str], relax_version: bool = False, relax_title: bool = False, - ) -> Optional[List[str]]: + ) -> Optional[Set[str]]: """ Predict List of CPE uris for triplet (vendor, product_name, list_of_version). The prediction is made as follows: 1. Sanitize all strings @@ -109,9 +111,10 @@ class CPEClassifier(BaseEstimator): ] threshold = self.match_threshold if not relax_version else 100 final_matches_aux: List[Tuple[float, CPE]] = list(filter(lambda x: x[0] >= threshold, zip(ratings, candidates))) - final_matches: Optional[List[str]] = [ - x[1].uri for x in final_matches_aux[: self.n_max_matches] if x[1].uri is not None - ] + final_matches_aux = sorted(final_matches_aux, key=operator.itemgetter(0, 1), reverse=True) + final_matches: Optional[Set[str]] = set( + [x[1].uri for x in final_matches_aux[: self.n_max_matches] if x[1].uri is not None] + ) if not relax_title and not final_matches: final_matches = self.predict_single_cert( @@ -120,7 +123,7 @@ class CPEClassifier(BaseEstimator): if not relax_version and not final_matches: final_matches = self.predict_single_cert( - vendor, product_name, ["-"], relax_version=True, relax_title=relax_title + vendor, product_name, {constants.CPE_VERSION_NA}, relax_version=True, relax_title=relax_title ) return final_matches if final_matches else None @@ -129,8 +132,8 @@ class CPEClassifier(BaseEstimator): self, cpe: CPE, product_name: str, - candidate_vendors: Optional[List[str]], - versions: List[str], + candidate_vendors: Optional[Set[str]], + versions: Set[str], relax_title: bool = False, ) -> float: """ @@ -183,13 +186,13 @@ class CPEClassifier(BaseEstimator): return string.replace("®", "").replace("™", "") @staticmethod - def _strip_manufacturer_and_version(string: str, manufacturers: Optional[List[str]], versions: List[str]) -> str: - to_strip = versions + manufacturers if manufacturers else versions + def _strip_manufacturer_and_version(string: str, manufacturers: Optional[Set[str]], versions: Set[str]) -> str: + to_strip = versions | manufacturers if manufacturers else versions for x in to_strip: string = string.lower().replace(CPEClassifier._replace_special_chars_with_space(x.lower()), "").strip() return string - def _process_manufacturer(self, manufacturer: str, result: Set) -> Optional[List[str]]: + def _process_manufacturer(self, manufacturer: str, result: Set) -> Set[str]: tokenized = manufacturer.split() if tokenized[0] in self.vendors_: result.add(tokenized[0]) @@ -208,20 +211,20 @@ class CPEClassifier(BaseEstimator): result.add("athena-scs") if tokenized[0] == "the" and not result: candidate_result = self.get_candidate_list_of_vendors(" ".join(tokenized[1:])) - return list(candidate_result) if candidate_result else None + return set(candidate_result) if candidate_result else set() - return list(result) if result else None + return set(result) if result else set() - def get_candidate_list_of_vendors(self, manufacturer: Optional[str]) -> Optional[List[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. """ + result: Set[str] = set() if not manufacturer: - return None + return result - result: Set = set() splits = re.compile(r"[,/]").findall(manufacturer) if splits: @@ -229,8 +232,8 @@ class CPEClassifier(BaseEstimator): 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_used = list(set(itertools.chain.from_iterable([x for x in result_aux if x]))) - return result_used if result_used else None + result_used = set(set(itertools.chain.from_iterable([x for x in result_aux if x]))) + return result_used if result_used else set() if manufacturer in self.vendors_: result.add(manufacturer) @@ -238,7 +241,7 @@ class CPEClassifier(BaseEstimator): return self._process_manufacturer(manufacturer, result) def get_candidate_vendor_version_pairs( - self, cert_candidate_cpe_vendors: Optional[List[str]], cert_candidate_versions: List[str] + 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 @@ -247,7 +250,7 @@ class CPEClassifier(BaseEstimator): @return: 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: List[str]) -> bool: + def is_cpe_version_among_cert_versions(cpe_version: Optional[str], cert_versions: Set[str]) -> bool: def simple_startswith(seeked_version: str, checked_string: str) -> bool: if seeked_version == checked_string: return True @@ -278,9 +281,7 @@ 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: Optional[List[str]], candidate_versions: List[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 diff --git a/sec_certs/model/dependency_finder.py b/sec_certs/model/dependency_finder.py index 15b0a31b..83d6cfb9 100644 --- a/sec_certs/model/dependency_finder.py +++ b/sec_certs/model/dependency_finder.py @@ -1,11 +1,11 @@ -from typing import Dict, List, Optional, Set, Tuple, Union +from typing import Dict, Optional, Set, Tuple from sec_certs.sample.common_criteria import CommonCriteriaCert Certificates = Dict[str, CommonCriteriaCert] -ReferencedByDirect = Dict[str, List[str]] +ReferencedByDirect = Dict[str, Set[str]] ReferencedByIndirect = Dict[str, Set[str]] -Dependencies = Dict[str, Dict[str, Union[Optional[List[str]], Optional[Set[str]]]]] +Dependencies = Dict[str, Dict[str, Optional[Set[str]]]] class DependencyFinder: @@ -15,9 +15,9 @@ class DependencyFinder: @staticmethod def _update_direct_references(referenced_by: ReferencedByDirect, cert_id: str, this_cert_id: str) -> None: if cert_id not in referenced_by: - referenced_by[cert_id] = [] + referenced_by[cert_id] = set() if this_cert_id not in referenced_by[cert_id]: - referenced_by[cert_id].append(this_cert_id) + referenced_by[cert_id].add(this_cert_id) @staticmethod def _process_references(referenced_by: ReferencedByDirect, referenced_by_indirect: ReferencedByIndirect) -> None: @@ -85,7 +85,7 @@ class DependencyFinder: return filter_indirect if filter_indirect else None @staticmethod - def _get_affected_directly(cert: str, referenced_by_direct: ReferencedByDirect) -> Optional[List[str]]: + def _get_affected_directly(cert: str, referenced_by_direct: ReferencedByDirect) -> Optional[Set[str]]: return referenced_by_direct.get(cert, None) @staticmethod @@ -118,9 +118,9 @@ class DependencyFinder: cert_id, referenced_by_indirect ) - def get_directly_affected_by(self, dgst: str) -> Optional[List[str]]: + def get_directly_affected_by(self, dgst: str) -> Optional[Set[str]]: res = self.dependencies[dgst].get("directly_affected_by", None) - return list(res) if res else None + return set(res) if res else None def get_indirectly_affected_by(self, dgst: str) -> Optional[Set[str]]: res = self.dependencies[dgst].get("indirectly_affected_by", None) diff --git a/sec_certs/model/evaluation.py b/sec_certs/model/evaluation.py index 5e51dbb2..6458f649 100644 --- a/sec_certs/model/evaluation.py +++ b/sec_certs/model/evaluation.py @@ -35,7 +35,7 @@ def compute_precision(y: np.ndarray, y_pred: np.ndarray, **kwargs) -> float: def evaluate( x_valid: List[Union[CommonCriteriaCert, FIPSCertificate]], - y_valid: List[Optional[List[str]]], + y_valid: List[Optional[Set[str]]], outpath: Optional[Union[Path, str]], cpe_dset: CPEDataset, ) -> None: @@ -48,10 +48,13 @@ def evaluate( n_newly_identified = 0 for cert, predicted_cpes, verified_cpes in zip(x_valid, y_pred, y_valid): - verified_cpes_set = set(verified_cpes) if verified_cpes else set() - verified_cpes_dict = {x: cpe_dset[x].title if cpe_dset[x].title else x for x in verified_cpes_set} - predicted_cpes_set = set(predicted_cpes) if predicted_cpes else set() - predicted_cpes_dict = {x: cpe_dset[x].title if cpe_dset[x].title else x for x in predicted_cpes_set} + if not verified_cpes: + verified_cpes = set() + verified_cpes_dict = {x: cpe_dset[x].title if cpe_dset[x].title else x for x in verified_cpes} + + if not predicted_cpes: + predicted_cpes = set() + predicted_cpes_dict = {x: cpe_dset[x].title if cpe_dset[x].title else x for x in predicted_cpes} cert_name = cert.name if isinstance(cert, CommonCriteriaCert) else cert.web_scan.module_name vendor = cert.manufacturer if isinstance(cert, CommonCriteriaCert) else cert.web_scan.vendor @@ -63,14 +66,14 @@ def evaluate( "manually_assigned_cpes": verified_cpes_dict, } - if verified_cpes_set.issubset(predicted_cpes_set): + if verified_cpes.issubset(predicted_cpes): correctly_classified.append(record) else: badly_classified.append(record) - if not verified_cpes_set and predicted_cpes_set: + if not verified_cpes and predicted_cpes: n_new_certs_with_match += 1 - n_newly_identified += len(predicted_cpes_set - verified_cpes_set) + n_newly_identified += len(predicted_cpes - verified_cpes) results = { "Precision": precision, @@ -85,4 +88,4 @@ def evaluate( if outpath: with Path(outpath).open("w") as handle: - json.dump(results, handle, indent=4, cls=CustomJSONEncoder) + json.dump(results, handle, indent=4, cls=CustomJSONEncoder, sort_keys=True) diff --git a/sec_certs/sample/certificate.py b/sec_certs/sample/certificate.py index a7f34d2a..c310dae2 100644 --- a/sec_certs/sample/certificate.py +++ b/sec_certs/sample/certificate.py @@ -4,7 +4,7 @@ import json import logging from abc import ABC, abstractmethod from pathlib import Path -from typing import Any, Dict, Generic, Type, TypeVar, Union +from typing import Any, Dict, Generic, Optional, Set, Type, TypeVar, Union from sec_certs.dataset.cve import CVEDataset from sec_certs.model.cpe_matching import CPEClassifier @@ -13,10 +13,16 @@ from sec_certs.serialization.json import ComplexSerializableType, CustomJSONDeco logger = logging.getLogger(__name__) T = TypeVar("T", bound="Certificate") +H = TypeVar("H", bound="Heuristics") -class Certificate(Generic[T], ABC, ComplexSerializableType): - heuristics: Any +class Heuristics: + cpe_matches: Optional[Set[str]] + related_cves: Optional[Set[str]] + + +class Certificate(Generic[T, H], ABC, ComplexSerializableType): + heuristics: H def __init__(self, *args, **kwargs): pass @@ -59,7 +65,7 @@ class Certificate(Generic[T], ABC, ComplexSerializableType): return json.load(handle, cls=CustomJSONDecoder) @abstractmethod - def compute_heuristics_version(self) -> None: + def _compute_heuristics_version(self) -> None: raise NotImplementedError("Not meant to be implemented") @abstractmethod diff --git a/sec_certs/sample/common_criteria.py b/sec_certs/sample/common_criteria.py index 5dda7326..68258f67 100644 --- a/sec_certs/sample/common_criteria.py +++ b/sec_certs/sample/common_criteria.py @@ -12,7 +12,7 @@ from bs4 import Tag from sec_certs import constants as constants from sec_certs import helpers from sec_certs.model.cpe_matching import CPEClassifier -from sec_certs.sample.certificate import Certificate, logger +from sec_certs.sample.certificate import Certificate, Heuristics, logger from sec_certs.sample.protection_profile import ProtectionProfile from sec_certs.serialization.json import ComplexSerializableType from sec_certs.serialization.pandas import PandasSerializableType @@ -26,7 +26,11 @@ HEADERS = { } -class CommonCriteriaCert(Certificate["CommonCriteriaCert"], PandasSerializableType, ComplexSerializableType): +class CommonCriteriaCert( + Certificate["CommonCriteriaCert", "CommonCriteriaCert.CCHeuristics"], + PandasSerializableType, + ComplexSerializableType, +): cc_url = "http://www.commoncriteriaportal.org" empty_st_url = "http://www.commoncriteriaportal.org/files/epfiles/" @@ -198,27 +202,21 @@ class CommonCriteriaCert(Certificate["CommonCriteriaCert"], PandasSerializableTy return processed if (processed := self.processed_cert_id) else self.keywords_cert_id @dataclass - class CCHeuristics(ComplexSerializableType): - extracted_versions: Optional[List[str]] = field(default=None) + class CCHeuristics(Heuristics, ComplexSerializableType): + 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) related_cves: Optional[Set[str]] = field(default=None) cert_lab: Optional[List[str]] = field(default=None) cert_id: Optional[str] = field(default=None) - directly_affected_by: Optional[List[str]] = field(default=None) + directly_affected_by: Optional[Set[str]] = field(default=None) indirectly_affected_by: Optional[Set[str]] = field(default=None) directly_affecting: Optional[Set[str]] = field(default=None) indirectly_affecting: Optional[Set[str]] = field(default=None) - cpe_candidate_vendors: Optional[List[str]] = field(init=False) @property def serialized_attributes(self) -> List[str]: - all_vars = copy.deepcopy(super().serialized_attributes) - all_vars.remove("cpe_candidate_vendors") - return all_vars - - def __post_init__(self) -> None: - self.cpe_candidate_vendors = None + return copy.deepcopy(super().serialized_attributes) pandas_columns: ClassVar[List[str]] = [ "dgst", @@ -284,18 +282,9 @@ class CommonCriteriaCert(Certificate["CommonCriteriaCert"], PandasSerializableTy self.manufacturer_web = helpers.sanitize_link(manufacturer_web) self.protection_profiles = protection_profiles self.maintenance_updates = maintenance_updates - - if state is None: - state = self.InternalState() - self.state = state - - if pdf_data is None: - pdf_data = self.PdfData() - self.pdf_data = pdf_data - - if heuristics is None: - heuristics = self.CCHeuristics() - self.heuristics = heuristics + 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 @property def dgst(self) -> str: @@ -643,10 +632,13 @@ class CommonCriteriaCert(Certificate["CommonCriteriaCert"], PandasSerializableTy cert.state.errors.append(response) return cert - def compute_heuristics_version(self) -> None: + def _compute_heuristics_version(self) -> None: self.heuristics.extracted_versions = helpers.compute_heuristics_version(self.name) def compute_heuristics_cpe_match(self, cpe_classifier: CPEClassifier) -> None: + self._compute_heuristics_version() + assert self.heuristics.extracted_versions is not None + self.heuristics.cpe_matches = cpe_classifier.predict_single_cert( self.manufacturer, self.name, self.heuristics.extracted_versions ) @@ -770,6 +762,8 @@ class CommonCriteriaCert(Certificate["CommonCriteriaCert"], PandasSerializableTy return new_cert_id 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() if CommonCriteriaCert._is_anssi_cert(cert_id): @@ -794,10 +788,13 @@ class CommonCriteriaCert(Certificate["CommonCriteriaCert"], PandasSerializableTy "ocsi": CommonCriteriaCert._fix_ocsi_cert_id, } - cert_lab = self.get_cert_laboratory() + try: + cert_lab = self.get_cert_laboratory() + except ValueError: + return None # No need for any fix, bcs we do not know how - if self.heuristics.cert_id is None or cert_lab == "unknown": + if cert_lab == "unknown": return None self.heuristics.cert_id = fix_methods[cert_lab](self.pdf_data.cert_id) diff --git a/sec_certs/sample/cpe.py b/sec_certs/sample/cpe.py index a5ce6921..bbbf1c39 100644 --- a/sec_certs/sample/cpe.py +++ b/sec_certs/sample/cpe.py @@ -2,6 +2,7 @@ from dataclasses import dataclass from functools import lru_cache from typing import Any, ClassVar, Dict, List, Optional, Tuple +from sec_certs import constants, helpers from sec_certs.serialization.json import ComplexSerializableType from sec_certs.serialization.pandas import PandasSerializableType @@ -37,17 +38,26 @@ class CPE(PandasSerializableType, ComplexSerializableType): ): super().__init__() self.uri = uri - self.vendor = " ".join(self.uri.split(":")[3].split("_")) - self.item_name = " ".join(self.uri.split(":")[4].split("_")) - self.version = self.uri.split(":")[5] + + splitted = helpers.split_unescape(self.uri, ":") + self.vendor = " ".join(splitted[3].split("_")) + self.item_name = " ".join(splitted[4].split("_")) + self.version = self.normalize_version(" ".join(splitted[5].split("_"))) self.title = title self.start_version = start_version self.end_version = end_version def __lt__(self, other: "CPE") -> bool: - if self.title is None or other.title is None: - raise RuntimeError("Cannot compare CPEs because title is missing.") - return self.title < other.title + return self.uri < other.uri + + @staticmethod + def normalize_version(version: str) -> str: + """ + Maps common empty versions (empty '', asterisk '*') to unified empty version (constants.CPE_VERSION_NA) + """ + if version in {"", "*"}: + return constants.CPE_VERSION_NA + return version @classmethod def from_dict(cls, dct: Dict[str, Any]) -> "CPE": @@ -81,12 +91,7 @@ class CPE(PandasSerializableType, ComplexSerializableType): return hash((self.uri, self.start_version, self.end_version)) def __eq__(self, other: object) -> bool: - return ( - isinstance(other, self.__class__) - and self.uri == other.uri - and self.start_version == other.start_version - and self.end_version == other.end_version - ) + return isinstance(other, self.__class__) and self.uri == other.uri @lru_cache(maxsize=4096) diff --git a/sec_certs/sample/fips.py b/sec_certs/sample/fips.py index 562a4a3e..97fd30bd 100644 --- a/sec_certs/sample/fips.py +++ b/sec_certs/sample/fips.py @@ -17,12 +17,12 @@ from sec_certs.config.configuration import config from sec_certs.constants import LINE_SEPARATOR from sec_certs.helpers import fips_dgst, load_cert_file, normalize_match_string, save_modified_cert_file from sec_certs.model.cpe_matching import CPEClassifier -from sec_certs.sample.certificate import Certificate, logger +from sec_certs.sample.certificate import Certificate, Heuristics, logger from sec_certs.sample.cpe import CPE from sec_certs.serialization.json import ComplexSerializableType -class FIPSCertificate(Certificate["FIPSCertificate"], ComplexSerializableType): +class FIPSCertificate(Certificate["FIPSCertificate", "FIPSCertificate.FIPSHeuristics"], ComplexSerializableType): FIPS_BASE_URL: ClassVar[str] = "https://csrc.nist.gov" FIPS_MODULE_URL: ClassVar[ str @@ -167,17 +167,16 @@ class FIPSCertificate(Certificate["FIPSCertificate"], ComplexSerializableType): return str(self.cert_id) @dataclass(eq=True) - class FIPSHeuristics(ComplexSerializableType): - keywords: Optional[Dict[str, Dict]] + class FIPSHeuristics(Heuristics, ComplexSerializableType): + keywords: Dict[str, Dict] algorithms: List[Dict[str, Dict]] connections: List[str] unmatched_algs: int - extracted_versions: Optional[List[str]] = field(default=None) + extracted_versions: Optional[Set[str]] = field(default=None) cpe_matches: Optional[Set[str]] = field(default=None) verified_cpe_matches: Optional[Set[CPE]] = field(default=None) - related_cves: Optional[List[str]] = field(default=None) - cpe_candidate_vendors: Optional[List[str]] = field(init=False) + related_cves: Optional[Set[str]] = field(default=None) directly_affected_by: Optional[Set] = field(default=None) indirectly_affected_by: Optional[Set] = field(default=None) @@ -186,12 +185,7 @@ class FIPSCertificate(Certificate["FIPSCertificate"], ComplexSerializableType): @property def serialized_attributes(self) -> List[str]: - all_vars = copy.deepcopy(super().serialized_attributes) - all_vars.remove("cpe_candidate_vendors") - return all_vars - - def __post_init__(self) -> None: - self.cpe_candidate_vendors = None + return copy.deepcopy(super().serialized_attributes) @property def dgst(self) -> str: @@ -240,7 +234,7 @@ class FIPSCertificate(Certificate["FIPSCertificate"], ComplexSerializableType): self.cert_id = cert_id self.web_scan = web_scan self.pdf_scan = pdf_scan - self.heuristics = heuristics + self.heuristics: "FIPSCertificate.FIPSHeuristics" = heuristics self.state = state @classmethod @@ -544,7 +538,7 @@ class FIPSCertificate(Certificate["FIPSCertificate"], ComplexSerializableType): [] if not initialized else initialized.pdf_scan.algorithms, [], # connections ), - FIPSCertificate.FIPSHeuristics(None, [], [], 0), + FIPSCertificate.FIPSHeuristics(dict(), [], [], 0), state, ) @@ -840,7 +834,7 @@ class FIPSCertificate(Certificate["FIPSCertificate"], ComplexSerializableType): ) return vendor_split[0][:4] if len(vendor_split) > 0 else vendor - def compute_heuristics_version(self) -> None: + def _compute_heuristics_version(self) -> None: versions_for_extraction = "" if self.web_scan.module_name: versions_for_extraction += f" {self.web_scan.module_name}" @@ -851,6 +845,9 @@ class FIPSCertificate(Certificate["FIPSCertificate"], ComplexSerializableType): self.heuristics.extracted_versions = helpers.compute_heuristics_version(versions_for_extraction) def compute_heuristics_cpe_match(self, cpe_classifier: CPEClassifier) -> None: + self._compute_heuristics_version() + assert self.heuristics.extracted_versions is not None + if not self.web_scan.module_name: self.heuristics.cpe_matches = None else: diff --git a/tests/test_cc_heuristics.py b/tests/test_cc_heuristics.py index ec8a49ed..94567574 100644 --- a/tests/test_cc_heuristics.py +++ b/tests/test_cc_heuristics.py @@ -5,6 +5,7 @@ from typing import ClassVar, Dict, List from unittest import TestCase import tests.data.test_cc_heuristics +from sec_certs import constants from sec_certs.dataset.common_criteria import CCDataset from sec_certs.dataset.cpe import CPEDataset from sec_certs.dataset.cve import CVEDataset @@ -123,6 +124,40 @@ class TestCommonCriteriaHeuristics(TestCase): "The CPE lookup dictionary (vendor,version)->cpe does not match the template.", ) + def test_cpe_parsing(self): + potentially_problematic_cpes = { + 'cpe:2.3:a:bayashi:dopvstar\::0091:*:*:*:*:*:*:*"': ("bayashi", "dopvstar:", "0091"), # noqa: W605 + "cpe:2.3:a:moundlabs:\:\:mound\:\::2.1.6:*:*:*:*:*:*:*": ("moundlabs", "::mound::", "2.1.6"), # noqa: W605 + "cpe:2.3:a:lemonldap-ng:lemonldap\:\::*:*:*:*:*:*:*:*": ( # noqa: W605 + "lemonldap-ng", + "lemonldap::", + constants.CPE_VERSION_NA, + ), + "cpe:2.3:o:cisco:nx-os:5.0\\\\\\(3\\\\\\)u5\\\\\\(1g\\\\\\):*:*:*:*:*:*:*": ( + "cisco", + "nx-os", + "5.0\\(3\\)u5\\(1g\\)", + ), + "cpe:2.3:a:\\@thi.ng\\/egf_project:\\@thi.ng\\/egf:-:*:*:*:*:node.js:*:*": ( + "@thi.ng/egf project", + "@thi.ng/egf", + "-", + ), + "cpe:2.3:a:oracle:communications_diameter_signaling_router_idih\\:::*:*:*:*:*:*:*": ( + "oracle", + "communications diameter signaling router idih:", + constants.CPE_VERSION_NA, + ), + } + + for uri, tpl in potentially_problematic_cpes.items(): + cpe = CPE(uri) + self.assertEqual(cpe.vendor, tpl[0], "Parsed CPE vendor differs from expected vendor. Broken escaping?") + self.assertEqual( + cpe.item_name, tpl[1], "Parsed CPE item name differs from expected item name. Broken escaping?" + ) + self.assertEqual(cpe.version, tpl[2], "Parsed CPE version differs from expected version. Broken escaping?") + def test_cve_lookup_dicts(self): alt_lookup = {x: set(y) for x, y in self.cve_dset.cpe_to_cve_ids_lookup.items()} self.assertEqual( @@ -162,7 +197,7 @@ class TestCommonCriteriaHeuristics(TestCase): def test_version_extraction(self): self.assertEqual( self.cc_dset["ebd276cca70fd723"].heuristics.extracted_versions, - ["8.2"], + {"8.2"}, "The version extracted from the sample does not match the template", ) new_cert = CommonCriteriaCert( @@ -184,7 +219,7 @@ class TestCommonCriteriaHeuristics(TestCase): None, None, ) - new_cert.compute_heuristics_version() + new_cert._compute_heuristics_version() self.assertEqual( set(new_cert.heuristics.extracted_versions), {"5.4", "1.0"}, @@ -241,7 +276,7 @@ class TestCommonCriteriaHeuristics(TestCase): dependency_dataset._compute_dependencies() test_cert = dependency_dataset["692e91451741ef49"] - self.assertEqual(test_cert.heuristics.directly_affected_by, ["BSI-DSZ-CC-0370-2006"]) + self.assertEqual(test_cert.heuristics.directly_affected_by, {"BSI-DSZ-CC-0370-2006"}) self.assertEqual(test_cert.heuristics.indirectly_affected_by, {"BSI-DSZ-CC-0370-2006", "BSI-DSZ-CC-0517-2009"}) self.assertEqual(test_cert.heuristics.directly_affecting, {"BSI-DSZ-CC-0268-2005"}) self.assertEqual(test_cert.heuristics.indirectly_affecting, {"BSI-DSZ-CC-0268-2005"}) |
