Merge pull request 'Zar-Branch' (#17) from Zar-Branch into master

Reviewed-on: brotoskyj/AirlockTools#17

AirlockTools.py

@@ -19,10 +19,10 @@ 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
 
 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
 dotenv.load_dotenv()
@@ -118,10 +118,11 @@ def menu_prepare_to_enforce():
 	first_policy = " "
 	second_policy = " "
 	
+
 	#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("dataframe_csv"): os.makedirs("dataframe_csv")
-	if not os.path.exists("approvals"): os.makedirs("approvals")
+	if not os.path.exists("manuallyapproved"): os.makedirs("manuallyapproved")
 
 	df_aggregated_combo = pd.DataFrame()
 	while True:
@@ -164,23 +165,28 @@ def menu_prepare_to_enforce():
 		else:
 			print(ct.colorText("  [✗]  This step has not been completed","red"))
 
-		print(ct.colorText("5. Determine if path exclusions are possible", "cyan"))
+		print(ct.colorText("5. Categorize your hashes ", "cyan"))
-		
-		if os.path.exists(f"dataframe_csv\\df_path_eligible_{first_policy}_{second_policy}.csv") == True:
-			print(ct.colorText("  [✓]  This step has been completed","green"))
-		else:
-			print(ct.colorText("  [✗]  This step has not been completed", "red"))
-
-		print(ct.colorText("6. Categorize your hashes ", "cyan"))
 				   
 		if os.path.isfile(f"dataframe_csv\\df_hashes_needing_approval_{first_policy}_{second_policy}.csv") and os.path.isfile(f"dataframe_csv\\df_automatically_approved_hashes_{first_policy}_{second_policy}.csv") and os.path.isfile(f"dataframe_csv\\df_unapproved_hashes__{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"))
 
