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

MemOpt #18

Merged
mysticmomba merged 30 commits from MemOpt into master 2025-09-02 17:53:43 -04:00
Conversation 0 Commits 30 Files Changed 9
+440 -158
14 changed files with 440 additions and 158 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 67615f364c - Show all commits
AirlockTools.py
+306 -144
View File
@@ -14,6 +14,7 @@
# 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 json
import os import os
import pandas as pd import pandas as pd
@@ -32,23 +33,13 @@ dotenv.load_dotenv()
#Constants #Constants
url = os.getenv('url') url = os.getenv('url')
badpublisherlist = ["Brave Software, Inc.", "Zoom Video Communications, Inc."] badpublisherlist = ["Brave Software, Inc.", "Zoom Video Communications, Inc."]
badpathparts = ["users", "inet\\wwwroot", "windows\\temp", "windows\\temp", "windows\\task", "windows\\system32", "startup", "windows\\fonts", "Recycle.Bin", "AppData", "programdata"]
path_exclusion_constant = 3 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():
print(ct.colorText(r"""
_____ .__ .__ __ ___________ .__
/ _ \ |__|______| | ____ ____ | | __ \__ ___/___ ____ | | ______
/ /_\ \| \_ __ \ | / _ \_/ ___\| |/ / | | / _ \ / _ \| | / ___/
/ | \ || | \/ |_( <_> ) \___| < | |( <_> | <_> ) |__\___ \
\____|__ /__||__| |____/\____/ \___ >__|_ \ |____| \____/ \____/|____/____ >
\/ \/ \/ \/
""", "cyan"))
print(ct.colorText("=================================================================================", "cyan"))
print(ct.colorText("======================== Welcome to the Airlock API Tool ========================", "cyan"))
print(ct.colorText("=================================================================================", "cyan"))
match os.getenv('APIKEY'): match os.getenv('APIKEY'):
case '': case '':
print(ct.colorText("Please add your API Key to the .env file", "red")) print(ct.colorText("Please add your API Key to the .env file", "red"))
@@ -67,11 +58,25 @@ def tryToReadCSV(csv):
df = pd.DataFrame() # Create an empty DataFrame as fallback df = pd.DataFrame() # Create an empty DataFrame as fallback
return df return df
def tryToReadParquet(parquet):
try:
df = pd.read_parquet(parquet)
if df.empty:
print(ct.colorText("Error: Parquet file has headers but no data rows.", "red"))
else:
print(ct.colorText(f"Data loaded successfully from {parquet}", "green"))
except pd.errors.EmptyDataError:
print(ct.colorText("Notice : Parquet file is completely empty (no headers, no data), falling back to empty frame", "white"))
df = pd.DataFrame() # Create an empty DataFrame as fallback
return df
def deduplicate_list(lst):
seen = set()
return [x for x in lst if not (x in seen or seen.add(x))]
def menu_main(): def menu_main():
while True: while True:
print(ct.colorText("\n-----------------------------------", "magenta")) ct.displayIntro();
print(ct.colorText("------------ Main Menu ------------", "magenta"))
print(ct.colorText("-----------------------------------", "magenta"))
print(ct.colorText("1. Get All Events for Single Device", "yellow")) print(ct.colorText("1. Get All Events for Single Device", "yellow"))
print(ct.colorText("2. Placeholder for Local Approval", "yellow")) print(ct.colorText("2. Placeholder for Local Approval", "yellow"))
print(ct.colorText("3. Placeholder for Another Tool", "yellow")) print(ct.colorText("3. Placeholder for Another Tool", "yellow"))
@@ -138,9 +143,9 @@ def menu_prepare_to_enforce():
df_aggregated_combo = pd.DataFrame() df_aggregated_combo = pd.DataFrame()
#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("dataframe_html"): os.makedirs("dataframe_html") if not os.path.exists("parquet"): os.makedirs("parquet")
if not os.path.exists("dataframe_csv"): os.makedirs("dataframe_csv") if not os.path.exists("needs_approved"): os.makedirs("needs_approved")
if not os.path.exists("manuallyapproved"): os.makedirs("manuallyapproved") if not os.path.exists("approved"): os.makedirs("approved")
if not os.path.exists("preflight"): os.makedirs("preflight") if not os.path.exists("preflight"): os.makedirs("preflight")
while True: while True:
@@ -163,54 +168,55 @@ def menu_prepare_to_enforce():
print(ct.colorText("2. Pulls and stages event history, combines the histories, adds hash info, then categorizes the hashes", "cyan")) print(ct.colorText("2. Pulls and stages event history, combines the histories, adds hash info, then categorizes the hashes", "cyan"))
if os.path.exists(f"dataframe_csv\\executionhist_{first_policy}.csv"): if os.path.exists(f"parquet\\execution_history_{first_policy}.parquet"):
print(ct.colorText(f" [✓] Execution history has been compiled for {first_policy}","green")) print(ct.colorText(f" [✓] Execution history has been compiled for {first_policy}","green"))
elif not os.path.exists(f"dataframe_csv\\executionhist_{first_policy}.csv"): elif not os.path.exists(f"parquet\\execution_history_{first_policy}.parquet"):
print(ct.colorText(f" [✗] Execution history has not been compiled for {first_policy}","red")) print(ct.colorText(f" [✗] Execution history has not been compiled for {first_policy}","red"))
elif second_policy is not first_policy and os.path.exists(f"dataframe_csv\\executionhist_{second_policy}.csv"): elif second_policy is not first_policy and os.path.exists(f"parquet\\execution_history_{second_policy}.parquet"):
print(ct.colorText(f" [✓] Execution history has been compiled for {second_policy}","green")) print(ct.colorText(f" [✓] Execution history has been compiled for {second_policy}","green"))
elif second_policy is not first_policy and not os.path.exists(f"dataframe_csv\\executionhist_{second_policy}.csv"): elif second_policy is not first_policy and not os.path.exists(f"parquet\\execution_history_{second_policy}.parquet"):
print(ct.colorText(f" [✗] Execution history has not been compiled for {second_policy}","red")) print(ct.colorText(f" [✗] Execution history has not been compiled for {second_policy}","red"))
if os.path.exists(f"dataframe_csv\\execuctionhist_combined_{first_policy}_{second_policy}.csv"): if os.path.exists(f"parquet\\combined_hashlist_{first_policy}_{second_policy}.parquet"):
print(ct.colorText(f" [✓] Execution history has been_combined_for {first_policy} and_{second_policy}", "green"))
else:
print(ct.colorText(f" [✗] Execution history has not been_combined_for {first_policy} and_{second_policy}", "red"))
if os.path.exists(f"dataframe_csv\\augmented_combo_{first_policy}_{second_policy}.csv"):
print(ct.colorText(f" [✓] Hash Info has been added to the combined execution history", "green")) print(ct.colorText(f" [✓] Hash Info has been added to the combined execution history", "green"))
else: else:
print(ct.colorText(f" [✗] Hash Info has not been added to the combined execution history", "red")) print(ct.colorText(f" [✗] Hash Info has not been added to the combined execution history", "red"))
if os.path.exists(f"dataframe_csv\\hashes_needing_approval_{first_policy}_{second_policy}.csv") and os.path.exists(f"dataframe_csv\\automatically_approved_hashes_{first_policy}_{second_policy}.csv") and os.path.exists(f"dataframe_csv\\unapproved_hashes__{first_policy}_{second_policy}.csv"): 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(ct.colorText(f" [✓] Hashes have been cateogrized", "green")) print(ct.colorText(f" [✓] Hashes have been cateogrized", "green"))
else: else:
print(ct.colorText(f" [✗] Hashes have not been cateogrized", "red")) print(ct.colorText(f" [✗] Hashes have not been cateogrized", "red"))
if os.path.exists(f"parquet\\condensed_executions_{first_policy}_{second_policy}.parquet"):
print(ct.colorText(f" [✓] Execution history has been_combined_for {first_policy} and_{second_policy}", "green"))
else:
print(ct.colorText(f" [✗] Execution history has not been_combined_for {first_policy} and_{second_policy}", "red"))
print(ct.colorText(f"3. Manually review the files:","cyan")) print(ct.colorText(f"3. Manually review the files:","cyan"))
print(ct.colorText(" '\\dataframe_csv\\hashes_needing_approval_{first_policy}_{second_policy}.csv' and 'dataframe_csv\\automatically_approved_hashes_{first_policy}_{second_policy}.csv'", "cyan")) print(ct.colorText(" 'needs_approved\\hashes_rep_good_{first_policy}_{second_policy}.csv' and 'needs_approved\\hashes_rep_unknown_{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(" 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(" If metarules need to be created, please make note of them, and remove the row from the csv.", "cyan"))
print(ct.colorText(" When complete, save both csv files to the directory 'manuallyapproved' and choose this option.","cyan")) print(ct.colorText(" When complete, save both csv files to the directory 'approved' and choose this option.","cyan"))
print(ct.colorText(" This will combine these approved hashes with the automatically approved hashes and generate a list of paths to be reviewed", "cyan")) print(ct.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"manuallyapproved\\hashes_needing_approval_{first_policy}_{second_policy}.csv") and os.path.exists(f"manuallyapproved\\automatically_approved_hashes_{first_policy}_{second_policy}.csv"): 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(ct.colorText(" [✓] Reviewed hashes have been loaded","green")) print(ct.colorText(" [✓] Reviewed hashes have been loaded","green"))
else: else:
print(ct.colorText(" [✗] Reviewed hashes have not been loaded","red")) print(ct.colorText(" [✗] Reviewed hashes have not been loaded","red"))
if os.path.exists(f"dataframe_csv\\all_approved_hashes_{first_policy}_{second_policy}.csv"): if os.path.exists(f"parquet\\all_approved_hashes_{first_policy}_{second_policy}.parquet"):
print(ct.colorText(" [✓] The combined approved hashes list has been generated","green")) print(ct.colorText(" [✓] The combined approved hashes list has been generated","green"))
else: else:
print(ct.colorText(" [✗] The combined approved hashes list has not been generated","red")) print(ct.colorText(" [✗] The combined approved hashes list has not been generated","red"))
if os.path.exists(f"dataframe_csv\\allinfo_{first_policy}_{second_policy}.csv"): if os.path.exists(f"parquet\\approved_hashes_with_paths_{first_policy}_{second_policy}.parquet"):
print(ct.colorText(" [✓] Longest common filepaths have been generated and appended to hash info","green")) print(ct.colorText(" [✓] Longest common filepaths have been generated and appended to hash info","green"))
else: else:
print(ct.colorText(" [✗] Longest common filepaths have not been generated","red")) print(ct.colorText(" [✗] Longest common filepaths have not been generated","red"))
if os.path.exists(f"dataframe_html\\lcf_needs_appoved_{first_policy}_{second_policy}.html"): if os.path.exists(f"needs_approved\\path_needs_approved_{first_policy}_{second_policy}.csv"):
print(ct.colorText(" [✓] Path review list created","green")) print(ct.colorText(" [✓] Path review list created","green"))
else: else:
print(ct.colorText(" [✗] Path review list has not been created","red")) print(ct.colorText(" [✗] Path review list has not been created","red"))
@@ -218,20 +224,20 @@ def menu_prepare_to_enforce():
print(ct.colorText(f"4. Manually review the file 'paths_needing_review_{first_policy}_{second_policy}.csv'", "cyan")) print(ct.colorText(f"4. Manually review the file 'paths_needing_review_{first_policy}_{second_policy}.csv'", "cyan"))
print(ct.colorText(" Remove the rows containing path exclusions you do not approve of" , "cyan")) print(ct.colorText(" Remove the rows containing path exclusions you do not approve of" , "cyan"))
print(ct.colorText(" When complete, save the csv file to the directory 'manuallyapproved'", "cyan")) print(ct.colorText(" When complete, save the csv file to the directory 'approved'", "cyan"))
print(ct.colorText(" Preflight Lists will be generated", "cyan")) print(ct.colorText(" Preflight Lists will be generated", "cyan"))
if os.path.exists(f"manuallyapproved\\lcf_needs_approved_{first_policy}_{second_policy}.csv"): if os.path.exists(f"approved\\path_needs_approved_{first_policy}_{second_policy}.csv"):
print(ct.colorText(" [✓] Reviewed path list detected","green")) print(ct.colorText(" [✓] Reviewed path list detected","green"))
else: else:
print(ct.colorText(" [✗] Path review list has not been detected","red")) print(ct.colorText(" [✗] Path review list has not been detected","red"))
if os.path.exists(f"preflight\\final_path_exclusions_{first_policy}_{second_policy}.csv"): if os.path.exists(f"preflight\\final_path_exclusions_{first_policy}_{second_policy}.html"):
print(ct.colorText(" [✓] Preflight Path Exclusion List has been generated","green")) print(ct.colorText(" [✓] Preflight Path Exclusion List has been generated","green"))
else: else:
print(ct.colorText(" [✗] Preflight Path Exclusion List has not been generated","red")) print(ct.colorText(" [✗] Preflight Path Exclusion List has not been generated","red"))
if os.path.exists(f"preflight\\final_hash_approvals_{first_policy}_{second_policy}.csv"): if os.path.exists(f"preflight\\final_hash_approvals_{first_policy}_{second_policy}.html"):
print(ct.colorText(" [✓] Preflight hash approval list has been generated","green")) print(ct.colorText(" [✓] Preflight hash approval list has been generated","green"))
else: else:
print(ct.colorText(" [✗] Preflight hash approval list has not been generated","red")) print(ct.colorText(" [✗] Preflight hash approval list has not been generated","red"))
@@ -277,114 +283,262 @@ def menu_prepare_to_enforce():
print(ct.colorText("Please answer with 'yes' or 'no'.", "red")) print(ct.colorText("Please answer with 'yes' or 'no'.", "red"))
elif choice == "2": elif choice == "2":
if not os.path.exists(f"dataframe_csv\\executionhist_{first_policy}.csv"): if not os.path.exists(f"parquet\\execution_history_{first_policy}.parquet"):
print(choice) print(choice)
print(first_policy) print(first_policy)
exe1 = utils.allowlist.pullPolicyExechistories(url, choice, first_policy, True)
exe1 = utils.allowlist.pullPolicyExechistories(url, choice, first_policy ,True)
data = json.loads(exe1) data = json.loads(exe1)
executionhist_policy1 = pd.DataFrame(data["response"]["exechistories"]) executionhist_policy1 = pd.DataFrame(data["response"]["exechistories"])
if not executionhist_policy1.empty:
executionhist_policy1 = executionhist_policy1.sort_values(by=['sha256','filename'])
executionhist_policy1.to_csv(f"dataframe_csv\\executionhist_{first_policy}.csv", index=False) if not executionhist_policy1.empty:
ct.style_dataframe_dark(executionhist_policy1, f"dataframe_html\\executionhist_{first_policy}.html") executionhist_policy1 = executionhist_policy1[['sha256', 'publisher', 'filename', 'hostname', 'username', 'pprocess', 'gprocess', 'commandline']]
print(ct.colorText(f"Staging of Execution history for policy: {first_policy} is complete","green")) executionhist_policy1 = executionhist_policy1.drop_duplicates(subset=['sha256', 'filename', 'hostname'])
executionhist_policy1 = executionhist_policy1.sort_values(by=['sha256', 'filename'])
if not os.path.exists(f"dataframe_csv\\executionhist_{second_policy}.csv"):
exe2 = utils.allowlist.pullPolicyExechistories(url,choice, second_policy,True) 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 executionhist_policy1
gc.collect()
if not os.path.exists(f"parquet\\execution_history_{second_policy}.parquet"):
exe2 = utils.allowlist.pullPolicyExechistories(url, choice, second_policy, True)
data2 = json.loads(exe2) data2 = json.loads(exe2)
executionhist_policy2 = pd.DataFrame(data2["response"]["exechistories"]) executionhist_policy2 = pd.DataFrame(data2["response"]["exechistories"])
if not executionhist_policy2.empty: if not executionhist_policy2.empty:
executionhist_policy2 = executionhist_policy2.sort_values(by=['sha256','filename']) executionhist_policy2 = executionhist_policy2[['sha256', 'publisher', 'filename', 'hostname', 'username', 'pprocess', 'gprocess', 'commandline']]
executionhist_policy2 = executionhist_policy2.drop_duplicates(subset=['sha256', 'filename', 'hostname'])
executionhist_policy2.to_csv(f"dataframe_csv\\executionhist_{second_policy}.csv", index=False) executionhist_policy2 = executionhist_policy2.sort_values(by=['sha256', 'filename'])
ct.style_dataframe_dark(executionhist_policy2, f"dataframe_html\\executionhist_{second_policy}.html")
print(ct.colorText(f"Staging of Exection history for policy: {first_policy} is complete","green"))
#Combine the two policies execution histories 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
gc.collect()
if not os.path.exists(f"parquet\\combined_hashlist_{first_policy}_{second_policy}.parquet"):
combined_hashes = pd.DataFrame(columns=['sha256', 'publisher'])
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"):
# Categorize the hashes
categorized = utils.hashfunctions.categorizeHashes(
pd.read_parquet(f"parquet\\combined_hashlist_{first_policy}_{second_policy}.parquet"),
threat_tolerance_constant,
badpublisherlist
)
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 second_policy is first_policy: if not os.path.exists(f"parquet\\condensed_executions_{first_policy}_{second_policy}.parquet"):
execuctionhist_combined = executionhist_policy1 # Condense execution history
execuctionhist_combined.to_csv(f"dataframe_csv\\execuctionhist_combined_{first_policy}_{second_policy}.csv", index=False) try:
ct.style_dataframe_dark(df_aggregated_combo, f"dataframe_html\\execuctionhist_combined_{first_policy}_{second_policy}.html") exe1 = pd.read_parquet(f"parquet\\execution_history_{first_policy}.parquet")
print(ct.colorText(f"Dataframes have been combined","green")) utils.pathfunctions.inspect_parquet(f"parquet\\execution_history_{first_policy}.parquet")
if not exe1.empty:
condensed_exe1 = exe1.groupby('sha256').agg(lambda x: list(set(x))).reset_index()
else:
print("⚠️ First dataframe is empty.")
except Exception as e:
print(f"❌ Error reading first Parquet file: {e}")
elif os.path.exists(f"dataframe_csv\\executionhist_{first_policy}.csv") and os.path.exists(f"dataframe_csv\\executionhist_{second_policy}.csv"): try:
execuctionhist_combined = pd.concat([tryToReadCSV(f"dataframe_csv\\executionhist_{first_policy}.csv") , tryToReadCSV(f"dataframe_csv\\executionhist_{second_policy}.csv")], ignore_index=True).sort_values(by=['sha256','filename']) exe2 = pd.read_parquet(f"parquet\\execution_history_{second_policy}.parquet")
execuctionhist_combined.to_csv(f"dataframe_csv\\execuctionhist_combined_{first_policy}_{second_policy}.csv", index=False) utils.pathfunctions.inspect_parquet(f"parquet\\execution_history_{second_policy}.parquet")
ct.style_dataframe_dark(df_aggregated_combo, f"dataframe_html\\execuctionhist_combined_{first_policy}_{second_policy}.html") if not exe2.empty:
print(ct.colorText(f"Dataframes have been combined","green")) condensed_exe2 = exe2.groupby('sha256').agg(lambda x: list(set(x))).reset_index()
else:
print("⚠️ Second dataframe is empty.")
except Exception as e:
print(f"❌ Error reading second Parquet file: {e}")
#Keep only unique combinations of hash, filename, and hostname if not exe1.empty and not exe2.empty:
if f"dataframe_csv\\execuctionhist_combined_{first_policy}_{second_policy}.csv": condensed_combo = pd.concat([condensed_exe1, condensed_exe2], ignore_index=True)
unique_executions = tryToReadCSV(f"dataframe_csv\\execuctionhist_combined_{first_policy}_{second_policy}.csv").drop_duplicates(subset=['sha256', 'filename', 'hostname']) print(f"✅ Combined {len(condensed_combo)} hashes.")
unique_executions.to_csv(f"dataframe_csv\\unique_executions{first_policy}_{second_policy}.csv") elif exe1.empty:
ct.style_dataframe_dark(unique_executions, f"dataframe_html\\unique_execuctions.html") condensed_combo = condensed_exe2
elif exe2.empty:
condensed_combo = condensed_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()
#Add Hash info to the combined execution history if not os.path.exists(f"needs_approved\\hashes_rep_unknown_{first_policy}_{second_policy}.csv"):
if not os.path.exists(f"dataframe_html\\augmented_combo_{first_policy}_{second_policy}.html"):
print(ct.colorText(f"Preparing to pull hash info","green"))
augmented_combo= utils.hashfunctions.augmentAggregatedHashes(url,tryToReadCSV(f"dataframe_csv\\unique_executions{first_policy}_{second_policy}.csv"))
augmented_combo.to_csv(f"dataframe_csv\\augmented_combo_{first_policy}_{second_policy}.csv", index=False)
ct.style_dataframe_dark(augmented_combo, f"dataframe_html\\augmented_combo_{first_policy}_{second_policy}.html")
print(ct.colorText(f"Hash reputation info added to dataframe","green"))
#Categorize the hashes
if os.path.exists(f"dataframe_csv\\hashes_needing_approval_{first_policy}_{second_policy}.csv") and os.path.exists(f"dataframe_csv\\automatically_approved_hashes_{first_policy}_{second_policy}.csv") and os.path.exists(f"dataframe_csv\\unapproved_hashes__{first_policy}_{second_policy}.csv"):
break
else:
categorized = utils.hashfunctions.categorizeHashes(pd.read_csv(f"dataframe_csv\\augmented_combo_{first_policy}_{second_policy}.csv"), threat_tolerance_constant, badpublisherlist)
if not categorized[0].empty: condensed_combo = pd.read_parquet(f"parquet\\condensed_executions_{first_policy}_{second_policy}.parquet")
categorized[0].sort_values(by=['sha256','filename_x']) needsapproval= pd.read_parquet(f"parquet\\hashes_rep_unknown_{first_policy}_{second_policy}.parquet")
categorized[0].to_csv(f"dataframe_csv\\hashes_needing_approval_{first_policy}_{second_policy}.csv", index=False)
ct.style_dataframe_dark(categorized[0], f"dataframe_html\\hashes_needing_approval_{first_policy}_{second_policy}.html")
if not categorized[1].empty:
categorized[1].sort_values(by=['sha256','filename_x'])
categorized[1].to_csv(f"dataframe_csv\\automatically_approved_hashes_{first_policy}_{second_policy}.csv", index=False)
ct.style_dataframe_dark(categorized[1], f"dataframe_html\\automatically_approved_hashes_{first_policy}_{second_policy}.html")
if not categorized[2].empty:
categorized[2].sort_values(by=['sha256','filename_x'])
categorized[2].to_csv(f"dataframe_csv\\unapproved_hashes__{first_policy}_{second_policy}.csv", index=False) #Pull hash info for the entries in the needs approval table
ct.style_dataframe_dark(categorized[2], f"dataframe_html\\unapproved_hashes_{first_policy}_{second_policy}.html") needsapproval = pd.merge(condensed_combo, needsapproval, on='sha256', how='inner')
print(ct.colorText(f"Hashes have been categorized","green")) #Deduplicate lists in the columns
for col in needsapproval.columns:
if needsapproval[col].apply(lambda x: isinstance(x, list)).all():
needsapproval[col] = needsapproval[col].apply(deduplicate_list)
#Rename Publisher, Keep and reorder columns we want
needsapproval = needsapproval.rename(columns={'publisher_x': 'publisher'})
needsapproval = needsapproval[['sha256', 'publisher', 'description', 'filename', 'hostname', 'username', 'productname', 'productversion','reputation_lastseen', 'reputation_scannermatch', 'reputation_scannercount','reputation_status', 'reputation_threatlevel', 'reputation_threatname','reputation_timestamp', 'pprocess', 'gprocess', 'commandline']]
needsapproval.to_csv(f"needs_approved\\hashes_rep_unknown_{first_policy}_{second_policy}.csv",index=False)
needsapproval.to_parquet(f"parquet\\hashes_rep_unknown_{first_policy}_{second_policy}.parquet",index=False)
ct.style_dataframe_dark(needsapproval, f"needs_approved\\hashes_rep_unknown_{first_policy}_{second_policy}.html")
del needsapproval
del condensed_combo
gc.collect()
if not os.path.exists(f"needs_approved\\hashes_rep_good_{first_policy}_{second_policy}.csv"):
condensed_combo = pd.read_parquet(f"parquet\\condensed_executions_{first_policy}_{second_policy}.parquet")
needsapproval= pd.read_parquet(f"parquet\\hashes_rep_good_{first_policy}_{second_policy}.parquet")
#Pull hash info for the entries in the needs approval table
needsapproval = pd.merge(condensed_combo, needsapproval, on='sha256', how='inner')
#Deduplicate lists in the columns
for col in needsapproval.columns:
if needsapproval[col].apply(lambda x: isinstance(x, list)).all():
needsapproval[col] = needsapproval[col].apply(deduplicate_list)
#Rename Publisher, Keep and reorder columns we want
needsapproval = needsapproval.rename(columns={'publisher_x': 'publisher'})
needsapproval = needsapproval[['sha256', 'publisher', 'description', 'filename', 'hostname', 'username', 'productname', 'productversion','reputation_lastseen', 'reputation_scannermatch', 'reputation_scannercount','reputation_status', 'reputation_threatlevel', 'reputation_threatname','reputation_timestamp', 'pprocess', 'gprocess', 'commandline']]
needsapproval.to_csv(f"needs_approved\\hashes_rep_good_{first_policy}_{second_policy}.csv",index=False)
needsapproval.to_parquet(f"parquet\\hashes_rep_good_{first_policy}_{second_policy}.parquet", index=False)
ct.style_dataframe_dark(needsapproval, f"needs_approved\\hashes_rep_good_{first_policy}_{second_policy}.html")
del needsapproval
del condensed_combo
gc.collect()
if not os.path.exists(f"needs_approved\\hashes_rep_bad_{first_policy}_{second_policy}.html"):
condensed_combo = pd.read_parquet(f"parquet\\condensed_executions_{first_policy}_{second_policy}.parquet")
needsapproval= pd.read_parquet(f"parquet\\hashes_rep_bad_{first_policy}_{second_policy}.parquet")
#Pull hash info for the entries in the needs approval table
needsapproval = pd.merge(condensed_combo, needsapproval, on='sha256', how='inner')
#Deduplicate lists in the columns
for col in needsapproval.columns:
if needsapproval[col].apply(lambda x: isinstance(x, list)).all():
needsapproval[col] = needsapproval[col].apply(deduplicate_list)
#Rename Publisher, Keep and reorder columns we want
needsapproval = needsapproval.rename(columns={'publisher_x': 'publisher'})
needsapproval = needsapproval[['sha256', 'publisher', 'description', 'filename', 'hostname', 'username', 'productname', 'productversion','reputation_lastseen', 'reputation_scannermatch', 'reputation_scannercount','reputation_status', 'reputation_threatlevel', 'reputation_threatname','reputation_timestamp', 'pprocess', 'gprocess', 'commandline']]
needsapproval.to_parquet(f"parquet\\hashes_rep_bad_{first_policy}_{second_policy}.parquet", index=False)
ct.style_dataframe_dark(needsapproval, f"needs_approved\\hashes_rep_bad_{first_policy}_{second_policy}.html")
del needsapproval
del condensed_combo
gc.collect()
elif choice == "3": elif choice == "3":
if os.path.exists(f"manuallyapproved\\hashes_needing_approval_{first_policy}_{second_policy}.csv") and os.path.exists(f"manuallyapproved\\automatically_approved_hashes_{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"):
df1 = tryToReadCSV(f"manuallyapproved\\hashes_needing_approval_{first_policy}_{second_policy}.csv")
df2 = tryToReadCSV(f"manuallyapproved\\automatically_approved_hashes_{first_policy}_{second_policy}.csv") if not os.path.exists(f"parquet\\all_approved_hashes_{first_policy}_{second_policy}.parquet"):
all_approved_hashes = pd.concat([df1 , df2], ignore_index=True).sort_values(by=['sha256','filename_x']) df1 = tryToReadCSV(f"approved\\hashes_rep_unknown_{first_policy}_{second_policy}.csv")
print(ct.colorText(f"Approved hash lists have been combined","green")) df2 = tryToReadCSV(f"approved\\hashes_rep_good_{first_policy}_{second_policy}.csv")
all_approved_hashes.to_csv(f"dataframe_csv\\all_approved_hashes_{first_policy}_{second_policy}.csv", index=False)
ct.style_dataframe_dark(all_approved_hashes, f"dataframe_html\\all_approved_hashes_{first_policy}_{second_policy}.html")
print(ct.colorText(f"Beginning calculating longest common filepaths for path exceptions","green"))
grouped_df_view, df_with_groups_appended = utils.pathfunctions.export_groups_for_review(all_approved_hashes,"filename_x","longestcfp",min_files_for_path,path_exclusion_constant)
df_with_groups_appended.to_csv(f"dataframe_csv\\allinfo_{first_policy}_{second_policy}.csv", index=False)
ct.style_dataframe_dark(df_with_groups_appended, f"dataframe_html\\allinfo_{first_policy}_{second_policy}.html")
forbidden_lcfp = grouped_df_view["longestcfp"].str.contains(r"(?i)(?:\\Users|\\c\$\\Users|inetpub\\wwwroot|windows\\temp)", na=False)
grouped_df_view = grouped_df_view[~forbidden_lcfp]
print(ct.colorText(f"Removing forbidden filepaths for path exceptions","green"))
grouped_df_view.to_csv(f"dataframe_csv\\lcf_needs_approved_{first_policy}_{second_policy}.csv", index=False) all_approved_hashes = pd.concat([df1 , df2], ignore_index=True).sort_values(by=['sha256','filename'])
ct.style_dataframe_dark(grouped_df_view, f"dataframe_html\\lcf_needs_appoved_{first_policy}_{second_policy}.html") 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"))
grouped_df_view, df_with_groups_appended = utils.pathfunctions.export_groups_for_review(all_approved_hashes,"filename","longestcfp",min_files_for_path,path_exclusion_constant)
df_with_groups_appended.to_parquet(f"parquet\\approved_hashes_with_paths_{first_policy}_{second_policy}.parquet", index=False)
forbidden = utils.pathfunctions.regulator(badpathparts, True)
forbidden_lcfp = grouped_df_view["longestcfp"].str.contains(forbidden, na=False)
grouped_df_view = grouped_df_view[~forbidden_lcfp]
print(ct.colorText(f"Removing forbidden filepaths for path exceptions","green"))
grouped_df_view.to_parquet(f"parquet\\path_needs_approved_{first_policy}_{second_policy}.parquet", index=False)
grouped_df_view.to_csv(f"needs_approved\\path_needs_approved_{first_policy}_{second_policy}.csv", index=False)
ct.style_dataframe_dark(grouped_df_view, f"needs_approved\\path_needs_approved_{first_policy}_{second_policy}.html")
del grouped_df_view
del df_with_groups_appended
gc.collect()
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"))
@@ -392,17 +546,29 @@ def menu_prepare_to_enforce():
elif choice == "4": elif choice == "4":
if os.path.exists(f"dataframe_csv\\allinfo_{first_policy}_{second_policy}.csv") and os.path.exists(f"manuallyapproved\\lcf_needs_approved_{first_policy}_{second_policy}.csv"): if os.path.exists(f"parquet\\approved_hashes_with_paths_{first_policy}_{second_policy}.parquet") and os.path.exists(f"approved\\path_needs_approved_{first_policy}_{second_policy}.csv"):
df1 = tryToReadCSV(f"dataframe_csv\\allinfo_{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"):
allowbyhash = utils.pathfunctions.mask_from_csv(df1, f"manuallyapproved\\lcf_needs_approved_{first_policy}_{second_policy}.csv","longestcfp") df1 = pd.read_parquet(f"parquet\\approved_hashes_with_paths_{first_policy}_{second_policy}.parquet")
pathexclusions = tryToReadCSV(f"manuallyapproved\\lcf_needs_approved_{first_policy}_{second_policy}.csv") allowbyhash = utils.pathfunctions.mask_from_csv(df1, f"approved\\path_needs_approved_{first_policy}_{second_policy}.csv","longestcfp")
pathexclusions.to_csv(f"preflight\\final_path_exclusions_{first_policy}_{second_policy}.csv", index=False)
ct.style_dataframe_dark(pathexclusions, f"preflight\\final_path_exclusions_{first_policy}_{second_policy}.html") 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.to_parquet(f"parquet\\final_hash_approvals_{first_policy}_{second_policy}.parquet", index=False)
easyview = allowbyhash.groupby('sha256').agg(list).reset_index()
ct.style_dataframe_dark(easyview, 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 df1
del allowbyhash
del pathexclusions
del easyview
gc.collect()
allowbyhash.to_csv(f"preflight\\final_hash_approvals_{first_policy}_{second_policy}.csv", index=False)
easyview = allowbyhash.groupby('sha256').agg(list).reset_index()
ct.style_dataframe_dark(easyview, f"preflight\\final_hash_approvals_{first_policy}_{second_policy}.html")
elif choice == "5": elif choice == "5":
@@ -425,18 +591,12 @@ def menu_prepare_to_enforce():
allowlist_child_id = allowid[choice] allowlist_child_id = allowid[choice]
elif choice == "6": elif choice == "6":
if os.path.exists(f"preflight\\final_path_exclusions_{first_policy}_{second_policy}.csv") and os.path.exists(f"preflight\\final_hash_approvals_{first_policy}_{second_policy}.csv") 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 != " ":
pathexclusions = tryToReadCSV(f"preflight\\final_path_exclusions_{first_policy}_{second_policy}.csv") pathexclusions = pd.read_parquet(f"parquet\\final_path_exclusions_{first_policy}_{second_policy}.parquet")
allowbyhash = tryToReadCSV(f"preflight\\final_hash_approvals_{first_policy}_{second_policy}.csv") allowbyhash = pd.read_parquet(f"parquet\\final_hash_approvals_{first_policy}_{second_policy}.parquet")
print(ct.colorText(f"*******************************************************************************************************************************************","red")) ct.areYouSure()
print(ct.colorText(f"*=========================================================================================================================================*","yellow"))
print(ct.colorText(f"*=========================================================================================================================================*","red"))
print(ct.colorText(f"*-------------This program will now begin to make changes to the Airlock Console. Do you understand and agree to proceed? ----------------*", "yellow"))
print(ct.colorText(f"*=========================================================================================================================================*","red"))
print(ct.colorText(f"*=========================================================================================================================================*","yellow"))
print(ct.colorText(f"*******************************************************************************************************************************************","red"))
confirmation = input(ct.colorText("Type 'I AGREE' to continue: ","white")) confirmation = input(ct.colorText("Type 'I AGREE' to continue: ","white"))
if confirmation.strip().upper() == "I AGREE": if confirmation.strip().upper() == "I AGREE":
@@ -449,11 +609,13 @@ def menu_prepare_to_enforce():
allowlist_parenthashlist = allowbyhash[allowbyhash['reputation_status'] == 'KNOWN']['sha256'].unique().tolist() allowlist_parenthashlist = allowbyhash[allowbyhash['reputation_status'] == 'KNOWN']['sha256'].unique().tolist()
utils.policyfunctions.addHash(allowlist_parent_id,allowlist_parenthashlist) utils.policyfunctions.addHash(allowlist_parent_id,allowlist_parenthashlist)
print(ct.colorText(f"Adding hashes to {allowlist_child_name}", "yellow")) print(ct.colorText(f"Adding hashes to {allowlist_child_name}", "yellow"))
allowlist_childhashlist = allowbyhash[allowbyhash['reputation_status'] == 'UNKNOWN']['sha256'].unique().tolist() allowlist_childhashlist = allowbyhash[allowbyhash['reputation_status'] == 'UNKNOWN']['sha256'].unique().tolist()
utils.policyfunctions.addHash(allowlist_child_id, allowlist_childhashlist) utils.policyfunctions.addHash(allowlist_child_id, allowlist_childhashlist)
ct.locked()
exit()
else: else:
print(ct.colorText("Operation aborted. You MUST EXPLICITLY AGREE to proceed.", "red")) print(ct.colorText("Operation aborted. You MUST EXPLICITLY AGREE to proceed.", "red"))
parquet/all_approved_hashes_Developers_Dev Audit.parquet
BIN
View File
Binary file not shown.
parquet/approved_hashes_with_paths_Developers_Dev Audit.parquet
BIN
View File
Binary file not shown.
parquet/combined_hashlist_Developers_Dev Audit.parquet
BIN
View File
Binary file not shown.
parquet/condensed_executions_Developers_Dev Audit.parquet
BIN
View File
Binary file not shown.
parquet/execution_history_Dev Audit.parquet
BIN
View File
Binary file not shown.
parquet/execution_history_Developers.parquet
BIN
View File
Binary file not shown.
parquet/hashes_rep_bad_Developers_Dev Audit.parquet
BIN
View File
Binary file not shown.
parquet/hashes_rep_good_Developers_Dev Audit.parquet
BIN
View File
Binary file not shown.
parquet/hashes_rep_unknown_Developers_Dev Audit.parquet
BIN
View File
Binary file not shown.
parquet/path_needs_approved_Developers_Dev Audit.parquet
BIN
View File
Binary file not shown.
utils/hashfunctions.py
+33 -14
View File
@@ -41,23 +41,28 @@ def aggregateHashes(executions_json) -> pd.DataFrame:
return agg_df return agg_df
def augmentAggregatedHashes(url, agg_df: pd.DataFrame) -> pd.DataFrame: def augmentAggregatedHashes(url, agg_df: pd.DataFrame) -> pd.DataFrame:
""" """
Takes output of aggregatedHashes, queries API for those hashes, flattens response while keeping one row per hash, Takes output of aggregatedHashes, queries API for those hashes, flattens response while keeping one row per hash,
aggregate applications and baselines into lists, then merges results back into agg_df to create a aggregate applications and baselines into lists, then merges results back into agg_df to create a
""" """
if 'sha256' not in agg_df.columns or agg_df.empty:
print("⚠️ 'sha256' column missing or DataFrame is empty. Skipping API query.")
return agg_df.copy() # Return as-is to avoid breaking downstream logic
endpoint = url + '/v1/hash/query' endpoint = url + '/v1/hash/query'
payload = { payload = {
"hashes": agg_df['sha256'].tolist() "hashes": agg_df['sha256'].tolist()
} }
headers = {"X-APIKey": os.getenv('APIKEY')} headers = {"X-APIKey": os.getenv('APIKEY')}
payload = json.dumps(payload) payload = json.dumps(payload)
response = requests.post(endpoint, headers=headers, data=payload, verify=False) response = requests.post(endpoint, headers=headers, data=payload, verify=False)
data = response.json() data = response.json()
results = data.get("response", {}).get("results", []) results = data.get("response", {}).get("results", [])
rows = [] rows = []
for res in results: for res in results:
row = {"sha256": res.get("sha256"), "result": res.get("result")} row = {"sha256": res.get("sha256"), "result": res.get("result")}
@@ -80,10 +85,24 @@ def augmentAggregatedHashes(url, agg_df: pd.DataFrame) -> pd.DataFrame:
df_api = pd.DataFrame(rows) df_api = pd.DataFrame(rows)
if 'sha256' not in df_api.columns:
print("⚠️ API response missing 'sha256'. Skipping merge.")
return agg_df.copy()
df = agg_df.merge(df_api, on="sha256", how="left") df = agg_df.merge(df_api, on="sha256", how="left")
aug_df = df[['sha256', 'filename_x', 'description', 'productname', 'productversion', 'publisher_y', 'publisher_x', 'netdomain', 'hostname', 'username', 'reputation_lastseen', 'reputation_scannermatch', 'reputation_scannercount', 'reputation_status', 'reputation_threatlevel', 'reputation_threatname', 'reputation_timestamp', 'pprocess', 'gprocess', 'commandline', ]]
# Only include columns that exist to avoid KeyErrors
expected_columns = ['sha256', 'filename_x', 'description', 'productname', 'productversion',
'publisher_y', 'publisher_x', 'netdomain', 'hostname', 'username',
'reputation_lastseen', 'reputation_scannermatch', 'reputation_scannercount',
'reputation_status', 'reputation_threatlevel', 'reputation_threatname',
'reputation_timestamp', 'pprocess', 'gprocess', 'commandline']
available_columns = [col for col in expected_columns if col in df.columns]
aug_df = df[available_columns]
return aug_df return aug_df
def categorizeHashes(aug_df: pd.DataFrame, threat_tolerance: int, untrusted_publishers: list): def categorizeHashes(aug_df: pd.DataFrame, threat_tolerance: int, untrusted_publishers: list):
if untrusted_publishers is None: if untrusted_publishers is None:
@@ -94,29 +113,29 @@ def categorizeHashes(aug_df: pd.DataFrame, threat_tolerance: int, untrusted_publ
def reputationtool(row): def reputationtool(row):
val = row["reputation_scannermatch"] val = row["reputation_scannermatch"]
if pd.isna(val) or val == "N/A": if pd.isna(val) or val == "N/A":
return row["publisher_y"] == "Not Signed" return row["publisher"] == "Not Signed"
try: try:
return int(val) > threat_tolerance return int(val) > threat_tolerance
except (ValueError, TypeError): except (ValueError, TypeError):
return row["publisher_y"] == "Not Signed" return row["publisher"] == "Not Signed"
df["reputation_flag"] = df.apply(reputationtool, axis=1) df["reputation_flag"] = df.apply(reputationtool, axis=1)
mask_needsreview = ( mask_needsreview = (
((df["publisher_y"] == "Not Signed") & df["reputation_flag"]) | ((df["publisher"] == "Not Signed") & df["reputation_flag"]) |
(df["reputation_status"] == "UNKNOWN") (df["reputation_status"] == "UNKNOWN")
) )
mask_approved = ( mask_approved = (
( (
(df["publisher_y"] != "Not Signed") & (df["publisher"] != "Not Signed") &
~df["publisher_y"].isin(untrusted_publishers) & ~df["publisher"].isin(untrusted_publishers) &
~df["reputation_status"].isna() ~df["reputation_status"].isna()
) | ) |
( (
(df["publisher_y"] == "Not Signed") & (df["publisher"] == "Not Signed") &
~df["reputation_flag"] & ~df["reputation_flag"] &
~df["publisher_y"].isin(untrusted_publishers) & ~df["publisher"].isin(untrusted_publishers) &
~df["reputation_status"].isna() ~df["reputation_status"].isna()
) )
) )
utils/pathfunctions.py
+26
View File
@@ -17,6 +17,7 @@ import pandas as pd
import os import os
from itertools import chain from itertools import chain
import ast import ast
import re
@@ -134,3 +135,28 @@ def filter_and_drop(approved, eligiblepaths, min_hashes):
filtered = filtered[filtered['sha256'].apply(len) >= min_hashes] filtered = filtered[filtered['sha256'].apply(len) >= min_hashes]
return filtered return filtered
def inspect_parquet(path):
try:
df = pd.read_parquet(path)
print(f"✅ Successfully read: {path}")
print(f"📄 Columns: {df.columns.tolist()}")
print(f"🔢 Rows: {len(df)}")
return df
except Exception as e:
print(f"❌ Error reading {path}: {e}")
return pd.DataFrame()
def regulator(paths, case_insensitive=True):
"""
Build a Python raw string regex that matches any of the given Windows path fragments.
"""
escaped = [re.escape(p) for p in paths]
pattern = "(?:" + "|".join(escaped) + ")"
if case_insensitive:
pattern = pattern
print(f"Regulator is providing {pattern}")
return f'r"{pattern}"'
utils/pretty.py
+75
View File
@@ -136,3 +136,78 @@ def style_dataframe_dark(df, output_html_path=None, overwrite=True):
print(f"✅ Styled table saved to temporary file: {temp_path}") print(f"✅ Styled table saved to temporary file: {temp_path}")
else: else:
return styled_html return styled_html
def displayIntro():
print(colorText(r"""
███
████ ░████████
█████████████ ███████████████
█████████████████████ █████████████████████
███████████████████ ██████████████████████▓
███████████████████ ██████████████████████
█████████████████████ ███████████████████████
████████████████████████████████████████████████████████
█████████ ██ ██ █████████
█████████ ██ ███ █ █████████
█████████ ██ ████ █████ █████████████
█████████ ██ ██████ █████████████
████████ ██ ███████ ████████████░
███████ ██ ██▓ ██████ ████████████
██████ ██ ████ █████ ███████████
█████████████████████████████████████████████████
▒████████████████████ ██████████████████
███████████████████ ███████████████▒
███████████████ █████████████
██████████ ███████████
████████
████
""", "yellow"))
print(colorText(r"""
_____ .__ .__ __ ___________ .__
/ _ \ |__|______| | ____ ____ | | __ \__ ___/___ ____ | | ______
/ /_\ \| \_ __ \ | / _ \_/ ___\| |/ / | | / _ \ / _ \| | / ___/
/ | \ || | \/ |_( <_> ) \___| < | |( <_> | <_> ) |__\___ \
\____|__ /__||__| |____/\____/ \___ >__|_ \ |____| \____/ \____/|____/____ >
\/ \/ \/ \/
""", "cyan"))
print(colorText("=================================================================================", "cyan"))
print(colorText("======================== Welcome to the Airlock API Tool ========================", "cyan"))
print(colorText("=================================================================================", "cyan"))
def areYouSure():
print(colorText(f"*******************************************************************************************************************************************","red"))
print(colorText(f"*=========================================================================================================================================*","yellow"))
print(colorText(f"*=========================================================================================================================================*","red"))
print(colorText(f"*-------------This program will now begin to make changes to the Airlock Console. Do you understand and agree to proceed? ----------------*", "yellow"))
print(colorText(f"*=========================================================================================================================================*","red"))
print(colorText(f"*=========================================================================================================================================*","yellow"))
print(colorText(f"*******************************************************************************************************************************************","red"))
def locked():
print(colorText(r"""
████████████████████████████████████████████████████████████████
███ ██
██ ██████ ███
██ ████████████ ███
██ ████ ███ ███
██ ███ ███ ███
██ ███ ███ ███
██ ▒████████████████████ ███
██ ██████████████████████ ███
██ ██████████████████████ ███
██ ██████████████████████ ███
██ ██████████████████████ ███
██ ██████████████████████ ███
██ ███
███ ███
████████████████████████████████████████████████████████████████████
▒██████████████████████████████████████████████████████████████████▒
▒████
▒████
▓██████████████████████████████████████████
█████████████████████████████████████████████░
""", "yellow"))