diff --git a/.gitignore b/.gitignore index 5b05e41..7e0cdf3 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,4 @@ .env *.html +*.csv *__pycache__* \ No newline at end of file diff --git a/AirlockTools.py b/AirlockTools.py index 7b11d80..a44c1c2 100644 --- a/AirlockTools.py +++ b/AirlockTools.py @@ -7,6 +7,8 @@ import utils.pathfunctions import utils.allowfunctions import urllib3 import pandas as pd +import sqlite3 +import pathlib urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) dotenv.load_dotenv() @@ -92,6 +94,9 @@ def menu_feature2(): print("Invalid choice. Please try again.") def menu_prepare_to_enforce(): + + first_policy = " " + second_policy = " 4" history_pol1_staged = True history_pol2_staged = True history_staged = True @@ -104,16 +109,15 @@ def menu_prepare_to_enforce(): while True: print("\n--- Prepare to Enforce Policy ---") print("Sequentually follow steps to prepare for policy enforcement") - #print("1. Choose first policy, Typically the audit version of the policy - Currently selected first policy is: " + str([[first_policy[0]])) - #print("2. Choose second policy, If an enforcement policy of that type exists, include it here - Currently selected second policy is: " + str(second_policy[0])) + print("1. Choose first policy, Typically the audit version of the policy - Currently selected first policy is: " + first_policy) + print("2. Choose second policy, If an enforcement policy of that type exists, include it here - Currently selected second policy is: " + second_policy) print("3. Pull and stage event history for the first policy - This step has been done - " + str(history_pol1_staged)) print("4. Pull and stage event history for the second policy - This step has been done - " + str(history_pol2_staged)) print("5. Combine aggregated policies -") - - print("4. Pull Hash info - This step has been done - " + str(hashes_threat_pulled)) - print("5. Determine if path exclusions are possible: - This step has been done - " + str(path_exclusions_calculated)) - print("6. Categorize your hashes - This step has been done - " + str(hashes_categorized)) - print("7. Combine Potential path exclusions and allowed hashes - This step has been done" + str(allowed_paths_determined)) + print("6. Pull Hash info - This step has been done - " + str(hashes_threat_pulled)) + print("7. Determine if path exclusions are possible: - This step has been done - " + str(path_exclusions_calculated)) + print("8. Categorize your hashes - This step has been done - " + str(hashes_categorized)) + print("9. Combine Potential path exclusions and allowed hashes - This step has been done" + str(allowed_paths_determined)) print("11. Exit") choice = input("Enter your choice: ") @@ -128,57 +132,68 @@ def menu_prepare_to_enforce(): elif choice == "3": executionhist_policy1 = utils.allowlist.pullPolicyExechistories(url,first_policy_tuple[0], first_policy_tuple[1],True) df_aggregated_policy1 = utils.hashfunctions.aggregateHashes(executionhist_policy1) - df_aggregated_policy1.to_html(f"df_aggregated_{first_policy}.html", index=False) + df_aggregated_policy1.to_html(f"dataframe_html\\df_aggregated_{first_policy}.html", index=False) + df_aggregated_policy1.to_csv(f"dataframe_csv\\df_aggregated_{first_policy}.csv", index=False) + history_pol1_staged =True elif choice == "4": executionhist_policy2 = utils.allowlist.pullPolicyExechistories(url,second_policy_tuple[0], second_policy_tuple[1],True) df_aggregated_policy2 = utils.hashfunctions.aggregateHashes(executionhist_policy2) - df_aggregated_policy2.to_html(f"df_aggregated_{second_policy}.html", index=False) + df_aggregated_policy2.to_html(f"dataframe_html\\df_aggregated_{second_policy}.html", index=False) + df_aggregated_policy2.to_csv(f"dataframe_csv\\df_aggregated_{second_policy}.csv", index=False) history_pol2_staged =True elif choice == "5": if history_pol1_staged == True & history_pol2_staged == True: - df1 = pd.read_html(f"df_aggregated_{first_policy}.html")[0] - df2 = pd.read_html(f"df_aggregated_{second_policy}.html")[0] + df1 = tryToReadCSV(f"df_aggregated_{first_policy}.csv") + df2 = tryToReadCSV(f"df_aggregated_{second_policy}.csv") df_aggregated_combo= pd.concat([df1 , df2], ignore_index=True) - df_aggregated_combo.to_html(f"df_aggregated_combo_{first_policy}_{second_policy}.html", index=False) + df_aggregated_combo.to_html(f"dataframe_html\\df_aggregated_combo_{first_policy}_{second_policy}.html", index=False) + df_aggregated_combo.to_csv(f"dataframe_csv\\df_aggregated_combo_{first_policy}_{second_policy}.csv", index=False) history_staged = True else: print("Both Policies have to be staged to combine them") elif choice == "6": if history_staged == True: - df_augmented = utils.hashfunctions.augmentAggregatedHashes(url, pd.read_html(f"df_aggregated_combo_{first_policy}_{second_policy}.html")[0]) - df_augmented.to_html(f"df_augmented_combo_{first_policy}_{second_policy}.html", index=False) + df_augmented = utils.hashfunctions.augmentAggregatedHashes(url,tryToReadCSV(f"df_aggregated_combo_{first_policy}_{second_policy}.csv")) + df_augmented.to_html(f"dataframe_html\\df_augmented_combo_{first_policy}_{second_policy}.html", index=False) + df_augmented.to_csv(f"dataframe_csv\\df_augmented_combo_{first_policy}_{second_policy}.csv", index=False) hashes_threat_pulled = True else: print("Data not yet staged, please complete earlier steps") elif choice == "7": if hashes_threat_pulled == True: - path_eligible, path_ineligible = utils.pathfunctions.filepathInitialGroup(pd.read_html(f"df_augmented_combo_{first_policy}_{second_policy}.html")[0]) - path_eligible.to_html(f"df_path_eligible_{first_policy}_{second_policy}.html", index=False) - path_ineligible.to_html(f"df_path_ineligible_{first_policy}_{second_policy}.html", index=False) + path_eligible, path_ineligible = utils.pathfunctions.filepathInitialGroup(pd.read_csv(f"df_augmented_combo_{first_policy}_{second_policy}.csv")) + path_eligible.to_html(f"dataframe_html\\df_path_eligible_{first_policy}_{second_policy}.html", index=False) + path_eligible.to_csv(f"dataframe_csv\\df_path_eligible_{first_policy}_{second_policy}.csv", index=False) + path_ineligible.to_html(f"dataframe_html\\df_path_ineligible_{first_policy}_{second_policy}.html", index=False) + path_ineligible.to_csv(f"dataframe_csv\\df_path_ineligible_{first_policy}_{second_policy}.csv", index=False) path_exclusions_calculated = True else: print("Step 6 not complete, Please complete step 6") elif choice == "8": if path_exclusions_calculated == True: - categorized = utils.hashfunctions.categorizeHashes(pd.read_html(f"df_augmented_combo_{first_policy}_{second_policy}.html")[0], treat_tolerance_constant, badpublisherlist) - categorized[0].to_html(f"df_hashes_needing_approval_{first_policy}_{second_policy}.html", index=False) - categorized[1].to_html(f"df_automatically_approved_hashes_{first_policy}_{second_policy}.html", index=False) - categorized[2].to_html(f"df_remaining_hashes__{first_policy}_{second_policy}.html", index=False) + categorized = utils.hashfunctions.categorizeHashes(pd.read_csv(f"df_augmented_combo_{first_policy}_{second_policy}.csv"), treat_tolerance_constant, badpublisherlist) + categorized[0].to_html(f"dataframe_html\\df_hashes_needing_approval_{first_policy}_{second_policy}.html", index=False) + categorized[0].to_csv(f"dataframe_csv\\df_hashes_needing_approval_{first_policy}_{second_policy}.csv", index=False) + categorized[1].to_html(f"dataframe_html\\df_automatically_approved_hashes_{first_policy}_{second_policy}.html", index=False) + categorized[1].to_csv(f"dataframe_csv\\df_automatically_approved_hashes_{first_policy}_{second_policy}.csv", index=False) + categorized[2].to_html(f"dataframe_html\\df_remaining_hashes__{first_policy}_{second_policy}.html", index=False) + categorized[2].to_csv(f"dataframe_csv\\df_remaining_hashes__{first_policy}_{second_policy}.csv", index=False) hashes_categorized = True else: print("Step 7 not complete, Please complete step 7") elif choice == "9": if hashes_categorized == True: - allowpaths = utils.allowfunctions.filter_and_drop(pd.read_html(f"df_automatically_approved_hashes_{first_policy}_{second_policy}.html")[0], pd.read_html(f"df_path_eligible_{first_policy}_{second_policy}.html")[0], path_exclusion_constant) - allowpaths.to_html(f"df_allowed_paths_{first_policy}_{second_policy}.html", index=False) + allowpaths = utils.allowfunctions.filter_and_drop(pd.read_csv(f"df_automatically_approved_hashes_{first_policy}_{second_policy}.csv"),tryToReadCSV(f"df_path_eligible_{first_policy}_{second_policy}.csv"), path_exclusion_constant) + allowpaths.to_html(f"dataframe_html\\df_allowed_paths_{first_policy}_{second_policy}.html", index=False) + allowpaths.to_csv(f"dataframe_csv\\df_allowed_paths_{first_policy}_{second_policy}.csv", index=False) allowed_paths_determined = True else: print("Step 8 not complete, Please complete step 8") @@ -186,12 +201,18 @@ def menu_prepare_to_enforce(): break else: print("Invalid choice. Please try again.") - - - - - +def tryToReadCSV(csv): + try: + df =tryToReadCSV(csv) + if df.empty: + print("CSV file has headers but no data rows.") + else: + print("Data loaded successfully.") + except pd.errors.EmptyDataError: + print("CSV file is completely empty (no headers, no data).") + df = pd.DataFrame() # Create an empty DataFrame as fallback + return df if __name__ == "__main__": apivalidation() diff --git a/hashtest.py b/hashtest.py deleted file mode 100644 index b0dab88..0000000 --- a/hashtest.py +++ /dev/null @@ -1,5 +0,0 @@ -hashes = '' -while True: - inputhash = input("Hash: ") - hashes = hashes + ',' + inputhash - print(hashes) \ No newline at end of file