User Management CLI Enhancements

This addresses the requirement of user confirmation from cli
for all the high-risk activities that may affect service/management.

Change-Id: I25a544e4df8fd1ac217c023a87bc4f6977cdaea6
Signed-off-by: Sabyasachi Nayak <sabyasachi.nayak@windriver.com>
This commit is contained in:
Sabyasachi Nayak 2024-10-21 03:35:59 -04:00
parent 448bb06c3b
commit ac8ddeb1b6

@ -32,6 +32,8 @@ import argparse
import dateutil
import prettytable
import textwrap
import signal
import sys
from datetime import datetime
from dateutil import parser
@ -228,7 +230,9 @@ def define_command(subparsers, command, callback, cmd_mapper):
formatter_class=HelpFormatter)
subparser.add_argument('-h', '--help', action='help',
help=argparse.SUPPRESS)
if _is_service_impacting_command(command):
subparser.add_argument('--yes', action='store_true', help=f"Automatically confirm the action: {command}")
callback = require_confirmation(callback, command.split("-")[0])
# Are we a list command?
if _does_command_need_no_wrap(callback):
# then decorate it with wrapping data formatter functionality
@ -578,3 +582,55 @@ def print_list(objs, fields, field_labels, formatters={}, sortby=0,
return print_long_list(objs, fields, field_labels, formatters=formatters, sortby=sortby,
reversesort=reversesort, no_wrap_fields=no_wrap_fields,
no_paging=True, printer=printer)
def input_with_timeout(prompt, timeout):
def timeout_handler(signum, frame):
raise TimeoutError
# Set the timeout handler
signal.signal(signal.SIGALRM, timeout_handler)
signal.alarm(timeout) # Set the alarm for the timeout
try:
# Try to get input from the user
result = input(prompt)
signal.alarm(0) # Cancel the alarm if input is received in time
return result
except TimeoutError:
print("\nError: No response received within the time limit.")
sys.exit(1)
def require_confirmation(func, target_object, timeout=10):
"""Decorator that asks for user confirmation before running the function."""
def wrapper(*args, **kwargs):
YELLOW = '\033[93m'
RESET = '\033[0m'
BOLD = '\033[1m'
if hasattr(args[1], 'yes') and args[1].yes:
# Skip confirmation if --yes was passed
return func(*args, **kwargs)
confirmation = input_with_timeout(
f"{BOLD}{YELLOW}WARNING: This is a high-risk operation that may cause a service interruption{RESET}\n"
f"{BOLD}{YELLOW}Do you want to continue? (yes/No): {RESET}", timeout
)
if confirmation is None:
print("\nError: No response received within the time limit.")
return
elif confirmation.lower() != 'yes':
print("Operation cancelled by the user")
sys.exit(1)
return func(*args, **kwargs)
return wrapper
def _is_service_impacting_command(command):
service_impacting_fm_commands = [
"event-suppress",
"event-unsuppress",
"event-unsuppress-all"
]
return command in service_impacting_fm__commands or 'delete' in command or 'remove' in command