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

Fixed Memory Leak Issue \

Browse Source 
Refactored Some Sorting for better deduplication
This commit is contained in:
brotoskyj
2025-08-25 20:22:51 -04:00
parent d767ba9621
commit e603ed38d3
1 changed files with 15 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils/allowlist.py
+15 -8
View File
@@ -18,7 +18,7 @@ import json
import os
import time
import utils.colortext as ct
import ijson
def pullPolicyExechistories(url, choice, policiesnames, outputjson: bool):
headers = {
@@ -33,18 +33,20 @@ def pullPolicyExechistories(url, choice, policiesnames, outputjson: bool):
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"))
print(ct.colorText(f"All checkpoints from this execution have been processed. Stepping to the subsequent checkpoint. {item['checkpoint']}", "blue"))
break
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
if (datetime.date.today() - datetime.timedelta(days=30) > datetime.datetime.strptime(item['datetime'].replace(' +0000 UTC', ''), '%Y-%m-%dT%H:%M:%SZ').date()):
print(item['datetime'])
else:
for output in json_response_data['response']['exechistories']:
json_output['response']['exechistories'].append(output)
print(item['datetime'])
json_output['response']['exechistories'].append(item)
json_output = json.dumps(json_output)
if outputjson == True:
return json_output
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],
@@ -52,8 +54,13 @@ def checkpoint_stomper(checkpoint, url, policy, headers):
"policy": [policy]
}
payload = json.dumps(payload_dict)
response = requests.request("POST", endpoint, headers=headers, data=payload, verify=False)
parse_text = json.loads(response.text)
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')), item.get('datetime')
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):