=
472 lines
26 KiB
Python
472 lines
26 KiB
Python
# Copyright (C) 2025 James Brotosky, Brandon Wickline
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU Affero General Public License as published
|
|
# by the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU Affero General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
|
|
|
import dotenv
|
|
import json
|
|
import os
|
|
import pandas as pd
|
|
import urllib3
|
|
import utils.allowlist
|
|
import utils.getdeviceevents
|
|
import utils.hashfunctions
|
|
import utils.pathfunctions
|
|
import utils.policyfunctions
|
|
import utils.pretty as ct
|
|
|
|
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
|
|
|
dotenv.load_dotenv()
|
|
|
|
#Constants
|
|
url = os.getenv('url')
|
|
badpublisherlist = ["Brave Software, Inc.", "Zoom Video Communications, Inc."]
|
|
path_exclusion_constant = 3
|
|
min_files_for_path = 4
|
|
threat_tolerance_constant = 4
|
|
|
|
|
|
def apivalidation():
|
|
print(ct.colorText(r"""
|
|
_____ .__ .__ __ ___________ .__
|
|
/ _ \ |__|______| | ____ ____ | | __ \__ ___/___ ____ | | ______
|
|
/ /_\ \| \_ __ \ | / _ \_/ ___\| |/ / | | / _ \ / _ \| | / ___/
|
|
/ | \ || | \/ |_( <_> ) \___| < | |( <_> | <_> ) |__\___ \
|
|
\____|__ /__||__| |____/\____/ \___ >__|_ \ |____| \____/ \____/|____/____ >
|
|
\/ \/ \/ \/
|
|
""", "cyan"))
|
|
print(ct.colorText("=================================================================================", "cyan"))
|
|
print(ct.colorText("======================== Welcome to the Airlock API Tool ========================", "cyan"))
|
|
print(ct.colorText("=================================================================================", "cyan"))
|
|
match os.getenv('APIKEY'):
|
|
case '':
|
|
print(ct.colorText("Please add your API Key to the .env file", "red"))
|
|
case _:
|
|
menu_main()
|
|
|
|
def tryToReadCSV(csv):
|
|
try:
|
|
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 menu_main():
|
|
while True:
|
|
print(ct.colorText("\n-----------------------------------", "magenta"))
|
|
print(ct.colorText("------------ Main Menu ------------", "magenta"))
|
|
print(ct.colorText("-----------------------------------", "magenta"))
|
|
print(ct.colorText("1. Get All Events for Single Device", "yellow"))
|
|
print(ct.colorText("2. Placeholder for Local Approval", "yellow"))
|
|
print(ct.colorText("3. Placeholder for Another Tool", "yellow"))
|
|
print(ct.colorText("4. Prepare Policy For Enforcement", "yellow"))
|
|
print(ct.colorText("Q. Quit", "yellow"))
|
|
|
|
choice = input(ct.colorText("\nEnter Menu Item: ", "white"))
|
|
if choice == '1':
|
|
utils.getdeviceevents.devicehistory(url,False)
|
|
elif choice == "2":
|
|
menu_local_approve()
|
|
elif choice == "3":
|
|
menu_feature2()
|
|
elif choice == "4":
|
|
menu_prepare_to_enforce()
|
|
elif choice == "Q":
|
|
break
|
|
else:
|
|
print(ct.colorText("Invalid choice. Please try again.","red"))
|
|
|
|
def menu_local_approve():
|
|
while True:
|
|
print("\n--- Submenu ---")
|
|
print("1. Sub-option A")
|
|
print("2. Sub-option B")
|
|
print("3. Return to Main Menu")
|
|
choice = input("Enter your choice: ")
|
|
|
|
if choice == "1":
|
|
print("You selected Sub-option A")
|
|
elif choice == "2":
|
|
print("You selected Sub-option B")
|
|
elif choice == "3":
|
|
print("Returning to Main Menu...")
|
|
break
|
|
else:
|
|
print("Invalid choice. Please try again.")
|
|
|
|
def menu_feature2():
|
|
while True:
|
|
print("\n--- Submenu ---")
|
|
print("1. Sub-option A")
|
|
print("2. Sub-option B")
|
|
print("3. Return to Main Menu")
|
|
choice = input("Enter your choice: ")
|
|
|
|
if choice == "1":
|
|
print("You selected Sub-option A")
|
|
elif choice == "2":
|
|
print("You selected Sub-option B")
|
|
elif choice == "3":
|
|
print("Returning to Main Menu...")
|
|
break
|
|
else:
|
|
print("Invalid choice. Please try again.")
|
|
|
|
def menu_prepare_to_enforce():
|
|
|
|
first_policy = " "
|
|
second_policy = " "
|
|
allowlist_parent_name = " "
|
|
allowlist_child_name = " "
|
|
destination_name = " "
|
|
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")
|
|
if not os.path.exists("dataframe_csv"): os.makedirs("dataframe_csv")
|
|
if not os.path.exists("manuallyapproved"): os.makedirs("manuallyapproved")
|
|
if not os.path.exists("preflight"): os.makedirs("preflight")
|
|
|
|
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 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"))
|
|
elif first_policy != " " and second_policy is first_policy:
|
|
print(ct.colorText(f" [✓] {first_policy} has been selected,", "green"))
|
|
elif first_policy != " " and second_policy != " ":
|
|
print(ct.colorText(f" [✓] {first_policy} has been selected as Policy 1","green"))
|
|
print(ct.colorText(f" [✓] {second_policy} has been selected as Policy 2","green"))
|
|
|
|
|
|
print(ct.colorText("2. Pulls and stages event history, combines the histories, adds hash info, then categorizes the hashes", "cyan"))
|
|
|
|
if os.path.exists(f"dataframe_csv\\executionhist_{first_policy}.csv"):
|
|
print(ct.colorText(f" [✓] Execution history has been compiled for {first_policy}","green"))
|
|
elif not os.path.exists(f"dataframe_csv\\executionhist_{first_policy}.csv"):
|
|
print(ct.colorText(f" [✗] Execution history has not been compiled for {first_policy}","red"))
|
|
elif second_policy is not first_policy and os.path.exists(f"dataframe_csv\\executionhist_{second_policy}.csv"):
|
|
print(ct.colorText(f" [✓] Execution history has been compiled for {second_policy}","green"))
|
|
elif second_policy is not first_policy and not os.path.exists(f"dataframe_csv\\executionhist_{second_policy}.csv"):
|
|
print(ct.colorText(f" [✗] Execution history has not been compiled for {second_policy}","red"))
|
|
|
|
if os.path.exists(f"dataframe_csv\\execuctionhist_combined_{first_policy}_{second_policy}.csv"):
|
|
print(ct.colorText(f" [✓] Execution history has been_combined_for {first_policy} and_{second_policy}", "green"))
|
|
else:
|
|
print(ct.colorText(f" [✗] Execution history has not been_combined_for {first_policy} and_{second_policy}", "red"))
|
|
|
|
if os.path.exists(f"dataframe_csv\\augmented_combo_{first_policy}_{second_policy}.csv"):
|
|
print(ct.colorText(f" [✓] Hash Info has been added to the combined execution history", "green"))
|
|
else:
|
|
print(ct.colorText(f" [✗] Hash Info has not been added to the combined execution history", "red"))
|
|
|
|
if os.path.exists(f"dataframe_csv\\hashes_needing_approval_{first_policy}_{second_policy}.csv") and os.path.exists(f"dataframe_csv\\automatically_approved_hashes_{first_policy}_{second_policy}.csv") and os.path.exists(f"dataframe_csv\\unapproved_hashes__{first_policy}_{second_policy}.csv"):
|
|
print(ct.colorText(f" [✓] Hashes have been cateogrized", "green"))
|
|
else:
|
|
print(ct.colorText(f" [✗] Hashes have not been cateogrized", "red"))
|
|
|
|
|
|
print(ct.colorText(f"3. Manually review the files:","cyan"))
|
|
print(ct.colorText(" '\\dataframe_csv\\hashes_needing_approval_{first_policy}_{second_policy}.csv' and 'dataframe_csv\\automatically_approved_hashes_{first_policy}_{second_policy}.csv'", "cyan"))
|
|
print(ct.colorText(" Remove the rows containing hashes you do not approve of, and those you would not approve of without metarules.", "cyan"))
|
|
print(ct.colorText(" If metarules need to be created, please make note of them, and remove the row from the csv.", "cyan"))
|
|
print(ct.colorText(" When complete, save both csv files to the directory 'manuallyapproved' and choose this option.","cyan"))
|
|
print(ct.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"manuallyapproved\\hashes_needing_approval_{first_policy}_{second_policy}.csv") and os.path.exists(f"manuallyapproved\\automatically_approved_hashes_{first_policy}_{second_policy}.csv"):
|
|
print(ct.colorText(" [✓] Reviewed hashes have been loaded","green"))
|
|
else:
|
|
print(ct.colorText(" [✗] Reviewed hashes have not been loaded","red"))
|
|
|
|
if os.path.exists(f"dataframe_csv\\all_approved_hashes_{first_policy}_{second_policy}.csv"):
|
|
print(ct.colorText(" [✓] The combined approved hashes list has been generated","green"))
|
|
else:
|
|
print(ct.colorText(" [✗] The combined approved hashes list has not been generated","red"))
|
|
|
|
if os.path.exists(f"dataframe_csv\\allinfo_{first_policy}_{second_policy}.csv"):
|
|
print(ct.colorText(" [✓] Longest common filepaths have been generated and appended to hash info","green"))
|
|
else:
|
|
print(ct.colorText(" [✗] Longest common filepaths have not been generated","red"))
|
|
|
|
if os.path.exists(f"dataframe_html\\lcf_needs_appoved_{first_policy}_{second_policy}.html"):
|
|
print(ct.colorText(" [✓] Path review list created","green"))
|
|
else:
|
|
print(ct.colorText(" [✗] Path review list has not been created","red"))
|
|
|
|
|
|
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'", "cyan"))
|
|
print(ct.colorText(" Preflight Lists will be generated", "cyan"))
|
|
|
|
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(" [✗] 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_name policy and parent and child allow list", "cyan"))
|
|
if allowlist_child_name == " " and allowlist_parent_name== " ":
|
|
print(ct.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(ct.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(ct.colorText(f" [✓] {allowlist_parent_name} has been selected as Parent Policy","green"))
|
|
print(ct.colorText(f" [✓] {allowlist_child_name} has been selected as Child Policy","green"))
|
|
if destination_name == " ":
|
|
print(ct.colorText(f" [✗] No destination_name policy has been chosen","red"))
|
|
else:
|
|
print(ct.colorText(f" [✓] destination_name policy is {destination_name}","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"))
|
|
|
|
choice = input(ct.colorText("\nEnter your choice: ", "white"))
|
|
|
|
if choice == "1":
|
|
|
|
choice, policynames, policyid = utils.allowlist.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.allowlist.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":
|
|
|
|
if not os.path.exists(f"dataframe_csv\\executionhist_{first_policy}.csv"):
|
|
print(choice)
|
|
print(first_policy)
|
|
|
|
|
|
exe1 = utils.allowlist.pullPolicyExechistories(url, choice, first_policy ,True)
|
|
print(exe1)
|
|
data = json.loads(exe1)
|
|
executionhist_policy1 = pd.DataFrame(data["response"]["exechistories"])
|
|
|
|
if not executionhist_policy1.empty:
|
|
executionhist_policy1 = executionhist_policy1.sort_values(by=['sha256','filename'])
|
|
|
|
executionhist_policy1.to_csv(f"dataframe_csv\\executionhist_{first_policy}.csv", index=False)
|
|
ct.style_dataframe_dark(executionhist_policy1, f"dataframe_html\\executionhist_{first_policy}.html")
|
|
print(ct.colorText(f"Staging of Execution history for policy: {first_policy} is complete","green"))
|
|
|
|
if not os.path.exists(f"dataframe_csv\\executionhist_{second_policy}.csv"):
|
|
exe2 = utils.allowlist.pullPolicyExechistories(url,choice, second_policy,True)
|
|
data2 = json.loads(exe2)
|
|
executionhist_policy2 = pd.DataFrame(data2["response"]["exechistories"])
|
|
|
|
if not executionhist_policy2.empty:
|
|
executionhist_policy2 = executionhist_policy2.sort_values(by=['sha256','filename'])
|
|
|
|
executionhist_policy2.to_csv(f"dataframe_csv\\executionhist_{second_policy}.csv", index=False)
|
|
ct.style_dataframe_dark(executionhist_policy2, f"dataframe_html\\executionhist_{second_policy}.html")
|
|
print(ct.colorText(f"Staging of Exection history for policy: {first_policy} is complete","green"))
|
|
|
|
#Combine the two policies execution histories
|
|
|
|
if second_policy is first_policy:
|
|
execuctionhist_combined = executionhist_policy1
|
|
execuctionhist_combined.to_csv(f"dataframe_csv\\execuctionhist_combined_{first_policy}_{second_policy}.csv", index=False)
|
|
ct.style_dataframe_dark(df_aggregated_combo, f"dataframe_html\\execuctionhist_combined_{first_policy}_{second_policy}.html")
|
|
print(ct.colorText(f"Dataframes have been combined","green"))
|
|
|
|
elif os.path.exists(f"dataframe_csv\\executionhist_{first_policy}.csv") and os.path.exists(f"dataframe_csv\\executionhist_{second_policy}.csv"):
|
|
execuctionhist_combined = pd.concat([tryToReadCSV(f"dataframe_csv\\executionhist_{first_policy}.csv") , tryToReadCSV(f"dataframe_csv\\executionhist_{second_policy}.csv")], ignore_index=True).sort_values(by=['sha256','filename'])
|
|
execuctionhist_combined.to_csv(f"dataframe_csv\\execuctionhist_combined_{first_policy}_{second_policy}.csv", index=False)
|
|
ct.style_dataframe_dark(df_aggregated_combo, f"dataframe_html\\execuctionhist_combined_{first_policy}_{second_policy}.html")
|
|
print(ct.colorText(f"Dataframes have been combined","green"))
|
|
|
|
#Keep only unique combinations of hash, filename, and hostname
|
|
if f"dataframe_csv\\execuctionhist_combined_{first_policy}_{second_policy}.csv":
|
|
unique_executions = tryToReadCSV(f"dataframe_csv\\execuctionhist_combined_{first_policy}_{second_policy}.csv").drop_duplicates(subset=['sha256', 'filename', 'hostname'])
|
|
unique_executions.to_csv(f"dataframe_csv\\unique_executions{first_policy}_{second_policy}.csv")
|
|
ct.style_dataframe_dark(unique_executions, f"dataframe_html\\unique_execuctions.html")
|
|
|
|
|
|
#Add Hash info to the combined execution history
|
|
if not os.path.exists(f"dataframe_html\\augmented_combo_{first_policy}_{second_policy}.html"):
|
|
print(ct.colorText(f"Preparing to pull hash info","green"))
|
|
augmented_combo= utils.hashfunctions.augmentAggregatedHashes(url,tryToReadCSV(f"dataframe_csv\\unique_executions{first_policy}_{second_policy}.csv"))
|
|
augmented_combo.to_csv(f"dataframe_csv\\augmented_combo_{first_policy}_{second_policy}.csv", index=False)
|
|
ct.style_dataframe_dark(augmented_combo, f"dataframe_html\\augmented_combo_{first_policy}_{second_policy}.html")
|
|
print(ct.colorText(f"Hash reputation info added to dataframe","green"))
|
|
|
|
#Categorize the hashes
|
|
if os.path.exists(f"dataframe_csv\\hashes_needing_approval_{first_policy}_{second_policy}.csv") and os.path.exists(f"dataframe_csv\\automatically_approved_hashes_{first_policy}_{second_policy}.csv") and os.path.exists(f"dataframe_csv\\unapproved_hashes__{first_policy}_{second_policy}.csv"):
|
|
break
|
|
else:
|
|
categorized = utils.hashfunctions.categorizeHashes(pd.read_csv(f"dataframe_csv\\augmented_combo_{first_policy}_{second_policy}.csv"), threat_tolerance_constant, badpublisherlist)
|
|
|
|
if not categorized[0].empty:
|
|
categorized[0].sort_values(by=['sha256','filename_x'])
|
|
|
|
categorized[0].to_csv(f"dataframe_csv\\hashes_needing_approval_{first_policy}_{second_policy}.csv", index=False)
|
|
ct.style_dataframe_dark(categorized[0], f"dataframe_html\\dashes_needing_approval_{first_policy}_{second_policy}.html")
|
|
|
|
if not categorized[1].empty:
|
|
categorized[1].sort_values(by=['sha256','filename_x'])
|
|
|
|
categorized[1].to_csv(f"dataframe_csv\\automatically_approved_hashes_{first_policy}_{second_policy}.csv", index=False)
|
|
ct.style_dataframe_dark(categorized[1], f"dataframe_html\\automatically_approved_hashes_{first_policy}_{second_policy}.html")
|
|
|
|
if not categorized[2].empty:
|
|
categorized[2].sort_values(by=['sha256','filename_x'])
|
|
|
|
categorized[2].to_csv(f"dataframe_csv\\unapproved_hashes__{first_policy}_{second_policy}.csv", index=False)
|
|
ct.style_dataframe_dark(categorized[2], f"dataframe_html\\unapproved_hashes_{first_policy}_{second_policy}.html")
|
|
|
|
print(ct.colorText(f"Hashes have been categorized","green"))
|
|
|
|
|
|
elif choice == "3":
|
|
|
|
if os.path.exists(f"manuallyapproved\\hashes_needing_approval_{first_policy}_{second_policy}.csv") and os.path.exists(f"manuallyapproved\\automatically_approved_hashes_{first_policy}_{second_policy}.csv"):
|
|
df1 = tryToReadCSV(f"manuallyapproved\\hashes_needing_approval_{first_policy}_{second_policy}.csv")
|
|
df2 = tryToReadCSV(f"manuallyapproved\\automatically_approved_hashes_{first_policy}_{second_policy}.csv")
|
|
|
|
all_approved_hashes = pd.concat([df1 , df2], ignore_index=True).sort_values(by=['sha256','filename_x'])
|
|
print(ct.colorText(f"Approved hash lists have been combined","green"))
|
|
all_approved_hashes.to_csv(f"dataframe_csv\\all_approved_hashes_{first_policy}_{second_policy}.csv", index=False)
|
|
ct.style_dataframe_dark(all_approved_hashes, f"dataframe_html\\all_approved_hashes_{first_policy}_{second_policy}.html")
|
|
|
|
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,"filename_x","longestcfp",min_files_for_path,path_exclusion_constant)
|
|
|
|
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")
|
|
|
|
forbidden_lcfp = grouped_df_view["longestcfp"].str.contains(r"(?i)(?:\\Users|\\c\$\\Users|inetpub\\wwwroot|windows\\temp)", na=False)
|
|
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_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":
|
|
|
|
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")
|
|
allowbyhash = utils.pathfunctions.mask_from_csv(df1, f"manuallyapproved\\lcf_needs_approved_{first_policy}_{second_policy}.csv","longestcfp")
|
|
pathexclusions = tryToReadCSV(f"manuallyapproved\\lcf_needs_approved_{first_policy}_{second_policy}.csv")
|
|
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)
|
|
|
|
easyview = allowbyhash.groupby('sha256').agg(list).reset_index()
|
|
ct.style_dataframe_dark(easyview, f"preflight\\final_hash_approvals_{first_policy}_{second_policy}.html")
|
|
|
|
elif choice == "5":
|
|
|
|
print(ct.colorText(f"Please choose destination_name Policy for Path Exclusions","white"))
|
|
choice, policynames, policyid = utils.allowlist.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.allowlist.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.allowlist.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}.csv") and os.path.exists(f"preflight\\final_hash_approvals_{first_policy}_{second_policy}.csv") and allowlist_parent_name != " " and allowlist_child_name != " " and destination_name != " ":
|
|
|
|
pathexclusions = tryToReadCSV(f"preflight\\final_path_exclusions_{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"))
|
|
print(ct.colorText(f"*-------------This program will now begin to make changes to the Airlock Console. Do you understand and agree to proceed? ----------------*", "yellow"))
|
|
print(ct.colorText(f"*=========================================================================================================================================*","red"))
|
|
print(ct.colorText(f"*=========================================================================================================================================*","yellow"))
|
|
print(ct.colorText(f"*******************************************************************************************************************************************","red"))
|
|
confirmation = input(ct.colorText("Type 'I AGREE' to continue: ","white"))
|
|
|
|
if confirmation.strip().upper() == "I AGREE":
|
|
print(ct.colorText("Proceeding with the code...", "yellow"))
|
|
print(ct.colorText(f"Adding path exclusions to {destination_name}", "yellow"))
|
|
pathexcludelist = pathexclusions['longestcfp'].unique().tolist()
|
|
utils.policyfunctions.addPath(destination_id,pathexcludelist)
|
|
|
|
print(ct.colorText(f"Adding hashes to {allowlist_parent_name}", "yellow"))
|
|
|
|
allowlist_parenthashlist = allowbyhash[allowbyhash['reputation_status'] == 'KNOWN']['sha256'].unique().tolist()
|
|
utils.policyfunctions.addHash(allowlist_parent_id,allowlist_parenthashlist)
|
|
|
|
|
|
print(ct.colorText(f"Adding hashes to {allowlist_child_name}", "yellow"))
|
|
allowlist_childhashlist = allowbyhash[allowbyhash['reputation_status'] == 'UNKNOWN']['sha256'].unique().tolist()
|
|
utils.policyfunctions.addHash(allowlist_child_id, allowlist_childhashlist)
|
|
|
|
else:
|
|
print(ct.colorText("Operation aborted. You MUST EXPLICITLY AGREE to proceed.", "red"))
|
|
break
|
|
|
|
elif choice == "Q":
|
|
break
|
|
else:
|
|
print(ct.colorText("Invalid choice. Please try again.", "red"))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
apivalidation()
|
|
|