From 7b83e6bf16cf4d9ac11e93f9c7e3c7838ea4ec03 Mon Sep 17 00:00:00 2001 From: = <=> Date: Mon, 25 Aug 2025 10:46:30 -0400 Subject: [PATCH] Tweaked Path Function --- AirlockTools.py | 8 ++++---- utils/hashfunctions.py | 4 ++-- utils/pathfunctions.py | 44 +++++++++++------------------------------- 3 files changed, 17 insertions(+), 39 deletions(-) diff --git a/AirlockTools.py b/AirlockTools.py index b4ef28e..4ad9e9f 100644 --- a/AirlockTools.py +++ b/AirlockTools.py @@ -173,7 +173,7 @@ def menu_prepare_to_enforce(): 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_remaining_hashes__{first_policy}_{second_policy}.csv"): + 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")) @@ -262,14 +262,14 @@ def menu_prepare_to_enforce(): 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_remaining_hashes__{first_policy}_{second_policy}.html", index=False) - categorized[2].to_csv(f"dataframe_csv\\df_remaining_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_remaining_hashes__{first_policy}_{second_policy}.csv"): + 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) diff --git a/utils/hashfunctions.py b/utils/hashfunctions.py index 80d260d..c193284 100644 --- a/utils/hashfunctions.py +++ b/utils/hashfunctions.py @@ -122,6 +122,6 @@ def categorizeHashes(aug_df: pd.DataFrame, threat_tolerance: int, untrusted_publ needsreview_df = df[mask_needsreview] approved_df = df[mask_approved] - remaining_df = df[~(mask_needsreview | mask_approved)] + unapproved_df = df[~(mask_needsreview | mask_approved)] - return needsreview_df, approved_df, remaining_df + return needsreview_df, approved_df, unapproved_df diff --git a/utils/pathfunctions.py b/utils/pathfunctions.py index 31ce2d7..ab8e166 100644 --- a/utils/pathfunctions.py +++ b/utils/pathfunctions.py @@ -12,6 +12,7 @@ # # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . + import pandas as pd import os from itertools import chain @@ -25,11 +26,13 @@ def filepathInitialGroup(df: pd.DataFrame): # Step 2: Explode the list so each filepath becomes its own row df = df.explode("filename_x", ignore_index=True) - # Step 3: Clean up whitespace + # Step 3: Clean up whitespace and normalize paths df["filename_x"] = df["filename_x"].str.strip() + df["filename_x"] = df["filename_x"].str.replace(r"\\\\", r"\\", regex=True) + df["filename_x"] = df["filename_x"].apply(lambda x: os.path.normpath(x) if pd.notna(x) else "") # Step 4: Extract directory and filename from each filepath - df["directory"] = df["filename_x"].apply(lambda x: os.path.dirname(x) if pd.notna(x) else "") + df["directory"] = df["filename_x"].apply(lambda x: os.path.normpath(os.path.dirname(x)) if pd.notna(x) else "") df["filename"] = df["filename_x"].apply(lambda x: os.path.basename(x) if pd.notna(x) else "") # Step 5: Drop the original raw filepath column @@ -37,10 +40,10 @@ def filepathInitialGroup(df: pd.DataFrame): # Helper functions for path manipulation def get_parts(path): - return path.strip("\\").split("\\") + return os.path.normpath(path).split(os.sep) def join_parts(parts): - return "\\".join(parts) + return os.path.normpath(os.sep.join(parts)) def longest_common_prefix(paths): split_paths = [get_parts(p) for p in paths] @@ -54,45 +57,21 @@ def filepathInitialGroup(df: pd.DataFrame): break return join_parts(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. - """ + # Step 6: Group directories by shared prefix 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]) @@ -120,7 +99,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 excluded directories 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] @@ -133,8 +112,7 @@ def filepathInitialGroup(df: pd.DataFrame): df[col] = df[col].apply(lambda x: list({str(item): item for item in chain.from_iterable(x if isinstance(x[0], list) else [x])}.values())) return df - path_eligible = deduplicate_lists(path_eligible) path_ineligible = deduplicate_lists(path_ineligible) - return path_eligible, path_ineligible + return path_eligible, path_ineligible \ No newline at end of file