flows/prepPolicy.py

@@ -132,6 +132,7 @@ def buildPathsandPublishers(selected_policies: List[Policy], split):
     all_approved_hashes = pd.DataFrame()
     path1 = f"{working_dir}\\Approved\\{selected_policies[0].name}_approved_executions.csv"
     path2 = f"{working_dir}\\Approved\\{selected_policies[0].name}_needs_review_executions.csv"
+    path_exclusion_constant = get_protected_value("PATH_EXCLUSION_CONST", cast_type= int)
 
     if os.path.exists(path1):
         df1 = pd.read_csv(path1)
@@ -154,29 +155,31 @@ def buildPathsandPublishers(selected_policies: List[Policy], split):
         else:
             logger.warning("Warning: 'filename' column not found in concatenated DataFrame.")
 
-    if not all_approved_hashes.empty:
+    if not all_approved_hashes.empty and path_exclusion_constant:
+
         primary_path_exclusions = calculatePath(
-            all_approved_hashes,
+            all_approved_hashes, path_exclusion_constant,
             split,
         )
         remaining_hashes = all_approved_hashes[
             ~all_approved_hashes["sha256"].isin(primary_path_exclusions["sha256"])
         ]
         secondary_path_exclusions = calculatePath(
-            remaining_hashes, split
+            remaining_hashes,(path_exclusion_constant - 1), split
         )
         remaining_hashes = remaining_hashes[
             ~remaining_hashes["sha256"].isin(secondary_path_exclusions["sha256"])
         ]
         dataframes = {
+            "all_approved_hashes" : all_approved_hashes,
             "primary_Paths": primary_path_exclusions,
             "secondary_Paths": secondary_path_exclusions,
-            "hashes_to_add": remaining_hashes,
+            "hashes_not_approvable_by_path": remaining_hashes
         }
         logger.debug("Preparing to sort dataframes")
         for name, df in dataframes.items():
             logger.debug(f" DataFrame headers: {list(df.columns)}")
-            if name == "hashes_to_add":   df.sort_values(by="filename", inplace=True)
+            if "hashes" in name :   df.sort_values(by="filename", inplace=True)
             else: df.sort_values(by="longestcfp", inplace=True)
           
             df.to_csv(f"{working_dir}\\Needs_Review\\Review_Second\\{selected_policies[0].name}_{name}.csv", index=False)
@@ -193,6 +196,8 @@ def buildPathsandPublishers(selected_policies: List[Policy], split):
             publist = publist[["publisher"]]
             publist.sort_values(by="publisher", inplace=True)
             publist.to_csv(f"{working_dir}\\Needs_Review\\Review_Second\\{selected_policies[0].name}_publishers.csv", index=False)
+    else: 
+        logger.debug("Approved Hashes list appears empty")
 
 def buildPreflights(selected_policies: List[Policy]):
     working_dir = load_env("WORKING_DIR")
@@ -202,17 +207,13 @@ def buildPreflights(selected_policies: List[Policy]):
     approved_hashes = pd.DataFrame()
     approved_publishers = pd.DataFrame()
 
-    hash = f"{working_dir}\\Approved\\{selected_policies[0].name}_hashes_to_add.csv"
+    hash = f"{working_dir}\\Needs_Review\\Review_Second\\{selected_policies[0].name}_all_approved_hashes.csv"
     path1 = f"{working_dir}\\Approved\\{selected_policies[0].name}_primary_Paths.csv"
     path2 = f"{working_dir}\\Approved\\{selected_policies[0].name}_secondary_Paths.csv"
     publishers = f"{working_dir}\\Approved\\{selected_policies[0].name}_publishers.csv"
 
-    if os.path.exists(hash):
-        approved_hashes = pd.read_csv(hash)
-
-    else:
-        logger.warning(f"File not found: {hash}")
 
+    #Read in and combine the two path generations
     if os.path.exists(path1):
         df1 = pd.read_csv(path1)
     else:
@@ -229,6 +230,19 @@ def buildPreflights(selected_policies: List[Policy]):
     else:
         approved_paths = pd.concat([df1, df2], ignore_index=True)
     
