From 6e4e35fa34385027b58cad99f89960e9e6a661b4 Mon Sep 17 00:00:00 2001 From: Zarithas Date: Tue, 7 Oct 2025 16:14:25 -0400 Subject: [PATCH] Fixed a breaking sort change, added some more logic to give insight on why functions arent running --- AirlockTools.py | 4 +-- flows/prepPolicy.py | 12 ++++++++- models/execution.py | 60 ++++++++++++++++++++++++++------------------- services/setup.py | 40 ++++++++++++++++-------------- utils/menus.py | 49 ++++++++++++++++++++++++------------ 5 files changed, 103 insertions(+), 62 deletions(-) diff --git a/AirlockTools.py b/AirlockTools.py index 5f9416c..5cdc396 100644 --- a/AirlockTools.py +++ b/AirlockTools.py @@ -45,7 +45,7 @@ def main(): working_dir = setup() logger = logging.getLogger(__name__) - + logger.debug("🔍 Logging test: this should appear in both console and file.") dotenv.load_dotenv(dotenv_path=working_dir / ".env") try: @@ -101,7 +101,7 @@ def main(): """ else: # Interactive logic - + raw = os.getenv("POLICY_MAP_ENF_AUD", "{}") diff --git a/flows/prepPolicy.py b/flows/prepPolicy.py index b80a07f..50ffe7e 100644 --- a/flows/prepPolicy.py +++ b/flows/prepPolicy.py @@ -176,8 +176,12 @@ def buildPathsandPublishers(split): "secondary_Paths": secondary_path_exclusions, "hashes_to_add": 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_exec", inplace=True) + else: df.sort_values(by="longestcfp", inplace=True) + df.to_csv(f"{working_dir}\\Needs_Review\\Review_Second\\{name}.csv", index=False) formatHTML(df, f"{working_dir}\\Needs_Review\\HTML\\{name}.html") @@ -190,6 +194,7 @@ def buildPathsandPublishers(split): pattern = regulator(load_env_json("BAD_PUBLISHERS","[]")) publist = publist[~publist["publisher_hash"].str.contains(pattern, na=False)] publist = publist[["publisher_hash"]] + publist.sort_values(by="publisher_hash", inplace=True) publist.to_csv(f"{working_dir}\\Needs_Review\\Review_Second\\publishers.csv", index=False) def buildPreflights(): @@ -236,6 +241,11 @@ def buildPreflights(): dataframes = {"approved_paths": approved_paths, "approved_hashes": approved_hashes, "approved_publishers": approved_publishers} for name, df in dataframes.items(): + logger.debug(f" DataFrame headers: {list(df.columns)}") + if name == "approved_paths":df.sort_values(by="longestcfp", inplace=True) + elif name == "approved_hashes":df.sort_values(by="filename_exec", inplace=True) + elif name == "approved_publishers" : df.sort_values(by="publisher_hash", inplace=True) + df.to_csv(f"{working_dir}\\Preflight\\{name}.csv", index=False) formatHTML(df, f"{working_dir}\\Preflight\\HTML\\{name}.html") diff --git a/models/execution.py b/models/execution.py index 139777e..73d903d 100644 --- a/models/execution.py +++ b/models/execution.py @@ -126,7 +126,6 @@ class Hash: @classmethod def categorize_hashes(cls, hashes): - threat_tolerance = load_env("VT_THREAT_TOLERANCE", cast_type=int) bad_publishers_pattern = regulator(load_env_json("BAD_PUBLISHERS", "[]")) pups_pattern = regulator(load_env_json("PUPS", "[]")) @@ -135,38 +134,48 @@ class Hash: approved = [] unapproved = [] - def reputationtool(hash_obj): - val = hash_obj.reputation.get("scannermatch") if isinstance(hash_obj.reputation, dict) else None - if val in [None, "N/A"]: - return hash_obj.publisher == "Not Signed" - try: - return int(val) > threat_tolerance - except (ValueError, TypeError): - return hash_obj.publisher == "Not Signed" - for hash_obj in hashes: publisher = hash_obj.publisher or "" description = hash_obj.description or "" reputation = hash_obj.reputation if isinstance(hash_obj.reputation, dict) else {} - rep_status = reputation.get("status") + scannermatch = reputation.get("scannermatch") - rep_flag = reputationtool(hash_obj) + logger.debug(f"Evaluating hash: {hash_obj}") + logger.debug(f"Publisher: {publisher}, Description: {description}, Scannermatch: {scannermatch}") - is_signed = publisher != "Not Signed" - is_untrusted = re.search(bad_publishers_pattern, publisher, re.IGNORECASE) is not None - is_pup = re.search(pups_pattern, description, re.IGNORECASE) is not None - has_known_status = rep_status == "KNOWN" - - if (not is_signed and rep_flag) or rep_status == "UNKNOWN": - needs_review.append(hash_obj) - elif ( - (is_signed and not is_untrusted and has_known_status and not is_pup) or - (not is_signed and not rep_flag and not is_untrusted and has_known_status and not is_pup) - ): - approved.append(hash_obj) - else: + # 1. Unapproved: bad publisher or PUP + if re.search(bad_publishers_pattern, publisher, re.IGNORECASE): + logger.debug("Unapproved: Publisher matches bad publisher pattern.") unapproved.append(hash_obj) + continue + if re.search(pups_pattern, description, re.IGNORECASE): + logger.debug("Unapproved: Description matches PUP pattern.") + unapproved.append(hash_obj) + continue + + # 2. Approved: signed + if publisher != "Not Signed": + logger.debug("Approved: File is signed and not flagged.") + approved.append(hash_obj) + continue + + # 3. Approved or Unapproved based on threat level + try: + score = int(scannermatch) + logger.debug(f"Parsed scannermatch score: {score}") + if score > threat_tolerance: + logger.debug("Unapproved: Unsigned file with high threat score.") + unapproved.append(hash_obj) + else: + logger.debug("Approved: Unsigned file with low threat score.") + approved.append(hash_obj) + except (ValueError, TypeError): + logger.debug("Needs Review: Scannermatch score is missing or invalid.") + needs_review.append(hash_obj) + + + logger.debug(f"Final counts — Needs Review: {len(needs_review)}, Approved: {len(approved)}, Unapproved: {len(unapproved)}") return needs_review, approved, unapproved @@ -287,6 +296,7 @@ class ExecutionHistoryRecord: how="left", # Preserve all executions, enrich where possible suffixes=("_exec", "_hash") ) + merged_df.sort_values(by="filename_exec", inplace=True) logger.info(f"Merged {len(merged_df)} rows. Non-null hash matches: {merged_df['sha256'].notna().sum()}") filename = f"{label}_executions.csv" diff --git a/services/setup.py b/services/setup.py index 7ed20aa..5483f71 100644 --- a/services/setup.py +++ b/services/setup.py @@ -48,27 +48,31 @@ def configure_logging(log_dir: Path, log_level: str = "DEBUG"): logger = logging.getLogger() logger.setLevel(getattr(logging, log_level.upper(), logging.DEBUG)) - if not logger.handlers: - file_handler = logging.handlers.RotatingFileHandler( - log_file, maxBytes=5_000_000, backupCount=5, encoding='utf-8' - ) - file_handler.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')) - logger.addHandler(file_handler) + # 🔧 Clear existing handlers + for handler in logger.handlers[:]: + logger.removeHandler(handler) - console_handler = logging.StreamHandler() - console_handler.setFormatter(logging.Formatter('%(levelname)s - %(message)s')) - logger.addHandler(console_handler) + file_handler = logging.handlers.RotatingFileHandler( + log_file, maxBytes=5_000_000, backupCount=5, encoding='utf-8' + ) + file_handler.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')) + logger.addHandler(file_handler) - 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) - except Exception as e: - logger.warning(f"Could not attach Windows Event Log handler: {e}") + console_handler = logging.StreamHandler() + console_handler.setFormatter(logging.Formatter('%(levelname)s - %(message)s')) + logger.addHandler(console_handler) + + 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) + except Exception as e: + logger.warning(f"Could not attach Windows Event Log handler: {e}") + + logger.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__)))) diff --git a/utils/menus.py b/utils/menus.py index d59e13e..631eca6 100644 --- a/utils/menus.py +++ b/utils/menus.py @@ -149,22 +149,15 @@ def menu_policy_enforce(api: AirlockAPIWrapper): pathexclusions = pd.read_csv(f"{working_dir}\\Preflight\\approved_paths.csv") hashes = pd.read_csv(f"{working_dir}\\Preflight\\approved_hashes.csv") - # Get unique combinations of longestcfp and file_extension - unique_combinations = pathexclusions[ - ["longestcfp", "file_extension"] - ].drop_duplicates() + unique_combinations = pathexclusions[["longestcfp", "file_extension"]].drop_duplicates() - # Regex to match a Windows drive letter at the start (e.g., C:\) drive_letter_pattern = re.compile(r"^[a-zA-Z]:\\") - - # Build processed paths like \\path\\**.exe or C:\path\**.jar processed_paths = [ (path if drive_letter_pattern.match(path) else f"\\\\{path}") + f"\\**{ext}" for path, ext in unique_combinations.itertuples(index=False, name=None) ] print(processed_paths) - print(colorText("These publishers would added", "yellow")) if os.path.exists(f"{working_dir}\\Preflight\\approved_publishers.csv"): @@ -173,13 +166,12 @@ def menu_policy_enforce(api: AirlockAPIWrapper): print(colorText("The publishers list is empty.", "red")) else: processed_publishers = ( - publishers[publishers["publisher_hash"] != "Not Signed"] - ["publisher_hash"] - .drop_duplicates() - .tolist() - ) - - print(processed_publishers) + publishers[publishers["publisher_hash"] != "Not Signed"] + ["publisher_hash"] + .drop_duplicates() + .tolist() + ) + print(processed_publishers) print(colorText("These hashes would be added to:", "yellow")) print(destination_allowlist) @@ -189,6 +181,22 @@ def menu_policy_enforce(api: AirlockAPIWrapper): if processed_paths and processed_hashes: tested = True + else: + # Log which condition(s) failed + missing_items = [] + if not os.path.exists(f"{working_dir}\\Preflight\\approved_paths.csv"): + missing_items.append("approved_paths.csv not found") + if not os.path.exists(f"{working_dir}\\Preflight\\approved_hashes.csv"): + missing_items.append("approved_hashes.csv not found") + if not destination_policy: + missing_items.append("destination_policy is empty or None") + if not destination_allowlist: + missing_items.append("destination_allowlist is empty or None") + + logger.error("Preflight check failed due to the following:") + for item in missing_items: + logger.error(f" - {item}") + elif choice == "7": areYouSure() @@ -205,7 +213,16 @@ 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) - + else: + logger.error("Confirmation block failed. Reasons:") + if not tested: + logger.error(" - Preflight checks were not completed successfully (`tested` is False).") + if not destination_policy: + logger.error(" - `destination_policy` is missing or invalid.") + if not destination_allowlist: + logger.error(" - `destination_allowlist` is missing or invalid.") + if confirmation.strip().upper() != "I AGREE": + logger.error(" - User did not confirm with 'I AGREE'. Received: '%s'", confirmation.strip()) elif choice == "F": open_directory(working_dir)