Merge pull request #31 from petrs/petrs/analyze_affecting

Petrs/analyze affecting

4 files changed, 122 insertions, 38 deletions

diff --git a/process_certificates.py b/process_certificates.py
index f5bbcf39..5272fc67 100755
--- a/process_certificates.py
+++ b/process_certificates.py
@@ -21,15 +21,16 @@ from sec_certs.cert_rules import rules as cc_search_rules
 @click.option("--do-analysis", "do_analysis", is_flag=True, help="Whether to analyse certificates.")
 @click.option("--do-analysis-fips", "do_analysis_fips", is_flag=True, help="Whether to analyse fips certificates.")
 @click.option("--do-find-affected", "do_find_affected", help="Find affected certs.", multiple=True, type=str, metavar="certificate id")
+@click.option("--do-find-affecting", "do_find_affecting", help="Find certificates affecting the provided one", multiple=True, type=str, metavar="certificate id")
 @click.option("--do-find-affected-keyword", "do_find_affected_keywords", help="Find certs referencing all certs with specific keyword.", multiple=True, type=str, metavar="keyword")
 @click.option("--analysis-label", "analysis_label", help="Optional custom label for analysis results", multiple=False, type=str, metavar="cutsom label")
 @click.option("-t", "--threads", "threads", type=int, default=4, help="Amount of threads to use.")
 def main(directory, do_complete_extraction: bool, do_download_meta: bool, do_extraction_meta: bool,
          do_download_certs: bool, do_pdftotext: bool, do_extraction_certs: bool,
          do_pairing: bool, do_processing: bool, do_analysis: bool, do_analysis_fips: bool, do_find_affected: list,
-         do_find_affected_keywords: list, analysis_label: str, threads: int):
-    directory = Path(directory)
+         do_find_affected_keywords: list, do_find_affecting: list, analysis_label: str, threads: int):
 
+    directory = Path(directory)
     web_dir = directory / "web"
     walk_dir = directory / "certs"
     certs_dir = walk_dir / "certs"