+    approved_paths = approved_paths.drop_duplicates(subset="longestcfp", keep ="first")
+
+    #We create a list of hashes that are left over if we exclude the ones that are covered by the path exclusions.
+    if os.path.exists(hash):
+        hashes = pd.read_csv(hash)
+        approved_hashes = hashes[~hashes['filename'].isin(approved_paths['longestcfp'])]
+
+        approved_hashes = approved_hashes.drop_duplicates(subset="sha256", keep ="first")
+
+    else:
+            logger.warning(f"File not found: {hash}")
+
+
     if os.path.exists(publishers):
         approved_publishers = pd.read_csv(publishers)
 
@@ -246,8 +260,7 @@ def buildPreflights(selected_policies: List[Policy]):
         df.to_csv(f"{working_dir}\\Preflight\\{selected_policies[0].name}_{name}.csv", index=False)
         formatHTML(df, f"{working_dir}\\Preflight\\HTML\\{selected_policies[0].name}_{name}.html")
 
-def splitFilepathsGrouped(df, col="filename"):
+def splitFilepathsGrouped(df, path_exclusion_constant, col="filename"):
-    path_exclusion_constant = get_protected_value("PATH_EXCLUSION_CONST", cast_type= int)
     min_files_for_path = get_protected_value("MIN_FILES_FOR_PATH", cast_type= int)
 
     def clean_split(path):
@@ -307,7 +320,7 @@ def splitFilepathsGrouped(df, col="filename"):
 
     return pd.DataFrame(new_rows).drop(columns=["group_key"])
 
-def calculatePath(approved_hashes, split):
+def calculatePath(approved_hashes, path_exclusion_constant, split):
     if split:
         dfs_by_policy = [group for _, group in approved_hashes.groupby("policy")]
     else:
@@ -319,7 +332,7 @@ def calculatePath(approved_hashes, split):
     processed_dfs = []
 
     for df in dfs_by_policy:
-        haslcp = splitFilepathsGrouped(df, "filename")
+        haslcp = splitFilepathsGrouped(df, path_exclusion_constant, "filename")
         haslcp = haslcp.drop_duplicates()
 
         forbidden = regulator(badpathparts, True)

utils/menus.py

@@ -41,6 +41,7 @@ from utils.utils import (
     get_sanitized_input,
     open_directory,
     printEnforceChecklist,
+    locked
 )
 
 logger = logging.getLogger(__name__)
@@ -99,7 +100,7 @@ def menu_main(api: AirlockAPIWrapper):
         else:
             print(colorText("Invalid choice. Please try again.", "red"))
 
-def menu_policy_enforce(api: AirlockAPIWrapper):
+def menu_policy_enforce(api: AirlockAPIWrapper): #TODO Need to clean up 6 and 7 into functions
     selected_policies = []
     destination_policy = []
     destination_allowlist = []
@@ -139,7 +140,7 @@ def menu_policy_enforce(api: AirlockAPIWrapper):
                 print("File not found. Please make sure it's saved correctly and try again.")
 
         elif choice == "5":
-            if os.path.exists(f"{working_dir}\\Approved\\{selected_policies[0].name}_hashes_to_add.csv") and os.path.exists(
+            if os.path.exists(f"{working_dir}\\Approved\\{selected_policies[0].name}_approved_executions.csv") and os.path.exists(
                 f"{working_dir}\\Approved\\{selected_policies[0].name}_primary_Paths.csv"
             ):
                 buildPreflights(selected_policies)
@@ -167,11 +168,13 @@ def menu_policy_enforce(api: AirlockAPIWrapper):
                     for path, ext in unique_combinations.itertuples(index=False, name=None)
                 ]
 
-                print(processed_paths)
+                for path in processed_paths:
+                    print(path)
+
                 print(colorText("These publishers would added", "yellow"))
 
                 if os.path.exists(f"{working_dir}\\Preflight\\{selected_policies[0].name}_approved_publishers.csv"):
