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

Added Pup Filter and Path Dedupe

Browse Source
This commit is contained in:
=
2025-09-02 20:15:55 -04:00
parent 6a684d71ec
commit e323ac71e9
3 changed files with 173 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils/hashfunctions.py
+11 -7
View File
@@ -17,8 +17,10 @@ import pandas as pd
import requests
import os
import json
import utils.pathfunctions as pathf
import utils.pretty as ct
def aggregateHashes(executions_json) -> pd.DataFrame:
"""
Takes the executions, aggregates all the data with sha256 as primary, then returns aggregated dataframe
@@ -101,11 +103,9 @@ def augmentAggregatedHashes(url, agg_df: pd.DataFrame) -> pd.DataFrame:
return aug_df
def categorizeHashes(aug_df: pd.DataFrame, threat_tolerance: int, untrusted_publishers: list):
if untrusted_publishers is None:
untrusted_publishers = []
df = aug_df.copy()
def categorizeHashes(df: pd.DataFrame, threat_tolerance: int, untrusted_publishers: list, pups: list):
if untrusted_publishers is None: untrusted_publishers = []
if pups is None: pups = []
def reputationtool(row):
val = row["reputation_scannermatch"]
@@ -127,13 +127,15 @@ def categorizeHashes(aug_df: pd.DataFrame, threat_tolerance: int, untrusted_publ
(
(df["publisher"] != "Not Signed") &
~df["publisher"].isin(untrusted_publishers) &
~df["reputation_status"].isna()
~df["reputation_status"].isna() &
~df["description"].str.contains(pathf.regulator(pups), case=False, na=False)
) |
(
(df["publisher"] == "Not Signed") &
~df["reputation_flag"] &
~df["publisher"].isin(untrusted_publishers) &
~df["reputation_status"].isna()
~df["reputation_status"].isna() &
~df["description"].str.contains(pathf.regulator(pups), case=False, na=False)
)
)
@@ -143,6 +145,8 @@ def categorizeHashes(aug_df: pd.DataFrame, threat_tolerance: int, untrusted_publ
return needsreview_df, approved_df, unapproved_df
def explode_and_deduplicate(df):
df['sha256'] = df['sha256'].str.split(',')
df = df.explode('sha256')
utils/pretty.py
+118
View File
@@ -1,3 +1,5 @@
import os
def colorText(text: str, color: str) -> str:
colors = {
@@ -176,6 +178,122 @@ def displayIntro():
print(colorText("======================== Welcome to the Airlock API Tool ========================", "cyan"))
print(colorText("=================================================================================", "cyan"))
def printEnforceChecklist(first_policy, second_policy, allowlist_child_name, allowlist_parent_name, destination_name):
print(colorText("\n --------------------------------------------------------------------", "cyan"))
print(colorText(" -------------------- Prepare to Enforce Policy ---------------------", "cyan"))
print(colorText(" --------------------------------------------------------------------", "cyan"))
print(colorText("\nSequentually follow these steps to prepare a policy for enforcement:", "white"))
print(colorText("\n1. Choose which originating policy or policies to move to enforcement", "cyan"))
if first_policy == " " and second_policy == " ":
print(colorText(f" [✗] No policies have been chosen","red"))
elif first_policy != " " and second_policy is first_policy:
print(colorText(f" [✓] {first_policy} has been selected,", "green"))
elif first_policy != " " and second_policy != " ":
print(colorText(f" [✓] {first_policy} has been selected as Policy 1","green"))
print(colorText(f" [✓] {second_policy} has been selected as Policy 2","green"))
print(colorText("2. Pull and stage event history, combine the histories, add hash info, then categorize the hashes", "cyan"))
if os.path.exists(f"parquet\\execution_history_{first_policy}.parquet"):
print(colorText(f" [✓] Execution history has been compiled for {first_policy}","green"))
elif not os.path.exists(f"parquet\\execution_history_{first_policy}.parquet"):
print(colorText(f" [✗] Execution history has not been compiled for {first_policy}","red"))
elif second_policy is not first_policy and os.path.exists(f"parquet\\execution_history_{second_policy}.parquet"):
print(colorText(f" [✓] Execution history has been compiled for {second_policy}","green"))
elif second_policy is not first_policy and not os.path.exists(f"parquet\\execution_history_{second_policy}.parquet"):
print(colorText(f" [✗] Execution history has not been compiled for {second_policy}","red"))
if os.path.exists(f"parquet\\combined_hashlist_{first_policy}_{second_policy}.parquet"):
print(colorText(f" [✓] Hash Info has been added to the combined execution history", "green"))
else:
print(colorText(f" [✗] Hash Info has not been added to the combined execution history", "red"))
if os.path.exists(f"parquet\\hashes_rep_unknown_{first_policy}_{second_policy}.parquet") and os.path.exists(f"parquet\\hashes_rep_good_{first_policy}_{second_policy}.parquet") and os.path.exists(f"parquet\\hashes_rep_bad_{first_policy}_{second_policy}.parquet"):
print(colorText(f" [✓] Hashes have been cateogrized", "green"))
else:
print(colorText(f" [✗] Hashes have not been cateogrized", "red"))
if os.path.exists(f"parquet\\condensed_executions_{first_policy}_{second_policy}.parquet"):
print(colorText(f" [✓] Execution history has been_combined_for {first_policy} and_{second_policy}", "green"))
else:
print(colorText(f" [✗] Execution history has not been_combined_for {first_policy} and_{second_policy}", "red"))
print(colorText(f"3. Manually review the files:","cyan"))
print(colorText(" 'needs_approved\\hashes_rep_good_{first_policy}_{second_policy}.csv' and 'needs_approved\\hashes_rep_unknown_{first_policy}_{second_policy}.csv'", "cyan"))
print(colorText(" Remove the rows containing hashes you do not approve of, and those you would not approve of without metarules.", "cyan"))
print(colorText(" If metarules need to be created, please make note of them, and remove the row from the csv.", "cyan"))
print(colorText(" When complete, save both csv files to the directory 'approved' and choose this option.","cyan"))
print(colorText(" This will combine these approved hashes with the automatically approved hashes and generate a list of paths to be reviewed", "cyan"))
if os.path.exists(f"approved\\hashes_rep_good_{first_policy}_{second_policy}.csv") and os.path.exists(f"approved\\hashes_rep_unknown_{first_policy}_{second_policy}.csv"):
print(colorText(" [✓] Reviewed hashes have been loaded","green"))
else:
print(colorText(" [✗] Reviewed hashes have not been loaded","red"))
if os.path.exists(f"parquet\\all_approved_hashes_{first_policy}_{second_policy}.parquet"):
print(colorText(" [✓] The combined approved hashes list has been generated","green"))
else:
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"):
print(colorText(" [✓] Path review list created","green"))
else:
print(colorText(" [✗] Path review list has not been created","red"))
print(colorText(f"4. Manually review the file 'needs_approved\\paths_needing_review_{first_policy}_{second_policy}.csv'", "cyan"))
print(colorText(" Remove the rows containing path exclusions you do not approve of" , "cyan"))
print(colorText(" When complete, save the csv file to the directory 'approved'", "cyan"))
print(colorText(" Preflight Lists will be generated", "cyan"))
if os.path.exists(f"approved\\path_needs_approved_{first_policy}_{second_policy}.csv"):
print(colorText(" [✓] Reviewed path list detected","green"))
else:
print(colorText(" [✗] Path review list has not been detected","red"))
if os.path.exists(f"preflight\\final_path_exclusions_{first_policy}_{second_policy}.html"):
print(colorText(" [✓] Preflight Path Exclusion List has been generated","green"))
else:
print(colorText(" [✗] Preflight Path Exclusion List has not been generated","red"))
if os.path.exists(f"preflight\\final_hash_approvals_{first_policy}_{second_policy}.html"):
print(colorText(" [✓] Preflight hash approval list has been generated","green"))
else:
print(colorText(" [✗] Preflight hash approval list has not been generated","red"))
print(colorText(f"5. Choose the destination policy and parent and child allow list", "cyan"))
if allowlist_child_name == " " and allowlist_parent_name== " ":
print(colorText(f" [✗] No allowlists have been chosen","red"))
elif allowlist_parent_name != " " and allowlist_child_name != " " and allowlist_parent_name is allowlist_child_name:
print(colorText(f" [✓] [✗] Only {allowlist_parent_name} has been selected this is unusual, but potentially valid case, double check before proceeding,", "yellow"))
elif allowlist_parent_name != " " and allowlist_child_name != " " and allowlist_parent_name is not allowlist_child_name:
print(colorText(f" [✓] {allowlist_parent_name} has been selected as Parent Policy","green"))
print(colorText(f" [✓] {allowlist_child_name} has been selected as Child Policy","green"))
if destination_name == " ":
print(colorText(f" [✗] No destination policy has been chosen","red"))
else:
print(colorText(f" [✓] destination policy is {destination_name}","green"))
print(colorText(f"6. Liftoff ------------------------------------------------------", "cyan"))
print(colorText(f" Apply path exclusions according to allowed and approved paths", "cyan"))
print(colorText(f" Apply signed or attested hashes to Parent Allow List", "cyan"))
print(colorText(f" Apply approved, but unsigned hashes to the Child Allow List", "cyan"))
print(colorText("Q. Quit", "cyan"))
def areYouSure():
print(colorText(f"*******************************************************************************************************************************************","red"))