aboutsummaryrefslogtreecommitdiffhomepage
path: root/src
diff options
context:
space:
mode:
authorJ08nY2020-10-15 21:56:24 +0200
committerJ08nY2020-10-18 11:43:10 +0200
commit64bf9c880828fa36f8625fdabad468668df62575 (patch)
tree68902488bcce91b5cdf8c39a809ca90cf465fe50 /src
parent7ed51240bd5e5e56a5847df28c8a1a9708307315 (diff)
downloadsec-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.py828
-rw-r--r--src/cert_rules.py357
-rw-r--r--src/download_files.py36
-rw-r--r--src/extract_certificates.py2435
-rw-r--r--src/fips_certificates.py444
-rw-r--r--src/process_certificates.py377
-rw-r--r--src/sanity.py53
-rw-r--r--src/tags_constants.py27
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('&#x3a;', ' ') # ':' is not allowed in dot
- sanitized_name = sanitized_name.replace('&#x2f;', ' ') # '/' 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('&nbsp;', ' ')
- whole_text = whole_text.replace('&amp;', '&')
-
- # 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