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

Cleaned up menu by functionalizing

Browse Source
This commit is contained in:
=
2025-09-04 12:18:22 -04:00
parent 73811ba4d2
commit 781edcfa99
5 changed files with 299 additions and 309 deletions
Download Patch File Download Diff File Expand all files Collapse all files
AirlockTools.py
+26 -290
View File
@@ -14,11 +14,8 @@
# along with this program. If not, see <https://www.gnu.org/licenses/>. # along with this program. If not, see <https://www.gnu.org/licenses/>.
import dotenv import dotenv
import gc
import json
import os import os
import pandas as pd import pandas as pd
import re
import urllib3 import urllib3
import utils.allowlist import utils.allowlist
import utils.getdeviceevents import utils.getdeviceevents
@@ -41,7 +38,6 @@ path_exclusion_constant = 3
min_files_for_path = 4 min_files_for_path = 4
threat_tolerance_constant = 4 threat_tolerance_constant = 4
def apivalidation(): def apivalidation():
match os.getenv('APIKEY'): match os.getenv('APIKEY'):
case '': case '':
@@ -140,9 +136,12 @@ def menu_prepare_to_enforce():
first_policy = " " first_policy = " "
second_policy = " " second_policy = " "
allowlist_parent_name = " "
allowlist_child_name = " "
destination_name = " " destination_name = " "
destination_id = " "
allowlist_parent_name = " "
allowlist_parent_id = " "
allowlist_child_name = " "
allowlist_child_id = " "
#If the directorys where we're going to store our output dont exist, make them. #If the directorys where we're going to store our output dont exist, make them.
if not os.path.exists("parquet"): os.makedirs("parquet") if not os.path.exists("parquet"): os.makedirs("parquet")
@@ -176,247 +175,34 @@ def menu_prepare_to_enforce():
elif choice == "2": elif choice == "2":
if not os.path.exists(f"parquet\\execution_history_{first_policy}.parquet"): if not os.path.exists(f"parquet\\execution_history_{first_policy}.parquet"):
print(choice) utils.policyfunctions.getPolicyInfo(url, first_policy, 60)
print(first_policy)
exe1 = utils.allowlist.pullPolicyExechistories(url, first_policy, 60, True)
data = json.loads(exe1)
executionhist_policy1 = pd.DataFrame(data["response"]["exechistories"])
if not executionhist_policy1.empty:
executionhist_policy1 = executionhist_policy1[['sha256', 'publisher', 'filename', 'hostname', 'username', 'pprocess', 'gprocess', 'commandline']]
executionhist_policy1 = executionhist_policy1.drop_duplicates(subset=['sha256', 'filename', 'hostname'])
executionhist_policy1 = executionhist_policy1.sort_values(by=['sha256', 'filename'])
executionhist_policy1.to_parquet(f"parquet\\execution_history_{first_policy}.parquet", index=False)
print(ct.colorText(f"Staging of Execution history for policy: {first_policy} is complete", "green"))
del data
del exe1
del executionhist_policy1
gc.collect()
if not os.path.exists(f"parquet\\execution_history_{second_policy}.parquet"): if not os.path.exists(f"parquet\\execution_history_{second_policy}.parquet"):
exe2 = utils.allowlist.pullPolicyExechistories(url,second_policy, 60, True) utils.policyfunctions.getPolicyInfo(url, second_policy, 60)
data2 = json.loads(exe2)
executionhist_policy2 = pd.DataFrame(data2["response"]["exechistories"])
if not executionhist_policy2.empty:
executionhist_policy2 = executionhist_policy2[['sha256', 'publisher', 'filename', 'hostname', 'username', 'pprocess', 'gprocess', 'commandline']]
executionhist_policy2 = executionhist_policy2.drop_duplicates(subset=['sha256', 'filename', 'hostname'])
executionhist_policy2 = executionhist_policy2.sort_values(by=['sha256', 'filename'])
executionhist_policy2.to_parquet(f"parquet\\execution_history_{second_policy}.parquet", index=False)
print(ct.colorText(f"Staging of Execution history for policy: {second_policy} is complete", "green"))
del executionhist_policy2
del data2
del exe2
gc.collect()
if not os.path.exists(f"parquet\\combined_hashlist_{first_policy}_{second_policy}.parquet"): if not os.path.exists(f"parquet\\combined_hashlist_{first_policy}_{second_policy}.parquet"):
combined_hashes = pd.DataFrame(columns=['sha256', 'publisher']) utils.hashfunctions.combineHashes(url, first_policy, second_policy)
hashes = []
try:
hash1 = pd.read_parquet(f"parquet\\execution_history_{first_policy}.parquet", columns=['sha256', 'publisher'])
utils.pathfunctions.inspect_parquet(f"parquet\\execution_history_{first_policy}.parquet")
if not hash1.empty:
hashes.append(hash1)
else:
print("⚠️ First dataframe is empty.")
except Exception as e:
print(f"❌ Error reading first Parquet file: {e}")
try:
hash2 = pd.read_parquet(f"parquet\\execution_history_{second_policy}.parquet", columns=['sha256', 'publisher'])
utils.pathfunctions.inspect_parquet(f"parquet\\execution_history_{second_policy}.parquet")
if not hash2.empty:
hashes.append(hash2)
else:
print("⚠️ Second dataframe is empty.")
except Exception as e:
print(f"❌ Error reading second Parquet file: {e}")
if hashes:
combined_hashes = pd.concat(hashes, ignore_index=True)
print(f"✅ Combined {len(combined_hashes)} hashes.")
else:
print("⚠️ No valid dataframes to combine.")
combined_hashes = combined_hashes.drop_duplicates(subset=['sha256'])
augmented_combo = utils.hashfunctions.augmentAggregatedHashes(url, combined_hashes)
numeric_reputation_cols = [
'reputation_scannermatch',
'reputation_scannercount',
'reputation_threatlevel'
]
for col in numeric_reputation_cols:
if col in augmented_combo.columns:
augmented_combo[col] = pd.to_numeric(augmented_combo[col].replace('N/A', pd.NA), errors='coerce')
augmented_combo = augmented_combo.rename(columns={'publisher_x': 'publisher'})
augmented_combo = augmented_combo[['sha256', 'publisher', 'description', 'productname', 'productversion',
'reputation_lastseen', 'reputation_scannermatch', 'reputation_scannercount',
'reputation_status', 'reputation_threatlevel', 'reputation_threatname',
'reputation_timestamp']]
augmented_combo = augmented_combo.sort_values(by=['publisher', 'description', 'productname'])
augmented_combo.to_parquet(f"parquet\\combined_hashlist_{first_policy}_{second_policy}.parquet", index=False)
del combined_hashes
del augmented_combo
gc.collect()
print(ct.colorText("Hash reputation info added to dataframe", "green"))
if not os.path.exists(f"parquet\\hashes_rep_unknown_{first_policy}_{second_policy}.parquet") and not os.path.exists(f"parquet\\hashes_rep_good_{first_policy}_{second_policy}.parquet") and not os.path.exists(f"parquet\\hashes_rep_bad_{first_policy}_{second_policy}.parquet"): if not os.path.exists(f"parquet\\hashes_rep_unknown_{first_policy}_{second_policy}.parquet") and not os.path.exists(f"parquet\\hashes_rep_good_{first_policy}_{second_policy}.parquet") and not os.path.exists(f"parquet\\hashes_rep_bad_{first_policy}_{second_policy}.parquet"):
utils.hashfunctions.categorizeHashes(
# Categorize the hashes first_policy,
categorized = utils.hashfunctions.categorizeHashes( second_policy,
pd.read_parquet(f"parquet\\combined_hashlist_{first_policy}_{second_policy}.parquet"), pd.read_parquet(f"parquet\\combined_hashlist_{first_policy}_{second_policy}.parquet"),
threat_tolerance_constant, threat_tolerance_constant,
bad_publisher_list, bad_publisher_list,
pups pups
) )
categorized[0].to_parquet(f"parquet\\hashes_rep_unknown_{first_policy}_{second_policy}.parquet", index=False)
categorized[1].to_parquet(f"parquet\\hashes_rep_good_{first_policy}_{second_policy}.parquet", index=False)
categorized[2].to_parquet(f"parquet\\hashes_rep_bad_{first_policy}_{second_policy}.parquet", index=False)
del categorized
gc.collect()
if not os.path.exists(f"parquet\\condensed_executions_{first_policy}_{second_policy}.parquet"): if not os.path.exists(f"parquet\\condensed_executions_{first_policy}_{second_policy}.parquet"):
# Condense execution history utils.hashfunctions.condenseExecutions(first_policy,second_policy)
try:
exe1 = pd.read_parquet(f"parquet\\execution_history_{first_policy}.parquet")
utils.pathfunctions.inspect_parquet(f"parquet\\execution_history_{first_policy}.parquet")
if not exe1.empty:
print()
else:
print("⚠️ First dataframe is empty.")
except Exception as e:
print(f"❌ Error reading first Parquet file: {e}")
try:
exe2 = pd.read_parquet(f"parquet\\execution_history_{second_policy}.parquet")
utils.pathfunctions.inspect_parquet(f"parquet\\execution_history_{second_policy}.parquet")
if not exe2.empty:
print()
else:
print("⚠️ Second dataframe is empty.")
except Exception as e:
print(f"❌ Error reading second Parquet file: {e}")
if not exe1.empty and not exe2.empty:
condensed_combo = pd.concat([exe1, exe2], ignore_index=True)
print(f"✅ Combined {len(condensed_combo)} hashes.")
elif exe1.empty:
condensed_combo = exe2
elif exe2.empty:
condensed_combo = exe1
else:
print("⚠️ No valid dataframes to combine.")
condensed_combo.to_parquet(f"parquet\\condensed_executions_{first_policy}_{second_policy}.parquet", index=False)
del condensed_combo
gc.collect()
if os.path.exists(f"parquet\\hashes_rep_unknown_{first_policy}_{second_policy}.parquet") & os.path.exists(f"parquet\\hashes_rep_good_{first_policy}_{second_policy}.parquet") & os.path.exists(f"parquet\\hashes_rep_bad_{first_policy}_{second_policy}.parquet"): if os.path.exists(f"parquet\\hashes_rep_unknown_{first_policy}_{second_policy}.parquet") & os.path.exists(f"parquet\\hashes_rep_good_{first_policy}_{second_policy}.parquet") & os.path.exists(f"parquet\\hashes_rep_bad_{first_policy}_{second_policy}.parquet"):
utils.hashfunctions.divideSortedHashExecutions(first_policy,second_policy,pups)
utils.hashfunctions.combineHashAndHist(f"parquet\\hashes_rep_unknown_{first_policy}_{second_policy}.parquet", first_policy, second_policy)
utils.hashfunctions.combineHashAndHist(f"parquet\\hashes_rep_good_{first_policy}_{second_policy}.parquet", first_policy, second_policy)
utils.hashfunctions.combineHashAndHist(f"parquet\\hashes_rep_bad_{first_policy}_{second_policy}.parquet", first_policy, second_policy)
unknown = pd.read_parquet(f"parquet\\hashes_rep_unknown_{first_policy}_{second_policy}.parquet")
good = pd.read_parquet(f"parquet\\hashes_rep_good_{first_policy}_{second_policy}.parquet")
bad = pd.read_parquet(f"parquet\\hashes_rep_bad_{first_policy}_{second_policy}.parquet")
# Build regex pattern once
pattern = utils.pathfunctions.regulator(pups)
# Move matching rows from unknown and good to bad
bad = pd.concat([
bad,
unknown[unknown["filename"].str.contains(pattern, na=False)],
good[good["filename"].str.contains(pattern, na=False)]
], ignore_index=True)
# Remove matching rows from unknown and good
unknown = unknown[~unknown["filename"].str.contains(pattern, na=False)]
good = good[~good["filename"].str.contains(pattern, na=False)]
unknown.to_csv(f"needs_approved\\hashes_rep_unknown_{first_policy}_{second_policy}.csv",index=False)
good.to_csv(f"needs_approved\\hashes_rep_good_{first_policy}_{second_policy}.csv",index=False)
bad.to_csv(f"needs_approved\\hashes_rep_bad_{first_policy}_{second_policy}.csv",index=False)
ct.style_dataframe_dark(unknown, f"needs_approved\\hashes_rep_unknown_{first_policy}_{second_policy}.html")
ct.style_dataframe_dark(good, f"needs_approved\\hashes_rep_good_{first_policy}_{second_policy}.html")
ct.style_dataframe_dark(bad, f"needs_approved\\hashes_rep_bad_{first_policy}_{second_policy}.html")
elif choice == "3": elif choice == "3":
if os.path.exists(f"approved\\hashes_rep_unknown_{first_policy}_{second_policy}.csv") and os.path.exists(f"approved\\hashes_rep_good_{first_policy}_{second_policy}.csv"): if os.path.exists(f"approved\\hashes_rep_unknown_{first_policy}_{second_policy}.csv") and os.path.exists(f"approved\\hashes_rep_good_{first_policy}_{second_policy}.csv"):
utils.pathfunctions.generatePathReview(first_policy, second_policy, badpathparts, min_files_for_path)
if not os.path.exists(f"parquet\\all_approved_hashes_{first_policy}_{second_policy}.parquet"):
df1 = tryToReadCSV(f"approved\\hashes_rep_unknown_{first_policy}_{second_policy}.csv")
df2 = tryToReadCSV(f"approved\\hashes_rep_good_{first_policy}_{second_policy}.csv")
all_approved_hashes = pd.concat([df1 , df2], ignore_index=True).sort_values(by=['filename'])
print(ct.colorText(f"Approved hash lists have been combined","green"))
all_approved_hashes.to_parquet(f"parquet\\all_approved_hashes_{first_policy}_{second_policy}.parquet", index=False)
del all_approved_hashes
gc.collect()
if not os.path.exists(f"parquet\\path_needs_approved_{first_policy}_{second_policy}.parquet"):
all_approved_hashes = pd.read_parquet(f"parquet\\all_approved_hashes_{first_policy}_{second_policy}.parquet")
print(ct.colorText(f"Beginning calculating longest common filepaths for path exceptions","green"))
haslcp = utils.pathfunctions.split_filepaths_grouped(all_approved_hashes)
haslcp.drop_duplicates()
forbidden = utils.pathfunctions.regulator(badpathparts, True)
forbidden_lcfp = haslcp["longestcfp"].str.contains(forbidden, na=False)
print(ct.colorText("Removing forbidden filepaths for path exceptions", "green"))
# Make a real DataFrame copy before modifying
lcp_not_forbidden = haslcp[~forbidden_lcfp].copy()
#For the review, drop down to only the columns we care, and then group by the commmon file path, consolidating and dropping dupes
lcp_not_forbidden_review = lcp_not_forbidden[['longestcfp', 'middle', 'filename_only', 'sha256']]
# Count unique sha256 per longestcfp
unique_sha_counts = lcp_not_forbidden_review.groupby('longestcfp')['sha256'].nunique().reset_index()
unique_sha_counts.columns = ['longestcfp', 'unique_sha256_count']
# Merge the count back into the original DataFrame
lcp_not_forbidden_review = lcp_not_forbidden_review.merge(unique_sha_counts, on='longestcfp', how='left')
lcp_not_forbidden_review = lcp_not_forbidden_review[lcp_not_forbidden_review['unique_sha256_count'] >= min_files_for_path]
lcp_not_forbidden_review.to_parquet(f"parquet\\path_needs_approved_{first_policy}_{second_policy}.parquet",index=False)
lcp_not_forbidden_review.to_csv(f"needs_approved\\path_needs_approved_{first_policy}_{second_policy}.csv",index=False)
else: else:
print(ct.colorText(f"Please manually approve hashes prior to this step","red")) print(ct.colorText(f"Please manually approve hashes prior to this step","red"))
@@ -424,26 +210,7 @@ def menu_prepare_to_enforce():
if os.path.exists(f"approved\\path_needs_approved_{first_policy}_{second_policy}.csv"): if os.path.exists(f"approved\\path_needs_approved_{first_policy}_{second_policy}.csv"):
if not os.path.exists(f"parquet\\final_hash_approvals_{first_policy}_{second_policy}.parquet") and not os.path.exists(f"parquet\\final_path_exclusions_{first_policy}_{second_policy}.parquet"): if not os.path.exists(f"parquet\\final_hash_approvals_{first_policy}_{second_policy}.parquet") and not os.path.exists(f"parquet\\final_path_exclusions_{first_policy}_{second_policy}.parquet"):
allhashes = pd.read_parquet(f"parquet\\all_approved_hashes_{first_policy}_{second_policy}.parquet") utils.hashfunctions.generatePreflights(first_policy, second_policy)
pathexclusions = tryToReadCSV(f"approved\\path_needs_approved_{first_policy}_{second_policy}.csv")
pathexclusions.to_parquet(f"parquet\\final_path_exclusions_{first_policy}_{second_policy}.parquet", index=False)
allowbyhash = allhashes[~allhashes['sha256'].isin(pathexclusions['sha256'])]
allowbyhash.to_parquet(f"parquet\\final_hash_approvals_{first_policy}_{second_policy}.parquet", index=False)
allowbyhash.sort_values(by=["filename"])
ct.style_dataframe_dark(allowbyhash, f"preflight\\final_hash_approvals_{first_policy}_{second_policy}.html")
ct.style_dataframe_dark(pathexclusions, f"preflight\\final_path_exclusions_{first_policy}_{second_policy}.html")
del allowbyhash
del pathexclusions
gc.collect()
elif choice == "5": elif choice == "5":
@@ -467,47 +234,16 @@ def menu_prepare_to_enforce():
elif choice == "6": elif choice == "6":
if os.path.exists(f"preflight\\final_path_exclusions_{first_policy}_{second_policy}.html") and os.path.exists(f"preflight\\final_hash_approvals_{first_policy}_{second_policy}.html") and allowlist_parent_name != " " and allowlist_child_name != " " and destination_name != " ": if os.path.exists(f"preflight\\final_path_exclusions_{first_policy}_{second_policy}.html") and os.path.exists(f"preflight\\final_hash_approvals_{first_policy}_{second_policy}.html") and allowlist_parent_name != " " and allowlist_child_name != " " and destination_name != " ":
utils.policyfunctions.sendToPolicy(
pathexclusions = pd.read_parquet(f"parquet\\final_path_exclusions_{first_policy}_{second_policy}.parquet") first_policy,
allowbyhash = pd.read_parquet(f"parquet\\final_hash_approvals_{first_policy}_{second_policy}.parquet") second_policy,
destination_name,
ct.areYouSure() destination_id,
confirmation = input(ct.colorText("Type 'I AGREE' to continue: ","white")) allowlist_parent_name,
allowlist_parent_id,
if confirmation.strip().upper() == "I AGREE": allowlist_child_name,
print(ct.colorText("Proceeding with the code...", "yellow")) allowlist_child_id
print(ct.colorText(f"Adding path exclusions to {destination_name}", "yellow")) )
pathexcludelist = pathexclusions['longestcfp'].unique().tolist()
# Regex to match a Windows drive letter at the start (e.g., C:\)
drive_letter_pattern = re.compile(r'^[a-zA-Z]:\\')
# Processed list
processed_paths = [
(path if drive_letter_pattern.match(path) else f"\\\\{path}") + "**"
for path in pathexcludelist
]
utils.policyfunctions.addPath(destination_id,processed_paths)
print(ct.colorText(f"Adding hashes to {allowlist_parent_name}", "yellow"))
allowlist_parenthashlist = allowbyhash[allowbyhash['reputation_status'] == 'KNOWN']['sha256'].unique().tolist()
utils.policyfunctions.addHash(allowlist_parent_id,allowlist_parenthashlist)
print(ct.colorText(f"Adding hashes to {allowlist_child_name}", "yellow"))
allowlist_childhashlist = allowbyhash[allowbyhash['reputation_status'] == 'UNKNOWN']['sha256'].unique().tolist()
utils.policyfunctions.addHash(allowlist_child_id, allowlist_childhashlist)
ct.locked()
exit()
else:
print(ct.colorText("Operation aborted. You MUST EXPLICITLY AGREE to proceed.", "red"))
break
elif choice == "Q": elif choice == "Q":
break break
utils/hashfunctions.py
+159 -8
View File
@@ -12,15 +12,15 @@
# #
# You should have received a copy of the GNU Affero General Public License # 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/>. # along with this program. If not, see <https://www.gnu.org/licenses/>.
import gc
import json
import os
import pandas as pd import pandas as pd
import requests import requests
import os
import json
import utils.pathfunctions as pathf import utils.pathfunctions as pathf
import utils.hashfunctions as hashf
import utils.pretty as ct import utils.pretty as ct
import gc from AirlockTools import tryToReadCSV
def aggregateHashes(executions_json) -> pd.DataFrame: def aggregateHashes(executions_json) -> pd.DataFrame:
""" """
@@ -104,7 +104,7 @@ def augmentAggregatedHashes(url, agg_df: pd.DataFrame) -> pd.DataFrame:
return aug_df return aug_df
def categorizeHashes(df: pd.DataFrame, threat_tolerance: int, untrusted_publishers, pups: list): def categorizeHashes(first_policy, second_policy, df: pd.DataFrame, threat_tolerance: int, untrusted_publishers, pups: list):
if untrusted_publishers is None: untrusted_publishers = [] if untrusted_publishers is None: untrusted_publishers = []
if pups is None: pups = [] if pups is None: pups = []
@@ -144,9 +144,14 @@ def categorizeHashes(df: pd.DataFrame, threat_tolerance: int, untrusted_publishe
approved_df = df[mask_approved] approved_df = df[mask_approved]
unapproved_df = df[~(mask_needsreview | mask_approved)] unapproved_df = df[~(mask_needsreview | mask_approved)]
return needsreview_df, approved_df, unapproved_df needsreview_df.to_parquet(f"parquet\\hashes_rep_unknown_{first_policy}_{second_policy}.parquet", index=False)
approved_df.to_parquet(f"parquet\\hashes_rep_good_{first_policy}_{second_policy}.parquet", index=False)
unapproved_df.to_parquet(f"parquet\\hashes_rep_bad_{first_policy}_{second_policy}.parquet", index=False)
del needsreview_df
del approved_df
del unapproved_df
gc.collect()
def explode_and_deduplicate(df): def explode_and_deduplicate(df):
df['sha256'] = df['sha256'].str.split(',') df['sha256'] = df['sha256'].str.split(',')
@@ -221,4 +226,150 @@ def combineHashAndHist(path, first_policy, second_policy):
df.to_parquet(path, index=False) df.to_parquet(path, index=False)
del df del df
del condensed_combo del condensed_combo
gc.collect()
def combineHashes(url, first_policy, second_policy):
combined_hashes = pd.DataFrame(columns=['sha256', 'publisher'])
hashes = []
try:
hash1 = pd.read_parquet(f"parquet\\execution_history_{first_policy}.parquet", columns=['sha256', 'publisher'])
pathf.inspect_parquet(f"parquet\\execution_history_{first_policy}.parquet")
if not hash1.empty:
hashes.append(hash1)
else:
print("⚠️ First dataframe is empty.")
except Exception as e:
print(f"❌ Error reading first Parquet file: {e}")
try:
hash2 = pd.read_parquet(f"parquet\\execution_history_{second_policy}.parquet", columns=['sha256', 'publisher'])
pathf.inspect_parquet(f"parquet\\execution_history_{second_policy}.parquet")
if not hash2.empty:
hashes.append(hash2)
else:
print("⚠️ Second dataframe is empty.")
except Exception as e:
print(f"❌ Error reading second Parquet file: {e}")
if hashes:
combined_hashes = pd.concat(hashes, ignore_index=True)
print(f"✅ Combined {len(combined_hashes)} hashes.")
else:
print("⚠️ No valid dataframes to combine.")
combined_hashes = combined_hashes.drop_duplicates(subset=['sha256'])
augmented_combo = hashf.augmentAggregatedHashes(url, combined_hashes)
numeric_reputation_cols = [
'reputation_scannermatch',
'reputation_scannercount',
'reputation_threatlevel'
]
for col in numeric_reputation_cols:
if col in augmented_combo.columns:
augmented_combo[col] = pd.to_numeric(augmented_combo[col].replace('N/A', pd.NA), errors='coerce')
augmented_combo = augmented_combo.rename(columns={'publisher_x': 'publisher'})
augmented_combo = augmented_combo[['sha256', 'publisher', 'description', 'productname', 'productversion',
'reputation_lastseen', 'reputation_scannermatch', 'reputation_scannercount',
'reputation_status', 'reputation_threatlevel', 'reputation_threatname',
'reputation_timestamp']]
augmented_combo = augmented_combo.sort_values(by=['publisher', 'description', 'productname'])
augmented_combo.to_parquet(f"parquet\\combined_hashlist_{first_policy}_{second_policy}.parquet", index=False)
del combined_hashes
del augmented_combo
gc.collect()
print(ct.colorText("Hash reputation info added to dataframe", "green"))
def condenseExecutions(first_policy,second_policy):
exe1 = pd.DataFrame()
exe2 = pd.DataFrame()
condensed_combo = pd.DataFrame()
try:
exe1 = pd.read_parquet(f"parquet\\execution_history_{first_policy}.parquet")
pathf.inspect_parquet(f"parquet\\execution_history_{first_policy}.parquet")
if not exe1.empty:
print()
else:
print("⚠️ First dataframe is empty.")
except Exception as e:
print(f"❌ Error reading first Parquet file: {e}")
try:
exe2 = pd.read_parquet(f"parquet\\execution_history_{second_policy}.parquet")
pathf.inspect_parquet(f"parquet\\execution_history_{second_policy}.parquet")
if not exe2.empty:
print()
else:
print("⚠️ Second dataframe is empty.")
except Exception as e:
print(f"❌ Error reading second Parquet file: {e}")
if not exe1.empty and not exe2.empty:
condensed_combo = pd.concat([exe1, exe2], ignore_index=True)
print(f"✅ Combined {len(condensed_combo)} hashes.")
elif exe1.empty:
condensed_combo = exe2
elif exe2.empty:
condensed_combo = exe1
else:
print("⚠️ No valid dataframes to combine.")
condensed_combo.to_parquet(f"parquet\\condensed_executions_{first_policy}_{second_policy}.parquet", index=False)
del condensed_combo
gc.collect()
def divideSortedHashExecutions(first_policy,second_policy, pups):
combineHashAndHist(f"parquet\\hashes_rep_unknown_{first_policy}_{second_policy}.parquet", first_policy, second_policy)
combineHashAndHist(f"parquet\\hashes_rep_good_{first_policy}_{second_policy}.parquet", first_policy, second_policy)
combineHashAndHist(f"parquet\\hashes_rep_bad_{first_policy}_{second_policy}.parquet", first_policy, second_policy)
unknown = pd.read_parquet(f"parquet\\hashes_rep_unknown_{first_policy}_{second_policy}.parquet")
good = pd.read_parquet(f"parquet\\hashes_rep_good_{first_policy}_{second_policy}.parquet")
bad = pd.read_parquet(f"parquet\\hashes_rep_bad_{first_policy}_{second_policy}.parquet")
# Build regex pattern once
pattern = pathf.regulator(pups)
# Move matching rows from unknown and good to bad
bad = pd.concat([
bad,
unknown[unknown["filename"].str.contains(pattern, na=False)],
good[good["filename"].str.contains(pattern, na=False)]
], ignore_index=True)
# Remove matching rows from unknown and good
unknown = unknown[~unknown["filename"].str.contains(pattern, na=False)]
good = good[~good["filename"].str.contains(pattern, na=False)]
unknown.to_csv(f"needs_approved\\hashes_rep_unknown_{first_policy}_{second_policy}.csv",index=False)
good.to_csv(f"needs_approved\\hashes_rep_good_{first_policy}_{second_policy}.csv",index=False)
bad.to_csv(f"needs_approved\\hashes_rep_bad_{first_policy}_{second_policy}.csv",index=False)
ct.style_dataframe_dark(unknown, f"needs_approved\\hashes_rep_unknown_{first_policy}_{second_policy}.html")
ct.style_dataframe_dark(good, f"needs_approved\\hashes_rep_good_{first_policy}_{second_policy}.html")
ct.style_dataframe_dark(bad, f"needs_approved\\hashes_rep_bad_{first_policy}_{second_policy}.html")
def generatePreflights(first_policy, second_policy):
allhashes = pd.read_parquet(f"parquet\\all_approved_hashes_{first_policy}_{second_policy}.parquet")
pathexclusions = tryToReadCSV(f"approved\\path_needs_approved_{first_policy}_{second_policy}.csv")
pathexclusions.to_parquet(f"parquet\\final_path_exclusions_{first_policy}_{second_policy}.parquet", index=False)
allowbyhash = allhashes[~allhashes['sha256'].isin(pathexclusions['sha256'])]
allowbyhash.to_parquet(f"parquet\\final_hash_approvals_{first_policy}_{second_policy}.parquet", index=False)
allowbyhash.sort_values(by=["filename"])
ct.style_dataframe_dark(allowbyhash, f"preflight\\final_hash_approvals_{first_policy}_{second_policy}.html")
ct.style_dataframe_dark(pathexclusions, f"preflight\\final_path_exclusions_{first_policy}_{second_policy}.html")
del allowbyhash
del pathexclusions
gc.collect() gc.collect()
utils/pathfunctions.py
+52 -3
View File
@@ -12,11 +12,14 @@
# #
# You should have received a copy of the GNU Affero General Public License # 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/>. # along with this program. If not, see <https://www.gnu.org/licenses/>.
import pandas as pd
import os
import ast import ast
import gc
import os
import pandas as pd
import re import re
import utils.pathfunctions as pathf
import utils.pretty as ct
from AirlockTools import tryToReadCSV
def split_filepaths_grouped(df, col="filename", group_parts=3, min_parts=3): def split_filepaths_grouped(df, col="filename", group_parts=3, min_parts=3):
@@ -133,3 +136,49 @@ def regulator(paths, case_insensitive=True):
print(f"Regulator is providing: {pattern}") print(f"Regulator is providing: {pattern}")
return pattern return pattern
def generatePathReview(first_policy, second_policy, badpathparts, min_files_for_path):
if not os.path.exists(f"parquet\\all_approved_hashes_{first_policy}_{second_policy}.parquet"):
df1 = tryToReadCSV(f"approved\\hashes_rep_unknown_{first_policy}_{second_policy}.csv")
df2 = tryToReadCSV(f"approved\\hashes_rep_good_{first_policy}_{second_policy}.csv")
all_approved_hashes = pd.concat([df1 , df2], ignore_index=True).sort_values(by=['filename'])
print(ct.colorText(f"Approved hash lists have been combined","green"))
all_approved_hashes.to_parquet(f"parquet\\all_approved_hashes_{first_policy}_{second_policy}.parquet", index=False)
del all_approved_hashes
gc.collect()
if not os.path.exists(f"parquet\\path_needs_approved_{first_policy}_{second_policy}.parquet"):
all_approved_hashes = pd.read_parquet(f"parquet\\all_approved_hashes_{first_policy}_{second_policy}.parquet")
print(ct.colorText(f"Beginning calculating longest common filepaths for path exceptions","green"))
haslcp = pathf.split_filepaths_grouped(all_approved_hashes)
haslcp.drop_duplicates()
forbidden = pathf.regulator(badpathparts, True)
forbidden_lcfp = haslcp["longestcfp"].str.contains(forbidden, na=False)
print(ct.colorText("Removing forbidden filepaths for path exceptions", "green"))
# Make a real DataFrame copy before modifying
lcp_not_forbidden = haslcp[~forbidden_lcfp].copy()
#For the review, drop down to only the columns we care, and then group by the commmon file path, consolidating and dropping dupes
lcp_not_forbidden_review = lcp_not_forbidden[['longestcfp', 'middle', 'filename_only', 'sha256']]
# Count unique sha256 per longestcfp
unique_sha_counts = lcp_not_forbidden_review.groupby('longestcfp')['sha256'].nunique().reset_index()
unique_sha_counts.columns = ['longestcfp', 'unique_sha256_count']
# Merge the count back into the original DataFrame
lcp_not_forbidden_review = lcp_not_forbidden_review.merge(unique_sha_counts, on='longestcfp', how='left')
lcp_not_forbidden_review = lcp_not_forbidden_review[lcp_not_forbidden_review['unique_sha256_count'] >= min_files_for_path]
lcp_not_forbidden_review.to_parquet(f"parquet\\path_needs_approved_{first_policy}_{second_policy}.parquet",index=False)
lcp_not_forbidden_review.to_csv(f"needs_approved\\path_needs_approved_{first_policy}_{second_policy}.csv",index=False)
utils/policyfunctions.py
+62 -2
View File
@@ -12,11 +12,17 @@
# #
# You should have received a copy of the GNU Affero General Public License # 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/>. # along with this program. If not, see <https://www.gnu.org/licenses/>.
import gc
import requests
import json import json
import os import os
import pandas as pd
import re
import requests
import utils.pretty as ct import utils.pretty as ct
import utils.allowlist
def addHash(policy, hash): def addHash(policy, hash):
print(f"Adding the following hashes to {policy}:") print(f"Adding the following hashes to {policy}:")
@@ -65,3 +71,57 @@ def addPathReal(url, grouplistID, pathlist):
except requests.exceptions.RequestException as e: except requests.exceptions.RequestException as e:
return {"error": str(e)} return {"error": str(e)}
def getPolicyInfo(url, policy, days):
executionhist_policy = pd.DataFrame()
exehist = utils.allowlist.pullPolicyExechistories(url, policy, days, True)
data = json.loads(exehist)
executionhist_policy = pd.DataFrame(data["response"]["exechistories"])
if not executionhist_policy.empty:
executionhist_policyxecutionhist_policy = executionhist_policy[['sha256', 'publisher', 'filename', 'hostname', 'username', 'pprocess', 'gprocess', 'commandline']]
executionhist_policy = executionhist_policy.drop_duplicates(subset=['sha256', 'filename', 'hostname'])
executionhist_policy = executionhist_policy.sort_values(by=['sha256', 'filename'])
executionhist_policy.to_parquet(f"parquet\\execution_history_{policy}.parquet", index=False)
print(ct.colorText(f"Staging of Execution history for policy: {policy} is complete", "green"))
del data
del exehist
gc.collect()
return executionhist_policy
def sendToPolicy(first_policy, second_policy, destination_name, destination_id, allowlist_parent_name, allowlist_parent_id, allowlist_child_name, allowlist_child_id):
pathexclusions = pd.read_parquet(f"parquet\\final_path_exclusions_{first_policy}_{second_policy}.parquet")
allowbyhash = pd.read_parquet(f"parquet\\final_hash_approvals_{first_policy}_{second_policy}.parquet")
ct.areYouSure()
confirmation = input(ct.colorText("Type 'I AGREE' to continue: ","white"))
if confirmation.strip().upper() == "I AGREE":
print(ct.colorText("Proceeding with the code...", "yellow"))
print(ct.colorText(f"Adding path exclusions to {destination_name}", "yellow"))
pathexcludelist = pathexclusions['longestcfp'].unique().tolist()
# Regex to match a Windows drive letter at the start (e.g., C:\)
drive_letter_pattern = re.compile(r'^[a-zA-Z]:\\')
# Processed list
processed_paths = [
(path if drive_letter_pattern.match(path) else f"\\\\{path}") + "**"
for path in pathexcludelist
]
addPath(destination_id,processed_paths)
print(ct.colorText(f"Adding hashes to {allowlist_parent_name}", "yellow"))
allowlist_parenthashlist = allowbyhash[allowbyhash['reputation_status'] == 'KNOWN']['sha256'].unique().tolist()
addHash(allowlist_parent_id,allowlist_parenthashlist)
print(ct.colorText(f"Adding hashes to {allowlist_child_name}", "yellow"))
allowlist_childhashlist = allowbyhash[allowbyhash['reputation_status'] == 'UNKNOWN']['sha256'].unique().tolist()
addHash(allowlist_child_id, allowlist_childhashlist)
ct.locked()
exit()
else:
print(ct.colorText("Operation aborted. You MUST EXPLICITLY AGREE to proceed.", "red"))
utils/pretty.py
-6
View File
@@ -1,6 +1,5 @@
import os import os
def colorText(text: str, color: str) -> str: def colorText(text: str, color: str) -> str:
colors = { colors = {
"red": "\033[91m", "red": "\033[91m",
@@ -240,11 +239,6 @@ def printEnforceChecklist(first_policy, second_policy, allowlist_child_name, all
else: else:
print(colorText(" [✗] The combined approved hashes list has not been generated","red")) print(colorText(" [✗] The combined approved hashes list has not been generated","red"))
if os.path.exists(f"parquet\\approved_hashes_with_paths_{first_policy}_{second_policy}.parquet"):
print(colorText(" [✓] Longest common filepaths have been generated and appended to hash info","green"))
else:
print(colorText(" [✗] Longest common filepaths have not been generated","red"))
if os.path.exists(f"needs_approved\\path_needs_approved_{first_policy}_{second_policy}.csv"): if os.path.exists(f"needs_approved\\path_needs_approved_{first_policy}_{second_policy}.csv"):
print(colorText(" [✓] Path review list created","green")) print(colorText(" [✓] Path review list created","green"))
else: else: