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

Zar-Branch #17

Merged
mysticmomba merged 6 commits from Zar-Branch into master 2025-08-26 13:15:34 -04:00
Conversation 0 Commits 6 Files Changed 8
+184 -36
6 changed files with 184 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit ebfb0d80a9 - Show all commits
AirlockTools.py
+27 -20
View File
@@ -19,7 +19,7 @@ import utils.getdeviceevents
import utils.allowlist
import utils.hashfunctions
import utils.pathfunctions
import utils.colortext as ct
import utils.pretty as ct
import urllib3
import pandas as pd
import ast
@@ -186,7 +186,7 @@ def menu_prepare_to_enforce():
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' and choose this option to generate the proposed list of changes", "cyan"))
if os.path.isfile(f"manuallyapproved\\df_paths_needing_review_{first_policy}_{second_policy}.csv"):
if os.path.isfile(f"manuallyapproved\\df_paths_needing_review_{first_policy}_{second_policy}.csv") and os.path.isfile(f"dataframe_csv\\df_hashdestination_{first_policy}_{second_policy}.csv"):
print(ct.colorText(" [✓] This step has been completed","green"))
else:
print(ct.colorText(" [✗] This step has not been completed","red"))
@@ -215,43 +215,43 @@ def menu_prepare_to_enforce():
if not os.path.exists("dataframe_csv\\df_aggregated_{first_policy}.csv"):
executionhist_policy1 = utils.allowlist.pullPolicyExechistories(url,first_policy_tuple[0], first_policy_tuple[1],True)
df_aggregated_policy1 = utils.hashfunctions.aggregateHashes(executionhist_policy1)
df_aggregated_policy1.to_html(f"dataframe_html\\df_aggregated_{first_policy}.html", index=False)
df_aggregated_policy1.to_csv(f"dataframe_csv\\df_aggregated_{first_policy}.csv", index=False)
ct.style_dataframe_dark(df_aggregated_policy1, f"dataframe_html\\df_aggregated_{first_policy}.html")
print(ct.colorText(f"Staging of Exection history for policy: {first_policy} is complete","green"))
if not os.path.exists("dataframe_csv\\df_aggregated_{second_policy}.csv"):
executionhist_policy2 = utils.allowlist.pullPolicyExechistories(url,second_policy_tuple[0], second_policy_tuple[1],True)
df_aggregated_policy2 = utils.hashfunctions.aggregateHashes(executionhist_policy2)
df_aggregated_policy2.to_html(f"dataframe_html\\df_aggregated_{second_policy}.html", index=False)
df_aggregated_policy2.to_csv(f"dataframe_csv\\df_aggregated_{second_policy}.csv", index=False)
ct.style_dataframe_dark(df_aggregated_policy2, f"dataframe_html\\df_aggregated_{second_policy}.html")
print(ct.colorText(f"Staging of Exection history for policy: {second_policy} is complete","green"))
elif choice == "3":
if second_policy is first_policy and os.path.exists(f"dataframe_csv\\df_aggregated_{first_policy}.csv"):
df1 = tryToReadCSV(f"dataframe_csv\\df_aggregated_{first_policy}.csv")
df_aggregated_combo = df1
df_aggregated_combo.to_html(f"dataframe_html\\df_aggregated_combo_{first_policy}_{second_policy}.html", index=False)
df_aggregated_combo.to_csv(f"dataframe_csv\\df_aggregated_combo_{first_policy}_{second_policy}.csv", index=False)
ct.style_dataframe_dark(df_aggregated_combo, f"dataframe_html\\df_aggregated_combo_{first_policy}_{second_policy}.html")
print(ct.colorText(f"Dataframes have been aggregated (combined)","green"))
elif os.path.exists(f"dataframe_csv\\df_aggregated_{first_policy}.csv") and os.path.exists(f"dataframe_csv\\df_aggregated_{second_policy}.csv"):
df1 = tryToReadCSV(f"dataframe_csv\\df_aggregated_{first_policy}.csv")
df2 = tryToReadCSV(f"dataframe_csv\\df_aggregated_{second_policy}.csv")
df_aggregated_combo = pd.concat([df1 , df2], ignore_index=True)
df_aggregated_combo.to_html(f"dataframe_html\\df_aggregated_combo_{first_policy}_{second_policy}.html", index=False)
df_aggregated_combo.to_csv(f"dataframe_csv\\df_aggregated_combo_{first_policy}_{second_policy}.csv", index=False)
ct.style_dataframe_dark(df_aggregated_combo, f"dataframe_html\\df_aggregated_combo_{first_policy}_{second_policy}.html")
print(ct.colorText(f"Dataframes have been aggregated (combined)","green"))
else:
print(ct.colorText(f"Please stage your data before attempting this step","red"))
elif choice == "4":
if os.path.exists(f"dataframe_csv\\df_aggregated_combo_{first_policy}_{second_policy}.csv"):
df_augmented = utils.hashfunctions.augmentAggregatedHashes(url,tryToReadCSV(f"dataframe_csv\\df_aggregated_combo_{first_policy}_{second_policy}.csv"))
df_augmented.to_html(f"dataframe_html\\df_augmented_combo_{first_policy}_{second_policy}.html", index=False)
df_augmented.to_csv(f"dataframe_csv\\df_augmented_combo_{first_policy}_{second_policy}.csv", index=False)
ct.style_dataframe_dark(df_augmented, f"dataframe_html\\df_augmented_combo_{first_policy}_{second_policy}.html")
print(ct.colorText(f"Hash reputation info added to dataframe","green"))
else:
print(ct.colorText(f"Please combine your data with step 3 prior to attempting this step","red"))
@@ -259,16 +259,17 @@ def menu_prepare_to_enforce():
if os.path.exists(f"dataframe_csv\\df_augmented_combo_{first_policy}_{second_policy}.csv"):
categorized = utils.hashfunctions.categorizeHashes(pd.read_csv(f"dataframe_csv\\df_augmented_combo_{first_policy}_{second_policy}.csv"), threat_tolerance_constant, badpublisherlist)
categorized[0].to_html(f"dataframe_html\\df_hashes_needing_approval_{first_policy}_{second_policy}.html", index=False)
categorized[0].to_csv(f"dataframe_csv\\df_hashes_needing_approval_{first_policy}_{second_policy}.csv", index=False)
ct.style_dataframe_dark(categorized[0], f"dataframe_html\\df_hashes_needing_approval_{first_policy}_{second_policy}.html")
categorized[1].to_html(f"dataframe_html\\df_automatically_approved_hashes_{first_policy}_{second_policy}.html", index=False)
categorized[1].to_csv(f"dataframe_csv\\df_automatically_approved_hashes_{first_policy}_{second_policy}.csv", index=False)
ct.style_dataframe_dark(categorized[1], f"dataframe_html\\df_automatically_approved_hashes_{first_policy}_{second_policy}.html")
categorized[2].to_html(f"dataframe_html\\df_unapproved_hashes__{first_policy}_{second_policy}.html", index=False)
categorized[2].to_csv(f"dataframe_csv\\df_unapproved_hashes__{first_policy}_{second_policy}.csv", index=False)
ct.style_dataframe_dark(categorized[2], f"dataframe_html\\df_unapproved_hashes_{first_policy}_{second_policy}.html")
print(ct.colorText(f"Hashes have been categorized","green"))
else:
print(ct.colorText(f"Please Augment your data with hash threat info using step 4 prior to attempting this step","red"))
@@ -278,19 +279,19 @@ def menu_prepare_to_enforce():
df2 = tryToReadCSV(f"manuallyapproved\\df_automatically_approved_hashes_{first_policy}_{second_policy}.csv")
df_all_approved_hashes = pd.concat([df1 , df2], ignore_index=True)
df_all_approved_hashes.to_html(f"dataframe_html\\df_all_approved_hashes_{first_policy}_{second_policy}.html", index=False)
df_all_approved_hashes.to_csv(f"dataframe_csv\\df_all_approved_hashes_{first_policy}_{second_policy}.csv", index=False)
ct.style_dataframe_dark(df_all_approved_hashes, f"dataframe_html\\df_all_approved_hashes_{first_policy}_{second_policy}.html")
df_paths_needing_review, df_path_ineligible = utils.pathfunctions.filepathInitialGroup(pd.read_csv(f"dataframe_csv\\df_all_approved_hashes_{first_policy}_{second_policy}.csv"))
df_paths_needing_review.to_html(f"dataframe_html\\df_paths_needing_review_{first_policy}_{second_policy}.html", index=False)
df_paths_needing_review.to_csv(f"dataframe_csv\\df_paths_needing_review_{first_policy}_{second_policy}.csv", index=False)
ct.style_dataframe_dark(df_paths_needing_review, f"dataframe_html\\df_paths_needing_review_{first_policy}_{second_policy}.html")
df_path_ineligible.to_html(f"dataframe_html\\df_path_ineligible_{first_policy}_{second_policy}.html", index=False)
df_path_ineligible.to_csv(f"dataframe_csv\\df_path_ineligible_{first_policy}_{second_policy}.csv", index=False)
ct.style_dataframe_dark(df_path_ineligible, f"dataframe_html\\df_path_ineligible_{first_policy}_{second_policy}.html")
print(ct.colorText(f"Eligible paths determined","green"))
else:
print(ct.colorText(f"Please manually approve hashes prior to this step","red"))
@@ -299,8 +300,14 @@ def menu_prepare_to_enforce():
if os.path.exists(f"manuallyapproved\\df_paths_needing_review_{first_policy}_{second_policy}.csv"):
df1 = tryToReadCSV(f"manuallyapproved\\df_paths_needing_review_{first_policy}_{second_policy}.csv")
df2 = tryToReadCSV(f"dataframe_csv\\df_path_ineligible_{first_policy}_{second_policy}.csv")
print(df1['grouped_directory'])
print(df2['sha256'])
df3 = tryToReadCSV(f"manuallyapproved\\df_automatically_approved_hashes_{first_policy}_{second_policy}.csv")
df_hashdestination = utils.hashfunctions.destinationbuilder(df2,df3)
df_hashdestination.to_csv("dataframe_csv\\df_hashdestination_{first_policy}_{second_policy}.csv")
ct.style_dataframe_dark(df_hashdestination,f"dataframe_html\\df_hashdestination_{first_policy}_{second_policy}.html")
else:
print(ct.colorText(f"Please manually approve suggested paths prior to this step","red"))
utils/allowlist.py
+2 -1
View File
@@ -17,8 +17,9 @@ import requests
import json
import os
import time
import utils.colortext as ct
import utils.pretty as ct
import ijson
def pullPolicyExechistories(url, choice, policiesnames, outputjson: bool):
headers = {
"X-APIKey": os.getenv('APIKEY')
utils/colortext.py
-14
View File
@@ -1,14 +0,0 @@
def colorText(text: str, color: str) -> str:
colors = {
"red": "\033[91m",
"green": "\033[92m",
"yellow": "\033[93m",
"blue": "\033[94m",
"magenta": "\033[95m",
"cyan": "\033[96m",
"white": "\033[97m",
"reset": "\033[0m"
}
return f"{colors.get(color, colors['reset'])}{text}{colors['reset']}"
utils/getdeviceevents.py
+1 -1
View File
@@ -16,7 +16,7 @@ import datetime
import requests
import json
import os
import utils.colortext as ct
import utils.pretty as ct
def devicehistory(url, outputjson: bool):
endpoint = url + '/v1/getexechistory'
utils/hashfunctions.py
+16
View File
@@ -125,3 +125,19 @@ def categorizeHashes(aug_df: pd.DataFrame, threat_tolerance: int, untrusted_publ
unapproved_df = df[~(mask_needsreview | mask_approved)]
return needsreview_df, approved_df, unapproved_df
def destinationbuilder(df, df2):
# Step 1: Explode the 'sha256' list in df2 to create one row per sha256 value
df_expanded = df.explode('sha256')
# Step 2: Create a new dataframe for the result
df_hashdestination = df_expanded.copy()
# Step 3: Populate the 'Destination Allowlist' column based on comparison with df3
df_hashdestination['Destination Allowlist'] = df_hashdestination['sha256'].apply(
lambda x: 'Parent Policy Baseline' if x in df2['sha256'].values else "Destination Policy Allowlist"
)
# Step 4: Return the new dataframe
return df_hashdestination
utils/pretty.py
+138
View File
@@ -0,0 +1,138 @@
def colorText(text: str, color: str) -> str:
colors = {
"red": "\033[91m",
"green": "\033[92m",
"yellow": "\033[93m",
"blue": "\033[94m",
"magenta": "\033[95m",
"cyan": "\033[96m",
"white": "\033[97m",
"reset": "\033[0m"
}
return f"{colors.get(color, colors['reset'])}{text}{colors['reset']}"
def style_dataframe_dark(df, output_html_path=None, overwrite=True):
from datetime import datetime
# Get current date and filename for subtitle
today = datetime.now().strftime("%d %B %Y") # Changed to "Day Month Year"
filename = output_html_path.replace('.html', '') if output_html_path else "Report"
dark_css = """
<style>
body {
background-color: #000000;
margin: 0;
padding: 0;
font-family: 'Segoe UI', Tahoma, Geneva, Verdana, sans-serif;
color: #f8f8f2;
}
.header {
text-align: center;
margin: 20px auto;
padding: 10px;
border-bottom: 2px solid #ffd700;
max-width: 95%;
}
.header h1 {
color: #ffd700;
margin: 0;
font-size: 32px;
}
.header p {
color: #00bfff;
margin: 5px 0 0 0;
font-size: 18px;
}
.table-container {
overflow-y: scroll;
margin: 0 auto;
width: 95%;
max-height: calc(80vh - 100px);
display: block;
border: 1px solid #3a3a4d;
margin-bottom: 0;
}
table {
border-collapse: collapse;
font-size: 14px;
background-color: #1e1e2f;
color: #f8f8f2;
width: max-content;
}
th, td {
border: 1px solid #3a3a4d;
text-align: left;
padding: 10px;
max-width: 300px;
word-wrap: break-word;
overflow-wrap: break-word;
}
/* First column: no wrap */
td:nth-child(1), th:nth-child(1) {
white-space: nowrap;
max-width: none !important;
word-wrap: normal !important;
}
th {
background-color: #2e2e40;
color: #ffd700;
position: sticky;
top: 0;
z-index: 10;
}
tr:nth-child(even) {
background-color: #262638;
}
tr:hover {
background-color: #33334d;
color: #00bfff;
}
/* Custom scrollbar styling */
.table-container::-webkit-scrollbar {
width: 12px;
}
.table-container::-webkit-scrollbar-track {
background: #1e1e2f;
}
.table-container::-webkit-scrollbar-thumb {
background-color: #3a3a4d;
border-radius: 6px;
}
</style>
"""
header = f"""
<div class="header">
<h1>Airlock Tools</h1>
<p>{filename} - {today}</p>
</div>
"""
html_table = df.to_html(index=False, escape=False)
styled_html = (
f"<html>\n"
f"<head><title>Airlock Tools Report</title></head>\n"
f"<body>\n"
f"{dark_css}\n"
f"{header}\n"
f"<div class='table-container'>\n"
f" {html_table}\n"
f"</div>\n"
f"</body>\n"
f"</html>"
)
if output_html_path:
with open(output_html_path, "w", encoding="utf-8") as f:
f.write(styled_html)
print(f"✅ Styled table saved to '{output_html_path}'")
elif overwrite:
import tempfile
temp_path = tempfile.mktemp(suffix=".html")
with open(temp_path, "w", encoding="utf-8") as f:
f.write(styled_html)
print(f"✅ Styled table saved to temporary file: {temp_path}")
else:
return styled_html