-                    publishers = pd.read_csv(f"{working_dir}\\Preflight\\{selected_policies[0].name}_approved_hashes.csv")
+                    publishers = pd.read_csv(f"{working_dir}\\Preflight\\{selected_policies[0].name}_approved_publishers.csv")
                     if publishers.empty:
                         print(colorText("The publishers list is empty.", "red"))
                     else:
@@ -181,13 +184,14 @@ def menu_policy_enforce(api: AirlockAPIWrapper):
                             .drop_duplicates()
                             .tolist()
                         )
-                        print(processed_publishers)
+                        for publisher in processed_publishers:
+                            print(publisher)
 
                 print(colorText("These hashes would be added to:", "yellow"))
                 print(destination_allowlist)
 
                 processed_hashes = hashes["sha256"].unique().tolist()
-                print(processed_hashes)
+                print_three_wide(processed_hashes)
 
                 if processed_paths and processed_hashes:
                     tested = True
@@ -207,7 +211,6 @@ def menu_policy_enforce(api: AirlockAPIWrapper):
                 for item in missing_items:
                     logger.error(f" - {item}")
 
-
         elif choice == "7":
             areYouSure()
             confirmation = get_sanitized_input("Type 'I AGREE' to continue: ")
@@ -222,6 +225,9 @@ def menu_policy_enforce(api: AirlockAPIWrapper):
                 api.policy_add_path_exclusions(destination_policy[0].groupid, processed_paths)
                 if processed_publishers:
                     api.policy_add_publishers(destination_policy[0].groupid, processed_publishers)
+
+                locked()
+                
             else:
                 logger.error("Confirmation block failed. Reasons:")
                 if not tested:
@@ -237,7 +243,7 @@ def menu_policy_enforce(api: AirlockAPIWrapper):
             open_directory(working_dir)
         elif choice.upper() == "S":
             menu_settings()
-        elif choice.upper == "B":
+        elif choice.upper() == "B":
             break
 
        
@@ -314,3 +320,8 @@ def menu_settings():
             break
         else:
             print("Invalid choice. Please try again.")
+
+def print_three_wide(items):
+    for i in range(0, len(items), 3):
+        row = items[i:i+3]
+        print(" | ".join(row))

utils/setup.py

@@ -15,6 +15,7 @@
 
 import json
 import logging
+import logging.config
 import logging.handlers
 import os
 import platform
@@ -36,49 +37,69 @@ def get_base_directory() -> Path:
     else:
         return home / '.local' / 'share' / "AirlockTools"
 
+
 def configure_logging(log_dir: Path, log_level: str = "DEBUG"):
     log_file = log_dir / "airlocktools.log"
-    logger = logging.getLogger()
 
-    # Always allow all messages to propagate to handlers
+    config = {
-    logger.setLevel(logging.DEBUG)
+        "version": 1,  # Required key for dictConfig format version
+        "disable_existing_loggers": False,  # Keeps existing loggers active
+        "formatters": {
+            "detailed": {
+                "format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
+                # Includes timestamp, logger name, level, and message
+            },
+            "simple": {
+                "format": "%(levelname)s - %(message)s"
+                # Minimal format for console output
+            },
+        },
+        "handlers": {
+            "file": {
+                "class": "logging.handlers.TimedRotatingFileHandler",
+                "filename": str(log_file),
+                "when": "midnight",       # Rotate logs at midnight
+                "interval": 1,            # Every 1 day
+                "backupCount": 7,         # Keep 7 days of logs
+                "encoding": "utf-8",        # Ensure UTF-8 encoding
+                "level": "DEBUG",           # Always log DEBUG and above
+                "formatter": "detailed",    # Use detailed format
+            },
+            "console": {
+                "class": "logging.StreamHandler",
+                "level": log_level.upper(),  # Configurable log level
+                "formatter": "simple",       # Use simple format
+            },
+        },
+        "root": {
+            "level": "DEBUG",  # Root logger level
+            "handlers": ["file", "console"],  # Attach both handlers
+        },
+    }
 
