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

Cleanup

Browse Source
This commit is contained in:
=
2025-09-02 08:29:42 -04:00
parent 7eeb072117
commit 6b8c849db8
4 changed files with 1 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils/hashfunctions.py
+1 -64
View File
@@ -12,14 +12,13 @@
#
# 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/>.
import pandas as pd
import requests
import os
import json
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
@@ -41,7 +40,6 @@ def aggregateHashes(executions_json) -> pd.DataFrame:
return agg_df
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,
@@ -103,7 +101,6 @@ 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 = []
@@ -146,9 +143,6 @@ def categorizeHashes(aug_df: pd.DataFrame, threat_tolerance: int, untrusted_publ
return needsreview_df, approved_df, unapproved_df
import pandas as pd
def explode_and_deduplicate(df):
df['sha256'] = df['sha256'].str.split(',')
df = df.explode('sha256')
@@ -205,60 +199,3 @@ def destinationHashes(
# Concatenate results
df_hashdestination = pd.concat([df_paths, df_hashes], ignore_index=True)
return df_hashdestination
"""
def explode_and_deduplicate(df):
df['sha256'] = df['sha256'].str.split(',')
df = df.explode('sha256')
return df.drop_duplicates().reset_index(drop=True)
def clean_sha256(df, column='sha256'):
df[column] = df[column].astype(str).str.strip("'[]\" ")
return df
def destinationHashes(
df_approved_paths: pd.DataFrame,
df_approved_hashes: pd.DataFrame,
df_hashes_auto_approved: pd.DataFrame,
df_hashes_manually_approved: pd.DataFrame,
):
# Deduplicate and explode all input DataFrames
df_approved_paths = explode_and_deduplicate(df_approved_paths)
df_approved_hashes = explode_and_deduplicate(df_approved_hashes)
df_hashes_auto_approved = explode_and_deduplicate(df_hashes_auto_approved)
df_hashes_manually_approved = explode_and_deduplicate(df_hashes_manually_approved)
# Clean sha256 values in all relevant DataFrames
df_approved_hashes = clean_sha256(df_approved_hashes)
df_hashes_auto_approved = clean_sha256(df_hashes_auto_approved)
df_hashes_manually_approved = clean_sha256(df_hashes_manually_approved)
# Create sets for faster lookup
auto_approved_sha256 = set(df_hashes_auto_approved['sha256'].values)
manually_approved_sha256 = set(df_hashes_manually_approved['sha256'].values)
# Debug: Print unmatched hashes
unmatched = set(df_approved_hashes['sha256']) - (auto_approved_sha256 | manually_approved_sha256)
print(f"Unmatched hashes: {unmatched}")
# Process df_approved_paths
df_paths = df_approved_paths.assign(
destination='Path Exclusion'
)[['sha256', 'description', 'destination', 'grouped_directory']]
# Process df_approved_hashes
df_hashes = df_approved_hashes.copy()
df_hashes['destination'] = df_hashes['sha256'].apply(
lambda x: 'Parent Policy Baseline' if x in auto_approved_sha256
else ('Child Policy Allowlist' if x in manually_approved_sha256 else None)
)
df_hashes = df_hashes.dropna(subset=['destination'])
df_hashes = df_hashes.assign(grouped_directory=None)[['sha256', 'description', 'destination', 'grouped_directory']]
# Concatenate results
df_hashdestination = pd.concat([df_paths, df_hashes], ignore_index=True)
return df_hashdestination
"""