From df4fd49197992a113de5819ebc0c9f29c0c8f974 Mon Sep 17 00:00:00 2001 From: brotoskyj Date: Tue, 2 Sep 2025 09:14:22 -0400 Subject: [PATCH] Rebased from Memopt --- utils/allowlist.py | 46 +++++++++++++++++++++------------------------- 1 file changed, 21 insertions(+), 25 deletions(-) diff --git a/utils/allowlist.py b/utils/allowlist.py index d533d79..3b96de9 100644 --- a/utils/allowlist.py +++ b/utils/allowlist.py @@ -20,6 +20,9 @@ import pandas import time import utils.pretty as ct import ijson +import os +from bson import ObjectId +import datetime def pullPolicyExechistories(url, choice, policiesnames, outputjson: bool): file_path = 'chunkinator.json' @@ -28,10 +31,14 @@ def pullPolicyExechistories(url, choice, policiesnames, outputjson: bool): } checkpoint = '000000000000000000000000' json_output = {'error': 'Success', 'response': {'exechistories': []}} + while True: json_response_data = checkpoint_stomper(checkpoint, url, policiesnames, headers) - if not json_response_data['response']['exechistories']: + histories = json_response_data['response']['exechistories'] + if not histories: break + + array_dividend = max(round(len(histories) / 20), 1) match_found = False array_dividend = round(len(json_response_data['response']['exechistories'])/20) if array_dividend == 0: @@ -41,10 +48,7 @@ def pullPolicyExechistories(url, choice, policiesnames, outputjson: bool): match_found = True break checkpoints_processed = round(len(json_response_data['response']['exechistories'])/array_dividend) - 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")) + print(ct.colorText(f"{checkpoints_processed} checkpoints from this execution have been processed. Stepping to the subsequent checkpoint. {item['checkpoint']}", "blue")) checkpoint = item['checkpoint'] if match_found == True: for index, item in enumerate(json_response_data['response']['exechistories']): @@ -99,6 +103,7 @@ def checkpoint_stomper(checkpoint, url, policy, headers): json_output['response']['exechistories'].append(item) parse_text = json.loads(json.dumps(json_output)) return parse_text + def listPolicies(url): endpoint = url + '/v1/group' print(ct.colorText("[+] Grabbing All Policies", "cyan")) @@ -142,25 +147,16 @@ def listAllowlists(url): return choice, policiesnames, policyids #Need else and catch for upper bound +def skipback(days): + """ + Generate a MongoDB ObjectId for a given number of days ago from today. + Adds 1 extra day to the input to look further back. + """ + adjusted_days = days + 1 + date_days_ago = datetime.datetime.now(datetime.UTC) - datetime.timedelta(days=adjusted_days) + timestamp = int(date_days_ago.timestamp()) + hex_timestamp = format(timestamp, '08x') + objectid_hex = hex_timestamp + '0000000000000000' + return ObjectId(objectid_hex) -def listCategories(url): - endpoint = url + '/v1/application/categories' - print(ct.colorText("[+] Grabbing All Categories", "cyan")) - payload = {} - headers = { - "X-APIKey": os.getenv('APIKEY') - } - response = requests.request("POST", endpoint, headers=headers, data=payload, verify=False) - parse_text = json.loads(response.text) - policiesnames = [] - policyids = [] - for index, list in enumerate(parse_text['response']['categories'], start=1): - print(ct.colorText(f"{index}. {list['name']}", "yellow")) - policiesnames.append(list['name']) - policyids.append(list['categoryid']) - - - choice = input(ct.colorText("Select Policy Group: ", "white")) - choice = int(choice) - 1 - return choice, policiesnames, policyids