diff --git a/.gitignore b/.gitignore
index 2eea525..7e0cdf3 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1 +1,4 @@
-.env
\ No newline at end of file
+.env
+*.html
+*.csv
+*__pycache__*
\ No newline at end of file
diff --git a/AirlockTools.py b/AirlockTools.py
index 0995fe8..4ad9e9f 100644
--- a/AirlockTools.py
+++ b/AirlockTools.py
@@ -1,29 +1,302 @@
+# 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 .
+
import dotenv
import os
import utils.getdeviceevents
+import utils.allowlist
+import utils.hashfunctions
+import utils.pathfunctions
+import utils.allowfunctions
+import utils.colortext as ct
+import urllib3
+import pandas as pd
-
+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
+
def apivalidation():
- print("=== Welcome to the Airlock API Tool ===")
+ 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("Please add your API Key to the .env file")
+ print(ct.colorText("Please add your API Key to the .env file", "red"))
case _:
- menu()
+ menu_main()
-def menu():
- print("\n--- Main Menu ---")
- print("1. Get All Events for Single Device")
- print("2. Policy Enforcement Readiness")
- print("3. Get Device with Highest Blocks in 7 Days")
+def menu_main():
while True:
- choice = input("Enter Menu Item: ")
+ 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)
+ 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 = " "
+
+ #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("approvals"): os.makedirs("approvals")
+
+ 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"))
+
+ 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. Pull and stage event history", "cyan"))
+
+ if os.path.exists(f"dataframe_csv\\df_aggregated_{first_policy}.csv") == True:
+ print(ct.colorText(f" [✓] This has been completed for {first_policy}","green"))
+ elif os.path.exists(f"dataframe_csv\\df_aggregated_{first_policy}.csv") == False:
+ print(ct.colorText(f" [✗] This step has not been completed","red"))
+ elif second_policy is not first_policy and os.path.exists(f"dataframe_csv\\df_aggregated_{second_policy}.csv") == True:
+ print(ct.colorText(f" [✓] This has been completed for {second_policy}","green"))
+ elif second_policy is not first_policy and os.path.exists(f"dataframe_csv\\df_aggregated_{second_policy}.csv") == False:
+ print(ct.colorText(f" [✓] This has not been completed for {second_policy}","red"))
+
+ print(ct.colorText("3. Combine Staged policies", "cyan"))
+
+ if os.path.exists(f"dataframe_csv\\df_aggregated_combo_{first_policy}_{second_policy}.csv") == True:
+ print(ct.colorText(" [✓] This step has been completed","green"))
+ else:
+ print(ct.colorText(" [✗] This step has not been completed","red"))
+
+ print(ct.colorText("4. Add hash threat information to list of executions", "cyan"))
+
+ if os.path.exists(f"dataframe_csv\\df_augmented_combo_{first_policy}_{second_policy}.csv") == True:
+ print(ct.colorText(" [✓] This step has been completed","green"))
+ else:
+ print(ct.colorText(" [✗] This step has not been completed","red"))
+
+ print(ct.colorText("5. Determine if path exclusions are possible", "cyan"))
+
+ if os.path.exists(f"dataframe_csv\\df_path_eligible_{first_policy}_{second_policy}.csv") == True:
+ print(ct.colorText(" [✓] This step has been completed","green"))
+ else:
+ print(ct.colorText(" [✗] This step has not been completed", "red"))
+
+ print(ct.colorText("6. Categorize your hashes ", "cyan"))
+
+ if os.path.isfile(f"dataframe_csv\\df_hashes_needing_approval_{first_policy}_{second_policy}.csv") and os.path.isfile(f"dataframe_csv\\df_automatically_approved_hashes_{first_policy}_{second_policy}.csv") and os.path.isfile(f"dataframe_csv\\df_unapproved_hashes__{first_policy}_{second_policy}.csv"):
+ print(ct.colorText(" [✓] This step has been completed","green"))
+ else:
+ print(ct.colorText(" [✗] This step has not been completed","red"))
+
+ print(ct.colorText("7. Compare potential path exclusions with allowed hashes", "cyan"))
+
+ if os.path.exists(f"dataframe_csv\\df_allowed_paths_{first_policy}_{second_policy}.csv") == True:
+ print(ct.colorText(" [✓] This step has been completed","green"))
+ else:
+ print(ct.colorText(" [✗] This step has not been completed","red"))
+
+
+ print(ct.colorText("Q. Quit", "cyan"))
+
+ choice = input(ct.colorText("\nEnter your choice: ", "white"))
+ if choice == "1":
+ first_policy_tuple = utils.allowlist.listPolicies(url)
+ first_policy = first_policy_tuple[1][first_policy_tuple[0]]
+ 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"):
+ second_policy_tuple = utils.allowlist.listPolicies(url)
+ second_policy = second_policy_tuple[1][second_policy_tuple[0]]
+ break
+ elif answer in ("no", "n"):
+ second_policy_tuple = first_policy_tuple
+ second_policy = first_policy
+ break
+ else:
+ print(ct.colorText("Please answer with 'yes' or 'no'.", "red"))
+
+ elif choice == "2":
+ if not os.path.exists("dataframe_csv\\df_aggregated_{first_policy}.csv"):
+ 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)
+ print(ct.colorText(f"Staging of Exection history for policy: {first_policy} is complete","green"))
+
+ if not os.path.exists("dataframe_csv\\df_aggregated_{second_policy}.csv"):
+ 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)
+ print(ct.colorText(f"Staging of Exection history for policy: {second_policy} is complete","green"))
+
+ elif choice == "3":
+ if second_policy is first_policy and os.path.exists(f"dataframe_csv\\df_aggregated_{first_policy}.csv"):
+ df1 = tryToReadCSV(f"dataframe_csv\\df_aggregated_{first_policy}.csv")
+ df_aggregated_combo = df1
+ 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)
+ elif os.path.exists(f"dataframe_csv\\df_aggregated_{first_policy}.csv") and os.path.exists(f"dataframe_csv\\df_aggregated_{second_policy}.csv"):
+ 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)
+ else:
+ print(ct.colorText(f"Please stage your data before attempting this step","red"))
+
+ elif choice == "4":
+ if os.path.exists(f"dataframe_csv\\df_aggregated_combo_{first_policy}_{second_policy}.csv"):
+ 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)
+ print(ct.colorText(f"Hash reputation info added to dataframe","green"))
+ else:
+ print(ct.colorText(f"Please combine your data with step 3 prior to attempting this step","red"))
+
+ elif choice == "5":
+ if os.path.exists(f"dataframe_html\\df_augmented_combo_{first_policy}_{second_policy}.html"):
+ 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)
+ print(ct.colorText(f"Eligible paths determined","green"))
+ else:
+ print(ct.colorText(f"Please Augment your data with hash threat info using step 4 prior to attempting this step","red"))
+
+ elif choice == "6":
+ if os.path.exists(f"dataframe_csv\\df_augmented_combo_{first_policy}_{second_policy}.csv"):
+ categorized = utils.hashfunctions.categorizeHashes(pd.read_csv(f"dataframe_csv\\df_augmented_combo_{first_policy}_{second_policy}.csv"), threat_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_unapproved_hashes__{first_policy}_{second_policy}.html", index=False)
+ categorized[2].to_csv(f"dataframe_csv\\df_unapproved_hashes__{first_policy}_{second_policy}.csv", index=False)
+ print(ct.colorText(f"Hashes have been categorized","green"))
+ else:
+ print(ct.colorText(f"Please Augment your data with hash threat info using step 4 prior to attempting this step","red"))
+
+ elif choice == "7":
+ if os.path.exists(f"dataframe_csv\\df_hashes_needing_approval_{first_policy}_{second_policy}.csv") and os.path.exists(f"dataframe_csv\\df_automatically_approved_hashes_{first_policy}_{second_policy}.csv") and os.path.exists(f"dataframe_csv\\df_unapproved_hashes__{first_policy}_{second_policy}.csv"):
+ 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)
+ print(ct.colorText(f"Allowable paths determined","green"))
+ else:
+ print(ct.colorText(f"Please complete step 6 prior to attempting this step","red"))
+
+ elif choice == "Q":
+ break
+
+ else:
+ print(ct.colorText("Invalid choice. Please try again.", "red"))
+
+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("Data loaded successfully.", "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
+
+
if __name__ == "__main__":
- apivalidation()
\ No newline at end of file
+ apivalidation()
+
diff --git a/LICENSE.md b/LICENSE.md
index 0ad25db..2b80397 100644
--- a/LICENSE.md
+++ b/LICENSE.md
@@ -1,7 +1,7 @@
GNU AFFERO GENERAL PUBLIC LICENSE
Version 3, 19 November 2007
- Copyright (C) 2007 Free Software Foundation, Inc.
+ Copyright (C) 2025 James Brotosky, Brandon Wickline
Everyone is permitted to copy and distribute verbatim copies
of this license document, but changing it is not allowed.
diff --git a/README.md b/README.md
index ec65353..86edfdb 100644
--- a/README.md
+++ b/README.md
@@ -8,3 +8,10 @@ Python based Carbon Black App Control feature implementation for Airlock
- "Local Approval Initialization"
This programmatically scans devices in audit mode within Airlock and subsequently adds the identified blocks to a user-specified whitelist.
+## License
+**AirlockTools** is licensed under the **GNU Affero General Public License v3.0**.
+
+You may copy, distribute, and modify the software under the terms of the AGPL-3.0 license.
+
+See the [LICENSE](LICENSE.md) file for full details, or visit
+[https://www.gnu.org/license/agpl-3.0.html](https://www.gnu.org/license/agpl-3.0.html)
\ No newline at end of file
diff --git a/requirements.txt b/requirements.txt
index 4de820b..3ba3d18 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,2 +1,4 @@
+pandas==2.3.2
python-dotenv==1.1.1
Requests==2.32.5
+urllib3==2.5.0
diff --git a/utils/__pycache__/getdeviceevents.cpython-313.pyc b/utils/__pycache__/getdeviceevents.cpython-313.pyc
deleted file mode 100644
index 3b24f94..0000000
Binary files a/utils/__pycache__/getdeviceevents.cpython-313.pyc and /dev/null differ
diff --git a/utils/allowfunctions.py b/utils/allowfunctions.py
new file mode 100644
index 0000000..3ec4959
--- /dev/null
+++ b/utils/allowfunctions.py
@@ -0,0 +1,16 @@
+import pandas as pd
+
+def filter_and_drop(approved, eligiblepaths, min_hashes):
+ """
+ Filters eligiblepaths to rows where all hashes are in approved,
+ then drops rows with fewer than min_hashes hashes.
+ """
+ approved_hashes = set(approved['sha256'])
+
+ def all_hashes_approved(row):
+ return all(h in approved_hashes for h in row['sha256'])
+
+ filtered = eligiblepaths[eligiblepaths.apply(all_hashes_approved, axis=1)]
+ filtered = filtered[filtered['sha256'].apply(len) >= min_hashes]
+
+ return filtered
diff --git a/utils/allowlist.py b/utils/allowlist.py
new file mode 100644
index 0000000..6cc6e2d
--- /dev/null
+++ b/utils/allowlist.py
@@ -0,0 +1,76 @@
+# 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 .
+import datetime
+import requests
+import json
+import os
+import time
+import utils.colortext as ct
+
+def pullPolicyExechistories(url, choice, policiesnames, outputjson: bool):
+
+ headers = {
+ "X-APIKey": os.getenv('APIKEY')
+ }
+ checkpoint = '000000000000000000000000'
+ json_output = {'error': 'Success', 'response': {'exechistories': []}}
+ while True:
+ json_response_data = checkpoint_stomper(checkpoint, url, policiesnames[choice], headers)
+ if not json_response_data['response']['exechistories']:
+ break
+ for index, item in enumerate(json_response_data['response']['exechistories']):
+ if index == len(json_response_data['response']['exechistories']) -1:
+ checkpoint = item['checkpoint']
+ print(ct.colorText(f"Date Greater than 30 Days, Stepping to new Checkpoint. {item['checkpoint']}", "blue"))
+ else:
+ if (datetime.date.today() - datetime.timedelta(days=10) > datetime.datetime.strptime(item['datetime'].replace(' +0000 UTC', ''), '%Y-%m-%dT%H:%M:%SZ').date()):
+ pass
+ else:
+ for output in json_response_data['response']['exechistories']:
+ json_output['response']['exechistories'].append(output)
+ json_output = json.dumps(json_output)
+ if outputjson == True:
+ return json_output
+
+def checkpoint_stomper(checkpoint, url, policy, headers):
+ endpoint = url + '/v1/logging/exechistories'
+ payload_dict = {
+ "type":[1,2,6,7],
+ "checkpoint": checkpoint,
+ "policy": [policy]
+ }
+ payload = json.dumps(payload_dict)
+ response = requests.request("POST", endpoint, headers=headers, data=payload, verify=False)
+ parse_text = json.loads(response.text)
+ return parse_text
+
+def listPolicies(url):
+ endpoint = url + '/v1/group'
+ print(ct.colorText("[+] Grabbing All Policies", "cyan"))
+ 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(ct.colorText(f"{index}. {list['name']}", "yellow"))
+ policiesnames.append(list['name'])
+ policyids.append(list['groupid'])
+ choice = input(ct.colorText("Select Policy Group: ", "white"))
+ choice = int(choice) - 1
+ return choice, policiesnames
\ No newline at end of file
diff --git a/utils/colortext.py b/utils/colortext.py
new file mode 100644
index 0000000..fcd101c
--- /dev/null
+++ b/utils/colortext.py
@@ -0,0 +1,14 @@
+
+def colorText(text: str, color: str) -> str:
+ colors = {
+ "red": "\033[91m",
+ "green": "\033[92m",
+ "yellow": "\033[93m",
+ "blue": "\033[94m",
+ "magenta": "\033[95m",
+ "cyan": "\033[96m",
+ "white": "\033[97m",
+ "reset": "\033[0m"
+ }
+
+ return f"{colors.get(color, colors['reset'])}{text}{colors['reset']}"
\ No newline at end of file
diff --git a/utils/getdeviceevents.py b/utils/getdeviceevents.py
index d9261ae..85bd8dd 100644
--- a/utils/getdeviceevents.py
+++ b/utils/getdeviceevents.py
@@ -1,17 +1,32 @@
+# 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 .
import datetime
import requests
import json
import os
+import utils.colortext as ct
-def devicehistory(url):
+def devicehistory(url, outputjson: bool):
endpoint = url + '/v1/getexechistory'
print("\n")
- print("1. Today")
- print("2. Last 24 Hours")
- print("3. Past 7 Days")
- print("4. Past 30 Days")
- print("5. Custom Date Range")
- choice = input("\nSelect Date Range: ")
+ print(ct.colorText("1. Today", "yellow"))
+ print(ct.colorText("2. Last 24 Hours", "yellow"))
+ print(ct.colorText("3. Past 7 Days", "yellow"))
+ print(ct.colorText("4. Past 30 Days", "yellow"))
+ print(ct.colorText("5. Custom Date Range","yellow"))
+ choice = input(ct.colorText("\nSelect Date Range: ", "white"))
today = datetime.date.today()
today = today.strftime("%Y-%m-%d")
if choice == '1':
@@ -26,29 +41,40 @@ def devicehistory(url):
date_selected = datetime.date.today() - datetime.timedelta(days=30)
date_selected = date_selected.strftime('%Y-%m-%d')
elif choice == "5":
- print("Please Input Dates as YYYY-MM-DD")
- date_selected = input("From: ")
- today = input("Date To: ")
- print("WARNING: Device Name is Case Sensitive")
- device = input("Enter Device Name: ")
+ print(ct.colorText("Please Input Dates as YYYY-MM-DD", "cyan"))
+ date_selected = input(ct.colorText("From: ", "white"))
+ today = input(ct.colorText("Date To: ", "white"))
+ print(ct.colorText("WARNING: Device Name is Case Sensitive", "red"))
+ device = input(ct.colorText("Enter Device Name: ", "white"))
payload_dict = {
"datefrom": date_selected,
"dateto": today,
"hostname": device
}
payload = json.dumps(payload_dict)
- print(payload)
+ print(ct.colorText(payload, "green"))
headers = {
"X-APIKey": os.getenv('APIKEY')
}
response = requests.request("POST", endpoint, headers=headers, data=payload, verify=False)
+
+ if outputjson:
+ return response
+
parse_text = json.loads(response.text)
- for block in parse_text['response']['exechistory']:
- print(f"Command: {block['commandline']}")
- print(f"Date: {block['datetime']}")
- print(f"Filename: {block['filename']}")
- print(f"Policy Name: {block['policyname']}")
- print(f"Hostname: {block['hostname']}")
- print(f"Hash: {block['sha256']}")
- print("\n")
\ No newline at end of file
+
+ # Safely get exechistory
+ exechistory = parse_text.get('response', {}).get('exechistory')
+
+ if isinstance(exechistory, list):
+ for block in exechistory:
+ print(ct.colorText(f"Command: {block.get('commandline', 'N/A')}", "green"))
+ print(ct.colorText(f"Date: {block.get('datetime', 'N/A')}", "green"))
+ print(ct.colorText(f"Filename: {block.get('filename', 'N/A')}", "green"))
+ print(ct.colorText(f"Policy Name: {block.get('policyname', 'N/A')}", "green"))
+ print(ct.colorText(f"Hostname: {block.get('hostname', 'N/A')}", "green"))
+ print(ct.colorText(f"Hash: {block.get('sha256', 'N/A')}", "green"))
+ print("\n")
+ else:
+ print(ct.colorText("No execution history found or data is not in expected format.", "red"))
\ No newline at end of file
diff --git a/utils/hashfunctions.py b/utils/hashfunctions.py
new file mode 100644
index 0000000..c193284
--- /dev/null
+++ b/utils/hashfunctions.py
@@ -0,0 +1,127 @@
+# 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 .
+import pandas as pd
+import requests
+import os
+import json
+
+
+
+def aggregateHashes(executions_json) -> pd.DataFrame:
+ """
+ Takes the executions, aggregates all the data with sha256 as primary, then returns aggregated dataframe
+ """
+ data = json.loads(executions_json)
+ df = pd.DataFrame(data["response"]["exechistories"])
+
+ if df.empty:
+ return df
+ print(df)
+ # Aggregate by sha256, deduplicate lists, and preserve order
+ agg_df = df.groupby("sha256").agg(lambda x: list(dict.fromkeys(x))).reset_index()
+
+ # Add a column for the number of unique hostnames
+ agg_df["num_devices"] = agg_df["hostname"].apply(len)
+
+ # Sort by num_devices in descending order
+ agg_df = agg_df.sort_values("num_devices", ascending=False)
+
+ return agg_df
+
+def augmentAggregatedHashes(url, agg_df: pd.DataFrame) -> pd.DataFrame:
+ """
+ Takes output of aggregatedHashes, queries API for those hashes, flattens response while keeping one row per hash,
+ aggregate applications and baselines into lists, then merges results back into agg_df to create a
+ """
+ endpoint = url + '/v1/hash/query'
+ payload = {
+ "hashes": agg_df['sha256'].tolist()
+ }
+
+ headers = {"X-APIKey": os.getenv('APIKEY')}
+ payload = json.dumps(payload)
+
+ response = requests.post(endpoint, headers=headers, data=payload, verify=False)
+ data = response.json()
+ results = data.get("response", {}).get("results", [])
+
+ rows = []
+ for res in results:
+ row = {"sha256": res.get("sha256"), "result": res.get("result")}
+
+ if "data" in res:
+ d = res["data"]
+ for key in ["filename", "filepath", "description", "filesize", "md5",
+ "productname", "productversion", "publisher", "createtime", "modtime",
+ "sha128", "sha384", "sha512", "datetime"]:
+ row[key] = d.get(key)
+
+ row["applications"] = d.get("applications", [])
+ row["baselines"] = d.get("baselines", [])
+
+ reputation = d.get("reputation", {})
+ for k, v in reputation.items():
+ row[f"reputation_{k}"] = v
+
+ rows.append(row)
+
+ df_api = pd.DataFrame(rows)
+
+ df = agg_df.merge(df_api, on="sha256", how="left")
+ aug_df = df[['sha256', 'filename_x', 'description', 'productname', 'productversion', 'publisher_y', 'publisher_x', 'netdomain', 'hostname', 'username', 'pprocess', 'gprocess', 'commandline', 'reputation_lastseen', 'reputation_scannercount', 'reputation_scannermatch', 'reputation_status', 'reputation_threatlevel', 'reputation_threatname', 'reputation_timestamp']]
+ return aug_df
+
+
+def categorizeHashes(aug_df: pd.DataFrame, threat_tolerance: int, untrusted_publishers: list):
+ if untrusted_publishers is None:
+ untrusted_publishers = []
+
+ df = aug_df.copy()
+
+ def reputationtool(row):
+ val = row["reputation_scannermatch"]
+ if pd.isna(val) or val == "N/A":
+ return row["publisher_y"] == "Not Signed"
+ try:
+ return int(val) > threat_tolerance
+ except (ValueError, TypeError):
+ return row["publisher_y"] == "Not Signed"
+
+ df["reputation_flag"] = df.apply(reputationtool, axis=1)
+
+ mask_needsreview = (
+ ((df["publisher_y"] == "Not Signed") & df["reputation_flag"]) |
+ (df["reputation_status"] == "UNKNOWN")
+ )
+
+ mask_approved = (
+ (
+ (df["publisher_y"] != "Not Signed") &
+ ~df["publisher_y"].isin(untrusted_publishers) &
+ ~df["reputation_status"].isna()
+ ) |
+ (
+ (df["publisher_y"] == "Not Signed") &
+ ~df["reputation_flag"] &
+ ~df["publisher_y"].isin(untrusted_publishers) &
+ ~df["reputation_status"].isna()
+ )
+ )
+
+ needsreview_df = df[mask_needsreview]
+ approved_df = df[mask_approved]
+ unapproved_df = df[~(mask_needsreview | mask_approved)]
+
+ return needsreview_df, approved_df, unapproved_df
diff --git a/utils/pathfunctions.py b/utils/pathfunctions.py
new file mode 100644
index 0000000..ab8e166
--- /dev/null
+++ b/utils/pathfunctions.py
@@ -0,0 +1,118 @@
+# 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 .
+
+import pandas as pd
+import os
+from itertools import chain
+
+def filepathInitialGroup(df: pd.DataFrame):
+ original_columns = df.columns.tolist()
+
+ # Step 1: Split comma-separated filepaths into lists
+ df["filename_x"] = df["filename_x"].str.split(",")
+
+ # Step 2: Explode the list so each filepath becomes its own row
+ df = df.explode("filename_x", ignore_index=True)
+
+ # Step 3: Clean up whitespace and normalize paths
+ df["filename_x"] = df["filename_x"].str.strip()
+ df["filename_x"] = df["filename_x"].str.replace(r"\\\\", r"\\", regex=True)
+ df["filename_x"] = df["filename_x"].apply(lambda x: os.path.normpath(x) if pd.notna(x) else "")
+
+ # Step 4: Extract directory and filename from each filepath
+ df["directory"] = df["filename_x"].apply(lambda x: os.path.normpath(os.path.dirname(x)) if pd.notna(x) else "")
+ df["filename"] = df["filename_x"].apply(lambda x: os.path.basename(x) if pd.notna(x) else "")
+
+ # Step 5: Drop the original raw filepath column
+ df = df.drop(columns=["filename_x"])
+
+ # Helper functions for path manipulation
+ def get_parts(path):
+ return os.path.normpath(path).split(os.sep)
+
+ def join_parts(parts):
+ return os.path.normpath(os.sep.join(parts))
+
+ def longest_common_prefix(paths):
+ split_paths = [get_parts(p) for p in paths]
+ min_len = min(len(p) for p in split_paths)
+ prefix = []
+ for i in range(min_len):
+ current = split_paths[0][i]
+ if all(p[i] == current for p in split_paths):
+ prefix.append(current)
+ else:
+ break
+ return join_parts(prefix)
+
+ # Step 6: Group directories by shared prefix
+ directories = df["directory"].tolist()
+ groups = []
+ used = set()
+
+ for i, path in enumerate(directories):
+ if path in used:
+ continue
+ group = [path]
+ parts_i = get_parts(path)
+
+ for j in range(i + 1, len(directories)):
+ parts_j = get_parts(directories[j])
+ common = os.path.commonprefix([parts_i, parts_j])
+
+ if (len(parts_i) > 3 and len(common) >= 3) or (len(parts_i) == 3 and len(common) >= 2):
+ group.append(directories[j])
+ used.add(directories[j])
+ elif len(common) == len(parts_i) - 1 and len(parts_i) > 3:
+ group.append(directories[j])
+ used.add(directories[j])
+ used.add(path)
+ groups.append(group)
+
+ # Step 7: Map each original directory to its grouped prefix
+ prefix_map = {dir: longest_common_prefix(group) for group in groups for dir in group}
+ df["grouped_directory"] = df["directory"].map(prefix_map)
+
+ # Step 8: Group the DataFrame by grouped_directory
+ aggregation = {col: (lambda x: list(x)) for col in original_columns if col not in ["filename_x"]}
+ aggregation.update({
+ "directory": lambda x: list(x),
+ "filename": lambda x: list(x)
+ })
+
+ grouped_df = df.groupby("grouped_directory", as_index=False).agg(aggregation)
+
+ # Step 9: Split into eligible and ineligible paths based on depth
+ grouped_df["depth"] = grouped_df["grouped_directory"].apply(lambda x: len(get_parts(x)))
+ path_eligible = grouped_df[grouped_df["depth"] > 2].drop(columns=["depth"])
+ path_ineligible = grouped_df[grouped_df["depth"] <= 2].drop(columns=["depth"])
+
+ # Step 10: Move entries from eligible to ineligible if grouped_directory contains excluded directories
+ mask = path_eligible["grouped_directory"].str.contains(r"(?i)(?:\\Users|\\c\$\\Users|inetpub\\wwwroot|windows\\temp)", na=False)
+ move_to_ineligible = path_eligible[mask]
+ path_eligible = path_eligible[~mask]
+ path_ineligible = pd.concat([path_ineligible, move_to_ineligible], ignore_index=True)
+
+ # Step 11: Deduplicate list elements in all columns
+ def deduplicate_lists(df):
+ for col in df.columns:
+ if df[col].apply(lambda x: isinstance(x, list)).all():
+ df[col] = df[col].apply(lambda x: list({str(item): item for item in chain.from_iterable(x if isinstance(x[0], list) else [x])}.values()))
+ return df
+
+ path_eligible = deduplicate_lists(path_eligible)
+ path_ineligible = deduplicate_lists(path_ineligible)
+
+ return path_eligible, path_ineligible
\ No newline at end of file