From 26ea73392b346d6069f416ed54496ad7c4ccc358 Mon Sep 17 00:00:00 2001 From: = <=> Date: Fri, 19 Sep 2025 12:19:35 -0400 Subject: [PATCH] Cleaned up Generate Unexclueded Function to keep its files in check --- AirlockTools.py | 30 ++++++++++++------------------ utils/pathfunctions.py | 12 ++++++------ utils/policyfunctions.py | 20 +------------------- 3 files changed, 19 insertions(+), 43 deletions(-) diff --git a/AirlockTools.py b/AirlockTools.py index 00b52de..0f30b2e 100644 --- a/AirlockTools.py +++ b/AirlockTools.py @@ -121,7 +121,7 @@ def menu_main(): ct.displayIntro(); print(ct.colorText("1. Get All Events for Single Device", "yellow")) print(ct.colorText("2. OTP", "yellow")) - print(ct.colorText("3. Divide by Exclusion", "yellow")) + print(ct.colorText("3. Find Unexcluded from 24hr Execution", "yellow")) print(ct.colorText("4. Prepare Policy For Enforcement", "yellow")) print(ct.colorText("5. Update Audit Policies from Enforcement Policies", "yellow")) print(ct.colorText("Q. Quit", "yellow")) @@ -132,7 +132,7 @@ def menu_main(): elif choice == "2": menu_otp() elif choice == "3": - menu_feature2() + find_unexcluded() elif choice == "4": menu_prepare_to_enforce() elif choice == "5": @@ -162,15 +162,17 @@ def menu_otp(): else: print("Invalid choice. Please try again.") -def menu_feature2(): +def find_unexcluded(): while True: - print("\n--- Submenu ---") + print("\n--- Follow Steps Sequentially, Files will land in directory named 'exclusions'") print("1. Pull last 24 hours execution for all ATPolicys") print("2. Generate Paths") print("3. Merge") print("4. Pull existing paths") print("5. Divide By Excluded or not excluded by path") print("Q. Exit") + + if not os.path.exists("exclusions"): os.makedirs("exclusions") choice = input("Enter your choice: ") if choice == "1": @@ -203,7 +205,7 @@ def menu_feature2(): combined_df = pd.DataFrame() print("⚠️ No data was retrieved.") - combined_df.to_csv("allATPolicyExecs.csv", index=False) + combined_df.to_csv("exclusions\\allATPolicyExecs.csv", index=False) elif choice == "2": @@ -248,12 +250,12 @@ def menu_feature2(): print("⚠️ No paths were retrieved.") # Save to CSV - paths_df.to_csv("paths.csv", index=False) + paths_df.to_csv("exclusions\\paths.csv", index=False) elif choice == "5": # Load and clean data - filenames_df = pd.read_csv('merged_output.csv') # Contains 'PolicyName' and 'filename' - exclusions_df = pd.read_csv('paths.csv') # Contains 'PolicyName' and 'Path' + filenames_df = pd.read_csv('exclusions\\merged_output.csv') # Contains 'PolicyName' and 'filename' + exclusions_df = pd.read_csv('exclusions\\paths.csv') # Contains 'PolicyName' and 'Path' # Clean and normalize columns filenames_df['filename'] = filenames_df['filename'].fillna('').astype(str).str.strip() @@ -308,16 +310,8 @@ def menu_feature2(): # Save results - excluded.to_csv('excluded_filenames.csv', index=False) - not_excluded.to_csv('non_excluded_filenames.csv', index=False) - - - elif choice == "6": - - pattern = utils.pathfunctions.wildcardRegex("C:\\Windows\\SystemTemp\\????????\\????????.dll") - test_path = "C:\\Windows\\SystemTemp\\fsq2xzua\\fsq2xzua.dll" - - print("Match:", pattern.match(test_path) is not None) + excluded.to_csv('exclusions\\excluded_filenames.csv', index=False) + not_excluded.to_csv('exclusions\\non_excluded_filenames.csv', index=False) elif choice == "Q": diff --git a/utils/pathfunctions.py b/utils/pathfunctions.py index fbaec32..658b870 100644 --- a/utils/pathfunctions.py +++ b/utils/pathfunctions.py @@ -202,7 +202,7 @@ def allATpaths(url, pups, untrusted_publishers, badpathparts, threat_tolerance, import pandas as pd # Load raw data - hashes = pd.read_csv("allATPolicyExecs.csv") + hashes = pd.read_csv("exclusions\\allATPolicyExecs.csv") # Deduplicate hashes before augmentation deduped_hashes = hashes.drop_duplicates(subset=['sha256']).copy() @@ -245,7 +245,7 @@ def allATpaths(url, pups, untrusted_publishers, badpathparts, threat_tolerance, if 'PolicyName_x' in final_augmented.columns: final_augmented = final_augmented.rename(columns={'PolicyName_x': 'PolicyName'}) - final_augmented.to_csv("testing.csv", index=False) + final_augmented.to_csv("exclusions\\testing.csv", index=False) # Reputation flag logic def reputationtool(row): @@ -306,21 +306,21 @@ def allATpaths(url, pups, untrusted_publishers, badpathparts, threat_tolerance, review_df = review_df.merge(sha_counts, on='longestcfp', how='left') review_df = review_df[review_df['unique_sha256_count'] >= min_files_for_path] - review_df.to_csv("ALL_AT_PATHS.csv", index=False) + review_df.to_csv("exclusions\\ALL_AT_PATHS.csv", index=False) # Cleanup del lcp_not_forbidden, sha_counts, review_df def mergeTesting(): # Load the two CSVs - testing_df = pd.read_csv("testing.csv") - paths_df = pd.read_csv("ALL_AT_PATHS.csv") + testing_df = pd.read_csv("exclusions\\testing.csv") + paths_df = pd.read_csv("exclusions\\ALL_AT_PATHS.csv") # Merge on 'sha256' with testing as the left DataFrame merged_df = testing_df.merge(paths_df, on="sha256", how="left") # Save the merged result - merged_df.to_csv("merged_output.csv", index=False) + merged_df.to_csv("exclusions\\merged_output.csv", index=False) print(f"Merged DataFrame saved with {len(merged_df)} rows.") diff --git a/utils/policyfunctions.py b/utils/policyfunctions.py index cd2acf0..eda13d0 100644 --- a/utils/policyfunctions.py +++ b/utils/policyfunctions.py @@ -354,25 +354,7 @@ def sendToPolicyTest(url, first_policy, second_policy, destination_name, destina allowlist_childhashlist = allowbyhash[allowbyhash['reputation_status'] == 'UNKNOWN']['sha256'].unique().tolist() addHash(url, allowlist_child_id, allowlist_childhashlist) -def getMetaRules(url,appid): - endpoint = url + '/v1/application/export' - payload = { - "applicationid" : {appid} - } - headers = { - "X-APIKey": os.getenv('APIKEY') - } - response = requests.request("POST", endpoint, headers=headers, data=payload, verify=False) - parse_text = json.loads(response.text) - policiesnames = [] - policyids = [] - for index, list in enumerate(parse_text['response']['groups'], start=1): - print(ct.colorText(f"{index}. {list['name']}", "yellow")) - policiesnames.append(list['name']) - policyids.append(list['groupid']) - choice = input(ct.colorText("Select Policy Group: ", "white")) - choice = int(choice) - 1 - return choice, policiesnames, policyids + def updateAuditPoliciesFromEnforcementPolices(url, policy_relationship_map): for enforcement_policy, audit_policy in policy_relationship_map.items():