From 57d0f12000dbc9604a52ad5027844b4d71f8394a Mon Sep 17 00:00:00 2001 From: Zarithas Date: Mon, 15 Dec 2025 17:01:58 -0500 Subject: [PATCH] feat: Complete Policy Prep Workflow with UX upgrades, Liftoff API, and TUI merge - Added intro screen with workflow overview, time estimate, and onboarding controls - Improved visuals: cleaner checkboxes (/), better loading screen layout - Enforced mandatory tab reviews for critical steps with warnings and blocked navigation - Optimized logging: INFO for milestones, DEBUG for internals; cleaner production logs - Implemented Liftoff API integration: paths, publishers, hashes with granular error handling - Color-coded completion feedback ( success, failure, partial) and detailed summaries - Consolidated architecture: merged TUI.py into Loxide.py (single entry point, no circular imports) - Fixed race condition in table creation with concurrency locks --- Loxide.py | 427 ++++++++++++++++++++++- TUI/Screens/policyprepworkflowscreen.py | 439 +++++++++++++++++++---- TUI/TUI.py | 444 ------------------------ requirements.txt | 22 +- 4 files changed, 820 insertions(+), 512 deletions(-) delete mode 100644 TUI/TUI.py diff --git a/Loxide.py b/Loxide.py index 7c3e974..912ba22 100644 --- a/Loxide.py +++ b/Loxide.py @@ -23,19 +23,438 @@ import logging import os +from typing import Optional +import dotenv +from textual.app import App, ComposeResult +from textual.containers import Vertical +from textual.message import Message +from textual.reactive import reactive +from textual.screen import Screen +from textual.widgets import ( + Button, + DirectoryTree, + Footer, + Header, + Static, + Tab, + Tabs, +) import urllib3 +from models.agent import Agent +from models.policy import Policy 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 setup -from utils.utils import irtang +from TUI.Screens.moveagentworkflowscreen import MoveAgentWorkflowScreen +from TUI.Screens.otpactivityscreen import OTPActivitiesScreen +from TUI.Screens.otprevokescreen import OTPRevokeScreen +from TUI.Screens.otpworkflowscreen import OTPWorkflowScreen +from TUI.Screens.policyprepworkflowscreen import PolicyPrepWorkflowScreen +from TUI.Screens.quietagentworkflowscreen import QuietAgentWorkflowScreen +from TUI.Themes.theme_amber_terminal import get_amber_terminal_theme +from TUI.Themes.theme_retro_terminal import get_retro_terminal_theme +from TUI.Themes.themeselector import ThemeSelector +from TUI.Widgets.agentmoveoperations import AgentMoveOperations +from TUI.Widgets.multiagentselector import MultiAgentSelector +from TUI.Widgets.policytreewidget import PolicyTreeWidget +from TUI.Widgets.resultsdisplay import ResultsDisplay +from utils.configmanager import ( + get_system_value, + get_user_value, + load_env, + save_user_config, +) +from utils.setup import get_base_directory, setup +from utils.utils import irtang, open_directory +dotenv.load_dotenv() urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) +# --------------------------------------------------------------------------- +# GLOBAL STASH +# --------------------------------------------------------------------------- +_APP_RESTART_REASON = None + +logger = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# helper to persist TEXTUAL_THEME to *user* config and mirror to .env +# --------------------------------------------------------------------------- +def _persist_user_theme(theme_name: str) -> None: + """ + 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" + + try: + 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) + + +# --------------------------------------------------------------------------- +# 1) SCREEN +# --------------------------------------------------------------------------- +class MainMenuScreen(Screen): + api: AirlockAPIWrapper + current_tab = reactive("") + + BUTTON_DEFS = { + "agent_actions": [ + ( + "šŸ–„ļø - Find agent, Move agent, or Generate One Time Pass", + "move_agent_workflow_button", + ), + ("šŸŽ« - Review and approve OTP Activities", "otp_activities_button"), + ("šŸ›‘ - Revoke Active OTP Session", "otp_revoke_button"), + ], + "policy": [ + ("āš–ļø - Prepare Policy For Enforcement", "policy_prep_button"), + ("šŸ”• - Find and Move Quiet Hosts to Enforcement", "find_quiet_button"), + ], + } + + def __init__(self) -> None: + super().__init__() + 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() + self.working_dir = wd + + def _make_buttons_for(self, tab_id: str) -> Vertical: + defs = self.BUTTON_DEFS.get(tab_id, []) + buttons = [] + for label, btn_id in defs: + btn = Button(label, id=btn_id) + btn.styles.width = "100%" + buttons.append(btn) + return Vertical(*buttons) + + def compose(self) -> ComposeResult: + yield Header(show_clock=True, icon="āš™") + + tabs = [ + Tab("Tree View", id="p_tree"), + Tab("Agents", id="agent_actions"), + Tab("Directory", id="dir"), + Tab("Settings", id="settings"), + ] + + if self.extras == "POLICYPREP": + tabs.insert(2, Tab("Policy Prep", id="policy")) + + yield Tabs(*tabs, id="tabs") + yield Vertical(id="content") + yield Footer() + + def on_mount(self) -> None: + self.switch_tab("agent_actions") + + # focus helpers + def _get_content_buttons(self) -> list[Button]: + content = self.query_one("#content", Vertical) + return list(content.query(Button)) + + def _focus_first_button(self) -> None: + buttons = self._get_content_buttons() + if buttons: + buttons[0].focus() + + def _focus_tabs(self) -> None: + tabs = self.query_one("#tabs", Tabs) + tabs.focus() + + def _focus_nearby_button(self, direction: int) -> None: + buttons = self._get_content_buttons() + if not buttons: + return + + try: + current = next(i for i, b in enumerate(buttons) if b.has_focus) + except StopIteration: + if direction > 0: + buttons[0].focus() + else: + buttons[-1].focus() + return + + if direction < 0 and current == 0: + self._focus_tabs() + return + + new_index = current + direction + if 0 <= new_index < len(buttons): + buttons[new_index].focus() + + def switch_tab(self, tab_id: str) -> None: + self.current_tab = tab_id + content = self.query_one("#content", Vertical) + content.remove_children() + + if tab_id in self.BUTTON_DEFS: + content.mount(self._make_buttons_for(tab_id)) + self.call_later(self._focus_first_button) + elif tab_id == "dir": + content.mount(DirectoryTree(self.working_dir, id="dir_tree")) + elif tab_id == "p_tree": + content.mount(PolicyTreeWidget(self.app.policies, self.app.devices)) + elif tab_id == "settings": + content.mount(ThemeSelector()) + else: + content.mount(Static(f"Unknown tab: {tab_id}")) + + def on_tabs_tab_activated(self, event: Tabs.TabActivated) -> None: + self.switch_tab(event.tab.id) + + def on_multi_agent_selector_agents_selected( + self, message: MultiAgentSelector.AgentsSelected + ) -> None: + """Handle selected agents from AgentSelector.""" + global _APP_RESTART_REASON + selected_agents = message.selected_agents + logger.info("Selected agents: %s", selected_agents) + # TODO: Implement actual handling of selected agents + _APP_RESTART_REASON = ("multi_agent_action", selected_agents) + self.app.exit() + + def on_theme_selector_theme_selected( + self, message: ThemeSelector.ThemeSelected + ) -> None: + """Handle theme selection from ThemeSelector.""" + global _APP_RESTART_REASON + _persist_user_theme(message.theme_name) + _APP_RESTART_REASON = ("restart",) + self.app.exit() + + def on_agent_move_operations_operation_complete( + self, message: AgentMoveOperations.OperationComplete + ) -> None: + """Handle completion of agent move operation - show results.""" + logger.info( + "Agent move operation completed: %s, %d successful, %d unsuccessful", + message.operation, + len(message.successful), + len(message.unsuccessful), + ) + + # Format results for display + successful_text = "\n".join( + [f"{agent.hostname}" for agent, _ in message.successful] + ) + unsuccessful_text = "\n".join( + [f"{agent.hostname}: {error}" for agent, error in message.unsuccessful] + ) + + # Remove the operations widget + try: + ops_widget = self.query_one(AgentMoveOperations) + ops_widget.remove() + except Exception: + pass + + # Show results + self.query_one("#content", Vertical).mount( + ResultsDisplay(message.operation, successful_text, unsuccessful_text) + ) + + def on_results_display_go_back(self, message: ResultsDisplay.GoBack) -> None: + """Handle back button from results display.""" + try: + results_widget = self.query_one(ResultsDisplay) + results_widget.remove() + except Exception: + pass + # Return to main menu + self.app.pop_screen() + + def on_directory_tree_file_selected( + self, event: DirectoryTree.FileSelected + ) -> None: + path = event.path + logger.debug("Directory file selected: %s", path) + try: + open_directory(str(path)) + except Exception as exc: + logger.error("Failed to open %s: %s", path, exc) + self.app.bell() + + def on_button_pressed(self, event: Button.Pressed) -> None: + button_id = event.button.id + logger.debug("Button pressed: %s", button_id) + + match button_id: + case "move_agent_workflow_button": + self.app.push_screen(MoveAgentWorkflowScreen(self.app.devices)) + event.stop() + + case "otp_generate_button": + self.app.push_screen(OTPWorkflowScreen(self.app.devices)) + event.stop() + + case "find_quiet_button": + self.app.push_screen( + QuietAgentWorkflowScreen(self.app.api, self.app.policies) + ) + event.stop() + return + + case "otp_activities_button": + self.app.push_screen(OTPActivitiesScreen()) + event.stop() + return + + case "otp_revoke_button": + self.app.push_screen(OTPRevokeScreen()) + event.stop() + return + + case "policy_prep_button": + # Use the new TUI workflow screen instead of legacy + self.app.push_screen( + PolicyPrepWorkflowScreen(self.app.api, self.app.policies) + ) + event.stop() + return + + case _: + self.app.bell() + logger.warning("Unknown button pressed: %s", button_id) + return + + +# --------------------------------------------------------------------------- +# 2) APP +# --------------------------------------------------------------------------- +class Loxide(App[Message]): + api: AirlockAPIWrapper + working_dir: str + policies: Optional[list[Policy]] + devices: Optional[list[Agent]] + + CSS = """ + #logo { + width: 100%; + content-align: center middle; + text-align: center; + } + """ + BINDINGS = [ + ("q", "quit", "Quit"), + ("f", "open_fe", "Launch Explorer"), + ("r", "refresh", "Refresh"), + ] + + def __init__(self, api: AirlockAPIWrapper): + self._textual_theme = get_user_value("TEXTUAL_THEME", str, "textual-dark") + super().__init__() + self.api = api + wd = load_env("WORKING_DIR") or os.getcwd() + if not os.path.isdir(wd): + wd = os.getcwd() + self.working_dir = wd + # Initial data load + self.refresh_data() + + def refresh_data(self) -> None: + """Public method to refresh policies and devices from the API.""" + try: + self.policies = [ + Policy(**row.to_dict()) + for _, row in self.api.policy_find_all().iterrows() + ] + self.devices = [ + Agent(**row.to_dict()) + for _, row in self.api.agent_find_all().iterrows() + ] + if self.policies and self.devices: + for agent in self.devices: + agent.enrich_with_policies(self.policies) + logger.debug( + f"Enriched {len(self.devices)} agents with policy information" + ) + except Exception as exc: + logger.error("Failed to load policies/devices: %s", exc) + self.policies = None + self.devices = None + + def on_mount(self, api: AirlockAPIWrapper) -> None: + self.register_theme(get_retro_terminal_theme()) + self.register_theme(get_amber_terminal_theme()) + self.theme = self._textual_theme + self.push_screen(MainMenuScreen()) + + def action_refresh(self) -> None: + self.refresh_data() + + def action_quit(self) -> None: + global _APP_RESTART_REASON + _APP_RESTART_REASON = None + self.exit() + + def action_open_fe(self) -> None: + """Open the working directory in the OS file manager (footer binding).""" + path_to_open = self.working_dir or os.getcwd() + try: + open_directory(path_to_open) + except Exception as exc: + logger.error("Failed to open directory %s: %s", path_to_open, exc) + self.bell() # optional feedback + + +# --------------------------------------------------------------------------- +# 3) PUBLIC ENTRYPOINT +# --------------------------------------------------------------------------- +def run_Loxide(api: AirlockAPIWrapper) -> None: + global _APP_RESTART_REASON + base_dir = get_base_directory() + env_path = base_dir / ".env" + dotenv.load_dotenv(dotenv_path=env_path, override=True) + + max_attempts = 5 + attempts = 0 + + while attempts < max_attempts: + attempts += 1 + logger.debug("Starting app loop iteration (attempt %d)", attempts) + _APP_RESTART_REASON = None + app = Loxide(api) + + try: + app.run() + except SystemExit as exc: + if exc.code != 0: + logger.debug("Caught SystemExit from Textual: %s", exc) + raise + + reason = _APP_RESTART_REASON + logger.debug("After app.run(), _APP_RESTART_REASON = %r", reason) + + if not reason: + logger.debug("No restart reason, exiting loop") + break + + if reason[0] == "restart": + logger.debug("Restarting app loop") + continue + + if reason[0] == "multi_agent_action": + logger.info("Multi-agent action with selected agents: %s", reason[1]) + continue + + logger.error("Unknown restart reason: %r", reason) + break + + +# --------------------------------------------------------------------------- +# 4) MAIN FUNCTION +# --------------------------------------------------------------------------- def main(): irtang() # Determine working directory, setup directory, configure logging, sent env, get API and URL if not already stored diff --git a/TUI/Screens/policyprepworkflowscreen.py b/TUI/Screens/policyprepworkflowscreen.py index ceec0d6..ec9b03a 100644 --- a/TUI/Screens/policyprepworkflowscreen.py +++ b/TUI/Screens/policyprepworkflowscreen.py @@ -165,6 +165,18 @@ class PolicyPrepWorkflowScreen(Screen): # Track if we're navigating with keyboard (to prevent selection) self._keyboard_navigation = False + # Tab review tracking for Step 5 (First Review) + self.approved_tab_reviewed = False + self.needs_review_tab_reviewed = False + + # Tab review tracking for Step 6 (Path Review) + self.paths_tab_reviewed = False + self.publishers_tab_reviewed = False + + # Lock to prevent concurrent table creation + self._creating_review_table = False + self._creating_path_table = False + def compose(self) -> ComposeResult: """Build the UI layout for the workflow screen.""" yield Header(show_clock=True, icon="āš™ļø") @@ -191,7 +203,7 @@ class PolicyPrepWorkflowScreen(Screen): def on_mount(self) -> None: """Initialize the screen when mounted.""" self._update_checklist() - self._show_source_policy_selection() + self._show_introduction() def watch_workflow_stage(self, old_value: str, new_value: str) -> None: """React to workflow stage changes.""" @@ -348,6 +360,86 @@ class PolicyPrepWorkflowScreen(Screen): step8.styles.text_style = "dim" col2.mount(step8) + def _show_introduction(self) -> None: + """Show workflow introduction and overview.""" + self.workflow_stage = "introduction" + content = self.query_one("#content_area", Vertical) + content.remove_children() + + # Title + title = Static("Welcome to Policy Preparation Workflow") + title.styles.margin = (1, 1) + title.styles.text_style = "bold" + title.styles.text_align = "center" + content.mount(title) + + # Description + description = Static( + "This workflow will help you:\n" + " ā€¢ Fetch execution history from selected policies\n" + " ā€¢ Review and approve safe executions\n" + " ā€¢ Calculate efficient path exclusions\n" + " ā€¢ Generate publisher trust rules\n" + " ā€¢ Apply changes to your destination policy" + ) + description.styles.margin = (1, 2) + content.mount(description) + + # Process steps + steps_title = Static("The Process:") + steps_title.styles.margin = (1, 2, 0, 2) + steps_title.styles.text_style = "bold" + content.mount(steps_title) + + steps = Static( + " šŸ“‹ Step 1: Select source policies (data collection)\n" + " šŸŽÆ Step 2: Select destination policy (where changes go)\n" + " šŸ“ Step 3: Select destination allowlist\n" + " šŸ“Š Step 4: Fetch execution data (may take 1-2 minutes)\n" + " āœ… Step 5: Review approved/needs review executions\n" + " šŸ“ Step 6: Review path exclusions and publishers\n" + " šŸ” Step 7: Preview changes before applying\n" + " šŸš€ Step 8: Liftoff - Apply to production" + ) + steps.styles.margin = (0, 2) + content.mount(steps) + + # Time estimate + estimate = Static("ā±ļø Estimated Time: 15-30 minutes depending on data size") + estimate.styles.margin = (1, 2) + estimate.styles.color = "cyan" + content.mount(estimate) + + # Tips + tips_title = Static("šŸ’” Tips:") + tips_title.styles.margin = (1, 2, 0, 2) + tips_title.styles.text_style = "bold" + content.mount(tips_title) + + tips = Static( + " ā€¢ Start with a test policy first\n" + " ā€¢ Review carefully - changes affect all agents\n" + " ā€¢ Use path rules when possible (more efficient)\n" + " ā€¢ Publishers are powerful - use cautiously" + ) + tips.styles.margin = (0, 2) + tips.styles.color = "yellow" + content.mount(tips) + + # Buttons + button_container = Horizontal() + button_container.styles.margin = (2, 2) + button_container.styles.align = ("center", "middle") + content.mount(button_container) + + continue_btn = Button( + "Continue to Policy Selection", id="start_workflow", variant="success" + ) + cancel_btn = Button("Cancel", id="cancel_workflow", variant="default") + + button_container.mount(continue_btn) + button_container.mount(cancel_btn) + def _show_source_policy_selection(self) -> None: """Show the source policy selection screen.""" self.workflow_stage = "select_source" @@ -366,7 +458,7 @@ class PolicyPrepWorkflowScreen(Screen): table.zebra_stripes = True # Add columns - checkbox first, then data columns - table.add_columns("ā˜", "Name", "ID", "Parent") + table.add_columns("ā—‹", "Name", "ID", "Parent") # Sort policies by name for easier selection sorted_policies = sorted(self.policies, key=lambda p: p.name.lower()) @@ -376,7 +468,7 @@ class PolicyPrepWorkflowScreen(Screen): # Skip parent policies if policy.parent == "global-policy-settings": continue - checkbox = "ā˜" # All start unchecked + checkbox = "ā—‹" # All start unchecked table.add_row( checkbox, policy.name, @@ -739,7 +831,7 @@ class PolicyPrepWorkflowScreen(Screen): def _show_fetch_results(self) -> None: """Show the results of data fetching.""" - logger.info("=== _show_fetch_results called ===") + logger.debug("=== _show_fetch_results called ===") self.workflow_stage = "first_review" content = self.query_one("#content_area", Vertical) content.remove_children() @@ -792,15 +884,39 @@ class PolicyPrepWorkflowScreen(Screen): logger.info("Mounted tab buttons") # Show approved table by default - logger.info("About to call _show_review_table('approved')") + logger.debug("About to call _show_review_table('approved')") self._show_review_table("approved") - logger.info("=== _show_fetch_results complete ===") + logger.debug("=== _show_fetch_results complete ===") def _show_review_table(self, table_type: str) -> None: """Show an editable DataTable for reviewing executions.""" - logger.info(f"=== _show_review_table called with type: {table_type} ===") + # Prevent concurrent execution + if self._creating_review_table: + logger.warning( + f"Already creating review table, ignoring duplicate call for {table_type}" + ) + return + + self._creating_review_table = True + + try: + self._show_review_table_impl(table_type) + finally: + self._creating_review_table = False + + def _show_review_table_impl(self, table_type: str) -> None: + """Internal implementation of _show_review_table.""" + logger.debug(f"_show_review_table called with type: {table_type}") content = self.query_one("#content_area", Vertical) + # Mark tab as reviewed + if table_type == "approved": + self.approved_tab_reviewed = True + logger.debug("Marked approved tab as reviewed") + else: + self.needs_review_tab_reviewed = True + logger.debug("Marked needs_review tab as reviewed") + # Determine which dataframe and table ID to show if table_type == "approved": df = self.approved_df @@ -841,6 +957,18 @@ class PolicyPrepWorkflowScreen(Screen): except Exception as e: logger.debug(f"Error removing existing tables: {e}") + # Remove existing instruction and help text (they accumulate without removal) + # Remove ALL Static widgets - they're just text that needs to be replaced + try: + existing_statics = content.query("Static") + logger.debug( + f"Found {len(existing_statics)} existing Static widgets to remove" + ) + for static in existing_statics: + static.remove() + except Exception as e: + logger.debug(f"Error removing Static widgets: {e}") + # Force a refresh to ensure removals are processed try: content.refresh() @@ -850,7 +978,7 @@ class PolicyPrepWorkflowScreen(Screen): # Note: We no longer remove review_controls or review_continue_container # They are reused between tabs to avoid DuplicateIds errors - logger.info( + logger.debug( f"DataFrame for {table_type}: {'empty' if df is None or df.empty else f'{len(df)} rows'}" ) @@ -861,13 +989,13 @@ class PolicyPrepWorkflowScreen(Screen): logger.info(f"No data for {table_type}, mounted empty message") return - # Instructions + # Instructions (no ID needed - we remove all Statics anyway) instruction = Static(title) instruction.styles.margin = (1, 1) instruction.styles.text_style = "bold" content.mount(instruction) - # Help text + # Help text (no ID needed) help_text = Static( "Click to toggle, 'r' for range select (click start, press 'r', click end)\n" "Space to toggle cursor row, 'd' to delete, 'a' select all, arrows navigate" @@ -876,6 +1004,19 @@ class PolicyPrepWorkflowScreen(Screen): help_text.styles.text_style = "dim" content.mount(help_text) + # CRITICAL: Check if table already exists in content (should not happen after removal above) + try: + existing_check = content.query_one(f"#{table_id}", DataTable) + if existing_check: + logger.error( + f"Table {table_id} STILL EXISTS after removal! This should not happen." + ) + # Don't create a new one - just return + return + except Exception: + # Good - table doesn't exist, proceed with creation + pass + # Create the review table review_table = DataTable(id=table_id) review_table.styles.height = "40vh" # Increased since we removed button rows @@ -899,15 +1040,15 @@ class PolicyPrepWorkflowScreen(Screen): if available_cols: # Add checkbox column first - review_table.add_columns("ā˜", *available_cols) + review_table.add_columns("ā—‹", *available_cols) # Add rows with row keys for tracking for idx, row in df.iterrows(): - checkbox = "ā˜" # All start unchecked + checkbox = "ā—‹" # All start unchecked row_data = [str(row.get(col, "")) for col in available_cols] review_table.add_row(checkbox, *row_data, key=str(idx)) - logger.info(f"About to mount {table_id}") + logger.debug(f"About to mount {table_id}") # Final safety check - make sure no table with this ID exists before mounting try: @@ -923,7 +1064,7 @@ class PolicyPrepWorkflowScreen(Screen): pass content.mount(review_table) - logger.info(f"Successfully mounted {table_id} with {len(df)} rows") + logger.debug(f"Successfully mounted {table_id} with {len(df)} rows") # Row count display only (removed Select All, Clear, Delete buttons) try: @@ -1044,13 +1185,19 @@ class PolicyPrepWorkflowScreen(Screen): content = self.query_one("#content_area", Vertical) content.remove_children() - # Clear message + # Add spacer to push text to bottom + spacer = Static("") + spacer.styles.height = "1fr" + content.mount(spacer) + + # Loading message at bottom (above checklist) status = Static( - "Building path exclusions and publisher lists...\n\n" + "Building path exclusions and publisher lists...\n" "This may take a moment for large datasets." ) - status.styles.margin = (2, 1) + status.styles.margin = (1, 1) status.styles.text_align = "center" + status.styles.color = "cyan" content.mount(status) # Force UI refresh to show the loading screen @@ -1311,8 +1458,8 @@ class PolicyPrepWorkflowScreen(Screen): logger.info( f"=== _calculate_paths called with path_exclusion_constant={path_exclusion_constant} ===" ) - logger.info(f"Input DataFrame: {len(df)} rows") - logger.info(f"Columns: {list(df.columns) if not df.empty else 'empty'}") + logger.debug(f"Input DataFrame: {len(df)} rows") + logger.debug(f"Columns: {list(df.columns) if not df.empty else 'empty'}") if df.empty: logger.warning("Input DataFrame is empty") @@ -1327,7 +1474,7 @@ class PolicyPrepWorkflowScreen(Screen): haslcp = self._split_filepaths_grouped(df, path_exclusion_constant, "filename") haslcp = haslcp.drop_duplicates() - logger.info(f"After split_filepaths_grouped: {len(haslcp)} rows") + logger.debug(f"After split_filepaths_grouped: {len(haslcp)} rows") # Filter forbidden paths badpathparts = get_system_list("BAD_PATH_PARTS") @@ -1337,7 +1484,7 @@ class PolicyPrepWorkflowScreen(Screen): forbidden_pattern, case=False, na=False, regex=True ) - logger.info( + logger.debug( f"Removing forbidden filepaths: {forbidden_lcfp.sum()} paths filtered" ) lcp_not_forbidden = haslcp[~forbidden_lcfp].copy() @@ -1347,7 +1494,7 @@ class PolicyPrepWorkflowScreen(Screen): ) lcp_not_forbidden = haslcp.copy() - logger.info(f"After forbidden filtering: {len(lcp_not_forbidden)} rows") + logger.debug(f"After forbidden filtering: {len(lcp_not_forbidden)} rows") # Select relevant columns if "policyname" in lcp_not_forbidden.columns: @@ -1399,11 +1546,11 @@ class PolicyPrepWorkflowScreen(Screen): lcp_not_forbidden_review = lcp_not_forbidden_review[ lcp_not_forbidden_review["unique_sha256_count"] >= min_files_for_path ] - logger.info( + logger.debug( f"After MIN_FILES_FOR_PATH filter ({min_files_for_path}): {len(lcp_not_forbidden_review)} rows (removed {before_filter - len(lcp_not_forbidden_review)})" ) - logger.info( + logger.debug( f"Final result: {len(lcp_not_forbidden_review)} rows with columns: {list(lcp_not_forbidden_review.columns)}" ) @@ -1441,7 +1588,7 @@ class PolicyPrepWorkflowScreen(Screen): def _show_path_results(self) -> None: """Show the results of path building.""" - logger.info("=== _show_path_results called ===") + logger.debug("=== _show_path_results called ===") self.workflow_stage = "second_review" content = self.query_one("#content_area", Vertical) content.remove_children() @@ -1490,15 +1637,39 @@ class PolicyPrepWorkflowScreen(Screen): logger.info("Mounted tab buttons") # Show paths table by default - logger.info("About to call _show_path_review_table('paths')") + logger.debug("About to call _show_path_review_table('paths')") self._show_path_review_table("paths") - logger.info("=== _show_path_results complete ===") + logger.debug("=== _show_path_results complete ===") def _show_path_review_table(self, table_type: str) -> None: """Show an editable DataTable for reviewing paths/publishers.""" - logger.info(f"=== _show_path_review_table called with type: {table_type} ===") + # Prevent concurrent execution + if self._creating_path_table: + logger.warning( + f"Already creating path table, ignoring duplicate call for {table_type}" + ) + return + + self._creating_path_table = True + + try: + self._show_path_review_table_impl(table_type) + finally: + self._creating_path_table = False + + def _show_path_review_table_impl(self, table_type: str) -> None: + """Internal implementation of _show_path_review_table.""" + logger.debug(f"_show_path_review_table called with type: {table_type}") content = self.query_one("#content_area", Vertical) + # Mark tab as reviewed (only paths and publishers, not remaining) + if table_type == "paths": + self.paths_tab_reviewed = True + logger.debug("Marked paths tab as reviewed") + elif table_type == "publishers": + self.publishers_tab_reviewed = True + logger.debug("Marked publishers tab as reviewed") + # Determine which dataframe to show if table_type == "paths": # Combine primary and secondary paths for review @@ -1608,6 +1779,18 @@ class PolicyPrepWorkflowScreen(Screen): except Exception as e: logger.debug(f"Error removing existing controls: {e}") + # Remove existing instruction and help text (they accumulate without removal) + # Remove ALL Static widgets - they're just text that needs to be replaced + try: + existing_statics = content.query("Static") + logger.debug( + f"Found {len(existing_statics)} existing Static widgets to remove" + ) + for static in existing_statics: + static.remove() + except Exception as e: + logger.debug(f"Error removing Static widgets: {e}") + # Force refresh to ensure removals complete try: content.refresh() @@ -1620,13 +1803,13 @@ class PolicyPrepWorkflowScreen(Screen): content.mount(empty_msg) return - # Instructions + # Instructions (no ID needed) instruction = Static(title) instruction.styles.margin = (1, 1) instruction.styles.text_style = "bold" content.mount(instruction) - # Help text (different for remaining hashes) + # Help text (different for remaining hashes, no ID needed) if table_type != "remaining": help_text = Static( "Click to toggle, 'r' for range select (click start, press 'r', click end)\n" @@ -1653,11 +1836,11 @@ class PolicyPrepWorkflowScreen(Screen): available_cols = [col for col in columns if col in df.columns] if available_cols: # Add checkbox column first - review_table.add_columns("ā˜", *available_cols) + review_table.add_columns("ā—‹", *available_cols) # Add rows with row keys for tracking for idx, row in df.iterrows(): - checkbox = "ā˜" # All start unchecked + checkbox = "ā—‹" # All start unchecked row_data = [] for col in available_cols: value = row.get(col, "") @@ -1669,7 +1852,7 @@ class PolicyPrepWorkflowScreen(Screen): row_data.append(str(value)) review_table.add_row(checkbox, *row_data, key=str(idx)) - logger.info(f"About to mount {table_id}") + logger.debug(f"About to mount {table_id}") # Final safety check before mounting table try: @@ -1682,7 +1865,7 @@ class PolicyPrepWorkflowScreen(Screen): pass content.mount(review_table) - logger.info(f"Successfully mounted {table_id}") + logger.debug(f"Successfully mounted {table_id}") # Row count display only (removed Select All, Clear, Delete buttons) if table_type != "remaining": @@ -1999,53 +2182,166 @@ class PolicyPrepWorkflowScreen(Screen): """Perform the actual application of changes.""" try: results = [] + errors = [] - # Apply path exclusions to policy - if self.destination_policy and self.primary_paths_df is not None: - # This would call the actual API methods - results.append("Applied path exclusions to policy") + # Apply path exclusions to policy (primary + secondary) + if self.destination_policy: + path_rules = [] + + # Process primary paths + if ( + self.primary_paths_df is not None + and not self.primary_paths_df.empty + ): + logger.info( + f"Processing {len(self.primary_paths_df)} primary paths" + ) + for _, row in self.primary_paths_df.groupby( + ["longestcfp", "file_extension"] + ): + path = row.iloc[0]["longestcfp"] + ext = row.iloc[0]["file_extension"] + # Format: C:\Path\**.ext + path_rule = f"{path}\\**{ext}" + path_rules.append(path_rule) + + # Process secondary paths + if ( + self.secondary_paths_df is not None + and not self.secondary_paths_df.empty + ): + logger.info( + f"Processing {len(self.secondary_paths_df)} secondary paths" + ) + for _, row in self.secondary_paths_df.groupby( + ["longestcfp", "file_extension"] + ): + path = row.iloc[0]["longestcfp"] + ext = row.iloc[0]["file_extension"] + path_rule = f"{path}\\**{ext}" + path_rules.append(path_rule) + + # Apply path rules to policy + if path_rules: + try: + logger.info( + f"Applying {len(path_rules)} path exclusions to policy {self.destination_policy.name}" + ) + response = self.api.policy_add_path_exclusions( + str(self.destination_policy.groupid), path_rules + ) + results.append( + f"āœ“ Added {len(path_rules)} path exclusions to policy" + ) + logger.info(f"Path exclusions applied successfully: {response}") + except Exception as e: + error_msg = f"āœ— Failed to add path exclusions: {str(e)}" + errors.append(error_msg) + logger.error(error_msg, exc_info=True) # Apply publishers to policy - if self.destination_policy and self.publishers_df is not None: - # This would call the actual API methods - results.append("Applied approved publishers to policy") + if ( + self.destination_policy + and self.publishers_df is not None + and not self.publishers_df.empty + ): + try: + publishers = self.publishers_df["publisher"].unique().tolist() + logger.info( + f"Applying {len(publishers)} publishers to policy {self.destination_policy.name}" + ) + response = self.api.policy_add_publishers( + str(self.destination_policy.groupid), publishers + ) + results.append( + f"āœ“ Added {len(publishers)} trusted publishers to policy" + ) + logger.info(f"Publishers applied successfully: {response}") + except Exception as e: + error_msg = f"āœ— Failed to add publishers: {str(e)}" + errors.append(error_msg) + logger.error(error_msg, exc_info=True) # Apply hashes to allowlist - if self.destination_allowlist and self.approved_df is not None: - # This would call the actual API methods - results.append("Applied approved hashes to allowlist") + if ( + self.destination_allowlist + and self.approved_df is not None + and not self.approved_df.empty + ): + try: + # Get unique hashes + hashes = self.approved_df["sha256"].unique().tolist() + logger.info( + f"Applying {len(hashes)} hashes to allowlist {self.destination_allowlist.name}" + ) + response = self.api.hash_add_to_allowlist( + str(self.destination_allowlist.applicationid), hashes + ) + results.append( + f"āœ“ Added {len(hashes):,} approved hashes to allowlist" + ) + logger.info(f"Hashes applied successfully: {response}") + except Exception as e: + error_msg = f"āœ— Failed to add hashes: {str(e)}" + errors.append(error_msg) + logger.error(error_msg, exc_info=True) - self._show_completion(results) + # Show completion with both results and errors + all_results = results + errors + self._show_completion(all_results, has_errors=len(errors) > 0) except Exception as e: - logger.error(f"Failed to apply changes: {e}", exc_info=True) - self.app.notify(f"Failed to apply changes: {str(e)}", severity="error") + logger.error(f"Critical failure in _perform_apply: {e}", exc_info=True) + self.app.notify(f"Critical failure: {str(e)}", severity="error") self._show_test_screen() - def _show_completion(self, results: List[str]) -> None: + def _show_completion(self, results: List[str], has_errors: bool = False) -> None: """Show completion screen.""" self.workflow_stage = "complete" content = self.query_one("#content_area", Vertical) content.remove_children() - summary = Static( - "Policy Preparation Complete!\n\n" - "The following changes have been applied:" - ) + # Title depends on whether there were errors + if has_errors: + title_text = "Policy Preparation Completed with Errors\n\n" "Results:" + title_color = "yellow" + else: + title_text = ( + "Policy Preparation Complete!\n\n" + "The following changes have been applied:" + ) + title_color = "green" + + summary = Static(title_text) summary.styles.margin = (1, 1) + summary.styles.text_style = "bold" + summary.styles.color = title_color content.mount(summary) for result in results: result_widget = Static(f" {result}") result_widget.styles.margin = (0, 2) + # Color based on success/failure + if result.startswith("āœ“"): + result_widget.styles.color = "green" + elif result.startswith("āœ—"): + result_widget.styles.color = "red" content.mount(result_widget) # Final message - final = Static( - f"\nPolicy '{self.destination_policy.name}' is now ready for enforcement!" - ) - final.styles.margin = (2, 1) - final.styles.color = "green" + if has_errors: + final = Static( + f"\nāš ļø Policy '{self.destination_policy.name}' was partially updated.\n" + "Please review errors above and retry failed operations manually." + ) + final.styles.margin = (2, 1) + final.styles.color = "yellow" + else: + final = Static( + f"\nāœ… Policy '{self.destination_policy.name}' is now ready for enforcement!" + ) + final.styles.margin = (2, 1) + final.styles.color = "green" content.mount(final) # Done button @@ -2078,7 +2374,7 @@ class PolicyPrepWorkflowScreen(Screen): ) # Determine if this row should be checked is_selected = row_key_str in selected_keys - checkbox = "ā˜‘ļø" if is_selected else "ā˜" + checkbox = "āœ“" if is_selected else "ā—‹" # Update the checkbox cell (first column, index 0) try: @@ -2315,8 +2611,15 @@ class PolicyPrepWorkflowScreen(Screen): """Handle button presses.""" button_id = event.button.id + # Introduction screen buttons + if button_id == "start_workflow": + self._show_source_policy_selection() + + elif button_id == "cancel_workflow": + self.app.pop_screen() + # Source policy selection buttons - if button_id == "select_none_source": + elif button_id == "select_none_source": self.selected_source_policy_ids.clear() # Refresh checkbox display self._refresh_table_checkboxes( @@ -2434,6 +2737,15 @@ class PolicyPrepWorkflowScreen(Screen): ) elif button_id == "continue_from_review": + # Check if both tabs have been reviewed + if not self.approved_tab_reviewed or not self.needs_review_tab_reviewed: + self.app.notify( + "Please review both 'Approved' and 'Needs Review' tabs before continuing.", + severity="warning", + timeout=5, + ) + return + # Validate that review is complete if (self.approved_df is None or self.approved_df.empty) and ( self.needs_review_df is None or self.needs_review_df.empty @@ -2449,6 +2761,15 @@ class PolicyPrepWorkflowScreen(Screen): self._show_path_building_screen() elif button_id == "build_preflight": + # Check if both required tabs have been reviewed + if not self.paths_tab_reviewed or not self.publishers_tab_reviewed: + self.app.notify( + "Please review both 'Paths' and 'Publishers' tabs before continuing.", + severity="warning", + timeout=5, + ) + return + # Validate that path review is complete if (self.primary_paths_df is None or self.primary_paths_df.empty) and ( self.publishers_df is None or self.publishers_df.empty diff --git a/TUI/TUI.py b/TUI/TUI.py deleted file mode 100644 index 1a6f603..0000000 --- a/TUI/TUI.py +++ /dev/null @@ -1,444 +0,0 @@ -# Copyright (C) 2025 James Brotosky, Brandon Wickline -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU Affero General Public License as published -# by the Free Software Foundation, either version 3 of the License, or -# (at your option) any later version. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU Affero General Public License for more details. -# -# You should have received a copy of the GNU Affero General Public License -# along with this program. If not, see . - -import logging -import os -from typing import Optional - -import dotenv -from textual.app import App, ComposeResult -from textual.containers import Vertical -from textual.message import Message -from textual.reactive import reactive -from textual.screen import Screen -from textual.widgets import ( - Button, - DirectoryTree, - Footer, - Header, - Static, - Tab, - Tabs, -) - -from models.agent import Agent -from models.policy import Policy -from services.API import AirlockAPIWrapper -from TUI.Screens.moveagentworkflowscreen import MoveAgentWorkflowScreen -from TUI.Screens.otpactivityscreen import OTPActivitiesScreen -from TUI.Screens.otprevokescreen import OTPRevokeScreen -from TUI.Screens.otpworkflowscreen import OTPWorkflowScreen -from TUI.Screens.policyprepworkflowscreen import PolicyPrepWorkflowScreen -from TUI.Screens.quietagentworkflowscreen import QuietAgentWorkflowScreen -from TUI.Themes.theme_amber_terminal import get_amber_terminal_theme -from TUI.Themes.theme_retro_terminal import get_retro_terminal_theme -from TUI.Themes.themeselector import ThemeSelector -from TUI.Widgets.agentmoveoperations import AgentMoveOperations -from TUI.Widgets.multiagentselector import MultiAgentSelector -from TUI.Widgets.policytreewidget import PolicyTreeWidget -from TUI.Widgets.resultsdisplay import ResultsDisplay -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() - -# --------------------------------------------------------------------------- -# GLOBAL STASH -# --------------------------------------------------------------------------- - -_APP_RESTART_REASON = None - -logger = logging.getLogger(__name__) - - -# --------------------------------------------------------------------------- -# helper to persist TEXTUAL_THEME to *user* config and mirror to .env -# --------------------------------------------------------------------------- -def _persist_user_theme(theme_name: str) -> None: - """ - 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" - - try: - 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) - - -# --------------------------------------------------------------------------- -# 1) SCREEN -# --------------------------------------------------------------------------- -class MainMenuScreen(Screen): - api: AirlockAPIWrapper - current_tab = reactive("") - - BUTTON_DEFS = { - "agent_actions": [ - ( - "šŸ–„ļø - Find agent, Move agent, or Generate One Time Pass", - "move_agent_workflow_button", - ), - ("šŸŽ« - Review and approve OTP Activities", "otp_activities_button"), - ("šŸ›‘ - Revoke Active OTP Session", "otp_revoke_button"), - ], - "policy": [ - ("āš–ļø - Prepare Policy For Enforcement", "policy_prep_button"), - ("šŸ”• - Find and Move Quiet Hosts to Enforcement", "find_quiet_button"), - ], - } - - def __init__(self) -> None: - super().__init__() - 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() - self.working_dir = wd - - def _make_buttons_for(self, tab_id: str) -> Vertical: - defs = self.BUTTON_DEFS.get(tab_id, []) - buttons = [] - for label, btn_id in defs: - btn = Button(label, id=btn_id) - btn.styles.width = "100%" - buttons.append(btn) - return Vertical(*buttons) - - def compose(self) -> ComposeResult: - yield Header(show_clock=True, icon="āš™") - - tabs = [ - Tab("Tree View", id="p_tree"), - Tab("Agents", id="agent_actions"), - Tab("Directory", id="dir"), - Tab("Settings", id="settings"), - ] - - if self.extras == "POLICYPREP": - tabs.insert(2, Tab("Policy Prep", id="policy")) - - yield Tabs(*tabs, id="tabs") - yield Vertical(id="content") - yield Footer() - - def on_mount(self) -> None: - self.switch_tab("agent_actions") - - # focus helpers - def _get_content_buttons(self) -> list[Button]: - content = self.query_one("#content", Vertical) - return list(content.query(Button)) - - def _focus_first_button(self) -> None: - buttons = self._get_content_buttons() - if buttons: - buttons[0].focus() - - def _focus_tabs(self) -> None: - tabs = self.query_one("#tabs", Tabs) - tabs.focus() - - def _focus_nearby_button(self, direction: int) -> None: - buttons = self._get_content_buttons() - if not buttons: - return - - try: - current = next(i for i, b in enumerate(buttons) if b.has_focus) - except StopIteration: - if direction > 0: - buttons[0].focus() - else: - buttons[-1].focus() - return - - if direction < 0 and current == 0: - self._focus_tabs() - return - - new_index = current + direction - if 0 <= new_index < len(buttons): - buttons[new_index].focus() - - def switch_tab(self, tab_id: str) -> None: - self.current_tab = tab_id - content = self.query_one("#content", Vertical) - content.remove_children() - - if tab_id in self.BUTTON_DEFS: - content.mount(self._make_buttons_for(tab_id)) - self.call_later(self._focus_first_button) - elif tab_id == "dir": - content.mount(DirectoryTree(self.working_dir, id="dir_tree")) - elif tab_id == "p_tree": - content.mount(PolicyTreeWidget(self.app.policies, self.app.devices)) - elif tab_id == "settings": - content.mount(ThemeSelector()) - else: - content.mount(Static(f"Unknown tab: {tab_id}")) - - def on_tabs_tab_activated(self, event: Tabs.TabActivated) -> None: - self.switch_tab(event.tab.id) - - def on_multi_agent_selector_agents_selected( - self, message: MultiAgentSelector.AgentsSelected - ) -> None: - """Handle selected agents from AgentSelector.""" - global _APP_RESTART_REASON - selected_agents = message.selected_agents - logger.info("Selected agents: %s", selected_agents) - # TODO: Implement actual handling of selected agents - _APP_RESTART_REASON = ("multi_agent_action", selected_agents) - self.app.exit() - - def on_theme_selector_theme_selected( - self, message: ThemeSelector.ThemeSelected - ) -> None: - """Handle theme selection from ThemeSelector.""" - global _APP_RESTART_REASON - _persist_user_theme(message.theme_name) - _APP_RESTART_REASON = ("restart",) - self.app.exit() - - def on_agent_move_operations_operation_complete( - self, message: AgentMoveOperations.OperationComplete - ) -> None: - """Handle completion of agent move operation - show results.""" - logger.info( - "Agent move operation completed: %s, %d successful, %d unsuccessful", - message.operation, - len(message.successful), - len(message.unsuccessful), - ) - - # Format results for display - successful_text = "\n".join( - [f"{agent.hostname}" for agent, _ in message.successful] - ) - unsuccessful_text = "\n".join( - [f"{agent.hostname}: {error}" for agent, error in message.unsuccessful] - ) - - # Remove the operations widget - try: - ops_widget = self.query_one(AgentMoveOperations) - ops_widget.remove() - except Exception: - pass - - # Show results - self.query_one("#content", Vertical).mount( - ResultsDisplay(message.operation, successful_text, unsuccessful_text) - ) - - def on_results_display_go_back(self, message: ResultsDisplay.GoBack) -> None: - """Handle back button from results display.""" - try: - results_widget = self.query_one(ResultsDisplay) - results_widget.remove() - except Exception: - pass - # Return to main menu - self.app.pop_screen() - - def on_directory_tree_file_selected( - self, event: DirectoryTree.FileSelected - ) -> None: - path = event.path - logger.debug("Directory file selected: %s", path) - try: - open_directory(str(path)) - except Exception as exc: - logger.error("Failed to open %s: %s", path, exc) - self.app.bell() - - def on_button_pressed(self, event: Button.Pressed) -> None: - button_id = event.button.id - logger.debug("Button pressed: %s", button_id) - - match button_id: - case "move_agent_workflow_button": - self.app.push_screen(MoveAgentWorkflowScreen(self.app.devices)) - event.stop() - - case "otp_generate_button": - self.app.push_screen(OTPWorkflowScreen(self.app.devices)) - event.stop() - - case "find_quiet_button": - self.app.push_screen( - QuietAgentWorkflowScreen(self.app.api, self.app.policies) - ) - event.stop() - return - - case "otp_activities_button": - self.app.push_screen(OTPActivitiesScreen()) - event.stop() - return - - case "otp_revoke_button": - self.app.push_screen(OTPRevokeScreen()) - event.stop() - return - - case "policy_prep_button": - # Use the new TUI workflow screen instead of legacy - self.app.push_screen( - PolicyPrepWorkflowScreen(self.app.api, self.app.policies) - ) - event.stop() - return - - case _: - self.app.bell() - logger.warning("Unknown button pressed: %s", button_id) - return - - -# --------------------------------------------------------------------------- -# 2) APP -# --------------------------------------------------------------------------- -class Loxide(App[Message]): - api: AirlockAPIWrapper - working_dir: str - policies: Optional[list[Policy]] - devices: Optional[list[Agent]] - - CSS = """ - #logo { - width: 100%; - content-align: center middle; - text-align: center; - } - """ - BINDINGS = [ - ("q", "quit", "Quit"), - ("f", "open_fe", "Launch Explorer"), - ("r", "refresh", "Refresh"), - ] - - def __init__(self, api: AirlockAPIWrapper): - self._textual_theme = get_user_value("TEXTUAL_THEME", str, "textual-dark") - super().__init__() - self.api = api - wd = load_env("WORKING_DIR") or os.getcwd() - if not os.path.isdir(wd): - wd = os.getcwd() - self.working_dir = wd - # Initial data load - self.refresh_data() - - def refresh_data(self) -> None: - """Public method to refresh policies and devices from the API.""" - try: - self.policies = [ - Policy(**row.to_dict()) - for _, row in self.api.policy_find_all().iterrows() - ] - self.devices = [ - Agent(**row.to_dict()) - for _, row in self.api.agent_find_all().iterrows() - ] - if self.policies and self.devices: - for agent in self.devices: - agent.enrich_with_policies(self.policies) - logger.debug( - f"Enriched {len(self.devices)} agents with policy information" - ) - except Exception as exc: - logger.error("Failed to load policies/devices: %s", exc) - self.policies = None - self.devices = None - - def on_mount(self, api: AirlockAPIWrapper) -> None: - self.register_theme(get_retro_terminal_theme()) - self.register_theme(get_amber_terminal_theme()) - self.theme = self._textual_theme - self.push_screen(MainMenuScreen()) - - def action_refresh(self) -> None: - self.refresh_data() - - def action_quit(self) -> None: - global _APP_RESTART_REASON - _APP_RESTART_REASON = None - self.exit() - - def action_open_fe(self) -> None: - """Open the working directory in the OS file manager (footer binding).""" - path_to_open = self.working_dir or os.getcwd() - try: - open_directory(path_to_open) - except Exception as exc: - logger.error("Failed to open directory %s: %s", path_to_open, exc) - self.bell() # optional feedback - - -# --------------------------------------------------------------------------- -# 3) PUBLIC ENTRYPOINT -# --------------------------------------------------------------------------- -def run_Loxide(api: AirlockAPIWrapper) -> None: - global _APP_RESTART_REASON - base_dir = get_base_directory() - env_path = base_dir / ".env" - dotenv.load_dotenv(dotenv_path=env_path, override=True) - - max_attempts = 5 - attempts = 0 - - while attempts < max_attempts: - attempts += 1 - logger.debug("Starting app loop iteration (attempt %d)", attempts) - _APP_RESTART_REASON = None - app = Loxide(api) - - try: - app.run() - except SystemExit as exc: - if exc.code != 0: - logger.debug("Caught SystemExit from Textual: %s", exc) - raise - - reason = _APP_RESTART_REASON - logger.debug("After app.run(), _APP_RESTART_REASON = %r", reason) - - if not reason: - logger.debug("No restart reason, exiting loop") - break - - if reason[0] == "restart": - logger.debug("Restarting app loop") - continue - - if reason[0] == "multi_agent_action": - logger.info("Multi-agent action with selected agents: %s", reason[1]) - continue - - logger.error("Unknown restart reason: %r", reason) - break - - -# --------------------------------------------------------------------------- -# 4) DEV -# --------------------------------------------------------------------------- -if __name__ == "__main__": - api = AirlockAPIWrapper() - run_Loxide(api) diff --git a/requirements.txt b/requirements.txt index 6492e2b..18ab1da 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,14 +1,26 @@ +# Core TUI dependencies +textual==6.5.0 + +# API and data handling +Requests==2.32.5 +pandas==2.3.3 +numpy==2.3.4 + +# Database +pymongo==4.15.3 + +# Security and encryption cryptography==46.0.3 keyring==25.6.0 -numpy==2.3.4 -pandas==2.3.3 -pymongo==4.15.3 + +# Environment management python-dotenv==1.2.1 -Requests==2.32.5 -textual==6.5.0 + +# Utilities tqdm==4.67.1 urllib3==2.5.0 pyperclip==1.11.0 +# Custom/Private packages --extra-index-url https://git.racooncity.org/api/packages/brotoskyj/pypi/simple/ airlock_libs==6.0.0 \ No newline at end of file