diff --git a/AirlockTools.py b/AirlockTools.py index 098c737..d614270 100644 --- a/AirlockTools.py +++ b/AirlockTools.py @@ -33,7 +33,7 @@ dotenv.load_dotenv() #Constants url = os.getenv('url') 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 min_files_for_path = 4 threat_tolerance_constant = 4 diff --git a/utils/allowlist.py b/utils/allowlist.py index 2eae0ea..4f9af7f 100644 --- a/utils/allowlist.py +++ b/utils/allowlist.py @@ -23,81 +23,61 @@ from bson import ObjectId import datetime def pullPolicyExechistories(url, policiesnames, days, outputjson: bool): - 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): with open(file_path, 'w') as 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: print(f"File '{file_path}' already exists.") - headers = {"X-APIKey": os.getenv('APIKEY')} - checkpoint = str(skipback(days)) json_output = {'error': 'Success', 'response': {'exechistories': []}} - while True: json_response_data = checkpoint_stomper(checkpoint, url, policiesnames, headers) histories = json_response_data['response']['exechistories'] if not histories: break - - array_dividend = max(round(len(histories) / 20), 1) + array_dividend = max(round(len(histories) / 20), 1) match_found = False - 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 item_date >= datetime.date.today() - datetime.timedelta(days): + if (datetime.date.today() - datetime.timedelta(days=days) <= datetime.datetime.strptime(item['datetime'].replace(' +0000 UTC', ''), '%Y-%m-%dT%H:%M:%SZ').date()): match_found = True break - checkpoints_processed = round(len(histories) / array_dividend) - print(ct.colorText( - f"{index + 1}/{checkpoints_processed} checkpoint(s) processed. " # type: ignore - f"{'Found with Date Match.' if match_found else ''} Last Checkpoint: {item['checkpoint']}.", "blue")) # type: ignore - - checkpoint = item['checkpoint'] # type: ignore - - if match_found: - for item in histories: - item_date = datetime.datetime.strptime(item['datetime'].replace(' +0000 UTC', ''), '%Y-%m-%dT%H:%M:%SZ').date() - if item_date >= datetime.date.today() - datetime.timedelta(days): - json_output['response']['exechistories'].append(item) - - # Deduplicate and write to file - seen = {} - if os.path.exists(file_path): - with open(file_path, 'r') as file: - existing_data = json.load(file) - combined = existing_data['response']['exechistories'] + json_output['response']['exechistories'] - else: - combined = json_output['response']['exechistories'] - - for item in combined: - key = (item.get('sha256'), item.get('filename'), item.get('hostname')) - seen[key] = item - - deduplicated = list(seen.values()) - with open(file_path, 'w') as file: - json.dump({'error': 'Success', 'response': {'exechistories': deduplicated}}, file) - - # Reset output to free memory - json_output['response']['exechistories'].clear() - - # Final output + if ( index + 1 ) < checkpoints_processed: + print(ct.colorText(f"{index + 1}/{checkpoints_processed} checkpoint(s) from this execution have been processed with date match. Last Checkpoint: {checkpoint}", "blue")) + else: + print(ct.colorText(f"{index + 1}/{checkpoints_processed} checkpoint(s) Processed. Last Checkpoint: {checkpoint}", "blue")) + if match_found == True: + for index, item in enumerate(histories): + if index == len(histories) - 1: + print(ct.colorText(f"All Events Processed for {checkpoint}", "blue")) + checkpoint = item['checkpoint'] + break + else: + if (datetime.date.today() - datetime.timedelta(days=days) > datetime.datetime.strptime(item['datetime'].replace(' +0000 UTC', ''), '%Y-%m-%dT%H:%M:%SZ').date()): + pass + else: json_output['response']['exechistories'].append(item) + seen = {} + if os.path.exists(file_path): + with open(file_path, 'r') as file: + existing_data = json.load(file) + combined = existing_data['response']['exechistories'] + json_output['response']['exechistories'] + else: + combined = json_output['response']['exechistories'] + for item in combined: + key = (item.get('sha256'), item.get('filename'), item.get('hostname')) + seen[key] = item + deduplicated = list(seen.values()) + with open(file_path, 'w') as file: + json.dump({'error': 'Success', 'response': {'exechistories': deduplicated}}, file) + json_output['response']['exechistories'].clear() with open(file_path, 'r') as file: final_output = json.load(file) os.remove(file_path) - return json.dumps(final_output) if outputjson else None - def checkpoint_stomper(checkpoint, url, policy, headers): json_output = {'error': 'Success', 'response': {'exechistories': []}} endpoint = url + '/v1/logging/exechistories' @@ -169,6 +149,4 @@ def skipback(days): timestamp = int(date_days_ago.timestamp()) hex_timestamp = format(timestamp, '08x') objectid_hex = hex_timestamp + '0000000000000000' - return ObjectId(objectid_hex) - - + return ObjectId(objectid_hex) \ No newline at end of file