AirlockTools.py

@@ -14,25 +14,27 @@
 #    along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
 import dotenv
+import json
 import os
-import utils.getdeviceevents
+import pandas as pd
+import urllib3
 import utils.allowlist
+import utils.getdeviceevents
 import utils.hashfunctions
 import utils.pathfunctions
 import utils.policyfunctions
 import utils.pretty as ct
-import urllib3
-import pandas as pd
-import json
-import ast
 
 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
+
 dotenv.load_dotenv()
 
+#Constants
+url = os.getenv('url')
 badpublisherlist = ["Brave Software, Inc.",	"Zoom Video Communications, Inc."]
 path_exclusion_constant = 3
+min_files_for_path = 4
 threat_tolerance_constant = 4
-url = os.getenv('url')
 
 
 def apivalidation(): 
@@ -53,8 +55,6 @@ def apivalidation():
 		case _:
 			menu_main()
 
-
-
 def tryToReadCSV(csv):
 	try:
 		df =pd.read_csv(csv)
@@ -148,6 +148,8 @@ def menu_prepare_to_enforce():
 		print(ct.colorText(" -------------------- Prepare to Enforce Policy ---------------------", "cyan"))
 		print(ct.colorText(" --------------------------------------------------------------------", "cyan"))
 		print(ct.colorText("\nSequentually follow these steps to prepare a policy for enforcement:", "white"))
+		
+		
 		print(ct.colorText("\n1. Choose which originating policy or policies to move to enforcement", "cyan"))
 		
 		if first_policy == " " and second_policy == " ": 
@@ -158,6 +160,7 @@ def menu_prepare_to_enforce():
 			print(ct.colorText(f"  [✓]  {first_policy} has been selected as Policy 1","green"))
 			print(ct.colorText(f"  [✓]  {second_policy} has been selected as Policy 2","green"))
 
+	
 		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"): 
@@ -184,6 +187,7 @@ def menu_prepare_to_enforce():
 		else:
 			print(ct.colorText(f"  [✗]  Hashes have not been cateogrized", "red"))
 
+		
 		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("     Remove the rows containing hashes you do not approve of, and those you would not approve of without metarules.", "cyan"))
@@ -211,6 +215,7 @@ def menu_prepare_to_enforce():
 		else:
 			print(ct.colorText("  [✗]  Path review list has not been created","red"))	
 
+	
 		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("     When complete, save the csv file to the directory 'manuallyapproved'", "cyan"))
@@ -231,6 +236,7 @@ def menu_prepare_to_enforce():
 		else:
 			print(ct.colorText("  [✗]  Preflight hash approval list has not been generated","red"))	
 
+
 		print(ct.colorText(f"5. Choose the destination_name policy and parent and child allow list", "cyan"))	
 		if allowlist_child_name == " " and allowlist_parent_name== " ": 
 			print(ct.colorText(f"  [✗]  No allowlists have been chosen","red"))
@@ -249,11 +255,12 @@ def menu_prepare_to_enforce():
 		print(ct.colorText(f"     Apply signed or attested hashes to Parent Allow List", "cyan"))
 		print(ct.colorText(f"     Apply approved, but unsigned hashes to the Child Allow List", "cyan"))
 
-
 		print(ct.colorText("Q. Quit", "cyan"))
 
 		choice = input(ct.colorText("\nEnter your choice: ", "white"))
+		
 		if choice == "1":
+			
 			choice, policynames, policyid = utils.allowlist.listPolicies(url)
 			first_policy = policynames[choice]
 			while True:
@@ -269,6 +276,7 @@ def menu_prepare_to_enforce():
 					print(ct.colorText("Please answer with 'yes' or 'no'.", "red"))
 
 		elif choice == "2":
+			
 			if  not os.path.exists(f"dataframe_csv\\executionhist_{first_policy}.csv"):
 				exe1 = utils.allowlist.pullPolicyExechistories(url,choice, policynames,True)
 				data = json.loads(exe1)
@@ -333,6 +341,7 @@ def menu_prepare_to_enforce():
 
 			
 		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"):
 				df1 = tryToReadCSV(f"manuallyapproved\\hashes_needing_approval_{first_policy}_{second_policy}.csv")
 				df2 = tryToReadCSV(f"manuallyapproved\\automatically_approved_hashes_{first_policy}_{second_policy}.csv")
@@ -343,7 +352,7 @@ def menu_prepare_to_enforce():
 				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)
+				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")
@@ -384,7 +393,6 @@ def menu_prepare_to_enforce():
 			#print(allowlist_parent_tuple)
 			allowlist_parent_name = allowlists[choice]
 			allowlist_parent_id = allowid[choice]
-			
 
 			print(ct.colorText(f"Please choose Child Allowlist for Less-Known Hashes","white"))
 			choice, allowlists, allowid = utils.allowlist.listAllowlists(url)
@@ -431,9 +439,6 @@ def menu_prepare_to_enforce():
 		else:
 			print(ct.colorText("Invalid choice. Please try again.", "red"))    
 		
-		
-
-
 
 if __name__ == "__main__":
 	apivalidation()

utils/pathfunctions.py

@@ -72,13 +72,19 @@ def add_longest_common_two_local(df, col="filename_x", new_col="longestcfp", min
     return df
 
 
-def export_groups_for_review(df, col="filename_x", group_col="longestcfp"):
+def export_groups_for_review(df, col, group_col, min_number_in_group, path_length_constant):
     """
     Compute longest common paths, group filepaths, write CSV for review.
     """
     df = add_longest_common_two_local(df, col=col, new_col=group_col)
     grouped = df.groupby(group_col)[col].apply(list).reset_index()
-    return grouped, df
+    grouped = grouped.sort_values(by=col)
+    print("Before filtering:", len(grouped))
+    grouped = grouped[grouped[col].apply(lambda x: len(x) >= min_number_in_group)]
+    filtered = grouped[grouped[group_col].apply(lambda x: len(os.path.normpath(x).split(os.sep)) >= path_length_constant)]
+    print("After filtering:", len(grouped))
+
+    return filtered, df