Major UI Improvment with blessing
This commit is contained in:
@@ -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__":
|
||||
|
||||
@@ -1,20 +1,59 @@
|
||||
[](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)
|
||||
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)
|
||||
|
||||
@@ -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<PyAny>) {
|
||||
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<PyAny>, 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
|
||||
}
|
||||
@@ -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 <https://www.gnu.org/licenses/>.
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
|
||||
+6
-6
@@ -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()
|
||||
|
||||
+212
-1
@@ -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"))
|
||||
|
||||
+2
-1
@@ -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__)
|
||||
|
||||
|
||||
+2
-2
@@ -14,13 +14,13 @@
|
||||
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
|
||||
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
|
||||
|
||||
@@ -14,12 +14,12 @@
|
||||
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
|
||||
|
||||
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
|
||||
|
||||
@@ -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)
|
||||
@@ -14,17 +14,17 @@
|
||||
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
|
||||
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
|
||||
|
||||
+16
-1
@@ -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 <https://www.gnu.org/licenses/>.
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
import sys
|
||||
from typing import Callable, Optional, TypeVar
|
||||
|
||||
T = TypeVar("T")
|
||||
|
||||
-316
@@ -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 <https://www.gnu.org/licenses/>.
|
||||
|
||||
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.")
|
||||
|
||||
@@ -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 <https://www.gnu.org/licenses/>.
|
||||
import logging
|
||||
from typing import Any, Callable, List, Optional, Union
|
||||
|
||||
|
||||
+1
-1
@@ -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
|
||||
|
||||
|
||||
+215
@@ -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 <https://www.gnu.org/licenses/>.
|
||||
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)
|
||||
@@ -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():
|
||||
|
||||
Reference in New Issue
Block a user