# Copyright (C) 2025 James Brotosky, Brandon Wickline # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as published # by the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. # # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . import logging import os import platform import re import subprocess import tempfile import tkinter as tk from tkinter import filedialog import pandas as pd logger = logging.getLogger(__name__) def import_to_dataframe(file_path: str) -> pd.DataFrame: df = pd.DataFrame() try: if not os.path.exists(file_path): print(colorText(f"Error: File '{file_path}' does not exist.", "red")) return df ext = os.path.splitext(file_path)[1].lower() if ext == ".csv": df = pd.read_csv(file_path) elif ext == ".parquet": df = pd.read_parquet(file_path) else: print(colorText(f"Error: Unsupported file extension '{ext}'.", "red")) return df if df.empty: print(colorText("Error: File has headers but no data rows.", "red")) else: print(colorText(f"Data loaded successfully from {file_path}", "green")) return df except pd.errors.EmptyDataError: print( colorText( "Notice: CSV file is completely empty, falling back to empty frame", "white", ) ) return pd.DataFrame() except Exception as e: print(colorText(f"Error reading file: {e}", "red")) return pd.DataFrame() def choose_directory(): root = tk.Tk() root.withdraw() # Hide the main window directory = filedialog.askdirectory(title="Select a Directory") print("Selected directory:", directory) return directory def choose_file(initial_directory=None, required_substring=None): """Open a file dialog and ensure the selected file contains a required substring.""" while True: root = tk.Tk() root.withdraw() # Hide the main window file_path = filedialog.askopenfilename(initialdir=initial_directory) if not file_path: print("No file selected.") return None if required_substring and required_substring not in file_path: print( f"The selected file must contain '{required_substring}' in its path or name. Please try again." ) else: return file_path def get_sanitized_input(prompt: str) -> str: while True: user_input = input(prompt) if user_input.strip() == "": return user_input # Allow blank lines if re.match(r"^[a-zA-Z0-9_\- .]+$", user_input.strip()): return user_input else: print( "Invalid input. Only letters, numbers, underscores, spaces, hyphens, and periods are allowed." ) def regulator(paths, case_insensitive=True): """ Build a regex pattern that matches any of the given Windows path fragments. """ escaped = [re.escape(p) for p in paths] pattern = "(?:" + "|".join(escaped) + ")" if case_insensitive: pattern = "(?i)" + pattern # Add inline case-insensitive flag print(f"Regulator is providing: {pattern}") return pattern def irtang(): print( colorText( r""" ███ ████ ░████████ █████████████ ███████████████ █████████████████████ █████████████████████ ███████████████████ ██████████████████████▓ ███████████████████ ██████████████████████ █████████████████████ ███████████████████████ ████████████████████████████████████████████████████████ █████████ ██ ██ █████████ █████████ ██ ███ █ █████████ █████████ ██ ████ █████ █████████████ █████████ ██ ██████ █████████████ ████████ ██ ███████ ████████████░ ███████ ██ ██▓ ██████ ████████████ ██████ ██ ████ █████ ███████████ █████████████████████████████████████████████████ ▒████████████████████ ██████████████████ ███████████████████ ███████████████▒ ███████████████ █████████████ ██████████ ███████████ ████████ ████ """, "yellow", ) ) def section_header(title): print( colorText( "\n --------------------------------------------------------------------", "cyan", ) ) print(colorText(f" ------------- {title} -------------", "cyan")) print( colorText( " --------------------------------------------------------------------", "cyan", ) ) def areYouSure(): print( colorText( "🛑****************************************************************************************************************************************🛑", "red", ) ) print( colorText( "⚠️=========================================================================================================================================⚠️", "yellow", ) ) print( colorText( "🛑========================================================================================================================================🛑", "red", ) ) print( colorText( "⚠️-------------This program will now begin to make changes to the Airlock Console. Do you understand and agree to proceed? ----------------⚠️", "yellow", ) ) print( colorText( "🛑========================================================================================================================================🛑", "red", ) ) print( colorText( "⚠️=========================================================================================================================================⚠️", "yellow", ) ) print( colorText( "🛑****************************************************************************************************************************************🛑", "red", ) ) def locked(): print( colorText( r""" ████████████████████████████████████████████████████████████████ ███ ██ ██ ██████ ███ ██ ████████████ ███ ██ ████ ███ ███ ██ ███ ███ ███ ██ ███ ███ ███ ██ ▒████████████████████ ███ ██ ██████████████████████ ███ ██ ██████████████████████ ███ ██ ██████████████████████ ███ ██ ██████████████████████ ███ ██ ██████████████████████ ███ ██ ███ ███ ███ ████████████████████████████████████████████████████████████████████ ▒██████████████████████████████████████████████████████████████████▒ ▒████ ▒████ ▓██████████████████████████████████████████ █████████████████████████████████████████████░ """, "yellow", ) ) def printDeviceEnforceChecklist(): print( colorText( "\n --------------------------------------------------------------------", "cyan", ) ) print( colorText( " ------------- 🛠️ 🔒 Prepare to Enforce Policy 🛠️ 🔒 ------------------", "cyan", ) ) print( colorText( " --------------------------------------------------------------------", "cyan", ) ) print( colorText( "\nSequentually follow these steps to prepare a policy for enforcement:", "white", ) ) print( colorText( "\n1. Choose which originating policy or policies to move to enforcement", "cyan", ) ) print( colorText( "2. Pull and stage event history, combine the histories, add hash info, then categorize the hashes", "cyan", ) ) print(colorText("3. Manually review the files:", "cyan")) print( colorText( " 'needs_approved\\good_{first_policy}_{second_policy}.csv' and 'needs_approved\\unknown_{first_policy}_{second_policy}.csv'", "cyan", ) ) print( colorText( " Remove the rows containing hashes you do not approve of, and those you would not approve of without metarules.", "cyan", ) ) print( colorText( " If metarules need to be created, please make note of them, and remove the row from the csv.", "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( "4. Manually review the file 'needs_approved\\paths_needing_review.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 forthe same directories", "cyan", ) ) print(colorText(" Preflight Lists will be generated", "cyan")) print( colorText( "5. Choose the destination policy and parent and child allow list", "cyan" ) ) print( colorText( "6. Test ------------------------------------------------------", "cyan" ) ) print(colorText(" Print rather than apply selected data.", "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( "R. Remove/Reset Generated data - will prompt to allow keeping execution history", "cyan", ) ) print(colorText("B. Back", "cyan")) def colorText(text, color): 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 formatHTML(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 = """ """ header = f"""

Airlock Tools

{filename} - {today}

""" html_table = df.to_html(index=False, escape=False) styled_html = ( f"\n" f"Airlock Tools Report\n" f"\n" f"{dark_css}\n" f"{header}\n" f"
\n" f" {html_table}\n" f"
\n" f"\n" f"" ) 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: with tempfile.NamedTemporaryFile( suffix=".html", delete=False, mode="w", encoding="utf-8" ) as f: f.write(styled_html) temp_path = f.name print(f"✅ Styled table saved to temporary file: {temp_path}") else: return styled_html def open_directory(path): system = platform.system() if system == "Windows": os.startfile(path) elif system == "Linux": subprocess.run(["xdg-open", path]) else: raise OSError(f"Unsupported operating system: {system}") def print_x_wide(items: list, width: int): for i in range(0, len(items), width): row = items[i : i + width] print(" | ".join(row)) def clear_screen(): os.system("cls" if os.name == "nt" else "clear")