From fccf787c1ddca5ffd4095ef1c3cb99f8feb8c78a Mon Sep 17 00:00:00 2001 From: Adam Janovsky Date: Fri, 19 Feb 2021 17:39:09 +0100 Subject: Improves parallel processing function - Adds option to unpack arguments --- sec_certs/cert_processing.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/sec_certs/cert_processing.py b/sec_certs/cert_processing.py index 739bfa9f..b5519aab 100644 --- a/sec_certs/cert_processing.py +++ b/sec_certs/cert_processing.py @@ -5,13 +5,16 @@ import time def process_parallel(func: Callable, items: Iterable, max_workers: int, callback: Optional[Callable] = None, - use_threading: bool = True, progress_bar: bool = True): + use_threading: bool = True, progress_bar: bool = True, unpack: bool = False): if use_threading is True: pool = ThreadPool(max_workers) else: pool = Pool(max_workers) - results = [pool.apply_async(func, (i, ), callback=callback) for i in items] + if unpack is False: + results = [pool.apply_async(func, (i, ), callback=callback) for i in items] + else: + results = [pool.apply_async(func, (*i, ), callback=callback) for i in items] if progress_bar is True: bar = tqdm(total=len(results)) -- cgit v1.3.1 From a060fd587830cd410503a4baeacf44518e5ac452 Mon Sep 17 00:00:00 2001 From: Adam Janovsky Date: Fri, 19 Feb 2021 17:39:56 +0100 Subject: Fix in Web source parsing, pdf metadata parsing - Rows of csv that are corrupted are now skipped and logged - Links from csv and html are now identical and thus correctly matched - Implements extraction of pdf metadata from ST and reports --- sec_certs/certificate.py | 50 ++++++++++++++++++++++++++++++++++++++++++------ sec_certs/dataset.py | 40 +++++++++++++++++++++++++++++++++----- sec_certs/helpers.py | 22 +++++++++++++++++++++ 3 files changed, 101 insertions(+), 11 deletions(-) diff --git a/sec_certs/certificate.py b/sec_certs/certificate.py index 39c56a3d..04dd41d1 100644 --- a/sec_certs/certificate.py +++ b/sec_certs/certificate.py @@ -743,8 +743,8 @@ class FIPSCertificate(Certificate, ComplexSerializableType): class CommonCriteriaCert(Certificate, ComplexSerializableType): - cc_url = 'http://www.commoncriteriaportal.org' - empty_st_url = 'http://www.commoncriteriaportal.org/files/epfiles/' + cc_url = 'http://commoncriteriaportal.org' + empty_st_url = 'http://commoncriteriaportal.org/files/epfiles/' @dataclass(eq=True, frozen=True) class MaintainanceReport(ComplexSerializableType): @@ -803,21 +803,43 @@ class CommonCriteriaCert(Certificate, ComplexSerializableType): report_link_ok: bool st_convert_ok: bool report_convert_ok: bool + st_extract_ok: bool + report_extract_ok: bool st_pdf_path: Path report_pdf_path: Path st_txt_path: Path report_txt_path: Path def __init__(self, st_link_ok: bool = True, report_link_ok: bool = True, - st_convert_ok: bool = True, report_convert_ok: bool = True): + st_convert_ok: bool = True, report_convert_ok: bool = True, + st_extract_ok: bool = True, report_extract_ok: bool = True): self.st_link_ok = st_link_ok self.report_link_ok = report_link_ok self.st_convert_ok = st_convert_ok self.report_convert_ok = report_convert_ok + self.st_extract_ok = st_extract_ok + self.report_extract_ok = report_extract_ok def to_dict(self): return {'st_link_ok': self.st_link_ok, 'report_link_ok': self.report_link_ok, - 'st_convert_ok': self.st_convert_ok, 'report_convert_ok': self.report_convert_ok} + 'st_convert_ok': self.st_convert_ok, 'report_convert_ok': self.report_convert_ok, + 'st_extract_ok': self.st_extract_ok, 'report_extract_ok': self.report_extract_ok} + + @classmethod + def from_dict(cls, dct: Dict[str, bool]): + return cls(*tuple(dct.values())) + + @dataclass(init=False) + class PdfData(ComplexSerializableType): + report_metadata: Dict[str, str] + st_metadata: Dict[str, str] + + def __init__(self, report_metadata: Optional[Dict[str, str]] = None, st_metadata: Optional[Dict[str, str]] = None): + self.report_metadata = report_metadata + self.st_metadata = st_metadata + + def to_dict(self): + return {'report_metadata': self.report_metadata, 'st_metadata': self.st_metadata} @classmethod def from_dict(cls, dct: Dict[str, bool]): @@ -829,7 +851,8 @@ class CommonCriteriaCert(Certificate, ComplexSerializableType): manufacturer_web: Optional[str], protection_profiles: set, maintainance_updates: set, - state: Optional[InternalState]): + state: Optional[InternalState], + pdf_data: Optional[PdfData]): super().__init__() self.category = category @@ -852,6 +875,11 @@ class CommonCriteriaCert(Certificate, ComplexSerializableType): else: self.state = self.InternalState() + if pdf_data is not None: + self.pdf_data = pdf_data + else: + self.pdf_data = self.PdfData() + @property def dgst(self) -> str: """ @@ -1005,7 +1033,7 @@ class CommonCriteriaCert(Certificate, ComplexSerializableType): maintainance_div) if maintainance_div else set() return cls(category, name, manufacturer, scheme, security_level, not_valid_before, not_valid_after, report_link, - st_link, 'html', cert_link, manufacturer_web, protection_profiles, maintainances, None) + st_link, 'html', cert_link, manufacturer_web, protection_profiles, maintainances, None, None) def set_local_paths(self, report_pdf_dir: Optional[Union[str, Path]], @@ -1055,3 +1083,13 @@ class CommonCriteriaCert(Certificate, ComplexSerializableType): logger.error(f'Cert dgst: {cert.dgst} failed to convert security target pdf->txt') cert.state.st_convert_ok = False return cert + + @staticmethod + def extract_st_pdf_metadata(cert: 'CommonCriteriaCert') -> 'CommonCriteriaCert': + cert.pdf_data.st_metadata = helpers.extract_pdf_metadata(cert.state.st_pdf_path)[1] + return cert + + @staticmethod + def extract_report_pdf_metadata(cert: 'CommonCriteriaCert') -> 'CommonCriteriaCert': + cert.pdf_data.report_metadata = helpers.extract_pdf_metadata(cert.state.report_pdf_path)[1] + return cert \ No newline at end of file diff --git a/sec_certs/dataset.py b/sec_certs/dataset.py index b852e6bc..0b0633e9 100644 --- a/sec_certs/dataset.py +++ b/sec_certs/dataset.py @@ -107,7 +107,8 @@ class Dataset(ABC): def _download_parallel(urls: Collection[str], paths: Collection[Path], prune_corrupted: bool = True): exit_codes = cert_processing.process_parallel(helpers.download_file, list(zip(urls, paths)), - constants.N_THREADS) + constants.N_THREADS, + unpack=True) n_successful = len([e for e in exit_codes if e == requests.codes.ok]) logger.info(f'Successfully downloaded {n_successful} files, {len(exit_codes) - n_successful} failed.') @@ -282,9 +283,9 @@ class CCDataset(Dataset, ComplexSerializableType): 'not_valid_before', 'not_valid_after', 'report_link', 'st_link', 'maintainance_date', 'maintainance_title', 'maintainance_report_link', 'maintainance_st_link'] - df = pd.read_csv(file, engine='python', encoding='windows-1250') - df = df.rename( - columns={x: y for (x, y) in zip(list(df.columns), csv_header)}) + # 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 = df.rename(columns={x: y for (x, y) in zip(list(df.columns), csv_header)}) df['is_maintainance'] = ~df.maintainance_title.isnull() df = df.fillna(value='') @@ -317,7 +318,7 @@ class CCDataset(Dataset, ComplexSerializableType): certs = {x.dgst: CommonCriteriaCert(x.category, x.cert_name, x.manufacturer, x.scheme, x.security_level, x.not_valid_before, x.not_valid_after, x.report_link, x.st_link, 'csv', - None, None, profiles.get(x.dgst, None), updates.get(x.dgst, None), None) for + None, None, profiles.get(x.dgst, None), updates.get(x.dgst, None), None, None) for x in df_base.itertuples()} return certs @@ -475,6 +476,35 @@ class CCDataset(Dataset, ComplexSerializableType): logger.info('Attempting to re-convert failed target pdfs') self._convert_targets_to_txt(False) + def _extract_report_metadata(self, fresh: bool = True): + if fresh is True: + certs_to_process = self.certs.values() + else: + certs_to_process = [x for x in self.certs.values() if not x.state.st_extract_ok] + cert_processing.process_parallel(CommonCriteriaCert.extract_report_pdf_metadata, certs_to_process, constants.N_THREADS) + + def _extract_targets_metadata(self, fresh: bool = True): + if fresh is True: + certs_to_process = self.certs.values() + else: + certs_to_process = [x for x in self.certs.values() if not x.state.st_extract_ok] + cert_processing.process_parallel(CommonCriteriaCert.extract_st_pdf_metadata, certs_to_process, constants.N_THREADS) + + def extract_pdf_metadata(self, fresh: bool = True): + logger.info('Extracting pdf metadata from CC dataset') + self._extract_report_metadata(fresh) + self._extract_targets_metadata(fresh) + + def extract_all_keywords(self, fresh: bool = True): + pass + + def extract_pdf_frontpage(self, fresh: bool = True): + pass + + def extract_data(self, fresh:bool = True): + # TODO: Extract metadata, keywords, frontpage, attempt to re-do once if fails + pass + class FIPSDataset(Dataset, ComplexSerializableType): FIPS_BASE_URL: ClassVar[str] = 'https://csrc.nist.gov' diff --git a/sec_certs/helpers.py b/sec_certs/helpers.py index 67b90347..f843a121 100644 --- a/sec_certs/helpers.py +++ b/sec_certs/helpers.py @@ -15,6 +15,8 @@ import pandas as pd import subprocess import sec_certs.constants as constants +from PyPDF2 import PdfFileReader + logger = logging.getLogger(__name__) @@ -165,3 +167,23 @@ def repair_pdf(file: Path): def convert_pdf_file(pdf_path: Path, txt_path: Path, options): return subprocess.run(['pdftotext', *options, pdf_path, txt_path], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=60).returncode + + +def extract_pdf_metadata(filepath: Path): + metadata = dict() + metadata['pdf_file_size_bytes'] = filepath.stat().st_size + + try: + with filepath.open('rb') as handle: + pdf = PdfFileReader(handle) + + metadata['pdf_is_encrypted'] = pdf.getIsEncrypted() + metadata['pdf_number_of_pages'] = pdf.getNumPages() + + for key, val in pdf.getDocumentInfo().items(): + metadata[key] = str(val) + + except Exception as e: + logger.error(f'Failed to read metadata of {filepath}, error: {e}') + + return filepath, metadata -- cgit v1.3.1 From b2ce38886a3b20002a3ec75e1ec7f914ca6cef48 Mon Sep 17 00:00:00 2001 From: Adam Janovsky Date: Fri, 19 Feb 2021 22:22:02 +0100 Subject: Plug-in old extractions methods to new OOP frame - Tests for txt file processing still missing - The internals of the txt processing methods are still authored by petrs - They are however fully parallel and called in an OOP fashion - Overall, all results are gathered in a single json file - All results are implicitly paired --- cc_oop_demo.py | 7 +- process_certificates.py | 3 +- sec_certs/certificate.py | 44 +++- sec_certs/dataset.py | 60 +++++- sec_certs/extract_certificates.py | 12 +- sec_certs/helpers.py | 277 +++++++++++++++++++++++++- test/data/test_cc_oop/cc_products_active.csv | 4 +- test/data/test_cc_oop/cc_products_active.html | 67 +++---- test/data/test_cc_oop/fictional_cert.json | 13 +- test/data/test_cc_oop/toy_dataset.json | 38 +++- test/test_cc_oop.py | 34 ++-- 11 files changed, 471 insertions(+), 88 deletions(-) diff --git a/cc_oop_demo.py b/cc_oop_demo.py index a49a2e7d..49729400 100644 --- a/cc_oop_demo.py +++ b/cc_oop_demo.py @@ -22,7 +22,7 @@ def main(): dset = CCDataset({}, Path('./debug_dataset'), 'sample_dataset', 'sample dataset description') # Load metadata for certificates from CSV and HTML sources - dset.get_certs_from_web() + dset.get_certs_from_web(to_download=False) logger.info(f'Finished parsing. Have dataset with {len(dset)} certificates.') # # Dump dataset into JSON @@ -39,6 +39,11 @@ def main(): # Convert pdfs to text dset.convert_all_pdfs() + # Extract data from txt files + dset.extract_data() + + dset.to_json('./debug_dataset/cc_full_dataset.json') + end = datetime.now() logger.info(f'The computation took {(end-start)} seconds.') diff --git a/process_certificates.py b/process_certificates.py index badd84e8..f5bbcf39 100755 --- a/process_certificates.py +++ b/process_certificates.py @@ -5,6 +5,7 @@ from sec_certs.files import load_json_files from sec_certs.extract_certificates import * from sec_certs.analyze_certificates import * from sec_certs.download import download_cc_web, download_cc +from sec_certs.cert_rules import rules as cc_search_rules @click.command() @@ -118,7 +119,7 @@ def main(directory, do_complete_extraction: bool, do_download_meta: bool, do_ext convert_pdf_files(walk_dir, threads, ["-raw"]) if do_extraction_certs: - all_keywords = extract_certificates_keywords_parallel(walk_dir, fragments_dir, 'certificate', threads) + all_keywords = extract_certificates_keywords_parallel(walk_dir, fragments_dir, 'certificate', cc_search_rules, threads) with open(results_dir / "certificate_data_keywords_all.json", "w") as write_file: json.dump(all_keywords, write_file, indent=4, sort_keys=True) diff --git a/sec_certs/certificate.py b/sec_certs/certificate.py index 04dd41d1..775e3d07 100644 --- a/sec_certs/certificate.py +++ b/sec_certs/certificate.py @@ -833,13 +833,24 @@ class CommonCriteriaCert(Certificate, ComplexSerializableType): class PdfData(ComplexSerializableType): report_metadata: Dict[str, str] st_metadata: Dict[str, str] - - def __init__(self, report_metadata: Optional[Dict[str, str]] = None, st_metadata: Optional[Dict[str, str]] = None): + report_frontpage: Dict[str, str] + st_frontpage: Dict[str, str] + report_keywords: Dict[str, str] + st_keywords: Dict[str, str] + + def __init__(self, report_metadata: Optional[Dict[str, str]] = None, st_metadata: Optional[Dict[str, str]] = None, + report_frontpage: Optional[Dict[str, str]] = None, st_frontpage: Optional[Dict[str, str]] = None, + report_keywords: Optional[Dict[str,str]] = None, st_keywords: Optional[Dict[str, str]] = None): self.report_metadata = report_metadata self.st_metadata = st_metadata + self.report_frontpage = report_frontpage + self.st_frontpage = st_frontpage + self.report_keywords = report_keywords + self.st_keywords = st_keywords def to_dict(self): - return {'report_metadata': self.report_metadata, 'st_metadata': self.st_metadata} + return {'report_metadata': self.report_metadata, 'st_metadata': self.st_metadata, 'report_frontpage': self.report_frontpage, + 'st_frontpage': self.st_frontpage, 'report_keywords': self.report_keywords, 'st_keywords': self.st_keywords} @classmethod def from_dict(cls, dct: Dict[str, bool]): @@ -1092,4 +1103,29 @@ class CommonCriteriaCert(Certificate, ComplexSerializableType): @staticmethod def extract_report_pdf_metadata(cert: 'CommonCriteriaCert') -> 'CommonCriteriaCert': cert.pdf_data.report_metadata = helpers.extract_pdf_metadata(cert.state.report_pdf_path)[1] - return cert \ No newline at end of file + return cert + + @staticmethod + def extract_st_pdf_frontpage(cert: 'CommonCriteriaCert') -> 'CommonCriteriaCert': + cert.pdf_data.st_frontpage = dict() + cert.pdf_data.st_frontpage['bsi'] = helpers.search_only_headers_bsi(cert.state.st_txt_path) + cert.pdf_data.st_frontpage['anssi'] = helpers.search_only_headers_anssi(cert.state.st_txt_path) + return cert + + @staticmethod + def extract_report_pdf_frontpage(cert: 'CommonCriteriaCert') -> 'CommonCriteriaCert': + cert.pdf_data.report_frontpage = dict() + cert.pdf_data.report_frontpage['bsi'] = helpers.search_only_headers_bsi(cert.state.report_txt_path) + cert.pdf_data.report_frontpage['anssi'] = helpers.search_only_headers_anssi(cert.state.report_txt_path) + return cert + + @staticmethod + def extract_report_pdf_keywords(cert: 'CommonCriteriaCert') -> 'CommonCriteriaCert': + cert.pdf_data.report_keywords = helpers.extract_keywords(cert.state.report_txt_path) + return cert + + @staticmethod + def extract_st_pdf_keywords(cert: 'CommonCriteriaCert') -> 'CommonCriteriaCert': + cert.pdf_data.st_keywords = helpers.extract_keywords(cert.state.st_txt_path) + return cert + diff --git a/sec_certs/dataset.py b/sec_certs/dataset.py index 0b0633e9..80eb49bd 100644 --- a/sec_certs/dataset.py +++ b/sec_certs/dataset.py @@ -480,7 +480,7 @@ class CCDataset(Dataset, ComplexSerializableType): if fresh is True: certs_to_process = self.certs.values() else: - certs_to_process = [x for x in self.certs.values() if not x.state.st_extract_ok] + certs_to_process = [x for x in self.certs.values() if not x.state.report_extract_ok] cert_processing.process_parallel(CommonCriteriaCert.extract_report_pdf_metadata, certs_to_process, constants.N_THREADS) def _extract_targets_metadata(self, fresh: bool = True): @@ -495,15 +495,61 @@ class CCDataset(Dataset, ComplexSerializableType): self._extract_report_metadata(fresh) self._extract_targets_metadata(fresh) - def extract_all_keywords(self, fresh: bool = True): - pass + def _extract_targets_frontpage(self, fresh: bool = True): + if fresh is True: + certs_to_process = self.certs.values() + else: + certs_to_process = [x for x in self.certs.values() if not x.state.st_extract_ok] + cert_processing.process_parallel(CommonCriteriaCert.extract_st_pdf_frontpage, certs_to_process, constants.N_THREADS) + + def _extract_report_frontpage(self, fresh: bool = True): + if fresh is True: + certs_to_process = self.certs.values() + else: + certs_to_process = [x for x in self.certs.values() if not x.state.report_extract_ok] + cert_processing.process_parallel(CommonCriteriaCert.extract_report_pdf_frontpage, certs_to_process, constants.N_THREADS) def extract_pdf_frontpage(self, fresh: bool = True): - pass + logger.info('Extracting pdf frontpages from CC dataset.') + self._extract_report_frontpage(fresh) + self._extract_targets_frontpage(fresh) - def extract_data(self, fresh:bool = True): - # TODO: Extract metadata, keywords, frontpage, attempt to re-do once if fails - pass + def _extract_report_keywords(self, fresh: bool = True): + if fresh is True: + certs_to_process = self.certs.values() + else: + certs_to_process = [x for x in self.certs.values() if not x.state.report_extract_ok] + cert_processing.process_parallel(CommonCriteriaCert.extract_report_pdf_keywords, certs_to_process, constants.N_THREADS) + + def _extract_targets_keywords(self, fresh: bool = True): + if fresh is True: + certs_to_process = self.certs.values() + else: + certs_to_process = [x for x in self.certs.values() if not x.state.st_extract_ok] + cert_processing.process_parallel(CommonCriteriaCert.extract_st_pdf_keywords, certs_to_process, constants.N_THREADS) + + def extract_pdf_keywords(self, fresh: bool = True): + logger.info('Extracting pdf keywords from CC dataset.') + self._extract_report_keywords(fresh) + self._extract_targets_keywords(fresh) + + def extract_data(self, fresh: bool = True): + logger.info('Extracting various stuff from converted txt filed from CC dataset.') + self.extract_pdf_metadata(fresh) + self.extract_pdf_frontpage(fresh) + self.extract_pdf_keywords(fresh) + + if fresh is True: + if any(filter(lambda x: not x.state.report_extract_ok, self.certs.values())): + logger.info('Attempting to re-extract failed data from report txts') + self._extract_report_metadata(False) + self._extract_report_frontpage(False) + self._extract_report_keywords(False) + if any(filter(lambda x: not x.state.st_extract_ok, self.certs.values())): + logger.info('Attempting to re-extract failed data from ST txts') + self._extract_targets_metadata(False) + self._extract_targets_frontpage(False) + self._extract_targets_keywords(False) class FIPSDataset(Dataset, ComplexSerializableType): diff --git a/sec_certs/extract_certificates.py b/sec_certs/extract_certificates.py index aada02b7..7de9a053 100644 --- a/sec_certs/extract_certificates.py +++ b/sec_certs/extract_certificates.py @@ -24,6 +24,10 @@ from .cert_rules import rules, fips_rules, REGEXEC_SEP from .files import search_files, load_cert_html_file, FILE_ERRORS_STRATEGY from .constants import * + +import sec_certs.constants as constants +import sec_certs.helpers as helpers + plt.rcdefaults() # if True, then exception is raised when unexpect intermediate number is obtained @@ -416,11 +420,11 @@ def search_only_headers_bsi(walk_dir: Path): if end_pos != -1: developer = developer[:end_pos] - items_found[const.TAG_CERT_ID] = normalize_match_string(cert_id) - items_found[const.TAG_CERT_ITEM] = normalize_match_string( + items_found[constants.TAG_CERT_ID] = normalize_match_string(cert_id) + items_found[constants.TAG_CERT_ITEM] = normalize_match_string( certified_item) - items_found[const.TAG_DEVELOPER] = normalize_match_string(developer) - items_found[const.TAG_CERT_LAB] = 'BSI' + items_found[constants.TAG_DEVELOPER] = normalize_match_string(developer) + items_found[constants.TAG_CERT_LAB] = 'BSI' # # Process page with more detailed certificate info diff --git a/sec_certs/helpers.py b/sec_certs/helpers.py index f843a121..bbb1be42 100644 --- a/sec_certs/helpers.py +++ b/sec_certs/helpers.py @@ -14,9 +14,13 @@ import numpy as np import pandas as pd import subprocess import sec_certs.constants as constants +from enum import Enum from PyPDF2 import PdfFileReader +import sec_certs.extract_certificates as extract_certificates +from sec_certs.cert_rules import REGEXEC_SEP +from sec_certs.cert_rules import rules as cc_search_rules logger = logging.getLogger(__name__) @@ -166,7 +170,8 @@ def repair_pdf(file: Path): def convert_pdf_file(pdf_path: Path, txt_path: Path, options): - return subprocess.run(['pdftotext', *options, pdf_path, txt_path], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=60).returncode + return subprocess.run(['pdftotext', *options, pdf_path, txt_path], stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, timeout=60).returncode def extract_pdf_metadata(filepath: Path): @@ -187,3 +192,273 @@ def extract_pdf_metadata(filepath: Path): logger.error(f'Failed to read metadata of {filepath}, error: {e}') return filepath, metadata + + +# TODO: Please, refactor me. I reallyyyyyyyyyyyyy need it!!!!!! +def search_only_headers_anssi(filepath: Path): + class HEADER_TYPE(Enum): + HEADER_FULL = 1 + HEADER_MISSING_CERT_ITEM_VERSION = 2 + HEADER_MISSING_PROTECTION_PROFILES = 3 + HEADER_DUPLICITIES = 4 + + rules_certificate_preface = [ + (HEADER_TYPE.HEADER_FULL, + 'Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.*)Conformité à un profil de protection(.+)Critères d\'évaluation et version(.+)Niveau d\'évaluation(.+)Développeurs(.+)Centre d\'évaluation(.+)Accords de reconnaissance applicables'), + (HEADER_TYPE.HEADER_FULL, + 'Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.*)Conformité à un profil de protection(.+)Critères d’évaluation et version(.+)Niveau d’évaluation(.+)Développeurs(.+)Centre d’évaluation(.+)Accords de reconnaissance applicables'), + (HEADER_TYPE.HEADER_FULL, + 'Référence du rapport de certification(.+)Nom du produit(.+)()Conformité à un profil de protection(.+)Critères d\'évaluation et version(.+)Niveau d\'évaluation(.+)Développeur (.+)Centre d\'évaluation(.+)Accords de reconnaissance applicables'), + (HEADER_TYPE.HEADER_FULL, + 'Référence du rapport de certification(.+)Nom des produits(.+)Référence/version des produits(.+)Conformité à un profil de protection(.+)Critères d\'évaluation et version(.+)Niveau d\'évaluation(.+)Développeur\(s\)(.+)Centre d\'évaluation(.+)Accords de reconnaissance applicables'), + (HEADER_TYPE.HEADER_FULL, + 'Référence du rapport de certification(.+)Nom des produits(.+)Référence/version des produits(.+)Conformité à un profil de protection(.+)Critères d\'évaluation et version(.+)Niveau d\'évaluation(.+)Développeur (.+)Centre d\'évaluation(.+)Accords de reconnaissance'), + (HEADER_TYPE.HEADER_FULL, + 'Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.+)Conformité aux profils de protection(.+)Critères d’évaluation et version(.+)Niveau d’évaluation(.+)Développeur\(s\)(.+)Centre d’évaluation(.+)Accords de reconnaissance applicables'), + (HEADER_TYPE.HEADER_FULL, + 'Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.+)Conformité à un profil de protection(.+)Critères d’évaluation et version(.+)Niveau d’évaluation(.+)Développeur\(s\)(.+)Centre d’évaluation(.+)Accords de reconnaissance applicables'), + (HEADER_TYPE.HEADER_FULL, + 'Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.+)Conformité à un profil de protection(.+)Critères d’évaluation et version(.+)Niveau d’évaluation(.+)Développeur (.+)Centre d’évaluation(.+)Accords de reconnaissance applicables'), + (HEADER_TYPE.HEADER_FULL, + 'Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.+)Conformité à des profils de protection(.+)Critères d’évaluation et version(.+)Niveau d’évaluation(.+)Développeurs(.+)Centre d’évaluation(.+)Accords de reconnaissance applicables'), + (HEADER_TYPE.HEADER_FULL, + 'Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.+)Conformité aux profils de protection(.+)Critères d\’évaluation et version(.+)Niveau d\’évaluation(.+)Développeurs(.+)Centre d\’évaluation(.+)Accords de reconnaissance applicables'), + (HEADER_TYPE.HEADER_FULL, + 'Référence du rapport de certification(.+)Nom du produit \(référence/version\)(.+)Nom de la TOE \(référence/version\)(.+)Conformité à un profil de protection(.+)Critères d\’évaluation et version(.+)Niveau d\’évaluation(.+)Développeurs(.+)Centre d’évaluation(.+)Accords de reconnaissance applicables'), + (HEADER_TYPE.HEADER_FULL, + 'Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.+)Conformité aux profil de protection(.+)Critères d’évaluation et version(.+)Niveau d’évaluation(.+)Développeur\(s\)(.+)Centre d’évaluation(.+)Accords de reconnaissance applicables'), + (HEADER_TYPE.HEADER_FULL, + 'Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.+)Conformité à un profil de protection(.+)Critères d\'évaluation et version(.+)Niveau d\'évaluation(.+)Développeur\(s\)(.+)Centre d\'évaluation(.+)Accords de reconnaissance applicables'), + (HEADER_TYPE.HEADER_FULL, + 'Référence du rapport de certification(.+)Nom du produit \(référence/version\)(.+)Nom de la TOE \(référence/version\)(.+)Conformité à un profil de protection(.+)Critères d\'évaluation et version(.+)Niveau d\'évaluation(.+)Développeurs(.+)Centre d\'évaluation(.+)Accords de reconnaissance applicables'), + (HEADER_TYPE.HEADER_FULL, + 'Référence du rapport de certification(.+)Nom du produit(.+)Référence du produit(.+)Conformité à un profil de protection(.+)Critères d\'évaluation et version(.+)Niveau d\'évaluation(.+)Développeurs(.+)Centre d\'évaluation(.+)Accords de reconnaissance applicables'), + (HEADER_TYPE.HEADER_FULL, + 'Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.+)Conformité aux profils de protection(.+)Critères d\'évaluation et version(.+)Niveau d\'évaluation(.+)Développeurs(.+)Centre d\'évaluation(.+)Accords de reconnaissance applicables'), + + (HEADER_TYPE.HEADER_FULL, + 'Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.+)Conformité àun profil de protection(.+)Critères d’évaluation et version(.+)Niveau d’évaluation(.+)Développeurs(.+)Centre d’évaluation(.+)Accords de reconnaissance applicables'), + (HEADER_TYPE.HEADER_FULL, + 'Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.+)Conformité àun profil de protection(.+)Critères d’évaluation et version(.+)Niveau d’évaluation(.+)Développeur\(s\)(.+)d’évaluation(.+)Accords de reconnaissance applicables'), + (HEADER_TYPE.HEADER_FULL, + 'Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.+)Conformité àun profil de protection(.+)Critères d’évaluation et version(.+)Niveau d’évaluation(.+)Développeur (.+)Centre d’évaluation(.+)Accords de reconnaissance applicables'), + (HEADER_TYPE.HEADER_FULL, + 'Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.+)Conformité àdes profils de protection(.+)Critères d’évaluation et version(.+)Niveau d’évaluation(.+)Développeurs(.+)Centre d’évaluation(.+)Accords de reconnaissance applicables'), + (HEADER_TYPE.HEADER_FULL, + 'Référence du rapport de certification(.+)Nom du produit \(référence/version\)(.+)Nom de la TOE \(référence/version\)(.+)Conformité àun profil de protection(.+)Critères d’évaluation et version(.+)Niveau d’évaluation(.+)Développeurs(.+)Centre d’évaluation(.+)Accords de reconnaissance applicables'), + (HEADER_TYPE.HEADER_FULL, + 'Certification Report(.+)Nom du produit(.+)Référence/version du produit(.*)Conformité à un profil de protection(.+)Critères d\'évaluation et version(.+)Niveau d\'évaluation(.+)Développeurs(.+)Centre d\'évaluation(.+)Accords de reconnaissance applicables'), + (HEADER_TYPE.HEADER_FULL, + 'Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.+)Conformité aux profisl de protection(.+)Critères d’évaluation et version(.+)Niveau d’évaluation(.+)Développeurs(.+)Centre d’évaluation(.+)Accords de reconnaissance applicables'), + (HEADER_TYPE.HEADER_FULL, + 'Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.+)Conformité àun profil de protection(.+)Critères d’évaluation et version(.+)Niveau d’évaluation(.+)Développeur (.+)Centres d’évaluation(.+)Accords de reconnaissance applicables'), + (HEADER_TYPE.HEADER_FULL, + 'Référence du rapport de certification(.+)Nom du produit(.+)Version du produit(.+)Conformité àun profil de protection(.+)Critères d’évaluation et version(.+)Niveau d’évaluation(.+)Développeur (.+)Centre d’évaluation(.+)Accords de reconnaissance applicables'), + (HEADER_TYPE.HEADER_FULL, + 'Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.+)Conformité aux profils de protection(.+)Critères d’évaluation et version(.+)Niveau d’évaluation(.+)Développeur\(s\)(.+)Centre d’évaluation(.+)Accords de reconnaissance applicables'), + (HEADER_TYPE.HEADER_FULL, + 'Référence du rapport de certification(.+)Nom du produit(.+)Versions du produit(.+)Conformité àun profil de protection(.+)Critères d’évaluation et version(.+)Niveau d’évaluation(.+)Développeur (.+)Centre d’évaluation(.+)Accords de reconnaissance applicables'), + (HEADER_TYPE.HEADER_FULL, + 'Référence du rapport de certification(.+)Nom du produit(.+)Référence du produit(.+)Conformité àun profil de protection(.+)Critères d’évaluation et version(.+)Niveau d’évaluation(.+)Développeurs(.+)Centre d’évaluation(.+)Accords de reconnaissance applicables'), + (HEADER_TYPE.HEADER_FULL, + 'Certification report reference(.+)Product name(.+)Product reference(.+)Protection profile conformity(.+)Evaluation criteria and version(.+)Evaluation level(.+)Developer (.+)Evaluation facility(.+)Recognition arrangements'), + (HEADER_TYPE.HEADER_FULL, + 'Certification report reference(.+)Product name(.+)Product reference(.+)Protection profile conformity(.+)Evaluation criteria and version(.+)Evaluation level(.+)Developer (.+)Evaluation facility(.+)Mutual Recognition Agreements'), + (HEADER_TYPE.HEADER_FULL, + 'Certification report reference(.+)Product name(.+)Product reference(.+)Protection profile conformity(.+)Evaluation criteria and version(.+)Evaluation level(.+)Developers(.+)Evaluation facility(.+)Recognition arrangements'), + (HEADER_TYPE.HEADER_FULL, + 'Certification report reference(.+)Product name(.+)Product reference(.+)Protection profile conformity(.+)Evaluation criteria and version(.+)Evaluation level(.+)Developer\(s\)(.+)Evaluation facility(.+)Recognition arrangements'), + (HEADER_TYPE.HEADER_FULL, + 'Certification report reference(.+)Products names(.+)Products references(.+)protection profile conformity(.+)Evaluation criteria and version(.+)Evaluation level(.+)Developers(.+)Evaluation facility(.+)Recognition arrangements'), + (HEADER_TYPE.HEADER_FULL, + 'Certification report reference(.+)Product name \(reference / version\)(.+)TOE name \(reference / version\)(.+)Protection profile conformity(.+)Evaluation criteria and version(.+)Evaluation level(.+)Developers(.+)Evaluation facility(.+)Recognition arrangements'), + (HEADER_TYPE.HEADER_FULL, + 'Certification report reference(.+)TOE name(.+)Product\'s reference/ version(.+)TOE\'s reference/ version(.+)Conformité à un profil de protection(.+)Evaluation criteria and version(.+)Evaluation level(.+)Developer (.+)Evaluation facility(.+)Recognition arrangements'), + + # corrupted text (duplicities) + (HEADER_TYPE.HEADER_DUPLICITIES, + 'Référencce du rapport de d certification n(.+)Nom du p produit(.+)Référencce/version du produit(.+)Conformiité àun profil de d protection(.+)Critères d d’évaluation ett version(.+)Niveau d’’évaluation(.+)Développ peurs(.+)Centre d’’évaluation(.+)Accords d de reconnaisssance applicab bles'), + + # rules without product version + (HEADER_TYPE.HEADER_MISSING_CERT_ITEM_VERSION, + 'Référence du rapport de certification(.+)Nom et version du produit(.+)Conformité à un profil de protection(.+)Critères d\'évaluation et version(.+)Niveau d\'évaluation(.+)Développeurs(.+)Centre d\'évaluation(.+)Accords de reconnaissance applicables'), + (HEADER_TYPE.HEADER_MISSING_CERT_ITEM_VERSION, + 'Référence du rapport de certification(.+)Nom et version du produit(.+)Conformité à un profil de protection(.+)Critères d\'évaluation et version(.+)Niveau d\'évaluation(.+)Développeur (.+)Centre d\'évaluation(.+)Accords de reconnaissance applicables'), + (HEADER_TYPE.HEADER_MISSING_CERT_ITEM_VERSION, + 'Référence du rapport de certification(.+)Nom du produit(.+)Conformité à un profil de protection(.+)Critères d\'évaluation et version(.+)Niveau d\'évaluation(.+)Développeurs(.+)Centre d\'évaluation(.+)Accords de reconnaissance applicables'), + + # rules without protection profile + (HEADER_TYPE.HEADER_MISSING_PROTECTION_PROFILES, + 'Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.+)Critères d\'évaluation et version(.+)Niveau d\'évaluation(.+)Développeurs(.+)Centre d\'évaluation(.+)Accords de reconnaissance applicables'), + ] + + # statistics about rules success rate + num_rules_hits = {} + for rule in rules_certificate_preface: + num_rules_hits[rule[1]] = 0 + + items_found = {} + whole_text, whole_text_with_newlines, was_unicode_decode_error = extract_certificates.load_cert_file(filepath) + + # for ANSII and DCSSI certificates, front page starts only on third page after 2 newpage signs + pos = whole_text.find('') + if pos != -1: + pos = whole_text.find('', pos) + if pos != -1: + whole_text = whole_text[pos:] + + no_match_yet = True + other_rule_already_match = False + rule_index = -1 + for rule in rules_certificate_preface: + rule_index += 1 + rule_and_sep = rule[1] + REGEXEC_SEP + + for m in re.finditer(rule_and_sep, whole_text): + if no_match_yet: + items_found[constants.TAG_HEADER_MATCH_RULES] = [] + no_match_yet = False + + # insert rule if at least one match for it was found + if rule not in items_found[constants.TAG_HEADER_MATCH_RULES]: + items_found[constants.TAG_HEADER_MATCH_RULES].append(rule[1]) + + if not other_rule_already_match: + other_rule_already_match = True + else: + logger.warning(f'WARNING: multiple rules are matching same certification document: {filepath}') + + num_rules_hits[rule[1]] += 1 # add hit to this rule + match_groups = m.groups() + index_next_item = 0 + items_found[constants.TAG_CERT_ID] = extract_certificates.normalize_match_string(match_groups[index_next_item]) + index_next_item += 1 + + items_found[constants.TAG_CERT_ITEM] = extract_certificates.normalize_match_string(match_groups[index_next_item]) + index_next_item += 1 + + if rule[0] == HEADER_TYPE.HEADER_MISSING_CERT_ITEM_VERSION: + items_found[constants.TAG_CERT_ITEM_VERSION] = '' + else: + items_found[constants.TAG_CERT_ITEM_VERSION] = extract_certificates.normalize_match_string(match_groups[index_next_item]) + index_next_item += 1 + + if rule[0] == HEADER_TYPE.HEADER_MISSING_PROTECTION_PROFILES: + items_found[constants.TAG_REFERENCED_PROTECTION_PROFILES] = '' + else: + items_found[constants.TAG_REFERENCED_PROTECTION_PROFILES] = extract_certificates.normalize_match_string(match_groups[index_next_item]) + index_next_item += 1 + + items_found[constants.TAG_CC_VERSION] = extract_certificates.normalize_match_string(match_groups[index_next_item]) + index_next_item += 1 + + items_found[constants.TAG_CC_SECURITY_LEVEL] = extract_certificates.normalize_match_string(match_groups[index_next_item]) + index_next_item += 1 + + items_found[constants.TAG_DEVELOPER] = extract_certificates.normalize_match_string(match_groups[index_next_item]) + index_next_item += 1 + + items_found[constants.TAG_CERT_LAB] = extract_certificates.normalize_match_string(match_groups[index_next_item]) + index_next_item += 1 + + # if True: + # print('# hits for rule') + # sorted_rules = sorted(num_rules_hits.items(), + # key=operator.itemgetter(1), reverse=True) + # used_rules = [] + # for rule in sorted_rules: + # print('{:4d} : {}'.format(rule[1], rule[0])) + # if rule[1] > 0: + # used_rules.append(rule[0]) + + return items_found + +# TODO: Please refactor me. I need it so badlyyyyyy!!! +def search_only_headers_bsi(filepath: Path): + LINE_SEPARATOR_STRICT = ' ' + NUM_LINES_TO_INVESTIGATE = 15 + rules_certificate_preface = [ + '(BSI-DSZ-CC-.+?) (?:for|For) (.+?) from (.*)', + '(BSI-DSZ-CC-.+?) zu (.+?) der (.*)', + ] + + items_found = {} + no_match_yet = True + # + # Process front page with info: cert_id, certified_item and developer + # + whole_text, whole_text_with_newlines, was_unicode_decode_error = extract_certificates.load_cert_file(filepath, NUM_LINES_TO_INVESTIGATE, LINE_SEPARATOR_STRICT) + + for rule in rules_certificate_preface: + rule_and_sep = rule + REGEXEC_SEP + + for m in re.finditer(rule_and_sep, whole_text): + if no_match_yet: + items_found[constants.TAG_HEADER_MATCH_RULES] = [] + no_match_yet = False + + # insert rule if at least one match for it was found + if rule not in items_found[constants.TAG_HEADER_MATCH_RULES]: + items_found[constants.TAG_HEADER_MATCH_RULES].append(rule) + + match_groups = m.groups() + cert_id = match_groups[0] + certified_item = match_groups[1] + developer = match_groups[2] + + FROM_KEYWORD_LIST = [' from ', ' der '] + for from_keyword in FROM_KEYWORD_LIST: + from_keyword_len = len(from_keyword) + if certified_item.find(from_keyword) != -1: + logger.warning(f'string {from_keyword} detected in certified item - shall not be here, fixing...') + certified_item_first = certified_item[:certified_item.find(from_keyword)] + developer = certified_item[certified_item.find(from_keyword) + from_keyword_len:] + certified_item = certified_item_first + continue + + end_pos = developer.find('\f-') + if end_pos == -1: + end_pos = developer.find('\fBSI') + if end_pos == -1: + end_pos = developer.find('Bundesamt') + if end_pos != -1: + developer = developer[:end_pos] + + items_found[constants.TAG_CERT_ID] = extract_certificates.normalize_match_string(cert_id) + items_found[constants.TAG_CERT_ITEM] = extract_certificates.normalize_match_string(certified_item) + items_found[constants.TAG_DEVELOPER] = extract_certificates.normalize_match_string(developer) + items_found[constants.TAG_CERT_LAB] = 'BSI' + + # Process page with more detailed certificate info + # PP Conformance, Functionality, Assurance + rules_certificate_third = ['PP Conformance: (.+)Functionality: (.+)Assurance: (.+)The IT Product identified'] + + whole_text, whole_text_with_newlines, was_unicode_decode_error = extract_certificates.load_cert_file(filepath) + + for rule in rules_certificate_third: + rule_and_sep = rule + REGEXEC_SEP + + for m in re.finditer(rule_and_sep, whole_text): + # check if previous rules had at least one match + if not constants.TAG_CERT_ID in items_found.keys(): + logger.error('ERROR: front page not found for file: {}'.format(filepath)) + + match_groups = m.groups() + ref_protection_profiles = match_groups[0] + cc_version = match_groups[1] + cc_security_level = match_groups[2] + + items_found[constants.TAG_REFERENCED_PROTECTION_PROFILES] = extract_certificates.normalize_match_string(ref_protection_profiles) + items_found[constants.TAG_CC_VERSION] = extract_certificates.normalize_match_string(cc_version) + items_found[constants.TAG_CC_SECURITY_LEVEL] = extract_certificates.normalize_match_string(cc_security_level) + + # print('\n*** Certificates without detected preface:') + # for file_name in files_without_match: + # print('No hits for {}'.format(file_name)) + # print('Total no hits files: {}'.format(len(files_without_match))) + # print('\n**********************************') + + return items_found + +def extract_keywords(filepath: Path) -> Dict[str, str]: + return extract_certificates.parse_cert_file(filepath, cc_search_rules, -1, extract_certificates.LINE_SEPARATOR)[0] \ No newline at end of file diff --git a/test/data/test_cc_oop/cc_products_active.csv b/test/data/test_cc_oop/cc_products_active.csv index 2561a372..5b41d581 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://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,,,, +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,,,, diff --git a/test/data/test_cc_oop/cc_products_active.html b/test/data/test_cc_oop/cc_products_active.html index 463fc51a..6c44e3c0 100644 --- a/test/data/test_cc_oop/cc_products_active.html +++ b/test/data/test_cc_oop/cc_products_active.html @@ -149,8 +149,7 @@ - + @@ -159,8 +158,7 @@
@@ -197,19 +195,23 @@ ICCC