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

Minor Changes to Quality of Life Update

Browse Source
This commit is contained in:
brotoskyj
2025-09-03 16:42:56 -04:00
parent c428a4a989
commit 40def73075
1 changed files with 41 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils/allowlist.py
+41 -40
View File
@@ -22,7 +22,7 @@ import os
from bson import ObjectId from bson import ObjectId
import datetime import datetime
import tqdm import tqdm
import time import sys
def pullPolicyExechistories(url, policiesnames, days, outputjson: bool): def pullPolicyExechistories(url, policiesnames, days, outputjson: bool):
file_path = 'chunkinator.json' file_path = 'chunkinator.json'
@@ -35,46 +35,47 @@ def pullPolicyExechistories(url, policiesnames, days, outputjson: bool):
headers = {"X-APIKey": os.getenv('APIKEY')} headers = {"X-APIKey": os.getenv('APIKEY')}
checkpoint = str(skipback(days)) checkpoint = str(skipback(days))
json_output = {'error': 'Success', 'response': {'exechistories': []}} json_output = {'error': 'Success', 'response': {'exechistories': []}}
with tqdm.tqdm(total=100, desc=f"Total of {policiesnames} Complete: ") as pbar: with tqdm.tqdm(file=sys.stdout, leave=True, total=10000, desc=f"Checkpoint Progess: {checkpoint}", colour="blue", initial=1) as filebar:
while True: with tqdm.tqdm(file=sys.stdout, leave=True, total=100, desc=f"Total of {policiesnames} Complete: ") as pbar:
json_response_data = checkpoint_stomper(checkpoint, url, policiesnames, headers) while True:
histories = json_response_data['response']['exechistories'] json_response_data = checkpoint_stomper(checkpoint, url, policiesnames, headers)
if not histories: histories = json_response_data['response']['exechistories']
break filebar.total=len(histories)
array_dividend = max(round(len(histories) / 20), 1) if not histories:
match_found = False
for index, item in enumerate(histories[::array_dividend]):
if (datetime.date.today() - datetime.timedelta(days=days) <= datetime.datetime.strptime(item['datetime'].replace(' +0000 UTC', ''), '%Y-%m-%dT%H:%M:%SZ').date()):
match_found = True
break
if match_found == True:
for index, item in enumerate(tqdm.tqdm(histories, desc=f"Processing events for {checkpoint}", unit="events", colour="blue", initial=1)):
if index == len(histories) - 1:
checkpoint = item['checkpoint']
break 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: else:
if (datetime.date.today() - datetime.timedelta(days=days) > datetime.datetime.strptime(item['datetime'].replace(' +0000 UTC', ''), '%Y-%m-%dT%H:%M:%SZ').date()): combined = json_output['response']['exechistories']
pass for item in combined:
else: json_output['response']['exechistories'].append(item) key = (item.get('sha256'), item.get('filename'), item.get('hostname'))
seen = {} seen[key] = item
if os.path.exists(file_path): deduplicated = list(seen.values())
with open(file_path, 'r') as file: with open(file_path, 'w') as file:
existing_data = json.load(file) json.dump({'error': 'Success', 'response': {'exechistories': deduplicated}}, file)
combined = existing_data['response']['exechistories'] + json_output['response']['exechistories'] json_output['response']['exechistories'].clear()
else: date_diff = datetime.date.today() - datetime.datetime.strptime(item['datetime'].replace(' +0000 UTC', ''), '%Y-%m-%dT%H:%M:%SZ').date()
combined = json_output['response']['exechistories'] percentage_diff = (((days + 10) - date_diff.days) / (days + 10)) * 100
for item in combined: pbar.n = round(percentage_diff)
key = (item.get('sha256'), item.get('filename'), item.get('hostname')) pbar.set_description_str(f"Total of {policiesnames} Complete: ")
seen[key] = item pbar.refresh()
deduplicated = list(seen.values()) filebar.n = 1
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 - date_diff.days) / 60) * 100
pbar.n = round(percentage_diff)
pbar.set_description_str(f"Total of {policiesnames} Complete: ({percentage_diff:.1f}%)")
pbar.refresh
with open(file_path, 'r') as file: with open(file_path, 'r') as file:
final_output = json.load(file) final_output = json.load(file)
os.remove(file_path) os.remove(file_path)
@@ -146,7 +147,7 @@ def skipback(days):
Generate a MongoDB ObjectId for a given number of days ago from today. Generate a MongoDB ObjectId for a given number of days ago from today.
Adds 1 extra day to the input to look further back. Adds 1 extra day to the input to look further back.
""" """
adjusted_days = days + 1 adjusted_days = days + 10
date_days_ago = datetime.datetime.now(datetime.UTC) - datetime.timedelta(days=adjusted_days) date_days_ago = datetime.datetime.now(datetime.UTC) - datetime.timedelta(days=adjusted_days)
timestamp = int(date_days_ago.timestamp()) timestamp = int(date_days_ago.timestamp())
hex_timestamp = format(timestamp, '08x') hex_timestamp = format(timestamp, '08x')