Skip to content

audit-alert view command enhancement #1538

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
craiglurey merged 2 commits into from
Aug 1, 2025
Merged

audit-alert view command enhancement #1538

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
4465959
Improve 'audit-alert view' command
sdubey-ks Aug 1, 2025
8824e15
Merge pull request #1536 from Keeper-Security/audit-alert_view-int-re…
sdubey-ks Aug 1, 2025
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
331 changes: 261 additions & 70 deletions keepercommander/commands/audit_alerts.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,8 +31,13 @@

alert_target_parser = argparse.ArgumentParser(add_help=False)
alert_target_parser.add_argument('target', metavar='ALERT', help='Alert ID or Name.')

alert_view_parser = argparse.ArgumentParser(prog='audit-alert view', parents=[alert_target_parser])
alert_view_parser = argparse.ArgumentParser(prog='audit-alert view')
alert_view_parser.add_argument('target', metavar='ALERT', nargs='?', help='Alert ID or Name')
alert_view_parser.add_argument('--all', dest='all', action='store_true', help='View all alerts')
alert_view_parser.add_argument('--format', dest='format', action='store', choices=['table', 'csv', 'json', 'pdf'],
default='table', help='format of output')
alert_view_parser.add_argument('--output', dest='output', action='store',
help='path to resulting output file (ignored for "table" format)')
alert_history_parser = argparse.ArgumentParser(
prog='audit-alert history', parents=[report_output_parser, alert_target_parser])
alert_reset_counts_parser = argparse.ArgumentParser(prog='audit-alert reset-counts', parents=[alert_target_parser])
Expand Down Expand Up @@ -414,84 +419,270 @@ def get_parser(self):
return alert_view_parser

def execute(self, params, **kwargs):
show_recipient = True
show_filter = True
show_stat = True
if kwargs.get('recipient_only'):
show_recipient = True
show_filter = False
show_stat = False
fmt = kwargs.get('format', '')
view_all = kwargs.get('all') or not kwargs.get('target')

if view_all:
self._execute_view_all(params, kwargs, fmt)
else:
self._execute_view_single(params, kwargs, fmt)

alert = AuditSettingMixin.get_alert_configuration(params, kwargs.get('target'))
def _get_display_options(self, kwargs):
"""Determine what sections to show based on kwargs."""
if kwargs.get('recipient_only'):
return {'recipient': True, 'filter': False, 'stat': False}
return {'recipient': True, 'filter': True, 'stat': True}

def _get_headers(self, show_options, fmt):
"""Get appropriate headers based on display options."""
if show_options['recipient'] and not show_options['filter'] and not show_options['stat']:
headers = ['alert_id', 'alert_name', 'status', 'recipients']
elif show_options['stat']:
headers = ['alert_id', 'alert_name', 'status', 'frequency', 'occurrences',
'sent_counter', 'last_sent', 'event_types', 'users', 'shared_folders',
'records', 'recipients']
else:
headers = ['alert_id', 'alert_name', 'status', 'event_types', 'users',
'shared_folders', 'records', 'recipients']

return headers if fmt == 'json' else [field_to_title(x) for x in headers]

def _format_datetime(self, date_str):
"""Format datetime string to local time."""
if not date_str:
return ''
try:
dt = datetime.datetime.strptime(date_str, '%Y-%m-%dT%H:%M:%S.%fZ')
dt = dt.replace(microsecond=0, tzinfo=datetime.timezone.utc).astimezone()
return dt.isoformat()
except ValueError:
return ''

def _get_event_names(self, event_ids):
"""Convert event IDs to names."""
if not event_ids:
return []
event_lookup = {i: n for i, n in self.EVENT_TYPES or []}
return [event_lookup[x] for x in event_ids if x in event_lookup]

def _get_usernames(self, user_ids, params):
"""Convert user IDs to usernames."""
if not user_ids:
return []
user_lookup = {x['user_id']: x['username'] for x in params.enterprise.get('users', [])}
return [user_lookup[x] for x in user_ids if x in user_lookup]

def _get_selected_ids(self, items):
"""Get IDs of selected items."""
if not items:
return []
return [x['id'] for x in items if x.get('selected')]

def _build_recipient_data(self, recipient):
"""Build recipient data dictionary."""
data = {
"Recipient ID": recipient.get('id', ''),
"Name": recipient.get('name', ''),
"Status": 'Disabled' if recipient.get('disabled') else 'Enabled'
}

if 'webhook' in recipient:
wh = recipient['webhook']
data["Webhook URL"] = wh.get('url', '')
if wh.get('template'):
data["HTTP Body"] = wh.get('template', '')
if wh.get('token'):
data["Webhook Token"] = wh.get('token', '')
data["Certificate Errors"] = 'Ignore' if wh.get('allowUnverifiedCertificate') else 'Enforce'

