From bb3cfb9bd46e7ee40118690c9b3cbee7828a901a Mon Sep 17 00:00:00 2001 From: = <=> Date: Fri, 29 Aug 2025 11:46:57 -0400 Subject: [PATCH] Fixed some breaking changes and added some QOL improvements on HMTL Preflight --- AirlockTools.py | 40 +++++++++++++++++++++++++++++++++------- utils/allowlist.py | 3 +-- 2 files changed, 34 insertions(+), 9 deletions(-) diff --git a/AirlockTools.py b/AirlockTools.py index 2a63442..8877c2e 100644 --- a/AirlockTools.py +++ b/AirlockTools.py @@ -268,6 +268,7 @@ def menu_prepare_to_enforce(): if answer in ("yes", "y"): choice, policynames, policyid = utils.allowlist.listPolicies(url) second_policy = policynames[choice] + break elif answer in ("no", "n"): second_policy = first_policy @@ -278,17 +279,30 @@ def menu_prepare_to_enforce(): elif choice == "2": if not os.path.exists(f"dataframe_csv\\executionhist_{first_policy}.csv"): - exe1 = utils.allowlist.pullPolicyExechistories(url,choice, policynames,True) + print(choice) + print(first_policy) + + + exe1 = utils.allowlist.pullPolicyExechistories(url, choice, first_policy ,True) + print(exe1) data = json.loads(exe1) executionhist_policy1 = pd.DataFrame(data["response"]["exechistories"]) + + if not executionhist_policy1.empty: + executionhist_policy1 = executionhist_policy1.sort_values(by=['sha256','filename']) + executionhist_policy1.to_csv(f"dataframe_csv\\executionhist_{first_policy}.csv", index=False) ct.style_dataframe_dark(executionhist_policy1, f"dataframe_html\\executionhist_{first_policy}.html") print(ct.colorText(f"Staging of Execution history for policy: {first_policy} is complete","green")) if not os.path.exists(f"dataframe_csv\\executionhist_{second_policy}.csv"): - exe2 = utils.allowlist.pullPolicyExechistories(url,choice, policynames,True) + exe2 = utils.allowlist.pullPolicyExechistories(url,choice, second_policy,True) data2 = json.loads(exe2) executionhist_policy2 = pd.DataFrame(data2["response"]["exechistories"]) + + if not executionhist_policy2.empty: + executionhist_policy2 = executionhist_policy2.sort_values(by=['sha256','filename']) + executionhist_policy2.to_csv(f"dataframe_csv\\executionhist_{second_policy}.csv", index=False) ct.style_dataframe_dark(executionhist_policy2, f"dataframe_html\\executionhist_{second_policy}.html") print(ct.colorText(f"Staging of Exection history for policy: {first_policy} is complete","green")) @@ -302,7 +316,7 @@ def menu_prepare_to_enforce(): print(ct.colorText(f"Dataframes have been combined","green")) elif os.path.exists(f"dataframe_csv\\executionhist_{first_policy}.csv") and os.path.exists(f"dataframe_csv\\executionhist_{second_policy}.csv"): - execuctionhist_combined = pd.concat([tryToReadCSV(f"dataframe_csv\\executionhist_{first_policy}.csv") , tryToReadCSV(f"dataframe_csv\\executionhist_{second_policy}.csv")], ignore_index=True) + execuctionhist_combined = pd.concat([tryToReadCSV(f"dataframe_csv\\executionhist_{first_policy}.csv") , tryToReadCSV(f"dataframe_csv\\executionhist_{second_policy}.csv")], ignore_index=True).sort_values(by=['sha256','filename']) execuctionhist_combined.to_csv(f"dataframe_csv\\execuctionhist_combined_{first_policy}_{second_policy}.csv", index=False) ct.style_dataframe_dark(df_aggregated_combo, f"dataframe_html\\execuctionhist_combined_{first_policy}_{second_policy}.html") print(ct.colorText(f"Dataframes have been combined","green")) @@ -327,13 +341,22 @@ def menu_prepare_to_enforce(): break else: categorized = utils.hashfunctions.categorizeHashes(pd.read_csv(f"dataframe_csv\\augmented_combo_{first_policy}_{second_policy}.csv"), threat_tolerance_constant, badpublisherlist) + + if not categorized[0].empty: + categorized[0].sort_values(by=['sha256','filename_x']) categorized[0].to_csv(f"dataframe_csv\\hashes_needing_approval_{first_policy}_{second_policy}.csv", index=False) ct.style_dataframe_dark(categorized[0], f"dataframe_html\\dashes_needing_approval_{first_policy}_{second_policy}.html") + if not categorized[1].empty: + categorized[1].sort_values(by=['sha256','filename_x']) + categorized[1].to_csv(f"dataframe_csv\\automatically_approved_hashes_{first_policy}_{second_policy}.csv", index=False) ct.style_dataframe_dark(categorized[1], f"dataframe_html\\automatically_approved_hashes_{first_policy}_{second_policy}.html") + if not categorized[2].empty: + categorized[2].sort_values(by=['sha256','filename_x']) + categorized[2].to_csv(f"dataframe_csv\\unapproved_hashes__{first_policy}_{second_policy}.csv", index=False) ct.style_dataframe_dark(categorized[2], f"dataframe_html\\unapproved_hashes_{first_policy}_{second_policy}.html") @@ -346,7 +369,7 @@ def menu_prepare_to_enforce(): df1 = tryToReadCSV(f"manuallyapproved\\hashes_needing_approval_{first_policy}_{second_policy}.csv") df2 = tryToReadCSV(f"manuallyapproved\\automatically_approved_hashes_{first_policy}_{second_policy}.csv") - all_approved_hashes = pd.concat([df1 , df2], ignore_index=True) + all_approved_hashes = pd.concat([df1 , df2], ignore_index=True).sort_values(by=['sha256','filename_x']) print(ct.colorText(f"Approved hash lists have been combined","green")) all_approved_hashes.to_csv(f"dataframe_csv\\all_approved_hashes_{first_policy}_{second_policy}.csv", index=False) ct.style_dataframe_dark(all_approved_hashes, f"dataframe_html\\all_approved_hashes_{first_policy}_{second_policy}.html") @@ -378,7 +401,9 @@ def menu_prepare_to_enforce(): ct.style_dataframe_dark(pathexclusions, f"preflight\\final_path_exclusions_{first_policy}_{second_policy}.html") allowbyhash.to_csv(f"preflight\\final_hash_approvals_{first_policy}_{second_policy}.csv", index=False) - ct.style_dataframe_dark(allowbyhash, f"preflight\\final_hash_approvals_{first_policy}_{second_policy}.html") + + easyview = allowbyhash.groupby('sha256').agg(list).reset_index() + ct.style_dataframe_dark(easyview, f"preflight\\final_hash_approvals_{first_policy}_{second_policy}.html") elif choice == "5": @@ -422,12 +447,13 @@ def menu_prepare_to_enforce(): utils.policyfunctions.addPath(destination_id,pathexcludelist) print(ct.colorText(f"Adding hashes to {allowlist_parent_name}", "yellow")) - allowlist_parenthashlist = allowbyhash[allowbyhash['reputation_status'] == 'KNOWN']['sha256'].tolist() + + allowlist_parenthashlist = allowbyhash[allowbyhash['reputation_status'] == 'KNOWN']['sha256'].unique().tolist() utils.policyfunctions.addHash(allowlist_parent_id,allowlist_parenthashlist) print(ct.colorText(f"Adding hashes to {allowlist_child_name}", "yellow")) - allowlist_childhashlist = allowbyhash[allowbyhash['reputation_status'] == 'UNKNOWN']['sha256'].tolist() + allowlist_childhashlist = allowbyhash[allowbyhash['reputation_status'] == 'UNKNOWN']['sha256'].unique().tolist() utils.policyfunctions.addHash(allowlist_child_id, allowlist_childhashlist) else: diff --git a/utils/allowlist.py b/utils/allowlist.py index 5b6b262..758ec08 100644 --- a/utils/allowlist.py +++ b/utils/allowlist.py @@ -27,7 +27,7 @@ def pullPolicyExechistories(url, choice, policiesnames, outputjson: bool): checkpoint = '000000000000000000000000' json_output = {'error': 'Success', 'response': {'exechistories': []}} while True: - json_response_data = checkpoint_stomper(checkpoint, url, policiesnames[choice], headers) + json_response_data = checkpoint_stomper(checkpoint, url, policiesnames, headers) if not json_response_data['response']['exechistories']: break match_found = False @@ -76,7 +76,6 @@ def checkpoint_stomper(checkpoint, url, policy, headers): json_output['response']['exechistories'].append(item) parse_text = json.loads(json.dumps(json_output)) return parse_text - def listPolicies(url): endpoint = url + '/v1/group' print(ct.colorText("[+] Grabbing All Policies", "cyan"))