mysticmomba/AirlockTools
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

Merge branch 'Zar-Branch'

Browse Source
This commit is contained in:
brotoskyj
2025-08-25 10:55:44 -04:00
parent 88a1bae574 7b83e6bf16
commit 9da036d105
12 changed files with 697 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+3
View File
@@ -1 +1,4 @@
.env
*.html
*.csv
*__pycache__*
AirlockTools.py
+284 -11
View File
@@ -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 <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 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()
LICENSE.md
+1 -1
View File
@@ -1,7 +1,7 @@
GNU AFFERO GENERAL PUBLIC LICENSE
Version 3, 19 November 2007
Copyright (C) 2007 Free Software Foundation, Inc. <https://fsf.org/>
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.
README.md
+7
View File
@@ -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)
requirements.txt
+2
View File
@@ -1,2 +1,4 @@
pandas==2.3.2
python-dotenv==1.1.1
Requests==2.32.5
urllib3==2.5.0
utils/__pycache__/getdeviceevents.cpython-313.pyc
BIN
View File
Binary file not shown.
utils/allowfunctions.py
+16
View File
@@ -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
utils/allowlist.py
+76
View File
@@ -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 <https://www.gnu.org/licenses/>.
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
utils/colortext.py
+14
View File
@@ -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']}"
utils/getdeviceevents.py
+46 -20
View File
@@ -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 <https://www.gnu.org/licenses/>.
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']}")
# 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"))
utils/hashfunctions.py
+127
View File
@@ -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 <https://www.gnu.org/licenses/>.
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
utils/pathfunctions.py
+118
View File
@@ -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 <https://www.gnu.org/licenses/>.
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