diff --git a/AirlockTools.py b/AirlockTools.py index 4ad9e9f..b944afd 100644 --- a/AirlockTools.py +++ b/AirlockTools.py @@ -19,8 +19,6 @@ import utils.getdeviceevents import utils.allowlist import utils.hashfunctions import utils.pathfunctions -import utils.allowfunctions -import utils.colortext as ct import urllib3 import pandas as pd @@ -66,235 +64,62 @@ def menu_main(): choice = input(ct.colorText("\nEnter Menu Item: ", "white")) if choice == '1': utils.getdeviceevents.devicehistory(url,False) - elif choice == "2": - menu_local_approve() - elif choice == "3": - menu_feature2() - elif choice == "4": - menu_prepare_to_enforce() - elif choice == "Q": - break - else: - print(ct.colorText("Invalid choice. Please try again.","red")) + if choice == '2': + utils.allowlist.allowlistexechistories(url,False) + if choice == '3': + executionhist = utils.allowlist.allowlistexechistories(url,True) + print(executionhist) + aggregated = utils.hashfunctions.aggregateHashes(executionhist) + print(aggregated) + augmented = utils.hashfunctions.augmentAggregatedHashes(url,aggregated) + print(augmented) -def menu_local_approve(): - while True: - print("\n--- Submenu ---") - print("1. Sub-option A") - print("2. Sub-option B") - print("3. Return to Main Menu") - choice = input("Enter your choice: ") + augmented.to_html("augmentedlist.html", index=False) - if choice == "1": - print("You selected Sub-option A") - elif choice == "2": - print("You selected Sub-option B") - elif choice == "3": - print("Returning to Main Menu...") - break - else: - print("Invalid choice. Please try again.") + badpublisherlist = [] + categorized = utils.hashfunctions.categorizeHashes(augmented, 5, badpublisherlist) + categorized[0].to_html("needsreview.html", index=False) + categorized[1].to_html("approved.html", index=False) + categorized[2].to_html("remaining.html", index=False) -def menu_feature2(): - while True: - print("\n--- Submenu ---") - print("1. Sub-option A") - print("2. Sub-option B") - print("3. Return to Main Menu") - choice = input("Enter your choice: ") + if choice == '4': + html_file = "augmentedlist.html" + augmented_df = pd.read_html(html_file) + print(augmented_df) + combined_df = pd.concat(augmented_df, ignore_index=True) - if choice == "1": - print("You selected Sub-option A") - elif choice == "2": - print("You selected Sub-option B") - elif choice == "3": - print("Returning to Main Menu...") - break - else: - print("Invalid choice. Please try again.") - -def menu_prepare_to_enforce(): - - first_policy = " " - second_policy = " " - - #If the directorys where we're going to store our output dont exist, make them. - if not os.path.exists("dataframe_html"): os.makedirs("dataframe_html") - if not os.path.exists("dataframe_csv"): os.makedirs("dataframe_csv") - if not os.path.exists("approvals"): os.makedirs("approvals") - - df_aggregated_combo = pd.DataFrame() - while True: - print(ct.colorText("\n --------------------------------------------------------------------", "cyan")) - print(ct.colorText(" -------------------- Prepare to Enforce Policy ---------------------", "cyan")) - print(ct.colorText(" --------------------------------------------------------------------", "cyan")) - print(ct.colorText("\nSequentually follow these steps to prepare a policy for enforcement:", "white")) - print(ct.colorText("\n1. Choose which policy or policies to work with - : ", "cyan")) + path_eligible, path_ineligible = utils.pathfunctions.filepathInitialGroup(combined_df) + path_eligible.to_html("EligblePaths.html", index=False) + path_ineligible.to_html("IneligiblePaths.html",index=False) - if first_policy == " " and second_policy == " ": - print(ct.colorText(f" [✗] No policies have been chosen","red")) - elif first_policy != " " and second_policy is first_policy: - print(ct.colorText(f" [✓] {first_policy} has been selected,", "green")) - elif first_policy != " " and second_policy != " ": - print(ct.colorText(f" [✓] {first_policy} has been selected as Policy 1","green")) - print(ct.colorText(f" [✓] {second_policy} has been selected as Policy 2","green")) + if choice == '5': + + executionhist = utils.allowlist.allowlistexechistories(url,True) + print(executionhist) - print(ct.colorText("2. Pull and stage event history", "cyan")) + aggregated = utils.hashfunctions.aggregateHashes(executionhist) + print(aggregated) + + augmented = utils.hashfunctions.augmentAggregatedHashes(url,aggregated) + print(augmented) + + augmented.to_html("augmentedlist.html", index=False) + html_file = "augmentedlist.html" + augmented_df = pd.read_html(html_file) + combined_df = pd.concat(augmented_df, ignore_index=True) + + path_eligible, path_ineligible = utils.pathfunctions.filepathInitialGroup(combined_df) + path_eligible.to_html("EligblePaths.html", index=False) + path_ineligible.to_html("IneligiblePaths.html",index=False) - if os.path.exists(f"dataframe_csv\\df_aggregated_{first_policy}.csv") == True: - print(ct.colorText(f" [✓] This has been completed for {first_policy}","green")) - elif os.path.exists(f"dataframe_csv\\df_aggregated_{first_policy}.csv") == False: - print(ct.colorText(f" [✗] This step has not been completed","red")) - elif second_policy is not first_policy and os.path.exists(f"dataframe_csv\\df_aggregated_{second_policy}.csv") == True: - print(ct.colorText(f" [✓] This has been completed for {second_policy}","green")) - elif second_policy is not first_policy and os.path.exists(f"dataframe_csv\\df_aggregated_{second_policy}.csv") == False: - print(ct.colorText(f" [✓] This has not been completed for {second_policy}","red")) - - print(ct.colorText("3. Combine Staged policies", "cyan")) - - if os.path.exists(f"dataframe_csv\\df_aggregated_combo_{first_policy}_{second_policy}.csv") == True: - print(ct.colorText(" [✓] This step has been completed","green")) - else: - print(ct.colorText(" [✗] This step has not been completed","red")) - - print(ct.colorText("4. Add hash threat information to list of executions", "cyan")) + badpublisherlist = [] + categorized = utils.hashfunctions.categorizeHashes(augmented, 5, badpublisherlist) + categorized[0].to_html("needsreview.html", index=False) + categorized[1].to_html("approved.html", index=False) + categorized[2].to_html("remaining.html", index=False) + - if os.path.exists(f"dataframe_csv\\df_augmented_combo_{first_policy}_{second_policy}.csv") == True: - print(ct.colorText(" [✓] This step has been completed","green")) - else: - print(ct.colorText(" [✗] This step has not been completed","red")) - - print(ct.colorText("5. Determine if path exclusions are possible", "cyan")) - - if os.path.exists(f"dataframe_csv\\df_path_eligible_{first_policy}_{second_policy}.csv") == True: - print(ct.colorText(" [✓] This step has been completed","green")) - else: - print(ct.colorText(" [✗] This step has not been completed", "red")) - - print(ct.colorText("6. Categorize your hashes ", "cyan")) - - if os.path.isfile(f"dataframe_csv\\df_hashes_needing_approval_{first_policy}_{second_policy}.csv") and os.path.isfile(f"dataframe_csv\\df_automatically_approved_hashes_{first_policy}_{second_policy}.csv") and os.path.isfile(f"dataframe_csv\\df_unapproved_hashes__{first_policy}_{second_policy}.csv"): - print(ct.colorText(" [✓] This step has been completed","green")) - else: - print(ct.colorText(" [✗] This step has not been completed","red")) - print(ct.colorText("7. Compare potential path exclusions with allowed hashes", "cyan")) - - if os.path.exists(f"dataframe_csv\\df_allowed_paths_{first_policy}_{second_policy}.csv") == True: - print(ct.colorText(" [✓] This step has been completed","green")) - else: - print(ct.colorText(" [✗] This step has not been completed","red")) - - - print(ct.colorText("Q. Quit", "cyan")) - - choice = input(ct.colorText("\nEnter your choice: ", "white")) - if choice == "1": - first_policy_tuple = utils.allowlist.listPolicies(url) - first_policy = first_policy_tuple[1][first_policy_tuple[0]] - while True: - answer = input(ct.colorText(f"{"Do you want to load a second policy?"} (yes/no): ", "white").strip().lower()) - if answer in ("yes", "y"): - second_policy_tuple = utils.allowlist.listPolicies(url) - second_policy = second_policy_tuple[1][second_policy_tuple[0]] - break - elif answer in ("no", "n"): - second_policy_tuple = first_policy_tuple - second_policy = first_policy - break - else: - print(ct.colorText("Please answer with 'yes' or 'no'.", "red")) - - elif choice == "2": - if not os.path.exists("dataframe_csv\\df_aggregated_{first_policy}.csv"): - executionhist_policy1 = utils.allowlist.pullPolicyExechistories(url,first_policy_tuple[0], first_policy_tuple[1],True) - df_aggregated_policy1 = utils.hashfunctions.aggregateHashes(executionhist_policy1) - df_aggregated_policy1.to_html(f"dataframe_html\\df_aggregated_{first_policy}.html", index=False) - df_aggregated_policy1.to_csv(f"dataframe_csv\\df_aggregated_{first_policy}.csv", index=False) - print(ct.colorText(f"Staging of Exection history for policy: {first_policy} is complete","green")) - - if not os.path.exists("dataframe_csv\\df_aggregated_{second_policy}.csv"): - executionhist_policy2 = utils.allowlist.pullPolicyExechistories(url,second_policy_tuple[0], second_policy_tuple[1],True) - df_aggregated_policy2 = utils.hashfunctions.aggregateHashes(executionhist_policy2) - df_aggregated_policy2.to_html(f"dataframe_html\\df_aggregated_{second_policy}.html", index=False) - df_aggregated_policy2.to_csv(f"dataframe_csv\\df_aggregated_{second_policy}.csv", index=False) - print(ct.colorText(f"Staging of Exection history for policy: {second_policy} is complete","green")) - - elif choice == "3": - if second_policy is first_policy and os.path.exists(f"dataframe_csv\\df_aggregated_{first_policy}.csv"): - df1 = tryToReadCSV(f"dataframe_csv\\df_aggregated_{first_policy}.csv") - df_aggregated_combo = df1 - df_aggregated_combo.to_html(f"dataframe_html\\df_aggregated_combo_{first_policy}_{second_policy}.html", index=False) - df_aggregated_combo.to_csv(f"dataframe_csv\\df_aggregated_combo_{first_policy}_{second_policy}.csv", index=False) - elif os.path.exists(f"dataframe_csv\\df_aggregated_{first_policy}.csv") and os.path.exists(f"dataframe_csv\\df_aggregated_{second_policy}.csv"): - df1 = tryToReadCSV(f"dataframe_csv\\df_aggregated_{first_policy}.csv") - df2 = tryToReadCSV(f"dataframe_csv\\df_aggregated_{second_policy}.csv") - df_aggregated_combo = pd.concat([df1 , df2], ignore_index=True) - df_aggregated_combo.to_html(f"dataframe_html\\df_aggregated_combo_{first_policy}_{second_policy}.html", index=False) - df_aggregated_combo.to_csv(f"dataframe_csv\\df_aggregated_combo_{first_policy}_{second_policy}.csv", index=False) - else: - print(ct.colorText(f"Please stage your data before attempting this step","red")) - - elif choice == "4": - if os.path.exists(f"dataframe_csv\\df_aggregated_combo_{first_policy}_{second_policy}.csv"): - df_augmented = utils.hashfunctions.augmentAggregatedHashes(url,tryToReadCSV(f"dataframe_csv\\df_aggregated_combo_{first_policy}_{second_policy}.csv")) - df_augmented.to_html(f"dataframe_html\\df_augmented_combo_{first_policy}_{second_policy}.html", index=False) - df_augmented.to_csv(f"dataframe_csv\\df_augmented_combo_{first_policy}_{second_policy}.csv", index=False) - print(ct.colorText(f"Hash reputation info added to dataframe","green")) - else: - print(ct.colorText(f"Please combine your data with step 3 prior to attempting this step","red")) - - elif choice == "5": - if os.path.exists(f"dataframe_html\\df_augmented_combo_{first_policy}_{second_policy}.html"): - path_eligible, path_ineligible = utils.pathfunctions.filepathInitialGroup(pd.read_csv(f"dataframe_csv\\df_augmented_combo_{first_policy}_{second_policy}.csv")) - path_eligible.to_html(f"dataframe_html\\df_path_eligible_{first_policy}_{second_policy}.html", index=False) - path_eligible.to_csv(f"dataframe_csv\\df_path_eligible_{first_policy}_{second_policy}.csv", index=False) - path_ineligible.to_html(f"dataframe_html\\df_path_ineligible_{first_policy}_{second_policy}.html", index=False) - path_ineligible.to_csv(f"dataframe_csv\\df_path_ineligible_{first_policy}_{second_policy}.csv", index=False) - print(ct.colorText(f"Eligible paths determined","green")) - else: - print(ct.colorText(f"Please Augment your data with hash threat info using step 4 prior to attempting this step","red")) - - elif choice == "6": - if os.path.exists(f"dataframe_csv\\df_augmented_combo_{first_policy}_{second_policy}.csv"): - categorized = utils.hashfunctions.categorizeHashes(pd.read_csv(f"dataframe_csv\\df_augmented_combo_{first_policy}_{second_policy}.csv"), threat_tolerance_constant, badpublisherlist) - categorized[0].to_html(f"dataframe_html\\df_hashes_needing_approval_{first_policy}_{second_policy}.html", index=False) - categorized[0].to_csv(f"dataframe_csv\\df_hashes_needing_approval_{first_policy}_{second_policy}.csv", index=False) - categorized[1].to_html(f"dataframe_html\\df_automatically_approved_hashes_{first_policy}_{second_policy}.html", index=False) - categorized[1].to_csv(f"dataframe_csv\\df_automatically_approved_hashes_{first_policy}_{second_policy}.csv", index=False) - categorized[2].to_html(f"dataframe_html\\df_unapproved_hashes__{first_policy}_{second_policy}.html", index=False) - categorized[2].to_csv(f"dataframe_csv\\df_unapproved_hashes__{first_policy}_{second_policy}.csv", index=False) - print(ct.colorText(f"Hashes have been categorized","green")) - else: - print(ct.colorText(f"Please Augment your data with hash threat info using step 4 prior to attempting this step","red")) - - elif choice == "7": - if os.path.exists(f"dataframe_csv\\df_hashes_needing_approval_{first_policy}_{second_policy}.csv") and os.path.exists(f"dataframe_csv\\df_automatically_approved_hashes_{first_policy}_{second_policy}.csv") and os.path.exists(f"dataframe_csv\\df_unapproved_hashes__{first_policy}_{second_policy}.csv"): - allowpaths = utils.allowfunctions.filter_and_drop(pd.read_csv(f"dataframe_csv\\df_automatically_approved_hashes_{first_policy}_{second_policy}.csv"),tryToReadCSV(f"dataframe_csv\\df_path_eligible_{first_policy}_{second_policy}.csv"), path_exclusion_constant) - allowpaths.to_html(f"dataframe_html\\df_allowed_paths_{first_policy}_{second_policy}.html", index=False) - allowpaths.to_csv(f"dataframe_csv\\df_allowed_paths_{first_policy}_{second_policy}.csv", index=False) - print(ct.colorText(f"Allowable paths determined","green")) - else: - print(ct.colorText(f"Please complete step 6 prior to attempting this step","red")) - - elif choice == "Q": - break - - else: - print(ct.colorText("Invalid choice. Please try again.", "red")) - -def tryToReadCSV(csv): - try: - df =pd.read_csv(csv) - if df.empty: - print(ct.colorText("Error: CSV file has headers but no data rows.", "red")) - else: - print(ct.colorText("Data loaded successfully.", "green")) - except pd.errors.EmptyDataError: - print(ct.colorText("Notice : CSV file is completely empty (no headers, no data), falling back to empty frame", "white")) - df = pd.DataFrame() # Create an empty DataFrame as fallback - return df - if __name__ == "__main__": diff --git a/hashtest.py b/hashtest.py new file mode 100644 index 0000000..b0dab88 --- /dev/null +++ b/hashtest.py @@ -0,0 +1,5 @@ +hashes = '' +while True: + inputhash = input("Hash: ") + hashes = hashes + ',' + inputhash + print(hashes) \ No newline at end of file diff --git a/utils/allowlist.py b/utils/allowlist.py index 6cc6e2d..a3ebf85 100644 --- a/utils/allowlist.py +++ b/utils/allowlist.py @@ -17,7 +17,6 @@ import requests import json import os import time -import utils.colortext as ct def pullPolicyExechistories(url, choice, policiesnames, outputjson: bool): @@ -73,4 +72,71 @@ def listPolicies(url): policyids.append(list['groupid']) choice = input(ct.colorText("Select Policy Group: ", "white")) choice = int(choice) - 1 - return choice, policiesnames \ No newline at end of file + checkpoint = '000000000000000000000000' + json_output = {'error': 'Success', 'response': {'exechistories': []}} + while True: + json_response_data = checkpoint_stomper(checkpoint, url, policiesnames[choice], headers) + if not json_response_data['response']['exechistories']: + break + for index, item in enumerate(json_response_data['response']['exechistories']): + if index == len(json_response_data['response']['exechistories']) -1: + checkpoint = item['checkpoint'] + print(f"Date Greater than 30 Days, Stepping to new Checkpoint. {item['checkpoint']}") + 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 + else: + #json_output['response']['exechistories'].append(json_response_data['response']['exechistories'][1]) + for output in json_response_data['response']['exechistories']: + json_output['response']['exechistories'].append(output) + json_output = json.dumps(json_output) + if outputjson == True: + return json_output + #endpoint = url + '/v1/logging/exechistories' + #payload_dict = { + # "type":[1, 2, 6, 7], + # "checkpoint":"000000000000000000000000", + # "policy": [policiesnames[choice]] + #} + #payload = json.dumps(payload_dict) + #print(payload) + #response = requests.request("POST", endpoint, headers=headers, data=payload, verify=False) + #parse_text = json.loads(response.text) + #text_response = checkpoint_stomper(parse_text['response']['exechistories'], url, policiesnames[choice]) +# + #if outputjson == False: + # return response + # + #parse_text = json.loads(response.text) + # + #for item in parse_text['response']['exechistories']: + # print(item['checkpoint']) + # print(item['datetime']) + # print(item['hostname']) + # print(item['filename']) + # checkpoint_stomper(item['checkpoint'], endpoint, headers, policiesnames[choice]) + +def checkpoint_stomper(checkpoint, url, policy, headers): + endpoint = url + '/v1/logging/exechistories' + payload_dict = { + "type":[1,2,6,7], + "checkpoint": checkpoint, + "policy": [policy] + } + payload = json.dumps(payload_dict) + response = requests.request("POST", endpoint, headers=headers, data=payload, verify=False) + parse_text = json.loads(response.text) + return parse_text + #for index, item in enumerate(parse_text): + # if index == len(parse_text) - 1: + # checkpoint = item['checkpoint'] + # print(f"Time: {item['datetime']} Checkpoint: {item['checkpoint']}") + # response_fuzzer(checkpoint, url, policyname) + # else: + # if (datetime.date.today() - datetime.timedelta(days=30) > datetime.datetime.strptime(item['datetime'].replace( ' +0000 UTC', ''), '%Y-%m-%dT%H:%M:%SZ').date()): + # pass + # else: + # response_fuzzer(checkpoint, url, policyname) + + + print("Finished") diff --git a/utils/hashfunctions.py b/utils/hashfunctions.py index c193284..9b9c59b 100644 --- a/utils/hashfunctions.py +++ b/utils/hashfunctions.py @@ -90,38 +90,28 @@ def categorizeHashes(aug_df: pd.DataFrame, threat_tolerance: int, untrusted_publ df = aug_df.copy() - def reputationtool(row): - val = row["reputation_scannermatch"] - if pd.isna(val) or val == "N/A": - return row["publisher_y"] == "Not Signed" + def reputationtool(row, threat_tolerance): + if row["reputation_scannermatch"] == "N/A": + return True try: return int(val) > threat_tolerance except (ValueError, TypeError): return row["publisher_y"] == "Not Signed" - df["reputation_flag"] = df.apply(reputationtool, axis=1) - - mask_needsreview = ( - ((df["publisher_y"] == "Not Signed") & df["reputation_flag"]) | - (df["reputation_status"] == "UNKNOWN") - ) + mask_needsreview = (df["publisher_y"] == "Not Signed") & df.apply(lambda row: reputationtool(row, threat_tolerance), axis=1) mask_approved = ( - ( - (df["publisher_y"] != "Not Signed") & - ~df["publisher_y"].isin(untrusted_publishers) & - ~df["reputation_status"].isna() - ) | - ( - (df["publisher_y"] == "Not Signed") & - ~df["reputation_flag"] & - ~df["publisher_y"].isin(untrusted_publishers) & - ~df["reputation_status"].isna() - ) + ((df["publisher_y"] != "Not Signed") & (~df["publisher_y"].isin(untrusted_publishers))) & + (df["publisher_y"] != "Not Signed") # explicitly signed + ) | ( + (df["publisher_y"] == "Not Signed") & + (~df.apply(lambda row: reputationtool(row, threat_tolerance), axis=1)) & + (~df["publisher_y"].isin(untrusted_publishers)) # exclude untrusted even if unsigned ) + needsreview_df = df[mask_needsreview] approved_df = df[mask_approved] - unapproved_df = df[~(mask_needsreview | mask_approved)] + remaining_df = df[~(mask_needsreview | mask_approved)] - return needsreview_df, approved_df, unapproved_df + return needsreview_df, approved_df, remaining_df diff --git a/utils/pathfunctions.py b/utils/pathfunctions.py index ab8e166..72bdae2 100644 --- a/utils/pathfunctions.py +++ b/utils/pathfunctions.py @@ -57,21 +57,45 @@ def filepathInitialGroup(df: pd.DataFrame): break return join_parts(prefix) - # Step 6: Group directories by shared prefix + # Step 6: Group directories by shared prefix using custom logic + """ + Loop through each directory path + directories: list of all directory paths. + groups: will hold lists of grouped directories. + used: tracks which directories have already been grouped. + """ directories = df["directory"].tolist() groups = [] used = set() + #For Each directory, compare it with others + """ + Skip if already grouped. + Start a new group with the current path. + parts_i is the list of folder names in the path (e.g., ["C:", "Users", "John", "Documents"]). + """ + for i, path in enumerate(directories): if path in used: continue group = [path] parts_i = get_parts(path) + #Compare with all other directories: For each other directory, split it into parts and find the common prefix (shared folder structure). + """ + Logic: + If the directory is deep (>3 parts) and shares at least 3 parts → group it. + If it's exactly 3 parts long and shares at least 2 → group it. + Or, if it shares all but one part and is deep → group it. + These rules are designed to: + Group directories that are closely related in structure. + Avoid grouping unrelated paths that just happen to start similarly. + """ + for j in range(i + 1, len(directories)): parts_j = get_parts(directories[j]) common = os.path.commonprefix([parts_i, parts_j]) - + #Apply grouping rules if (len(parts_i) > 3 and len(common) >= 3) or (len(parts_i) == 3 and len(common) >= 2): group.append(directories[j]) used.add(directories[j]) @@ -99,7 +123,7 @@ def filepathInitialGroup(df: pd.DataFrame): path_eligible = grouped_df[grouped_df["depth"] > 2].drop(columns=["depth"]) path_ineligible = grouped_df[grouped_df["depth"] <= 2].drop(columns=["depth"]) - # Step 10: Move entries from eligible to ineligible if grouped_directory contains excluded directories + # Step 10: Move entries from eligible to ineligible if grouped_directory contains 'C:\Users' or 'c$\Users' mask = path_eligible["grouped_directory"].str.contains(r"(?i)(?:\\Users|\\c\$\\Users|inetpub\\wwwroot|windows\\temp)", na=False) move_to_ineligible = path_eligible[mask] path_eligible = path_eligible[~mask]