brotoskyj
235 lines
10 KiB
Python
235 lines
10 KiB
Python
# 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 <https://www.gnu.org/licenses/>.
|
|
|
|
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 = 3
|
|
treat_tolerance_constant = 4
|
|
|
|
def apivalidation():
|
|
print(r"""
|
|
_____ .__ .__ __ ___________ .__
|
|
/ _ \ |__|______| | ____ ____ | | __ \__ ___/___ ____ | | ______
|
|
/ /_\ \| \_ __ \ | / _ \_/ ___\| |/ / | | / _ \ / _ \| | / ___/
|
|
/ | \ || | \/ |_( <_> ) \___| < | |( <_> | <_> ) |__\___ \ 4
|
|
\____|__ /__||__| |____/\____/ \___ >__|_ \ |____| \____/ \____/|____/____ >
|
|
\/ \/ \/ \/
|
|
""")
|
|
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()
|
|
|