# Copyright (C) 2025 James Brotosky, Brandon Wicklines # # 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 . import dotenv import os import utils.getdeviceevents import utils.allowlist import utils.hashfunctions 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() 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_main() 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) 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.") 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 = " 4" history_pol1_staged = True history_pol2_staged = True history_staged = True hashes_threat_pulled = True hashes_categorized = True path_exclusions_calculated = True allowed_paths_determined = True 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: " + 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("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: ") if choice == "1": first_policy_tuple = utils.allowlist.listPolicies(url) first_policy = first_policy_tuple[1][first_policy_tuple[0]] elif choice == "2": second_policy_tuple = utils.allowlist.listPolicies(url) second_policy = second_policy_tuple[1][second_policy_tuple[0]] 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"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"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 = tryToReadCSV(f"dataframe_csv\\df_aggregated_{first_policy}.csv") df2 = tryToReadCSV(f"dataframe_csv\\df_aggregated_{second_policy}.csv") df_aggregated_combo= pd.concat([df1 , df2], ignore_index=True) 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,tryToReadCSV(f"dataframe_csv\\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_csv(f"dataframe_csv\\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_csv(f"dataframe_csv\\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_csv(f"dataframe_csv\\df_automatically_approved_hashes_{first_policy}_{second_policy}.csv"),tryToReadCSV(f"dataframe_csv\\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") elif choice == "11": break else: print("Invalid choice. Please try again.") def tryToReadCSV(csv): try: df =pd.read_csv(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()