@@ -249,7 +250,7 @@ def main(directory, do_complete_extraction: bool, do_download_meta: bool, do_ext
 
         # set output folder according to analysis label
         out_folder = analysis_label + '_' + name_results
-        results_out_dir = results_out_dir / out_folder
+        results_out_dir = results_dir / out_folder
 
         do_analysis_affected(all_cert_items, results_out_dir, list(do_find_affected), analysis_label)
 
@@ -264,6 +265,16 @@ def main(directory, do_complete_extraction: bool, do_download_meta: bool, do_ext
         results_out_dir = results_dir / analysis_label
         do_analysis_affected(all_cert_items, results_out_dir, list(do_find_affected), analysis_label)
 
+    # find all certificates which are potentially affecting security of the provided one ()
+    # example: --do-find-affecting ANSSI-CC-2013/55     # Estonia estID
+    # example: --do-find-affecting ANSSI-CC-2020/44     # eTravel v2.2 EAC/BAC on MultiApp v4.0.1 platform with Filter Set 1.0 version 1.0
+    if len(do_find_affecting) > 0:
+        with open(results_dir / 'certificate_data_complete_processed_analyzed.json') as json_file:
+            all_cert_items = json.load(json_file)
+        # set output folder according to analysis label
+        results_out_dir = results_dir / analysis_label
+        do_analysis_affecting(all_cert_items, results_out_dir, list(do_find_affecting), analysis_label)
+
     # analysis of fips extracted data
     if do_analysis_fips:
         with open(results_dir / 'fips_full_dataset.json') as json_file:
diff --git a/sec_certs/analyze_certificates.py b/sec_certs/analyze_certificates.py
index 42c5854f..f20a0e2e 100644
--- a/sec_certs/analyze_certificates.py
+++ b/sec_certs/analyze_certificates.py
@@ -22,6 +22,8 @@ plt.rcdefaults()
 
 STOP_ON_UNEXPECTED_NUMS = False
 
+GRAPHS_COLOR_PALETTE = plt.cm.get_cmap('tab20').colors
+
 printable = set(string.printable)
 
 
@@ -70,6 +72,31 @@ def plot_bar_graph(data, x_data_labels, y_label, title, file_name):
     plt.close()
 
 
+def plot_bar_multi_graph(data, x_data_labels, y_label, title, file_name):
+    max_len = 0
+    for item in data:
+        if max_len < len(item):
+            max_len = len(item)
+
+    fig_width = round(max_len / 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))
+    BAR_WIDTH = 0.4
+    for i in range(0, len(data)):
+        plt.bar(y_pos + i * BAR_WIDTH, data[i], alpha=0.5, width=BAR_WIDTH)
+    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:
@@ -439,6 +466,7 @@ def plot_schemes_multi_line_graph(x_ticks, data, prominent_data, x_label, y_labe
     line_types = ['-', ':', '-.', '--']
     num_lines_plotted = 0
     data_sorted = sorted(data.keys())
+    color_index = 0
     for group in data_sorted:
         items_in_year = []
         for item in sorted(data[group]):
@@ -446,13 +474,16 @@ def plot_schemes_multi_line_graph(x_ticks, data, prominent_data, x_label, y_labe
             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)
+            plt.plot(x_ticks, items_in_year, line_types[num_lines_plotted % len(line_types)],
+                     label=group, linewidth=3, color=GRAPHS_COLOR_PALETTE[color_index])
         else:
             # plot non-prominent data as dashed
-            plt.plot(x_ticks, items_in_year, line_types[num_lines_plotted % len(line_types)], label=group, linewidth=2)
+            plt.plot(x_ticks, items_in_year, line_types[num_lines_plotted % len(line_types)],
+                     label=group, linewidth=2, color=GRAPHS_COLOR_PALETTE[color_index])
 
         # change line type to prevent color repetitions
         num_lines_plotted += 1
+        color_index += 1
 
     plt.rcParams.update({'font.size': 16})
     plt.legend(loc=2)
@@ -488,7 +519,7 @@ def plot_schemes_stacked_graph(x_ticks, data, prominent_data, x_label, y_label,
     if len(data_stacked) == 0:
         data_stacked = [0] * len(x_ticks)
 
-    plt.stackplot(x_ticks, data_stacked, colors=plt.cm.get_cmap('tab20').colors, labels=data_sorted)
+    plt.stackplot(x_ticks, data_stacked, colors=GRAPHS_COLOR_PALETTE, labels=data_sorted)
 
     plt.rcParams.update({'font.size': 16})
     plt.legend(loc=2)
@@ -652,7 +683,6 @@ def analyze_cert_years_frequency(all_cert_items, filter_label, force_plot_end_ye
                         # 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')
@@ -697,7 +727,7 @@ def analyze_cert_years_frequency(all_cert_items, filter_label, force_plot_end_ye
     plot_schemes_multi_graph(years, plot_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_graph(years, plot_level_date, ['EAL4+', 'EAL5+','EAL2+', 'Protection Profile'], 'Year of issuance', 'Number of certificates issued', fig_label('Certificates issuance frequency per level and year', filter_label), 'num_certs_level_in_years')
     plot_schemes_multi_graph(years, plot_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_graph(years, plot_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_graph(years, plot_pp_date, [], 'Year of issuance', 'Number of certificates issued', fig_label('Certificates conforming to Protection Profile(s)', filter_label), 'num_certs_pp_in_years')
     plot_schemes_multi_graph(years, plot_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_graph(years, plot_top_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')
 
@@ -738,8 +768,8 @@ def analyze_eal_frequency(all_cert_items, filter_label):
             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']
+    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:
@@ -840,7 +870,6 @@ def analyze_sc_frequency(all_cert_items, filter_label, sec_component_label):
                         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 = []
@@ -917,7 +946,6 @@ def generate_dot_graphs(all_items_found, filter_label, highlight_certs_id=None):
 
 
 def do_all_analysis(all_cert_items, filter_label, highlight_certs_id = None):
-    generate_dot_graphs(all_cert_items, filter_label, highlight_certs_id)
     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)
@@ -925,6 +953,7 @@ def do_all_analysis(all_cert_items, filter_label, highlight_certs_id = None):
     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)
+    generate_dot_graphs(all_cert_items, filter_label, highlight_certs_id)
 
 
 def do_analysis_everything(all_cert_items, current_dir: Path):
@@ -969,32 +998,67 @@ def do_analysis_affected(all_cert_items: dict, results_dir: Path, target_certs_i
     filter_direct.update(target_certs_id)
     filter_indirect.update(target_certs_id)
 
+    label = get_label_from_certids(target_certs_id)
+    filter_and_do_all_analysis_filtered(all_cert_items, target_certs_id, target_certs_id, results_dir,
+                                        '{}_target_id_{}'.format(set_name_label, label))
+    filter_and_do_all_analysis_filtered(all_cert_items, filter_direct, target_certs_id, results_dir,
+                                        '{}_directly_referring_to_{}'.format(set_name_label, label))
+    filter_and_do_all_analysis_filtered(all_cert_items, filter_indirect, target_certs_id, results_dir,
+                                        '{}_indirectly_referring_to_{}'.format(set_name_label, label))
+
+
+def get_label_from_certids(target_certs_id: list):
     # concatenate all target ids into single string label
     label = ''
     for target_id in target_certs_id:
         target_id_sanitized = target_id.replace('\\', '')
-        target_id_sanitized = target_id.replace('/', '')
+        target_id_sanitized = target_id_sanitized.replace('/', '')
         label = label + target_id_sanitized + '__'
     MAX_LABEL_LENGTH = 100
     if len(label) > MAX_LABEL_LENGTH:
         label = label[:MAX_LABEL_LENGTH - 6] + '__more'
 
+    return label
+
+
+def filter_and_do_all_analysis_filtered(all_cert_items: dict, target_certs_id: list, highlight_certs_id: list,
+                                        results_dir: Path, label: str):
     target_cert_items = {x: all_cert_items[x] for x in all_cert_items if
                                     is_in_dict(all_cert_items[x], ['processed', 'cert_id']) and
                                     all_cert_items[x]['processed']['cert_id'] in target_certs_id}
-    do_all_analysis_filtered(target_cert_items, results_dir,
-                             '{}_target_id_{}'.format(set_name_label, label), target_certs_id)
-    # find only direct and indirect references of target_certs_id, filter original dataset, run atop it
-    direct_referenced_cert_items = {x: all_cert_items[x] for x in all_cert_items if
-                                    is_in_dict(all_cert_items[x], ['processed', 'cert_id']) and
-                                    all_cert_items[x]['processed']['cert_id'] in filter_direct}
-    do_all_analysis_filtered(direct_referenced_cert_items, results_dir,
-                             '{}_directly_referring_to_{}'.format(set_name_label, label), target_certs_id)
-    direct_referenced_cert_items = {x: all_cert_items[x] for x in all_cert_items if
-                                    is_in_dict(all_cert_items[x], ['processed', 'cert_id']) and
-                                    all_cert_items[x]['processed']['cert_id'] in filter_indirect}
-    do_all_analysis_filtered(direct_referenced_cert_items, results_dir,
-                             '{}_indirectly_referring_to_{}'.format(set_name_label, label), target_certs_id)
+    do_all_analysis_filtered(target_cert_items, results_dir, label, highlight_certs_id)
+
+
+def do_analysis_affecting(all_cert_items: dict, results_dir: Path, target_certs_id: list, set_name_label: str):
+    # build references graph
+    referenced_by_direct, referenced_by_indirect = build_cert_references(['rules_cert_id'], all_cert_items)
+
+    # extract only such references which are relevant to target_certs_id
+    # idea - loop over all cert ids, add to set when target id(s) is referenced
+    filter_direct = set()
+    filter_indirect = set()
+
+    for cert_id in referenced_by_direct.keys():
+        for target_cert_id in target_certs_id:
+            if target_cert_id in referenced_by_direct[cert_id]:
+                filter_direct.update([cert_id])
+
+    for cert_id in referenced_by_indirect.keys():
+        for target_cert_id in target_certs_id:
+            if target_cert_id in referenced_by_indirect[cert_id]:
+                filter_indirect.update([cert_id])
+
+    # add also target_certs_id into list of certificates to analyze
+    filter_direct.update(target_certs_id)
+    filter_indirect.update(target_certs_id)
+
+    label = get_label_from_certids(target_certs_id)
+    filter_and_do_all_analysis_filtered(all_cert_items, target_certs_id, target_certs_id, results_dir,
+                                        '{}_target_id_{}'.format(set_name_label, label))
+    filter_and_do_all_analysis_filtered(all_cert_items, filter_direct, target_certs_id, results_dir,
+                                        '{}_directly_referenced_by{}'.format(set_name_label, label))
+    filter_and_do_all_analysis_filtered(all_cert_items, filter_indirect, target_certs_id, results_dir,
+                                        '{}_indirectly_referenced_by_{}'.format(set_name_label, label))
 
 
 def do_all_analysis_filtered(limited_cert_items, current_dir: Path, filter_label: str, highlight_certs_id: list):
@@ -1033,13 +1097,7 @@ def do_analysis_only_filtered(all_cert_items, current_dir: Path, filter_path, fi
         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 = filter_items(all_cert_items, filter_path, filter_value)
 
     print(len(cert_items))
     do_all_analysis(cert_items, '{}={}'.format(filter_string, filter_value))
@@ -1156,3 +1214,15 @@ def process_matched_keywords(all_cert_items: dict, all_keywords: dict, do_find_a
 
 
     return certs_with_keywords
+
+
+def filter_items(items: dict, filter_path: list, filter_value: str):
+    filtered_items = {}
+    for item_key in items.keys():
+        item = get_item_from_dict(items[item_key], filter_path)
+        if item is not None:
+            if item == filter_value:
+                # Match found, include
+                filtered_items[item_key] = items[item_key]
+
+    return filtered_items
diff --git a/sec_certs/cert_rules.py b/sec_certs/cert_rules.py
index 034dd0b9..3be285c5 100644
--- a/sec_certs/cert_rules.py
+++ b/sec_certs/cert_rules.py
@@ -7,7 +7,6 @@ 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
@@ -58,8 +57,7 @@ rules_protection_profiles = [
     'PP-SSCD.+?',
     'PP_DBMS_.+?'
     #    'Protection Profile',
-    'CCMB-20.+?',
-    'CCMB-20[0-9]+?-[0-9]+?-[0-9]+?',
+    #'CCMB-20.+?',
     'BSI-CCPP-.+?',
     'ANSSI-CC-PP.+?',
     'WBIS_V[0-9]\\.[0-9]',
@@ -68,6 +66,7 @@ rules_protection_profiles = [
 
 rules_technical_reports = [
     'BSI[ ]*TR-[0-9]+?(?:-[0-9]+?|)',
+    'BSI [0-9]+?',  # German BSI document containing list of issued certificates in some period
 ]
 
 rules_device_id = [
@@ -97,6 +96,8 @@ rules_standard_id = [
     '[Xx]\\.509',
     'RFC [0-9]+',
     '(?:SCP|scp)[ \']*[0-9][0-9]',
+    'CC[I]*MB-20[0-9]+?-[0-9]+?-[0-9]+?',   # Common Criteria methodology
+    'CCIMB-9[0-9]-[0-9]+?'                  # Common Criteria methodology old
 ]
 
 rules_security_level = [
@@ -507,7 +508,8 @@ fips_rules['rules_to_remove'] = rules_fips_to_remove
 fips_rules['rules_security_level'] = rules_fips_security_level
 fips_rules['rules_cert_id'] = rules_fips_cert
 fips_common_rules = copy.deepcopy(common_rules)  # make separate copy not to process cc rules by fips's re.compile
+#fips_rules.update(fips_common_rules)
 
 for rule in fips_rules:
     for current_rule in range(len(fips_rules[rule])):
-        fips_rules[rule][current_rule] = re.compile(fips_rules[rule][current_rule])
+        fips_rules[rule][current_rule] = re.compile(fips_rules[rule][current_rule])
\ No newline at end of file
diff --git a/sec_certs/constants.py b/sec_certs/constants.py
index adf30869..41a4f3b7 100644
--- a/sec_certs/constants.py
+++ b/sec_certs/constants.py
@@ -43,4 +43,5 @@ TAG_PP_EDITOR = 'pp_editor'
 TAG_PP_REVIEWER = 'pp_reviewer'
 TAG_KEYWORDS = 'keywords'
 FIPS_NOT_AVAILABLE_CERT_SIZE = 10000
-FIPS_ALG_URL = 'https://csrc.nist.gov/projects/cryptographic-algorithm-validation-program/validation-search?searchMode=validation&page='
\ No newline at end of file
+FIPS_ALG_URL = 'https://csrc.nist.gov/projects/cryptographic-algorithm-validation-program/validation-search?searchMode=validation&page='
+