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

Zar-Branch (#6)

Browse Source 
Co-authored-by: = <=>
Co-authored-by: Driven-Element <129630760+Driven-Element@users.noreply.github.com>
Reviewed-on: brotoskyj/AirlockTools#6
Co-authored-by: brotoskyj <jbrotosky@gmail.com>
Co-committed-by: brotoskyj <jbrotosky@gmail.com>
This commit was merged in pull request #6.
This commit is contained in:
brotoskyj
2025-08-22 09:27:16 -04:00
committed by brotoskyj
parent 4d25a21ddb
commit ccbf716418
13 changed files with 159 additions and 62606 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils/__pycache__/allowlist.cpython-313.pyc
BIN
View File
Binary file not shown.
utils/__pycache__/getdeviceevents.cpython-313.pyc
BIN
View File
Binary file not shown.
utils/__pycache__/hashfunctions.cpython-313.pyc
BIN
View File
Binary file not shown.
utils/__pycache__/pathfunctions.cpython-313.pyc
BIN
View File
Binary file not shown.
utils/allowfunctions.py
+16
View File
@@ -0,0 +1,16 @@
import pandas as pd
def filter_and_drop(approved, eligiblepaths, min_hashes):
"""
Filters eligiblepaths to rows where all hashes are in approved,
then drops rows with fewer than min_hashes hashes.
"""
approved_hashes = set(approved['sha256'])
def all_hashes_approved(row):
return all(h in approved_hashes for h in row['sha256'])
filtered = eligiblepaths[eligiblepaths.apply(all_hashes_approved, axis=1)]
filtered = filtered[filtered['sha256'].apply(len) >= min_hashes]
return filtered
utils/allowlist.py
+58 -24
View File
@@ -2,6 +2,7 @@ import datetime
import requests
import json
import os
import time
def allowlistexechistories(url, outputjson: bool):
endpoint = url + '/v1/group'
@@ -20,38 +21,71 @@ def allowlistexechistories(url, outputjson: bool):
policyids.append(list['groupid'])
choice = input("Select Policy Group: ")
choice = int(choice) - 1
endpoint = url + '/v1/logging/exechistories'
payload_dict = {
"type":[1, 2, 6, 7],
"checkpoint":"68a153c23963989b484541b4",
"policy": [policiesnames[choice]]
}
payload = json.dumps(payload_dict)
print(payload)
response = requests.request("POST", endpoint, headers=headers, data=payload, verify=False)
checkpoint = '000000000000000000000000'
json_output = {'error': 'Success', 'response': {'exechistories': []}}
while True:
json_response_data = checkpoint_stomper(checkpoint, url, policiesnames[choice], headers)
if not json_response_data['response']['exechistories']:
break
for index, item in enumerate(json_response_data['response']['exechistories']):
if index == len(json_response_data['response']['exechistories']) -1:
checkpoint = item['checkpoint']
print(f"Date Greater than 30 Days, Stepping to new Checkpoint. {item['checkpoint']}")
else:
if (datetime.date.today() - datetime.timedelta(days=10) > datetime.datetime.strptime(item['datetime'].replace(' +0000 UTC', ''), '%Y-%m-%dT%H:%M:%SZ').date()):
pass
else:
#json_output['response']['exechistories'].append(json_response_data['response']['exechistories'][1])
for output in json_response_data['response']['exechistories']:
json_output['response']['exechistories'].append(output)
json_output = json.dumps(json_output)
if outputjson == True:
return response
parse_text = json.loads(response.text)
for item in parse_text['response']['exechistories']:
print(item['checkpoint'])
print(item['datetime'])
print(item['hostname'])
print(item['filename'])
return json_output
#endpoint = url + '/v1/logging/exechistories'
#payload_dict = {
# "type":[1, 2, 6, 7],
# "checkpoint":"000000000000000000000000",
# "policy": [policiesnames[choice]]
#}
#payload = json.dumps(payload_dict)
#print(payload)
#response = requests.request("POST", endpoint, headers=headers, data=payload, verify=False)
#parse_text = json.loads(response.text)
#text_response = checkpoint_stomper(parse_text['response']['exechistories'], url, policiesnames[choice])
#
#if outputjson == False:
# return response
#
#parse_text = json.loads(response.text)
#
#for item in parse_text['response']['exechistories']:
# print(item['checkpoint'])
# print(item['datetime'])
# print(item['hostname'])
# print(item['filename'])
# checkpoint_stomper(item['checkpoint'], endpoint, headers, policiesnames[choice])
def checkpoint_stomper(checkpoint, endpoint, headers, policyname):
print(checkpoint)
def checkpoint_stomper(checkpoint, url, policy, headers):
endpoint = url + '/v1/logging/exechistories'
payload_dict = {
"type":[1,2,6,7],
"checkpoint": checkpoint,
"policy":[policyname]
"policy": [policy]
}
payload = json.dumps(payload_dict)
response = requests.request("POST", endpoint, headers=headers, data=payload, verify=False)
parse_text = json.loads(response.text)
# Send Whole JSON Response
for item in parse_text['response']['exechistories']:
print("test")
return parse_text
#for index, item in enumerate(parse_text):
# if index == len(parse_text) - 1:
# checkpoint = item['checkpoint']
# print(f"Time: {item['datetime']} Checkpoint: {item['checkpoint']}")
# response_fuzzer(checkpoint, url, policyname)
# else:
# if (datetime.date.today() - datetime.timedelta(days=30) > datetime.datetime.strptime(item['datetime'].replace( ' +0000 UTC', ''), '%Y-%m-%dT%H:%M:%SZ').date()):
# pass
# else:
# response_fuzzer(checkpoint, url, policyname)
print("Finished")
utils/hashfunctions.py
+24 -14
View File
@@ -8,12 +8,12 @@ def aggregateHashes(executions_json) -> pd.DataFrame:
"""
Takes the executions, aggregates all the data with sha256 as primary, then returns aggregated dataframe
"""
data = executions_json.json()
data = json.loads(executions_json)
df = pd.DataFrame(data["response"]["exechistories"])
if df.empty:
return df
print(df)
# Aggregate by sha256, deduplicate lists, and preserve order
agg_df = df.groupby("sha256").agg(lambda x: list(dict.fromkeys(x))).reset_index()
@@ -64,16 +64,17 @@ def augmentAggregatedHashes(url, agg_df: pd.DataFrame) -> pd.DataFrame:
df_api = pd.DataFrame(rows)
aug_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', 'pprocess', 'gprocess', 'commandline', 'reputation_lastseen', 'reputation_scannercount', 'reputation_scannermatch', 'reputation_status', 'reputation_threatlevel', 'reputation_threatname', 'reputation_timestamp']]
return aug_df
def categorizeHashes(aug_df: pd.DataFrame, threat_tolerance: int, untrusted_publishers: list):
if untrusted_publishers is None:
untrusted_publishers = []
if untrusted_publishers is None:
untrusted_publishers = []
df = aug_df.copy()
def reputationtool(row, threat_tolerance):
df = aug_df.copy()
def reputationtool(row, threat_tolerance):
if row["reputation_scannermatch"] == "N/A":
return True
try:
@@ -83,11 +84,20 @@ def categorizeHashes(aug_df: pd.DataFrame, threat_tolerance: int, untrusted_publ
pass
return False
mask_needsreview = (df["publisher_y"] == "Not Signed") & df.apply(lambda row: reputationtool(row, threat_tolerance), axis=1)
mask_approved = (df["publisher_y"] != "Not Signed") & (~df["publisher_y"].isin(untrusted_publishers))
mask_needsreview = (df["publisher_y"] == "Not Signed") & df.apply(lambda row: reputationtool(row, threat_tolerance), axis=1)
needsreview_df = df[mask_needsreview]
approved_df = df[mask_approved]
remaining_df = df[~(mask_needsreview | mask_approved)]
mask_approved = (
((df["publisher_y"] != "Not Signed") & (~df["publisher_y"].isin(untrusted_publishers))) &
(df["publisher_y"] != "Not Signed") # explicitly signed
) | (
(df["publisher_y"] == "Not Signed") &
(~df.apply(lambda row: reputationtool(row, threat_tolerance), axis=1)) &
(~df["publisher_y"].isin(untrusted_publishers)) # exclude untrusted even if unsigned
)
return needsreview_df, approved_df, remaining_df
needsreview_df = df[mask_needsreview]
approved_df = df[mask_approved]
remaining_df = df[~(mask_needsreview | mask_approved)]
return needsreview_df, approved_df, remaining_df
utils/pathfunctions.py
+27 -1
View File
@@ -41,18 +41,44 @@ def filepathInitialGroup(df: pd.DataFrame):
return join_parts(prefix)
# Step 6: Group directories by shared prefix using custom logic
"""
Loop through each directory path
directories: list of all directory paths.
groups: will hold lists of grouped directories.
used: tracks which directories have already been grouped.
"""
directories = df["directory"].tolist()
groups = []
used = set()
#For Each directory, compare it with others
"""
Skip if already grouped.
Start a new group with the current path.
parts_i is the list of folder names in the path (e.g., ["C:", "Users", "John", "Documents"]).
"""
for i, path in enumerate(directories):
if path in used:
continue
group = [path]
parts_i = get_parts(path)
#Compare with all other directories: For each other directory, split it into parts and find the common prefix (shared folder structure).
"""
Logic:
If the directory is deep (>3 parts) and shares at least 3 parts → group it.
If it's exactly 3 parts long and shares at least 2 → group it.
Or, if it shares all but one part and is deep → group it.
These rules are designed to:
Group directories that are closely related in structure.
Avoid grouping unrelated paths that just happen to start similarly.
"""
for j in range(i + 1, len(directories)):
parts_j = get_parts(directories[j])
common = os.path.commonprefix([parts_i, parts_j])
#Apply grouping rules
if (len(parts_i) > 3 and len(common) >= 3) or (len(parts_i) == 3 and len(common) >= 2):
group.append(directories[j])
used.add(directories[j])
@@ -81,7 +107,7 @@ def filepathInitialGroup(df: pd.DataFrame):
path_ineligible = grouped_df[grouped_df["depth"] <= 2].drop(columns=["depth"])
# Step 10: Move entries from eligible to ineligible if grouped_directory contains 'C:\Users' or 'c$\Users'
mask = path_eligible["grouped_directory"].str.contains(r"(?i)(?:\\Users|\\c\$\\Users)")
mask = path_eligible["grouped_directory"].str.contains(r"(?i)(?:\\Users|\\c\$\\Users|inetpub\\wwwroot|windows\\temp)", na=False)
move_to_ineligible = path_eligible[mask]
path_eligible = path_eligible[~mask]
path_ineligible = pd.concat([path_ineligible, move_to_ineligible], ignore_index=True)