if recipient.get('email'):
data["Email To"] = recipient.get('email')

if recipient.get('phone'):
phone_country = recipient.get('phoneCountry')
phone_display = f"(+{phone_country}) {recipient.get('phone')}" if phone_country else recipient.get('phone')
data["Text To"] = phone_display

return data

def _format_list_for_cell(self, items, fmt, max_lines=5):
"""Format a list of items for display in a table cell, with line limits for PDF."""
if not items:
return ''

if fmt == 'pdf' and len(items) > max_lines:
displayed_items = items[:max_lines]
additional_count = len(items) - max_lines
return '\n'.join(displayed_items) + f'\n+ {additional_count} more'

return '\n'.join(items)

def _format_recipients_for_cell(self, recipients_complete, fmt):
"""Format recipients data for display in a table cell, with limits for PDF."""
recipients_json = json.dumps(recipients_complete, indent=2)

if fmt == 'pdf':
lines = recipients_json.split('\n')
if len(lines) > 5:
truncated_lines = lines[:5]
additional_count = len(lines) - 5
return '\n'.join(truncated_lines) + f'\n+ {additional_count} more lines'

return recipients_json

def _build_alert_row(self, alert, show_options, params, fmt=''):
"""Build a single alert row for the table."""
alert_id = alert.get('id', '')
ctx = AuditSettingMixin.get_alert_context(alert_id) or alert
alert_filter_data = alert.get('filter', {})

row = [
alert_id,
alert.get('name', ''),
'Disabled' if ctx.get('disabled') else 'Enabled'
]

if show_options['stat'] and not show_options.get('recipient_only'):
row.extend([
self.frequency_to_text(alert.get('frequency')) or '',
ctx.get('counter', ''),
ctx.get('sentCounter', ''),
self._format_datetime(ctx.get('lastSent'))
])

if not show_options.get('recipient_only'):
events = self._get_event_names(alert_filter_data.get('events'))
row.append(self._format_list_for_cell(events, fmt))

users = self._get_usernames(alert_filter_data.get('userIds'), params)
row.append(self._format_list_for_cell(users, fmt))

folders = self._get_selected_ids(alert_filter_data.get('sharedFolderUids'))
row.append(self._format_list_for_cell(folders, fmt))

records = self._get_selected_ids(alert_filter_data.get('recordUids'))
row.append(self._format_list_for_cell(records, fmt))

recipients_data = [self._build_recipient_data(r) for r in alert.get('recipients', [])]
recipients_complete = {
"Send To Originator": alert.get('sendToOriginator', False),
"Recipients": recipients_data
}
row.append(self._format_recipients_for_cell(recipients_complete, fmt))

return row

def _execute_view_all(self, params, kwargs, fmt):
"""Execute view all alerts functionality."""
show_options = self._get_display_options(kwargs)

alerts = self.load_settings(params, kwargs.get('reload', False))
if not isinstance(alerts, dict):
logging.info('No alerts found')
return

alert_filter = alerts.get('AuditAlertFilter')
if not isinstance(alert_filter, list):
logging.info('No alerts found')
return

headers = self._get_headers(show_options, fmt)
table = [self._build_alert_row(alert, show_options, params, fmt) for alert in alert_filter]

dump_report_data(table, headers, fmt=fmt, filename=kwargs.get('output'), sort_by=None)

def _add_filter_data_to_table(self, table, alert_filter, params):
"""Add filter data to table for single alert view."""
table.extend([['', ''], ['Alert Filter:', '']])

if 'events' in alert_filter:
events = self._get_event_names(alert_filter.get('events'))
table.append(['Event Types', events])

if 'userIds' in alert_filter:
users = self._get_usernames(alert_filter.get('userIds'), params)
table.append(['User', users])

if 'sharedFolderUids' in alert_filter:
folders = self._get_selected_ids(alert_filter['sharedFolderUids'])
table.append(['Shared Folder', folders])

if 'recordUids' in alert_filter:
records = self._get_selected_ids(alert_filter['recordUids'])
table.append(['Record', records])

def _add_recipient_data_to_table(self, table, alert):
"""Add recipient data to table for single alert view."""
table.extend([['', ''], ['Recipients:', ''],
['Send To Originator (*)', alert.get('sendToOriginator', False)]])

for recipient in alert.get('recipients', []):
table.extend([
['', ''],
['Recipient ID', recipient.get('id')],
['Name', recipient.get('name')],
['Status', 'Disabled' if recipient.get('disabled') else 'Enabled']
])

if 'webhook' in recipient:
wh = recipient['webhook']
table.append(['Webhook URL', wh.get('url')])
if wh.get('template'):
table.append(['HTTP Body', wh.get('template')])
table.append(['Webhook Token', wh.get('token')])
table.append(['Certificate Errors', 'Ignore' if wh.get('allowUnverifiedCertificate') else 'Enforce'])

