diff --git a/AirlockTools.py b/AirlockTools.py
index 86b1842..ecfdedb 100644
--- a/AirlockTools.py
+++ b/AirlockTools.py
@@ -16,11 +16,8 @@ import argparse
import dotenv
import os
import pandas as pd
-import time
import urllib3
-import utils.allowlist
import utils.clientfunctions
-import utils.getdeviceevents
import utils.hashfunctions
import utils.otpfunctions
import utils.pathfunctions
@@ -112,16 +109,16 @@ def menu_main():
while True:
ct.displayIntro();
print(ct.colorText("1. Get All Events for Single Device", "yellow"))
- print(ct.colorText("2. Placeholder for Local Approval", "yellow"))
+ print(ct.colorText("2. OTP", "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,False)
+ utils.clientfunctions.devicehistory(url,False)
elif choice == "2":
- menu_local_approve()
+ menu_otp()
elif choice == "3":
menu_feature2()
elif choice == "4":
@@ -131,23 +128,25 @@ def menu_main():
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: ")
+def menu_otp():
+ while True:
+ print(ct.colorText("\n--- OTP Submenu ---","cyan"))
+ print(ct.colorText("1. Generate OTP","cyan"))
+ #print(ct.colorText("2. Sub-option B","cyan"))
+ print(ct.colorText("Q. Return to Main Menu","cyan"))
+ 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.")
+ if choice == "1":
+ utils.otpfunctions.generateOTP(url, utils.clientfunctions.findAgentID(url))
+ break
+
+ elif choice == "2":
+ print("You selected Sub-option B")
+ elif choice == "Q":
+ print("Returning to Main Menu...")
+ break
+ else:
+ print("Invalid choice. Please try again.")
def menu_feature2():
while True:
@@ -192,12 +191,12 @@ def menu_prepare_to_enforce():
if choice == "1":
- choice, policynames, policyid = utils.allowlist.listPolicies(url)
+ choice, policynames, policyid = utils.policyfunctions.listPolicies(url)
first_policy = policynames[choice]
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"):
- choice, policynames, policyid = utils.allowlist.listPolicies(url)
+ choice, policynames, policyid = utils.policyfunctions.listPolicies(url)
second_policy = policynames[choice]
break
@@ -250,19 +249,19 @@ def menu_prepare_to_enforce():
elif choice == "5":
print(ct.colorText(f"Please choose destination_name Policy for Path Exclusions","white"))
- choice, policynames, policyid = utils.allowlist.listPolicies(url)
+ choice, policynames, policyid = utils.policyfunctions.listPolicies(url)
#print(allowlist_parent_tuple)
destination_name = policynames[choice]
destination_id = policyid[choice]
print(ct.colorText(f"Please choose Parent Allowlist for Known Hashes","white"))
- choice, allowlists,allowid = utils.allowlist.listAllowlists(url)
+ choice, allowlists,allowid = utils.policyfunctions.listAllowlists(url)
#print(allowlist_parent_tuple)
allowlist_parent_name = allowlists[choice]
allowlist_parent_id = allowid[choice]
print(ct.colorText(f"Please choose Child Allowlist for Less-Known Hashes","white"))
- choice, allowlists, allowid = utils.allowlist.listAllowlists(url)
+ choice, allowlists, allowid = utils.policyfunctions.listAllowlists(url)
#print(allowlist_child_tuple)
allowlist_child_name = allowlists[choice]
allowlist_child_id = allowid[choice]
diff --git a/utils/allowlist.py b/utils/allowlist.py
deleted file mode 100644
index 0f02958..0000000
--- a/utils/allowlist.py
+++ /dev/null
@@ -1,155 +0,0 @@
-# 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.pretty as ct
-import ijson
-import os
-from bson import ObjectId
-import datetime
-import tqdm
-import sys
-
-def pullPolicyExechistories(url, policiesnames, days, outputjson: bool):
- file_path = 'chunkinator.json'
- if not os.path.exists(file_path):
- with open(file_path, 'w') as file:
- json.dump({'error': 'Success', 'response': {'exechistories': []}}, file)
- print(f"File '{file_path}' has been crated.")
- else:
- print(f"File '{file_path}' already exists.")
- headers = {"X-APIKey": os.getenv('APIKEY')}
- checkpoint = str(skipback(days))
- json_output = {'error': 'Success', 'response': {'exechistories': []}}
- with tqdm.tqdm(file=sys.stdout, leave=True, total=10000, desc=f"Checkpoint Progess: {checkpoint}", colour="blue", initial=1) as filebar:
- with tqdm.tqdm(file=sys.stdout, leave=True, total=100, desc=f"Total of {policiesnames} Complete: ") as pbar:
- while True:
- json_response_data = checkpoint_stomper(checkpoint, url, policiesnames, headers)
- histories = json_response_data['response']['exechistories']
- filebar.total=len(histories)
- if not histories:
- break
- match_found = True
- if match_found == True:
- for index, item in enumerate(histories):
- if index == len(histories) - 1:
- checkpoint = item['checkpoint']
- filebar.desc = f"Checkpoint Progress: {checkpoint}"
- break
- else:
- if (datetime.date.today() - datetime.timedelta(days=days) > datetime.datetime.strptime(item['datetime'].replace(' +0000 UTC', ''), '%Y-%m-%dT%H:%M:%SZ').date()):
- pass
- else: json_output['response']['exechistories'].append(item)
- filebar.update(1)
- filebar.refresh()
- seen = {}
- if os.path.exists(file_path):
- with open(file_path, 'r') as file:
- existing_data = json.load(file)
- combined = existing_data['response']['exechistories'] + json_output['response']['exechistories']
- else:
- combined = json_output['response']['exechistories']
- for item in combined:
- key = (item.get('sha256'), item.get('filename'), item.get('hostname'))
- seen[key] = item
- deduplicated = list(seen.values())
- with open(file_path, 'w') as file:
- json.dump({'error': 'Success', 'response': {'exechistories': deduplicated}}, file)
- json_output['response']['exechistories'].clear()
- date_diff = datetime.date.today() - datetime.datetime.strptime(item['datetime'].replace(' +0000 UTC', ''), '%Y-%m-%dT%H:%M:%SZ').date()
- percentage_diff = (((days + 10) - date_diff.days) / (days + 10)) * 100
- pbar.n = round(percentage_diff)
- pbar.set_description_str(f"Total of {policiesnames} Complete: ")
- pbar.refresh()
- filebar.n = 1
- with open(file_path, 'r') as file:
- final_output = json.load(file)
- os.remove(file_path)
- return json.dumps(final_output) if outputjson else None
-
-def checkpoint_stomper(checkpoint, url, policy, headers):
- json_output = {'error': 'Success', 'response': {'exechistories': []}}
- endpoint = url + '/v1/logging/exechistories'
- payload_dict = {
- "type":[1,2,6,7],
- "checkpoint": checkpoint,
- "policy": [policy]
- }
- payload = json.dumps(payload_dict)
- with requests.request("POST", endpoint, headers=headers, data=payload, verify=False, stream=True) as response:
- parser = ijson.items(response.raw, 'response.exechistories.item')
- for item in parser:
- key = (item.get('sha256'), item.get('hostname'))
- if key not in json_output:
- json_output['response']['exechistories'].append(item)
- parse_text = json.loads(json.dumps(json_output))
- 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, policyids
-
-def listAllowlists(url):
- endpoint = url + '/v1/application'
- print(ct.colorText("[+] Grabbing All Allowlists", "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']['applications'], start=1):
- if index >= 38:
- print(ct.colorText(f"{index}. {list['name']}", "yellow"))
- policiesnames.append(list['name'])
- policyids.append(list['applicationid'])
- choice = int(input(ct.colorText("Select allowlist: ", "white")))
- if choice < 38:
- print(ct.colorText("Please only choose an allowlist designed for this use - '38+'","red"))
- elif choice >= 38:
- choice = choice - 38
- return choice, policiesnames, policyids
- #Need else and catch for upper bound
-
-def skipback(days):
- """
- Generate a MongoDB ObjectId for a given number of days ago from today.
- Adds 1 extra day to the input to look further back.
- """
- adjusted_days = days + 10
- date_days_ago = datetime.datetime.now(datetime.UTC) - datetime.timedelta(days=adjusted_days)
- timestamp = int(date_days_ago.timestamp())
- hex_timestamp = format(timestamp, '08x')
- objectid_hex = hex_timestamp + '0000000000000000'
- return ObjectId(objectid_hex)
\ No newline at end of file
diff --git a/utils/clientfunctions.py b/utils/clientfunctions.py
index 323c7a9..486a1e0 100644
--- a/utils/clientfunctions.py
+++ b/utils/clientfunctions.py
@@ -5,6 +5,25 @@ import pandas as pd
import utils.pretty as ct
+def findAgentID(url):
+
+ print(ct.colorText("WARNING: Device Name is Case Sensitive", "red"))
+ hostname = input(ct.colorText("Enter Device Name: ", "white"))
+
+ endpoint = url + '/v1/agent/find'
+ payload = {
+ "hostname" : f"{hostname}"
+ }
+
+ headers = {"X-APIKey": os.getenv('APIKEY')}
+ payload = json.dumps(payload)
+
+
+ response = requests.post(endpoint, headers=headers, data=payload, verify=False)
+ result = json.loads(response.text)
+ data = pd.DataFrame(result["response"]["agents"])
+ return data.loc[0, "agentid"]
+
def getDestAllowlistFromClientID(url, clientid):
#This is dependant on their being an allowlist in the policy containing the name "localapproval"
@@ -76,4 +95,71 @@ def getPolicyName(url, groupid):
result = json.loads(response.text)
data = pd.DataFrame(result["response"]["groups"])
name = data.loc[data['groupid'] == f"{groupid}", 'name'].values[0]
- return name
\ No newline at end of file
+ return name
+import datetime
+import requests
+import json
+import os
+import utils.pretty as ct
+
+def devicehistory(url, outputjson: bool):
+ endpoint = url + '/v1/getexechistory'
+ print("\n")
+ 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':
+ date_selected = today
+ elif choice == '2':
+ date_selected = datetime.date.today() - datetime.timedelta(days=1)
+ date_selected = date_selected.strftime('%Y-%m-%d')
+ elif choice == '3':
+ date_selected = datetime.date.today() - datetime.timedelta(days=7)
+ date_selected = date_selected.strftime('%Y-%m-%d')
+ elif choice == '4':
+ date_selected = datetime.date.today() - datetime.timedelta(days=30)
+ date_selected = date_selected.strftime('%Y-%m-%d')
+ elif choice == "5":
+ 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(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)
+
+ # 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"))
+
diff --git a/utils/getdeviceevents.py b/utils/getdeviceevents.py
deleted file mode 100644
index fbb5d66..0000000
--- a/utils/getdeviceevents.py
+++ /dev/null
@@ -1,80 +0,0 @@
-# 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.pretty as ct
-
-def devicehistory(url, outputjson: bool):
- endpoint = url + '/v1/getexechistory'
- print("\n")
- 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':
- date_selected = today
- elif choice == '2':
- date_selected = datetime.date.today() - datetime.timedelta(days=1)
- date_selected = date_selected.strftime('%Y-%m-%d')
- elif choice == '3':
- date_selected = datetime.date.today() - datetime.timedelta(days=7)
- date_selected = date_selected.strftime('%Y-%m-%d')
- elif choice == '4':
- date_selected = datetime.date.today() - datetime.timedelta(days=30)
- date_selected = date_selected.strftime('%Y-%m-%d')
- elif choice == "5":
- 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(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)
-
- # 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/otpfunctions.py b/utils/otpfunctions.py
index 55e4fb4..d78fe85 100644
--- a/utils/otpfunctions.py
+++ b/utils/otpfunctions.py
@@ -11,10 +11,6 @@ import utils.pathfunctions as pathf
import utils.policyfunctions as policyf
from utils.perstscheduler import register_function, run_once_job, recurring_job, reload_jobs, start_scheduler
-
-
-
-
def getActiveOTP(url):
endpoint = url + f'/v1/otp/usage'
@@ -36,7 +32,6 @@ def getActiveOTP(url):
if not otp.empty:
ct.style_dataframe_dark(otp, f"newest_active_OTP.html")
-
def getOTPActivities(url, otpid):
endpoint = url + f'/v1/otp/activities'
payload = {"otpid": f"{otpid}"}
@@ -137,10 +132,6 @@ def monitorOTP(url, pups):
os.remove(f"OTP\\PARQ\\otp_activities_{pid}.parquet")
del finalhashesadded
-
-
-
-
def addOTPHashes(url, clientid, otpid, pups):
path = f"OTP\\PARQ\\otp_activities_{otpid}.parquet"
activities = pd.read_parquet(path)
@@ -160,7 +151,7 @@ def addOTPHashes(url, clientid, otpid, pups):
# Add hashes to policy
if hashes_to_add:
- policyf.addHashReal(url, allowlist, hashes_to_add)
+ policyf.addHash(url, allowlist, hashes_to_add)
# Update 'hash_added' column
activities["hash_added"] = activities.apply(
@@ -171,3 +162,39 @@ def addOTPHashes(url, clientid, otpid, pups):
# Save the updated DataFrame
activities.to_parquet(path)
+
+def generateOTP(url, agentid):
+ purpose = input(ct.colorText(" Please enter the purpose for the OTP: ", "white"))
+
+ possible_durations = [15, 60, 360, 1440, 10080]
+ duration_selected = " "
+
+ print(ct.colorText("Please select a duration:", "white"))
+ for i, option in enumerate(possible_durations, start=1):
+ print(f"{i}. {option}")
+
+ try:
+ choice = int(input("Enter the number of your choice: "))
+ if 1 <= choice <= len(possible_durations):
+ duration_selected = possible_durations[choice - 1]
+ print(ct.colorText(f"You selected: {duration_selected}", "yellow"))
+ else:
+ print(ct.colorText("Invalid choice.", "red"))
+ except ValueError:
+ print(ct.colorText("Invalid input. Please enter a number.", "red"))
+
+ endpoint = url + '/v1/otp/retrieve'
+ payload = {
+ "duration" : f"{duration_selected}",
+ "agentid" : f"{agentid}",
+ "purpose" : f"{purpose}"
+ }
+
+ headers = {"X-APIKey": os.getenv('APIKEY')}
+ payload = json.dumps(payload)
+
+
+ response = requests.post(endpoint, headers=headers, data=payload, verify=False)
+ result = json.loads(response.text)
+ otpcode = result["response"]["otpcode"]
+ print(ct.colorText(f"The OPT code is: {otpcode}", "yellow"))
\ No newline at end of file
diff --git a/utils/policyfunctions.py b/utils/policyfunctions.py
index f6cfb71..227a48d 100644
--- a/utils/policyfunctions.py
+++ b/utils/policyfunctions.py
@@ -19,7 +19,13 @@ import pandas as pd
import re
import requests
import utils.pretty as ct
-import utils.allowlist
+import datetime
+import requests
+import ijson
+from bson import ObjectId
+import datetime
+import tqdm
+import sys
@@ -69,7 +75,7 @@ def addPathReal(url, grouplistID, pathlist):
def getPolicyInfo(url, policy, days):
executionhist_policy = pd.DataFrame()
- exehist = utils.allowlist.pullPolicyExechistories(url, policy, days, True)
+ exehist = pullPolicyExechistories(url, policy, days, True)
data = json.loads(exehist)
executionhist_policy = pd.DataFrame(data["response"]["exechistories"])
if not executionhist_policy.empty:
@@ -120,4 +126,134 @@ def sendToPolicy(url, first_policy, second_policy, destination_name, destination
else:
print(ct.colorText("Operation aborted. You MUST EXPLICITLY AGREE to proceed.", "red"))
+
+def pullPolicyExechistories(url, policiesnames, days, outputjson: bool):
+ file_path = 'chunkinator.json'
+ if not os.path.exists(file_path):
+ with open(file_path, 'w') as file:
+ json.dump({'error': 'Success', 'response': {'exechistories': []}}, file)
+ print(f"File '{file_path}' has been crated.")
+ else:
+ print(f"File '{file_path}' already exists.")
+ headers = {"X-APIKey": os.getenv('APIKEY')}
+ checkpoint = str(skipback(days))
+ json_output = {'error': 'Success', 'response': {'exechistories': []}}
+ with tqdm.tqdm(file=sys.stdout, leave=True, total=10000, desc=f"Checkpoint Progess: {checkpoint}", colour="blue", initial=1) as filebar:
+ with tqdm.tqdm(file=sys.stdout, leave=True, total=100, desc=f"Total of {policiesnames} Complete: ") as pbar:
+ while True:
+ json_response_data = checkpoint_stomper(checkpoint, url, policiesnames, headers)
+ histories = json_response_data['response']['exechistories']
+ filebar.total=len(histories)
+ if not histories:
+ break
+ match_found = True
+ if match_found == True:
+ for index, item in enumerate(histories):
+ if index == len(histories) - 1:
+ checkpoint = item['checkpoint']
+ filebar.desc = f"Checkpoint Progress: {checkpoint}"
+ break
+ else:
+ if (datetime.date.today() - datetime.timedelta(days=days) > datetime.datetime.strptime(item['datetime'].replace(' +0000 UTC', ''), '%Y-%m-%dT%H:%M:%SZ').date()):
+ pass
+ else: json_output['response']['exechistories'].append(item)
+ filebar.update(1)
+ filebar.refresh()
+ seen = {}
+ if os.path.exists(file_path):
+ with open(file_path, 'r') as file:
+ existing_data = json.load(file)
+ combined = existing_data['response']['exechistories'] + json_output['response']['exechistories']
+ else:
+ combined = json_output['response']['exechistories']
+ for item in combined:
+ key = (item.get('sha256'), item.get('filename'), item.get('hostname'))
+ seen[key] = item
+ deduplicated = list(seen.values())
+ with open(file_path, 'w') as file:
+ json.dump({'error': 'Success', 'response': {'exechistories': deduplicated}}, file)
+ json_output['response']['exechistories'].clear()
+ date_diff = datetime.date.today() - datetime.datetime.strptime(item['datetime'].replace(' +0000 UTC', ''), '%Y-%m-%dT%H:%M:%SZ').date()
+ percentage_diff = (((days + 10) - date_diff.days) / (days + 10)) * 100
+ pbar.n = round(percentage_diff)
+ pbar.set_description_str(f"Total of {policiesnames} Complete: ")
+ pbar.refresh()
+ filebar.n = 1
+ with open(file_path, 'r') as file:
+ final_output = json.load(file)
+ os.remove(file_path)
+ return json.dumps(final_output) if outputjson else None
+
+def checkpoint_stomper(checkpoint, url, policy, headers):
+ json_output = {'error': 'Success', 'response': {'exechistories': []}}
+ endpoint = url + '/v1/logging/exechistories'
+ payload_dict = {
+ "type":[1,2,6,7],
+ "checkpoint": checkpoint,
+ "policy": [policy]
+ }
+ payload = json.dumps(payload_dict)
+ with requests.request("POST", endpoint, headers=headers, data=payload, verify=False, stream=True) as response:
+ parser = ijson.items(response.raw, 'response.exechistories.item')
+ for item in parser:
+ key = (item.get('sha256'), item.get('hostname'))
+ if key not in json_output:
+ json_output['response']['exechistories'].append(item)
+ parse_text = json.loads(json.dumps(json_output))
+ 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, policyids
+
+def listAllowlists(url):
+ endpoint = url + '/v1/application'
+ print(ct.colorText("[+] Grabbing All Allowlists", "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']['applications'], start=1):
+ if index >= 38:
+ print(ct.colorText(f"{index}. {list['name']}", "yellow"))
+ policiesnames.append(list['name'])
+ policyids.append(list['applicationid'])
+ choice = int(input(ct.colorText("Select allowlist: ", "white")))
+ if choice < 38:
+ print(ct.colorText("Please only choose an allowlist designed for this use - '38+'","red"))
+ elif choice >= 38:
+ choice = choice - 38
+ return choice, policiesnames, policyids
+ #Need else and catch for upper bound
+
+def skipback(days):
+ """
+ Generate a MongoDB ObjectId for a given number of days ago from today.
+ Adds 1 extra day to the input to look further back.
+ """
+ adjusted_days = days + 10
+ date_days_ago = datetime.datetime.now(datetime.UTC) - datetime.timedelta(days=adjusted_days)
+ timestamp = int(date_days_ago.timestamp())
+ hex_timestamp = format(timestamp, '08x')
+ objectid_hex = hex_timestamp + '0000000000000000'
+ return ObjectId(objectid_hex)
\ No newline at end of file