Skip to content

Commit dd4d4c3

Browse files
authored
Implement two-tiered Ctrl+C handling for graceful session navigation (#491)
Implements a two-tiered Ctrl+C handling mechanism for better UX: - First Ctrl+C at prompt: Shows hint message - Second Ctrl+C (within 2s): Returns to session menu - Ctrl+C during module: Interrupts and returns to prompt - Ctrl+C at session menu: Exits Pacu (unchanged) Fixes #474 Co-authored-by: Son Sulung Suryahatta Asnan <Synchx00@users.noreply.github.com>
1 parent 3f841dd commit dd4d4c3

File tree

1 file changed

+91
-13
lines changed

1 file changed

+91
-13
lines changed

pacu/main.py

Lines changed: 91 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -155,6 +155,9 @@ def display_pacu_help():
155155
console/open_console Generate a URL that will log the current user/role in to
156156
the AWS web console
157157
debug Display the contents of the error log file
158+
Keyboard Shortcuts:
159+
Ctrl+C Stop the currently running module and return to the prompt
160+
Press twice at the prompt to return to session selection
158161
""")
159162

160163

@@ -194,15 +197,46 @@ class Main:
194197
'swap_keys', 'swap_session', 'unset_ua_suffix', 'update_regions', 'use', 'whoami', 'debug', 'clear'
195198
]
196199

200+
# Time window (in seconds) for detecting double Ctrl+C
201+
INTERRUPT_TIMEOUT = 2.0
202+
197203
def __init__(self):
198204
# NOTE: self.database is the sqlalchemy session since 'session' is reserved for PacuSession objects.
199205
self.database: orm.session.Session = None
200206
self.running_module_names: List[str] = []
201207
self.CATEGORIES: set = load_categories()
202208

209+
# Keyboard interrupt handling state (Issue #474)
210+
self._module_running: bool = False
211+
self._interrupt_count: int = 0
212+
self._last_interrupt_time: float = 0.0
213+
203214
# Hack so we can use session names without passing around Main.
204215
lib.get_active_session = self.get_active_session
205216

217+
def _reset_interrupt_state(self) -> None:
218+
"""Reset the interrupt counter and timestamp."""
219+
self._interrupt_count = 0
220+
self._last_interrupt_time = 0.0
221+
222+
def _record_interrupt(self) -> int:
223+
"""
224+
Record an interrupt and return the count within the timeout window.
225+
226+
Returns:
227+
The number of interrupts within the timeout window.
228+
"""
229+
current_time = time.time()
230+
231+
# Check if we're within the timeout window
232+
if current_time - self._last_interrupt_time > self.INTERRUPT_TIMEOUT:
233+
self._interrupt_count = 1
234+
else:
235+
self._interrupt_count += 1
236+
237+
self._last_interrupt_time = current_time
238+
return self._interrupt_count
239+
206240
# Utility methods
207241
def log_error(self, text, exception_info=None, session=None, local_data=None, global_data=None) -> None:
208242
""" Write an error to the file at log_file_path, or a default log file
@@ -1039,6 +1073,9 @@ def exec_module(self, command: List[str]) -> None:
10391073
return
10401074

10411075
self.running_module_names.append(module.module_info['name'])
1076+
self._module_running = True
1077+
self._reset_interrupt_state()
1078+
10421079
try:
10431080
summary_data = module.main(command[2:], self)
10441081
# If the module's return value is None, it exited early.
@@ -1054,11 +1091,13 @@ def exec_module(self, command: List[str]) -> None:
10541091

10551092
self.print('{} completed.\n'.format(module.module_info['name']))
10561093
self.print('MODULE SUMMARY:\n\n{}\n'.format(summary.strip('\n')))
1094+
10571095
except SystemExit as exception_value:
10581096
exception_type, _, tb = sys.exc_info()
10591097

10601098
if 'SIGINT called' in exception_value.args:
1061-
self.print('^C\nExiting the currently running module.')
1099+
# Handle Ctrl+C during module execution (Issue #474)
1100+
self.print('\n[!] Module {} interrupted by user (Ctrl+C). Returning to Pacu prompt...'.format(module.module_info['name']))
10621101
else:
10631102
traceback_text = '\nTraceback (most recent call last):\n{}{}: {}\n\n'.format(
10641103
''.join(traceback.format_tb(tb)), str(exception_type), str(exception_value)
@@ -1072,6 +1111,7 @@ def exec_module(self, command: List[str]) -> None:
10721111
global_data=global_data
10731112
)
10741113
finally:
1114+
self._module_running = False
10751115
self.running_module_names.pop()
10761116
elif module_name in self.COMMANDS:
10771117
print('Error: "{}" is the name of a Pacu command, not a module. Try using it without "run" or "exec" in front.'.format(module_name))
@@ -1739,21 +1779,52 @@ def complete(completer, text, state):
17391779
pass
17401780

17411781
def exit(self) -> None:
1742-
sys.exit('SIGINT called')
1743-
1744-
def idle(self) -> None:
1782+
sys.exit('Pacu exit')
1783+
1784+
def idle(self) -> bool:
1785+
"""
1786+
The main command input loop.
1787+
Returns:
1788+
True if should return to session selection menu
1789+
False if should exit Pacu entirely
1790+
"""
17451791
session = self.get_active_session()
17461792

1747-
if session.key_alias:
1748-
alias = session.key_alias
1749-
else:
1750-
alias = 'No Keys Set'
1793+
while True:
1794+
try:
1795+
if session.key_alias:
1796+
alias = session.key_alias
1797+
else:
1798+
alias = 'No Keys Set'
17511799

1752-
command = input('Pacu ({}:{}) > '.format(session.name, alias))
1800+
command = input('Pacu ({}:{}) > '.format(session.name, alias))
1801+
self.parse_command(command)
1802+
1803+
# Reset interrupt state only AFTER successful command
1804+
self._reset_interrupt_state()
1805+
1806+
except (KeyboardInterrupt, SystemExit) as e:
1807+
# Handle Ctrl+C at the prompt (Issue #474)
1808+
if isinstance(e, SystemExit):
1809+
# 'Pacu exit' = exit command, 'SIGINT called' = Ctrl+C
1810+
if e.args and 'SIGINT called' in str(e.args):
1811+
# This is Ctrl+C, handle as interrupt
1812+
pass
1813+
else:
1814+
# This is exit command or other SystemExit, re-raise
1815+
raise
17531816

1754-
self.parse_command(command)
1817+
interrupt_count = self._record_interrupt()
1818+
if interrupt_count >= 2:
1819+
print('\n[*] Returning to session selection menu...')
1820+
return True
1821+
else:
1822+
print('\n[*] Press Ctrl+C again within 2 seconds to return to session menu.')
1823+
continue
17551824

1756-
self.idle()
1825+
except EOFError:
1826+
print('\n[*] EOF detected. Returning to session selection...')
1827+
return True
17571828

17581829
def run_cli(self, *args) -> None:
17591830
self.database = get_database_connection(settings.DATABASE_CONNECTION_PATH)
@@ -1897,13 +1968,20 @@ def run_gui(self, quiet=False) -> None:
18971968
idle_ready = True
18981969

18991970
self.check_user_agent()
1900-
self.idle()
1971+
1972+
# Session command loop with interrupt handling (Issue #474)
1973+
should_return_to_session_menu = self.idle()
1974+
1975+
if should_return_to_session_menu:
1976+
# User pressed Ctrl+C twice, go back to session selection
1977+
idle_ready = False
1978+
continue
19011979

19021980
except (Exception, SystemExit) as exception_value:
19031981
exception_type, _, tb = sys.exc_info()
19041982

19051983
if exception_type == SystemExit:
1906-
if 'SIGINT called' in exception_value.args:
1984+
if 'SIGINT called' in exception_value.args or 'Pacu exit' in exception_value.args:
19071985
print('\nBye!')
19081986
return
19091987
else:

0 commit comments

Comments
 (0)