diff options
| author | J08nY | 2020-10-15 21:56:24 +0200 |
|---|---|---|
| committer | J08nY | 2020-10-18 11:43:10 +0200 |
| commit | 64bf9c880828fa36f8625fdabad468668df62575 (patch) | |
| tree | 68902488bcce91b5cdf8c39a809ca90cf465fe50 /src | |
| parent | 7ed51240bd5e5e56a5847df28c8a1a9708307315 (diff) | |
| download | sec-certs-64bf9c880828fa36f8625fdabad468668df62575.tar.gz sec-certs-64bf9c880828fa36f8625fdabad468668df62575.tar.zst sec-certs-64bf9c880828fa36f8625fdabad468668df62575.zip | |
Add a proper setup, move into a package.
Diffstat (limited to 'src')
| -rw-r--r-- | src/analyze_certificates.py | 828 | ||||
| -rw-r--r-- | src/cert_rules.py | 357 | ||||
| -rw-r--r-- | src/download_files.py | 36 | ||||
| -rw-r--r-- | src/extract_certificates.py | 2435 | ||||
| -rw-r--r-- | src/fips_certificates.py | 444 | ||||
| -rw-r--r-- | src/process_certificates.py | 377 | ||||
| -rw-r--r-- | src/sanity.py | 53 | ||||
| -rw-r--r-- | src/tags_constants.py | 27 |
8 files changed, 0 insertions, 4557 deletions
diff --git a/src/analyze_certificates.py b/src/analyze_certificates.py deleted file mode 100644 index cc0beba4..00000000 --- a/src/analyze_certificates.py +++ /dev/null @@ -1,828 +0,0 @@ -import operator -from graphviz import Digraph -from tabulate import tabulate -import matplotlib.pyplot as plt; - -import sanity - -plt.rcdefaults() -import numpy as np -import matplotlib.pyplot as plt -from matplotlib.pyplot import figure -from dateutil import parser -import datetime -from tags_constants import * -import string -import os - -STOP_ON_UNEXPECTED_NUMS = False - -printable = set(string.printable) - - -def is_in_dict(target_dict, path): - current_level = target_dict - for item in path: - if item not in current_level: - return False - else: - current_level = current_level[item] - return True - - -def get_item_from_dict(target_dict, path): - current_level = target_dict - for item in path: - if item not in current_level: - return None - else: - current_level = current_level[item] - return current_level - - -def fig_label(title, filter): - if filter != '': - return '{}\nfilter: {}'.format(title, filter) - else: - return title - - -def plot_bar_graph(data, x_data_labels, y_label, title, file_name): - fig_width = round(len(data) / 2) - if fig_width < 10: - fig_width = 10 - figure(num=None, figsize=(fig_width, 8), dpi=200, facecolor='w', edgecolor='k') - y_pos = np.arange(len(x_data_labels)) - plt.bar(y_pos, data, align='center', alpha=0.5) - plt.xticks(y_pos, x_data_labels) - plt.xticks(rotation=45) - plt.ylabel(y_label) - plt.title(title) - x1, x2, y1, y2 = plt.axis() - 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): - 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') - #sns.heatmap(data_matrix, cmap=color_map, linewidth=0.5) - x_pos = np.arange(len(y_data_ticks)) - plt.yticks(x_pos, y_data_ticks) - y_pos = np.arange(len(x_data_ticks)) - plt.xticks(y_pos, x_data_ticks) - plt.xticks(rotation=90, ha='center') - plt.gca().invert_yaxis() - x1, x2, y1, y2 = plt.axis() - plt.axis((x1, x2, y1 - 0.5, y2)) - plt.xlabel(x_label) - plt.ylabel(y_label) - plt.title(title) - 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): - hist_refs = np.histogram(data, bins) - hist_labels = [] - for index in range(0, len(bins) - 1): - if bins[index] == bins[index + 1] - 1: - hist_labels.append('{}'.format(bins[index])) - else: - hist_labels.append('{}-{}'.format(bins[index], bins[index + 1])) - # plot bar graph with number of certificates referenced by given number of other certificates - plot_bar_graph(hist_refs[0], hist_labels, y_label, title, file_name) - - -def depricated_print_dot_graph_keywordsonly(filter_rules_group, all_items_found, cert_id, filter_label, out_dot_name, thick_as_occurences): - # print dot - dot = Digraph(comment='Certificate ecosystem: {}'.format(filter_rules_group)) - dot.attr('graph', label='{}'.format(filter_label), labelloc='t', fontsize='30') - dot.attr('node', style='filled') - - # insert nodes believed to be cert id for the processed certificates - for cert in cert_id.keys(): - if cert != "": - dot.attr('node', color='green') - dot.node(cert_id[cert]) - - dot.attr('node', color='gray') - for file_name in all_items_found.keys(): - just_file_name = file_name - this_cert_id = cert_id[file_name] - - 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 != "": - dot.edge(this_cert_id, this_cert_id, label=just_file_name) - - items_found_group = all_items_found[file_name] - for rules_group in items_found_group.keys(): - - # process only specified rule groups - if rules_group not in filter_rules_group: - continue - - items_found = items_found_group[rules_group] - for rule in items_found.keys(): - for match in items_found[rule]: - if match != this_cert_id: - if thick_as_occurences: - num_occurrences = str(items_found[rule][match][TAG_MATCH_COUNTER] / 3 + 1) - else: - num_occurrences = '1' - label = str(items_found[rule][match][TAG_MATCH_COUNTER]) # label with number of occurrences - if this_cert_id != "": - dot.edge(this_cert_id, match, color='orange', style='solid', label=label, penwidth=num_occurrences) - - # Generate dot graph using GraphViz into pdf - dot.render(out_dot_name, view=False) - print('{} pdf rendered'.format(out_dot_name)) - - -def get_cert_node_label(cert_item, print_item_name): - if print_item_name: - sanitized_name = ''.join(filter(lambda x: x in printable, cert_item['csv_scan']['cert_item_name'])) - #sanitized_name = cert_item['csv_scan']['cert_item_name'].encode('ascii', 'ignore') - #sanitized_name = cert_item['csv_scan']['cert_item_name'].encode("ascii") - sanitized_name = sanitized_name.replace(':', ' ') # ':' is not allowed in dot - sanitized_name = sanitized_name.replace('/', ' ') # '/' is not allowed in dot - return '{}\n{}'.format(cert_item['processed']['cert_id'], sanitized_name) - else: - return cert_item['processed']['cert_id'] - - -def print_dot_graph(filter_rules_group, all_items_found, filter_label, out_dot_name, thick_as_occurences, print_item_name): - # print dot - dot = Digraph(comment='Certificate ecosystem: {}'.format(filter_rules_group)) - dot.attr('graph', label='{}'.format(filter_label), labelloc='t', fontsize='30') - dot.attr('node', style='filled') - - # insert nodes believed to be cert id for the processed certificates - cert_id_to_long_id_map = {} - for cert_long_id in all_items_found.keys(): - if is_in_dict(all_items_found[cert_long_id], ['processed', 'cert_id']): - dot.attr('node', color='green') # URL='https://www.commoncriteriaportal.org/' doesn't work for pdf - this_cert_node_label = get_cert_node_label(all_items_found[cert_long_id], print_item_name) - # basic node id is cert id, but possibly add additional info - dot.node(all_items_found[cert_long_id]['processed']['cert_id'], label=this_cert_node_label) - cert_id_to_long_id_map[all_items_found[cert_long_id]['processed']['cert_id']] = cert_long_id - - dot.attr('node', color='gray') - for cert_long_id in all_items_found.keys(): - # do not continue if no keywords were extracted - if 'keywords_scan' not in all_items_found[cert_long_id].keys(): - continue - - cert = all_items_found[cert_long_id] - this_cert_id = '' - if is_in_dict(cert, ['processed', 'cert_id']): - this_cert_id = cert['processed']['cert_id'] - - just_file_name = cert['csv_scan']['link_cert_report_file_name'] - - # insert file name and identified probable certification id - if this_cert_id != "": - dot.edge(this_cert_id, this_cert_id, label=just_file_name) - - items_found_group = all_items_found[cert_long_id]['keywords_scan'] - for rules_group in items_found_group.keys(): - - # process only specified rule groups - if rules_group not in filter_rules_group: - continue - - items_found = items_found_group[rules_group] - for rule in items_found.keys(): - for match in items_found[rule]: - if match != this_cert_id: - if thick_as_occurences: - num_occurrences = str(items_found[rule][match][TAG_MATCH_COUNTER] / 3 + 1) - else: - num_occurrences = '1' - label = str(items_found[rule][match][TAG_MATCH_COUNTER]) # label with number of occurrences - if this_cert_id != "": - #if is_in_dict(cert_id_to_long_id_map, [match]): - # other_cert_node_label = get_cert_node_label(all_items_found[cert_id_to_long_id_map[match]], print_item_name) - #else: - # other_cert_node_label = match - #dot.edge(this_cert_node_label, other_cert_node_label, color='orange', style='solid', label=label, penwidth=num_occurrences) - dot.edge(this_cert_id, match, color='orange', style='solid', label=label, penwidth=num_occurrences) - - # Generate dot graph using GraphViz into pdf - dot.render(out_dot_name, view=False) - - print('{} pdf rendered'.format(out_dot_name)) - - -def plot_certid_to_item_graph(item_path, all_items_found, filter_label, out_dot_name, thick_as_occurences): - # print dot - dot = Digraph(comment='Certificate ecosystem: {}'.format(item_path)) - dot.attr('graph', label='{}'.format(filter_label), labelloc='t', fontsize='30') - dot.attr('node', style='filled') - - # insert nodes believed to be cert id for the processed certificates - for cert_long_id in all_items_found.keys(): - if is_in_dict(all_items_found[cert_long_id], ['processed', 'cert_id']): - dot.attr('node', color='green') # URL='https://www.commoncriteriaportal.org/' doesn't work for pdf - dot.node(all_items_found[cert_long_id]['processed']['cert_id']) - - dot.attr('node', color='gray') - for cert_long_id in all_items_found.keys(): - # do not continue if no values with specified path were extracted - if item_path[0] not in all_items_found[cert_long_id].keys(): - continue - - cert = all_items_found[cert_long_id] - this_cert_id = '' - if is_in_dict(cert, ['processed', 'cert_id']): - this_cert_id = cert['processed']['cert_id'] - - if is_in_dict(cert, [item_path[0], item_path[1]]): - items_found = cert[item_path[0]][item_path[1]] - for rule in items_found: - for match in items_found[rule]: - if match != this_cert_id: - if thick_as_occurences: - num_occurrences = str(items_found[rule][match][TAG_MATCH_COUNTER] / 3 + 1) - else: - num_occurrences = '1' - label = str(items_found[rule][match][TAG_MATCH_COUNTER]) # label with number of occurrences - if this_cert_id != "": - dot.edge(this_cert_id, match, color='orange', style='solid', label=label, penwidth=num_occurrences) - - # Generate dot graph using GraphViz into pdf - 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 = {} - for cert_long_id in all_items_found.keys(): - cert = all_items_found[cert_long_id] - if is_in_dict(cert, ['processed', 'cert_id']): - if is_in_dict(cert, ['frontpage_scan', 'cert_item']): - this_cert_id = cert['processed']['cert_id'] - if this_cert_id not in certid_info.keys(): - - certid_info[this_cert_id] = {} - certid_info[this_cert_id]['cert_item'] = cert['frontpage_scan']['cert_item'] - - # build cert_id to cert_long_id mapping - cert_id_to_long_id_mapping = {} - for cert_long_id in all_items_found.keys(): - cert = all_items_found[cert_long_id] - if is_in_dict(cert, ['processed', 'cert_id']): - if is_in_dict(cert, ['frontpage_scan', 'cert_item']): - this_cert_id = cert['processed']['cert_id'] - if this_cert_id in cert_id_to_long_id_mapping.keys(): - print('ERROR: duplicate cert_id for multiple cert_long_id: {}, {} already used by {}'.format(this_cert_id, cert_long_id, cert_id_to_long_id_mapping[this_cert_id])) - else: - cert_id_to_long_id_mapping[this_cert_id] = cert_long_id - - # build list of references - referenced_by = {} - for cert_long_id in all_items_found.keys(): - # do not continue if no keywords were extracted () - if 'keywords_scan' not in all_items_found[cert_long_id].keys(): - continue - - cert = all_items_found[cert_long_id] - this_cert_id = '' - if is_in_dict(cert, ['processed', 'cert_id']): - this_cert_id = cert['processed']['cert_id'] - - items_found_group = all_items_found[cert_long_id]['keywords_scan'] - for rules_group in items_found_group.keys(): - - # process only specified rule groups - if rules_group not in filter_rules_group: - continue - - items_found = items_found_group[rules_group] - for rule in items_found.keys(): - for match in items_found[rule]: - if match != this_cert_id: - if this_cert_id != "": - # add this_cert_id to the list of references of match item - if match not in referenced_by: - referenced_by[match] = [] - if this_cert_id not in referenced_by[match]: - referenced_by[match].append(this_cert_id) - - # - # process direct references - # - referenced_by_direct_nums = {} - for cert_id in referenced_by.keys(): - referenced_by_direct_nums[cert_id] = len(referenced_by[cert_id]) - if cert_id in cert_id_to_long_id_mapping.keys(): - all_items_found[cert_id_to_long_id_mapping[cert_id]]['processed']['direct_refs'] = len(referenced_by[cert_id]) - - print('### Certificates sorted by number of other certificates directly referencing them:') - sorted_ref_direct = sorted(referenced_by_direct_nums.items(), key=operator.itemgetter(1), reverse=False) - direct_refs = [] - for cert_id in sorted_ref_direct: - direct_refs.append(cert_id[1]) - try: - if is_in_dict(certid_info, [cert_id[0], 'cert_item']): - print(' {} : {}x directly: {}'.format(cert_id[0], cert_id[1], certid_info[cert_id[0]]['cert_item'])) - else: - print(' {} : {}x directly'.format(cert_id[0], cert_id[1])) - except UnicodeEncodeError: - print('ERROR: UnicodeEncodeError occured') - print(' Total number of certificates referenced at least once: {}'.format(len(sorted_ref_direct))) - - step = 5 - if len(direct_refs) == 0: - max_refs = 0 - else: - max_refs = max(direct_refs) + step - bins = [1, 2, 3, 4, 5] + list(range(6, max_refs + 1, step)) - compute_and_plot_hist(direct_refs, bins, 'Number of certificates', fig_label('# certificates with specific number of direct references', filter_label), 'cert_direct_refs_frequency.png') - - sanity.check_certs_referenced_once(len(sorted_ref_direct)) - - - # - # compute indirect num of references - # - referenced_by_indirect = {} - for cert_id in referenced_by.keys(): - referenced_by_indirect[cert_id] = set() - for item in referenced_by[cert_id]: - referenced_by_indirect[cert_id].add(item) - - new_change_detected = True - while new_change_detected: - new_change_detected = False - - certids_list = referenced_by.keys() - for cert_id in certids_list: - tmp_referenced_by_indirect_nums = referenced_by_indirect[cert_id].copy() - for referencing in tmp_referenced_by_indirect_nums: - if referencing in referenced_by.keys(): - tmp_referencing = referenced_by_indirect[referencing].copy() - for in_referencing in tmp_referencing: - if in_referencing not in referenced_by_indirect[cert_id]: - new_change_detected = True - referenced_by_indirect[cert_id].add(in_referencing) - - print('### Certificates sorted by number of other certificates indirectly referencing them:') - referenced_by_indirect_nums = {} - for cert_id in referenced_by_indirect.keys(): - referenced_by_indirect_nums[cert_id] = len(referenced_by_indirect[cert_id]) - if cert_id in cert_id_to_long_id_mapping.keys(): - all_items_found[cert_id_to_long_id_mapping[cert_id]]['processed']['indirect_refs'] = referenced_by_indirect_nums[cert_id] - - sorted_ref_indirect = sorted(referenced_by_indirect_nums.items(), key=operator.itemgetter(1), reverse=False) - indirect_refs = [] - for cert_id in sorted_ref_indirect: - indirect_refs.append(cert_id[1]) - try: - if is_in_dict(certid_info, [cert_id[0], 'cert_item']): - print(' {} : {}x indirectly: {}'.format(cert_id[0], cert_id[1], certid_info[cert_id[0]]['cert_item'])) - else: - print(' {} : {}x indirectly'.format(cert_id[0], cert_id[1])) - except UnicodeEncodeError: - print('ERROR: UnicodeEncodeError occured') - - step = 5 - if len(indirect_refs) == 0: - max_refs = 0 - else: - max_refs = max(indirect_refs) + step - bins = [1, 2, 3, 4, 5] + list(range(6, max_refs + 1, step)) - compute_and_plot_hist(indirect_refs, bins, 'Number of certificates', fig_label('# certificates with specific number of indirect references', filter_label), 'cert_indirect_refs_frequency.png') - - -def plot_schemes_multi_line_graph(x_ticks, data, prominent_data, x_label, y_label, title, file_name): - - figure(num=None, figsize=(16, 8), dpi=200, facecolor='w', edgecolor='k') - - line_types = ['-', ':', '-.', '--'] - num_lines_plotted = 0 - data_sorted = sorted(data.keys()) - for group in data_sorted: - items_in_year = [] - for item in sorted(data[group]): - num = len(data[group][item]) - items_in_year.append(num) - - if group in prominent_data: - plt.plot(x_ticks, items_in_year, line_types[num_lines_plotted % len(line_types)], label=group, linewidth=3) - else: - # plot minor suppliers dashed - plt.plot(x_ticks, items_in_year, line_types[num_lines_plotted % len(line_types)], label=group, linewidth=2) - - # change line type to prevent color repetitions - num_lines_plotted += 1 - - plt.rcParams.update({'font.size': 16}) - plt.legend(loc=2) - plt.xticks(x_ticks, rotation=45) - 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') - plt.close() - - -def analyze_cert_years_frequency(all_cert_items, filter_label): - scheme_date = {} - level_date = {} - category_date = {} - pp_date = {} - labs_date = {} - archive_date = {} - validity_length = {} - valid_in_years = {} - manufacturer_date = {} - manufacturer_items = {} - START_YEAR = 1997 - END_YEAR = datetime.datetime.now().year + 1 - ARCHIVE_OFFSET = 10 - - for i in range(END_YEAR - START_YEAR + ARCHIVE_OFFSET): - validity_length[i] = [] - - valid_in_years['active'] = {} - valid_in_years['archived'] = {} - for year in range(START_YEAR, END_YEAR + ARCHIVE_OFFSET): - valid_in_years['active'][year] = [] - valid_in_years['archived'][year] = [] - - for cert_long_id in all_cert_items.keys(): - cert = all_cert_items[cert_long_id] - if is_in_dict(cert, ['csv_scan', 'cc_certification_date']): - # extract year of certification - cert_date = cert['csv_scan']['cc_certification_date'] - parsed_date = parser.parse(cert_date) - cert_year = parsed_date.year - # try to extract year of archivation (if provided) - archived_year = None - if is_in_dict(cert, ['csv_scan', 'cc_archived_date']): - cert_archive_date = cert['csv_scan']['cc_archived_date'] - if cert_archive_date != '': - archived_year = parser.parse(cert_archive_date).year - - # extract EAL level - if is_in_dict(cert, ['processed', 'cc_security_level']): - level_out = cert['processed']['cc_security_level'] - - if level_out not in level_date.keys(): - level_date[level_out] = {} - for year in range(START_YEAR, END_YEAR): - level_date[level_out][year] = [] - level_date[level_out][cert_year].append(cert_long_id) - - # extract certificate category - if is_in_dict(cert, ['csv_scan', 'cc_category']): - category = cert['csv_scan']['cc_category'] - - if category not in category_date.keys(): - category_date[category] = {} - for year in range(START_YEAR, END_YEAR): - category_date[category][year] = [] - category_date[category][cert_year].append(cert_long_id) - - # extract conformance to protection profile - if is_in_dict(cert, ['csv_scan', 'cc_protection_profiles']): - pp_id = cert['csv_scan']['cc_protection_profiles'] - if pp_id == '': - pp_id = 'No Protection Profile' - else: - pp_id = 'Protection Profile' - if pp_id not in pp_date.keys(): - pp_date[pp_id] = {} - for year in range(START_YEAR, END_YEAR): - pp_date[pp_id][year] = [] - pp_date[pp_id][cert_year].append(cert_long_id) - - - # extract scheme - if is_in_dict(cert, ['csv_scan', 'cc_scheme']): - cc_scheme = cert['csv_scan']['cc_scheme'] - if cc_scheme not in scheme_date.keys(): - scheme_date[cc_scheme] = {} - for year in range(START_YEAR, END_YEAR): - scheme_date[cc_scheme][year] = [] - scheme_date[cc_scheme][cert_year].append(cert_long_id) - - # extract manufacturer(s) - if 'cc_manufacturer_simple_list' in cert['processed']: - for manufacturer in cert['processed']['cc_manufacturer_simple_list']: - if manufacturer not in manufacturer_date.keys(): - manufacturer_date[manufacturer] = {} - for year in range(START_YEAR, END_YEAR): - manufacturer_date[manufacturer][year] = [] - if manufacturer not in manufacturer_items: - manufacturer_items[manufacturer] = 0 - - manufacturer_date[manufacturer][cert_year].append(cert_long_id) - manufacturer_items[manufacturer] += 1 - - # extract laboratory - if is_in_dict(cert, ['processed', 'cert_lab']): - lab = cert['processed']['cert_lab'] - - if lab not in labs_date.keys(): - labs_date[lab] = {} - for year in range(START_YEAR, END_YEAR): - labs_date[lab][year] = [] - labs_date[lab][cert_year].append(cert_long_id) - - # extract cert archival status - if archived_year is not None: - valid_years = archived_year - cert_year + 1 - validity_length[valid_years].append(cert_long_id) - cert['processed']['cert_lifetime_length'] = valid_years - - if 'archived_date' not in archive_date.keys(): - archive_date['archived_date'] = {} - for year in range(START_YEAR, END_YEAR + ARCHIVE_OFFSET): # archive year can be quite in future - archive_date['archived_date'][year] = [] - - archive_date['archived_date'][archived_year].append(cert_long_id) - - # establish certificates active / archived in give year - for year in range(START_YEAR, END_YEAR + ARCHIVE_OFFSET): - if archived_year is not None: - # archived date is set - if year >= cert_year: - if year <= archived_year: - # certificate is valid in year - valid_in_years['active'][year].append(cert_long_id) - else: - # certificate is NOT valid in given year - valid_in_years['archived'][year].append(cert_long_id) - else: - # no archival date set => active - if year >= cert_year: - # certificate is valid in year - valid_in_years['active'][year].append(cert_long_id) - - # print manufacturers frequency - sorted_by_occurence = sorted(manufacturer_items.items(), key=operator.itemgetter(1)) - print('\n### Frequency of certificates per company') - print(' # companies: {}'.format(len(manufacturer_items))) - print(' # companies with more than 1 cert: {}'.format(len([i for i in sorted_by_occurence if i[1] > 1]))) - print(' # companies with more than 10 cert: {}'.format(len([i for i in sorted_by_occurence if i[1] > 10]))) - print(' # companies with more than 50 cert: {}\n'.format(len([i for i in sorted_by_occurence if i[1] > 50]))) - for manufacturer in sorted_by_occurence: - print(' {}: {}x'.format(manufacturer[0], manufacturer[1])) - - # plot graphs showing cert. scheme and EAL in years - years = np.arange(START_YEAR, END_YEAR) - years_extended = np.arange(START_YEAR, END_YEAR + ARCHIVE_OFFSET) - plot_schemes_multi_line_graph(years, scheme_date, ['DE', 'JP', 'FR', 'US', 'CA'], 'Year of issuance', 'Number of certificates issued', fig_label('CC certificates issuance frequency per scheme and year', filter_label), 'num_certs_in_years') - plot_schemes_multi_line_graph(years, level_date, ['EAL4+', 'EAL5+','EAL2+', 'Protection Profile'], 'Year of issuance', 'Number of certificates issued', fig_label('Certificates issuance frequency per EAL and year', filter_label), 'num_certs_eal_in_years') - plot_schemes_multi_line_graph(years, category_date, [], 'Year of issuance', 'Number of certificates issued', fig_label('Category of certificates issued in given year', filter_label), 'num_certs_category_in_years') - plot_schemes_multi_line_graph(years, pp_date, [], 'Year of issuance', 'Number of certificates issued', fig_label('Certificates with/without conforming to Protection Profile', filter_label), 'num_certs_pp_in_years') - plot_schemes_multi_line_graph(years, labs_date, [], 'Year of issuance', 'Number of certificates issued', fig_label('Number of certificates certified by laboratory in given year', filter_label), 'num_certs_by_lab_in_years') - plot_schemes_multi_line_graph(years_extended, archive_date, [], 'Year of issuance', 'Number of certificates', fig_label('Number of certificates archived or planned for archival in a given year', filter_label), 'num_certs_archived_in_years') - plot_schemes_multi_line_graph(years_extended, valid_in_years, [], 'Year', 'Number of certificates', fig_label('Number of certificates active and archived in given year', filter_label), 'num_certs_active_archived_in_years') - - sc_manufacturers = ['Gemalto', 'NXP Semiconductors', 'Samsung', 'STMicroelectronics', 'Oberthur Technologies', - 'Infineon Technologies AG', 'G+D Mobile Security GmbH', 'ATMEL Smart Card ICs', 'Idemia', - 'Athena Smartcard', 'Renesas', 'Philips Semiconductors GmbH', 'Oberthur Card Systems'] - - # plot only top manufacturers - top_manufacturers = dict(sorted_by_occurence[len(sorted_by_occurence) - 20:]).keys() # top 20 manufacturers - plot_manufacturers_date = {} - for manuf in manufacturer_date.keys(): - if manuf in top_manufacturers: - plot_manufacturers_date[manuf] = manufacturer_date[manuf] - plot_schemes_multi_line_graph(years, plot_manufacturers_date, sc_manufacturers, 'Year of issuance', 'Number of certificates issued', fig_label('Top 20 manufacturers of certified items per year', filter_label), 'manufacturer_in_years') - - # plot only smartcard manufacturers - plot_manufacturers_date = {} - for manuf in manufacturer_date.keys(): - if manuf in sc_manufacturers: - plot_manufacturers_date[manuf] = manufacturer_date[manuf] - # plot_schemes_multi_line_graph(years, plot_manufacturers_date, [], 'Year of issuance', 'Number of certificates issued', fig_label('Smartcard-related manufacturers of certified items per year', filter_label), 'manufacturer_sc_in_years') - - # plot certificate validity lengths - print('### Certificates validity period lengths:') - validity_length_numbers = [] - for length in sorted(validity_length.keys()): - print(' {} year(s): {}x {}'.format(length, len(validity_length[length]), validity_length[length])) - validity_length_numbers.append(len(validity_length[length])) - plot_bar_graph(validity_length_numbers, sorted(validity_length.keys()), 'Number of certificates', fig_label('Number of certificates with specific validity length', filter_label), 'cert_validity_length_frequency') - - -def analyze_eal_frequency(all_cert_items, filter_label): - scheme_level = {} - for cert_long_id in all_cert_items.keys(): - cert = all_cert_items[cert_long_id] - if is_in_dict(cert, ['csv_scan', 'cc_scheme']): - if is_in_dict(cert, ['processed', 'cc_security_level']): - cc_scheme = cert['csv_scan']['cc_scheme'] - level = cert['processed']['cc_security_level'] - if level.find(',') != -1: - level = level[:level.find(',')] # trim list of augmented items - if cc_scheme not in scheme_level.keys(): - scheme_level[cc_scheme] = {} - if level not in scheme_level[cc_scheme]: - scheme_level[cc_scheme][level] = 0 - scheme_level[cc_scheme][level] += 1 - - print('\n### CC EAL levels based on the certification scheme:') - for cc_scheme in sorted(scheme_level.keys()): - print(cc_scheme) - for level in sorted(scheme_level[cc_scheme].keys()): - print(' {:5}: {}x'.format(level, scheme_level[cc_scheme][level])) - - print('\n') - eal_headers = ['EAL0+', 'EAL1', 'EAL1+','EAL2', 'EAL2+','EAL3', 'EAL3+','EAL4', 'EAL4+','EAL5', - 'EAL5+','EAL6', 'EAL6+','EAL7', 'EAL7+', 'None'] - - total_eals = {} - for level in eal_headers: - total_eals[level] = 0 - - cc_eal_freq = [] - sum_total = 0 - for cc_scheme in sorted(scheme_level.keys()): - this_scheme_levels = [cc_scheme] - total = 0 - for level in eal_headers: - if level in scheme_level[cc_scheme]: - this_scheme_levels.append(scheme_level[cc_scheme][level]) - total += scheme_level[cc_scheme][level] - total_eals[level] += scheme_level[cc_scheme][level] - else: - this_scheme_levels.append(0) - - this_scheme_levels.append(total) - sum_total += total - cc_eal_freq.append(this_scheme_levels) - - total_eals_row = [] - for level in sorted(total_eals.keys()): - total_eals_row.append(total_eals[level]) - - # plot bar graph with frequency of CC EAL levels - plot_bar_graph(total_eals_row, eal_headers, 'Number of certificates', fig_label('Number of certificates of specific EAL level', filter_label), 'cert_eal_frequency') - - # Print table with results over national schemes - total_eals_row.append(sum_total) - cc_eal_freq.append(['Total'] + total_eals_row) - print(tabulate(cc_eal_freq, ['CC scheme'] + eal_headers + ['Total'])) - - -def analyze_pdfmeta(all_cert_items, filter_label): - pdf_tags = {} - pdf_values = {} - for cert_long_id in all_cert_items.keys(): - cert = all_cert_items[cert_long_id] - - if is_in_dict(cert, ['pdfmeta_scan']): - pdf_scan = cert['pdfmeta_scan'] - for tag in pdf_scan.keys(): - if tag not in pdf_tags.keys(): - pdf_tags[tag] = 1 # count number of occurence of particular pdf tag - pdf_values[tag] = {} # collect values for particular tag and their frequencies - pdf_values[tag][pdf_scan[tag]] = [cert_long_id] - else: - pdf_tags[tag] += 1 - if pdf_scan[tag] not in pdf_values[tag].keys(): - pdf_values[tag][pdf_scan[tag]] = [cert_long_id] - else: - pdf_values[tag][pdf_scan[tag]].append(cert_long_id) - - - print('\n### Extracted pdf tags frequency:') - sorted_by_occurence = sorted(pdf_tags.items(), key=operator.itemgetter(1)) - for tag in sorted_by_occurence: - print('{:10}: {}x'.format(tag[0], tag[1])) - if tag[0] in ['pdf_file_size_bytes', '/ModDate', 'pdf_number_of_pages', '/CreationDate']: - print(' Would be too many separate outputs, print supressed') - else: - detected_items_count = {} - for item in pdf_values[tag[0]].items(): - detected_items_count[item[0]] = len(item[1]) - sorted_by_occurence_values = sorted(detected_items_count.items(), key=operator.itemgetter(1)) - for value in sorted_by_occurence_values: - print(' {:10}: {}x : {}'.format(value[0], value[1], pdf_values[tag[0]][value[0]])) - - -def analyze_security_assurance_component_frequency(all_cert_items, filter_label): - return analyze_sc_frequency(all_cert_items, filter_label, 'assurance') - - -def analyze_security_functional_component_frequency(all_cert_items, filter_label): - return analyze_sc_frequency(all_cert_items, filter_label, 'functional') - - -def analyze_sc_frequency(all_cert_items, filter_label, sec_component_label): - sars_freq = {} - key_name = 'invalid' - shortcut = 'invalid' - if sec_component_label == 'functional': - key_name = 'rules_security_functional_components' - shortcut = 'sfr' - if sec_component_label == 'assurance': - key_name = 'rules_security_assurance_components' - shortcut = 'sar' - - for cert_long_id in all_cert_items.keys(): - cert = all_cert_items[cert_long_id] - if is_in_dict(cert, ['keywords_scan', key_name]): - sars = cert['keywords_scan'][key_name] - for sar_rule in sars: - for sar_hit in sars[sar_rule]: - if sar_hit not in sars_freq.keys(): - sars_freq[sar_hit] = 0 - sars_freq[sar_hit] += 1 - - - print('\n### CC security ' + sec_component_label + ' components frequency:') - sars_labels = sorted(sars_freq.keys()) - sars_freq_nums = [] - for sar in sars_labels: - print('{:10}: {}x'.format(sar, sars_freq[sar])) - sars_freq_nums.append(sars_freq[sar]) - - print('\n### CC security ' + sec_component_label + ' components frequency sorted by num occurences:') - sorted_by_occurence = sorted(sars_freq.items(), key=operator.itemgetter(1)) - for sar in sorted_by_occurence: - print('{:10}: {}x'.format(sar[0], sar[1])) - - # plot bar graph with frequency of CC SARs - plot_bar_graph(sars_freq_nums, sars_labels, 'Number of certificates', fig_label('Number of certificates mentioning specific security ' + sec_component_label + ' component (' + shortcut + ')\nAll listed occured at least once', filter_label), 'cert_' + shortcut + '_frequency') - if len(sars_freq_nums) > 0 and len(sars_labels) > 0: - sars_freq_nums, sars_labels = (list(t) for t in zip(*sorted(zip(sars_freq_nums, sars_labels), reverse=True))) - plot_bar_graph(sars_freq_nums, sars_labels, 'Number of certificates', fig_label( - 'Number of certificates mentioning specific security ' + sec_component_label + ' component (' + shortcut + ')\nAll listed occured at least once', - filter_label), 'cert_' + shortcut + '_frequency_sorted') - else: - print('ERROR: len(sars_freq_nums) < 1') - - # plot heatmap of SARs frequencies based on type (row) and level (column) - sars_labels = sorted(sars_freq.keys()) - sars_unique_names = [] - for sar in sars_labels: - if sar.find('.') != -1: - name = sar[:sar.find('.')] - else: - name = sar - if name not in sars_unique_names: - sars_unique_names.append(name) - - sars_unique_names = sorted(sars_unique_names) - max_sar_level = 8 - num_sars = len(sars_unique_names) - sar_heatmap = [] - sar_matrix = [] - for i in range(1, max_sar_level + 1): - sar_row = [] - for name in sars_unique_names: - sar_row.append(0) - sar_matrix.append(sar_row) - - for sar in sorted_by_occurence: - if sar[0].find('.') != -1: - name = sar[0][:sar[0].find('.')] - name_index = sars_unique_names.index(name) - level = int(sar[0][sar[0].find('.') + 1:]) - sar_matrix[level - 1][name_index] = sar[1] - - # plot heatmap graph with frequency of SAR levels - y_data_labels = range(1, max_sar_level + 2) - plot_heatmap_graph(sar_matrix, sars_unique_names, y_data_labels, 'Security ' + sec_component_label + ' component (' + shortcut + ') class', 'Security ' + sec_component_label + ' components (' + shortcut + ') level', fig_label('Frequency of achieved levels for Security ' + sec_component_label + ' component (' + shortcut + ') classes', filter_label), 'cert_' + shortcut + '_heatmap') - - -def generate_dot_graphs(all_items_found, filter_label): - # with name of certified items - print_dot_graph(['rules_cert_id'], all_items_found, filter_label, 'certidname_graph.dot', True, True) - # without name of certified items - print_dot_graph(['rules_cert_id'], all_items_found, filter_label, 'certid_graph.dot', True, False) - # link between device and its javacard version - print_dot_graph(['rules_javacard'], all_items_found, filter_label, 'cert_javacard_graph.dot', False, True) - - # print_dot_graph(['rules_security_level'], all_items_found, filter_label, 'cert_security_level_graph.dot', True) - # print_dot_graph(['rules_crypto_libs'], all_items_found, filter_label, 'cert_crypto_libs_graph.dot', False) - # print_dot_graph(['rules_vendor'], all_items_found, filter_label, 'rules_vendor.dot', False) - # print_dot_graph(['rules_crypto_algs'], all_items_found, filter_label, 'rules_crypto_algs.dot', False) - # print_dot_graph(['rules_protection_profiles'], all_items_found, filter_label, 'rules_protection_profiles.dot', False) - # print_dot_graph(['rules_defenses'], all_items_found, filter_label, 'rules_defenses.dot', False) - diff --git a/src/cert_rules.py b/src/cert_rules.py deleted file mode 100644 index 8623996c..00000000 --- a/src/cert_rules.py +++ /dev/null @@ -1,357 +0,0 @@ -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 - '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 - '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 - 'KECS-ISIS-[0-9]+?-[0-9][0-9][0-9][0-9]', # Korea - 'CRP-C[0-9]+?-[0-9]+?', # Japan - 'ISCB-[0-9]+?-RPT-[0-9]+?', # Malaysia - 'OCSI/CERT/.+?', # Italia - '[0-9\\.]+?/TSE-CCCS-[0-9]+?', # Turkis CCCS - 'BTBD-.+?', # Turkis CCCS -] - -rules_vendor = [ - 'NXP', - 'Infineon', - 'Samsung', - '(?:STMicroelectronics|STM)', - 'Feitian', - 'Gemalto', - 'Gemplus', - 'Axalto', - '(?:Oberthur|OBERTHUR)', - '(?:Idemia|IDEMIA)', - '(?:G\&D|G\+D|Giesecke+Devrient|Giesecke \& Devrient)', - 'Philips', - 'Sagem', -] - -rules_eval_facilities = [ - 'Serma Technologies', - 'THALES - CEACI' -] - -rules_protection_profiles = [ - 'BSI-(?:CC[-_]|)PP[-_]*.+?', - 'PP-SSCD.+?', - 'PP_DBMS_.+?' - # 'Protection Profile', - 'CCMB-20.+?', - 'CCMB-20[0-9]+?-[0-9]+?-[0-9]+?', - 'BSI-CCPP-.+?', - '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]+?', - 'FIPS ?(?:PUB )?[0-9]+?', - 'NIST SP [0-9]+-[0-9]+?[a-zA-Z]?', - 'PKCS[ #]*[1-9]+', - 'TLS[ ]*v[0-9\\.]+', - 'TLS[ ]*v[0-9\\.]+', - 'BSI-AIS[ ]*[0-9]+?', - 'AIS[ ]*[0-9]+?', - 'RFC[ ]*[0-9]+?', - 'ISO/IEC[ ]*[0-9]+[-]*[0-9]*', - 'ISO/IEC[ ]*[0-9]+:[ 0-9]+', - 'ISO/IEC[ ]*[0-9]+', - '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]|)', - r'ACM_[A-Z]{3}(?:\.[0-9]|)', - r'ACO_[A-Z]{3}(?:\.[0-9]|)', - r'ADO_[A-Z]{3}(?:\.[0-9]|)', - r'ADV_[A-Z]{3}(?:\.[0-9]|)', - r'AGD_[A-Z]{3}(?:\.[0-9]|)', - r'ALC_[A-Z]{3}(?:\.[0-9]|)', - r'ATE_[A-Z]{3}(?:\.[0-9]|)', - r'AVA_[A-Z]{3}(?:\.[0-9]|)', - r'AMA_[A-Z]{3}(?:\.[0-9]|)', - r'APE_[A-Z]{3}(?:\.[0-9]|)', - r'ASE_[A-Z]{3}(?:\.[0-9]|)' -] - -rules_security_functional_components = [ - r'FAU_[A-Z]{3}(?:\.[0-9]|)', - r'FCO_[A-Z]{3}(?:\.[0-9]|)', - r'FCS_[A-Z]{3}(?:\.[0-9]|)', - r'FDP_[A-Z]{3}(?:\.[0-9]|)', - r'FIA_[A-Z]{3}(?:\.[0-9]|)', - r'FMT_[A-Z]{3}(?:\.[0-9]|)', - r'FPR_[A-Z]{3}(?:\.[0-9]|)', - r'FPT_[A-Z]{3}(?:\.[0-9]|)', - r'FRU_[A-Z]{3}(?:\.[0-9]|)', - r'FTA_[A-Z]{3}(?:\.[0-9]|)', - r'FTP_[A-Z]{3}(?:\.[0-9]|)' -] - -rules_javacard = [ - #'(?:Java Card|JavaCard)', - #'(?:Global Platform|GlobalPlatform)', - r'(?:Java Card|JavaCard) [2-3]\.[0-9](?:\.[0-9]|)', - 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)', - 'RSASSAPKCS1-[Vv]1_5', - 'SHA[-]*(?:160|224|256|384|512)', - 'AES[-]*(?:128|192|256|)', - 'SHA-1', - 'MD5', - 'HMAC', - 'HMAC-SHA-(?:160|224|256|384|512)', - '(Diffie-Hellman|DH)', - 'ECDH', - 'ECDSA', - 'EdDSA', - '3?DES', - 'ECC', - 'DTRNG', - 'TRNG', - 'RN[GD]', - 'RBG', -] - -rules_block_cipher_modes = [ - 'ECB', - 'CBC', - 'CTR', - 'CFB', - 'OFB', - 'GCM', -] - -rules_ecc_curves = [ - 'P-(192|224|256|384|521)', - 'brainpoolP[0-9]{3}[rkt][12]', - '(sec|ansi)[pt].+?[rk][12]', - 'prime[0-9]{3}v[123]', - 'c2[pto]nb[0-9]{3}[vw][123]', - 'FRP256v1', - 'Curve(25519|1174|4417|22103|67254|383187|41417)', - 'Ed(25519|448)', - 'ssc-(160|192|224|256|288|320|384|512)', - 'Tweedle(dee|dum)', - 'JubJub', - 'BLS(12|24)-[0-9]{3}' -] - -rules_cplc = [ - 'IC[ ]*Fabricator', - 'IC[ ]*Type', - 'IC[ ]*Version', -] - -rules_crypto_engines = [ - 'TORNADO', - 'SmartMX', - 'SmartMX2' - 'NesCrypt', -] - -rules_crypto_libs = [ - '(?:NesLib|NESLIB) [v]*[0-9.]+', - 'AT1 Secure .{1,30}? Library [v]*[0-9.]+', - 'AT1 Secure RSA/ECC/SHA library', - 'Crypto Library [v]*[0-9.]+', - 'ATMEL Toolbox [0-9.]+', - 'v1.02.013' # Infineon's ROCA-vulnerable library -] - -rules_IC_data_groups = [ - r'EF\.DG[1-9][0-6]?', - r'EF\.COM', - r'EF\.CardAccess', - r'EF\.SOD', - r'EF\.ChipSecurity' -] - -rules_defenses = [ - '[Mm]alfunction', - 'Leak-Inherent', - '[Pp]hysical [Pp]robing', - '[pP]hysical [tT]ampering', - '[Ss]ide.channels?', - 'SPA', - 'DPA', - 'DFA', - '[Ff]+ault induction', - 'ROCA', -] - -rules_certification_process = [ - '[oO]ut of [sS]cope', - '[\\.\\(].{0,100}?.[oO]ut of [sS]cope..{0,100}?[\\.\\)]', - '.{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]*://.+?/' -] - -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_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 -rules['rules_protection_profiles'] = rules_protection_profiles -rules['rules_technical_reports'] = rules_technical_reports -rules['rules_device_id'] = rules_device_id -rules['rules_os'] = rules_os -rules['rules_standard_id'] = rules_standard_id -rules['rules_security_level'] = rules_security_level -rules['rules_security_assurance_components'] = rules_security_assurance_components -rules['rules_security_functional_components'] = rules_security_functional_components -rules['rules_javacard'] = rules_javacard -rules['rules_crypto_algs'] = rules_crypto_algs -rules['block_cipher_modes'] = rules_block_cipher_modes -rules['rules_ecc_curves'] = rules_ecc_curves -rules['rules_cplc'] = rules_cplc -rules['rules_crypto_engines'] = rules_crypto_engines -rules['rules_crypto_libs'] = rules_crypto_libs -rules['rules_IC_data_groups'] = rules_IC_data_groups -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 deleted file mode 100644 index a877593c..00000000 --- a/src/download_files.py +++ /dev/null @@ -1,36 +0,0 @@ -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 deleted file mode 100644 index 07ff1707..00000000 --- a/src/extract_certificates.py +++ /dev/null @@ -1,2435 +0,0 @@ -from PyPDF2 import PdfFileReader - -import sanity -from tags_constants import * -import re -import os -import operator -from graphviz import Digraph -import json -import csv -import string - -from analyze_certificates import is_in_dict -from cert_rules import rules, fips_rules -from enum import Enum -import matplotlib.pyplot as plt - -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 -APPEND_DETAILED_MATCH_MATCHES = False -VERBOSE = False -FILE_ERRORS_STRATEGY = 'surrogateescape' -'replace' -# 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 - -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] - - -def get_line_number(lines, line_length_compensation, match_start_index): - line_chars_offset = 0 - line_number = 1 - for line in lines: - line_chars_offset += len(line) + line_length_compensation - if line_chars_offset > match_start_index: - # we found the line - return line_number - line_number += 1 - # not found - return -1 - - -def load_cert_file(file_name, limit_max_lines=-1, line_separator=LINE_SEPARATOR): - lines = [] - was_unicode_decode_error = False - with open(file_name, 'r', errors=FILE_ERRORS_STRATEGY) as f: - try: - lines = f.readlines() - except UnicodeDecodeError: - f.close() - was_unicode_decode_error = True - print(' WARNING: UnicodeDecodeError, opening as utf8') - - with open(file_name, encoding="utf8", errors=FILE_ERRORS_STRATEGY) as f2: - # coding failure, try line by line - line = ' ' - while line: - try: - line = f2.readline() - lines.append(line) - except UnicodeDecodeError: - # ignore error - continue - - whole_text = '' - whole_text_with_newlines = '' - # we will estimate the line for searched matches - # => we need to known how much lines were modified (removal of eoln..) - # 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: - break - - whole_text_with_newlines += line - line = line.replace('\n', '') - whole_text += line - whole_text += line_separator - lines_included += 1 - - return whole_text, whole_text_with_newlines, was_unicode_decode_error - - -def load_cert_html_file(file_name): - with open(file_name, 'r', errors=FILE_ERRORS_STRATEGY) as f: - try: - whole_text = f.read() - except UnicodeDecodeError: - f.close() - with open(file_name, encoding="utf8", errors=FILE_ERRORS_STRATEGY) as f2: - try: - whole_text = f2.read() - except UnicodeDecodeError: - print('### ERROR: failed to read file {}'.format(file_name)) - return whole_text - - -def normalize_match_string(match): - # normalize match - match = match.strip() - match = match.rstrip(']') - match = match.rstrip(os.sep) - match = match.rstrip(';') - match = match.rstrip('.') - match = match.rstrip('”') - match = match.rstrip('"') - match = match.rstrip(':') - match = match.rstrip(')') - match = match.rstrip('(') - match = match.rstrip(',') - match = match.replace(' ', ' ') # two spaces into one - - sanitized = ''.join(filter(lambda x: x in printable, match)) - - return sanitized - - -def set_match_string(items, key_name, new_value): - if key_name not in items.keys(): - 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)) - - -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(): - if rule_group not in items_found_all: - items_found_all[rule_group] = {} - - 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_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] = {} - items_found[rule][match][TAG_MATCH_COUNTER] = 0 - if APPEND_DETAILED_MATCH_MATCHES: - items_found[rule][match][TAG_MATCH_MATCHES] = [] - # else: - # items_found[rule][match][TAG_MATCH_MATCHES] = ['List of matches positions disabled. Set APPEND_DETAILED_MATCH_MATCHES to True'] - - items_found[rule][match][TAG_MATCH_COUNTER] += 1 - match_span = m.span() - # 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]) - if APPEND_DETAILED_MATCH_MATCHES: - 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 - 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)) - 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) - for file_name_count in sorted_certid_count: - print('{:03d}: {}'.format(file_name_count[1], file_name_count[0])) - - -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(os.sep) != -1: - just_file_name = just_file_name[just_file_name.rfind(os.sep) + 1:] - print('{:30s}: {}'.format(double[1], just_file_name)) - - -def print_all_results(items_found_all): - # print results - for rule_group in items_found_all.keys(): - print(rule_group) - items_found = items_found_all[rule_group] - for rule in items_found.keys(): - print(' ' + rule) - for match in items_found[rule]: - print(' {}: {}'.format(match, items_found[rule][match])) - - -def count_num_items_found(items_found_all): - num_items_found = 0 - 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]: - num_items_found += 1 - - return num_items_found - - -def estimate_cert_id(frontpage_scan, keywords_scan, file_name): - # check if cert id was extracted from frontpage (most priority) - frontpage_cert_id = '' - if frontpage_scan != None: - if 'cert_id' in frontpage_scan.keys(): - frontpage_cert_id = frontpage_scan['cert_id'] - - keywords_cert_id = '' - if keywords_scan != None: - # find certificate ID which is the most common - num_items_found_certid_group = 0 - max_occurences = 0 - items_found = keywords_scan['rules_cert_id'] - for rule in items_found.keys(): - for match in items_found[rule]: - num_occurences = items_found[rule][match][TAG_MATCH_COUNTER] - if num_occurences > max_occurences: - max_occurences = num_occurences - 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)) - - # 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( - 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])) - filename_cert_id = matches[0] - - if VERBOSE: - print('Identified cert ids for {}:'.format(file_name)) - print(' frontpage_cert_id: {}'.format(frontpage_cert_id)) - print(' filename_cert_id: {}'.format(filename_cert_id)) - print(' keywords_cert_id: {}'.format(keywords_cert_id)) - - if frontpage_cert_id != '': - return frontpage_cert_id - if filename_cert_id != '': - return filename_cert_id - if keywords_cert_id != '': - return keywords_cert_id - - return '' - - -def save_modified_cert_file(target_file, modified_cert_file_text, is_unicode_text): - if is_unicode_text: - write_file = open(target_file, "w", encoding="utf8", errors="replace") - else: - write_file = open(target_file, "w", errors="replace") - - try: - write_file.write(modified_cert_file_text) - except UnicodeEncodeError as e: - write_file.close() - print('UnicodeDecodeError while writing file fragments back') - - write_file.close() - - -def process_raw_header(items_found): - return items_found - - -def print_specified_property_sorted(section_name, item_name, items_found_all): - specific_item_values = [] - 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]) - else: - 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) - for item in sorted_items: - print(item) - - -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_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_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 = [ - '(BSI-DSZ-CC-.+?) (?:for|For) (.+?) from (.*)', - '(BSI-DSZ-CC-.+?) zu (.+?) der (.*)', - ] - - items_found_all = {} - items_found = {} - files_without_match = [] - for file_name in search_files(walk_dir): - if not os.path.isfile(file_name): - continue - file_ext = file_name[file_name.rfind('.'):] - if file_ext != '.txt': - continue - # 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) - - for rule in rules_certificate_preface: - rule_and_sep = rule + REGEXEC_SEP - - for m in re.finditer(rule_and_sep, whole_text): - if no_match_yet: - items_found_all[file_name] = {} - items_found_all[file_name] = {} - items_found = items_found_all[file_name] - 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) - - match_groups = m.groups() - cert_id = match_groups[0] - certified_item = match_groups[1] - developer = match_groups[2] - - FROM_KEYWORD_LIST = [' from ', ' der '] - for from_keyword in FROM_KEYWORD_LIST: - from_keyword_len = len(from_keyword) - if certified_item.find(from_keyword) != -1: - 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 - - end_pos = developer.find('\f-') - if end_pos == -1: - end_pos = developer.find('\fBSI') - if end_pos == -1: - end_pos = developer.find('Bundesamt') - if end_pos != -1: - developer = developer[:end_pos] - - items_found[TAG_CERT_ID] = normalize_match_string(cert_id) - items_found[TAG_CERT_ITEM] = normalize_match_string( - certified_item) - items_found[TAG_DEVELOPER] = normalize_match_string(developer) - items_found[TAG_CERT_LAB] = 'BSI' - - # - # Process page with more detailed certificate info - # PP Conformance, Functionality, Assurance - rules_certificate_third = [ - 'PP Conformance: (.+)Functionality: (.+)Assurance: (.+)The IT Product identified', - ] - - whole_text, whole_text_with_newlines, was_unicode_decode_error = load_cert_file( - file_name) - - for rule in rules_certificate_third: - rule_and_sep = rule + REGEXEC_SEP - - for m in re.finditer(rule_and_sep, whole_text): - # check if previous rules had at least one match - if not TAG_CERT_ID in items_found.keys(): - print('ERROR: front page not found for file: {}'.format(file_name)) - - match_groups = m.groups() - ref_protection_profiles = match_groups[0] - 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) - - if no_match_yet: - files_without_match.append(file_name) - - if False: - print_found_properties(items_found_all) - - with open("certificate_data_bsiheader.json", "w", errors=FILE_ERRORS_STRATEGY) as write_file: - write_file.write(json.dumps(items_found_all, indent=4, sort_keys=True)) - - print('\n*** Certificates without detected preface:') - for file_name in files_without_match: - print('No hits for {}'.format(file_name)) - print('Total no hits files: {}'.format(len(files_without_match))) - print('\n**********************************') - - return items_found_all, files_without_match - - -def search_only_headers_anssi(walk_dir): - class HEADER_TYPE(Enum): - HEADER_FULL = 1 - HEADER_MISSING_CERT_ITEM_VERSION = 2 - HEADER_MISSING_PROTECTION_PROFILES = 3 - HEADER_DUPLICITIES = 4 - - rules_certificate_preface = [ - (HEADER_TYPE.HEADER_FULL, - 'Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.*)Conformité à un profil de protection(.+)Critères d\'évaluation et version(.+)Niveau d\'évaluation(.+)Développeurs(.+)Centre d\'évaluation(.+)Accords de reconnaissance applicables'), - (HEADER_TYPE.HEADER_FULL, - 'Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.*)Conformité à un profil de protection(.+)Critères d’évaluation et version(.+)Niveau d’évaluation(.+)Développeurs(.+)Centre d’évaluation(.+)Accords de reconnaissance applicables'), - (HEADER_TYPE.HEADER_FULL, - 'Référence du rapport de certification(.+)Nom du produit(.+)()Conformité à un profil de protection(.+)Critères d\'évaluation et version(.+)Niveau d\'évaluation(.+)Développeur (.+)Centre d\'évaluation(.+)Accords de reconnaissance applicables'), - (HEADER_TYPE.HEADER_FULL, - 'Référence du rapport de certification(.+)Nom des produits(.+)Référence/version des produits(.+)Conformité à un profil de protection(.+)Critères d\'évaluation et version(.+)Niveau d\'évaluation(.+)Développeur\(s\)(.+)Centre d\'évaluation(.+)Accords de reconnaissance applicables'), - (HEADER_TYPE.HEADER_FULL, - 'Référence du rapport de certification(.+)Nom des produits(.+)Référence/version des produits(.+)Conformité à un profil de protection(.+)Critères d\'évaluation et version(.+)Niveau d\'évaluation(.+)Développeur (.+)Centre d\'évaluation(.+)Accords de reconnaissance'), - (HEADER_TYPE.HEADER_FULL, - 'Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.+)Conformité aux profils de protection(.+)Critères d’évaluation et version(.+)Niveau d’évaluation(.+)Développeur\(s\)(.+)Centre d’évaluation(.+)Accords de reconnaissance applicables'), - (HEADER_TYPE.HEADER_FULL, - 'Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.+)Conformité à un profil de protection(.+)Critères d’évaluation et version(.+)Niveau d’évaluation(.+)Développeur\(s\)(.+)Centre d’évaluation(.+)Accords de reconnaissance applicables'), - (HEADER_TYPE.HEADER_FULL, - 'Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.+)Conformité à un profil de protection(.+)Critères d’évaluation et version(.+)Niveau d’évaluation(.+)Développeur (.+)Centre d’évaluation(.+)Accords de reconnaissance applicables'), - (HEADER_TYPE.HEADER_FULL, - 'Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.+)Conformité à des profils de protection(.+)Critères d’évaluation et version(.+)Niveau d’évaluation(.+)Développeurs(.+)Centre d’évaluation(.+)Accords de reconnaissance applicables'), - (HEADER_TYPE.HEADER_FULL, - 'Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.+)Conformité aux profils de protection(.+)Critères d\’évaluation et version(.+)Niveau d\’évaluation(.+)Développeurs(.+)Centre d\’évaluation(.+)Accords de reconnaissance applicables'), - (HEADER_TYPE.HEADER_FULL, - 'Référence du rapport de certification(.+)Nom du produit \(référence/version\)(.+)Nom de la TOE \(référence/version\)(.+)Conformité à un profil de protection(.+)Critères d\’évaluation et version(.+)Niveau d\’évaluation(.+)Développeurs(.+)Centre d’évaluation(.+)Accords de reconnaissance applicables'), - (HEADER_TYPE.HEADER_FULL, - 'Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.+)Conformité aux profil de protection(.+)Critères d’évaluation et version(.+)Niveau d’évaluation(.+)Développeur\(s\)(.+)Centre d’évaluation(.+)Accords de reconnaissance applicables'), - (HEADER_TYPE.HEADER_FULL, - 'Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.+)Conformité à un profil de protection(.+)Critères d\'évaluation et version(.+)Niveau d\'évaluation(.+)Développeur\(s\)(.+)Centre d\'évaluation(.+)Accords de reconnaissance applicables'), - (HEADER_TYPE.HEADER_FULL, - 'Référence du rapport de certification(.+)Nom du produit \(référence/version\)(.+)Nom de la TOE \(référence/version\)(.+)Conformité à un profil de protection(.+)Critères d\'évaluation et version(.+)Niveau d\'évaluation(.+)Développeurs(.+)Centre d\'évaluation(.+)Accords de reconnaissance applicables'), - (HEADER_TYPE.HEADER_FULL, - 'Référence du rapport de certification(.+)Nom du produit(.+)Référence du produit(.+)Conformité à un profil de protection(.+)Critères d\'évaluation et version(.+)Niveau d\'évaluation(.+)Développeurs(.+)Centre d\'évaluation(.+)Accords de reconnaissance applicables'), - (HEADER_TYPE.HEADER_FULL, - 'Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.+)Conformité aux profils de protection(.+)Critères d\'évaluation et version(.+)Niveau d\'évaluation(.+)Développeurs(.+)Centre d\'évaluation(.+)Accords de reconnaissance applicables'), - - (HEADER_TYPE.HEADER_FULL, - 'Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.+)Conformité à un profil de protection(.+)Critères d’évaluation et version(.+)Niveau d’évaluation(.+)Développeurs(.+)Centre d’évaluation(.+)Accords de reconnaissance applicables'), - (HEADER_TYPE.HEADER_FULL, - 'Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.+)Conformité à un profil de protection(.+)Critères d’évaluation et version(.+)Niveau d’évaluation(.+)Développeur\(s\)(.+)d’évaluation(.+)Accords de reconnaissance applicables'), - (HEADER_TYPE.HEADER_FULL, - 'Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.+)Conformité à un profil de protection(.+)Critères d’évaluation et version(.+)Niveau d’évaluation(.+)Développeur (.+)Centre d’évaluation(.+)Accords de reconnaissance applicables'), - (HEADER_TYPE.HEADER_FULL, - 'Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.+)Conformité à des profils de protection(.+)Critères d’évaluation et version(.+)Niveau d’évaluation(.+)Développeurs(.+)Centre d’évaluation(.+)Accords de reconnaissance applicables'), - (HEADER_TYPE.HEADER_FULL, - 'Référence du rapport de certification(.+)Nom du produit \(référence/version\)(.+)Nom de la TOE \(référence/version\)(.+)Conformité à un profil de protection(.+)Critères d’évaluation et version(.+)Niveau d’évaluation(.+)Développeurs(.+)Centre d’évaluation(.+)Accords de reconnaissance applicables'), - (HEADER_TYPE.HEADER_FULL, - 'Certification Report(.+)Nom du produit(.+)Référence/version du produit(.*)Conformité à un profil de protection(.+)Critères d\'évaluation et version(.+)Niveau d\'évaluation(.+)Développeurs(.+)Centre d\'évaluation(.+)Accords de reconnaissance applicables'), - (HEADER_TYPE.HEADER_FULL, - 'Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.+)Conformité aux profisl de protection(.+)Critères d’évaluation et version(.+)Niveau d’évaluation(.+)Développeurs(.+)Centre d’évaluation(.+)Accords de reconnaissance applicables'), - (HEADER_TYPE.HEADER_FULL, - 'Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.+)Conformité à un profil de protection(.+)Critères d’évaluation et version(.+)Niveau d’évaluation(.+)Développeur (.+)Centres d’évaluation(.+)Accords de reconnaissance applicables'), - (HEADER_TYPE.HEADER_FULL, - 'Référence du rapport de certification(.+)Nom du produit(.+)Version du produit(.+)Conformité à un profil de protection(.+)Critères d’évaluation et version(.+)Niveau d’évaluation(.+)Développeur (.+)Centre d’évaluation(.+)Accords de reconnaissance applicables'), - (HEADER_TYPE.HEADER_FULL, - 'Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.+)Conformité aux profils de protection(.+)Critères d’évaluation et version(.+)Niveau d’évaluation(.+)Développeur\(s\)(.+)Centre d’évaluation(.+)Accords de reconnaissance applicables'), - (HEADER_TYPE.HEADER_FULL, - 'Référence du rapport de certification(.+)Nom du produit(.+)Versions du produit(.+)Conformité à un profil de protection(.+)Critères d’évaluation et version(.+)Niveau d’évaluation(.+)Développeur (.+)Centre d’évaluation(.+)Accords de reconnaissance applicables'), - (HEADER_TYPE.HEADER_FULL, - 'Référence du rapport de certification(.+)Nom du produit(.+)Référence du produit(.+)Conformité à un profil de protection(.+)Critères d’évaluation et version(.+)Niveau d’évaluation(.+)Développeurs(.+)Centre d’évaluation(.+)Accords de reconnaissance applicables'), - (HEADER_TYPE.HEADER_FULL, - 'Certification report reference(.+)Product name(.+)Product reference(.+)Protection profile conformity(.+)Evaluation criteria and version(.+)Evaluation level(.+)Developer (.+)Evaluation facility(.+)Recognition arrangements'), - (HEADER_TYPE.HEADER_FULL, - 'Certification report reference(.+)Product name(.+)Product reference(.+)Protection profile conformity(.+)Evaluation criteria and version(.+)Evaluation level(.+)Developer (.+)Evaluation facility(.+)Mutual Recognition Agreements'), - (HEADER_TYPE.HEADER_FULL, - 'Certification report reference(.+)Product name(.+)Product reference(.+)Protection profile conformity(.+)Evaluation criteria and version(.+)Evaluation level(.+)Developers(.+)Evaluation facility(.+)Recognition arrangements'), - (HEADER_TYPE.HEADER_FULL, - 'Certification report reference(.+)Product name(.+)Product reference(.+)Protection profile conformity(.+)Evaluation criteria and version(.+)Evaluation level(.+)Developer\(s\)(.+)Evaluation facility(.+)Recognition arrangements'), - (HEADER_TYPE.HEADER_FULL, - 'Certification report reference(.+)Products names(.+)Products references(.+)protection profile conformity(.+)Evaluation criteria and version(.+)Evaluation level(.+)Developers(.+)Evaluation facility(.+)Recognition arrangements'), - (HEADER_TYPE.HEADER_FULL, - 'Certification report reference(.+)Product name \(reference / version\)(.+)TOE name \(reference / version\)(.+)Protection profile conformity(.+)Evaluation criteria and version(.+)Evaluation level(.+)Developers(.+)Evaluation facility(.+)Recognition arrangements'), - (HEADER_TYPE.HEADER_FULL, - 'Certification report reference(.+)TOE name(.+)Product\'s reference/ version(.+)TOE\'s reference/ version(.+)Conformité à un profil de protection(.+)Evaluation criteria and version(.+)Evaluation level(.+)Developer (.+)Evaluation facility(.+)Recognition arrangements'), - - # corrupted text (duplicities) - (HEADER_TYPE.HEADER_DUPLICITIES, - 'Référencce du rapport de d certification n(.+)Nom du p produit(.+)Référencce/version du produit(.+)Conformiité à un profil de d protection(.+)Critères d d’évaluation ett version(.+)Niveau d’’évaluation(.+)Développ peurs(.+)Centre d’’évaluation(.+)Accords d de reconnaisssance applicab bles'), - - # rules without product version - (HEADER_TYPE.HEADER_MISSING_CERT_ITEM_VERSION, - 'Référence du rapport de certification(.+)Nom et version du produit(.+)Conformité à un profil de protection(.+)Critères d\'évaluation et version(.+)Niveau d\'évaluation(.+)Développeurs(.+)Centre d\'évaluation(.+)Accords de reconnaissance applicables'), - (HEADER_TYPE.HEADER_MISSING_CERT_ITEM_VERSION, - 'Référence du rapport de certification(.+)Nom et version du produit(.+)Conformité à un profil de protection(.+)Critères d\'évaluation et version(.+)Niveau d\'évaluation(.+)Développeur (.+)Centre d\'évaluation(.+)Accords de reconnaissance applicables'), - (HEADER_TYPE.HEADER_MISSING_CERT_ITEM_VERSION, - 'Référence du rapport de certification(.+)Nom du produit(.+)Conformité à un profil de protection(.+)Critères d\'évaluation et version(.+)Niveau d\'évaluation(.+)Développeurs(.+)Centre d\'évaluation(.+)Accords de reconnaissance applicables'), - - # rules without protection profile - (HEADER_TYPE.HEADER_MISSING_PROTECTION_PROFILES, - 'Référence du rapport de certification(.+)Nom du produit(.+)Référence/version du produit(.+)Critères d\'évaluation et version(.+)Niveau d\'évaluation(.+)Développeurs(.+)Centre d\'évaluation(.+)Accords de reconnaissance applicables'), - ] - - # 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): - if not os.path.isfile(file_name): - continue - file_ext = file_name[file_name.rfind('.'):] - if file_ext != '.txt': - continue - # print('*** {} ***'.format(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('') - if pos != -1: - pos = whole_text.find('', pos) - if pos != -1: - whole_text = whole_text[pos:] - - no_match_yet = True - other_rule_already_match = False - other_rule = '' - rule_index = -1 - for rule in rules_certificate_preface: - rule_index += 1 - rule_and_sep = rule[1] + REGEXEC_SEP - - for m in re.finditer(rule_and_sep, whole_text): - if no_match_yet: - items_found_all[file_name] = {} - items_found_all[file_name] = {} - items_found = items_found_all[file_name] - 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 not other_rule_already_match: - other_rule_already_match = True - other_rule = rule - else: - print( - 'WARNING: multiple rules are matching same certification document: ' + file_name) - - num_rules_hits[rule[1]] += 1 # add hit to this rule - - match_groups = m.groups() - - index_next_item = 0 - - 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]) - 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]) - 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]) - index_next_item += 1 - - 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]) - index_next_item += 1 - - 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]) - index_next_item += 1 - - if no_match_yet: - files_without_match.append(file_name) - - if False: - print_found_properties(items_found_all) - - # store results into file with fixed name and also with time appendix - with open("certificate_data_anssiheader.json", "w", errors=FILE_ERRORS_STRATEGY) as write_file: - write_file.write(json.dumps(items_found_all, indent=4, sort_keys=True)) - - print('\n*** Certificates without detected preface:') - for file_name in files_without_match: - print('No hits for {}'.format(file_name)) - print('Total no hits files: {}'.format(len(files_without_match))) - print('\n**********************************') - - if True: - print('# hits for rule') - sorted_rules = sorted(num_rules_hits.items(), - key=operator.itemgetter(1), reverse=True) - used_rules = [] - for rule in sorted_rules: - print('{:4d} : {}'.format(rule[1], rule[0])) - if rule[1] > 0: - used_rules.append(rule[0]) - - return items_found_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) - - print('*** Files without detected header') - 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))) - - items_found_all = {**anssi_items_found, **bsi_items_found} - # store results into file with fixed name and also with time appendix - - if write_output_file: - with open("certificate_data_frontpage_all.json", "w", errors=FILE_ERRORS_STRATEGY) as write_file: - write_file.write(json.dumps( - items_found_all, indent=4, sort_keys=True)) - - return items_found_all - - -def search_pp_only_headers(walk_dir): - # LINE_SEPARATOR_STRICT = ' ' - # NUM_LINES_TO_INVESTIGATE = 15 - # rules_certificate_preface = [ - # '(Common Criteria Protection Profile .+)?(BSI-PP-CC-.+?)Federal Office for Information Security', - # '(Protection Profile for the .+)?Schutzprofil für das .+?Certification-ID (BSI-CC-PP-[0-9]+?) ', - # # 'Protection Profile for the Security Module of a Smart Meter Mini-HSM (Mini-HSM Security Module PP) Schutzprofil für das Sicherheitsmodul des Smart Meter Mini-HSM Mini-HSM SecMod-PP Version 1.0 – 23 June 2017 Certification-ID BSI-CC-PP-0095 Mini-HSM Security Module PP Bundesamt für Sicherheit in der Informationstechnik' - # ] - - class HEADER_TYPE(Enum): - BSI_TYPE1 = 1 - BSI_TYPE2 = 2 - - DCSSI_TYPE1 = 11 - DCSSI_TYPE2 = 12 - FRONT_DCSSI_TYPE3 = 13 - FRONT_DCSSI_TYPE4 = 14 - DCSSI_TYPE5 = 15 - DCSSI_TYPE6 = 16 - - ANSSI_TYPE1 = 41 - 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'), - (HEADER_TYPE.BSI_TYPE2, - 'PP Reference.+?Title: (.+)?Version: (.+)?Date: (.+)?Authors: (.+)?Registration: (.+)?Certification-ID: (.+)?Evaluation Assurance Level: (.+)?CC Version: (.+)?Keywords: (.+)?Specific Terms'), - (HEADER_TYPE.ANSSI_TYPE1, - 'PROTECTION PROFILE IDENTIFICATION.+?Title: (.+)?Version: (.+)?Publication date: (.+)?Certified by: (.+)?Sponsor: (.+)?Editor: (.+)?Review Committee: (.+)?This Protection Profile is conformant to the Common Criteria version (.+)?The minimum assurance level for this Protection Profile is (.+)?PROTECTION PROFILE PRESENTATION'), - (HEADER_TYPE.ANSSI_TYPE2, - 'PP reference.+?Title : (.+)?Version : (.+)?Authors : (.+)?Evaluation Assurance Level : (.+)?Registration : (.+)?Conformant to Version (.+)?of Common Criteria.+?Key words : (.+)?A glossary of terms'), - (HEADER_TYPE.ANSSI_TYPE3, - 'Introduction.+?Title: (.+)?Identifications: (.+)?Editor: (.+)?Date: (.+)?Version: (.+)?Sponsor: (.+)?CC Version: (.+)? This Protection Profile'), - (HEADER_TYPE.DCSSI_TYPE1, - 'Protection profile reference[ ]*Title: (.+)?Reference: (.+)?, Version (.+)?, (.+)?Author: (.+)?Context'), - (HEADER_TYPE.DCSSI_TYPE2, - '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 (.+)?(?: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' - (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') - ] - print("***PP HEADER SEARCH***") - items_found_all = {} - files_without_match = [] - for file_name in search_files(walk_dir): - if not os.path.isfile(file_name): - continue - file_ext = file_name[file_name.rfind('.'):] - if file_ext != '.txt': - continue - # 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) - - no_match_yet = True - for rule in rules_pp_third: - rule_and_sep = rule[1] + REGEXEC_SEP - - for m in re.finditer(rule_and_sep, whole_text): - if no_match_yet: - items_found_all[file_name] = {} - items_found_all[file_name] = {} - items_found = items_found_all[file_name] - items_found[TAG_HEADER_MATCH_RULES] = [] - no_match_yet = False - - # insert rule if at least one match for it was found - if rule[1] not in items_found[TAG_HEADER_MATCH_RULES]: - items_found[TAG_HEADER_MATCH_RULES].append(rule[1]) - - match_groups = m.groups() - index = 0 - - if rule[0] == HEADER_TYPE.BSI_TYPE1: - 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])) - index += 1 - 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])) - index += 1 - 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])) - index += 1 - keywords = match_groups[index].lstrip(' ') - 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') - - if rule[0] == HEADER_TYPE.BSI_TYPE2: - 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])) - index += 1 - 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])) - index += 1 - 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])) - index += 1 - 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])) - index += 1 - keywords = match_groups[index].lstrip(' ') - 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') - - if rule[0] == HEADER_TYPE.ANSSI_TYPE1: - 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])) - index += 1 - 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])) - index += 1 - 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])) - index += 1 - 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])) - index += 1 - level = match_groups[index].lstrip(' ') - 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') - - if rule[0] == HEADER_TYPE.ANSSI_TYPE2: - 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])) - index += 1 - 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])) - index += 1 - 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])) - index += 1 - 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') - - if rule[0] == HEADER_TYPE.ANSSI_TYPE3: - 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])) - index += 1 - 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])) - index += 1 - 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])) - index += 1 - ccversion = match_groups[index].lstrip(' ') - 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') - - if rule[0] == HEADER_TYPE.DCSSI_TYPE1: - 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])) - index += 1 - 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])) - index += 1 - author = match_groups[index].lstrip(' ') - 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') - - if rule[0] == HEADER_TYPE.DCSSI_TYPE2: - 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])) - index += 1 - version = match_groups[index].lstrip(' ') - 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') - - 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])) - index += 1 - 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])) - index += 1 - 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])) - index += 1 - - 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])) - index += 1 - 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])) - index += 1 - 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])) - index += 1 - ccversion = match_groups[index].lstrip(' ') - 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])) - index += 1 - 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])) - index += 1 - 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])) - index += 1 - 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])) - index += 1 - - set_match_string( - items_found, TAG_PP_REGISTRATOR_SIMPLIFIED, 'DCSSI') - - if no_match_yet: - files_without_match.append(file_name) - - if False: - print_found_properties(items_found_all) - - with open("pp_data_header.json", "w", errors=FILE_ERRORS_STRATEGY) as write_file: - write_file.write(json.dumps(items_found_all, indent=4, sort_keys=True)) - - print('\n*** Protection profiles without detected header:') - for file_name in files_without_match: - print('No hits for {}'.format(file_name)) - print('Total no hits files: {}'.format(len(files_without_match))) - print('\n**********************************') - - return items_found_all, files_without_match - - -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') - for file_name in pp_files_without_match: - print(file_name) - print('Total no hits files: {}'.format(len(pp_files_without_match))) - - # store results into file with fixed name and also with time appendix - if write_output_file: - with open("pp_data_frontpage_all.json", "w", errors=FILE_ERRORS_STRATEGY) as write_file: - 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, - 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 = {} - for file_name in search_files(walk_dir): - if not os.path.isfile(file_name): - continue - file_ext = file_name[file_name.rfind('.'):] - if file_ext != '.txt': - continue - - # 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, 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_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(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 - 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)) - - # print('\nTotal matches found in separate files:') - # print_total_matches_in_files(all_items_found_count) - - # 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) - - total_items_found = 0 - for file_name in all_items_found: - total_items_found += count_num_items_found(all_items_found[file_name]) - - PRINT_MATCHES = False - 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: - print(match) - # all_matches.append(match) - - sorted_all_matches = sorted(all_matches) - # for match in sorted_all_matches: - # print(match) - - # verify total matches found - print('\nTotal matches found: {}'.format(total_items_found)) - - return all_items_found - - -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): - if not os.path.isfile(file_name): - continue - file_ext = file_name[file_name.rfind('.'):] - if file_ext != '.pdf': - continue - - print("***EXTRACT PDFMETA***") - # print('*** {} ***'.format(file_name)) - - item = {} - item['pdf_file_size_bytes'] = os.path.getsize(file_name) - try: - with open(file_name, 'rb') as f: - pdf = PdfFileReader(f) - # store additional interesting info - item['pdf_is_encrypted'] = pdf.getIsEncrypted() - item['pdf_number_of_pages'] = pdf.getNumPages() - - # extract pdf metadata (as dict) and save it - info = pdf.getDocumentInfo() - if info is not None: - for key in info: - item[key] = str(info[key]) - except Exception as e: - item['error'] = str(e) - - # test save of the data extracted to prevent error only very later - # try: - # with open("{}_temp.json".format(file_prefix), "w") as write_file: - # write_file.write(json.dumps(item, indent=4, sort_keys=True)) - # except Exception: - # print(' ERROR: invalid data from pdf') - - all_items_found[file_name] = item - - 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)) - 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)) - - return all_items_found - - -def extract_file_name_from_url(url): - file_name = url[url.rfind('/') + 1:] - file_name = file_name.replace('%20', ' ') - return file_name - - -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' - rule_without_maintainance_ST = '.*?([0-9]+?-[0-9]+?-[0-9]+?) (.+?)\<br style=' \ - '.*?\<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) - while start_pos != -1: - end_pos = updates_chunk.find('</li>', start_pos) - report_chunk = updates_chunk[start_pos:end_pos] - - start_pos = updates_chunk.find('<li>', end_pos) - - # decide which search rule to use 1) one that matches also Maintenance ST or 2) without it - if report_chunk.find('Maintenance ST') != -1: - rule = rule_with_maintainance_ST - else: - rule = rule_without_maintainance_ST - - items_found = {} - 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]) - index_next_item += 1 - 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]) - 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]) - 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']) - items_found['link_cert_report_file_name'] = cert_file_name - 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)) - - maintenance_reports.append(items_found) - - return maintenance_reports - - -def parse_security_level(security_level): - start_pos = security_level.find('<br>') - eal_level = security_level - eal_augmented = [] - if start_pos != -1: - eal_level = normalize_match_string(security_level[:start_pos]) - # some augmented items found - augm_chunk = security_level[start_pos:] - augm_chunk += ' ' - # 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() - eal_augmented.append(normalize_match_string(match_groups[0])) - - return eal_level, eal_augmented - - -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)) - - whole_text = load_cert_html_file(file_name) - - whole_text = whole_text.replace('\n', ' ') - whole_text = whole_text.replace(' ', ' ') - whole_text = whole_text.replace('&', '&') - - # First find end extract chunks between <tr class=""> ... </tr> - start_pos = whole_text.find('<tfoot class="hilite7"') - start_pos = whole_text.find('<tr class="', start_pos) - - chunks_found = 0 - chunks_matched = 0 - - while start_pos != -1: - end_pos = whole_text.find('</tr>', start_pos) - - chunk = whole_text[start_pos:end_pos] - - even_start_pos = whole_text.find('<tr class="even">', start_pos + 1) - odd_start_pos = whole_text.find('<tr class="">', start_pos + 1) - - start_pos = min(even_start_pos, odd_start_pos) - - # skip chunks which are not cert item chunks - if chunk.find('This list was generated on') != -1: - continue - - chunks_found += 1 - - class HEADER_TYPE(Enum): - HEADER_FULL = 1 - HEADER_MISSING_VENDOR_WEB = 2 - - # 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_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 - - rule_and_sep = rule[1] - - for m in re.finditer(rule_and_sep, chunk): - if no_match_yet: - chunks_matched += 1 - 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]) - - match_groups = m.groups() - - index_next_item = 0 - 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]) - 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]) - index_next_item += 1 - 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_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_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)) - index_next_item += 1 - - 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]) - index_next_item += 1 - 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) - 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) - 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 - else: - print('{} already in'.format(cert_file_name)) - - continue # we are interested only in first match - - if no_match_yet: - print('No match found in block #{}'.format(chunks_found)) - - print('Chunks found: {}, Chunks matched: {}'.format( - chunks_found, chunks_matched)) - if chunks_found != chunks_matched: - print('WARNING: not all chunks found were matched') - - return items_found_all, download_files_certs, download_files_updates - - -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 sanity.STOP_ON_UNEXPECTED_NUMS: - 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)) - 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)}') - line_count += 1 - else: - if no_further_maintainance: - items_found = {} - 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)) - # 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 - if row[13].find('http://') != -1: - # name - row[11] = row[11] + row[12] - row.remove(row[12]) - - # check if some maintainance reports are present. If yes, then extract these to list of updates - if len(row[10]) > 0: - no_further_maintainance = False - else: - no_further_maintainance = True - - 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]) - 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]) - 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]) - 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]) - 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]) - 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]) - 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]) - 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]) - 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]) - 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) - 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']) - items_found['link_security_target_file_name'] = st_file_name - index_next_item += 1 - - if 'maintainance_updates' not in items_found: - items_found['maintainance_updates'] = [] - - maintainance = {} - 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]) - index_next_item += 1 - 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]) - index_next_item += 1 - # add this maintainance to parent item only when not empty - if len(maintainance['cc_maintainance_title']) > 0: - items_found['maintainance_updates'].append(maintainance) - - if no_further_maintainance: - # 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) - 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 sanity.STOP_ON_UNEXPECTED_NUMS: - raise ValueError( - 'ERROR: Stopping as value is not unique') - - line_count += 1 - - return items_found_all - - -def fix_pp_url(original_url): - # 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', '') - return original_url - - -def extract_pp_metadata_csv(file_name): - items_found_all = {} - download_files_certs = [] - download_files_maintainance = [] - expected_columns = -1 - with open(file_name, errors=FILE_ERRORS_STRATEGY) as csv_file: - 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) - line_count += 1 - else: - if no_further_maintainance: - items_found = {} - 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)) - # trying to fix - if len(row) == expected_columns + 2: - row[9] = row[9] + row[10] + row[11] - row[10] = row[12] - row[11] = row[13] - del row[13] - del row[12] - - # check if some maintainance reports (based on presence of maintainance date - row[8]) are present. - # If yes, then extract these to list of updates - if len(row[8]) > 0: - no_further_maintainance = False - else: - no_further_maintainance = True - - 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]) - 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]) - 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]) - 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]) - 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]) - 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]) - 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']) - 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']) - index_next_item += 1 - 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]) - index_next_item += 1 - 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']) - index_next_item += 1 - - # add this maintainance to parent item only when not empty - if len(maintainance['cc_pp_maintainance_title']) > 0: - items_found['maintainance_updates'].append(maintainance) - - 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', ' ') - item_unique_name = pp_document_file_name - 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)) - if sanity.STOP_ON_UNEXPECTED_NUMS: - 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, - items_found['link_pp_document'], pp_document_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)) - - line_count += 1 - - return items_found_all, download_files_certs, download_files_maintainance - - -def generate_download_script(file_name, certs_dir, targets_dir, base_url, download_files_certs): - with open(file_name, "w", errors=FILE_ERRORS_STRATEGY) as write_file: - # certs files - if certs_dir != '': - write_file.write('mkdir \"{}\"\n'.format(certs_dir)) - write_file.write('cd \"{}\"\n\n'.format(certs_dir)) - for cert in download_files_certs: - # double %% is necessary to prevent replacement of %2 within script (second argument of script) - file_name_short_web = cert[0].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[1])) - else: - # insert base url - 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: - # security targets file - if targets_dir != '': - write_file.write('\n\ncd ..\n') - write_file.write('mkdir \"{}\"\n'.format(targets_dir)) - write_file.write('cd \"{}\"\n\n'.format(targets_dir)) - for cert in download_files_certs: - # double %% is necessary to prevent replacement of %2 within script (second argument of script) - 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])) - 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])) - - -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) - 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)) - - 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) - 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)) - - 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)) - - return items_found_all - - -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(): - items_found_all_active[item]['csv_scan']['cert_status'] = 'active' - - file_name = '{}cc_products_archived.csv'.format(base_dir) - items_found_all_archived = extract_certificates_metadata_csv(file_name) - for item in items_found_all_archived.keys(): - items_found_all_archived[item]['csv_scan']['cert_status'] = 'archived' - - items_found_all = {**items_found_all_active, **items_found_all_archived} - - if write_output_file: - with open("certificate_data_csv_all.json", "w", errors=FILE_ERRORS_STRATEGY) as write_file: - 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): - 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) - 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) - - 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) - 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) - - 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)) - - return items_found_all - - -def check_expected_cert_results(all_html, all_csv, all_front, all_keywords, all_pdf_meta): - # CSV - sanity.check_certs_min_items_found_csv(len(all_csv)) - # HTML - sanity.check_certs_min_items_found_html(len(all_html)) - # FRONTPAGE - sanity.check_certs_min_items_found_frontpage(len(all_front)) - # KEYWORDS - total_items_found = 0 - for file_name in all_keywords.keys(): - total_items_found += count_num_items_found(all_keywords[file_name]) - sanity.check_certs_min_items_found_keywords(total_items_found) - - -def check_expected_pp_results(all_html, all_csv, all_front, all_keywords): - # CSV - sanity.check_pp_min_items_found_csv(len(all_csv)) - # HTML - sanity.check_pp_min_items_found_html(len(all_html)) - # FRONTPAGE - sanity.check_pp_min_items_found_frontpage(len(all_front)) - # KEYWORDS - total_items_found = 0 - for file_name in all_keywords.keys(): - total_items_found += count_num_items_found(all_keywords[file_name]) - sanity.check_pp_min_items_found_keywords(total_items_found) - - -def collate_certificates_data(all_html, all_csv, all_front, all_keywords, all_pdf_meta, file_name_key): - print('\n\nPairing results from different scans ***') - - file_name_to_html_name_mapping = {} - for long_file_name in all_html.keys(): - 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(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(os.sep) + 1:] - if short_file_name != '': - 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(os.sep) + 1:] - if short_file_name != '': - 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 - for file_name in all_csv.keys(): - pairing_found = False - - 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'] - 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_txt = file_name_st[:file_name_st.rfind('.')] + '.txt' - else: - file_name_st_txt = 'security_target_which_doesnt_exists' - - # for file_and_id in all_html.keys(): - # # in items extracted from html, names are in form of 'file_name.pdf__number' - # if file_and_id.find(file_name_pdf + '__') != -1: - if 'processed' not in all_cert_items[file_name].keys(): - all_cert_items[file_name]['processed'] = {} - pairing_found = True - frontpage_scan = None - keywords_scan = None - - 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 - else: - 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]] - # 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]] - # 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]] - # 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) - - # pair pairing in maintainance updates - for file_name in all_csv.keys(): - pairing_found = False - - # 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_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_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' - 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( - 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])) - # 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]] - 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])) - # 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)) - - for file_and_id in file_name_to_pdfmeta_name_mapping.keys(): - 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' - # BUGBUG: mismatch in character case will result in missing paiing (e.g., st_vid3014-vr.pdf) - 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])) - # 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])) - # 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('*** 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 - print(' {}'.format(file_name_to_keywords_name_mapping[item][0])) - - # display all record which were not paired - print('\n\nRecords with missing pairing of frontpage:') - num_frontpage_missing = 0 - for item in all_cert_items.keys(): - this_item = all_cert_items[item] - if 'frontpage_scan' not in this_item.keys(): - print('WARNING: {} no frontpage scan detected'.format(item)) - num_frontpage_missing += 1 - - print('\n\nRecords with missing pairing of keywords:') - num_keywords_missing = 0 - for item in all_cert_items.keys(): - this_item = all_cert_items[item] - if 'keywords_scan' not in this_item.keys(): - print('WARNING: {} no keywords scan detected'.format(item)) - num_keywords_missing += 1 - - print('\n\nRecords with missing pairing of pdfmeta:') - num_pdfmeta_missing = 0 - for item in all_cert_items.keys(): - this_item = all_cert_items[item] - if 'pdfmeta_scan' not in this_item.keys(): - 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)) - - return all_cert_items - - -def get_manufacturer_simple_name(long_manufacturer, reduction_list): - if long_manufacturer in reduction_list: - return reduction_list[long_manufacturer] - else: - return long_manufacturer - - -def build_pp_id_mapping(all_pp_items): - # read mapping between protection profile id as used in CSV and - # key used in pp_data_complete_processed.json - pp_id_mapping = {} - for pp in all_pp_items: - if is_in_dict(all_pp_items[pp], ['pp_analysis', 'separate_profiles']): - for profile in all_pp_items[pp]['pp_analysis']['separate_profiles']: - pp_id_mapping[profile['pp_id_csv']] = pp - - return pp_id_mapping - - -def process_certificates_data(all_cert_items, all_pp_items): - print('\n\nExtracting useful info from collated files ***') - - # - # Process 'cc_manufacturer' CSV field - # 1. separate multiple manufacturers (',' '-' '/' 'and') - # 2. map different names of a same manufacturer to the same - manufacturers = [] - for file_name in all_cert_items.keys(): - cert = all_cert_items[file_name] - # extract manufacturer - if is_in_dict(cert, ['csv_scan', 'cc_manufacturer']): - manufacturer = cert['csv_scan']['cc_manufacturer'] - - if manufacturer != '': - if manufacturer not in manufacturers: - manufacturers.append(manufacturer) - - sorted_manufacturers = sorted(manufacturers) - for manuf in sorted_manufacturers: - print('{}'.format(manuf)) - - print('\n\n') - mapping_csvmanuf_separated = {} - for manuf in sorted_manufacturers: - mapping_csvmanuf_separated[manuf] = [] - - for manuf in sorted_manufacturers: - # 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'] - multiple_manuf_detected = False - for sep in separators: - list_manuf = manuf.split(sep) - for i in range(0, len(list_manuf)): - list_manuf[i] = list_manuf[i].strip() - if len(list_manuf) > 1: - all_separated_exists = True - for separated_manuf in list_manuf: - if separated_manuf in sorted_manufacturers: - continue - else: - print('Problematic separator \'{}\' in {}'.format(sep, manuf)) - all_separated_exists = False - break - - if all_separated_exists: - for x in list_manuf: - mapping_csvmanuf_separated[manuf].append(x) - multiple_manuf_detected = True - - if not multiple_manuf_detected: - mapping_csvmanuf_separated[manuf].append(manuf) - - print('### Multiple manufactures detected and split:') - for manuf in mapping_csvmanuf_separated: - if len(mapping_csvmanuf_separated[manuf]) > 1: - print(' {}:{}'.format(manuf, mapping_csvmanuf_separated[manuf])) - - manuf_starts = {} - already_reduced = {} - 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)) - 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])) - - # 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('node', style='filled') - already_inserted_edges = [] - for file_name in all_cert_items.keys(): - cert = all_cert_items[file_name] - if is_in_dict(cert, ['csv_scan', 'cc_manufacturer']): - 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) - 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') - already_inserted_edges.append(edge_name) - - # plot naming hierarchies - file_name = 'manufacturer_naming_dependency.dot' - dot.render(file_name, view=False) - print('{} pdf rendered'.format(file_name)) - - pp_id_mapping = build_pp_id_mapping(all_pp_items) - - # update dist with processed list of manufactures - all_cert_items_keys = list(all_cert_items.keys()) - for file_name in all_cert_items_keys: - cert = all_cert_items[file_name] - # extract manufacturer - if is_in_dict(cert, ['csv_scan', 'cc_manufacturer']): - manufacturer = cert['csv_scan']['cc_manufacturer'] - - if manufacturer != '': - if 'processed' not in cert: - cert['processed'] = {} - - # insert extracted manufacturers by full name - cert['processed']['cc_manufacturer_list'] = mapping_csvmanuf_separated[manufacturer] - - # 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)) - - cert['processed']['cc_manufacturer_simple_list'] = simple_manufacturers - cert['processed']['cc_manufacturer_simple'] = get_manufacturer_simple_name( - manufacturer, already_reduced) - - # extract certification lab - if is_in_dict(cert, ['frontpage_scan', 'cert_lab']): - lab = cert['frontpage_scan']['cert_lab'] - - if lab != '': - lab = lab.upper() - if 'processed' not in cert: - cert['processed'] = {} - - # insert extracted lab - only the first words, changed to uppercase, omitting the rest - pos1 = lab.find(' ') - if pos1 != -1: - cert['processed']['cert_lab'] = lab[:pos1] - else: - cert['processed']['cert_lab'] = lab - - # extract security level EAL - if is_in_dict(cert, ['csv_scan', 'cc_security_level']): - level = cert['csv_scan']['cc_security_level'] - level_split = level.split(",") - if level_split[0] == 'None': - if len(level_split[0]) > 1: - level_split[0] = 'EAL0+' - # REMOVE 20201016: security level from protection profiles done later - #if level.find(',') != -1: - # level = level[:level.find(',')] # trim list of augmented items - #level_out = level_split[0] - #if level == 'None': - # if cert['csv_scan']['cc_protection_profiles'] != '': - # level_out = 'Protection Profile' - - cert['processed']['cc_security_level'] = level_split[0] - cert['processed']['cc_security_level_augments'] = level_split[1:] - - # pair cert with its protection profile(s) - if is_in_dict(cert, ['csv_scan', 'cc_protection_profiles']): - pp_id_csv = cert['csv_scan']['cc_protection_profiles'] - if pp_id_csv != '': - # find corresponding protection profile - if pp_id_csv in pp_id_mapping.keys() and \ - pp_id_mapping[pp_id_csv] in all_pp_items.keys(): - pp = all_pp_items[pp_id_mapping[pp_id_csv]] - - # security level of certificate will be equal to level of protection profile - if is_in_dict(cert, ['processed', 'cc_security_level']): - if cert['processed']['cc_security_level'] == 'None': - cert['processed']['cc_security_level'] = pp['csv_scan']['cc_security_level'] - else: - cert['processed']['cc_security_level'] = pp['csv_scan']['cc_security_level'] - if cert['processed']['cc_security_level'] != pp['csv_scan']['cc_security_level']: - print('WARNING: {} cc_security_level level already set differently than inferred from PP: {} vs. {}'.format(file_name, cert['processed']['cc_security_level'], pp['csv_scan']['cc_security_level'])) - - # there might be more protection profiles in single file - search the right one - for sub_profile in pp['pp_analysis']['separate_profiles']: - if sub_profile['pp_id_csv'] == pp_id_csv: - cert['processed']['cc_pp_name'] = pp['csv_scan']['cc_pp_name'] - # set protection profile id as ID extracted from pp pdf (if not found csv is used) - cert['processed']['cc_pp_id'] = pp_id_csv - if sub_profile['pp_id_legacy'] != '': - cert['processed']['cc_pp_id'] = sub_profile['pp_id_legacy'] - # set path to protection profile file - cert['processed']['pp_filename'] = sub_profile['pp_filename'] - - return all_cert_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/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/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') - - -def generate_failed_download_script(base_dir): - # obtain list of all downloaded pdf files and their size - # check for pdf files with too small length - # generate download script again (single one) - - # visit all relevant subfolders - sub_folders = ['active/certs', 'active/targets', 'active_update/certs', 'active_update/targets', - 'archived/certs', 'archived/targets', 'archived_update/certs', 'archived_update/targets'] - - # the smallest correct certificate downloaded was 71kB, if server error occurred, it was only 1245 bytes - MIN_CORRECT_CERT_SIZE = 5000 - download_again = [] - for sub_folder in sub_folders: - 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: - # process only .pdf files - if not os.path.isfile(file_name): - continue - file_ext = file_name[file_name.rfind('.'):].upper() - if file_ext != '.PDF' and file_ext != '.DOC' and file_ext != '.DOCX': - continue - - # obtain size of file - 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(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))) diff --git a/src/fips_certificates.py b/src/fips_certificates.py deleted file mode 100644 index c7486906..00000000 --- a/src/fips_certificates.py +++ /dev/null @@ -1,444 +0,0 @@ -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 - - 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 - - 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 deleted file mode 100644 index 167f6998..00000000 --- a/src/process_certificates.py +++ /dev/null @@ -1,377 +0,0 @@ -import sys - -from extract_certificates import * -from analyze_certificates import * - -import os -import json - - -def do_all_analysis(all_cert_items, filter_label): - generate_dot_graphs(all_cert_items, filter_label) - analyze_cert_years_frequency(all_cert_items, filter_label) - analyze_references_graph(['rules_cert_id'], all_cert_items, filter_label) - analyze_eal_frequency(all_cert_items, filter_label) - analyze_security_assurance_component_frequency(all_cert_items, filter_label) - analyze_security_functional_component_frequency(all_cert_items, filter_label) - analyze_pdfmeta(all_cert_items, filter_label) - plot_certid_to_item_graph(['keywords_scan', 'rules_protection_profiles'], all_cert_items, filter_label, 'certid_pp_graph.dot', False) - - -def do_analysis_everything(all_cert_items, current_dir): - if not os.path.exists(current_dir): - os.makedirs(current_dir) - os.chdir(current_dir) - do_all_analysis(all_cert_items, '') - - -def do_analysis_09_01_2019_archival(all_cert_items, current_dir): - target_folder = current_dir + '\\results_archived01092019_only\\' - if not os.path.exists(target_folder): - os.makedirs(target_folder) - os.chdir(target_folder) - archived_date = '09/01/2019' - limited_cert_items = {x: all_cert_items[x] for x in all_cert_items if is_in_dict(all_cert_items[x], ['csv_scan', 'cc_archived_date']) and all_cert_items[x]['csv_scan']['cc_archived_date'] == archived_date} - do_all_analysis(limited_cert_items, 'cc_archived_date={}'.format(archived_date)) - - -def do_analysis_manufacturers(all_cert_items, current_dir): - # analyze only Infineon certificates - do_analysis_only_filtered(all_cert_items, current_dir, - ['processed', 'cc_manufacturer_simple'], 'Infineon Technologies AG') - # analyze only NXP certificates - do_analysis_only_filtered(all_cert_items, current_dir, - ['processed', 'cc_manufacturer_simple'], 'NXP Semiconductors') - # analyze only Red Hat certificates - do_analysis_only_filtered(all_cert_items, current_dir, - ['processed', 'cc_manufacturer_simple'], 'Red Hat, Inc') - # analyze only Suse certificates - do_analysis_only_filtered(all_cert_items, current_dir, - ['processed', 'cc_manufacturer_simple'], 'SUSE Linux Products Gmbh') - - -def do_analysis_only_filtered(all_cert_items, current_dir, filter_path, filter_value): - filter_string = '' - for item in filter_path: - if len(filter_string) > 0: - filter_string = filter_string + '__' - filter_string = filter_string + item - target_folder = current_dir + '\\{}={}\\'.format(filter_string, filter_value) - if not os.path.exists(target_folder): - os.makedirs(target_folder) - os.chdir(target_folder) - - cert_items = {} - for cert_item_key in all_cert_items.keys(): - item = get_item_from_dict(all_cert_items[cert_item_key], filter_path) - if item is not None: - if item == filter_value: - # Match found, include - cert_items[cert_item_key] = all_cert_items[cert_item_key] - - #cert_items = {x: all_cert_items[x] for x in all_cert_items if is_in_dict(all_cert_items[x], ['csv_scan', filter_key]) and all_cert_items[x]['csv_scan'][filter_key] == filter_value} - - print(len(cert_items)) - do_all_analysis(cert_items, '{}={}'.format(filter_string, filter_value)) - - -def do_analysis_only_category(all_cert_items, current_dir, category): - do_analysis_only_filtered(all_cert_items, current_dir, ['csv_scan', 'cc_category'], category) - - -def do_analysis_only_smartcards(all_cert_items, current_dir): - do_analysis_only_filtered(all_cert_items, current_dir, ['csv_scan', 'cc_category'], 'ICs, Smart Cards and Smart Card-Related Devices and Systems') - - -def do_analysis_only_operatingsystems(all_cert_items, current_dir): - do_analysis_only_category(all_cert_items, current_dir, ['csv_scan', 'cc_category'], 'Operating Systems') - - -def load_json_files(files_list): - loaded_jsons = [] - for file_name in files_list: - with open(file_name) as json_file: - loaded_items = json.load(json_file) - loaded_jsons.append(loaded_items) - print('{} loaded, total items = {}'.format(file_name, len(loaded_items))) - return tuple(loaded_jsons) - - -def sanitize_all_strings(data): - printable = set(string.printable) - - if isinstance(data, dict): - for k, v in data.items(): - if isinstance(v, dict) or isinstance(v, list) or isinstance(v, tuple): - sanitize_all_strings(v) - elif isinstance(v, str): - sanitized = ''.join(filter(lambda x: x in printable, v)) - data[k] = ''.join(filter(lambda x: x in printable, v)) - - if isinstance(data, list) or isinstance(data, tuple): - for v in data: - if isinstance(v, dict) or isinstance(v, list) or isinstance(v, tuple): - sanitize_all_strings(v) - elif isinstance(v, str): - sanitized = ''.join(filter(lambda x: x in printable, v)) - v = ''.join(filter(lambda x: x in printable, v)) - - -def main(): - # Paths for certificates downloaded on 20191208 - paths_20191208 = {} - paths_20191208['id'] = '20191208' - paths_20191208['cc_web_files_dir'] = 'c:\\Certs\\cc_certs_20191208\\web\\' - paths_20191208['walk_dir'] = 'c:\\Certs\\cc_certs_20191208\\cc_certs\\' - #paths_20191208['walk_dir'] = 'c:\\Certs\\cc_certs_20191208\\cc_certs_test1\\' - paths_20191208['pp_dir'] = 'c:\\Certs\\cc_certs_20191208\\cc_pp\\' - #paths_20191208['pp_dir'] = 'c:\\Certs\\cc_certs_20191208\\cc_pp_test1\\' - paths_20191208['fragments_dir'] = 'c:\\Certs\\cc_certs_20191208\\cc_certs_txt_fragments\\' - paths_20191208['pp_fragments_dir'] = 'c:\\Certs\\cc_certs_20191208\\cc_pp_txt_fragments\\' - - # Paths for certificates downloaded on 20200225 - paths_20200225 = {} - paths_20200225['id'] = '20200225' - paths_20200225['cc_web_files_dir'] = 'c:\\Certs\\cc_certs_20200225\\web\\' - paths_20200225['walk_dir'] = 'c:\\Certs\\cc_certs_20200225\\cc_certs\\' - paths_20200225['pp_dir'] = 'c:\\Certs\\cc_certs_20200225\\cc_pp\\' - paths_20200225['fragments_dir'] = 'c:\\Certs\\cc_certs_20200225\\cc_certs_txt_fragments\\' - paths_20200225['pp_fragments_dir'] = 'c:\\Certs\\cc_certs_20200225\\cc_pp_txt_fragments\\' - - # Paths for certificates downloaded on 20200904 - paths_20200904 = {} - paths_20200904['id'] = '20200904' - paths_20200904['cc_web_files_dir'] = 'c:\\Certs\\cc_certs_20200904\\web\\' - paths_20200904['cc_pp_web_files_dir'] = 'c:\\Certs\\certs_pp_20201008\\pp_web\\' - paths_20200904['walk_dir'] = 'c:\\Certs\\cc_certs_20200904\\cc_certs\\' - paths_20200904['pp_dir'] = 'c:\\Certs\\certs_pp_20201008\\cc_pp\\' - paths_20200904['fragments_dir'] = 'c:\\Certs\\cc_certs_20200904\\cc_certs_txt_fragments\\' - paths_20200904['pp_fragments_dir'] = 'c:\\Certs\\certs_pp_20201008\\cc_pp_txt_fragments\\' - - # initialize paths based on the profile used - #paths_used = paths_20191208 - #paths_used = paths_20200225 - paths_used = paths_20200904 - #paths_used['id'] = 'temp' # change id for temporary debugging - - cc_web_files_dir = paths_used['cc_web_files_dir'] - walk_dir = paths_used['walk_dir'] - fragments_dir = paths_used['fragments_dir'] - - # results folder includes unique identification of input dataset - results_folder = '{}\\..\\results_{}\\'.format(os.getcwd(), paths_used['id']) - # ensure existence of results folder - if not os.path.exists(results_folder): - os.makedirs(results_folder) - # change current directory to store results into results file - os.chdir(results_folder) - - # 1. generate_basic_download_script - # 2. run and download basic cc files from webpage (no certs yet) - - # - # Start processing - # - generate_basic_download_script() - generate_failed_download_script(walk_dir) - - do_complete_extraction = True - do_download_certs = True - do_extraction = True - do_pairing = True - do_processing = True - do_analysis = True - do_analysis_filtered = True - - if do_complete_extraction: - # analyze all files from scratch, set 'previous' state to empty dict - prev_csv = {} - prev_html = {} - prev_front = {} - prev_keywords = {} - prev_pdf_meta = {} - else: - # load previously analyzed results - prev_csv, prev_html, prev_front, prev_keywords, prev_pdf_meta = load_json_files( - ['certificate_data_csv_all.json', 'certificate_data_html_all.json', 'certificate_data_frontpage_all.json', - 'certificate_data_keywords_all.json', 'certificate_data_pdfmeta_all.json']) - - if do_download_certs: - # extract_certificates_html() will generate download scripts for cert documents - # NOTE: download scripts must be run manually now - current_html = extract_certificates_html(cc_web_files_dir, False) - - # NOTE: Code below is preparation for differetian download of only new certificates - # - unfinished now - # print('*** Items: {} vs. {}'.format(len(current_html.keys()), len(prev_html.keys()))) - # current_html_keys = sorted(current_html.keys()) - # prev_html_keys = sorted(prev_html.keys()) - # new_items = list(set(current_html.keys()) - set(prev_html.keys())) - # print('*** New items detected: {}'.format(len(new_items))) - # - # # find new items which are not yet processed based on the value of raw csv line - # new_items = [] - # for current_item_key in current_csv.keys(): - # current_item = current_csv[current_item_key] - # current_raw_csv = current_item['csv_scan']['raw_csv_line'] - # match_found = False - # for prev_item_key in prev_csv.keys(): - # prev_item = prev_csv[prev_item_key] - # prev_raw_csv = prev_item['csv_scan']['raw_csv_line'] - # if current_raw_csv == prev_raw_csv: - # match_found = True - # break - # if not match_found: - # # we found new item - # new_items.append(current_item_key) - # - # print('*** New items detected: {}'.format(len(new_items))) - - if do_extraction: - all_csv = extract_certificates_csv(cc_web_files_dir, False) - all_html = extract_certificates_html(cc_web_files_dir, False) - all_front = extract_certificates_frontpage(walk_dir, False) - all_keywords = extract_certificates_keywords(walk_dir, fragments_dir, 'certificate', False) - all_pdf_meta = extract_certificates_pdfmeta(walk_dir, 'certificate', False) - - # save joined results - with open("certificate_data_csv_all.json", "w") as write_file: - write_file.write(json.dumps(all_csv, indent=4, sort_keys=True)) - with open("certificate_data_html_all.json", "w") as write_file: - write_file.write(json.dumps(all_html, indent=4, sort_keys=True)) - with open("certificate_data_frontpage_all.json", "w") as write_file: - write_file.write(json.dumps(all_front, indent=4, sort_keys=True)) - with open("certificate_data_keywords_all.json", "w") as write_file: - write_file.write(json.dumps(all_keywords, indent=4, sort_keys=True)) - with open("certificate_data_pdfmeta_all.json", "w") as write_file: - write_file.write(json.dumps(all_pdf_meta, indent=4, sort_keys=True)) - - # if do_extraction_pp: - # all_pp_csv = extract_protectionprofiles_csv(cc_pp_web_files_dir) - # all_pp_front = extract_protectionprofiles_frontpage(pp_dir) - # all_pp_keywords = extract_certificates_keywords(pp_dir, pp_fragments_dir, 'pp') - # all_pp_pdf_meta = extract_certificates_pdfmeta(pp_dir, 'pp') - # - # # save joined results - # with open("pp_data_csv_all.json", "w") as write_file: - # write_file.write(json.dumps(all_pp_csv, indent=4, sort_keys=True)) - # with open("pp_data_frontpage_all.json", "w") as write_file: - # write_file.write(json.dumps(all_pp_front, indent=4, sort_keys=True)) - # with open("pp_data_keywords_all.json", "w") as write_file: - # write_file.write(json.dumps(all_pp_keywords, indent=4, sort_keys=True)) - # with open("pp_data_pdfmeta_all.json", "w") as write_file: - # write_file.write(json.dumps(all_pp_pdf_meta, indent=4, sort_keys=True)) - - if do_pairing: - # # PROTECTION PROFILES - # # load results from previous step - # all_pp_csv, all_pp_front, all_pp_keywords, all_pp_pdf_meta = load_json_files( - # ['pp_data_csv_all.json', 'pp_data_frontpage_all.json', - # 'pp_data_keywords_all.json', 'pp_data_pdfmeta_all.json']) - # # check for unexpected results - # check_expected_pp_results({}, all_pp_csv, {}, all_pp_keywords) - # # collate all results into single file - # all_pp_items = collate_certificates_data({}, all_pp_csv, all_pp_front, all_pp_keywords, all_pp_pdf_meta, 'link_pp_document') - # # write collated result - # with open("pp_data_complete.json", "w") as write_file: - # write_file.write(json.dumps(all_pp_items, indent=4, sort_keys=True)) - - # CERTIFICATES - # load results from previous step - all_csv, all_html, all_front, all_keywords, all_pdf_meta = load_json_files( - ['certificate_data_csv_all.json', 'certificate_data_html_all.json', 'certificate_data_frontpage_all.json', - 'certificate_data_keywords_all.json', 'certificate_data_pdfmeta_all.json']) - # check for unexpected results - check_expected_cert_results(all_html, all_csv, all_front, all_keywords, all_pdf_meta) - # collate all results into single file - all_cert_items = collate_certificates_data(all_html, all_csv, all_front, all_keywords, all_pdf_meta, 'link_security_target') - - # write collated result - with open("certificate_data_complete.json", "w") as write_file: - write_file.write(json.dumps(all_cert_items, indent=4, sort_keys=True)) - - if do_processing: - # load information about protection profiles as extracted by sec-certs-pp tool - with open('pp_data_complete_processed.json') as json_file: - all_pp_items = json.load(json_file) - - with open('certificate_data_complete.json') as json_file: - all_cert_items = json.load(json_file) - - all_cert_items = process_certificates_data(all_cert_items, all_pp_items) - - with open("certificate_data_complete_processed.json", "w") as write_file: - write_file.write(json.dumps(all_cert_items, indent=4, sort_keys=True)) - - if do_analysis: - with open('certificate_data_complete_processed.json') as json_file: - all_cert_items = json.load(json_file) - - if do_analysis_filtered: - # analyze only smartcards - do_analysis_only_filtered(all_cert_items, results_folder, - ['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 all certificates together - do_analysis_everything(all_cert_items, results_folder) - - with open("certificate_data_complete_processed_analyzed.json", "w") as write_file: - write_file.write(json.dumps(all_cert_items, indent=4, sort_keys=True)) - - -if __name__ == "__main__": - main() - - - # TODO - # add saving of logs into file - # include parsing from protection profiles repo - # add differential partial download of new files only + processing + combine - # generate download script only for new files (need to have previous version of files stored) - # option for extraction of info just for single file? - # allow for late extraction of keywords (only newly added regexes) - # extraction of keywords done with the provided cert_rules_dict => cert_rules.py and cert_rules_new.py - # detect archival of certificates - # add tests - few selected files - # add detection of overly long regex matches - # add analysis of target CC version - # extract even more pdf file metadata https://github.com/pdfminer/pdfminer.six - # protection profiles dependency graph similarly as certid dependency graph is done - # If None == protection profile => Match PP with its assurance level and recompute - # extract info about protection profiles, download and parse pdf, map to referencing files - # analysis of PP only: which PP is the most popular?, what schemes/countries are doing most... - # analysis of certificates in time (per year) (different schemes) - # how many certificates are extended? How many times - # analysis of use of protection profiles - # analysis of security targets documents - # analysis of big cert clusters - # improve logging (info, warnings, errors, final summary) - # save as json, named segments (start_segment('name'), end_segment('name'), print('log_line', level) - # other schemes: FIPS140-2 certs, EMVCo, Visa Certification, American Express Certification, MasterCard Certification - # download and analyse CC documentation - # solve treatment of unicode characters - # analyze bibliography - # Statistics about number of characters (length), words, pages - histogram of pdf length & extracted text length - # add keywords extraction for trademarks (e.g, from 0963V2b_pdf.pdf) - # FRONTPAGE - # extract frontpage also from other than anssi and bsi certificates (US, BE...) - # add extraction of frontpage for protection profiles - # PORTABILITY - # check functionality on Linux (script %%20 expansions..., \\ vs. /) - # Add processing of docx files - # search for ATR, Response APDU, and custom commands specifying the IC type (CPLC + others) - # e.g., KECS-CR-15-105 XSmart e-Passport V1.4 EAC with SAC on M7892(eng).txt - # use pdf2text -raw switch to preserve better tables (needs to be checked wrt existing regexes) - # add tool for language detection, and if required, use automatic translation into english (https://pypi.org/project/googletrans/) - # analyze technical decisions: https://www.niap-ccevs.org/Documents_and_Guidance/view_tds.cfm - # extract names of IC (or other devices) from certificates => results in the list of certified chips in smartcards etc. - # analyze SARs and SFRs in correlation with specific company (what level of SAR/SFR can company achieve?) diff --git a/src/sanity.py b/src/sanity.py deleted file mode 100644 index a3298d48..00000000 --- a/src/sanity.py +++ /dev/null @@ -1,53 +0,0 @@ -STOP_ON_UNEXPECTED_NUMS = False -EXPECTED_CERTS_REFERENCED_ONCE = 942 -EXPECTED_CERTS_MIN_ITEMS_FOUND_CSV = 4415 -EXPECTED_CERTS_MIN_ITEMS_FOUND_HTML = 4413 -EXPECTED_CERTS_MIN_ITEMS_FOUND_KEYWORDS = 494655 -EXPECTED_CERTS_MIN_ITEMS_FOUND_FRONTPAGE = 1438 -EXPECTED_PP_MIN_ITEMS_FOUND_CSV = 1000 -EXPECTED_PP_MIN_ITEMS_FOUND_HTML = 1000 -EXPECTED_PP_MIN_ITEMS_FOUND_KEYWORDS = 1000 -EXPECTED_PP_MIN_ITEMS_FOUND_FRONTPAGE = 1000 - - -def check_and_handle(expected, obtained, message): - if expected != obtained: - print('SANITY: {}: {} vs. {} expected'.format(message, expected, obtained)) - if STOP_ON_UNEXPECTED_NUMS: - raise ValueError('ERROR: Stopping on unexpected intermediate numbers') - - -def check_certs_referenced_once(num_items): - check_and_handle(EXPECTED_CERTS_REFERENCED_ONCE, num_items, 'Different than expected num certificates referenced at least once') - - -def check_certs_min_items_found_csv(num_items): - check_and_handle(EXPECTED_CERTS_MIN_ITEMS_FOUND_CSV, num_items, 'Different than expected number of CSV records found!') - - -def check_certs_min_items_found_html(num_items): - check_and_handle(EXPECTED_CERTS_MIN_ITEMS_FOUND_HTML, num_items, 'Different than expected number of HTML records found!') - - -def check_certs_min_items_found_frontpage(num_items): - check_and_handle(EXPECTED_CERTS_MIN_ITEMS_FOUND_FRONTPAGE, num_items, 'Different than expected number of frontpage records found!') - - -def check_certs_min_items_found_keywords(num_items): - check_and_handle(EXPECTED_CERTS_MIN_ITEMS_FOUND_KEYWORDS, num_items, 'Different than expected number of keywords found!') - - -def check_pp_min_items_found_csv(num_items): - check_and_handle(EXPECTED_PP_MIN_ITEMS_FOUND_CSV, num_items, 'Different than expected number of CSV records found!') - - -def check_pp_min_items_found_html(num_items): - check_and_handle(EXPECTED_PP_MIN_ITEMS_FOUND_HTML, num_items, 'Different than expected number of HTML records found!') - - -def check_pp_min_items_found_frontpage(num_items): - check_and_handle(EXPECTED_PP_MIN_ITEMS_FOUND_FRONTPAGE, num_items, 'Different than expected number of frontpage records found!') - - -def check_pp_min_items_found_keywords(num_items): - check_and_handle(EXPECTED_PP_MIN_ITEMS_FOUND_KEYWORDS, num_items, 'Different than expected number of keywords found!')
\ No newline at end of file diff --git a/src/tags_constants.py b/src/tags_constants.py deleted file mode 100644 index 01625265..00000000 --- a/src/tags_constants.py +++ /dev/null @@ -1,27 +0,0 @@ -TAG_MATCH_COUNTER = 'count' -TAG_MATCH_MATCHES = 'matches' - -TAG_CERT_HEADER_PROCESSED = 'cert_header_processed' - -TAG_CERT_ID = 'cert_id' -TAG_CC_SECURITY_LEVEL = 'cc_security_level' -TAG_CC_VERSION = 'cc_version' -TAG_CERT_LAB = 'cert_lab' -TAG_CERT_ITEM = 'cert_item' -TAG_CERT_ITEM_VERSION = 'cert_item_version' -TAG_DEVELOPER = 'developer' -TAG_REFERENCED_PROTECTION_PROFILES = 'ref_protection_profiles' -TAG_HEADER_MATCH_RULES = 'match_rules' -TAG_PP_TITLE = 'pp_title' -TAG_PP_GENERAL_STATUS = 'pp_general_status' -TAG_PP_VERSION_NUMBER = 'pp_version_number' -TAG_PP_ID = 'pp_id' -TAG_PP_ID_REGISTRATOR = 'pp_id_registrator' -TAG_PP_DATE = 'pp_date' -TAG_PP_AUTHORS = 'pp_authors' -TAG_PP_REGISTRATOR = 'pp_registrator' -TAG_PP_REGISTRATOR_SIMPLIFIED = 'pp_registrator_simplified' -TAG_PP_SPONSOR = 'pp_sponsor' -TAG_PP_EDITOR = 'pp_editor' -TAG_PP_REVIEWER = 'pp_reviewer' -TAG_KEYWORDS = 'keywords'
\ No newline at end of file |
