diff --git a/keepercommander/commands/ksm.py b/keepercommander/commands/ksm.py index e6215b531..45d0aa61c 100644 --- a/keepercommander/commands/ksm.py +++ b/keepercommander/commands/ksm.py @@ -11,11 +11,13 @@ import argparse import base64 +import calendar import datetime import hmac import json import logging import os +import shlex import time import urllib.parse from itertools import product, groupby @@ -24,7 +26,9 @@ from typing import Sequence, List, Optional -from .base import Command, dump_report_data, user_choice, as_boolean +from .base import ( + Command, dump_report_data, user_choice, as_boolean, report_output_parser, + suppress_exit, raise_parse_exception, expand_cmd_args, normalize_output_param, ParseError) from . import record from ..nested_share_folder.common import get_folder_key, get_record_key from .nested_share_folder.helpers import ( @@ -119,6 +123,31 @@ Adds one or more one-time access tokens to an existing KSM application. Equivalent to: secrets-manager client add --app [APP NAME OR UID] + {bcolors.BOLD}Usage Report:{bcolors.ENDC} + {bcolors.OKGREEN}secrets-manager usage{bcolors.ENDC} + Modes: + (default) : Top Application Usage - owner (email), count + --detail : Full User Usage - owner (name+email), device, API usage + --by-device : Top Usage by Device - device, count (+ Exist with --exists) + --detail --by-device : Full Device Usage - device, app UID, owner, API usage + --summary : Total API usage, applications, devices, avg calls/user + --timeline : Event timeline over all KSM event types + --range [24h|7d|30d] : Timeline window (default 30d) + --all : Export All rows (Date / Event / Number of Events) + Options: + --created [RANGE] : Override usage/summary date window (default: KSM billing cycle) + --limit [N] : Max rows to fetch (paginated; -1 for all) + --sort [count|name] : Sort by usage count (default) or name/title (groups related rows) + --exists : Add an Exist column to --by-device (Y=live, T=in Trash, N=purged, + ?=unknown) resolved from enterprise compliance data. + {bcolors.WARNING}*** VERY SLOW: runs a FULL enterprise compliance sync (can take + minutes on large tenants). ***{bcolors.ENDC} Only applies to --by-device; it is the + only way to detect deleted KSM apps, since the local Trash API cannot + list v5 records. + --format [table|csv|json] --output [FILE] : Output format / file + Note: usage covers only applications with recorded activity and may include ones since deleted. + To list all live applications you have access to, run: {bcolors.OKGREEN}secrets-manager app list{bcolors.ENDC} + ----- Note: If the UID you are using contains a dash (-) in the beginning, the value should be wrapped in quotes and prepended with an equal sign. For example: @@ -172,6 +201,69 @@ help='Output format (table, json)') +# KSM audit event types, mirrors Admin Console `ksmEventTypes` (const.ts). Only `app_client_access` +# drives the usage-metrics report; the full list is used for the event timeline / Export All. +KSM_EVENT_TYPES = [ + 'app_record_shared', 'app_folder_shared', 'app_record_removed', 'app_folder_removed', + 'app_record_share_changed', 'app_folder_share_changed', 'app_client_added', 'app_client_removed', + 'app_client_connected', 'app_client_access', 'app_client_record_update', 'app_client_access_denied', + 'record_rotation_on_demand_fail', 'record_rotation_on_demand_ok', + 'record_rotation_scheduled_fail', 'record_rotation_scheduled_ok', +] +KSM_USAGE_EVENT_TYPE = 'app_client_access' +# Shown after every usage table. +USAGE_APPS_TAIL = ('Usage covers only applications with recorded activity and may include ones since ' + 'deleted. Run "secrets-manager app list" to see all live applications you can access.') +# Legend for the Exist column, shown only with --exists (on --by-device). Existence is resolved from +# enterprise compliance data - deleted KSM apps (v5) cannot be listed via the local Trash API (the +# backend excludes version >= 4), so this tenant-wide check is the only way to detect them. N (purged) +# and T (in trash) are best-effort: only as complete/current as the compliance dataset. +USAGE_EXIST_LEGEND = ( + 'Exist: Y=live, T=in Trash (restorable), N=purged, ?=unknown (compliance data unavailable). ' + 'Resolved from enterprise compliance data (best-effort).') +# Composite-key separator, mirrors Admin Console (`getSecretsManagerMetrics`). +_USAGE_KEY_SEP = '~|~' +# Timeline presets -> (audit report_type, console preset). Mirrors reportsDatePresets + auditTimeline.js. +KSM_TIMELINE_RANGES = { + '24h': ('hour', 'last_24_hours', 1), + '7d': ('day', 'last_7_days', 7), + '30d': ('day', 'last_30_days', 30), +} + +ksm_usage_parser = argparse.ArgumentParser( + prog='secrets-manager usage', parents=[report_output_parser], add_help=False, + description='Keeper Secrets Manager usage report') +ksm_usage_parser.add_argument('--detail', dest='detail', action='store_true', + help='Full detail table (per owner+device, or per device with --by-device)') +ksm_usage_parser.add_argument('--by-device', dest='by_device', action='store_true', + help='Aggregate by device instead of by application owner') +ksm_usage_parser.add_argument('--summary', dest='summary', action='store_true', + help='Usage metrics summary (total API usage, apps, devices, avg calls/user)') +ksm_usage_parser.add_argument('--timeline', dest='timeline', action='store_true', + help='Secrets Manager event timeline (all KSM event types)') +ksm_usage_parser.add_argument('--range', dest='range', action='store', choices=list(KSM_TIMELINE_RANGES.keys()), + help='Timeline window: 24h, 7d or 30d (default 30d). Timeline only') +ksm_usage_parser.add_argument('--all', dest='export_all', action='store_true', + help='Timeline only: Export All rows (Date / Event / Number of Events)') +ksm_usage_parser.add_argument('--created', dest='created', action='store', + help='Override usage/summary date window (same syntax as audit-report --created). ' + 'Default: current KSM billing cycle') +ksm_usage_parser.add_argument('--limit', dest='limit', type=int, action='store', + help='Max rows to fetch (paginated). Default: all. Use -1 for all') +ksm_usage_parser.add_argument('--exists', dest='exists', action='store_true', + help='Add an Exist column to --by-device showing whether each app is ' + 'live / in Trash / purged, resolved from enterprise compliance data. ' + '*** VERY SLOW *** - runs a FULL enterprise compliance sync (can take ' + 'minutes on large tenants). Only applies to --by-device; ignored on ' + 'other views (deleted KSM apps cannot be detected any other way)') +ksm_usage_parser.add_argument('--sort', dest='sort', action='store', choices=['count', 'name'], + default='count', + help='Sort rows by usage count (default, descending) or by name/title ' + '(ascending) - use "name" to group related rows, e.g. all "Playground" devices') +ksm_usage_parser.add_argument('--help', '-h', dest='helpflag', action='store_true', help='Display help') +ksm_usage_parser.error = raise_parse_exception +ksm_usage_parser.exit = suppress_exit + def ms_to_str(ms, frmt='%Y-%m-%d %H:%M:%S'): dt = datetime.datetime.fromtimestamp(ms // 1000) @@ -187,6 +279,25 @@ class KSMCommand(Command): def get_parser(self): return ksm_parser + def execute_args(self, params, args, **kwargs): + # Route the `usage` verb to its own parser so its many flags stay out of the shared ksm_parser. + try: + tokens = shlex.split(args) if args else [] + except ValueError: + tokens = [] + if tokens and tokens[0].lower() == 'usage': + raw = normalize_output_param(expand_cmd_args(args, params.environment_variables)) + try: + opts = ksm_usage_parser.parse_args(shlex.split(raw)[1:]) + except ParseError as e: + logging.error(e) + return + d = {} + d.update(kwargs) + d.update(opts.__dict__) + return KSMCommand.execute_usage(params, **d) + return super(KSMCommand, self).execute_args(params, args, **kwargs) + def execute(self, params, **kwargs): ksm_command = kwargs.get('command') # type: List[str] @@ -475,6 +586,378 @@ def execute(self, params, **kwargs): print(f"{bcolors.WARNING}Unknown combination of KSM commands. " + f"Type 'secrets-manager' for more details'{bcolors.ENDC}") + # ------------------------------------------------------------------ # + # secrets-manager usage # + # ------------------------------------------------------------------ # + + @staticmethod + def _ts_to_ms(value): + """Normalize an epoch timestamp to milliseconds (Console stores add-on/billing dates in ms, + but Commander protos may deliver seconds). Values below 1e11 are treated as seconds.""" + try: + value = int(value) + except (TypeError, ValueError): + return 0 + if value <= 0: + return 0 + return value if value >= 100_000_000_000 else value * 1000 + + @staticmethod + def _ksm_pick_license(params): + licenses = params.enterprise.get('licenses', []) if params.enterprise else [] + if not licenses: + return None + # Prefer a license carrying the secrets_manager add-on, else the first (Console uses licenses[0]). + for lic in licenses: + if any(a.get('name') == 'secrets_manager' for a in lic.get('add_ons', [])): + return lic + return licenses[0] + + @staticmethod + def _set_month(dt, month0, year): + """Mirror Console checkDateChangeRisk: set month (0-based) + year, clamping day to month length.""" + month = month0 + 1 + last_day = calendar.monthrange(year, month)[1] + return dt.replace(year=year, month=month, day=min(dt.day, last_day)) + + @staticmethod + def _ksm_billing_cycle(params, run_on_ms=None): + """Port of Admin Console getKSMBillingCycle (yearly/enterprise path). + Returns (min_epoch_sec, max_epoch_sec) for the current billing cycle, or None if unavailable.""" + lic = KSMCommand._ksm_pick_license(params) + if not lic: + return None + nb_ms = KSMCommand._ts_to_ms(lic.get('next_billing_date')) + if not nb_ms: + return None + + nb = datetime.datetime.fromtimestamp(nb_ms / 1000) + nb_day = nb.day + run = datetime.datetime.fromtimestamp(run_on_ms / 1000) if run_on_ms else datetime.datetime.now() + run_month0 = run.month - 1 + run_year = run.year + + def case_current_to_next(): + start = KSMCommand._set_month(nb, run_month0, run_year) + end_month0 = 0 if run_month0 + 1 > 11 else run_month0 + 1 + end_year = run_year + 1 if run_month0 + 1 > 11 else run_year + end = KSMCommand._set_month(nb, end_month0, end_year) + return start, end + + def case_prev_to_current(): + start_month0 = run_month0 - 1 if run_month0 else 11 + start_year = run_year if run_month0 else run_year - 1 + start = KSMCommand._set_month(nb, start_month0, start_year) + end = KSMCommand._set_month(nb, run_month0, run_year) + return start, end + + if run.day > nb_day: + start, end = case_current_to_next() + elif run.day < nb_day: + start, end = case_prev_to_current() + else: + run_tod = (run.hour * 3600 + run.minute * 60 + run.second) * 1000 + run.microsecond // 1000 + nb_tod = (nb.hour * 3600 + nb.minute * 60 + nb.second) * 1000 + nb.microsecond // 1000 + start, end = case_prev_to_current() if run_tod < nb_tod else case_current_to_next() + + # Never report before KSM was actually created (Console clamps to max(created, activationTime); + # Commander add-ons expose only `created`). + ksm_created_ms = 0 + for a in lic.get('add_ons', []): + if a.get('name') == 'secrets_manager': + ksm_created_ms = KSMCommand._ts_to_ms(a.get('created')) + break + if ksm_created_ms and ksm_created_ms > int(start.timestamp() * 1000): + start = datetime.datetime.fromtimestamp(ksm_created_ms / 1000) + + return int(start.timestamp()), int(end.timestamp()) + + @staticmethod + def _user_fullname_map(params): + result = {} + for u in (params.enterprise.get('users', []) if params.enterprise else []): + data = u.get('data') or {} + name = data.get('displayname') + if u.get('username'): + result[u['username']] = name or '' + return result + + @staticmethod + def _compliance_app_status(params): + """Tenant-wide record existence via compliance/SOX data. Returns (live_uids, trash_uids) or + None if compliance reporting is unavailable. VERY SLOW: performs a FULL enterprise compliance + sync (can take minutes on large tenants) - callers must gate this behind the opt-in --exists + flag. This is the ONLY way to detect deleted KSM apps - the local Trash API (get_deleted_records) + excludes version >= 4 in the backend, so deleted v5 app records are never returned to the client. + `in_trash` is point-in-time (last sync); a record restored after the sync may read stale, and + absence from the dataset is treated as 'purged' only as best effort.""" + try: + from .. import sox + if not sox.is_compliance_reporting_enabled(params): + logging.warning('secrets-manager usage --exists: compliance reporting is not enabled for ' + 'this account; the Exist column will be reported as unknown (?)') + return None + nodes = params.enterprise.get('nodes') or [] + enterprise_id = next(((n['node_id'] >> 32) for n in nodes), 0) + root_node_id = nodes[0]['node_id'] if nodes else 0 + sd = sox.get_compliance_data(params, root_node_id, enterprise_id, rebuild=False, min_updated=0) + live, trash = set(), set() + for rec in sd.get_records().values(): + (trash if rec.in_trash else live).add(rec.record_uid) + return live, trash + except Exception as e: + logging.warning('secrets-manager usage --exists: could not load compliance data (%s); ' + 'the Exist column will be reported as unknown (?)', e) + return None + + @staticmethod + def _resolve_usage_window(params, kwargs): + """Return (min_sec, max_sec) for the usage/summary path. --created overrides billing cycle.""" + from .aram import AuditReportCommand + created = kwargs.get('created') + if created: + if created in ('today', 'yesterday', 'last_7_days', 'last_30_days', 'month_to_date', + 'last_month', 'year_to_date', 'last_year'): + # Named ranges are resolved server-side; return the token for the filter. + return created + flt = AuditReportCommand.get_filter(created, AuditReportCommand.convert_date) + return flt + cycle = KSMCommand._ksm_billing_cycle(params) + if cycle: + return {'min': cycle[0], 'max': cycle[1], 'exclude_max': True} + # No billing-cycle anchor available (license has no next_billing_date); fall back to a 30-day window. + logging.warning('KSM billing cycle unavailable; defaulting to the last 30 days') + return 'last_30_days' + + @staticmethod + def execute_usage(params, **kwargs): + if kwargs.get('helpflag'): + ksm_usage_parser.print_help() + return + if not params.enterprise: + print(f'{bcolors.WARNING}secrets-manager usage requires an enterprise account.{bcolors.ENDC}') + return + + detail = kwargs.get('detail', False) + by_device = kwargs.get('by_device', False) + summary = kwargs.get('summary', False) + timeline = kwargs.get('timeline', False) + export_all = kwargs.get('export_all', False) + time_range = kwargs.get('range') + + # Validate mutually-exclusive families. + if timeline and (detail or by_device or summary): + print(f'{bcolors.WARNING}--timeline cannot be combined with --detail/--by-device/--summary.{bcolors.ENDC}') + return + if summary and (detail or by_device): + print(f'{bcolors.WARNING}--summary is its own report; do not combine with --detail/--by-device.{bcolors.ENDC}') + return + if not timeline and (export_all or time_range): + print(f'{bcolors.WARNING}--all and --range apply only to --timeline.{bcolors.ENDC}') + return + + if timeline: + return KSMCommand._usage_timeline(params, **kwargs) + return KSMCommand._usage_metrics(params, **kwargs) + + @staticmethod + def _usage_metrics(params, **kwargs): + from .aram import fetch_audit_events + detail = kwargs.get('detail', False) + by_device = kwargs.get('by_device', False) + summary = kwargs.get('summary', False) + fmt = kwargs.get('format') or 'table' + output = kwargs.get('output') + limit = kwargs.get('limit') + + created_filter = KSMCommand._resolve_usage_window(params, kwargs) + audit_filter = {'audit_event_type': [KSM_USAGE_EVENT_TYPE], 'created': created_filter} + rows = fetch_audit_events( + params, audit_filter, columns=['app_uid', 'device_name', 'username'], + aggregate=['occurrences'], report_type='span', limit=limit) + + # Aggregation maps mirror Console getSecretsManagerMetrics. The *_apps maps track which + # application UID(s) contributed to each row so we can flag rows whose app has been deleted + # (audit events outlive the app, so usage can reference apps no longer in the vault inventory). + users = {} # username -> occurrences + devices = {} # device_name -> occurrences + applications = set() # distinct app_uid + by_owner = {} # username~|~device_name -> occurrences + by_dev = {} # device_name~|~app_uid~|~username -> occurrences + users_apps = {} # username -> {app_uid} (for the aggregate Exist column) + devices_apps = {} # device_name -> {app_uid} + # (device_name, app_uid) -> occurrences. Device names are NOT unique - the same name can + # belong to different apps (e.g. several "Playground Gateway" from repeated sample-data + # imports). Aggregating by name alone would merge distinct apps into one row with a blurred + # Exist; keying by (name, app_uid) keeps them separate so each row maps to exactly one app. + dev_app = {} + # username~|~device_name -> {app_uid}. Not consumed today (detail views omit the Exist column), + # but kept so a per-owner+device existence flag can be re-added without touching aggregation. + by_owner_apps = {} + for r in rows: + username = r.get('username') or '' + device_name = r.get('device_name') or '' + app_uid = r.get('app_uid') or '' + occ = int(r.get('occurrences') or 0) + users[username] = users.get(username, 0) + occ + devices[device_name] = devices.get(device_name, 0) + occ + dev_app[(device_name, app_uid)] = dev_app.get((device_name, app_uid), 0) + occ + if app_uid: + applications.add(app_uid) + users_apps.setdefault(username, set()).add(app_uid) + devices_apps.setdefault(device_name, set()).add(app_uid) + by_owner_apps.setdefault(_USAGE_KEY_SEP.join((username, device_name)), set()).add(app_uid) + by_owner[_USAGE_KEY_SEP.join((username, device_name))] = \ + by_owner.get(_USAGE_KEY_SEP.join((username, device_name)), 0) + occ + by_dev[_USAGE_KEY_SEP.join((device_name, app_uid, username))] = \ + by_dev.get(_USAGE_KEY_SEP.join((device_name, app_uid, username)), 0) + occ + + name_map = KSMCommand._user_fullname_map(params) + + # Exist column: shown only with --exists, and only on --by-device (each row is a single app). + # Existence comes solely from enterprise compliance data - deleted KSM apps (v5) cannot be + # listed via any local API (get_deleted_records excludes version >= 4 in the backend), so the + # compliance sync is the only source. That sync is VERY SLOW (full tenant scan) - hence opt-in + # behind --exists. Unknown ('?') when compliance data is unavailable. + show_exist = by_device and not detail and bool(kwargs.get('exists')) + sox_live, sox_trash, sox_loaded = set(), set(), False + if show_exist: + logging.warning(f'{bcolors.WARNING}secrets-manager usage --exists is VERY SLOW: it runs a ' + f'FULL enterprise compliance sync (can take minutes on large tenants). ' + f'Resolving app existence...{bcolors.ENDC}') + status = KSMCommand._compliance_app_status(params) + if status is not None: + sox_live, sox_trash = status + sox_loaded = True + + def app_status(a): # -> 'Y' (live) | 'T' (in trash) | 'N' (purged) | '?' (unknown) + if not a or not sox_loaded: + return '?' + if a in sox_live: + return 'Y' + if a in sox_trash: + return 'T' + return 'N' # absent from compliance data -> purged (best effort) + + if summary: + total = sum(users.values()) + users_count = len(users) + avg = (total / users_count) if users_count else 0 + table = [ + ['Total API Usage This Cycle', total], + ['Applications', len(applications)], + ['Devices', len(devices)], + ['Average API Calls Per User', round(avg, 2)], + ] + headers = ['Metric', 'Value'] + elif detail and by_device: + # Full Device Usage: device, app_uid, owner (name+email), usage + table = [] + for key, occ in sorted(by_dev.items(), key=lambda kv: kv[1], reverse=True): + device_name, app_uid, username = key.split(_USAGE_KEY_SEP) + full = name_map.get(username, '') + table.append([device_name, app_uid, full, username, occ]) + headers = ['Device', 'Application UID', 'Owner', 'Email', 'API Usage per Month'] + elif detail: + # Full User Usage: owner (name+email), device, usage + table = [] + for key, occ in sorted(by_owner.items(), key=lambda kv: kv[1], reverse=True): + username, device_name = key.split(_USAGE_KEY_SEP) + full = name_map.get(username, '') + table.append([full, username, device_name, occ]) + headers = ['Owner', 'Email', 'Device', 'API Usage per Month'] + elif by_device: + # Top Usage by Device: one row per (device, app). Suffix the name with the app UID only + # when that device name is shared by more than one app, so each row is unambiguous. The + # Exist column is appended only with --exists (compliance-based; see above). + name_counts = {} + for (dev, _app) in dev_app: + name_counts[dev] = name_counts.get(dev, 0) + 1 + table = [] + for (dev, app), occ in sorted(dev_app.items(), key=lambda kv: kv[1], reverse=True): + label = dev if name_counts.get(dev, 0) <= 1 else f'{dev} ({app or "no-app-uid"})' + row = [label, occ] + if show_exist: + row.append(app_status(app)) + table.append(row) + headers = ['Device', 'Count', 'Exist'] if show_exist else ['Device', 'Count'] + else: + # Default: Top Application Usage (by owner/user): email, count. No Exist column - a single + # owner aggregates many apps, so a combined existence verdict would be misleading. + table = [[user, occ] + for user, occ in sorted(users.items(), key=lambda kv: kv[1], reverse=True)] + headers = ['Application Owner', 'Count'] + + # --sort name: re-sort by the first (name/title) column so related rows group together + # (e.g. all "Playground*" devices). Default 'count' keeps the per-view descending-count order. + # Skipped for --summary, whose rows are fixed metric labels. + if kwargs.get('sort') == 'name' and not summary: + table.sort(key=lambda r: str(r[0]).casefold()) + + result = dump_report_data(table, headers=headers, fmt=fmt, filename=output) + # Table mode: always print the app-list pointer; add the Exist legend only when the Exist + # column is shown (--by-device --exists). + if fmt == 'table': + if show_exist: + print(f'\n{bcolors.OKBLUE}{USAGE_EXIST_LEGEND} {USAGE_APPS_TAIL}{bcolors.ENDC}') + else: + print(f'\n{bcolors.OKBLUE}{USAGE_APPS_TAIL}{bcolors.ENDC}') + return result + + @staticmethod + def _usage_timeline(params, **kwargs): + from .aram import fetch_audit_events + from ..constants import AUDIT_EVENT_STATE_MAPPING + fmt = kwargs.get('format') or 'table' + output = kwargs.get('output') + limit = kwargs.get('limit') + export_all = kwargs.get('export_all', False) + time_range = kwargs.get('range') or '30d' + report_type, _preset, _days = KSM_TIMELINE_RANGES[time_range] + + # Window: mirror Console reportsDatePresets / auditTimeline.js. + now = datetime.datetime.now() + if time_range == '24h': + to_dt = now.replace(minute=0, second=0, microsecond=0) + datetime.timedelta(hours=1) + from_dt = to_dt - datetime.timedelta(hours=24) + else: + to_dt = now.replace(hour=0, minute=0, second=0, microsecond=0) + datetime.timedelta(days=1) + from_dt = to_dt - datetime.timedelta(days=_days + 1) + created_filter = {'min': int(from_dt.timestamp()), 'max': int(to_dt.timestamp())} + + audit_filter = {'audit_event_type': KSM_EVENT_TYPES, 'created': created_filter} + rows = fetch_audit_events( + params, audit_filter, columns=['audit_event_type'], aggregate=['occurrences'], + report_type=report_type, limit=limit, order='descending') + + def label(evt): + return AUDIT_EVENT_STATE_MAPPING.get(evt, evt) + + def fmt_time(created): + dt = datetime.datetime.fromtimestamp(int(created)) + return dt.strftime('%Y-%m-%d %H:%M') if report_type == 'hour' else dt.strftime('%Y-%m-%d') + + if export_all: + # Export All: one row per (time bucket x event), columns Date / Event / Number of Events. + table = [] + for r in sorted(rows, key=lambda x: int(x.get('created') or 0)): + evt = r.get('audit_event_type') or '' + table.append([fmt_time(r.get('created')), label(evt), int(r.get('occurrences') or 0)]) + headers = ['Date', 'Event', 'Number of Events'] + return dump_report_data(table, headers=headers, fmt=fmt, filename=output) + + # Default timeline: event totals over the range, with % of total. + totals = {} + for r in rows: + evt = r.get('audit_event_type') or '' + totals[evt] = totals.get(evt, 0) + int(r.get('occurrences') or 0) + grand = sum(totals.values()) + table = [] + for evt, cnt in sorted(totals.items(), key=lambda kv: kv[1], reverse=True): + pct = round(cnt * 100.0 / grand, 1) if grand else 0 + table.append([label(evt), cnt, f'{pct}%']) + return dump_report_data(table, headers=['Event', 'Count', '% of Total'], fmt=fmt, filename=output) + @staticmethod def share_app(params, app_name_or_uid, email, is_admin=False, unshare=False): # type: (KeeperParams, str, List[str], Optional[bool], Optional[bool]) -> None diff --git a/unit-tests/test_command_ksm.py b/unit-tests/test_command_ksm.py index b2f18521d..ab0c98716 100644 --- a/unit-tests/test_command_ksm.py +++ b/unit-tests/test_command_ksm.py @@ -1,8 +1,9 @@ """Unit tests for secrets-manager (KSM) CLI commands.""" +import datetime import unittest from unittest.mock import MagicMock, patch -from keepercommander.commands.ksm import KSMCommand +from keepercommander.commands.ksm import KSMCommand, KSM_EVENT_TYPES, KSM_USAGE_EVENT_TYPE class TestKSMSecretResolution(unittest.TestCase): @@ -215,5 +216,283 @@ def test_token_add_missing_app_prints_help(self): assert result is None, f"Expected None, got {result!r}" +class TestKSMUsage(unittest.TestCase): + """secrets-manager usage report - aggregation, billing cycle, timeline, flag matrix.""" + + SAMPLE_ROWS = [ + {'username': 'a@x.com', 'device_name': 'D1', 'app_uid': 'APP1', 'occurrences': 10}, + {'username': 'a@x.com', 'device_name': 'D2', 'app_uid': 'APP1', 'occurrences': 5}, + {'username': 'b@x.com', 'device_name': 'D1', 'app_uid': 'APP2', 'occurrences': 3}, + ] + + def _make_params(self): + params = MagicMock() + params.environment_variables = {} + params.enterprise = { + 'users': [ + {'username': 'a@x.com', 'data': {'displayname': 'Alice A'}}, + {'username': 'b@x.com', 'data': {'displayname': 'Bob B'}}, + ], + 'licenses': [{ + 'next_billing_date': int(datetime.datetime(2026, 1, 15, 12, 0, 0).timestamp() * 1000), + 'add_ons': [{'name': 'secrets_manager', + 'created': int(datetime.datetime(2025, 1, 1).timestamp() * 1000)}], + }], + } + return params + + # ---- unit helpers ---- + + def test_event_type_constants(self): + self.assertEqual(len(KSM_EVENT_TYPES), 16) + self.assertIn('app_client_access', KSM_EVENT_TYPES) + self.assertEqual(KSM_USAGE_EVENT_TYPE, 'app_client_access') + + def test_ts_to_ms_normalization(self): + self.assertEqual(KSMCommand._ts_to_ms(1_700_000_000), 1_700_000_000_000) # seconds -> ms + self.assertEqual(KSMCommand._ts_to_ms(1_700_000_000_000), 1_700_000_000_000) # ms passthrough + self.assertEqual(KSMCommand._ts_to_ms(0), 0) + self.assertEqual(KSMCommand._ts_to_ms(None), 0) + + def test_set_month_day_clamp(self): + jan31 = datetime.datetime(2026, 1, 31, 9, 0, 0) + feb = KSMCommand._set_month(jan31, 1, 2026) # month0=1 -> February + self.assertEqual((feb.month, feb.day), (2, 28)) + + def test_billing_cycle_run_after_billing_day(self): + params = self._make_params() + run_ms = int(datetime.datetime(2026, 7, 16, 9, 0, 0).timestamp() * 1000) + min_sec, max_sec = KSMCommand._ksm_billing_cycle(params, run_on_ms=run_ms) + start = datetime.datetime.fromtimestamp(min_sec) + end = datetime.datetime.fromtimestamp(max_sec) + self.assertEqual((start.year, start.month, start.day), (2026, 7, 15)) + self.assertEqual((end.year, end.month, end.day), (2026, 8, 15)) + self.assertLess(min_sec, max_sec) + + def test_billing_cycle_none_without_billing_date(self): + params = self._make_params() + params.enterprise['licenses'][0]['next_billing_date'] = 0 + self.assertIsNone(KSMCommand._ksm_billing_cycle(params)) + + # ---- aggregation / rendering (assert the table handed to dump_report_data) ---- + + # One device -> one app, for clean per-device Exist assertions. + ONE_DEV_ROWS = [{'username': 'u@x.com', 'device_name': 'Donly', 'app_uid': 'APPX', 'occurrences': 7}] + + def _run_metrics(self, params, sox=None, rows=None, **flags): + # sox: None (compliance unavailable) or (live_uids, trash_uids) returned by the --exists path. + rows = list(self.SAMPLE_ROWS if rows is None else rows) + with patch('keepercommander.commands.aram.fetch_audit_events', return_value=rows), \ + patch('keepercommander.commands.ksm.KSMCommand._compliance_app_status', return_value=sox), \ + patch('keepercommander.commands.ksm.dump_report_data') as mock_dump: + KSMCommand._usage_metrics(params, format='json', **flags) + self.assertTrue(mock_dump.called) + _args, kwargs = mock_dump.call_args + table = _args[0] + headers = kwargs.get('headers') or (_args[1] if len(_args) > 1 else None) + return table, headers + + def test_default_top_application_usage_no_exist_column(self): + table, headers = self._run_metrics(self._make_params()) + self.assertEqual(headers, ['Application Owner', 'Count']) + self.assertEqual(table, [['a@x.com', 15], ['b@x.com', 3]]) + + def test_by_device_no_exist_column_by_default(self): + # Without --exists there is NO Exist column - just Device/Count. Names shared by multiple apps + # are still split with a UID suffix so each row maps to one app. + table, headers = self._run_metrics(self._make_params(), by_device=True) + self.assertEqual(headers, ['Device', 'Count']) + self.assertEqual(table, [['D1 (APP1)', 10], ['D2', 5], ['D1 (APP2)', 3]]) + + def test_by_device_sort_by_name(self): + # --sort name groups related device names alphabetically instead of by count. + rows = [{'username': 'u@x.com', 'device_name': 'zeta', 'app_uid': 'A1', 'occurrences': 100}, + {'username': 'u@x.com', 'device_name': 'Playground B', 'app_uid': 'A2', 'occurrences': 5}, + {'username': 'u@x.com', 'device_name': 'Playground A', 'app_uid': 'A3', 'occurrences': 50}] + table, _ = self._run_metrics(self._make_params(), rows=rows, by_device=True, sort='name') + self.assertEqual([r[0] for r in table], ['Playground A', 'Playground B', 'zeta']) + # default (count) keeps descending-usage order + table2, _ = self._run_metrics(self._make_params(), rows=rows, by_device=True, sort='count') + self.assertEqual([r[0] for r in table2], ['zeta', 'Playground A', 'Playground B']) + + def test_by_device_unique_name_no_suffix(self): + rows = [{'username': 'u@x.com', 'device_name': 'Solo', 'app_uid': 'APP1', 'occurrences': 9}] + table, headers = self._run_metrics(self._make_params(), rows=rows, by_device=True) + self.assertEqual(headers, ['Device', 'Count']) + self.assertEqual(table, [['Solo', 9]]) # single app -> no UID suffix + + def test_by_device_exists_live_and_purged(self): + # --exists: compliance says APP1 live, APP2 unknown -> purged (N). + table, headers = self._run_metrics(self._make_params(), sox=(frozenset({'APP1'}), frozenset()), + by_device=True, exists=True) + self.assertEqual(headers, ['Device', 'Count', 'Exist']) + self.assertEqual(table, [['D1 (APP1)', 10, 'Y'], ['D2', 5, 'Y'], ['D1 (APP2)', 3, 'N']]) + + def test_by_device_exists_trash(self): + table, _ = self._run_metrics(self._make_params(), sox=(frozenset(), frozenset({'APPX'})), + rows=self.ONE_DEV_ROWS, by_device=True, exists=True) + self.assertEqual(table, [['Donly', 7, 'T']]) # in_trash per compliance + + def test_by_device_exists_unknown_when_compliance_unavailable(self): + # --exists but compliance data can't be loaded -> '?' + table, _ = self._run_metrics(self._make_params(), sox=None, + rows=self.ONE_DEV_ROWS, by_device=True, exists=True) + self.assertEqual(table, [['Donly', 7, '?']]) + + def test_exists_not_queried_without_flag(self): + params = self._make_params() + with patch('keepercommander.commands.aram.fetch_audit_events', return_value=list(self.SAMPLE_ROWS)), \ + patch('keepercommander.commands.ksm.KSMCommand._compliance_app_status') as mock_sox, \ + patch('keepercommander.commands.ksm.dump_report_data'): + KSMCommand._usage_metrics(params, format='json', by_device=True) # no --exists + mock_sox.assert_not_called() + + def test_exists_ignored_off_by_device(self): + # --exists only applies to --by-device; on default/summary/detail the compliance sync must not run. + params = self._make_params() + for flags in ({}, {'summary': True}, {'detail': True}, {'detail': True, 'by_device': True}): + with patch('keepercommander.commands.aram.fetch_audit_events', return_value=list(self.SAMPLE_ROWS)), \ + patch('keepercommander.commands.ksm.KSMCommand._compliance_app_status') as mock_sox, \ + patch('keepercommander.commands.ksm.dump_report_data'): + KSMCommand._usage_metrics(params, format='json', exists=True, **flags) + mock_sox.assert_not_called() + + def test_detail_full_user_usage_no_exist_column(self): + table, headers = self._run_metrics(self._make_params(), detail=True) + self.assertEqual(headers, ['Owner', 'Email', 'Device', 'API Usage per Month']) + self.assertEqual(table, [['Alice A', 'a@x.com', 'D1', 10], + ['Alice A', 'a@x.com', 'D2', 5], + ['Bob B', 'b@x.com', 'D1', 3]]) + + def test_detail_by_device_full_device_usage_no_exist_column(self): + table, headers = self._run_metrics(self._make_params(), detail=True, by_device=True) + self.assertEqual(headers, ['Device', 'Application UID', 'Owner', 'Email', 'API Usage per Month']) + self.assertEqual(table, [['D1', 'APP1', 'Alice A', 'a@x.com', 10], + ['D2', 'APP1', 'Alice A', 'a@x.com', 5], + ['D1', 'APP2', 'Bob B', 'b@x.com', 3]]) + + def test_summary(self): + table, headers = self._run_metrics(self._make_params(), summary=True) + self.assertEqual(headers, ['Metric', 'Value']) + as_dict = {row[0]: row[1] for row in table} + self.assertEqual(as_dict['Total API Usage This Cycle'], 18) + self.assertEqual(as_dict['Applications'], 2) + self.assertEqual(as_dict['Devices'], 2) + self.assertEqual(as_dict['Average API Calls Per User'], 9.0) + self.assertNotIn(' live (Y)', as_dict) + self.assertNotIn(' purged (N)', as_dict) + + def _capture_table_footer(self, sox=None, **flags): + import io + import contextlib + params = self._make_params() + buf = io.StringIO() + with patch('keepercommander.commands.aram.fetch_audit_events', return_value=list(self.SAMPLE_ROWS)), \ + patch('keepercommander.commands.ksm.KSMCommand._compliance_app_status', return_value=sox), \ + contextlib.redirect_stdout(buf): + KSMCommand._usage_metrics(params, format='table', **flags) + return buf.getvalue() + + def test_footer_exist_legend_only_with_exists(self): + # No --exists anywhere -> app-list note present, Exist legend absent. + for flags in ({}, {'by_device': True}, {'detail': True}, {'summary': True}): + out = self._capture_table_footer(**flags) + self.assertIn('secrets-manager app list', out) + self.assertNotIn('Exist:', out) + # --by-device --exists -> legend present. + out = self._capture_table_footer(sox=(frozenset({'APP1'}), frozenset()), by_device=True, exists=True) + self.assertIn('Exist:', out) + + def test_usage_uses_only_app_client_access_filter(self): + params = self._make_params() + with patch('keepercommander.commands.aram.fetch_audit_events', return_value=[]) as mock_fetch, \ + patch('keepercommander.commands.ksm.dump_report_data'): + KSMCommand._usage_metrics(params, format='table') + _args, kwargs = mock_fetch.call_args + audit_filter = _args[1] + self.assertEqual(audit_filter['audit_event_type'], ['app_client_access']) + self.assertEqual(kwargs.get('columns'), ['app_uid', 'device_name', 'username']) + self.assertEqual(kwargs.get('aggregate'), ['occurrences']) + + # ---- timeline ---- + + def test_timeline_export_all_row_shape(self): + params = self._make_params() + rows = [ + {'audit_event_type': 'app_client_access', 'created': 1_700_000_100, 'occurrences': 4}, + {'audit_event_type': 'app_client_added', 'created': 1_700_000_000, 'occurrences': 2}, + ] + with patch('keepercommander.commands.aram.fetch_audit_events', return_value=rows) as mock_fetch, \ + patch('keepercommander.commands.ksm.dump_report_data') as mock_dump: + KSMCommand._usage_timeline(params, format='csv', export_all=True, range='7d') + # full KSM event set requested for timeline + _fargs, fkwargs = mock_fetch.call_args + self.assertEqual(_fargs[1]['audit_event_type'], KSM_EVENT_TYPES) + self.assertEqual(fkwargs.get('report_type'), 'day') + _args, kwargs = mock_dump.call_args + table = _args[0] + headers = kwargs.get('headers') + self.assertEqual(headers, ['Date', 'Event', 'Number of Events']) + # sorted ascending by created; count in last column + self.assertEqual([r[2] for r in table], [2, 4]) + self.assertEqual(len(table[0]), 3) + + def test_timeline_default_totals(self): + params = self._make_params() + rows = [ + {'audit_event_type': 'app_client_access', 'created': 1_700_000_000, 'occurrences': 6}, + {'audit_event_type': 'app_client_access', 'created': 1_700_100_000, 'occurrences': 4}, + {'audit_event_type': 'app_client_added', 'created': 1_700_000_000, 'occurrences': 10}, + ] + with patch('keepercommander.commands.aram.fetch_audit_events', return_value=rows), \ + patch('keepercommander.commands.ksm.dump_report_data') as mock_dump: + KSMCommand._usage_timeline(params, format='table', range='24h') + _args, kwargs = mock_dump.call_args + table = _args[0] + self.assertEqual(kwargs.get('headers'), ['Event', 'Count', '% of Total']) + counts = {row[0]: row[1] for row in table} + # app_client_access totals to 10, app_client_added to 10 + self.assertEqual(sorted(counts.values(), reverse=True), [10, 10]) + + # ---- flag validation & routing ---- + + def test_timeline_conflicts_with_detail(self): + params = self._make_params() + with patch('keepercommander.commands.aram.fetch_audit_events') as mock_fetch: + KSMCommand.execute_usage(params, timeline=True, detail=True) + mock_fetch.assert_not_called() + + def test_range_without_timeline_rejected(self): + params = self._make_params() + with patch('keepercommander.commands.aram.fetch_audit_events') as mock_fetch: + KSMCommand.execute_usage(params, range='7d') + mock_fetch.assert_not_called() + + def test_requires_enterprise(self): + params = self._make_params() + params.enterprise = None + with patch('keepercommander.commands.aram.fetch_audit_events') as mock_fetch: + KSMCommand.execute_usage(params) + mock_fetch.assert_not_called() + + def test_execute_args_routes_usage_verb(self): + params = self._make_params() + cmd = KSMCommand() + with patch('keepercommander.commands.ksm.KSMCommand.execute_usage') as mock_usage: + cmd.execute_args(params, 'usage --by-device') + self.assertTrue(mock_usage.called) + _args, kwargs = mock_usage.call_args + self.assertTrue(kwargs.get('by_device')) + + def test_execute_args_non_usage_delegates(self): + params = self._make_params() + cmd = KSMCommand() + with patch('keepercommander.commands.ksm.KSMCommand.execute_usage') as mock_usage, \ + patch('keepercommander.commands.base.Command.execute_args') as mock_super: + cmd.execute_args(params, 'app list') + mock_usage.assert_not_called() + mock_super.assert_called_once() + + if __name__ == '__main__': unittest.main()