From 9357918d2f44a6754e7f21a17ea038ca8452e522 Mon Sep 17 00:00:00 2001 From: J08nY Date: Fri, 16 Oct 2020 00:14:27 +0200 Subject: Add basic CLI options (directory and processing step). --- sec_certs/extract_certificates.py | 41 +++++++-------- sec_certs/process_certificates.py | 103 +++++++++++++++----------------------- setup.py | 9 ++-- 3 files changed, 67 insertions(+), 86 deletions(-) diff --git a/sec_certs/extract_certificates.py b/sec_certs/extract_certificates.py index ed09e41a..cc0bc8af 100644 --- a/sec_certs/extract_certificates.py +++ b/sec_certs/extract_certificates.py @@ -6,6 +6,7 @@ import operator import string from enum import Enum +from pathlib import Path import matplotlib.pyplot as plt from PyPDF2 import PdfFileReader from graphviz import Digraph @@ -359,7 +360,7 @@ def print_found_properties(items_found_all): print_specified_property_sorted(TAG_CERT_LAB, items_found_all) -def search_only_headers_bsi(walk_dir): +def search_only_headers_bsi(walk_dir: Path): print('BSI HEADER SEARCH') LINE_SEPARATOR_STRICT = ' ' NUM_LINES_TO_INVESTIGATE = 15 @@ -482,7 +483,7 @@ def search_only_headers_bsi(walk_dir): return items_found_all, files_without_match -def search_only_headers_anssi(walk_dir): +def search_only_headers_anssi(walk_dir: Path): class HEADER_TYPE(Enum): HEADER_FULL = 1 HEADER_MISSING_CERT_ITEM_VERSION = 2 @@ -709,7 +710,7 @@ def search_only_headers_anssi(walk_dir): return items_found_all, files_without_match -def extract_certificates_frontpage(walk_dir, write_output_file=True): +def extract_certificates_frontpage(walk_dir: Path, write_output_file=True): anssi_items_found, anssi_files_without_match = search_only_headers_anssi( walk_dir) bsi_items_found, bsi_files_without_match = search_only_headers_bsi( @@ -733,7 +734,7 @@ def extract_certificates_frontpage(walk_dir, write_output_file=True): return items_found_all -def search_pp_only_headers(walk_dir): +def search_pp_only_headers(walk_dir: Path): # LINE_SEPARATOR_STRICT = ' ' # NUM_LINES_TO_INVESTIGATE = 15 # rules_certificate_preface = [ @@ -1090,7 +1091,7 @@ def search_pp_only_headers(walk_dir): return items_found_all, files_without_match -def extract_protectionprofiles_frontpage(walk_dir, write_output_file=True): +def extract_protectionprofiles_frontpage(walk_dir: Path, write_output_file=True): pp_items_found, pp_files_without_match = search_pp_only_headers(walk_dir) print('*** Files without detected protection profiles header') @@ -1107,7 +1108,7 @@ def extract_protectionprofiles_frontpage(walk_dir, write_output_file=True): return pp_items_found -def extract_certificates_keywords(walk_dir, fragments_dir, file_prefix, write_output_file=True, +def extract_certificates_keywords(walk_dir: Path, fragments_dir: Path, file_prefix, write_output_file=True, should_censure_right_away=False, fips_items=None): # ensure existence of fragments folder if not os.path.exists(fragments_dir): @@ -1183,7 +1184,7 @@ def extract_certificates_keywords(walk_dir, fragments_dir, file_prefix, write_ou return all_items_found -def extract_certificates_pdfmeta(walk_dir, file_prefix, write_output_file=True): +def extract_certificates_pdfmeta(walk_dir: Path, file_prefix, write_output_file=True): all_items_found = {} counter = 0 for file_name in search_files(walk_dir): @@ -1825,8 +1826,8 @@ def generate_download_script(file_name, certs_dir, targets_dir, base_url, downlo PDF2TEXT_CONVERT, cert[3])) -def extract_certificates_html(base_dir, write_output_file=True): - file_name = '{}cc_products_active.html'.format(base_dir) +def extract_certificates_html(base_dir: Path, write_output_file: bool = True): + file_name = base_dir / 'cc_products_active.html' items_found_all_active, download_files_certs, download_files_updates = extract_certificates_metadata_html( file_name) for item in items_found_all_active.keys(): @@ -1842,7 +1843,7 @@ def extract_certificates_html(base_dir, write_output_file=True): generate_download_script('download_active_updates.bat', 'certs', 'targets', CC_WEB_URL, download_files_updates) - file_name = '{}cc_products_archived.html'.format(base_dir) + file_name = base_dir / 'cc_products_archived.html' items_found_all_archived, download_files_certs, download_files_updates = extract_certificates_metadata_html( file_name) for item in items_found_all_archived.keys(): @@ -1868,13 +1869,13 @@ def extract_certificates_html(base_dir, write_output_file=True): return items_found_all -def extract_certificates_csv(base_dir, write_output_file=True): - file_name = '{}cc_products_active.csv'.format(base_dir) +def extract_certificates_csv(base_dir: Path, write_output_file: bool = True): + file_name = base_dir / 'cc_products_active.csv' items_found_all_active = extract_certificates_metadata_csv(file_name) for item in items_found_all_active.keys(): items_found_all_active[item]['csv_scan']['cert_status'] = 'active' - file_name = '{}cc_products_archived.csv'.format(base_dir) + file_name = base_dir / 'cc_products_archived.csv' items_found_all_archived = extract_certificates_metadata_csv(file_name) for item in items_found_all_archived.keys(): items_found_all_archived[item]['csv_scan']['cert_status'] = 'archived' @@ -1889,8 +1890,8 @@ def extract_certificates_csv(base_dir, write_output_file=True): return items_found_all -def extract_protectionprofiles_csv(base_dir, write_output_file=True): - file_name = '{}cc_pp_active.csv'.format(base_dir) +def extract_protectionprofiles_csv(base_dir: Path, write_output_file: bool = True): + file_name = base_dir / 'cc_pp_active.csv' items_found_all_active, download_files_pp, download_files_pp_updates = extract_pp_metadata_csv( file_name) for item in items_found_all_active.keys(): @@ -1901,7 +1902,7 @@ def extract_protectionprofiles_csv(base_dir, write_output_file=True): generate_download_script('download_active_pp_updates.bat', 'pp_updates', '', CC_WEB_URL, download_files_pp_updates) - file_name = '{}cc_pp_archived.csv'.format(base_dir) + file_name = base_dir / 'cc_pp_archived.csv' items_found_all_archived, download_files_pp, download_files_pp_updates = extract_pp_metadata_csv( file_name) for item in items_found_all_archived.keys(): @@ -2366,8 +2367,8 @@ def process_certificates_data(all_cert_items, all_pp_items): return all_cert_items -def generate_basic_download_script(): - with open('download_cc_web.bat', 'w', errors=FILE_ERRORS_STRATEGY) as file: +def generate_basic_download_script(web_dir: Path): + with open(web_dir / 'download_cc_web.bat', 'w', errors=FILE_ERRORS_STRATEGY) as file: file.write( 'curl \"https://www.commoncriteriaportal.org/products/\" -o cc_products_active.html\n') file.write( @@ -2394,7 +2395,7 @@ def generate_basic_download_script(): 'curl \"https://www.commoncriteriaportal.org/pps/pps-archived.csv\" -o cc_pp_archived.csv\n\n') -def generate_failed_download_script(base_dir): +def generate_failed_download_script(base_dir: Path): # obtain list of all downloaded pdf files and their size # check for pdf files with too small length # generate download script again (single one) @@ -2407,7 +2408,7 @@ def generate_failed_download_script(base_dir): MIN_CORRECT_CERT_SIZE = 5000 download_again = [] for sub_folder in sub_folders: - target_dir = os.path.join(base_dir, sub_folder) + target_dir = base_dir / sub_folder # obtain list of all downloaded pdf files and their size files = search_files(target_dir) for file_name in files: diff --git a/sec_certs/process_certificates.py b/sec_certs/process_certificates.py index 3612eafe..41f44247 100755 --- a/sec_certs/process_certificates.py +++ b/sec_certs/process_certificates.py @@ -1,6 +1,9 @@ #!/usr/bin/env python3 import os import json +from pathlib import Path + +import click from extract_certificates import * from analyze_certificates import * @@ -101,69 +104,45 @@ def sanitize_all_strings(data): if isinstance(data, dict): for k, v in data.items(): - if isinstance(v, dict) or isinstance(v, list) or isinstance(v, tuple): + if isinstance(v, (dict, list, tuple)): sanitize_all_strings(v) elif isinstance(v, str): sanitized = ''.join(filter(lambda x: x in printable, v)) data[k] = ''.join(filter(lambda x: x in printable, v)) - if isinstance(data, list) or isinstance(data, tuple): + if isinstance(data, (list, tuple)): for v in data: - if isinstance(v, dict) or isinstance(v, list) or isinstance(v, tuple): + if isinstance(v, (dict, list, tuple)): sanitize_all_strings(v) elif isinstance(v, str): sanitized = ''.join(filter(lambda x: x in printable, v)) v = ''.join(filter(lambda x: x in printable, v)) -def main(): - # Paths for certificates downloaded on 20191208 - paths_20191208 = {} - paths_20191208['id'] = '20191208' - paths_20191208['cc_web_files_dir'] = 'c:\\Certs\\cc_certs_20191208\\web\\' - paths_20191208['walk_dir'] = 'c:\\Certs\\cc_certs_20191208\\cc_certs\\' - #paths_20191208['walk_dir'] = 'c:\\Certs\\cc_certs_20191208\\cc_certs_test1\\' - paths_20191208['pp_dir'] = 'c:\\Certs\\cc_certs_20191208\\cc_pp\\' - #paths_20191208['pp_dir'] = 'c:\\Certs\\cc_certs_20191208\\cc_pp_test1\\' - paths_20191208['fragments_dir'] = 'c:\\Certs\\cc_certs_20191208\\cc_certs_txt_fragments\\' - paths_20191208['pp_fragments_dir'] = 'c:\\Certs\\cc_certs_20191208\\cc_pp_txt_fragments\\' - - # Paths for certificates downloaded on 20200225 - paths_20200225 = {} - paths_20200225['id'] = '20200225' - paths_20200225['cc_web_files_dir'] = 'c:\\Certs\\cc_certs_20200225\\web\\' - paths_20200225['walk_dir'] = 'c:\\Certs\\cc_certs_20200225\\cc_certs\\' - paths_20200225['pp_dir'] = 'c:\\Certs\\cc_certs_20200225\\cc_pp\\' - paths_20200225['fragments_dir'] = 'c:\\Certs\\cc_certs_20200225\\cc_certs_txt_fragments\\' - paths_20200225['pp_fragments_dir'] = 'c:\\Certs\\cc_certs_20200225\\cc_pp_txt_fragments\\' - - # Paths for certificates downloaded on 20200904 - paths_20200904 = {} - paths_20200904['id'] = '20200904' - paths_20200904['cc_web_files_dir'] = 'c:\\Certs\\cc_certs_20200904\\web\\' - paths_20200904['cc_pp_web_files_dir'] = 'c:\\Certs\\certs_pp_20201008\\pp_web\\' - paths_20200904['walk_dir'] = 'c:\\Certs\\cc_certs_20200904\\cc_certs\\' - paths_20200904['pp_dir'] = 'c:\\Certs\\certs_pp_20201008\\cc_pp\\' - paths_20200904['fragments_dir'] = 'c:\\Certs\\cc_certs_20200904\\cc_certs_txt_fragments\\' - paths_20200904['pp_fragments_dir'] = 'c:\\Certs\\certs_pp_20201008\\cc_pp_txt_fragments\\' - - # initialize paths based on the profile used - #paths_used = paths_20191208 - #paths_used = paths_20200225 - paths_used = paths_20200904 - #paths_used['id'] = 'temp' # change id for temporary debugging - - cc_web_files_dir = paths_used['cc_web_files_dir'] - walk_dir = paths_used['walk_dir'] - fragments_dir = paths_used['fragments_dir'] - - # results folder includes unique identification of input dataset - results_folder = '{}\\..\\results_{}\\'.format(os.getcwd(), paths_used['id']) - # ensure existence of results folder - if not os.path.exists(results_folder): - os.makedirs(results_folder) - # change current directory to store results into results file - os.chdir(results_folder) +@click.command() +@click.option("-d", "--dir", "directory", type=str, help="The directory to use.", required=True) +@click.option("--fresh", "do_complete_extraction", is_flag=True, help="Whether to extract from a fresh state.") +@click.option("--do-download", "do_download_certs", is_flag=True, help="Whether to download certificate pages.") +@click.option("--do-extraction", "do_extraction", is_flag=True, help="Whether to extract information from the certs.") +@click.option("--do-pairing", "do_pairing", is_flag=True, help="Whether to pair PP stuff.") +@click.option("--do-processing", "do_processing", is_flag=True, help="Whether to process certificates.") +@click.option("--do-anaysis", "do_analysis", is_flag=True, help="Whether to analyse certificates.") +def main(directory, do_complete_extraction, do_download_certs, do_extraction, do_pairing, do_processing, do_analysis): + directory = Path(directory) + + web_dir = directory / "web" + walk_dir = directory / "certs" + pp_dir = directory / "pp" + fragments_dir = directory / "cert_fragments" + pp_fragments_dir = directory / "pp_fragments" + results_folder = directory / "results" + + web_dir.mkdir(parents=True, exist_ok=True) + walk_dir.mkdir(parents=True, exist_ok=True) + pp_dir.mkdir(parents=True, exist_ok=True) + fragments_dir.mkdir(parents=True, exist_ok=True) + pp_fragments_dir.mkdir(parents=True, exist_ok=True) + results_folder.mkdir(parents=True, exist_ok=True) # 1. generate_basic_download_script # 2. run and download basic cc files from webpage (no certs yet) @@ -171,16 +150,16 @@ def main(): # # Start processing # - generate_basic_download_script() + generate_basic_download_script(web_dir) generate_failed_download_script(walk_dir) - do_complete_extraction = True - do_download_certs = True - do_extraction = True - do_pairing = True - do_processing = True - do_analysis = True - do_analysis_filtered = True + #do_complete_extraction = True + #do_download_certs = True + #do_extraction = True + #do_pairing = True + #do_processing = True + #do_analysis = True + do_analysis_filtered = False if do_complete_extraction: # analyze all files from scratch, set 'previous' state to empty dict @@ -198,7 +177,7 @@ def main(): if do_download_certs: # extract_certificates_html() will generate download scripts for cert documents # NOTE: download scripts must be run manually now - current_html = extract_certificates_html(cc_web_files_dir, False) + current_html = extract_certificates_html(web_dir, False) # NOTE: Code below is preparation for differetian download of only new certificates # - unfinished now @@ -227,8 +206,8 @@ def main(): # print('*** New items detected: {}'.format(len(new_items))) if do_extraction: - all_csv = extract_certificates_csv(cc_web_files_dir, False) - all_html = extract_certificates_html(cc_web_files_dir, False) + all_csv = extract_certificates_csv(web_dir, False) + all_html = extract_certificates_html(web_dir, False) all_front = extract_certificates_frontpage(walk_dir, False) all_keywords = extract_certificates_keywords(walk_dir, fragments_dir, 'certificate', False) all_pdf_meta = extract_certificates_pdfmeta(walk_dir, 'certificate', False) diff --git a/setup.py b/setup.py index 4f59a739..9d1469ba 100644 --- a/setup.py +++ b/setup.py @@ -1,12 +1,12 @@ #!/usr/bin/env python3 -from setuptools import setup +from setuptools import setup, find_packages setup( name='sec-certs', author='Petr Svenda', - author_email='', + author_email='svenda@fi.muni.cz', version='0.0.0', - packages=['sec_certs'], + packages=find_packages(), license='MIT', description="Tool for analysis of security certificates", long_description=open("README.md").read(), @@ -27,6 +27,7 @@ setup( "numpy", "tabulate", "tabula-py", - "pikepdf" + "pikepdf", + "Click" ] ) -- cgit v1.3.1 From edaa68a3bdbe40027cb1e809d7b5790c480d52cd Mon Sep 17 00:00:00 2001 From: J08nY Date: Sun, 18 Oct 2020 17:42:56 +0200 Subject: Replace json.dumps with json.dump to file. --- sec_certs/extract_certificates.py | 38 ++++++++++++++------------------------ sec_certs/fips_certificates.py | 9 ++++----- sec_certs/process_certificates.py | 16 ++++++++-------- 3 files changed, 26 insertions(+), 37 deletions(-) diff --git a/sec_certs/extract_certificates.py b/sec_certs/extract_certificates.py index cc0bc8af..ae5bb2cc 100644 --- a/sec_certs/extract_certificates.py +++ b/sec_certs/extract_certificates.py @@ -472,7 +472,7 @@ def search_only_headers_bsi(walk_dir: Path): print_found_properties(items_found_all) with open("certificate_data_bsiheader.json", "w", errors=FILE_ERRORS_STRATEGY) as write_file: - write_file.write(json.dumps(items_found_all, indent=4, sort_keys=True)) + json.dump(items_found_all, write_file, indent=4, sort_keys=True) print('\n*** Certificates without detected preface:') for file_name in files_without_match: @@ -689,7 +689,7 @@ def search_only_headers_anssi(walk_dir: Path): # store results into file with fixed name and also with time appendix with open("certificate_data_anssiheader.json", "w", errors=FILE_ERRORS_STRATEGY) as write_file: - write_file.write(json.dumps(items_found_all, indent=4, sort_keys=True)) + json.dump(items_found_all, write_file, indent=4, sort_keys=True) print('\n*** Certificates without detected preface:') for file_name in files_without_match: @@ -728,8 +728,7 @@ def extract_certificates_frontpage(walk_dir: Path, write_output_file=True): if write_output_file: with open("certificate_data_frontpage_all.json", "w", errors=FILE_ERRORS_STRATEGY) as write_file: - write_file.write(json.dumps( - items_found_all, indent=4, sort_keys=True)) + json.dump(items_found_all, write_file, indent=4, sort_keys=True) return items_found_all @@ -1080,7 +1079,7 @@ def search_pp_only_headers(walk_dir: Path): print_found_properties(items_found_all) with open("pp_data_header.json", "w", errors=FILE_ERRORS_STRATEGY) as write_file: - write_file.write(json.dumps(items_found_all, indent=4, sort_keys=True)) + json.dump(items_found_all, write_file, indent=4, sort_keys=True) print('\n*** Protection profiles without detected header:') for file_name in files_without_match: @@ -1102,8 +1101,7 @@ def extract_protectionprofiles_frontpage(walk_dir: Path, write_output_file=True) # store results into file with fixed name and also with time appendix if write_output_file: with open("pp_data_frontpage_all.json", "w", errors=FILE_ERRORS_STRATEGY) as write_file: - write_file.write(json.dumps( - pp_items_found, indent=4, sort_keys=True)) + json.dump(pp_items_found, write_file, indent=4, sort_keys=True) return pp_items_found @@ -1139,15 +1137,14 @@ def extract_certificates_keywords(walk_dir: Path, fragments_dir: Path, file_pref # save report text with highlighted/replaced matches into \\fragments\\ directory base_path = file_name[:file_name.rfind(os.sep)] file_name_short = file_name[file_name.rfind(os.sep) + 1:] - target_file = '{}{}{}'.format(fragments_dir, os.sep, file_name_short) + target_file = fragments_dir / file_name_short save_modified_cert_file( target_file, modified_cert_file[0], modified_cert_file[1]) # store results into file with fixed name if write_output_file: with open("{}_data_keywords_all.json".format(file_prefix), "w", errors=FILE_ERRORS_STRATEGY) as write_file: - write_file.write(json.dumps( - all_items_found, indent=4, sort_keys=True)) + json.dump(all_items_found, write_file, indent=4, sort_keys=True) # print('\nTotal matches found in separate files:') # print_total_matches_in_files(all_items_found_count) @@ -1227,15 +1224,13 @@ def extract_certificates_pdfmeta(walk_dir: Path, file_prefix, write_output_file= # store results into file with fixed name with open("{}_data_pdfmeta_{}.json".format(file_prefix, counter), "w", errors=FILE_ERRORS_STRATEGY) as write_file: - write_file.write(json.dumps( - all_items_found, indent=4, sort_keys=True)) + json.dump(all_items_found, write_file, indent=4, sort_keys=True) counter += 1 # store allresults into file with fixed name if write_output_file: with open("{}_data_pdfmeta_all.json".format(file_prefix), "w", errors=FILE_ERRORS_STRATEGY) as write_file: - write_file.write(json.dumps( - all_items_found, indent=4, sort_keys=True)) + json.dump(all_items_found, write_file, indent=4, sort_keys=True) return all_items_found @@ -1835,8 +1830,7 @@ def extract_certificates_html(base_dir: Path, write_output_file: bool = True): if write_output_file: with open("certificate_data_html_active.json", "w", errors=FILE_ERRORS_STRATEGY) as write_file: - write_file.write(json.dumps( - items_found_all_active, indent=4, sort_keys=True)) + json.dump(items_found_all_active, write_file, indent=4, sort_keys=True) generate_download_script('download_active_certs.bat', 'certs', 'targets', CC_WEB_URL, download_files_certs) @@ -1851,8 +1845,7 @@ def extract_certificates_html(base_dir: Path, write_output_file: bool = True): if write_output_file: with open("certificate_data_html_archived.json", "w", errors=FILE_ERRORS_STRATEGY) as write_file: - write_file.write(json.dumps( - items_found_all_archived, indent=4, sort_keys=True)) + json.dump(items_found_all_archived, write_file, indent=4, sort_keys=True) generate_download_script('download_archived_certs.bat', 'certs', 'targets', CC_WEB_URL, download_files_certs) @@ -1863,8 +1856,7 @@ def extract_certificates_html(base_dir: Path, write_output_file: bool = True): if write_output_file: with open("certificate_data_html_all.json", "w", errors=FILE_ERRORS_STRATEGY) as write_file: - write_file.write(json.dumps( - items_found_all, indent=4, sort_keys=True)) + json.dump(items_found_all, write_file, indent=4, sort_keys=True) return items_found_all @@ -1884,8 +1876,7 @@ def extract_certificates_csv(base_dir: Path, write_output_file: bool = True): if write_output_file: with open("certificate_data_csv_all.json", "w", errors=FILE_ERRORS_STRATEGY) as write_file: - write_file.write(json.dumps( - items_found_all, indent=4, sort_keys=True)) + json.dump(items_found_all, write_file, ndent=4, sort_keys=True) return items_found_all @@ -1917,8 +1908,7 @@ def extract_protectionprofiles_csv(base_dir: Path, write_output_file: bool = Tru if write_output_file: with open("pp_data_csv_all.json", "w", errors=FILE_ERRORS_STRATEGY) as write_file: - write_file.write(json.dumps( - items_found_all, indent=4, sort_keys=True)) + json.dump(items_found_all, write_file, indent=4, sort_keys=True) return items_found_all diff --git a/sec_certs/fips_certificates.py b/sec_certs/fips_certificates.py index 3f473ecd..1647f255 100755 --- a/sec_certs/fips_certificates.py +++ b/sec_certs/fips_certificates.py @@ -149,8 +149,7 @@ def fips_search_html(base_dir, output_file, dump_to_file=False): if dump_to_file: with open(output_file, 'w', errors=FILE_ERRORS_STRATEGY) as write_file: - write_file.write(json.dumps( - all_found_items, indent=4, sort_keys=True)) + json.dump(all_found_items, write_file, indent=4, sort_keys=True) return all_found_items @@ -412,7 +411,7 @@ def main(): os.path.join(FIPS_BASE_DIR, 'fragments'), 'fips', fips_items=fips_items, should_censure_right_away=True, write_output_file=True) with open(FIPS_RESULTS_DIR + 'fips_data_keywords_all.json', 'w') as f: - f.write(json.dumps(items, indent=4, sort_keys=True)) + json.dump(items, f, indent=4, sort_keys=True) break print("EXTRACTION DONE") @@ -423,7 +422,7 @@ def main(): print("NOT DECODED:", not_decoded) with open(FIPS_RESULTS_DIR + 'broken_files.json', 'w') as f: - f.write(json.dumps(not_decoded)) + json.dump(not_decoded, f) print("REMOVING ALGORITHMS") remove_algorithms_from_extracted_data(items, html) @@ -431,7 +430,7 @@ def main(): print("VALIDATING RESULTS") validate_results(items, html) with open(FIPS_RESULTS_DIR + 'fips_html_all.json', 'w') as f: - f.write(json.dumps(html, indent=4, sort_keys=True)) + json.dump(html, f, indent=4, sort_keys=True) print("PLOTTING GRAPH") get_dot_graph(html, 'output') diff --git a/sec_certs/process_certificates.py b/sec_certs/process_certificates.py index 41f44247..ee5f35bb 100755 --- a/sec_certs/process_certificates.py +++ b/sec_certs/process_certificates.py @@ -214,15 +214,15 @@ def main(directory, do_complete_extraction, do_download_certs, do_extraction, do # save joined results with open("certificate_data_csv_all.json", "w") as write_file: - write_file.write(json.dumps(all_csv, indent=4, sort_keys=True)) + json.dump(all_csv, write_file, indent=4, sort_keys=True) with open("certificate_data_html_all.json", "w") as write_file: - write_file.write(json.dumps(all_html, indent=4, sort_keys=True)) + json.dump(all_html, write_file, indent=4, sort_keys=True) with open("certificate_data_frontpage_all.json", "w") as write_file: - write_file.write(json.dumps(all_front, indent=4, sort_keys=True)) + json.dump(all_front, write_file, indent=4, sort_keys=True) with open("certificate_data_keywords_all.json", "w") as write_file: - write_file.write(json.dumps(all_keywords, indent=4, sort_keys=True)) + json.dump(all_keywords, write_file, indent=4, sort_keys=True) with open("certificate_data_pdfmeta_all.json", "w") as write_file: - write_file.write(json.dumps(all_pdf_meta, indent=4, sort_keys=True)) + json.dump(all_pdf_meta, write_file, indent=4, sort_keys=True) # if do_extraction_pp: # all_pp_csv = extract_protectionprofiles_csv(cc_pp_web_files_dir) @@ -266,7 +266,7 @@ def main(directory, do_complete_extraction, do_download_certs, do_extraction, do # write collated result with open("certificate_data_complete.json", "w") as write_file: - write_file.write(json.dumps(all_cert_items, indent=4, sort_keys=True)) + json.dump(all_cert_items, write_file, indent=4, sort_keys=True) if do_processing: # load information about protection profiles as extracted by sec-certs-pp tool @@ -279,7 +279,7 @@ def main(directory, do_complete_extraction, do_download_certs, do_extraction, do all_cert_items = process_certificates_data(all_cert_items, all_pp_items) with open("certificate_data_complete_processed.json", "w") as write_file: - write_file.write(json.dumps(all_cert_items, indent=4, sort_keys=True)) + json.dump(all_cert_items, write_file, indent=4, sort_keys=True) if do_analysis: with open('certificate_data_complete_processed.json') as json_file: @@ -303,7 +303,7 @@ def main(directory, do_complete_extraction, do_download_certs, do_extraction, do do_analysis_everything(all_cert_items, results_folder) with open("certificate_data_complete_processed_analyzed.json", "w") as write_file: - write_file.write(json.dumps(all_cert_items, indent=4, sort_keys=True)) + json.dump(all_cert_items, write_file, indent=4, sort_keys=True) if __name__ == "__main__": -- cgit v1.3.1 From 23c9b3fd4702c48fecd3b541eaf9cb492872f93d Mon Sep 17 00:00:00 2001 From: J08nY Date: Sun, 18 Oct 2020 19:13:49 +0200 Subject: Fixup paths in fips processing. --- sec_certs/extract_certificates.py | 78 +++++----------------------- sec_certs/fips_certificates.py | 86 ++++++++++++++++--------------- sec_certs/process_certificates.py | 104 +++++++++++++++----------------------- setup.py | 7 ++- 4 files changed, 106 insertions(+), 169 deletions(-) diff --git a/sec_certs/extract_certificates.py b/sec_certs/extract_certificates.py index ae5bb2cc..e2d645c1 100644 --- a/sec_certs/extract_certificates.py +++ b/sec_certs/extract_certificates.py @@ -32,8 +32,6 @@ REGEXEC_SEP = '[ ,;\]”)(]' LINE_SEPARATOR = ' ' # LINE_SEPARATOR = '' # if newline is not replaced with space, long string included in matches are found -printable = set(string.printable) - def search_files(folder): for root, dirs, files in os.walk(folder): @@ -101,7 +99,7 @@ def load_cert_html_file(file_name): whole_text = f.read() except UnicodeDecodeError: f.close() - with open(file_name, encoding="utf8", errors=FILE_ERRORS_STRATEGY) as f2: + with open(file_name, "r", encoding="utf8", errors=FILE_ERRORS_STRATEGY) as f2: try: whole_text = f2.read() except UnicodeDecodeError: @@ -124,7 +122,7 @@ def normalize_match_string(match): match = match.rstrip(',') match = match.replace(' ', ' ') # two spaces into one - sanitized = ''.join(filter(lambda x: x in printable, match)) + sanitized = ''.join(filter(str.isprintable, match)) return sanitized @@ -471,9 +469,6 @@ def search_only_headers_bsi(walk_dir: Path): if False: print_found_properties(items_found_all) - with open("certificate_data_bsiheader.json", "w", errors=FILE_ERRORS_STRATEGY) as write_file: - json.dump(items_found_all, write_file, indent=4, sort_keys=True) - print('\n*** Certificates without detected preface:') for file_name in files_without_match: print('No hits for {}'.format(file_name)) @@ -687,10 +682,6 @@ def search_only_headers_anssi(walk_dir: Path): if False: print_found_properties(items_found_all) - # store results into file with fixed name and also with time appendix - with open("certificate_data_anssiheader.json", "w", errors=FILE_ERRORS_STRATEGY) as write_file: - json.dump(items_found_all, write_file, indent=4, sort_keys=True) - print('\n*** Certificates without detected preface:') for file_name in files_without_match: print('No hits for {}'.format(file_name)) @@ -710,7 +701,7 @@ def search_only_headers_anssi(walk_dir: Path): return items_found_all, files_without_match -def extract_certificates_frontpage(walk_dir: Path, write_output_file=True): +def extract_certificates_frontpage(walk_dir: Path): anssi_items_found, anssi_files_without_match = search_only_headers_anssi( walk_dir) bsi_items_found, bsi_files_without_match = search_only_headers_bsi( @@ -726,10 +717,6 @@ def extract_certificates_frontpage(walk_dir: Path, write_output_file=True): items_found_all = {**anssi_items_found, **bsi_items_found} # store results into file with fixed name and also with time appendix - if write_output_file: - with open("certificate_data_frontpage_all.json", "w", errors=FILE_ERRORS_STRATEGY) as write_file: - json.dump(items_found_all, write_file, indent=4, sort_keys=True) - return items_found_all @@ -1078,9 +1065,6 @@ def search_pp_only_headers(walk_dir: Path): if False: print_found_properties(items_found_all) - with open("pp_data_header.json", "w", errors=FILE_ERRORS_STRATEGY) as write_file: - json.dump(items_found_all, write_file, indent=4, sort_keys=True) - print('\n*** Protection profiles without detected header:') for file_name in files_without_match: print('No hits for {}'.format(file_name)) @@ -1090,7 +1074,7 @@ def search_pp_only_headers(walk_dir: Path): return items_found_all, files_without_match -def extract_protectionprofiles_frontpage(walk_dir: Path, write_output_file=True): +def extract_protectionprofiles_frontpage(walk_dir: Path): pp_items_found, pp_files_without_match = search_pp_only_headers(walk_dir) print('*** Files without detected protection profiles header') @@ -1098,16 +1082,10 @@ def extract_protectionprofiles_frontpage(walk_dir: Path, write_output_file=True) print(file_name) print('Total no hits files: {}'.format(len(pp_files_without_match))) - # store results into file with fixed name and also with time appendix - if write_output_file: - with open("pp_data_frontpage_all.json", "w", errors=FILE_ERRORS_STRATEGY) as write_file: - json.dump(pp_items_found, write_file, indent=4, sort_keys=True) - return pp_items_found -def extract_certificates_keywords(walk_dir: Path, fragments_dir: Path, file_prefix, write_output_file=True, - should_censure_right_away=False, fips_items=None): +def extract_certificates_keywords(walk_dir: Path, fragments_dir: Path, file_prefix, should_censure_right_away=False, fips_items=None): # ensure existence of fragments folder if not os.path.exists(fragments_dir): os.makedirs(fragments_dir) @@ -1141,11 +1119,6 @@ def extract_certificates_keywords(walk_dir: Path, fragments_dir: Path, file_pref save_modified_cert_file( target_file, modified_cert_file[0], modified_cert_file[1]) - # store results into file with fixed name - if write_output_file: - with open("{}_data_keywords_all.json".format(file_prefix), "w", errors=FILE_ERRORS_STRATEGY) as write_file: - json.dump(all_items_found, write_file, indent=4, sort_keys=True) - # print('\nTotal matches found in separate files:') # print_total_matches_in_files(all_items_found_count) @@ -1181,7 +1154,7 @@ def extract_certificates_keywords(walk_dir: Path, fragments_dir: Path, file_pref return all_items_found -def extract_certificates_pdfmeta(walk_dir: Path, file_prefix, write_output_file=True): +def extract_certificates_pdfmeta(walk_dir: Path, file_prefix, results_dir: Path): all_items_found = {} counter = 0 for file_name in search_files(walk_dir): @@ -1227,11 +1200,6 @@ def extract_certificates_pdfmeta(walk_dir: Path, file_prefix, write_output_file= json.dump(all_items_found, write_file, indent=4, sort_keys=True) counter += 1 - # store allresults into file with fixed name - if write_output_file: - with open("{}_data_pdfmeta_all.json".format(file_prefix), "w", errors=FILE_ERRORS_STRATEGY) as write_file: - json.dump(all_items_found, write_file, indent=4, sort_keys=True) - return all_items_found @@ -1821,32 +1789,24 @@ def generate_download_script(file_name, certs_dir, targets_dir, base_url, downlo PDF2TEXT_CONVERT, cert[3])) -def extract_certificates_html(base_dir: Path, write_output_file: bool = True): - file_name = base_dir / 'cc_products_active.html' +def extract_certificates_html(web_dir: Path): + file_name = web_dir / 'cc_products_active.html' items_found_all_active, download_files_certs, download_files_updates = extract_certificates_metadata_html( file_name) for item in items_found_all_active.keys(): items_found_all_active[item]['html_scan']['cert_status'] = 'active' - if write_output_file: - with open("certificate_data_html_active.json", "w", errors=FILE_ERRORS_STRATEGY) as write_file: - json.dump(items_found_all_active, write_file, indent=4, sort_keys=True) - generate_download_script('download_active_certs.bat', 'certs', 'targets', CC_WEB_URL, download_files_certs) generate_download_script('download_active_updates.bat', 'certs', 'targets', CC_WEB_URL, download_files_updates) - file_name = base_dir / 'cc_products_archived.html' + file_name = web_dir / 'cc_products_archived.html' items_found_all_archived, download_files_certs, download_files_updates = extract_certificates_metadata_html( file_name) for item in items_found_all_archived.keys(): items_found_all_archived[item]['html_scan']['cert_status'] = 'archived' - if write_output_file: - with open("certificate_data_html_archived.json", "w", errors=FILE_ERRORS_STRATEGY) as write_file: - json.dump(items_found_all_archived, write_file, indent=4, sort_keys=True) - generate_download_script('download_archived_certs.bat', 'certs', 'targets', CC_WEB_URL, download_files_certs) generate_download_script('download_archived_updates.bat', @@ -1854,34 +1814,26 @@ def extract_certificates_html(base_dir: Path, write_output_file: bool = True): items_found_all = {**items_found_all_active, **items_found_all_archived} - if write_output_file: - with open("certificate_data_html_all.json", "w", errors=FILE_ERRORS_STRATEGY) as write_file: - json.dump(items_found_all, write_file, indent=4, sort_keys=True) - return items_found_all -def extract_certificates_csv(base_dir: Path, write_output_file: bool = True): - file_name = base_dir / 'cc_products_active.csv' +def extract_certificates_csv(web_dir: Path, results_dir: Path): + file_name = web_dir / 'cc_products_active.csv' items_found_all_active = extract_certificates_metadata_csv(file_name) for item in items_found_all_active.keys(): items_found_all_active[item]['csv_scan']['cert_status'] = 'active' - file_name = base_dir / 'cc_products_archived.csv' + file_name = web_dir / 'cc_products_archived.csv' items_found_all_archived = extract_certificates_metadata_csv(file_name) for item in items_found_all_archived.keys(): items_found_all_archived[item]['csv_scan']['cert_status'] = 'archived' items_found_all = {**items_found_all_active, **items_found_all_archived} - if write_output_file: - with open("certificate_data_csv_all.json", "w", errors=FILE_ERRORS_STRATEGY) as write_file: - json.dump(items_found_all, write_file, ndent=4, sort_keys=True) - return items_found_all -def extract_protectionprofiles_csv(base_dir: Path, write_output_file: bool = True): +def extract_protectionprofiles_csv(base_dir: Path): file_name = base_dir / 'cc_pp_active.csv' items_found_all_active, download_files_pp, download_files_pp_updates = extract_pp_metadata_csv( file_name) @@ -1906,10 +1858,6 @@ def extract_protectionprofiles_csv(base_dir: Path, write_output_file: bool = Tru items_found_all = {**items_found_all_active, **items_found_all_archived} - if write_output_file: - with open("pp_data_csv_all.json", "w", errors=FILE_ERRORS_STRATEGY) as write_file: - json.dump(items_found_all, write_file, indent=4, sort_keys=True) - return items_found_all diff --git a/sec_certs/fips_certificates.py b/sec_certs/fips_certificates.py index 1647f255..60d2d6ed 100755 --- a/sec_certs/fips_certificates.py +++ b/sec_certs/fips_certificates.py @@ -3,9 +3,12 @@ import json import os import re import time +from pathlib import Path +from typing import Set, Optional from graphviz import Digraph from PyPDF2 import PdfFileReader, utils +import click import pikepdf # from camelot import read_pdf from tabula import read_pdf @@ -17,12 +20,9 @@ from cert_rules import rules_fips_htmls as RE_FIPS_HTMLS FILE_ERRORS_STRATEGY = extract_certificates.FILE_ERRORS_STRATEGY FIPS_BASE_URL = 'https://csrc.nist.gov' FIPS_MODULE_URL = 'https://csrc.nist.gov/projects/cryptographic-module-validation-program/certificate/' -FIPS_RESULTS_DIR = '/home/stan/sec-certs/fips_results/' -FIPS_BASE_DIR = '/home/stan/sec-certs/files/fips/' -SECURITY_POLICIES_DIR = '/home/stan/sec-certs/files/fips/security_policies/' -def extract_filename(file): +def extract_filename(file: str) -> str: return os.path.splitext(os.path.basename(file))[0] @@ -269,11 +269,11 @@ def validate_results(items, html): count = 0 -def parse_list_of_tables(txt): +def parse_list_of_tables(txt: str) -> Set[str]: """ Parses list of tables from function find_tables(), finds ones that mention algorithms :param txt: chunk of text - :return: list of all pages mentioning algorithm table + :return: set of all pages mentioning algorithm table """ rr = re.compile(r"^.+?(?:[Ff]unction|[Aa]lgorithm).+?(?P\d+)$", re.MULTILINE) pages = set() @@ -282,7 +282,7 @@ def parse_list_of_tables(txt): return pages -def extract_page_number(txt): +def extract_page_number(txt: str) -> Optional[str]: """ Parses chunks of text that are supposed to be mentioning table and having a footer :param txt: input chunk @@ -342,40 +342,38 @@ def find_tables(txt, file_name, num_pages): return footers -def repair_pdf_page_count(file): +def repair_pdf_page_count(file: str) -> int: pdf = pikepdf.Pdf.open(file, allow_overwriting_input=True) pdf.save(file) return len(pdf.pages) def extract_certs_from_tables(list_of_files, html_items): - global count - not_decoded = [] - for REDHAT_FILE in list_of_files: - if '.txt' not in REDHAT_FILE: + for cert_file in list_of_files: + if '.txt' not in cert_file: continue - if html_items[extract_filename(REDHAT_FILE[:-8])]['tables_done']: + if html_items[extract_filename(cert_file[:-8])]['tables_done']: continue - with open(REDHAT_FILE, 'r') as f: + with open(cert_file, 'r') as f: try: - pages = repair_pdf_page_count(REDHAT_FILE[:-4]) + pages = repair_pdf_page_count(cert_file[:-4]) except pikepdf._qpdf.PdfError: - not_decoded.append(REDHAT_FILE) + not_decoded.append(cert_file) continue - tables = find_tables(f.read(), REDHAT_FILE, pages) + tables = find_tables(f.read(), cert_file, pages) # If we find any tables with page numbers, we process them if tables: lst = [] - print("~~~~~~~~~~~~~~~", REDHAT_FILE, "~~~~~~~~~~~~~~~~~~~~~~~") + print("~~~~~~~~~~~~~~~", cert_file, "~~~~~~~~~~~~~~~~~~~~~~~") try: - data = read_pdf(REDHAT_FILE[:-4], pages=tables, silent=True) + data = read_pdf(cert_file[:-4], pages=tables, silent=True) except Exception: - not_decoded.append(REDHAT_FILE) + not_decoded.append(cert_file) continue # find columns with cert numbers @@ -387,41 +385,48 @@ def extract_certs_from_tables(list_of_files, html_items): # Parse again if someone picks not so descriptive column names lst += parse_algorithms(df.to_string(index=False)) if lst: - if 'fips_algorithms' not in html_items[extract_filename(REDHAT_FILE[:-8])]: - html_items[extract_filename(REDHAT_FILE[:-8])]['fips_algorithms'] = lst + if 'fips_algorithms' not in html_items[extract_filename(cert_file[:-8])]: + html_items[extract_filename(cert_file[:-8])]['fips_algorithms'] = lst else: - html_items[extract_filename(REDHAT_FILE[:-8])]['fips_algorithms'] += lst + html_items[extract_filename(cert_file[:-8])]['fips_algorithms'] += lst - html_items[extract_filename(REDHAT_FILE[:-8])]['tables_done'] = True + html_items[extract_filename(cert_file[:-8])]['tables_done'] = True return not_decoded -def main(): +@click.command() +@click.argument("directory", required=True, type=str, help="The directory to use.") +def main(directory): + start = time.time() + directory = Path(directory) + results_dir = directory / "results" + policies_dir = directory / "security_policies" + files_to_load = [ - FIPS_RESULTS_DIR + 'fips_data_keywords_all.json', - FIPS_RESULTS_DIR + 'fips_html_all.json' + results_dir / 'fips_data_keywords_all.json', + results_dir / 'fips_html_all.json' ] for file in files_to_load: if not os.path.isfile(file): - fips_items = fips_search_html(os.path.join(FIPS_BASE_DIR, 'html'), - os.path.join(FIPS_RESULTS_DIR,'fips_html_all.json'), True) + fips_items = fips_search_html(directory / 'html', + results_dir / 'fips_html_all.json', True) items = extract_certificates.extract_certificates_keywords( - os.path.join(FIPS_BASE_DIR, 'security_policies'), - os.path.join(FIPS_BASE_DIR, 'fragments'), 'fips', fips_items=fips_items, - should_censure_right_away=True, write_output_file=True) - with open(FIPS_RESULTS_DIR + 'fips_data_keywords_all.json', 'w') as f: + directory / 'security_policies', + directory / 'fragments', 'fips', fips_items=fips_items, + should_censure_right_away=True) + with open(results_dir / 'fips_data_keywords_all.json', 'w') as f: json.dump(items, f, indent=4, sort_keys=True) break print("EXTRACTION DONE") - (items, html) = load_json_files(files_to_load) + items, html = load_json_files(files_to_load) print("FINDING TABLES") - not_decoded = extract_certs_from_tables(extract_certificates.search_files(SECURITY_POLICIES_DIR), html) + not_decoded = extract_certs_from_tables(extract_certificates.search_files(policies_dir), html) print("NOT DECODED:", not_decoded) - with open(FIPS_RESULTS_DIR + 'broken_files.json', 'w') as f: + with open(results_dir / 'broken_files.json', 'w') as f: json.dump(not_decoded, f) print("REMOVING ALGORITHMS") @@ -429,15 +434,14 @@ def main(): print("VALIDATING RESULTS") validate_results(items, html) - with open(FIPS_RESULTS_DIR + 'fips_html_all.json', 'w') as f: + with open(results_dir / 'fips_html_all.json', 'w') as f: json.dump(html, f, indent=4, sort_keys=True) print("PLOTTING GRAPH") get_dot_graph(html, 'output') + end = time.time() + print("TIME:", end - start) + print("COUNT:", count) if __name__ == '__main__': - start = time.time() main() - end = time.time() - print("TIME:", end - start) - print("COUNT:", count) diff --git a/sec_certs/process_certificates.py b/sec_certs/process_certificates.py index ee5f35bb..00be17a0 100755 --- a/sec_certs/process_certificates.py +++ b/sec_certs/process_certificates.py @@ -20,14 +20,14 @@ def do_all_analysis(all_cert_items, filter_label): plot_certid_to_item_graph(['keywords_scan', 'rules_protection_profiles'], all_cert_items, filter_label, 'certid_pp_graph.dot', False) -def do_analysis_everything(all_cert_items, current_dir): +def do_analysis_everything(all_cert_items, current_dir: Path): if not os.path.exists(current_dir): os.makedirs(current_dir) os.chdir(current_dir) do_all_analysis(all_cert_items, '') -def do_analysis_09_01_2019_archival(all_cert_items, current_dir): +def do_analysis_09_01_2019_archival(all_cert_items, current_dir: Path): target_folder = os.path.join(current_dir, 'results_archived01092019_only') if not os.path.exists(target_folder): os.makedirs(target_folder) @@ -37,7 +37,7 @@ def do_analysis_09_01_2019_archival(all_cert_items, current_dir): do_all_analysis(limited_cert_items, 'cc_archived_date={}'.format(archived_date)) -def do_analysis_manufacturers(all_cert_items, current_dir): +def do_analysis_manufacturers(all_cert_items, current_dir: Path): # analyze only Infineon certificates do_analysis_only_filtered(all_cert_items, current_dir, ['processed', 'cc_manufacturer_simple'], 'Infineon Technologies AG') @@ -52,13 +52,13 @@ def do_analysis_manufacturers(all_cert_items, current_dir): ['processed', 'cc_manufacturer_simple'], 'SUSE Linux Products Gmbh') -def do_analysis_only_filtered(all_cert_items, current_dir, filter_path, filter_value): +def do_analysis_only_filtered(all_cert_items, current_dir: Path, filter_path, filter_value): filter_string = '' for item in filter_path: if len(filter_string) > 0: filter_string = filter_string + '__' filter_string = filter_string + item - target_folder = current_dir + '\\{}={}\\'.format(filter_string, filter_value) + target_folder = current_dir / '{}={}'.format(filter_string, filter_value) if not os.path.exists(target_folder): os.makedirs(target_folder) os.chdir(target_folder) @@ -77,15 +77,15 @@ def do_analysis_only_filtered(all_cert_items, current_dir, filter_path, filter_v do_all_analysis(cert_items, '{}={}'.format(filter_string, filter_value)) -def do_analysis_only_category(all_cert_items, current_dir, category): +def do_analysis_only_category(all_cert_items, current_dir: Path, category): do_analysis_only_filtered(all_cert_items, current_dir, ['csv_scan', 'cc_category'], category) -def do_analysis_only_smartcards(all_cert_items, current_dir): +def do_analysis_only_smartcards(all_cert_items, current_dir: Path): do_analysis_only_category(all_cert_items, current_dir, 'ICs, Smart Cards and Smart Card-Related Devices and Systems') -def do_analysis_only_operatingsystems(all_cert_items, current_dir): +def do_analysis_only_operatingsystems(all_cert_items, current_dir: Path): do_analysis_only_category(all_cert_items, current_dir, 'Operating Systems') @@ -99,35 +99,15 @@ def load_json_files(files_list): return tuple(loaded_jsons) -def sanitize_all_strings(data): - printable = set(string.printable) - - if isinstance(data, dict): - for k, v in data.items(): - if isinstance(v, (dict, list, tuple)): - sanitize_all_strings(v) - elif isinstance(v, str): - sanitized = ''.join(filter(lambda x: x in printable, v)) - data[k] = ''.join(filter(lambda x: x in printable, v)) - - if isinstance(data, (list, tuple)): - for v in data: - if isinstance(v, (dict, list, tuple)): - sanitize_all_strings(v) - elif isinstance(v, str): - sanitized = ''.join(filter(lambda x: x in printable, v)) - v = ''.join(filter(lambda x: x in printable, v)) - - @click.command() -@click.option("-d", "--dir", "directory", type=str, help="The directory to use.", required=True) +@click.argument("directory", required=True, type=str) @click.option("--fresh", "do_complete_extraction", is_flag=True, help="Whether to extract from a fresh state.") @click.option("--do-download", "do_download_certs", is_flag=True, help="Whether to download certificate pages.") @click.option("--do-extraction", "do_extraction", is_flag=True, help="Whether to extract information from the certs.") @click.option("--do-pairing", "do_pairing", is_flag=True, help="Whether to pair PP stuff.") @click.option("--do-processing", "do_processing", is_flag=True, help="Whether to process certificates.") @click.option("--do-anaysis", "do_analysis", is_flag=True, help="Whether to analyse certificates.") -def main(directory, do_complete_extraction, do_download_certs, do_extraction, do_pairing, do_processing, do_analysis): +def main(directory, do_complete_extraction: bool, do_download_certs: bool, do_extraction: bool, do_pairing: bool, do_processing: bool, do_analysis: bool): directory = Path(directory) web_dir = directory / "web" @@ -135,14 +115,14 @@ def main(directory, do_complete_extraction, do_download_certs, do_extraction, do pp_dir = directory / "pp" fragments_dir = directory / "cert_fragments" pp_fragments_dir = directory / "pp_fragments" - results_folder = directory / "results" + results_dir = directory / "results" web_dir.mkdir(parents=True, exist_ok=True) walk_dir.mkdir(parents=True, exist_ok=True) pp_dir.mkdir(parents=True, exist_ok=True) fragments_dir.mkdir(parents=True, exist_ok=True) pp_fragments_dir.mkdir(parents=True, exist_ok=True) - results_folder.mkdir(parents=True, exist_ok=True) + results_dir.mkdir(parents=True, exist_ok=True) # 1. generate_basic_download_script # 2. run and download basic cc files from webpage (no certs yet) @@ -171,13 +151,13 @@ def main(directory, do_complete_extraction, do_download_certs, do_extraction, do else: # load previously analyzed results prev_csv, prev_html, prev_front, prev_keywords, prev_pdf_meta = load_json_files( - ['certificate_data_csv_all.json', 'certificate_data_html_all.json', 'certificate_data_frontpage_all.json', - 'certificate_data_keywords_all.json', 'certificate_data_pdfmeta_all.json']) + map(lambda x: results_dir / x, ['certificate_data_csv_all.json', 'certificate_data_html_all.json', 'certificate_data_frontpage_all.json', + 'certificate_data_keywords_all.json', 'certificate_data_pdfmeta_all.json'])) if do_download_certs: # extract_certificates_html() will generate download scripts for cert documents # NOTE: download scripts must be run manually now - current_html = extract_certificates_html(web_dir, False) + current_html = extract_certificates_html(web_dir) # NOTE: Code below is preparation for differetian download of only new certificates # - unfinished now @@ -206,26 +186,26 @@ def main(directory, do_complete_extraction, do_download_certs, do_extraction, do # print('*** New items detected: {}'.format(len(new_items))) if do_extraction: - all_csv = extract_certificates_csv(web_dir, False) - all_html = extract_certificates_html(web_dir, False) - all_front = extract_certificates_frontpage(walk_dir, False) - all_keywords = extract_certificates_keywords(walk_dir, fragments_dir, 'certificate', False) - all_pdf_meta = extract_certificates_pdfmeta(walk_dir, 'certificate', False) + all_csv = extract_certificates_csv(web_dir) + all_html = extract_certificates_html(web_dir) + all_front = extract_certificates_frontpage(walk_dir) + all_keywords = extract_certificates_keywords(walk_dir, fragments_dir, 'certificate') + all_pdf_meta = extract_certificates_pdfmeta(walk_dir, 'certificate', results_dir) # save joined results - with open("certificate_data_csv_all.json", "w") as write_file: + with open(results_dir / "certificate_data_csv_all.json", "w") as write_file: json.dump(all_csv, write_file, indent=4, sort_keys=True) - with open("certificate_data_html_all.json", "w") as write_file: + with open(results_dir / "certificate_data_html_all.json", "w") as write_file: json.dump(all_html, write_file, indent=4, sort_keys=True) - with open("certificate_data_frontpage_all.json", "w") as write_file: + with open(results_dir / "certificate_data_frontpage_all.json", "w") as write_file: json.dump(all_front, write_file, indent=4, sort_keys=True) - with open("certificate_data_keywords_all.json", "w") as write_file: + with open(results_dir / "certificate_data_keywords_all.json", "w") as write_file: json.dump(all_keywords, write_file, indent=4, sort_keys=True) - with open("certificate_data_pdfmeta_all.json", "w") as write_file: + with open(results_dir / "certificate_data_pdfmeta_all.json", "w") as write_file: json.dump(all_pdf_meta, write_file, indent=4, sort_keys=True) # if do_extraction_pp: - # all_pp_csv = extract_protectionprofiles_csv(cc_pp_web_files_dir) + # all_pp_csv = extract_protectionprofiles_csv(web_dir) # all_pp_front = extract_protectionprofiles_frontpage(pp_dir) # all_pp_keywords = extract_certificates_keywords(pp_dir, pp_fragments_dir, 'pp') # all_pp_pdf_meta = extract_certificates_pdfmeta(pp_dir, 'pp') @@ -257,53 +237,53 @@ def main(directory, do_complete_extraction, do_download_certs, do_extraction, do # CERTIFICATES # load results from previous step all_csv, all_html, all_front, all_keywords, all_pdf_meta = load_json_files( - ['certificate_data_csv_all.json', 'certificate_data_html_all.json', 'certificate_data_frontpage_all.json', - 'certificate_data_keywords_all.json', 'certificate_data_pdfmeta_all.json']) + map(lambda x: results_dir / x, ['certificate_data_csv_all.json', 'certificate_data_html_all.json', 'certificate_data_frontpage_all.json', + 'certificate_data_keywords_all.json', 'certificate_data_pdfmeta_all.json'])) # check for unexpected results check_expected_cert_results(all_html, all_csv, all_front, all_keywords, all_pdf_meta) # collate all results into single file all_cert_items = collate_certificates_data(all_html, all_csv, all_front, all_keywords, all_pdf_meta, 'link_security_target') # write collated result - with open("certificate_data_complete.json", "w") as write_file: - json.dump(all_cert_items, write_file, indent=4, sort_keys=True) + with open(results_dir / "certificate_data_complete.json", "w") as write_file: + json.dump(all_cert_items, write_file, indent=4, sort_keys=True) if do_processing: # load information about protection profiles as extracted by sec-certs-pp tool - with open('pp_data_complete_processed.json') as json_file: + with open(results_dir / 'pp_data_complete_processed.json') as json_file: all_pp_items = json.load(json_file) - with open('certificate_data_complete.json') as json_file: + with open(results_dir / 'certificate_data_complete.json') as json_file: all_cert_items = json.load(json_file) all_cert_items = process_certificates_data(all_cert_items, all_pp_items) - with open("certificate_data_complete_processed.json", "w") as write_file: - json.dump(all_cert_items, write_file, indent=4, sort_keys=True) + with open(results_dir / "certificate_data_complete_processed.json", "w") as write_file: + json.dump(all_cert_items, write_file, indent=4, sort_keys=True) if do_analysis: - with open('certificate_data_complete_processed.json') as json_file: + with open(results_dir / 'certificate_data_complete_processed.json') as json_file: all_cert_items = json.load(json_file) if do_analysis_filtered: # analyze only smartcards - do_analysis_only_filtered(all_cert_items, results_folder, + do_analysis_only_filtered(all_cert_items, results_dir, ['csv_scan', 'cc_category'], 'ICs, Smart Cards and Smart Card-Related Devices and Systems') # analyze only operating systems - do_analysis_only_filtered(all_cert_items, results_folder, + do_analysis_only_filtered(all_cert_items, results_dir, ['csv_scan', 'cc_category'], 'Operating Systems') # analyze separate manufacturers - do_analysis_manufacturers(all_cert_items, results_folder) + do_analysis_manufacturers(all_cert_items, results_dir) # archived on 09/01/2019 - do_analysis_09_01_2019_archival(all_cert_items, results_folder) + do_analysis_09_01_2019_archival(all_cert_items, results_dir) # analyze all certificates together - do_analysis_everything(all_cert_items, results_folder) + do_analysis_everything(all_cert_items, results_dir) - with open("certificate_data_complete_processed_analyzed.json", "w") as write_file: - json.dump(all_cert_items, write_file, indent=4, sort_keys=True) + with open(results_dir / "certificate_data_complete_processed_analyzed.json", "w") as write_file: + json.dump(all_cert_items, write_file, indent=4, sort_keys=True) if __name__ == "__main__": diff --git a/setup.py b/setup.py index 9d1469ba..124a3957 100644 --- a/setup.py +++ b/setup.py @@ -29,5 +29,10 @@ setup( "tabula-py", "pikepdf", "Click" - ] + ], + entry_points = """ + [console_scripts] + process-certs=sec_certs.process_certificates:main + fips-certs=sec_certs.fips_certificates:main + """ ) -- cgit v1.3.1