Updated OTP Generate with new workflow
This commit is contained in:
@@ -0,0 +1,346 @@
|
||||
import logging
|
||||
from typing import List
|
||||
|
||||
from textual.containers import Horizontal, Vertical
|
||||
from textual.css.query import NoMatches
|
||||
from textual.message import Message
|
||||
from textual.reactive import reactive
|
||||
from textual.widget import Widget
|
||||
from textual.widgets import Button, Input, RadioButton, RadioSet, Static, TextArea
|
||||
|
||||
from models.agent import Agent
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class OTPGenerator(Widget):
|
||||
# Reactive properties to track form completion
|
||||
requestor_filled = reactive(False)
|
||||
reasoning_filled = reactive(False)
|
||||
duration_selected = reactive(True) # Default is selected
|
||||
otp_generated = reactive(False)
|
||||
|
||||
class OTPInfo(Message):
|
||||
def __init__(
|
||||
self, devices: List[Agent], requestor: str, reasoning: str, duration: int
|
||||
):
|
||||
super().__init__()
|
||||
self.devices = devices
|
||||
self.requestor = requestor
|
||||
self.reasoning = reasoning
|
||||
self.duration = duration
|
||||
|
||||
# Duration options in minutes
|
||||
DURATION_OPTIONS = [
|
||||
(15, "15 minutes"),
|
||||
(60, "1 hour"),
|
||||
(360, "6 hours"),
|
||||
(1440, "1 day"),
|
||||
(10080, "7 days"),
|
||||
]
|
||||
|
||||
def __init__(self, devices: List[Agent]):
|
||||
"""Initialize with a list of Agent objects."""
|
||||
super().__init__()
|
||||
self.devices = devices
|
||||
|
||||
def watch_requestor_filled(self, old_value: bool, new_value: bool) -> None:
|
||||
"""Update button state when requestor changes."""
|
||||
self._update_button_state()
|
||||
|
||||
def watch_reasoning_filled(self, old_value: bool, new_value: bool) -> None:
|
||||
"""Update button state when reasoning changes."""
|
||||
self._update_button_state()
|
||||
|
||||
def watch_otp_generated(self, old_value: bool, new_value: bool) -> None:
|
||||
"""Update button state when OTP is generated."""
|
||||
self._update_button_state()
|
||||
|
||||
def _update_button_state(self) -> None:
|
||||
"""Enable/disable the generate button based on form state."""
|
||||
try:
|
||||
button = self.query_one("#generate_button", Button)
|
||||
# Enable only if all fields filled and OTP not yet generated
|
||||
button.disabled = not (
|
||||
self.requestor_filled
|
||||
and self.reasoning_filled
|
||||
and not self.otp_generated
|
||||
)
|
||||
except NoMatches:
|
||||
pass
|
||||
|
||||
def compose(self):
|
||||
title_text = Static(
|
||||
f"🎫 Generate One Time Passes for {len(self.devices)} device(s)",
|
||||
id="otpgen_title",
|
||||
)
|
||||
title_text.styles.margin = (0, 0, 1, 0)
|
||||
yield title_text
|
||||
|
||||
with Horizontal() as main_layout:
|
||||
main_layout.styles.height = "auto"
|
||||
|
||||
# Left side - Inputs and controls
|
||||
with Vertical() as left_side:
|
||||
left_side.styles.width = "1fr"
|
||||
left_side.styles.height = "auto"
|
||||
|
||||
# Requestor input
|
||||
requestor_label = Static("Who is requesting OTP?")
|
||||
requestor_label.styles.margin = (0, 0, 0, 0)
|
||||
yield requestor_label
|
||||
|
||||
requestor_box = Input(
|
||||
placeholder="Enter requestor name", id="requestor_input"
|
||||
)
|
||||
requestor_box.styles.margin = (0, 0, 1, 0)
|
||||
yield requestor_box
|
||||
|
||||
# Reasoning input
|
||||
reasoning_label = Static("What work are they doing?")
|
||||
reasoning_label.styles.margin = (0, 0, 0, 0)
|
||||
yield reasoning_label
|
||||
|
||||
reasoning_box = Input(
|
||||
placeholder="Enter reason for OTP", id="reasoning_input"
|
||||
)
|
||||
reasoning_box.styles.margin = (0, 0, 1, 0)
|
||||
yield reasoning_box
|
||||
|
||||
# Duration selection
|
||||
duration_label = Static("Duration:")
|
||||
duration_label.styles.margin = (0, 0, 0, 0)
|
||||
yield duration_label
|
||||
|
||||
with RadioSet(id="duration_radio") as radio_set:
|
||||
radio_set.styles.margin = (0, 0, 1, 0)
|
||||
for minutes, label in self.DURATION_OPTIONS:
|
||||
radio = RadioButton(label, id=f"duration_{minutes}")
|
||||
if minutes == 360: # Default to 6 hours
|
||||
radio.value = True
|
||||
yield radio
|
||||
|
||||
# Buttons in a horizontal layout
|
||||
with Horizontal() as button_row:
|
||||
button_row.styles.height = "auto"
|
||||
button_row.styles.margin = (1, 0, 0, 0)
|
||||
|
||||
back_button = Button("← Back", id="back_button")
|
||||
back_button.styles.width = "1fr"
|
||||
yield back_button
|
||||
|
||||
generate_button = Button(
|
||||
"Generate OTP", id="generate_button", variant="primary"
|
||||
)
|
||||
generate_button.styles.width = "2fr"
|
||||
yield generate_button
|
||||
|
||||
# Right side - Show device list initially, then output after generation
|
||||
with Vertical() as right_side:
|
||||
right_side.styles.width = "2fr"
|
||||
right_side.styles.height = "100%"
|
||||
|
||||
output_label = Static(
|
||||
f"Selected Devices ({len(self.devices)}):", id="output_label"
|
||||
)
|
||||
output_label.styles.margin = (0, 0, 0, 0)
|
||||
yield output_label
|
||||
|
||||
# Container for either device list or output
|
||||
with Vertical(id="output_container") as output_container:
|
||||
output_container.styles.height = "1fr"
|
||||
output_container.styles.margin = (1, 0, 0, 0)
|
||||
output_container.styles.overflow_y = "auto"
|
||||
output_container.styles.border = ("round", "green")
|
||||
|
||||
# Show device list initially
|
||||
device_list_text = "\n".join(
|
||||
f"• {device.hostname}" for device in self.devices
|
||||
)
|
||||
device_display = Static(device_list_text, id="device_display")
|
||||
yield device_display
|
||||
|
||||
# Copy to clipboard button (hidden initially)
|
||||
copy_button = Button("📋 Copy to Clipboard", id="copy_clipboard_button")
|
||||
copy_button.styles.margin = (1, 0, 0, 0)
|
||||
copy_button.styles.display = "none"
|
||||
yield copy_button
|
||||
|
||||
def on_mount(self) -> None:
|
||||
"""Set initial button state."""
|
||||
self._update_button_state()
|
||||
|
||||
def on_input_changed(self, event: Input.Changed) -> None:
|
||||
"""Handle input field changes."""
|
||||
input_id = event.input.id
|
||||
|
||||
if input_id == "requestor_input":
|
||||
self.requestor_filled = bool(event.value.strip())
|
||||
elif input_id == "reasoning_input":
|
||||
self.reasoning_filled = bool(event.value.strip())
|
||||
|
||||
def on_button_pressed(self, event: Button.Pressed):
|
||||
btn_id = event.button.id
|
||||
|
||||
if btn_id == "back_button":
|
||||
self.app.pop_screen()
|
||||
event.stop()
|
||||
|
||||
elif btn_id == "copy_clipboard_button":
|
||||
try:
|
||||
output_area = self.query_one("#otp_output", TextArea)
|
||||
text_to_copy = output_area.text
|
||||
|
||||
import pyperclip
|
||||
|
||||
pyperclip.copy(text_to_copy)
|
||||
self.app.notify(
|
||||
"✅ Copied to clipboard!", severity="information", timeout=2
|
||||
)
|
||||
except ImportError:
|
||||
self.app.notify(
|
||||
"⚠️ pyperclip not installed. Run: pip install pyperclip",
|
||||
severity="warning",
|
||||
)
|
||||
except Exception as e:
|
||||
self.app.notify(f"❌ Failed to copy: {str(e)}", severity="error")
|
||||
event.stop()
|
||||
|
||||
elif btn_id == "generate_button":
|
||||
try:
|
||||
requestor = self.query_one("#requestor_input", Input).value.strip()
|
||||
reasoning = self.query_one("#reasoning_input", Input).value.strip()
|
||||
|
||||
radio_set = self.query_one("#duration_radio", RadioSet)
|
||||
selected_button_id = (
|
||||
radio_set.pressed_button.id if radio_set.pressed_button else None
|
||||
)
|
||||
|
||||
if not selected_button_id:
|
||||
self._show_error("Please select a duration")
|
||||
return
|
||||
|
||||
duration = int(selected_button_id.replace("duration_", ""))
|
||||
|
||||
if not requestor or not reasoning:
|
||||
self._show_error("Please fill in all fields")
|
||||
return
|
||||
|
||||
self.otp_generated = True
|
||||
|
||||
# Access API from the app - this is the key change!
|
||||
api = self.app.api
|
||||
|
||||
output_lines = [
|
||||
"=" * 60,
|
||||
"OTP GENERATION RESULTS",
|
||||
"=" * 60,
|
||||
]
|
||||
|
||||
otp_dict = {}
|
||||
for device in self.devices:
|
||||
try:
|
||||
otp_code = api.otp_generate(device.agentid, duration, reasoning)
|
||||
otp_dict[device.hostname] = otp_code
|
||||
logger.debug(f"Generated OTP for {device.hostname}: {otp_code}")
|
||||
except Exception as e:
|
||||
otp_dict[device.hostname] = f"ERROR: {str(e)}"
|
||||
logger.error(
|
||||
f"Failed to generate OTP for {device.hostname}: {e}"
|
||||
)
|
||||
|
||||
for hostname, otp_code in otp_dict.items():
|
||||
output_lines.append(f"{hostname:30} | {otp_code}")
|
||||
|
||||
output_lines.append("=" * 60)
|
||||
result_text = "\n".join(output_lines)
|
||||
self._show_result(result_text)
|
||||
|
||||
# Post message with the OTP info
|
||||
self.post_message(
|
||||
self.OTPInfo(self.devices, requestor, reasoning, duration)
|
||||
)
|
||||
event.stop()
|
||||
|
||||
except NoMatches:
|
||||
self._show_error("UI elements not found")
|
||||
except Exception as e:
|
||||
self._show_error(f"Error: {str(e)}")
|
||||
logger.exception("Error generating OTP")
|
||||
|
||||
def _show_error(self, message: str):
|
||||
"""Display error message in output area."""
|
||||
try:
|
||||
container = self.query_one("#output_container", Vertical)
|
||||
try:
|
||||
device_display = self.query_one("#device_display", Static)
|
||||
device_display.remove()
|
||||
except NoMatches:
|
||||
pass
|
||||
|
||||
try:
|
||||
output_area = self.query_one("#otp_output", TextArea)
|
||||
except NoMatches:
|
||||
output_area = TextArea(id="otp_output", read_only=True)
|
||||
container.mount(output_area)
|
||||
|
||||
output_area.text = f"❌ ERROR: {message}"
|
||||
except Exception as e:
|
||||
logger.debug(f"Error showing error message: {e}")
|
||||
|
||||
def _show_result(self, message: str):
|
||||
"""Display result message in output area."""
|
||||
try:
|
||||
container = self.query_one("#output_container", Vertical)
|
||||
try:
|
||||
device_display = self.query_one("#device_display", Static)
|
||||
device_display.remove()
|
||||
except NoMatches:
|
||||
pass
|
||||
|
||||
try:
|
||||
output_area = self.query_one("#otp_output", TextArea)
|
||||
except NoMatches:
|
||||
output_area = TextArea(id="otp_output", read_only=True)
|
||||
container.mount(output_area)
|
||||
|
||||
output_area.text = message
|
||||
output_label = self.query_one("#output_label", Static)
|
||||
output_label.update("Generated OTP Details:")
|
||||
copy_button = self.query_one("#copy_clipboard_button", Button)
|
||||
copy_button.styles.display = "block"
|
||||
except Exception as e:
|
||||
logger.debug(f"Error showing result: {e}")
|
||||
|
||||
def display_otp_result(self, result_text: str):
|
||||
"""Display OTP generation result in the output area."""
|
||||
try:
|
||||
container = self.query_one("#output_container", Vertical)
|
||||
try:
|
||||
device_display = self.query_one("#device_display", Static)
|
||||
device_display.remove()
|
||||
except NoMatches:
|
||||
pass
|
||||
|
||||
try:
|
||||
output_area = self.query_one("#otp_output", TextArea)
|
||||
except NoMatches:
|
||||
output_area = TextArea(id="otp_output", read_only=True)
|
||||
container.mount(output_area)
|
||||
|
||||
output_area.text = result_text
|
||||
except Exception as e:
|
||||
logger.debug(f"Error displaying OTP result: {e}")
|
||||
|
||||
def clear_form(self):
|
||||
"""Clear all input fields and reset state."""
|
||||
try:
|
||||
self.query_one("#requestor_input", Input).value = ""
|
||||
self.query_one("#reasoning_input", Input).value = ""
|
||||
self.query_one("#otp_output", TextArea).text = ""
|
||||
self.otp_generated = False
|
||||
self.requestor_filled = False
|
||||
self.reasoning_filled = False
|
||||
self._update_button_state()
|
||||
except NoMatches:
|
||||
pass
|
||||
Reference in New Issue
Block a user