if recipient.get('email'):
table.append(['Email To', recipient.get('email')])

if recipient.get('phone'):
phone_country = recipient.get('phoneCountry')
phone_display = (f'(+{phone_country}) {recipient.get("phone")}'
if phone_country else recipient.get('phone'))
table.append(['Text To', phone_display])

def _execute_view_single(self, params, kwargs, fmt):
"""Execute view single alert functionality."""
show_options = self._get_display_options(kwargs)
target = kwargs.get('target')

alert = AuditSettingMixin.get_alert_configuration(params, target)
alert_id = alert.get('id')
ctx = AuditSettingMixin.get_alert_context(alert_id) or alert
table = []
header = ['name', 'value']
table.append(['Alert ID', alert_id])
table.append(['Alert name', alert.get('name')])
table.append(['Status', 'Disabled' if ctx.get('disabled') is True else 'Enabled'])
if show_stat:
table.append(['Frequency', self.frequency_to_text(alert.get('frequency'))])
table.append(['Occurrences', ctx.get('counter')])
table.append(['Sent Counter', ctx.get('sentCounter')])

header = ['name', 'value'] if fmt == 'json' else [field_to_title(x) for x in ['name', 'value']]

table = [
['Alert ID', alert_id],
['Alert name', alert.get('name')],
['Status', 'Disabled' if ctx.get('disabled') else 'Enabled']
]

if show_options['stat']:
last_sent = ctx.get('lastSent')
if last_sent:
try:
last_sent = datetime.datetime.strptime(last_sent, '%Y-%m-%dT%H:%M:%S.%fZ')
last_sent = last_sent.replace(microsecond=0, tzinfo=datetime.timezone.utc).astimezone()
except ValueError:
pass
table.append(['Last Sent', last_sent.isoformat() if last_sent else ''])

if show_filter:
alert_filter = alert.get('filter') or {}
table.append(['', ''])
table.append(['Alert Filter:', ''])
if 'events' in alert_filter:
event_lookup = {i: n for i, n in self.EVENT_TYPES or []}
events = [event_lookup[x] for x in alert_filter.get('events') or [] if x in event_lookup]
table.append(['Event Types', events])
if 'userIds' in alert_filter:
user_lookup = {x['user_id']: x['username'] for x in params.enterprise.get('users') or []}
users = [user_lookup[x] for x in alert_filter.get('userIds') or [] if x in user_lookup]
table.append(['User', users])
if 'sharedFolderUids' in alert_filter:
table.append(['Shared Folder', [x['id'] for x in alert_filter['sharedFolderUids'] if x['selected']]])
if 'recordUids' in alert_filter:
table.append(['Record', [x['id'] for x in alert_filter['recordUids'] if x['selected']]])

if show_recipient:
recipients = []
for r in alert.get('recipients') or []:
recipients.append(['', ''])
recipients.append(['Recipient ID', r.get('id')])
recipients.append(['Name', r.get('name')])
recipients.append(['Status', 'Disabled' if r.get('disabled') is True else 'Enabled'])
if 'webhook' in r:
wh = r['webhook']
recipients.append(['Webhook URL', wh.get('url')])
http_body = wh.get('template')
if http_body:
recipients.append(['HTTP Body', http_body])
recipients.append(['Webhook Token', wh.get('token')])
recipients.append(['Certificate Errors', 'Ignore' if wh.get('allowUnverifiedCertificate') else 'Enforce'])
email = r.get('email')
if email:
recipients.append(['Email To', email])
phone = r.get('phone')
if phone:
phone_country = r.get('phoneCountry')
if phone_country:
recipients.append(['Text To', f'(+{r.get("phoneCountry")}) {phone}'])
else:
recipients.append(['Text To', phone])
table.append(['', ''])
table.append(['Recipients:', ''])
table.append(['Send To Originator (*)', alert.get('sendToOriginator') or False])
table.extend(recipients)

dump_report_data(table, header, no_header=True, right_align=(0,))


table.extend([
['Frequency', self.frequency_to_text(alert.get('frequency'))],
['Occurrences', ctx.get('counter')],
['Sent Counter', ctx.get('sentCounter')],
['Last Sent', last_sent.isoformat() if last_sent else '']
])

if show_options['filter']:
alert_filter = alert.get('filter', {})
self._add_filter_data_to_table(table, alert_filter, params)

if show_options['recipient']:
self._add_recipient_data_to_table(table, alert)

dump_report_data(table, header, no_header=True, right_align=(0,),
fmt=fmt, filename=kwargs.get('output'), sort_by=None)

class AuditAlertHistory(EnterpriseCommand, AuditSettingMixin):
def get_parser(self):
Expand Down