487 lines
18 KiB
Python
487 lines
18 KiB
Python
# 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 <https://www.gnu.org/licenses/>
|
|
|
|
import logging
|
|
import re
|
|
from typing import Optional
|
|
|
|
import pandas as pd
|
|
from textual.containers import Horizontal, Vertical
|
|
from textual.message import Message
|
|
from textual.widget import Widget
|
|
from textual.widgets import Button, DataTable, Static, TextArea
|
|
|
|
from models.policy import Policy
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class PolicySelector(Widget):
|
|
"""
|
|
A Textual widget for selecting a target policy for agent operations.
|
|
|
|
This widget displays available policies in a table and allows users to select
|
|
one policy as the destination for bulk agent movements. It automatically excludes:
|
|
- Parent/logical policies (where parent == "global-policy-settings")
|
|
- Specified policy IDs (e.g., the current policy)
|
|
|
|
Features:
|
|
- Wildcard filtering (* and ?)
|
|
- Interactive table for policy browsing
|
|
- Explicit confirm button for selection
|
|
- Use escape key to go back
|
|
|
|
Attributes:
|
|
policies (list[Policy]): List of available Policy objects to display.
|
|
excluded_policy_ids (set[str]): Set of policy IDs to exclude from selection.
|
|
selected_policy (Optional[Policy]): The currently selected policy (if any).
|
|
|
|
Automatically Filtered Out:
|
|
- Policies with parent == "global-policy-settings" (parent policies for organization)
|
|
- Any policies in excluded_policy_ids set
|
|
|
|
Example:
|
|
```python
|
|
policies = [policy1, policy2, policy3]
|
|
widget = PolicySelector(policies, excluded_policy_ids={current_policy.groupid})
|
|
```
|
|
"""
|
|
|
|
class PolicySelected(Message):
|
|
"""
|
|
Message posted when a policy is selected.
|
|
|
|
Attributes:
|
|
policy (Policy): The selected policy object.
|
|
"""
|
|
|
|
def __init__(self, policy: Policy):
|
|
super().__init__()
|
|
self.policy = policy
|
|
|
|
def __init__(self, policies: list):
|
|
"""
|
|
Initialize the PolicySelector widget.
|
|
|
|
Args:
|
|
policies (list): List of Policy objects or DataFrame rows to display.
|
|
Can be a list of Policy objects or a pandas DataFrame of policy data.
|
|
"""
|
|
super().__init__()
|
|
self.policies = policies
|
|
self.selected_policy: Optional[Policy] = None
|
|
self._filtered_policies = []
|
|
self._displayed_policies = [] # Track what's currently shown in the table
|
|
self._filter_text = ""
|
|
|
|
def compose(self):
|
|
"""
|
|
Build the UI layout for the PolicySelector widget.
|
|
|
|
The layout includes:
|
|
- Title indicating policy selection
|
|
- Search/filter text area with wildcard support
|
|
- Filter help text showing wildcard options
|
|
- Apply Filter button
|
|
- Clear Filter button
|
|
- Confirm Selection button
|
|
- Policy table displaying available policies
|
|
- Use escape key to go back
|
|
"""
|
|
title_text = Static(
|
|
"Select Target Policy",
|
|
id="policy_selector_title",
|
|
)
|
|
title_text.styles.margin = (0, 0, 1, 0)
|
|
yield title_text
|
|
|
|
with Horizontal() as main_layout:
|
|
main_layout.styles.height = "auto"
|
|
|
|
# Left side - Filter and controls
|
|
with Vertical() as left_side:
|
|
left_side.styles.width = "1fr"
|
|
left_side.styles.height = "auto"
|
|
left_side.styles.margin = (0, 1, 0, 1)
|
|
|
|
filter_label = Static("Filter Policies:")
|
|
filter_label.styles.margin = (0, 0, 0, 0)
|
|
yield filter_label
|
|
|
|
filter_input = TextArea(
|
|
id="policy_filter",
|
|
text="",
|
|
)
|
|
filter_input.styles.height = 3
|
|
filter_input.styles.margin = (0, 0, 1, 0)
|
|
yield filter_input
|
|
|
|
filter_help = Static("(Use * and ? for wildcards)", id="filter_help")
|
|
filter_help.styles.margin = (0, 0, 1, 0)
|
|
yield filter_help
|
|
|
|
apply_button = Button("🔍 Apply Filter", id="filter_button")
|
|
apply_button.styles.width = "100%"
|
|
apply_button.styles.margin = (0, 0, 1, 0)
|
|
yield apply_button
|
|
|
|
clear_button = Button("🧹 Clear Filter", id="clear_filter_button")
|
|
clear_button.styles.width = "100%"
|
|
clear_button.styles.margin = (0, 0, 1, 0)
|
|
yield clear_button
|
|
|
|
confirm_button = Button("✅ Confirm Selection", id="confirm_button")
|
|
confirm_button.styles.width = "100%"
|
|
confirm_button.styles.margin = (1, 0, 1, 0)
|
|
yield confirm_button
|
|
|
|
selected_label = Static("", id="selected_policy_label")
|
|
selected_label.styles.margin = (2, 0, 1, 0)
|
|
yield selected_label
|
|
|
|
# Right side - Policy table
|
|
with Vertical() as right_side:
|
|
right_side.styles.width = "2fr"
|
|
right_side.styles.height = "auto"
|
|
|
|
table_label = Static("Available Policies:")
|
|
table_label.styles.margin = (0, 0, 0, 0)
|
|
yield table_label
|
|
|
|
policy_table = DataTable(id="policy_table", cursor_type="row")
|
|
policy_table.styles.height = "1fr"
|
|
policy_table.styles.margin = (1, 0, 1, 0)
|
|
yield policy_table
|
|
|
|
def on_mount(self) -> None:
|
|
"""
|
|
Initialize the policy table when the widget is mounted.
|
|
|
|
Populates the table with column (Policy Name) and rows for each
|
|
available policy (excluding those in excluded_policy_ids and parent policies).
|
|
Sets up event handlers for table row selection.
|
|
|
|
Filters out:
|
|
- Parent policies (where parent == "global-policy-settings")
|
|
"""
|
|
table = self.query_one("#policy_table", DataTable)
|
|
|
|
# Configure table for row selection
|
|
table.cursor_type = "row"
|
|
table.zebra_stripes = True
|
|
|
|
# Only add Policy Name column
|
|
table.add_columns("Policy Name")
|
|
|
|
# Filter out excluded policies and convert to list if DataFrame
|
|
if isinstance(self.policies, pd.DataFrame):
|
|
policies_list = self.policies.to_dict("records")
|
|
else:
|
|
policies_list = self.policies
|
|
policies_list = sorted(policies_list)
|
|
|
|
self._filtered_policies = []
|
|
self._displayed_policies = [] # Initialize displayed list
|
|
|
|
for policy_data in policies_list:
|
|
# Handle both Policy objects and dict/DataFrame rows
|
|
if isinstance(policy_data, Policy):
|
|
policy_id = policy_data.groupid
|
|
policy_name = policy_data.name
|
|
parent = policy_data.parent
|
|
else:
|
|
policy_id = policy_data.get("groupid", "Unknown")
|
|
policy_name = policy_data.get("name", "Unknown")
|
|
parent = policy_data.get("parent", None)
|
|
|
|
# Skip parent policies (logical policies that shouldn't have devices)
|
|
if parent == "global-policy-settings":
|
|
logger.debug(f"Skipping parent policy: {policy_name}")
|
|
continue
|
|
|
|
self._filtered_policies.append(policy_data)
|
|
self._displayed_policies.append(policy_data) # Add to displayed list
|
|
|
|
table.add_row(
|
|
policy_name,
|
|
key=policy_id,
|
|
)
|
|
|
|
def on_button_pressed(self, event: Button.Pressed):
|
|
"""
|
|
Handle button press events from the widget.
|
|
|
|
Routes to:
|
|
- filter_button (Apply Filter): Filter policies with wildcard support
|
|
- clear_filter_button: Clear filter and show all policies
|
|
- confirm_button: Confirm selection and post message
|
|
|
|
Args:
|
|
event (Button.Pressed): The button press event.
|
|
"""
|
|
btn_id = event.button.id
|
|
|
|
if btn_id == "filter_button":
|
|
self._apply_filter()
|
|
event.stop()
|
|
|
|
elif btn_id == "clear_filter_button":
|
|
self._clear_filter()
|
|
event.stop()
|
|
|
|
elif btn_id == "confirm_button":
|
|
self._confirm_selection()
|
|
event.stop()
|
|
|
|
def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None:
|
|
"""
|
|
Handle row selection in the policy table.
|
|
|
|
Updates the selected_policy and displays the selection in the UI.
|
|
|
|
Args:
|
|
event: DataTable.RowSelected event containing the selected row data.
|
|
"""
|
|
try:
|
|
# Get the row key from the event
|
|
row_key = event.row_key
|
|
if row_key is None:
|
|
return
|
|
|
|
# Find the policy with matching groupid
|
|
for policy_data in self._displayed_policies:
|
|
if isinstance(policy_data, Policy):
|
|
if policy_data.groupid == row_key.value:
|
|
self.selected_policy = policy_data
|
|
break
|
|
else:
|
|
if policy_data.get("groupid") == row_key.value:
|
|
self.selected_policy = Policy(
|
|
groupid=policy_data.get("groupid"),
|
|
hidden=policy_data.get("hidden", False),
|
|
name=policy_data.get("name"),
|
|
parent=policy_data.get("parent"),
|
|
)
|
|
break
|
|
|
|
if self.selected_policy:
|
|
# Update selection display
|
|
label = self.query_one("#selected_policy_label", Static)
|
|
label.update(f"Selected: {self.selected_policy.name}")
|
|
|
|
# Log for debugging
|
|
logger.debug(
|
|
f"Selected policy: {self.selected_policy.name} (ID: {self.selected_policy.groupid})"
|
|
)
|
|
self.app.notify(
|
|
f"Selected: {self.selected_policy.name}",
|
|
severity="information",
|
|
timeout=1,
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error handling row selection: {e}")
|
|
self.app.notify(f"Selection error: {str(e)}", severity="error")
|
|
|
|
def on_data_table_row_highlighted(self, event: DataTable.RowHighlighted) -> None:
|
|
"""
|
|
Handle row highlighting (cursor movement) in the table.
|
|
|
|
This provides immediate visual feedback when navigating rows.
|
|
"""
|
|
try:
|
|
# Get the row key from the event
|
|
row_key = event.row_key
|
|
if row_key is None:
|
|
return
|
|
|
|
# Find the highlighted policy
|
|
highlighted_name = None
|
|
for policy_data in self._displayed_policies:
|
|
if isinstance(policy_data, Policy):
|
|
if policy_data.groupid == row_key.value:
|
|
highlighted_name = policy_data.name
|
|
break
|
|
else:
|
|
if policy_data.get("groupid") == row_key.value:
|
|
highlighted_name = policy_data.get("name")
|
|
break
|
|
|
|
if highlighted_name:
|
|
label = self.query_one("#selected_policy_label", Static)
|
|
label.update(f"Highlighting: {highlighted_name}")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error handling row highlight: {e}")
|
|
|
|
def _apply_filter(self) -> None:
|
|
"""
|
|
Apply filter text to policy list with wildcard support.
|
|
|
|
Supports wildcards:
|
|
- * matches any sequence of characters
|
|
- ? matches a single character
|
|
|
|
Examples:
|
|
- "policy*" matches "policy_prod", "policy_dev", etc.
|
|
- "policy?" matches "policy1", "policy2", etc.
|
|
- "*audit*" matches anything containing "audit"
|
|
- "*test*" matches "AT Testing", "test_policy", etc.
|
|
|
|
Filters policies by name or ID (case-insensitive) and refreshes the table display
|
|
with only matching policies. Only filters from already-filtered list
|
|
(which excludes parent policies and excluded IDs).
|
|
"""
|
|
try:
|
|
filter_input = self.query_one("#policy_filter", TextArea)
|
|
filter_text = filter_input.text.strip()
|
|
|
|
table = self.query_one("#policy_table", DataTable)
|
|
table.clear()
|
|
|
|
# Clear the displayed policies list
|
|
self._displayed_policies = []
|
|
|
|
# Compile wildcard pattern if filter text is provided
|
|
pattern = None
|
|
if filter_text:
|
|
# Escape special regex chars but preserve wildcards
|
|
pattern_text = re.escape(filter_text.lower())
|
|
pattern_text = pattern_text.replace(r"\*", ".*").replace(r"\?", ".")
|
|
# Use search() for partial matching
|
|
pattern = re.compile(pattern_text, re.IGNORECASE)
|
|
|
|
# Filter policies based on search text
|
|
for policy_data in self._filtered_policies:
|
|
# Handle both Policy objects and dict/DataFrame rows
|
|
if isinstance(policy_data, Policy):
|
|
policy_name = policy_data.name.lower()
|
|
policy_id = policy_data.groupid.lower()
|
|
display_name = policy_data.name
|
|
key_id = policy_data.groupid
|
|
else:
|
|
policy_name = str(policy_data.get("name", "")).lower()
|
|
policy_id = str(policy_data.get("groupid", "Unknown")).lower()
|
|
display_name = policy_data.get("name")
|
|
key_id = policy_data.get("groupid")
|
|
|
|
# Match against filter text with wildcard support
|
|
if pattern:
|
|
# Use search() for partial matching
|
|
matches = pattern.search(policy_name) or pattern.search(policy_id)
|
|
else:
|
|
matches = True
|
|
|
|
if matches:
|
|
# Add to displayed policies list
|
|
self._displayed_policies.append(policy_data)
|
|
|
|
# Add row to table
|
|
table.add_row(
|
|
display_name,
|
|
key=key_id,
|
|
)
|
|
|
|
displayed_count = len(self._displayed_policies)
|
|
status_text = (
|
|
f"Showing {displayed_count} of {len(self._filtered_policies)} policies"
|
|
)
|
|
self.app.notify(status_text, severity="information", timeout=2)
|
|
|
|
# Clear selection when filter is applied
|
|
self.selected_policy = None
|
|
label = self.query_one("#selected_policy_label", Static)
|
|
label.update("")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error applying filter: {e}")
|
|
self.app.notify(f"❌ Filter error: {str(e)}", severity="error")
|
|
|
|
def _clear_filter(self) -> None:
|
|
"""
|
|
Clear the filter and display all available policies.
|
|
|
|
Resets the filter text and refreshes the table to show all policies
|
|
(already excluding parent policies and excluded IDs).
|
|
"""
|
|
try:
|
|
filter_input = self.query_one("#policy_filter", TextArea)
|
|
filter_input.text = ""
|
|
|
|
table = self.query_one("#policy_table", DataTable)
|
|
table.clear()
|
|
|
|
# Reset displayed policies to all filtered policies
|
|
self._displayed_policies = list(self._filtered_policies)
|
|
|
|
# Reload all policies
|
|
for policy_data in self._filtered_policies:
|
|
if isinstance(policy_data, Policy):
|
|
policy_id = policy_data.groupid
|
|
policy_name = policy_data.name
|
|
else:
|
|
policy_id = policy_data.get("groupid", "Unknown")
|
|
policy_name = policy_data.get("name", "Unknown")
|
|
|
|
# Add row with only policy name
|
|
table.add_row(
|
|
policy_name,
|
|
key=policy_id,
|
|
)
|
|
|
|
self.selected_policy = None
|
|
label = self.query_one("#selected_policy_label", Static)
|
|
label.update("")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error clearing filter: {e}")
|
|
|
|
def on_text_area_changed(self, event) -> None:
|
|
"""
|
|
Handle TextArea change events - specifically for Enter key in filter.
|
|
|
|
When the user types in the filter TextArea and the text ends with a newline,
|
|
treat it as pressing Enter and apply the filter.
|
|
"""
|
|
if event.text_area.id == "policy_filter":
|
|
# Check if the text ends with a newline (Enter was pressed)
|
|
if event.text_area.text.endswith("\n"):
|
|
# Remove the newline that was added
|
|
event.text_area.text = event.text_area.text.rstrip("\n")
|
|
# Apply the filter
|
|
self._apply_filter()
|
|
|
|
def _confirm_selection(self) -> None:
|
|
"""
|
|
Confirm the selected policy and post selection message.
|
|
|
|
Posts a PolicySelected message to the parent widget/screen with the
|
|
selected policy. If no policy is selected, displays an error notification.
|
|
"""
|
|
if self.selected_policy is None:
|
|
self.app.notify(
|
|
"Please select a policy first by clicking on a row in the table",
|
|
severity="warning",
|
|
timeout=3,
|
|
)
|
|
return
|
|
|
|
# Log confirmation for debugging
|
|
logger.info(f"Confirming selection of policy: {self.selected_policy.name}")
|
|
self.app.notify(
|
|
f"Confirmed: {self.selected_policy.name}", severity="success", timeout=2
|
|
)
|
|
self.post_message(self.PolicySelected(self.selected_policy))
|