diff --git a/TUI/TUI.py b/TUI/TUI.py
index 990cde3..2c5d4fc 100644
--- a/TUI/TUI.py
+++ b/TUI/TUI.py
@@ -4,7 +4,6 @@ import sys
from typing import Optional
import dotenv
-from dotenv import set_key
from textual.app import App, ComposeResult
from textual.containers import Vertical
from textual.message import Message
@@ -38,8 +37,8 @@ from TUI.resultsdisplay import ResultsDisplay
from TUI.theme_amber_terminal import get_amber_terminal_theme
from TUI.theme_retro_terminal import get_retro_terminal_theme
from TUI.themeselector import ThemeSelector
-from utils.configmanager import load_env
-from utils.setup import get_base_directory, load_user_config
+from utils.configmanager import get_user_value, load_env, save_user_config
+from utils.setup import get_base_directory
from utils.utils import open_directory
dotenv.load_dotenv()
@@ -58,47 +57,17 @@ logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
def _persist_user_theme(theme_name: str) -> None:
"""
- Store the chosen Textual theme in the user's config:
- /config/user_config.json
- and also mirror to /.env so load_env(...) sees it.
+ Store the chosen Textual theme in the user's config using the config manager.
+ No need to touch .env - config manager handles everything.
"""
base_dir = get_base_directory()
config_dir = base_dir / "config"
- user_config_path = config_dir / "user_config.json"
- env_path = base_dir / ".env"
- # ensure dirs / files exist similarly to setup()
- config_dir.mkdir(parents=True, exist_ok=True)
- if not user_config_path.exists():
- # minimal default like your load_user_config does
- user_config_path.write_text(
- '{"URL": "", "LOG_LEVEL": "INFO"}\n', encoding="utf-8"
- )
-
- # load existing user config
- user_conf = load_user_config(config_dir)
- user_conf["TEXTUAL_THEME"] = theme_name
-
- # write it back
- user_config_path.write_text(
- # pretty print so it stays human-readable
- __import__("json").dumps(user_conf, indent=4),
- encoding="utf-8",
- )
- logger.debug("Updated user_config.json with TEXTUAL_THEME=%s", theme_name)
-
- # mirror to .env (like setup.write_config_to_env does)
- env_path.parent.mkdir(parents=True, exist_ok=True)
- if not env_path.exists():
- env_path.touch()
try:
- set_key(str(env_path), "TEXTUAL_THEME", theme_name)
- except Exception as exc: # keep going even if .env write fails
- logger.warning("Failed to mirror TEXTUAL_THEME to .env: %s", exc)
-
- # reload so load_env(...) sees the new value right now
- dotenv.load_dotenv(dotenv_path=env_path, override=True)
- logger.debug("Reloaded .env from %s", env_path)
+ save_user_config(config_dir, {"TEXTUAL_THEME": theme_name})
+ logger.debug("Updated user config with TEXTUAL_THEME=%s", theme_name)
+ except Exception as exc:
+ logger.error("Failed to save TEXTUAL_THEME: %s", exc)
# ---------------------------------------------------------------------------
@@ -115,7 +84,7 @@ class MainMenuScreen(Screen):
"move_agent_workflow_button",
),
("📊 - Review and appove OTP Activities", "otp_activities_button"),
- ("🔇 - Find and Move Quiet Hosts to Enforcement", "find_quiet_button"),
+ ("📇 - Find and Move Quiet Hosts to Enforcement", "find_quiet_button"),
],
"policy": [
("🔒 - Prepare Policy For Enforcement", "policy_prep_button"),
@@ -126,7 +95,7 @@ class MainMenuScreen(Screen):
def __init__(self) -> None:
super().__init__()
- self.extras = load_env("EXTRAS")
+ self.extras = get_user_value("EXTRAS", str, "NOTTODAY")
wd = load_env("WORKING_DIR") or os.getcwd()
if not os.path.isdir(wd):
wd = os.getcwd()
@@ -380,10 +349,11 @@ class Loxide(App[Message]):
BINDINGS = [
("q", "quit", "Quit"),
("f", "open_fe", "Launch Explorer"),
+ ("r", "refresh", "Refresh"),
]
def __init__(self, api: AirlockAPIWrapper):
- self._textual_theme = load_env("TEXTUAL_THEME") or "nord"
+ self._textual_theme = get_user_value("TEXTUAL_THEME", str, "nord")
super().__init__()
self.api = api
wd = load_env("WORKING_DIR") or os.getcwd()
@@ -421,6 +391,9 @@ class Loxide(App[Message]):
self.theme = self._textual_theme
self.push_screen(MainMenuScreen())
+ def action_refresh(self) -> None:
+ self.refresh_data()
+
def action_quit(self) -> None:
global _PENDING_JOB
_PENDING_JOB = None
diff --git a/TUI/agentmoveoperations.py b/TUI/agentmoveoperations.py
index d3615f0..26032a0 100644
--- a/TUI/agentmoveoperations.py
+++ b/TUI/agentmoveoperations.py
@@ -183,21 +183,21 @@ class AgentMoveOperations(Widget):
f"Operation: {operation_name}",
f"{'=' * 50}",
"",
- f"✅ Successful ({len(successful)}):",
+ f"✅ Successful ({len(successful)}):",
]
if successful:
for agent, result in successful:
- results_lines.append(f" ✅ {agent.hostname}")
+ results_lines.append(f" ✅ {agent.hostname}")
else:
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)")
@@ -232,9 +232,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)
@@ -269,32 +269,32 @@ 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)
@@ -305,7 +305,7 @@ class AgentMoveOperations(Widget):
status_label.styles.margin = (2, 0, 0, 0)
yield status_label
- back_button = Button("← Back", id="back_button")
+ back_button = Button("↠Back", id="back_button")
back_button.styles.width = "50%"
back_button.styles.margin = (0, 1, 1, 0)
yield back_button
@@ -369,17 +369,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()
@@ -427,7 +427,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
@@ -462,12 +462,12 @@ 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
self.operation_in_progress = False
- status_label.update("✅ Operation complete!")
+ status_label.update("✅ Operation complete!")
# Display results in the widget
self._display_results("Local Approval Mode", successful, unsuccessful)
@@ -511,9 +511,9 @@ class AgentMoveOperations(Widget):
file_path = os.path.join(str(path), filename)
df.to_csv(file_path, index=False)
successful.append(file_path)
- status_label.update(f"✅ Exported to {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
@@ -557,7 +557,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
@@ -567,9 +567,9 @@ class AgentMoveOperations(Widget):
try:
from services.agenthandler import moveAgentToRelatedPolicy
- from utils.configmanager import get_protected_json
+ from utils.configmanager import get_system_json
- policy_relationship_map = get_protected_json("POLICY_MAP_ENF_AUD", "{}")
+ policy_relationship_map = get_system_json("POLICY_MAP_ENF_AUD", "{}")
for agent in self.agents:
try:
@@ -593,12 +593,12 @@ 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
self.operation_in_progress = False
- status_label.update("✅ Operation complete!")
+ status_label.update("✅ Operation complete!")
# Display results in the widget
self._display_results("Toggle Audit/Enforcement", successful, unsuccessful)
@@ -663,7 +663,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")
diff --git a/airlock_libs/airlock_libs.pyi b/airlock_libs/airlock_libs.pyi
index 84bee85..137dc92 100644
--- a/airlock_libs/airlock_libs.pyi
+++ b/airlock_libs/airlock_libs.pyi
@@ -30,7 +30,7 @@ def history_logging(
checkpoint_number: str,
policy_names: str,
) -> List[Dict[str, Any]]:
- """
+ """
Query execution history logs from the Airlock API.
Parameters
diff --git a/default_system_config.json b/default_system_config.json
index decfd79..2c003af 100644
--- a/default_system_config.json
+++ b/default_system_config.json
@@ -2,15 +2,38 @@
"APPNAME": "Loxide",
"URL": "https://server:3129",
"LOG_LEVEL": "INFO",
- "BAD_PATH_PARTS": ["users","wwwroot","windows\\temp","windows\\task","windows\\system32","startup", "windows\\fonts","Recycle.Bin","AppData","programdata", "Solarwinds","kaseya"],
- "BAD_PUBLISHERS": ["Brave", "Zoom", "GlavSoft", "VNC"],
- "PUPS":["logmein","invalid","nmap","LTSvc","VNC","Kaseya","Solarwinds","mRemoteNG"],
+ "BAD_PATH_PARTS": [
+ "users",
+ "wwwroot",
+ "windows\\temp",
+ "windows\\task",
+ "windows\\system32",
+ "startup",
+ "windows\\fonts",
+ "Recycle.Bin",
+ "AppData",
+ "programdata",
+ "Solarwinds",
+ "kaseya"
+ ],
+ "BAD_PUBLISHERS": [
+ "Brave",
+ "Zoom",
+ "GlavSoft",
+ "VNC"
+ ],
+ "PUPS": [
+ "logmein",
+ "invalid",
+ "nmap",
+ "LTSvc",
+ "VNC",
+ "Kaseya",
+ "Solarwinds",
+ "mRemoteNG"
+ ],
"PATH_EXCLUSION_CONST": 4,
"MIN_FILES_FOR_PATH": 4,
"VT_THREAT_TOLERANCE": 4,
- "TELEMETRY": "FALSE",
- "TELEM_URL": "",
- "POLICY_MAP_ENF_AUD": {
-
- }
+ "POLICY_MAP_ENF_AUD": {}
}
\ No newline at end of file
diff --git a/flows/localApproval.py b/flows/localApproval.py
index da89c8c..f1054f1 100644
--- a/flows/localApproval.py
+++ b/flows/localApproval.py
@@ -10,7 +10,7 @@ from typing import List, Optional
from models.agent import Agent
from services.agenthandler import moveAgentToRelatedPolicy, selectAgents
from services.API import AirlockAPIWrapper
-from utils.configmanager import get_protected_json
+from utils.configmanager import get_system_json
from utils.utils import colorText, get_sanitized_input
logger = logging.getLogger(__name__)
@@ -28,7 +28,7 @@ class LocalApprovalRequestor:
username: Username creating the approvals (for tracking)
"""
self.api = api
- self.policy_map = get_protected_json("POLICY_MAP_ENF_AUD", "{}")
+ self.policy_map = get_system_json("POLICY_MAP_ENF_AUD", "{}")
self.username = (
username or os.getenv("USERNAME") or os.getenv("USER") or "unknown"
)
@@ -51,7 +51,7 @@ class LocalApprovalRequestor:
batch_id = int(time.time())
purpose = (
- f"🎫 Local Approval 🎫 - {duration_minutes} mins - "
+ f"🎫 Local Approval 🎫 - {duration_minutes} mins - "
f"batch:{batch_id} Client:{agent_id} User:{self.username}"
)
@@ -103,10 +103,10 @@ class LocalApprovalRequestor:
success_count = 0
failure_count = 0
- print(colorText(f"\n📦 Processing batch {batch_id}...", "cyan"))
- print(colorText(f"👤 Requested by: {self.username}", "cyan"))
+ print(colorText(f"\n📦 Processing batch {batch_id}...", "cyan"))
+ print(colorText(f"👤 Requested by: {self.username}", "cyan"))
print(
- colorText(f"📊 Moving {len(agents)} agent(s) to local approval\n", "cyan")
+ colorText(f"📊 Moving {len(agents)} agent(s) to local approval\n", "cyan")
)
for agent in agents:
@@ -125,11 +125,11 @@ class LocalApprovalRequestor:
if not move_success:
raise Exception("Failed to move to audit policy")
- print(colorText(f"✓ {agent.hostname}", "green"))
+ print(colorText(f"✓ {agent.hostname}", "green"))
success_count += 1
except Exception as e:
- print(colorText(f"✗ {agent.hostname}: {e}", "red"))
+ print(colorText(f"✗ {agent.hostname}: {e}", "red"))
logger.error(f"Error processing agent {agent.hostname}: {e}")
failure_count += 1
@@ -152,7 +152,7 @@ class LocalApprovalRequestor:
]
# Display duration options
- print(colorText("\n⏱️ Select Local Approval Duration:", "white"))
+ print(colorText("\nâ±ï¸ Select Local Approval Duration:", "white"))
print(colorText("=" * 50, "white"))
for i, (minutes, label) in enumerate(duration_options, start=1):
@@ -166,36 +166,36 @@ class LocalApprovalRequestor:
if 1 <= choice <= len(duration_options):
duration_minutes, duration_label = duration_options[choice - 1]
- print(colorText(f"✓ Selected: {duration_label}", "green"))
+ print(colorText(f"✓ Selected: {duration_label}", "green"))
logger.info(f"User selected duration: {duration_minutes} minutes")
else:
- print(colorText("❌ Invalid choice.", "red"))
+ print(colorText("⌠Invalid choice.", "red"))
logger.warning("Invalid duration choice")
return
except ValueError:
- print(colorText("❌ Invalid input. Please enter a number.", "red"))
+ print(colorText("⌠Invalid input. Please enter a number.", "red"))
logger.warning("Invalid input for duration selection")
return
# Select agents
- print(colorText("\n🎯 Select Agents for Local Approval:", "white"))
+ print(colorText("\n🎯 Select Agents for Local Approval:", "white"))
agents = selectAgents(self.api)
if not agents:
- print(colorText("❌ No agents found or error retrieving agents.", "red"))
+ print(colorText("⌠No agents found or error retrieving agents.", "red"))
logger.warning("No agents selected or error retrieving agents")
return
# Confirm with user
- print(colorText("\n📋 Summary:", "cyan"))
+ print(colorText("\n📋 Summary:", "cyan"))
print(colorText(f" Duration: {duration_label}", "white"))
print(colorText(f" Agents: {len(agents)}", "white"))
confirm = get_sanitized_input("\nProceed? (y/n): ").lower()
if confirm != "y":
- print(colorText("❌ Operation cancelled.", "yellow"))
+ print(colorText("⌠Operation cancelled.", "yellow"))
return
# Process the batch
@@ -219,24 +219,25 @@ class LocalApprovalRequestor:
failure_count: Number of failed operations
"""
print(colorText(f"\n{'=' * 60}", "white"))
- print(colorText("📊 Local Approval Summary", "cyan"))
+ print(colorText("📊 Local Approval Summary", "cyan"))
print(colorText("=" * 60, "white"))
- print(colorText(f"✓ Successfully processed: {success_count}", "green"))
+ print(colorText(f"✓ Successfully processed: {success_count}", "green"))
if failure_count > 0:
- print(colorText(f"✗ Failed: {failure_count}", "red"))
+ print(colorText(f"✗ Failed: {failure_count}", "red"))
- print(colorText(f"\n📦 Batch ID: {batch_id}", "cyan"))
- print(colorText(f"⏱️ Duration: {duration_label}", "cyan"))
+ print(colorText(f"\n📦 Batch ID: {batch_id}", "cyan"))
+ print(colorText(f"â±ï¸ Duration: {duration_label}", "cyan"))
print(colorText("=" * 60, "white"))
- print(colorText("\n💡 Next Steps:", "yellow"))
- print(colorText(" • Agents have been moved to audit policies", "white"))
- print(colorText(" • Local approvals are active", "white"))
+ print(colorText("\n💡 Next Steps:", "yellow"))
+ print(colorText(" • Agents have been moved to audit policies", "white"))
+ print(colorText(" • Local approvals are active", "white"))
print(
colorText(
- f" • Agents will return to enforcement after {duration_label}", "white"
+ f" • Agents will return to enforcement after {duration_label}",
+ "white",
)
)
print(colorText("=" * 60 + "\n", "white"))
diff --git a/flows/prepPolicy.py b/flows/prepPolicy.py
index 6f4243c..53f7ede 100644
--- a/flows/prepPolicy.py
+++ b/flows/prepPolicy.py
@@ -25,7 +25,7 @@ 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_protected_value, load_env, load_env_json
+from utils.configmanager import get_system_list, get_system_value, load_env
from utils.selector import Selector
from utils.utils import (
areYouSure,
@@ -88,7 +88,7 @@ def sortHashes(
):
working_dir = load_env("WORKING_DIR")
history_days = Selector.select_value(
- prompt="Enter how many days of history to pull (1–150): ",
+ prompt="Enter how many days of history to pull (1–150): ",
value_type=int,
valid_range=(1, 150),
)
@@ -157,7 +157,7 @@ def buildPathsandPublishers(selected_policies: List[Policy], split):
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_protected_value("PATH_EXCLUSION_CONST", cast_type=int)
+ path_exclusion_constant = get_system_value("PATH_EXCLUSION_CONST", cast_type=int)
if os.path.exists(path1):
df1 = pd.read_csv(path1)
@@ -227,7 +227,7 @@ def buildPathsandPublishers(selected_policies: List[Policy], split):
all_approved_hashes["publisher"] != "Not Signed"
].drop_duplicates(subset=["publisher"])
# Remove Bad publisher if somehow they made it this far
- pattern = regulator(load_env_json("BAD_PUBLISHERS", "[]"))
+ 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)
@@ -313,7 +313,7 @@ def buildPreflights(selected_policies: List[Policy]):
def splitFilepathsGrouped(df, path_exclusion_constant, col="filename"):
- min_files_for_path = get_protected_value("MIN_FILES_FOR_PATH", cast_type=int)
+ 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)):
@@ -385,8 +385,8 @@ def calculatePath(approved_hashes, path_exclusion_constant, split):
else:
dfs_by_policy = [approved_hashes]
- badpathparts = load_env_json("BAD_PATH_PARTS", "[]")
- min_files_for_path = get_protected_value("MIN_FILES_FOR_PATH", cast_type=int)
+ badpathparts = get_system_list("BAD_PATH_PARTS")
+ min_files_for_path = get_system_value("MIN_FILES_FOR_PATH", cast_type=int)
processed_dfs = []
@@ -655,7 +655,7 @@ def section_header(title):
def printEnforceChecklist(selected_policies, destination_policy, destination_allowlist):
working_dir = load_env("WORKING_DIR")
- section_header("🛠️ 🔒 Prepare to Enforce Policy 🛠️ 🔒")
+ section_header("ðŸ› ï¸ ðŸ”’ Prepare to Enforce Policy ðŸ› ï¸ ðŸ”’")
print(
colorText(
"\nSequentially follow these steps to prepare a policy for enforcement:",
@@ -670,11 +670,11 @@ def printEnforceChecklist(selected_policies, destination_policy, destination_all
)
)
if not selected_policies:
- print(colorText(" [✗] No policies have been chosen", "red"))
+ 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"))
+ print(colorText(f" [✓] {policy.name}", "green"))
# Step 2: Destination Policy and Allowlist
print(
@@ -683,22 +683,22 @@ def printEnforceChecklist(selected_policies, destination_policy, destination_all
if destination_policy:
print(
colorText(
- f" [✓] {destination_policy[0].name} has been selected as the destination policy",
+ f" [✓] {destination_policy[0].name} has been selected as the destination policy",
"green",
)
)
else:
- print(colorText(" [✗] No destination policy has been chosen", "red"))
+ print(colorText(" [✗] No destination policy has been chosen", "red"))
if destination_allowlist:
print(
colorText(
- f" [✓] {destination_allowlist[0].name} has been selected as allowlist",
+ f" [✓] {destination_allowlist[0].name} has been selected as allowlist",
"green",
)
)
else:
- print(colorText(" [✗] No allowlist has been chosen", "red"))
+ print(colorText(" [✗] No allowlist has been chosen", "red"))
# Step 3: Data Preparation
print(
@@ -713,9 +713,9 @@ def printEnforceChecklist(selected_policies, destination_policy, destination_all
print(
colorText(
(
- " [✓] Data has been fetched"
+ " [✓] Data has been fetched"
if os.path.exists(review_path)
- else " [✗] Data has not been fetched"
+ else " [✗] Data has not been fetched"
),
"green" if os.path.exists(review_path) else "red",
)
@@ -723,7 +723,7 @@ def printEnforceChecklist(selected_policies, destination_policy, destination_all
else:
print(
colorText(
- " [✗] No policies selected, cannot check data fetch status", "red"
+ " [✗] No policies selected, cannot check data fetch status", "red"
)
)
@@ -756,9 +756,9 @@ def printEnforceChecklist(selected_policies, destination_policy, destination_all
print(
colorText(
(
- " [✓] Reviewed hashes have been loaded"
+ " [✓] Reviewed hashes have been loaded"
if os.path.exists(approved_path)
- else " [✗] Reviewed hashes have not been loaded"
+ else " [✗] Reviewed hashes have not been loaded"
),
"green" if os.path.exists(approved_path) else "red",
)
@@ -766,9 +766,9 @@ def printEnforceChecklist(selected_policies, destination_policy, destination_all
print(
colorText(
(
- " [✓] Path review list created"
+ " [✓] Path review list created"
if os.path.exists(second_review_path)
- else " [✗] Path review list has not been created"
+ else " [✗] Path review list has not been created"
),
"green" if os.path.exists(second_review_path) else "red",
)
@@ -776,7 +776,7 @@ def printEnforceChecklist(selected_policies, destination_policy, destination_all
else:
print(
colorText(
- " [✗] No policies selected, cannot check reviewed hashes or path list",
+ " [✗] No policies selected, cannot check reviewed hashes or path list",
"red",
)
)
@@ -812,9 +812,9 @@ def printEnforceChecklist(selected_policies, destination_policy, destination_all
print(
colorText(
(
- " [✓] Reviewed path list detected"
+ " [✓] Reviewed path list detected"
if os.path.exists(reviewed_path)
- else " [✗] Path review list has not been detected"
+ else " [✗] Path review list has not been detected"
),
"green" if os.path.exists(reviewed_path) else "red",
)
@@ -825,9 +825,9 @@ def printEnforceChecklist(selected_policies, destination_policy, destination_all
print(
colorText(
(
- " [✓] Preflight Path Exclusion List has been generated"
+ " [✓] Preflight Path Exclusion List has been generated"
if preflight_ready
- else " [✗] Preflight Path Exclusion List has not been generated"
+ else " [✗] Preflight Path Exclusion List has not been generated"
),
"green" if preflight_ready else "red",
)
@@ -835,7 +835,7 @@ def printEnforceChecklist(selected_policies, destination_policy, destination_all
else:
print(
colorText(
- " [✗] No policies selected, cannot check preflight status", "red"
+ " [✗] No policies selected, cannot check preflight status", "red"
)
)
@@ -866,5 +866,5 @@ def printEnforceChecklist(selected_policies, destination_policy, destination_all
print(colorText(" Apply approved hashes to allowlist", "cyan"))
# Utility Options
- print(colorText("F. 📂 - Open Working Directory", "cyan"))
- print(colorText("B. 🔚 - Back", "cyan"))
+ print(colorText("F. 📂 - Open Working Directory", "cyan"))
+ print(colorText("B. 🔚 - Back", "cyan"))
diff --git a/flows/quietAgent.py b/flows/quietAgent.py
deleted file mode 100644
index d723da7..0000000
--- a/flows/quietAgent.py
+++ /dev/null
@@ -1,134 +0,0 @@
-# Copyright (C) 2025 James Brotosky, Brandon Wickline
-#
-# This program is free software: you can redistribute it and/or modify
-# it under the terms of the GNU Affero General Public License as published
-# by the Free Software Foundation, either version 3 of the License, or
-# (at your option) any later version.
-#
-# This program is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-# GNU Affero General Public License for more details.
-#
-# You should have received a copy of the GNU Affero General Public License
-# along with this program. If not, see .
-
-import datetime
-import logging
-
-import dotenv
-import pandas as pd
-
-from flows.prepPolicy import selectPolicies
-from services.API import AirlockAPIWrapper
-from services.policyhandler import getPolicyInfo
-from utils.configmanager import load_env
-from utils.selector import Selector
-from utils.utils import colorText, get_sanitized_input
-
-logger = logging.getLogger(__name__)
-
-
-dotenv.load_dotenv()
-
-
-def findQuietAgents(api: AirlockAPIWrapper):
- working_dir = load_env("WORKING_DIR")
- # Get policy selection and agent list
- selected_policy = selectPolicies(api, False)
- if selected_policy:
- agents = api.agents_find_by_group(selected_policy[0].groupid)
-
- # Prompt user for history range
- history_days = Selector.select_value(
- prompt="Enter how many days of history to pull (1–150): ",
- value_type=int,
- valid_range=(1, 150),
- )
- required_quiet = Selector.select_value(
- prompt="Enter how many days without an untrusted execution before these are considered ready for enforcement? (1–365): ",
- value_type=int,
- valid_range=(1, 150),
- )
-
- confirm = Selector.confirm(
- f"Do you wish to proceed to pull history for {selected_policy[0].name}? Y/N : "
- )
- # Get execution history as a DataFrame
- if confirm:
- policy_exec_history = getPolicyInfo(
- api, selected_policy[0], [1, 2, 6, 7], history_days
- )
-
- if policy_exec_history.empty:
- logging.info(
- "No execution history found for the selected policy and time range."
- )
- get_sanitized_input("Press enter to continue")
- return
-
- # Convert 'datetime' column to timezone-aware datetime objects
- policy_exec_history["datetime"] = pd.to_datetime(
- policy_exec_history["datetime"], format="%Y-%m-%dT%H:%M:%SZ", utc=True
- )
-
- # Get current UTC time
- now = datetime.datetime.now(datetime.timezone.utc)
-
- # Calculate days ago
- policy_exec_history["days_ago"] = policy_exec_history["datetime"].apply(
- lambda dt: (now - dt).days
- )
-
- # Count total executions per hostname
- hostname_counts = policy_exec_history["hostname"].value_counts()
-
- # Map execution counts to agents
- agents["execution_count"] = (
- agents["hostname"].map(hostname_counts).fillna(0).astype(int)
- )
-
- # Find most recent execution per hostname
- most_recent_exec = policy_exec_history.sort_values(
- by="days_ago"
- ).drop_duplicates(subset="hostname", keep="first")
-
- # Map most recent execution age to agents
- agents["days_since"] = agents["hostname"].map(
- most_recent_exec.set_index("hostname")["days_ago"]
- )
-
- # Check for enforcement readiness
- agents["required_quiet"] = required_quiet
- agents["enforce_ready"] = agents["days_since"].apply(
- lambda x: True if pd.isna(x) or x > required_quiet else False
- )
-
- # Sort agents by execution count and hostname
- agents = agents.sort_values(
- by=["execution_count", "hostname"], ascending=[True, True]
- )
-
- # Save to CSV
- filename = f"{working_dir}\\{selected_policy[0].name}_agents_last_{history_days}_days.csv"
- logging.debug(f"Saving CSV to {filename}")
- print(colorText(f"Saving CSV to {filename}", "green"))
- agents.to_csv(filename, index=False)
-
- # Summary statistics
- total_agents = len(agents)
- ready_agents = agents["enforce_ready"].sum()
- not_ready_agents = total_agents - ready_agents
- ready_percentage = (ready_agents / total_agents) * 100
-
- # Print results
-
- message = (
- f"Total agents: {total_agents}\n"
- f"Agents marked as 'enforce_ready': {ready_agents}\n"
- f"Agents not ready: {not_ready_agents}\n"
- f"Percentage ready for enforcement: {ready_percentage:.2f}%"
- )
- logger.debug(message)
- colorText(message, "green")
- get_sanitized_input("Press enter to continue")
diff --git a/models/execution.py b/models/execution.py
index e1cf514..462193d 100644
--- a/models/execution.py
+++ b/models/execution.py
@@ -28,7 +28,7 @@ import pandas as pd
import airlock_libs
from services.API import AirlockAPIWrapper
-from utils.configmanager import get_protected_value, load_env_json
+from utils.configmanager import get_system_list, get_system_value
from utils.utils import colorText, regulator
logger = logging.getLogger(__name__)
@@ -90,9 +90,9 @@ class Hash:
@classmethod
def categorize_hashes(cls, hashes):
- threat_tolerance = get_protected_value("VT_THREAT_TOLERANCE", cast_type=int)
- bad_publishers_pattern = regulator(load_env_json("BAD_PUBLISHERS", "[]"))
- pups_pattern = regulator(load_env_json("PUPS", "[]"))
+ threat_tolerance = get_system_value("VT_THREAT_TOLERANCE", cast_type=int)
+ bad_publishers_pattern = regulator(get_system_list("BAD_PUBLISHERS"))
+ pups_pattern = regulator(get_system_list("PUPS"))
approved_count = 0
unapproved_count = 0
@@ -145,13 +145,13 @@ class Hash:
approved_count += 1
except (ValueError, TypeError):
logger.debug(
- "Needs Review: Scannermatch score is missing or invalid. — {e}"
+ "Needs Review: Scannermatch score is missing or invalid. — {e}"
)
hash_obj.at_decision = "needs_review"
needs_review_count += 1
logger.debug(
- f"Final counts — Needs Review: {needs_review_count}, Approved: {approved_count}, Unapproved: {unapproved_count}"
+ f"Final counts — Needs Review: {needs_review_count}, Approved: {approved_count}, Unapproved: {unapproved_count}"
)
return hashes
@@ -384,9 +384,9 @@ class ExecutionHistoryRecord:
Returns:
List[ExecutionHistoryRecord]: The same list, with hash_obj.at_decision updated.
"""
- threat_tolerance = get_protected_value("VT_THREAT_TOLERANCE", cast_type=int)
- bad_publishers_pattern = regulator(load_env_json("BAD_PUBLISHERS", "[]"))
- pups_pattern = regulator(load_env_json("PUPS", "[]"))
+ threat_tolerance = get_system_value("VT_THREAT_TOLERANCE", cast_type=int)
+ bad_publishers_pattern = regulator(get_system_list("BAD_PUBLISHERS"))
+ pups_pattern = regulator(get_system_list("PUPS"))
approved_count = 0
unapproved_count = 0
@@ -443,13 +443,13 @@ class ExecutionHistoryRecord:
approved_count += 1
except (ValueError, TypeError) as e:
logger.debug(
- f"Needs Review: Scannermatch score is missing or invalid. — {e}"
+ f"Needs Review: Scannermatch score is missing or invalid. — {e}"
)
hash_obj.at_decision = "needs_review"
needs_review_count += 1
logger.debug(
- f"Final counts — Needs Review: {needs_review_count}, "
+ f"Final counts — Needs Review: {needs_review_count}, "
f"Approved: {approved_count}, Unapproved: {unapproved_count}"
)
diff --git a/services/agenthandler.py b/services/agenthandler.py
index 0ec3d3b..7a788e6 100644
--- a/services/agenthandler.py
+++ b/services/agenthandler.py
@@ -28,7 +28,7 @@ from flows.prepPolicy import selectPolicies
from models.agent import Agent
from models.policy import Policy
from services.API import AirlockAPIWrapper
-from utils.configmanager import get_protected_json, load_env
+from utils.configmanager import get_system_json, load_env
from utils.selector import Selector
from utils.utils import colorText, get_sanitized_input
@@ -38,7 +38,7 @@ logger = logging.getLogger(__name__)
def devicehistory(api: AirlockAPIWrapper, outputjson: bool):
agents = selectAgents(api)
history_days = Selector.select_value(
- prompt="Enter how many days of history to pull (1–150): ",
+ prompt="Enter how many days of history to pull (1–150): ",
value_type=int,
valid_range=(1, 150),
)
@@ -60,7 +60,7 @@ def devicehistory(api: AirlockAPIWrapper, outputjson: bool):
except Exception as e:
print(
colorText(
- f"❌ Error retrieving history for {agent.hostname}: {e}", "red"
+ f"⌠Error retrieving history for {agent.hostname}: {e}", "red"
)
)
continue
@@ -139,7 +139,7 @@ def findAgents(api, return_dataframe):
print(
colorText(
- f"\n✅ Matched devices exported to: {working_dir}\\{filename}",
+ f"\n✅ Matched devices exported to: {working_dir}\\{filename}",
"green",
)
)
@@ -148,7 +148,7 @@ def findAgents(api, return_dataframe):
def collect_device_names() -> List[str]:
- print(colorText("🔍 Device Search", "cyan"))
+ print(colorText("🔠Device Search", "cyan"))
print(
colorText(
"Enter the device hostnames you'd like to search for, one per line.", "cyan"
@@ -185,7 +185,7 @@ def collect_device_names() -> List[str]:
else:
print(
colorText(
- f"⚠️ Invalid input: '{stripped_line}' — only letters, numbers, underscores, spaces, and hyphens are allowed.",
+ f"âš ï¸ Invalid input: '{stripped_line}' — only letters, numbers, underscores, spaces, and hyphens are allowed.",
"yellow",
)
)
@@ -235,8 +235,8 @@ def show_unmatched(
]
if unmatched:
- logger.debug(f"⚠️ No matches for: {', '.join(unmatched)}")
- print(colorText(f"⚠️ No matches for: {', '.join(unmatched)}", "yellow"))
+ logger.debug(f"âš ï¸ No matches for: {', '.join(unmatched)}")
+ print(colorText(f"âš ï¸ No matches for: {', '.join(unmatched)}", "yellow"))
def enrich_agents(agents: List["Agent"], policies: List["Policy"]):
@@ -248,7 +248,7 @@ def selectAgents(api: "AirlockAPIWrapper") -> List["Agent"]:
device_names = collect_device_names()
if not device_names:
logger.debug("No device names entered")
- print(colorText("⚠️ No device names entered.", "red"))
+ print(colorText("âš ï¸ No device names entered.", "red"))
return []
use_exact = choose_match_type()
@@ -261,11 +261,11 @@ def selectAgents(api: "AirlockAPIWrapper") -> List["Agent"]:
show_unmatched(device_names, matched_agents, use_exact)
if not matched_agents:
- logger.debug("❌ No matching devices found.")
- print(colorText("❌ No matching devices found.", "red"))
+ logger.debug("⌠No matching devices found.")
+ print(colorText("⌠No matching devices found.", "red"))
return []
- print(colorText(f"✅ Found {len(matched_agents)} matching device(s).", "green"))
+ print(colorText(f"✅ Found {len(matched_agents)} matching device(s).", "green"))
logger.info("Matched agent hostnames:")
rows = (len(matched_agents) + 2) // 3 # 3 columns
for row in range(rows):
@@ -283,8 +283,8 @@ def selectAgents(api: "AirlockAPIWrapper") -> List["Agent"]:
)
if not matched_agents:
- logger.debug("❌ No matching devices remain after refinement.")
- print(colorText("❌ No matching devices remain after refinement.", "red"))
+ logger.debug("⌠No matching devices remain after refinement.")
+ print(colorText("⌠No matching devices remain after refinement.", "red"))
return []
enrich_agents(matched_agents, policies)
@@ -302,10 +302,10 @@ def moveAgentToRelatedPolicy(
Args:
api: AirlockAPIWrapper instance.
agent: Agent object.
- policy_relationship_map: Dict mapping enforcement → audit.
+ policy_relationship_map: Dict mapping enforcement → audit.
mode: 'audit' to move to audit, 'enforcement' to move to enforcement.
"""
- policy_relationship_map = get_protected_json("POLICY_MAP_ENF_AUD", "{}")
+ policy_relationship_map = get_system_json("POLICY_MAP_ENF_AUD", "{}")
if mode == "audit":
if agent.groupid in policy_relationship_map:
diff --git a/services/policyhandler.py b/services/policyhandler.py
index c32cdfc..0fa6a11 100644
--- a/services/policyhandler.py
+++ b/services/policyhandler.py
@@ -27,7 +27,7 @@ import tqdm
from models.policy import Policy
from services.API import AirlockAPIWrapper
-from utils.configmanager import get_protected_json
+from utils.configmanager import get_system_json
from utils.setup import get_base_directory
from utils.utils import areYouSure, colorText, get_sanitized_input
@@ -236,7 +236,7 @@ def skipback(days):
def updateAuditPoliciesFromEnforcementPolices(api: AirlockAPIWrapper):
- policy_relationship_map = get_protected_json("POLICY_MAP_ENF_AUD", "{}")
+ policy_relationship_map = get_system_json("POLICY_MAP_ENF_AUD", "{}")
for enforcement_policy, audit_policy in policy_relationship_map.items():
api.policy_clone(enforcement_policy, audit_policy)
api.policy_set_auditmode(audit_policy, "1")
diff --git a/utils/configmanager.py b/utils/configmanager.py
index 7b5aea5..7ca49ed 100644
--- a/utils/configmanager.py
+++ b/utils/configmanager.py
@@ -18,119 +18,279 @@ import logging
import os
from pathlib import Path
import sys
-from typing import Callable, Optional, TypeVar
+from typing import Any, Callable, Optional, TypeVar
T = TypeVar("T")
logger = logging.getLogger(__name__)
-PROTECTED_KEYS = [
+# System config keys - these are immutable and come from system_config.json (bundled in exe)
+SYSTEM_CONFIG_KEYS = [
"URL",
"TELEM_URL",
"APPNAME",
"LOG_LEVEL",
+ "BAD_PATH_PARTS",
+ "BAD_PUBLISHERS",
+ "PUPS",
"PATH_EXCLUSION_CONST",
"MIN_FILES_FOR_PATH",
"VT_THREAT_TOLERANCE",
"POLICY_MAP_ENF_AUD",
]
-_protected_config = {}
+# User config keys - these can be changed by the end user
+USER_CONFIG_KEYS = [
+ "TELEMETRY", # User opt-in/out for telemetry
+ "TEXTUAL_THEME", # UI theme preference
+ "EXTRAS", # Feature flags
+]
+
+# In-memory config storage
+_system_config = {}
+_user_config = {}
def get_system_config_path() -> Path:
+ """
+ Get path to system_config.json.
+ Priority:
+ 1. Bundled in exe (_MEIPASS)
+ 2. Next to this file (development)
+ """
# Check inside bundled EXE directory first
bundled_dir = Path(getattr(sys, "_MEIPASS", ""))
bundled_path = bundled_dir / "system_config.json"
if bundled_path.exists():
return bundled_path
- # Fallback to external location
+ # Fallback to development location (next to this file)
return Path(__file__).parent.parent / "system_config.json"
-def load_protected_config() -> dict:
- global _protected_config
+def load_system_config() -> dict:
+ """
+ Load system configuration from system_config.json.
+ This should only be called once at startup.
+ Returns the full system config dict.
+ """
+ global _system_config
+
try:
- with open(get_system_config_path(), "r") as f:
- system_config = json.load(f)
+ config_path = get_system_config_path()
+ with open(config_path, "r") as f:
+ _system_config = json.load(f)
+ logger.debug(f"✅ Loaded system config from {config_path}")
except FileNotFoundError:
- logging.warning("⚠️ system_config.json not found. Using built-in defaults.")
- system_config = {
+ logger.warning("⚠️ system_config.json not found. Using minimal defaults.")
+ # Minimal defaults for development without system_config.json
+ _system_config = {
"APPNAME": "Loxide",
+ "LOG_LEVEL": "INFO",
"PATH_EXCLUSION_CONST": 4,
"MIN_FILES_FOR_PATH": 4,
"VT_THREAT_TOLERANCE": 4,
- "POLICY_MAP_ENF_AUD": {"enforced_id": "audit_id"},
+ "POLICY_MAP_ENF_AUD": {},
}
- _protected_config = {key: system_config[key] for key in PROTECTED_KEYS}
- return _protected_config
+ return _system_config
-def get_protected_value(
+def get_system_value(
key: str, cast_type: Callable[[str], T] = str, default: Optional[T] = None
) -> Optional[T]:
- value = _protected_config.get(key)
+ """
+ Get a value from system config (immutable).
+
+ Parameters:
+ key: The config key to retrieve
+ cast_type: Function to cast the value to desired type
+ default: Default value if key not found
+
+ Returns:
+ The config value cast to the desired type, or default
+ """
+ value = _system_config.get(key)
if value is None:
- logging.warning(f"Protected config key '{key}' not found.")
+ logger.warning(f"System config key '{key}' not found.")
return default
+
try:
if isinstance(value, str):
value = value.strip("'\"")
return cast_type(value)
except (ValueError, TypeError):
- logging.warning(
- f"Invalid value for protected key '{key}': {value}. Expected type {cast_type.__name__}."
+ logger.warning(
+ f"Invalid value for system key '{key}': {value}. Expected type {cast_type.__name__}."
)
return default
-def get_protected_json(key: str, default: str = "{}") -> dict:
- raw = _protected_config.get(key, default)
+def get_system_json(key: str, default: Optional[dict] = None) -> dict:
+ """
+ Get a JSON/dict value from system config.
+ Handles both dict values and JSON strings.
+ """
+ if default is None:
+ default = {}
+
+ raw = _system_config.get(key, default)
if isinstance(raw, dict):
return raw
+
try:
return json.loads(raw)
- except json.JSONDecodeError:
- try:
- escaped = raw.encode("unicode_escape").decode("utf-8")
- return json.loads(escaped)
- except Exception as e:
- logging.error(f"Failed to parse protected JSON key '{key}': {e}")
- return json.loads(default)
+ except (json.JSONDecodeError, TypeError) as e:
+ logger.error(f"Failed to parse system JSON key '{key}': {e}")
+ return default
-def load_env_json(key: str, default: str):
- raw = os.getenv(key, default)
+def get_system_list(key: str, default: Optional[list] = None) -> list:
+ """
+ Get a list value from system config.
+ Handles both list values and JSON strings.
+
+ Parameters:
+ key: The config key to retrieve
+ default: Default value if key not found or parsing fails
+
+ Returns:
+ The list value or default
+ """
+ if default is None:
+ default = []
+
+ raw = _system_config.get(key, default)
+ if isinstance(raw, list):
+ return raw
+
try:
- return json.loads(raw)
- except json.JSONDecodeError:
- try:
- escaped = raw.encode("unicode_escape").decode("utf-8")
- return json.loads(escaped)
- except Exception as e:
- logging.error(f"Failed to parse {key}: {e}")
- return json.loads(default)
+ result = json.loads(raw) if isinstance(raw, str) else raw
+ if isinstance(result, list):
+ return result
+ logger.warning(f"System config key '{key}' is not a list: {type(result)}")
+ return default
+ except (json.JSONDecodeError, TypeError) as e:
+ logger.error(f"Failed to parse system list key '{key}': {e}")
+ return default
+
+
+def load_user_config(config_dir: Path) -> dict:
+ """
+ Load user configuration from user_config.json.
+ Creates the file with defaults if it doesn't exist.
+
+ Parameters:
+ config_dir: Directory containing user_config.json
+
+ Returns:
+ The user config dict
+ """
+ global _user_config
+
+ user_config_path = config_dir / "user_config.json"
+
+ if not user_config_path.exists():
+ # Create default user config
+ default_user_config = {
+ "TELEMETRY": "false",
+ "TEXTUAL_THEME": "gruvbox",
+ "EXTRAS": "NOTTODAY",
+ }
+ user_config_path.parent.mkdir(parents=True, exist_ok=True)
+ with open(user_config_path, "w") as f:
+ json.dump(default_user_config, f, indent=4)
+ logger.debug(f"Created default user config at {user_config_path}")
+ _user_config = default_user_config
+ else:
+ with open(user_config_path, "r") as f:
+ _user_config = json.load(f)
+ logger.debug(f"✅ Loaded user config from {user_config_path}")
+
+ return _user_config
+
+
+def save_user_config(config_dir: Path, updates: dict) -> None:
+ """
+ Save updates to user configuration.
+ Only keys in USER_CONFIG_KEYS are allowed.
+
+ Parameters:
+ config_dir: Directory containing user_config.json
+ updates: Dict of key-value pairs to update
+ """
+ global _user_config
+
+ # Validate that only user-configurable keys are being updated
+ invalid_keys = [k for k in updates.keys() if k not in USER_CONFIG_KEYS]
+ if invalid_keys:
+ logger.error(f"Attempted to save invalid user config keys: {invalid_keys}")
+ raise ValueError(f"Cannot modify system config keys: {invalid_keys}")
+
+ # Update in-memory config
+ _user_config.update(updates)
+
+ # Write to file
+ user_config_path = config_dir / "user_config.json"
+ user_config_path.parent.mkdir(parents=True, exist_ok=True)
+ with open(user_config_path, "w") as f:
+ json.dump(_user_config, f, indent=4)
+
+ logger.debug(f"✅ Saved user config to {user_config_path}: {updates}")
+
+
+def get_user_value(
+ key: str, cast_type: Callable[[str], T] = str, default: Optional[T] = None
+) -> Optional[T]:
+ """
+ Get a value from user config (mutable).
+
+ Parameters:
+ key: The config key to retrieve
+ cast_type: Function to cast the value to desired type
+ default: Default value if key not found
+
+ Returns:
+ The config value cast to the desired type, or default
+ """
+ value = _user_config.get(key)
+ if value is None:
+ logger.warning(f"User config key '{key}' not found.")
+ return default
+
+ try:
+ if isinstance(value, str):
+ value = value.strip("'\"")
+ return cast_type(value)
+ except (ValueError, TypeError):
+ logger.warning(
+ f"Invalid value for user key '{key}': {value}. Expected type {cast_type.__name__}."
+ )
+ return default
def load_env(
key: str, cast_type: Callable[[str], T] = str, default: Optional[T] = None
) -> Optional[T]:
"""
- Safely retrieves an environment variable and casts it to the desired type.
+ Safely retrieves an environment variable from .env and casts it to the desired type.
+ This should ONLY be used for runtime/dynamic values like WORKING_DIR.
+
+ For system config, use get_system_value().
+ For user config, use get_user_value().
Parameters:
- key (str): The name of the environment variable.
- cast_type (Callable[[str], T], optional): Function to cast the value. Defaults to str.
- default (Optional[T], optional): Default value if the variable is not set or invalid.
+ key: The name of the environment variable
+ cast_type: Function to cast the value. Defaults to str
+ default: Default value if the variable is not set or invalid
Returns:
- Optional[T]: The casted value or the default.
+ The casted value or the default
"""
value = os.getenv(key)
if value is None:
- logger.warning(f"Environment variable '{key}' not set.")
+ logger.debug(f"Environment variable '{key}' not set, using default.")
return default
+
try:
value = value.strip("'\"") # Strip surrounding quotes
return cast_type(value)
@@ -139,3 +299,46 @@ def load_env(
f"Invalid value for env var '{key}': {value}. Expected type {cast_type.__name__}."
)
return default
+
+
+def load_env_json(key: str, default: str = "[]") -> Any:
+ """
+ Load a JSON value from environment or system config.
+
+ DEPRECATED: This function is kept for backward compatibility.
+ - For system config lists (BAD_PUBLISHERS, PUPS, BAD_PATH_PARTS), use get_system_list()
+ - For system config dicts, use get_system_json()
+ - For actual .env JSON values, parse manually
+
+ This function automatically redirects known system config keys to system config.
+ """
+ # Known system config list keys - redirect to system config
+ system_list_keys = ["BAD_PUBLISHERS", "PUPS", "BAD_PATH_PARTS"]
+ if key in system_list_keys:
+ logger.debug(f"Redirecting load_env_json('{key}') to get_system_list()")
+ return get_system_list(key, json.loads(default) if default else [])
+
+ # Known system config dict keys - redirect to system config
+ system_dict_keys = ["POLICY_MAP_ENF_AUD"]
+ if key in system_dict_keys:
+ logger.debug(f"Redirecting load_env_json('{key}') to get_system_json()")
+ return get_system_json(key, json.loads(default) if default else {})
+
+ # Fall back to reading from .env (backward compatibility for unknown keys)
+ raw = os.getenv(key, default)
+ try:
+ return json.loads(raw)
+ except json.JSONDecodeError:
+ try:
+ escaped = raw.encode("unicode_escape").decode("utf-8")
+ return json.loads(escaped)
+ except Exception as e:
+ logger.error(f"Failed to parse {key}: {e}")
+ return json.loads(default)
+
+
+# Backwards compatibility aliases (deprecated - use get_system_value instead)
+get_protected_value = get_system_value
+get_protected_json = get_system_json
+load_protected_config = load_system_config
+PROTECTED_KEYS = SYSTEM_CONFIG_KEYS # For backwards compatibility
diff --git a/utils/setup.py b/utils/setup.py
index 77bc5a0..ee4a384 100644
--- a/utils/setup.py
+++ b/utils/setup.py
@@ -13,18 +13,20 @@
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see .
-import json
import logging
import logging.config
import logging.handlers
import os
from pathlib import Path
import platform
-import sys
from dotenv import load_dotenv, set_key
-from utils.configmanager import PROTECTED_KEYS, load_protected_config
+from utils.configmanager import (
+ get_system_value,
+ load_system_config,
+ load_user_config,
+)
def get_base_directory() -> Path:
@@ -38,7 +40,7 @@ def get_base_directory() -> Path:
return home / ".local" / "share" / "Loxide"
-def configure_logging(log_dir: Path, log_level: str = "DEBUG"):
+def configure_logging(log_dir: Path, log_level: str = "INFO"):
log_file = log_dir / "Loxide.log"
config = {
@@ -62,12 +64,12 @@ def configure_logging(log_dir: Path, log_level: str = "DEBUG"):
"interval": 1, # Every 1 day
"backupCount": 7, # Keep 7 days of logs
"encoding": "utf-8", # Ensure UTF-8 encoding
- "level": "DEBUG", # Always log DEBUG and above
+ "level": "DEBUG", # Always log DEBUG and above to file
"formatter": "detailed", # Use detailed format
},
"console": {
"class": "logging.StreamHandler",
- "level": log_level.upper(), # Configurable log level
+ "level": log_level.upper(), # System-configured level for console
"formatter": "simple", # Use simple format
},
},
@@ -95,59 +97,15 @@ def configure_logging(log_dir: Path, log_level: str = "DEBUG"):
logging.getLogger().debug("✅ Logging configured.")
-def get_system_config_path() -> Path:
- base_path = Path(
- getattr(sys, "_MEIPASS", os.path.dirname(os.path.abspath(__file__)))
- )
- return base_path.parent / "system_config.json"
-
-
-def load_system_config() -> dict:
- try:
- config_path = get_system_config_path()
- with open(config_path, "r") as f:
- return json.load(f)
- except FileNotFoundError:
- logging.warning("⚠️ system_config.json not found. Using built-in defaults.")
- return {
- "APPNAME": "Loxide",
- "LOG_LEVEL": "DEBUG",
- "PATH_EXCLUSION_CONST": 4,
- "MIN_FILES_FOR_PATH": 4,
- "VT_THREAT_TOLERANCE": 4,
- "POLICY_MAP_ENF_AUD": {"enforced_id": "audit_id"},
- }
-
-
-def load_user_config(config_dir: Path) -> dict:
- user_config_path = config_dir / "user_config.json"
- if not user_config_path.exists():
- default_user_config = {
- "TELEMETRY": "FALSE",
- "TEXTUAL_THEME": "gruvbox",
- "EXTRAS": "NOTTODAY",
- }
- with open(user_config_path, "w") as f:
- json.dump(default_user_config, f, indent=4)
- logging.debug(f"Created user config at {user_config_path}")
- with open(user_config_path, "r") as f:
- return json.load(f)
-
-
-def write_config_to_env(config: dict, env_path: Path):
- for key, value in config.items():
- if key in PROTECTED_KEYS:
- continue # Skip protected keys
- try:
- serialized = (
- json.dumps(value) if isinstance(value, (list, dict)) else str(value)
- )
- set_key(env_path, key, serialized)
- except Exception as e:
- logging.warning(f"Failed to write {key} to .env: {e}")
-
-
def setup():
+ """
+ Initialize the application environment:
+ 1. Create directory structure
+ 2. Load system config (immutable, from system_config.json)
+ 3. Load user config (mutable, from user_config.json)
+ 4. Configure logging
+ 5. Set up .env with WORKING_DIR only
+ """
base_dir = get_base_directory()
dirs = {
"config": base_dir / "config",
@@ -159,20 +117,30 @@ def setup():
path.mkdir(parents=True, exist_ok=True)
logging.debug(f"{name.capitalize()} directory ensured at: {path}")
+ # Load system config (immutable)
system_config = load_system_config()
- configure_logging(dirs["logs"], system_config.get("LOG_LEVEL", "DEBUG"))
+ # Configure logging with system-defined log level
+ log_level = get_system_value("LOG_LEVEL", str, "INFO")
+ configure_logging(dirs["logs"], log_level)
+
+ # Load user config (mutable)
+ user_config = load_user_config(dirs["config"])
+
+ # Set up .env file - ONLY for WORKING_DIR (runtime-configurable value)
env_path = base_dir / ".env"
if not env_path.exists():
env_path.touch()
load_dotenv(dotenv_path=env_path, override=True)
+ # Set up working directory (only dynamic value in .env)
working_dir = Path(os.getenv("WORKING_DIR") or (base_dir / "data"))
working_dir.mkdir(parents=True, exist_ok=True)
- set_key(env_path, "WORKING_DIR", str(working_dir))
+ set_key(str(env_path), "WORKING_DIR", str(working_dir))
os.environ["WORKING_DIR"] = str(working_dir)
logging.debug(f"Working directory set to: {working_dir}")
+ # Create folder structure in working directory
folders_structure = {
"Approved": [],
"Needs_Review": ["Review_First", "Review_Second", "HTML"],
@@ -189,23 +157,4 @@ def setup():
subfolder_path.mkdir(parents=True, exist_ok=True)
logging.debug(f" └─ '{subfolder}' subfolder created at: {subfolder_path}")
- user_config = load_user_config(dirs["config"])
- merged_config = {**system_config, **user_config}
-
- protected_config = load_protected_config()
- merged_config.update(protected_config)
-
- # ✅ URL resolution order: system_config → .env → user prompt
- url = system_config.get("URL")
- if not url:
- url = os.getenv("URL")
- if not url:
- url = input(
- "🌐 Enter the service URL (e.g., https://example.com/api): "
- ).strip()
- merged_config["URL"] = url
- set_key(env_path, "URL", url)
- os.environ["URL"] = url
- logging.debug(f"Service URL set to: {url}")
-
- write_config_to_env(merged_config, env_path)
+ logging.info("✅ Setup complete")
diff --git a/utils/test.py b/utils/test.py
deleted file mode 100644
index e69de29..0000000