diff options
| author | adamjanovsky | 2021-04-19 11:31:04 +0200 |
|---|---|---|
| committer | GitHub | 2021-04-19 11:31:04 +0200 |
| commit | fba10c7c2f874bb8debd1a357cc154da0613fed4 (patch) | |
| tree | 052909e8233cacf492ddc88ef733b8d475119da5 | |
| parent | 29823f8b9f92cb937b34b63f44dc7dc3da920d81 (diff) | |
| parent | 3a223490e85517647a3c585739d6d46fd6a90252 (diff) | |
| download | sec-certs-fba10c7c2f874bb8debd1a357cc154da0613fed4.tar.gz sec-certs-fba10c7c2f874bb8debd1a357cc154da0613fed4.tar.zst sec-certs-fba10c7c2f874bb8debd1a357cc154da0613fed4.zip | |
Merge pull request #50 from crocs-muni/cpe_matching_notebook
CPE matching
| -rw-r--r-- | .gitignore | 1 | ||||
| -rw-r--r-- | README.md | 8 | ||||
| -rw-r--r-- | examples/cc_cpe_labeling.py | 32 | ||||
| -rw-r--r-- | examples/cc_oop_demo.py (renamed from cc_oop_demo.py) | 10 | ||||
| -rw-r--r-- | examples/fips_oop_demo.py (renamed from fips_oop_demo.py) | 0 | ||||
| -rw-r--r-- | examples/readme.md | 46 | ||||
| -rw-r--r-- | sec_certs/certificate.py | 112 | ||||
| -rw-r--r-- | sec_certs/constants.py | 3 | ||||
| -rw-r--r-- | sec_certs/cpe.py | 237 | ||||
| -rw-r--r-- | sec_certs/cve.py | 185 | ||||
| -rw-r--r-- | sec_certs/dataset.py | 161 | ||||
| -rw-r--r-- | sec_certs/helpers.py | 6 | ||||
| -rw-r--r-- | test/data/test_cc_oop/cc_products_active.csv | 6 | ||||
| -rw-r--r-- | test/data/test_cc_oop/fictional_cert.json | 8 | ||||
| -rw-r--r-- | test/data/test_cc_oop/toy_dataset.json | 32 | ||||
| -rw-r--r-- | test/data/test_cpe_cve/auxillary_datasets/cpe_dataset.json | 6 | ||||
| -rw-r--r-- | test/data/test_cpe_cve/auxillary_datasets/cve_dataset.json | 33 | ||||
| -rw-r--r-- | test/data/test_cpe_cve/vulnerable_dataset.json | 68 | ||||
| -rw-r--r-- | test/test_cc_oop.py | 35 | ||||
| -rw-r--r-- | test/test_cve_cpe_matching.py | 62 |
20 files changed, 971 insertions, 80 deletions
@@ -1,5 +1,6 @@ # idea files .idea +.vscode # results folder results*/ @@ -4,7 +4,7 @@ Tool for analysis of security certificates and their security targets (Common Cr This project is developed by the [Centre for Research On Cryptography and Security](https://crocs.fi.muni.cz) at Faculty of Informatics, Masaryk University. -## Usage (CC) +## Installation (CC) The tool requires several Python packages as well as the `pdftotext` binary somewhere on the `PATH`. The easiest way to setup the tool is to install it in a virtual environment, e.g.: @@ -20,6 +20,12 @@ python3 -m venv virt pip install -e . ``` +## Examples + +Some examples are documented in [examples](https://github.com/crocs-muni/sec-certs/blob/master/examples/) + +## Old API + The following steps will do a full extraction and analysis of CC certificates: 1. Make a directory in which the certificates will be downloaded and processing will take place. diff --git a/examples/cc_cpe_labeling.py b/examples/cc_cpe_labeling.py new file mode 100644 index 00000000..fa96a575 --- /dev/null +++ b/examples/cc_cpe_labeling.py @@ -0,0 +1,32 @@ +from datetime import datetime +import logging +from pathlib import Path + +from sec_certs.dataset import CCDataset +import sec_certs.constants as constants + +logger = logging.getLogger(__name__) + + +def main(): + file_handler = logging.FileHandler(constants.LOGS_FILENAME) + stream_handler = logging.StreamHandler() + formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') + file_handler.setFormatter(formatter) + stream_handler.setFormatter(formatter) + logging.basicConfig(level=logging.INFO, handlers=[file_handler, stream_handler]) + start = datetime.now() + + dset = CCDataset({}, Path('./my_debug_datset'), 'cc_full_dataset', 'Full CC dataset') + dset.get_certs_from_web(to_download=True) + dset.compute_heuristics() + dset.manually_verify_cpe_matches() + + logger.info(f'{dset.json_path} should now contain fully labeled dataset.') + + end = datetime.now() + logger.info(f'The computation took {(end-start)} seconds.') + + +if __name__ == '__main__': + main() diff --git a/cc_oop_demo.py b/examples/cc_oop_demo.py index a0fcdde9..a78fe147 100644 --- a/cc_oop_demo.py +++ b/examples/cc_oop_demo.py @@ -5,6 +5,7 @@ from pathlib import Path from datetime import datetime import logging import json +import pandas as pd logger = logging.getLogger(__name__) @@ -43,6 +44,15 @@ def main(): # transform to pandas DataFrame df = dset.to_pandas() + # Compute heuristics on the dataset + dset.compute_heuristics(update_json=True) + + # Manually verify CPE findings and compute related cves + # dset.manually_verify_cpe_matches(update_json=True) + # dset.compute_related_cves() + + + end = datetime.now() logger.info(f'The computation took {(end-start)} seconds.') diff --git a/fips_oop_demo.py b/examples/fips_oop_demo.py index 7015574e..7015574e 100644 --- a/fips_oop_demo.py +++ b/examples/fips_oop_demo.py diff --git a/examples/readme.md b/examples/readme.md new file mode 100644 index 00000000..66f43661 --- /dev/null +++ b/examples/readme.md @@ -0,0 +1,46 @@ +## New CommonCriteria API + +New object oriented API. The old one should not be used, unless you explicitly want that. Demo of the tool's capabilities with CommonCriteria dataset can be found in [cc_oop_demo.py](https://github.com/crocs-muni/sec-certs/blob/master/examples/cc_oop_demo.py). Also, comments are provided on separate actions that may serve as a temprorary API documentation :). + +To download and build whole dataset can take up to several hours. + + +## Manual CPE labeling + +The tool contains a fuzzy procedure that attempts to map [CPE names](https://nvd.nist.gov/products/cpe) to CC certificates. Result is a list of potentially promising matchings that should be manually evaluated by an analyst to obtain ground truth labeling. The analyst should run the file [cc_cpe_labeling.py](https://github.com/crocs-muni/sec-certs/blob/master/examples/cc_cpe_labeling.py) + +```python +dset = CCDataset({}, Path('./my_debug_datset'), 'cc_full_dataset', 'Full CC dataset') +dset.get_certs_from_web(to_download=True, update_json=True) +dset.compute_heuristics() +dset.manually_verify_cpe_matches() +``` + +For each of the certificates, the user is then prompted for an expert knowledge, see example below: + +``` +[0/1512] Vendor: NetIQ Corporation, Name: NetIQ Identity Manager 4.7 + - [0]: CPE(uri='cpe:2.3:a:netiq:sentinel:-:*:*:*:*:*:*:*', title='NetIQ Sentinel', version='-', vendor='netiq', item_name='sentinel') + - [1]: CPE(uri='cpe:2.3:a:netiq:sentinel_agent_manager:-:*:*:*:*:*:*:*', title='NetIQ Sentinel Agent Manager', version='-', vendor='netiq', item_name='sentinel agent manager') + - [A]: All are fitting + - [X]: No fitting match +Select fitting CPE matches (split with comma if choosing more): +``` + +Here, one should type `X` (case insensitive) and press enter, since all guesses are false positives. In different case + +``` +[1/1512] Vendor: NetIQ, Incorporated, Name: NetIQ Access Manager 4.5 + - [0]: CPE(uri='cpe:2.3:a:netiq:access_manager:4.5:hotfix1:*:*:*:*:*:*', title='NetIQ Access Manager 4.5 Hotfix 1', version='4.5', vendor='netiq', item_name='access manager') + - [1]: CPE(uri='cpe:2.3:a:netiq:access_manager:4.5:sp1:*:*:*:*:*:*', title='NetIQ Access Manager 4.5 Service Pack 1', version='4.5', vendor='netiq', item_name='access manager') + - [2]: CPE(uri='cpe:2.3:a:netiq:access_manager:4.5:-:*:*:*:*:*:*', title='NetIQ Access Manager 4.5', version='4.5', vendor='netiq', item_name='access manager') + - [A]: All are fitting + - [X]: No fitting match +Select fitting CPE matches (split with comma if choosing more): +``` + +one may answer with `0,1,2` as all CPEs may be releated to the certificate. + +The progress of the expert is periodically saved. Currently, there's no way to gracefully exit the process, just do keyboard interrupt if you want to stop. The json will be updated and next time you will get prompted only for the unlabeled certificates. + +We strongly suggest you try the process with `dset.manually_verify_cpe_matches(update_json=False)` to experiment with correct inputs/outputs. While you will get prompted again if the input is recognized incorrect, the `update_json=False` will not store the results so you can experiment with the tool without loosing your results or creating bad labels. diff --git a/sec_certs/certificate.py b/sec_certs/certificate.py index 7a85facc..b873ac37 100644 --- a/sec_certs/certificate.py +++ b/sec_certs/certificate.py @@ -1,6 +1,6 @@ import re from datetime import datetime, date -from dataclasses import dataclass +from dataclasses import dataclass, field import logging from pathlib import Path import os @@ -8,6 +8,7 @@ import copy import json import requests from dateutil import parser +import itertools from abc import ABC, abstractmethod from bs4 import Tag, BeautifulSoup, NavigableString @@ -22,6 +23,8 @@ from sec_certs.extract_certificates import load_cert_file, normalize_match_strin LINE_SEPARATOR, APPEND_DETAILED_MATCH_MATCHES from sec_certs.cert_rules import fips_rules, fips_common_rules from sec_certs.configuration import config +from sec_certs.cpe import CPE, CPEDataset +from sec_certs.cve import CVE, CVEDataset logger = logging.getLogger(__name__) @@ -56,7 +59,7 @@ class Certificate(ABC): def to_json(self, output_path: Union[Path, str]): with Path(output_path).open('w') as handle: - json.dump(self, handle, indent=4, cls=CustomJSONEncoder) + json.dump(self, handle, indent=4, cls=CustomJSONEncoder, ensure_ascii=False) @classmethod def from_json(cls, input_path: Union[Path, str]): @@ -744,8 +747,8 @@ class FIPSCertificate(Certificate, ComplexSerializableType): class CommonCriteriaCert(Certificate, ComplexSerializableType): - cc_url = 'http://commoncriteriaportal.org' - empty_st_url = 'http://commoncriteriaportal.org/files/epfiles/' + cc_url = 'http://www.commoncriteriaportal.org' + empty_st_url = 'http://www.commoncriteriaportal.org/files/epfiles/' @dataclass(eq=True, frozen=True) class MaintainanceReport(ComplexSerializableType): @@ -868,8 +871,36 @@ class CommonCriteriaCert(Certificate, ComplexSerializableType): def from_dict(cls, dct: Dict[str, bool]): return cls(*tuple(dct.values())) - pandas_serialization_vars = ['dgst', 'name', 'manufacturer', 'scheme', 'security_level', 'not_valid_before', - 'not_valid_after', 'report_link', 'st_link', 'src', 'manufacturer_web'] + @dataclass(init=False) + class Heuristics(ComplexSerializableType): + extracted_versions: List[str] + cpe_candidate_vendors: Optional[List[str]] = field(init=False) + cpe_matches: Optional[List[Tuple[float, CPE]]] + verified_cpe_matches: Optional[List[CPE]] + related_cves: Optional[List[str]] + + def __init__(self, + extracted_versions: Optional[List[str]] = None, + cpe_matches: Optional[List[str]] = None, + verified_cpe_matches: Optional[List[str]] = None, + related_cves: Optional[List[CVE]] = None): + self.extracted_versions = extracted_versions + self.cpe_matches = cpe_matches + self.cpe_candidate_vendors = None + self.verified_cpe_matches = verified_cpe_matches + self.related_cves = related_cves + + def to_dict(self): + return {'extracted_versions': self.extracted_versions, 'cpe_matches': self.cpe_matches, 'verified_cpe_matches': self.verified_cpe_matches, 'related_cves': self.related_cves} + + @classmethod + def from_dict(cls, dct: Dict[str, str]): + return cls(*tuple(dct.values())) + + pandas_columns = ['dgst', 'name', 'status', 'category', 'manufacturer', 'scheme', 'security_level', + 'not_valid_before', 'not_valid_after', 'report_link', 'st_link', + 'manufacturer_web', 'extracted_versions', 'cpe_matches', 'verified_cpe_matches', + 'related_cves'] def __init__(self, status: str, category: str, name: str, manufacturer: str, scheme: str, security_level: Union[str, set], not_valid_before: date, @@ -879,7 +910,7 @@ class CommonCriteriaCert(Certificate, ComplexSerializableType): maintainance_updates: set, state: Optional[InternalState], pdf_data: Optional[PdfData], - cpe_matching: Optional[List[Tuple[str]]]): + heuristics: Optional[Heuristics]): super().__init__() self.status = status @@ -906,9 +937,9 @@ class CommonCriteriaCert(Certificate, ComplexSerializableType): pdf_data = self.PdfData() self.pdf_data = pdf_data - if cpe_matching is None: - cpe_matching = [] - self.cpe_matching = cpe_matching + if heuristics is None: + heuristics = self.Heuristics() + self.heuristics = heuristics @property def dgst(self) -> str: @@ -917,8 +948,15 @@ class CommonCriteriaCert(Certificate, ComplexSerializableType): """ return helpers.get_first_16_bytes_sha256(self.category + self.name + self.report_link) + def __str__(self): + return self.manufacturer + ' ' + self.name + ' dgst: ' + self.dgst + def to_pandas_tuple(self): - return tuple(getattr(self, i) for i in self.pandas_serialization_vars) + return self.dgst, self.name, self.status, self.category, self.manufacturer, self.scheme, self.security_level,\ + self.not_valid_before, self.not_valid_after, self.report_link, self.st_link, self.manufacturer_web, \ + self.heuristics.extracted_versions, self.heuristics.cpe_matches, self.heuristics.verified_cpe_matches, \ + self.heuristics.related_cves + def merge(self, other: 'CommonCriteriaCert'): """ @@ -1083,6 +1121,13 @@ class CommonCriteriaCert(Certificate, ComplexSerializableType): if st_txt_dir is not None: self.state.st_txt_path = Path(st_txt_dir) / (self.dgst + '.txt') + @property + def best_cpe_match(self): + clean = [x for x in self.cpe_matching if len(x[0]) > 5] + cpe_match_ranking = [x[1] for x in clean] + argmax = cpe_match_ranking.index(max(cpe_match_ranking)) + return clean[argmax] + @staticmethod def download_pdf_report(cert: 'CommonCriteriaCert') -> 'CommonCriteriaCert': exit_code = helpers.download_file(cert.report_link, cert.state.report_pdf_path) @@ -1189,3 +1234,48 @@ class CommonCriteriaCert(Certificate, ComplexSerializableType): cert.state.st_extract_ok = False cert.state.errors.append(response) return cert + + def compute_heuristics_version(self): + """ + Will extract possible versions from the name + """ + at_least_something = r'(\b(\d)+\b)' + just_numbers = r'(\d{1,5})(\.\d{1,5})' + + without_version = r'(' + just_numbers + r'+)' + long_version = r'(' + r'(\bversion)\s*' + just_numbers + r'+)' + short_version = r'(' + r'\bv\s*' + just_numbers + r'+)' + full_regex_string = r'|'.join([without_version, short_version, long_version]) + normalizer = r'(\d+\.*)+' + + matched_strings = set([max(x, key=len) for x in re.findall(full_regex_string, self.name, re.IGNORECASE)]) + if not matched_strings: + matched_strings = set([max(x, key=len) for x in re.findall(at_least_something, self.name, re.IGNORECASE)]) + + if matched_strings: + self.heuristics.extracted_versions = [re.search(normalizer, x).group() for x in matched_strings] + else: + self.heuristics.extracted_versions = ['-'] + + def compute_heuristics_cpe_vendors(self, cpe_dataset: CPEDataset): + """ + With the help of the CPE dataset, will find CPE vendors that could match the given certificate vendor + """ + self.heuristics.cpe_candidate_vendors = cpe_dataset.get_candidate_list_of_vendors(self.manufacturer) + + def compute_heuristics_cpe_match(self, cpe_dataset: CPEDataset): + self.compute_heuristics_cpe_vendors(cpe_dataset) + self.heuristics.cpe_matches = cpe_dataset.get_cpe_matches(self.name, + self.heuristics.cpe_candidate_vendors, + self.heuristics.extracted_versions, + n_max_matches=constants.CPE_MAX_MATCHES, + threshold=constants.CPE_MATCHING_THRESHOLD) + + def compute_heuristics_related_cves(self, cve_dataset: CVEDataset): + if self.heuristics.verified_cpe_matches: + related_cves = [cve_dataset.get_cves_for_cpe(x.uri) for x in self.heuristics.verified_cpe_matches] + related_cves = list(filter(lambda x: x is not None, related_cves)) + if related_cves: + self.heuristics.related_cves = list(itertools.chain.from_iterable(related_cves)) + else: + self.heuristics.related_cves = None
\ No newline at end of file diff --git a/sec_certs/constants.py b/sec_certs/constants.py index 41a4f3b7..610e4cc9 100644 --- a/sec_certs/constants.py +++ b/sec_certs/constants.py @@ -6,6 +6,9 @@ RETURNCODE_OK = 'ok' RETURNCODE_NOK = 'nok' REQUEST_TIMEOUT = 10 +CPE_MATCHING_THRESHOLD = 70 +CPE_MAX_MATCHES = 10 + MIN_CORRECT_CERT_SIZE = 5000 LOGS_FILENAME = './cert_processing_log.txt' diff --git a/sec_certs/cpe.py b/sec_certs/cpe.py new file mode 100644 index 00000000..e3a456f5 --- /dev/null +++ b/sec_certs/cpe.py @@ -0,0 +1,237 @@ +from dataclasses import dataclass, field +import logging +import json +from typing import Optional, List, Dict, Tuple, Set, Union, ClassVar +import itertools +import re +from rapidfuzz import process, fuzz +import tempfile +from pathlib import Path +import zipfile +import operator + +import sec_certs.helpers as helpers +from sec_certs.serialization import ComplexSerializableType + +import pandas as pd +import xml.etree.ElementTree as ET + +logger = logging.getLogger(__name__) + + +@dataclass(init=False) +class CPE(ComplexSerializableType): + uri: str + title: str + version: str + vendor: str + item_name: str + + def __init__(self, uri: Optional[str] = None, title: Optional[str] = None): + self.uri = uri + self.title = title + + if self.uri: + self.vendor = ' '.join(self.uri.split(':')[3].split('_')) + self.item_name = ' '.join(self.uri.split(':')[4].split('_')) + self.version = self.uri.split(':')[5] + + def to_dict(self): + return {'uri': self.uri, 'title': self.title} + + @classmethod + def from_dict(cls, dct: Dict[str, str]): + return cls(*tuple(dct.values())) + + def __hash__(self): + return hash(self.uri) + + +def build_cpe_uri_to_title_dict(input_xml_filepath: str, output_filepath: str): + """ + Will parse CPE XML file into dictionary cpe_uri: cpe_title and dump the dict into json + """ + logger.info(f'Extracting dictionary cpe_uri:cpe_title from {input_xml_filepath} to {output_filepath}') + root = ET.parse(input_xml_filepath).getroot() + dct = {} + for cpe_item in root.findall('{http://cpe.mitre.org/dictionary/2.0}cpe-item'): + title = cpe_item.find('{http://cpe.mitre.org/dictionary/2.0}title').text + cpe_uri = cpe_item.find('{http://scap.nist.gov/schema/cpe-extension/2.3}cpe23-item').attrib['name'] + dct[cpe_uri] = title + with open(output_filepath, 'w') as handle: + json.dump(dct, handle, indent=4) + + +# TODO: Make this ComplexSerializableType +@dataclass +class CPEDataset: + cpes: Dict[str, CPE] + vendor_to_versions: Dict[str, Set[str]] = field(init=False) # Look-up dict cpe_vendor: list of viable versions + vendor_version_to_cpe: Dict[Tuple[str, str], Set[CPE]] = field(init=False) # Look-up dict (cpe_vendor, cpe_version): List of viable cpe items + vendors: Set[str] = field(init=False) + + 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' + + def __iter__(self): + yield from self.cpes.values() + + def __getitem__(self, item: str) -> CPE: + return self.cpes.__getitem__(item.lower()) + + def __setitem__(self, key: str, value: CPE): + self.cpes.__setitem__(key.lower(), value) + + def __len__(self) -> int: + return len(self.cpes) + + def __post_init__(self): + """ + Will build look-up dictionaries that are used for fast matching + """ + logging.info('CPE dataset: building lookup dictionaries.') + self.vendor_to_versions = {x.vendor: set() for x in self} + self.vendor_version_to_cpe = dict() + self.vendors = set(self.vendor_to_versions.keys()) + for cpe in self: + self.vendor_to_versions[cpe.vendor].add(cpe.version) + if (cpe.vendor, cpe.version) not in self.vendor_version_to_cpe: + self.vendor_version_to_cpe[(cpe.vendor, cpe.version)] = {cpe} + else: + self.vendor_version_to_cpe[(cpe.vendor, cpe.version)].add(cpe) + + @classmethod + def from_json(cls, json_path: Union[str, Path]): + with Path(json_path).open('r') as handle: + data = json.load(handle) + return cls({x: CPE(x, y) for x, y in data.items()}) + + def to_json(self, json_path: str): + with open(json_path, 'w') as handle: + json.dump({x: y.title for x, y in self.cpes.items()}, handle, indent=4) + + @classmethod + def from_web(cls): + 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') + helpers.download_file(cls.cpe_url, zip_path) + + with zipfile.ZipFile(zip_path, 'r') as zip_ref: + zip_ref.extractall(tmp_dir) + + return cls.from_xml(xml_path) + + @classmethod + def from_xml(cls, xml_path: Union[str, Path]): + logger.info('Loading CPE dataset from XML.') + root = ET.parse(xml_path).getroot() + dct = {} + for cpe_item in root.findall('{http://cpe.mitre.org/dictionary/2.0}cpe-item'): + title = cpe_item.find('{http://cpe.mitre.org/dictionary/2.0}title').text + cpe_uri = cpe_item.find('{http://scap.nist.gov/schema/cpe-extension/2.3}cpe23-item').attrib['name'] + dct[cpe_uri] = CPE(cpe_uri, title) + return cls(dct) + + def to_pandas(self): + if not self.cpes: + return None + else: + columns = list(CPE.__annotations__.keys()) + data = [list(x.__dict__.values()) for x in self] + df = pd.DataFrame(data, columns=columns) + df = df.set_index('uri') + + return df + + def get_candidate_list_of_vendors(self, cert_vendor: str) -> Optional[List[str]]: + """ + Will return List of CPE vendors that could match the cert_vendor. + """ + result = set() + if not isinstance(cert_vendor, str): + return None + lower = cert_vendor.lower() + if ' / ' in cert_vendor: + chain = [self.get_candidate_list_of_vendors(x) for x in cert_vendor.split(' / ')] + chain = [x for x in chain if x] + return list(set(itertools.chain(*chain))) + if lower in self.vendors: + result.add(lower) + if ' ' in lower and (y := lower.split(' ')[0]) in self.vendors: + result.add(y) + if ',' in lower and (y := lower.split(',')[0]) in self.vendors: + result.add(y) + if not result: + return None + return list(result) + + def get_candidate_vendor_version_pairs(self, cert_candidate_cpe_vendors: List[str], cert_candidate_versions: List[str]) -> Optional[List[Tuple[str, str]]]: + """ + Given parameters, will return Pairs (cpe_vendor, cpe_version) that should are relevant to a given certificate + Parameters + :param cert_candidate_cpe_vendors: list of CPE vendors relevant to a certificate + :param cert_candidate_versions: List of versions heuristically extracted from the certificate name + :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: str, cert_versions: List[str]) -> bool: + just_numbers = r'(\d{1,5})(\.\d{1,5})' # TODO: The use of this should be double-checked + for v in cert_versions: + if (v.startswith(cpe_version) and re.search(just_numbers, cpe_version)) or cpe_version.startswith(v): + return True + return False + + if not cert_candidate_cpe_vendors: + return None + + candidate_vendor_version_pairs: List[Tuple[str, str]] = [] + for vendor in cert_candidate_cpe_vendors: + viable_cpe_versions = self.vendor_to_versions[vendor] + matched_cpe_versions = [x for x in viable_cpe_versions if is_cpe_version_among_cert_versions(x, cert_candidate_versions)] + candidate_vendor_version_pairs.extend([(vendor, x) for x in matched_cpe_versions]) + return candidate_vendor_version_pairs + + def get_candidate_cpe_items(self, cert_candidate_cpe_vendors: List[str], cert_candidate_versions: List[str]) -> Optional[List[CPE]]: + candidate_vendor_version_pairs = self.get_candidate_vendor_version_pairs(cert_candidate_cpe_vendors, cert_candidate_versions) + + if not candidate_vendor_version_pairs: + return [] + + return list(itertools.chain.from_iterable([self.vendor_version_to_cpe[x] for x in candidate_vendor_version_pairs])) + + def get_cpe_matches(self, cert_name: str, cert_candidate_cpe_vendors: List[str], cert_candidate_versions: List[str], relax_version: bool = False, n_max_matches=10, threshold: int = 60) -> Optional[List[Tuple[float, CPE]]]: + replace_non_letter_non_numbers_with_space = re.compile(r"(?ui)\W") + + def sanitize_matched_string(string: str): + string = string.replace('®', '').replace('™', '').lower() + return replace_non_letter_non_numbers_with_space.sub(' ', string) + candidates = self.get_candidate_cpe_items(cert_candidate_cpe_vendors, cert_candidate_versions) + + sanitized_cert_name = sanitize_matched_string(cert_name) + reasonable_matches = [] + for c in candidates: + sanitized_title = sanitize_matched_string(c.title) + sanitized_item_name = sanitize_matched_string(c.item_name) + set_match_title = fuzz.token_set_ratio(sanitized_cert_name, sanitized_title) + partial_match_title = fuzz.partial_ratio(sanitized_cert_name, sanitized_title) + set_match_item = fuzz.token_set_ratio(sanitized_cert_name, sanitized_item_name) + partial_match_item = fuzz.partial_ratio(sanitized_cert_name, sanitized_item_name) + + potential = max([set_match_title, partial_match_title, set_match_item, partial_match_item]) + + if potential > threshold: + reasonable_matches.append((potential, c)) + + if reasonable_matches: + reasonable_matches = sorted(reasonable_matches, key=operator.itemgetter(0), reverse=True) + + # possibly filter short titles to avoid false positives + # reasonable_matches = list(filter(lambda x: len(x[1].item_name) > 4, reasonable_matches)) + + return reasonable_matches[:n_max_matches] + + if not reasonable_matches and not relax_version: + return self.get_cpe_matches(cert_name, cert_candidate_cpe_vendors, ['-'], relax_version=True, n_max_matches=n_max_matches, threshold=threshold) + + return None
\ No newline at end of file diff --git a/sec_certs/cve.py b/sec_certs/cve.py new file mode 100644 index 00000000..a8cc9742 --- /dev/null +++ b/sec_certs/cve.py @@ -0,0 +1,185 @@ +from dataclasses import dataclass, field +from typing import Dict, List, Optional, Union, ClassVar +import copy +import datetime +from pathlib import Path +import tempfile +import zipfile +import logging +import glob +import tqdm +import json + +from sec_certs.cert_processing import process_parallel +import sec_certs.constants as constants +import sec_certs.helpers as helpers +from sec_certs.serialization import ComplexSerializableType, CustomJSONDecoder, CustomJSONEncoder + +logger = logging.getLogger(__name__) + + +@dataclass(eq=True, frozen=True) +class CVE(ComplexSerializableType): + @dataclass(eq=True) + class Impact(ComplexSerializableType): + base_score: float + severity: str + explotability_score: float + impact_score: float + + def to_dict(self): + return copy.deepcopy(self.__dict__) + + @classmethod + def from_dict(cls, dct: Dict): + return cls(*tuple(dct.values())) + + @classmethod + def from_nist_dict(cls, dct: Dict): + """ + Will load Impact from dictionary defined at https://nvd.nist.gov/feeds/json/cve/1.1 + """ + if not dct['impact']: + return cls(0, '', 0, 0) + elif 'baseMetricV3' in dct['impact']: + return cls(dct['impact']['baseMetricV3']['cvssV3']['baseScore'], + dct['impact']['baseMetricV3']['cvssV3']['baseSeverity'], + dct['impact']['baseMetricV3']['exploitabilityScore'], + dct['impact']['baseMetricV3']['impactScore']) + elif 'baseMetricV2' in dct['impact']: + return cls(dct['impact']['baseMetricV2']['cvssV2']['baseScore'], + dct['impact']['baseMetricV2']['severity'], + dct['impact']['baseMetricV2']['exploitabilityScore'], + dct['impact']['baseMetricV2']['impactScore']) + + cve_id: str + vulnerable_cpes: List[str] + impact: Impact + + def to_dict(self): + return copy.deepcopy(self.__dict__) + + @classmethod + def from_dict(cls, dct: Dict): + return cls(*tuple(dct.values())) + + @classmethod + def from_nist_dict(cls, dct: Dict) -> 'CVE': + """ + Will load CVE from dictionary defined at https://nvd.nist.gov/feeds/json/cve/1.1 + """ + def get_vulnerable_cpes_from_nist_dict(dct: Dict) -> List[str]: + def get_vulnerable_cpes_from_node(node: Dict) -> List[str]: + cpe_uris = [] + if 'children' in node: + for child in node['children']: + cpe_uris += get_vulnerable_cpes_from_node(child) + if 'cpe_match' in node: + lst = node['cpe_match'] + for x in lst: + if x['vulnerable']: + cpe_uris.append(x['cpe23Uri']) + return cpe_uris + + vulnerable_cpes = [] + for node in dct['configurations']['nodes']: + vulnerable_cpes.extend(get_vulnerable_cpes_from_node(node)) + + return vulnerable_cpes + + cve_id = dct['cve']['CVE_data_meta']['ID'] + impact = cls.Impact.from_nist_dict(dct) + vulnerable_cpes = get_vulnerable_cpes_from_nist_dict(dct) + + return CVE(cve_id, vulnerable_cpes, impact) + + +@dataclass(eq=True) +class CVEDataset(ComplexSerializableType): + cves: Dict[str, CVE] + cpes_to_cve_lookup: Dict[str, List[str]] = field(init=False) + + cve_url: ClassVar[str] = 'https://nvd.nist.gov/feeds/json/cve/1.1/nvdcve-1.1-' + + def to_dict(self): + return copy.deepcopy({'cves': self.cves}) + + @classmethod + def from_dict(cls, dct: Dict): + return cls(*tuple(dct.values())) + + def __post_init__(self): + self.cpes_to_cve_lookup = dict() + for cve in self: + for cpe in cve.vulnerable_cpes: + if not cpe in self.cpes_to_cve_lookup: + self.cpes_to_cve_lookup[cpe] = [cve.cve_id] + else: + self.cpes_to_cve_lookup[cpe].append(cve.cve_id) + + def __iter__(self): + yield from self.cves.values() + + def __getitem__(self, item: str) -> CVE: + return self.cves.__getitem__(item.lower()) + + def __setitem__(self, key: str, value: CVE): + self.cves.__setitem__(key.lower(), value) + + def __len__(self) -> int: + return len(self.cves) + + def download_cves(self, output_path: str, start_year: int, end_year: int): + output_path = Path(output_path) + if not output_path.exists: + output_path.mkdir() + + urls = [self.cve_url + str(x) + '.json.zip' for x in range(start_year, end_year + 1)] + + logger.info(f'Identified {len(urls)} CVE files to fetch from nist.gov. Downloading them into {output_path}') + with tempfile.TemporaryDirectory() as tmp_dir: + outpaths = [Path(tmp_dir) / Path(x).name.rstrip('.zip') for x in urls] + responses = list(zip(*helpers.download_parallel(list(zip(urls, outpaths)), num_threads=constants.N_THREADS)))[1] + + for o, u, r in zip(outpaths, urls, responses): + if r == constants.RESPONSE_OK: + with zipfile.ZipFile(o, 'r') as zip_handle: + zip_handle.extractall(output_path) + else: + logger.info(f'Failed to download from {u}, got status code {r}') + + @classmethod + def from_nist_json(cls, input_path: str) -> 'CVEDataset': + with Path(input_path).open('r') as handle: + data = json.load(handle) + cves = [CVE.from_nist_dict(x) for x in data['CVE_Items']] + return cls({x.cve_id: x for x in cves}) + + @classmethod + def from_web(cls, start_year: int = 2002, end_year: int = datetime.datetime.now().year): + logger.info(f'Building CVE dataset from nist.gov website.') + with tempfile.TemporaryDirectory() as tmp_dir: + cls.download_cves(tmp_dir, start_year, end_year) + json_files = glob.glob(tmp_dir + '/*.json') + + all_cves = dict() + logger.info(f'Downloaded required resources. Building CVEDataset from jsons.') + results = process_parallel(cls.from_nist_json, json_files, constants.N_THREADS, use_threading=False) + for r in results: + all_cves.update(r.cves) + return cls(all_cves) + + def to_json(self, output_path: str): + with Path(output_path).open('w') as handle: + json.dump(self, handle, indent=4, cls=CustomJSONEncoder, ensure_ascii=False) + + @classmethod + def from_json(cls, input_path: Union[str, Path]): + with Path(input_path).open('r') as handle: + dset = json.load(handle, cls=CustomJSONDecoder) + return dset + + def get_cves_for_cpe(self, cpe_uri: str) -> Optional[List[str]]: + if not isinstance(cpe_uri, str): + return None + return self.cpes_to_cve_lookup.get(cpe_uri, None) diff --git a/sec_certs/dataset.py b/sec_certs/dataset.py index a9854edd..5845e4d3 100644 --- a/sec_certs/dataset.py +++ b/sec_certs/dataset.py @@ -6,6 +6,7 @@ from typing import Dict, List, ClassVar, Collection, Union, Set, Tuple, Optional from itertools import groupby from dataclasses import dataclass import copy +import time import json from abc import ABC, abstractmethod @@ -27,6 +28,8 @@ import sec_certs.files as files from sec_certs.certificate import CommonCriteriaCert, Certificate, FIPSCertificate from sec_certs.serialization import ComplexSerializableType, CustomJSONDecoder, CustomJSONEncoder from sec_certs.configuration import config +from sec_certs.cpe import CPEDataset +from sec_certs.cve import CVEDataset logger = logging.getLogger(__name__) @@ -51,10 +54,14 @@ class Dataset(ABC): raise FileNotFoundError('Root directory for Dataset does not exist') self._root_dir = new_path + @property + def json_path(self) -> Path: + return self.root_dir / (self.name + '.json') + def __iter__(self): yield from self.certs.values() - def __getitem__(self, item: str) -> 'Certificate': + def __getitem__(self, item: str): return self.certs.__getitem__(item.lower()) def __setitem__(self, key: str, value: 'Certificate'): @@ -83,9 +90,12 @@ class Dataset(ABC): f'The actual number of certs in dataset ({len(dset)}) does not match the claimed number ({claimed}).') return dset - def to_json(self, output_path: Union[str, Path]): + def to_json(self, output_path: Union[str, Path] = None): + if not output_path: + output_path = self.json_path + with Path(output_path).open('w') as handle: - json.dump(self, handle, indent=4, cls=CustomJSONEncoder) + json.dump(self, handle, indent=4, cls=CustomJSONEncoder, ensure_ascii=False) @classmethod def from_json(cls, input_path: Union[str, Path]): @@ -144,20 +154,34 @@ class CCDataset(Dataset, ComplexSerializableType): def from_dict(cls, dct: Dict[str, bool]): return cls(*tuple(dct.values())) - def __init__(self, certs: Dict[str, 'Certificate'], root_dir: Path, name: str = 'dataset name', + certs: Dict[str, 'CommonCriteriaCert'] + + def __init__(self, certs: Dict[str, 'CommonCriteriaCert'], root_dir: Path, name: str = 'dataset name', description: str = 'dataset_description', state: Optional[DatasetInternalState] = None): super().__init__(certs, root_dir, name, description) + if state is None: state = self.DatasetInternalState() self.state = state + def __iter__(self) -> CommonCriteriaCert: + yield from self.certs.values() + def to_dict(self): return {**{'state': self.state}, **super().to_dict()} def to_pandas(self): tuples = [x.to_pandas_tuple() for x in self.certs.values()] - cols = CommonCriteriaCert.pandas_serialization_vars - return pd.DataFrame(tuples, columns=cols).set_index('dgst') + cols = CommonCriteriaCert.pandas_columns + + df = pd.DataFrame(tuples, columns=cols) + df = df.set_index('dgst') + + df.not_valid_before = pd.to_datetime(df.not_valid_before, infer_datetime_format=True) + df.not_valid_after = pd.to_datetime(df.not_valid_after, infer_datetime_format=True) + df = df.astype({'category': 'category', 'status': 'category', 'scheme': 'category'}) + + return df @classmethod def from_dict(cls, dct: Dict): @@ -171,10 +195,6 @@ class CCDataset(Dataset, ComplexSerializableType): self.set_local_paths() @property - def json_path(self) -> Path: - return self.root_dir / (self.name + '.json') - - @property def web_dir(self) -> Path: return self.root_dir / 'web' @@ -206,6 +226,18 @@ class CCDataset(Dataset, ComplexSerializableType): def targets_txt_dir(self) -> Path: return self.targets_dir / 'txt' + @property + def auxillary_datasets_path(self) -> Path: + return self.root_dir / 'auxillary_datasets' + + @property + def cve_dataset_path(self) -> Path: + return self.auxillary_datasets_path / 'cve_dataset.json' + + @property + def cpe_dataset_path(self) -> Path: + return self.auxillary_datasets_path / 'cpe_dataset.json' + html_products = { 'cc_products_active.html': 'https://www.commoncriteriaportal.org/products/', 'cc_products_archived.html': 'https://www.commoncriteriaportal.org/products/index.cfm?archived=1', @@ -253,7 +285,7 @@ class CCDataset(Dataset, ComplexSerializableType): f'Added {len(will_be_added)} new and merged further {n_merged} certificates to the dataset.') def get_certs_from_web(self, to_download: bool = True, keep_metadata: bool = True, get_active: bool = True, - get_archived: bool = True, update_json: bool = False): + get_archived: bool = True, update_json: bool = True): """ Downloads all metadata about certificates from CSV and HTML sources """ @@ -336,7 +368,7 @@ class CCDataset(Dataset, ComplexSerializableType): 'maintainance_title', 'maintainance_report_link', 'maintainance_st_link'] # TODO: Now skipping bad lines, smarter heuristics to be built for dumb files - df = pd.read_csv(file, engine='python', encoding='windows-1250', error_bad_lines=False) + df = pd.read_csv(file, engine='python', encoding='windows-1252', error_bad_lines=False) df = df.rename(columns={x: y for (x, y) in zip(list(df.columns), csv_header)}) df['is_maintainance'] = ~df.maintainance_title.isnull() @@ -650,34 +682,100 @@ class CCDataset(Dataset, ComplexSerializableType): if update_json is True: self.to_json(self.json_path) - # TODO: Probably breaks a logic of ceritifcate managing itself. Needs design refactoring - def fuzzy_match_cpe(self, cpe_path: Path, update_json: bool = False): - def get_cpe_titles(cpe_path: Path): - root = ET.parse(str(cpe_path)).getroot() - return [child.text for child in root.findall( - '{http://cpe.mitre.org/dictionary/2.0}cpe-item/{http://cpe.mitre.org/dictionary/2.0}title')] + def prepare_cpe_dataset(self, download_fresh_cpes: bool = False) -> CPEDataset: + logger.info('Preparing CPE dataset.') + if not self.auxillary_datasets_path.exists(): + self.auxillary_datasets_path.mkdir(parents=True) - digests = [x for x in self.certs.keys()] - cpe_titles = get_cpe_titles(cpe_path) + if not self.cpe_dataset_path.exists() or download_fresh_cpes is True: + cpe_dataset = CPEDataset.from_web() + cpe_dataset.to_json(str(self.cpe_dataset_path)) + else: + cpe_dataset = CPEDataset.from_json(str(self.cpe_dataset_path)) + + return cpe_dataset + + def prepare_cve_dataset(self, download_fresh_cves: bool = False) -> CVEDataset: + logger.info('Preparing CVE dataset.') + if not self.auxillary_datasets_path.exists(): + self.auxillary_datasets_path.mkdir(parents=True) + + if not self.cve_dataset_path.exists() or download_fresh_cves is True: + cve_dataset = CVEDataset.from_web() + cve_dataset.to_json(str(self.cve_dataset_path)) + else: + cve_dataset = CVEDataset.from_json(str(self.cve_dataset_path)) - def chunk_list(a: List, n: int): - k, m = divmod(len(a), n) - return (a[i * k + min(i, m):(i + 1) * k + min(i + 1, m)] for i in range(n)) + return cve_dataset - chunks = chunk_list(digests, constants.N_THREADS) - chunks_dicts = [{x: self[x].name for x in y} for y in chunks] + def compute_heuristics(self, update_json=True, download_fresh_cpes: bool = False): + def compute_candidate_versions(): + logger.info('Computing heuristics: possible product versions in certificate name') + for cert in self: + cert.compute_heuristics_version() - results = cert_processing.process_parallel(helpers.match_certs, list( - zip(chunks_dicts, [cpe_titles for _ in range(constants.N_THREADS)])), constants.N_THREADS, - use_threading=False, unpack=True) + def compute_cpe_matches(cpe_dataset: CPEDataset): + logger.info('Computing heuristics: Finding CPE matches for certificates') + for cert in self: + cert.compute_heuristics_cpe_match(cpe_dataset) - for chunk in results: - for digest, matches in chunk.items(): - self[digest].cpe_matching = matches + compute_candidate_versions() + cpe_dset = self.prepare_cpe_dataset(download_fresh_cpes) + compute_cpe_matches(cpe_dset) if update_json is True: self.to_json(self.json_path) + def manually_verify_cpe_matches(self, update_json=True): + def verify_certs(certificates_to_verify: List[CommonCriteriaCert]): + n_certs_to_verify = len(certificates_to_verify) + for i, x in enumerate(certificates_to_verify): + print(f'\n[{i}/{n_certs_to_verify}] Vendor: {x.manufacturer}, Name: {x.name}') + for index, c in enumerate(x.heuristics.cpe_matches): + print(f'\t- {[index]}: {c[1]}') + print(f'\t- [A]: All are fitting') + print(f'\t- [X]: No fitting match') + inpts = input('Select fitting CPE matches (split with comma if choosing more):').strip().split(',') + + if 'X' not in inpts and 'x' not in inpts: + if 'A' in inpts or 'a' in inpts: + inpts = [x for x in range(0, len(x.heuristics.cpe_matches))] + try: + inpts = [int(x) for x in inpts] + if min(inpts) < 0 or max(inpts) > len(x.heuristics.cpe_matches) - 1: + raise ValueError(f'Incorrect number chosen, choose in range 0-{len(x.heuristics.cpe_matches) - 1}') + except ValueError as e: + logger.error(f'Bad input from user, repeating instance: {e}') + print(f'Bad input from user, repeating instance: {e}') + time.sleep(0.05) + verify_certs([x]) + else: + matches = [x.heuristics.cpe_matches[y][1] for y in inpts] + self[x.dgst].heuristics.verified_cpe_matches = matches + + if i != 0 and not i % 10 and update_json: + print(f'Saving progress.') + self.to_json() + + certs_to_verify: List[CommonCriteriaCert] = [x for x in self if (x.heuristics.cpe_matches and not x.heuristics.verified_cpe_matches)] + logger.info('Manually verifying CPE matches') + time.sleep(0.05) # easier than flushing the logger + verify_certs(certs_to_verify) + + if update_json is True: + self.to_json() + + def compute_related_cves(self, download_fresh_cves: bool = False): + logger.info('Retrieving related CVEs to verified CPE matches') + cve_dset = self.prepare_cve_dataset(download_fresh_cves) + + verified_cpe_rich_certs = [x for x in self if x.heuristics.verified_cpe_matches] + if not verified_cpe_rich_certs: + logger.error('No certificates with verified CPE match detected. You must run dset.manually_verify_cpe_matches() first. Returning.') + return + for cert in verified_cpe_rich_certs: + cert.compute_heuristics_related_cves(cve_dset) + class FIPSDataset(Dataset, ComplexSerializableType): FIPS_BASE_URL: ClassVar[str] = 'https://csrc.nist.gov' @@ -740,7 +838,6 @@ class FIPSDataset(Dataset, ComplexSerializableType): return output - def download_all_pdfs(self): sp_paths, sp_urls = [], [] self.policies_dir.mkdir(exist_ok=True) diff --git a/sec_certs/helpers.py b/sec_certs/helpers.py index a8711f73..87f6e25e 100644 --- a/sec_certs/helpers.py +++ b/sec_certs/helpers.py @@ -491,12 +491,6 @@ def extract_keywords(filepath: Path) -> Tuple[int, Optional[Dict[str, str]]]: return constants.RETURNCODE_OK, result -def match_certs(certs: Dict[str, str], cpes: List[str]): - results = {} - for dgst, cert_name in certs.items(): - results[dgst] = process.extract(cert_name, cpes, scorer=fuzz.token_set_ratio, limit=10) - return results - def analyze_matched_algs(data: Dict): pd_data = pd.Series(data) pd_data.hist(bins=50) diff --git a/test/data/test_cc_oop/cc_products_active.csv b/test/data/test_cc_oop/cc_products_active.csv index 5b41d581..a361ba35 100644 --- a/test/data/test_cc_oop/cc_products_active.csv +++ b/test/data/test_cc_oop/cc_products_active.csv @@ -1,3 +1,3 @@ -Category,Name,Manufacturer,Scheme,Assurance Level,Protection Profile(s),Certification Date,Archived Date,Certification Report URL,Security Target URL,Maintenance Date,Maintenance Title,Maintenance Report,Maintenance ST -Access Control Devices and Systems,NetIQ Identity Manager 4.7,NetIQ Corporation,SE,"EAL3+,ALC_FLR.2",,06/15/2020,06/15/2025,http://commoncriteriaportal.org:443/files/epfiles/Certification Report - NetIQ® Identity Manager 4.7.pdf,http://commoncriteriaportal.org:443/files/epfiles/ST - NetIQ Identity Manager 4.7.pdf,,,, -Access Control Devices and Systems,Magic SSO V4.0,"Dreamsecurity Co., Ltd.",KR,None,KECS-PP-0822-2017 SSO V1.0,11/15/2019,11/15/2024,http://commoncriteriaportal.org:443/files/epfiles/KECS-CR-19-70 Magic SSO V4.0(eng) V1.0.pdf,http://commoncriteriaportal.org:443/files/epfiles/Magic_SSO_V4.0-ST-v1.4_EN.pdf,,,, +"Category","Name","Manufacturer","Scheme","Assurance Level","Protection Profile(s)","Certification Date","Archived Date","Certification Report URL","Security Target URL","Maintenance Date","Maintenance Title","Maintenance Report","Maintenance ST"
+"Access Control Devices and Systems","NetIQ Identity Manager 4.7","NetIQ Corporation","SE","EAL3+,ALC_FLR.2","","06/15/2020","06/15/2025","http://www.commoncriteriaportal.org:443/files/epfiles/Certification Report - NetIQ® Identity Manager 4.7.pdf","http://www.commoncriteriaportal.org:443/files/epfiles/ST - NetIQ Identity Manager 4.7.pdf","","","",""
+"Access Control Devices and Systems","Magic SSO V4.0","Dreamsecurity Co., Ltd.","KR","None","KECS-PP-0822-2017 SSO V1.0","11/15/2019","11/15/2024","http://www.commoncriteriaportal.org:443/files/epfiles/KECS-CR-19-70 Magic SSO V4.0(eng) V1.0.pdf","http://www.commoncriteriaportal.org:443/files/epfiles/Magic_SSO_V4.0-ST-v1.4_EN.pdf","","","",""
\ No newline at end of file diff --git a/test/data/test_cc_oop/fictional_cert.json b/test/data/test_cc_oop/fictional_cert.json index d062ff76..3856548d 100644 --- a/test/data/test_cc_oop/fictional_cert.json +++ b/test/data/test_cc_oop/fictional_cert.json @@ -51,5 +51,11 @@ "report_keywords": null, "st_keywords": null }, - "cpe_matching": [] + "heuristics": { + "_type": "Heuristics", + "extracted_versions": null, + "cpe_matches": null, + "verified_cpe_matches": null, + "related_cves": null + } }
\ No newline at end of file diff --git a/test/data/test_cc_oop/toy_dataset.json b/test/data/test_cc_oop/toy_dataset.json index 8fda0b54..62ea04de 100644 --- a/test/data/test_cc_oop/toy_dataset.json +++ b/test/data/test_cc_oop/toy_dataset.json @@ -16,7 +16,7 @@ "certs": [ { "_type": "CommonCriteriaCert", - "dgst": "7ef8227c1aed06bb", + "dgst": "869415cc4b91282e", "status": "active", "category": "Access Control Devices and Systems", "name": "NetIQ Identity Manager 4.7", @@ -28,10 +28,10 @@ ], "not_valid_before": "2020-06-15", "not_valid_after": "2025-06-15", - "report_link": "http://commoncriteriaportal.org/files/epfiles/Certification%20Report%20-%20NetIQ\u00ae%20Identity%20Manager%204.7.pdf", - "st_link": "http://commoncriteriaportal.org/files/epfiles/ST%20-%20NetIQ%20Identity%20Manager%204.7.pdf", + "report_link": "http://www.commoncriteriaportal.org/files/epfiles/Certification%20Report%20-%20NetIQ®%20Identity%20Manager%204.7.pdf", + "st_link": "http://www.commoncriteriaportal.org/files/epfiles/ST%20-%20NetIQ%20Identity%20Manager%204.7.pdf", "src": "csv + html", - "cert_link": "http://commoncriteriaportal.org/files/epfiles/Certifikat%20CCRA%20-%20NetIQ%20Identity%20Manager%204.7_signed.pdf", + "cert_link": "http://www.commoncriteriaportal.org/files/epfiles/Certifikat%20CCRA%20-%20NetIQ%20Identity%20Manager%204.7_signed.pdf", "manufacturer_web": "https://www.netiq.com/", "protection_profiles": [], "maintainance_updates": [], @@ -54,11 +54,17 @@ "report_keywords": null, "st_keywords": null }, - "cpe_matching": [] + "heuristics": { + "_type": "Heuristics", + "extracted_versions": null, + "cpe_matches": null, + "verified_cpe_matches": null, + "related_cves": null + } }, { "_type": "CommonCriteriaCert", - "dgst": "561a012d9a30e960", + "dgst": "2d010ecfb604747a", "status": "active", "category": "Access Control Devices and Systems", "name": "Magic SSO V4.0", @@ -67,8 +73,8 @@ "security_level": [], "not_valid_before": "2019-11-15", "not_valid_after": "2024-11-15", - "report_link": "http://commoncriteriaportal.org/files/epfiles/KECS-CR-19-70%20Magic%20SSO%20V4.0(eng)%20V1.0.pdf", - "st_link": "http://commoncriteriaportal.org/files/epfiles/Magic_SSO_V4.0-ST-v1.4_EN.pdf", + "report_link": "http://www.commoncriteriaportal.org/files/epfiles/KECS-CR-19-70%20Magic%20SSO%20V4.0(eng)%20V1.0.pdf", + "st_link": "http://www.commoncriteriaportal.org/files/epfiles/Magic_SSO_V4.0-ST-v1.4_EN.pdf", "src": "csv + html", "cert_link": null, "manufacturer_web": "https://www.dreamsecurity.com/", @@ -76,7 +82,7 @@ { "_type": "ProtectionProfile", "pp_name": "Korean National Protection Profile for Single Sign On V1.0", - "pp_link": "http://commoncriteriaportal.org/files/ppfiles/KECS-PP-0822-2017%20Korean%20National%20PP%20for%20Single%20Sign%20On%20V1.0(eng).pdf" + "pp_link": "http://www.commoncriteriaportal.org/files/ppfiles/KECS-PP-0822-2017%20Korean%20National%20PP%20for%20Single%20Sign%20On%20V1.0(eng).pdf" } ], "maintainance_updates": [], @@ -99,7 +105,13 @@ "report_keywords": null, "st_keywords": null }, - "cpe_matching": [] + "heuristics": { + "_type": "Heuristics", + "extracted_versions": null, + "cpe_matches": null, + "verified_cpe_matches": null, + "related_cves": null + } } ] }
\ No newline at end of file diff --git a/test/data/test_cpe_cve/auxillary_datasets/cpe_dataset.json b/test/data/test_cpe_cve/auxillary_datasets/cpe_dataset.json new file mode 100644 index 00000000..7bb22f1c --- /dev/null +++ b/test/data/test_cpe_cve/auxillary_datasets/cpe_dataset.json @@ -0,0 +1,6 @@ +{ + "cpe:2.3:a:ibm:security_access_manager_for_enterprise_single_sign-on:8.2.2:*:*:*:*:*:*:*": "IBM Security Access Manager For Enterprise Single Sign-On 8.2.2", + "cpe:2.3:a:ibm:security_key_lifecycle_manager:2.6.0.1:*:*:*:*:*:*:*": "IBM Security Key Lifecycle Manager 2.6.0.1", + "cpe:2.3:a:semperplugins:all_in_one_seo_pack:1.3.6.4:*:*:*:*:wordpress:*:*": "Semper Plugins All in One SEO Pack 1.3.6.4 for WordPress", + "cpe:2.3:a:tracker-software:pdf-xchange_lite_printer:6.0.320.0:*:*:*:*:*:*:*": "Tracker Software PDF-XChange Lite Printer 6.0.320.0" +}
\ No newline at end of file diff --git a/test/data/test_cpe_cve/auxillary_datasets/cve_dataset.json b/test/data/test_cpe_cve/auxillary_datasets/cve_dataset.json new file mode 100644 index 00000000..0572310b --- /dev/null +++ b/test/data/test_cpe_cve/auxillary_datasets/cve_dataset.json @@ -0,0 +1,33 @@ +{ + "_type": "CVEDataset", + "cves": { + "CVE-2017-1732": { + "_type": "CVE", + "cve_id": "CVE-2017-1732", + "vulnerable_cpes": [ + "cpe:2.3:a:ibm:security_access_manager_for_enterprise_single_sign-on:8.2.2:*:*:*:*:*:*:*" + ], + "impact": { + "_type": "Impact", + "base_score": 5.3, + "severity": "MEDIUM", + "explotability_score": 3.9, + "impact_score": 1.4 + } + }, + "CVE-2019-4513": { + "_type": "CVE", + "cve_id": "CVE-2019-4513", + "vulnerable_cpes": [ + "cpe:2.3:a:ibm:security_access_manager_for_enterprise_single_sign-on:8.2.2:*:*:*:*:*:*:*" + ], + "impact": { + "_type": "Impact", + "base_score": 8.2, + "severity": "HIGH", + "explotability_score": 3.9, + "impact_score": 4.2 + } + } + } +}
\ No newline at end of file diff --git a/test/data/test_cpe_cve/vulnerable_dataset.json b/test/data/test_cpe_cve/vulnerable_dataset.json new file mode 100644 index 00000000..30a62017 --- /dev/null +++ b/test/data/test_cpe_cve/vulnerable_dataset.json @@ -0,0 +1,68 @@ +{ + "_type": "CCDataset", + "state": { + "_type": "DatasetInternalState", + "meta_sources_parsed": true, + "pdfs_downloaded": false, + "pdfs_converted": false, + "txt_data_extracted": false, + "certs_analyzed": false + }, + "timestamp": "2021-04-16 15:05:18.386794", + "sha256_digest": "not implemented", + "name": "cc_full_dataset", + "description": "sample dataset description", + "n_certs": 1, + "certs": [ + { + "_type": "CommonCriteriaCert", + "dgst": "c01e5375331b25dc", + "status": "active", + "category": "Access Control Devices and Systems", + "name": "IBM Security Access Manager for Enterprise Single Sign-On Version 8.2", + "manufacturer": "IBM Corporation", + "scheme": "DE", + "security_level": [ + "ALC_FLR.1", + "EAL3+" + ], + "not_valid_before": "2014-12-05", + "not_valid_after": null, + "report_link": "http://www.commoncriteriaportal.org/files/epfiles/0683a_pdf.pdf", + "st_link": "http://www.commoncriteriaportal.org/files/epfiles/0683b_pdf.pdf", + "src": "csv + html", + "cert_link": null, + "manufacturer_web": "http://www.ibm.com", + "protection_profiles": [], + "maintainance_updates": [], + "state": { + "_type": "InternalState", + "st_link_ok": true, + "report_link_ok": true, + "st_convert_ok": true, + "report_convert_ok": true, + "st_extract_ok": true, + "report_extract_ok": true, + "errors": [] + }, + "pdf_data": { + "_type": "PdfData", + "report_metadata": null, + "st_metadata": null, + "report_frontpage": null, + "st_frontpage": null, + "report_keywords": null, + "st_keywords": null + }, + "heuristics": { + "_type": "Heuristics", + "extracted_versions": [ + "8.2" + ], + "cpe_matches": null, + "verified_cpe_matches": null, + "related_cves": null + } + } + ] +}
\ No newline at end of file diff --git a/test/test_cc_oop.py b/test/test_cc_oop.py index f1df4517..3eac6be1 100644 --- a/test/test_cc_oop.py +++ b/test/test_cc_oop.py @@ -24,10 +24,10 @@ class TestCommonCriteriaOOP(TestCase): 'EAL3+'}, date(2020, 6, 15), date(2025, 6, 15), - 'http://commoncriteriaportal.org/files/epfiles/Certification%20Report%20-%20NetIQ\u00ae%20Identity%20Manager%204.7.pdf', - 'http://commoncriteriaportal.org/files/epfiles/ST%20-%20NetIQ%20Identity%20Manager%204.7.pdf', + 'http://www.commoncriteriaportal.org/files/epfiles/Certification%20Report%20-%20NetIQ®%20Identity%20Manager%204.7.pdf', + 'http://www.commoncriteriaportal.org/files/epfiles/ST%20-%20NetIQ%20Identity%20Manager%204.7.pdf', 'csv + html', - 'http://commoncriteriaportal.org/files/epfiles/Certifikat%20CCRA%20-%20NetIQ%20Identity%20Manager%204.7_signed.pdf', + 'http://www.commoncriteriaportal.org/files/epfiles/Certifikat%20CCRA%20-%20NetIQ%20Identity%20Manager%204.7_signed.pdf', 'https://www.netiq.com/', set(), set(), @@ -43,13 +43,13 @@ class TestCommonCriteriaOOP(TestCase): set(), date(2019, 11, 15), date(2024, 11, 15), - 'http://commoncriteriaportal.org/files/epfiles/KECS-CR-19-70%20Magic%20SSO%20V4.0(eng)%20V1.0.pdf', - 'http://commoncriteriaportal.org/files/epfiles/Magic_SSO_V4.0-ST-v1.4_EN.pdf', + 'http://www.commoncriteriaportal.org/files/epfiles/KECS-CR-19-70%20Magic%20SSO%20V4.0(eng)%20V1.0.pdf', + 'http://www.commoncriteriaportal.org/files/epfiles/Magic_SSO_V4.0-ST-v1.4_EN.pdf', 'csv + html', None, 'https://www.dreamsecurity.com/', {CommonCriteriaCert.ProtectionProfile('Korean National Protection Profile for Single Sign On V1.0', - 'http://commoncriteriaportal.org/files/ppfiles/KECS-PP-0822-2017%20Korean%20National%20PP%20for%20Single%20Sign%20On%20V1.0(eng).pdf')}, + 'http://www.commoncriteriaportal.org/files/ppfiles/KECS-PP-0822-2017%20Korean%20National%20PP%20for%20Single%20Sign%20On%20V1.0(eng).pdf')}, set(), None, None, @@ -79,17 +79,17 @@ class TestCommonCriteriaOOP(TestCase): self.template_dataset.timestamp = datetime(2020, 11, 16, hour=17, minute=4, second=14, microsecond=770153) self.template_dataset.state.meta_sources_parsed = True - self.template_report_pdf_hashes = {'7ef8227c1aed06bb': '774c41fbba980191ca40ae610b2f61484c5997417b3325b6fd68b345173bde52', - '561a012d9a30e960': '533a5995ef8b736cc48cfda30e8aafec77d285511471e0e5a9e8007c8750203a'} - self.template_target_pdf_hashes = {'7ef8227c1aed06bb': 'b9a45995d9e40b2515506bbf5945e806ef021861820426c6d0a6a074090b47a9', - '561a012d9a30e960': '3c8614338899d956e9e56f1aa88d90e37df86f3310b875d9d14ec0f71e4759be'} + self.template_report_pdf_hashes = {'869415cc4b91282e': '774c41fbba980191ca40ae610b2f61484c5997417b3325b6fd68b345173bde52', + '2d010ecfb604747a': '533a5995ef8b736cc48cfda30e8aafec77d285511471e0e5a9e8007c8750203a'} + self.template_target_pdf_hashes = {'869415cc4b91282e': 'b9a45995d9e40b2515506bbf5945e806ef021861820426c6d0a6a074090b47a9', + '2d010ecfb604747a': '3c8614338899d956e9e56f1aa88d90e37df86f3310b875d9d14ec0f71e4759be'} self.template_report_txt_path = self.test_data_dir / 'report_869415cc4b91282e.txt' self.template_target_txt_path = self.test_data_dir / 'target_869415cc4b91282e.txt' def test_certificate_input_sanity(self): self.assertEqual(self.crt_one.report_link, - 'http://commoncriteriaportal.org/files/epfiles/Certification%20Report%20-%20NetIQ®%20Identity%20Manager%204.7.pdf', + 'http://www.commoncriteriaportal.org/files/epfiles/Certification%20Report%20-%20NetIQ®%20Identity%20Manager%204.7.pdf', 'Report link contains some improperly escaped characters.') def test_download_and_convert_pdfs(self): @@ -106,14 +106,14 @@ class TestCommonCriteriaOOP(TestCase): self.assertEqual(actual_report_pdf_hashes, self.template_report_pdf_hashes, 'Hashes of downloaded pdfs (certificate report) do not the template') self.assertEqual(actual_target_pdf_hashes, self.template_target_pdf_hashes, 'Hashes of downloaded pdfs (security target) do not match the template') - self.assertTrue(dset['7ef8227c1aed06bb'].state.report_txt_path.exists()) - self.assertTrue(dset['7ef8227c1aed06bb'].state.st_txt_path.exists()) + self.assertTrue(dset['869415cc4b91282e'].state.report_txt_path.exists()) + self.assertTrue(dset['869415cc4b91282e'].state.st_txt_path.exists()) - self.assertAlmostEqual(dset['7ef8227c1aed06bb'].state.st_txt_path.stat().st_size, + self.assertAlmostEqual(dset['869415cc4b91282e'].state.st_txt_path.stat().st_size, self.template_target_txt_path.stat().st_size, delta=1000) - self.assertAlmostEqual(dset['7ef8227c1aed06bb'].state.report_txt_path.stat().st_size, + self.assertAlmostEqual(dset['869415cc4b91282e'].state.report_txt_path.stat().st_size, self.template_report_txt_path.stat().st_size, delta=1000) @@ -155,11 +155,14 @@ class TestCommonCriteriaOOP(TestCase): shutil.copyfile(self.test_data_dir / 'cc_products_active.html', dataset_path / 'web' / 'cc_products_active.html') dset = CCDataset({}, dataset_path, 'sample_dataset', 'sample dataset description') - dset.get_certs_from_web(keep_metadata=False, to_download=False, get_archived=False, get_active=True) + dset.get_certs_from_web(keep_metadata=False, to_download=False, get_archived=False, get_active=True, update_json=False) self.assertEqual(len(os.listdir(dataset_path)), 0, 'Meta files (csv, html) were not deleted properly albeit this was explicitly required.') + for cert in dset: + print(repr(cert.report_link)) + self.assertEqual(len(dset), 2, 'The dataset should contain 2 files.') self.assertTrue(self.crt_one in dset, 'The dataset does not contain the template certificate.') diff --git a/test/test_cve_cpe_matching.py b/test/test_cve_cpe_matching.py new file mode 100644 index 00000000..f84242a8 --- /dev/null +++ b/test/test_cve_cpe_matching.py @@ -0,0 +1,62 @@ +from unittest import TestCase +from sec_certs.dataset import CCDataset +from sec_certs.certificate import CommonCriteriaCert +from sec_certs.cpe import CPEDataset, CPE +from sec_certs.cve import CVEDataset, CVE +from pathlib import Path + + +class TestCPEandCVEMatching(TestCase): + def setUp(self) -> None: + self.test_data_dir = Path(__file__).parent / 'data' / 'test_cpe_cve' + self.cc_dset = CCDataset.from_json(self.test_data_dir / 'vulnerable_dataset.json') + self.cc_dset.compute_heuristics(update_json=False) + + self.cpes = [CPE("cpe:2.3:a:ibm:security_access_manager_for_enterprise_single_sign-on:8.2.2:*:*:*:*:*:*:*", "IBM Security Access Manager For Enterprise Single Sign-On 8.2.2"), + CPE("cpe:2.3:a:ibm:security_key_lifecycle_manager:2.6.0.1:*:*:*:*:*:*:*", "IBM Security Key Lifecycle Manager 2.6.0.1"), + CPE("cpe:2.3:a:semperplugins:all_in_one_seo_pack:1.3.6.4:*:*:*:*:wordpress:*:*", "Semper Plugins All in One SEO Pack 1.3.6.4 for WordPress"), + CPE("cpe:2.3:a:tracker-software:pdf-xchange_lite_printer:6.0.320.0:*:*:*:*:*:*:*", "Tracker Software PDF-XChange Lite Printer 6.0.320.0")] + self.cpe_dset = CPEDataset({x.uri: x for x in self.cpes}) + + self.cves = [CVE('CVE-2017-1732', ['cpe:2.3:a:ibm:security_access_manager_for_enterprise_single_sign-on:8.2.2:*:*:*:*:*:*:*'], CVE.Impact(5.3, 'MEDIUM', 3.9, 1.4)), + CVE('CVE-2019-4513', ['cpe:2.3:a:ibm:security_access_manager_for_enterprise_single_sign-on:8.2.2:*:*:*:*:*:*:*'], CVE.Impact(8.2, 'HIGH', 3.9, 4.2))] + self.cve_dset = CVEDataset({x.cve_id: x for x in self.cves}) + + def test_load_cpe_dataset(self): + json_cpe_dset = CPEDataset.from_json(self.test_data_dir / 'auxillary_datasets' / 'cpe_dataset.json') + self.assertEqual(self.cpe_dset, json_cpe_dset, 'CPE template dataset does not match CPE dataset loaded from json.') + + def test_cpe_lookup_dicts(self): + self.assertEqual(self.cpe_dset.vendors, {'ibm', 'tracker-software', 'semperplugins'}, + 'The set of versions in CPE dataset does not match template') + self.assertEqual(self.cpe_dset.vendor_to_versions, {'ibm': {'8.2.2', '2.6.0.1'}, 'semperplugins': {'1.3.6.4'}, 'tracker-software': {'6.0.320.0'}}, + 'The CPE lookup dictionary vendor->version of CPE dataset does not match template.') + self.assertEqual(self.cpe_dset.vendor_version_to_cpe, {('ibm', '8.2.2'): {CPE('cpe:2.3:a:ibm:security_access_manager_for_enterprise_single_sign-on:8.2.2:*:*:*:*:*:*:*', 'IBM Security Access Manager For Enterprise Single Sign-On 8.2.2')}, ('ibm', '2.6.0.1'): {CPE('cpe:2.3:a:ibm:security_key_lifecycle_manager:2.6.0.1:*:*:*:*:*:*:*', 'IBM Security Key Lifecycle Manager 2.6.0.1')}, ('semperplugins', '1.3.6.4'): {CPE('cpe:2.3:a:semperplugins:all_in_one_seo_pack:1.3.6.4:*:*:*:*:wordpress:*:*', 'Semper Plugins All in One SEO Pack 1.3.6.4 for WordPress')}, ('tracker-software', '6.0.320.0'): {CPE('cpe:2.3:a:tracker-software:pdf-xchange_lite_printer:6.0.320.0:*:*:*:*:*:*:*', 'Tracker Software PDF-XChange Lite Printer 6.0.320.0')}}, + 'The CPE lookup dictionary (vendor,version)->cpe does not match the template.') + + def test_cve_lookup_dicts(self): + alt_lookup = {x: set(y) for x,y in self.cve_dset.cpes_to_cve_lookup.items()} + self.assertEqual(alt_lookup, {'cpe:2.3:a:ibm:security_access_manager_for_enterprise_single_sign-on:8.2.2:*:*:*:*:*:*:*': {'CVE-2017-1732', 'CVE-2019-4513'}}, + 'The CVE lookup dicionary cve-> affected cpes does not match the template') + + def test_load_cve_dataset(self): + json_cve_dset = CVEDataset.from_json(self.test_data_dir / 'auxillary_datasets' / 'cve_dataset.json') + self.assertEqual(self.cve_dset, json_cve_dset, 'CVE template dataset does not match CVE dataset loaded from json.') + + def test_match_cpe(self): + self.assertTrue(self.cpes[0] in [x[1] for x in self.cc_dset['c01e5375331b25dc'].heuristics.cpe_matches], 'The CPE matching algorithm did not find the right CPE.') + self.assertTrue(len(self.cc_dset['c01e5375331b25dc'].heuristics.cpe_matches) == 1, 'Exactly one CPE match should be found.') + + def test_find_related_cves(self): + self.cc_dset['c01e5375331b25dc'].heuristics.verified_cpe_matches = [self.cpes[0]] + self.cc_dset.compute_related_cves() + self.assertCountEqual([x.cve_id for x in self.cves], self.cc_dset['c01e5375331b25dc'].heuristics.related_cves, 'The computed CVEs do not match the excpected CVEs') + + def test_version_extraction(self): + self.assertEqual(self.cc_dset['c01e5375331b25dc'].heuristics.extracted_versions, ['8.2'], 'The version extracted from the certificate does not match the template') + new_cert = CommonCriteriaCert('', '', 'IDOneClassIC Card : ID-One Cosmo 64 RSA v5.4 and applet IDOneClassIC v1.0 embedded on P5CT072VOP', '', '', + '', None, None, '', '', '', '', '', set(), set(), None, None, None) + new_cert.compute_heuristics_version() + self.assertEqual(set(new_cert.heuristics.extracted_versions), {'5.4', '1.0'}, 'The extracted versions do not match the template.') + + |
