"
+
+
+"""
+executions = ExecutionHistoryRecord.from_policies(api, selected_policies, type_=[0,1,3], history_days=30)
+
+ExecutionHistoryRecord.enrich_with_hashes_and_export(executions, hash_objects, "C:/Users/Brandon/Documents/EnrichedExports")
+"""
diff --git a/Development/WIP/Sync Code/menus.py b/Development/WIP/Sync Code/menus.py
new file mode 100644
index 0000000..01295a6
--- /dev/null
+++ b/Development/WIP/Sync Code/menus.py
@@ -0,0 +1,316 @@
+# 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 re
+
+import dotenv
+import pandas as pd
+
+import services.policyhandler as policyh
+from flows.otp import generate, otp_activities_by_agent, revoke
+from flows.prepPolicy import (
+ buildPathsandPublishers,
+ buildPreflights,
+ selectAllowlists,
+ selectPolicies,
+ sortHashes,
+)
+from flows.quietAgent import findQuietAgents
+from services.agenthandler import findAgents, moveAgentToRelatedPolicy, selectAgents
+from services.API import AirlockAPIWrapper
+from utils.configmanager import load_env
+from utils.selector import Selector
+from utils.utils import (
+ areYouSure,
+ colorText,
+ displayIntro,
+ get_sanitized_input,
+ open_directory,
+ printEnforceChecklist,
+)
+
+logger = logging.getLogger(__name__)
+
+dotenv.load_dotenv()
+
+def menu_main(api: AirlockAPIWrapper):
+ working_dir = load_env("WORKING_DIR")
+ extras = load_env("EXTRAS")
+ while True:
+ displayIntro()
+ # Add Settings, and give option to change working dir
+ print(colorText("1. โ
- Move Device(s) to local approval", "yellow"))
+ print(colorText("2. ๐ซ - OTP", "yellow"))
+ print(colorText("3. ๐ - Move to Audit/Enforcement", "yellow"))
+ print(colorText("4. ๐ - Device Search", "yellow"))
+ print(colorText("5. ๐ - Find Quiet Hosts", "yellow"))
+ if extras == "POLICYPREP" : print(colorText("6. ๐ก๏ธ - Policy Enforcement Tools", "yellow"))
+ print(colorText("F. ๐ - Open Working Directory", "yellow"))
+ print(colorText("S. ๐ ๏ธ - Settings", "yellow"))
+ print(colorText("Q. ๐ - Quit", "yellow"))
+
+ choice = get_sanitized_input("\nEnter Menu Item: ")
+ if choice == "1":
+ print("This Feature is still in development")
+ get_sanitized_input("Press enter to continue")
+ elif choice == "2":
+ menu_otp(api)
+ elif choice == "3":
+ choices = ["audit", "enforcement"]
+ print(colorText("Move devices to which state?:", "yellow"))
+ direction = Selector.select_string(choices, False, False)
+ devices = selectAgents(api)
+ print(colorText("Would you like to continue with these devices?","white"))
+ for device in devices:
+ print(device.hostname)
+ confirm = Selector.confirm()
+ if direction and devices and confirm:
+ for device in devices:
+ moveAgentToRelatedPolicy(api,device, direction[0])
+ elif choice == "4":
+ findAgents(api,False)
+ elif choice == "5":
+ findQuietAgents(api)
+ elif choice == "6":
+ if extras == "POLICYPREP": menu_policymanagment(api)
+ elif choice.upper() == "F":
+ open_directory(working_dir)
+ elif choice.upper() == "S":
+ menu_settings()
+ elif choice.upper() == "Q":
+ break
+ else:
+ print(colorText("Invalid choice. Please try again.", "red"))
+
+
+
+def menu_policy_enforce(api: AirlockAPIWrapper):
+ selected_policies = []
+ destination_policy = []
+ destination_allowlist = []
+ processed_paths = []
+ processed_hashes = []
+ processed_publishers = []
+ tested = False
+ working_dir = load_env("WORKING_DIR")
+
+ while True:
+ printEnforceChecklist(selected_policies, destination_policy, destination_allowlist)
+ choice = get_sanitized_input("\nEnter your choice: ")
+
+ if choice == "1":
+ selected_policies = selectPolicies(api,True)
+
+ elif choice == "2":
+ print(colorText("Please choose destination_name Policy for Path Exclusions", "white"))
+
+ destination_policy = selectPolicies(api, False)
+
+ print(colorText("Please choose Allowlist for Hashes", "white"))
+
+ destination_allowlist = selectAllowlists(api, destination_policy, False)
+
+ elif choice == "3":
+ sortHashes(
+ api,
+ selected_policies,
+ type=[1, 2, 6, 7],
+ )
+
+ elif choice == "4":
+ if os.path.exists(f"{working_dir}\\Needs_Review\\Review_First\\approved_executions.csv"):
+ buildPathsandPublishers(False)
+ else:
+ 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\\hashes_to_add.csv") and os.path.exists(
+ f"{working_dir}\\Approved\\primary_Paths.csv"
+ ):
+ buildPreflights()
+ else:
+ print("File not found. Please make sure it's saved correctly and try again.")
+
+ elif choice == "6":
+ if (
+ os.path.exists(f"{working_dir}\\Preflight\\approved_paths.csv")
+ and os.path.exists(f"{working_dir}\\Preflight\\approved_hashes.csv")
+ and destination_policy
+ and destination_allowlist
+ ):
+ print(colorText("These path exclusions would be added to:", "yellow"))
+ print(destination_policy)
+
+ pathexclusions = pd.read_csv(f"{working_dir}\\Preflight\\approved_paths.csv")
+ hashes = pd.read_csv(f"{working_dir}\\Preflight\\approved_hashes.csv")
+
+ unique_combinations = pathexclusions[["longestcfp", "file_extension"]].drop_duplicates()
+
+ drive_letter_pattern = re.compile(r"^[a-zA-Z]:\\")
+ 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"):
+ publishers = pd.read_csv(f"{working_dir}\\Preflight\\approved_hashes.csv")
+ if publishers.empty:
+ 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)
+
+ print(colorText("These hashes would be added to:", "yellow"))
+ print(destination_allowlist)
+
+ processed_hashes = hashes["sha256"].unique().tolist()
+ print(processed_hashes)
+
+ 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()
+ confirmation = get_sanitized_input("Type 'I AGREE' to continue: ")
+ if (
+ tested
+ and destination_policy
+ and destination_allowlist
+ and confirmation.strip() == "I AGREE"
+ ):
+ print(colorText("Proceeding with the code...", "yellow"))
+ api.hash_add_to_allowlist(destination_allowlist[0].applicationid, processed_hashes)
+ 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() != "I AGREE":
+ logger.error(" - User did not confirm with 'I AGREE'. Received: '%s'", confirmation.strip())
+
+ elif choice.upper() == "F":
+ open_directory(working_dir)
+ elif choice.upper() == "S":
+ menu_settings()
+ elif choice.upper == "B":
+ break
+
+
+ else:
+ print(colorText("Invalid choice. Please try again.", "red"))
+
+
+def menu_otp(api: AirlockAPIWrapper):
+ working_dir = load_env("WORKING_DIR")
+ while True:
+
+ print(colorText("\n--- ๐ซ OTP Submenu ๐ซ ---", "cyan"))
+ print(colorText("1. ๐ -Generate OTPs", "cyan"))
+ print(colorText("2. ๐ -OTP Activities By Agent", "cyan"))
+ print(colorText("3. โ -Revoke OTPs", "cyan"))
+ print(colorText("F. ๐ - Open Working Directory", "yellow"))
+ print(colorText("S. ๐ ๏ธ - Settings", "yellow"))
+ print(colorText("B. ๐ - Back", "yellow"))
+
+ choice = get_sanitized_input("Enter your choice: ")
+
+ if choice == "1":
+ otp_list = generate(api)
+ print(colorText(otp_list,"green"))
+ elif choice == "2":
+ otp_activities_by_agent(api)
+ elif choice == "3":
+ revoke(api)
+ elif choice.upper() == "F":
+ open_directory(working_dir)
+ elif choice.upper() == "S":
+ menu_settings()
+ elif choice.upper() == "B":
+ break
+
+def menu_policymanagment(api: AirlockAPIWrapper):
+ working_dir = load_env("WORKING_DIR")
+ while True:
+ print(colorText("1. ๐ - Prepare Policy For Enforcement", "yellow"))
+ print(colorText("2. ๐ - Update Audit Policies from Enforcement Policies", "yellow"))
+ print(colorText("F. ๐ - Open Working Directory", "yellow"))
+ print(colorText("S. ๐ ๏ธ - Settings", "yellow"))
+ print(colorText("B. ๐ - Back", "yellow"))
+ choice = get_sanitized_input("\n Enter Menu Item: ")
+
+ if choice == "1":
+ menu_policy_enforce(api)
+ elif choice == "2":
+ areYouSure()
+ confirmation = get_sanitized_input("Type 'I AGREE' to continue: ")
+ if confirmation.strip() == "I AGREE":
+ policyh.updateAuditPoliciesFromEnforcementPolices(api)
+
+ elif choice.upper() == "F":
+ open_directory(working_dir)
+ elif choice.upper() == "S":
+ menu_settings()
+ elif choice.upper() == "B":
+ break
+ else:
+ print(colorText("Invalid choice. Please try again.", "red"))
+def menu_settings():
+ while True:
+ print(colorText("\n--- ๐ ๏ธ Settings Submenu ๐ ๏ธ ---", "cyan"))
+ print(colorText("This Feature is still in development", "cyan"))
+ # print(colorText("2. Sub-option B","cyan"))
+ print(colorText("B. ๐ - Back", "yellow"))
+ choice = get_sanitized_input("Enter your choice: ")
+
+ if choice == "1":
+ pass #TODO ADD CHANGE WORKDIR CODE
+
+ elif choice.upper() == "B":
+ print("Returning to Main Menu...")
+ break
+ else:
+ print("Invalid choice. Please try again.")
diff --git a/Development/WIP/Sync Code/otp.py b/Development/WIP/Sync Code/otp.py
new file mode 100644
index 0000000..4f5d52b
--- /dev/null
+++ b/Development/WIP/Sync Code/otp.py
@@ -0,0 +1,69 @@
+# 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
+
+from services.agenthandler import selectAgents
+from services.API import AirlockAPIWrapper
+from utils.selector import Selector
+from utils.utils import colorText, get_sanitized_input
+
+logger = logging.getLogger(__name__)
+
+
+
+
+def generate(api: AirlockAPIWrapper):
+ otp_dict = {}
+ agents = selectAgents(api)
+ print(colorText("Would you like to continue with these devices?","white"))
+ for agent in agents:
+ print(agent.hostname)
+ confirm = Selector.confirm()
+ if agents and confirm:
+ requester = get_sanitized_input("Who is requesting the OTP: ")
+ because = get_sanitized_input("Why/What work are they doing?: ")
+
+ purpose = f"Requester: {requester} - for : {because}"
+ possible_durations = [15, 60, 360, 1440, 10080]
+
+ print(colorText("Please select a duration in minutes: ", "white"))
+ print(colorText("15 mins, 60 mins, 360 mins(6 Hours), 1440 mins (24 Hours), 10080 mins (7 Days):", "white"))
+ duration_selected = Selector.select_int(possible_durations)
+ if duration_selected:
+ for agent in agents:
+ otp_code = api.otp_generate(agent.agentid, duration_selected, purpose)
+ logger.info(f"Generated OTP for {agent.hostname}: {otp_code}")
+ otp_dict[agent.hostname] = otp_code
+
+ return otp_dict
+
+def otp_activities_by_agent(api: AirlockAPIWrapper):
+ agents = selectAgents(api)
+ otp_dict = {}
+ for agent in agents:
+ otp_info = api.otp_find_by_agent(agent.agentid)
+ otp_dict[agent.hostname] = otp_info
+
+ return otp_dict
+
+def revoke(api: AirlockAPIWrapper):
+ otp_dict = otp_activities_by_agent(api)
+ list_to_revoke = [entry["otpid"] for entry in otp_dict]
+ if otp_dict and list_to_revoke:
+ for revokee in list_to_revoke:
+ api.otp_revoke(revokee)
\ No newline at end of file
diff --git a/Development/WIP/Sync Code/policyhandler.py b/Development/WIP/Sync Code/policyhandler.py
new file mode 100644
index 0000000..f76b969
--- /dev/null
+++ b/Development/WIP/Sync Code/policyhandler.py
@@ -0,0 +1,237 @@
+# 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 datetime
+import gc
+import json
+import logging
+import os
+import sys
+
+import pandas as pd
+import tqdm
+from bson import ObjectId
+
+from models.policy import Policy
+from services.API import AirlockAPIWrapper
+from utils.configmanager import get_protected_json
+from utils.setup import get_base_directory
+from utils.utils import colorText
+
+logger = logging.getLogger(__name__)
+
+
+
+def pullPolicyExechistories(
+ api: AirlockAPIWrapper,
+ policy: Policy,
+ type: list,
+ days,
+ outputjson: bool,
+):
+
+ file_path = f"{get_base_directory()}\\cache\\chunkinator.json"
+
+ # Ensure the file exists
+ if not os.path.exists(file_path):
+ with open(file_path, "w") as file:
+ json.dump({"error": "Success", "response": {"exechistories": []}}, file)
+ logger.debug(f"File '{file_path}' has been created.")
+ else:
+ logger.debug(f"File '{file_path}' already exists.")
+
+ checkpoint = str(skipback(days))
+ json_output = {"error": "Success", "response": {"exechistories": []}}
+
+ with tqdm.tqdm(
+ file=sys.stdout,
+ leave=True,
+ total=10000,
+ desc=f"Checkpoint Progress: {checkpoint}",
+ colour="blue",
+ initial=1,
+ ) as filebar:
+ with tqdm.tqdm(
+ file=sys.stdout,
+ leave=True,
+ total=100,
+ desc=f"Total of {policy} Complete: ",
+ ) as pbar:
+ while True:
+ histories = api.history_logging(
+ type=type, checkpoint=checkpoint, policy= [policy.name]
+ )
+
+ # Ensure histories is a list of dictionaries
+ if not isinstance(histories, list) or not all(
+ isinstance(h, dict) for h in histories
+ ):
+ logger.error(
+ "Unexpected response format from API. Expected list of dictionaries."
+ )
+ break
+
+ filebar.total = len(histories)
+
+ if not histories:
+ break
+
+ for index, history_item in enumerate(histories):
+ if (
+ "checkpoint" not in history_item
+ or "datetime" not in history_item
+ ):
+ continue # Skip malformed entries
+
+ # Update checkpoint on last item
+ if index == len(histories) - 1:
+ checkpoint = history_item["checkpoint"] # pyright: ignore[reportArgumentType]
+ filebar.desc = f"Checkpoint Progress: {checkpoint}"
+ break
+
+ try:
+ history_date = datetime.datetime.strptime(
+ history_item["datetime"].replace(" +0000 UTC", ""), # pyright: ignore[reportArgumentType]
+ "%Y-%m-%dT%H:%M:%SZ",
+ ).date()
+ except ValueError:
+ continue # Skip if date format is invalid
+
+ if (
+ datetime.date.today() - datetime.timedelta(days=days)
+ ) <= history_date:
+ json_output["response"]["exechistories"].append(history_item)
+
+ filebar.update(1)
+ filebar.refresh()
+
+ # Deduplicate entries
+ seen = {}
+ if os.path.exists(file_path):
+ with open(file_path, "r") as file:
+ existing_data = json.load(file)
+ combined = (
+ existing_data["response"]["exechistories"]
+ + json_output["response"]["exechistories"]
+ )
+ else:
+ combined = json_output["response"]["exechistories"]
+
+ for entry in combined:
+ key = (
+ entry.get("sha256"),
+ entry.get("filename"),
+ entry.get("hostname"),
+ )
+ seen[key] = entry
+
+ deduplicated = list(seen.values())
+ with open(file_path, "w") as file:
+ json.dump(
+ {
+ "error": "Success",
+ "response": {"exechistories": deduplicated},
+ },
+ file,
+ )
+
+ json_output["response"]["exechistories"].clear()
+
+ # Update progress bar based on last valid item
+ try:
+ last_date = datetime.datetime.strptime(
+ history_item["datetime"].replace(" +0000 UTC", ""), # type: ignore
+ "%Y-%m-%dT%H:%M:%SZ",
+ ).date()
+ date_diff = datetime.date.today() - last_date
+ percentage_diff = (
+ ((days + 10) - date_diff.days) / (days + 10)
+ ) * 100
+ pbar.n = round(percentage_diff)
+ pbar.set_description_str(f"Total of {policy} Complete: ")
+ pbar.refresh()
+ except Exception:
+ pass
+
+ filebar.n = 1
+
+ # Final output
+ with open(file_path, "r") as file:
+ final_output = json.load(file)
+ os.remove(file_path)
+
+ return json.dumps(final_output) if outputjson else None
+
+
+def getPolicyInfo(api: AirlockAPIWrapper, policy, type, days):
+ executionhist_policy = pd.DataFrame()
+ exehist = pullPolicyExechistories(api, policy, type, days, True)
+ if exehist is not None:
+ data = json.loads(exehist)
+ executionhist_policy = pd.DataFrame(data["response"]["exechistories"])
+ if not executionhist_policy.empty:
+ executionhist_policy = executionhist_policy[
+ [
+ "datetime",
+ "sha256",
+ "publisher",
+ "filename",
+ "hostname",
+ "username",
+ "pprocess",
+ "gprocess",
+ "commandline",
+ ]
+ ]
+ executionhist_policy["policy"] = policy # Add policy column here
+ executionhist_policy = executionhist_policy.drop_duplicates(
+ subset=["sha256", "filename", "hostname"]
+ )
+ executionhist_policy = executionhist_policy.sort_values(
+ by=["sha256", "filename"]
+ )
+ logger.debug( f"Staging of Execution history for policy: {policy} is complete")
+ print(
+ colorText(
+ f"Staging of Execution history for policy: {policy} is complete",
+ "green",
+ )
+ )
+ del data
+ del exehist
+ gc.collect()
+ return executionhist_policy
+
+
+def skipback(days):
+ """
+ Generate a MongoDB ObjectId for a given number of days ago from today.
+ """
+ adjusted_days = days
+ date_days_ago = datetime.datetime.now(datetime.UTC) - datetime.timedelta(
+ days=adjusted_days
+ )
+ timestamp = int(date_days_ago.timestamp())
+ hex_timestamp = format(timestamp, "08x")
+ objectid_hex = hex_timestamp + "0000000000000000"
+ return ObjectId(objectid_hex)
+
+
+def updateAuditPoliciesFromEnforcementPolices(api: AirlockAPIWrapper):
+ policy_relationship_map = get_protected_json("POLICY_MAP_ENF_AUD", "{}")
+ for enforcement_policy, audit_policy in policy_relationship_map.items():
+ api.policy_clone(enforcement_policy, audit_policy)
+ api.policy_set_auditmode(audit_policy, "1")
diff --git a/Development/WIP/Sync Code/prepPolicy.py b/Development/WIP/Sync Code/prepPolicy.py
new file mode 100644
index 0000000..249edc8
--- /dev/null
+++ b/Development/WIP/Sync Code/prepPolicy.py
@@ -0,0 +1,365 @@
+# 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 os.path
+from typing import List, Optional
+
+import dotenv
+import pandas as pd
+
+from models.execution import ExecutionHistoryRecord, Hash
+from models.policy import Allowlist, Policy
+from services.API import AirlockAPIWrapper
+from utils.configmanager import get_protected_value, load_env, load_env_json
+from utils.selector import Selector
+from utils.utils import (
+ colorText,
+ formatHTML,
+ import_to_dataframe,
+ regulator,
+)
+
+logger = logging.getLogger(__name__)
+
+dotenv.load_dotenv()
+
+
+
+def selectPolicies(api: AirlockAPIWrapper, allow_multiple=True) -> List[Policy]:
+
+ policies = [Policy(**row.to_dict()) for _, row in api.policy_find_all().iterrows()]
+ logger.debug("Prompting for Policies")
+ print(colorText("Please select policy/policies", "white"))
+ selected = Selector.select_objects(policies, allow_multiple, prompt_each=True)
+
+ if selected is None:
+ return []
+
+ # Normalize to always return a list
+ logger.debug("Returning {selected.dict}")
+ return selected if isinstance(selected, list) else [selected]
+
+
+def selectAllowlists(api: AirlockAPIWrapper, policy = all, allow_multiple=True) -> List[Allowlist]:
+ if policy == "all": allowlists = [Allowlist(**row.to_dict()) for _, row in api.allowlist_find_all().iterrows()]
+ else: allowlists = [Allowlist(**row.to_dict()) for _, row in api.policy_list_allowlists(policy[0].groupid).iterrows()]
+ logger.debug("Prompting for Allowlist(s)")
+ print(colorText("Please select allowlist(s)", "white"))
+ selected = Selector.select_objects(allowlists, allow_multiple, prompt_each=True)
+
+ if selected is None:
+ return []
+
+ # Normalize to always return a list
+ logger.debug(f"Returning {selected}")
+ return selected if isinstance(selected, list) else [selected]
+
+
+
+def sortHashes(
+ api: AirlockAPIWrapper,
+ selected_policies: List[Policy],
+ type=[1, 2, 6, 7],
+ history_days: Optional[int] = None
+):
+
+
+ if history_days is None:
+ history_days = Selector.select_value(
+ prompt="Enter how many days of history to pull (1โ150): ",
+ value_type=int,
+ valid_range=(1, 150),
+ )
+
+
+ logger.debug(f"{history_days} day selected for history")
+
+ if history_days is None:
+ logging.warning("No history range selected. Aborting.")
+ return
+
+ executions = []
+ hashes = []
+ working_dir = load_env("WORKING_DIR")
+ # Pull execution histories for each policy
+
+ policy_executions = ExecutionHistoryRecord.from_policies(
+ api, selected_policies, type_=type, history_days=history_days
+ )
+
+ logger.debug(f"Policy_executions is {policy_executions}")
+
+ executions.extend(policy_executions)
+ logger.debug(f"Executions contains {executions}")
+ if executions:
+ hashes = [Hash(sha256=row["sha256"], **row["data"]) for _, row in api.hash_query([record.sha256 for record in executions]).iterrows()
+ ]
+
+ if hashes:
+ unique_hashes = Hash.deduplicate(hashes)
+
+ needs_review, approved, unapproved = Hash.categorize_hashes(
+ hashes=unique_hashes
+ )
+
+ categories = {
+ "needs_review": needs_review,
+ "approved": approved,
+ "unapproved": unapproved,
+ }
+
+ for label, category in categories.items():
+ csv_path = f"{working_dir}\\Needs_Review\\Review_First\\{selected_policies[0].name}_{label}_executions.csv"
+ html_path = f"{working_dir}\\Needs_Review\\HTML\\{label}.html"
+
+ ExecutionHistoryRecord.enrich_with_hashes_and_export(
+ executions, category, f"{working_dir}\\Needs_Review\\Review_First", label=label
+ )
+ df = import_to_dataframe(csv_path)
+ formatHTML(df, html_path)
+
+def buildPathsandPublishers(split):
+ working_dir = load_env("WORKING_DIR")
+ df1 = pd.DataFrame()
+ df2 = pd.DataFrame()
+ all_approved_hashes = pd.DataFrame()
+ path1 = f"{working_dir}\\Approved\\approved_executions.csv"
+ path2 = f"{working_dir}\\Approved\\needs_review_executions.csv"
+
+ if os.path.exists(path1):
+ df1 = pd.read_csv(path1)
+ else:
+ logger.warning(f"File not found: {path1}")
+
+ if os.path.exists(path2):
+ df2 = pd.read_csv(path2)
+ else:
+ logger.warning(f"File not found: {path2}")
+
+ if df1.empty and df2.empty:
+ logger.warning("Both DataFrames are empty. Skipping sort.")
+ all_approved_hashes = pd.DataFrame()
+ logger.debug(all_approved_hashes.head)
+ else:
+ all_approved_hashes = pd.concat([df1, df2], ignore_index=True)
+ if "filename_exec" in all_approved_hashes.columns:
+ all_approved_hashes = all_approved_hashes.sort_values(by="filename_exec")
+ else:
+ logger.warning("Warning: 'filename_exec' column not found in concatenated DataFrame.")
+
+ if not all_approved_hashes.empty:
+ primary_path_exclusions = calculatePath(
+ all_approved_hashes,
+ split,
+ )
+ remaining_hashes = all_approved_hashes[
+ ~all_approved_hashes["sha256"].isin(primary_path_exclusions["sha256"])
+ ]
+ secondary_path_exclusions = calculatePath(
+ remaining_hashes, split
+ )
+ remaining_hashes = remaining_hashes[
+ ~remaining_hashes["sha256"].isin(secondary_path_exclusions["sha256"])
+ ]
+ dataframes = {
+ "primary_Paths": primary_path_exclusions,
+ "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")
+
+ if not all_approved_hashes.empty:
+ # Drop all not signed, only keep unique values
+ publist = all_approved_hashes[
+ all_approved_hashes["publisher_hash"] != "Not Signed"
+ ].drop_duplicates(subset=["publisher_hash"])
+ # Remove Bad publisher if somehow they made it this far
+ 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():
+ working_dir = load_env("WORKING_DIR")
+
+ df1 = pd.DataFrame()
+ df2 = pd.DataFrame()
+ approved_hashes = pd.DataFrame()
+ approved_publishers = pd.DataFrame()
+
+ hash = f"{working_dir}\\Approved\\hashes_to_add.csv"
+ path1 = f"{working_dir}\\Approved\\primary_Paths.csv"
+ path2 = f"{working_dir}\\Approved\\secondary_Paths.csv"
+ publishers = f"{working_dir}\\Approved\\publishers.csv"
+
+ if os.path.exists(hash):
+ approved_hashes = pd.read_csv(hash)
+
+ else:
+ logger.warning(f"File not found: {hash}")
+
+ if os.path.exists(path1):
+ df1 = pd.read_csv(path1)
+ else:
+ logger.warning(f"File not found: {path1}")
+
+ if os.path.exists(path2):
+ df2 = pd.read_csv(path2)
+ else:
+ logger.warning(f"File not found: {path2}")
+
+ if df1.empty and df2.empty:
+ logger.warning("Both DataFrames are empty. Skipping sort.")
+ approved_paths = pd.DataFrame()
+ else:
+ approved_paths = pd.concat([df1, df2], ignore_index=True)
+
+ if os.path.exists(publishers):
+ approved_publishers = pd.read_csv(publishers)
+
+ else:
+ logger.warning(f"File not found: {publishers}")
+
+ 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")
+
+def splitFilepathsGrouped(df, 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):
+ if not isinstance(path, (str, bytes, os.PathLike)):
+ return []
+ parts = os.path.normpath(path).split(os.sep)
+ parts = [p for p in parts if p] # Remove empty strings
+ return parts
+
+ # Diagnostic: log any non-string entries
+ non_string_entries = df[~df[col].apply(lambda x: isinstance(x, (str, bytes, os.PathLike)))]
+ if not non_string_entries.empty:
+ print(f"[WARNING] Non-string entries found in column '{col}':")
+ print(non_string_entries)
+
+ df = df.copy()
+ split_paths = df[col].apply(clean_split)
+
+ # Filter out paths with fewer than `min_files_for_path` components
+ df = df[split_paths.apply(lambda parts: len(parts) >= min_files_for_path)].copy()
+ split_paths = split_paths[df.index]
+
+ df["group_key"] = split_paths.apply(lambda parts: os.sep.join(parts[:path_exclusion_constant]))
+ grouped = df.groupby("group_key")
+ new_rows = []
+
+ for _, group_df in grouped:
+ paths = group_df[col].tolist()
+ split_parts = [clean_split(p) for p in paths]
+
+ def longest_common_prefix(paths):
+ if not paths:
+ return []
+ prefix = paths[0]
+ for path in paths[1:]:
+ prefix = [a for a, b in zip(prefix, path) if a == b]
+ if not prefix:
+ break
+ return prefix
+
+ common_prefix = longest_common_prefix(split_parts)
+ prefix_str = os.sep.join(common_prefix)
+
+ for i, parts in enumerate(split_parts):
+ filename = parts[-1]
+ middle = (
+ os.sep.join(parts[len(common_prefix):-1])
+ if len(parts) > len(common_prefix) + 1
+ else ""
+ )
+ row = group_df.iloc[i].copy()
+ row["longestcfp"] = prefix_str
+ row["middle"] = middle
+ row["filename_only"] = filename
+ row["file_extension"] = os.path.splitext(filename)[1].lower()
+ new_rows.append(row)
+
+ return pd.DataFrame(new_rows).drop(columns=["group_key"])
+
+def calculatePath(approved_hashes, split):
+ if split:
+ dfs_by_policy = [group for _, group in approved_hashes.groupby("policy")]
+ else:
+ dfs_by_policy = [approved_hashes]
+
+ badpathparts = load_env_json("BAD_PATH_PARTS", "[]")
+ min_files_for_path = get_protected_value("MIN_FILES_FOR_PATH", cast_type = int)
+
+ processed_dfs = []
+
+ for df in dfs_by_policy:
+ haslcp = splitFilepathsGrouped(df, "filename_exec")
+ haslcp = haslcp.drop_duplicates()
+
+ forbidden = regulator(badpathparts, True)
+ forbidden_lcfp = haslcp["longestcfp"].str.contains(forbidden, na=False)
+
+ logger.debug("Removing forbidden filepaths for path exceptions")
+ print(colorText("Removing forbidden filepaths for path exceptions", "green"))
+ lcp_not_forbidden = haslcp[~forbidden_lcfp].copy()
+
+ lcp_not_forbidden_review = lcp_not_forbidden[
+ [
+ "policyname",
+ "longestcfp",
+ "middle",
+ "filename_only",
+ "file_extension",
+ "sha256",
+ ]
+ ]
+
+ unique_sha_counts = (
+ lcp_not_forbidden_review.groupby("longestcfp")["sha256"].nunique().reset_index()
+ )
+ unique_sha_counts.columns = ["longestcfp", "unique_sha256_count"]
+
+ lcp_not_forbidden_review = lcp_not_forbidden_review.merge(
+ unique_sha_counts, on="longestcfp", how="left"
+ )
+ lcp_not_forbidden_review = lcp_not_forbidden_review[
+ lcp_not_forbidden_review["unique_sha256_count"] >= min_files_for_path
+ ]
+ processed_dfs.append(lcp_not_forbidden_review)
+
+ pathExclusions = pd.concat(processed_dfs, ignore_index=True)
+
+ return pathExclusions
diff --git a/Development/WIP/Sync Code/quietAgent.py b/Development/WIP/Sync Code/quietAgent.py
new file mode 100644
index 0000000..1a8d79b
--- /dev/null
+++ b/Development/WIP/Sync Code/quietAgent.py
@@ -0,0 +1,105 @@
+# 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 datetime
+import logging
+
+import dotenv
+import pandas as pd
+
+from flows.prepPolicy import selectPolicies
+from services.API import AirlockAPIWrapper
+from services.policyhandler import getPolicyInfo
+from utils.selector import Selector
+from utils.utils import colorText, load_env
+
+logger = logging.getLogger(__name__)
+
+
+dotenv.load_dotenv()
+
+
+def findQuietAgents(api: AirlockAPIWrapper):
+ working_dir = load_env("WORKING_DIR")
+ selected_policy = selectPolicies(api, False)
+ if selected_policy:
+ agents = api.agents_find_by_group(selected_policy[0].groupid)
+
+ history_days = Selector.select_value(
+ prompt="Enter how many days of history to pull (1โ150): ",
+ value_type=int,
+ valid_range=(1, 150),
+ )
+ required_quiet = Selector.select_value(
+ prompt="Enter how many days without an untrusted execution before these are considered ready for enforcement? (1โ150): ",
+ value_type=int,
+ valid_range=(1, 150),
+ )
+
+ policy_exec_history = getPolicyInfo(
+ api, selected_policy[0], [1, 2, 6, 7], history_days
+ )
+
+ if policy_exec_history.empty:
+ logging.info("No execution history found for the selected policy and time range.")
+ return
+
+ policy_exec_history.loc[:, "datetime"] = pd.to_datetime(
+ policy_exec_history["datetime"], format="%Y-%m-%dT%H:%M:%SZ", utc=True
+ )
+
+ now = datetime.datetime.now(datetime.timezone.utc)
+
+ policy_exec_history.loc[:, "days_ago"] = policy_exec_history["datetime"].apply(
+ lambda dt: (now - dt).days
+ )
+
+ hostname_counts = policy_exec_history["hostname"].value_counts()
+
+ agents.loc[:, "execution_count"] = agents["hostname"].map(hostname_counts).fillna(0).astype(int)
+
+ most_recent_exec = policy_exec_history.sort_values(by="days_ago").drop_duplicates(
+ subset="hostname", keep="first"
+ )
+
+ agents.loc[:, "days_since"] = agents["hostname"].map(
+ most_recent_exec.set_index("hostname")["days_ago"]
+ )
+
+ agents.loc[:, "required_quiet"] = required_quiet
+ agents.loc[:, "enforce_ready"] = agents["days_since"].apply(
+ lambda x: True if pd.isna(x) or x > required_quiet else False
+ )
+
+ agents = agents.sort_values(by=["execution_count", "hostname"], ascending=[True, True])
+
+ filename = f"{working_dir}\\{selected_policy[0].name}_agents_last_{history_days}_days.csv"
+ logging.debug(f"Saving CSV to {filename}")
+ print(colorText(f"Saving CSV to {filename}", "green"))
+ agents.to_csv(filename, index=False)
+
+ total_agents = len(agents)
+ ready_agents = agents["enforce_ready"].sum()
+ not_ready_agents = total_agents - ready_agents
+ ready_percentage = (ready_agents / total_agents) * 100
+
+ message = (
+ f"Total agents: {total_agents}\n"
+ f"Agents marked as 'enforce_ready': {ready_agents}\n"
+ f"Agents not ready: {not_ready_agents}\n"
+ f"Percentage ready for enforcement: {ready_percentage:.2f}%"
+ )
+ logger.info(message)
+ colorText(message, "green")
diff --git a/Development/WIP/Sync Code/security.py b/Development/WIP/Sync Code/security.py
new file mode 100644
index 0000000..841902e
--- /dev/null
+++ b/Development/WIP/Sync Code/security.py
@@ -0,0 +1,159 @@
+# 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 base64
+import logging
+import os
+import platform
+import re
+from getpass import getpass
+
+import keyring
+from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives.ciphers.aead import AESGCM
+from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
+from utils.utils import colorText
+from sys import exit
+
+# Constants
+KDF_ITERATIONS = 200_000
+SALT_SIZE = 16 # 128-bit Salt
+NONCE_SIZE = 12 # AES-GCM
+KEY_SIZE = 32 # AES-256
+
+
+def _derive_key(password: bytes, salt: bytes) -> bytes:
+ kdf = PBKDF2HMAC(
+ algorithm=hashes.SHA256(),
+ length=KEY_SIZE,
+ salt=salt,
+ iterations=KDF_ITERATIONS,
+ )
+ return kdf.derive(password)
+
+
+def configure_keyring_backend():
+ system = platform.system()
+ if system == "Windows":
+ import keyring.backends.Windows
+ keyring.set_keyring(keyring.backends.Windows.WinVaultKeyring())
+ elif system == "Linux":
+ import keyring.backends.kwallet
+ keyring.set_keyring(keyring.backends.kwallet.DBusKeyring())
+ else:
+ raise EnvironmentError(f"Unsupported OS: {system}")
+
+
+def store_api_key(service: str, username: str, api_key: str, password: str):
+ configure_keyring_backend()
+ salt = os.urandom(SALT_SIZE)
+ key = _derive_key(password.encode(), salt)
+ aesgcm = AESGCM(key)
+ nonce = os.urandom(NONCE_SIZE)
+ ct = aesgcm.encrypt(nonce, api_key.encode(), associated_data=None)
+ blob = salt + nonce + ct
+ b64 = base64.b64encode(blob).decode()
+ keyring.set_password(service, username, b64)
+
+
+def retrieve_api_key(service: str, username: str, password: str) -> str:
+ configure_keyring_backend()
+ b64 = keyring.get_password(service, username)
+ if b64 is None:
+ raise ValueError("No stored secret for this service/username.")
+ blob = base64.b64decode(b64)
+ salt = blob[:SALT_SIZE]
+ nonce = blob[SALT_SIZE:SALT_SIZE + NONCE_SIZE]
+ ct = blob[SALT_SIZE + NONCE_SIZE:]
+ key = _derive_key(password.encode(), salt)
+ aesgcm = AESGCM(key)
+ pt = aesgcm.decrypt(nonce, ct, associated_data=None)
+ return pt.decode()
+
+
+def api_key_exists(service: str, username: str) -> bool:
+ configure_keyring_backend()
+ return keyring.get_password(service, username) is not None
+
+
+def check_password_complexity(password: str) -> bool:
+ if len(password) < 12:
+ return False
+ if not re.search(r"[A-Z]", password):
+ return False
+ if not re.search(r"[a-z]", password):
+ return False
+ if not re.search(r"[0-9]", password):
+ return False
+ if not re.search(r"[^A-Za-z0-9]", password):
+ return False
+ return True
+
+
+def getAPI(USERNAME, SERVICE_NAME):
+ logging.debug(
+ f"Checking for stored API key for user '{USERNAME}' in service '{SERVICE_NAME}'..."
+ )
+
+ if api_key_exists(SERVICE_NAME, USERNAME):
+ for attempt in range(1, 4):
+ password = getpass(f"Attempt {attempt}/3 - Enter password to unlock your API key: ")
+ try:
+ apikey = retrieve_api_key(SERVICE_NAME, USERNAME, password)
+ logging.debug("API key successfully retrieved.")
+ return apikey
+ except Exception as e:
+ logging.warning(f"Attempt {attempt} failed: {str(e)}")
+ logging.error("Failed to retrieve API key after 3 incorrect attempts.")
+ print(colorText("โ Authentication failed. Exiting.","red"))
+ exit(1) # Exit cleanly without traceback
+ raise ValueError("Failed to retrieve API key after 3 incorrect attempts.")
+ else:
+ logging.warning(f"No API key found for user '{USERNAME}' in service '{SERVICE_NAME}'.")
+ api_key = getpass(f"๐๏ธ No API key found. Please enter your API key for '{SERVICE_NAME}': ").strip()
+ print("Please exit and relaunch program after saving your credential to avoid errors")
+
+ while True:
+ password = getpass("๐ Create a password to encrypt your API key: ")
+ confirm_password = getpass("๐ Confirm your password: ")
+
+ if password != confirm_password:
+ logging.warning("โ Passwords do not match. Try again.")
+ continue
+
+ if check_password_complexity(password):
+ try:
+ store_api_key(SERVICE_NAME, USERNAME, api_key, password)
+ logging.info("API key stored securely.")
+ break
+ except Exception as e:
+ logging.error(f"Failed to store API key: {e}")
+ break
+ else:
+ logging.warning("โ Password does not meet complexity requirements. Try again.")
+
+
+class APIKeyManager:
+ _api_key = None
+
+ @classmethod
+ def load(cls, service: str, username: str, password: str):
+ cls._api_key = retrieve_api_key(service, username, password)
+
+ @classmethod
+ def get(cls) -> str:
+ if cls._api_key is None:
+ raise ValueError("API key not loaded. Call APIKeyManager.load() first.")
+ return cls._api_key
\ No newline at end of file
diff --git a/Development/WIP/Sync Code/selector.py b/Development/WIP/Sync Code/selector.py
new file mode 100644
index 0000000..9385eb8
--- /dev/null
+++ b/Development/WIP/Sync Code/selector.py
@@ -0,0 +1,171 @@
+# 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
+from typing import Any, Callable, List, Optional, Union
+
+from utils.utils import get_sanitized_input
+
+logger = logging.getLogger(__name__)
+
+
+class Selector:
+ @staticmethod
+ def _display_choices(
+ items: List[Any],
+ label_func: Callable[[Any], str],
+ num_columns: int = 3,
+ header: str = "Available Choices:"
+ ) -> None:
+ sorted_items = sorted(items, key=lambda item: label_func(item).lower())
+ rows = (len(sorted_items) + num_columns - 1) // num_columns
+ print(f"\n{header}")
+ for row in range(rows):
+ line = ""
+ for col in range(num_columns):
+ idx = row + col * rows
+ if idx < len(sorted_items):
+ label = label_func(sorted_items[idx])
+ line += f"{idx + 1}: {label:<30}"
+ print(line)
+
+ @staticmethod
+ def _select_from_list(
+ items: List[Any],
+ label_func: Callable[[Any], str],
+ allow_multiple: bool = False,
+ prompt_each: bool = False,
+ header: str = "Available Choices:"
+ ) -> Union[Optional[Any], List[Any]]:
+ if not items:
+ logger.warning("No items available for selection.")
+ return None
+
+ Selector._display_choices(items, label_func, header=header)
+ sorted_items = sorted(items, key=lambda item: label_func(item).lower())
+ selected = []
+
+ if allow_multiple:
+ while True:
+ choice = get_sanitized_input("Select an item by number (or Q to finish): ").strip().lower()
+ if choice == "q":
+ break
+ try:
+ index = int(choice)
+ if 1 <= index <= len(sorted_items):
+ item = sorted_items[index - 1]
+ if item not in selected:
+ selected.append(item)
+ if prompt_each:
+ logger.info(f"Selected: {label_func(item)}")
+ else:
+ logger.warning("Item already selected.")
+ else:
+ logger.warning("Selection out of range. Try again.")
+ except ValueError:
+ logger.warning("Invalid input. Enter a number or 'Q' to quit.")
+ return selected if selected else None
+ else:
+ try:
+ choice = int(get_sanitized_input("Select one item by number: "))
+ if 1 <= choice <= len(sorted_items):
+ selected_item = sorted_items[choice - 1]
+ logger.info(f"Selected: {label_func(selected_item)}")
+ return selected_item
+ else:
+ logger.warning("Selection out of range.")
+ except ValueError:
+ logger.warning("Invalid input.")
+ return None
+
+ @staticmethod
+ def select_objects(
+ objects: List[Any],
+ allow_multiple: bool = False,
+ prompt_each: bool = False
+ ) -> Union[Optional[Any], List[Any]]:
+ return Selector._select_from_list(
+ objects,
+ label_func=lambda obj: getattr(obj, "name", str(obj)),
+ allow_multiple=allow_multiple,
+ prompt_each=prompt_each,
+ header="Available Objects:"
+ )
+
+ @staticmethod
+ def select_string(
+ options: List[str],
+ allow_multiple: bool = False,
+ prompt_each: bool = False
+ ) -> Union[Optional[str], List[str]]:
+ return Selector._select_from_list(
+ options,
+ label_func=str,
+ allow_multiple=allow_multiple,
+ prompt_each=prompt_each,
+ header="Available Options:"
+ )
+
+ @staticmethod
+ def select_int(
+ options: List[int],
+ allow_multiple: bool = False,
+ prompt_each: bool = False
+ ) -> Union[Optional[int], List[int]]:
+ return Selector._select_from_list(
+ options,
+ label_func=lambda x: str(x),
+ allow_multiple=allow_multiple,
+ prompt_each=prompt_each,
+ header="Available Integers:"
+ )
+
+ @staticmethod
+ def select_value(
+ prompt: str,
+ value_type: type = int,
+ valid_range: Optional[tuple] = None,
+ allow_quit: bool = False
+ ) -> Optional[Any]:
+ while True:
+ user_input = get_sanitized_input(prompt).strip().lower()
+ if allow_quit and user_input == "q":
+ logger.info("User opted to quit value selection.")
+ return None
+ try:
+ value = value_type(user_input)
+ if valid_range:
+ min_val, max_val = valid_range
+ if not (min_val <= value <= max_val):
+ logger.warning(f"Value out of range ({min_val}โ{max_val}).")
+ continue
+ logger.info(f"User selected value: {value}")
+ return value
+ except ValueError:
+ logger.warning(f"Invalid input. Expected a {value_type.__name__}.")
+
+ @staticmethod
+ def confirm(prompt: str = "Are you sure? (Y/N): ") -> bool:
+ while True:
+ response = get_sanitized_input(prompt).strip().lower()
+ if response in ["y", "yes"]:
+ logger.info("User confirmed action.")
+ return True
+ elif response in ["n", "no"]:
+ logger.info("User declined action.")
+ return False
+ else:
+ logger.warning("Invalid confirmation input. Expected 'Y' or 'N'.")
\ No newline at end of file
diff --git a/Development/WIP/Sync Code/setup.py b/Development/WIP/Sync Code/setup.py
new file mode 100644
index 0000000..d2d70ef
--- /dev/null
+++ b/Development/WIP/Sync Code/setup.py
@@ -0,0 +1,175 @@
+# 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 json
+import logging
+import logging.handlers
+import os
+import platform
+import sys
+from pathlib import Path
+
+from dotenv import load_dotenv, set_key
+
+from utils.configmanager import PROTECTED_KEYS, load_protected_config
+
+
+def get_base_directory() -> Path:
+ system = platform.system()
+ home = Path.home()
+ if system == 'Windows':
+ return Path(os.getenv('APPDATA', home / 'AppData' / 'Roaming')) / "AirlockTools"
+ elif system == 'Darwin':
+ return home / 'Library' / 'Application Support' / "AirlockTools"
+ 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()
+ logger.setLevel(getattr(logging, log_level.upper(), logging.DEBUG))
+
+ # ๐ง Clear existing handlers
+ for handler in logger.handlers[:]:
+ logger.removeHandler(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)
+
+ 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.")
+
+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()
+ with open(config_path, "r") as f:
+ return json.load(f)
+ except FileNotFoundError:
+ logging.warning("โ ๏ธ system_config.json not found. Using built-in defaults.")
+ return {
+ "APPNAME": "AirlockTools",
+ "LOG_LEVEL": "DEBUG",
+ "PATH_EXCLUSION_CONST": 4,
+ "MIN_FILES_FOR_PATH": 4,
+ "VT_THREAT_TOLERANCE": 4,
+ "POLICY_MAP_ENF_AUD": {
+ "enforced_id": "audit_id"
+ }
+ }
+
+def load_user_config(config_dir: Path) -> dict:
+ user_config_path = config_dir / "user_config.json"
+ if not user_config_path.exists():
+ default_user_config = {
+ "URL": "",
+ "LOG_LEVEL": "INFO"
+ }
+ with open(user_config_path, "w") as f:
+ json.dump(default_user_config, f, indent=4)
+ logging.debug(f"Created user config at {user_config_path}")
+ with open(user_config_path, "r") as f:
+ return json.load(f)
+
+def write_config_to_env(config: dict, env_path: Path):
+ for key, value in config.items():
+ if key in PROTECTED_KEYS:
+ continue # Skip protected keys
+ try:
+ serialized = json.dumps(value) if isinstance(value, (list, dict)) else str(value)
+ set_key(env_path, key, serialized)
+ except Exception as e:
+ logging.warning(f"Failed to write {key} to .env: {e}")
+
+def setup() -> Path:
+ base_dir = get_base_directory()
+ dirs = {
+ 'config': base_dir / 'config',
+ 'cache': base_dir / 'cache',
+ 'logs': base_dir / 'logs',
+ }
+
+ for name, path in dirs.items():
+ path.mkdir(parents=True, exist_ok=True)
+ logging.debug(f"{name.capitalize()} directory ensured at: {path}")
+
+ system_config = load_system_config()
+ configure_logging(dirs['logs'], system_config.get("LOG_LEVEL", "DEBUG"))
+
+ env_path = base_dir / ".env"
+ if not env_path.exists():
+ env_path.touch()
+ load_dotenv(dotenv_path=env_path, override=True)
+
+ working_dir = Path(os.getenv("WORKING_DIR") or (base_dir / "data"))
+ working_dir.mkdir(parents=True, exist_ok=True)
+ set_key(env_path, "WORKING_DIR", str(working_dir))
+ os.environ["WORKING_DIR"] = str(working_dir)
+ logging.debug(f"Working directory set to: {working_dir}")
+
+ folders_structure = {
+ "Approved": [],
+ "Needs_Review": ["Review_First", "Review_Second", "HTML"],
+ "Preflight": ["HTML"],
+ "Archived": []
+ }
+
+ for folder_name, subfolders in folders_structure.items():
+ folder_path = working_dir / folder_name
+ folder_path.mkdir(parents=True, exist_ok=True)
+ logging.debug(f"'{folder_name}' folder ensured at: {folder_path}")
+ for subfolder in subfolders:
+ subfolder_path = folder_path / subfolder
+ subfolder_path.mkdir(parents=True, exist_ok=True)
+ logging.debug(f" โโ '{subfolder}' subfolder created at: {subfolder_path}")
+
+ user_config = load_user_config(dirs['config'])
+ merged_config = {**system_config, **user_config}
+
+ protected_config = load_protected_config()
+ merged_config.update(protected_config)
+
+ # โ
URL resolution order: system_config โ .env โ user prompt
+ url = system_config.get("URL")
+ if not url:
+ url = os.getenv("URL")
+ if not url:
+ url = input("๐ Enter the service URL (e.g., https://example.com/api): ").strip()
+ merged_config["URL"] = url
+ set_key(env_path, "URL", url)
+ os.environ["URL"] = url
+ logging.debug(f"Service URL set to: {url}")
+
+ write_config_to_env(merged_config, env_path)
+
+ return working_dir
\ No newline at end of file
diff --git a/Development/WIP/Sync Code/utils.py b/Development/WIP/Sync Code/utils.py
new file mode 100644
index 0000000..6803411
--- /dev/null
+++ b/Development/WIP/Sync Code/utils.py
@@ -0,0 +1,708 @@
+# 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
+
+from utils.configmanager import load_env
+
+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 displayIntro():
+ print(
+ colorText(
+ r"""
+ โโโ
+ โโโโ โโโโโโโโโ
+ โโโโโโโโโโโโโ โโโโโโโโโโโโโโโ
+ โโโโโโโโโโโโโโโโโโโโโ โโโโโโโโโโโโโโโโโโโโโ
+ โโโโโโโโโโโโโโโโโโโ โโโโโโโโโโโโโโโโโโโโโโโ
+ โโโโโโโโโโโโโโโโโโโ โโโโโโโโโโโโโโโโโโโโโโ
+ โโโโโโโโโโโโโโโโโโโโโ โโโโโโโโโโโโโโโโโโโโโโโ
+ โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
+ โโโโโโโโโ โโ โโ โโโโโโโโโ
+ โโโโโโโโโ โโ โโโ โ โโโโโโโโโ
+ โโโโโโโโโ โโ โโโโ โโโโโ โโโโโโโโโโโโโ
+ โโโโโโโโโ โโ โโโโโโ โโโโโโโโโโโโโ
+ โโโโโโโโ โโ โโโโโโโ โโโโโโโโโโโโโ
+ โโโโโโโ โโ โโโ โโโโโโ โโโโโโโโโโโโ
+ โโโโโโ โโ โโโโ โโโโโ โโโโโโโโโโโ
+ โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
+ โโโโโโโโโโโโโโโโโโโโโ โโโโโโโโโโโโโโโโโโ
+ โโโโโโโโโโโโโโโโโโโ โโโโโโโโโโโโโโโโ
+ โโโโโโโโโโโโโโโ โโโโโโโโโโโโโ
+ โโโโโโโโโโ โโโโโโโโโโโ
+ โโโโโโโโ
+ โโโโ
+""",
+ "yellow",
+ )
+ )
+ print(
+ colorText(
+ r"""
+ _____ .__ .__ __ ___________ .__
+ / _ \ |__|______| | ____ ____ | | __ \__ ___/___ ____ | | ______
+ / /_\ \| \_ __ \ | / _ \_/ ___\| |/ / | | / _ \ / _ \| | / ___/
+/ | \ || | \/ |_( <_> ) \___| < | |( <_> | <_> ) |__\___ \
+\____|__ /__||__| |____/\____/ \___ >__|_ \ |____| \____/ \____/|____/____ >
+ \/ \/ \/ \/
+""",
+ "cyan",
+ )
+ )
+ print(
+ colorText(
+ "=================================================================================",
+ "cyan",
+ )
+ )
+ print(
+ colorText(
+ "======================== Welcome to the Airlock API Tool ========================",
+ "cyan",
+ )
+ )
+ print(
+ colorText(
+ "=================================================================================",
+ "cyan",
+ )
+ )
+
+
+def printEnforceChecklist(selected_policies, destination_policy, destination_allowlist):
+ working_dir = load_env("WORKING_DIR")
+ 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",
+ )
+ )
+ if not selected_policies:
+ print(colorText(" [โ] No policies have been chosen", "red"))
+ else:
+ print(colorText("The following policies have been choosen:", "green"))
+ for policy in selected_policies:
+ print(colorText(f" [โ] {policy.name}", "green"))
+
+ print(colorText("2. Choose the destination policy and allowlist", "cyan"))
+
+ if not destination_policy:
+ print(colorText(" [โ] No destination policy has been chosen", "red"))
+ elif destination_policy:
+ print(colorText(f" [โ] {destination_policy[0].name} has been selected as the destination policy", "green"))
+
+
+
+ if not destination_allowlist:
+ print(colorText(" [โ] No allowlist has been chosen", "red"))
+ elif destination_allowlist:
+ print(
+ colorText(
+ f" [โ] {destination_allowlist[0].name} has been selected as allowlist",
+ "green",
+ )
+ )
+
+
+
+ print(
+ colorText(
+ "3. Pull and stage event history, combine the histories, add hash info, then categorize the hashes",
+ "cyan",
+ )
+ )
+ if not selected_policies:
+ print(colorText(" [โ] No policies have been chosen", "red"))
+ else:
+ if os.path.exists(f"{working_dir}\\Needs_Review\\Review_First\\approved_executions.csv"):
+ print(colorText(" [โ] Data has been fetched", "green"))
+ else:
+ print(colorText(" [โ] Data has not been fetched", "red"))
+
+ print(colorText("4. Manually review the files:", "cyan"))
+ 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, 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",
+ )
+ )
+
+ if os.path.exists(f"{working_dir}\\Approved\\approved_executions.csv"):
+ print(colorText(" [โ] Reviewed hashes have been loaded", "green"))
+ else:
+ print(colorText(" [โ] Reviewed hashes have not been loaded", "red"))
+
+ if os.path.exists(
+ f"{working_dir}\\Needs_Review\\Review_Second\\primary_Paths.csv",
+ ):
+ print(colorText(" [โ] Path review list created", "green"))
+ else:
+ print(colorText(" [โ] Path review list has not been created", "red"))
+
+ 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. The secondary list can be not added at all if nothing is useful",
+ "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"))
+
+ if os.path.exists(
+ f"{working_dir}\\Approved\\primary_Paths.csv",
+ ):
+ print(colorText(" [โ] Reviewed path list detected", "green"))
+ else:
+ print(colorText(" [โ] Path review list has not been detected", "red"))
+
+ if os.path.exists(f"{working_dir}\\Preflight\\approved_paths.csv") and os.path.exists(
+ f"{working_dir}\\Preflight\\approved_hashes.csv"
+ ):
+ print(colorText(" [โ] Preflight Path Exclusion List has been generated", "green"))
+ else:
+ print(colorText(" [โ] Preflight Path Exclusion List has not been generated", "red"))
+
+ 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("F. ๐ - Open Working Directory", "cyan"))
+ print(colorText("Q. ๐ - Quit", "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"""
+
+ """
+
+ 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}")
+
+
diff --git a/Development/WIP/dynamicAsyncTaskQueue.py b/Development/WIP/dynamicAsyncTaskQueue.py
new file mode 100644
index 0000000..4f847e9
--- /dev/null
+++ b/Development/WIP/dynamicAsyncTaskQueue.py
@@ -0,0 +1,135 @@
+import asyncio
+import logging
+from typing import Callable, Any
+import psutil # For CPU and memory monitoring
+
+# Set up the logger
+logger = logging.getLogger("AsyncTaskQueue")
+logger.setLevel(logging.INFO)
+handler = logging.StreamHandler()
+formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+handler.setFormatter(formatter)
+logger.addHandler(handler)
+
+class AsyncTaskQueue:
+ def __init__(
+ self,
+ min_workers: int = 1,
+ max_workers: int = 10,
+ cpu_threshold: float = 90.0,
+ memory_threshold: float = 90.0,
+ ):
+ self.queue = asyncio.Queue()
+ self.min_workers = min_workers
+ self.max_workers = max_workers
+ self.cpu_threshold = cpu_threshold
+ self.memory_threshold = memory_threshold
+ self.workers = []
+ self._stop_event = asyncio.Event()
+ self._scale_lock = asyncio.Lock() # Prevent race conditions
+
+ async def _scale_up(self):
+ """Add a worker if under max limit and resources allow."""
+ async with self._scale_lock:
+ if (
+ len(self.workers) < self.max_workers
+ and psutil.cpu_percent() < self.cpu_threshold
+ and psutil.virtual_memory().percent < self.memory_threshold
+ ):
+ worker = asyncio.create_task(self.worker_loop(f"Worker-{len(self.workers) + 1}"))
+ self.workers.append(worker)
+ logger.info(f"Scaled up. Workers: {len(self.workers)}")
+
+ async def _scale_down(self):
+ """Remove a worker if above min limit."""
+ async with self._scale_lock:
+ if len(self.workers) > self.min_workers:
+ worker = self.workers.pop()
+ worker.cancel()
+ logger.info(f"Scaled down. Workers: {len(self.workers)}")
+
+ async def _monitor_resources(self):
+ """Monitor CPU and memory usage, and adjust workers."""
+ while not self._stop_event.is_set():
+ cpu_usage = psutil.cpu_percent()
+ memory_usage = psutil.virtual_memory().percent
+
+ if (
+ cpu_usage > self.cpu_threshold
+ or memory_usage > self.memory_threshold
+ ):
+ await self._scale_down()
+ elif (
+ len(self.workers) < self.max_workers
+ and self.queue.qsize() > 2 # Only scale up if there's work
+ ):
+ await self._scale_up()
+
+ await asyncio.sleep(2) # Check every 2 seconds
+
+ async def start_workers(self):
+ """Start initial workers and the resource monitor."""
+ for i in range(self.min_workers):
+ worker = asyncio.create_task(self.worker_loop(f"Worker-{i+1}"))
+ self.workers.append(worker)
+ # Start the resource monitor
+ asyncio.create_task(self._monitor_resources())
+ logger.info(f"Started {self.min_workers} workers and resource monitor.")
+
+ async def stop_workers(self):
+ """Stop all workers and the resource monitor."""
+ logger.info("Stopping workers...")
+ self._stop_event.set()
+ await self.queue.join() # Wait for all tasks to complete
+ for worker in self.workers:
+ worker.cancel()
+ await asyncio.gather(*self.workers, return_exceptions=True)
+ logger.info("All workers stopped.")
+
+ async def worker_loop(self, name: str):
+ """Worker loop: Process tasks from the queue."""
+ logger.info(f"{name} started.")
+ while not self._stop_event.is_set():
+ try:
+ task = await self.queue.get()
+ logger.info(f"{name} processing: {task['name']}")
+ await task['func'](*task['args'])
+ except Exception as e:
+ logger.error(f"Error in {name}: {e}", exc_info=True)
+ finally:
+ self.queue.task_done()
+ logger.info(f"{name} exited.")
+
+ async def enqueue(self, name: str, func: Callable, *args: Any):
+ """Add a task to the queue."""
+ logger.info(f"Enqueuing task: {name}")
+ await self.queue.put({'name': name, 'func': func, 'args': args})
+
+async def run_sync_task_in_thread(func: Callable, *args: Any):
+ """Run a synchronous function in a separate thread."""
+ await asyncio.to_thread(func, *args)
+"""import asyncio
+
+async def example_async_task(name: str, duration: int):
+ print(f"{name} started, will sleep for {duration} seconds")
+ await asyncio.sleep(duration)
+ print(f"{name} finished")
+
+async def main():
+ queue = AsyncTaskQueue(
+ min_workers=2,
+ max_workers=10,
+ cpu_threshold=90.0,
+ memory_threshold=90.0,
+ )
+ await queue.start_workers()
+
+ # Enqueue tasks
+ for i in range(20):
+ await queue.enqueue(f"Task{i}", example_async_task, f"Task{i}", 1)
+
+ await asyncio.sleep(10) # Let tasks run
+ await queue.stop_workers()
+
+asyncio.run(main())
+"""
\ No newline at end of file
diff --git a/Development/WIP/gu2i.py b/Development/WIP/gu2i.py
new file mode 100644
index 0000000..758690b
--- /dev/null
+++ b/Development/WIP/gu2i.py
@@ -0,0 +1,179 @@
+from textual.app import App, ComposeResult
+from textual.screen import Screen
+from textual.widgets import Header, Tabs, Tab, Static, Footer, DirectoryTree, Button
+from textual.containers import Horizontal
+import logging
+import dotenv
+import os
+
+from Development_Stubs.WIP.localApproval import moveToLocalApproval
+from flows.prepPolicy import (
+ buildPathsandPublishers,
+ buildPreflights,
+ selectAllowlists,
+ selectPolicies,
+ sortHashes,
+)
+from flows.quietAgent import findQuietAgents
+from services.agenthandler import findAgents
+from services.API import AirlockAPIWrapper
+import services.policyhandler as policyh
+from utils.utils import (
+ areYouSure,
+ colorText,
+ displayIntro,
+ load_env,
+ open_directory,
+ printEnforceChecklist,
+)
+from screens.agent_results import AgentResultsScreen
+logger = logging.getLogger(__name__)
+dotenv.load_dotenv()
+
+ASCII_ART = displayIntro()
+
+class BaseScreen(Screen):
+ def compose(self) -> ComposeResult:
+ yield Header()
+ yield Static("This is a dummy screen.", id="content")
+ yield Footer()
+
+class LandingScreen(BaseScreen):
+ def compose(self) -> ComposeResult:
+ yield Header()
+ yield Static(ASCII_ART, id="ascii-art", markup=False)
+ yield Horizontal(
+ Button("Get Started", id="get-started-button"),
+ Button("Settings", id="settings-button"),
+ )
+ yield Footer()
+
+ def on_button_pressed(self, event: Button.Pressed) -> None:
+ if event.button.id == "get-started-button":
+ # Go back to the previous screen (MainMenuScreen)
+ self.app.pop_screen()
+
+class LocalApprovalScreen(BaseScreen):
+ def compose(self) -> ComposeResult:
+ yield Header()
+
+ yield Footer()
+
+class OTPScreen(BaseScreen):
+ def compose(self) -> ComposeResult:
+ yield Header()
+ yield Static("OTP: Enter your one-time password here.", id="content")
+ yield Footer()
+
+class SearchScreen(BaseScreen):
+ def compose(self) -> ComposeResult:
+ yield Header()
+ yield Static("Search: Find items in the system.", id="content")
+ yield Horizontal(
+ Button("Search for Agents", id="searchAgent-button"),
+ )
+ yield Footer()
+
+ def on_button_pressed(self, event: Button.Pressed) -> None:
+ if event.button.id == "searchAgent-button":
+ self.app.push_screen(AgentResultsScreen(self.app.api, self.app.working_dir))
+
+class QuietHostsScreen(BaseScreen):
+ def compose(self) -> ComposeResult:
+ yield Header()
+ yield Static("Quiet Hosts: Manage hosts that are not responding.", id="content")
+ yield Footer()
+
+class PolicyPrepScreen(BaseScreen):
+ def compose(self) -> ComposeResult:
+ yield Header()
+ yield Static("Policy Prep: Prepare policies for deployment.", id="content")
+ yield Footer()
+
+class UpdatePoliciesScreen(BaseScreen):
+ def compose(self) -> ComposeResult:
+ yield Header()
+ yield Static("Update Policies: Update existing policies.", id="content")
+ yield Footer()
+
+class SettingsScreen(BaseScreen):
+ def compose(self) -> ComposeResult:
+ yield Header()
+ yield Static("Settings: Configure application settings.", id="content")
+ yield Footer()
+
+class DirectoryTreeScreen(Screen):
+ def compose(self) -> ComposeResult:
+ yield Header()
+ yield DirectoryTree("./")
+ yield Footer()
+
+class MainMenuScreen(Screen):
+ def compose(self) -> ComposeResult:
+ yield Header()
+ yield Tabs(
+ Tab("Home", id="home_screen"),
+ Tab("Local Approval", id="local_approval"),
+ Tab("OTP", id="otp"),
+ Tab("Search", id="search"),
+ Tab("Quiet Hosts", id="quiet"),
+ Tab("Policy Prep", id="policy"),
+ Tab("Update Policies", id="update"),
+ id="tabs"
+ )
+ yield Static("Select a tab to begin.", id="content")
+ yield Footer()
+
+ def on_tabs_tab_activated(self, event: Tabs.TabActivated) -> None:
+ match event.tab.id:
+ case "home_screen":
+ self.app.push_screen(LandingScreen())
+ case "local_approval":
+ self.app.push_screen(LocalApprovalScreen())
+ case "otp":
+ self.app.push_screen(OTPScreen())
+ case "search":
+ self.app.push_screen(SearchScreen())
+ case "quiet":
+ self.app.push_screen(QuietHostsScreen())
+ case "policy":
+ self.app.push_screen(PolicyPrepScreen())
+ case "update":
+ self.app.push_screen(UpdatePoliciesScreen())
+ case "settings":
+ self.app.push_screen(SettingsScreen())
+
+class AirlockTools(App):
+ BINDINGS = [
+ ("q", "quit", "Quit"),
+ ("d", "open_dir", "Open Directory"),
+ ("b", "back", "Go Back"),
+ ]
+
+ def __init__(self, api: AirlockAPIWrapper, working_dir: str):
+ super().__init__()
+ self.api = api
+ self.working_dir = working_dir
+
+ def on_mount(self) -> None:
+ self.push_screen(MainMenuScreen())
+
+ def action_quit(self) -> None:
+ self.exit()
+
+ def action_open_dir(self) -> None:
+ self.push_screen(DirectoryTreeScreen())
+
+ def action_back(self) -> None:
+ if len(self.screen_stack) > 2:
+ self.pop_screen()
+ else:
+ self.bell()
+
+
+if __name__ == "__main__":
+ api = AirlockAPIWrapper(base_url=str(os.getenv("URL")),api_key = "b82ab7e2b12430fff0cbf0e798f79e16c1aea2c37ffee5dc884de7ba11d3a3a4")
+ working_dir = "./" # Or a dynamic path
+ AirlockTools(api, working_dir).run()
+
+
diff --git a/services/security.py b/services/security.py
index 093b58d..bbf2bf4 100644
--- a/services/security.py
+++ b/services/security.py
@@ -32,6 +32,8 @@ SALT_SIZE = 16 # 128-bit Salt
NONCE_SIZE = 12 # AES-GCM
KEY_SIZE = 32 # AES-256
+logger = logging.getLogger(__name__)
+
def _derive_key(password: bytes, salt: bytes) -> bytes:
kdf = PBKDF2HMAC(
@@ -66,6 +68,20 @@ def store_api_key(service: str, username: str, api_key: str, password: str):
b64 = base64.b64encode(blob).decode()
keyring.set_password(service, username, b64)
+
+ logger.debug(f"API key for service '{service}' and user '{username}' stored successfully.")
+
+ print("\nโ
API key stored securely.")
+ print("The program will now exit. Press Enter to continue...")
+
+ try:
+ _ = input()
+ except Exception:
+ pass
+
+ _ = None
+ sys.exit(0)
+
def retrieve_api_key(service: str, username: str, password: str) -> str:
configure_keyring_backend()
@@ -80,20 +96,6 @@ def retrieve_api_key(service: str, username: str, password: str) -> str:
aesgcm = AESGCM(key)
pt = aesgcm.decrypt(nonce, ct, associated_data=None)
return pt.decode()
-
- logger.debug(f"API key for service '{service}' and user '{username}' stored successfully.")
-
- print("\nโ
API key stored securely.")
- print("The program will now exit. Press Enter to continue...")
-
- try:
- _ = input()
- except Exception:
- pass
-
- _ = None
- sys.exit(0)
-
def api_key_exists(service: str, username: str) -> bool: