diff --git a/AirlockTools.py b/AirlockTools.py index 30eed4e..2d35e2d 100644 --- a/AirlockTools.py +++ b/AirlockTools.py @@ -19,6 +19,7 @@ import utils.getdeviceevents import utils.allowlist import utils.hashfunctions import utils.pathfunctions +import utils.policyfunctions import utils.pretty as ct import urllib3 import pandas as pd @@ -28,11 +29,10 @@ import ast urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) dotenv.load_dotenv() -url = "https://172.17.22.240:3129" badpublisherlist = ["Brave Software, Inc.", "Zoom Video Communications, Inc."] path_exclusion_constant = 3 threat_tolerance_constant = 4 - +url = os.getenv('url') def apivalidation(): @@ -132,7 +132,9 @@ def menu_prepare_to_enforce(): first_policy = " " second_policy = " " - + parentpolicy = " " + childpolicy = " " + df_aggregated_combo = pd.DataFrame() #If the directorys where we're going to store our output dont exist, make them. if not os.path.exists("dataframe_html"): os.makedirs("dataframe_html") @@ -140,13 +142,12 @@ def menu_prepare_to_enforce(): if not os.path.exists("manuallyapproved"): os.makedirs("manuallyapproved") if not os.path.exists("preflight"): os.makedirs("preflight") - df_aggregated_combo = pd.DataFrame() while True: print(ct.colorText("\n --------------------------------------------------------------------", "cyan")) print(ct.colorText(" -------------------- Prepare to Enforce Policy ---------------------", "cyan")) print(ct.colorText(" --------------------------------------------------------------------", "cyan")) print(ct.colorText("\nSequentually follow these steps to prepare a policy for enforcement:", "white")) - print(ct.colorText("\n1. Choose which policy or policies to work with - : ", "cyan")) + print(ct.colorText("\n1. Choose which originating policy or policies to move to enforcement", "cyan")) if first_policy == " " and second_policy == " ": print(ct.colorText(f" [✗] No policies have been chosen","red")) @@ -209,12 +210,37 @@ def menu_prepare_to_enforce(): print(ct.colorText(f"4. Manually review the file 'paths_needing_review_{first_policy}_{second_policy}.csv'", "cyan")) print(ct.colorText(" Remove the rows containing path exclusions you do not approve of" , "cyan")) - print(ct.colorText(" When complete, save the csv file to the directory 'manuallyapproved' and choose this option to generate the preflight lists", "cyan")) + print(ct.colorText(" When complete, save the csv file to the directory 'manuallyapproved'", "cyan")) + print(ct.colorText(" Preflight Lists will be generated", "cyan")) - if os.path.isfile(f"manuallyapproved\\paths_needing_review_{first_policy}_{second_policy}.csv") and os.path.isfile("dataframe_csv\\hashdestination_{first_policy}_{second_policy}.csv") and os.path.isfile(f"dataframe_csv\\addtochildpolicy_{first_policy}_{second_policy}.csv") and os.path.isfile(f"dataframe_csv\\addtobaseline_{first_policy}_{second_policy}.csv"): - print(ct.colorText(" [✓] This step has been completed","green")) + if os.path.exists(f"manuallyapproved\\lcf_needs_approved_{first_policy}_{second_policy}.csv"): + print(ct.colorText(" [✓] Reviewed path list detected","green")) else: - print(ct.colorText(" [✗] This step has not been completed","red")) + print(ct.colorText(" [✗] Path review list has not been detected","red")) + + if os.path.exists(f"preflight\\final_path_exclusions_{first_policy}_{second_policy}.csv"): + print(ct.colorText(" [✓] Preflight Path Exclusion List has been generated","green")) + else: + print(ct.colorText(" [✗] Preflight Path Exclusion List has not been generated","red")) + + if os.path.exists(f"preflight\\final_hash_approvals_{first_policy}_{second_policy}.csv"): + print(ct.colorText(" [✓] Preflight hash approval list has been generated","green")) + else: + print(ct.colorText(" [✗] Preflight hash approval list has not been generated","red")) + + print(ct.colorText(f"5. Choose the destination parent and child policy allow list", "cyan")) + if childpolicy == " " and parentpolicy == " ": + print(ct.colorText(f" [✗] No policies have been chosen","red")) + elif parentpolicy != " " and childpolicy != " " and parentpolicy is childpolicy: + print(ct.colorText(f" [✓] [✗] Only {parentpolicy} has been selected this is unusual, but potentially valid case, double check before proceeding,", "yellow")) + elif parentpolicy != " " and childpolicy != " " and parentpolicy is not childpolicy: + print(ct.colorText(f" [✓] {parentpolicy} has been selected as Parent Policy","green")) + print(ct.colorText(f" [✓] {childpolicy} has been selected as Child Policy","green")) + + print(ct.colorText(f"6. Liftoff ------------------------------------------------------", "cyan")) + print(ct.colorText(f" Apply path exclusions according to allowed and approved paths", "cyan")) + print(ct.colorText(f" Apply signed or attested hashes to Parent Allow List", "cyan")) + print(ct.colorText(f" Apply approved, but unsigned hashes to the Child Allow List", "cyan")) print(ct.colorText("Q. Quit", "cyan")) @@ -311,7 +337,7 @@ def menu_prepare_to_enforce(): ct.style_dataframe_dark(all_approved_hashes, f"dataframe_html\\all_approved_hashes_{first_policy}_{second_policy}.html") print(ct.colorText(f"Beginning calculating longest common filepaths for path exceptions","green")) - grouped_df_view, df_with_groups_appended = utils.pathfunctions.export_groups_for_review(all_approved_hashes, csv_path="filegroups_review.csv") + grouped_df_view, df_with_groups_appended = utils.pathfunctions.export_groups_for_review(all_approved_hashes) df_with_groups_appended.to_csv(f"dataframe_csv\\allinfo_{first_policy}_{second_policy}.csv", index=False) ct.style_dataframe_dark(df_with_groups_appended, f"dataframe_html\\allinfo_{first_policy}_{second_policy}.html") @@ -320,24 +346,61 @@ def menu_prepare_to_enforce(): grouped_df_view = grouped_df_view[~forbidden_lcfp] print(ct.colorText(f"Removing forbidden filepaths for path exceptions","green")) - grouped_df_view.to_csv(f"dataframe_csv\\lcf_{first_policy}_{second_policy}.csv", index=False) - ct.style_dataframe_dark(grouped_df_view, f"dataframe_html\\lcf_{first_policy}_{second_policy}.html") - - - - - + grouped_df_view.to_csv(f"dataframe_csv\\lcf_needs_approved_{first_policy}_{second_policy}.csv", index=False) + ct.style_dataframe_dark(grouped_df_view, f"dataframe_html\\lcf_needs_appoved_{first_policy}_{second_policy}.html") else: print(ct.colorText(f"Please manually approve hashes prior to this step","red")) - + elif choice == "4": - pass + + if os.path.exists(f"dataframe_csv\\allinfo_{first_policy}_{second_policy}.csv") and os.path.exists(f"manuallyapproved\\lcf_needs_approved_{first_policy}_{second_policy}.csv"): + df1 = tryToReadCSV(f"dataframe_csv\\allinfo_{first_policy}_{second_policy}.csv") + pathexclusions,allowbyhash = utils.pathfunctions.mask_from_csv(df1, f"manuallyapproved\\lcf_needs_approved_{first_policy}_{second_policy}.csv","longestcfp") + + pathexclusions.to_csv(f"preflight\\final_path_exclusions_{first_policy}_{second_policy}.csv", index=False) + ct.style_dataframe_dark(pathexclusions, f"preflight\\final_path_exclusions_{first_policy}_{second_policy}.html") + allowbyhash.to_csv(f"preflight\\final_hash_approvals_{first_policy}_{second_policy}.csv", index=False) + ct.style_dataframe_dark(allowbyhash, f"preflight\\final_hash_approvals_{first_policy}_{second_policy}.html") + elif choice == "5": + + parent_policy_tuple = utils.allowlist.listPolicies(url) + parentpolicy = parent_policy_tuple[1][parent_policy_tuple[0]] + child_policy_tuple = utils.allowlist.listPolicies(url) + childpolicy = child_policy_tuple[1][child_policy_tuple[0]] + elif choice == "6": + if os.path.exists(f"preflight\\final_pathexclusions_{first_policy}_{second_policy}.csv") and os.path.exists(f"preflight\\final_hash_approvals_{first_policy}_{second_policy}.csv") and parentpolicy != " " and childpolicy != " ": + + pathexclusions = tryToReadCSV(f"preflight\\final_pathexclusions_{first_policy}_{second_policy}.csv") + allowbyhash = tryToReadCSV(f"preflight\\final_hash_approvals_{first_policy}_{second_policy}.csv") + + print(ct.colorText(f"===========================================================================================================================================","red")) + print(ct.colorText(f"===========================================================================================================================================","yellow")) + print(ct.colorText(f"===========================================================================================================================================","red")) + confirmation = input("This program will now begin to make changes to the Airlock Console. Do you understand and agree to proceed? Type 'I AGREE' to continue: ") + print(ct.colorText(f"===========================================================================================================================================","red")) + print(ct.colorText(f"===========================================================================================================================================","yellow")) + print(ct.colorText(f"===========================================================================================================================================","red")) + + if confirmation.strip().upper() == "I AGREE": + print(ct.colorText("Proceeding with the code...", "yellow")) + for _, row in pathexclusions.iterrows(): + utils.policyfunctions.addPath(childpolicy,row['Sha256']) + + for _, row in allowbyhash.iterrows(): + if row['reputation_status'] == 'KNOWN': + utils.policyfunctions.addHash(parentpolicy,row['Sha256']) + else: + utils.policyfunctions.addHash(childpolicy,row['Sha256']) + + else: + print(ct.colorText("Operation aborted. You MUST EXPLICITLY AGREE to proceed.", "red")) + break elif choice == "Q": break