mysticmomba/AirlockTools
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

feat: add version updater and statistics enhancements (fixes #29)

Browse Source 
- Implemented version checking system with update notifications
- Integrated Git for fetching and downloading the latest version
- Added statistics updates
- Removed unused code across the project
- Condensed project structure
- Updated README
- Cleaned up UI
This commit is contained in:
Zarithas
2025-12-19 16:31:36 -05:00
parent 09d2c125cd
commit e1e0cb7ac7
26 changed files with 1906 additions and 3806 deletions
Download Patch File Download Diff File Expand all files Collapse all files
TUI/Widgets/OTP_generate.py
-1
View File
@@ -43,7 +43,6 @@ class OTPGenerator(Widget):
# Reactive properties to track form completion
requestor_filled = reactive(False)
reasoning_filled = reactive(False)
duration_selected = reactive(True) # Default is selected
otp_generated = reactive(False)
class OTPInfo(Message):
TUI/Widgets/agentmoveoperations.py
+77 -25
View File
@@ -29,14 +29,71 @@ from textual.widget import Widget
from textual.widgets import Button, DataTable, Footer, Header, Static, TextArea
from models.agent import Agent
from services.API import AirlockAPIWrapper
from TUI.Screens.executionhistoryscreen import ExecutionHistoryScreen
from TUI.Screens.otpworkflowscreen import OTPWorkflowScreen
from TUI.Screens.policyselectorscreen import PolicySelectorScreen
from TUI.Widgets.OTP_generate import OTPGenerator
from utils.configmanager import get_system_json
logger = logging.getLogger(__name__)
def moveAgentToRelatedPolicy(
api: AirlockAPIWrapper,
agent: Agent,
mode: str = "audit",
):
"""
Moves an agent between audit and enforcement policies based on the mode.
Args:
api: AirlockAPIWrapper instance.
agent: Agent object.
mode: 'audit' to move to audit, 'enforcement' to move to enforcement.
"""
policy_relationship_map = get_system_json("POLICY_MAP_ENF_AUD", "{}")
if mode == "audit":
if agent.groupid in policy_relationship_map:
target_policy = policy_relationship_map[agent.groupid]
elif agent.groupid in policy_relationship_map.values():
logger.debug(
f"Agent {agent.hostname} is already in an audit group. No action needed."
)
print(
f"Agent {agent.hostname} is already in an audit group. No action needed."
)
return
else:
logger.warning(
f"Error: No corresponding audit policy found for groupid: {agent.groupid}."
)
return
elif mode == "enforcement":
inverse_map = {v: k for k, v in policy_relationship_map.items()}
if agent.groupid in inverse_map:
target_policy = inverse_map[agent.groupid]
elif agent.groupid in inverse_map.values():
logger.info(
f"Agent {agent.hostname} is already in an enforcement group. No action needed."
)
return
else:
logger.warning(
f"Error: No corresponding enforcement policy found for groupid: {agent.groupid}."
)
return
else:
logger.error(f"Unknown mode '{mode}'. Use 'audit' or 'enforcement'.")
return
result = api.agent_move(agent.agentid, target_policy)
return result
class AgentMoveOperations(Widget):
"""
A Textual widget for managing bulk agent operations and policy migrations.
@@ -216,11 +273,11 @@ class AgentMoveOperations(Widget):
results_lines.append(" (none)")
results_lines.append("")
results_lines.append(f"❌ Failed ({len(unsuccessful)}):")
results_lines.append(f"❌ Failed ({len(unsuccessful)}):")
if unsuccessful:
for agent, error in unsuccessful:
results_lines.append(f" ❌ {agent.hostname}: {error}")
results_lines.append(f" ❌ {agent.hostname}: {error}")
else:
results_lines.append(" (none)")
@@ -255,9 +312,9 @@ class AgentMoveOperations(Widget):
- Operations panel: 1/3 width
- Results area: Initially hidden, shown after operation completion
"""
yield Header(show_clock=True, icon="⚙️")
yield Header(show_clock=True, icon="âš™❗")
title_text = Static(
f"🖥️ Agent Operations - {len(self.agents)} device(s) selected",
f"🖥❗ Agent Operations - {len(self.agents)} device(s) selected",
id="move_ops_title",
)
title_text.styles.margin = (0, 0, 1, 0)
@@ -292,39 +349,39 @@ class AgentMoveOperations(Widget):
yield operations_label
# Operation buttons
export_csv_btn = Button("📄 Export CSV", id="export_csv_btn")
export_csv_btn = Button("📄 Export CSV", id="export_csv_btn")
export_csv_btn.styles.width = "100%"
export_csv_btn.styles.margin = (0, 0, 1, 0)
yield export_csv_btn
local_approval_btn = Button(
"✔️ Local Approval Mode", id="local_approval_btn"
"✔❗ Local Approval Mode", id="local_approval_btn"
)
local_approval_btn.styles.width = "100%"
local_approval_btn.styles.margin = (0, 0, 1, 0)
yield local_approval_btn
otp_gen_btn = Button("🎫 Generate One Time Passes", id="otp_gen_btn")
otp_gen_btn = Button("🎫 Generate One Time Passes", id="otp_gen_btn")
otp_gen_btn.styles.width = "100%"
otp_gen_btn.styles.margin = (0, 0, 1, 0)
yield otp_gen_btn
toggle_enforcement_btn = Button(
"🔄 Toggle Audit/Enforcement", id="toggle_enforcement_btn"
"🔄 Toggle Audit/Enforcement", id="toggle_enforcement_btn"
)
toggle_enforcement_btn.styles.width = "100%"
toggle_enforcement_btn.styles.margin = (0, 0, 1, 0)
yield toggle_enforcement_btn
other_policy_btn = Button(
"🔀 Move to Other Policy", id="other_policy_btn"
"🔀 Move to Other Policy", id="other_policy_btn"
)
other_policy_btn.styles.width = "100%"
other_policy_btn.styles.margin = (0, 0, 1, 0)
yield other_policy_btn
exec_history_btn = Button(
"📊 View Execution History", id="exec_history_btn"
"📊 View Execution History", id="exec_history_btn"
)
exec_history_btn.styles.width = "100%"
exec_history_btn.styles.margin = (0, 0, 1, 0)
@@ -391,17 +448,17 @@ class AgentMoveOperations(Widget):
pyperclip.copy(results_text.text)
self.app.notify(
"📋✅ Results copied to clipboard!",
"📋✅ Results copied to clipboard!",
severity="information",
timeout=2,
)
except ImportError:
self.app.notify(
"❌ pyperclip not installed. Run: pip install pyperclip",
"❌ pyperclip not installed. Run: pip install pyperclip",
severity="warning",
)
except Exception as e:
self.app.notify(f"❌ Failed to copy: {str(e)}", severity="error")
self.app.notify(f"❌ Failed to copy: {str(e)}", severity="error")
event.stop()
elif btn_id == "export_csv_btn":
self._start_export_csv_operation()
@@ -452,7 +509,7 @@ class AgentMoveOperations(Widget):
self.operation_in_progress = True
status_label = self.query_one("#status_label", Static)
status_label.update("✔️ Moving agents to local approval...")
status_label.update("✔❗ Moving agents to local approval...")
# Get API from app
api = self.app.api
@@ -463,8 +520,6 @@ class AgentMoveOperations(Widget):
try:
import time
from services.agenthandler import moveAgentToRelatedPolicy
# Generate batch ID
batch = int(time.time())
duration = 360 # Default 6 hours, could make this configurable
@@ -487,7 +542,7 @@ class AgentMoveOperations(Widget):
except Exception as e:
logger.error(f"Error during local approval operation: {e}")
status_label.update(f"❌ Error: {str(e)}")
status_label.update(f"❌ Error: {str(e)}")
self.operation_in_progress = False
return
@@ -537,7 +592,7 @@ class AgentMoveOperations(Widget):
successful.append(file_path)
status_label.update(f"✅ Exported to {file_path}")
except Exception:
status_label.update("❌ Failed")
status_label.update("❌ Failed")
self.operation_in_progress = False
@@ -581,7 +636,7 @@ class AgentMoveOperations(Widget):
self.operation_in_progress = True
status_label = self.query_one("#status_label", Static)
status_label.update("🔄 Toggling enforcement mode...")
status_label.update("🔄 Toggling enforcement mode...")
# Get API from app
api = self.app.api
@@ -590,9 +645,6 @@ class AgentMoveOperations(Widget):
unsuccessful = []
try:
from services.agenthandler import moveAgentToRelatedPolicy
from utils.configmanager import get_system_json
policy_relationship_map = get_system_json("POLICY_MAP_ENF_AUD", "{}")
for agent in self.agents:
@@ -617,7 +669,7 @@ class AgentMoveOperations(Widget):
except Exception as e:
logger.error(f"Error during toggle enforcement operation: {e}")
status_label.update(f"❌ Error: {str(e)}")
status_label.update(f"❌ Error: {str(e)}")
self.operation_in_progress = False
return
@@ -687,7 +739,7 @@ class AgentMoveOperations(Widget):
except Exception as e:
logger.error(f"Error loading policies: {e}")
status_label.update(f"❌ Error: {str(e)}")
status_label.update(f"❌ Error: {str(e)}")
self.operation_in_progress = False
self.selected_operation = ""
self.app.notify(f"Failed to load policies: {str(e)}", severity="error")
@@ -722,7 +774,7 @@ class AgentMoveOperations(Widget):
)
except Exception as e:
logger.error(f"Failed to open execution history viewer: {e}")
status_label.update(f"❌ Error: {str(e)}")
status_label.update(f"❌ Error: {str(e)}")
self.app.notify(
f"Failed to open execution history: {str(e)}", severity="error"
)
TUI/Widgets/prepPolicy.py
-870
View File
@@ -1,870 +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 os.path
import re
from typing import List
import dotenv
import pandas as pd
from models.execution import ExecutionHistoryRecord
from models.policy import Allowlist, Policy
from services.API import AirlockAPIWrapper
from utils.configmanager import get_system_list, get_system_value, load_env
from utils.selector import Selector
from utils.utils import (
areYouSure,
clear_screen,
colorText,
formatHTML,
get_sanitized_input,
locked,
open_directory,
print_x_wide,
regulator,
)
logger = logging.getLogger(__name__)
dotenv.load_dotenv()
def selectPolicies(api: AirlockAPIWrapper, allow_multiple=True) -> List[Policy]:
policies = [Policy(**row.to_dict()) for _, row in api.policy_find_all().iterrows()]
logger.debug("Prompting for Policies")
print(colorText("Please select policy/policies", "white"))
selected = Selector.select_objects(policies, allow_multiple, prompt_each=True)
if selected is None:
return []
# Normalize to always return a list
logger.debug("Returning {selected.dict}")
return selected if isinstance(selected, list) else [selected]
def selectAllowlists(
api: AirlockAPIWrapper, policy=all, allow_multiple=True
) -> List[Allowlist]:
if policy == "all":
allowlists = [
Allowlist(**row.to_dict()) for _, row in api.allowlist_find_all().iterrows()
]
else:
allowlists = [
Allowlist(**row.to_dict())
for _, row in api.policy_list_allowlists(policy[0].groupid).iterrows()
]
logger.debug("Prompting for Allowlist(s)")
print(colorText("Please select allowlist(s)", "white"))
selected = Selector.select_objects(allowlists, allow_multiple, prompt_each=True)
if selected is None:
return []
# Normalize to always return a list
logger.debug(f"Returning {selected}")
return selected if isinstance(selected, list) else [selected]
def sortHashes(
api: AirlockAPIWrapper, selected_policies: List[Policy], type=[1, 2, 6, 7]
):
working_dir = load_env("WORKING_DIR")
history_days = Selector.select_value(
prompt="Enter how many days of history to pull (1-365): ",
value_type=int,
valid_range=(1, 365),
)
logger.debug(f"{history_days} day selected for history")
if history_days is None:
logging.warning("No history range selected. Aborting.")
return
policy_executions = ExecutionHistoryRecord.from_policies(
api, selected_policies, type_=type, history_days=history_days
)
logger.debug(f"Executions contains {policy_executions}")
enriched_executions = ExecutionHistoryRecord.enrich_with_hashes(
api, policy_executions
)
categorized_executions = (
ExecutionHistoryRecord.categorize_executions_by_hash_decision(
enriched_executions
)
)
approved, unapproved, needs_review, unknown = (
ExecutionHistoryRecord.sort_by_hash_decision(categorized_executions)
)
categories = {
"needs_review": needs_review,
"approved": approved,
"unapproved": unapproved,
"leftover": unknown,
}
for label, records in categories.items():
if not records:
continue # Skip empty or falsy categories
csv_path = f"{working_dir}\\Needs_Review\\Review_First\\{selected_policies[0].name}_{label}_executions.csv"
html_path = f"{working_dir}\\Needs_Review\\HTML\\{selected_policies[0].name}_{label}.html"
# Convert ExecutionHistoryRecord objects to dictionaries
df = pd.DataFrame([r.__dict__ for r in records])
# Optional: flatten hash_obj if needed
if not df.empty and "hash_obj" in df.columns:
hash_df = df["hash_obj"].apply(lambda h: h.to_dict() if h else {})
df = pd.concat([df.drop(columns=["hash_obj"]), hash_df], axis=1)
# Save to CSV
df.to_csv(csv_path, index=False)
logger.info(f"Saved {label} executions to {csv_path}")
# Generate HTML
formatHTML(df, html_path)
logger.info(f"Generated HTML report at {html_path}")
def buildPathsandPublishers(selected_policies: List[Policy], split):
working_dir = load_env("WORKING_DIR")
df1 = pd.DataFrame()
df2 = pd.DataFrame()
all_approved_hashes = pd.DataFrame()
path1 = (
f"{working_dir}\\Approved\\{selected_policies[0].name}_approved_executions.csv"
)
path2 = f"{working_dir}\\Approved\\{selected_policies[0].name}_needs_review_executions.csv"
path_exclusion_constant = get_system_value("PATH_EXCLUSION_CONST", cast_type=int)
if os.path.exists(path1):
df1 = pd.read_csv(path1)
else:
logger.warning(f"File not found: {path1}")
if os.path.exists(path2):
df2 = pd.read_csv(path2)
else:
logger.warning(f"File not found: {path2}")
if df1.empty and df2.empty:
logger.warning("Both DataFrames are empty. Skipping sort.")
all_approved_hashes = pd.DataFrame()
logger.debug(all_approved_hashes.head)
else:
all_approved_hashes = pd.concat([df1, df2], ignore_index=True)
if "filename" in all_approved_hashes.columns:
all_approved_hashes = all_approved_hashes.sort_values(by="filename")
else:
logger.warning(
"Warning: 'filename' column not found in concatenated DataFrame."
)
if not all_approved_hashes.empty and path_exclusion_constant:
primary_path_exclusions = calculatePath(
all_approved_hashes,
path_exclusion_constant,
split,
)
remaining_hashes = all_approved_hashes[
~all_approved_hashes["sha256"].isin(primary_path_exclusions["sha256"])
]
secondary_path_exclusions = calculatePath(
remaining_hashes, (path_exclusion_constant - 1), split
)
remaining_hashes = remaining_hashes[
~remaining_hashes["sha256"].isin(secondary_path_exclusions["sha256"])
]
dataframes = {
"all_approved_hashes": all_approved_hashes,
"primary_Paths": primary_path_exclusions,
"secondary_Paths": secondary_path_exclusions,
"hashes_not_approvable_by_path": remaining_hashes,
}
logger.debug("Preparing to sort dataframes")
for name, df in dataframes.items():
logger.debug(f" DataFrame headers: {list(df.columns)}")
if "hashes" in name:
df.sort_values(by="filename", inplace=True)
else:
df.sort_values(by="longestcfp", inplace=True)
df.to_csv(
f"{working_dir}\\Needs_Review\\Review_Second\\{selected_policies[0].name}_{name}.csv",
index=False,
)
formatHTML(
df,
f"{working_dir}\\Needs_Review\\HTML\\{selected_policies[0].name}_{name}.html",
)
if not all_approved_hashes.empty:
# Drop all not signed, only keep unique values
publist = all_approved_hashes[
all_approved_hashes["publisher"] != "Not Signed"
].drop_duplicates(subset=["publisher"])
# Remove Bad publisher if somehow they made it this far
pattern = regulator(get_system_list("BAD_PUBLISHERS"))
publist = publist[~publist["publisher"].str.contains(pattern, na=False)]
publist = publist[["publisher"]]
publist.sort_values(by="publisher", inplace=True)
publist.to_csv(
f"{working_dir}\\Needs_Review\\Review_Second\\{selected_policies[0].name}_publishers.csv",
index=False,
)
else:
logger.debug("Approved Hashes list appears empty")
def buildPreflights(selected_policies: List[Policy]):
working_dir = load_env("WORKING_DIR")
df1 = pd.DataFrame()
df2 = pd.DataFrame()
approved_hashes = pd.DataFrame()
approved_publishers = pd.DataFrame()
hash = f"{working_dir}\\Needs_Review\\Review_Second\\{selected_policies[0].name}_all_approved_hashes.csv"
path1 = f"{working_dir}\\Approved\\{selected_policies[0].name}_primary_Paths.csv"
path2 = f"{working_dir}\\Approved\\{selected_policies[0].name}_secondary_Paths.csv"
publishers = f"{working_dir}\\Approved\\{selected_policies[0].name}_publishers.csv"
# Read in and combine the two path generations
if os.path.exists(path1):
df1 = pd.read_csv(path1)
else:
logger.warning(f"File not found: {path1}")
if os.path.exists(path2):
df2 = pd.read_csv(path2)
else:
logger.warning(f"File not found: {path2}")
if df1.empty and df2.empty:
logger.warning("Both DataFrames are empty. Skipping sort.")
approved_paths = pd.DataFrame()
else:
approved_paths = pd.concat([df1, df2], ignore_index=True)
approved_paths = approved_paths.drop_duplicates(subset="longestcfp", keep="first")
# We create a list of hashes that are left over if we exclude the ones that are covered by the path exclusions.
if os.path.exists(hash):
hashes = pd.read_csv(hash)
approved_hashes = hashes[~hashes["filename"].isin(approved_paths["longestcfp"])]
approved_hashes = approved_hashes.drop_duplicates(subset="sha256", keep="first")
else:
logger.warning(f"File not found: {hash}")
if os.path.exists(publishers):
approved_publishers = pd.read_csv(publishers)
else:
logger.warning(f"File not found: {publishers}")
dataframes = {
"approved_paths": approved_paths,
"approved_hashes": approved_hashes,
"approved_publishers": approved_publishers,
}
for name, df in dataframes.items():
logger.debug(f" DataFrame headers: {list(df.columns)}")
if name == "approved_paths":
df.sort_values(by="longestcfp", inplace=True)
elif name == "approved_hashes":
df.sort_values(by="filename", inplace=True)
elif name == "approved_publishers":
df.sort_values(by="publisher", inplace=True)
df.to_csv(
f"{working_dir}\\Preflight\\{selected_policies[0].name}_{name}.csv",
index=False,
)
formatHTML(
df,
f"{working_dir}\\Preflight\\HTML\\{selected_policies[0].name}_{name}.html",
)
def splitFilepathsGrouped(df, path_exclusion_constant, col="filename"):
min_files_for_path = get_system_value("MIN_FILES_FOR_PATH", cast_type=int)
def clean_split(path):
if not isinstance(path, (str, bytes, os.PathLike)):
return []
parts = str(os.path.normpath(path)).split(os.sep)
parts = [p for p in parts if p] # Remove empty strings
return parts
# Diagnostic: log any non-string entries
non_string_entries = df[
~df[col].apply(lambda x: isinstance(x, (str, bytes, os.PathLike)))
]
if not non_string_entries.empty:
print(f"[WARNING] Non-string entries found in column '{col}':")
print(non_string_entries)
df = df.copy()
split_paths = df[col].apply(clean_split)
if min_files_for_path is not None:
df = df[
split_paths.apply(lambda parts: len(parts) >= min_files_for_path)
].copy()
split_paths = split_paths[df.index]
df["group_key"] = split_paths.apply(
lambda parts: os.sep.join(parts[:path_exclusion_constant])
)
grouped = df.groupby("group_key")
new_rows = []
for _, group_df in grouped:
paths = group_df[col].tolist()
split_parts = [clean_split(p) for p in paths]
def longest_common_prefix(paths):
if not paths:
return []
prefix = paths[0]
for path in paths[1:]:
prefix = [a for a, b in zip(prefix, path) if a == b]
if not prefix:
break
return prefix
common_prefix = longest_common_prefix(split_parts)
prefix_str = os.sep.join(common_prefix)
for i, parts in enumerate(split_parts):
filename = parts[-1]
middle = (
os.sep.join(parts[len(common_prefix) : -1])
if len(parts) > len(common_prefix) + 1
else ""
)
row = group_df.iloc[i].copy()
row["longestcfp"] = prefix_str
row["middle"] = middle
row["filename_only"] = filename
row["file_extension"] = os.path.splitext(filename)[1].lower()
new_rows.append(row)
return pd.DataFrame(new_rows).drop(columns=["group_key"])
def calculatePath(approved_hashes, path_exclusion_constant, split):
if split:
dfs_by_policy = [group for _, group in approved_hashes.groupby("policy")]
else:
dfs_by_policy = [approved_hashes]
badpathparts = get_system_list("BAD_PATH_PARTS")
min_files_for_path = get_system_value("MIN_FILES_FOR_PATH", cast_type=int)
processed_dfs = []
for df in dfs_by_policy:
haslcp = splitFilepathsGrouped(df, path_exclusion_constant, "filename")
haslcp = haslcp.drop_duplicates()
forbidden = regulator(badpathparts, True)
forbidden_lcfp = haslcp["longestcfp"].str.contains(forbidden, na=False)
logger.debug("Removing forbidden filepaths for path exceptions")
print(colorText("Removing forbidden filepaths for path exceptions", "green"))
lcp_not_forbidden = haslcp[~forbidden_lcfp].copy()
lcp_not_forbidden_review = lcp_not_forbidden[
[
"policyname",
"longestcfp",
"middle",
"filename_only",
"file_extension",
"sha256",
]
]
unique_sha_counts = (
lcp_not_forbidden_review.groupby("longestcfp")["sha256"]
.nunique()
.reset_index()
)
unique_sha_counts.columns = ["longestcfp", "unique_sha256_count"]
lcp_not_forbidden_review = lcp_not_forbidden_review.merge(
unique_sha_counts, on="longestcfp", how="left"
)
lcp_not_forbidden_review = lcp_not_forbidden_review[
lcp_not_forbidden_review["unique_sha256_count"] >= min_files_for_path
]
processed_dfs.append(lcp_not_forbidden_review)
pathExclusions = pd.concat(processed_dfs, ignore_index=True)
return pathExclusions
def testChange(selected_policies, destination_policy, destination_allowlist):
working_dir = load_env("WORKING_DIR")
logger.info("These path exclusions would be added to:")
logger.info(destination_policy)
pathexclusions = pd.read_csv(
f"{working_dir}\\Preflight\\{selected_policies[0].name}_approved_paths.csv"
)
hashes = pd.read_csv(
f"{working_dir}\\Preflight\\{selected_policies[0].name}_approved_hashes.csv"
)
unique_combinations = pathexclusions[
["longestcfp", "file_extension"]
].drop_duplicates()
drive_letter_pattern = re.compile(r"^[a-zA-Z]:\\")
processed_paths = [
(path if drive_letter_pattern.match(path) else f"\\\\{path}") + f"\\**{ext}"
for path, ext in unique_combinations.itertuples(index=False, name=None)
]
for path in processed_paths:
logger.info(path)
print(colorText("These publishers would added", "yellow"))
processed_publishers = []
if os.path.exists(
f"{working_dir}\\Preflight\\{selected_policies[0].name}_approved_publishers.csv"
):
publishers = pd.read_csv(
f"{working_dir}\\Preflight\\{selected_policies[0].name}_approved_publishers.csv"
)
if publishers.empty:
print(colorText("The publishers list is empty.", "red"))
else:
processed_publishers = (
publishers[publishers["publisher"] != "Not Signed"]["publisher"]
.drop_duplicates()
.tolist()
)
for publisher in processed_publishers:
print(publisher)
print(colorText("These hashes would be added to:", "yellow"))
print(destination_allowlist)
processed_hashes = hashes["sha256"].unique().tolist()
print_x_wide(processed_hashes, 3)
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"))
TUI/Widgets/settingswidget.py
+567
View File
@@ -0,0 +1,567 @@
# 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/>.
"""
Settings widget combining theme selection and update checking.
"""
import logging
import webbrowser
from textual.containers import Horizontal, Vertical, VerticalScroll
from textual.message import Message
from textual.widget import Widget
from textual.widgets import Button, Rule, Static
from utils.versionchecker import (
RELEASES_PAGE_URL,
UpdateCheckResult,
check_for_updates,
get_current_version,
get_version_checker,
)
logger = logging.getLogger(__name__)
class SettingsWidget(Widget):
"""Widget for application settings including themes and updates."""
DEFAULT_CSS = """
SettingsWidget {
height: 1fr;
}
/* Update section buttons - add margin between them */
#update_buttons Button {
margin-right: 1;
}
/* Theme buttons - consistent width within columns, slightly smaller */
.theme_btn {
width: 100%;
margin-bottom: 1;
}
/* Column headers */
.theme_column_header {
text-align: center;
text-style: bold;
margin-bottom: 1;
}
/* Section titles */
.settings_section_title {
text-style: bold;
margin-bottom: 1;
}
/* Theme columns - reduce overall width */
#theme_columns {
width: 80%;
}
/* Theme columns spacing */
#dark_themes_col1, #dark_themes_col2 {
margin-right: 1;
}
#light_themes_col {
margin-left: 1;
}
"""
class ThemeSelected(Message):
"""Message posted when a theme is selected."""
def __init__(self, theme_name: str):
super().__init__()
self.theme_name = theme_name
# Dark themes - Column 1
DARK_THEMES_COL1 = [
("Textual Dark", "textual-dark"),
("Nord", "nord"),
("Gruvbox", "gruvbox"),
("Dracula", "dracula"),
]
# Dark themes - Column 2
DARK_THEMES_COL2 = [
("Catppuccin Mocha", "catppuccin-mocha"),
("Tokyo Night", "tokyo-night"),
("Monokai", "monokai"),
]
# Light themes (third column)
LIGHT_THEMES = [
("Textual Light", "textual-light"),
("Flexoki", "flexoki"),
("Catppuccin Latte", "catppuccin-latte"),
("Solarized Light", "solarized-light"),
]
# Combined for backward compatibility
DARK_THEMES = DARK_THEMES_COL1 + DARK_THEMES_COL2
AVAILABLE_THEMES = DARK_THEMES + LIGHT_THEMES
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._update_result: UpdateCheckResult | None = None
self._checking = False
def compose(self):
# Wrap everything in a scrollable container with auto height children
with VerticalScroll(id="settings_scroll"):
# Version & Updates Section
with Vertical(id="updates_section") as updates:
updates.styles.height = "auto"
yield Static(
"📦 Version & Updates",
id="updates_title",
classes="settings_section_title",
)
version_text = f"Current Version: v{get_current_version()}"
yield Static(version_text, id="current_version")
with Horizontal(id="update_buttons") as btn_row:
btn_row.styles.height = "auto"
yield Button("🔍 Check for Updates", id="check_updates_btn")
yield Button("📥 View Releases", id="view_releases_btn")
yield Static("", id="update_status")
yield Rule()
# Theme Section - Three columns: Dark 1, Dark 2, Light
with Vertical(id="themes_section") as themes:
themes.styles.height = "auto"
yield Static(
"🎨 Theme Options",
id="theme_title",
classes="settings_section_title",
)
with Horizontal(id="theme_columns") as cols:
cols.styles.height = "auto"
# Dark themes section (2 columns under one header)
with Vertical(id="dark_themes_section") as dark_section:
dark_section.styles.width = "2fr"
dark_section.styles.height = "auto"
yield Static(
"🌙 Dark Themes",
classes="theme_column_header",
id="dark_header",
)
with Horizontal(id="dark_columns") as dark_cols:
dark_cols.styles.height = "auto"
# Dark themes column 1
with Vertical(id="dark_themes_col1") as dark_col1:
dark_col1.styles.width = "1fr"
dark_col1.styles.height = "auto"
for label, btn_id in self.DARK_THEMES_COL1:
yield Button(
label,
id=f"set_theme_{btn_id}",
classes="theme_btn",
)
# Dark themes column 2
with Vertical(id="dark_themes_col2") as dark_col2:
dark_col2.styles.width = "1fr"
dark_col2.styles.height = "auto"
for label, btn_id in self.DARK_THEMES_COL2:
yield Button(
label,
id=f"set_theme_{btn_id}",
classes="theme_btn",
)
# Light themes column
with Vertical(id="light_themes_col") as light_col:
light_col.styles.width = "1fr"
light_col.styles.height = "auto"
yield Static("☀️ Light Themes", classes="theme_column_header")
for label, btn_id in self.LIGHT_THEMES:
yield Button(
label, id=f"set_theme_{btn_id}", classes="theme_btn"
)
def on_mount(self) -> None:
"""Check for cached update result on mount."""
checker = get_version_checker()
cached_result = checker.get_last_result()
if cached_result and cached_result.update_available:
self._update_result = cached_result
self._show_update_available(cached_result)
def on_button_pressed(self, event: Button.Pressed) -> None:
button_id = event.button.id
if button_id == "check_updates_btn":
self._check_for_updates()
event.stop()
elif button_id == "view_releases_btn":
self._open_releases_page()
event.stop()
elif button_id == "download_update_btn":
self._download_update()
event.stop()
elif button_id == "dismiss_update_btn":
self._dismiss_update()
event.stop()
elif button_id and button_id.startswith("set_theme_"):
theme_name = button_id.replace("set_theme_", "")
self.post_message(self.ThemeSelected(theme_name))
event.stop()
def _check_for_updates(self) -> None:
"""Check for updates and update UI."""
if self._checking:
return
self._checking = True
status = self.query_one("#update_status", Static)
check_btn = self.query_one("#check_updates_btn", Button)
# Show checking status
check_btn.disabled = True
check_btn.label = "⏳ Checking..."
status.update("🔄 Checking for updates...")
# Run check in worker to avoid blocking UI
self.run_worker(self._do_update_check, exclusive=True)
async def _do_update_check(self) -> None:
"""Worker to perform update check."""
try:
result = check_for_updates()
self._update_result = result
# Since we're in an async worker (not a thread), we can call directly
self._update_check_complete(result)
except Exception as e:
logger.error(f"Update check failed: {e}")
self._update_check_failed(str(e))
finally:
self._checking = False
def _update_check_complete(self, result: UpdateCheckResult) -> None:
"""Handle completed update check."""
check_btn = self.query_one("#check_updates_btn", Button)
check_btn.disabled = False
check_btn.label = "🔍 Check for Updates"
if result.error:
self._update_check_failed(result.error)
return
if result.update_available:
self._show_update_available(result)
self.app.notify(
f"🆕 Update available: {result.latest_version}",
title="Update Available",
severity="information",
timeout=8,
)
else:
status = self.query_one("#update_status", Static)
status.update(f"✅ Loxide is up to date (v{result.current_version})")
self.app.notify(
"✅ Loxide is up to date!",
severity="information",
timeout=5,
)
def _update_check_failed(self, error: str) -> None:
"""Handle failed update check."""
check_btn = self.query_one("#check_updates_btn", Button)
check_btn.disabled = False
check_btn.label = "🔍 Check for Updates"
status = self.query_one("#update_status", Static)
status.update(f"⚠️ Could not check for updates: {error}")
def _show_update_available(self, result: UpdateCheckResult) -> None:
"""Show update available UI with release notes."""
status = self.query_one("#update_status", Static)
msg = f"🆕 New version available: {result.latest_version}\n"
msg += f" Current: v{result.current_version}"
if result.release_info and result.release_info.body:
# Show release notes (truncate if very long)
notes = result.release_info.body.strip()
# Limit to ~500 chars to avoid overwhelming the UI
if len(notes) > 500:
notes = notes[:500] + "\n..."
msg += f"\n\n📋 Release Notes:\n{notes}"
status.update(msg)
# Add download/dismiss buttons if not already there
try:
self.query_one("#download_update_btn")
except Exception:
# Buttons don't exist, add them
button_container = self.query_one("#update_buttons", Horizontal)
download_btn = Button(
"📥 Download Update", id="download_update_btn", variant="success"
)
dismiss_btn = Button(
"✖ Dismiss", id="dismiss_update_btn", variant="default"
)
button_container.mount(download_btn)
button_container.mount(dismiss_btn)
def _open_releases_page(self) -> None:
"""Open the releases page in browser."""
try:
webbrowser.open(RELEASES_PAGE_URL)
self.app.notify("📂 Opened releases page in browser", timeout=3)
except Exception as e:
logger.error(f"Could not open browser: {e}")
self.app.notify(f"⚠️ Could not open browser: {e}", severity="warning")
def _download_update(self) -> None:
"""Download the update exe file."""
import os
from pathlib import Path
if not self._update_result or not self._update_result.release_info:
self.app.notify("⚠️ No update information available", severity="warning")
return
download_url = self._update_result.release_info.download_url
if not download_url:
# Fall back to opening the release page
url = self._update_result.release_info.html_url
try:
webbrowser.open(url)
self.app.notify(
"📥 Opened download page in browser (no direct download available)",
timeout=5,
)
except Exception as e:
logger.error(f"Could not open browser: {e}")
self.app.notify(f"⚠️ Could not open browser: {e}", severity="warning")
return
# Determine destination path
if os.name == "nt": # Windows
downloads_dir = Path.home() / "Downloads"
else:
downloads_dir = Path.home() / "Downloads"
if not downloads_dir.exists():
downloads_dir = Path.home()
# Extract filename from URL
filename = download_url.split("/")[-1]
if not filename.endswith(".exe"):
filename = f"Loxide_{self._update_result.latest_version}.exe"
dest_path = downloads_dir / filename
# Disable the button while downloading
try:
btn = self.query_one("#download_update_btn", Button)
btn.disabled = True
btn.label = "⏳ Downloading..."
except Exception:
pass
self.app.notify(
f"📥 Downloading to:\n{dest_path}", title="Download Starting", timeout=5
)
# Small delay so user sees the "downloading to" toast before download completes
self.set_timer(
0.5,
lambda: self.run_worker(
self._do_download(download_url, dest_path), exclusive=True
),
)
async def _do_download(self, download_url: str, dest_path) -> None:
"""Worker to download the update file."""
try:
# Download the file
import requests
response = requests.get(download_url, stream=True, timeout=60)
response.raise_for_status()
with open(dest_path, "wb") as f:
for chunk in response.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)
# Success
self.app.notify(
f"✅ Downloaded to:\n{dest_path}",
title="Download Complete",
severity="information",
timeout=10,
)
logger.info(f"Update downloaded to {dest_path}")
# Re-enable button
try:
btn = self.query_one("#download_update_btn", Button)
btn.disabled = False
btn.label = "📥 Download Again"
except Exception:
pass
except Exception as e:
logger.error(f"Download failed: {e}")
self.app.notify(f"❌ Download failed: {e}", severity="error", timeout=10)
# Re-enable button
try:
btn = self.query_one("#download_update_btn", Button)
btn.disabled = False
btn.label = "📥 Download Update"
except Exception:
pass
def _dismiss_update(self) -> None:
"""Dismiss the current update notification."""
if self._update_result and self._update_result.latest_version:
checker = get_version_checker()
checker.dismiss_update(self._update_result.latest_version)
# Remove the extra buttons
try:
self.query_one("#download_update_btn").remove()
self.query_one("#dismiss_update_btn").remove()
except Exception:
pass
status = self.query_one("#update_status", Static)
status.update(f"✓ Dismissed update {self._update_result.latest_version}")
self._update_result = None
# Keep ThemeSelector as a standalone for backward compatibility
class ThemeSelector(Widget):
"""Widget for selecting and applying Textual themes.
DEPRECATED: Use SettingsWidget instead for combined settings UI.
"""
DEFAULT_CSS = """
ThemeSelector {
height: 1fr;
}
/* Theme buttons - consistent width within columns */
.theme_btn {
width: 100%;
margin-bottom: 1;
}
/* Column headers */
.theme_column_header {
text-align: center;
text-style: bold;
margin-bottom: 1;
}
/* Theme columns - reduce overall width */
#theme_columns {
width: 80%;
}
/* Theme columns spacing */
#dark_themes_col1, #dark_themes_col2 {
margin-right: 1;
}
#light_themes_col {
margin-left: 1;
}
"""
class ThemeSelected(Message):
"""Message posted when a theme is selected."""
def __init__(self, theme_name: str):
super().__init__()
self.theme_name = theme_name
DARK_THEMES_COL1 = SettingsWidget.DARK_THEMES_COL1
DARK_THEMES_COL2 = SettingsWidget.DARK_THEMES_COL2
DARK_THEMES = SettingsWidget.DARK_THEMES
LIGHT_THEMES = SettingsWidget.LIGHT_THEMES
AVAILABLE_THEMES = SettingsWidget.AVAILABLE_THEMES
def compose(self):
with VerticalScroll(id="theme_scroll"):
yield Static("Theme Options", id="theme_title")
with Horizontal(id="theme_columns") as cols:
cols.styles.height = "auto"
# Dark themes section (2 columns under one header)
with Vertical(id="dark_themes_section") as dark_section:
dark_section.styles.width = "2fr"
dark_section.styles.height = "auto"
yield Static(
"🌙 Dark Themes",
classes="theme_column_header",
id="dark_header",
)
with Horizontal(id="dark_columns") as dark_cols:
dark_cols.styles.height = "auto"
# Dark themes column 1
with Vertical(id="dark_themes_col1") as dark_col1:
dark_col1.styles.width = "1fr"
dark_col1.styles.height = "auto"
for label, btn_id in self.DARK_THEMES_COL1:
yield Button(
label, id=f"set_theme_{btn_id}", classes="theme_btn"
)
# Dark themes column 2
with Vertical(id="dark_themes_col2") as dark_col2:
dark_col2.styles.width = "1fr"
dark_col2.styles.height = "auto"
for label, btn_id in self.DARK_THEMES_COL2:
yield Button(
label, id=f"set_theme_{btn_id}", classes="theme_btn"
)
# Light themes column
with Vertical(id="light_themes_col") as light_col:
light_col.styles.width = "1fr"
light_col.styles.height = "auto"
yield Static("☀️ Light Themes", classes="theme_column_header")
for label, btn_id in self.LIGHT_THEMES:
yield Button(
label, id=f"set_theme_{btn_id}", classes="theme_btn"
)
def on_button_pressed(self, event: Button.Pressed) -> None:
button_id = event.button.id
if button_id and button_id.startswith("set_theme_"):
theme_name = button_id.replace("set_theme_", "")
self.post_message(self.ThemeSelected(theme_name))
TUI/Widgets/themeselector.py
+60
View File
@@ -0,0 +1,60 @@
# 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/>.
from textual.containers import Vertical
from textual.message import Message
from textual.widget import Widget
from textual.widgets import Button, Static
class ThemeSelector(Widget):
"""Widget for selecting and applying Textual themes."""
class ThemeSelected(Message):
"""Message posted when a theme is selected."""
def __init__(self, theme_name: str):
super().__init__()
self.theme_name = theme_name
AVAILABLE_THEMES = [
("Textual Dark", "textual-dark"),
("Textual Light", "textual-light"),
("Nord", "nord"),
("Gruvbox", "gruvbox"),
("Catppuccin Mocha", "catppuccin-mocha"),
("Dracula", "dracula"),
("Tokyo Night", "tokyo-night"),
("Monokai", "monokai"),
("Flexoki", "flexoki"),
("Catppuccin Latte", "catppuccin-latte"),
("Solarized Light", "solarized-light"),
("Retro Terminal", "retro-terminal"),
("Amber Terminal", "amber-terminal"), # your custom theme
]
def compose(self):
yield Static("Theme Options", id="theme_title")
with Vertical() as column:
column.styles.width = "1fr"
column.styles.height = "auto"
for label, btn_id in self.AVAILABLE_THEMES:
yield Button(label, id=f"set_theme_{btn_id}", compact=True)
def on_button_pressed(self, event: Button.Pressed) -> None:
button_id = event.button.id
if button_id and button_id.startswith("set_theme_"):
theme_name = button_id.replace("set_theme_", "")
self.post_message(self.ThemeSelected(theme_name))