-    # Remove existing handlers
+    # Add Windows Event Log handler if on Windows
-    for handler in logger.handlers[:]:
-        logger.removeHandler(handler)
-
-    # File handler always logs DEBUG and above
-    file_handler = logging.handlers.RotatingFileHandler(
-        log_file, maxBytes=5_000_000, backupCount=5, encoding='utf-8'
-    )
-    file_handler.setLevel(logging.DEBUG)
-    file_handler.setFormatter(logging.Formatter(
-        '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
-    ))
-    logger.addHandler(file_handler)
-
-    # Console handler respects the configured log level
-    console_handler = logging.StreamHandler()
-    console_handler.setLevel(getattr(logging, log_level.upper(), logging.INFO))
-    console_handler.setFormatter(logging.Formatter('%(levelname)s - %(message)s'))
-    logger.addHandler(console_handler)
-
-    # Optional Windows Event Log handler
     if platform.system() == "Windows":
         try:
-            event_handler = logging.handlers.NTEventLogHandler("AirlockTools")
+            config["handlers"]["eventlog"] = {
-            event_handler.setLevel(logging.CRITICAL)
+                "class": "logging.handlers.NTEventLogHandler",
-            event_handler.setFormatter(logging.Formatter('%(levelname)s: %(message)s'))
+                "appname": "AirlockTools",  # Event log source name
-            logger.addHandler(event_handler)
+                "level": "CRITICAL",        # Only log critical errors
+                "formatter": "simple",      # Use simple format
+            }
+            config["root"]["handlers"].append("eventlog")
         except Exception as e:
-            logger.warning(f"Could not attach Windows Event Log handler: {e}")
+            logging.warning(f"Could not attach Windows Event Log handler: {e}")
+
+    # Apply the logging configuration
+    logging.config.dictConfig(config)
+    logging.getLogger().debug("✅ Logging configured.")
 
-    logger.debug("✅ Logging configured.")
     
 def get_system_config_path() -> Path:
     base_path = Path(getattr(sys, '_MEIPASS', os.path.dirname(os.path.abspath(__file__))))
     return base_path.parent / "system_config.json"
 
+
 def load_system_config() -> dict:
     try:
         config_path = get_system_config_path()
@@ -180,4 +201,3 @@ def setup():
     logging.debug(f"Service URL set to: {url}")
 
     write_config_to_env(merged_config, env_path)
-

utils/utils.py

@@ -184,20 +184,18 @@ def displayIntro():
         )
     )
 
-
+def section_header(title):
-def printEnforceChecklist(selected_policies, destination_policy, destination_allowlist):
-    working_dir = load_env("WORKING_DIR")
-
-    def section_header(title):
     print(colorText("\n --------------------------------------------------------------------", "cyan"))
     print(colorText(f" ------------- {title} -------------", "cyan"))
     print(colorText(" --------------------------------------------------------------------", "cyan"))
 
+def printEnforceChecklist(selected_policies, destination_policy, destination_allowlist):
+    working_dir = load_env("WORKING_DIR")
     section_header("🛠️ 🔒 Prepare to Enforce Policy 🛠️ 🔒")
     print(colorText("\nSequentially follow these steps to prepare a policy for enforcement:", "white"))
 
     # Step 1: Originating Policies
-    print(colorText("\n1. Choose which originating policy or policies to move to enforcement", "cyan"))
+    print(colorText("\n1. Choose which policy or policies to gather execution info from", "cyan"))
     if not selected_policies:
         print(colorText("  [✗]  No policies have been chosen", "red"))
     else:
