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

feat: Multiple UI improvements and new server log functionality

Browse Source 
- Add Server Log tab with DataTable display of server activity logs

- Fix keyboard navigation bug in agents tab

- Add execution history viewer for selected agents

- Improve policy tree widget functionality by adding single device operations

- Integrate logging notifications into TUI
  - Add TextualNotificationHandler to setup.py
  - Display ERROR/WARNING/CRITICAL logs as toast notifications
  - Remove terminal output to prevent interference with TUI
  - Logs still written to Loxide.log file

closes #45
This commit is contained in:
Zarithas
2025-12-16 16:16:49 -05:00
parent 57d0f12000
commit fc17c869fc
7 changed files with 1183 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Loxide.py
+133 -19
View File
@@ -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__":