diff --git a/AirlockTools_Client.py b/AirlockTools_Client.py index db572f7..8f55d97 100644 --- a/AirlockTools_Client.py +++ b/AirlockTools_Client.py @@ -27,10 +27,10 @@ import os import dotenv import urllib3 -import utils.menus as menus from services.API import AirlockAPIWrapper from services.security import getAPI from utils.setup import get_base_directory, setup +from utils.tui import run_menu from utils.utils import irtang urllib3.disable_warnings( @@ -40,7 +40,6 @@ urllib3.disable_warnings( def main(): irtang() #Determine working directory, setup directory, configure logging, sent env, get API and URL if not already stored - setup() base_dir = get_base_directory() logger = logging.getLogger(__name__) @@ -72,7 +71,7 @@ def main(): api_key=api_key, ) - menus.menu_main(api) + run_menu(api) if __name__ == "__main__": diff --git a/README.md b/README.md index 6207bdc..626650e 100644 --- a/README.md +++ b/README.md @@ -1,20 +1,59 @@ -[![AGPL License](https://img.shields.io/badge/license-AGPL-blue.svg)](http://www.gnu.org/licenses/agpl-3.0) -# Airlock Digital Local Approval +# 🛡️ Airlock Tools -Python based Carbon Black App Control feature implementation for Airlock -## Features +Python toolkit for secure, auditable, and automated airlock agent and policy management. Designed for enterprise environments, it supports advanced policy workflows, device tracking, and terminal-based interaction. -- "Local Approval Initialization" -This programmatically scans devices in audit mode within Airlock and subsequently adds the identified blocks to a user-specified whitelist. - -- KDE Wallet must be running on Linux (`kwalletd`) +--- -## License +## 🚀 Features +- 🔍 **Fuzzy Device Search** + Quickly locate devices using partial or approximate matches. + +- 📦 **Batch Move Devices** + Move multiple devices between groups or policies easily. + +- 🔄 **Toggle Enforcement/Audit Policies** + Seamlessly switch devices between enforcement and audit modes. + +- 🕵️‍♂️ **Device History Search** + Track agent executions. + +- 🧰 **Prepare Policies for Enforcement** + Validate and stage policies before pushing them to enforcement. + +- 💤 **Find Quiet Hosts** + Identify devices ready for enforcement. + +- 🎛️ **TUI** + Navigate with arrow keys and F-key shortcuts using a custom ANSI-colored terminal UI. + +--- + +## 🧭 Roadmap + +- ⚙️ **Rust-based Async API Calls** + Improve performance and concurrency with a Rust-powered backend. + +- ✅ **Carbon Black-style Local Approval** + Enable local user approvals for policy exceptions and enforcement actions. + +- 📊 **Audit Logging & Export** + Add detailed logging and export capabilities for compliance and analysis. + +--- + +## 🧑‍💻 Requirements + +TBD + +--- + +## 📜 License + **AirlockTools** is licensed under the **GNU Affero General Public License v3.0**. You may copy, distribute, and modify the software under the terms of the AGPL-3.0 license. -See the [LICENSE](LICENSE.md) file for full details, or visit -[https://www.gnu.org/license/agpl-3.0.html](https://www.gnu.org/license/agpl-3.0.html) \ No newline at end of file +See the [LICENSE](LICENSE.md) file for full details, or visit +[https://www.gnu.org/license/agpl-3.0.html](https://www.gnu.org/license/agpl-3.0.html) diff --git a/Rust/services.rs b/Rust/services.rs deleted file mode 100644 index 9ba17ce..0000000 --- a/Rust/services.rs +++ /dev/null @@ -1,65 +0,0 @@ -use std::str::FromStr; - -use pyo3::prelude::*; -use reqwest::{Client, header::{HeaderMap, HeaderName, HeaderValue}}; -use serde::{Deserialize, Serialize}; -use serde_json::{Value, json}; - -#[pyfunction] -pub fn pull_policy_exec_histories() { - println!("Hello World from Rust!"); -} - -#[pyfunction] -pub fn api(py: Python<'_>, x: Py) { - let base_url: String = x.getattr(py, "base_url").unwrap().to_string(); - println!("{}", base_url); -} - -#[tokio::main] -#[pyfunction] -pub async fn history_logging(py: Python<'_>, py_self: Py, exec_types: String, checkpoint_number: String, policy_names: String ) -> String { - #[derive(Serialize, Debug)] - struct Payload { - checkpoint: String, - } - let base_url = py_self.getattr(py, "base_url").unwrap().to_string(); - let headers = py_self.getattr(py, "headers").unwrap().to_string(); - let headers_replace = headers.replace('\'', "\""); - let parsed: Value = serde_json::from_str(&headers_replace.as_str()).unwrap(); - let mut header_map = HeaderMap::new(); - if let Some(obj) = parsed.as_object() { - for (key, value) in obj { - if let Some(v) = value.as_str() { - let val = HeaderValue::from_str(v).unwrap(); - header_map.insert(HeaderName::from_str("X-APIKey").unwrap(), val); - } - } - } - let exec_types = exec_types.replace(" ", ""); - let payload_dict: Payload = Payload { - checkpoint:exec_types.to_string(), - }; - println!("{:?}", payload_dict); - let client = Client::builder() - .danger_accept_invalid_certs(true) - .default_headers(header_map) - .timeout(std::time::Duration::from_secs(30)) - .build() - .unwrap(); - let res = client - .post(format!("{}/v1/logging/exechistories", base_url)) - .json(&payload_dict) - .send() - .await; - match res { - Ok(res) => { - println!("{:?}", res.text().await.unwrap()); - } - Err(e) => { - eprintln!("{}", e); - } - } - let testingstring: String = "Testing".to_string(); - return testingstring -} \ No newline at end of file diff --git a/Server/scheduler_async.py b/Server/scheduler_async.py index e20c123..6f90d36 100644 --- a/Server/scheduler_async.py +++ b/Server/scheduler_async.py @@ -1,3 +1,18 @@ +# 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 asyncio import json import logging diff --git a/flows/otp.py b/flows/otp.py index d299799..61b44fc 100644 --- a/flows/otp.py +++ b/flows/otp.py @@ -15,9 +15,9 @@ +from datetime import datetime import logging import os -from datetime import datetime import pandas as pd @@ -30,9 +30,7 @@ from utils.utils import colorText, get_sanitized_input logger = logging.getLogger(__name__) - - -def generate(api: AirlockAPIWrapper): +def otp_generate(api: AirlockAPIWrapper): otp_dict = {} agents = selectAgents(api) print(colorText("Would you like to continue with these devices?","white")) @@ -59,7 +57,9 @@ def generate(api: AirlockAPIWrapper): logger.info(f"Generated OTP for {agent.hostname}: {otp_code}") otp_dict[agent.hostname] = otp_code - return otp_dict + print(colorText("Requested Codes:", "green")) + for key, value in otp_dict.items(): + print(colorText(f"{key} | {value}","green")) def otp_activities_by_agent(api: AirlockAPIWrapper): activeagents = api.otp_find_active() @@ -128,7 +128,7 @@ def otp_activities_by_agent(api: AirlockAPIWrapper): -def revoke(api: AirlockAPIWrapper): +def otp_revoke(api: AirlockAPIWrapper): activeagents = api.otp_find_active() awaitingagents = api.otp_find_awaiting() diff --git a/flows/prepPolicy.py b/flows/prepPolicy.py index 6f58b14..f66eee7 100644 --- a/flows/prepPolicy.py +++ b/flows/prepPolicy.py @@ -27,7 +27,17 @@ 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, print_x_wide, regulator +from utils.utils import ( + areYouSure, + clear_screen, + colorText, + formatHTML, + get_sanitized_input, + locked, + open_directory, + print_x_wide, + regulator, +) logger = logging.getLogger(__name__) @@ -414,3 +424,204 @@ def testChange(selected_policies, destination_policy, destination_allowlist): return processed_paths, processed_hashes, processed_publishers +def menu_policy_enforce(api: AirlockAPIWrapper): #TODO Need to clean up 6 and 7 into functions + selected_policies = [] + destination_policy = [] + destination_allowlist = [] + processed_paths = [] + processed_hashes = [] + processed_publishers = [] + 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": + clear_screen() + selected_policies = selectPolicies(api,True) + + elif choice == "2": + clear_screen() + 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": + clear_screen() + sortHashes( + api, + selected_policies, + type=[1, 2, 6, 7], + ) + + elif choice == "4": + clear_screen() + if os.path.exists(f"{working_dir}\\Needs_Review\\Review_First\\{selected_policies[0].name}_approved_executions.csv"): + buildPathsandPublishers(selected_policies, False) + else: + print("File not found. Please make sure it's saved correctly and try again.") + + elif choice == "5": + clear_screen() + if os.path.exists(f"{working_dir}\\Approved\\{selected_policies[0].name}_approved_executions.csv") and os.path.exists( + f"{working_dir}\\Approved\\{selected_policies[0].name}_primary_Paths.csv" + ): + buildPreflights(selected_policies) + else: + print("File not found. Please make sure it's saved correctly and try again.") + + elif choice == "6": + clear_screen() + if ( + os.path.exists(f"{working_dir}\\Preflight\\{selected_policies[0].name}_approved_paths.csv") + and os.path.exists(f"{working_dir}\\Preflight\\{selected_policies[0].name}_approved_hashes.csv") + and destination_policy + and destination_allowlist + ): + processed_paths, processed_hashes, processed_publishers = testChange(selected_policies, destination_policy, destination_allowlist) + else: + # Log which condition(s) failed + missing_items = [] + if not os.path.exists(f"{working_dir}\\Preflight\\{selected_policies[0].name}_approved_paths.csv"): + missing_items.append("approved_paths.csv not found") + if not os.path.exists(f"{working_dir}\\Preflight\\{selected_policies[0].name}_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": + clear_screen() + areYouSure() + confirmation = get_sanitized_input("Type 'I AGREE' to continue: ") + if ( + processed_paths + and processed_hashes + and processed_publishers + 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) + + locked() + + else: + logger.error("Confirmation block failed. Reasons:") + if not processed_publishers or processed_hashes or processed_paths: + logger.error(" - Test not performed.") + 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() == "B": + break + + + else: + print(colorText("Invalid choice. Please try again.", "red")) + +def section_header(title): + print(colorText("\n --------------------------------------------------------------------", "cyan")) + print(colorText(f" ------------- {title} -------------", "cyan")) + print(colorText(" --------------------------------------------------------------------", "cyan")) + + +def printEnforceChecklist(selected_policies, destination_policy, destination_allowlist): + working_dir = load_env("WORKING_DIR") + section_header("🛠️ 🔒 Prepare to Enforce Policy 🛠️ 🔒") + print(colorText("\nSequentially follow these steps to prepare a policy for enforcement:", "white")) + + # Step 1: Originating Policies + print(colorText("\n1. Choose which policy or policies to gather execution info from", "cyan")) + if not selected_policies: + print(colorText(" [✗] No policies have been chosen", "red")) + else: + print(colorText("The following policies have been chosen:", "green")) + for policy in selected_policies: + print(colorText(f" [✓] {policy.name}", "green")) + + # Step 2: Destination Policy and Allowlist + print(colorText("2. Choose the destination policy and associated allowlist", "cyan")) + if destination_policy: + print(colorText(f" [✓] {destination_policy[0].name} has been selected as the destination policy", "green")) + else: + print(colorText(" [✗] No destination policy has been chosen", "red")) + + if destination_allowlist: + print(colorText(f" [✓] {destination_allowlist[0].name} has been selected as allowlist", "green")) + else: + print(colorText(" [✗] No allowlist has been chosen", "red")) + + # Step 3: Data Preparation + print(colorText(f"3. Select to begin pulling execution history. The executions will be sorted and placed in {working_dir}\\data\\Needs_Review", "cyan")) + if selected_policies: + policy_id = selected_policies[0].name + review_path = f"{working_dir}\\Needs_Review\\Review_First\\{policy_id}_approved_executions.csv" + print(colorText(" [✓] Data has been fetched" if os.path.exists(review_path) else " [✗] Data has not been fetched", "green" if os.path.exists(review_path) else "red")) + else: + print(colorText(" [✗] No policies selected, cannot check data fetch status", "red")) + + # Step 4: Manual Review + print(colorText("4. Manually review the files:", "cyan")) + print(colorText(" Remove the rows containing hashes you do not approve of", "cyan")) + print(colorText(f" When complete, save both csv files to {working_dir}\\data\\Approved and choose this option.", "cyan")) + print(colorText(" This will start the process to generate possible filepath approvals", "cyan")) + + if selected_policies: + policy_id = selected_policies[0].name + approved_path = f"{working_dir}\\Approved\\{policy_id}_approved_executions.csv" + second_review_path = f"{working_dir}\\Needs_Review\\Review_Second\\{policy_id}_primary_Paths.csv" + print(colorText(" [✓] Reviewed hashes have been loaded" if os.path.exists(approved_path) else " [✗] Reviewed hashes have not been loaded", "green" if os.path.exists(approved_path) else "red")) + print(colorText(" [✓] Path review list created" if os.path.exists(second_review_path) else " [✗] Path review list has not been created", "green" if os.path.exists(second_review_path) else "red")) + else: + print(colorText(" [✗] No policies selected, cannot check reviewed hashes or path list", "red")) + + # Step 5: Path Review + print(colorText(f"5. Manually review the files in {working_dir}\\Needs_Review\\Review_Second\\", "cyan")) + print(colorText(" Remove the rows containing path exclusions or publishers you do not approve of.", "cyan")) + print(colorText(f" When complete, save the files to {working_dir}\\data\\Approved", "cyan")) + print(colorText(" Choose this option when done to build your preflights", "cyan")) + + if selected_policies: + policy_id = selected_policies[0].name + reviewed_path = f"{working_dir}\\Approved\\{policy_id}_primary_Paths.csv" + preflight_paths = f"{working_dir}\\Preflight\\{policy_id}_approved_paths.csv" + preflight_hashes = f"{working_dir}\\Preflight\\{policy_id}_approved_hashes.csv" + print(colorText(" [✓] Reviewed path list detected" if os.path.exists(reviewed_path) else " [✗] Path review list has not been detected", "green" if os.path.exists(reviewed_path) else "red")) + preflight_ready = os.path.exists(preflight_paths) and os.path.exists(preflight_hashes) + print(colorText(" [✓] Preflight Path Exclusion List has been generated" if preflight_ready else " [✗] Preflight Path Exclusion List has not been generated", "green" if preflight_ready else "red")) + else: + print(colorText(" [✗] No policies selected, cannot check preflight status", "red")) + + # Final Steps + print(colorText("6. Test ------------------------------------------------------", "cyan")) + print(colorText(" Prints to console the changes that would be made, must be done to proceed. ", "cyan")) + + print(colorText("7. Liftoff ------------------------------------------------------", "cyan")) + print(colorText(" Apply path exclusions and approved publishers to selected policy", "cyan")) + print(colorText(" Apply approved hashes to allowlist", "cyan")) + + + # Utility Options + print(colorText("F. 📂 - Open Working Directory", "cyan")) + print(colorText("B. 🔚 - Back", "cyan")) diff --git a/flows/quietAgent.py b/flows/quietAgent.py index fd5e1a3..6bc05ea 100644 --- a/flows/quietAgent.py +++ b/flows/quietAgent.py @@ -22,8 +22,9 @@ import pandas as pd from flows.prepPolicy import selectPolicies from services.API import AirlockAPIWrapper from services.policyhandler import getPolicyInfo +from utils.configmanager import load_env from utils.selector import Selector -from utils.utils import colorText, get_sanitized_input, load_env +from utils.utils import colorText, get_sanitized_input logger = logging.getLogger(__name__) diff --git a/models/execution.py b/models/execution.py index a4231fc..c5375df 100644 --- a/models/execution.py +++ b/models/execution.py @@ -14,13 +14,13 @@ # along with this program. If not, see . import dataclasses +from dataclasses import asdict, dataclass +from datetime import datetime import inspect import json import logging import os import re -from dataclasses import asdict, dataclass -from datetime import datetime from typing import List, Optional, Tuple import dotenv diff --git a/services/agenthandler.py b/services/agenthandler.py index 5a89991..531fc41 100644 --- a/services/agenthandler.py +++ b/services/agenthandler.py @@ -14,12 +14,12 @@ # along with this program. If not, see . +from dataclasses import asdict +from datetime import datetime, timedelta import json import logging import os import re -from dataclasses import asdict -from datetime import datetime, timedelta from typing import List import pandas as pd diff --git a/services/policyhandler.py b/services/policyhandler.py index f76b969..5acc422 100644 --- a/services/policyhandler.py +++ b/services/policyhandler.py @@ -21,15 +21,15 @@ import logging import os import sys +from bson import ObjectId 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 +from utils.utils import areYouSure, colorText, get_sanitized_input logger = logging.getLogger(__name__) @@ -235,3 +235,10 @@ def updateAuditPoliciesFromEnforcementPolices(api: AirlockAPIWrapper): for enforcement_policy, audit_policy in policy_relationship_map.items(): api.policy_clone(enforcement_policy, audit_policy) api.policy_set_auditmode(audit_policy, "1") + + +def confirmUpdateAfromE(api: AirlockAPIWrapper): + areYouSure() + confirmation = get_sanitized_input("Type 'I AGREE' to continue: ") + if confirmation.strip() == "I AGREE": + updateAuditPoliciesFromEnforcementPolices(api) \ No newline at end of file diff --git a/services/security.py b/services/security.py index 7574824..aa95929 100644 --- a/services/security.py +++ b/services/security.py @@ -14,17 +14,17 @@ # along with this program. If not, see . import base64 +from getpass import getpass import logging import os import platform import re import sys -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 +import keyring # Constants KDF_ITERATIONS = 200_000 diff --git a/utils/configmanager.py b/utils/configmanager.py index 56c8f3c..62c2bdf 100644 --- a/utils/configmanager.py +++ b/utils/configmanager.py @@ -1,8 +1,23 @@ +# 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 os -import sys from pathlib import Path +import sys from typing import Callable, Optional, TypeVar T = TypeVar("T") diff --git a/utils/menus.py b/utils/menus.py deleted file mode 100644 index 2363a58..0000000 --- a/utils/menus.py +++ /dev/null @@ -1,316 +0,0 @@ -# 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 dotenv - -import services.policyhandler as policyh -from flows.otp import generate, otp_activities_by_agent, revoke -from flows.prepPolicy import ( - buildPathsandPublishers, - buildPreflights, - selectAllowlists, - selectPolicies, - sortHashes, - testChange, -) -from flows.quietAgent import findQuietAgents -from services.agenthandler import ( - findAgents, - moveAgents, - toggleEnforcement, -) -from services.API import AirlockAPIWrapper -from utils.configmanager import load_env -from utils.utils import ( - areYouSure, - clear_screen, - colorText, - displayIntro, - get_sanitized_input, - locked, - open_directory, - printEnforceChecklist, - welcome, -) - -logger = logging.getLogger(__name__) - -dotenv.load_dotenv() - -def menu_main(api: AirlockAPIWrapper): - working_dir = load_env("WORKING_DIR") - extras = load_env("EXTRAS") - while True: - clear_screen() - displayIntro() - welcome() - print(colorText("1. 🔍 - Device Search", "yellow")) - print(colorText("2. 🔀 - Move Device(s)", "yellow")) - print(colorText("3. 🎫 - One Time Pass (OTP)", "yellow")) - print(colorText("4. 🔇 - Find Quiet Hosts", "yellow")) - if extras == "POLICYPREP" : print(colorText("5. 🛡️ - Policy Enforcement Tools", "yellow")) - print(colorText("=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=", "cyan")) - 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": - findAgents(api,False) - elif choice == "2": - menu_move(api) - elif choice == "3": - menu_otp(api) - elif choice == "4": - findQuietAgents(api) - elif choice == "5": - 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): #TODO Need to clean up 6 and 7 into functions - selected_policies = [] - destination_policy = [] - destination_allowlist = [] - processed_paths = [] - processed_hashes = [] - processed_publishers = [] - 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\\{selected_policies[0].name}_approved_executions.csv"): - buildPathsandPublishers(selected_policies, 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\\{selected_policies[0].name}_approved_executions.csv") and os.path.exists( - f"{working_dir}\\Approved\\{selected_policies[0].name}_primary_Paths.csv" - ): - buildPreflights(selected_policies) - 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\\{selected_policies[0].name}_approved_paths.csv") - and os.path.exists(f"{working_dir}\\Preflight\\{selected_policies[0].name}_approved_hashes.csv") - and destination_policy - and destination_allowlist - ): - processed_paths, processed_hashes, processed_publishers = testChange(selected_policies, destination_policy, destination_allowlist) - else: - # Log which condition(s) failed - missing_items = [] - if not os.path.exists(f"{working_dir}\\Preflight\\{selected_policies[0].name}_approved_paths.csv"): - missing_items.append("approved_paths.csv not found") - if not os.path.exists(f"{working_dir}\\Preflight\\{selected_policies[0].name}_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 ( - processed_paths - and processed_hashes - and processed_publishers - 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) - - locked() - - else: - logger.error("Confirmation block failed. Reasons:") - if not processed_publishers or processed_hashes or processed_paths: - logger.error(" - Test not performed.") - 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_move(api: AirlockAPIWrapper): - working_dir = load_env("WORKING_DIR") - while True: - clear_screen() - displayIntro() - print(colorText("=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-= ↔️ Agent Movement ↔️ =-=-=-=-=-=-=-=-=-=-=-=-=-=-=", "cyan")) - print(colorText("1. ✅ - Move to local approval (Placeholder)", "yellow")) - print(colorText("2. 🔄 - Move to Audit/Enforcement", "yellow")) - print(colorText("3. 🔀 - Move - Other", "yellow")) - print(colorText("=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=", "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": - print("This Feature is still in development") - get_sanitized_input("Press enter to continue") - elif choice == "2": - toggleEnforcement(api) - elif choice == "3": - moveAgents(api) - elif choice.upper() == "F": - open_directory(working_dir) - elif choice.upper() == "S": - menu_settings() - elif choice.upper() == "B": - break - - -def menu_otp(api: AirlockAPIWrapper): - working_dir = load_env("WORKING_DIR") - while True: - clear_screen() - displayIntro() - print(colorText("=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-= 🎫 OTP 🎫 =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=", "cyan")) - print(colorText("1. 🔐 -Generate OTPs", "cyan")) - print(colorText("2. 📊 -OTP Activities By Agent", "cyan")) - print(colorText("3. ❌ -Revoke OTPs", "cyan")) - print(colorText("=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=", "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("Requested Codes:", "green")) - for key, value in otp_list.items(): - print(colorText(f"{key} | {value}","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: - clear_screen() - displayIntro() - print(colorText("=-=-=-=-=-=-=-=-=-=-=-=-=-=-= 🛡️ Policy Tools 🛡️ =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=", "cyan")) - print(colorText("1. 🔒 - Prepare Policy For Enforcement", "yellow")) - print(colorText("2. 🔄 - Update Audit Policies from Enforcement Policies", "yellow")) - print(colorText("=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=", "cyan")) - 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: - clear_screen() - displayIntro() - print(colorText("=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-= 🛠️ Settings 🛠️ =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=", "cyan")) - print(colorText("This Feature is still in development, if you have ideas for options you would like to see, let us know.", "cyan")) - print(colorText("=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=", "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/utils/selector.py b/utils/selector.py index 8659a41..714e197 100644 --- a/utils/selector.py +++ b/utils/selector.py @@ -1,3 +1,17 @@ +# 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 diff --git a/utils/setup.py b/utils/setup.py index e12bef5..f7ba7f2 100644 --- a/utils/setup.py +++ b/utils/setup.py @@ -18,9 +18,9 @@ import logging import logging.config import logging.handlers import os +from pathlib import Path import platform import sys -from pathlib import Path from dotenv import load_dotenv, set_key diff --git a/utils/tui.py b/utils/tui.py new file mode 100644 index 0000000..5643b0f --- /dev/null +++ b/utils/tui.py @@ -0,0 +1,215 @@ +# 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 + +from blessed import Terminal +import dotenv + +from flows.otp import otp_activities_by_agent, otp_generate, otp_revoke +from flows.prepPolicy import menu_policy_enforce +from flows.quietAgent import findQuietAgents +from services.agenthandler import ( + findAgents, + moveAgents, + toggleEnforcement, +) +from services.API import AirlockAPIWrapper +from services.policyhandler import confirmUpdateAfromE +from utils.configmanager import load_env + +logger = logging.getLogger(__name__) +dotenv.load_dotenv() +term = Terminal() + + + + +logo = r""" + _____ .__ .__ __ ___________ .__ + / _ \ |__|______| | ____ ____ | | __ \__ ___/___ ____ | | ______ + / /_\ \| \_ __ \ | / _ \_/ ___\| |/ / | | / _ \ / _ \| | / ___/ +/ | \ || | \/ |_( <_> ) \___| < | |( <_> | <_> ) |__\___ \ +\____|__ /__||__| |____/\____/ \___ >__|_ \ |____| \____/ \____/|____/____ > + \/ \/ \/ \/ +""" + + + +footer_keys = ["F: Folder", "S: Settings", "B: Back", "Q: Quit"] + +# Box drawing characters +TOP_LEFT = '╔' +TOP_RIGHT = '╗' +BOTTOM_LEFT = '╚' +BOTTOM_RIGHT = '╝' +HORIZONTAL = '═' +VERTICAL = '║' +SHADOW = '░' + +logo_lines = logo.splitlines() +logo_width = max(len(line) for line in logo_lines) + +def draw_screen(title, items, selected_index): + print(term.clear) + center_x = (term.width - logo_width) // 2 + top = len(logo_lines) + 2 + height = len(items) + 4 + + for i, line in enumerate(logo_lines): + print(term.move_xy(center_x, i) + term.cyan(line)) + + box_color = term.bright_magenta # or term.cyan, term.green, etc. + + print(term.move_xy(center_x, top) + box_color(TOP_LEFT + HORIZONTAL * logo_width + TOP_RIGHT)) + for i in range(height): + print(term.move_xy(center_x, top + 1 + i) + box_color(VERTICAL) + ' ' * logo_width + box_color(VERTICAL)) + print(term.move_xy(center_x, top + 1 + height) + box_color(BOTTOM_LEFT + HORIZONTAL * logo_width + BOTTOM_RIGHT)) + + for i in range(1,height + 2): + print(term.move_xy(center_x + logo_width + 2, top + i) + term.darkgray(SHADOW)) + print(term.move_xy(center_x + 1, top + height + 2) + term.darkgray(SHADOW * (logo_width + 2))) + + print(term.move_xy(center_x + 4, top) + term.bold_magenta(title)) + for i, item in enumerate(items): + style = term.reverse if i == selected_index else term.bold_yellow + print(term.move_xy(center_x + 4, top + 2 + i) + style(f"{i+1}. {item}")) + + footer_text = " ".join([term.bold_cyan(k) for k in footer_keys]) + footer_x = (term.width - len(footer_text)) // 2 + print(term.move_xy(footer_x, term.height - 2) + footer_text) + print(term.move_xy(footer_x, term.height - 4) + term.bold("Use ↑/↓ or number keys. Press Enter to select."), end='', flush=True) + + +def run_legacy_function(func, *args, **kwargs): + try: + # Exit fullscreen and restore terminal state + print(term.exit_fullscreen, end='', flush=True) + print(term.normal_cursor, end='', flush=True) + + # Reset terminal state on Linux + if platform.system() != 'Windows': + os.system('stty sane') + + # Clear screen + os.system('cls' if os.name == 'nt' else 'clear') + + # Run the legacy function + func(*args, **kwargs) + input("Press Enter to return to the previous menu...") + + except Exception as e: + print(f"Error during legacy function: {e}") + input("Press Enter to return to the Main Menu...") + + +def run_menu(api: AirlockAPIWrapper, start_menu="Main Menu"): + current_menu = start_menu + extras = load_env("EXTRAS") + selected_index = 0 + history = [] + menus = { + "Main Menu": [ + "🔍 - Device Search", + "🔀 - Move Device(s)", + "🎫 - One Time Pass (OTP)", + "🔇 - Find Quiet Hosts", + ], + "🔀 - Move Device(s)": [ + "✅ - Move to local approval (Placeholder)", + "🔄 - Move to Audit/Enforcement", + "🔀 - Move - Other" + ], + "🎫 - One Time Pass (OTP)": [ + "🔐 - Generate OTPs", + "📊 - OTP Activities By Agent", + "❌ - Revoke OTPs" + ], + "Settings": [ + "(Placeholder) Change Working Directory" + ] + } + + if extras == "POLICYPREP": + menus["Main Menu"].insert(4, "🛡️ - Policy Enforcement Tools") + menus["🛡️ - Policy Enforcement Tools"] = [ + "🔒 - Prepare Policy For Enforcement", + "🔄 - Update Audit Policies" + ] + + actions = { + "🔍 - Device Search": (findAgents, [api, False]), + "🔇 - Find Quiet Hosts": (findQuietAgents, [api]), + "🔄 - Move to Audit/Enforcement": (toggleEnforcement, [api]), + "🔀 - Move - Other": (moveAgents, [api]), + "🔐 - Generate OTPs": (otp_generate, [api]), + "📊 - OTP Activities By Agent": (otp_activities_by_agent, [api]), + "❌ - Revoke OTPs": (otp_revoke, [api]), + "🔄 - Update Audit Policies": (confirmUpdateAfromE, [api]), + "🔒 - Prepare Policy For Enforcement": (menu_policy_enforce, [api]) + } + + while True: + func_to_run = None + args_to_run = [] + + with term.fullscreen(), term.cbreak(), term.hidden_cursor(): + draw_screen(current_menu, menus[current_menu], selected_index) + key = term.inkey() + + items = menus[current_menu] + if key.name == "KEY_UP": + selected_index = (selected_index - 1) % len(items) + elif key.name == "KEY_DOWN": + selected_index = (selected_index + 1) % len(items) + elif key.name == "KEY_ENTER" or key == "\n": + selected_item = items[selected_index] + if selected_item in menus: + history.append(current_menu) + current_menu = selected_item + selected_index = 0 + elif selected_item in actions: + func_to_run, args_to_run = actions[selected_item] + elif key.upper() == "Q": + return + elif key.upper() == "B": + if history: + current_menu = history.pop() + selected_index = 0 + elif key.upper() == "F": + print(term.move_xy(4, term.height - 6) + term.bold_green("Folder selected")) + term.inkey(timeout=2) + + elif key.upper() == "S": + history.append(current_menu) # Add this line + current_menu = "Settings" + selected_index = 0 + + elif key.isdigit(): + num = int(key) + if 1 <= num <= len(items): + selected_index = num - 1 + selected_item = items[selected_index] + if selected_item in menus: + history.append(current_menu) + current_menu = selected_item + selected_index = 0 + elif selected_item in actions: + func_to_run, args_to_run = actions[selected_item] + + # Run legacy function outside of terminal context + if func_to_run: + run_legacy_function(func_to_run, *args_to_run) diff --git a/utils/utils.py b/utils/utils.py index 76df8fe..539a588 100644 --- a/utils/utils.py +++ b/utils/utils.py @@ -25,8 +25,6 @@ from tkinter import filedialog import pandas as pd -from utils.configmanager import load_env - logger = logging.getLogger(__name__) @@ -191,86 +189,6 @@ def section_header(title): print(colorText(f" ------------- {title} -------------", "cyan")) print(colorText(" --------------------------------------------------------------------", "cyan")) -def printEnforceChecklist(selected_policies, destination_policy, destination_allowlist): - working_dir = load_env("WORKING_DIR") - section_header("🛠️ 🔒 Prepare to Enforce Policy 🛠️ 🔒") - print(colorText("\nSequentially follow these steps to prepare a policy for enforcement:", "white")) - - # Step 1: Originating Policies - print(colorText("\n1. Choose which policy or policies to gather execution info from", "cyan")) - if not selected_policies: - print(colorText(" [✗] No policies have been chosen", "red")) - else: - print(colorText("The following policies have been chosen:", "green")) - for policy in selected_policies: - print(colorText(f" [✓] {policy.name}", "green")) - - # Step 2: Destination Policy and Allowlist - print(colorText("2. Choose the destination policy and associated allowlist", "cyan")) - if destination_policy: - print(colorText(f" [✓] {destination_policy[0].name} has been selected as the destination policy", "green")) - else: - print(colorText(" [✗] No destination policy has been chosen", "red")) - - if destination_allowlist: - print(colorText(f" [✓] {destination_allowlist[0].name} has been selected as allowlist", "green")) - else: - print(colorText(" [✗] No allowlist has been chosen", "red")) - - # Step 3: Data Preparation - print(colorText(f"3. Select to begin pulling execution history. The executions will be sorted and placed in {working_dir}\\data\\Needs_Review", "cyan")) - if selected_policies: - policy_id = selected_policies[0].name - review_path = f"{working_dir}\\Needs_Review\\Review_First\\{policy_id}_approved_executions.csv" - print(colorText(" [✓] Data has been fetched" if os.path.exists(review_path) else " [✗] Data has not been fetched", "green" if os.path.exists(review_path) else "red")) - else: - print(colorText(" [✗] No policies selected, cannot check data fetch status", "red")) - - # Step 4: Manual Review - print(colorText("4. Manually review the files:", "cyan")) - print(colorText(" Remove the rows containing hashes you do not approve of", "cyan")) - print(colorText(f" When complete, save both csv files to {working_dir}\\data\\Approved and choose this option.", "cyan")) - print(colorText(" This will start the process to generate possible filepath approvals", "cyan")) - - if selected_policies: - policy_id = selected_policies[0].name - approved_path = f"{working_dir}\\Approved\\{policy_id}_approved_executions.csv" - second_review_path = f"{working_dir}\\Needs_Review\\Review_Second\\{policy_id}_primary_Paths.csv" - print(colorText(" [✓] Reviewed hashes have been loaded" if os.path.exists(approved_path) else " [✗] Reviewed hashes have not been loaded", "green" if os.path.exists(approved_path) else "red")) - print(colorText(" [✓] Path review list created" if os.path.exists(second_review_path) else " [✗] Path review list has not been created", "green" if os.path.exists(second_review_path) else "red")) - else: - print(colorText(" [✗] No policies selected, cannot check reviewed hashes or path list", "red")) - - # Step 5: Path Review - print(colorText(f"5. Manually review the files in {working_dir}\\Needs_Review\\Review_Second\\", "cyan")) - print(colorText(" Remove the rows containing path exclusions or publishers you do not approve of.", "cyan")) - print(colorText(f" When complete, save the files to {working_dir}\\data\\Approved", "cyan")) - print(colorText(" Choose this option when done to build your preflights", "cyan")) - - if selected_policies: - policy_id = selected_policies[0].name - reviewed_path = f"{working_dir}\\Approved\\{policy_id}_primary_Paths.csv" - preflight_paths = f"{working_dir}\\Preflight\\{policy_id}_approved_paths.csv" - preflight_hashes = f"{working_dir}\\Preflight\\{policy_id}_approved_hashes.csv" - print(colorText(" [✓] Reviewed path list detected" if os.path.exists(reviewed_path) else " [✗] Path review list has not been detected", "green" if os.path.exists(reviewed_path) else "red")) - preflight_ready = os.path.exists(preflight_paths) and os.path.exists(preflight_hashes) - print(colorText(" [✓] Preflight Path Exclusion List has been generated" if preflight_ready else " [✗] Preflight Path Exclusion List has not been generated", "green" if preflight_ready else "red")) - else: - print(colorText(" [✗] No policies selected, cannot check preflight status", "red")) - - # Final Steps - print(colorText("6. Test ------------------------------------------------------", "cyan")) - print(colorText(" Prints to console the changes that would be made, must be done to proceed. ", "cyan")) - - print(colorText("7. Liftoff ------------------------------------------------------", "cyan")) - print(colorText(" Apply path exclusions and approved publishers to selected policy", "cyan")) - print(colorText(" Apply approved hashes to allowlist", "cyan")) - - - # Utility Options - print(colorText("R. Remove/Reset Generated data - will prompt to allow keeping execution history", "cyan")) - print(colorText("F. 📂 - Open Working Directory", "cyan")) - print(colorText("B. 🔚 - Back", "cyan")) def areYouSure():