@@ -206,7 +204,7 @@ def printEnforceChecklist(selected_policies, destination_policy, destination_all
             print(colorText(f"  [✓] {policy.name}", "green"))
 
     # Step 2: Destination Policy and Allowlist
-    print(colorText("2. Choose the destination policy and allowlist", "cyan"))
+    print(colorText("2. Choose the destination policy and associated allowlist", "cyan"))
     if destination_policy:
         print(colorText(f"  [✓] {destination_policy[0].name} has been selected as the destination policy", "green"))
     else:
@@ -218,7 +216,7 @@ def printEnforceChecklist(selected_policies, destination_policy, destination_all
         print(colorText("  [✗]  No allowlist has been chosen", "red"))
 
     # Step 3: Data Preparation
-    print(colorText("3. Pull and stage event history, combine the histories, add hash info, then categorize the hashes", "cyan"))
+    print(colorText(f"3. Select to begin pulling execution history. The executions will be sorted and placed in {working_dir}\\data\\Needs_Review", "cyan"))
     if selected_policies:
         policy_id = selected_policies[0].name
         review_path = f"{working_dir}\\Needs_Review\\Review_First\\{policy_id}_approved_executions.csv"
@@ -228,15 +226,9 @@ def printEnforceChecklist(selected_policies, destination_policy, destination_all
 
     # Step 4: Manual Review
     print(colorText("4. Manually review the files:", "cyan"))
-    if selected_policies:
+    print(colorText("     Remove the rows containing hashes you do not approve of", "cyan"))
-        policy_id = selected_policies[0].name
+    print(colorText(f"     When complete, save both csv files to {working_dir}\\data\\Approved and choose this option.", "cyan"))
-        print(colorText(f"     'prepare_policy\\needs_approved\\good_hashes.csv' and 'prepare_policy\\needs_approved\\{policy_id}_unknown_hashes.csv'\n", "cyan"))
+    print(colorText("     This will start the process to generate possible filepath approvals", "cyan"))
-    else:
-        print(colorText("     'prepare_policy\\needs_approved\\good_hashes.csv' and 'prepare_policy\\needs_approved\\<policy>_unknown_hashes.csv'\n", "cyan"))
-
-    print(colorText("     Remove the rows containing hashes you do not approve of...", "cyan"))
-    print(colorText("     When complete, save both csv files to the directory 'approved' and choose this option.", "cyan"))
-    print(colorText("     This will combine these approved hashes with the automatically approved hashes and generate a list of paths to be reviewed", "cyan"))
 
     if selected_policies:
         policy_id = selected_policies[0].name
@@ -248,11 +240,10 @@ def printEnforceChecklist(selected_policies, destination_policy, destination_all
         print(colorText("  [✗]  No policies selected, cannot check reviewed hashes or path list", "red"))
 
     # Step 5: Path Review
-    print(colorText("5. Manually review the files \n     'prepare_policy\\needs_approved\\primary_Paths.csv'\n     'prepare_policy\\needs_approved\\secondary_Paths.csv'", "cyan"))
+    print(colorText(f"5. Manually review the files in {working_dir}\\Needs_Review\\Review_Second\\", "cyan"))
-    print(colorText("     Remove the rows containing path exclusions you do not approve of...", "cyan"))
+    print(colorText("     Remove the rows containing path exclusions or publishers you do not approve of.", "cyan"))
-    print(colorText("     When complete, save the csv file to the directory 'approved'", "cyan"))
+    print(colorText(f"     When complete, save the files to {working_dir}\\data\\Approved", "cyan"))
-    print(colorText("     Do the same process with the list of publishers for the same directories", "cyan"))
+    print(colorText("     Choose this option when done to build your preflights", "cyan"))
-    print(colorText("     Preflight Lists will be generated", "cyan"))
 
     if selected_policies:
         policy_id = selected_policies[0].name
@@ -267,12 +258,12 @@ def printEnforceChecklist(selected_policies, destination_policy, destination_all
 
     # Final Steps
     print(colorText("6. Test ------------------------------------------------------", "cyan"))
-    print(colorText("     Print rather than apply selected data.", "cyan"))
+    print(colorText("     Prints to console the changes that would be made, must be done to proceed. ", "cyan"))
 
     print(colorText("7. Liftoff ------------------------------------------------------", "cyan"))
-    print(colorText("     Apply path exclusions according to allowed and approved paths", "cyan"))
+    print(colorText("     Apply path exclusions and approved publishers to selected policy", "cyan"))
-    print(colorText("     Apply signed or attested hashes to Parent Allow List", "cyan"))
+    print(colorText("     Apply approved hashes to allowlist", "cyan"))
-    print(colorText("     Apply approved, but unsigned hashes to the Child Allow List", "cyan"))
+
 
     # Utility Options
     print(colorText("R. Remove/Reset Generated data - will prompt to allow keeping execution history", "cyan"))