diff --git a/AirlockTools.py b/AirlockTools.py index 1be9010..04f94c5 100644 --- a/AirlockTools.py +++ b/AirlockTools.py @@ -15,15 +15,11 @@ import argparse import dotenv import os -import re -import pandas as pd import urllib3 import utils.clientfunctions -import utils.hashfunctions import utils.otpfunctions -import utils.pathfunctions import utils.policyfunctions -import utils.pretty as ct +import utils.utils as ct from utils.perstscheduler import register_function, run_once_job, recurring_job, reload_jobs, start_scheduler @@ -47,456 +43,95 @@ policy_relationship_map ={ #Enforcement : Audit #"bf0b1f9b-bfea-4f44-97c0-80e27ff61712" : "d126db36-72ed-4937-adc7-d88b7509a5b5" #AT Servers General, AT Testing } - - - + def main(): - parser = argparse.ArgumentParser(description="Your script description") - parser.add_argument('--monitorOTP', action='store_true', help='Run in non-interactive mode') - # Add other arguments as needed + parser = argparse.ArgumentParser(description="Your script description") + parser.add_argument('--monitorOTP', action='store_true', help='Run in non-interactive mode') + # Add other arguments as needed - args = parser.parse_args() + args = parser.parse_args() - if args.monitorOTP: + if args.monitorOTP: # Non-interactive logic - print(f"Running non-interactively to start monitoring OTP") - - os.makedirs("scheduling", exist_ok=True) - os.makedirs("OTP/HTML", exist_ok=True) - os.makedirs("OTP/PARQ", exist_ok=True) + print(f"Running non-interactively to start monitoring OTP") + + os.makedirs("scheduling", exist_ok=True) + os.makedirs("OTP/HTML", exist_ok=True) + os.makedirs("OTP/PARQ", exist_ok=True) - apivalidation() - - register_function("monitorOTP", utils.otpfunctions.monitorOTP) - register_function("updateAudit", utils.policyfunctions.updateAuditPoliciesFromEnforcementPolices) - - if not os.path.exists("scheduling\\jobs.json"): - recurring_job("monitorOTP", "monitorOTP", interval=60, unit="seconds", args=[url, pups]) - recurring_job("updateAudit", "updateAudit", interval=10, unit="minutes", args=[url, policy_relationship_map]) - else: - reload_jobs() - start_scheduler() + ct.apivalidation() + + register_function("monitorOTP", utils.otpfunctions.monitorOTP) + register_function("updateAudit", utils.policyfunctions.updateAuditPoliciesFromEnforcementPolices) + + if not os.path.exists("scheduling\\jobs.json"): + recurring_job("monitorOTP", "monitorOTP", interval=60, unit="seconds", args=[url, pups]) + recurring_job("updateAudit", "updateAudit", interval=10, unit="minutes", args=[url, policy_relationship_map]) + else: + reload_jobs() + start_scheduler() - else: + else: # Interactive logic - apivalidation() - menu_main() - - -def apivalidation(): - match os.getenv('APIKEY'): - case '': - print(ct.colorText("Please add your API Key to the .env file", "red")) - - -def tryToReadCSV(csv): - try: - if not os.path.exists(csv): - print(ct.colorText(f"Error: File '{csv}' does not exist.", "red")) - return pd.DataFrame() # Return empty DataFrame if file doesn't exist - - df = pd.read_csv(csv) - if df.empty: - print(ct.colorText("Error: CSV file has headers but no data rows.", "red")) - else: - print(ct.colorText(f"Data loaded successfully from {csv}", "green")) - except pd.errors.EmptyDataError: - print(ct.colorText("Notice: CSV file is completely empty (no headers, no data), falling back to empty frame", "white")) - df = pd.DataFrame() # Create an empty DataFrame as fallback - return df - - -def tryToReadParquet(parquet): - try: - df = pd.read_parquet(parquet) - if df.empty: - print(ct.colorText("Error: Parquet file has headers but no data rows.", "red")) - else: - print(ct.colorText(f"Data loaded successfully from {parquet}", "green")) - except pd.errors.EmptyDataError: - print(ct.colorText("Notice : Parquet file is completely empty (no headers, no data), falling back to empty frame", "white")) - df = pd.DataFrame() # Create an empty DataFrame as fallback - return df - -def deduplicate_list(lst): - seen = set() - return [x for x in lst if not (x in seen or seen.add(x))] + ct.apivalidation() + menu_main() def menu_main(): - while True: - ct.displayIntro(); - print(ct.colorText("1. šŸ–„ļø - Get All Events for Single Device", "yellow")) - print(ct.colorText("2. šŸŽ« - OTP", "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("6 šŸ” - Device Search", "yellow")) - print(ct.colorText("Q. šŸ”š - Quit", "yellow")) + while True: + ct.displayIntro(); + print(ct.colorText("1. šŸ–„ļø - Get All Events for Single Device", "yellow")) + print(ct.colorText("2. šŸŽ« - OTP", "yellow")) + print(ct.colorText("3. ā±ļø - Placeholder", "yellow")) + print(ct.colorText("4. šŸ”’ - Prepare Policy For Enforcement", "yellow")) + print(ct.colorText("5. šŸ”„ - Update Audit Policies from Enforcement Policies", "yellow")) + print(ct.colorText("6 šŸ” - Device Search", "yellow")) + print(ct.colorText("Q. šŸ”š - Quit", "yellow")) - choice = input(ct.colorText("\nEnter Menu Item: ", "white")) - if choice == '1': - utils.clientfunctions.devicehistory(url,False) - elif choice == "2": - menu_otp() - elif choice == "3": - find_unexcluded() - elif choice == "4": - menu_prepare_to_enforce() - elif choice == "5": - utils.policyfunctions.updateAuditPoliciesFromEnforcementPolices(url, policy_relationship_map) - elif choice == "6": - device_input_str = utils.clientfunctions.promptForDevices() - utils.clientfunctions.findAgents(url, device_input_str) - elif choice == "Q": - break - else: - print(ct.colorText("Invalid choice. Please try again.","red")) + choice = input(ct.colorText("\nEnter Menu Item: ", "white")) + if choice == '1': + utils.clientfunctions.devicehistory(url,False) + elif choice == "2": + menu_otp() + elif choice == "3": + pass + elif choice == "4": + utils.policyfunctions.prepare_to_enforce(url, bad_publisher_list, pups, badpathparts, threat_tolerance_constant, path_exclusion_constant, min_files_for_path) + elif choice == "5": + utils.policyfunctions.updateAuditPoliciesFromEnforcementPolices(url, policy_relationship_map) + elif choice == "6": + devicelist = utils.clientfunctions.promptForDevices() + utils.clientfunctions.findAgents(url, devicelist, False) + + elif choice == "7": + devicelist = utils.clientfunctions.promptForDevices() + device_df = utils.clientfunctions.findAgents(url, devicelist, True) + utils.clientfunctions.returnToEnforcement(url, device_df, policy_relationship_map, bad_publisher_list, pups, badpathparts, threat_tolerance_constant, path_exclusion_constant, min_files_for_path) + elif choice == "Q": + break + else: + print(ct.colorText("Invalid choice. Please try again.","red")) def menu_otp(): - while True: - print(ct.colorText("\n--- šŸŽ« OTP Submenu šŸŽ« ---","cyan")) - print(ct.colorText("1. Generate OTP","cyan")) - #print(ct.colorText("2. Sub-option B","cyan")) - print(ct.colorText("Q. Return to Main Menu","cyan")) - choice = input("Enter your choice: ") + while True: + print(ct.colorText("\n--- šŸŽ« OTP Submenu šŸŽ« ---","cyan")) + print(ct.colorText("1. Generate OTP","cyan")) + #print(ct.colorText("2. Sub-option B","cyan")) + print(ct.colorText("Q. Return to Main Menu","cyan")) + choice = input("Enter your choice: ") - if choice == "1": - utils.otpfunctions.generateOTP(url, utils.clientfunctions.findAgentID(url)) - break + if choice == "1": + utils.otpfunctions.generateOTP(url, utils.clientfunctions.findAgentID(url)) + break - elif choice == "2": - print("You selected Sub-option B") - elif choice == "Q": - print("Returning to Main Menu...") - break - else: - print("Invalid choice. Please try again.") + elif choice == "2": + print("You selected Sub-option B") + elif choice == "Q": + print("Returning to Main Menu...") + break + else: + print("Invalid choice. Please try again.") -def find_unexcluded(): - while True: - print("\n--- Follow Steps Sequentially, Files will land in directory named 'exclusions'") - print("1. Pull last 24-72 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": - - # Get the AT policies dictionary - atpolicies = utils.policyfunctions.listATPolicies(url) - - # List to hold each policy's DataFrame - all_dfs = [] - - while True: - try: - history_days = int(input("Enter the how many days in of history do you want to pull - select a number between 1 and 150: ")) - if 1 <= history_days <= 3: - break - else: - print("Invalid input. Please enter a number between 1 and 3.") - except ValueError: - print("Invalid input. Please enter a valid integer.") - - - - - # Loop through each policy name - for policy_name in atpolicies: - try: - # Get the policy info DataFrame - df = utils.policyfunctions.getPolicyInfo(url, policy_name, [1, 2, 6, 7], history_days, False) - - # Add a column to indicate the policy name - df['PolicyName'] = policy_name - - # Append to the list - all_dfs.append(df) - except Exception as e: - print(f"Error processing policy '{policy_name}': {e}") - - # Combine all DataFrames into one - if all_dfs: - combined_df = pd.concat(all_dfs, ignore_index=True) - print("āœ… Combined DataFrame created.") - else: - combined_df = pd.DataFrame() - print("āš ļø No data was retrieved.") - - combined_df.to_csv("exclusions\\allATPolicyExecs.csv", index=False) - - - elif choice == "2": - utils.pathfunctions.allATpaths(url,pups, bad_publisher_list, badpathparts, threat_tolerance_constant, path_exclusion_constant, min_files_for_path) - - elif choice == "3": - utils.pathfunctions.mergeTesting() - - elif choice == "4": - - - # Get the AT policies dictionary - atpolicies = utils.policyfunctions.listATPolicies(url) - - # List to hold each policy's path data - all_paths = [] - - # Loop through each policy name and group ID - for policy_name, group_id in atpolicies.items(): - try: - # Get the list of paths using the group ID - paths = utils.pathfunctions.listPaths(url, group_id) - - # Ensure each path is a string and append properly - for path in paths: - if isinstance(path, str): - all_paths.append({ - 'PolicyName': policy_name, - 'GroupID': group_id, - 'Path': path - }) - except Exception as e: - print(f"Error retrieving paths for policy '{policy_name}': {e}") - - # Convert to DataFrame - if all_paths: - paths_df = pd.DataFrame(all_paths) - print(paths_df.head()) # Optional: preview first few rows - print("āœ… Paths DataFrame created.") - else: - paths_df = pd.DataFrame() - print("āš ļø No paths were retrieved.") - - # Save to CSV - paths_df.to_csv("exclusions\\paths.csv", index=False) - elif choice == "5": - - # Load and clean data - 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() - filenames_df['PolicyName'] = filenames_df['PolicyName'].fillna('').astype(str).str.strip() - exclusions_df['Path'] = exclusions_df['Path'].fillna('').astype(str).str.strip() - exclusions_df['PolicyName'] = exclusions_df['PolicyName'].fillna('').astype(str).str.strip() - - # Decode escaped backslashes in exclusion patterns - exclusions_df['Path'] = exclusions_df['Path'].apply(lambda p: p.encode('utf-8').decode('unicode_escape')) - - # Build exclusion map: {PolicyName: [compiled regex patterns]} - exclusion_map = {} - for _, row in exclusions_df.iterrows(): - policy = row['PolicyName'] - raw_pattern = row['Path'] - try: - regex = utils.pathfunctions.wildcardRegex(raw_pattern) - print(f"[EXCLUSION MAP] Policy: {policy}, Pattern: {raw_pattern} ā†’ Regex: {regex.pattern}") - exclusion_map.setdefault(policy, []).append(regex) - except Exception as e: - print(f"[ERROR] Failed to compile pattern for Policy: {policy}, Path: {raw_pattern}, Error: {e}") - - # Function to check if a filename matches any exclusion pattern for its PolicyName - def is_excluded(row): - policy = row['PolicyName'].strip() - path = os.path.normpath(row['filename'].strip()) - patterns = exclusion_map.get(policy, []) - for r in patterns: - if r.match(path): - print(f"[MATCH] {policy}: {path} matches {r.pattern}") - return True - print(f"[NO MATCH] {policy}: {path}") - return False - - # Apply matching - excluded = filenames_df[filenames_df.apply(is_excluded, axis=1)] - not_excluded = filenames_df[~filenames_df.apply(is_excluded, axis=1)] - - - # Desired column order - column_order = [ - 'PolicyName', 'filename', 'longestcfp', - 'pprocess', 'gprocess', 'sha256', 'publisher', 'description', 'productname','commandline', 'middle', 'filename_only', - 'file_extension', 'unique_sha256_count', 'hostname', 'username', 'productversion', - 'reputation_lastseen', 'reputation_scannermatch', 'reputation_scannercount', - 'reputation_status', 'reputation_threatlevel', 'reputation_threatname', 'reputation_timestamp' - ] - - # Reorder columns (ignore missing ones) - excluded = excluded[[col for col in column_order if col in excluded.columns]] - not_excluded = not_excluded[[col for col in column_order if col in not_excluded.columns]] - - - # Save results - excluded.to_csv('exclusions\\excluded_filenames.csv', index=False) - not_excluded.to_csv('exclusions\\non_excluded_filenames.csv', index=False) - - - elif choice == "Q": - break - else: - print("Invalid choice. Please try again.") - -def menu_prepare_to_enforce(): - - first_policy = " " - second_policy = " " - destination_name = " " - destination_id = " " - allowlist_parent_name = " " - allowlist_parent_id = " " - allowlist_child_name = " " - allowlist_child_id = " " - history_days = 1 - - #If the directorys where we're going to store our output dont exist, make them. - if not os.path.exists("parquet"): os.makedirs("parquet") - if not os.path.exists("needs_approved"): os.makedirs("needs_approved") - if not os.path.exists("approved"): os.makedirs("approved") - if not os.path.exists("preflight"): os.makedirs("preflight") - - while True: - - ct.printEnforceChecklist(first_policy, second_policy, allowlist_child_name, allowlist_parent_name, destination_name) - - choice = input(ct.colorText("\nEnter your choice: ", "white")) - - if choice == "1": - - choice, policynames, policyid = utils.policyfunctions.listPolicies(url) - first_policy = policynames[choice] - while True: - answer = input(ct.colorText(f"{"Do you want to load a second policy?"} (yes/no): ", "white").strip().lower()) - if answer in ("yes", "y"): - choice, policynames, policyid = utils.policyfunctions.listPolicies(url) - second_policy = policynames[choice] - - break - elif answer in ("no", "n"): - second_policy = first_policy - break - else: - print(ct.colorText("Please answer with 'yes' or 'no'.", "red")) - - elif choice == "2": - - - while True: - try: - history_days = int(input("Enter the how many days in of history do you want to pull - select a number between 1 and 150: ")) - if 1 <= history_days <= 150: - break - else: - print("Invalid input. Please enter a number between 1 and 150.") - except ValueError: - print("Invalid input. Please enter a valid integer.") - - - if not os.path.exists(f"parquet\\execution_history_{first_policy}.parquet"): - utils.policyfunctions.getPolicyInfo(url, first_policy, [1,2,6,7], history_days) - - if not os.path.exists(f"parquet\\execution_history_{second_policy}.parquet"): - utils.policyfunctions.getPolicyInfo(url, second_policy, [1,2,6,7], history_days) - - if not os.path.exists(f"parquet\\combined_hashlist_{first_policy}_{second_policy}.parquet"): - utils.hashfunctions.combineHashes(url, first_policy, second_policy) - - if not os.path.exists(f"parquet\\hashes_rep_unknown_{first_policy}_{second_policy}.parquet") and not os.path.exists(f"parquet\\hashes_rep_good_{first_policy}_{second_policy}.parquet") and not os.path.exists(f"parquet\\hashes_rep_bad_{first_policy}_{second_policy}.parquet"): - utils.hashfunctions.categorizeHashes( - first_policy, - second_policy, - pd.read_parquet(f"parquet\\combined_hashlist_{first_policy}_{second_policy}.parquet"), - threat_tolerance_constant, - bad_publisher_list, - pups - ) - - if not os.path.exists(f"parquet\\condensed_executions_{first_policy}_{second_policy}.parquet"): - utils.hashfunctions.condenseExecutions(first_policy,second_policy) - - if ( - os.path.exists(f"parquet\\hashes_rep_unknown_{first_policy}_{second_policy}.parquet") and - os.path.exists(f"parquet\\hashes_rep_good_{first_policy}_{second_policy}.parquet") and - os.path.exists(f"parquet\\hashes_rep_bad_{first_policy}_{second_policy}.parquet") and - not os.path.exists(f"needs_approved\\hashes_rep_unknown_{first_policy}_{second_policy}.csv") - ):utils.hashfunctions.divideSortedHashExecutions(first_policy,second_policy,pups) - - elif choice == "3": - - if os.path.exists(f"approved\\hashes_rep_unknown_{first_policy}_{second_policy}.csv") and os.path.exists(f"approved\\hashes_rep_good_{first_policy}_{second_policy}.csv"): - utils.pathfunctions.generatePathReview(first_policy, second_policy, badpathparts,path_exclusion_constant, min_files_for_path) - utils.hashfunctions.generatePublist(first_policy,second_policy,bad_publisher_list) - else: - print(ct.colorText(f"Please manually approve hashes prior to this step","red")) - - elif choice == "4": - - if os.path.exists(f"approved\\path_needs_approved_{first_policy}_{second_policy}.csv"): - if not os.path.exists(f"parquet\\final_hash_approvals_{first_policy}_{second_policy}.parquet") and not os.path.exists(f"parquet\\final_path_exclusions_{first_policy}_{second_policy}.parquet"): - utils.hashfunctions.generatePreflights(first_policy, second_policy) - - elif choice == "5": - - print(ct.colorText(f"Please choose destination_name Policy for Path Exclusions","white")) - choice, policynames, policyid = utils.policyfunctions.listPolicies(url) - #print(allowlist_parent_tuple) - destination_name = policynames[choice] - destination_id = policyid[choice] - - print(ct.colorText(f"Please choose Parent Allowlist for Known Hashes","white")) - choice, allowlists,allowid = utils.policyfunctions.listAllowlists(url) - #print(allowlist_parent_tuple) - allowlist_parent_name = allowlists[choice] - allowlist_parent_id = allowid[choice] - - print(ct.colorText(f"Please choose Child Allowlist for Less-Known Hashes","white")) - choice, allowlists, allowid = utils.policyfunctions.listAllowlists(url) - #print(allowlist_child_tuple) - allowlist_child_name = allowlists[choice] - allowlist_child_id = allowid[choice] - - - elif choice == "6": - if os.path.exists(f"preflight\\final_path_exclusions_{first_policy}_{second_policy}.html") and os.path.exists(f"preflight\\final_hash_approvals_{first_policy}_{second_policy}.html") and allowlist_parent_name != " " and allowlist_child_name != " " and destination_name != " ": - utils.policyfunctions.sendToPolicyTest( - url, - first_policy, - second_policy, - destination_name, - destination_id, - allowlist_parent_name, - allowlist_parent_id, - allowlist_child_name, - allowlist_child_id - ) - - elif choice == "7": - if os.path.exists(f"preflight\\final_path_exclusions_{first_policy}_{second_policy}.html") and os.path.exists(f"preflight\\final_hash_approvals_{first_policy}_{second_policy}.html") and allowlist_parent_name != " " and allowlist_child_name != " " and destination_name != " ": - utils.policyfunctions.sendToPolicy( - url, - first_policy, - second_policy, - destination_name, - destination_id, - allowlist_parent_name, - allowlist_parent_id, - allowlist_child_name, - allowlist_child_id - ) - elif choice == "R": - - utils.pathfunctions.clean_folders_enforcement_prep() - - elif choice == "Q": - break - else: - print(ct.colorText("Invalid choice. Please try again.", "red")) - + if __name__ == "__main__": main() \ No newline at end of file diff --git a/utils/clientfunctions.py b/utils/clientfunctions.py index 6a1f1fc..bdee8b8 100644 --- a/utils/clientfunctions.py +++ b/utils/clientfunctions.py @@ -14,7 +14,10 @@ # along with this program. If not, see . #Local Imports -import utils.pretty as ct +import utils.hashfunctions as hashf +import utils.utils as ct +import utils.pathfunctions as pathf +import utils.policyfunctions as policyf #Standard Libary Imports: import datetime @@ -27,7 +30,7 @@ import pandas as pd import requests def findAgentID(url): - + print(ct.colorText("WARNING: Device Name is Case Sensitive", "red")) hostname = input(ct.colorText("Enter Device Name: ", "white")) @@ -130,66 +133,67 @@ def getPolicyName(url, groupid): def devicehistory(url, outputjson: bool): - endpoint = url + '/v1/getexechistory' - print("\n") - print(ct.colorText("1. Today", "yellow")) - print(ct.colorText("2. Last 24 Hours", "yellow")) - print(ct.colorText("3. Past 7 Days", "yellow")) - print(ct.colorText("4. Past 30 Days", "yellow")) - print(ct.colorText("5. Custom Date Range","yellow")) - choice = input(ct.colorText("\nSelect Date Range: ", "white")) - today = datetime.date.today() - today = today.strftime("%Y-%m-%d") - if choice == '1': - date_selected = today - elif choice == '2': - date_selected = datetime.date.today() - datetime.timedelta(days=1) - date_selected = date_selected.strftime('%Y-%m-%d') - elif choice == '3': - date_selected = datetime.date.today() - datetime.timedelta(days=7) - date_selected = date_selected.strftime('%Y-%m-%d') - elif choice == '4': - date_selected = datetime.date.today() - datetime.timedelta(days=30) - date_selected = date_selected.strftime('%Y-%m-%d') - elif choice == "5": - print(ct.colorText("Please Input Dates as YYYY-MM-DD", "cyan")) - date_selected = input(ct.colorText("From: ", "white")) - today = input(ct.colorText("Date To: ", "white")) - print(ct.colorText("WARNING: Device Name is Case Sensitive", "red")) - device = input(ct.colorText("Enter Device Name: ", "white")) - payload_dict = { - "datefrom": date_selected, - "dateto": today, - "hostname": device - } - payload = json.dumps(payload_dict) - print(ct.colorText(payload, "green")) - headers = { - "X-APIKey": os.getenv('APIKEY') - } + endpoint = url + '/v1/getexechistory' + print("\n") + print(ct.colorText("1. Today", "yellow")) + print(ct.colorText("2. Last 24 Hours", "yellow")) + print(ct.colorText("3. Past 7 Days", "yellow")) + print(ct.colorText("4. Past 30 Days", "yellow")) + print(ct.colorText("5. Custom Date Range","yellow")) + choice = input(ct.colorText("\nSelect Date Range: ", "white")) + today = datetime.date.today() + today = today.strftime("%Y-%m-%d") + date_selected = " " + if choice == '1': + date_selected = today + elif choice == '2': + date_selected = datetime.date.today() - datetime.timedelta(days=1) + date_selected = date_selected.strftime('%Y-%m-%d') + elif choice == '3': + date_selected = datetime.date.today() - datetime.timedelta(days=7) + date_selected = date_selected.strftime('%Y-%m-%d') + elif choice == '4': + date_selected = datetime.date.today() - datetime.timedelta(days=30) + date_selected = date_selected.strftime('%Y-%m-%d') + elif choice == "5": + print(ct.colorText("Please Input Dates as YYYY-MM-DD", "cyan")) + date_selected = input(ct.colorText("From: ", "white")) + today = input(ct.colorText("Date To: ", "white")) + print(ct.colorText("WARNING: Device Name is Case Sensitive", "red")) + device = input(ct.colorText("Enter Device Name: ", "white")) + payload_dict = { + "datefrom": date_selected, + "dateto": today, + "hostname": device + } + payload = json.dumps(payload_dict) + print(ct.colorText(payload, "green")) + headers = { + "X-APIKey": os.getenv('APIKEY') + } - response = requests.request("POST", endpoint, headers=headers, data=payload, verify=False) + response = requests.request("POST", endpoint, headers=headers, data=payload, verify=False) - if outputjson: - return response + if outputjson: + return response - parse_text = json.loads(response.text) + parse_text = json.loads(response.text) - # Safely get exechistory - exechistory = parse_text.get('response', {}).get('exechistory') + # Safely get exechistory + exechistory = parse_text.get('response', {}).get('exechistory') - if isinstance(exechistory, list): - for block in exechistory: - print(ct.colorText(f"Command: {block.get('commandline', 'N/A')}", "green")) - print(ct.colorText(f"Date: {block.get('datetime', 'N/A')}", "green")) - print(ct.colorText(f"Filename: {block.get('filename', 'N/A')}", "green")) - print(ct.colorText(f"Policy Name: {block.get('policyname', 'N/A')}", "green")) - print(ct.colorText(f"Hostname: {block.get('hostname', 'N/A')}", "green")) - print(ct.colorText(f"Hash: {block.get('sha256', 'N/A')}", "green")) - print("\n") - else: - print(ct.colorText("No execution history found or data is not in expected format.", "red")) - + if isinstance(exechistory, list): + for block in exechistory: + print(ct.colorText(f"Command: {block.get('commandline', 'N/A')}", "green")) + print(ct.colorText(f"Date: {block.get('datetime', 'N/A')}", "green")) + print(ct.colorText(f"Filename: {block.get('filename', 'N/A')}", "green")) + print(ct.colorText(f"Policy Name: {block.get('policyname', 'N/A')}", "green")) + print(ct.colorText(f"Hostname: {block.get('hostname', 'N/A')}", "green")) + print(ct.colorText(f"Hash: {block.get('sha256', 'N/A')}", "green")) + print("\n") + else: + print(ct.colorText("No execution history found or data is not in expected format.", "red")) + def findAllAgents(url): endpoint = url + '/v1/agent/find' payload = {} @@ -218,7 +222,7 @@ def findAllAgents(url): data['status'] = data['status'].map(status_map) return(data) -def findAgents(url, device_input_str): +def findAgents(url, device_input_str, return_dataframe): os.makedirs("device_search", exist_ok=True) df = findAllAgents(url) @@ -240,7 +244,8 @@ def findAgents(url, device_input_str): matched_df.to_csv(f"device_search\\{filename}", index=False) print(ct.colorText(f"\nāœ… Matched devices exported to: device_search\\{filename}","green")) - + if return_dataframe : return matched_df + def promptForDevices(): print(ct.colorText("šŸ” Device Search", "cyan")) @@ -266,3 +271,82 @@ def promptForDevices(): return device_input_str + +def returnToEnforcement(url, device_df, policy_relationship_map, bad_publisher_list, pups, badpathparts, threat_tolerance_constant, path_exclusion_constant, min_files_for_path): + policylist = sorted(device_df['policy_name'].unique().tolist()) + type = [1, 2, 6, 7] + + parq_base_dir = "prepare_policy\\parquet\\" + appr_base_dir = "prepare_policy\\approved\\" + needappr_base_dir = "prepare_policy\\needs_approved" + pflight_base_dir = "prepare_policy\\preflight" + + #If the directorys where we're going to store our output dont exist, make them. + os.makedirs(parq_base_dir, exist_ok=True) + os.makedirs(needappr_base_dir, exist_ok=True) + os.makedirs(appr_base_dir, exist_ok=True) + os.makedirs(pflight_base_dir, exist_ok=True) + + while True: + + #ct.printDeviceEnforceChecklist() + + choice = input(ct.colorText("\nEnter your choice: ", "white")) + + if choice == "1": + + policyf.buildExecHistory(url, + policylist, + parq_base_dir, + needappr_base_dir, + type, + threat_tolerance_constant, + bad_publisher_list, + pups, + ) + + elif choice == "2": + + pathf.generateApprovables(needappr_base_dir, appr_base_dir, parq_base_dir, badpathparts, bad_publisher_list, path_exclusion_constant, min_files_for_path, True) + + elif choice == "3": + policyf.savePreFlights(appr_base_dir, parq_base_dir, pflight_base_dir) + + """ + + elif choice == "4": + if os.path.exists(f"preflight\\final_path_exclusions.html") and os.path.exists(f"preflight\\final_hash_approvals.html") and allowlist_parent_name != " " and allowlist_child_name != " " and destination_name != " ": + sendToPolicyTest( + url, + first_policy, + second_policy, + destination_name, + destination_id, + allowlist_parent_name, + allowlist_parent_id, + allowlist_child_name, + allowlist_child_id + ) + + elif choice == "5": + if os.path.exists(f"preflight\\final_path_exclusions.html") and os.path.exists(f"preflight\\final_hash_approvals.html") and allowlist_parent_name != " " and allowlist_child_name != " " and destination_name != " ": + sendToPolicy( + url, + first_policy, + second_policy, + destination_name, + destination_id, + allowlist_parent_name, + allowlist_parent_id, + allowlist_child_name, + allowlist_child_id + ) + elif choice == "R": + + pathf.clean_folders(enforcement_prep) + + elif choice == "Q": + break + else: + print(ct.colorText("Invalid choice. Please try again.", "red")) + """ \ No newline at end of file diff --git a/utils/hashfunctions.py b/utils/hashfunctions.py index 0a69c21..f1097fa 100644 --- a/utils/hashfunctions.py +++ b/utils/hashfunctions.py @@ -14,10 +14,8 @@ # along with this program. If not, see . #Local Imports -import utils.hashfunctions as hashf import utils.pathfunctions as pathf -import utils.pretty as ct -from AirlockTools import tryToReadCSV +import utils.utils as ct #Standard Libary Imports: import gc @@ -99,7 +97,7 @@ def augmentAggregatedHashes(url, agg_df: pd.DataFrame) -> pd.DataFrame: df = agg_df.merge(df_api, on="sha256", how="left") # Only include columns that exist to avoid KeyErrors - expected_columns = ['sha256', 'filename_x', 'description', 'productname', 'productversion', + expected_columns = ['policy','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', @@ -110,7 +108,7 @@ def augmentAggregatedHashes(url, agg_df: pd.DataFrame) -> pd.DataFrame: return aug_df -def categorizeHashes(first_policy, second_policy, df: pd.DataFrame, threat_tolerance: int, untrusted_publishers, pups: list): +def categorizeHashes(df: pd.DataFrame, threat_tolerance: int, untrusted_publishers, pups: list) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]: if untrusted_publishers is None: untrusted_publishers = [] if pups is None: pups = [] @@ -149,122 +147,56 @@ def categorizeHashes(first_policy, second_policy, df: pd.DataFrame, threat_toler needsreview_df = df[mask_needsreview] approved_df = df[mask_approved] unapproved_df = df[~(mask_needsreview | mask_approved)] + return needsreview_df, approved_df, unapproved_df - needsreview_df.to_parquet(f"parquet\\hashes_rep_unknown_{first_policy}_{second_policy}.parquet", index=False) - approved_df.to_parquet(f"parquet\\hashes_rep_good_{first_policy}_{second_policy}.parquet", index=False) - unapproved_df.to_parquet(f"parquet\\hashes_rep_bad_{first_policy}_{second_policy}.parquet", index=False) - - del needsreview_df - del approved_df - del unapproved_df - gc.collect() - -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) - - # 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 combineHashAndHist(path, first_policy, second_policy): - - condensed_combo = pd.read_parquet(f"parquet\\condensed_executions_{first_policy}_{second_policy}.parquet") - df = pd.read_parquet(path) - - #Pull hash info for the entries in the needs approval table +def combineHashAndHist(hash_path, condensed_path): + # Load both datasets + condensed_combo = pd.read_parquet(condensed_path) + df = pd.read_parquet(hash_path) + # Merge on sha256 df = pd.merge(condensed_combo, df, on='sha256', how='inner') - - #Rename Publisher, Keep and reorder columns we want + df.to_csv("testing4.csv") + # Rename and reorder columns df = df.rename(columns={'publisher_x': 'publisher'}) - df = df[['sha256', 'publisher', 'description', 'filename', 'hostname', 'username', 'productname', 'productversion','reputation_lastseen', 'reputation_scannermatch', 'reputation_scannercount','reputation_status', 'reputation_threatlevel', 'reputation_threatname','reputation_timestamp', 'pprocess', 'gprocess', 'commandline']] + df = df.rename(columns={'policy_x': 'policy'}) + df = df[['policy','sha256', 'publisher', 'description', 'filename', 'hostname', 'username', + 'productname', 'productversion', 'reputation_lastseen', 'reputation_scannermatch', + 'reputation_scannercount', 'reputation_status', 'reputation_threatlevel', + 'reputation_threatname', 'reputation_timestamp', 'pprocess', 'gprocess', 'commandline']] df = df.sort_values(by='filename') - df.to_parquet(path, index=False) + # Overwrite the original hash file + df.to_parquet(hash_path, index=False) + + # Cleanup del df del condensed_combo gc.collect() -def combineHashes(url, first_policy, second_policy): - combined_hashes = pd.DataFrame(columns=['sha256', 'publisher']) +def combineHashes(url, parquet_files) -> pd.DataFrame: + combined_hashes = pd.DataFrame() hashes = [] - try: - hash1 = pd.read_parquet(f"parquet\\execution_history_{first_policy}.parquet", columns=['sha256', 'publisher']) - pathf.inspect_parquet(f"parquet\\execution_history_{first_policy}.parquet") - if not hash1.empty: - hashes.append(hash1) - else: - print("āš ļø First dataframe is empty.") - except Exception as e: - print(f"āŒ Error reading first Parquet file: {e}") - try: - hash2 = pd.read_parquet(f"parquet\\execution_history_{second_policy}.parquet", columns=['sha256', 'publisher']) - pathf.inspect_parquet(f"parquet\\execution_history_{second_policy}.parquet") - if not hash2.empty: - hashes.append(hash2) - else: - print("āš ļø Second dataframe is empty.") - except Exception as e: - print(f"āŒ Error reading second Parquet file: {e}") + for file_path in parquet_files: + try: + hash_df = pd.read_parquet(file_path) + pathf.inspect_parquet(file_path) + + if not hash_df.empty: + hashes.append(hash_df) + else: + print(f"āš ļø Dataframe is empty: {file_path}") + except Exception as e: + print(f"āŒ Error reading Parquet file '{file_path}': {e}") if hashes: combined_hashes = pd.concat(hashes, ignore_index=True) - print(f"āœ… Combined {len(combined_hashes)} hashes.") + print(f"āœ… Combined {len(combined_hashes)} hashes from {len(hashes)} files.") else: print("āš ļø No valid dataframes to combine.") combined_hashes = combined_hashes.drop_duplicates(subset=['sha256']) - augmented_combo = hashf.augmentAggregatedHashes(url, combined_hashes) + augmented_combo = augmentAggregatedHashes(url, combined_hashes) numeric_reputation_cols = [ 'reputation_scannermatch', @@ -282,62 +214,45 @@ def combineHashes(url, first_policy, second_policy): 'reputation_status', 'reputation_threatlevel', 'reputation_threatname', 'reputation_timestamp']] augmented_combo = augmented_combo.sort_values(by=['publisher', 'description', 'productname']) - augmented_combo.to_parquet(f"parquet\\combined_hashlist_{first_policy}_{second_policy}.parquet", index=False) - del combined_hashes - del augmented_combo gc.collect() print(ct.colorText("Hash reputation info added to dataframe", "green")) + return augmented_combo -def condenseExecutions(first_policy,second_policy): - exe1 = pd.DataFrame() - exe2 = pd.DataFrame() - condensed_combo = pd.DataFrame() - - try: - exe1 = pd.read_parquet(f"parquet\\execution_history_{first_policy}.parquet") - pathf.inspect_parquet(f"parquet\\execution_history_{first_policy}.parquet") - if not exe1.empty: - print() - else: - print("āš ļø First dataframe is empty.") - except Exception as e: - print(f"āŒ Error reading first Parquet file: {e}") +def condenseExecutions(parquet_paths): + combined_df = pd.DataFrame() + valid_files = [] - try: - exe2 = pd.read_parquet(f"parquet\\execution_history_{second_policy}.parquet") - pathf.inspect_parquet(f"parquet\\execution_history_{second_policy}.parquet") - if not exe2.empty: - print() - else: - print("āš ļø Second dataframe is empty.") - except Exception as e: - print(f"āŒ Error reading second Parquet file: {e}") + for file_path in parquet_paths: + try: + df = pd.read_parquet(file_path) + # Optional: pathf.inspect_parquet(file_path) + if not df.empty: + combined_df = pd.concat([combined_df, df], ignore_index=True) + valid_files.append(file_path) + print(f"āœ… Loaded {len(df)} rows from {file_path}") + else: + print(f"āš ļø DataFrame from '{file_path}' is empty.") + except Exception as e: + print(f"āŒ Error reading Parquet file '{file_path}': {e}") - if not exe1.empty and not exe2.empty: - condensed_combo = pd.concat([exe1, exe2], ignore_index=True) - - print(f"āœ… Combined {len(condensed_combo)} hashes.") - elif exe1.empty: - condensed_combo = exe2 - elif exe2.empty: - condensed_combo = exe1 + if not combined_df.empty: + print(f"āœ… Combined {len(combined_df)} rows from {len(valid_files)} files.") else: print("āš ļø No valid dataframes to combine.") - condensed_combo.to_parquet(f"parquet\\condensed_executions_{first_policy}_{second_policy}.parquet", index=False) - del condensed_combo - gc.collect() - -def divideSortedHashExecutions(first_policy,second_policy, pups): + return combined_df - combineHashAndHist(f"parquet\\hashes_rep_unknown_{first_policy}_{second_policy}.parquet", first_policy, second_policy) - combineHashAndHist(f"parquet\\hashes_rep_good_{first_policy}_{second_policy}.parquet", first_policy, second_policy) - combineHashAndHist(f"parquet\\hashes_rep_bad_{first_policy}_{second_policy}.parquet", first_policy, second_policy) +def divideSortedHashExecutions(unknown_parq, good_parq, bad_parq, condensed_parq, pups) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]: + # Run combineHashAndHist on each file + combineHashAndHist(unknown_parq, condensed_parq) + combineHashAndHist(good_parq, condensed_parq) + combineHashAndHist(bad_parq, condensed_parq) - unknown = pd.read_parquet(f"parquet\\hashes_rep_unknown_{first_policy}_{second_policy}.parquet") - good = pd.read_parquet(f"parquet\\hashes_rep_good_{first_policy}_{second_policy}.parquet") - bad = pd.read_parquet(f"parquet\\hashes_rep_bad_{first_policy}_{second_policy}.parquet") + # Load data + unknown = pd.read_parquet(unknown_parq) + good = pd.read_parquet(good_parq) + bad = pd.read_parquet(bad_parq) # Build regex pattern once pattern = pathf.regulator(pups) @@ -353,52 +268,30 @@ def divideSortedHashExecutions(first_policy,second_policy, pups): unknown = unknown[~unknown["filename"].str.contains(pattern, na=False)] good = good[~good["filename"].str.contains(pattern, na=False)] - unknown.to_csv(f"needs_approved\\hashes_rep_unknown_{first_policy}_{second_policy}.csv",index=False) - good.to_csv(f"needs_approved\\hashes_rep_good_{first_policy}_{second_policy}.csv",index=False) - bad.to_csv(f"needs_approved\\hashes_rep_bad_{first_policy}_{second_policy}.csv",index=False) + return unknown, good, bad - ct.style_dataframe_dark(unknown, f"needs_approved\\hashes_rep_unknown_{first_policy}_{second_policy}.html") - ct.style_dataframe_dark(good, f"needs_approved\\hashes_rep_good_{first_policy}_{second_policy}.html") - ct.style_dataframe_dark(bad, f"needs_approved\\hashes_rep_bad_{first_policy}_{second_policy}.html") - -def generatePreflights(first_policy, second_policy): - allhashes = pd.read_parquet(f"parquet\\all_approved_hashes_{first_policy}_{second_policy}.parquet") - - primarypathexclusions = tryToReadCSV(f"approved\\path_needs_approved_{first_policy}_{second_policy}.csv") - secondarypathexclusions = tryToReadCSV(f"approved\\secondary_paths_{first_policy}_{second_policy}.csv") +def generatePreflights(hashes, primary_path, secondary_path): + all_hashes = pd.read_parquet(hashes) + + primarypathexclusions = ct.tryToReadCSV(primary_path) + secondarypathexclusions = ct.tryToReadCSV(secondary_path) pathexclusions = pd.concat([primarypathexclusions, secondarypathexclusions], ignore_index=True) - - publishers = tryToReadCSV(f"approved\\publishers_{first_policy}_{second_policy}.csv") - publishers.to_parquet(f"parquet\\publishers_{first_policy}_{second_policy}.parquet", index=False) - allowbyhash = allhashes[~allhashes['sha256'].isin(pathexclusions['sha256'])] - - allowbyhash.to_parquet(f"parquet\\final_hash_approvals_{first_policy}_{second_policy}.parquet", index=False) + allowbyhash = all_hashes[~all_hashes['sha256'].isin(pathexclusions['sha256'])] allowbyhash.sort_values(by=["filename"]) - ct.style_dataframe_dark(allowbyhash, f"preflight\\final_hash_approvals_{first_policy}_{second_policy}.html") - ct.style_dataframe_dark(pathexclusions, f"preflight\\final_path_exclusions_{first_policy}_{second_policy}.html") - ct.style_dataframe_dark(publishers, f"preflight\\publishers_{first_policy}_{second_policy}.html") + return pathexclusions, allowbyhash - del allowbyhash - del pathexclusions - gc.collect() - -def generatePublist(first_policy,second_policy,bad_publisher_list): - - try: - publist = pd.read_parquet(f"parquet\\all_approved_hashes_{first_policy}_{second_policy}.parquet", columns=['publisher']) - except Exception as e: - print(f"Error reading parquet file: {e}") - publist = pd.DataFrame() +def generatePublist(all_hashes, bad_publisher_list): + all_approved_hashes = ct.tryToReadParquet(all_hashes) #Drop all not signed, only keep unique values - publist = publist[publist['publisher'] != "Not Signed"].drop_duplicates(subset='publisher') + publist = all_approved_hashes[all_approved_hashes['publisher'] != "Not Signed"].drop_duplicates(subset='publisher') #Remove Bad publisher if somehow they made it this far pattern = pathf.regulator(bad_publisher_list) publist = publist[~publist["publisher"].str.contains(pattern, na=False)] - publist.to_csv(f"needs_approved\\publishers_{first_policy}_{second_policy}.csv", index=False) + return publist \ No newline at end of file diff --git a/utils/otpfunctions.py b/utils/otpfunctions.py index 31932f1..47bcd68 100644 --- a/utils/otpfunctions.py +++ b/utils/otpfunctions.py @@ -17,7 +17,7 @@ import utils.clientfunctions as clientf import utils.pathfunctions as pathf import utils.policyfunctions as policyf -import utils.pretty as ct +import utils.utils as ct from utils.perstscheduler import register_function, run_once_job, recurring_job, reload_jobs, start_scheduler, find_and_prioritize_jobs_by_pid #Standard Libary Imports: diff --git a/utils/pathfunctions.py b/utils/pathfunctions.py index 9c5e4cd..7082edb 100644 --- a/utils/pathfunctions.py +++ b/utils/pathfunctions.py @@ -16,8 +16,7 @@ #Local Imports import utils.hashfunctions as hashf import utils.pathfunctions as pathf -import utils.pretty as ct -from AirlockTools import tryToReadCSV +import utils.utils as ct #Standard Libary Imports: import ast @@ -77,51 +76,6 @@ def split_filepaths_grouped(df, col="filename", group_parts=4, min_parts=4): return pd.DataFrame(new_rows).drop(columns=["group_key"]) -def mask_from_csv(df, csv_path, filepath_col): - """ - Reads reviewed CSV of groups, keeps only files in approved groups. - """ - review_df = pd.read_csv(csv_path) - - def parse_paths(val): - if isinstance(val, str): - try: - # Try to parse as a list - parsed = ast.literal_eval(val) - # If it's not a list, wrap it - return parsed if isinstance(parsed, list) else [parsed] - except (ValueError, SyntaxError): - # If parsing fails, treat it as a single path - return [val] - return [val] - - review_df[filepath_col] = review_df[filepath_col].apply(parse_paths) - - # Flatten all approved file paths into a set for masking - approved_files = set() - for paths in review_df[filepath_col]: - approved_files.update(paths) - - # Keep only rows in df that are in approved_files - masked_df = df[df[filepath_col].isin(approved_files)].copy() - remainder = df[~df[filepath_col].isin(approved_files)].copy() - return remainder - -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 - def inspect_parquet(path): try: df = pd.read_parquet(path) @@ -133,7 +87,6 @@ def inspect_parquet(path): print(f"āŒ Error reading {path}: {e}") return pd.DataFrame() - def regulator(paths, case_insensitive=True): """ Build a regex pattern that matches any of the given Windows path fragments. @@ -145,269 +98,54 @@ def regulator(paths, case_insensitive=True): print(f"Regulator is providing: {pattern}") return pattern +def calculatePath(approved_hashes, badpathparts, path_exclusion_constant, min_files_for_path, split): + + if split : dfs_by_policy = [group for _, group in approved_hashes.groupby('policy')] + else : dfs_by_policy = [approved_hashes] -def generatePathReview(first_policy, second_policy, badpathparts, path_exclusion_constant, min_files_for_path): + processed_dfs = [] - if not os.path.exists(f"parquet\\all_approved_hashes_{first_policy}_{second_policy}.parquet"): + for df in dfs_by_policy: + haslcp = pathf.split_filepaths_grouped(df, "filename", path_exclusion_constant, min_files_for_path) + haslcp = haslcp.drop_duplicates() - df1 = tryToReadCSV(f"approved\\hashes_rep_unknown_{first_policy}_{second_policy}.csv") - df2 = tryToReadCSV(f"approved\\hashes_rep_good_{first_policy}_{second_policy}.csv") + forbidden = pathf.regulator(badpathparts, True) + forbidden_lcfp = haslcp["longestcfp"].str.contains(forbidden, na=False) + + print(ct.colorText("Removing forbidden filepaths for path exceptions", "green")) + lcp_not_forbidden = haslcp[~forbidden_lcfp].copy() + + lcp_not_forbidden_review = lcp_not_forbidden[['policy', 'longestcfp', 'middle', 'filename_only', 'file_extension', 'sha256']] + + unique_sha_counts = lcp_not_forbidden_review.groupby('longestcfp')['sha256'].nunique().reset_index() + unique_sha_counts.columns = ['longestcfp', 'unique_sha256_count'] + + lcp_not_forbidden_review = lcp_not_forbidden_review.merge(unique_sha_counts, on='longestcfp', how='left') + lcp_not_forbidden_review = lcp_not_forbidden_review[lcp_not_forbidden_review['unique_sha256_count'] >= min_files_for_path] + processed_dfs.append(lcp_not_forbidden_review) + + pathExclusions = pd.concat(processed_dfs, ignore_index=True) + + return pathExclusions + +def generatePathReview(unknown, good, badpathparts, path_exclusion_constant, min_files_for_path, split = False): + + df1 = ct.tryToReadCSV(unknown) + df2 = ct.tryToReadCSV(good) all_approved_hashes = pd.concat([df1 , df2], ignore_index=True).sort_values(by=['filename']) - print(ct.colorText(f"Approved hash lists have been combined","green")) + primary_path_exclusions = calculatePath(all_approved_hashes, badpathparts, path_exclusion_constant, min_files_for_path, split) - all_approved_hashes.to_parquet(f"parquet\\all_approved_hashes_{first_policy}_{second_policy}.parquet", index=False) - del all_approved_hashes - gc.collect() + remaining_hashes = all_approved_hashes[~all_approved_hashes['sha256'].isin(primary_path_exclusions['sha256'])] - if not os.path.exists(f"parquet\\path_needs_approved_{first_policy}_{second_policy}.parquet"): - all_approved_hashes = pd.read_parquet(f"parquet\\all_approved_hashes_{first_policy}_{second_policy}.parquet") - print(ct.colorText(f"Beginning calculating longest common filepaths for path exceptions","green")) - - haslcp = pathf.split_filepaths_grouped(all_approved_hashes,"filename",path_exclusion_constant, min_files_for_path) - haslcp.drop_duplicates() + secondary_path_exclusions = calculatePath(remaining_hashes, badpathparts, 3, min_files_for_path, split) - forbidden = pathf.regulator(badpathparts, True) - forbidden_lcfp = haslcp["longestcfp"].str.contains(forbidden, na=False) + remaining_hashes = remaining_hashes[~remaining_hashes['sha256'].isin(secondary_path_exclusions['sha256'])] - - print(ct.colorText("Removing forbidden filepaths for path exceptions", "green")) + return all_approved_hashes, primary_path_exclusions, secondary_path_exclusions, remaining_hashes - # Make a real DataFrame copy before modifying - lcp_not_forbidden = haslcp[~forbidden_lcfp].copy() - - #For the review, drop down to only the columns we care, and then group by the commmon file path, consolidating and dropping dupes - lcp_not_forbidden_review = lcp_not_forbidden[['longestcfp', 'middle', 'filename_only', 'file_extension', 'sha256']] - - # Count unique sha256 per longestcfp - unique_sha_counts = lcp_not_forbidden_review.groupby('longestcfp')['sha256'].nunique().reset_index() - unique_sha_counts.columns = ['longestcfp', 'unique_sha256_count'] - - # Merge the count back into the original DataFrame - lcp_not_forbidden_review = lcp_not_forbidden_review.merge(unique_sha_counts, on='longestcfp', how='left') - lcp_not_forbidden_review = lcp_not_forbidden_review[lcp_not_forbidden_review['unique_sha256_count'] >= min_files_for_path] - - lcp_not_forbidden_review.to_parquet(f"parquet\\path_needs_approved_{first_policy}_{second_policy}.parquet",index=False) - lcp_not_forbidden_review.to_csv(f"needs_approved\\path_needs_approved_{first_policy}_{second_policy}.csv",index=False) - ct.style_dataframe_dark(lcp_not_forbidden_review,f"needs_approved\\path_needs_approved_{first_policy}_{second_policy}.html", True) - - del lcp_not_forbidden - del unique_sha_counts - del lcp_not_forbidden_review - - if not os.path.exists(f"parquet\\secondary_paths_{first_policy}_{second_policy}.parquet"): - all_approved_hashes = pd.read_parquet(f"parquet\\all_approved_hashes_{first_policy}_{second_policy}.parquet") - recommended_paths = pd.read_parquet(f"parquet\\path_needs_approved_{first_policy}_{second_policy}.parquet") - - remaining_hashes = all_approved_hashes[~all_approved_hashes['sha256'].isin(recommended_paths['sha256'])] - - print(ct.colorText(f"Beginning calculating longest common filepaths for path exceptions","green")) - - haslcp = pathf.split_filepaths_grouped(remaining_hashes,"filename", 3, min_files_for_path) - haslcp.drop_duplicates() - - forbidden = pathf.regulator(badpathparts, True) - forbidden_lcfp = haslcp["longestcfp"].str.contains(forbidden, na=False) - - - print(ct.colorText("Removing forbidden filepaths for path exceptions", "green")) - - # Make a real DataFrame copy before modifying - lcp_not_forbidden = haslcp[~forbidden_lcfp].copy() - - #For the review, drop down to only the columns we care, and then group by the commmon file path, consolidating and dropping dupes - lcp_not_forbidden_review = lcp_not_forbidden[['longestcfp', 'middle', 'filename_only', 'file_extension', 'sha256']] - - # Count unique sha256 per longestcfp - unique_sha_counts = lcp_not_forbidden_review.groupby('longestcfp')['sha256'].nunique().reset_index() - unique_sha_counts.columns = ['longestcfp', 'unique_sha256_count'] - - # Merge the count back into the original DataFrame - lcp_not_forbidden_review = lcp_not_forbidden_review.merge(unique_sha_counts, on='longestcfp', how='left') - lcp_not_forbidden_review = lcp_not_forbidden_review[lcp_not_forbidden_review['unique_sha256_count'] >= min_files_for_path] - - lcp_not_forbidden_review.to_parquet(f"parquet\\secondary_paths_{first_policy}_{second_policy}.parquet",index=False) - lcp_not_forbidden_review.to_csv(f"needs_approved\\secondary_paths_{first_policy}_{second_policy}.csv",index=False) - ct.style_dataframe_dark(lcp_not_forbidden_review,f"needs_approved\\secondary_paths_{first_policy}_{second_policy}.html", True) - - remaining_hashes = remaining_hashes[~remaining_hashes['sha256'].isin(lcp_not_forbidden_review['sha256'])] - - remaining_hashes.to_csv(f"needs_approved\\not_covered_by_path_exclusion_{first_policy}_{second_policy}.csv",index=False) - - del lcp_not_forbidden - del unique_sha_counts - del lcp_not_forbidden_review - -def allATpaths(url, pups, untrusted_publishers, badpathparts, threat_tolerance, path_exclusion_constant, min_files_for_path): - import pandas as pd - - # Load raw data - hashes = pd.read_csv("exclusions\\allATPolicyExecs.csv") - - # Deduplicate hashes before augmentation - deduped_hashes = hashes.drop_duplicates(subset=['sha256']).copy() - - # Save policy name mapping (before deduplication) - policyname_map = hashes[['hostname', 'PolicyName']].drop_duplicates() - - # Handle None inputs - untrusted_publishers = untrusted_publishers or [] - pups = pups or [] - - # Augment deduplicated hashes - augmented = hashf.augmentAggregatedHashes(url, deduped_hashes) - - # Merge policy names - augmented = augmented.merge(policyname_map, on='hostname', how='left') - - # Clean numeric reputation fields - for col in ['reputation_scannermatch', 'reputation_scannercount', 'reputation_threatlevel']: - if col in augmented.columns: - augmented[col] = pd.to_numeric(augmented[col].replace('N/A', pd.NA), errors='coerce') - - # Rename and select relevant columns - augmented = augmented.rename(columns={'publisher_x': 'publisher'}) - augmented = augmented[[ - 'PolicyName', 'sha256', 'publisher', 'description', 'productname', 'productversion', - 'reputation_lastseen', 'reputation_scannermatch', 'reputation_scannercount', - 'reputation_status', 'reputation_threatlevel', 'reputation_threatname', - 'reputation_timestamp' - ]].sort_values(by=['publisher', 'description', 'productname']) - - # Merge with deduplicated hashes to enrich data - final_augmented = augmented.merge(deduped_hashes, on='sha256', how='left') - - # Clean up column names before applying reputation logic - if 'publisher_y' in final_augmented.columns: - final_augmented = final_augmented.drop(columns=['publisher_y']) - if 'publisher_x' in final_augmented.columns: - final_augmented = final_augmented.rename(columns={'publisher_x': 'publisher'}) - if 'PolicyName_x' in final_augmented.columns: - final_augmented = final_augmented.rename(columns={'PolicyName_x': 'PolicyName'}) - - final_augmented.to_csv("exclusions\\testing.csv", index=False) - - # Reputation flag logic - def reputationtool(row): - val = row["reputation_scannermatch"] - if pd.isna(val): - return row["publisher"] == "Not Signed" - try: - return int(val) > threat_tolerance - except (ValueError, TypeError): - return row["publisher"] == "Not Signed" - - df = final_augmented.copy() - df["reputation_flag"] = df.apply(reputationtool, axis=1) - - # Filtering logic - mask_needsreview = ( - ((df["publisher"] == "Not Signed") & df["reputation_flag"]) | - (df["reputation_status"] == "UNKNOWN") - ) - - mask_approved = ( - ( - (df["publisher"] != "Not Signed") & - ~df["publisher"].str.contains(pathf.regulator(untrusted_publishers), case=False, na=False) & - ~df["reputation_status"].isna() & - ~df["description"].str.contains(pathf.regulator(pups), case=False, na=False) - ) | - ( - (df["publisher"] == "Not Signed") & - ~df["reputation_flag"] & - ~df["publisher"].str.contains(pathf.regulator(untrusted_publishers), case=False, na=False) & - ~df["reputation_status"].isna() & - ~df["description"].str.contains(pathf.regulator(pups), case=False, na=False) - ) - ) - - needsreview_df = df[mask_needsreview] - approved_df = df[mask_approved] - all_approved_hashes = pd.concat([needsreview_df, approved_df], ignore_index=True) - - print(ct.colorText("Beginning calculating longest common filepaths for path exceptions", "green")) - - # Path analysis - haslcp = pathf.split_filepaths_grouped(all_approved_hashes, "filename", path_exclusion_constant, min_files_for_path) - haslcp = haslcp.drop_duplicates() - - # Remove forbidden paths - forbidden = pathf.regulator(badpathparts, True) - lcp_not_forbidden = haslcp[~haslcp["longestcfp"].str.contains(forbidden, na=False)].copy() - - print(ct.colorText("Removing forbidden filepaths for path exceptions", "green")) - - # Reviewable paths - review_df = lcp_not_forbidden[['longestcfp', 'middle', 'filename_only', 'file_extension', 'sha256']] - sha_counts = review_df.groupby('longestcfp')['sha256'].nunique().reset_index() - sha_counts.columns = ['longestcfp', 'unique_sha256_count'] - - 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("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("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("exclusions\\merged_output.csv", index=False) - - print(f"Merged DataFrame saved with {len(merged_df)} rows.") - - -def listPaths(url, group): - endpoint = url + '/v1/group/policies' - print(ct.colorText("[+] Grabbing All Paths", "cyan")) - - payload = { - "groupid": [group], - } - headers = { - "X-APIKey": os.getenv('APIKEY') - } - - try: - response = requests.post(endpoint, headers=headers, data=payload, verify=False) - response.raise_for_status() - parse_text = response.json() - - pathnames = [] - print(parse_text) # Optional: for debugging - - for item in parse_text.get('response', {}).get('paths', []): - path = item.get('name') - if path: - pathnames.append(path) - - return pathnames - - except requests.exceptions.RequestException as e: - print(ct.colorText(f"[!] Request failed: {e}", "red")) - return [] - except (KeyError, json.JSONDecodeError) as e: - print(ct.colorText(f"[!] Failed to parse response: {e}", "red")) - return [] - -def wildcardRegex(pattern): - pattern = pattern.replace("\\", "\\\\") - pattern = pattern.replace("**", "___RECURSIVE___") - pattern = pattern.replace("*", "[^\\\\]*") - pattern = pattern.replace("?", ".") - pattern = pattern.replace("___RECURSIVE___", ".*") - return re.compile(f"^{pattern}$", re.IGNORECASE) - -def clean_folders_enforcement_prep(): +def clean_folders(parq_base_dir, appr_base_dir, needappr_base_dir, pflight_base_dir): """ Prompts user to choose whether to delete all .parquet files or preserve execution_history ones. Then deletes .csv, .html, and .parquet files accordingly from specified folders. @@ -416,7 +154,7 @@ def clean_folders_enforcement_prep(): user_input = input("Do you want to delete *all* .parquet files including execution_history ones? (yes/y or no/n): ").strip().lower() delete_execution_hist = user_input in ["yes", "y"] - folders = ["approved", "exclusions", "needs_approved", "parquet", "preflight"] + folders = [parq_base_dir, appr_base_dir, needappr_base_dir, pflight_base_dir] for folder in folders: folder_path = os.path.abspath(folder) @@ -443,3 +181,35 @@ def clean_folders_enforcement_prep(): if delete_execution_hist or not filename.startswith("execution_history"): os.remove(file_path) +def generateApprovables(needappr_base_dir, appr_base_dir, parq_base_dir, badpathparts, bad_publisher_list, path_exclusion_constant, min_files_for_path, split= False ): + + if os.path.exists(f"{needappr_base_dir}unknown_hashes.csv") and os.path.exists(f"{needappr_base_dir}good_hashes.csv"): + + all_hashes, primary_paths, secondary_paths, remaining = pathf.generatePathReview(f"{appr_base_dir}unknown_hashes.csv", f"{appr_base_dir}good_hashes.csv", badpathparts,path_exclusion_constant, min_files_for_path, split) + + all_hashes.to_parquet(f"{parq_base_dir}all_hashes.parquet", index=False) + + dataframes = { + "all_hashes" : all_hashes, + "primary_Paths": primary_paths, + "secondary_Paths": secondary_paths, + "remaining": remaining + + } + + for name, df in dataframes.items(): + df.to_csv(f"{needappr_base_dir}{name}.csv", index=False) + df.to_parquet(f"{parq_base_dir}{name}.parquet", index=False) + ct.style_dataframe_dark(df, f"{needappr_base_dir}{name}.html") + + + + publishers = hashf.generatePublist(f"{parq_base_dir}all_hashes.parquet",bad_publisher_list) + + publishers.to_csv(f"{needappr_base_dir}publishers.csv", index=False) + publishers.to_parquet(f"{parq_base_dir}publishers.parquet", index=False) + ct.style_dataframe_dark(publishers, f"{needappr_base_dir}publishers.html") + + else: + print(ct.colorText(f"Please manually approve hashes prior to this step","red")) + \ No newline at end of file diff --git a/utils/perstscheduler.py b/utils/perstscheduler.py index 71e0489..81e6762 100644 --- a/utils/perstscheduler.py +++ b/utils/perstscheduler.py @@ -123,7 +123,7 @@ def _schedule_once(job_id: str, func_name: str, run_at_timestamp: float, args=No print(f"[WARN] Job {job_id} scheduled in the past. Skipping.") return # Schedule via schedule library - schedule.every(delay_seconds).seconds.do(job_wrapper).tag(job_id) + schedule.every(int(delay_seconds)).seconds.do(job_wrapper).tag(job_id) def _schedule_recurring(job_id: str, func_name: str, interval: int, unit: str, args=None, kwargs=None): args = args or [] diff --git a/utils/policyfunctions.py b/utils/policyfunctions.py index eda13d0..e39bbaf 100644 --- a/utils/policyfunctions.py +++ b/utils/policyfunctions.py @@ -14,7 +14,11 @@ # along with this program. If not, see . #Local Imports -import utils.pretty as ct +import utils.utils as ct +import utils.hashfunctions as hashf +import utils.pathfunctions as pathf +import utils.policyfunctions as policyf + #Standard Libary Imports: import datetime import gc @@ -44,7 +48,6 @@ def addPub(url, policy, publist): print(f"Adding the following Publishers to {policy}:") for p in publist: print(p) - def addHashReal(url, allowlistID, hashlist): endpoint = url + '/v1/hash/application/add' @@ -62,7 +65,6 @@ def addHashReal(url, allowlistID, hashlist): parse_text = json.loads(response.text) print(parse_text) - def addPathReal(url, grouplistID, pathlist): endpoint = url + '/v1/group/path/add' payload = { @@ -77,7 +79,6 @@ def addPathReal(url, grouplistID, pathlist): response = requests.request("POST", endpoint, headers=headers, data=payload, verify=False) print(response.text) - def addPubReal(url, grouplistID, publist): endpoint = url + '/v1/group/publisher/add' payload = { @@ -92,27 +93,29 @@ def addPubReal(url, grouplistID, publist): response = requests.request("POST", endpoint, headers=headers, data=payload, verify=False) print(response.text) - def getPolicyInfo(url, policy, type, days, parquet=True): executionhist_policy = pd.DataFrame() exehist = pullPolicyExechistories(url, policy, type, days, True) - data = json.loads(exehist) - executionhist_policy = pd.DataFrame(data["response"]["exechistories"]) - if not executionhist_policy.empty: - executionhist_policy = executionhist_policy[['sha256', 'publisher', 'filename', 'hostname', 'username', 'pprocess', 'gprocess', 'commandline']] - executionhist_policy = executionhist_policy.drop_duplicates(subset=['sha256', 'filename', 'hostname']) - executionhist_policy = executionhist_policy.sort_values(by=['sha256', 'filename']) - if parquet: executionhist_policy.to_parquet(f"parquet\\execution_history_{policy}.parquet", index=False) - print(ct.colorText(f"Staging of Execution history for policy: {policy} is complete", "green")) - del data - del exehist - gc.collect() + if exehist is not None: + data = json.loads(exehist) + executionhist_policy = pd.DataFrame(data["response"]["exechistories"]) + if not executionhist_policy.empty: + executionhist_policy = executionhist_policy[['sha256', 'publisher', 'filename', 'hostname', 'username', 'pprocess', 'gprocess', 'commandline']] + executionhist_policy['policy'] = policy # Add policy column here + executionhist_policy = executionhist_policy.drop_duplicates(subset=['sha256', 'filename', 'hostname']) + executionhist_policy = executionhist_policy.sort_values(by=['sha256', 'filename']) + if parquet: + executionhist_policy.to_parquet(f"prepare_policy\\parquet\\execution_history_{policy}.parquet", index=False) + print(ct.colorText(f"Staging of Execution history for policy: {policy} is complete", "green")) + del data + del exehist + gc.collect() return executionhist_policy -def sendToPolicy(url, first_policy, second_policy, destination_name, destination_id, allowlist_parent_name, allowlist_parent_id, allowlist_child_name, allowlist_child_id): - pathexclusions = pd.read_parquet(f"parquet\\final_path_exclusions_{first_policy}_{second_policy}.parquet") - allowbyhash = pd.read_parquet(f"parquet\\final_hash_approvals_{first_policy}_{second_policy}.parquet") - publishers = pd.read_parquet(f"parquet\\publishers_{first_policy}_{second_policy}.parquet") +def sendToPolicy(url, paths, hashes, publishers, destination_name, destination_id, allowlist_name, allowlist_id): + pathexclusions = pd.read_parquet(paths) + allowbyhash = pd.read_parquet(hashes) + publishers = ct.tryToReadCSV(publishers) ct.areYouSure() confirmation = input(ct.colorText("Type 'I AGREE' to continue: ","white")) @@ -137,22 +140,22 @@ def sendToPolicy(url, first_policy, second_policy, destination_name, destination addPathReal(url, destination_id,processed_paths) print(ct.colorText(f"Adding publishers to {destination_name}", "yellow")) + + if publishers.empty: + print(ct.colorText("The publishers list is empty.", "red")) + else: + publisher_list = publishers['publisher'].tolist() + addPubReal(url, destination_id, publisher_list) - publisher_list = publishers['publisher'].tolist() - addPubReal(url, destination_id, publisher_list) - print(ct.colorText(f"Adding hashes to {allowlist_parent_name}", "yellow")) + print(ct.colorText(f"These hashes would be added to {allowlist_name}", "yellow")) - allowlist_parenthashlist = allowbyhash[allowbyhash['reputation_status'] == 'KNOWN']['sha256'].unique().tolist() - addHashReal(url, allowlist_parent_id,allowlist_parenthashlist) - - print(ct.colorText(f"Adding hashes to {allowlist_child_name}", "yellow")) - allowlist_childhashlist = allowbyhash[allowbyhash['reputation_status'] == 'UNKNOWN']['sha256'].unique().tolist() - addHashReal(url, allowlist_child_id, allowlist_childhashlist) + allowlist = allowbyhash['sha256'].unique().tolist() + addHash(url, allowlist_id,allowlist) ct.locked() - exit() + exit() else: print(ct.colorText("Operation aborted. You MUST EXPLICITLY AGREE to proceed.", "red")) @@ -281,8 +284,7 @@ def listATPolicies(url): print(ct.colorText(f"[!] Failed to parse response: {e}", "red")) return {} - -def listAllowlists(url): +def listAllowlists(url: str) -> tuple[int, list, list]: endpoint = url + '/v1/application' print(ct.colorText("[+] Grabbing All Allowlists", "cyan")) payload = {} @@ -293,18 +295,23 @@ def listAllowlists(url): parse_text = json.loads(response.text) policiesnames = [] policyids = [] - for index, list in enumerate(parse_text['response']['applications'], start=1): + + for index, item in enumerate(parse_text['response']['applications'], start=1): if index >= 38: - print(ct.colorText(f"{index}. {list['name']}", "yellow")) - policiesnames.append(list['name']) - policyids.append(list['applicationid']) - choice = int(input(ct.colorText("Select allowlist: ", "white"))) - if choice < 38: - print(ct.colorText("Please only choose an allowlist designed for this use - '38+'","red")) - elif choice >= 38: - choice = choice - 38 - return choice, policiesnames, policyids - #Need else and catch for upper bound + print(ct.colorText(f"{index}. {item['name']}", "yellow")) + policiesnames.append(item['name']) + policyids.append(item['applicationid']) + + while True: + try: + choice = int(input(ct.colorText("Select allowlist: ", "white"))) + if choice < 38 or choice > len(parse_text['response']['applications']): + print(ct.colorText("Please only choose an allowlist designed for this use - '38+'", "red")) + else: + adjusted_choice = choice - 38 + return adjusted_choice, policiesnames, policyids + except ValueError: + print(ct.colorText("Invalid input. Please enter a number.", "red")) def skipback(days): """ @@ -318,10 +325,10 @@ def skipback(days): objectid_hex = hex_timestamp + '0000000000000000' return ObjectId(objectid_hex) -def sendToPolicyTest(url, first_policy, second_policy, destination_name, destination_id, allowlist_parent_name, allowlist_parent_id, allowlist_child_name, allowlist_child_id): - pathexclusions = pd.read_parquet(f"parquet\\final_path_exclusions_{first_policy}_{second_policy}.parquet") - allowbyhash = pd.read_parquet(f"parquet\\final_hash_approvals_{first_policy}_{second_policy}.parquet") - publishers = pd.read_parquet(f"parquet\\publishers_{first_policy}_{second_policy}.parquet") +def sendToPolicyTest(url, paths, hashes, publishers, destination_name, destination_id, allowlist_id, allowlist_name): + pathexclusions = pd.read_parquet(paths) + allowbyhash = pd.read_parquet(hashes) + publishers = ct.tryToReadCSV(publishers) print(ct.colorText(f"These path exclusions would be added to {destination_name}", "yellow")) @@ -342,20 +349,17 @@ def sendToPolicyTest(url, first_policy, second_policy, destination_name, destina print(ct.colorText(f"These publishers would added to {destination_name}", "yellow")) - publisher_list = publishers['publisher'].tolist() - addPub(url, destination_id, publisher_list) + if publishers.empty: + print(ct.colorText("The publishers list is empty.", "red")) + else: + publisher_list = publishers['publisher'].tolist() + addPub(url, destination_id, publisher_list) - print(ct.colorText(f"These hashes would be added to {allowlist_parent_name}", "yellow")) + print(ct.colorText(f"These hashes would be added to {allowlist_name}", "yellow")) - allowlist_parenthashlist = allowbyhash[allowbyhash['reputation_status'] == 'KNOWN']['sha256'].unique().tolist() - addHash(url, allowlist_parent_id,allowlist_parenthashlist) + allowlist = allowbyhash['sha256'].unique().tolist() + addHash(url, allowlist_id,allowlist) - print(ct.colorText(f"These hashes would be added to {allowlist_child_name}", "yellow")) - allowlist_childhashlist = allowbyhash[allowbyhash['reputation_status'] == 'UNKNOWN']['sha256'].unique().tolist() - addHash(url, allowlist_child_id, allowlist_childhashlist) - - - def updateAuditPoliciesFromEnforcementPolices(url, policy_relationship_map): for enforcement_policy, audit_policy in policy_relationship_map.items(): assignPoliciesfromGroup(url, enforcement_policy, audit_policy) @@ -376,7 +380,6 @@ def assignPoliciesfromGroup(url, source_policy_id, target_policy_id): parse_text = json.loads(response.text) print(parse_text) - def turnOnAudit(url, policyid): endpoint = url + '/v1/group/settings/auditmode' @@ -404,3 +407,205 @@ def agentsInPolicy(url, policyid): response = requests.request("POST", endpoint, headers=headers, data=payload, verify=False) parse_text = json.loads(response.text) print(parse_text) + +def prepare_to_enforce(url, bad_publisher_list, pups, badpathparts, threat_tolerance_constant = 4, path_exclusion_constant = 4, min_files_for_path = 4): + + destination_name = " " + destination_id = " " + allowlist_name = " " + allowlist_id = " " + policylist = [] + type = [1, 2, 6, 7] + + parq_base_dir = "prepare_policy\\parquet\\" + appr_base_dir = "prepare_policy\\approved\\" + needappr_base_dir = "prepare_policy\\needs_approved\\" + pflight_base_dir = "prepare_policy\\preflight\\" + + #If the directorys where we're going to store our output dont exist, make them. + os.makedirs(parq_base_dir, exist_ok=True) + os.makedirs(needappr_base_dir, exist_ok=True) + os.makedirs(appr_base_dir, exist_ok=True) + os.makedirs(pflight_base_dir, exist_ok=True) + + while True: + + ct.printEnforceChecklist(parq_base_dir,appr_base_dir, needappr_base_dir, pflight_base_dir, policylist, allowlist_name, destination_name) + + choice = input(ct.colorText("\nEnter your choice: ", "white")) + + if choice == "1": + + while True: + choice, policynames, policyid = listPolicies(url) + selected_policy = policynames[choice] + + if selected_policy not in policylist: + policylist.append(selected_policy) + + while True: + answer = input(ct.colorText("Do you want to load another policy? (yes/no): ", "white")).strip().lower() + if answer in ("no", "n"): + break # Exit the inner loop and then the outer loop + elif answer in ("yes", "y"): + break # Exit the inner loop and continue the outer loop + else: + print(ct.colorText("Please answer with 'yes' or 'no'.", "red")) + + if answer in ("no", "n"): + break + + print(policylist) + + elif choice == "2": + + print(ct.colorText(f"Please choose destination_name Policy for Path Exclusions","white")) + choice, policynames, policyid = listPolicies(url) + #print(allowlist_parent_tuple) + destination_name = policynames[choice] + destination_id = policyid[choice] + + print(ct.colorText(f"Please choose Allowlist for Hashes","white")) + choice, allowlists,allowid = listAllowlists(url) + #print(allowlist_parent_tuple) + allowlist_name = allowlists[choice] + allowlist_id = allowid[choice] + + print(destination_name, allowlist_name) + + + elif choice == "3": + + policyf.buildExecHistory(url, + policylist, + parq_base_dir, + needappr_base_dir, + type, + threat_tolerance_constant, + bad_publisher_list, + pups, + ) + csvs = [f"{needappr_base_dir}unknown_hashes.csv", f"{needappr_base_dir}good_hashes.csv"] + #Since we want to build paths as if they were all in the same policy to begin with, lets group them that way + for csv in csvs: + df = ct.tryToReadCSV(csv) + df['policy'] = destination_name + df.to_csv(csv) + + + elif choice == "4": + pathf.generateApprovables(needappr_base_dir, appr_base_dir, parq_base_dir, badpathparts, bad_publisher_list, path_exclusion_constant, min_files_for_path) + + elif choice == "5": + savePreFlights(appr_base_dir, parq_base_dir, pflight_base_dir) + + + elif choice == "6": + if os.path.exists(f"{pflight_base_dir}final_path_exclusions.html") and os.path.exists(f"{pflight_base_dir}\\final_hash_approvals.html") and allowlist_name != " " and destination_name != " ": + sendToPolicyTest( + url, + f"{parq_base_dir}final_path_exclusions.parquet", + f"{parq_base_dir}final_hash_approvals.parquet", + f"{appr_base_dir}publishers.parquet", + destination_name, + destination_id, + allowlist_name, + allowlist_id + ) + + elif choice == "7": + if os.path.exists(f"{pflight_base_dir}final_path_exclusions.html") and os.path.exists(f"{pflight_base_dir}\\final_hash_approvals.html") and allowlist_name != " " and destination_name != " ": + sendToPolicy( + url, + f"{parq_base_dir}final_path_exclusions.parquet", + f"{parq_base_dir}final_hash_approvals.parquet", + f"{appr_base_dir}publishers.parquet", + destination_name, + destination_id, + allowlist_name, + allowlist_id, + ) + elif choice == "R": + + pathf.clean_folders(parq_base_dir, appr_base_dir, needappr_base_dir, pflight_base_dir) + + elif choice == "Q": + break + else: + print(ct.colorText("Invalid choice. Please try again.", "red")) + +def buildExecHistory(url, + policylist, + parq_base_dir, + needappr_base_dir, + type, + threat_tolerance_constant, + bad_publisher_list, + pups + ): + exe_hist_parq_list = [] + while True: + try: + history_days = int(input("Enter the how many days in of history do you want to pull - select a number between 1 and 150: ")) + if 1 <= history_days <= 150: + break + else: + print("Invalid input. Please enter a number between 1 and 150.") + except ValueError: + print("Invalid input. Please enter a valid integer.") + for policy in policylist: + policy_exec_history = policyf.getPolicyInfo(url, policy, type, history_days) + policy_exec_history['policy'] = policy + policy_exec_history.to_parquet(f"{parq_base_dir}Exec_Hist_{policy}.parquet") + exe_hist_parq_list.append(f"{parq_base_dir}Exec_Hist_{policy}.parquet") + + augmented_hashlist = hashf.combineHashes(url, exe_hist_parq_list) + augmented_hashlist.to_parquet(f"{parq_base_dir}augmentedHashlist.parquet",index=False) + + needsreview_df, approved_df, unapproved_df = hashf.categorizeHashes(augmented_hashlist, threat_tolerance_constant, bad_publisher_list, pups) + needsreview_df.to_parquet(f"{parq_base_dir}needsreview.parquet",index=False) + approved_df.to_parquet(f"{parq_base_dir}approved.parquet",index=False) + unapproved_df.to_parquet(f"{parq_base_dir}unapproved.parquet",index=False) + + + condensed_executions = hashf.condenseExecutions(exe_hist_parq_list) + condensed_executions.to_parquet(f"{parq_base_dir}condensed_executions.parquet", index=False) + + unknown, good, bad = hashf.divideSortedHashExecutions( + f"{parq_base_dir}needsreview.parquet", + f"{parq_base_dir}approved.parquet", + f"{parq_base_dir}unapproved.parquet", + f"{parq_base_dir}condensed_executions.parquet", + pups + ) + + dataframes = { + "unknown_hashes" : unknown, + "good_hashes": good, + "bad_hashes": bad + } + + for name, df in dataframes.items(): + df.to_csv(f"{needappr_base_dir}{name}.csv", index=False) + df.to_parquet(f"{parq_base_dir}{name}.parquet", index=False) + ct.style_dataframe_dark(df, f"{needappr_base_dir}{name}.html") + +def savePreFlights(appr_base_dir, parq_base_dir, pflight_base_dir): + if os.path.exists(f"{appr_base_dir}primary_Paths.csv"): + if not os.path.exists(f"{parq_base_dir}final_hash_approvals.parquet") and not os.path.exists(f"{parq_base_dir}final_path_exclusions.parquet"): + + pathexclusions, allowbyhash = hashf.generatePreflights( + f"{parq_base_dir}all_hashes.parquet", + f"{appr_base_dir}primary_Paths.csv", + f"{appr_base_dir}secondary_Paths.csv") + + dataframes = { + "final_path_exclusions" : pathexclusions, + "final_hash_approvals": allowbyhash + } + + for name, df in dataframes.items(): + df.to_csv(f"{pflight_base_dir}{name}.csv", index=False) + df.to_parquet(f"{parq_base_dir}{name}.parquet", index=False) + ct.style_dataframe_dark(df, f"{pflight_base_dir}{name}.html") + diff --git a/utils/pretty.py b/utils/utils.py similarity index 71% rename from utils/pretty.py rename to utils/utils.py index 5b151e4..3a5a4b2 100644 --- a/utils/pretty.py +++ b/utils/utils.py @@ -15,6 +15,7 @@ #Standard Libary Imports: import os +import pandas as pd def colorText(text: str, color: str) -> str: colors = { @@ -25,7 +26,7 @@ def colorText(text: str, color: str) -> str: "magenta": "\033[95m", "cyan": "\033[96m", "white": "\033[97m", - "reset": "\033[0m" + "reset": "\033[0m" } return f"{colors.get(color, colors['reset'])}{text}{colors['reset']}" @@ -193,7 +194,7 @@ def displayIntro(): print(colorText("======================== Welcome to the Airlock API Tool ========================", "cyan")) print(colorText("=================================================================================", "cyan")) -def printEnforceChecklist(first_policy, second_policy, allowlist_child_name, allowlist_parent_name, destination_name): +def printEnforceChecklist(parq_base_dir,appr_base_dir, needappr_base_dir, pflight_base_dir, policy_list, allowlist_name, destination_name): print(colorText("\n --------------------------------------------------------------------", "cyan")) print(colorText(" ------------- šŸ› ļø šŸ”’ Prepare to Enforce Policy šŸ› ļø šŸ”’ ------------------", "cyan")) @@ -201,106 +202,86 @@ def printEnforceChecklist(first_policy, second_policy, allowlist_child_name, all print(colorText("\nSequentually follow these steps to prepare a policy for enforcement:", "white")) print(colorText("\n1. Choose which originating policy or policies to move to enforcement", "cyan")) - if first_policy == " " and second_policy == " ": + if not policy_list: print(colorText(f" [āœ—] No policies have been chosen","red")) - elif first_policy != " " and second_policy is first_policy: - print(colorText(f" [āœ“] {first_policy} has been selected,", "green")) - elif first_policy != " " and second_policy != " ": - print(colorText(f" [āœ“] {first_policy} has been selected as Policy 1","green")) - print(colorText(f" [āœ“] {second_policy} has been selected as Policy 2","green")) - - - - print(colorText("2. Pull and stage event history, combine the histories, add hash info, then categorize the hashes", "cyan")) - - if os.path.exists(f"parquet\\execution_history_{first_policy}.parquet"): - print(colorText(f" [āœ“] Execution history has been compiled for {first_policy}","green")) - elif not os.path.exists(f"parquet\\execution_history_{first_policy}.parquet"): - print(colorText(f" [āœ—] Execution history has not been compiled for {first_policy}","red")) - elif second_policy is not first_policy and os.path.exists(f"parquet\\execution_history_{second_policy}.parquet"): - print(colorText(f" [āœ“] Execution history has been compiled for {second_policy}","green")) - elif second_policy is not first_policy and not os.path.exists(f"parquet\\execution_history_{second_policy}.parquet"): - print(colorText(f" [āœ—] Execution history has not been compiled for {second_policy}","red")) - - if os.path.exists(f"parquet\\combined_hashlist_{first_policy}_{second_policy}.parquet"): - print(colorText(f" [āœ“] Hash Info has been added to the combined execution history", "green")) else: - print(colorText(f" [āœ—] Hash Info has not been added to the combined execution history", "red")) - - if os.path.exists(f"parquet\\hashes_rep_unknown_{first_policy}_{second_policy}.parquet") and os.path.exists(f"parquet\\hashes_rep_good_{first_policy}_{second_policy}.parquet") and os.path.exists(f"parquet\\hashes_rep_bad_{first_policy}_{second_policy}.parquet"): - print(colorText(f" [āœ“] Hashes have been cateogrized", "green")) - else: - print(colorText(f" [āœ—] Hashes have not been cateogrized", "red")) - - if os.path.exists(f"parquet\\condensed_executions_{first_policy}_{second_policy}.parquet"): - print(colorText(f" [āœ“] Execution history has been_combined_for {first_policy} and_{second_policy}", "green")) - else: - print(colorText(f" [āœ—] Execution history has not been_combined_for {first_policy} and_{second_policy}", "red")) + print(colorText(f"The following policies have been choosen:", "green")) + for policy in policy_list: + print(colorText(f" [āœ“] {policy}","green")) - print(colorText(f"3. Manually review the files:","cyan")) - print(colorText(" 'needs_approved\\hashes_rep_good_{first_policy}_{second_policy}.csv' and 'needs_approved\\hashes_rep_unknown_{first_policy}_{second_policy}.csv'", "cyan")) - print(colorText(" Remove the rows containing hashes you do not approve of, and those you would not approve of without metarules.", "cyan")) - print(colorText(" If metarules need to be created, please make note of them, and remove the row from the csv.", "cyan")) - print(colorText(" When complete, save both csv files to the directory 'approved' and choose this option.","cyan")) - print(colorText(" This will combine these approved hashes with the automatically approved hashes and generate a list of paths to be reviewed", "cyan")) - - if os.path.exists(f"approved\\hashes_rep_good_{first_policy}_{second_policy}.csv") and os.path.exists(f"approved\\hashes_rep_unknown_{first_policy}_{second_policy}.csv"): - print(colorText(" [āœ“] Reviewed hashes have been loaded","green")) - else: - print(colorText(" [āœ—] Reviewed hashes have not been loaded","red")) - - if os.path.exists(f"parquet\\all_approved_hashes_{first_policy}_{second_policy}.parquet"): - print(colorText(" [āœ“] The combined approved hashes list has been generated","green")) - else: - print(colorText(" [āœ—] The combined approved hashes list has not been generated","red")) - - if os.path.exists(f"needs_approved\\path_needs_approved_{first_policy}_{second_policy}.csv"): - print(colorText(" [āœ“] Path review list created","green")) - else: - print(colorText(" [āœ—] Path review list has not been created","red")) - - - print(colorText(f"4. Manually review the file 'needs_approved\\paths_needing_review_{first_policy}_{second_policy}.csv'", "cyan")) - print(colorText(" Remove the rows containing path exclusions you do not approve of" , "cyan")) - print(colorText(" When complete, save the csv file to the directory 'approved'", "cyan")) - print(colorText(" Do the same process with the list of publishers forthe same directories", "cyan")) - print(colorText(" Preflight Lists will be generated", "cyan")) - - if os.path.exists(f"approved\\path_needs_approved_{first_policy}_{second_policy}.csv"): - print(colorText(" [āœ“] Reviewed path list detected","green")) - else: - print(colorText(" [āœ—] Path review list has not been detected","red")) - - if os.path.exists(f"preflight\\final_path_exclusions_{first_policy}_{second_policy}.html"): - print(colorText(" [āœ“] Preflight Path Exclusion List has been generated","green")) - else: - print(colorText(" [āœ—] Preflight Path Exclusion List has not been generated","red")) - - if os.path.exists(f"preflight\\final_hash_approvals_{first_policy}_{second_policy}.html"): - print(colorText(" [āœ“] Preflight hash approval list has been generated","green")) - else: - print(colorText(" [āœ—] Preflight hash approval list has not been generated","red")) - - - print(colorText(f"5. Choose the destination policy and parent and child allow list", "cyan")) - if allowlist_child_name == " " and allowlist_parent_name== " ": + print(colorText(f"2. Choose the destination policy and allowlist", "cyan")) + + if allowlist_name == " ": print(colorText(f" [āœ—] No allowlists have been chosen","red")) - elif allowlist_parent_name != " " and allowlist_child_name != " " and allowlist_parent_name is allowlist_child_name: - print(colorText(f" [āœ“] [āœ—] Only {allowlist_parent_name} has been selected this is unusual, but potentially valid case, double check before proceeding,", "yellow")) - elif allowlist_parent_name != " " and allowlist_child_name != " " and allowlist_parent_name is not allowlist_child_name: - print(colorText(f" [āœ“] {allowlist_parent_name} has been selected as Parent Policy","green")) - print(colorText(f" [āœ“] {allowlist_child_name} has been selected as Child Policy","green")) + elif allowlist_name != " " and allowlist_name != " " and allowlist_name is not allowlist_name: + print(colorText(f" [āœ“] {allowlist_name} has been selected as allowlist","green")) + if destination_name == " ": print(colorText(f" [āœ—] No destination policy has been chosen","red")) else: print(colorText(f" [āœ“] destination policy is {destination_name}","green")) + print(colorText("3. Pull and stage event history, combine the histories, add hash info, then categorize the hashes", "cyan")) + if not policy_list: + print(colorText(f" [āœ—] No policies have been chosen","red")) + else: + for policy in policy_list: + if os.path.exists(f"{parq_base_dir}Exec_Hist_{policy}.parquet"): print(colorText(f" [āœ“] Data for {policy} has been fetched","green")) + else: print(colorText(f" [āœ—] Data for {policy} has not been fetched","red")) + + print(colorText(f"4. Manually review the files:","cyan")) + print(colorText(" 'prepare_policy\\needs_approved\\good_hashes.csv' and 'prepare_policy\\needs_approved\\unknown_hashes.csv'\n", "cyan")) + print(colorText(" Remove the rows containing hashes you do not approve of, and those you would not approve of without metarules.", "cyan")) + print(colorText(" If metarules need to be created, please make note of them, and remove the row from the csv.", "cyan")) + print(colorText(" When complete, save both csv files to the directory 'approved' and choose this option.","cyan")) + print(colorText(" This will combine these approved hashes with the automatically approved hashes and generate a list of paths to be reviewed", "cyan")) + + if os.path.exists(f"{appr_base_dir}good_hashes.csv") and os.path.exists(f"{appr_base_dir}unknown_hashes.csv"): + print(colorText(" [āœ“] Reviewed hashes have been loaded","green")) + else: + print(colorText(" [āœ—] Reviewed hashes have not been loaded","red")) + + if os.path.exists(f"{parq_base_dir}all_hashes.parquet"): + print(colorText(" [āœ“] The combined approved hashes list has been generated","green")) + else: + print(colorText(" [āœ—] The combined approved hashes list has not been generated","red")) + + if os.path.exists(f"{needappr_base_dir}primary_Paths.csv"): + print(colorText(" [āœ“] Path review list created","green")) + else: + print(colorText(" [āœ—] Path review list has not been created","red")) + + + print(colorText(f"5. Manually review the files \n 'prepare_policy\\needs_approved\\primary_Paths.csv'\n 'prepare_policy\\needs_approved\\secondary_Paths.csv'", "cyan")) + print(colorText(" Remove the rows containing path exclusions you do not approve of. The secondary list can be not added at all if nothing is useful" , "cyan")) + print(colorText(" When complete, save the csv file to the directory 'approved'", "cyan")) + print(colorText(" Do the same process with the list of publishers forthe same directories", "cyan")) + print(colorText(" Preflight Lists will be generated", "cyan")) + + if os.path.exists(f"{appr_base_dir}primary_Paths.csv"): + print(colorText(" [āœ“] Reviewed path list detected","green")) + else: + print(colorText(" [āœ—] Path review list has not been detected","red")) + + if os.path.exists(f"{pflight_base_dir}final_path_exclusions.csv"): + print(colorText(" [āœ“] Preflight Path Exclusion List has been generated","green")) + else: + print(colorText(" [āœ—] Preflight Path Exclusion List has not been generated","red")) + + if os.path.exists(f"{pflight_base_dir}final_hash_approvals.csv"): + print(colorText(" [āœ“] Preflight hash approval list has been generated","green")) + else: + print(colorText(" [āœ—] Preflight hash approval list has not been generated","red")) + + + + print(colorText(f"6. Test ------------------------------------------------------", "cyan")) - print(colorText(f" Print rather than apply selected data.", "cyan")) + print(colorText(f" Print rather than apply selected data.", "cyan")) print(colorText(f"7. Liftoff ------------------------------------------------------", "cyan")) - print(colorText(f" Apply path exclusions according to allowed and approved paths", "cyan")) + print(colorText(f" Apply path exclusions according to allowed and approved paths", "cyan")) print(colorText(f" Apply signed or attested hashes to Parent Allow List", "cyan")) print(colorText(f" Apply approved, but unsigned hashes to the Child Allow List", "cyan")) @@ -341,4 +322,80 @@ def locked(): ā–’ā–ˆā–ˆā–ˆā–ˆ ā–“ā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆ ā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–ˆā–‘ -""", "yellow")) \ No newline at end of file +""", "yellow")) + +def printDeviceEnforceChecklist(): + + print(colorText("\n --------------------------------------------------------------------", "cyan")) + print(colorText(" ------------- šŸ› ļø šŸ”’ Prepare to Enforce Policy šŸ› ļø šŸ”’ ------------------", "cyan")) + print(colorText(" --------------------------------------------------------------------", "cyan")) + print(colorText("\nSequentually follow these steps to prepare a policy for enforcement:", "white")) + + print(colorText("\n1. Choose which originating policy or policies to move to enforcement", "cyan")) + print(colorText("2. Pull and stage event history, combine the histories, add hash info, then categorize the hashes", "cyan")) + print(colorText(f"3. Manually review the files:","cyan")) + print(colorText(" 'needs_approved\\good_{first_policy}_{second_policy}.csv' and 'needs_approved\\unknown_{first_policy}_{second_policy}.csv'", "cyan")) + print(colorText(" Remove the rows containing hashes you do not approve of, and those you would not approve of without metarules.", "cyan")) + print(colorText(" If metarules need to be created, please make note of them, and remove the row from the csv.", "cyan")) + print(colorText(" When complete, save both csv files to the directory 'approved' and choose this option.","cyan")) + print(colorText(" This will combine these approved hashes with the automatically approved hashes and generate a list of paths to be reviewed", "cyan")) + print(colorText(f"4. Manually review the file 'needs_approved\\paths_needing_review.csv'", "cyan")) + print(colorText(" Remove the rows containing path exclusions you do not approve of" , "cyan")) + print(colorText(" When complete, save the csv file to the directory 'approved'", "cyan")) + print(colorText(" Do the same process with the list of publishers forthe same directories", "cyan")) + print(colorText(" Preflight Lists will be generated", "cyan")) + + print(colorText(f"5. Choose the destination policy and parent and child allow list", "cyan")) + + print(colorText(f"6. Test ------------------------------------------------------", "cyan")) + print(colorText(f" Print rather than apply selected data.", "cyan")) + + print(colorText(f"7. Liftoff ------------------------------------------------------", "cyan")) + print(colorText(f" Apply path exclusions according to allowed and approved paths", "cyan")) + print(colorText(f" Apply signed or attested hashes to Parent Allow List", "cyan")) + print(colorText(f" Apply approved, but unsigned hashes to the Child Allow List", "cyan")) + + print(colorText("R. Remove/Reset Generated data - will prompt to allow keeping execution history", "cyan")) + + print(colorText("Q. Quit", "cyan")) + + + +def apivalidation(): + match os.getenv('APIKEY'): + case '': + print(colorText("Please add your API Key to the .env file", "red")) + + +def tryToReadCSV(csv): + try: + if not os.path.exists(csv): + print(colorText(f"Error: File '{csv}' does not exist.", "red")) + return pd.DataFrame() # Return empty DataFrame if file doesn't exist + + df = pd.read_csv(csv) + if df.empty: + print(colorText("Error: CSV file has headers but no data rows.", "red")) + else: + print(colorText(f"Data loaded successfully from {csv}", "green")) + except pd.errors.EmptyDataError: + print(colorText("Notice: CSV file is completely empty (no headers, no data), falling back to empty frame", "white")) + df = pd.DataFrame() # Create an empty DataFrame as fallback + return df + + +def tryToReadParquet(parquet): + try: + df = pd.read_parquet(parquet) + if df.empty: + print(colorText("Error: Parquet file has headers but no data rows.", "red")) + else: + print(colorText(f"Data loaded successfully from {parquet}", "green")) + except pd.errors.EmptyDataError: + print(colorText("Notice : Parquet file is completely empty (no headers, no data), falling back to empty frame", "white")) + df = pd.DataFrame() # Create an empty DataFrame as fallback + return df + +def deduplicate_list(lst): + seen = set() + return [x for x in lst if not (x in seen or seen.add(x))]