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

closes #15

Browse Source 
Created iterator to skip over items. This was causing issues because it was too many entries and it would cause a memory leak.
This commit is contained in:
brotoskyj
2025-08-25 23:36:24 -04:00
parent e603ed38d3
commit f081594a7e
1 changed files with 24 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils/allowlist.py
+24 -11
View File
@@ -20,7 +20,6 @@ import time
import utils.colortext as ct import utils.colortext as ct
import ijson import ijson
def pullPolicyExechistories(url, choice, policiesnames, outputjson: bool): def pullPolicyExechistories(url, choice, policiesnames, outputjson: bool):
headers = { headers = {
"X-APIKey": os.getenv('APIKEY') "X-APIKey": os.getenv('APIKEY')
} }
@@ -30,17 +29,31 @@ def pullPolicyExechistories(url, choice, policiesnames, outputjson: bool):
json_response_data = checkpoint_stomper(checkpoint, url, policiesnames[choice], headers) json_response_data = checkpoint_stomper(checkpoint, url, policiesnames[choice], headers)
if not json_response_data['response']['exechistories']: if not json_response_data['response']['exechistories']:
break break
for index, item in enumerate(json_response_data['response']['exechistories']): match_found = False
if index == len(json_response_data['response']['exechistories']) -1: array_dividend = round(len(json_response_data['response']['exechistories'])/20)
checkpoint = item['checkpoint'] if array_dividend == 0:
print(ct.colorText(f"All checkpoints from this execution have been processed. Stepping to the subsequent checkpoint. {item['checkpoint']}", "blue")) array_dividend == 1
for index, item in enumerate(json_response_data['response']['exechistories'][::array_dividend]):
if (datetime.date.today() - datetime.timedelta(days=30) <= datetime.datetime.strptime(item['datetime'].replace(' +0000 UTC', ''), '%Y-%m-%dT%H:%M:%SZ').date()):
print("Found Date Match")
match_found = True
break break
else: checkpoints_processed = round(len(json_response_data['response']['exechistories'])/array_dividend)
if (datetime.date.today() - datetime.timedelta(days=30) > datetime.datetime.strptime(item['datetime'].replace(' +0000 UTC', ''), '%Y-%m-%dT%H:%M:%SZ').date()): print(ct.colorText(f"{checkpoints_processed} checkpoints from this execution have been processed. Stepping to the subsequent checkpoint. {item['checkpoint']}", "blue"))
print(item['datetime']) checkpoint = item['checkpoint']
if match_found == True:
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"All checkpoints from this execution have been processed. Stepping to the subsequent checkpoint. {item['checkpoint']}", "blue"))
break
else: else:
print(item['datetime']) if (datetime.date.today() - datetime.timedelta(days=30) > datetime.datetime.strptime(item['datetime'].replace(' +0000 UTC', ''), '%Y-%m-%dT%H:%M:%SZ').date()):
json_output['response']['exechistories'].append(item) pass
else:
print(f"Added: {item['datetime']} | {item['checkpoint']} | {item['filename']} | {item['sha256']}")
json_output['response']['exechistories'].append(item)
match_found = False
json_output = json.dumps(json_output) json_output = json.dumps(json_output)
if outputjson == True: if outputjson == True:
return json_output return json_output
@@ -57,7 +70,7 @@ def checkpoint_stomper(checkpoint, url, policy, headers):
with requests.request("POST", endpoint, headers=headers, data=payload, verify=False, stream=True) as response: with requests.request("POST", endpoint, headers=headers, data=payload, verify=False, stream=True) as response:
parser = ijson.items(response.raw, 'response.exechistories.item') parser = ijson.items(response.raw, 'response.exechistories.item')
for item in parser: for item in parser:
key = (item.get('sha256'), item.get('hostname')), item.get('datetime') key = (item.get('sha256'), item.get('hostname'))
if key not in json_output: if key not in json_output:
json_output['response']['exechistories'].append(item) json_output['response']['exechistories'].append(item)
parse_text = json.loads(json.dumps(json_output)) parse_text = json.loads(json.dumps(json_output))