diff options
| author | Petr Svenda | 2020-10-15 17:30:15 +0200 |
|---|---|---|
| committer | GitHub | 2020-10-15 17:30:15 +0200 |
| commit | 4b0e406d5a223286aef32a88899fbb627e25008a (patch) | |
| tree | 229c8353a0555bd1df0bdb2b3f2b78869b21f2c7 | |
| parent | 08d37981b56d6e12c64994156d0a01704a9fa5dc (diff) | |
| parent | 79ed1029a1f2fa3395a35ab8d5382af886d7ea7b (diff) | |
| download | sec-certs-4b0e406d5a223286aef32a88899fbb627e25008a.tar.gz sec-certs-4b0e406d5a223286aef32a88899fbb627e25008a.tar.zst sec-certs-4b0e406d5a223286aef32a88899fbb627e25008a.zip | |
Merge branch 'master' into feature/more-rules
| -rw-r--r-- | .travis.yml | 10 | ||||
| -rw-r--r-- | setup.py | 30 | ||||
| -rw-r--r-- | src/analyze_certificates.py | 24 | ||||
| -rw-r--r-- | src/cert_rules.py | 167 | ||||
| -rw-r--r-- | src/download_files.py | 36 | ||||
| -rw-r--r-- | src/extract_certificates.py | 1242 | ||||
| -rw-r--r-- | src/fips_certificates.py | 444 | ||||
| -rw-r--r-- | src/process_certificates.py | 42 |
8 files changed, 1492 insertions, 503 deletions
diff --git a/.travis.yml b/.travis.yml new file mode 100644 index 00000000..493398bc --- /dev/null +++ b/.travis.yml @@ -0,0 +1,10 @@ +os: linux +language: python +dist: xenial +python: "3.8" + +install: + - pip install . + +script: + - echo "Test here" diff --git a/setup.py b/setup.py new file mode 100644 index 00000000..9f4d51b0 --- /dev/null +++ b/setup.py @@ -0,0 +1,30 @@ +#!/usr/bin/env python3 +from setuptools import setup + +setup( + name='sec-certs', + author='Petr Svenda', + author_email='', + version='0.0.0', + packages='sec-certs', + license='MIT', + description="Tool for analysis of security certificates", + long_description=open("README.md").read(), + long_description_content_type="text/markdown", + classifiers=[ + "Development Status :: 3 - Alpha", + "License :: OSI Approved :: MIT License", + "Topic :: Security", + "Topic :: Security :: Cryptography", + "Programming Language :: Python :: 3", + "Intended Audience :: Developers", + "Intended Audience :: Science/Research" + ], + install_requires=[ + "PyPDF2", + "matplotlib", + "graphviz", + "numpy", + "tabulate" + ] +) diff --git a/src/analyze_certificates.py b/src/analyze_certificates.py index 92091949..addb1711 100644 --- a/src/analyze_certificates.py +++ b/src/analyze_certificates.py @@ -9,6 +9,7 @@ from dateutil import parser import datetime from tags_constants import * import string +import os STOP_ON_UNEXPECTED_NUMS = False @@ -57,10 +58,14 @@ def plot_bar_graph(data, x_data_labels, y_label, title, file_name): plt.axis((x1, x2, y1 - 1, y2)) plt.savefig(file_name + '.png', bbox_inches='tight') plt.savefig(file_name + '.pdf', bbox_inches='tight') + plt.close() def plot_heatmap_graph(data_matrix, x_data_ticks, y_data_ticks, x_label, y_label, title, file_name): - plt.figure(figsize=(round(len(x_data_ticks) / 2), 8), dpi=200, facecolor='w', edgecolor='k') + fig_size = round(len(x_data_ticks) / 2) + if fig_size == 0: + fig_size = 8 + plt.figure(figsize=(fig_size, 8), dpi=200, facecolor='w', edgecolor='k') #color_map = 'BuGn' color_map = 'Purples' plt.imshow(data_matrix, cmap=color_map, interpolation='none', aspect='auto') @@ -76,8 +81,15 @@ def plot_heatmap_graph(data_matrix, x_data_ticks, y_data_ticks, x_label, y_label plt.xlabel(x_label) plt.ylabel(y_label) plt.title(title) - plt.savefig(file_name + '.png', bbox_inches='tight') - plt.savefig(file_name + '.pdf', bbox_inches='tight') + try: + plt.savefig(file_name + '.png', bbox_inches='tight') + except RuntimeError as e: + print('RuntimeError while writing {} file as png'.format(file_name + '.png')) + try: + plt.savefig(file_name + '.pdf', bbox_inches='tight') + except RuntimeError as e: + print('RuntimeError while writing {} file as pdf'.format(file_name + '.pdf')) + plt.close() def compute_and_plot_hist(data, bins, y_label, title, file_name): @@ -109,8 +121,8 @@ def depricated_print_dot_graph_keywordsonly(filter_rules_group, all_items_found, just_file_name = file_name this_cert_id = cert_id[file_name] - if file_name.rfind('\\') != -1: - just_file_name = file_name[file_name.rfind('\\') + 1:] + if file_name.rfind(os.sep) != -1: + just_file_name = file_name[file_name.rfind(os.sep) + 1:] # insert file name and identified probable certification id if this_cert_id != "": @@ -255,6 +267,7 @@ def plot_certid_to_item_graph(item_path, all_items_found, filter_label, out_dot_ dot.render(out_dot_name, view=False) print('{} pdf rendered'.format(out_dot_name)) + def analyze_references_graph(filter_rules_group, all_items_found, filter_label): # build cert_id to item name mapping certid_info = {} @@ -430,6 +443,7 @@ def plot_schemes_multi_line_graph(x_ticks, data, prominent_data, x_label, y_labe plt.title(title) plt.savefig(file_name + '.png', bbox_inches='tight') plt.savefig(file_name + '.pdf', bbox_inches='tight') + plt.close() def analyze_cert_years_frequency(all_cert_items, filter_label): diff --git a/src/cert_rules.py b/src/cert_rules.py index 67069f23..8623996c 100644 --- a/src/cert_rules.py +++ b/src/cert_rules.py @@ -1,21 +1,21 @@ rules_cert_id = [ - 'BSI-DSZ-CC-[0-9]+?-[0-9]+', # German BSI - 'BSI-DSZ-CC-[0-9]+?-(?:V|v)[0-9]+-[0-9]+', # German BSI - 'BSI-DSZ-CC-[0-9]+?-(?:V|v)[0-9]+', # German BSI - 'BSI [0-9]+?', # German BSI - #'CC-Zert-.+?', - 'ANSSI(?:-|-CC-)[0-9]+?/[0-9]+', # French - #'ANSSI-CC-CER-F-.+?', # French - 'DCSSI-[0-9]+?/[0-9]+?', # French - 'Certification Report [0-9]+?/[0-9]+?', # French - 'Rapport de certification [0-9]+?/[0-9]+?', # French - 'NSCIB-CC-[0-9][0-9][0-9][0-9].+?', # Netherlands - 'NSCIB-CC-[0-9][0-9][0-9][0-9][0-9]*-CR', # Netherlands + 'BSI-DSZ-CC-[0-9]+?-[0-9]+', # German BSI + 'BSI-DSZ-CC-[0-9]+?-(?:V|v)[0-9]+-[0-9]+', # German BSI + 'BSI-DSZ-CC-[0-9]+?-(?:V|v)[0-9]+', # German BSI + 'BSI [0-9]+?', # German BSI + # 'CC-Zert-.+?', + 'ANSSI(?:-|-CC-)[0-9]+?/[0-9]+', # French + # 'ANSSI-CC-CER-F-.+?', # French + 'DCSSI-[0-9]+?/[0-9]+?', # French + 'Certification Report [0-9]+?/[0-9]+?', # French + 'Rapport de certification [0-9]+?/[0-9]+?', # French + 'NSCIB-CC-[0-9][0-9][0-9][0-9].+?', # Netherlands + 'NSCIB-CC-[0-9][0-9][0-9][0-9][0-9]*-CR', # Netherlands 'NSCIB-CC-[0-9][0-9]-[0-9]+?-CR[0-9]+?', # Netherlands - 'SERTIT-[0-9]+?', # Norway - 'CCEVS-VR-(?:|VID)[0-9]+?-[0-9]+?', # US NSA - #'[0-9][0-9\-]+?-CR', # Canada - 'CRP[0-9][0-9][0-9][0-9]*?', # UK CESG + 'SERTIT-[0-9]+?', # Norway + 'CCEVS-VR-(?:|VID)[0-9]+?-[0-9]+?', # US NSA + # '[0-9][0-9\-]+?-CR', # Canada + 'CRP[0-9][0-9][0-9][0-9]*?', # UK CESG 'CERTIFICATION REPORT No. P[0-9]+?', # UK CESG '20[0-9][0-9]-[0-9]+-INF-[0-9]+?', # Spain 'KECS-CR-[0-9]+?-[0-9]+?', # Korea @@ -25,7 +25,7 @@ rules_cert_id = [ 'OCSI/CERT/.+?', # Italia '[0-9\\.]+?/TSE-CCCS-[0-9]+?', # Turkis CCCS 'BTBD-.+?', # Turkis CCCS - ] +] rules_vendor = [ 'NXP', @@ -41,12 +41,12 @@ rules_vendor = [ '(?:G\&D|G\+D|Giesecke+Devrient|Giesecke \& Devrient)', 'Philips', 'Sagem', - ] +] rules_eval_facilities = [ 'Serma Technologies', 'THALES - CEACI' - ] +] rules_protection_profiles = [ 'BSI-(?:CC[-_]|)PP[-_]*.+?', @@ -59,22 +59,21 @@ rules_protection_profiles = [ 'ANSSI-CC-PP.+?', 'WBIS_V[0-9]\\.[0-9]', 'EHCT_V.+?' - ] +] rules_technical_reports = [ 'BSI[ ]*TR-[0-9]+?(?:-[0-9]+?|)', - ] - +] rules_device_id = [ 'G87-.+?', 'ATMEL AT.+?', - ] +] rules_os = [ 'STARCOS(?: [0-9\\.]+?|)', 'JCOP[ ]*[0-9]' - ] +] rules_standard_id = [ 'FIPS ?(?:PUB )?[0-9]+-[0-9]+?', @@ -92,13 +91,13 @@ rules_standard_id = [ 'ICAO(?:-SAC|)', '[Xx]\\.509', 'RFC [0-9]+' - ] +] rules_security_level = [ 'EAL[ ]*[0-9+]+?', 'EAL[ ]*[0-9] augmented+?', 'ITSEC[ ]*E[1-9]*.+?', - ] +] rules_security_assurance_components = [ r'ACE_[A-Z]{3}(?:\.[0-9]|)', @@ -136,7 +135,7 @@ rules_javacard = [ r'(?:Java Card|JavaCard) \(version [2-3]\.[0-9](?:\.[0-9]|)\)', r'(?:Global Platform|GlobalPlatform) [2-3]\.[0-9]\.[0-9]', r'(?:Global Platform|GlobalPlatform) \(version [2-3]\.[0-9]\.[0-9]\)', - ] +] rules_crypto_algs = [ 'RSA[- ]*(?:512|768|1024|1280|1536|2048|3072|4096|8192)', @@ -156,7 +155,7 @@ rules_crypto_algs = [ 'DTRNG', 'TRNG', 'RN[GD]', - 'RBG' + 'RBG', ] rules_block_cipher_modes = [ @@ -165,7 +164,7 @@ rules_block_cipher_modes = [ 'CTR', 'CFB', 'OFB', - 'GCM' + 'GCM', ] rules_ecc_curves = [ @@ -187,15 +186,14 @@ rules_cplc = [ 'IC[ ]*Fabricator', 'IC[ ]*Type', 'IC[ ]*Version', - ] +] rules_crypto_engines = [ 'TORNADO', 'SmartMX', 'SmartMX2' 'NesCrypt', - ] - +] rules_crypto_libs = [ '(?:NesLib|NESLIB) [v]*[0-9.]+', @@ -203,8 +201,8 @@ rules_crypto_libs = [ 'AT1 Secure RSA/ECC/SHA library', 'Crypto Library [v]*[0-9.]+', 'ATMEL Toolbox [0-9.]+', - 'v1.02.013' # Infineon's ROCA-vulnerable library - ] + 'v1.02.013' # Infineon's ROCA-vulnerable library +] rules_IC_data_groups = [ r'EF\.DG[1-9][0-6]?', @@ -225,8 +223,7 @@ rules_defenses = [ 'DFA', '[Ff]+ault induction', 'ROCA', - ] - +] rules_certification_process = [ '[oO]ut of [sS]cope', @@ -234,19 +231,99 @@ rules_certification_process = [ '.{0,100}[oO]ut of [sS]cope.{0,100}', '.{0,100}confidential document.{0,100}', '[sS]ecurity [fF]unction SF\\.[a-zA-Z0-9_]', - ] +] rules_vulnerabilities = [ 'CVE-[0-9]+?-[0-9]+?', 'CWE-[0-9]+?', - ] +] rules_other = [ 'library', - #'http[s]*://.+?/' - ] + # 'http[s]*://.+?/' +] + +rules_fips_remove_algorithm_ids = [ + r"HMAC(?:-SHA)?[ -]*((?:160|224|256|384|512)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[, ]*?(?:#|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{4})", + r"HMAC(?:-SHA)?[ -]*((?:160|224|256|384|512)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[, ]*?(?:#|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{3})", + r"HMAC(?:-SHA)?[ -]*((?:160|224|256|384|512)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[, ]*?(?:#|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{2})", + r"HMAC(?:-SHA)?[ -]*((?:160|224|256|384|512)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[, ]*?(?:#|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{1})", + r"SH[SA][-]*(?:160|224|256|384|512)?(?:[\s]*?(?:KAT)?[\s,]*?[\s(]*?(?:#|Cert\.?|Certificate)?[\s#]*?)(\d{4})(?:\)?\[#?\d+\])?", + r"SH[SA][-]*(?:160|224|256|384|512)?(?:[\s]*?(?:KAT)?[\s,]*?[\s(]*?(?:#|Cert\.?|Certificate)?[\s#]*?)(\d{3})(?:\)?\[#?\d+\])?", + r"SH[SA][-]*(?:160|224|256|384|512)?(?:[\s]*?(?:KAT)?[\s,]*?[\s(]*?(?:#|Cert\.?|Certificate)?[\s#]*?)(\d{2})(?:\)?\[#?\d+\])?", + r"SH[SA][-]*(?:160|224|256|384|512)?(?:[\s]*?(?:KAT)?[\s,]*?[\s(]*?(?:#|Cert\.?|Certificate)?[\s#]*?)(\d{1})(?:\)?\[#?\d+\])?", + r"RSA(?:[- ]*(?:512|768|1024|1280|1536|2048|3072|4096|8192)\s]*?(?:(?:KAT|Verify|PSS|\s)*?)?[\s,]*?[\s(]*?(?:#|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{4})", + r"RSA(?:[- ]*(?:512|768|1024|1280|1536|2048|3072|4096|8192)\s]*?(?:(?:KAT|Verify|PSS|\s)*?)?[\s,]*?[\s(]*?(?:#|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{3})", + r"RSA(?:[- ]*(?:512|768|1024|1280|1536|2048|3072|4096|8192)\s]*?(?:(?:KAT|Verify|PSS|\s)*?)?[\s,]*?[\s(]*?(?:#|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{2})", + r"RSA(?:[- ]*(?:512|768|1024|1280|1536|2048|3072|4096|8192)\s]*?(?:(?:KAT|Verify|PSS|\s)*?)?[\s,]*?[\s(]*?(?:#|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{1})", + r"(?:RSA)?[- ]?(?:SSA)?[- ]?PKCS\s?#?\d(?:-[Vv]1_5| [Vv]1[-_]5)?\s?(\d{4})?", + r"(?:RSA)?[- ]?(?:SSA)?[- ]?PKCS\s?#?\d(?:-[Vv]1_5| [Vv]1[-_]5)?\s?(\d{3})?", + r"(?:RSA)?[- ]?(?:SSA)?[- ]?PKCS\s?#?\d(?:-[Vv]1_5| [Vv]1[-_]5)?\s?(\d{2})?", + r"(?:RSA)?[- ]?(?:SSA)?[- ]?PKCS\s?#?\d(?:-[Vv]1_5| [Vv]1[-_]5)?\s?(\d{1})?", + r"AES[- ]*((?:128|192|256|)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT|CMAC|CTR)*?[,\s(]*?(?:#|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{4})(?:\)?\[#?\d+\])?", + r"AES[- ]*((?:128|192|256|)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT|CMAC|CTR)*?[,\s(]*?(?:#|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{3})(?:\)?\[#?\d+\])?", + r"AES[- ]*((?:128|192|256|)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT|CMAC|CTR)*?[,\s(]*?(?:#|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{2})(?:\)?\[#?\d+\])?", + r"AES[- ]*((?:128|192|256|)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT|CMAC|CTR)*?[,\s(]*?(?:#|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{1})(?:\)?\[#?\d+\])?", + r"Diffie[- ]*Hellman[,\s(]*?(?:CVL|\s)*?(?:#|Cert\.?|Certificate)?[\s#]*?\s?(\d{4})", + r"Diffie[- ]*Hellman[,\s(]*?(?:CVL|\s)*?(?:#|Cert\.?|Certificate)?[\s#]*?\s?(\d{3})", + r"Diffie[- ]*Hellman[,\s(]*?(?:CVL|\s)*?(?:#|Cert\.?|Certificate)?[\s#]*?\s?(\d{2})", + r"Diffie[- ]*Hellman[,\s(]*?(?:CVL|\s)*?(?:#|Cert\.?|Certificate)?[\s#]*?\s?(\d{1})", + r"DRBG[ -]*((?:160|224|256|384|512)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[,\s]*?(?:#|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{4})", + r"DRBG[ -]*((?:160|224|256|384|512)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[,\s]*?(?:#|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{3})", + r"DRBG[ -]*((?:160|224|256|384|512)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[,\s]*?(?:#|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{2})", + r"DRBG[ -]*((?:160|224|256|384|512)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[,\s]*?(?:#|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{1})", + r"DES[ -]*((?:160|224|256|384|512)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[,\s]*?(?:#|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{4})", + r"DES[ -]*((?:160|224|256|384|512)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[,\s]*?(?:#|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{3})", + r"DES[ -]*((?:160|224|256|384|512)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[,\s]*?(?:#|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{2})", + r"DES[ -]*((?:160|224|256|384|512)?(?: |[Dd]ecrypt|[Ee]ncrypt|KAT)*?[,\s]*?(?:#|Cert\.?|Certificate)?[\s#]*?)?\s?(\d{1})", + r"CVL[\s#]*?(\d{4})", + r"CVL[\s#]*?(\d{3})", + r"CVL[\s#]*?(\d{2})", + r"CVL[\s#]*?(\d{1})", + r"PAA[: #]*?\d{4}", + r"PAA[: #]*?\d{3}", + r"PAA[: #]*?\d{2}", + r"PAA[: #]*?\d{1}", + r"(?:#|Cert\.?|Certificate)[\s#]*?(\d+)?\s*?(?:AES|SHS|SHA|RSA|HMAC|Diffie-Hellman|DRBG|DES|CVL)", + r"PKCS[ ]?#?\d+", + r"Survey #192" # why would they get an address like this /o\ cert 2079 +] +rules_fips_cert = [ + # r"(?:#\s?|Cert\.?[^. ]*?\s?)(?P<id>\d{4})", + # r"(?:#\s?|Cert\.?[^. ]*?\s?)(?P<id>\d{3})", + # r"(?:#\s?|Cert\.?[^. ]*?\s?)(?P<id>\d{2})", + # r"(?:#\s?|Cert\.?[^. ]*?\s?)(?P<id>\d{1}) + r"(?:#[^\S\r\n]?|Cert\.?(?!.\s)[^\S\r\n]?|Certificate[^\S\r\n]?)(?P<id>\d{4})", + r"(?:#[^\S\r\n]?|Cert\.?(?!.\s)[^\S\r\n]?|Certificate[^\S\r\n]?)(?P<id>\d{3})", + r"(?:#[^\S\r\n]?|Cert\.?(?!.\s)[^\S\r\n]?|Certificate[^\S\r\n]?)(?P<id>\d{2})", + r"(?:#[^\S\r\n]?|Cert\.?(?!.\s)[^\S\r\n]?|Certificate[^\S\r\n]?)(?P<id>\d{1})" +] + +# rule still too "general" +rules_fips_security_level = [ + r"[lL]evel (\d)" +] -#rules_security_target_class +rules_fips_htmls = [ + r"module-name\">\s*(?P<fips_module_name>[^<]*)", + r"module-standard\">\s*(?P<fips_standard>[^<]*)", + r"Status[\s\S]*?\">\s*(?P<fips_status>[^<]*)", + r"Sunset Date[\s\S]*?\">\s*(?P<fips_date_sunset>[^<]*)", + r"Validation Dates[\s\S]*?\">\s*(?P<fips_date_validation>[^<]*)", + r"Overall Level[\s\S]*?\">\s*(?P<fips_level>[^<]*)", + r"Caveat[\s\S]*?\">\s*(?P<fips_caveat>[^<]*)", + r"Security Level Exceptions[\s\S]*?\">\s*(?P<fips_exceptions><ul.*</ul>)", + r"Module Type[\s\S]*?\">\s*(?P<fips_type>[^<]*)", + r"Embodiment[\s\S]*?\">\s*(?P<fips_embodiment>[^<]*)", + r"Tested Configuration[\s\S]*?\">\s*(?P<fips_tested_conf><ul.*</ul>)", + r"FIPS Algorithms[\s\S]*?\">\s*(?P<fips_algorithms><tbody>[\s\S]*</tbody>)", + r"Allowed Algorithms[\s\S]*?\">\s*(?P<fips_allowed_algorithms>[^<]*)", + r"Software Versions[\s\S]*?\">\s*(?P<fips_software>[^<]*)", + r"Product URL[\s\S]*?\">\s*<a href=\"(?P<fips_url>.*)\"", + r"Vendor<\/h4>[\s\S]*?href=\".*?\">(?P<fips_vendor>.*?)<\/a>" +] + +# rules_security_target_class rules = {} rules['rules_vendor'] = rules_vendor rules['rules_cert_id'] = rules_cert_id @@ -270,3 +347,11 @@ rules['rules_defenses'] = rules_defenses rules['rules_certification_process'] = rules_certification_process rules['rules_vulnerabilities'] = rules_vulnerabilities rules['rules_other'] = rules_other + +# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ +# For FIPS +# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ +fips_rules = {} +fips_rules['rules_fips_algorithms'] = rules_fips_remove_algorithm_ids +fips_rules['rules_security_level'] = rules_fips_security_level +fips_rules['rules_cert_id'] = rules_fips_cert diff --git a/src/download_files.py b/src/download_files.py new file mode 100644 index 00000000..a877593c --- /dev/null +++ b/src/download_files.py @@ -0,0 +1,36 @@ +import extract_certificates + +FILE_ERRORS_STRATEGY = extract_certificates.FILE_ERRORS_STRATEGY + + +def generate_fips_basic_download_script(): + with open('download_fips_web.bat', 'w', errors=FILE_ERRORS_STRATEGY) as file: + file.write( + 'curl "https://csrc.nist.gov/projects/cryptographic-module-validation-program/validated-modules/search' + '/all" -o fips_modules_validated.html\n') + + +def generate_fips_download_script(file_name, fips_dir): + """generate_fips_download_script. + + :param file_name: name of the download file + :param fips_dir: directory for saved files + """ + html_dir = fips_dir + '/html/' + sp_dir = fips_dir + '/security_policies/' + + with open(file_name, 'w', errors=FILE_ERRORS_STRATEGY) as write_file: + # make directories for both html and security policies, scraping in one go + write_file.write('mkdir {}\n'.format(html_dir)) + write_file.write('mkdir {}\n\n'.format(sp_dir)) + + for cert_id in range(1, 4001): + write_file.write( + 'curl "https://csrc.nist.gov/projects/cryptographic-module-validation-program/certificate/{}" -o {}{}.html\n'.format( + cert_id, html_dir, cert_id)) + write_file.write( + 'curl "https://csrc.nist.gov/CSRC/media/projects/cryptographic-module-validation-program/documents' + '/security-policies/140sp{}.pdf" -o {}{}.pdf\n'.format( + cert_id, sp_dir, cert_id)) + write_file.write("{} {}{}.pdf\n".format( + extract_certificates.PDF2TEXT_CONVERT, sp_dir, cert_id)) diff --git a/src/extract_certificates.py b/src/extract_certificates.py index ada54be2..887f06a4 100644 --- a/src/extract_certificates.py +++ b/src/extract_certificates.py @@ -1,3 +1,5 @@ +from PyPDF2 import PdfFileReader +from tags_constants import * import re import os import operator @@ -7,12 +9,11 @@ import csv import string from analyze_certificates import is_in_dict -from cert_rules import rules +from cert_rules import rules, fips_rules from enum import Enum -import matplotlib.pyplot as plt; plt.rcdefaults() -from tags_constants import * +import matplotlib.pyplot as plt -from PyPDF2 import PdfFileReader +plt.rcdefaults() # if True, then exception is raised when unexpect intermediate number is obtained # Used as sanity check during development to detect sudden drop in number of extracted features @@ -21,16 +22,17 @@ APPEND_DETAILED_MATCH_MATCHES = False VERBOSE = False FILE_ERRORS_STRATEGY = 'surrogateescape' 'replace' -#FILE_ERRORS_STRATEGY = 'strict' +# FILE_ERRORS_STRATEGY = 'strict' CC_WEB_URL = 'https://www.commoncriteriaportal.org' PDF2TEXT_CONVERT = 'pdftotext -raw' REGEXEC_SEP = '[ ,;\]”)(]' LINE_SEPARATOR = ' ' -#LINE_SEPARATOR = '' # if newline is not replaced with space, long string included in matches are found +# 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): yield from [os.path.join(root, x) for x in files] @@ -49,7 +51,6 @@ def get_line_number(lines, line_length_compensation, match_start_index): return -1 - def load_cert_file(file_name, limit_max_lines=-1, line_separator=LINE_SEPARATOR): lines = [] was_unicode_decode_error = False @@ -76,7 +77,8 @@ def load_cert_file(file_name, limit_max_lines=-1, line_separator=LINE_SEPARATOR) whole_text_with_newlines = '' # we will estimate the line for searched matches # => we need to known how much lines were modified (removal of eoln..) - line_length_compensation = 1 - len(LINE_SEPARATOR) # for removed newline and for any added separator + # for removed newline and for any added separator + line_length_compensation = 1 - len(LINE_SEPARATOR) lines_included = 0 for line in lines: if limit_max_lines != -1 and lines_included >= limit_max_lines: @@ -109,7 +111,7 @@ def normalize_match_string(match): # normalize match match = match.strip() match = match.rstrip(']') - match = match.rstrip('/') + match = match.rstrip(os.sep) match = match.rstrip(';') match = match.rstrip('.') match = match.rstrip('”') @@ -131,12 +133,16 @@ def set_match_string(items, key_name, new_value): else: old_value = items[key_name] if old_value != new_value: - print(' WARNING: values mismatch, key=\'{}\', old=\'{}\', new=\'{}\''.format(key_name, old_value, new_value)) + print(' WARNING: values mismatch, key=\'{}\', old=\'{}\', new=\'{}\''.format( + key_name, old_value, new_value)) -def parse_cert_file(file_name, search_rules, limit_max_lines=-1, line_separator=LINE_SEPARATOR): - whole_text, whole_text_with_newlines, was_unicode_decode_error = load_cert_file(file_name, limit_max_lines, line_separator) +def parse_cert_file(file_name, search_rules, limit_max_lines=-1, line_separator=LINE_SEPARATOR, + should_censure_right_away=False, fips_items=None, ): + whole_text, whole_text_with_newlines, was_unicode_decode_error = load_cert_file( + file_name, limit_max_lines, line_separator) + file_name = os.path.splitext(os.path.splitext(os.path.basename(file_name))[0])[0] # apply all rules items_found_all = {} for rule_group in search_rules.keys(): @@ -146,15 +152,31 @@ def parse_cert_file(file_name, search_rules, limit_max_lines=-1, line_separator= items_found = items_found_all[rule_group] for rule in search_rules[rule_group]: + rule_and_sep = rule + REGEXEC_SEP - for m in re.finditer(rule_and_sep, whole_text): + for m in re.finditer(rule_and_sep, whole_text_with_newlines): # insert rule if at least one match for it was found if rule not in items_found: items_found[rule] = {} match = m.group() match = normalize_match_string(match) + is_algorithm = False + if fips_items and match != '': + certs = [x['Certificate'] + for x in fips_items[file_name]['fips_algorithms']] + + match_cert_id = ''.join(filter(str.isdigit, match)) + # if file_name == '/home/stan/sec-certs-master/files/fips/security_policies/3676.html.txt': + + for fips_cert in certs: + for actual_cert in fips_cert: + if actual_cert != '' and match_cert_id == ''.join(filter(str.isdigit, actual_cert)): + is_algorithm = True + + if is_algorithm: + continue if match not in items_found[rule]: items_found[rule][match] = {} @@ -169,28 +191,37 @@ def parse_cert_file(file_name, search_rules, limit_max_lines=-1, line_separator= # estimate line in original text file # line_number = get_line_number(lines, line_length_compensation, match_span[0]) # start index, end index, line number - #items_found[rule][match][TAG_MATCH_MATCHES].append([match_span[0], match_span[1], line_number]) + # items_found[rule][match][TAG_MATCH_MATCHES].append([match_span[0], match_span[1], line_number]) if APPEND_DETAILED_MATCH_MATCHES: - items_found[rule][match][TAG_MATCH_MATCHES].append([match_span[0], match_span[1]]) + items_found[rule][match][TAG_MATCH_MATCHES].append( + [match_span[0], match_span[1]]) + if should_censure_right_away: + whole_text_with_newlines = whole_text_with_newlines.replace( + match, 'x' * len(match)) # highlight all found strings from the input text and store the rest - for rule_group in items_found_all.keys(): - items_found = items_found_all[rule_group] - for rule in items_found.keys(): - for match in items_found[rule]: - whole_text_with_newlines = whole_text_with_newlines.replace(match, 'x' * len(match)) # warning - if AES string is removed before AES-128, -128 will be left in text (does it matter?) + if not should_censure_right_away: + for rule_group in items_found_all.keys(): + items_found = items_found_all[rule_group] + for rule in items_found.keys(): + for match in items_found[rule]: + # warning - if AES string is removed before AES-128, -128 will be left in text (does it matter?) + whole_text_with_newlines = whole_text_with_newlines.replace( + match, 'x' * len(match)) return items_found_all, (whole_text_with_newlines, was_unicode_decode_error) def print_total_matches_in_files(all_items_found_count): - sorted_all_items_found_count = sorted(all_items_found_count.items(), key=operator.itemgetter(1)) + sorted_all_items_found_count = sorted( + all_items_found_count.items(), key=operator.itemgetter(1)) for file_name_count in sorted_all_items_found_count: print('{:03d}: {}'.format(file_name_count[1], file_name_count[0])) def print_total_found_cert_ids(all_items_found_certid_count): - sorted_certid_count = sorted(all_items_found_certid_count.items(), key=operator.itemgetter(1), reverse=True) + sorted_certid_count = sorted( + all_items_found_certid_count.items(), key=operator.itemgetter(1), reverse=True) for file_name_count in sorted_certid_count: print('{:03d}: {}'.format(file_name_count[1], file_name_count[0])) @@ -199,8 +230,8 @@ def print_guessed_cert_id(cert_id): sorted_cert_id = sorted(cert_id.items(), key=operator.itemgetter(1)) for double in sorted_cert_id: just_file_name = double[0] - if just_file_name.rfind('\\') != -1: - just_file_name = just_file_name[just_file_name.rfind('\\') + 1:] + if just_file_name.rfind(os.sep) != -1: + just_file_name = just_file_name[just_file_name.rfind(os.sep) + 1:] print('{:30s}: {}'.format(double[1], just_file_name)) @@ -247,19 +278,21 @@ def estimate_cert_id(frontpage_scan, keywords_scan, file_name): keywords_cert_id = match num_items_found_certid_group += num_occurences if VERBOSE: - print(' -> most frequent cert id: {}, {}x'.format(keywords_cert_id, num_items_found_certid_group)) + print(' -> most frequent cert id: {}, {}x'.format(keywords_cert_id, + num_items_found_certid_group)) # try to search for certificate id directly in file name - if found, higher priority filename_cert_id = '' if file_name != None: file_name_no_suff = file_name[:file_name.rfind('.')] - file_name_no_suff = file_name_no_suff[file_name_no_suff.rfind('\\') + 1:] + file_name_no_suff = file_name_no_suff[file_name_no_suff.rfind( + os.sep) + 1:] for rule in rules['rules_cert_id']: file_name_no_suff += ' ' matches = re.findall(rule, file_name_no_suff) if len(matches) > 0: # we found cert id directly in name - #print(' -> cert id found directly in certificate name: {}'.format(matches[0])) + # print(' -> cert id found directly in certificate name: {}'.format(matches[0])) filename_cert_id = matches[0] if VERBOSE: @@ -302,9 +335,11 @@ def print_specified_property_sorted(section_name, item_name, items_found_all): for file_name in items_found_all.keys(): if section_name in items_found_all[file_name].keys(): if item_name in items_found_all[file_name][section_name].keys(): - specific_item_values.append(items_found_all[file_name][item_name]) + specific_item_values.append( + items_found_all[file_name][item_name]) else: - print('WARNING: Item {} not found in file {}'.format(item_name, file_name)) + print('WARNING: Item {} not found in file {}'.format( + item_name, file_name)) print('*** Occurrences of *{}* item'.format(item_name)) sorted_items = sorted(specific_item_values) @@ -314,16 +349,18 @@ def print_specified_property_sorted(section_name, item_name, items_found_all): def print_found_properties(items_found_all): print_specified_property_sorted(TAG_CERT_ID, items_found_all) - print_specified_property_sorted(TAG_CERT_ITEM , items_found_all) + print_specified_property_sorted(TAG_CERT_ITEM, items_found_all) print_specified_property_sorted(TAG_CERT_ITEM_VERSION, items_found_all) - print_specified_property_sorted(TAG_REFERENCED_PROTECTION_PROFILES, items_found_all) - print_specified_property_sorted(TAG_CC_VERSION , items_found_all) + print_specified_property_sorted( + TAG_REFERENCED_PROTECTION_PROFILES, items_found_all) + print_specified_property_sorted(TAG_CC_VERSION, items_found_all) print_specified_property_sorted(TAG_CC_SECURITY_LEVEL, items_found_all) - print_specified_property_sorted(TAG_DEVELOPER , items_found_all) + print_specified_property_sorted(TAG_DEVELOPER, items_found_all) print_specified_property_sorted(TAG_CERT_LAB, items_found_all) def search_only_headers_bsi(walk_dir): + print('BSI HEADER SEARCH') LINE_SEPARATOR_STRICT = ' ' NUM_LINES_TO_INVESTIGATE = 15 rules_certificate_preface = [ @@ -340,13 +377,14 @@ def search_only_headers_bsi(walk_dir): file_ext = file_name[file_name.rfind('.'):] if file_ext != '.txt': continue - print('*** {} ***'.format(file_name)) + # print('*** {} ***'.format(file_name)) 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 = load_cert_file(file_name, NUM_LINES_TO_INVESTIGATE, LINE_SEPARATOR_STRICT) + whole_text, whole_text_with_newlines, was_unicode_decode_error = load_cert_file( + file_name, NUM_LINES_TO_INVESTIGATE, LINE_SEPARATOR_STRICT) for rule in rules_certificate_preface: rule_and_sep = rule + REGEXEC_SEP @@ -372,9 +410,13 @@ def search_only_headers_bsi(walk_dir): for from_keyword in FROM_KEYWORD_LIST: from_keyword_len = len(from_keyword) if certified_item.find(from_keyword) != -1: - print('string **{}** detected in certified item - shall not be here, fixing...'.format(from_keyword)) - certified_item_first = certified_item[:certified_item.find(from_keyword)] - developer = certified_item[certified_item.find(from_keyword) + from_keyword_len:] + print( + 'string **{}** detected in certified item - shall not be here, fixing...'.format( + from_keyword)) + 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 @@ -387,7 +429,8 @@ def search_only_headers_bsi(walk_dir): developer = developer[:end_pos] items_found[TAG_CERT_ID] = normalize_match_string(cert_id) - items_found[TAG_CERT_ITEM] = normalize_match_string(certified_item) + items_found[TAG_CERT_ITEM] = normalize_match_string( + certified_item) items_found[TAG_DEVELOPER] = normalize_match_string(developer) items_found[TAG_CERT_LAB] = 'BSI' @@ -398,7 +441,8 @@ def search_only_headers_bsi(walk_dir): 'PP Conformance: (.+)Functionality: (.+)Assurance: (.+)The IT Product identified', ] - whole_text, whole_text_with_newlines, was_unicode_decode_error = load_cert_file(file_name) + whole_text, whole_text_with_newlines, was_unicode_decode_error = load_cert_file( + file_name) for rule in rules_certificate_third: rule_and_sep = rule + REGEXEC_SEP @@ -413,9 +457,12 @@ def search_only_headers_bsi(walk_dir): cc_version = match_groups[1] cc_security_level = match_groups[2] - items_found[TAG_REFERENCED_PROTECTION_PROFILES] = normalize_match_string(ref_protection_profiles) - items_found[TAG_CC_VERSION] = normalize_match_string(cc_version) - items_found[TAG_CC_SECURITY_LEVEL] = normalize_match_string(cc_security_level) + items_found[TAG_REFERENCED_PROTECTION_PROFILES] = normalize_match_string( + ref_protection_profiles) + items_found[TAG_CC_VERSION] = normalize_match_string( + cc_version) + items_found[TAG_CC_SECURITY_LEVEL] = normalize_match_string( + cc_security_level) if no_match_yet: files_without_match.append(file_name) @@ -443,64 +490,105 @@ def search_only_headers_anssi(walk_dir): 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é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'), + (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'), + (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'), + (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'), + (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'), ] -# rules_certificate_preface = [ -# (HEADER_TYPE.HEADER_FULL, 'ddddd'), -# ] + # rules_certificate_preface = [ + # (HEADER_TYPE.HEADER_FULL, 'ddddd'), + # ] # statistics about rules success rate num_rules_hits = {} for rule in rules_certificate_preface: num_rules_hits[rule[1]] = 0 + print('***ANSSI HEADER SEARCH***') items_found_all = {} files_without_match = [] for file_name in search_files(walk_dir): @@ -509,9 +597,10 @@ def search_only_headers_anssi(walk_dir): file_ext = file_name[file_name.rfind('.'):] if file_ext != '.txt': continue - print('*** {} ***'.format(file_name)) + # print('*** {} ***'.format(file_name)) - whole_text, whole_text_with_newlines, was_unicode_decode_error = load_cert_file(file_name) + whole_text, whole_text_with_newlines, was_unicode_decode_error = load_cert_file( + file_name) # for ANSII and DCSSI certificates, front page starts only on third page after 2 newpage signs pos = whole_text.find(' ') @@ -544,7 +633,8 @@ def search_only_headers_anssi(walk_dir): other_rule_already_match = True other_rule = rule else: - print('WARNING: multiple rules are matching same certification document: ' + file_name) + print( + 'WARNING: multiple rules are matching same certification document: ' + file_name) num_rules_hits[rule[1]] += 1 # add hit to this rule @@ -552,34 +642,42 @@ def search_only_headers_anssi(walk_dir): index_next_item = 0 - items_found[TAG_CERT_ID] = normalize_match_string(match_groups[index_next_item]) + items_found[TAG_CERT_ID] = normalize_match_string( + match_groups[index_next_item]) index_next_item += 1 - items_found[TAG_CERT_ITEM] = normalize_match_string(match_groups[index_next_item]) + items_found[TAG_CERT_ITEM] = normalize_match_string( + match_groups[index_next_item]) index_next_item += 1 if rule[0] == HEADER_TYPE.HEADER_MISSING_CERT_ITEM_VERSION: items_found[TAG_CERT_ITEM_VERSION] = '' else: - items_found[TAG_CERT_ITEM_VERSION] = normalize_match_string(match_groups[index_next_item]) + items_found[TAG_CERT_ITEM_VERSION] = normalize_match_string( + match_groups[index_next_item]) index_next_item += 1 if rule[0] == HEADER_TYPE.HEADER_MISSING_PROTECTION_PROFILES: items_found[TAG_REFERENCED_PROTECTION_PROFILES] = '' else: - items_found[TAG_REFERENCED_PROTECTION_PROFILES] = normalize_match_string(match_groups[index_next_item]) + items_found[TAG_REFERENCED_PROTECTION_PROFILES] = normalize_match_string( + match_groups[index_next_item]) index_next_item += 1 - items_found[TAG_CC_VERSION] = normalize_match_string(match_groups[index_next_item]) + items_found[TAG_CC_VERSION] = normalize_match_string( + match_groups[index_next_item]) index_next_item += 1 - items_found[TAG_CC_SECURITY_LEVEL] = normalize_match_string(match_groups[index_next_item]) + items_found[TAG_CC_SECURITY_LEVEL] = normalize_match_string( + match_groups[index_next_item]) index_next_item += 1 - items_found[TAG_DEVELOPER] = normalize_match_string(match_groups[index_next_item]) + items_found[TAG_DEVELOPER] = normalize_match_string( + match_groups[index_next_item]) index_next_item += 1 - items_found[TAG_CERT_LAB] = normalize_match_string(match_groups[index_next_item]) + items_found[TAG_CERT_LAB] = normalize_match_string( + match_groups[index_next_item]) index_next_item += 1 if no_match_yet: @@ -600,7 +698,8 @@ def search_only_headers_anssi(walk_dir): if True: print('# hits for rule') - sorted_rules = sorted(num_rules_hits.items(), key=operator.itemgetter(1), reverse=True) + 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])) @@ -610,12 +709,15 @@ def search_only_headers_anssi(walk_dir): return items_found_all, files_without_match -def extract_certificates_frontpage(walk_dir, 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(walk_dir) +def extract_certificates_frontpage(walk_dir, 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( + walk_dir) print('*** Files without detected header') - files_without_match = list(set(anssi_files_without_match) & set(bsi_files_without_match)) + files_without_match = list( + set(anssi_files_without_match) & set(bsi_files_without_match)) for file_name in files_without_match: print(file_name) print('Total no hits files: {}'.format(len(files_without_match))) @@ -625,7 +727,8 @@ def extract_certificates_frontpage(walk_dir, 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)) + write_file.write(json.dumps( + items_found_all, indent=4, sort_keys=True)) return items_found_all @@ -654,7 +757,6 @@ def search_pp_only_headers(walk_dir): ANSSI_TYPE2 = 42 ANSSI_TYPE3 = 43 - rules_pp_third = [ (HEADER_TYPE.BSI_TYPE1, 'PP Reference .+?Title (.+)?CC Version (.+)?Assurance Level (.+)?General Status (.+)?Version Number (.+)?Registration (.+)?Keywords (.+)?TOE Overview'), @@ -672,18 +774,19 @@ def search_pp_only_headers(walk_dir): 'Protection profile reference[ ]*Title: (.+)?Author: (.+)?Version: (.+)?Context'), (HEADER_TYPE.FRONT_DCSSI_TYPE3, 'Direction centrale de la sécurité des systèmes d\’information(.+)?(?:Creation date|Date)[ ]*[:]*(.+)?Reference[ ]*[:]*(.+)?Version[ ]*[:]*(.+)?Courtesy Translation[ ]*Courtesy translation.+?under the reference (DCSSI-PP-[0-9/]+)?\.[ ]*Page'), -# (HEADER_TYPE.FRONT_DCSSI_TYPE4, -# 'Direction centrale de la sécurité des systèmes d\’information(.+)?Date[ ]*:(.+)?Reference[ ]*:(.+)?Version[ ]*:(.+)?Courtesy Translation[ ]*Courtesy translation.+?under the reference (DCSSI-PP-[0-9/]+)?\.[ ]*Page'), + # (HEADER_TYPE.FRONT_DCSSI_TYPE4, + # 'Direction centrale de la sécurité des systèmes d\’information(.+)?Date[ ]*:(.+)?Reference[ ]*:(.+)?Version[ ]*:(.+)?Courtesy Translation[ ]*Courtesy translation.+?under the reference (DCSSI-PP-[0-9/]+)?\.[ ]*Page'), (HEADER_TYPE.FRONT_DCSSI_TYPE4, 'Direction centrale de la sécurité des systèmes d’information (.+)?(?:Creation date|Date)[ ]*:(.+)?Reference[ ]*:(.+)?Version[ ]*:(.+)?Courtesy Translation[ ]*Courtesy translation.+?under the reference (DCSSI-PP-[0-9/]+)?\.[ ]*Page'), - #'Direction centrale de la sécurité des systèmes d’information Time-stamping System Protection Profile Date : July 18, 2008 Reference : PP-SH-CCv3.1 Version : 1.7 Courtesy Translation Courtesy translation of the protection profile registered and certified by the French Certification Body under the reference DCSSI-PP-2008/07. Page' + # 'Direction centrale de la sécurité des systèmes d’information Time-stamping System Protection Profile Date : July 18, 2008 Reference : PP-SH-CCv3.1 Version : 1.7 Courtesy Translation Courtesy translation of the protection profile registered and certified by the French Certification Body under the reference DCSSI-PP-2008/07. Page' (HEADER_TYPE.DCSSI_TYPE5, 'Protection Profile identification[ ]*Title[ ]*[:]*(.+)?Author[ ]*[:]*(.+)?Version[ ]*[:]*(.+)?,(.+)?Sponsor[ ]*[:]*(.+)?CC version[ ]*[:]*(.+)?(?:Context|Protection Profile introduction)'), (HEADER_TYPE.DCSSI_TYPE6, 'PP reference.+?Title[ ]*:(.+)?Author[ ]*:(.+)?Version[ ]*:(.+)?Date[ ]*:(.+)?Sponsor[ ]*:(.+)?CC version[ ]*:(.+)?This protection profile.+?The evaluation assurance level required by this protection profile is (.+)?specified by the DCSSI qualification process'), -# (HEADER_TYPE.DCSSI_TYPE7, -# 'Protection Profile identification.+?Title[ ]*[:]*(.+)?Author[ ]*[:]*(.+)?Version[ ]*[:]*(.+)?,(.+)?Sponsor[ ]*[:]*(.+)?CC version[ ]*[:]*(.+)?Protection Profile introduction') + # (HEADER_TYPE.DCSSI_TYPE7, + # 'Protection Profile identification.+?Title[ ]*[:]*(.+)?Author[ ]*[:]*(.+)?Version[ ]*[:]*(.+)?,(.+)?Sponsor[ ]*[:]*(.+)?CC version[ ]*[:]*(.+)?Protection Profile introduction') ] + print("***PP HEADER SEARCH***") items_found_all = {} files_without_match = [] for file_name in search_files(walk_dir): @@ -692,13 +795,14 @@ def search_pp_only_headers(walk_dir): file_ext = file_name[file_name.rfind('.'):] if file_ext != '.txt': continue - print('*** {} ***'.format(file_name)) + # print('*** {} ***'.format(file_name)) # # Process page with more detailed protection profile info # PP Reference - whole_text, whole_text_with_newlines, was_unicode_decode_error = load_cert_file(file_name) + whole_text, whole_text_with_newlines, was_unicode_decode_error = load_cert_file( + file_name) no_match_yet = True for rule in rules_pp_third: @@ -720,180 +824,253 @@ def search_pp_only_headers(walk_dir): index = 0 if rule[0] == HEADER_TYPE.BSI_TYPE1: - set_match_string(items_found, TAG_PP_TITLE, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_TITLE, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_CC_VERSION, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_CC_VERSION, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_CC_SECURITY_LEVEL, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_CC_SECURITY_LEVEL, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_GENERAL_STATUS, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_GENERAL_STATUS, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_VERSION_NUMBER, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_VERSION_NUMBER, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_ID, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_ID, + normalize_match_string(match_groups[index])) index += 1 keywords = match_groups[index].lstrip(' ') - set_match_string(items_found, TAG_KEYWORDS, normalize_match_string(keywords[0:keywords.find(' ')])) + set_match_string(items_found, TAG_KEYWORDS, normalize_match_string( + keywords[0:keywords.find(' ')])) index += 1 set_match_string(items_found, TAG_PP_AUTHORS, 'BSI') - set_match_string(items_found, TAG_PP_REGISTRATOR_SIMPLIFIED, 'BSI') + set_match_string( + items_found, TAG_PP_REGISTRATOR_SIMPLIFIED, 'BSI') if rule[0] == HEADER_TYPE.BSI_TYPE2: - set_match_string(items_found, TAG_PP_TITLE, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_TITLE, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_VERSION_NUMBER, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_VERSION_NUMBER, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_DATE, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_DATE, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_AUTHORS, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_AUTHORS, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_REGISTRATOR, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_REGISTRATOR, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_ID, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_ID, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_CC_SECURITY_LEVEL, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_CC_SECURITY_LEVEL, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_CC_VERSION, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_CC_VERSION, + normalize_match_string(match_groups[index])) index += 1 keywords = match_groups[index].lstrip(' ') - set_match_string(items_found, TAG_KEYWORDS, normalize_match_string(keywords[0:keywords.find(' ')])) + set_match_string(items_found, TAG_KEYWORDS, normalize_match_string( + keywords[0:keywords.find(' ')])) index += 1 - set_match_string(items_found, TAG_PP_REGISTRATOR_SIMPLIFIED, 'BSI') + set_match_string( + items_found, TAG_PP_REGISTRATOR_SIMPLIFIED, 'BSI') if rule[0] == HEADER_TYPE.ANSSI_TYPE1: - set_match_string(items_found, TAG_PP_TITLE, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_TITLE, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_VERSION_NUMBER, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_VERSION_NUMBER, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_DATE, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_DATE, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_REGISTRATOR, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_REGISTRATOR, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_SPONSOR, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_SPONSOR, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_EDITOR, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_EDITOR, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_REVIEWER, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_REVIEWER, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_CC_VERSION, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_CC_VERSION, + normalize_match_string(match_groups[index])) index += 1 level = match_groups[index].lstrip(' ') - set_match_string(items_found, TAG_CC_SECURITY_LEVEL, normalize_match_string(level[0:level.find(' ')])) + set_match_string(items_found, TAG_CC_SECURITY_LEVEL, normalize_match_string( + level[0:level.find(' ')])) index += 1 - set_match_string(items_found, TAG_PP_REGISTRATOR_SIMPLIFIED, 'ANSSI') + set_match_string( + items_found, TAG_PP_REGISTRATOR_SIMPLIFIED, 'ANSSI') if rule[0] == HEADER_TYPE.ANSSI_TYPE2: - set_match_string(items_found, TAG_PP_TITLE, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_TITLE, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_VERSION_NUMBER, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_VERSION_NUMBER, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_AUTHORS, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_AUTHORS, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_CC_SECURITY_LEVEL, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_CC_SECURITY_LEVEL, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_REGISTRATOR, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_REGISTRATOR, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_CC_VERSION, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_CC_VERSION, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_KEYWORDS, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_KEYWORDS, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_REGISTRATOR_SIMPLIFIED, 'ANSSI') + set_match_string( + items_found, TAG_PP_REGISTRATOR_SIMPLIFIED, 'ANSSI') if rule[0] == HEADER_TYPE.ANSSI_TYPE3: - set_match_string(items_found, TAG_PP_TITLE, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_TITLE, + normalize_match_string(match_groups[index])) index += 1 # todo: parse if multiple pp ids are present - set_match_string(items_found, TAG_PP_ID, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_ID, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_EDITOR, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_EDITOR, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_DATE, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_DATE, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_VERSION_NUMBER, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_VERSION_NUMBER, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_SPONSOR, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_SPONSOR, + normalize_match_string(match_groups[index])) index += 1 ccversion = match_groups[index].lstrip(' ') - set_match_string(items_found, TAG_CC_VERSION, normalize_match_string(ccversion[0:ccversion.find(' ')])) + set_match_string(items_found, TAG_CC_VERSION, normalize_match_string( + ccversion[0:ccversion.find(' ')])) index += 1 - set_match_string(items_found, TAG_PP_REGISTRATOR_SIMPLIFIED, 'ANSSI') + set_match_string( + items_found, TAG_PP_REGISTRATOR_SIMPLIFIED, 'ANSSI') if rule[0] == HEADER_TYPE.DCSSI_TYPE1: - set_match_string(items_found, TAG_PP_TITLE, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_TITLE, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_ID, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_ID, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_VERSION_NUMBER, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_VERSION_NUMBER, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_DATE, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_DATE, + normalize_match_string(match_groups[index])) index += 1 author = match_groups[index].lstrip(' ') - set_match_string(items_found, TAG_PP_AUTHORS, normalize_match_string(author[0:author.find(' ')])) + set_match_string(items_found, TAG_PP_AUTHORS, normalize_match_string( + author[0:author.find(' ')])) index += 1 - set_match_string(items_found, TAG_PP_REGISTRATOR_SIMPLIFIED, 'DCSSI') + set_match_string( + items_found, TAG_PP_REGISTRATOR_SIMPLIFIED, 'DCSSI') if rule[0] == HEADER_TYPE.DCSSI_TYPE2: - set_match_string(items_found, TAG_PP_TITLE, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_TITLE, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_AUTHORS, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_AUTHORS, + normalize_match_string(match_groups[index])) index += 1 version = match_groups[index].lstrip(' ') - set_match_string(items_found, TAG_PP_VERSION_NUMBER, normalize_match_string(version[0:version.find(' ')])) + set_match_string(items_found, TAG_PP_VERSION_NUMBER, normalize_match_string( + version[0:version.find(' ')])) index += 1 - set_match_string(items_found, TAG_PP_REGISTRATOR_SIMPLIFIED, 'DCSSI') + set_match_string( + items_found, TAG_PP_REGISTRATOR_SIMPLIFIED, 'DCSSI') if rule[0] == HEADER_TYPE.FRONT_DCSSI_TYPE3 or rule[0] == HEADER_TYPE.FRONT_DCSSI_TYPE4: - set_match_string(items_found, TAG_PP_TITLE, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_TITLE, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_DATE, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_DATE, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_ID, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_ID, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_VERSION_NUMBER, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_VERSION_NUMBER, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_ID_REGISTRATOR, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_ID_REGISTRATOR, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_REGISTRATOR_SIMPLIFIED, 'DCSSI') + set_match_string( + items_found, TAG_PP_REGISTRATOR_SIMPLIFIED, 'DCSSI') if rule[0] == HEADER_TYPE.DCSSI_TYPE5: - set_match_string(items_found, TAG_PP_TITLE, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_TITLE, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_AUTHORS, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_AUTHORS, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_VERSION_NUMBER, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_VERSION_NUMBER, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_DATE, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_DATE, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_SPONSOR, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_SPONSOR, + normalize_match_string(match_groups[index])) index += 1 ccversion = match_groups[index].lstrip(' ') - set_match_string(items_found, TAG_CC_VERSION, normalize_match_string(ccversion[0:ccversion.find(' ')])) + set_match_string(items_found, TAG_CC_VERSION, normalize_match_string( + ccversion[0:ccversion.find(' ')])) index += 1 if rule[0] == HEADER_TYPE.DCSSI_TYPE6: - set_match_string(items_found, TAG_PP_TITLE, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_TITLE, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_AUTHORS, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_AUTHORS, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_VERSION_NUMBER, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_VERSION_NUMBER, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_DATE, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_DATE, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_PP_SPONSOR, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_PP_SPONSOR, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_CC_VERSION, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_CC_VERSION, + normalize_match_string(match_groups[index])) index += 1 - set_match_string(items_found, TAG_CC_SECURITY_LEVEL, normalize_match_string(match_groups[index])) + set_match_string(items_found, TAG_CC_SECURITY_LEVEL, + normalize_match_string(match_groups[index])) index += 1 - - set_match_string(items_found, TAG_PP_REGISTRATOR_SIMPLIFIED, 'DCSSI') + set_match_string( + items_found, TAG_PP_REGISTRATOR_SIMPLIFIED, 'DCSSI') if no_match_yet: files_without_match.append(file_name) @@ -913,7 +1090,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, write_output_file=True): pp_items_found, pp_files_without_match = search_pp_only_headers(walk_dir) print('*** Files without detected protection profiles header') @@ -924,18 +1101,20 @@ def extract_protectionprofiles_frontpage(walk_dir, 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)) + write_file.write(json.dumps( + pp_items_found, indent=4, sort_keys=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, fragments_dir, 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): os.makedirs(fragments_dir) - + print("***EXTRACT KEYWORDS***") all_items_found = {} - cert_id = {} + # cert_id = {} for file_name in search_files(walk_dir): if not os.path.isfile(file_name): continue @@ -943,32 +1122,39 @@ def extract_certificates_keywords(walk_dir, fragments_dir, file_prefix, write_ou if file_ext != '.txt': continue - print('*** {} ***'.format(file_name)) + # print('*** {} ***'.format(file_name)) + file_cert_name = os.path.splitext( + os.path.splitext(os.path.basename(file_name))[0])[0] # parse certificate, return all matches - all_items_found[file_name], modified_cert_file = parse_cert_file(file_name, rules, -1) + all_items_found[file_cert_name], modified_cert_file = parse_cert_file( + file_name, fips_rules if fips_items else rules, -1, should_censure_right_away=should_censure_right_away, + fips_items=fips_items) # try to establish the certificate id of the current certificate - cert_id[file_name] = estimate_cert_id(None, all_items_found[file_name], file_name) + # cert_id[file_cert_name] = estimate_cert_id( + # None, all_items_found[file_cert_name], file_name) # save report text with highlighted/replaced matches into \\fragments\\ directory - base_path = file_name[:file_name.rfind('\\')] - file_name_short = file_name[file_name.rfind('\\') + 1:] - target_file = '{}\\{}'.format(fragments_dir, file_name_short) - save_modified_cert_file(target_file, modified_cert_file[0], modified_cert_file[1]) + 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) + save_modified_cert_file( + target_file, modified_cert_file[0], modified_cert_file[1]) # store results into file with fixed name and also with time appendix 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)) + write_file.write(json.dumps( + all_items_found, indent=4, sort_keys=True)) - print('\nTotal matches found in separate files:') + # print('\nTotal matches found in separate files:') # print_total_matches_in_files(all_items_found_count) - print('\nFile name and estimated certificate ID:') + # print('\nFile name and estimated certificate ID:') # print_guessed_cert_id(cert_id) - #depricated_print_dot_graph_keywordsonly(['rules_cert_id'], all_items_found, cert_id, walk_dir, 'certid_graph_from_keywords.dot', True) + # depricated_print_dot_graph_keywordsonly(['rules_cert_id'], all_items_found, cert_id, walk_dir, 'certid_graph_from_keywords.dot', True) total_items_found = 0 for file_name in all_items_found: @@ -978,16 +1164,18 @@ def extract_certificates_keywords(walk_dir, fragments_dir, file_prefix, write_ou if PRINT_MATCHES: all_matches = [] for file_name in all_items_found: + print('*' * 10, "FILENAME:", file_name, '*' * 10) for rule_group in all_items_found[file_name].keys(): items_found = all_items_found[file_name][rule_group] for rule in items_found.keys(): for match in items_found[rule]: if match not in all_matches: - all_matches.append(match) + print(match) + # all_matches.append(match) sorted_all_matches = sorted(all_matches) - for match in sorted_all_matches: - print(match) + # for match in sorted_all_matches: + # print(match) # verify total matches found print('\nTotal matches found: {}'.format(total_items_found)) @@ -995,8 +1183,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, file_prefix, write_output_file=True): all_items_found = {} counter = 0 for file_name in search_files(walk_dir): @@ -1006,7 +1193,8 @@ def extract_certificates_pdfmeta(walk_dir, file_prefix, write_output_file = True if file_ext != '.pdf': continue - print('*** {} ***'.format(file_name)) + print("***EXTRACT PDFMETA***") + # print('*** {} ***'.format(file_name)) item = {} item['pdf_file_size_bytes'] = os.path.getsize(file_name) @@ -1036,15 +1224,17 @@ def extract_certificates_pdfmeta(walk_dir, file_prefix, write_output_file = True if counter % 100 == 0: # 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)) + 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)) 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)) + write_file.write(json.dumps( + all_items_found, indent=4, sort_keys=True)) return all_items_found @@ -1059,11 +1249,10 @@ def parse_product_updates(updates_chunk, link_files_updates): maintenance_reports = [] rule_with_maintainance_ST = '.*?([0-9]+?-[0-9]+?-[0-9]+?) (.+?)\<br style=' \ - '.*?\<a href="(.+?)" title="Maintenance Report' \ - '.*?\<a href="(.+?)" title="Maintenance ST' + '.*?\<a href="(.+?)" title="Maintenance Report' \ + '.*?\<a href="(.+?)" title="Maintenance ST' rule_without_maintainance_ST = '.*?([0-9]+?-[0-9]+?-[0-9]+?) (.+?)\<br style=' \ - '.*?\<a href="(.+?)" title="Maintenance Report'\ - + '.*?\<a href="(.+?)" title="Maintenance Report' if updates_chunk.find('Maintenance Report(s)') != -1: start_pos = updates_chunk.find('Maintenance Report(s)</div>') start_pos = updates_chunk.find('<li>', start_pos) @@ -1083,24 +1272,32 @@ def parse_product_updates(updates_chunk, link_files_updates): for m in re.finditer(rule, report_chunk): match_groups = m.groups() index_next_item = 0 - items_found['maintenance_date'] = normalize_match_string(match_groups[index_next_item]) + items_found['maintenance_date'] = normalize_match_string( + match_groups[index_next_item]) index_next_item += 1 - items_found['maintenance_item_name'] = normalize_match_string(match_groups[index_next_item]) + items_found['maintenance_item_name'] = normalize_match_string( + match_groups[index_next_item]) index_next_item += 1 - items_found['maintenance_link_cert_report'] = normalize_match_string(match_groups[index_next_item]) + items_found['maintenance_link_cert_report'] = normalize_match_string( + match_groups[index_next_item]) index_next_item += 1 if len(match_groups) > index_next_item: - items_found['maintenance_link_security_target'] = normalize_match_string(match_groups[index_next_item]) + items_found['maintenance_link_security_target'] = normalize_match_string( + match_groups[index_next_item]) index_next_item += 1 else: items_found['maintenance_link_security_target'] = "" - cert_file_name = extract_file_name_from_url(items_found['maintenance_link_cert_report']) + cert_file_name = extract_file_name_from_url( + items_found['maintenance_link_cert_report']) items_found['link_cert_report_file_name'] = cert_file_name - st_file_name = extract_file_name_from_url(items_found['maintenance_link_security_target']) + st_file_name = extract_file_name_from_url( + items_found['maintenance_link_security_target']) items_found['link_security_target_file_name'] = st_file_name - link_files_updates.append((items_found['maintenance_link_cert_report'], cert_file_name, items_found['maintenance_link_security_target'], st_file_name)) + link_files_updates.append( + (items_found['maintenance_link_cert_report'], cert_file_name, + items_found['maintenance_link_security_target'], st_file_name)) maintenance_reports.append(items_found) @@ -1116,7 +1313,8 @@ def parse_security_level(security_level): # some augmented items found augm_chunk = security_level[start_pos:] augm_chunk += ' ' - rule = '\<br\>(.+?) ' # items are in form of <br>AVA_VLA.4 <br>AVA_MSU.3 ... + # items are in form of <br>AVA_VLA.4 <br>AVA_MSU.3 ... + rule = '\<br\>(.+?) ' for m in re.finditer(rule, augm_chunk): match_groups = m.groups() @@ -1126,10 +1324,12 @@ def parse_security_level(security_level): def extract_certificates_metadata_html(file_name): + print("***HTML METADATA***") + print(file_name) items_found_all = {} download_files_certs = [] download_files_updates = [] - print('*** {} ***'.format(file_name)) + # print('*** {} ***'.format(file_name)) whole_text = load_cert_html_file(file_name) @@ -1166,31 +1366,33 @@ def extract_certificates_metadata_html(file_name): # IMPORTANT: order regexes based on their specificity - the most specific goes first rules_cc_html = [ - (HEADER_TYPE.HEADER_FULL, '\<tr class=(?:""|"even")\>[ ]+\<td class="b"\>(.+?)\<a name="(.+?)" style=.+?\<!-- \<a href="(.+?)" title="Vendor\'s web site" target="_blank"\>(.+?)</a> -->' - '.+?\<a href="(.+?)" title="Certification Report:.+?" target="_blank" class="button2"\>Certification Report\</a\>' - '.+?\<a href="(.+?)" title="Security Target:.+?" target="_blank" class="button2">Security Target</a>' - '.+?\<!-- ------ ------ ------ Product Updates ------ ------ ------ --\>' - '(.+?)<!-- ------ ------ ------ END Product Updates ------ ------ ------ --\>' - '.+?\<!--end-product-cell--\>' - '.+?\<td style="text-align:center"\>\<span title=".+?"\>(.+?)\</span\>\</td\>' - '.+?\<td style="text-align:center"\>(.*?)\</td\>' - '[ ]+?\<td>(.+?)\</td\>'), + (HEADER_TYPE.HEADER_FULL, + '\<tr class=(?:""|"even")\>[ ]+\<td class="b"\>(.+?)\<a name="(.+?)" style=.+?\<!-- \<a href="(.+?)" title="Vendor\'s web site" target="_blank"\>(.+?)</a> -->' + '.+?\<a href="(.+?)" title="Certification Report:.+?" target="_blank" class="button2"\>Certification Report\</a\>' + '.+?\<a href="(.+?)" title="Security Target:.+?" target="_blank" class="button2">Security Target</a>' + '.+?\<!-- ------ ------ ------ Product Updates ------ ------ ------ --\>' + '(.+?)<!-- ------ ------ ------ END Product Updates ------ ------ ------ --\>' + '.+?\<!--end-product-cell--\>' + '.+?\<td style="text-align:center"\>\<span title=".+?"\>(.+?)\</span\>\</td\>' + '.+?\<td style="text-align:center"\>(.*?)\</td\>' + '[ ]+?\<td>(.+?)\</td\>'), - (HEADER_TYPE.HEADER_MISSING_VENDOR_WEB,'\<tr class=(?:""|"even")\>[ ]+\<td class="b"\>(.+?)\<a name="(.+?)" style=.+?' - '.+?\<a href="(.+?)" title="Certification Report:.+?" target="_blank" class="button2"\>Certification Report\</a\>' - '.+?\<a href="(.+?)" title="Security Target:.+?" target="_blank" class="button2">Security Target</a>' - '.+?\<!-- ------ ------ ------ Product Updates ------ ------ ------ --\>' - '(.+?)<!-- ------ ------ ------ END Product Updates ------ ------ ------ --\>' - '.+?\<!--end-product-cell--\>' - '.+?\<td style="text-align:center"\>\<span title=".+?"\>(.+?)\</span\>\</td\>' - '.+?\<td style="text-align:center"\>(.*?)\</td\>' - '[ ]+?\<td>(.+?)\</td\>'), + (HEADER_TYPE.HEADER_MISSING_VENDOR_WEB, + '\<tr class=(?:""|"even")\>[ ]+\<td class="b"\>(.+?)\<a name="(.+?)" style=.+?' + '.+?\<a href="(.+?)" title="Certification Report:.+?" target="_blank" class="button2"\>Certification Report\</a\>' + '.+?\<a href="(.+?)" title="Security Target:.+?" target="_blank" class="button2">Security Target</a>' + '.+?\<!-- ------ ------ ------ Product Updates ------ ------ ------ --\>' + '(.+?)<!-- ------ ------ ------ END Product Updates ------ ------ ------ --\>' + '.+?\<!--end-product-cell--\>' + '.+?\<td style="text-align:center"\>\<span title=".+?"\>(.+?)\</span\>\</td\>' + '.+?\<td style="text-align:center"\>(.*?)\</td\>' + '[ ]+?\<td>(.+?)\</td\>'), ] no_match_yet = True for rule in rules_cc_html: if not no_match_yet: - continue # search only the first match + continue # search only the first match rule_and_sep = rule[1] @@ -1198,50 +1400,67 @@ def extract_certificates_metadata_html(file_name): if no_match_yet: chunks_matched += 1 items_found = {} - #items_found_all.append(items_found) + # items_found_all.append(items_found) items_found[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[TAG_HEADER_MATCH_RULES]: - # items_found[TAG_HEADER_MATCH_RULES].append(rule[1]) + # if rule not in items_found[TAG_HEADER_MATCH_RULES]: + # items_found[TAG_HEADER_MATCH_RULES].append(rule[1]) match_groups = m.groups() index_next_item = 0 - items_found['cert_item_name'] = normalize_match_string(match_groups[index_next_item]) + items_found['cert_item_name'] = normalize_match_string( + match_groups[index_next_item]) index_next_item += 1 - items_found['cc_cert_item_html_id'] = normalize_match_string(match_groups[index_next_item]) + items_found['cc_cert_item_html_id'] = normalize_match_string( + match_groups[index_next_item]) cert_item_id = items_found['cc_cert_item_html_id'] index_next_item += 1 if not rule[0] == HEADER_TYPE.HEADER_MISSING_VENDOR_WEB: - items_found['company_site'] = normalize_match_string(match_groups[index_next_item]) + items_found['company_site'] = normalize_match_string( + match_groups[index_next_item]) index_next_item += 1 - items_found['company_name'] = normalize_match_string(match_groups[index_next_item]) + items_found['company_name'] = normalize_match_string( + match_groups[index_next_item]) index_next_item += 1 - items_found['link_cert_report'] = normalize_match_string(match_groups[index_next_item]) - cert_file_name = extract_file_name_from_url(items_found['link_cert_report']) + items_found['link_cert_report'] = normalize_match_string( + match_groups[index_next_item]) + cert_file_name = extract_file_name_from_url( + items_found['link_cert_report']) items_found['link_cert_report_file_name'] = cert_file_name index_next_item += 1 - items_found['link_security_target'] = normalize_match_string(match_groups[index_next_item]) - st_file_name = extract_file_name_from_url(items_found['link_security_target']) + items_found['link_security_target'] = normalize_match_string( + match_groups[index_next_item]) + st_file_name = extract_file_name_from_url( + items_found['link_security_target']) items_found['link_security_target_file_name'] = st_file_name - download_files_certs.append((items_found['link_cert_report'], cert_file_name, items_found['link_security_target'], st_file_name)) + download_files_certs.append( + ( + items_found['link_cert_report'], cert_file_name, items_found['link_security_target'], + st_file_name)) index_next_item += 1 - items_found['maintainance_updates'] = parse_product_updates(match_groups[index_next_item], download_files_updates) + items_found['maintainance_updates'] = parse_product_updates( + match_groups[index_next_item], download_files_updates) index_next_item += 1 - items_found['date_cert_issued'] = normalize_match_string(match_groups[index_next_item]) + items_found['date_cert_issued'] = normalize_match_string( + match_groups[index_next_item]) index_next_item += 1 - items_found['date_cert_expiration'] = normalize_match_string(match_groups[index_next_item]) + items_found['date_cert_expiration'] = normalize_match_string( + match_groups[index_next_item]) index_next_item += 1 - cc_security_level = normalize_match_string(match_groups[index_next_item]) - items_found['cc_security_level'], items_found['cc_security_level_augmented'] = parse_security_level(cc_security_level) + cc_security_level = normalize_match_string( + match_groups[index_next_item]) + items_found['cc_security_level'], items_found['cc_security_level_augmented'] = parse_security_level( + cc_security_level) index_next_item += 1 # prepare unique name for dictionary (file name is not enough as multiple records reference same cert) - item_unique_name = '{}__{}'.format(cert_file_name, cert_item_id) + item_unique_name = '{}__{}'.format( + cert_file_name, cert_item_id) if item_unique_name not in items_found_all.keys(): items_found_all[item_unique_name] = {} items_found_all[item_unique_name]['html_scan'] = items_found @@ -1253,7 +1472,8 @@ def extract_certificates_metadata_html(file_name): if no_match_yet: print('No match found in block #{}'.format(chunks_found)) - print('Chunks found: {}, Chunks matched: {}'.format(chunks_found, chunks_matched)) + print('Chunks found: {}, Chunks matched: {}'.format( + chunks_found, chunks_matched)) if chunks_found != chunks_matched: print('WARNING: not all chunks found were matched') @@ -1264,21 +1484,24 @@ def check_if_new_or_same(target_dict, target_key, new_value): if target_key in target_dict.keys(): if target_dict[target_key] != new_value: if STOP_ON_UNEXPECTED_NUMS: - raise ValueError('ERROR: Stopping on unexpected intermediate numbers') + raise ValueError( + 'ERROR: Stopping on unexpected intermediate numbers') def extract_certificates_metadata_csv(file_name): + print("***CSV METADATA***") + print(file_name) items_found_all = {} expected_columns = -1 with open(file_name, errors=FILE_ERRORS_STRATEGY) as csv_file: - print('*** {} ***'.format(file_name)) + # print('*** {} ***'.format(file_name)) csv_reader = csv.reader(csv_file, delimiter=',') line_count = 0 no_further_maintainance = True for row in csv_reader: if line_count == 0: expected_columns = len(row) - #print(f'Column names are {", ".join(row)}') + # print(f'Column names are {", ".join(row)}') line_count += 1 else: if no_further_maintainance: @@ -1286,12 +1509,14 @@ def extract_certificates_metadata_csv(file_name): if len(row) == 0: break if len(row) != expected_columns: - print('WARNING: Incorrect number of columns in row {} (likely separator , in item name), going to fix...'.format(line_count)) + print( + 'WARNING: Incorrect number of columns in row {} (likely separator , in item name), going to fix...'.format( + line_count)) # trying to fix if row[4].find('EAL') == -1: - row[1] = row[1] + row[2] # fix name - row.remove(row[2]) # remove second part of name - if len(row[11]) > 0: # test if reassesment is filled + row[1] = row[1] + row[2] # fix name + row.remove(row[2]) # remove second part of name + if len(row[11]) > 0: # test if reassesment is filled if row[13].find('http://') != -1: # name row[11] = row[11] + row[12] @@ -1306,42 +1531,65 @@ def extract_certificates_metadata_csv(file_name): items_found['raw_csv_line'] = str(row) index_next_item = 0 - check_if_new_or_same(items_found, 'cc_category', normalize_match_string(row[index_next_item])) - items_found['cc_category'] = normalize_match_string(row[index_next_item]) + check_if_new_or_same( + items_found, 'cc_category', normalize_match_string(row[index_next_item])) + items_found['cc_category'] = normalize_match_string( + row[index_next_item]) index_next_item += 1 - check_if_new_or_same(items_found, 'cert_item_name', normalize_match_string(row[index_next_item])) - items_found['cert_item_name'] = normalize_match_string(row[index_next_item]) + check_if_new_or_same( + items_found, 'cert_item_name', normalize_match_string(row[index_next_item])) + items_found['cert_item_name'] = normalize_match_string( + row[index_next_item]) index_next_item += 1 - check_if_new_or_same(items_found, 'cc_manufacturer', normalize_match_string(row[index_next_item])) - items_found['cc_manufacturer'] = normalize_match_string(row[index_next_item]) + check_if_new_or_same( + items_found, 'cc_manufacturer', normalize_match_string(row[index_next_item])) + items_found['cc_manufacturer'] = normalize_match_string( + row[index_next_item]) index_next_item += 1 - check_if_new_or_same(items_found, 'cc_scheme', normalize_match_string(row[index_next_item])) - items_found['cc_scheme'] = normalize_match_string(row[index_next_item]) + check_if_new_or_same( + items_found, 'cc_scheme', normalize_match_string(row[index_next_item])) + items_found['cc_scheme'] = normalize_match_string( + row[index_next_item]) index_next_item += 1 - check_if_new_or_same(items_found, 'cc_security_level', normalize_match_string(row[index_next_item])) - items_found['cc_security_level'] = normalize_match_string(row[index_next_item]) + check_if_new_or_same( + items_found, 'cc_security_level', normalize_match_string(row[index_next_item])) + items_found['cc_security_level'] = normalize_match_string( + row[index_next_item]) index_next_item += 1 - check_if_new_or_same(items_found, 'cc_protection_profiles', normalize_match_string(row[index_next_item])) - items_found['cc_protection_profiles'] = normalize_match_string(row[index_next_item]) + check_if_new_or_same( + items_found, 'cc_protection_profiles', normalize_match_string(row[index_next_item])) + items_found['cc_protection_profiles'] = normalize_match_string( + row[index_next_item]) index_next_item += 1 - check_if_new_or_same(items_found, 'cc_certification_date', normalize_match_string(row[index_next_item])) - items_found['cc_certification_date'] = normalize_match_string(row[index_next_item]) + check_if_new_or_same( + items_found, 'cc_certification_date', normalize_match_string(row[index_next_item])) + items_found['cc_certification_date'] = normalize_match_string( + row[index_next_item]) index_next_item += 1 - check_if_new_or_same(items_found, 'cc_archived_date', normalize_match_string(row[index_next_item])) - items_found['cc_archived_date'] = normalize_match_string(row[index_next_item]) + check_if_new_or_same( + items_found, 'cc_archived_date', normalize_match_string(row[index_next_item])) + items_found['cc_archived_date'] = normalize_match_string( + row[index_next_item]) index_next_item += 1 - check_if_new_or_same(items_found, 'link_cert_report', normalize_match_string(row[index_next_item])) - items_found['link_cert_report'] = normalize_match_string(row[index_next_item]) + check_if_new_or_same( + items_found, 'link_cert_report', normalize_match_string(row[index_next_item])) + items_found['link_cert_report'] = normalize_match_string( + row[index_next_item]) link_cert_report = items_found['link_cert_report'] - cert_file_name = extract_file_name_from_url(items_found['link_cert_report']) - check_if_new_or_same(items_found, 'link_cert_report_file_name', cert_file_name) + cert_file_name = extract_file_name_from_url( + items_found['link_cert_report']) + check_if_new_or_same( + items_found, 'link_cert_report_file_name', cert_file_name) items_found['link_cert_report_file_name'] = cert_file_name cert_file_name = items_found['link_cert_report_file_name'] index_next_item += 1 - check_if_new_or_same(items_found, 'link_security_target', normalize_match_string(row[index_next_item])) - items_found['link_security_target'] = normalize_match_string(row[index_next_item]) - st_file_name = extract_file_name_from_url(items_found['link_security_target']) + check_if_new_or_same( + items_found, 'link_security_target', normalize_match_string(row[index_next_item])) + items_found['link_security_target'] = normalize_match_string( + row[index_next_item]) + st_file_name = extract_file_name_from_url( + items_found['link_security_target']) items_found['link_security_target_file_name'] = st_file_name index_next_item += 1 @@ -1349,13 +1597,17 @@ def extract_certificates_metadata_csv(file_name): items_found['maintainance_updates'] = [] maintainance = {} - maintainance['cc_maintainance_date'] = normalize_match_string(row[index_next_item]) + maintainance['cc_maintainance_date'] = normalize_match_string( + row[index_next_item]) index_next_item += 1 - maintainance['cc_maintainance_title'] = normalize_match_string(row[index_next_item]) + maintainance['cc_maintainance_title'] = normalize_match_string( + row[index_next_item]) index_next_item += 1 - maintainance['cc_maintainance_report_link'] = normalize_match_string(row[index_next_item]) + maintainance['cc_maintainance_report_link'] = normalize_match_string( + row[index_next_item]) index_next_item += 1 - maintainance['cc_maintainance_st_link'] = normalize_match_string(row[index_next_item]) + maintainance['cc_maintainance_st_link'] = normalize_match_string( + row[index_next_item]) index_next_item += 1 # add this maintainance to parent item only when not empty if len(maintainance['cc_maintainance_title']) > 0: @@ -1365,14 +1617,16 @@ def extract_certificates_metadata_csv(file_name): # prepare unique name for dictionary (file name is not enough as multiple records reference same cert) cert_file_name = cert_file_name.replace('%20', ' ') item_unique_name = cert_file_name - item_unique_name = '{}__{}'.format(cert_file_name, line_count) + item_unique_name = '{}__{}'.format( + cert_file_name, line_count) if item_unique_name not in items_found_all.keys(): items_found_all[item_unique_name] = {} items_found_all[item_unique_name]['csv_scan'] = items_found else: print(' ERROR: {} already in'.format(cert_file_name)) if STOP_ON_UNEXPECTED_NUMS: - raise ValueError('ERROR: Stopping as value is not unique') + raise ValueError( + 'ERROR: Stopping as value is not unique') line_count += 1 @@ -1380,7 +1634,8 @@ def extract_certificates_metadata_csv(file_name): def fix_pp_url(original_url): - if original_url.find('/epfiles/') != -1: # links to pp are incorrect - epfiles instead ppfiles + # links to pp are incorrect - epfiles instead ppfiles + if original_url.find('/epfiles/') != -1: original_url = original_url.replace('/epfiles/', '/ppfiles/') original_url = original_url.replace('http://', 'https://') original_url = original_url.replace(':443', '') @@ -1406,7 +1661,9 @@ def extract_pp_metadata_csv(file_name): if len(row) == 0: break if len(row) != expected_columns: - print('WARNING: Incorrect number of columns in row {} (likely separator , in item name), going to fix...'.format(line_count)) + print( + 'WARNING: Incorrect number of columns in row {} (likely separator , in item name), going to fix...'.format( + line_count)) # trying to fix if len(row) == expected_columns + 2: row[9] = row[9] + row[10] + row[11] @@ -1425,45 +1682,69 @@ def extract_pp_metadata_csv(file_name): items_found['raw_csv_line'] = str(row) index_next_item = 0 - check_if_new_or_same(items_found, 'cc_category', normalize_match_string(row[index_next_item])) - items_found['cc_category'] = normalize_match_string(row[index_next_item]) + check_if_new_or_same( + items_found, 'cc_category', normalize_match_string(row[index_next_item])) + items_found['cc_category'] = normalize_match_string( + row[index_next_item]) index_next_item += 1 - check_if_new_or_same(items_found, 'cc_pp_name', normalize_match_string(row[index_next_item])) - items_found['cc_pp_name'] = normalize_match_string(row[index_next_item]) + check_if_new_or_same( + items_found, 'cc_pp_name', normalize_match_string(row[index_next_item])) + items_found['cc_pp_name'] = normalize_match_string( + row[index_next_item]) index_next_item += 1 - check_if_new_or_same(items_found, 'cc_pp_version', normalize_match_string(row[index_next_item])) - items_found['cc_pp_version'] = normalize_match_string(row[index_next_item]) + check_if_new_or_same( + items_found, 'cc_pp_version', normalize_match_string(row[index_next_item])) + items_found['cc_pp_version'] = normalize_match_string( + row[index_next_item]) index_next_item += 1 - check_if_new_or_same(items_found, 'cc_security_level', normalize_match_string(row[index_next_item])) - items_found['cc_security_level'] = normalize_match_string(row[index_next_item]) + check_if_new_or_same( + items_found, 'cc_security_level', normalize_match_string(row[index_next_item])) + items_found['cc_security_level'] = normalize_match_string( + row[index_next_item]) index_next_item += 1 - check_if_new_or_same(items_found, 'cc_certification_date', normalize_match_string(row[index_next_item])) - items_found['cc_certification_date'] = normalize_match_string(row[index_next_item]) + check_if_new_or_same( + items_found, 'cc_certification_date', normalize_match_string(row[index_next_item])) + items_found['cc_certification_date'] = normalize_match_string( + row[index_next_item]) index_next_item += 1 - check_if_new_or_same(items_found, 'cc_archived_date', normalize_match_string(row[index_next_item])) - items_found['cc_archived_date'] = normalize_match_string(row[index_next_item]) + check_if_new_or_same( + items_found, 'cc_archived_date', normalize_match_string(row[index_next_item])) + items_found['cc_archived_date'] = normalize_match_string( + row[index_next_item]) index_next_item += 1 - check_if_new_or_same(items_found, 'link_pp_report', normalize_match_string(row[index_next_item])) - items_found['link_pp_report'] = normalize_match_string(row[index_next_item]) - items_found['link_pp_report'] = fix_pp_url(items_found['link_pp_report']) + check_if_new_or_same( + items_found, 'link_pp_report', normalize_match_string(row[index_next_item])) + items_found['link_pp_report'] = normalize_match_string( + row[index_next_item]) + items_found['link_pp_report'] = fix_pp_url( + items_found['link_pp_report']) index_next_item += 1 - pp_report_file_name = extract_file_name_from_url(items_found['link_pp_report']) - check_if_new_or_same(items_found, 'link_pp_document', normalize_match_string(row[index_next_item])) - items_found['link_pp_document'] = normalize_match_string(row[index_next_item]) - items_found['link_pp_document'] = fix_pp_url(items_found['link_pp_document']) + pp_report_file_name = extract_file_name_from_url( + items_found['link_pp_report']) + check_if_new_or_same( + items_found, 'link_pp_document', normalize_match_string(row[index_next_item])) + items_found['link_pp_document'] = normalize_match_string( + row[index_next_item]) + items_found['link_pp_document'] = fix_pp_url( + items_found['link_pp_document']) index_next_item += 1 - pp_document_file_name = extract_file_name_from_url(items_found['link_pp_document']) + pp_document_file_name = extract_file_name_from_url( + items_found['link_pp_document']) if 'maintainance_updates' not in items_found: items_found['maintainance_updates'] = [] maintainance = {} - maintainance['cc_pp_maintainance_date'] = normalize_match_string(row[index_next_item]) + maintainance['cc_pp_maintainance_date'] = normalize_match_string( + row[index_next_item]) index_next_item += 1 - maintainance['cc_pp_maintainance_title'] = normalize_match_string(row[index_next_item]) + maintainance['cc_pp_maintainance_title'] = normalize_match_string( + row[index_next_item]) index_next_item += 1 - maintainance['cc_maintainance_report_link'] = normalize_match_string(row[index_next_item]) - maintainance['cc_maintainance_report_link'] = fix_pp_url(maintainance['cc_maintainance_report_link']) + maintainance['cc_maintainance_report_link'] = normalize_match_string( + row[index_next_item]) + maintainance['cc_maintainance_report_link'] = fix_pp_url( + maintainance['cc_maintainance_report_link']) index_next_item += 1 # add this maintainance to parent item only when not empty @@ -1472,16 +1753,20 @@ def extract_pp_metadata_csv(file_name): if no_further_maintainance: # prepare unique name for dictionary (file name is not enough as multiple records reference same cert) - pp_document_file_name = pp_document_file_name.replace('%20', ' ') + pp_document_file_name = pp_document_file_name.replace( + '%20', ' ') item_unique_name = pp_document_file_name - item_unique_name = '{}__{}'.format(pp_document_file_name, line_count) + item_unique_name = '{}__{}'.format( + pp_document_file_name, line_count) if item_unique_name not in items_found_all.keys(): items_found_all[item_unique_name] = {} items_found_all[item_unique_name]['csv_scan'] = items_found else: - print(' ERROR: {} already in'.format(pp_document_file_name)) + print(' ERROR: {} already in'.format( + pp_document_file_name)) if STOP_ON_UNEXPECTED_NUMS: - raise ValueError('ERROR: Stopping as value is not unique') + raise ValueError( + 'ERROR: Stopping as value is not unique') # save download links for basic protection profile download_files_certs.append((items_found['link_pp_report'], pp_report_file_name, @@ -1489,8 +1774,10 @@ def extract_pp_metadata_csv(file_name): # save download links for maintainance updates protection profile for item in items_found['maintainance_updates']: if item['cc_maintainance_report_link'] != "": - pp_maintainainace_file_name = extract_file_name_from_url(item['cc_maintainance_report_link']) - download_files_maintainance.append((item['cc_maintainance_report_link'], pp_maintainainace_file_name)) + pp_maintainainace_file_name = extract_file_name_from_url( + item['cc_maintainance_report_link']) + download_files_maintainance.append( + (item['cc_maintainance_report_link'], pp_maintainainace_file_name)) line_count += 1 @@ -1509,10 +1796,12 @@ def generate_download_script(file_name, certs_dir, targets_dir, base_url, downlo if file_name_short_web.find(base_url) != -1: # base url already included - write_file.write('curl \"{}\" -o \"{}\"\n'.format(file_name_short_web, cert[1])) + write_file.write( + 'curl \"{}\" -o \"{}\"\n'.format(file_name_short_web, cert[1])) else: # insert base url - write_file.write('curl \"{}{}\" -o \"{}\"\n'.format(base_url, file_name_short_web, cert[1])) + write_file.write( + 'curl \"{}{}\" -o \"{}\"\n'.format(base_url, file_name_short_web, cert[1])) write_file.write('{} \"{}\"\n\n'.format(PDF2TEXT_CONVERT, cert[1])) if len(download_files_certs) > 0 and len(cert) > 2: @@ -1526,48 +1815,60 @@ def generate_download_script(file_name, certs_dir, targets_dir, base_url, downlo file_name_short_web = cert[2].replace(' ', '%%20') if file_name_short_web.find(base_url) != -1: # base url already included - write_file.write('curl \"{}\" -o \"{}\"\n'.format(file_name_short_web, cert[3])) + write_file.write( + 'curl \"{}\" -o \"{}\"\n'.format(file_name_short_web, cert[3])) else: # insert base url - write_file.write('curl \"{}{}\" -o \"{}\"\n'.format(base_url, file_name_short_web, cert[3])) - write_file.write('{} \"{}\"\n\n'.format(PDF2TEXT_CONVERT, cert[3])) + write_file.write( + 'curl \"{}{}\" -o \"{}\"\n'.format(base_url, file_name_short_web, cert[3])) + write_file.write('{} \"{}\"\n\n'.format( + PDF2TEXT_CONVERT, cert[3])) -def extract_certificates_html(base_dir, write_output_file = True): +def extract_certificates_html(base_dir, write_output_file=True): file_name = '{}cc_products_active.html'.format(base_dir) - items_found_all_active, download_files_certs, download_files_updates = extract_certificates_metadata_html(file_name) + 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: - write_file.write(json.dumps(items_found_all_active, indent=4, sort_keys=True)) + write_file.write(json.dumps( + items_found_all_active, 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) + 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 = '{}cc_products_archived.html'.format(base_dir) - items_found_all_archived, download_files_certs, download_files_updates = extract_certificates_metadata_html(file_name) + 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: - write_file.write(json.dumps(items_found_all_archived, indent=4, sort_keys=True)) + write_file.write(json.dumps( + items_found_all_archived, 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', 'certs', 'targets', CC_WEB_URL, download_files_updates) + generate_download_script('download_archived_certs.bat', + 'certs', 'targets', CC_WEB_URL, download_files_certs) + generate_download_script('download_archived_updates.bat', + 'certs', 'targets', CC_WEB_URL, download_files_updates) 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: - write_file.write(json.dumps(items_found_all, indent=4, sort_keys=True)) + write_file.write(json.dumps( + items_found_all, indent=4, sort_keys=True)) return items_found_all -def extract_certificates_csv(base_dir, write_output_file = True): +def extract_certificates_csv(base_dir, write_output_file=True): file_name = '{}cc_products_active.csv'.format(base_dir) items_found_all_active = extract_certificates_metadata_csv(file_name) for item in items_found_all_active.keys(): @@ -1582,34 +1883,41 @@ def extract_certificates_csv(base_dir, write_output_file = 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)) + write_file.write(json.dumps( + items_found_all, indent=4, sort_keys=True)) return items_found_all -def extract_protectionprofiles_csv(base_dir, write_output_file = True): +def extract_protectionprofiles_csv(base_dir, write_output_file=True): file_name = '{}cc_pp_active.csv'.format(base_dir) - items_found_all_active, download_files_pp, download_files_pp_updates = extract_pp_metadata_csv(file_name) + 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(): items_found_all_active[item]['csv_scan']['cert_status'] = 'active' - generate_download_script('download_active_pp.bat', 'pp_report', 'pp', CC_WEB_URL, download_files_pp) - generate_download_script('download_active_pp_updates.bat', 'pp_updates', '', CC_WEB_URL, download_files_pp_updates) - + generate_download_script('download_active_pp.bat', + 'pp_report', 'pp', CC_WEB_URL, download_files_pp) + 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) - items_found_all_archived, download_files_pp, download_files_pp_updates = extract_pp_metadata_csv(file_name) + 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(): items_found_all_archived[item]['csv_scan']['cert_status'] = 'archived' - generate_download_script('download_archived_pp.bat', 'pp_report', 'pp', CC_WEB_URL, download_files_pp) - generate_download_script('download_archived_pp_updates.bat', 'pp_updates', '', CC_WEB_URL, download_files_pp_updates) + generate_download_script('download_archived_pp.bat', + 'pp_report', 'pp', CC_WEB_URL, download_files_pp) + generate_download_script('download_archived_pp_updates.bat', + 'pp_updates', '', CC_WEB_URL, download_files_pp_updates) 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: - write_file.write(json.dumps(items_found_all, indent=4, sort_keys=True)) + write_file.write(json.dumps( + items_found_all, indent=4, sort_keys=True)) return items_found_all @@ -1621,9 +1929,11 @@ def check_expected_cert_results(all_html, all_csv, all_front, all_keywords, all_ MIN_ITEMS_FOUND_CSV = 4105 num_items = len(all_csv) if MIN_ITEMS_FOUND_CSV != num_items: - print('SANITY: different than expected number of CSV records found! ({} vs. {} expected)'.format(num_items, MIN_ITEMS_FOUND_CSV)) + print('SANITY: different than expected number of CSV records found! ({} vs. {} expected)'.format( + num_items, MIN_ITEMS_FOUND_CSV)) if STOP_ON_UNEXPECTED_NUMS: - raise ValueError('ERROR: Stopping on unexpected intermediate numbers') + raise ValueError( + 'ERROR: Stopping on unexpected intermediate numbers') # # HTML @@ -1631,9 +1941,11 @@ def check_expected_cert_results(all_html, all_csv, all_front, all_keywords, all_ MIN_ITEMS_FOUND_HTML = 4103 num_items = len(all_html) if MIN_ITEMS_FOUND_HTML != num_items: - print('SANITY: different than expected number of HTML records found! ({} vs. {} expected)'.format(num_items, MIN_ITEMS_FOUND_HTML)) + print('SANITY: different than expected number of HTML records found! ({} vs. {} expected)'.format( + num_items, MIN_ITEMS_FOUND_HTML)) if STOP_ON_UNEXPECTED_NUMS: - raise ValueError('ERROR: Stopping on unexpected intermediate numbers') + raise ValueError( + 'ERROR: Stopping on unexpected intermediate numbers') # # FRONTPAGE @@ -1641,9 +1953,11 @@ def check_expected_cert_results(all_html, all_csv, all_front, all_keywords, all_ MIN_ITEMS_FOUND_FRONTPAGE = 1369 num_items = len(all_front) if MIN_ITEMS_FOUND_FRONTPAGE != num_items: - print('SANITY: different than expected number of frontpage records found! ({} vs. {} expected)'.format(num_items, MIN_ITEMS_FOUND_FRONTPAGE)) + print('SANITY: different than expected number of frontpage records found! ({} vs. {} expected)'.format( + num_items, MIN_ITEMS_FOUND_FRONTPAGE)) if STOP_ON_UNEXPECTED_NUMS: - raise ValueError('ERROR: Stopping on unexpected intermediate numbers') + raise ValueError( + 'ERROR: Stopping on unexpected intermediate numbers') # # KEYWORDS @@ -1653,9 +1967,11 @@ def check_expected_cert_results(all_html, all_csv, all_front, all_keywords, all_ for file_name in all_keywords.keys(): total_items_found += count_num_items_found(all_keywords[file_name]) if MIN_ITEMS_FOUND_KEYWORDS != total_items_found: - print('SANITY: different than expected number of keywords found! ({} vs. {} expected)'.format(total_items_found, MIN_ITEMS_FOUND_KEYWORDS)) + print('SANITY: different than expected number of keywords found! ({} vs. {} expected)'.format( + total_items_found, MIN_ITEMS_FOUND_KEYWORDS)) if STOP_ON_UNEXPECTED_NUMS: - raise ValueError('ERROR: Stopping on unexpected intermediate numbers') + raise ValueError( + 'ERROR: Stopping on unexpected intermediate numbers') def check_expected_pp_results(all_html, all_csv, all_front, all_keywords): @@ -1665,9 +1981,11 @@ def check_expected_pp_results(all_html, all_csv, all_front, all_keywords): MIN_ITEMS_FOUND_CSV = 4105 num_items = len(all_csv) if MIN_ITEMS_FOUND_CSV != num_items: - print('SANITY: different than expected number of CSV records found! ({} vs. {} expected)'.format(num_items, MIN_ITEMS_FOUND_CSV)) + print('SANITY: different than expected number of CSV records found! ({} vs. {} expected)'.format( + num_items, MIN_ITEMS_FOUND_CSV)) if STOP_ON_UNEXPECTED_NUMS: - raise ValueError('ERROR: Stopping on unexpected intermediate numbers') + raise ValueError( + 'ERROR: Stopping on unexpected intermediate numbers') # # HTML @@ -1675,9 +1993,11 @@ def check_expected_pp_results(all_html, all_csv, all_front, all_keywords): MIN_ITEMS_FOUND_HTML = 4103 num_items = len(all_html) if MIN_ITEMS_FOUND_HTML != num_items: - print('SANITY: different than expected number of HTML records found! ({} vs. {} expected)'.format(num_items, MIN_ITEMS_FOUND_HTML)) + print('SANITY: different than expected number of HTML records found! ({} vs. {} expected)'.format( + num_items, MIN_ITEMS_FOUND_HTML)) if STOP_ON_UNEXPECTED_NUMS: - raise ValueError('ERROR: Stopping on unexpected intermediate numbers') + raise ValueError( + 'ERROR: Stopping on unexpected intermediate numbers') # # FRONTPAGE @@ -1685,9 +2005,11 @@ def check_expected_pp_results(all_html, all_csv, all_front, all_keywords): MIN_ITEMS_FOUND_FRONTPAGE = 1369 num_items = len(all_front) if MIN_ITEMS_FOUND_FRONTPAGE != num_items: - print('SANITY: different than expected number of frontpage records found! ({} vs. {} expected)'.format(num_items, MIN_ITEMS_FOUND_FRONTPAGE)) + print('SANITY: different than expected number of frontpage records found! ({} vs. {} expected)'.format( + num_items, MIN_ITEMS_FOUND_FRONTPAGE)) if STOP_ON_UNEXPECTED_NUMS: - raise ValueError('ERROR: Stopping on unexpected intermediate numbers') + raise ValueError( + 'ERROR: Stopping on unexpected intermediate numbers') # # KEYWORDS @@ -1697,10 +2019,11 @@ def check_expected_pp_results(all_html, all_csv, all_front, all_keywords): for file_name in all_keywords.keys(): total_items_found += count_num_items_found(all_keywords[file_name]) if MIN_ITEMS_FOUND_KEYWORDS != total_items_found: - print('SANITY: different than expected number of keywords found! ({} vs. {} expected)'.format(total_items_found, MIN_ITEMS_FOUND_KEYWORDS)) + print('SANITY: different than expected number of keywords found! ({} vs. {} expected)'.format( + total_items_found, MIN_ITEMS_FOUND_KEYWORDS)) if STOP_ON_UNEXPECTED_NUMS: - raise ValueError('ERROR: Stopping on unexpected intermediate numbers') - + raise ValueError( + 'ERROR: Stopping on unexpected intermediate numbers') def collate_certificates_data(all_html, all_csv, all_front, all_keywords, all_pdf_meta, file_name_key): @@ -1708,29 +2031,29 @@ def collate_certificates_data(all_html, all_csv, all_front, all_keywords, all_pd file_name_to_html_name_mapping = {} for long_file_name in all_html.keys(): - short_file_name = long_file_name[long_file_name.rfind('\\') + 1:] + short_file_name = long_file_name[long_file_name.rfind(os.sep) + 1:] if short_file_name != '': file_name_to_html_name_mapping[short_file_name] = long_file_name file_name_to_front_name_mapping = {} for long_file_name in all_front.keys(): - short_file_name = long_file_name[long_file_name.rfind('\\') + 1:] + short_file_name = long_file_name[long_file_name.rfind(os.sep) + 1:] if short_file_name != '': file_name_to_front_name_mapping[short_file_name] = long_file_name file_name_to_keywords_name_mapping = {} for long_file_name in all_keywords.keys(): - short_file_name = long_file_name[long_file_name.rfind('\\') + 1:] + short_file_name = long_file_name[long_file_name.rfind(os.sep) + 1:] if short_file_name != '': - file_name_to_keywords_name_mapping[short_file_name] = [long_file_name, 0] + file_name_to_keywords_name_mapping[short_file_name] = [ + long_file_name, 0] file_name_to_pdfmeta_name_mapping = {} for long_file_name in all_pdf_meta.keys(): - short_file_name = long_file_name[long_file_name.rfind('\\') + 1:] + short_file_name = long_file_name[long_file_name.rfind(os.sep) + 1:] if short_file_name != '': - file_name_to_pdfmeta_name_mapping[short_file_name] = [long_file_name, 0] - - + file_name_to_pdfmeta_name_mapping[short_file_name] = [ + long_file_name, 0] all_cert_items = all_csv # pair html data, csv data, front pages and keywords @@ -1739,9 +2062,10 @@ def collate_certificates_data(all_html, all_csv, all_front, all_keywords, all_pd file_name_pdf = file_name[:file_name.rfind('__')] file_name_txt = file_name_pdf[:file_name_pdf.rfind('.')] + '.txt' - #file_name_st = all_csv[file_name]['csv_scan']['link_security_target_file_name'] + # file_name_st = all_csv[file_name]['csv_scan']['link_security_target_file_name'] if is_in_dict(all_csv, [file_name, 'csv_scan', 'link_security_target']): - file_name_st = extract_file_name_from_url(all_csv[file_name]['csv_scan']['link_security_target']) + file_name_st = extract_file_name_from_url( + all_csv[file_name]['csv_scan']['link_security_target']) file_name_st_txt = file_name_st[:file_name_st.rfind('.')] + '.txt' else: file_name_st_txt = 'security_target_which_doesnt_exists' @@ -1757,25 +2081,33 @@ def collate_certificates_data(all_html, all_csv, all_front, all_keywords, all_pd if file_name_txt in file_name_to_html_name_mapping.keys(): all_cert_items[file_name]['html_scan'] = all_html[file_name_to_html_name_mapping[file_name_txt][0]] - file_name_to_html_name_mapping[file_name_txt][1] = 1 # was paired + file_name_to_html_name_mapping[file_name_txt][1] = 1 # was paired else: - print('WARNING: Corresponding HTML report not found for CSV item {}'.format(file_name)) + print('WARNING: Corresponding HTML report not found for CSV item {}'.format( + file_name)) if file_name_txt in file_name_to_front_name_mapping.keys(): all_cert_items[file_name]['frontpage_scan'] = all_front[file_name_to_front_name_mapping[file_name_txt]] frontpage_scan = all_front[file_name_to_front_name_mapping[file_name_txt]] if file_name_txt in file_name_to_keywords_name_mapping.keys(): - all_cert_items[file_name]['keywords_scan'] = all_keywords[file_name_to_keywords_name_mapping[file_name_txt][0]] - file_name_to_keywords_name_mapping[file_name_txt][1] = 1 # was paired + all_cert_items[file_name]['keywords_scan'] = all_keywords[ + file_name_to_keywords_name_mapping[file_name_txt][0]] + # was paired + file_name_to_keywords_name_mapping[file_name_txt][1] = 1 keywords_scan = all_keywords[file_name_to_keywords_name_mapping[file_name_txt][0]] if file_name_st_txt in file_name_to_keywords_name_mapping.keys(): - all_cert_items[file_name]['st_keywords_scan'] = all_keywords[file_name_to_keywords_name_mapping[file_name_st_txt][0]] - file_name_to_keywords_name_mapping[file_name_st_txt][1] = 1 # was paired + all_cert_items[file_name]['st_keywords_scan'] = all_keywords[ + file_name_to_keywords_name_mapping[file_name_st_txt][0]] + # was paired + file_name_to_keywords_name_mapping[file_name_st_txt][1] = 1 if file_name_pdf in file_name_to_pdfmeta_name_mapping.keys(): - all_cert_items[file_name]['pdfmeta_scan'] = all_pdf_meta[file_name_to_pdfmeta_name_mapping[file_name_pdf][0]] - file_name_to_pdfmeta_name_mapping[file_name_pdf][1] = 1 # was paired + all_cert_items[file_name]['pdfmeta_scan'] = all_pdf_meta[ + file_name_to_pdfmeta_name_mapping[file_name_pdf][0]] + # was paired + file_name_to_pdfmeta_name_mapping[file_name_pdf][1] = 1 else: print('ERROR: File {} not found in pdfmeta scan'.format(file_name_pdf)) - all_cert_items[file_name]['processed']['cert_id'] = estimate_cert_id(frontpage_scan, keywords_scan, file_name) + all_cert_items[file_name]['processed']['cert_id'] = estimate_cert_id( + frontpage_scan, keywords_scan, file_name) # pair pairing in maintainance updates for file_name in all_csv.keys(): @@ -1784,66 +2116,81 @@ def collate_certificates_data(all_html, all_csv, all_front, all_keywords, all_pd # process all maintainance updates for update in all_cert_items[file_name]['csv_scan']['maintainance_updates']: - file_name_pdf = extract_file_name_from_url(update['cc_maintainance_report_link']) + file_name_pdf = extract_file_name_from_url( + update['cc_maintainance_report_link']) file_name_txt = file_name_pdf[:file_name_pdf.rfind('.')] + '.txt' if is_in_dict(update, ['cc_maintainance_st_link']): - file_name_st = extract_file_name_from_url(update['cc_maintainance_st_link']) + file_name_st = extract_file_name_from_url( + update['cc_maintainance_st_link']) file_name_st_pdf = file_name_st file_name_st_txt = '' if len(file_name_st) > 0: - file_name_st_txt = file_name_st[:file_name_st.rfind('.')] + '.txt' + file_name_st_txt = file_name_st[:file_name_st.rfind( + '.')] + '.txt' else: file_name_st_pdf = 'file_name_which_doesnt_exists' file_name_st_txt = 'file_name_which_doesnt_exists' for file_and_id in all_keywords.keys(): - file_name_keyword_txt = file_and_id[file_and_id.rfind('\\') + 1:] + file_name_keyword_txt = file_and_id[file_and_id.rfind( + os.sep) + 1:] # in items extracted from html, names are in form of 'file_name.pdf__number' if file_name_keyword_txt == file_name_txt: pairing_found = True if file_name_txt in file_name_to_keywords_name_mapping.keys(): update['keywords_scan'] = all_keywords[file_name_to_keywords_name_mapping[file_name_txt][0]] if file_name_to_keywords_name_mapping[file_name_txt][1] == 1: - print('WARNING: {} already paired'.format(file_name_to_keywords_name_mapping[file_name_txt][0])) - file_name_to_keywords_name_mapping[file_name_txt][1] = 1 # was paired + print('WARNING: {} already paired'.format( + file_name_to_keywords_name_mapping[file_name_txt][0])) + # was paired + file_name_to_keywords_name_mapping[file_name_txt][1] = 1 if file_name_keyword_txt == file_name_st_txt: if file_name_st_txt in file_name_to_keywords_name_mapping.keys(): - update['st_keywords_scan'] = all_keywords[file_name_to_keywords_name_mapping[file_name_st_txt][0]] + update['st_keywords_scan'] = all_keywords[ + file_name_to_keywords_name_mapping[file_name_st_txt][0]] if file_name_to_keywords_name_mapping[file_name_st_txt][1] == 1: - print('WARNING: {} already paired'.format(file_name_to_keywords_name_mapping[file_name_st_txt][0])) - file_name_to_keywords_name_mapping[file_name_st_txt][1] = 1 # was paired + print('WARNING: {} already paired'.format( + file_name_to_keywords_name_mapping[file_name_st_txt][0])) + # was paired + file_name_to_keywords_name_mapping[file_name_st_txt][1] = 1 if not pairing_found: - print('WARNING: Corresponding keywords pairing not found for maintaince item {}'.format(file_name)) + print('WARNING: Corresponding keywords pairing not found for maintaince item {}'.format( + file_name)) for file_and_id in file_name_to_pdfmeta_name_mapping.keys(): - file_name_pdf = file_and_id[file_and_id.rfind('\\') + 1:] - file_name_pdfmeta_txt = file_name_pdf[:file_name_pdf.rfind('.')] + '.txt' + file_name_pdf = file_and_id[file_and_id.rfind(os.sep) + 1:] + file_name_pdfmeta_txt = file_name_pdf[:file_name_pdf.rfind( + '.')] + '.txt' # in items extracted from html, names are in form of 'file_name.pdf__number' if file_name_pdfmeta_txt == file_name_txt: pairing_found = True if file_name_pdf in file_name_to_pdfmeta_name_mapping.keys(): update['pdfmeta_scan'] = all_pdf_meta[file_name_to_pdfmeta_name_mapping[file_name_pdf][0]] if file_name_to_pdfmeta_name_mapping[file_name_pdf][1] == 1: - print('WARNING: {} already paired'.format(file_name_to_pdfmeta_name_mapping[file_name_pdf][0])) - file_name_to_pdfmeta_name_mapping[file_name_pdf][1] = 1 # was paired + print('WARNING: {} already paired'.format( + file_name_to_pdfmeta_name_mapping[file_name_pdf][0])) + # was paired + file_name_to_pdfmeta_name_mapping[file_name_pdf][1] = 1 if file_name_pdfmeta_txt == file_name_st_txt: if file_name_st_pdf in file_name_to_pdfmeta_name_mapping.keys(): update['st_pdfmeta_scan'] = all_pdf_meta[file_name_to_pdfmeta_name_mapping[file_name_st_pdf][0]] if file_name_to_pdfmeta_name_mapping[file_name_st_pdf][1] == 1: - print('WARNING: {} already paired'.format(file_name_to_pdfmeta_name_mapping[file_name_st_pdf][0])) - file_name_to_pdfmeta_name_mapping[file_name_st_pdf][1] = 1 # was paired + print('WARNING: {} already paired'.format( + file_name_to_pdfmeta_name_mapping[file_name_st_pdf][0])) + # was paired + file_name_to_pdfmeta_name_mapping[file_name_st_pdf][1] = 1 if not pairing_found: - print('WARNING: Corresponding pdfmeta pairing not found for maintaince item {}'.format(file_name)) - + print('WARNING: Corresponding pdfmeta pairing not found for maintaince item {}'.format( + file_name)) print('*** Files with keywords extracted, which were NOT matched to any CSV item:') for item in file_name_to_keywords_name_mapping: - if file_name_to_keywords_name_mapping[item][1] == 0: # not paired + if file_name_to_keywords_name_mapping[item][1] == 0: # not paired print(' {}'.format(file_name_to_keywords_name_mapping[item][0])) # display all record which were not paired @@ -1871,7 +2218,8 @@ def collate_certificates_data(all_html, all_csv, all_front, all_keywords, all_pd print('WARNING: {} no pdfmeta scan detected'.format(item)) num_pdfmeta_missing += 1 - print('Records without frontpage: {}\nRecords without keywords: {}\nRecords without pdfmeta: {}'.format(num_frontpage_missing, num_keywords_missing, num_pdfmeta_missing)) + print('Records without frontpage: {}\nRecords without keywords: {}\nRecords without pdfmeta: {}'.format( + num_frontpage_missing, num_keywords_missing, num_pdfmeta_missing)) return all_cert_items @@ -1914,7 +2262,7 @@ def process_certificates_data(all_cert_items, all_pp_items): # Manufacturer can be single, multiple, separated by - / and , # heuristics: if separated candidate manufacturer can be found in original list ( # => is sole manufacturer on another certificate => assumption of correct separation) - separators = [',', '/'] # , '/', ',', 'and'] + separators = [',', '/'] # , '/', ',', 'and'] multiple_manuf_detected = False for sep in separators: list_manuf = manuf.split(sep) @@ -1945,25 +2293,27 @@ def process_certificates_data(all_cert_items, all_pp_items): manuf_starts = {} already_reduced = {} - for manuf1 in sorted_manufacturers: # we are processing from the shorter to longer + for manuf1 in sorted_manufacturers: # we are processing from the shorter to longer if manuf1 == '': continue for manuf2 in sorted_manufacturers: if manuf1 != manuf2: if manuf2.startswith(manuf1): - print('Potential consolidation of manufacturers: {} vs. {}'.format(manuf1, manuf2)) + print('Potential consolidation of manufacturers: {} vs. {}'.format( + manuf1, manuf2)) if manuf1 not in manuf_starts: manuf_starts[manuf1] = set() manuf_starts[manuf1].add(manuf2) if manuf2 not in already_reduced: already_reduced[manuf2] = manuf1 else: - print(' Warning: \'{}\' prefixed by \'{}\' already reduced to \'{}\''.format(manuf2, manuf1, already_reduced[manuf2])) - + print(' Warning: \'{}\' prefixed by \'{}\' already reduced to \'{}\''.format( + manuf2, manuf1, already_reduced[manuf2])) # try to find manufacturers with multiple names and draw the map dot = Digraph(comment='Manufacturers naming simplifications') - dot.attr('graph', label='Manufacturers naming simplifications', labelloc='t', fontsize='30') + dot.attr('graph', label='Manufacturers naming simplifications', + labelloc='t', fontsize='30') dot.attr('node', style='filled') already_inserted_edges = [] for file_name in all_cert_items.keys(): @@ -1972,11 +2322,13 @@ def process_certificates_data(all_cert_items, all_pp_items): joint_manufacturer = cert['csv_scan']['cc_manufacturer'] if joint_manufacturer != '': for manuf in mapping_csvmanuf_separated[joint_manufacturer]: - simple_manuf = get_manufacturer_simple_name(manuf, already_reduced) + simple_manuf = get_manufacturer_simple_name( + manuf, already_reduced) if simple_manuf != manuf: edge_name = '{}<->{}'.format(simple_manuf, manuf) if edge_name not in already_inserted_edges: - dot.edge(simple_manuf, manuf, color='orange', style='solid') + dot.edge(simple_manuf, manuf, + color='orange', style='solid') already_inserted_edges.append(edge_name) # plot naming hierarchies @@ -1984,7 +2336,6 @@ def process_certificates_data(all_cert_items, all_pp_items): dot.render(file_name, view=False) print('{} pdf rendered'.format(file_name)) - # update dist with processed list of manufactures all_cert_items_keys = list(all_cert_items.keys()) for file_name in all_cert_items_keys: @@ -2003,10 +2354,12 @@ def process_certificates_data(all_cert_items, all_pp_items): # insert extracted manufacturers by simplified name simple_manufacturers = [] for manuf in mapping_csvmanuf_separated[manufacturer]: - simple_manufacturers.append(get_manufacturer_simple_name(manuf, already_reduced)) + simple_manufacturers.append( + get_manufacturer_simple_name(manuf, already_reduced)) cert['processed']['cc_manufacturer_simple_list'] = simple_manufacturers - cert['processed']['cc_manufacturer_simple'] = get_manufacturer_simple_name(manufacturer, already_reduced) + cert['processed']['cc_manufacturer_simple'] = get_manufacturer_simple_name( + manufacturer, already_reduced) # extract certification lab if is_in_dict(cert, ['frontpage_scan', 'cert_lab']): @@ -2035,20 +2388,30 @@ def process_certificates_data(all_cert_items, all_pp_items): def generate_basic_download_script(): with open('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('curl \"https://www.commoncriteriaportal.org/products/index.cfm?archived=1\" -o cc_products_archived.html\n\n') + file.write( + 'curl \"https://www.commoncriteriaportal.org/products/\" -o cc_products_active.html\n') + file.write( + 'curl \"https://www.commoncriteriaportal.org/products/index.cfm?archived=1\" -o cc_products_archived.html\n\n') - file.write('curl \"https://www.commoncriteriaportal.org/labs/\" -o cc_labs.html\n') + file.write( + 'curl \"https://www.commoncriteriaportal.org/labs/\" -o cc_labs.html\n') - file.write('curl \"https://www.commoncriteriaportal.org/products/certified_products.csv\" -o cc_products_active.csv\n') - file.write('curl \"https://www.commoncriteriaportal.org/products/certified_products-archived.csv\" -o cc_products_archived.csv\n\n') + file.write( + 'curl \"https://www.commoncriteriaportal.org/products/certified_products.csv\" -o cc_products_active.csv\n') + file.write( + 'curl \"https://www.commoncriteriaportal.org/products/certified_products-archived.csv\" -o cc_products_archived.csv\n\n') - file.write('curl \"https://www.commoncriteriaportal.org/pps/\" -o cc_pp_active.html\n') - file.write('curl \"https://www.commoncriteriaportal.org/pps/collaborativePP.cfm?cpp=1\" -o cc_pp_collaborative.html\n') - file.write('curl \"https://www.commoncriteriaportal.org/pps/index.cfm?archived=1\" -o cc_pp_archived.html\n\n') + file.write( + 'curl \"https://www.commoncriteriaportal.org/pps/\" -o cc_pp_active.html\n') + file.write( + 'curl \"https://www.commoncriteriaportal.org/pps/collaborativePP.cfm?cpp=1\" -o cc_pp_collaborative.html\n') + file.write( + 'curl \"https://www.commoncriteriaportal.org/pps/index.cfm?archived=1\" -o cc_pp_archived.html\n\n') - file.write('curl \"https://www.commoncriteriaportal.org/pps/pps.csv\" -o cc_pp_active.csv\n') - file.write('curl \"https://www.commoncriteriaportal.org/pps/pps-archived.csv\" -o cc_pp_archived.csv\n\n') + file.write( + 'curl \"https://www.commoncriteriaportal.org/pps/pps.csv\" -o cc_pp_active.csv\n') + file.write( + 'curl \"https://www.commoncriteriaportal.org/pps/pps-archived.csv\" -o cc_pp_archived.csv\n\n') def generate_failed_download_script(base_dir): @@ -2064,7 +2427,7 @@ def generate_failed_download_script(base_dir): MIN_CORRECT_CERT_SIZE = 5000 download_again = [] for sub_folder in sub_folders: - target_dir = '{}\\{}'.format(base_dir, sub_folder) + target_dir = '{}{}{}'.format(base_dir, os.sep, sub_folder) # obtain list of all downloaded pdf files and their size files = search_files(target_dir) for file_name in files: @@ -2079,12 +2442,13 @@ def generate_failed_download_script(base_dir): file_size = os.path.getsize(file_name) if file_size < MIN_CORRECT_CERT_SIZE: # too small file, likely failed download - retry - file_name_short = file_name[file_name.rfind('\\') + 1:] + file_name_short = file_name[file_name.rfind(os.sep) + 1:] # double %% is necessary to prevent replacement of %2 within script (second argument of script) file_name_short_web = file_name_short.replace(' ', '%%20') download_link = '/files/epfiles/{}'.format(file_name_short_web) download_again.append((download_link, file_name)) - generate_download_script('download_failed_certs.bat', '', '', CC_WEB_URL, download_again) - print('*** Number of files to be re-downloaded again (inside \'{}\'): {}'.format('download_failed_certs.bat', len(download_again))) - + generate_download_script('download_failed_certs.bat', + '', '', CC_WEB_URL, download_again) + print('*** Number of files to be re-downloaded again (inside \'{}\'): {}'.format( + 'download_failed_certs.bat', len(download_again))) diff --git a/src/fips_certificates.py b/src/fips_certificates.py new file mode 100644 index 00000000..5d7380ee --- /dev/null +++ b/src/fips_certificates.py @@ -0,0 +1,444 @@ +import json +import os +import re + +from graphviz import Digraph +from PyPDF2 import PdfFileReader, utils +import pikepdf +# from camelot import read_pdf +from tabula import read_pdf + +import extract_certificates +from process_certificates import load_json_files +from cert_rules import rules_fips_htmls as RE_FIPS_HTMLS + +import time + +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): + return os.path.splitext(os.path.basename(file))[0] + + +def parse_ul(text): + p = re.compile(r"<li>(.*?)<\/li>") + return p.findall(text) + + +def parse_table(text): + items_found_all = [] + + # find <tr>, in that look for "text-nowrap" and look if there is a cert mentioned + tr_pattern = re.compile(r"<tr>([\s\S]*?)<\/tr>") + name_pattern = re.compile(r"wrap\">(?P<name>[\s\S]*?)<\/td>") + cert_pattern_found = re.compile(r"<td>[ \S]*?#[ \S]*?\d+[ \S]*?<\/td>") + cert_pattern_localize = re.compile(r"#?[ \S]*?(?P<cert>\d+)") + + for tr_match in tr_pattern.finditer(text): + items_found = {} + current_tr = tr_match.group() + items_found['Name'] = name_pattern.search(current_tr).group('name') + cert_line = cert_pattern_found.search(current_tr) + + if cert_line is None: + items_found['Certificate'] = ['Not found'] + else: + items_found['Certificate'] = ['#' + x.group('cert') for x in cert_pattern_localize.finditer( + cert_line.group())] + + items_found_all.append(items_found) + + return items_found_all + + +def parse_algorithms(text, in_table=False): + # print("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~") + # print() + # print(text) + items_found = [] + for m in re.finditer(r"(?:#{}\s?|Cert\.?[^. ]*?\s?)(?:[Cc]\s)?(?P<id>\d+)".format('?' if in_table else ''), text): + items_found.append({'Certificate': m.group()}) + + return items_found + + +def parse_caveat(text): + items_found = [] + + for m in re.finditer(r"(?:#\s?|Cert\.?(?!.\s)\s?|Certificate\s?)(?P<id>\d+)", text): + items_found.append({r"(?:#\s?|Cert\.?(?!.\s)\s?|Certificate\s?)(?P<id>\d+?})": {m.group(): {'count': 1}}}) + + return items_found + + +def initialize_entry(input_dictionary): + input_dictionary['fips_exceptions'] = [] + input_dictionary['fips_tested_conf'] = [] + + input_dictionary['fips_algorithms'] = [] + input_dictionary['fips_caveat'] = [] + input_dictionary['tables_done'] = False + + +def fips_search_html(base_dir, output_file, dump_to_file=False): + """fips_search_html. + + :param base_dir: directory to search for html files + :param output_file: file to dump json to + :param dump_to_file: True/False + """ + + all_found_items = {} + + for file in extract_certificates.search_files(base_dir): + items_found = {} + initialize_entry(items_found) + text = extract_certificates.load_cert_html_file(file) + filename = os.path.splitext(os.path.basename(file))[0] + all_found_items[filename] = items_found + items_found['cert_fips_id'] = filename + + for rule in RE_FIPS_HTMLS: + m = re.search(rule, text) + if m is None: + # print("ERROR: For rule {} nothing found in file {}.".format(rule, file)) + continue + + group_dict = m.groupdict() + key = list(group_dict) + + # <ul> + if key[0] == 'fips_exceptions' or key[0] == 'fips_tested_conf': + items_found[key[0]] = parse_ul(group_dict[key[0]]) + + # <table> + elif key[0] == 'fips_algorithms': + if 'fips_algorithms' not in items_found: + items_found['fips_algorithms'] = parse_table(group_dict[key[0]]) + else: + items_found['fips_algorithms'] += parse_table( + group_dict[key[0]]) + + # allowed algorithms + elif key[0] == 'fips_allowed_algorithms': + if 'fips_algorithms' not in items_found: + items_found['fips_algorithms'] = parse_algorithms( + group_dict[key[0]]) + else: + items_found['fips_algorithms'] += parse_algorithms( + group_dict[key[0]]) + + # certificates in Caveat + elif key[0] == 'fips_caveat': + items_found['fips_mentioned_certs'] = parse_caveat(group_dict[key[0]]) + + # there are usually multiple dates separated by ";" + elif 'date' in key[0]: + items_found[key[0]] = group_dict[key[0]].replace('\n', '').replace( + '\t', '').replace(' ', ' ').strip().split(';') + + else: + items_found[key[0]] = group_dict[key[0]].replace( + '\n', '').replace('\t', '').replace(' ', ' ').strip() + + 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)) + + return all_found_items + + +def get_dot_graph(found_items, output_file_name): + dot = Digraph(comment='Certificate ecosystem') + single_dot = Digraph(comment='Modules with no dependencies') + single_dot.attr('graph', label='Single nodes', labelloc='t', fontsize='30') + single_dot.attr('node', style='filled') + dot.attr('graph', label='Dependencies', labelloc='t', fontsize='30') + dot.attr('node', style='filled') + + def found_interesting_cert(current_key): + if found_items[current_key]['fips_vendor'] == highlighted_vendor: + dot.attr('node', color='red') + if found_items[current_key]['fips_status'] == 'Revoked': + dot.attr('node', color='grey32') + if found_items[current_key]['fips_status'] == 'Historical': + dot.attr('node', color='gold3') + if found_items[current_key]['fips_vendor'] == "SUSE, LLC": + dot.attr('node', color='lightblue') + + def color_check(current_key): + dot.attr('node', color='lightgreen') + if found_items[current_key]['fips_status'] == 'Revoked': + dot.attr('node', color='lightgrey') + if found_items[current_key]['fips_status'] == 'Historical': + dot.attr('node', color='gold') + found_interesting_cert(current_key) + dot.node(current_key, label=current_key + '\n' + found_items[current_key]['fips_vendor'] + + ('\n' + found_items[current_key]['fips_module_name'] + if 'fips_module_name' in found_items[current_key] else '')) + + keys = 0 + edges = 0 + + highlighted_vendor = 'Red Hat®, Inc.' + for key in found_items: + if key != 'Not found' and found_items[key]['file_status']: + if found_items[key]['Connections']: + color_check(key) + keys += 1 + else: + single_dot.attr('node', color='lightblue') + found_interesting_cert(key) + single_dot.node(key, label=key + '\n' + found_items[key]['fips_vendor'] + ('\n' + found_items[key][ + 'fips_module_name'] if 'fips_module_name' in found_items[key] else '')) + + for key in found_items: + if key != 'Not found' and found_items[key]['file_status']: + for conn in found_items[key]['Connections']: + color_check(conn) + dot.edge(key, conn) + edges += 1 + + print("rendering {} keys and {} edges".format(keys, edges)) + + dot.render(output_file_name + 'connections', view=True) + single_dot.render(output_file_name + 'single', view=True) + + +def remove_algorithms_from_extracted_data(items, html): + for file_name in items: + items[file_name]['file_status'] = True + html[file_name]['file_status'] = True + if 'fips_mentioned_certs' in html[file_name]: + for item in html[file_name]['fips_mentioned_certs']: + items[file_name]['rules_cert_id'].update(item) + + for rule in items[file_name]['rules_cert_id']: + to_pop = set() + rr = re.compile(rule) + for cert in items[file_name]['rules_cert_id'][rule]: + for alg in items[file_name]['rules_fips_algorithms']: + for found in items[file_name]['rules_fips_algorithms'][alg]: + if rr.search(found) and rr.search(cert) and rr.search(found).group('id') == rr.search( + cert).group('id'): + to_pop.add(cert) + for r in to_pop: + items[file_name]['rules_cert_id'][rule].pop(r, None) + + items[file_name]['rules_cert_id'][rule].pop( + html[file_name]['cert_fips_id'], None) + + +def validate_results(items, html): + broken_files = set() + for file_name in items: + for rule in items[file_name]['rules_cert_id']: + for cert in items[file_name]['rules_cert_id'][rule]: + cert_id = ''.join(filter(str.isdigit, cert)) + + if cert_id == '' or cert_id not in html: + broken_files.add(file_name) + items[file_name]['file_status'] = False + html[file_name]['file_status'] = False + break + + print("WARNING: CERTIFICATE FILES WITH WRONG CERTIFICATES PARSED") + print(*sorted(list(broken_files)), sep='\n') + print("... skipping these...") + print("Total non-analyzable files:", len(broken_files)) + + for file_name in items: + html[file_name]['Connections'] = [] + if not items[file_name]['file_status']: + continue + if items[file_name]['rules_cert_id'] == {}: + continue + for rule in items[file_name]['rules_cert_id']: + for cert in items[file_name]['rules_cert_id'][rule]: + cert_id = ''.join(filter(str.isdigit, cert)) + if cert_id not in html[file_name]['Connections']: + html[file_name]['Connections'].append(cert_id) + + +count = 0 + + +def parse_list_of_tables(txt): + """ + 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 + """ + rr = re.compile(r"^.+?(?:[Ff]unction|[Aa]lgorithm).+?(?P<page_num>\d+)$", re.MULTILINE) + pages = set() + for m in rr.finditer(txt): + pages.add(m.group('page_num')) + return pages + + +def extract_page_number(txt): + """ + Parses chunks of text that are supposed to be mentioning table and having a footer + :param txt: input chunk + :return: page number + """ + # Page # of # + m = re.findall(r"(?P<pattern>(?:[Pp]age) (?P<page_num>\d+)(?: of \d+))", txt) + if m: + return m[-1][-1] + # Page # + m = re.findall(r"(?P<pattern>(?:[Pp]age) (?P<page_num>\d+)(?: of \d+)?)", txt) + if m: + return m[-1][-1] + # # of # + m = re.findall(r"(?P<pattern>(?:[Pp]age)? ?(?P<page_num>\d+)(?: of \d+))", txt) + if m: + return m[-1][-1] + # number alone + m = re.findall(r"(?P<pattern>(?:[Pp]age)? ?(?P<page_num>\d+)(?: of \d+)?)", txt) + return m[-1][-1] if m else None + + +def find_tables(txt, file_name, num_pages): + global count + + # Look for "List of Tables", where we can find exactly tables with page num + tables_regex = re.compile(r"^(?:(?:[Tt]able\s|[Ll]ist\s)(?:[Oo]f\s))[Tt]ables[\s\S]+?\f", re.MULTILINE) + table = tables_regex.search(txt) + if table: + count += 1 + rb = parse_list_of_tables(table.group()) + if rb: + return list(rb) + return None + + # Otherwise look for "Table" in text and \f representing footer, then extract page number from footer + print("~" * 20, file_name, '~' * 20) + footer_regex = re.compile(r"(?:Table[^\f]*)(?P<first>^[\S\t ]*$)\n(?P<second>(\f[ \t\S]+)$)(?P<third>\n^[ \t\S]+?$)?", + re.MULTILINE) + + # We have 2 groups, one is optional - trying to parse 2 lines (just in case) + footer1 = [m.group('first') for m in footer_regex.finditer(txt)] + footer2 = [m.group('second') for m in footer_regex.finditer(txt)] + footer3 = [m.group('third') for m in footer_regex.finditer(txt)] + + # if len(footer2) < len(footer1): + # footer2 += [''] * (len(footer1) - len(footer2)) + + # zipping them together + footer_complete = [m[0] + m[1] + m[2] for m in zip(footer1, footer2, footer3) if m[0] is not None and m[1] is not None and m[2] is not None] + + # removing None and duplicates + footers = [extract_page_number(x) for x in footer_complete] + footers = list(dict.fromkeys([x for x in footers if x is not None and 0 < int(x) < num_pages])) + print(footers) + if footers: + return footers + + +def repair_pdf_page_count(file): + 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 + list_of_files = json.loads(open(FIPS_RESULTS_DIR + 'bakup/broken_files.json').read()) + not_decoded = [] + for REDHAT_FILE in list_of_files: + if '.txt' not in REDHAT_FILE: + continue + # + # if html_items[extract_filename(REDHAT_FILE[:-8])]['tables_done']: + # continue + + with open(REDHAT_FILE, 'r') as f: + try: + pages = repair_pdf_page_count(REDHAT_FILE[:-4]) + except pikepdf._qpdf.PdfError: + not_decoded.append(REDHAT_FILE) + continue + tables = find_tables(f.read(), REDHAT_FILE, pages) + + # If we find any tables with page numbers, we process them + if tables: + lst = [] + print("~~~~~~~~~~~~~~~", REDHAT_FILE, "~~~~~~~~~~~~~~~~~~~~~~~") + + try: + data = read_pdf(REDHAT_FILE[:-4], pages=tables, silent=True) + except Exception: + not_decoded.append(REDHAT_FILE) + continue + + # find columns with cert numbers + for df in data: + for col in range(len(df.columns)): + if 'cert' in df.columns[col].lower() or 'algo' in df.columns[col].lower(): + lst += parse_algorithms(df.iloc[:, col].to_string(index=False), True) + + # 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 + else: + html_items[extract_filename(REDHAT_FILE[:-8])]['fips_algorithms'] += lst + print(lst) + html_items[extract_filename(REDHAT_FILE[:-8])]['tables_done'] = True + return not_decoded + + +def main(): + files_to_load = [ + FIPS_RESULTS_DIR + 'fips_data_keywords_all.json', + FIPS_RESULTS_DIR + 'fips_html_all.json' + ] + + for file in files_to_load: + if not os.path.isfile(file): + fips_items = fips_search_html(FIPS_BASE_DIR + 'html/', + FIPS_RESULTS_DIR + 'fips_html_all.json', True) + items = extract_certificates.extract_certificates_keywords( + FIPS_BASE_DIR + 'security_policies/', + 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)) + break + + print("EXTRACTION DONE") + (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) + + print("NOT DECODED:", not_decoded) + with open(FIPS_RESULTS_DIR + 'broken_files.json', 'w') as f: + f.write(json.dumps(not_decoded)) + + print("REMOVING ALGORITHMS") + remove_algorithms_from_extracted_data(items, html) + + 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)) + print("PLOTTING GRAPH") + get_dot_graph(html, 'output') + + +if __name__ == '__main__': + start = time.time() + main() + end = time.time() + print("TIME:", end - start) + print("COUNT:", count) diff --git a/src/process_certificates.py b/src/process_certificates.py index 49c60c20..82d92d52 100644 --- a/src/process_certificates.py +++ b/src/process_certificates.py @@ -139,15 +139,18 @@ def main(): paths_20200904['id'] = '20200904' paths_20200904['cc_html_files_dir'] = 'c:\\Certs\\cc_certs_20200904\\web\\' paths_20200904['walk_dir'] = 'c:\\Certs\\cc_certs_20200904\\cc_certs\\' - #paths_20200904['walk_dir'] = 'c:\\Certs\\cc_certs_20200904\\cc_certs_test\\' paths_20200904['pp_dir'] = 'c:\\Certs\\cc_certs_20200904\\cc_pp\\' paths_20200904['fragments_dir'] = 'c:\\Certs\\cc_certs_20200904\\cc_certs_txt_fragments\\' paths_20200904['pp_fragments_dir'] = 'c:\\Certs\\cc_certs_20200904\\cc_pp_txt_fragments\\' + paths_20200904_test = paths_20200904 + paths_20200904_test['walk_dir'] = 'c:\\Certs\\cc_certs_20200904\\cc_certs_test\\' + # initialize paths based on the profile used #paths_used = paths_20191208 #paths_used = paths_20200225 - paths_used = paths_20200904 + #paths_used = paths_20200904 + paths_used = paths_20200904_test #paths_used['id'] = 'temp' # change id for temporary debugging cc_html_files_dir = paths_used['cc_html_files_dir'] @@ -177,12 +180,13 @@ def main(): # with open("pp_data_complete.json", "w") as write_file: # write_file.write(json.dumps(all_pp_items, indent=4, sort_keys=True)) - do_complete_extraction = False - do_extraction = False - do_extraction_pp = False - do_pairing = False - do_processing = False + do_complete_extraction = True + do_extraction = True + do_extraction_pp = True + do_pairing = True + do_processing = True do_analysis = True + do_analysis_filtered = False # with open('certificate_data_complete_processed.json') as json_file: # all_cert_items = json.load(json_file) @@ -304,21 +308,24 @@ def main(): with open('certificate_data_complete_processed.json') as json_file: all_cert_items = json.load(json_file) - # analyze only smartcards - do_analysis_only_filtered(all_cert_items, results_folder, - ['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, - ['csv_scan', 'cc_category'], 'Operating Systems') + if do_analysis_filtered: + # analyze only smartcards + do_analysis_only_filtered(all_cert_items, results_folder, + ['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, + ['csv_scan', 'cc_category'], 'Operating Systems') + + # analyze separate manufacturers + do_analysis_manufacturers(all_cert_items, results_folder) + + # archived on 09/01/2019 + do_analysis_09_01_2019_archival(all_cert_items, results_folder) - # analyze separate manufacturers - do_analysis_manufacturers(all_cert_items, results_folder) # analyze all certificates together do_analysis_everything(all_cert_items, results_folder) - # archived on 09/01/2019 - do_analysis_09_01_2019_archival(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)) @@ -328,7 +335,6 @@ def main(): # TODO - # use os.sep to support properly Linux&Windows path separation # add saving of logs into file # include parsing from protection profiles repo # add differential partial download of new files only + processing + combine |