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

Working Chunkinantor

Browse Source
This commit is contained in:
brotoskyj
2025-09-01 13:16:43 -04:00
parent e9c7ba132a
commit 4b291a9521
2 changed files with 38 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils/allowlist.py
+29 -10
View File
@@ -22,9 +22,7 @@ import utils.pretty as ct
import ijson
def pullPolicyExechistories(url, choice, policiesnames, outputjson: bool):
if not os.path.exists('chunkinator.csv'):
with open('chunkinator.csv', 'x') as file:
pass
file_path = 'chunkinator.json'
headers = {
"X-APIKey": os.getenv('APIKEY')
}
@@ -39,27 +37,48 @@ def pullPolicyExechistories(url, choice, policiesnames, outputjson: bool):
if array_dividend == 0:
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")
if (datetime.date.today() - datetime.timedelta(days=60) <= datetime.datetime.strptime(item['datetime'].replace(' +0000 UTC', ''), '%Y-%m-%dT%H:%M:%SZ').date()):
match_found = True
break
checkpoints_processed = round(len(json_response_data['response']['exechistories'])/array_dividend)
print(ct.colorText(f"{checkpoints_processed} checkpoints from this execution have been processed. Stepping to the subsequent checkpoint. {item['checkpoint']}", "blue"))
if (index + 1) < checkpoints_processed:
print(ct.colorText(f"{index + 1}/{checkpoints_processed} checkpoint(s) from this execution have been processed and Found with Date Match. Last Checkpoint: {item['checkpoint']}.", "blue"))
else:
print(ct.colorText(f"{index + 1}/{checkpoints_processed} checkpoint(s) from this execution have been processed. Last Checkpoint: {item['checkpoint']}.", "blue"))
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"))
print(ct.colorText(f"All Events Processed. Next Checkpoint: {item['checkpoint']}", "blue"))
break
else:
if (datetime.date.today() - datetime.timedelta(days=30) > datetime.datetime.strptime(item['datetime'].replace(' +0000 UTC', ''), '%Y-%m-%dT%H:%M:%SZ').date()):
if (datetime.date.today() - datetime.timedelta(days=60) > datetime.datetime.strptime(item['datetime'].replace(' +0000 UTC', ''), '%Y-%m-%dT%H:%M:%SZ').date()):
pass
else:
print(f"Added: {item['datetime']} | {item['checkpoint']} | {item['filename']} | {item['sha256']}")
#print(f"Added: {item['datetime']} | {item['checkpoint']} | {item['filename']} | {item['sha256']}")
json_output['response']['exechistories'].append(item)
match_found = False
if os.path.exists(file_path):
seen = {}
with open(file_path, 'r') as file:
temp_file = json.load(file)
combined_data = temp_file['response']['exechistories'] + json_output['response']['exechistories']
for item in combined_data:
key = (item.get('sha256'), item.get('filename'), item.get('hostname'))
seen[key] = item
deduplicated_data = list(seen.values())
with open(file_path, 'w') as file:
json.dump({'error': 'Success', 'response': {'exechistories': deduplicated_data}}, file)
json_output = {'error': 'Success', 'response': {'exechistories': []}}
else:
with open(file_path, 'a') as file:
json.dump(json_output, file)
json_output = {'error': 'Success', 'response': {'exechistories': []}}
with open(file_path, 'r') as file:
json_output = json.load(file)
json_output = json.dumps(json_output)
os.remove(file_path)
if outputjson == True:
return json_output