diff --git a/IRT_icon_32-512.ico b/IRT_icon_32-512.ico new file mode 100644 index 0000000..103b72e Binary files /dev/null and b/IRT_icon_32-512.ico differ diff --git a/flows/prepPolicy.py b/flows/prepPolicy.py index d830931..53c1f89 100644 --- a/flows/prepPolicy.py +++ b/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: @@ -228,6 +229,19 @@ def buildPreflights(selected_policies: List[Policy]): approved_paths = pd.DataFrame() 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"): - path_exclusion_constant = get_protected_value("PATH_EXCLUSION_CONST", cast_type= int) +def splitFilepathsGrouped(df, path_exclusion_constant, col="filename"): 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) diff --git a/icon.ico b/icon.ico deleted file mode 100644 index eb6d558..0000000 Binary files a/icon.ico and /dev/null differ diff --git a/utils/menus.py b/utils/menus.py index 404dfbb..0794d8e 100644 --- a/utils/menus.py +++ b/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)) \ No newline at end of file diff --git a/utils/setup.py b/utils/setup.py index a530926..e12bef5 100644 --- a/utils/setup.py +++ b/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 - logger.setLevel(logging.DEBUG) + config = { + "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 - 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 + # Add Windows Event Log handler if on Windows if platform.system() == "Windows": try: - event_handler = logging.handlers.NTEventLogHandler("AirlockTools") - event_handler.setLevel(logging.CRITICAL) - event_handler.setFormatter(logging.Formatter('%(levelname)s: %(message)s')) - logger.addHandler(event_handler) + config["handlers"]["eventlog"] = { + "class": "logging.handlers.NTEventLogHandler", + "appname": "AirlockTools", # Event log source name + "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() @@ -179,5 +200,4 @@ def setup(): os.environ["URL"] = url logging.debug(f"Service URL set to: {url}") - write_config_to_env(merged_config, env_path) - + write_config_to_env(merged_config, env_path) \ No newline at end of file diff --git a/utils/utils.py b/utils/utils.py index fdb1d66..2f48ff3 100644 --- a/utils/utils.py +++ b/utils/utils.py @@ -184,20 +184,18 @@ def displayIntro(): ) ) +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") - - def section_header(title): - print(colorText("\n --------------------------------------------------------------------", "cyan")) - print(colorText(f" ------------- {title} -------------", "cyan")) - print(colorText(" --------------------------------------------------------------------", "cyan")) - 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: - policy_id = selected_policies[0].name - print(colorText(f" 'prepare_policy\\needs_approved\\good_hashes.csv' and 'prepare_policy\\needs_approved\\{policy_id}_unknown_hashes.csv'\n", "cyan")) - else: - print(colorText(" 'prepare_policy\\needs_approved\\good_hashes.csv' and 'prepare_policy\\needs_approved\\_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")) + print(colorText(" Remove the rows containing hashes you do not approve of", "cyan")) + print(colorText(f" When complete, save both csv files to {working_dir}\\data\\Approved and choose this option.", "cyan")) + print(colorText(" This will start the process to generate possible filepath approvals", "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(" Remove the rows containing path exclusions you do not approve of...", "cyan")) - print(colorText(" When complete, save the csv file to the directory 'approved'", "cyan")) - print(colorText(" Do the same process with the list of publishers for the same directories", "cyan")) - print(colorText(" Preflight Lists will be generated", "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 or publishers you do not approve of.", "cyan")) + print(colorText(f" When complete, save the files to {working_dir}\\data\\Approved", "cyan")) + print(colorText(" Choose this option when done to build your preflights", "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 signed or attested hashes to Parent Allow List", "cyan")) - print(colorText(" Apply approved, but unsigned hashes to the Child Allow List", "cyan")) + print(colorText(" Apply path exclusions and approved publishers to selected policy", "cyan")) + print(colorText(" Apply approved hashes to allowlist", "cyan")) + # Utility Options print(colorText("R. Remove/Reset Generated data - will prompt to allow keeping execution history", "cyan"))