diff --git a/Loxide.py b/Loxide.py index 7fa448c..fae9cc8 100644 --- a/Loxide.py +++ b/Loxide.py @@ -23,14 +23,13 @@ import logging import os -import tempfile -import dotenv import urllib3 from services.API import AirlockAPIWrapper from services.security import getAPI from TUI.TUI import run_Loxide +from utils.configmanager import get_system_value from utils.setup import get_base_directory, setup from utils.utils import irtang @@ -38,24 +37,14 @@ urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) def main(): - - if "NUITKA_ONEFILE_PARENT" in os.environ: - splash_filename = os.path.join( - tempfile.gettempdir(), - f"onefile_{int(os.environ['NUITKA_ONEFILE_PARENT'])}_splash_feedback.tmp", - ) - if os.path.exists(splash_filename): - os.unlink(splash_filename) - irtang() # Determine working directory, setup directory, configure logging, sent env, get API and URL if not already stored setup() base_dir = get_base_directory() logger = logging.getLogger(__name__) - dotenv.load_dotenv(dotenv_path=base_dir / ".env") try: - url = os.getenv("URL") + url = get_system_value("URL") username = os.getenv("USERNAME") if not url: @@ -75,7 +64,7 @@ def main(): raise ValueError("API key for Loxide is missing.") api = AirlockAPIWrapper( - base_url=str(os.getenv("URL")), + base_url=str(url), api_key=api_key, ) run_Loxide(api) diff --git a/TUI/allowlistselectionscreen.py b/TUI/allowlistselectionscreen.py index fd84f20..72426f2 100644 --- a/TUI/allowlistselectionscreen.py +++ b/TUI/allowlistselectionscreen.py @@ -32,7 +32,7 @@ class AllowlistSelectionWidget(Static): layout: vertical; } #allowlist_main { - height: 100%; + height: 1fr; width: 100%; } #left_panel { @@ -61,7 +61,8 @@ class AllowlistSelectionWidget(Static): margin: 1 0; } #action_buttons { - height: 10%; + height: auto; + min-height: 3; padding: 1; content-align: center middle; } @@ -149,7 +150,7 @@ class AllowlistSelectionWidget(Static): # Action buttons at bottom with Horizontal(id="action_buttons"): - self.back_btn = Button("← Back", id="back_btn") + self.back_btn = Button("⬅ Back", id="back_btn") self.add_btn = Button("➕ Add to Allowlist", id="add_to_allowlist_btn") self.back_btn.styles.width = "50%" @@ -176,7 +177,8 @@ class AllowlistSelectionWidget(Static): # First, try to get the host's policy if hostname is provided host_policy_allowlists = [] host_policy_ids = set() - policy_name = None + policy_name = "Unknown Policy" # Default value + group_id = None if self.hostname: try: @@ -185,8 +187,23 @@ class AllowlistSelectionWidget(Static): if not agents_df.empty: # Get the policy group ID for this host group_id = agents_df.iloc[0].get("groupid") - policy_name = agents_df.iloc[0].get( - "groupname", "Unknown Policy" + + # Look up the policy name from app's cached policies + if ( + group_id + and hasattr(self.app, "policies") + and self.app.policies + ): + for policy in self.app.policies: + if policy.groupid == group_id: + policy_name = policy.name + logger.info( + f"Found policy name: '{policy_name}' for group_id: {group_id}" + ) + break + + logger.info( + f"Found host '{self.hostname}' in policy '{policy_name}' (group_id: {group_id})" ) if group_id: @@ -208,6 +225,33 @@ class AllowlistSelectionWidget(Static): except Exception as e: logger.warning(f"Could not get host's policy allowlists: {e}") + # If we still don't have a policy name, try to get it from the first allowlist or use a default + if not policy_name: + # Get all policies and try to find which one has allowlists + try: + all_policies_df = self.api.policy_find_all() + if not all_policies_df.empty: + # If we have a group_id from somewhere, use it + if group_id: + policy_row = all_policies_df[ + all_policies_df["groupid"] == group_id + ] + if not policy_row.empty: + policy_name = policy_row.iloc[0].get( + "groupname", "Unknown Policy" + ) + else: + # Use the first policy as fallback + policy_name = all_policies_df.iloc[0].get( + "groupname", "Default Policy" + ) + logger.info(f"Using first available policy: {policy_name}") + else: + policy_name = "Unknown Policy" + except Exception as e: + logger.warning(f"Could not fetch policies: {e}") + policy_name = "Unknown Policy" + # Get all allowlists all_allowlists_df = self.api.allowlist_find_all() @@ -243,7 +287,7 @@ class AllowlistSelectionWidget(Static): # Add policy-associated allowlists if any if host_policy_allowlists: # Add section header - header_text = f"━━━ Policy: {policy_name or 'Host Policy'} ━━━" + header_text = f"=== Policy: {policy_name or 'Host Policy'} ===" self.allowlist_table.add_row(header_text, "", "", key="header_policy") current_row += 1 @@ -270,7 +314,7 @@ class AllowlistSelectionWidget(Static): current_row += 1 self.allowlist_table.add_row( - "━━━ Other Available Allowlists ━━━", "", "", key="header_other" + "=== Other Available Allowlists ===", "", "", key="header_other" ) current_row += 1 @@ -336,9 +380,9 @@ class AllowlistSelectionWidget(Static): if found_col: self.hash_column = found_col - preview_lines.append(f"✓ Found hash column: **{found_col}**\n") + preview_lines.append(f"✓ Found hash column: **{found_col}**\n") else: - preview_lines.append("⚠️ **No hash column found**\n") + preview_lines.append("⚠️ **No hash column found**\n") preview_lines.append("Available columns:\n") for col in self.selected_data.columns: if col != "_row_id": @@ -420,7 +464,7 @@ class AllowlistSelectionWidget(Static): self.selected_allowlist = self.allowlists[actual_allowlist_index] self.add_btn.disabled = False self.add_btn.label = ( - f"➕ Add to '{self.selected_allowlist.get('name', 'Unknown')}'" + f"➕ Add to '{self.selected_allowlist.get('name', 'Unknown')}'" ) # Update preview with selection @@ -542,11 +586,8 @@ class AllowlistSelectionWidget(Static): ) # Change button to "Done" - self.add_btn.label = "✅ Done - Close" - self.add_btn.disabled = False - - # When clicked again, close the screen - self.add_btn_success = True + self.add_btn.label = "✅ Done" + self.add_btn.disabled = True except Exception as exc: logger.exception(f"Failed to add hashes to allowlist: {exc}") @@ -559,7 +600,7 @@ class AllowlistSelectionWidget(Static): # Re-enable button self.add_btn.disabled = False - self.add_btn.label = "➕ Retry Add to Allowlist" + self.add_btn.label = "⟳ Retry Add to Allowlist" class AllowlistSelectionScreen(Screen): diff --git a/TUI/otpactivityscreen.py b/TUI/otpactivityscreen.py index 84847e0..5947cee 100644 --- a/TUI/otpactivityscreen.py +++ b/TUI/otpactivityscreen.py @@ -56,12 +56,13 @@ class OTPActivitiesWidget(Static): layout: vertical; } #activity_preview_container { - height: 75%; + height: 1fr; border: none; padding: 1 1; } #activity_buttons { - height: 25%; + height: auto; + min-height: 3; padding: 1 1; content-align: center middle; } @@ -481,16 +482,18 @@ class ActivityDetailWidget(Static): layout: vertical; } #detail_table_container { - height: 75%; + height: 1fr; padding: 1 1; } #selection_buttons { - height: 10%; + height: auto; + min-height: 3; padding: 1 1; content-align: center middle; } #detail_buttons { - height: 15%; + height: auto; + min-height: 3; padding: 1 1; content-align: center middle; } @@ -649,7 +652,7 @@ class ActivityDetailWidget(Static): logger.exception("Failed to sort by column %s: %s", column_key, exc) return - # ✅ Only refresh rows, not columns + # ✅ Only refresh rows, not columns await self._build_table(rebuild=False) async def on_button_pressed(self, event) -> None: @@ -728,13 +731,13 @@ class ActivityDetailWidget(Static): if self.activities_df is None or self.activities_df.empty: logger.info("No activities to export.") await self.mount( - Static("❌ No activities to export.", classes="notification") + Static("❌ No activities to export.", classes="notification") ) return if not self.selected_row_ids: logger.info("No rows selected for export.") await self.mount( - Static("❌ No rows selected for export.", classes="notification") + Static("❌ No rows selected for export.", classes="notification") ) return try: @@ -755,12 +758,12 @@ class ActivityDetailWidget(Static): logger.exception("Failed to export detail activities: %s", exc) await self.mount( Static( - "❌ Failed to export activities; check logs.", + "❌ Failed to export activities; check logs.", classes="notification", ) ) - # ✅ Helper methods + # ✅ Helper methods def get_selected_data(self) -> pd.DataFrame: """Return a DataFrame of the selected rows.""" if not self.selected_row_ids: diff --git a/services/API.py b/services/API.py index 7fab0ac..22cfa7d 100644 --- a/services/API.py +++ b/services/API.py @@ -64,6 +64,23 @@ class AirlockAPIWrapper: logger.error(f"API request failed: {e}") raise + def _post_raw(self, endpoint: str, payload: Optional[dict] = None) -> bytes: + """ + Send POST request and return raw response content (bytes). + Useful for XML endpoints. + """ + url = f"{self.base_url}{endpoint}" + data = json.dumps(payload or {}) + try: + logger.debug(f"POST Request to {url} with payload: {payload}") + response = requests.post(url, headers=self.headers, data=data, verify=False) + response.raise_for_status() + logger.debug(f"Raw response received from {url}") + return response.content # bytes + except requests.exceptions.RequestException as e: + logger.error(f"API request failed: {e}") + raise + # Allowlist Management def allowlist_find_all(self) -> pd.DataFrame: """ @@ -75,6 +92,12 @@ class AirlockAPIWrapper: result = self._post("/v1/application", {}) return pd.DataFrame(result["response"]["applications"]) + def allowlist_export(self, applicationid) -> bytes: + """Return Allowlist XML as bytes to save to file""" + payload = {"applicationid": applicationid} + result = self._post_raw("/v1/application/export", payload) + return result # should be bytes + # Agent Management def agent_find_all(self) -> pd.DataFrame: """Retrieve all agents.""" @@ -116,6 +139,30 @@ class AirlockAPIWrapper: result = self._post("/v1/agent/find", payload) return pd.DataFrame(result["response"]["agents"]) + # Baseline Managment + def baseline_find_all(self) -> pd.DataFrame: + """Retrieve all Baselines.""" + result = self._post("/v1/baseline", {}) + return pd.DataFrame(result["response"]["baselines"]) + + def baseline_export(self, baselineid) -> bytes: + """Return Baseline XML as bytes to save to file""" + payload = {"baselineid": baselineid} + result = self._post_raw("/v1/baseline/export", payload) + return result + + # Blocklist Managment + def blocklist_find_all(self) -> pd.DataFrame: + """Retrieve all Baselines.""" + result = self._post("/v1/blocklist", {}) + return pd.DataFrame(result["response"]["blocklists"]) + + def blocklist_export(self, blocklistid) -> bytes: + """Return Blocklist XML as bytes to save to file""" + payload = {"blocklistid": blocklistid} + result = self._post_raw("/v1/blocklist/export", payload) + return result + # Hash Management def hash_add_to_allowlist(self, applicationid: str, hashes: List[str]) -> dict: """Add hashes to the allowlist for a specific application.""" diff --git a/utils/configmanager.py b/utils/configmanager.py index a5f59bb..0d5391a 100644 --- a/utils/configmanager.py +++ b/utils/configmanager.py @@ -192,7 +192,7 @@ def load_user_config(config_dir: Path) -> dict: if not user_config_path.exists(): # Create default user config default_user_config = { - "TELEMETRY": "false", + "TELEMETRY": False, "TELEM_URL": "", "TEXTUAL_THEME": "gruvbox", "EXTRAS": "NOTTODAY",