From f2530f8ebd1e70e73802a992d649e0493090cc29 Mon Sep 17 00:00:00 2001 From: = <=> Date: Fri, 22 Aug 2025 11:14:32 -0400 Subject: [PATCH] Revamped Menu & Split Exec History Function --- AirlockTools.py | 197 +++++++++++++++++++++++++++++++-------------- utils/allowlist.py | 41 +++++----- 2 files changed, 161 insertions(+), 77 deletions(-) diff --git a/AirlockTools.py b/AirlockTools.py index 9b84ef7..ca89141 100644 --- a/AirlockTools.py +++ b/AirlockTools.py @@ -9,92 +9,171 @@ import urllib3 import pandas as pd urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) - - dotenv.load_dotenv() -url = "https://172.17.22.240:3129" +url = "https://172.17.22.240:3129" +badpublisherlist = ["Brave Software, Inc.", "Zoom Video Communications, Inc."] +path_exclusion_constant = 5 +treat_tolerance_constant = 4 def apivalidation(): + print(r""" + _____ .__ .__ __ ___________ .__ + / _ \ |__|______| | ____ ____ | | __ \__ ___/___ ____ | | ______ + / /_\ \| \_ __ \ | / _ \_/ ___\| |/ / | | / _ \ / _ \| | / ___/ +/ | \ || | \/ |_( <_> ) \___| < | |( <_> | <_> ) |__\___ \ +\____|__ /__||__| |____/\____/ \___ >__|_ \ |____| \____/ \____/|____/____ > + \/ \/ \/ \/ +""") print("=== Welcome to the Airlock API Tool ===") match os.getenv('APIKEY'): case '': print("Please add your API Key to the .env file") case _: - menu() + menu_main() -def menu(): - print("\n--- Main Menu ---") - print("1. Get All Events for Single Device") - print("2. Get Execution Histories for Allow List") +def menu_main(): while True: + print("\n--- Main Menu ---") + print("1. Get All Events for Single Device") + print("2. Placeholder for Local Approval") + print("3. Placeholder for Another Tool" ) + print("4. Prepare Policy For Enforcement") + print("11. Exit") + choice = input("Enter Menu Item: ") if choice == '1': utils.getdeviceevents.devicehistory(url,False) - if choice == '2': - utils.allowlist.allowlistexechistories(url,False) - if choice == '3': - executionhist = utils.allowlist.allowlistexechistories(url,True) - print(executionhist) - aggregated = utils.hashfunctions.aggregateHashes(executionhist) - print(aggregated) - augmented = utils.hashfunctions.augmentAggregatedHashes(url,aggregated) - print(augmented) + elif choice == "2": + menu_local_approve() + elif choice == "3": + menu_feature2() + elif choice == "4": + menu_prepare_to_enforce() + elif choice == "11": + break + else: + print("Invalid choice. Please try again.") - augmented.to_html("z_augmented_list.html", index=False) +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: ") - badpublisherlist = [] - categorized = utils.hashfunctions.categorizeHashes(augmented, 5, badpublisherlist) - categorized[0].to_html("z_needs_review.html", index=False) - categorized[1].to_html("z_approved_hashes.html", index=False) - categorized[2].to_html("z_remaining.html", index=False) + 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.") - if choice == '4': - html_file = "z_augmented_list.html" - augmented_df = pd.read_html(html_file) - print(augmented_df) - combined_df = pd.concat(augmented_df, ignore_index=True) +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: ") - path_eligible, path_ineligible = utils.pathfunctions.filepathInitialGroup(combined_df) - path_eligible.to_html("z_eligble_paths.html", index=False) - path_ineligible.to_html("z_ineligible_paths.html",index=False) - - if choice == '5': - - executionhist = utils.allowlist.allowlistexechistories(url,True) - #print(executionhist) + 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.") - aggregated = utils.hashfunctions.aggregateHashes(executionhist) - print(aggregated) +def menu_prepare_to_enforce(): + first_policy = "" + second_policy = "" + history_staged = False + hashes_threat_pulled = False + hashes_categorized = False + path_exclusions_calculated = False + allowed_paths_determined = False - augmented = utils.hashfunctions.augmentAggregatedHashes(url,aggregated) - print(augmented) + df_aggregated_combo = pd.DataFrame() + 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)) + print("2. Choose second policy, If an enforcement policy of that type exists, include it here - Currently selected second policy is: " + str(second_policy)) + print("3. Pull and stage event history for the selected policies - This step has been done - " + str(history_staged)) + 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("11. Exit") - augmented.to_html("z_augmented_list.html", index=False) - html_file = "z_augmented_list.html" - augmented_df = pd.read_html(html_file) - combined_df = pd.concat(augmented_df, ignore_index=True) + choice = input("Enter your choice: ") + if choice == "1": + first_policy = utils.allowlist.listPolicies(url) - path_eligible, path_ineligible = utils.pathfunctions.filepathInitialGroup(combined_df) - path_eligible.to_html("z_eligble_paths.html", index=False) - path_ineligible.to_html("z_ineligible_paths.html",index=False) - - badpublisherlist = ["Brave Software, Inc.", "Zoom Video Communications, Inc."] - categorized = utils.hashfunctions.categorizeHashes(augmented, 5, badpublisherlist) - categorized[0].to_html("z_needs_review.html", index=False) - categorized[1].to_html("z_approved_hashes.html", index=False) - categorized[2].to_html("z_remaining.html", index=False) + elif choice == "2": + second_policy = utils.allowlist.listPolicies(url) - allowpaths = utils.allowfunctions.filter_and_drop(categorized[1],path_eligible,4) - allowpaths.to_html("z_allowed_paths.html", index=False) + elif choice == "3": + executionhist_policy1 = utils.allowlist.pullPolicyExechistories(url,first_policy,True) + executionhist_policy2 = utils.allowlist.pullPolicyExechistories(url,second_policy,True) + df_aggregated_policy1 = utils.hashfunctions.aggregateHashes(executionhist_policy1) + df_aggregated_policy2 = utils.hashfunctions.aggregateHashes(executionhist_policy2) + df_aggregated_combo= pd.concat([df_aggregated_policy1, df_aggregated_policy2], ignore_index=True) + df_aggregated_combo.to_html(f"df_aggregated_combo_{first_policy}_{second_policy}.html", index=False) + history_staged = True - + elif choice == "4": + 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) + hashes_threat_pulled = True + else: + print("Data not yet staged, please complete steps 1-3") + elif choice == "5": + 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_exclusions_calculated = True + else: + print("Step 4 not complete, Please complete step 4") - + elif choice == "6": + 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) + hashes_categorized = True + else: + print("Step 5 not complete, Please complete step 5") + + elif choice == "7": + 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) + allowed_paths_determined = True + else: + print("Step 6 not complete, Please complete step 6") + elif choice == "11": + break + else: + print("Invalid choice. Please try again.") - + + + + if __name__ == "__main__": - apivalidation() \ No newline at end of file + apivalidation() + diff --git a/utils/allowlist.py b/utils/allowlist.py index 169c6ba..cdace0a 100644 --- a/utils/allowlist.py +++ b/utils/allowlist.py @@ -4,23 +4,8 @@ import json import os import time -def allowlistexechistories(url, outputjson: bool): - endpoint = url + '/v1/group' - print("[+] Grabbing All Policies") - payload = {} - headers = { - "X-APIKey": os.getenv('APIKEY') - } - response = requests.request("POST", endpoint, headers=headers, data=payload, verify=False) - parse_text = json.loads(response.text) - policiesnames = [] - policyids = [] - for index, list in enumerate(parse_text['response']['groups'], start=1): - print(f"{index}. {list['name']}") - policiesnames.append(list['name']) - policyids.append(list['groupid']) - choice = input("Select Policy Group: ") - choice = int(choice) - 1 +def pullPolicyExechistories(url, choice, outputjson: bool): + checkpoint = '000000000000000000000000' json_output = {'error': 'Success', 'response': {'exechistories': []}} while True: @@ -88,4 +73,24 @@ def checkpoint_stomper(checkpoint, url, policy, headers): # response_fuzzer(checkpoint, url, policyname) - print("Finished") \ No newline at end of file + print("Finished") + + +def listPolicies(url): + endpoint = url + '/v1/group' + print("[+] Grabbing All Policies") + payload = {} + headers = { + "X-APIKey": os.getenv('APIKEY') + } + response = requests.request("POST", endpoint, headers=headers, data=payload, verify=False) + parse_text = json.loads(response.text) + policiesnames = [] + policyids = [] + for index, list in enumerate(parse_text['response']['groups'], start=1): + print(f"{index}. {list['name']}") + policiesnames.append(list['name']) + policyids.append(list['groupid']) + choice = input("Select Policy Group: ") + choice = int(choice) - 1 + return choice \ No newline at end of file