diff --git a/AirlockTools.py b/AirlockTools.py index 8e400ab..4faf2ac 100644 --- a/AirlockTools.py +++ b/AirlockTools.py @@ -19,14 +19,18 @@ import utils.getdeviceevents import utils.allowlist import utils.hashfunctions import utils.pathfunctions +<<<<<<< HEAD import utils.allowfunctions <<<<<<< HEAD +======= +>>>>>>> b187d8cd79c04d364185a4e366d0432841a3372a import utils.colortext as ct ======= >>>>>>> ccbf716 (Zar-Branch (#6)) import urllib3 import pandas as pd + urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) dotenv.load_dotenv() @@ -125,7 +129,7 @@ def menu_prepare_to_enforce(): #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("manuallyapproved"): os.makedirs("approvals") + if not os.path.exists("approvals"): os.makedirs("approvals") df_aggregated_combo = pd.DataFrame() while True: @@ -234,25 +238,14 @@ def menu_prepare_to_enforce(): print(ct.colorText(" [✓] This step has been completed","green")) else: print(ct.colorText(" [✗] This step has not been completed","red")) - - print(ct.colorText(f"7. Manually review the files \\dataframe_csv\\df_hashes_needing_approval_{first_policy}_{second_policy}.csv and dataframe_csv\\df_automatically_approved_hashes_{first_policy}_{second_policy}.csv", "cyan")) - print(ct.colorText(" Remove the rows containing hashes you do not approve of, and those you would not approve of without metarules.", "cyan")) - print(ct.colorText(" If metarules need to be created, please make note of them, and remove the row from the csv.", "cyan")) - print(ct.colorText(" When complete, move both csv files to the directory 'manuallyapproved' and choose this option to combine these approved hashes with the automatically approved hashes", "cyan")) - - if os.path.isfile(f"dataframe_csv\\df_all_approved_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")) @@ -336,14 +329,6 @@ def menu_prepare_to_enforce(): 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.isfile(f"manuallyapproved\\df_hashes_needing_approval_{first_policy}_{second_policy}.csv"): - df1 = tryToReadCSV(f"manuallyapproved\\df_hashes_needing_approval_{first_policy}_{second_policy}.csv") - df2 = tryToReadCSV(f"manuallyapproved\\df_automatically_approved_hashes_{first_policy}_{second_policy}.csv") - df_all_approved_hashes = pd.concat([df1 , df2], ignore_index=True) - df_aggregated_combo.to_html(f"dataframe_html\\df_all_approved_hashes_{first_policy}_{second_policy}.html", index=False) - df_aggregated_combo.to_csv(f"dataframe_csv\\df_all_approved_hashes_{first_policy}_{second_policy}.csv", index=False) - """ 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) @@ -352,8 +337,7 @@ def menu_prepare_to_enforce(): 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 diff --git a/utils/allowlist.py b/utils/allowlist.py index 97467e3..9e1a914 100644 --- a/utils/allowlist.py +++ b/utils/allowlist.py @@ -27,6 +27,7 @@ import utils.colortext as ct >>>>>>> b187d8c (Added colors) + def pullPolicyExechistories(url, choice, policiesnames, outputjson: bool): headers = { @@ -81,9 +82,12 @@ def listPolicies(url): policyids.append(list['groupid']) choice = input(ct.colorText("Select Policy Group: ", "white")) choice = int(choice) - 1 +<<<<<<< HEAD <<<<<<< HEAD return choice, policiesnames ======= +======= +>>>>>>> b187d8cd79c04d364185a4e366d0432841a3372a checkpoint = '000000000000000000000000' json_output = {'error': 'Success', 'response': {'exechistories': []}} while True: @@ -152,4 +156,7 @@ def checkpoint_stomper(checkpoint, url, policy, headers): print("Finished") +<<<<<<< HEAD >>>>>>> ccbf716 (Zar-Branch (#6)) +======= +>>>>>>> b187d8cd79c04d364185a4e366d0432841a3372a diff --git a/utils/hashfunctions.py b/utils/hashfunctions.py index 43481b8..165a98c 100644 --- a/utils/hashfunctions.py +++ b/utils/hashfunctions.py @@ -90,6 +90,7 @@ def categorizeHashes(aug_df: pd.DataFrame, threat_tolerance: int, untrusted_publ df = aug_df.copy() +<<<<<<< HEAD <<<<<<< HEAD def reputationtool(row): val = row["reputation_scannermatch"] @@ -100,11 +101,17 @@ def categorizeHashes(aug_df: pd.DataFrame, threat_tolerance: int, untrusted_publ if row["reputation_scannermatch"] == "N/A": return True >>>>>>> ccbf716 (Zar-Branch (#6)) +======= + def reputationtool(row, threat_tolerance): + if row["reputation_scannermatch"] == "N/A": + return True +>>>>>>> b187d8cd79c04d364185a4e366d0432841a3372a try: return int(val) > threat_tolerance except (ValueError, TypeError): return row["publisher_y"] == "Not Signed" +<<<<<<< HEAD <<<<<<< HEAD df["reputation_flag"] = df.apply(reputationtool, axis=1) @@ -112,25 +119,25 @@ def categorizeHashes(aug_df: pd.DataFrame, threat_tolerance: int, untrusted_publ ((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) +>>>>>>> b187d8cd79c04d364185a4e366d0432841a3372a 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)] +<<<<<<< HEAD return needsreview_df, approved_df, unapproved_df ======= mask_needsreview = (df["publisher_y"] == "Not Signed") & df.apply(lambda row: reputationtool(row, threat_tolerance), axis=1) @@ -151,3 +158,6 @@ def categorizeHashes(aug_df: pd.DataFrame, threat_tolerance: int, untrusted_publ return needsreview_df, approved_df, remaining_df >>>>>>> ccbf716 (Zar-Branch (#6)) +======= + return needsreview_df, approved_df, remaining_df +>>>>>>> b187d8cd79c04d364185a4e366d0432841a3372a diff --git a/utils/pathfunctions.py b/utils/pathfunctions.py index 890f62f..58fe2b9 100644 --- a/utils/pathfunctions.py +++ b/utils/pathfunctions.py @@ -57,9 +57,12 @@ def filepathInitialGroup(df: pd.DataFrame): break return join_parts(prefix) +<<<<<<< HEAD <<<<<<< HEAD # Step 6: Group directories by shared prefix ======= +======= +>>>>>>> b187d8cd79c04d364185a4e366d0432841a3372a # Step 6: Group directories by shared prefix using custom logic """ Loop through each directory path @@ -67,7 +70,10 @@ def filepathInitialGroup(df: pd.DataFrame): groups: will hold lists of grouped directories. used: tracks which directories have already been grouped. """ +<<<<<<< HEAD >>>>>>> ccbf716 (Zar-Branch (#6)) +======= +>>>>>>> b187d8cd79c04d364185a4e366d0432841a3372a directories = df["directory"].tolist() groups = [] used = set() @@ -85,6 +91,7 @@ def filepathInitialGroup(df: pd.DataFrame): group = [path] parts_i = get_parts(path) +<<<<<<< HEAD <<<<<<< HEAD for j in range(i + 1, len(directories)): parts_j = get_parts(directories[j]) @@ -107,6 +114,23 @@ def filepathInitialGroup(df: pd.DataFrame): common = os.path.commonprefix([parts_i, parts_j]) #Apply grouping rules >>>>>>> ccbf716 (Zar-Branch (#6)) +======= + #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 +>>>>>>> b187d8cd79c04d364185a4e366d0432841a3372a 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]) @@ -134,11 +158,15 @@ 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"]) +<<<<<<< HEAD <<<<<<< HEAD # 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' >>>>>>> ccbf716 (Zar-Branch (#6)) +======= + # Step 10: Move entries from eligible to ineligible if grouped_directory contains 'C:\Users' or 'c$\Users' +>>>>>>> b187d8cd79c04d364185a4e366d0432841a3372a 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]