diff --git a/Loxide.py b/Loxide.py
index 912ba22..46f97af 100644
--- a/Loxide.py
+++ b/Loxide.py
@@ -46,6 +46,7 @@ from models.agent import Agent
from models.policy import Policy
from services.API import AirlockAPIWrapper
from services.security import getAPI
+from TUI.Screens.executionhistoryscreen import ExecutionHistoryScreen
from TUI.Screens.moveagentworkflowscreen import MoveAgentWorkflowScreen
from TUI.Screens.otpactivityscreen import OTPActivitiesScreen
from TUI.Screens.otprevokescreen import OTPRevokeScreen
@@ -59,6 +60,7 @@ 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 TUI.Widgets.serverlogwidget import ServerLogWidget
from utils.configmanager import (
get_system_value,
get_user_value,
@@ -107,16 +109,29 @@ class MainMenuScreen(Screen):
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"),
+ {
+ "label": "🖥️ - Multi-Agent Operations",
+ "id": "move_agent_workflow_button",
+ "description": "Select agents to: Move policies, Generate OTPs, Toggle audit/enforcement, View history, Export data",
+ },
+ {
+ "label": "🎫 - Review and approve OTP Activities",
+ "id": "otp_activities_button",
+ },
+ {
+ "label": "🛑 - Revoke Active OTP Session",
+ "id": "otp_revoke_button",
+ },
],
"policy": [
- ("⚖️ - Prepare Policy For Enforcement", "policy_prep_button"),
- ("🔕 - Find and Move Quiet Hosts to Enforcement", "find_quiet_button"),
+ {
+ "label": "⚖️ - Prepare Policy For Enforcement",
+ "id": "policy_prep_button",
+ },
+ {
+ "label": "🔕 - Find and Move Quiet Hosts to Enforcement",
+ "id": "find_quiet_button",
+ },
],
}
@@ -130,19 +145,40 @@ class MainMenuScreen(Screen):
def _make_buttons_for(self, tab_id: str) -> Vertical:
defs = self.BUTTON_DEFS.get(tab_id, [])
- buttons = []
- for label, btn_id in defs:
+ widgets = []
+ for item in defs:
+ # Support both old tuple format and new dict format
+ if isinstance(item, dict):
+ label = item["label"]
+ btn_id = item["id"]
+ description = item.get("description")
+ else:
+ # Old tuple format: (label, id)
+ label, btn_id = item
+ description = None
+
btn = Button(label, id=btn_id)
btn.styles.width = "100%"
- buttons.append(btn)
- return Vertical(*buttons)
+ widgets.append(btn)
+
+ # Add description text if provided
+ if description:
+ desc_text = Static(description, classes="button_description")
+ desc_text.styles.width = "100%"
+ desc_text.styles.color = "ansi_bright_black"
+ desc_text.styles.text_align = "center"
+ desc_text.styles.margin = (0, 0, 1, 0)
+ widgets.append(desc_text)
+
+ return Vertical(*widgets)
def compose(self) -> ComposeResult:
yield Header(show_clock=True, icon="⚙")
tabs = [
- Tab("Tree View", id="p_tree"),
Tab("Agents", id="agent_actions"),
+ Tab("Tree View", id="p_tree"),
+ Tab("Server Log", id="server_log"),
Tab("Directory", id="dir"),
Tab("Settings", id="settings"),
]
@@ -157,6 +193,18 @@ class MainMenuScreen(Screen):
def on_mount(self) -> None:
self.switch_tab("agent_actions")
+ def on_key(self, event) -> None:
+ """Handle up/down arrow keys for button navigation."""
+ if event.key == "down":
+ self._focus_nearby_button(1)
+ event.prevent_default()
+ event.stop()
+ elif event.key == "up":
+ self._focus_nearby_button(-1)
+ event.prevent_default()
+ event.stop()
+ # left/right are handled by Textual's default tab navigation
+
# focus helpers
def _get_content_buttons(self) -> list[Button]:
content = self.query_one("#content", Vertical)
@@ -200,7 +248,8 @@ class MainMenuScreen(Screen):
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 == "server_log":
+ content.mount(ServerLogWidget(self.app.api))
elif tab_id == "dir":
content.mount(DirectoryTree(self.working_dir, id="dir_tree"))
elif tab_id == "p_tree":
@@ -274,6 +323,66 @@ class MainMenuScreen(Screen):
# Return to main menu
self.app.pop_screen()
+ def on_policy_tree_widget_view_execution_history(
+ self, message: PolicyTreeWidget.ViewExecutionHistory
+ ) -> None:
+ """Handle request to view execution history for a device from tree view."""
+ logger.info("Viewing execution history for device: %s", message.device.hostname)
+ self.app.push_screen(ExecutionHistoryScreen([message.device]))
+ message.stop()
+
+ def on_policy_tree_widget_generate_otp(
+ self, message: PolicyTreeWidget.GenerateOTP
+ ) -> None:
+ """Handle request to generate OTP for a device from tree view."""
+ logger.info("Generating OTP for device: %s", message.device.hostname)
+ self.app.push_screen(OTPWorkflowScreen([message.device]))
+ message.stop()
+
+ def on_policy_tree_widget_toggle_enforcement(
+ self, message: PolicyTreeWidget.ToggleEnforcement
+ ) -> None:
+ """Handle request to toggle enforcement for a device from tree view."""
+ logger.info("Toggling enforcement for device: %s", message.device.hostname)
+
+ try:
+ from services.agenthandler import moveAgentToRelatedPolicy
+ from utils.configmanager import get_system_json
+
+ policy_relationship_map = get_system_json("POLICY_MAP_ENF_AUD", "{}")
+
+ # Determine current mode and toggle
+ if message.device.groupid in policy_relationship_map:
+ # Currently in enforcement, move to audit
+ result = moveAgentToRelatedPolicy(self.app.api, message.device, "audit")
+ mode = "audit"
+ else:
+ # Currently in audit, move to enforcement
+ result = moveAgentToRelatedPolicy(
+ self.app.api, message.device, "enforcement"
+ )
+ mode = "enforcement"
+
+ logger.info(f"Successfully toggled {message.device.hostname} to {mode}")
+
+ # Refresh data at the app level
+ self.app.refresh_data()
+
+ # Refresh the tree widget with new data
+ try:
+ tree_widget = self.query_one(PolicyTreeWidget)
+ tree_widget.refresh_data(self.app.policies, self.app.devices)
+ except:
+ pass
+
+ except Exception as e:
+ logger.error(
+ f"Failed to toggle enforcement for {message.device.hostname}: {e}"
+ )
+ self.app.bell()
+
+ message.stop()
+
def on_directory_tree_file_selected(
self, event: DirectoryTree.FileSelected
) -> None:
@@ -409,9 +518,9 @@ class Loxide(App[Message]):
# ---------------------------------------------------------------------------
-# 3) PUBLIC ENTRYPOINT
+# 3) PUBLIC ENTRYPOINT - Updated to accept attach_notification_handler
# ---------------------------------------------------------------------------
-def run_Loxide(api: AirlockAPIWrapper) -> None:
+def run_Loxide(api: AirlockAPIWrapper, attach_notification_handler=None) -> None:
global _APP_RESTART_REASON
base_dir = get_base_directory()
env_path = base_dir / ".env"
@@ -426,6 +535,10 @@ def run_Loxide(api: AirlockAPIWrapper) -> None:
_APP_RESTART_REASON = None
app = Loxide(api)
+ # Attach the notification handler if provided
+ if attach_notification_handler:
+ attach_notification_handler(app)
+
try:
app.run()
except SystemExit as exc:
@@ -453,12 +566,13 @@ def run_Loxide(api: AirlockAPIWrapper) -> None:
# ---------------------------------------------------------------------------
-# 4) MAIN FUNCTION
+# 4) MAIN FUNCTION - Updated to get and pass attach_notification_handler
# ---------------------------------------------------------------------------
def main():
irtang()
# Determine working directory, setup directory, configure logging, sent env, get API and URL if not already stored
- setup()
+ # setup() now returns a function to attach the notification handler
+ attach_notification_handler = setup()
logger = logging.getLogger(__name__)
try:
@@ -485,7 +599,7 @@ def main():
base_url=str(url),
api_key=api_key,
)
- run_Loxide(api)
+ run_Loxide(api, attach_notification_handler)
if __name__ == "__main__":
diff --git a/TUI/Screens/executionhistoryscreen.py b/TUI/Screens/executionhistoryscreen.py
new file mode 100644
index 0000000..abf0648
--- /dev/null
+++ b/TUI/Screens/executionhistoryscreen.py
@@ -0,0 +1,639 @@
+# 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 .
+
+
+from datetime import datetime, timedelta
+import logging
+import os
+from typing import List
+
+import pandas as pd
+from textual.app import ComposeResult
+from textual.binding import Binding
+from textual.containers import Horizontal, Vertical
+from textual.screen import Screen
+from textual.widgets import (
+ Button,
+ DataTable,
+ Footer,
+ Header,
+ Label,
+ Select,
+ Static,
+)
+
+from models.agent import Agent
+from models.execution import ExecutionHistoryRecord
+from utils.configmanager import load_env
+
+logger = logging.getLogger(__name__)
+
+
+class ExecutionHistoryScreen(Screen):
+ """
+ A screen for viewing and exporting execution history for selected agents.
+
+ This screen allows users to:
+ 1. Select a start date and end date using dropdown selects
+ 2. Fetch execution history for all selected agents
+ 3. View the results in a DataTable
+ 4. Export the results to CSV using a keybinding
+
+ Attributes:
+ agents (List[Agent]): List of agents to fetch execution history for
+ execution_data (pd.DataFrame): Combined execution history data
+ working_dir (str): Directory for CSV exports
+ """
+
+ DEFAULT_CSS = """
+ ExecutionHistoryScreen {
+ align: center top;
+ }
+
+ #main_container {
+ width: 95%;
+ height: 1fr;
+ border: solid $primary;
+ padding: 1;
+ }
+
+ #title {
+ text-style: bold;
+ color: $text;
+ text-align: center;
+ margin-bottom: 1;
+ }
+
+ #date_container {
+ height: auto;
+ margin-bottom: 1;
+ }
+
+ #start_date_row, #end_date_row {
+ height: auto;
+ align-horizontal: left;
+ margin-bottom: 1;
+ }
+
+ .date_label {
+ width: 8;
+ margin-right: 1;
+ }
+
+ .date_selector {
+ width: 18;
+ margin: 0 1;
+ }
+
+ #quick_buttons_row {
+ height: auto;
+ align-horizontal: center;
+ margin-bottom: 1;
+ }
+
+ .quick_select_btn {
+ margin: 0 1;
+ }
+
+ #button_row {
+ height: auto;
+ align-horizontal: center;
+ margin-top: 1;
+ margin-bottom: 1;
+ }
+
+ Button {
+ margin: 0 1;
+ }
+
+ #status_label {
+ text-align: center;
+ color: $accent;
+ margin-bottom: 1;
+ }
+
+ #results_container {
+ height: 1fr;
+ display: none;
+ }
+
+ #results_button_row {
+ height: auto;
+ align-horizontal: center;
+ margin-bottom: 1;
+ }
+
+ #history_table {
+ height: 1fr;
+ border: solid $primary;
+ }
+
+ DataTable > .datatable--header {
+ text-style: bold;
+ background: $primary 20%;
+ }
+ """
+
+ BINDINGS = [
+ Binding("escape", "close_screen", "Close"),
+ Binding("e", "export_csv", "Export CSV"),
+ Binding("q", "close_screen", "Quit"),
+ ]
+
+ def __init__(self, agents: List[Agent]):
+ """
+ Initialize the ExecutionHistoryScreen.
+
+ Args:
+ agents (List[Agent]): List of agents to fetch execution history for
+ """
+ super().__init__()
+ self.agents = agents
+ self.execution_data = pd.DataFrame()
+ self.working_dir = load_env("WORKING_DIR") or os.getcwd()
+
+ # Generate dropdown options
+ today = datetime.now().date()
+
+ # Month options - format is (display_text, value)
+ self.month_options = [
+ ("January", "01"),
+ ("February", "02"),
+ ("March", "03"),
+ ("April", "04"),
+ ("May", "05"),
+ ("June", "06"),
+ ("July", "07"),
+ ("August", "08"),
+ ("September", "09"),
+ ("October", "10"),
+ ("November", "11"),
+ ("December", "12"),
+ ]
+
+ # Day options (1-31) - format is (display_text, value)
+ self.day_options = [(f"{i}", f"{i:02d}") for i in range(1, 32)]
+
+ # Year options (current year back 5 years) - format is (display_text, value)
+ current_year = today.year
+ self.year_options = [
+ (str(year), str(year)) for year in range(current_year, current_year - 6, -1)
+ ]
+
+ # Default dates: last 30 days
+ start_date = today - timedelta(days=30)
+ self.start_month = f"{start_date.month:02d}"
+ self.start_day = f"{start_date.day:02d}"
+ self.start_year = str(start_date.year)
+
+ self.end_month = f"{today.month:02d}"
+ self.end_day = f"{today.day:02d}"
+ self.end_year = str(today.year)
+
+ def compose(self) -> ComposeResult:
+ """Build the UI layout."""
+ yield Header(show_clock=True, icon="📊")
+
+ with Vertical(id="main_container"):
+ title_text = f"Execution History - {len(self.agents)} Agent(s)"
+ yield Static(title_text, id="title")
+
+ # Date selection area
+ with Vertical(id="date_container"):
+ yield Label("Select Date Range:")
+
+ # Start date row
+ with Horizontal(id="start_date_row"):
+ yield Label("From:", classes="date_label")
+ yield Select(
+ options=self.month_options,
+ value=self.start_month,
+ id="start_month_select",
+ classes="date_selector",
+ )
+ yield Select(
+ options=self.day_options,
+ value=self.start_day,
+ id="start_day_select",
+ classes="date_selector",
+ )
+ yield Select(
+ options=self.year_options,
+ value=self.start_year,
+ id="start_year_select",
+ classes="date_selector",
+ )
+
+ # End date row
+ with Horizontal(id="end_date_row"):
+ yield Label("To:", classes="date_label")
+ yield Select(
+ options=self.month_options,
+ value=self.end_month,
+ id="end_month_select",
+ classes="date_selector",
+ )
+ yield Select(
+ options=self.day_options,
+ value=self.end_day,
+ id="end_day_select",
+ classes="date_selector",
+ )
+ yield Select(
+ options=self.year_options,
+ value=self.end_year,
+ id="end_year_select",
+ classes="date_selector",
+ )
+
+ # Quick select buttons
+ with Horizontal(id="quick_buttons_row"):
+ yield Button(
+ "1 Day",
+ id="quick_1day",
+ classes="quick_select_btn",
+ variant="default",
+ )
+ yield Button(
+ "1 Week",
+ id="quick_1week",
+ classes="quick_select_btn",
+ variant="default",
+ )
+ yield Button(
+ "30 Days",
+ id="quick_30days",
+ classes="quick_select_btn",
+ variant="default",
+ )
+
+ # Buttons
+ with Horizontal(id="button_row"):
+ yield Button("Fetch History", id="fetch_btn", variant="primary")
+ yield Button("Close", id="close_btn", variant="error")
+
+ # Status
+ yield Static(
+ "Select date range and click 'Fetch History'", id="status_label"
+ )
+
+ # Results container (hidden initially, shown after fetch)
+ with Vertical(id="results_container"):
+ with Horizontal(id="results_button_row"):
+ yield Button("Export CSV", id="export_btn", variant="success")
+ yield Button("Back", id="back_btn", variant="default")
+ yield DataTable(id="history_table")
+
+ yield Footer()
+
+ def on_mount(self) -> None:
+ """Initialize the table when screen is mounted."""
+ table = self.query_one("#history_table", DataTable)
+ table.cursor_type = "row"
+ table.zebra_stripes = True
+
+ # Initially empty - will populate after fetch
+ logger.info(f"ExecutionHistoryScreen mounted with {len(self.agents)} agents")
+
+ def on_select_changed(self, event: Select.Changed) -> None:
+ """Handle date selection changes."""
+ select_id = event.select.id
+
+ if select_id == "start_month_select":
+ self.start_month = event.value
+ logger.debug(f"Start month changed to: {self.start_month}")
+ elif select_id == "start_day_select":
+ self.start_day = event.value
+ logger.debug(f"Start day changed to: {self.start_day}")
+ elif select_id == "start_year_select":
+ self.start_year = event.value
+ logger.debug(f"Start year changed to: {self.start_year}")
+ elif select_id == "end_month_select":
+ self.end_month = event.value
+ logger.debug(f"End month changed to: {self.end_month}")
+ elif select_id == "end_day_select":
+ self.end_day = event.value
+ logger.debug(f"End day changed to: {self.end_day}")
+ elif select_id == "end_year_select":
+ self.end_year = event.value
+ logger.debug(f"End year changed to: {self.end_year}")
+
+ def _set_quick_date_range(self, days: int) -> None:
+ """Set the date range based on quick select button."""
+ today = datetime.now().date()
+ start_date = today - timedelta(days=days)
+
+ # Update internal values
+ self.start_month = f"{start_date.month:02d}"
+ self.start_day = f"{start_date.day:02d}"
+ self.start_year = str(start_date.year)
+
+ self.end_month = f"{today.month:02d}"
+ self.end_day = f"{today.day:02d}"
+ self.end_year = str(today.year)
+
+ # Update the Select widgets
+ try:
+ self.query_one("#start_month_select", Select).value = self.start_month
+ self.query_one("#start_day_select", Select).value = self.start_day
+ self.query_one("#start_year_select", Select).value = self.start_year
+
+ self.query_one("#end_month_select", Select).value = self.end_month
+ self.query_one("#end_day_select", Select).value = self.end_day
+ self.query_one("#end_year_select", Select).value = self.end_year
+
+ self.app.notify(
+ f"Date range set to last {days} day(s)",
+ severity="information",
+ timeout=2,
+ )
+ logger.info(f"Quick select: Set date range to last {days} days")
+ except Exception as e:
+ logger.error(f"Failed to update date selects: {e}")
+
+ def _show_date_selection(self) -> None:
+ """Show the date selection view and hide results."""
+ try:
+ self.query_one("#date_container").styles.display = "block"
+ self.query_one("#button_row").styles.display = "block"
+ self.query_one("#status_label").styles.display = "block"
+ self.query_one("#results_container").styles.display = "none"
+ except Exception as e:
+ logger.error(f"Failed to show date selection: {e}")
+
+ def _show_results(self) -> None:
+ """Hide date selection view and show results."""
+ try:
+ self.query_one("#date_container").styles.display = "none"
+ self.query_one("#button_row").styles.display = "none"
+ self.query_one("#status_label").styles.display = "none"
+ self.query_one("#results_container").styles.display = "block"
+ except Exception as e:
+ logger.error(f"Failed to show results: {e}")
+
+ def on_button_pressed(self, event: Button.Pressed) -> None:
+ """Handle button clicks."""
+ if event.button.id == "fetch_btn":
+ self._fetch_execution_history()
+ elif event.button.id == "export_btn":
+ self._export_to_csv()
+ elif event.button.id == "close_btn":
+ self.app.pop_screen()
+ elif event.button.id == "back_btn":
+ self._show_date_selection()
+ elif event.button.id == "quick_1day":
+ self._set_quick_date_range(days=1)
+ elif event.button.id == "quick_1week":
+ self._set_quick_date_range(days=7)
+ elif event.button.id == "quick_30days":
+ self._set_quick_date_range(days=30)
+
+ def _fetch_execution_history(self) -> None:
+ """Fetch execution history for all selected agents."""
+ status_label = self.query_one("#status_label", Static)
+ status_label.update("â³ Fetching execution history...")
+
+ # Disable buttons during fetch
+ fetch_btn = self.query_one("#fetch_btn", Button)
+ export_btn = self.query_one("#export_btn", Button)
+ fetch_btn.disabled = True
+ export_btn.disabled = True
+
+ api = self.app.api
+ all_history = []
+
+ try:
+ # Construct dates from dropdowns
+ start_date_str = f"{self.start_year}-{self.start_month}-{self.start_day}"
+ end_date_str = f"{self.end_year}-{self.end_month}-{self.end_day}"
+
+ # Validate dates
+ try:
+ start_dt = datetime.strptime(start_date_str, "%Y-%m-%d")
+ end_dt = datetime.strptime(end_date_str, "%Y-%m-%d")
+ except ValueError as e:
+ status_label.update(f"⌠Invalid date: {str(e)}")
+ fetch_btn.disabled = False
+ export_btn.disabled = False
+ self.app.notify(f"Invalid date selected: {str(e)}", severity="error")
+ return
+
+ if start_dt > end_dt:
+ status_label.update("⌠Error: Start date must be before end date")
+ fetch_btn.disabled = False
+ export_btn.disabled = False
+ return
+
+ # Fetch history for each agent
+ for i, agent in enumerate(self.agents):
+ try:
+ status_label.update(
+ f"â³ Fetching history for {agent.hostname} ({i+1}/{len(self.agents)})..."
+ )
+
+ # Call API - note the API expects 'dateto' first, then 'datefrom'
+ history = api.history_execution(
+ today=end_date_str,
+ date_selected=start_date_str,
+ agent_name=agent.hostname,
+ )
+
+ if history:
+ # Add agent hostname to each record for identification
+ for record in history:
+ record["agent_hostname"] = agent.hostname
+ all_history.extend(history)
+ logger.info(
+ f"Fetched {len(history)} records for {agent.hostname}"
+ )
+ else:
+ logger.info(f"No history found for {agent.hostname}")
+
+ except Exception as e:
+ logger.error(f"Failed to fetch history for {agent.hostname}: {e}")
+ self.app.notify(
+ f"Warning: Failed to fetch history for {agent.hostname}",
+ severity="warning",
+ )
+
+ # Convert to DataFrame
+ if all_history:
+ status_label.update(
+ "â³ Enriching execution data with hash information..."
+ )
+
+ # Normalize field names (handle API typos)
+ for record in all_history:
+ if "policver" in record and "policyver" not in record:
+ record["policyver"] = record.pop("policver")
+
+ # Convert dict records to ExecutionHistoryRecord objects
+ execution_records = []
+ for record in all_history:
+ try:
+ execution_records.append(ExecutionHistoryRecord(**record))
+ except TypeError as e:
+ logger.warning(f"Failed to create ExecutionHistoryRecord: {e}")
+ # If it fails, just keep the dict
+ continue
+
+ # Enrich with hash data if we have ExecutionHistoryRecord objects
+ if execution_records:
+ try:
+ enriched_records = ExecutionHistoryRecord.enrich_with_hashes(
+ api, execution_records
+ )
+ logger.info(
+ f"Enriched {len(enriched_records)} records with hash data"
+ )
+
+ # Convert back to DataFrame
+ self.execution_data = pd.DataFrame(
+ [r.__dict__ for r in enriched_records]
+ )
+
+ # Flatten hash_obj if present
+ if (
+ not self.execution_data.empty
+ and "hash_obj" in self.execution_data.columns
+ ):
+ hash_df = self.execution_data["hash_obj"].apply(
+ lambda h: (
+ h.to_dict() if h and hasattr(h, "to_dict") else {}
+ )
+ )
+ self.execution_data = pd.concat(
+ [
+ self.execution_data.drop(columns=["hash_obj"]),
+ hash_df,
+ ],
+ axis=1,
+ )
+ except Exception as e:
+ logger.warning(f"Failed to enrich with hashes: {e}")
+ # Fall back to plain DataFrame
+ self.execution_data = pd.DataFrame(all_history)
+ else:
+ # If we couldn't create any ExecutionHistoryRecord objects, just use raw data
+ self.execution_data = pd.DataFrame(all_history)
+
+ self._populate_table()
+ self._show_results() # Switch to results view
+ self.app.notify(
+ f"Successfully loaded {len(self.execution_data)} records",
+ severity="information",
+ )
+ else:
+ status_label.update(
+ "â„¹ï¸ No execution history found for selected agents/dates"
+ )
+ self.app.notify("No execution history found", severity="information")
+ self.execution_data = pd.DataFrame()
+
+ except Exception as e:
+ logger.error(f"Error fetching execution history: {e}")
+ status_label.update(f"⌠Error: {str(e)}")
+ self.app.notify(f"Failed to fetch history: {str(e)}", severity="error")
+
+ finally:
+ # Re-enable buttons
+ fetch_btn.disabled = False
+ export_btn.disabled = False
+
+ def _populate_table(self) -> None:
+ """Populate the DataTable with execution history data."""
+ table = self.query_one("#history_table", DataTable)
+ table.clear(columns=True)
+
+ if self.execution_data.empty:
+ return
+
+ # Define preferred column order (your specified order)
+ preferred_order = [
+ "policyname",
+ "policyver",
+ "hostname",
+ "username",
+ "publisher",
+ "filename",
+ "pprocess",
+ "gprocess",
+ "sha256",
+ "commandline",
+ "agent_hostname", # Our custom field
+ ]
+
+ # Get available columns in preferred order, then add any remaining columns
+ available_cols = []
+ for col in preferred_order:
+ if col in self.execution_data.columns:
+ available_cols.append(col)
+
+ # Add any remaining columns not in preferred order
+ for col in self.execution_data.columns:
+ if col not in available_cols:
+ available_cols.append(col)
+
+ # Add columns to table
+ for col in available_cols:
+ table.add_column(col, key=col)
+
+ # Add rows
+ for idx, row in self.execution_data.iterrows():
+ row_data = []
+ for col in available_cols:
+ value = row[col]
+ # Convert to string, handle None/NaN
+ if pd.isna(value):
+ row_data.append("")
+ else:
+ row_data.append(str(value))
+ table.add_row(*row_data, key=str(idx))
+
+ logger.info(f"Populated table with {len(self.execution_data)} rows")
+
+ def _export_to_csv(self) -> None:
+ """Export the current execution data to CSV."""
+ if self.execution_data.empty:
+ self.app.notify("No data to export", severity="warning")
+ return
+
+ try:
+ # Create filename with timestamp
+ timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+ filename = f"execution_history_{timestamp}.csv"
+ filepath = os.path.join(self.working_dir, filename)
+
+ # Export to CSV
+ self.execution_data.to_csv(filepath, index=False, encoding="utf-8-sig")
+
+ self.app.notify(
+ f"✅ Exported {len(self.execution_data)} records to: {filepath}",
+ severity="information",
+ timeout=5,
+ )
+ logger.info(f"Exported execution history to: {filepath}")
+
+ except Exception as e:
+ logger.error(f"Failed to export CSV: {e}")
+ self.app.notify(f"Failed to export CSV: {str(e)}", severity="error")
+
+ def action_export_csv(self) -> None:
+ """Keybinding action to export CSV."""
+ self._export_to_csv()
+
+ def action_close_screen(self) -> None:
+ """Close this screen and return to previous."""
+ self.app.pop_screen()
diff --git a/TUI/Widgets/agentmoveoperations.py b/TUI/Widgets/agentmoveoperations.py
index bf20a82..2986423 100644
--- a/TUI/Widgets/agentmoveoperations.py
+++ b/TUI/Widgets/agentmoveoperations.py
@@ -29,6 +29,7 @@ from textual.widget import Widget
from textual.widgets import Button, DataTable, Footer, Header, Static, TextArea
from models.agent import Agent
+from TUI.Screens.executionhistoryscreen import ExecutionHistoryScreen
from TUI.Screens.otpworkflowscreen import OTPWorkflowScreen
from TUI.Screens.policyselectorscreen import PolicySelectorScreen
from TUI.Widgets.OTP_generate import OTPGenerator
@@ -140,6 +141,7 @@ class AgentMoveOperations(Widget):
toggle_enforcement_btn = self.query_one("#toggle_enforcement_btn", Button)
other_policy_btn = self.query_one("#other_policy_btn", Button)
otp_gen_btn = self.query_one("#otp_gen_btn", Button)
+ exec_history_btn = self.query_one("#exec_history_btn", Button)
# If operation in progress, disable all
if self.operation_in_progress:
@@ -148,6 +150,7 @@ class AgentMoveOperations(Widget):
local_approval_btn.disabled = True
toggle_enforcement_btn.disabled = True
other_policy_btn.disabled = True
+ exec_history_btn.disabled = True
else:
# If an operation was selected, disable
if self.selected_operation:
@@ -162,6 +165,9 @@ class AgentMoveOperations(Widget):
other_policy_btn.disabled = (
self.selected_operation == "other_policy"
)
+ exec_history_btn.disabled = (
+ self.selected_operation == "exec_history"
+ )
else:
# Enable all buttons
otp_gen_btn = False
@@ -169,6 +175,7 @@ class AgentMoveOperations(Widget):
local_approval_btn.disabled = False
toggle_enforcement_btn.disabled = False
other_policy_btn.disabled = False
+ exec_history_btn.disabled = False
except NoMatches:
pass
@@ -199,21 +206,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)")
@@ -248,9 +255,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)
@@ -285,37 +292,44 @@ class AgentMoveOperations(Widget):
yield operations_label
# Operation buttons
- export_csv_btn = Button("📄 Export CSV", id="export_csv_btn")
+ export_csv_btn = Button("📄 Export CSV", id="export_csv_btn")
export_csv_btn.styles.width = "100%"
export_csv_btn.styles.margin = (0, 0, 1, 0)
yield export_csv_btn
local_approval_btn = Button(
- "✔️ Local Approval Mode", id="local_approval_btn"
+ "âœ”ï¸ Local Approval Mode", id="local_approval_btn"
)
local_approval_btn.styles.width = "100%"
local_approval_btn.styles.margin = (0, 0, 1, 0)
yield local_approval_btn
- otp_gen_btn = Button("🎫 Generate One Time Passes", id="otp_gen_btn")
+ otp_gen_btn = Button("🎫 Generate One Time Passes", id="otp_gen_btn")
otp_gen_btn.styles.width = "100%"
otp_gen_btn.styles.margin = (0, 0, 1, 0)
yield otp_gen_btn
toggle_enforcement_btn = Button(
- "🔄 Toggle Audit/Enforcement", id="toggle_enforcement_btn"
+ "🔄 Toggle Audit/Enforcement", id="toggle_enforcement_btn"
)
toggle_enforcement_btn.styles.width = "100%"
toggle_enforcement_btn.styles.margin = (0, 0, 1, 0)
yield toggle_enforcement_btn
other_policy_btn = Button(
- "🔀 Move to Other Policy", id="other_policy_btn"
+ "🔀 Move to Other Policy", id="other_policy_btn"
)
other_policy_btn.styles.width = "100%"
other_policy_btn.styles.margin = (0, 0, 1, 0)
yield other_policy_btn
+ exec_history_btn = Button(
+ "📊 View Execution History", id="exec_history_btn"
+ )
+ exec_history_btn.styles.width = "100%"
+ exec_history_btn.styles.margin = (0, 0, 1, 0)
+ yield exec_history_btn
+
# Status label
status_label = Static("", id="status_label")
status_label.styles.margin = (2, 0, 0, 0)
@@ -377,17 +391,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()
@@ -407,6 +421,9 @@ class AgentMoveOperations(Widget):
elif btn_id == "otp_gen_btn":
self._start_OTP_gen_operation()
event.stop()
+ elif btn_id == "exec_history_btn":
+ self._start_execution_history_operation()
+ event.stop()
def _start_local_approval_operation(self) -> None:
"""
@@ -435,7 +452,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
@@ -470,12 +487,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)
@@ -518,9 +535,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
@@ -564,7 +581,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
@@ -600,12 +617,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)
@@ -670,7 +687,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")
@@ -681,6 +698,35 @@ class AgentMoveOperations(Widget):
self.app.push_screen(OTPWorkflowScreen(self.agents))
+ def _start_execution_history_operation(self) -> None:
+ """
+ Launch the execution history viewer for selected agents.
+
+ This operation opens a new screen that allows the user to:
+ 1. Select a date range for execution history
+ 2. Fetch execution logs for all selected agents
+ 3. View the results in a table
+ 4. Export the results to CSV
+
+ The screen is pushed onto the screen stack, allowing the user to return
+ to this screen when done.
+ """
+ status_label = self.query_one("#status_label", Static)
+ status_label.update("Opening execution history viewer...")
+
+ try:
+ # Push the execution history screen
+ self.app.push_screen(ExecutionHistoryScreen(self.agents))
+ logger.info(
+ f"Opened execution history viewer for {len(self.agents)} agents"
+ )
+ except Exception as e:
+ logger.error(f"Failed to open execution history viewer: {e}")
+ status_label.update(f"❌ Error: {str(e)}")
+ self.app.notify(
+ f"Failed to open execution history: {str(e)}", severity="error"
+ )
+
def _execute_move_to_policy(self, target_policy) -> None:
"""
Execute the actual move of agents to the selected policy.
diff --git a/TUI/Widgets/policytreewidget.py b/TUI/Widgets/policytreewidget.py
index d51da78..c9811de 100644
--- a/TUI/Widgets/policytreewidget.py
+++ b/TUI/Widgets/policytreewidget.py
@@ -18,8 +18,9 @@ import logging
from rich.text import Text
from textual.containers import Horizontal, Vertical
+from textual.message import Message
from textual.widget import Widget
-from textual.widgets import Input, OptionList, Static, Switch, Tree
+from textual.widgets import Button, Input, OptionList, Static, Switch, Tree
from textual.widgets.option_list import Option
logger = logging.getLogger(__name__)
@@ -28,6 +29,27 @@ logger = logging.getLogger(__name__)
class PolicyTreeWidget(Widget):
"""Widget for displaying and searching a hierarchical policy tree."""
+ class ViewExecutionHistory(Message):
+ """Message sent when user wants to view execution history for a device."""
+
+ def __init__(self, device):
+ super().__init__()
+ self.device = device
+
+ class GenerateOTP(Message):
+ """Message sent when user wants to generate OTP for a device."""
+
+ def __init__(self, device):
+ super().__init__()
+ self.device = device
+
+ class ToggleEnforcement(Message):
+ """Message sent when user wants to toggle audit/enforcement for a device."""
+
+ def __init__(self, device):
+ super().__init__()
+ self.device = device
+
def __init__(self, policies, devices):
super().__init__()
self.policies = policies
@@ -35,6 +57,7 @@ class PolicyTreeWidget(Widget):
self.last_highlighted_node = None
self.leaf_counts = defaultdict(int)
self.match_type = "Count" # Default to sorting by count
+ self.selected_device = None # Track currently selected device
def compose(self):
# Create the switch and its label
@@ -56,6 +79,18 @@ class PolicyTreeWidget(Widget):
search_box = Input(
placeholder="Search policies or devices...", id="tree_search"
)
+ exec_history_button = Button(
+ "📊 Execution History", id="view_exec_history_button", disabled=True
+ )
+ exec_history_button.styles.margin = (0, 1, 0, 0) # Right margin
+
+ otp_button = Button("🎫 Generate OTP", id="generate_otp_button", disabled=True)
+ otp_button.styles.margin = (0, 1, 0, 0) # Right margin
+
+ toggle_enforcement_button = Button(
+ "🔄 Toggle Enforcement/Audit", id="toggle_enforcement_button", disabled=True
+ )
+ # No right margin on last button
details_pane = Static("", id="details_pane")
# Layout the UI
@@ -73,6 +108,12 @@ class PolicyTreeWidget(Widget):
# Add the search box and details pane
yield label
yield search_box
+ # Action buttons in a horizontal row
+ with Horizontal() as button_row:
+ button_row.styles.height = "auto"
+ yield exec_history_button
+ yield otp_button
+ yield toggle_enforcement_button
yield details_pane
def on_mount(self) -> None:
@@ -89,6 +130,32 @@ class PolicyTreeWidget(Widget):
# Expand the root node
policy_tree.root.expand()
+ def refresh_data(self, policies, devices):
+ """Refresh the widget with new data and rebuild the tree."""
+ self.policies = policies
+ self.devices = devices
+ self.selected_device = None
+
+ # Disable all buttons since selection is lost
+ try:
+ self.query_one("#view_exec_history_button", Button).disabled = True
+ self.query_one("#generate_otp_button", Button).disabled = True
+ self.query_one("#toggle_enforcement_button", Button).disabled = True
+ except:
+ pass
+
+ # Rebuild tree with new data
+ self._precompute_leaf_counts()
+ total_leaves = sum(
+ self.leaf_counts.get(policy.groupid, 0)
+ for policy in self.policies
+ if policy.parent == "global-policy-settings"
+ )
+ policy_tree = self.query_one("#policy_tree", Tree)
+ policy_tree.root.set_label(f"Agents in Policies: ({total_leaves})")
+ self._build_tree()
+ policy_tree.root.expand()
+
def _precompute_leaf_counts(self):
"""Precompute leaf counts for each policy group."""
device_counts = defaultdict(int)
@@ -184,6 +251,9 @@ class PolicyTreeWidget(Widget):
node = message.node
data = node.data
details_pane = self.query_one("#details_pane", Static)
+ exec_history_button = self.query_one("#view_exec_history_button", Button)
+ otp_button = self.query_one("#generate_otp_button", Button)
+ toggle_enforcement_button = self.query_one("#toggle_enforcement_button", Button)
if self.last_highlighted_node is not None:
original_label = str(self.last_highlighted_node.label).strip()
@@ -198,6 +268,20 @@ class PolicyTreeWidget(Widget):
node.set_label(highlighted_label)
self.last_highlighted_node = node
+ # Check if selected node is a device (has Agent data)
+ from models.agent import Agent
+
+ if data and isinstance(data, Agent):
+ self.selected_device = data
+ exec_history_button.disabled = False
+ otp_button.disabled = False
+ toggle_enforcement_button.disabled = False
+ else:
+ self.selected_device = None
+ exec_history_button.disabled = True
+ otp_button.disabled = True
+ toggle_enforcement_button.disabled = True
+
if data:
details = "\n".join(
f"{key}: {value}" for key, value in data.__dict__.items()
@@ -287,3 +371,18 @@ class PolicyTreeWidget(Widget):
option_list.remove()
except:
pass
+
+ def on_button_pressed(self, event: Button.Pressed) -> None:
+ """Handle button presses."""
+ if event.button.id == "view_exec_history_button":
+ if self.selected_device:
+ self.post_message(self.ViewExecutionHistory(self.selected_device))
+ event.stop()
+ elif event.button.id == "generate_otp_button":
+ if self.selected_device:
+ self.post_message(self.GenerateOTP(self.selected_device))
+ event.stop()
+ elif event.button.id == "toggle_enforcement_button":
+ if self.selected_device:
+ self.post_message(self.ToggleEnforcement(self.selected_device))
+ event.stop()
diff --git a/TUI/Widgets/serverlogwidget.py b/TUI/Widgets/serverlogwidget.py
new file mode 100644
index 0000000..79ef132
--- /dev/null
+++ b/TUI/Widgets/serverlogwidget.py
@@ -0,0 +1,159 @@
+# 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
+
+from bson import ObjectId
+from textual.app import ComposeResult
+from textual.containers import Container, Vertical
+from textual.widgets import Button, DataTable, Static
+
+from services.API import AirlockAPIWrapper
+
+logger = logging.getLogger(__name__)
+
+
+def skipback(days):
+ """
+ Generate a MongoDB ObjectId for a given number of days ago from today.
+ """
+ adjusted_days = days
+ date_days_ago = datetime.datetime.now(datetime.UTC) - datetime.timedelta(
+ days=adjusted_days
+ )
+ timestamp = int(date_days_ago.timestamp())
+ hex_timestamp = format(timestamp, "08x")
+ objectid_hex = hex_timestamp + "0000000000000000"
+ return ObjectId(objectid_hex)
+
+
+class ServerLogWidget(Vertical):
+ """Widget for displaying server activity logs in a DataTable."""
+
+ DEFAULT_CSS = """
+ ServerLogWidget {
+ width: 100%;
+ height: 100%;
+ }
+
+ ServerLogWidget #status_bar {
+ width: 100%;
+ height: auto;
+ background: $surface;
+ padding: 1;
+ margin-bottom: 1;
+ }
+
+ ServerLogWidget DataTable {
+ height: 1fr;
+ border: solid $primary;
+ }
+
+ ServerLogWidget #button_container {
+ width: 100%;
+ height: auto;
+ layout: horizontal;
+ padding: 1;
+ }
+
+ ServerLogWidget Button {
+ margin-right: 1;
+ }
+ """
+
+ def __init__(self, api: AirlockAPIWrapper):
+ super().__init__()
+ self.api = api
+
+ def compose(self) -> ComposeResult:
+ yield Static("Loading server logs (last 72 hours)...", id="status_bar")
+ yield DataTable(id="server_log_table")
+ with Container(id="button_container"):
+ yield Button("Refresh", id="refresh_button", variant="primary")
+
+ def on_mount(self) -> None:
+ """Initialize the DataTable and load server logs."""
+ self.load_logs()
+
+ def load_logs(self) -> None:
+ """Load server logs from the API and populate the DataTable."""
+ table = self.query_one("#server_log_table", DataTable)
+ status = self.query_one("#status_bar", Static)
+
+ try:
+ status.update("⏳ Loading server logs (last 72 hours)...")
+
+ # Create a fake checkpoint for 3 days ago (72 hours)
+ checkpoint = str(skipback(3))
+
+ # Get server logs from API
+ logs = self.api.server_logs(checkpoint=checkpoint)
+
+ if not logs:
+ status.update("ℹï¸ No server logs found in the last 72 hours.")
+ table.clear(columns=True)
+ return
+
+ # Clear existing data
+ table.clear(columns=True)
+
+ # Add columns based on the first log entry
+ if logs:
+ first_log = logs[0]
+ columns = [col for col in first_log.keys() if col != "checkpoint"]
+
+ for col in columns:
+ table.add_column(col, key=col)
+
+ # Add rows in reverse order so newest entries are at the top
+ for log_entry in reversed(logs):
+ row_data = []
+ for col in columns:
+ value = log_entry.get(col, "")
+ # Format datetime column to be more readable
+ if col == "datetime" and value:
+ try:
+ # Parse ISO format and convert to readable format
+ dt = datetime.datetime.fromisoformat(
+ str(value).replace("Z", "+00:00")
+ )
+ value = dt.strftime("%Y-%m-%d %H:%M:%S")
+ except Exception:
+ # If parsing fails, just use the original value
+ pass
+ row_data.append(str(value))
+ table.add_row(*row_data)
+
+ status.update(
+ f"✅ Loaded {len(logs)} log entries from the last 72 hours"
+ )
+ logger.info(f"Loaded {len(logs)} server log entries")
+ else:
+ status.update("ℹï¸ No log entries found.")
+
+ except Exception as exc:
+ error_msg = f"❌ Error loading server logs: {exc}"
+ status.update(error_msg)
+ logger.error(f"Failed to load server logs: {exc}", exc_info=True)
+ table.clear(columns=True)
+
+ def on_button_pressed(self, event: Button.Pressed) -> None:
+ """Handle button presses."""
+ button_id = event.button.id
+
+ if button_id == "refresh_button":
+ self.load_logs()
+ event.stop()
diff --git a/services/API.py b/services/API.py
index 22cfa7d..d9e2d36 100644
--- a/services/API.py
+++ b/services/API.py
@@ -351,6 +351,14 @@ class AirlockAPIWrapper:
result = self._post("/v1/getexechistory", payload)
return result["response"]["exechistory"]
+ def server_logs(self, checkpoint: str | None = None) -> str:
+ """Retrieves Server Activity History Logs."""
+ payload = {}
+ if checkpoint is not None:
+ payload["checkpoint"] = checkpoint
+ result = self._post("/v1/logging/svractivities?checkpoint", payload)
+ return result["response"]["svractivities"]
+
"""
from services.API import AirlockAPIWrapper
diff --git a/utils/setup.py b/utils/setup.py
index f28a9a3..36cf927 100644
--- a/utils/setup.py
+++ b/utils/setup.py
@@ -29,6 +29,49 @@ from utils.configmanager import (
)
+class TextualNotificationHandler(logging.Handler):
+ """
+ Custom logging handler that sends ERROR, WARNING, and CRITICAL logs
+ to Textual toast notifications.
+ """
+
+ def __init__(self, app):
+ super().__init__()
+ self.app = app
+
+ def emit(self, record):
+ try:
+ # Only handle ERROR, WARNING, and CRITICAL
+ if record.levelno >= logging.WARNING:
+ # Format the message
+ msg = self.format(record)
+
+ # Map log levels to Textual severity
+ severity_map = {
+ logging.WARNING: "warning",
+ logging.ERROR: "error",
+ logging.CRITICAL: "error",
+ }
+ severity = severity_map.get(record.levelno, "information")
+
+ # Send to Textual notification
+ # Use call_from_thread if logging from non-main thread
+ try:
+ self.app.notify(msg, severity=severity, timeout=5)
+ except Exception:
+ # If we're not on the main thread, schedule it
+ try:
+ self.app.call_from_thread(
+ self.app.notify, msg, severity=severity, timeout=5
+ )
+ except Exception:
+ # Silently fail to avoid breaking the logging system
+ pass
+ except Exception:
+ # Silently fail to avoid breaking the logging system
+ pass
+
+
def get_base_directory() -> Path:
system = platform.system()
home = Path.home()
@@ -44,38 +87,31 @@ def configure_logging(log_dir: Path, log_level: str = "INFO"):
log_file = log_dir / "Loxide.log"
config = {
- "version": 1, # Required key for dictConfig format version
- "disable_existing_loggers": False, # Keeps existing loggers active
+ "version": 1,
+ "disable_existing_loggers": False,
"formatters": {
"detailed": {
"format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
- # Includes timestamp, logger name, level, and message
- },
- "simple": {
- "format": "%(levelname)s - %(message)s"
- # Minimal format for console output
},
+ "simple": {"format": "%(levelname)s - %(message)s"},
+ "toast": {"format": "%(name)s: %(message)s"}, # Simpler format for toasts
},
"handlers": {
"file": {
"class": "logging.handlers.TimedRotatingFileHandler",
"filename": str(log_file),
- "when": "midnight", # Rotate logs at midnight
- "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 to file
- "formatter": "detailed", # Use detailed format
- },
- "console": {
- "class": "logging.StreamHandler",
- "level": log_level.upper(), # System-configured level for console
- "formatter": "simple", # Use simple format
+ "when": "midnight",
+ "interval": 1,
+ "backupCount": 7,
+ "encoding": "utf-8",
+ "level": "DEBUG",
+ "formatter": "detailed",
},
+ # REMOVED console handler - it interferes with Textual TUI
},
"root": {
- "level": "DEBUG", # Root logger level
- "handlers": ["file", "console"], # Attach both handlers
+ "level": "DEBUG",
+ "handlers": ["file"], # Only use file handler, not console
},
}
@@ -96,6 +132,18 @@ def configure_logging(log_dir: Path, log_level: str = "INFO"):
logging.config.dictConfig(config)
logging.getLogger().debug("✅ Logging configured.")
+ # Return a function to attach the notification handler once the app is created
+ def attach_notification_handler(app):
+ """Attach the Textual notification handler to the root logger."""
+ handler = TextualNotificationHandler(app)
+ handler.setLevel(logging.WARNING) # Only WARNING and above
+ formatter = logging.Formatter("%(name)s: %(message)s")
+ handler.setFormatter(formatter)
+ logging.getLogger().addHandler(handler)
+ logging.getLogger().debug("✅ Textual notification handler attached.")
+
+ return attach_notification_handler
+
def setup():
"""
@@ -105,6 +153,9 @@ def setup():
3. Load user config (mutable, from user_config.json)
4. Configure logging
5. Set up .env with WORKING_DIR only
+
+ Returns:
+ attach_notification_handler: Function to attach notification handler to TUI app
"""
base_dir = get_base_directory()
dirs = {
@@ -122,7 +173,7 @@ def setup():
# Configure logging with system-defined log level
log_level = get_system_value("LOG_LEVEL", str, "INFO")
- configure_logging(dirs["logs"], log_level)
+ attach_handler = configure_logging(dirs["logs"], log_level)
# Load user config (mutable)
load_user_config(dirs["config"])
@@ -145,7 +196,6 @@ def setup():
"Approved": [],
"Needs_Review": ["Review_First", "Review_Second", "HTML"],
"Preflight": ["HTML"],
- "Archived": [],
}
for folder_name, subfolders in folders_structure.items():
@@ -158,3 +208,6 @@ def setup():
logging.debug(f"'{subfolder}' subfolder created at: {subfolder_path}")
logging.info("✅ Setup complete")
+
+ # Return the attach handler function
+ return attach_handler