From ebfb0d80a9d136c3e7793378a09abbd6ad57f2fd Mon Sep 17 00:00:00 2001 From: = <=> Date: Tue, 26 Aug 2025 12:20:09 -0400 Subject: [PATCH] Pretty HTML and beginnings of Destination Policy --- AirlockTools.py | 47 +++++++------ utils/allowlist.py | 3 +- utils/colortext.py | 14 ---- utils/getdeviceevents.py | 2 +- utils/hashfunctions.py | 16 +++++ utils/pretty.py | 138 +++++++++++++++++++++++++++++++++++++++ 6 files changed, 184 insertions(+), 36 deletions(-) delete mode 100644 utils/colortext.py create mode 100644 utils/pretty.py diff --git a/AirlockTools.py b/AirlockTools.py index 20cfaf2..e0a512f 100644 --- a/AirlockTools.py +++ b/AirlockTools.py @@ -19,7 +19,7 @@ import utils.getdeviceevents import utils.allowlist import utils.hashfunctions import utils.pathfunctions -import utils.colortext as ct +import utils.pretty as ct import urllib3 import pandas as pd import ast @@ -186,7 +186,7 @@ def menu_prepare_to_enforce(): 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")) - if os.path.isfile(f"manuallyapproved\\df_paths_needing_review_{first_policy}_{second_policy}.csv"): + 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"): print(ct.colorText(" [✓] This step has been completed","green")) else: print(ct.colorText(" [✗] This step has not been completed","red")) @@ -215,43 +215,43 @@ def menu_prepare_to_enforce(): if not os.path.exists("dataframe_csv\\df_aggregated_{first_policy}.csv"): executionhist_policy1 = utils.allowlist.pullPolicyExechistories(url,first_policy_tuple[0], first_policy_tuple[1],True) df_aggregated_policy1 = utils.hashfunctions.aggregateHashes(executionhist_policy1) - df_aggregated_policy1.to_html(f"dataframe_html\\df_aggregated_{first_policy}.html", index=False) df_aggregated_policy1.to_csv(f"dataframe_csv\\df_aggregated_{first_policy}.csv", index=False) + ct.style_dataframe_dark(df_aggregated_policy1, f"dataframe_html\\df_aggregated_{first_policy}.html") print(ct.colorText(f"Staging of Exection history for policy: {first_policy} is complete","green")) if not os.path.exists("dataframe_csv\\df_aggregated_{second_policy}.csv"): executionhist_policy2 = utils.allowlist.pullPolicyExechistories(url,second_policy_tuple[0], second_policy_tuple[1],True) df_aggregated_policy2 = utils.hashfunctions.aggregateHashes(executionhist_policy2) - df_aggregated_policy2.to_html(f"dataframe_html\\df_aggregated_{second_policy}.html", index=False) df_aggregated_policy2.to_csv(f"dataframe_csv\\df_aggregated_{second_policy}.csv", index=False) + ct.style_dataframe_dark(df_aggregated_policy2, f"dataframe_html\\df_aggregated_{second_policy}.html") print(ct.colorText(f"Staging of Exection history for policy: {second_policy} is complete","green")) elif choice == "3": if second_policy is first_policy and os.path.exists(f"dataframe_csv\\df_aggregated_{first_policy}.csv"): df1 = tryToReadCSV(f"dataframe_csv\\df_aggregated_{first_policy}.csv") df_aggregated_combo = df1 - - df_aggregated_combo.to_html(f"dataframe_html\\df_aggregated_combo_{first_policy}_{second_policy}.html", index=False) df_aggregated_combo.to_csv(f"dataframe_csv\\df_aggregated_combo_{first_policy}_{second_policy}.csv", index=False) + ct.style_dataframe_dark(df_aggregated_combo, f"dataframe_html\\df_aggregated_combo_{first_policy}_{second_policy}.html") + print(ct.colorText(f"Dataframes have been aggregated (combined)","green")) elif os.path.exists(f"dataframe_csv\\df_aggregated_{first_policy}.csv") and os.path.exists(f"dataframe_csv\\df_aggregated_{second_policy}.csv"): df1 = tryToReadCSV(f"dataframe_csv\\df_aggregated_{first_policy}.csv") df2 = tryToReadCSV(f"dataframe_csv\\df_aggregated_{second_policy}.csv") - df_aggregated_combo = pd.concat([df1 , df2], ignore_index=True) - df_aggregated_combo.to_html(f"dataframe_html\\df_aggregated_combo_{first_policy}_{second_policy}.html", index=False) df_aggregated_combo.to_csv(f"dataframe_csv\\df_aggregated_combo_{first_policy}_{second_policy}.csv", index=False) + ct.style_dataframe_dark(df_aggregated_combo, f"dataframe_html\\df_aggregated_combo_{first_policy}_{second_policy}.html") + print(ct.colorText(f"Dataframes have been aggregated (combined)","green")) + else: print(ct.colorText(f"Please stage your data before attempting this step","red")) elif choice == "4": if os.path.exists(f"dataframe_csv\\df_aggregated_combo_{first_policy}_{second_policy}.csv"): df_augmented = utils.hashfunctions.augmentAggregatedHashes(url,tryToReadCSV(f"dataframe_csv\\df_aggregated_combo_{first_policy}_{second_policy}.csv")) - - df_augmented.to_html(f"dataframe_html\\df_augmented_combo_{first_policy}_{second_policy}.html", index=False) df_augmented.to_csv(f"dataframe_csv\\df_augmented_combo_{first_policy}_{second_policy}.csv", index=False) - + ct.style_dataframe_dark(df_augmented, f"dataframe_html\\df_augmented_combo_{first_policy}_{second_policy}.html") print(ct.colorText(f"Hash reputation info added to dataframe","green")) + else: print(ct.colorText(f"Please combine your data with step 3 prior to attempting this step","red")) @@ -259,16 +259,17 @@ def menu_prepare_to_enforce(): if os.path.exists(f"dataframe_csv\\df_augmented_combo_{first_policy}_{second_policy}.csv"): categorized = utils.hashfunctions.categorizeHashes(pd.read_csv(f"dataframe_csv\\df_augmented_combo_{first_policy}_{second_policy}.csv"), threat_tolerance_constant, badpublisherlist) - categorized[0].to_html(f"dataframe_html\\df_hashes_needing_approval_{first_policy}_{second_policy}.html", index=False) categorized[0].to_csv(f"dataframe_csv\\df_hashes_needing_approval_{first_policy}_{second_policy}.csv", index=False) + ct.style_dataframe_dark(categorized[0], f"dataframe_html\\df_hashes_needing_approval_{first_policy}_{second_policy}.html") - categorized[1].to_html(f"dataframe_html\\df_automatically_approved_hashes_{first_policy}_{second_policy}.html", index=False) categorized[1].to_csv(f"dataframe_csv\\df_automatically_approved_hashes_{first_policy}_{second_policy}.csv", index=False) + ct.style_dataframe_dark(categorized[1], f"dataframe_html\\df_automatically_approved_hashes_{first_policy}_{second_policy}.html") - categorized[2].to_html(f"dataframe_html\\df_unapproved_hashes__{first_policy}_{second_policy}.html", index=False) categorized[2].to_csv(f"dataframe_csv\\df_unapproved_hashes__{first_policy}_{second_policy}.csv", index=False) + ct.style_dataframe_dark(categorized[2], f"dataframe_html\\df_unapproved_hashes_{first_policy}_{second_policy}.html") print(ct.colorText(f"Hashes have been categorized","green")) + else: print(ct.colorText(f"Please Augment your data with hash threat info using step 4 prior to attempting this step","red")) @@ -278,19 +279,19 @@ def menu_prepare_to_enforce(): df2 = tryToReadCSV(f"manuallyapproved\\df_automatically_approved_hashes_{first_policy}_{second_policy}.csv") df_all_approved_hashes = pd.concat([df1 , df2], ignore_index=True) - df_all_approved_hashes.to_html(f"dataframe_html\\df_all_approved_hashes_{first_policy}_{second_policy}.html", index=False) df_all_approved_hashes.to_csv(f"dataframe_csv\\df_all_approved_hashes_{first_policy}_{second_policy}.csv", index=False) - + ct.style_dataframe_dark(df_all_approved_hashes, f"dataframe_html\\df_all_approved_hashes_{first_policy}_{second_policy}.html") df_paths_needing_review, df_path_ineligible = utils.pathfunctions.filepathInitialGroup(pd.read_csv(f"dataframe_csv\\df_all_approved_hashes_{first_policy}_{second_policy}.csv")) - df_paths_needing_review.to_html(f"dataframe_html\\df_paths_needing_review_{first_policy}_{second_policy}.html", index=False) df_paths_needing_review.to_csv(f"dataframe_csv\\df_paths_needing_review_{first_policy}_{second_policy}.csv", index=False) + ct.style_dataframe_dark(df_paths_needing_review, f"dataframe_html\\df_paths_needing_review_{first_policy}_{second_policy}.html") - df_path_ineligible.to_html(f"dataframe_html\\df_path_ineligible_{first_policy}_{second_policy}.html", index=False) df_path_ineligible.to_csv(f"dataframe_csv\\df_path_ineligible_{first_policy}_{second_policy}.csv", index=False) + ct.style_dataframe_dark(df_path_ineligible, f"dataframe_html\\df_path_ineligible_{first_policy}_{second_policy}.html") print(ct.colorText(f"Eligible paths determined","green")) + else: print(ct.colorText(f"Please manually approve hashes prior to this step","red")) @@ -299,8 +300,14 @@ def menu_prepare_to_enforce(): 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") - print(df1['grouped_directory']) - print(df2['sha256']) + 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") + + else: + print(ct.colorText(f"Please manually approve suggested paths prior to this step","red")) + diff --git a/utils/allowlist.py b/utils/allowlist.py index 46c48aa..c4fb983 100644 --- a/utils/allowlist.py +++ b/utils/allowlist.py @@ -17,8 +17,9 @@ import requests import json import os import time -import utils.colortext as ct +import utils.pretty as ct import ijson + def pullPolicyExechistories(url, choice, policiesnames, outputjson: bool): headers = { "X-APIKey": os.getenv('APIKEY') diff --git a/utils/colortext.py b/utils/colortext.py deleted file mode 100644 index fcd101c..0000000 --- a/utils/colortext.py +++ /dev/null @@ -1,14 +0,0 @@ - -def colorText(text: str, color: str) -> str: - colors = { - "red": "\033[91m", - "green": "\033[92m", - "yellow": "\033[93m", - "blue": "\033[94m", - "magenta": "\033[95m", - "cyan": "\033[96m", - "white": "\033[97m", - "reset": "\033[0m" - } - - return f"{colors.get(color, colors['reset'])}{text}{colors['reset']}" \ No newline at end of file diff --git a/utils/getdeviceevents.py b/utils/getdeviceevents.py index 85bd8dd..fbb5d66 100644 --- a/utils/getdeviceevents.py +++ b/utils/getdeviceevents.py @@ -16,7 +16,7 @@ import datetime import requests import json import os -import utils.colortext as ct +import utils.pretty as ct def devicehistory(url, outputjson: bool): endpoint = url + '/v1/getexechistory' diff --git a/utils/hashfunctions.py b/utils/hashfunctions.py index c193284..771cbe3 100644 --- a/utils/hashfunctions.py +++ b/utils/hashfunctions.py @@ -125,3 +125,19 @@ def categorizeHashes(aug_df: pd.DataFrame, threat_tolerance: int, untrusted_publ unapproved_df = df[~(mask_needsreview | mask_approved)] 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() + + # 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" + ) + + # Step 4: Return the new dataframe + return df_hashdestination + diff --git a/utils/pretty.py b/utils/pretty.py new file mode 100644 index 0000000..58d89d5 --- /dev/null +++ b/utils/pretty.py @@ -0,0 +1,138 @@ + +def colorText(text: str, color: str) -> str: + colors = { + "red": "\033[91m", + "green": "\033[92m", + "yellow": "\033[93m", + "blue": "\033[94m", + "magenta": "\033[95m", + "cyan": "\033[96m", + "white": "\033[97m", + "reset": "\033[0m" + } + + return f"{colors.get(color, colors['reset'])}{text}{colors['reset']}" + +def style_dataframe_dark(df, output_html_path=None, overwrite=True): + from datetime import datetime + + # Get current date and filename for subtitle + today = datetime.now().strftime("%d %B %Y") # Changed to "Day Month Year" + filename = output_html_path.replace('.html', '') if output_html_path else "Report" + + dark_css = """ + + """ + + header = f""" +
+

Airlock Tools

+

{filename} - {today}

+
+ """ + + html_table = df.to_html(index=False, escape=False) + styled_html = ( + f"\n" + f"Airlock Tools Report\n" + f"\n" + f"{dark_css}\n" + f"{header}\n" + f"
\n" + f" {html_table}\n" + f"
\n" + f"\n" + f"" + ) + if output_html_path: + with open(output_html_path, "w", encoding="utf-8") as f: + f.write(styled_html) + print(f"✅ Styled table saved to '{output_html_path}'") + elif overwrite: + import tempfile + temp_path = tempfile.mktemp(suffix=".html") + with open(temp_path, "w", encoding="utf-8") as f: + f.write(styled_html) + print(f"✅ Styled table saved to temporary file: {temp_path}") + else: + return styled_html