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

Merge branch 'MemOpt' of https://git.racooncity.org/brotoskyj/Airlocktools into MemOpt

Browse Source
This commit is contained in:
=
2025-09-02 18:16:31 -04:00
parent 52efd6c2fd fbd785d6ab
commit 6a684d71ec
2 changed files with 33 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
AirlockTools.py
+1 -1
View File
@@ -33,7 +33,7 @@ dotenv.load_dotenv()
#Constants #Constants
url = os.getenv('url') url = os.getenv('url')
badpublisherlist = ["Brave Software, Inc.", "Zoom Video Communications, Inc."] badpublisherlist = ["Brave Software, Inc.", "Zoom Video Communications, Inc."]
badpathparts = ["users", "inet\\wwwroot", "windows\\temp", "windows\\temp", "windows\\task", "windows\\system32", "startup", "windows\\fonts", "Recycle.Bin", "AppData", "programdata"] badpathparts = ["users", "inet\\wwwroot", "windows\\temp", "windows\\task", "windows\\system32", "startup", "windows\\fonts", "Recycle.Bin", "AppData", "programdata"]
path_exclusion_constant = 3 path_exclusion_constant = 3
min_files_for_path = 4 min_files_for_path = 4
threat_tolerance_constant = 4 threat_tolerance_constant = 4
utils/allowlist.py
+31 -53
View File
@@ -23,81 +23,61 @@ from bson import ObjectId
import datetime import datetime
def pullPolicyExechistories(url, policiesnames, days, outputjson: bool): def pullPolicyExechistories(url, policiesnames, days, outputjson: bool):
file_path = 'chunkinator.json' file_path = 'chunkinator.json'
# If chunkintor exists - kill and make new
if os.path.exists(file_path):
os.remove(file_path)
if not os.path.exists(file_path): if not os.path.exists(file_path):
with open(file_path, 'w') as file: with open(file_path, 'w') as file:
json.dump({'error': 'Success', 'response': {'exechistories': []}}, file) json.dump({'error': 'Success', 'response': {'exechistories': []}}, file)
print(f"File '{file_path}' has been created.") print(f"File '{file_path}' has been crated.")
else: else:
print(f"File '{file_path}' already exists.") print(f"File '{file_path}' already exists.")
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': []}}
while True: while True:
json_response_data = checkpoint_stomper(checkpoint, url, policiesnames, headers) json_response_data = checkpoint_stomper(checkpoint, url, policiesnames, headers)
histories = json_response_data['response']['exechistories'] histories = json_response_data['response']['exechistories']
if not histories: if not histories:
break break
array_dividend = max(round(len(histories) / 20), 1)
array_dividend = max(round(len(histories) / 20), 1)
match_found = False match_found = False
for index, item in enumerate(histories[::array_dividend]): for index, item in enumerate(histories[::array_dividend]):
item_date = datetime.datetime.strptime(item['datetime'].replace(' +0000 UTC', ''), '%Y-%m-%dT%H:%M:%SZ').date() if (datetime.date.today() - datetime.timedelta(days=days) <= datetime.datetime.strptime(item['datetime'].replace(' +0000 UTC', ''), '%Y-%m-%dT%H:%M:%SZ').date()):
if item_date >= datetime.date.today() - datetime.timedelta(days):
match_found = True match_found = True
break break
checkpoints_processed = round(len(histories) / array_dividend) checkpoints_processed = round(len(histories) / array_dividend)
print(ct.colorText( if ( index + 1 ) < checkpoints_processed:
f"{index + 1}/{checkpoints_processed} checkpoint(s) processed. " # type: ignore print(ct.colorText(f"{index + 1}/{checkpoints_processed} checkpoint(s) from this execution have been processed with date match. Last Checkpoint: {checkpoint}", "blue"))
f"{'Found with Date Match.' if match_found else ''} Last Checkpoint: {item['checkpoint']}.", "blue")) # type: ignore else:
print(ct.colorText(f"{index + 1}/{checkpoints_processed} checkpoint(s) Processed. Last Checkpoint: {checkpoint}", "blue"))
checkpoint = item['checkpoint'] # type: ignore if match_found == True:
for index, item in enumerate(histories):
if match_found: if index == len(histories) - 1:
for item in histories: print(ct.colorText(f"All Events Processed for {checkpoint}", "blue"))
item_date = datetime.datetime.strptime(item['datetime'].replace(' +0000 UTC', ''), '%Y-%m-%dT%H:%M:%SZ').date() checkpoint = item['checkpoint']
if item_date >= datetime.date.today() - datetime.timedelta(days): break
json_output['response']['exechistories'].append(item) else:
if (datetime.date.today() - datetime.timedelta(days=days) > datetime.datetime.strptime(item['datetime'].replace(' +0000 UTC', ''), '%Y-%m-%dT%H:%M:%SZ').date()):
# Deduplicate and write to file pass
seen = {} else: json_output['response']['exechistories'].append(item)
if os.path.exists(file_path): seen = {}
with open(file_path, 'r') as file: if os.path.exists(file_path):
existing_data = json.load(file) with open(file_path, 'r') as file:
combined = existing_data['response']['exechistories'] + json_output['response']['exechistories'] existing_data = json.load(file)
else: combined = existing_data['response']['exechistories'] + json_output['response']['exechistories']
combined = json_output['response']['exechistories'] else:
combined = json_output['response']['exechistories']
for item in combined: for item in combined:
key = (item.get('sha256'), item.get('filename'), item.get('hostname')) key = (item.get('sha256'), item.get('filename'), item.get('hostname'))
seen[key] = item seen[key] = item
deduplicated = list(seen.values())
deduplicated = list(seen.values()) with open(file_path, 'w') as file:
with open(file_path, 'w') as file: json.dump({'error': 'Success', 'response': {'exechistories': deduplicated}}, file)
json.dump({'error': 'Success', 'response': {'exechistories': deduplicated}}, file) json_output['response']['exechistories'].clear()
# Reset output to free memory
json_output['response']['exechistories'].clear()
# Final output
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)
return json.dumps(final_output) if outputjson else None return json.dumps(final_output) if outputjson else None
def checkpoint_stomper(checkpoint, url, policy, headers): def checkpoint_stomper(checkpoint, url, policy, headers):
json_output = {'error': 'Success', 'response': {'exechistories': []}} json_output = {'error': 'Success', 'response': {'exechistories': []}}
endpoint = url + '/v1/logging/exechistories' endpoint = url + '/v1/logging/exechistories'
@@ -170,5 +150,3 @@ def skipback(days):
hex_timestamp = format(timestamp, '08x') hex_timestamp = format(timestamp, '08x')
objectid_hex = hex_timestamp + '0000000000000000' objectid_hex = hex_timestamp + '0000000000000000'
return ObjectId(objectid_hex) return ObjectId(objectid_hex)