From b7082ffb5e0bf14b44ad53505d7381e5a213d35a Mon Sep 17 00:00:00 2001 From: = <=> Date: Tue, 26 Aug 2025 17:12:06 -0400 Subject: [PATCH] Added Preflights --- AirlockTools.py | 36 +++++++++--- utils/allowfunctions.py | 16 ------ utils/hashfunctions.py | 122 ++++++++++++++++++++++++++++++++++++---- utils/pathfunctions.py | 17 +++++- 4 files changed, 155 insertions(+), 36 deletions(-) delete mode 100644 utils/allowfunctions.py diff --git a/AirlockTools.py b/AirlockTools.py index e0a512f..b5c8ee5 100644 --- a/AirlockTools.py +++ b/AirlockTools.py @@ -123,6 +123,7 @@ def menu_prepare_to_enforce(): if not os.path.exists("dataframe_html"): os.makedirs("dataframe_html") if not os.path.exists("dataframe_csv"): os.makedirs("dataframe_csv") if not os.path.exists("manuallyapproved"): os.makedirs("manuallyapproved") + if not os.path.exists("preflight"): os.makedirs("preflight") df_aggregated_combo = pd.DataFrame() while True: @@ -184,9 +185,9 @@ def menu_prepare_to_enforce(): print(ct.colorText(f"7. Manually review the file df_paths_needing_review_{first_policy}_{second_policy}.csv", "cyan")) print(ct.colorText(" Remove the rows containing path exclusions you do not approve of" , "cyan")) - print(ct.colorText(" When complete, save the csv file to the directory 'manuallyapproved' and choose this option to generate the proposed list of changes", "cyan")) + print(ct.colorText(" When complete, save the csv file to the directory 'manuallyapproved' and choose this option to generate the preflight lists", "cyan")) - if os.path.isfile(f"manuallyapproved\\df_paths_needing_review_{first_policy}_{second_policy}.csv") and os.path.isfile(f"dataframe_csv\\df_hashdestination_{first_policy}_{second_policy}.csv"): + if os.path.isfile(f"manuallyapproved\\df_paths_needing_review_{first_policy}_{second_policy}.csv") and os.path.isfile("dataframe_csv\\df_hashdestination_{first_policy}_{second_policy}.csv") and os.path.isfile(f"dataframe_csv\\df_addtochildpolicy_{first_policy}_{second_policy}.csv") and os.path.isfile(f"dataframe_csv\\f_addtobaseline_{first_policy}_{second_policy}.csv"): print(ct.colorText(" [✓] This step has been completed","green")) else: print(ct.colorText(" [✗] This step has not been completed","red")) @@ -298,13 +299,30 @@ def menu_prepare_to_enforce(): elif choice == "7": if os.path.exists(f"manuallyapproved\\df_paths_needing_review_{first_policy}_{second_policy}.csv"): - df1 = tryToReadCSV(f"manuallyapproved\\df_paths_needing_review_{first_policy}_{second_policy}.csv") - df2 = tryToReadCSV(f"dataframe_csv\\df_path_ineligible_{first_policy}_{second_policy}.csv") - df3 = tryToReadCSV(f"manuallyapproved\\df_automatically_approved_hashes_{first_policy}_{second_policy}.csv") - df_hashdestination = utils.hashfunctions.destinationbuilder(df2,df3) - df_hashdestination.to_csv("dataframe_csv\\df_hashdestination_{first_policy}_{second_policy}.csv") - ct.style_dataframe_dark(df_hashdestination,f"dataframe_html\\df_hashdestination_{first_policy}_{second_policy}.html") - + df_approved = tryToReadCSV(f"manuallyapproved\\df_paths_needing_review_{first_policy}_{second_policy}.csv") + df_eligible = tryToReadCSV(f"dataframe_csv\\df_paths_needing_review_{first_policy}_{second_policy}.csv") + df_ineligible = tryToReadCSV(f"dataframe_csv\\df_path_ineligible_{first_policy}_{second_policy}.csv") + + approved_set = set([tuple(map(tuple, row)) for row in df_approved.values]) + + # Identify rows in eligible that are not in approved + not_approved_rows = df_eligible[~df_eligible.apply(lambda row: tuple(map(tuple, row)) in approved_set, axis=1)] + + # Append these rows to ineligible + df_ineligible= pd.concat([df_ineligible, not_approved_rows], ignore_index=True) + + df_approved.to_csv(f"preflight\\Approved_Path_Exclusions_{first_policy}_{second_policy}.csv") + ct.style_dataframe_dark(df_approved, f"preflight\\Approved_Path_Exclusions_{first_policy}_{second_policy}.html") + + #Seperate out what we arent excluding by path into those that will go into the baseline, and those that will b added to the child. + df_addtobaseline = df_ineligible[df_ineligible['reputation status'] == 'KNOWN'] + df_addtobaseline.to_csv(f"preflight\\Add_to_Baseline_{first_policy}_{second_policy}.csv", index=False) + ct.style_dataframe_dark(df_addtobaseline, f"preflight\\Add_to_Baseline_{first_policy}_{second_policy}.html") + + df_addtochildpolicy = df_ineligible[df_ineligible['reputation status'] == 'UNKNOWN'] + df_addtochildpolicy.to_csv(f"preflight\\Add_to_Child_Policy_{first_policy}_{second_policy}.csv", index=False) + ct.style_dataframe_dark(df_addtochildpolicy, f"preflight\\Add_to_Child_Policy_{first_policy}_{second_policy}.html") + else: print(ct.colorText(f"Please manually approve suggested paths prior to this step","red")) diff --git a/utils/allowfunctions.py b/utils/allowfunctions.py deleted file mode 100644 index 3ec4959..0000000 --- a/utils/allowfunctions.py +++ /dev/null @@ -1,16 +0,0 @@ -import pandas as pd - -def filter_and_drop(approved, eligiblepaths, min_hashes): - """ - Filters eligiblepaths to rows where all hashes are in approved, - then drops rows with fewer than min_hashes hashes. - """ - approved_hashes = set(approved['sha256']) - - def all_hashes_approved(row): - return all(h in approved_hashes for h in row['sha256']) - - filtered = eligiblepaths[eligiblepaths.apply(all_hashes_approved, axis=1)] - filtered = filtered[filtered['sha256'].apply(len) >= min_hashes] - - return filtered diff --git a/utils/hashfunctions.py b/utils/hashfunctions.py index 771cbe3..8cec0f9 100644 --- a/utils/hashfunctions.py +++ b/utils/hashfunctions.py @@ -16,6 +16,7 @@ import pandas as pd import requests import os import json +import utils.pretty as ct @@ -80,7 +81,7 @@ def augmentAggregatedHashes(url, agg_df: pd.DataFrame) -> pd.DataFrame: df_api = pd.DataFrame(rows) df = agg_df.merge(df_api, on="sha256", how="left") - aug_df = df[['sha256', 'filename_x', 'description', 'productname', 'productversion', 'publisher_y', 'publisher_x', 'netdomain', 'hostname', 'username', 'pprocess', 'gprocess', 'commandline', 'reputation_lastseen', 'reputation_scannercount', 'reputation_scannermatch', 'reputation_status', 'reputation_threatlevel', 'reputation_threatname', 'reputation_timestamp']] + aug_df = df[['sha256', 'filename_x', 'description', 'productname', 'productversion', 'publisher_y', 'publisher_x', 'netdomain', 'hostname', 'username', 'reputation_lastseen', 'reputation_scannermatch', 'reputation_scannercount', 'reputation_status', 'reputation_threatlevel', 'reputation_threatname', 'reputation_timestamp', 'pprocess', 'gprocess', 'commandline', ]] return aug_df @@ -126,18 +127,119 @@ def categorizeHashes(aug_df: pd.DataFrame, threat_tolerance: int, untrusted_publ return needsreview_df, approved_df, unapproved_df -def destinationbuilder(df, df2): - # Step 1: Explode the 'sha256' list in df2 to create one row per sha256 value - df_expanded = df.explode('sha256') - # Step 2: Create a new dataframe for the result - df_hashdestination = df_expanded.copy() +import pandas as pd - # Step 3: Populate the 'Destination Allowlist' column based on comparison with df3 - df_hashdestination['Destination Allowlist'] = df_hashdestination['sha256'].apply( - lambda x: 'Parent Policy Baseline' if x in df2['sha256'].values else "Destination Policy Allowlist" +def explode_and_deduplicate(df): + df['sha256'] = df['sha256'].str.split(',') + df = df.explode('sha256') + return df.drop_duplicates().reset_index(drop=True) + +def clean_sha256(df, column='sha256'): + """Discard quotes, brackets, and whitespace from sha256 values.""" + df[column] = df[column].astype(str).str.strip("'[]\" ") + return df + +def destinationHashes( + df_approved_paths: pd.DataFrame, + df_approved_hashes: pd.DataFrame, + df_hashes_auto_approved: pd.DataFrame, + df_hashes_manually_approved: pd.DataFrame, +): + # Deduplicate and explode all input DataFrames + df_approved_paths = explode_and_deduplicate(df_approved_paths) + df_approved_hashes = explode_and_deduplicate(df_approved_hashes) + df_hashes_auto_approved = explode_and_deduplicate(df_hashes_auto_approved) + df_hashes_manually_approved = explode_and_deduplicate(df_hashes_manually_approved) + + # Clean sha256 values in all relevant DataFrames + df_approved_hashes = clean_sha256(df_approved_hashes) + df_hashes_auto_approved = clean_sha256(df_hashes_auto_approved) + df_hashes_manually_approved = clean_sha256(df_hashes_manually_approved) + + # Create sets for faster lookup + auto_approved_sha256 = set(df_hashes_auto_approved['sha256'].values) + manually_approved_sha256 = set(df_hashes_manually_approved['sha256'].values) + + # Debug: Print unmatched hashes + unmatched = set(df_approved_hashes['sha256']) - (auto_approved_sha256 | manually_approved_sha256) + print(f"Unmatched hashes: {unmatched}") + + # Process df_approved_paths + df_paths = df_approved_paths.assign(destination='Path Exclusion') + df_paths = df_paths[['sha256', 'description', 'destination', 'grouped_directory', 'filename']] + + # Process df_approved_hashes + df_hashes = df_approved_hashes.copy() + df_hashes['destination'] = df_hashes['sha256'].apply( + lambda x: 'Parent Policy Baseline' if x in auto_approved_sha256 + else ('Child Policy Allowlist' if x in manually_approved_sha256 else None) ) + df_hashes = df_hashes.dropna(subset=['destination']) + df_hashes = df_hashes.assign(grouped_directory=None) - # Step 4: Return the new dataframe + # Use 'filename_x' only if it exists, otherwise fallback to 'filename' + filename_col = 'filename_x' if 'filename_x' in df_hashes.columns else 'filename' + selected_cols = ['sha256', 'description', 'destination', 'grouped_directory', filename_col] + df_hashes = df_hashes[selected_cols] + + # Concatenate results + df_hashdestination = pd.concat([df_paths, df_hashes], ignore_index=True) return df_hashdestination + + +""" + +def explode_and_deduplicate(df): + df['sha256'] = df['sha256'].str.split(',') + df = df.explode('sha256') + return df.drop_duplicates().reset_index(drop=True) + +def clean_sha256(df, column='sha256'): + df[column] = df[column].astype(str).str.strip("'[]\" ") + return df + +def destinationHashes( + df_approved_paths: pd.DataFrame, + df_approved_hashes: pd.DataFrame, + df_hashes_auto_approved: pd.DataFrame, + df_hashes_manually_approved: pd.DataFrame, +): + # Deduplicate and explode all input DataFrames + df_approved_paths = explode_and_deduplicate(df_approved_paths) + df_approved_hashes = explode_and_deduplicate(df_approved_hashes) + df_hashes_auto_approved = explode_and_deduplicate(df_hashes_auto_approved) + df_hashes_manually_approved = explode_and_deduplicate(df_hashes_manually_approved) + + # Clean sha256 values in all relevant DataFrames + df_approved_hashes = clean_sha256(df_approved_hashes) + df_hashes_auto_approved = clean_sha256(df_hashes_auto_approved) + df_hashes_manually_approved = clean_sha256(df_hashes_manually_approved) + + # Create sets for faster lookup + auto_approved_sha256 = set(df_hashes_auto_approved['sha256'].values) + manually_approved_sha256 = set(df_hashes_manually_approved['sha256'].values) + + # Debug: Print unmatched hashes + unmatched = set(df_approved_hashes['sha256']) - (auto_approved_sha256 | manually_approved_sha256) + print(f"Unmatched hashes: {unmatched}") + + # Process df_approved_paths + df_paths = df_approved_paths.assign( + destination='Path Exclusion' + )[['sha256', 'description', 'destination', 'grouped_directory']] + + # Process df_approved_hashes + df_hashes = df_approved_hashes.copy() + df_hashes['destination'] = df_hashes['sha256'].apply( + lambda x: 'Parent Policy Baseline' if x in auto_approved_sha256 + else ('Child Policy Allowlist' if x in manually_approved_sha256 else None) + ) + df_hashes = df_hashes.dropna(subset=['destination']) + df_hashes = df_hashes.assign(grouped_directory=None)[['sha256', 'description', 'destination', 'grouped_directory']] + + # Concatenate results + df_hashdestination = pd.concat([df_paths, df_hashes], ignore_index=True) + return df_hashdestination +""" \ No newline at end of file diff --git a/utils/pathfunctions.py b/utils/pathfunctions.py index ab8e166..a5953f6 100644 --- a/utils/pathfunctions.py +++ b/utils/pathfunctions.py @@ -115,4 +115,19 @@ def filepathInitialGroup(df: pd.DataFrame): path_eligible = deduplicate_lists(path_eligible) path_ineligible = deduplicate_lists(path_ineligible) - return path_eligible, path_ineligible \ No newline at end of file + return path_eligible, path_ineligible + +def filter_and_drop(approved, eligiblepaths, min_hashes): + """ + Filters eligiblepaths to rows where all hashes are in approved, + then drops rows with fewer than min_hashes hashes. + """ + approved_hashes = set(approved['sha256']) + + def all_hashes_approved(row): + return all(h in approved_hashes for h in row['sha256']) + + filtered = eligiblepaths[eligiblepaths.apply(all_hashes_approved, axis=1)] + filtered = filtered[filtered['sha256'].apply(len) >= min_hashes] + + return filtered