-		print(ct.colorText("7. Compare potential path exclusions with allowed hashes", "cyan"))
+		print(ct.colorText(f"6. Manually review the files \\dataframe_csv\\df_hashes_needing_approval_{first_policy}_{second_policy}.csv and dataframe_csv\\df_automatically_approved_hashes_{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("     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 to combine these approved hashes with the automatically approved hashes and generate a list of paths to be reviewed", "cyan"))
 
-		if os.path.exists(f"dataframe_csv\\df_allowed_paths_{first_policy}_{second_policy}.csv") == True:
+		if os.path.isfile(f"dataframe_csv\\df_paths_needing_review_{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"))
+
+		print(ct.colorText(f"7. Manually review the file  df_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("     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") 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"))
@@ -209,80 +215,111 @@ 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"))
 
 		elif choice == "5":
-			if  os.path.exists(f"dataframe_html\\df_augmented_combo_{first_policy}_{second_policy}.html"):
+			if os.path.exists(f"dataframe_csv\\df_augmented_combo_{first_policy}_{second_policy}.csv"):
-				path_eligible, path_ineligible = utils.pathfunctions.filepathInitialGroup(pd.read_csv(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)
-				path_eligible.to_html(f"dataframe_html\\df_path_eligible_{first_policy}_{second_policy}.html", index=False)
+
-				path_eligible.to_csv(f"dataframe_csv\\df_path_eligible_{first_policy}_{second_policy}.csv", index=False)
+				categorized[0].to_csv(f"dataframe_csv\\df_hashes_needing_approval_{first_policy}_{second_policy}.csv", index=False)
-				path_ineligible.to_html(f"dataframe_html\\df_path_ineligible_{first_policy}_{second_policy}.html", index=False)
+				ct.style_dataframe_dark(categorized[0], f"dataframe_html\\df_hashes_needing_approval_{first_policy}_{second_policy}.html")
-				path_ineligible.to_csv(f"dataframe_csv\\df_path_ineligible_{first_policy}_{second_policy}.csv", index=False)
+
-				print(ct.colorText(f"Eligible paths determined","green"))
+				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_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"))
 
 		elif choice == "6":
-			if os.path.exists(f"dataframe_csv\\df_augmented_combo_{first_policy}_{second_policy}.csv"):
+			if os.path.exists(f"manuallyapproved\\df_hashes_needing_approval_{first_policy}_{second_policy}.csv") and os.path.exists(f"manuallyapproved\\df_automatically_approved_hashes_{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)
+				df1 = tryToReadCSV(f"manuallyapproved\\df_hashes_needing_approval_{first_policy}_{second_policy}.csv")
-				categorized[0].to_html(f"dataframe_html\\df_hashes_needing_approval_{first_policy}_{second_policy}.html", index=False)
+				df2 = tryToReadCSV(f"manuallyapproved\\df_automatically_approved_hashes_{first_policy}_{second_policy}.csv")
-				categorized[0].to_csv(f"dataframe_csv\\df_hashes_needing_approval_{first_policy}_{second_policy}.csv", index=False)
+
-				categorized[1].to_html(f"dataframe_html\\df_automatically_approved_hashes_{first_policy}_{second_policy}.html", index=False)
+				df_all_approved_hashes = pd.concat([df1 , df2], ignore_index=True)
-				categorized[1].to_csv(f"dataframe_csv\\df_automatically_approved_hashes_{first_policy}_{second_policy}.csv", index=False)
+				df_all_approved_hashes.to_csv(f"dataframe_csv\\df_all_approved_hashes_{first_policy}_{second_policy}.csv", index=False)
-				categorized[2].to_html(f"dataframe_html\\df_unapproved_hashes__{first_policy}_{second_policy}.html", index=False)
+				ct.style_dataframe_dark(df_all_approved_hashes, f"dataframe_html\\df_all_approved_hashes_{first_policy}_{second_policy}.html")
-				categorized[2].to_csv(f"dataframe_csv\\df_unapproved_hashes__{first_policy}_{second_policy}.csv", index=False)
+				
-				print(ct.colorText(f"Hashes have been categorized","green"))
+				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_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_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 Augment your data with hash threat info using step 4 prior to attempting this step","red"))
+				print(ct.colorText(f"Please manually approve hashes prior to this step","red"))
+
 		
 		elif choice == "7":
-			if os.path.exists(f"dataframe_csv\\df_hashes_needing_approval_{first_policy}_{second_policy}.csv") and os.path.exists(f"dataframe_csv\\df_automatically_approved_hashes_{first_policy}_{second_policy}.csv") and os.path.exists(f"dataframe_csv\\df_unapproved_hashes__{first_policy}_{second_policy}.csv"):
+			if os.path.exists(f"manuallyapproved\\df_paths_needing_review_{first_policy}_{second_policy}.csv"):
-				allowpaths = utils.allowfunctions.filter_and_drop(pd.read_csv(f"dataframe_csv\\df_automatically_approved_hashes_{first_policy}_{second_policy}.csv"),tryToReadCSV(f"dataframe_csv\\df_path_eligible_{first_policy}_{second_policy}.csv"), path_exclusion_constant)
+				df1 = tryToReadCSV(f"manuallyapproved\\df_paths_needing_review_{first_policy}_{second_policy}.csv")
-				allowpaths.to_html(f"dataframe_html\\df_allowed_paths_{first_policy}_{second_policy}.html", index=False)
+				df2 = tryToReadCSV(f"dataframe_csv\\df_path_ineligible_{first_policy}_{second_policy}.csv")
-				allowpaths.to_csv(f"dataframe_csv\\df_allowed_paths_{first_policy}_{second_policy}.csv", index=False)
+				df3 = tryToReadCSV(f"manuallyapproved\\df_automatically_approved_hashes_{first_policy}_{second_policy}.csv")
-				print(ct.colorText(f"Allowable paths determined","green"))
+				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 complete step 6 prior to attempting this step","red"))
+				print(ct.colorText(f"Please manually approve suggested paths prior to this step","red"))
+			
+
+
+
+
 
 		elif choice == "Q":
 			break
-		
 		else:
 			print(ct.colorText("Invalid choice. Please try again.", "red"))    
 		
+		
+
 def tryToReadCSV(csv):
 	try:
 		df =pd.read_csv(csv)

hashtest.py

@@ -1,5 +0,0 @@
-hashes = ''
-while True:
-    inputhash = input("Hash: ")
-    hashes = hashes + ',' + inputhash
-    print(hashes)

utils/allowlist.py

@@ -17,9 +17,10 @@ import requests
 import json
 import os
 import time
+import utils.pretty as ct
+import ijson
 
 def pullPolicyExechistories(url, choice, policiesnames, outputjson: bool): 
-   
     headers = {
         "X-APIKey": os.getenv('APIKEY')
     }
@@ -29,21 +30,37 @@ def pullPolicyExechistories(url, choice, policiesnames, outputjson: bool):
         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']):
+        match_found = False
-            if index == len(json_response_data['response']['exechistories']) -1:
+        array_dividend = round(len(json_response_data['response']['exechistories'])/20)
-                checkpoint = item['checkpoint']
+        if array_dividend == 0:
-                print(ct.colorText(f"Date Greater than 30 Days, Stepping to new Checkpoint. {item['checkpoint']}", "blue"))
+            array_dividend == 1
-            else:
+        for index, item in enumerate(json_response_data['response']['exechistories'][::array_dividend]):
-                if (datetime.date.today() - datetime.timedelta(days=10) > datetime.datetime.strptime(item['datetime'].replace(' +0000 UTC', ''), '%Y-%m-%dT%H:%M:%SZ').date()):
+            if (datetime.date.today() - datetime.timedelta(days=30) <= datetime.datetime.strptime(item['datetime'].replace(' +0000 UTC', ''), '%Y-%m-%dT%H:%M:%SZ').date()):
-                    pass
+                print("Found Date Match")
+                match_found = True
+                break
+        checkpoints_processed = round(len(json_response_data['response']['exechistories'])/array_dividend)
+        print(ct.colorText(f"{checkpoints_processed} checkpoints from this execution have been processed. Stepping to the subsequent checkpoint. {item['checkpoint']}", "blue"))
+        checkpoint = item['checkpoint']
+        if match_found == True:
+            for index, item in enumerate(json_response_data['response']['exechistories']):
+                if index == len(json_response_data['response']['exechistories']) -1:
+                    checkpoint = item['checkpoint']
+                    print(ct.colorText(f"All checkpoints from this execution have been processed. Stepping to the subsequent checkpoint. {item['checkpoint']}", "blue"))
+                    break
                 else:
-                    for output in json_response_data['response']['exechistories']:
+                    if (datetime.date.today() - datetime.timedelta(days=30) > datetime.datetime.strptime(item['datetime'].replace(' +0000 UTC', ''), '%Y-%m-%dT%H:%M:%SZ').date()):
-                        json_output['response']['exechistories'].append(output)
+                        pass
+                    else:
+                        print(f"Added: {item['datetime']} | {item['checkpoint']} | {item['filename']} | {item['sha256']}")
+                        json_output['response']['exechistories'].append(item)
+            match_found = False
     json_output = json.dumps(json_output)
     if outputjson == True:
         return json_output
 
 def checkpoint_stomper(checkpoint, url, policy, headers):
+    json_output = {'error': 'Success', 'response': {'exechistories': []}}
     endpoint = url + '/v1/logging/exechistories'
     payload_dict = {
         "type":[1,2,6,7],
@@ -51,8 +68,13 @@ def checkpoint_stomper(checkpoint, url, policy, headers):
         "policy": [policy]
     }
     payload = json.dumps(payload_dict)
-    response = requests.request("POST", endpoint, headers=headers, data=payload, verify=False)
+    with requests.request("POST", endpoint, headers=headers, data=payload, verify=False, stream=True) as response:
-    parse_text = json.loads(response.text)
+        parser = ijson.items(response.raw, 'response.exechistories.item')
+        for item in parser:
+            key = (item.get('sha256'), item.get('hostname'))
+            if key not in json_output:
+                json_output['response']['exechistories'].append(item)   
+    parse_text = json.loads(json.dumps(json_output))
     return parse_text
 
 def listPolicies(url): 
@@ -72,71 +94,4 @@ def listPolicies(url):
         policyids.append(list['groupid'])
     choice = input(ct.colorText("Select Policy Group: ", "white"))
     choice = int(choice) - 1
-    checkpoint = '000000000000000000000000'
+    return choice, policiesnames
-    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 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, url, policy, headers):
-    endpoint = url + '/v1/logging/exechistories'
-    payload_dict = {
-        "type":[1,2,6,7],
-        "checkpoint": checkpoint,
-        "policy": [policy]
-    }
-    payload = json.dumps(payload_dict)
-    response = requests.request("POST", endpoint, headers=headers, data=payload, verify=False)
-    parse_text = json.loads(response.text)
-    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/colortext.py

@@ -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

@@ -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

@@ -90,28 +90,54 @@ def categorizeHashes(aug_df: pd.DataFrame, threat_tolerance: int, untrusted_publ
 
     df = aug_df.copy()
 
-    def reputationtool(row, threat_tolerance):
+    def reputationtool(row):
-        if row["reputation_scannermatch"] == "N/A":
+        val = row["reputation_scannermatch"]
-            return True
+        if pd.isna(val) or val == "N/A":
+            return row["publisher_y"] == "Not Signed"
         try:
             return int(val) > threat_tolerance
         except (ValueError, TypeError):
             return row["publisher_y"] == "Not Signed"
 
-    mask_needsreview = (df["publisher_y"] == "Not Signed") & df.apply(lambda row: reputationtool(row, threat_tolerance), axis=1)
+    df["reputation_flag"] = df.apply(reputationtool, axis=1)
 
-    mask_approved = (
+    mask_needsreview = (
-    ((df["publisher_y"] != "Not Signed") & (~df["publisher_y"].isin(untrusted_publishers))) &
+        ((df["publisher_y"] == "Not Signed") & df["reputation_flag"]) |
-    (df["publisher_y"] != "Not Signed")  # explicitly signed
+        (df["reputation_status"] == "UNKNOWN")
-    ) | (
-    (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
     )
 
+    mask_approved = (
+        (
+            (df["publisher_y"] != "Not Signed") &
+            ~df["publisher_y"].isin(untrusted_publishers) &
+            ~df["reputation_status"].isna()
+        ) |
+        (
+            (df["publisher_y"] == "Not Signed") &
+            ~df["reputation_flag"] &
+            ~df["publisher_y"].isin(untrusted_publishers) &
+            ~df["reputation_status"].isna()
+        )
+    )
 
     needsreview_df = df[mask_needsreview]
     approved_df = df[mask_approved]
-    remaining_df = df[~(mask_needsreview | mask_approved)]
+    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
 
-    return needsreview_df, approved_df, remaining_df

utils/pathfunctions.py

@@ -57,45 +57,21 @@ def filepathInitialGroup(df: pd.DataFrame):
                 break
         return join_parts(prefix)
 
-    # Step 6: Group directories by shared prefix using custom logic
+    # Step 6: Group directories by shared prefix
-    """
-    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])
@@ -123,7 +99,7 @@ def filepathInitialGroup(df: pd.DataFrame):
     path_eligible = grouped_df[grouped_df["depth"] > 2].drop(columns=["depth"])
     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'
+    # Step 10: Move entries from eligible to ineligible if grouped_directory contains excluded directories
     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]

utils/pretty.py

@@ -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