mysticmomba/AirlockTools
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

Reverting Back to Working Branch

Browse Source
This commit is contained in:
brotoskyj
2025-08-25 12:14:58 -04:00
parent 05ca60f1b6 b187d8cd79
commit 3c265c5817
4 changed files with 65 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
AirlockTools.py
+6 -22
View File
@@ -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:
@@ -235,24 +239,13 @@ def menu_prepare_to_enforce():
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,7 +337,6 @@ 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
utils/allowlist.py
+7
View File
@@ -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
utils/hashfunctions.py
+21 -11
View File
@@ -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["publisher_y"].isin(untrusted_publishers))) &
(df["publisher_y"] != "Not Signed") # explicitly signed
) | (
(df["publisher_y"] == "Not Signed") &
~df["reputation_flag"] &
~df["publisher_y"].isin(untrusted_publishers) &
~df["reputation_status"].isna()
)
(~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
utils/pathfunctions.py
+28
View File
@@ -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]