Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
88 changes: 88 additions & 0 deletions src/sot/_app.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
ProcessesWidget,
SotWidget,
)
from .widgets.confirmation_modal import ConfirmationModal


# Main SOT Application
Expand Down Expand Up @@ -51,6 +52,8 @@ def __init__(self, net_interface=None, log_file=None):
super().__init__()
self.net_interface = net_interface
self.log_file = log_file
self.pending_kill = None
self._waiting_for_kill_confirmation = False

# Set up logging if specified
if log_file:
Expand Down Expand Up @@ -115,6 +118,20 @@ def on_mount(self) -> None:
async def on_load(self, _):
self.bind("q", "quit")

def on_key(self, event) -> None:
"""Handle key events for kill confirmation."""
from textual import events

if self._waiting_for_kill_confirmation:
if event.key == "y":
self._waiting_for_kill_confirmation = False
self._kill_process(self.pending_kill["process_info"])
else:
# Any other key cancels
self._waiting_for_kill_confirmation = False
self.notify("❌ Kill cancelled", timeout=2)
event.prevent_default()

def on_processes_widget_process_selected(
self, message: ProcessesWidget.ProcessSelected
) -> None:
Expand Down Expand Up @@ -169,6 +186,77 @@ def on_processes_widget_process_selected(
timeout=5,
)

def on_processes_widget_kill_request(
self, message: ProcessesWidget.KillRequest
) -> None:
"""Handle kill request by showing a pending confirmation."""
process_info = message.process_info
process_name = process_info.get("name", "Unknown")
process_id = process_info.get("pid", "N/A")

# Store pending kill for confirmation
self.pending_kill = {
"process_info": process_info,
"process_name": process_name,
"process_id": process_id,
"timestamp": 0,
}

# Show confirmation prompt with danger emoji and error severity (red)
self.notify(
f"⚠️ KILL {process_name}? Press 'y' to confirm, any key to cancel",
severity="error",
timeout=10,
)

# Listen for confirmation key in the next on_key event
self._waiting_for_kill_confirmation = True

# Auto-cancel after 10 seconds (matching notification timeout)
def reset_confirmation():
if self._waiting_for_kill_confirmation:
self._waiting_for_kill_confirmation = False
self.notify("❌ Kill action expired", severity="error", timeout=2)

self.set_timer(reset_confirmation, 10.0)

def _kill_process(self, process_info: dict) -> None:
"""Execute the kill process action."""
import psutil

process_id = process_info.get("pid")
process_name = process_info.get("name", "Unknown")

if not process_id:
self.notify("❌ Invalid process ID", severity="error", timeout=3)
return

# Log the action attempt if logging is enabled
if self.log_file:
self.log(f"Attempting to kill process: {process_name} (PID: {process_id})")

try:
target_process = psutil.Process(process_id)
target_process.kill()
if self.log_file:
self.log(
f"Successfully killed process: {process_name} (PID: {process_id})"
)
self.notify(
f"💥 Killed {process_name} (PID: {process_id})",
severity="warning",
timeout=4,
)

except psutil.ZombieProcess:
self._handle_zombie_process_error(process_name, process_id)
except psutil.NoSuchProcess:
self._handle_no_such_process_error(process_id)
except psutil.AccessDenied:
self._handle_access_denied_error(process_name, process_id)
except Exception as error:
self._handle_general_process_error("kill", process_name, error)

def on_processes_widget_process_action(
self, message: ProcessesWidget.ProcessAction
) -> None:
Expand Down
82 changes: 82 additions & 0 deletions src/sot/widgets/confirmation_modal.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,82 @@
"""
Confirmation Modal Widget

Simple confirmation dialog using Textual best practices.
"""

from textual.app import ComposeResult
from textual.containers import Container, Horizontal
from textual.message import Message
from textual.screen import ModalScreen
from textual.widgets import Button, Label


class ConfirmationModal(ModalScreen[bool]):
"""A modal for confirming actions. Returns True if confirmed, False if cancelled."""

class Confirmed(Message):
"""Posted when user confirms."""

def __init__(self, action_data: dict) -> None:
self.action_data = action_data
super().__init__()

BINDINGS = [("escape", "cancel")]

CSS = """
ConfirmationModal {
align: center middle;
}

ConfirmationModal > Container {
width: 60;
height: 9;
border: solid $accent;
background: $panel;
}

ConfirmationModal #title {
margin: 1 2;
width: 1fr;
}

ConfirmationModal #message {
margin: 0 2;
width: 1fr;
}

ConfirmationModal #buttons {
margin: 1 2 0 2;
width: 1fr;
}

ConfirmationModal Button {
margin-right: 1;
}
"""

def __init__(self, title: str, message: str, action_data: dict = None) -> None:
super().__init__()
self.title_text = title
self.message_text = message
self.action_data = action_data or {}

def compose(self) -> ComposeResult:
with Container():
yield Label(self.title_text, id="title")
yield Label(self.message_text, id="message")
with Horizontal(id="buttons"):
yield Button("Kill", variant="error", id="confirm")
yield Button("Cancel", id="cancel")

def on_button_pressed(self, event: Button.Pressed) -> None:
"""Handle button presses."""
if event.button.id == "confirm":
self.app.post_message(self.Confirmed(self.action_data))
self.dismiss(True)
else:
self.dismiss(False)

def action_cancel(self) -> None:
"""Handle escape key."""
self.dismiss(False)
14 changes: 13 additions & 1 deletion src/sot/widgets/processes.py
Original file line number Diff line number Diff line change
Expand Up @@ -94,6 +94,13 @@ def __init__(self, action: str, process_info: dict) -> None:
self.process_info = process_info
super().__init__()

class KillRequest(Message):
"""Message sent when kill is requested on a process (requires confirmation)."""

def __init__(self, process_info: dict) -> None:
self.process_info = process_info
super().__init__()

def __init__(self, **kwargs):
super().__init__(title="Processes", **kwargs)
self.max_num_procs = 1000
Expand Down Expand Up @@ -242,7 +249,7 @@ def handle_action_keys(self, key_pressed: str) -> bool:
elif key_pressed == "k":
if 0 <= self.selected_process_index < len(self.process_list_data):
selected_process = self.process_list_data[self.selected_process_index]
self.post_message(self.ProcessAction("kill", selected_process))
self.post_message(self.KillRequest(selected_process))
return True
elif key_pressed == "t":
if 0 <= self.selected_process_index < len(self.process_list_data):
Expand All @@ -265,6 +272,11 @@ def handle_action_keys(self, key_pressed: str) -> bool:

def on_key(self, event: events.Key) -> None:
"""Handle keyboard navigation and actions with scrolling support."""
# Check if the app is waiting for kill confirmation
# If so, let it bubble up to the app's on_key handler
if hasattr(self.app, '_waiting_for_kill_confirmation') and self.app._waiting_for_kill_confirmation:
return

if not self.is_interactive_mode or not self.process_list_data:
return

Expand Down