Tweaked Path Function
This commit is contained in:
+4
-4
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
+10
-32
@@ -12,6 +12,7 @@
|
||||
#
|
||||
# You should have received a copy of the GNU Affero General Public License
|
||||
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
|
||||
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,7 +112,6 @@ 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)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user