From 68968a0db679f31e6086a7ec10833c5c69c25aba Mon Sep 17 00:00:00 2001 From: adeshmukh-ks Date: Thu, 27 Aug 2026 21:17:47 +0530 Subject: [PATCH 1/4] PAM Workflow SDK functions and CLI commands added (#229) --- .../src/keepercli/commands/pam/keeper_pam.py | 2 + .../keepercli/commands/pam/pam_workflow.py | 775 ++++++++++++++++ .../src/keepercli/register_commands.py | 2 + keepersdk-package/requirements.txt | 1 + keepersdk-package/setup.cfg | 2 +- .../keepersdk/helpers/workflow/__init__.py | 56 ++ .../src/keepersdk/helpers/workflow/helpers.py | 716 +++++++++++++++ .../keepersdk/helpers/workflow/workflow.py | 845 ++++++++++++++++++ .../src/keepersdk/proto/workflow_pb2.py | 79 ++ .../src/keepersdk/proto/workflow_pb2.pyi | 259 ++++++ keepersdk-package/unit_tests/test_workflow.py | 84 ++ 11 files changed, 2820 insertions(+), 1 deletion(-) create mode 100644 keepercli-package/src/keepercli/commands/pam/pam_workflow.py create mode 100644 keepersdk-package/src/keepersdk/helpers/workflow/__init__.py create mode 100644 keepersdk-package/src/keepersdk/helpers/workflow/helpers.py create mode 100644 keepersdk-package/src/keepersdk/helpers/workflow/workflow.py create mode 100644 keepersdk-package/src/keepersdk/proto/workflow_pb2.py create mode 100644 keepersdk-package/src/keepersdk/proto/workflow_pb2.pyi create mode 100644 keepersdk-package/unit_tests/test_workflow.py diff --git a/keepercli-package/src/keepercli/commands/pam/keeper_pam.py b/keepercli-package/src/keepercli/commands/pam/keeper_pam.py index cd99959b..d400edfd 100644 --- a/keepercli-package/src/keepercli/commands/pam/keeper_pam.py +++ b/keepercli-package/src/keepercli/commands/pam/keeper_pam.py @@ -10,6 +10,7 @@ from .pam_rotation import PAMCreateRecordRotationCommand, PAMListRecordRotationCommand, PAMRouterGetRotationInfo, PAMRouterScriptCommand from .pam_connection import PAMConnectionEditCommand from .pam_rbi import PAMRbiEditCommand +from .pam_workflow import PAMWorkflowCommand from .. import enterprise_utils from .. import base from ... import api @@ -83,6 +84,7 @@ def __init__(self): self.register_command(PAMRotationCommand(), 'rotation', 'r') self.register_command(PAMConnectionCommand(), 'connection', 'n') self.register_command(PAMRbiCommand(), 'rbi', 'b') + self.register_command(PAMWorkflowCommand(), 'workflow', 'w') class PAMGatewayCommand(base.GroupCommand): diff --git a/keepercli-package/src/keepercli/commands/pam/pam_workflow.py b/keepercli-package/src/keepercli/commands/pam/pam_workflow.py new file mode 100644 index 00000000..cd21bebe --- /dev/null +++ b/keepercli-package/src/keepercli/commands/pam/pam_workflow.py @@ -0,0 +1,775 @@ +import argparse +import json +import shlex +from datetime import datetime +from typing import Optional + +from keepersdk.helpers.workflow import ( + WorkflowError, + add_workflow_approvers, + approve_workflow, + can_configure_workflow_settings, + create_workflow, + delete_workflow, + deny_workflow, + end_workflow, + get_pending_approvals, + get_user_access_state, + get_workflow_state, + read_workflow, + remove_workflow_approvers, + request_workflow_access, + start_workflow, + update_workflow, +) + +from .. import base +from ... import api +from ...helpers import report_utils +from ...params import KeeperParams + + +logger = api.get_logger() + +_ADMIN_VERBS = frozenset({'create', 'update', 'delete', 'add-approver', 'remove-approver'}) + + +def _require_vault(context: KeeperParams): + base.require_login(context) + if context.vault is None: + raise base.CommandError('Vault is not initialized, login to initialize the vault.') + + +def _get_enterprise_data(context: KeeperParams): + """Best-effort enterprise directory access, used to resolve numeric user IDs and + teams the current user doesn't personally belong to. Returns None for non-admin + sessions (e.g. --skip-enterprise) where only vault-local resolution is possible.""" + try: + auth = context.auth + if auth is not None and auth.auth_context.is_enterprise_admin: + return context.enterprise_loader.enterprise_data + except AssertionError: + pass + return None + + +def _run_sdk(fn, *args, **kwargs): + try: + return fn(*args, **kwargs) + except WorkflowError as e: + raise base.CommandError(e.message) from e + + +def _is_json(kwargs) -> bool: + return kwargs.get('format') == 'json' + + +def _emit_json(payload: dict): + return json.dumps(payload, indent=2) + + +def _fmt_ts(ts_ms) -> str: + if not ts_ms: + return '' + dt = datetime.fromtimestamp(ts_ms / 1000).astimezone() + tz_label = dt.strftime('%Z') or dt.tzname() or 'local' + return dt.strftime(f'%Y-%m-%d %H:%M:%S {tz_label}') + + +def _fix_dash_uid_args(parser: argparse.ArgumentParser, args: str) -> str: + """Insert '--' before a base64url UID starting with '-' so argparse treats it as positional.""" + if not args: + return args + try: + tokens = shlex.split(args) + except ValueError: + return args + if '--' in tokens: + return args + + known_opts = set() + consumes_value = set() + for action in parser._actions: + for opt in action.option_strings: + known_opts.add(opt) + if action.nargs not in (0, '?'): + consumes_value.add(opt) + + result = [] + skip_next = False + for token in tokens: + if skip_next: + result.append(token) + skip_next = False + continue + opt_name = token.split('=', 1)[0] if token.startswith('--') and '=' in token else token + if opt_name in known_opts: + result.append(token) + if opt_name in consumes_value and token == opt_name: + skip_next = True + continue + if token.startswith('-'): + result.append('--') + result.append(token) + + if len(result) != len(tokens): + return ' '.join(shlex.quote(t) for t in result) + return args + + +class DashUidArgsMixin: + """Mixin for commands whose positional flow-UID arg may start with '-' (base64url).""" + + def execute_args(self, context, args, **kwargs): + args = _fix_dash_uid_args(self.get_parser(), args) + return super().execute_args(context, args, **kwargs) + + +class _WorkflowCommand(base.ArgparseCommand): + def execute(self, context: KeeperParams, **kwargs): + _require_vault(context) + return self.execute_workflow(context, **kwargs) + + def execute_workflow(self, context: KeeperParams, **kwargs): + raise NotImplementedError + + +class PAMWorkflowCommand(base.GroupCommand): + def __init__(self): + super().__init__('PAM Workflow') + self._current_context: Optional[KeeperParams] = None + self.register_command(WorkflowCreateCommand(), 'create', 'c') + self.register_command(WorkflowReadCommand(), 'read', 'r') + self.register_command(WorkflowUpdateCommand(), 'update', 'u') + self.register_command(WorkflowDeleteCommand(), 'delete', 'd') + self.register_command(WorkflowAddApproversCommand(), 'add-approver', 'aa') + self.register_command(WorkflowDeleteApproversCommand(), 'remove-approver', 'ra') + self.register_command(WorkflowGetApprovalRequestsCommand(), 'pending', 'p') + self.register_command(WorkflowApproveCommand(), 'approve', 'a') + self.register_command(WorkflowDenyCommand(), 'deny', 'dn') + self.register_command(WorkflowRequestAccessCommand(), 'request', 'rq') + self.register_command(WorkflowStartCommand(), 'start', 's') + self.register_command(WorkflowEndCommand(), 'end', 'e') + self.register_command(WorkflowGetStateCommand(), 'state', 'st') + self.register_command(WorkflowGetUserAccessStateCommand(), 'my-access', 'ma') + self.default_verb = 'state' + + def execute_args(self, context: KeeperParams, args, **kwargs): + self._current_context = context + _require_vault(context) + return super().execute_args(context, args, **kwargs) + + def print_help(self, **kwargs): + context = self._current_context + is_admin = bool(context and context.vault and can_configure_workflow_settings(context.vault)) + print(f'{kwargs.get("command")} command [--options]') + table = [] + headers = ['Command', 'Description'] + for verb, command in self.commands.items(): + if verb in _ADMIN_VERBS and not is_admin: + continue + table.append([verb, command.description()]) + print('') + report_utils.dump_report_data(table, headers=headers) + print('') + + +class WorkflowCreateCommand(_WorkflowCommand): + def __init__(self): + parser = argparse.ArgumentParser( + prog='pam workflow create', + description='Create workflow configuration for a PAM record', + allow_abbrev=False, + parents=[base.json_output_parser], + ) + self.add_arguments_to_parser(parser) + super().__init__(parser) + + @staticmethod + def add_arguments_to_parser(parser: argparse.ArgumentParser): + parser.add_argument('record', help='Record UID or name to configure workflow for') + parser.add_argument('-n', '--approvals-needed', type=int, default=1, + help='Number of approvals required (default: 1)') + parser.add_argument('-co', '--checkout', action='store_true', + help='Enable single-user check-in/check-out mode') + parser.add_argument('-sa', '--start-on-approval', action='store_true', + help='Start access timer when approved (vs when checked out)') + parser.add_argument('-rr', '--require-reason', action='store_true', + help='Require user to provide reason for access') + parser.add_argument('-rt', '--require-ticket', action='store_true', + help='Require user to provide ticket number') + parser.add_argument('-rm', '--require-mfa', action='store_true', + help='Require MFA verification for access') + parser.add_argument('-d', '--duration', type=str, default='1d', + help='Access duration (e.g., "2h", "30m", "1d"). Default: 1d') + parser.add_argument('--allowed-days', type=str, + help='Comma-separated allowed days (e.g., "mon,tue,wed,thu,fri")') + parser.add_argument('--time-range', type=str, + help='Allowed time range in HH:MM-HH:MM format (e.g., "09:00-17:00")') + parser.add_argument('-u', '--approver', action='append', + help='User email to add as an approver. Pass multiple times to ' + 'add several. Required when --approvals-needed > 0. ' + 'Duplicates are removed automatically.') + + def execute_workflow(self, context: KeeperParams, **kwargs): + result = _run_sdk( + create_workflow, + context.vault, + kwargs.get('record'), + approvals_needed=kwargs.get('approvals_needed', 1), + checkout=bool(kwargs.get('checkout')), + start_on_approval=bool(kwargs.get('start_on_approval')), + require_reason=bool(kwargs.get('require_reason')), + require_ticket=bool(kwargs.get('require_ticket')), + require_mfa=bool(kwargs.get('require_mfa')), + duration=kwargs.get('duration') or '1d', + allowed_days=kwargs.get('allowed_days'), + time_range=kwargs.get('time_range'), + approvers=kwargs.get('approver'), + ) + if _is_json(kwargs): + return _emit_json(result) + cfg = result.get('workflow_config') or {} + logger.info('Workflow created successfully') + logger.info('Record: %s (%s)', result.get('record_name'), result.get('record_uid')) + logger.info('Approvals needed: %s', cfg.get('approvals_needed')) + logger.info('Check-in/out: %s', 'Yes' if cfg.get('checkout_needed') else 'No') + logger.info('Duration: %s', cfg.get('access_duration')) + if cfg.get('require_reason'): + logger.info('Requires reason: Yes') + if cfg.get('require_ticket'): + logger.info('Requires ticket: Yes') + if cfg.get('require_mfa'): + logger.info('Requires MFA: Yes') + approvers = result.get('approvers') or [] + if approvers: + logger.info('Approvers: %s', ', '.join(approvers)) + elif cfg.get('approvals_needed'): + logger.warning( + 'Note: Add approvers with: pam workflow add-approver %s --user ', + result.get('record_uid'), + ) + + +class WorkflowReadCommand(_WorkflowCommand): + def __init__(self): + parser = argparse.ArgumentParser( + prog='pam workflow read', + description='Read and display workflow configuration', + parents=[base.json_output_parser], + ) + self.add_arguments_to_parser(parser) + super().__init__(parser) + + @staticmethod + def add_arguments_to_parser(parser: argparse.ArgumentParser): + parser.add_argument('record', help='Record UID or name') + + def execute_workflow(self, context: KeeperParams, **kwargs): + result = _run_sdk( + read_workflow, context.vault, kwargs.get('record'), + enterprise_data=_get_enterprise_data(context), + ) + if _is_json(kwargs): + return _emit_json(result) + if result.get('status') == 'no_workflow': + logger.warning('No workflow configured for this record') + logger.info('Record: %s (%s)', result.get('record_name'), result.get('record_uid')) + logger.info('To create a workflow, run:') + logger.info(' pam workflow create %s', result.get('record_uid')) + return + self._print_table(result) + + @staticmethod + def _print_table(result: dict): + logger.info('Workflow Configuration') + logger.info('Record: %s', result.get('record_name')) + logger.info('Record UID: %s', result.get('record_uid')) + if result.get('created_on'): + logger.info('Created: %s', _fmt_ts(result['created_on'])) + p = result.get('parameters') or {} + logger.info('Access Parameters:') + logger.info(' Approvals needed: %s', p.get('approvals_needed')) + logger.info(' Check-in/out required: %s', 'Yes' if p.get('checkout_needed') else 'No') + logger.info(' Access duration: %s', p.get('access_duration')) + logger.info(' Timer starts: %s', 'On approval' if p.get('start_access_on_approval') else 'On check-out') + logger.info('Requirements:') + logger.info(' Reason required: %s', 'Yes' if p.get('require_reason') else 'No') + logger.info(' Ticket required: %s', 'Yes' if p.get('require_ticket') else 'No') + logger.info(' MFA required: %s', 'Yes' if p.get('require_mfa') else 'No') + allowed = p.get('allowed_times') or {} + if allowed: + logger.info('Allowed Times:') + if allowed.get('allowed_days'): + logger.info(' Days: %s', ', '.join(allowed['allowed_days'])) + if allowed.get('time_ranges'): + for tr in allowed['time_ranges']: + logger.info(' Time: %s', tr.replace('-', ' - ')) + if allowed.get('timezone'): + logger.info(' Timezone: %s', allowed['timezone']) + approvers = result.get('approvers') or [] + if approvers: + logger.info('Approvers (%s):', len(approvers)) + for idx, approver in enumerate(approvers, 1): + esc_label = '' + if approver.get('escalation'): + esc_label = ' (Escalation' + if approver.get('escalation_after'): + esc_label += f' after {approver["escalation_after"]}' + esc_label += ')' + if approver.get('type') == 'user': + logger.info(' %s. User: %s%s', idx, approver.get('email'), esc_label) + elif approver.get('type') == 'user_id': + logger.info(' %s. User: %s%s', idx, approver.get('email') or approver.get('user_id'), esc_label) + elif approver.get('type') == 'team': + team_uid = approver.get('team_uid') + team_name = approver.get('team_name') + team_display = f'{team_name} ({team_uid})' if team_name else team_uid + logger.info(' %s. Team: %s%s', idx, team_display, esc_label) + else: + logger.warning('No approvers configured') + logger.info('Add approvers with: pam workflow add-approver %s --user ', result.get('record_uid')) + + +class WorkflowUpdateCommand(_WorkflowCommand): + def __init__(self): + parser = argparse.ArgumentParser( + prog='pam workflow update', + description='Update existing workflow configuration. ' + 'Only specified fields are changed; unspecified fields retain their current values.', + parents=[base.json_output_parser], + ) + self.add_arguments_to_parser(parser) + super().__init__(parser) + + @staticmethod + def add_arguments_to_parser(parser: argparse.ArgumentParser): + parser.add_argument('record', help='Record UID or name with workflow to update') + parser.add_argument('-n', '--approvals-needed', type=int, help='Number of approvals required') + parser.add_argument('-co', '--checkout', type=lambda x: x.lower() == 'true', + help='Enable/disable check-in/check-out (true/false)') + parser.add_argument('-sa', '--start-on-approval', type=lambda x: x.lower() == 'true', + help='Start timer on approval vs check-out (true/false)') + parser.add_argument('-rr', '--require-reason', type=lambda x: x.lower() == 'true', + help='Require reason (true/false)') + parser.add_argument('-rt', '--require-ticket', type=lambda x: x.lower() == 'true', + help='Require ticket (true/false)') + parser.add_argument('-rm', '--require-mfa', type=lambda x: x.lower() == 'true', + help='Require MFA (true/false)') + parser.add_argument('-d', '--duration', type=str, help='Access duration (e.g., "2h", "30m", "1d")') + parser.add_argument('--allowed-days', type=str, + help='Comma-separated allowed days (e.g., "mon,tue,wed,thu,fri")') + parser.add_argument('--time-range', type=str, + help='Allowed time range in HH:MM-HH:MM format (e.g., "09:00-17:00")') + + def execute_workflow(self, context: KeeperParams, **kwargs): + result = _run_sdk( + update_workflow, + context.vault, + kwargs.get('record'), + approvals_needed=kwargs.get('approvals_needed'), + checkout=kwargs.get('checkout'), + start_on_approval=kwargs.get('start_on_approval'), + require_reason=kwargs.get('require_reason'), + require_ticket=kwargs.get('require_ticket'), + require_mfa=kwargs.get('require_mfa'), + duration=kwargs.get('duration'), + allowed_days=kwargs.get('allowed_days'), + time_range=kwargs.get('time_range'), + ) + if _is_json(kwargs): + return _emit_json(result) + logger.info('Workflow updated successfully') + logger.info('Record: %s (%s)', result.get('record_name'), result.get('record_uid')) + + +class WorkflowDeleteCommand(_WorkflowCommand): + def __init__(self): + parser = argparse.ArgumentParser( + prog='pam workflow delete', + description='Delete workflow configuration from a record', + parents=[base.json_output_parser], + ) + self.add_arguments_to_parser(parser) + super().__init__(parser) + + @staticmethod + def add_arguments_to_parser(parser: argparse.ArgumentParser): + parser.add_argument('record', help='Record UID or name to remove workflow from') + + def execute_workflow(self, context: KeeperParams, **kwargs): + result = _run_sdk(delete_workflow, context.vault, kwargs.get('record')) + if _is_json(kwargs): + return _emit_json(result) + logger.info('Workflow deleted successfully') + logger.info('Record: %s (%s)', result.get('record_name'), result.get('record_uid')) + + +class WorkflowAddApproversCommand(_WorkflowCommand): + def __init__(self): + parser = argparse.ArgumentParser( + prog='pam workflow add-approver', + description='Add approvers to a workflow', + parents=[base.json_output_parser], + ) + self.add_arguments_to_parser(parser) + super().__init__(parser) + + @staticmethod + def add_arguments_to_parser(parser: argparse.ArgumentParser): + parser.add_argument('record', help='Record UID or name') + parser.add_argument('-u', '--user', action='append', + help='User email to add as approver (can specify multiple times)') + parser.add_argument('-t', '--team', action='append', + help='Team name or UID to add as approver (can specify multiple times)') + parser.add_argument('-e', '--escalation', action='store_true', help='Mark as escalation approver') + parser.add_argument('-ea', '--escalation-after', type=str, + help='Time before escalating to this approver (e.g., "30m", "1h", "2h"). ' + 'Only meaningful with --escalation') + + def execute_workflow(self, context: KeeperParams, **kwargs): + result = _run_sdk( + add_workflow_approvers, + context.vault, + kwargs.get('record'), + users=kwargs.get('user'), + teams=kwargs.get('team'), + escalation=bool(kwargs.get('escalation')), + escalation_after=kwargs.get('escalation_after'), + enterprise_data=_get_enterprise_data(context), + ) + if _is_json(kwargs): + return _emit_json(result) + logger.info('Approvers added successfully') + logger.info('Record: %s (%s)', result.get('record_name'), result.get('record_uid')) + logger.info('Added %s approver(s)', result.get('approvers_added')) + if result.get('escalation'): + esc_info = f' (after {result["escalation_after"]})' if result.get('escalation_after') else '' + logger.info('Type: Escalation approver%s', esc_info) + + +class WorkflowDeleteApproversCommand(_WorkflowCommand): + def __init__(self): + parser = argparse.ArgumentParser( + prog='pam workflow remove-approver', + description='Remove approvers from a workflow', + parents=[base.json_output_parser], + ) + self.add_arguments_to_parser(parser) + super().__init__(parser) + + @staticmethod + def add_arguments_to_parser(parser: argparse.ArgumentParser): + parser.add_argument('record', help='Record UID or name') + parser.add_argument('-u', '--user', action='append', help='User email to remove as approver') + parser.add_argument('-t', '--team', action='append', help='Team name or UID to remove as approver') + + def execute_workflow(self, context: KeeperParams, **kwargs): + result = _run_sdk( + remove_workflow_approvers, + context.vault, + kwargs.get('record'), + users=kwargs.get('user'), + teams=kwargs.get('team'), + enterprise_data=_get_enterprise_data(context), + ) + if _is_json(kwargs): + return _emit_json(result) + logger.info('Approvers removed successfully') + logger.info('Record: %s (%s)', result.get('record_name'), result.get('record_uid')) + logger.info('Removed %s approver(s)', result.get('approvers_removed')) + + +class WorkflowGetApprovalRequestsCommand(_WorkflowCommand): + def __init__(self): + parser = argparse.ArgumentParser( + prog='pam workflow pending', + description='Get pending approval requests', + parents=[base.json_output_parser], + ) + super().__init__(parser) + + def execute_workflow(self, context: KeeperParams, **kwargs): + result = _run_sdk(get_pending_approvals, context.vault, enterprise_data=_get_enterprise_data(context)) + requests = result.get('requests') or [] + if _is_json(kwargs): + return _emit_json(result) + if not requests: + logger.warning('No pending approval requests') + return + rows = [] + for req in requests: + rows.append([ + req.get('record_name') or '', + req.get('record_uid') or '', + req.get('flow_uid') or '', + req.get('requested_by') or '', + req.get('reason') or '', + req.get('ticket') or '', + _fmt_ts(req.get('started_on')), + _fmt_ts(req.get('expires_on')), + req.get('duration') or '', + ]) + headers = ['Record Name', 'Record UID', 'Flow UID', 'Requested By', 'Reason', + 'Ticket', 'Started', 'Expires', 'Duration'] + report_utils.dump_report_data(rows, headers=headers) + + +class WorkflowApproveCommand(DashUidArgsMixin, _WorkflowCommand): + def __init__(self): + parser = argparse.ArgumentParser( + prog='pam workflow approve', + description='Approve a workflow access request', + parents=[base.json_output_parser], + ) + self.add_arguments_to_parser(parser) + super().__init__(parser) + + @staticmethod + def add_arguments_to_parser(parser: argparse.ArgumentParser): + parser.add_argument('flow_uid', help='Flow UID of the workflow to approve') + + def execute_workflow(self, context: KeeperParams, **kwargs): + result = _run_sdk(approve_workflow, context.vault, kwargs.get('flow_uid')) + if _is_json(kwargs): + return _emit_json(result) + logger.info('Access request approved') + logger.info('Flow UID: %s', result.get('flow_uid')) + + +class WorkflowDenyCommand(DashUidArgsMixin, _WorkflowCommand): + def __init__(self): + parser = argparse.ArgumentParser( + prog='pam workflow deny', + description='Deny a workflow access request', + parents=[base.json_output_parser], + ) + self.add_arguments_to_parser(parser) + super().__init__(parser) + + @staticmethod + def add_arguments_to_parser(parser: argparse.ArgumentParser): + parser.add_argument('flow_uid', help='Flow UID of the workflow to deny') + parser.add_argument('-r', '--reason', help='Reason for denial') + + def execute_workflow(self, context: KeeperParams, **kwargs): + result = _run_sdk( + deny_workflow, context.vault, kwargs.get('flow_uid'), reason=kwargs.get('reason'), + enterprise_data=_get_enterprise_data(context), + ) + if _is_json(kwargs): + return _emit_json(result) + logger.info('Access request denied') + logger.info('Flow UID: %s', result.get('flow_uid')) + if result.get('reason'): + logger.info('Reason: %s', result['reason']) + + +class WorkflowRequestAccessCommand(_WorkflowCommand): + def __init__(self): + parser = argparse.ArgumentParser( + prog='pam workflow request', + description='Request access to a PAM resource, escalate, or cancel a pending request', + parents=[base.json_output_parser], + ) + self.add_arguments_to_parser(parser) + super().__init__(parser) + + @staticmethod + def add_arguments_to_parser(parser: argparse.ArgumentParser): + parser.add_argument('record', help='Record UID or name') + parser.add_argument('-r', '--reason', help='Reason for access request') + parser.add_argument('-t', '--ticket', help='External ticket/reference number') + parser.add_argument('-e', '--escalate', action='store_true', + help='Escalate a pending request to escalation approvers') + parser.add_argument('-c', '--cancel', action='store_true', + help='Cancel a pending or active workflow request') + + def execute_workflow(self, context: KeeperParams, **kwargs): + result = _run_sdk( + request_workflow_access, + context.vault, + kwargs.get('record'), + reason=kwargs.get('reason') or '', + ticket=kwargs.get('ticket') or '', + escalate=bool(kwargs.get('escalate')), + cancel=bool(kwargs.get('cancel')), + ) + if _is_json(kwargs): + return _emit_json(result) + if result.get('status') == 'exempt': + logger.warning('You are exempt from workflow restrictions on this record.') + logger.info('As a record owner or approver, you can access this resource directly.') + return + action = result.get('action') + if action == 'escalated': + logger.info('Request escalated') + logger.info('Record: %s (%s)', result.get('record_name'), result.get('record_uid')) + logger.info('Escalation approvers have been notified.') + elif action == 'cancelled': + logger.info('Workflow request cancelled') + logger.info('Record: %s (%s)', result.get('record_name'), result.get('record_uid')) + logger.info('Flow UID: %s', result.get('flow_uid')) + else: + logger.info('Access request sent') + logger.info('Record: %s (%s)', result.get('record_name'), result.get('record_uid')) + if result.get('reason'): + logger.info('Reason: %s', result['reason']) + if result.get('ticket'): + logger.info('Ticket: %s', result['ticket']) + logger.info('Approvers have been notified.') + + +class WorkflowStartCommand(DashUidArgsMixin, _WorkflowCommand): + def __init__(self): + parser = argparse.ArgumentParser( + prog='pam workflow start', + description='Start a workflow (check-out). Can use either record UID/name or flow UID.', + parents=[base.json_output_parser], + ) + self.add_arguments_to_parser(parser) + super().__init__(parser) + + @staticmethod + def add_arguments_to_parser(parser: argparse.ArgumentParser): + parser.add_argument('uid', help='Record UID, record name, or Flow UID') + + def execute_workflow(self, context: KeeperParams, **kwargs): + result = _run_sdk(start_workflow, context.vault, kwargs.get('uid')) + if _is_json(kwargs): + return _emit_json(result) + logger.info('Workflow started (checked out)') + if result.get('record_uid'): + logger.info('Record: %s (%s)', result.get('record_name'), result.get('record_uid')) + else: + logger.info('Flow UID: %s', result.get('flow_uid')) + + +class WorkflowEndCommand(DashUidArgsMixin, _WorkflowCommand): + def __init__(self): + parser = argparse.ArgumentParser( + prog='pam workflow end', + description='End a workflow (check-in).', + parents=[base.json_output_parser], + ) + self.add_arguments_to_parser(parser) + super().__init__(parser) + + @staticmethod + def add_arguments_to_parser(parser: argparse.ArgumentParser): + parser.add_argument('uid', help='Record UID, record name, or Flow UID') + parser.add_argument('-f', '--force', action='store_true', + help="force check-in: approvers can terminate another user's active session " + 'when single-user checkout is enabled.') + + def execute_workflow(self, context: KeeperParams, **kwargs): + result = _run_sdk(end_workflow, context.vault, kwargs.get('uid'), force=bool(kwargs.get('force'))) + if _is_json(kwargs): + return _emit_json(result) + if result.get('action') == 'force_checkin': + logger.info('Record force checked in') + else: + logger.info('Workflow ended (checked in)') + if result.get('record_uid'): + logger.info('Record: %s (%s)', result.get('record_name'), result.get('record_uid')) + if result.get('flow_uid'): + logger.info('Flow UID: %s', result.get('flow_uid')) + if result.get('action') == 'ended': + logger.info('Credentials may have been rotated.') + + +class WorkflowGetStateCommand(_WorkflowCommand): + def __init__(self): + parser = argparse.ArgumentParser( + prog='pam workflow state', + description='Get workflow state for a record', + parents=[base.json_output_parser], + ) + self.add_arguments_to_parser(parser) + super().__init__(parser) + + @staticmethod + def add_arguments_to_parser(parser: argparse.ArgumentParser): + parser.add_argument('record', help='Record UID or name') + + def execute_workflow(self, context: KeeperParams, **kwargs): + result = _run_sdk( + get_workflow_state, context.vault, kwargs.get('record'), + enterprise_data=_get_enterprise_data(context), + ) + if _is_json(kwargs): + return _emit_json(result) + if result.get('status') == 'exempt': + logger.warning('You are exempt from workflow restrictions on this record.') + logger.info('As a record owner or approver, you can access this resource directly.') + return + if result.get('status') == 'no_workflow': + logger.warning('No workflow found for this record') + return + logger.info('Workflow State') + record_label = result.get('record_name') or '' + rec_uid = result.get('record_uid') or '' + if record_label and rec_uid: + logger.info('Record: %s (%s)', record_label, rec_uid) + elif rec_uid: + logger.info('Record: %s', rec_uid) + if result.get('flow_uid'): + logger.info('Flow UID: %s', result['flow_uid']) + if result.get('stage'): + logger.info('Stage: %s', result['stage']) + conditions = result.get('conditions') or [] + if conditions: + logger.info('Conditions: %s', ', '.join(conditions)) + if result.get('checked_out_by'): + logger.info('Checked out by: %s', result['checked_out_by']) + if result.get('can_force_checkin'): + logger.info('Force check-in: Available') + if result.get('escalated'): + logger.info('Escalated: Yes') + if result.get('started_on'): + logger.info('Started: %s', _fmt_ts(result['started_on'])) + if result.get('expires_on'): + logger.info('Expires: %s', _fmt_ts(result['expires_on'])) + approved_by = result.get('approved_by') or [] + if approved_by: + logger.info('Approved by:') + for a in approved_by: + suffix = f" at {_fmt_ts(a.get('approved_on'))}" if a.get('approved_on') else '' + logger.info(' - %s%s', a.get('user'), suffix) + + +class WorkflowGetUserAccessStateCommand(_WorkflowCommand): + def __init__(self): + parser = argparse.ArgumentParser( + prog='pam workflow my-access', + description='Get all workflow states for current user', + parents=[base.json_output_parser], + ) + super().__init__(parser) + + def execute_workflow(self, context: KeeperParams, **kwargs): + result = _run_sdk(get_user_access_state, context.vault, enterprise_data=_get_enterprise_data(context)) + workflows = result.get('workflows') or [] + if _is_json(kwargs): + return _emit_json(result) + if not workflows: + logger.warning('No active workflows') + return + rows = [] + for wf in workflows: + approved_by = '' + if wf.get('approved_by'): + approved_by = '\n'.join(a.get('user') or '' for a in wf['approved_by']) + rows.append([ + wf.get('stage') or '', + wf.get('record_name') or '', + wf.get('record_uid') or '', + wf.get('flow_uid') or '', + wf.get('checked_out_by') or '', + approved_by, + _fmt_ts(wf.get('started_on')), + _fmt_ts(wf.get('expires_on')), + ]) + headers = ['Stage', 'Record Name', 'Record UID', 'Flow UID', 'Checked Out By', + 'Approved By', 'Started', 'Expires'] + report_utils.dump_report_data(rows, headers=headers) diff --git a/keepercli-package/src/keepercli/register_commands.py b/keepercli-package/src/keepercli/register_commands.py index 20138886..d41a9152 100644 --- a/keepercli-package/src/keepercli/register_commands.py +++ b/keepercli-package/src/keepercli/register_commands.py @@ -36,6 +36,7 @@ def register_commands(commands: base.CliCommands, scopes: Optional[base.CommandS record_type, secrets_manager, shares, password_report, trash, record_file_report, record_handling_commands, register, password_generate, verify_records, shared_records_report, share_report, nsf_commands) + from .commands.pam.pam_workflow import PAMWorkflowCommand commands.register_command('sync-down', vault.SyncDownCommand(), base.CommandScope.Vault, 'd') commands.register_command('cd', vault_folder.FolderCdCommand(), base.CommandScope.Vault) @@ -104,6 +105,7 @@ def register_commands(commands: base.CliCommands, scopes: Optional[base.CommandS commands.register_command('nsf-record-permission', nsf_commands.NsfRecordPermissionCommand(), base.CommandScope.Vault) commands.register_command('nsf-transfer-record', nsf_commands.NsfTransferRecordCommand(), base.CommandScope.Vault) commands.register_command('nsf-shortcut', nsf_commands.NsfShortcutCommand(), base.CommandScope.Vault) + commands.register_command('workflow', PAMWorkflowCommand(), base.CommandScope.Vault) if not scopes or bool(scopes & base.CommandScope.Enterprise): diff --git a/keepersdk-package/requirements.txt b/keepersdk-package/requirements.txt index 54ae42e8..c6bd5fd5 100644 --- a/keepersdk-package/requirements.txt +++ b/keepersdk-package/requirements.txt @@ -7,3 +7,4 @@ fido2>=2.0.0; python_version>='3.10' email-validator>=2.0.0 pydantic>=2.6.4; python_version>='3.8' google-api-core>=2.16.0 +tzlocal>=5.2.0 diff --git a/keepersdk-package/setup.cfg b/keepersdk-package/setup.cfg index 5b313776..b4e65ad4 100644 --- a/keepersdk-package/setup.cfg +++ b/keepersdk-package/setup.cfg @@ -38,7 +38,7 @@ install_requires = email-validator>=2.3.0 pydantic>=2.12.5 google-api-core>=2.25.2 - + tzlocal>=5.2.0 [options.package_data] keepersdk = diff --git a/keepersdk-package/src/keepersdk/helpers/workflow/__init__.py b/keepersdk-package/src/keepersdk/helpers/workflow/__init__.py new file mode 100644 index 00000000..9f011924 --- /dev/null +++ b/keepersdk-package/src/keepersdk/helpers/workflow/__init__.py @@ -0,0 +1,56 @@ +# _ __ +# | |/ /___ ___ _ __ ___ _ _ ® +# | ' str: + msg = str(error) + msg = _RESPONSE_CODE_RE.sub('', msg) + msg = _PROTO_DUMP_RE.sub('', msg) + msg = re.sub(r'\s+', ' ', msg).strip() + return msg or 'Unknown error' + + +def refresh_enforcements(vault: vault_online.VaultOnline, *, force: bool = False) -> None: + """Reload account summary so role enforcements take effect without re-login. + + Debounced with a short TTL: callers such as the CLI's admin-verb gate and + ensure_can_configure_workflow_settings() both request a refresh for the same + command, and without debouncing that means two account-summary round trips + per admin command. Pass force=True to bypass the debounce. + """ + ctx = vault.keeper_auth.auth_context + now = time.monotonic() + last_refresh = getattr(ctx, '_workflow_enforcement_refreshed_at', None) + if not force and last_refresh is not None and (now - last_refresh) < _ENFORCEMENT_REFRESH_TTL_SEC: + return + from google.protobuf.json_format import MessageToDict + try: + rs = keeper_auth.load_account_summary(vault.keeper_auth) + enf = MessageToDict(rs.Enforcements) + bools = {x['key']: x.get('value', False) for x in enf.get('booleans', []) if 'key' in x} + ctx.enforcements.update(bools) + if WORKFLOW_SETTINGS_ENFORCEMENT_KEY not in bools: + ctx.enforcements[WORKFLOW_SETTINGS_ENFORCEMENT_KEY] = False + ctx._workflow_enforcement_refreshed_at = now + except Exception as e: + logger.error('Failed to refresh enforcements: %s', e, exc_info=True) + + +def can_configure_workflow_settings( + vault: vault_online.VaultOnline, *, refresh: bool = False) -> bool: + """True when the user currently has allow_configure_workflow_settings.""" + if refresh: + refresh_enforcements(vault) + return bool(vault.keeper_auth.auth_context.enforcements.get(WORKFLOW_SETTINGS_ENFORCEMENT_KEY)) + + +def ensure_can_configure_workflow_settings( + vault: vault_online.VaultOnline, *, refresh: bool = True, action: str = 'manage') -> None: + if can_configure_workflow_settings(vault, refresh=refresh): + return + raise WorkflowError( + f'You do not have permission to manage workflow settings. ' + f'The "{action}" command requires the "Can manage workflow settings" ' + f'enforcement policy. Contact your Keeper administrator to enable this ' + f'for your role.' + ) + + +def get_record_key(vault: vault_online.VaultOnline, record_uid: str) -> Optional[bytes]: + key = vault.vault_data.get_record_key(record_uid) + if key: + return key + if vault.nsf_data: + entry = vault.nsf_data.get_record(record_uid) + if entry: + return entry.record_key + return None + + +def is_record_owner(vault: vault_online.VaultOnline, record_uid: str) -> bool: + info = vault.vault_data.get_record(record_uid) + if info and (info.flags & vault_record.RecordFlags.IsOwner): + return True + if vault.nsf and vault.nsf_data and vault.nsf_data.get_record(record_uid): + account_uid = utils.base64_url_encode(vault.keeper_auth.auth_context.account_uid) + for ra in vault.nsf.record_accesses.get_links_by_subject(record_uid): + if ra.owner and ra.access_type_uid == account_uid: + return True + return False + + +def is_on_approver_list(vault: vault_online.VaultOnline, config) -> bool: + if not config or not config.approvers: + return False + current_user = (vault.keeper_auth.auth_context.username or '').lower() + team_uids = {t.team_uid for t in vault.vault_data.teams()} + for approver in config.approvers: + if approver.user and approver.user.lower() == current_user: + return True + if approver.teamUid: + team_uid_b64 = utils.base64_url_encode(approver.teamUid) + if team_uid_b64 in team_uids: + return True + return False + + +def is_workflow_exempt(vault: vault_online.VaultOnline, record_uid: str, config=None) -> bool: + """Exempt = record owner OR on approver list. Transport failures fail closed.""" + if is_record_owner(vault, record_uid): + return True + if config is None: + try: + ref = ProtobufRefBuilder.record_ref(utils.base64_url_decode(record_uid), '') + config = post_to_router( + vault, 'read_workflow_config', request=ref, response_type=workflow_pb2.WorkflowConfig) + except Exception as e: + logger.debug('is_workflow_exempt config read failed for %s: %s', record_uid, e) + return False + return is_on_approver_list(vault, config) + + +def _load_nsf_typed_record( + vault: vault_online.VaultOnline, record_uid: str) -> Optional[vault_record.TypedRecord]: + if not vault.nsf_data or not vault.nsf_data.get_record(record_uid): + return None + try: + meta = nsf_management.load_nsf_record_metadata(vault, record_uid) + except nsf_management.NsfError: + return None + typed = vault_record.TypedRecord() + typed.record_uid = record_uid + typed.load_record_data({ + 'type': meta.get('type') or '', + 'title': meta.get('title') or record_uid, + 'notes': meta.get('notes') or '', + 'fields': meta.get('fields') or [], + 'custom': meta.get('custom') or [], + }) + entry = vault.nsf_data.get_record(record_uid) + if entry and entry.record_key: + typed.record_key = entry.record_key + return typed + + +def _validate_iana_timezone(tz: str) -> str: + if tz not in available_timezones(): + raise WorkflowError( + f'Invalid IANA timezone: "{tz}". ' + 'Set TZ to a valid value (e.g., TZ=America/New_York).' + ) + # Confirm ZoneInfo accepts it (available_timezones can lag on some platforms). + ZoneInfo(tz) + return tz + + +def _record_uids_by_title(vault: vault_online.VaultOnline, title: str) -> List[str]: + """Return record UIDs whose title matches exactly (case-insensitive). + + Builds a title index lazily and rebuilds when the vault record count changes. + """ + title_cf = title.casefold() + record_count = len(vault.vault_data) + cached = getattr(vault, '_workflow_title_lookup', None) + if cached is None or cached[0] != record_count: + index: dict[str, List[str]] = {} + for info in vault.vault_data.records(): + index.setdefault(info.title.casefold(), []).append(info.record_uid) + cached = (record_count, index) + vault._workflow_title_lookup = cached + return list(cached[1].get(title_cf, [])) + + +def load_typed_record( + vault: vault_online.VaultOnline, identifier: str) -> Optional[vault_record.TypedRecord]: + """Load a TypedRecord from classic vault or NSF by UID or exact title.""" + if not identifier: + return None + loaded = vault.vault_data.load_record(identifier) + if isinstance(loaded, vault_record.TypedRecord): + key = vault.vault_data.get_record_key(identifier) + if key: + loaded.record_key = key + return loaded + + matching_uids = _record_uids_by_title(vault, identifier) + matches: List[vault_record.TypedRecord] = [] + for record_uid in matching_uids: + rec = vault.vault_data.load_record(record_uid) + if isinstance(rec, vault_record.TypedRecord): + key = vault.vault_data.get_record_key(record_uid) + if key: + rec.record_key = key + matches.append(rec) + if len(matches) == 1: + return matches[0] + if len(matches) > 1: + raise WorkflowError(f'Multiple records match title "{identifier}"') + + if vault.nsf_data: + try: + nsf_uid = nsf_management.resolve_nsf_record_uid(vault, identifier) + except nsf_management.NsfError: + nsf_uid = None + if nsf_uid: + return _load_nsf_typed_record(vault, nsf_uid) + return None + + +class RecordResolver: + @staticmethod + def resolve( + vault: vault_online.VaultOnline, + record_input: Optional[str], + allow_missing: bool = False) -> Tuple[Optional[str], Optional[vault_record.TypedRecord]]: + if not record_input: + if allow_missing: + return None, None + raise WorkflowError('Record is required') + rec = load_typed_record(vault, record_input) + if rec: + return rec.record_uid, rec + if allow_missing: + return None, None + raise WorkflowError(f'Record "{record_input}" not found') + + @staticmethod + def validate_workflow_record_type(record: vault_record.KeeperRecord) -> None: + if not isinstance(record, vault_record.TypedRecord): + raise WorkflowError('Workflows are only supported on PAM records') + record_type = record.record_type or 'unknown' + if record_type not in WORKFLOW_RECORD_TYPES: + supported = ', '.join(sorted(WORKFLOW_RECORD_TYPES)) + raise WorkflowError( + f'Record "{record.title}" is of type "{record_type}" which does not support workflows. ' + f'Supported record types: {supported}' + ) + + @staticmethod + def resolve_name(vault: vault_online.VaultOnline, resource_ref) -> str: + if resource_ref.name: + return resource_ref.name + if resource_ref.value: + rec_uid = utils.base64_url_encode(resource_ref.value) + rec = load_typed_record(vault, rec_uid) + if rec: + return rec.title + info = vault.vault_data.get_record(rec_uid) + if info: + return info.title + return '' + + @staticmethod + def format_label(vault: vault_online.VaultOnline, resource_ref) -> str: + rec_uid = utils.base64_url_encode(resource_ref.value) if resource_ref.value else '' + rec_name = RecordResolver.resolve_name(vault, resource_ref) + if rec_name and rec_name != rec_uid: + return f'{rec_name} ({rec_uid})' + return rec_uid or 'Unknown' + + @staticmethod + def resolve_user( + vault: vault_online.VaultOnline, + user_id: int, + enterprise_data: Optional[enterprise_types.IEnterpriseData] = None) -> str: + """Resolve an enterprise numeric user ID (workflow proto `userId`) to an email. + + `vault.vault_data` only carries `account_uid` (a base64 UID), which is a different + identifier than the enterprise numeric user ID, so it cannot be used for this lookup. + Resolving requires enterprise directory access; pass `enterprise_data` (e.g. from + `EnterpriseLoader.enterprise_data`) when available, otherwise a placeholder is returned. + """ + if enterprise_data is not None: + try: + user = enterprise_data.users().get_entity(user_id) + if user and user.username: + return user.username + except Exception as e: + logger.debug('Failed to resolve enterprise user %s: %s', user_id, e) + return f'User ID {user_id}' + + @staticmethod + def resolve_team_name( + vault: vault_online.VaultOnline, + team_uid: str, + enterprise_data: Optional[enterprise_types.IEnterpriseData] = None) -> str: + team = vault.vault_data.get_team(team_uid) + if team: + return team.name + if enterprise_data is not None: + try: + ent_team = enterprise_data.teams().get_entity(team_uid) + if ent_team: + return ent_team.name + except Exception as e: + logger.debug('Failed to resolve enterprise team %s: %s', team_uid, e) + return '' + + @staticmethod + def validate_team( + vault: vault_online.VaultOnline, + team_input: str, + enterprise_data: Optional[enterprise_types.IEnterpriseData] = None) -> str: + """Resolve a team UID or name to a team UID. + + Checks the current user's team-membership cache first (cheap, always available), then + falls back to the full enterprise team directory when `enterprise_data` is supplied — + needed because a workflow admin managing approvers may add a team they don't personally + belong to. + """ + if vault.vault_data.get_team(team_input): + return team_input + for team in vault.vault_data.teams(): + if team.name.casefold() == team_input.casefold(): + return team.team_uid + if enterprise_data is not None: + try: + if enterprise_data.teams().get_entity(team_input): + return team_input + for ent_team in enterprise_data.teams().get_all_entities(): + if ent_team.name.casefold() == team_input.casefold(): + return ent_team.team_uid + except Exception as e: + logger.debug('Enterprise team lookup failed for "%s": %s', team_input, e) + raise WorkflowError(f'Team "{team_input}" not found. Use a valid team UID or team name.') + + +class ProtobufRefBuilder: + @staticmethod + def record_ref(record_uid_bytes: bytes, record_name: str = '') -> GraphSync_pb2.GraphSyncRef: + ref = GraphSync_pb2.GraphSyncRef() + ref.type = GraphSync_pb2.RFT_REC + ref.value = record_uid_bytes + if record_name: + ref.name = record_name + return ref + + @staticmethod + def workflow_ref(flow_uid_bytes: bytes) -> GraphSync_pb2.GraphSyncRef: + ref = GraphSync_pb2.GraphSyncRef() + ref.type = GraphSync_pb2.RFT_WORKFLOW + ref.value = flow_uid_bytes + return ref + + +class WorkflowFormatter: + STAGE_MAP = { + workflow_pb2.WS_READY_TO_START: 'Ready to Start', + workflow_pb2.WS_STARTED: 'Started', + workflow_pb2.WS_NEEDS_ACTION: 'Needs Action', + workflow_pb2.WS_WAITING: 'Waiting', + } + + CONDITION_MAP = { + workflow_pb2.AC_APPROVAL: 'Approval Required', + workflow_pb2.AC_CHECKIN: 'Check-in Required', + workflow_pb2.AC_MFA: 'MFA Required', + workflow_pb2.AC_TIME: 'Time Restriction', + workflow_pb2.AC_REASON: 'Reason Required', + workflow_pb2.AC_TICKET: 'Ticket Required', + } + + DURATION_MULTIPLIERS = {'d': 86_400_000, 'h': 3_600_000, 'm': 60_000} + + DAY_PARSE_MAP = { + 'mon': workflow_pb2.MONDAY, 'monday': workflow_pb2.MONDAY, + 'tue': workflow_pb2.TUESDAY, 'tuesday': workflow_pb2.TUESDAY, + 'wed': workflow_pb2.WEDNESDAY, 'wednesday': workflow_pb2.WEDNESDAY, + 'thu': workflow_pb2.THURSDAY, 'thursday': workflow_pb2.THURSDAY, + 'fri': workflow_pb2.FRIDAY, 'friday': workflow_pb2.FRIDAY, + 'sat': workflow_pb2.SATURDAY, 'saturday': workflow_pb2.SATURDAY, + 'sun': workflow_pb2.SUNDAY, 'sunday': workflow_pb2.SUNDAY, + } + + DAY_NAME_MAP = { + workflow_pb2.MONDAY: 'Monday', + workflow_pb2.TUESDAY: 'Tuesday', + workflow_pb2.WEDNESDAY: 'Wednesday', + workflow_pb2.THURSDAY: 'Thursday', + workflow_pb2.FRIDAY: 'Friday', + workflow_pb2.SATURDAY: 'Saturday', + workflow_pb2.SUNDAY: 'Sunday', + } + + BLOCKING_CONDITIONS = {workflow_pb2.AC_TIME, workflow_pb2.AC_APPROVAL} + + @staticmethod + def format_stage(stage: int, status=None) -> str: + if stage == workflow_pb2.WS_READY_TO_START and status is not None: + if status.conditions: + has_blocking = any(c in WorkflowFormatter.BLOCKING_CONDITIONS for c in status.conditions) + if has_blocking: + return 'Waiting' + return 'Ready to Start' + if status.approvedBy and not status.startedOn: + return 'Ready to Start' + if not status.startedOn and not status.approvedBy: + return 'Needs Action' + return WorkflowFormatter.STAGE_MAP.get(stage, f'Unknown ({stage})') + + @staticmethod + def format_conditions(conditions: Sequence[int]) -> str: + return ', '.join( + WorkflowFormatter.CONDITION_MAP.get(c, f'Unknown ({c})') + for c in conditions + ) + + @staticmethod + def parse_duration(duration_str: str) -> int: + duration_str = duration_str.lower().strip() + try: + for suffix, factor in WorkflowFormatter.DURATION_MULTIPLIERS.items(): + if duration_str.endswith(suffix): + value = int(duration_str[:-1]) + if value <= 0: + raise ValueError + return value * factor + value = int(duration_str) + if value <= 0: + raise ValueError + return value * 60_000 + except ValueError: + raise WorkflowError( + f'Invalid duration format: {duration_str}. ' + 'Use a positive value like "2h", "30m", or "1d"' + ) + + @staticmethod + def format_duration(milliseconds: int) -> str: + seconds = milliseconds // 1000 + minutes = seconds // 60 + hours = minutes // 60 + days = hours // 24 + if days > 0: + return f"{days} day{'s' if days != 1 else ''}" + if hours > 0: + return f"{hours} hour{'s' if hours != 1 else ''}" + if minutes > 0: + return f"{minutes} minute{'s' if minutes != 1 else ''}" + return f"{seconds} second{'s' if seconds != 1 else ''}" + + @staticmethod + def build_temporal_filter(allowed_days_str, time_range_str): + if not allowed_days_str and not time_range_str: + return None + temporal = workflow_pb2.TemporalAccessFilter() + if allowed_days_str: + for day_token in allowed_days_str.split(','): + day_token = day_token.strip().lower() + day_enum = WorkflowFormatter.DAY_PARSE_MAP.get(day_token) + if day_enum is None: + valid = ', '.join(sorted({k for k in WorkflowFormatter.DAY_PARSE_MAP if len(k) == 3})) + raise WorkflowError(f'Invalid day: "{day_token}". Valid: {valid}') + temporal.allowedDays.append(day_enum) + if time_range_str: + if '-' not in time_range_str: + raise WorkflowError('Time range must be in HH:MM-HH:MM format (e.g., "09:00-17:00")') + start_str, end_str = time_range_str.split('-', 1) + start_hhmm = WorkflowFormatter._parse_time_to_hhmm(start_str.strip()) + end_hhmm = WorkflowFormatter._parse_time_to_hhmm(end_str.strip()) + if start_hhmm >= end_hhmm: + raise WorkflowError( + f'Time range start must be before end (got "{start_str.strip()}-{end_str.strip()}"). ' + 'Use HH:MM-HH:MM with start earlier than end (e.g., "09:00-17:00").' + ) + time_range = workflow_pb2.TimeOfDayRange() + time_range.startTime = start_hhmm + time_range.endTime = end_hhmm + temporal.timeRanges.append(time_range) + temporal.timeZone = WorkflowFormatter._get_local_iana_timezone() + return temporal + + @staticmethod + def _get_local_iana_timezone() -> str: + """Return the local IANA timezone name for temporal access filters. + + Resolution order: + 1. ``TZ`` environment variable (must contain ``/``, e.g. ``America/New_York``). + 2. ``tzlocal`` (declared dependency in ``requirements.txt`` / ``setup.cfg``). + 3. ``datetime.now().astimezone().tzinfo.key`` when present (often unavailable on Windows). + + Raises ``WorkflowError`` if none succeed. On Windows, or when ``tzlocal`` is missing + or broken, set ``TZ`` explicitly (e.g. ``TZ=Asia/Kolkata``) before creating or + updating workflows with ``--allowed-days`` or ``--time-range``. + """ + tz = os.environ.get('TZ') + if tz and '/' in tz: + return _validate_iana_timezone(tz) + try: + from tzlocal import get_localzone_name + zone = get_localzone_name() + if zone: + return _validate_iana_timezone(zone) + except Exception as e: + logger.debug('tzlocal lookup failed: %s', e) + now = datetime.now().astimezone() + key = getattr(now.tzinfo, 'key', None) + if isinstance(key, str) and '/' in key: + return _validate_iana_timezone(key) + raise WorkflowError( + 'Could not detect local IANA timezone. ' + 'Set the TZ environment variable (e.g., TZ=Asia/Kolkata).' + ) + + @staticmethod + def _parse_time_to_hhmm(time_str: str) -> int: + try: + parts = time_str.split(':') + h = int(parts[0]) + m = int(parts[1]) if len(parts) > 1 else 0 + if not (0 <= h <= 23 and 0 <= m <= 59): + raise ValueError + return h * 100 + m + except (ValueError, IndexError): + raise WorkflowError(f'Invalid time format: "{time_str}". Use HH:MM (e.g., "09:00")') + + @staticmethod + def format_temporal_filter(at) -> Optional[dict]: + if not at: + return None + result = {} + if at.allowedDays: + result['allowed_days'] = [WorkflowFormatter.DAY_NAME_MAP.get(d, str(d)) for d in at.allowedDays] + if at.timeRanges: + ranges = [] + for tr in at.timeRanges: + sh, sm = divmod(tr.startTime, 100) + eh, em = divmod(tr.endTime, 100) + ranges.append(f'{sh:02d}:{sm:02d}-{eh:02d}:{em:02d}') + result['time_ranges'] = ranges + if at.timeZone: + result['timezone'] = at.timeZone + return result or None + + +def dedupe_approver_items(items: Optional[Iterable[str]], *, label: str = 'approver') -> List[str]: + """Strip, drop blanks, and de-duplicate approver entries preserving order.""" + raw = [x.strip() for x in (items or []) if x and x.strip()] + deduped = list(dict.fromkeys(raw)) + removed = len(raw) - len(deduped) + if removed: + logger.info('Removed %s duplicate %s(s)', removed, label) + return deduped + + +def submit_access_request( + vault: vault_online.VaultOnline, + record_uid: str, + record_name: str = '', + reason: str = '', + ticket: str = '') -> None: + """Send a workflow access request. Reason/ticket are encrypted with the record key.""" + record_uid_bytes = utils.base64_url_decode(record_uid) + record_key = None + if reason or ticket: + record_key = get_record_key(vault, record_uid) + if not record_key: + raise WorkflowError( + 'Record key not available — cannot encrypt reason/ticket. ' + 'You do not have sufficient access to this record to send encrypted parameters.' + ) + access_request = workflow_pb2.WorkflowAccessRequest() + access_request.resource.CopyFrom(ProtobufRefBuilder.record_ref(record_uid_bytes, record_name)) + if reason: + reason_bytes = reason.encode('utf-8') if isinstance(reason, str) else reason + access_request.reason = crypto.encrypt_aes_v2(reason_bytes, record_key) + if ticket: + ticket_bytes = ticket.encode('utf-8') if isinstance(ticket, str) else ticket + access_request.ticket = crypto.encrypt_aes_v2(ticket_bytes, record_key) + post_to_router(vault, 'request_workflow_access', request=access_request) + + +def start_workflow_for_record( + vault: vault_online.VaultOnline, record_uid: str, record_name: str = '') -> None: + record_uid_bytes = utils.base64_url_decode(record_uid) + state = workflow_pb2.WorkflowState() + state.resource.CopyFrom(ProtobufRefBuilder.record_ref(record_uid_bytes, record_name)) + post_to_router(vault, 'start_workflow', request=state) + + +def add_approvers_to_workflow( + vault: vault_online.VaultOnline, + record_uid: str, + record_name: str, + users: Optional[Iterable[str]] = None, + teams: Optional[Iterable[str]] = None, + is_escalation: bool = False, + escalation_after_ms: int = 0, + enterprise_data: Optional[enterprise_types.IEnterpriseData] = None) -> None: + record_uid_bytes = utils.base64_url_decode(record_uid) + config = workflow_pb2.WorkflowConfig() + config.parameters.resource.CopyFrom(ProtobufRefBuilder.record_ref(record_uid_bytes, record_name)) + for user_email in (users or []): + approver = workflow_pb2.WorkflowApprover() + approver.user = user_email + approver.escalation = is_escalation + if escalation_after_ms: + approver.escalationAfterMs = escalation_after_ms + config.approvers.append(approver) + for team_input in (teams or []): + resolved_team_uid = RecordResolver.validate_team(vault, team_input, enterprise_data) + approver = workflow_pb2.WorkflowApprover() + approver.teamUid = utils.base64_url_decode(resolved_team_uid) + approver.escalation = is_escalation + if escalation_after_ms: + approver.escalationAfterMs = escalation_after_ms + config.approvers.append(approver) + post_to_router(vault, 'add_workflow_approvers', request=config) + + +def workflow_state_to_dict( + vault: vault_online.VaultOnline, + wf, + enterprise_data: Optional[enterprise_types.IEnterpriseData] = None) -> dict: + st = wf.status + return { + 'flow_uid': utils.base64_url_encode(wf.flowUid) if wf.flowUid else None, + 'record_uid': utils.base64_url_encode(wf.resource.value) if wf.resource.value else '', + 'record_name': RecordResolver.resolve_name(vault, wf.resource), + 'stage': WorkflowFormatter.format_stage(st.stage, st) if st else None, + 'conditions': ( + [WorkflowFormatter.format_conditions([c]) for c in st.conditions] if st and st.conditions else [] + ), + 'escalated': bool(st.escalated) if st else False, + 'checked_out_by': (st.checkedOutBy or None) if st else None, + 'can_force_checkin': bool(st.canForceCheckIn) if st else False, + 'started_on': (st.startedOn or None) if st else None, + 'expires_on': (st.expiresOn or None) if st else None, + 'approved_by': [ + { + 'user': a.user if a.user else RecordResolver.resolve_user(vault, a.userId, enterprise_data), + 'approved_on': a.approvedOn or None, + } + for a in (st.approvedBy if st else []) + ], + } + + +def decrypt_workflow_param( + vault: vault_online.VaultOnline, record_uid: str, encrypted_bytes) -> Optional[str]: + if not encrypted_bytes: + return None + record_key = get_record_key(vault, record_uid) + if not record_key: + return 'No permission to view. Only users with record access can view this information.' + try: + return crypto.decrypt_aes_v2(encrypted_bytes, record_key).decode('utf-8') + except Exception: + logger.debug('Failed to decrypt workflow parameter for record %s', record_uid, exc_info=True) + return 'Unable to decrypt' + + +def extract_workflow_param(wf, key: str): + for p in wf.workflowParameters: + if p.key == key: + return p.data + return None diff --git a/keepersdk-package/src/keepersdk/helpers/workflow/workflow.py b/keepersdk-package/src/keepersdk/helpers/workflow/workflow.py new file mode 100644 index 00000000..f9b1f5f8 --- /dev/null +++ b/keepersdk-package/src/keepersdk/helpers/workflow/workflow.py @@ -0,0 +1,845 @@ +# _ __ +# | |/ /___ ___ _ __ ___ _ _ ® +# | ' dict: + """Create workflow configuration for a PAM record (`pam workflow create`).""" + ensure_can_configure_workflow_settings(vault, refresh=True, action='create') + record_uid, rec = RecordResolver.resolve(vault, record) + RecordResolver.validate_workflow_record_type(rec) + record_uid_bytes = utils.base64_url_decode(record_uid) + + existing = post_to_router( + vault, 'read_workflow_config', + request=ProtobufRefBuilder.record_ref(record_uid_bytes, rec.title), + response_type=workflow_pb2.WorkflowConfig, + ) + if existing: + raise WorkflowError( + f'Workflow already configured for "{rec.title}" ({record_uid}). ' + 'Use update_workflow to modify it, read_workflow to inspect it, ' + 'or delete_workflow then create_workflow to recreate it.' + ) + + if approvals_needed < 0: + raise WorkflowError('Approvals needed must be 0 or greater') + + approver_list = dedupe_approver_items(approvers, label='approver') + if approvals_needed > 0 and not approver_list: + raise WorkflowError( + 'At least one approver is required when approvals_needed > 0. ' + 'Pass approvers, or use approvals_needed=0 for a workflow that does not need approval.' + ) + if approver_list and approvals_needed == 0: + logger.warning( + 'approvers supplied but approvals_needed is 0 — approvers will ' + 'be recorded but no approval will ever be required.' + ) + + parameters = workflow_pb2.WorkflowParameters() + parameters.resource.CopyFrom(ProtobufRefBuilder.record_ref(record_uid_bytes, rec.title)) + parameters.approvalsNeeded = approvals_needed + parameters.checkoutNeeded = checkout + parameters.startAccessOnApproval = start_on_approval + parameters.requireReason = require_reason + parameters.requireTicket = require_ticket + parameters.requireMFA = require_mfa + parameters.accessLength = WorkflowFormatter.parse_duration(duration) + temporal_filter = WorkflowFormatter.build_temporal_filter(allowed_days, time_range) + if temporal_filter: + parameters.allowedTimes.CopyFrom(temporal_filter) + + try: + post_to_router(vault, 'create_workflow_config', request=parameters) + approvers_added: List[str] = [] + if approver_list: + try: + add_approvers_to_workflow(vault, record_uid, rec.title, users=approver_list) + approvers_added = list(approver_list) + except Exception as e: + logger.warning( + 'Workflow created, but failed to add approvers: %s. ' + 'Call add_workflow_approvers for record %s.', + sanitize_router_error(e), record_uid, + ) + + result = { + 'status': 'success', + 'record_uid': record_uid, + 'record_name': rec.title, + 'workflow_config': { + 'approvals_needed': parameters.approvalsNeeded, + 'checkout_needed': parameters.checkoutNeeded, + 'require_reason': parameters.requireReason, + 'require_ticket': parameters.requireTicket, + 'require_mfa': parameters.requireMFA, + 'access_duration': WorkflowFormatter.format_duration(parameters.accessLength), + }, + 'approvers': approvers_added, + } + logger.info('Workflow created for %s (%s)', rec.title, record_uid) + return result + except WorkflowError: + raise + except ROUTER_TRANSPORT_ERRORS as e: + raise WorkflowError(f'Failed to create workflow: {sanitize_router_error(e)}') from e + + +def read_workflow( + vault: vault_online.VaultOnline, + record: str, + enterprise_data: Optional[enterprise_types.IEnterpriseData] = None) -> dict: + """Read workflow configuration (`pam workflow read`).""" + record_uid, rec = RecordResolver.resolve(vault, record) + record_uid_bytes = utils.base64_url_decode(record_uid) + ref = ProtobufRefBuilder.record_ref(record_uid_bytes, rec.title) + try: + response = post_to_router( + vault, 'read_workflow_config', request=ref, response_type=workflow_pb2.WorkflowConfig) + if not response: + logger.warning('No workflow configured for record %s (%s)', rec.title, record_uid) + return { + 'status': 'no_workflow', + 'message': 'No workflow configured', + 'record_uid': record_uid, + 'record_name': rec.title, + } + p = response.parameters + result = { + 'status': 'success', + 'record_uid': record_uid, + 'record_name': RecordResolver.resolve_name(vault, p.resource) or rec.title, + 'created_on': response.createdOn or None, + 'parameters': { + 'approvals_needed': p.approvalsNeeded, + 'checkout_needed': p.checkoutNeeded, + 'start_access_on_approval': p.startAccessOnApproval, + 'require_reason': p.requireReason, + 'require_ticket': p.requireTicket, + 'require_mfa': p.requireMFA, + 'access_duration': WorkflowFormatter.format_duration(p.accessLength), + 'allowed_times': WorkflowFormatter.format_temporal_filter(p.allowedTimes), + }, + 'approvers': [], + } + for approver in response.approvers: + approver_info = {'escalation': approver.escalation} + if approver.escalationAfterMs: + approver_info['escalation_after'] = WorkflowFormatter.format_duration(approver.escalationAfterMs) + if approver.HasField('user'): + approver_info['type'] = 'user' + approver_info['email'] = approver.user + elif approver.HasField('userId'): + approver_info['type'] = 'user_id' + approver_info['user_id'] = approver.userId + approver_info['email'] = RecordResolver.resolve_user(vault, approver.userId, enterprise_data) + elif approver.HasField('teamUid'): + team_uid = utils.base64_url_encode(approver.teamUid) + approver_info['type'] = 'team' + approver_info['team_uid'] = team_uid + approver_info['team_name'] = RecordResolver.resolve_team_name(vault, team_uid, enterprise_data) + result['approvers'].append(approver_info) + if not result['approvers']: + logger.warning('No approvers configured for record %s', record_uid) + logger.info('Read workflow configuration for %s (%s)', rec.title, record_uid) + return result + except WorkflowError: + raise + except ROUTER_TRANSPORT_ERRORS as e: + raise WorkflowError(f'Failed to read workflow: {sanitize_router_error(e)}') from e + + +def update_workflow( + vault: vault_online.VaultOnline, + record: str, + *, + approvals_needed: Optional[int] = None, + checkout: Optional[bool] = None, + start_on_approval: Optional[bool] = None, + require_reason: Optional[bool] = None, + require_ticket: Optional[bool] = None, + require_mfa: Optional[bool] = None, + duration: Optional[str] = None, + allowed_days: Optional[str] = None, + time_range: Optional[str] = None) -> dict: + """Update existing workflow configuration (`pam workflow update`). Unspecified fields are kept.""" + ensure_can_configure_workflow_settings(vault, refresh=True, action='update') + record_uid, rec = RecordResolver.resolve(vault, record) + record_uid_bytes = utils.base64_url_decode(record_uid) + try: + ref = ProtobufRefBuilder.record_ref(record_uid_bytes, rec.title) + current_config = post_to_router( + vault, 'read_workflow_config', request=ref, response_type=workflow_pb2.WorkflowConfig) + if not current_config: + raise WorkflowError('No workflow found for record. Create one first with create_workflow.') + + parameters = workflow_pb2.WorkflowParameters() + parameters.CopyFrom(current_config.parameters) + updates = { + 'approvalsNeeded': approvals_needed, + 'checkoutNeeded': checkout, + 'startAccessOnApproval': start_on_approval, + 'requireReason': require_reason, + 'requireTicket': require_ticket, + 'requireMFA': require_mfa, + } + if approvals_needed is not None and approvals_needed < 0: + raise WorkflowError('Approvals needed must be 0 or greater') + + updates_provided = False + for proto_field, value in updates.items(): + if value is not None: + setattr(parameters, proto_field, value) + updates_provided = True + if duration is not None: + parameters.accessLength = WorkflowFormatter.parse_duration(duration) + updates_provided = True + temporal_filter = WorkflowFormatter.build_temporal_filter(allowed_days, time_range) + if temporal_filter: + parameters.allowedTimes.CopyFrom(temporal_filter) + updates_provided = True + if not updates_provided: + raise WorkflowError( + 'No updates provided. Specify at least one option to update ' + '(e.g., approvals_needed, duration).' + ) + post_to_router(vault, 'update_workflow_config', request=parameters) + logger.info('Workflow updated for %s (%s)', rec.title, record_uid) + return {'status': 'success', 'record_uid': record_uid, 'record_name': rec.title} + except WorkflowError: + raise + except ROUTER_TRANSPORT_ERRORS as e: + raise WorkflowError(f'Failed to update workflow: {sanitize_router_error(e)}') from e + + +def delete_workflow(vault: vault_online.VaultOnline, record: str) -> dict: + """Delete workflow configuration from a record (`pam workflow delete`).""" + ensure_can_configure_workflow_settings(vault, refresh=True, action='delete') + record_uid, rec = RecordResolver.resolve(vault, record) + record_uid_bytes = utils.base64_url_decode(record_uid) + ref = ProtobufRefBuilder.record_ref(record_uid_bytes, rec.title) + existing = post_to_router( + vault, 'read_workflow_config', request=ref, response_type=workflow_pb2.WorkflowConfig) + if not existing: + raise WorkflowError( + f'No workflow configured for "{rec.title}" ({record_uid}). Nothing to delete.' + ) + try: + post_to_router(vault, 'delete_workflow_config', request=ref) + logger.info('Workflow deleted for %s (%s)', rec.title, record_uid) + return {'status': 'success', 'record_uid': record_uid, 'record_name': rec.title} + except WorkflowError: + raise + except ROUTER_TRANSPORT_ERRORS as e: + raise WorkflowError(f'Failed to delete workflow: {sanitize_router_error(e)}') from e + + +def add_workflow_approvers( + vault: vault_online.VaultOnline, + record: str, + *, + users: Optional[Iterable[str]] = None, + teams: Optional[Iterable[str]] = None, + escalation: bool = False, + escalation_after: Optional[str] = None, + enterprise_data: Optional[enterprise_types.IEnterpriseData] = None) -> dict: + """Add approvers to a workflow (`pam workflow add-approver`).""" + ensure_can_configure_workflow_settings(vault, refresh=True, action='add-approver') + user_list = dedupe_approver_items(users, label='user approver') + team_list = dedupe_approver_items(teams, label='team approver') + if not user_list and not team_list: + raise WorkflowError('Must specify at least one user or team') + if escalation_after and not escalation: + raise WorkflowError('escalation_after requires escalation=True') + escalation_after_ms = WorkflowFormatter.parse_duration(escalation_after) if escalation_after else 0 + record_uid, rec = RecordResolver.resolve(vault, record) + try: + add_approvers_to_workflow( + vault, record_uid, rec.title, + users=user_list, teams=team_list, + is_escalation=escalation, escalation_after_ms=escalation_after_ms, + enterprise_data=enterprise_data, + ) + total = len(user_list) + len(team_list) + result = { + 'status': 'success', + 'record_uid': record_uid, + 'record_name': rec.title, + 'approvers_added': total, + 'escalation': escalation, + 'users': user_list, + 'teams': team_list, + } + if escalation_after_ms: + result['escalation_after'] = WorkflowFormatter.format_duration(escalation_after_ms) + logger.info('Added %s approver(s) to %s (%s)', total, rec.title, record_uid) + return result + except WorkflowError: + raise + except ROUTER_TRANSPORT_ERRORS as e: + raise WorkflowError(f'Failed to add approvers: {sanitize_router_error(e)}') from e + + +def remove_workflow_approvers( + vault: vault_online.VaultOnline, + record: str, + *, + users: Optional[Iterable[str]] = None, + teams: Optional[Iterable[str]] = None, + enterprise_data: Optional[enterprise_types.IEnterpriseData] = None) -> dict: + """Remove approvers from a workflow (`pam workflow remove-approver`).""" + ensure_can_configure_workflow_settings(vault, refresh=True, action='remove-approver') + user_list = dedupe_approver_items(users, label='user approver') + team_list = dedupe_approver_items(teams, label='team approver') + if not user_list and not team_list: + raise WorkflowError('Must specify at least one user or team') + record_uid, rec = RecordResolver.resolve(vault, record) + record_uid_bytes = utils.base64_url_decode(record_uid) + ref = ProtobufRefBuilder.record_ref(record_uid_bytes, rec.title) + existing = post_to_router( + vault, 'read_workflow_config', request=ref, response_type=workflow_pb2.WorkflowConfig) + if not existing: + raise WorkflowError('No workflow found for record. Create one first with create_workflow.') + config = workflow_pb2.WorkflowConfig() + config.parameters.resource.CopyFrom(ProtobufRefBuilder.record_ref(record_uid_bytes, rec.title)) + for user_email in user_list: + approver = workflow_pb2.WorkflowApprover() + approver.user = user_email + config.approvers.append(approver) + for team_input in team_list: + resolved_team_uid = RecordResolver.validate_team(vault, team_input, enterprise_data) + approver = workflow_pb2.WorkflowApprover() + approver.teamUid = utils.base64_url_decode(resolved_team_uid) + config.approvers.append(approver) + try: + post_to_router(vault, 'delete_workflow_approvers', request=config) + total = len(user_list) + len(team_list) + logger.info('Removed %s approver(s) from %s (%s)', total, rec.title, record_uid) + return { + 'status': 'success', + 'record_uid': record_uid, + 'record_name': rec.title, + 'approvers_removed': total, + } + except WorkflowError: + raise + except ROUTER_TRANSPORT_ERRORS as e: + raise WorkflowError(f'Failed to remove approvers: {sanitize_router_error(e)}') from e + + +def get_pending_approvals( + vault: vault_online.VaultOnline, + enterprise_data: Optional[enterprise_types.IEnterpriseData] = None) -> dict: + """Get pending approval requests (`pam workflow pending`).""" + try: + response = post_to_router( + vault, 'get_approval_requests', response_type=workflow_pb2.ApprovalRequests) + if not response or not response.workflows: + logger.info('No approval requests') + return {'status': 'success', 'requests': []} + + seen_flows = set() + unique = [] + for wf in response.workflows: + fuid = bytes(wf.flowUid) + if fuid not in seen_flows: + seen_flows.add(fuid) + unique.append(wf) + + current_user = vault.keeper_auth.auth_context.username + pending = _filter_already_approved(vault, unique, current_user) + if not pending: + logger.info('No pending approval requests') + return {'status': 'success', 'requests': []} + + requests = [] + for wf in pending: + rec_uid = utils.base64_url_encode(wf.resource.value) if wf.resource.value else '' + requested_by = wf.user or RecordResolver.resolve_user(vault, wf.userId, enterprise_data) + requests.append({ + 'flow_uid': utils.base64_url_encode(wf.flowUid), + 'requested_by': requested_by, + 'record_uid': rec_uid, + 'record_name': RecordResolver.resolve_name(vault, wf.resource), + 'started_on': wf.startedOn or None, + 'expires_on': wf.expiresOn or None, + 'escalated': wf.escalated, + 'duration': ( + WorkflowFormatter.format_duration(wf.expiresOn - wf.startedOn) + if wf.expiresOn and wf.startedOn else None + ), + 'reason': decrypt_workflow_param(vault, rec_uid, extract_workflow_param(wf, 'reason')), + 'ticket': decrypt_workflow_param(vault, rec_uid, extract_workflow_param(wf, 'ticket')), + }) + logger.info('Found %s pending approval request(s)', len(requests)) + return {'status': 'success', 'requests': requests} + except WorkflowError: + raise + except ROUTER_TRANSPORT_ERRORS as e: + raise WorkflowError(f'Failed to get approval requests: {sanitize_router_error(e)}') from e + + +def _already_approved_by_me(vault: vault_online.VaultOnline, wf, current_user: str) -> bool: + st = workflow_pb2.WorkflowState() + st.flowUid = wf.flowUid + ws = post_to_router( + vault, 'get_workflow_state', request=st, response_type=workflow_pb2.WorkflowState) + if ws and ws.status and ws.status.approvedBy: + for a in ws.status.approvedBy: + if a.user == current_user: + return True + return False + + +def _filter_already_approved( + vault: vault_online.VaultOnline, + workflows: List, + current_user: str) -> List: + """Drop requests the current user has already approved. + + Issues one ``get_workflow_state`` call per candidate. Parallelized when there + are multiple requests so large pending queues do not stall sequentially. + """ + if not workflows: + return [] + if len(workflows) == 1: + wf = workflows[0] + return [] if _already_approved_by_me(vault, wf, current_user) else [wf] + + approved: dict[int, bool] = {} + max_workers = min(10, len(workflows)) + with ThreadPoolExecutor(max_workers=max_workers) as pool: + futures = { + pool.submit(_already_approved_by_me, vault, wf, current_user): wf + for wf in workflows + } + for future, wf in futures.items(): + approved[id(wf)] = future.result() + return [wf for wf in workflows if not approved[id(wf)]] + + +def approve_workflow(vault: vault_online.VaultOnline, flow_uid: str) -> dict: + """Approve a workflow access request (`pam workflow approve`).""" + try: + flow_uid_bytes = utils.base64_url_decode(flow_uid) + except (ValueError, TypeError) as e: + raise WorkflowError(f'Invalid flow UID: "{flow_uid}"') from e + approval = workflow_pb2.WorkflowApprovalOrDenial() + approval.flowUid = flow_uid_bytes + approval.deny = False + try: + post_to_router(vault, 'approve_or_deny_workflow_access', request=approval) + logger.info('Access request approved for flow %s', flow_uid) + return {'status': 'success', 'flow_uid': flow_uid, 'action': 'approved'} + except WorkflowError: + raise + except ROUTER_TRANSPORT_ERRORS as e: + raise WorkflowError(f'Failed to approve request: {sanitize_router_error(e)}') from e + + +def deny_workflow( + vault: vault_online.VaultOnline, + flow_uid: str, + *, + reason: Optional[str] = None, + enterprise_data: Optional[enterprise_types.IEnterpriseData] = None) -> dict: + """Deny a workflow access request (`pam workflow deny`).""" + reason = reason or '' + try: + flow_uid_bytes = utils.base64_url_decode(flow_uid) + except (ValueError, TypeError) as e: + raise WorkflowError(f'Invalid flow UID: "{flow_uid}"') from e + denial = workflow_pb2.WorkflowApprovalOrDenial() + denial.flowUid = flow_uid_bytes + denial.deny = True + if reason: + encrypted = _encrypt_denial_reason(vault, flow_uid_bytes, reason.encode('utf-8'), enterprise_data) + if encrypted: + denial.denialReason = encrypted + else: + logger.warning( + 'Could not encrypt denial reason for the requester — reason will not be attached. ' + 'The denial itself will still be sent.' + ) + reason = '' + try: + post_to_router(vault, 'approve_or_deny_workflow_access', request=denial) + result = {'status': 'success', 'flow_uid': flow_uid, 'action': 'denied'} + if reason: + result['reason'] = reason + logger.info('Access request denied for flow %s', flow_uid) + return result + except WorkflowError: + raise + except ROUTER_TRANSPORT_ERRORS as e: + raise WorkflowError(f'Failed to deny request: {sanitize_router_error(e)}') from e + + +def _encrypt_denial_reason( + vault: vault_online.VaultOnline, + flow_uid_bytes: bytes, + reason_bytes: bytes, + enterprise_data: Optional[enterprise_types.IEnterpriseData] = None): + try: + response = post_to_router( + vault, 'get_approval_requests', response_type=workflow_pb2.ApprovalRequests) + if not response or not response.workflows: + return None + requester_email = None + for wf in response.workflows: + if wf.flowUid == flow_uid_bytes: + requester_email = wf.user or RecordResolver.resolve_user(vault, wf.userId, enterprise_data) + break + if not requester_email or requester_email.startswith('User ID '): + logger.debug('Could not resolve requester email for flow UID') + return None + vault.keeper_auth.load_user_public_keys([requester_email]) + public_keys = vault.keeper_auth.get_user_keys(requester_email) + if not public_keys: + logger.debug('Public key not available for %s', requester_email) + return None + if public_keys.ec: + return crypto.encrypt_ec(reason_bytes, crypto.load_ec_public_key(public_keys.ec)) + if public_keys.rsa: + return crypto.encrypt_rsa(reason_bytes, crypto.load_rsa_public_key(public_keys.rsa)) + except Exception: + logger.debug('Failed to encrypt denial reason with requester public key', exc_info=True) + return None + + +def request_workflow_access( + vault: vault_online.VaultOnline, + record: str, + *, + reason: str = '', + ticket: str = '', + escalate: bool = False, + cancel: bool = False) -> dict: + """Request, escalate, or cancel access (`pam workflow request`).""" + if cancel and escalate: + raise WorkflowError('cancel and escalate cannot be used together') + if cancel and (reason or ticket): + raise WorkflowError('cancel cannot be used with reason or ticket') + if cancel: + return _cancel_access_request(vault, record) + if escalate: + return _escalate_access_request(vault, record) + return _submit_access_request(vault, record, reason=reason, ticket=ticket) + + +def _submit_access_request( + vault: vault_online.VaultOnline, record: str, *, reason: str, ticket: str) -> dict: + record_uid, rec = RecordResolver.resolve(vault, record) + RecordResolver.validate_workflow_record_type(rec) + if is_workflow_exempt(vault, record_uid): + logger.warning( + 'You are exempt from workflow restrictions on this record. ' + 'As a record owner or approver, you can access this resource directly.' + ) + return {'status': 'exempt', 'message': 'Workflow not required', 'record_uid': record_uid} + try: + submit_access_request(vault, record_uid, rec.title, reason=reason, ticket=ticket) + result = { + 'status': 'success', + 'record_uid': record_uid, + 'record_name': rec.title, + 'message': 'Access request sent to approvers', + } + if reason: + result['reason'] = reason + if ticket: + result['ticket'] = ticket + logger.info('Access request sent for %s (%s)', rec.title, record_uid) + return result + except WorkflowError: + raise + except ROUTER_TRANSPORT_ERRORS as e: + raise WorkflowError(f'Failed to request access: {sanitize_router_error(e)}') from e + + +def _escalate_access_request(vault: vault_online.VaultOnline, record: str) -> dict: + record_uid, rec = RecordResolver.resolve(vault, record) + if is_workflow_exempt(vault, record_uid): + logger.warning( + 'You are exempt from workflow restrictions on this record. ' + 'As a record owner or approver, you can access this resource directly.' + ) + return {'status': 'exempt', 'message': 'Workflow not required', 'record_uid': record_uid} + record_uid_bytes = utils.base64_url_decode(record_uid) + state = workflow_pb2.WorkflowState() + state.resource.CopyFrom(ProtobufRefBuilder.record_ref(record_uid_bytes, rec.title)) + try: + post_to_router(vault, 'request_escalation', request=state) + logger.info('Request escalated for %s (%s)', rec.title, record_uid) + return { + 'status': 'success', + 'record_uid': record_uid, + 'record_name': rec.title, + 'action': 'escalated', + } + except WorkflowError: + raise + except ROUTER_TRANSPORT_ERRORS as e: + raise WorkflowError(f'Failed to escalate request: {sanitize_router_error(e)}') from e + + +def _cancel_access_request(vault: vault_online.VaultOnline, record: str) -> dict: + record_uid, rec = RecordResolver.resolve(vault, record) + record_uid_bytes = utils.base64_url_decode(record_uid) + try: + state_query = workflow_pb2.WorkflowState() + state_query.resource.CopyFrom( + ProtobufRefBuilder.record_ref(record_uid_bytes, rec.title if rec else '') + ) + workflow_state = post_to_router( + vault, 'get_workflow_state', + request=state_query, response_type=workflow_pb2.WorkflowState, + ) + if not workflow_state or not workflow_state.flowUid: + raise WorkflowError('No active workflow request found for this record.') + flow_ref = ProtobufRefBuilder.workflow_ref(workflow_state.flowUid) + post_to_router(vault, 'end_workflow', request=flow_ref) + flow_uid_str = utils.base64_url_encode(workflow_state.flowUid) + logger.info('Workflow request cancelled for %s (%s)', rec.title, record_uid) + return { + 'status': 'success', + 'record_uid': record_uid, + 'record_name': rec.title, + 'flow_uid': flow_uid_str, + 'action': 'cancelled', + } + except WorkflowError: + raise + except ROUTER_TRANSPORT_ERRORS as e: + raise WorkflowError(f'Failed to cancel request: {sanitize_router_error(e)}') from e + + +def start_workflow(vault: vault_online.VaultOnline, uid: str) -> dict: + """Start a workflow / check-out (`pam workflow start`). Accepts record UID/name or flow UID.""" + record_uid, rec = RecordResolver.resolve(vault, uid, allow_missing=True) + state = workflow_pb2.WorkflowState() + if record_uid: + record_uid_bytes = utils.base64_url_decode(record_uid) + state.resource.CopyFrom(ProtobufRefBuilder.record_ref(record_uid_bytes, rec.title)) + else: + try: + uid_bytes = utils.base64_url_decode(uid) + except (ValueError, TypeError) as e: + raise WorkflowError(f'"{uid}" is not a valid record UID/name or flow UID') from e + state.flowUid = uid_bytes + state.resource.CopyFrom(ProtobufRefBuilder.workflow_ref(uid_bytes)) + try: + post_to_router(vault, 'start_workflow', request=state) + result = {'status': 'success', 'action': 'checked_out'} + if rec: + result['record_uid'] = record_uid + result['record_name'] = rec.title + logger.info('Workflow started (checked out) for %s (%s)', rec.title, record_uid) + else: + result['flow_uid'] = uid + logger.info('Workflow started (checked out) for flow %s', uid) + return result + except WorkflowError: + raise + except ROUTER_TRANSPORT_ERRORS as e: + raise WorkflowError(f'Failed to start workflow: {sanitize_router_error(e)}') from e + + +def end_workflow(vault: vault_online.VaultOnline, uid: str, *, force: bool = False) -> dict: + """End a workflow / check-in (`pam workflow end`). force=True allows approver force check-in.""" + if force: + return _force_checkin(vault, uid) + record_uid, rec = RecordResolver.resolve(vault, uid, allow_missing=True) + if record_uid: + return _end_by_record(vault, record_uid, rec) + return _end_by_flow_uid(vault, uid) + + +def _force_checkin(vault: vault_online.VaultOnline, uid: str) -> dict: + if not uid or not uid.strip(): + raise WorkflowError('Record UID, record name, or Flow UID is required') + record_uid, rec = RecordResolver.resolve(vault, uid, allow_missing=True) + if record_uid: + ref = GraphSync_pb2.GraphSyncRef() + ref.type = GraphSync_pb2.RFT_REC + ref.value = utils.base64_url_decode(record_uid) + if rec: + ref.name = rec.title + else: + try: + uid_bytes = utils.base64_url_decode(uid) + except (ValueError, TypeError) as e: + raise WorkflowError(f'"{uid}" is not a valid record UID/name or flow UID') from e + ref = GraphSync_pb2.GraphSyncRef() + ref.type = GraphSync_pb2.RFT_WORKFLOW + ref.value = uid_bytes + try: + post_to_router(vault, 'force_checkin', request=ref) + result = {'status': 'success', 'action': 'force_checkin'} + if record_uid: + result['record_uid'] = record_uid + result['record_name'] = rec.title if rec else '' + logger.info('Record force checked in: %s (%s)', rec.title if rec else '', record_uid) + else: + result['flow_uid'] = uid + logger.info('Record force checked in for flow %s', uid) + return result + except WorkflowError: + raise + except ROUTER_TRANSPORT_ERRORS as e: + raise WorkflowError(f'Failed to force check-in: {sanitize_router_error(e)}') from e + + +def _end_by_record( + vault: vault_online.VaultOnline, + record_uid: str, + rec: Optional[vault_record.TypedRecord]) -> dict: + try: + state_query = workflow_pb2.WorkflowState() + state_query.resource.CopyFrom( + ProtobufRefBuilder.record_ref( + utils.base64_url_decode(record_uid), rec.title if rec else '') + ) + workflow_state = post_to_router( + vault, 'get_workflow_state', + request=state_query, response_type=workflow_pb2.WorkflowState, + ) + if not workflow_state or not workflow_state.flowUid: + raise WorkflowError( + 'No active workflow found for this record. ' + 'The workflow may have already ended or never started.' + ) + flow_ref = ProtobufRefBuilder.workflow_ref(workflow_state.flowUid) + post_to_router(vault, 'end_workflow', request=flow_ref) + flow_uid_str = utils.base64_url_encode(workflow_state.flowUid) + logger.info( + 'Workflow ended (checked in) for %s (%s). Credentials may have been rotated.', + rec.title if rec else record_uid, record_uid, + ) + return { + 'status': 'success', + 'flow_uid': flow_uid_str, + 'record_uid': record_uid, + 'record_name': rec.title if rec else '', + 'action': 'ended', + } + except WorkflowError: + raise + except ROUTER_TRANSPORT_ERRORS as e: + raise WorkflowError(f'Failed to end workflow: {sanitize_router_error(e)}') from e + + +def _end_by_flow_uid(vault: vault_online.VaultOnline, uid: str) -> dict: + try: + uid_bytes = utils.base64_url_decode(uid) + ref = ProtobufRefBuilder.workflow_ref(uid_bytes) + post_to_router(vault, 'end_workflow', request=ref) + logger.info('Workflow ended (checked in) for flow %s. Credentials may have been rotated.', uid) + return {'status': 'success', 'flow_uid': uid, 'action': 'ended'} + except WorkflowError: + raise + except (ValueError, TypeError) as e: + raise WorkflowError(f'"{uid}" is not a valid flow UID') from e + except ROUTER_TRANSPORT_ERRORS as e: + raise WorkflowError(f'Failed to end workflow: {sanitize_router_error(e)}') from e + + +def get_workflow_state( + vault: vault_online.VaultOnline, + record: str, + enterprise_data: Optional[enterprise_types.IEnterpriseData] = None) -> dict: + """Get workflow state for a record (`pam workflow state`).""" + record_uid, rec = RecordResolver.resolve(vault, record) + if is_workflow_exempt(vault, record_uid): + logger.warning( + 'You are exempt from workflow restrictions on this record. ' + 'As a record owner or approver, you can access this resource directly.' + ) + return {'status': 'exempt', 'message': 'Workflow not required', 'record_uid': record_uid} + + state = workflow_pb2.WorkflowState() + record_uid_bytes = utils.base64_url_decode(record_uid) + state.resource.CopyFrom(ProtobufRefBuilder.record_ref(record_uid_bytes, rec.title)) + try: + response = post_to_router( + vault, 'get_workflow_state', request=state, response_type=workflow_pb2.WorkflowState) + if response is None: + logger.warning('No workflow found for record %s (%s)', rec.title, record_uid) + return {'status': 'no_workflow', 'message': 'No workflow found', 'record_uid': record_uid} + result = workflow_state_to_dict(vault, response, enterprise_data) + result['status'] = 'success' + logger.info('Workflow state for %s (%s): %s', rec.title, record_uid, result.get('stage')) + return result + except WorkflowError: + raise + except ROUTER_TRANSPORT_ERRORS as e: + raise WorkflowError(f'Failed to get workflow state: {sanitize_router_error(e)}') from e + + +def get_user_access_state( + vault: vault_online.VaultOnline, + enterprise_data: Optional[enterprise_types.IEnterpriseData] = None) -> dict: + """Get all workflow states for the current user (`pam workflow my-access`).""" + try: + response = post_to_router( + vault, 'get_user_access_state', response_type=workflow_pb2.UserAccessState) + if not response or not response.workflows: + logger.info('No active workflows') + return {'status': 'success', 'workflows': []} + workflows = [workflow_state_to_dict(vault, wf, enterprise_data) for wf in response.workflows] + logger.info('Found %s active workflow(s) for current user', len(workflows)) + return {'status': 'success', 'workflows': workflows} + except WorkflowError: + raise + except ROUTER_TRANSPORT_ERRORS as e: + raise WorkflowError(f'Failed to get user access state: {sanitize_router_error(e)}') from e diff --git a/keepersdk-package/src/keepersdk/proto/workflow_pb2.py b/keepersdk-package/src/keepersdk/proto/workflow_pb2.py new file mode 100644 index 00000000..c24a4f19 --- /dev/null +++ b/keepersdk-package/src/keepersdk/proto/workflow_pb2.py @@ -0,0 +1,79 @@ +# -*- coding: utf-8 -*- +# Generated by the protocol buffer compiler. DO NOT EDIT! +# NO CHECKED-IN PROTOBUF GENCODE +# source: workflow.proto +# Protobuf Python Version: 5.29.5 +"""Generated protocol buffer code.""" +from google.protobuf import descriptor as _descriptor +from google.protobuf import descriptor_pool as _descriptor_pool +from google.protobuf import runtime_version as _runtime_version +from google.protobuf import symbol_database as _symbol_database +from google.protobuf.internal import builder as _builder +_runtime_version.ValidateProtobufRuntimeVersion( + _runtime_version.Domain.PUBLIC, + 5, + 29, + 5, + '', + 'workflow.proto' +) +# @@protoc_insertion_point(imports) + +_sym_db = _symbol_database.Default() + + +from . import GraphSync_pb2 as GraphSync__pb2 +from . import NotificationCenter_pb2 as NotificationCenter__pb2 + + +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(b'\n\x0eworkflow.proto\x12\x08Workflow\x1a\x0fGraphSync.proto\x1a\x18NotificationCenter.proto\"\x82\x01\n\x10WorkflowApprover\x12\x0e\n\x04user\x18\x01 \x01(\tH\x00\x12\x10\n\x06userId\x18\x02 \x01(\x05H\x00\x12\x11\n\x07teamUid\x18\x03 \x01(\x0cH\x00\x12\x12\n\nescalation\x18\x04 \x01(\x08\x12\x19\n\x11\x65scalationAfterMs\x18\x05 \x01(\x03\x42\n\n\x08\x61pprover\"\x9d\x02\n\x12WorkflowParameters\x12)\n\x08resource\x18\x01 \x01(\x0b\x32\x17.GraphSync.GraphSyncRef\x12\x17\n\x0f\x61pprovalsNeeded\x18\x02 \x01(\x05\x12\x16\n\x0e\x63heckoutNeeded\x18\x03 \x01(\x08\x12\x1d\n\x15startAccessOnApproval\x18\x04 \x01(\x08\x12\x15\n\rrequireReason\x18\x05 \x01(\x08\x12\x15\n\rrequireTicket\x18\x06 \x01(\x08\x12\x12\n\nrequireMFA\x18\x07 \x01(\x08\x12\x14\n\x0c\x61\x63\x63\x65ssLength\x18\x08 \x01(\x03\x12\x34\n\x0c\x61llowedTimes\x18\t \x01(\x0b\x32\x1e.Workflow.TemporalAccessFilter\"\x84\x01\n\x0eWorkflowConfig\x12\x30\n\nparameters\x18\x01 \x01(\x0b\x32\x1c.Workflow.WorkflowParameters\x12-\n\tapprovers\x18\x02 \x03(\x0b\x32\x1a.Workflow.WorkflowApprover\x12\x11\n\tcreatedOn\x18\x03 \x01(\x03\"\xff\x01\n\x0eWorkflowStatus\x12&\n\x05stage\x18\x01 \x01(\x0e\x32\x17.Workflow.WorkflowStage\x12-\n\nconditions\x18\x02 \x03(\x0e\x32\x19.Workflow.AccessCondition\x12.\n\napprovedBy\x18\x03 \x03(\x0b\x32\x1a.Workflow.WorkflowApproval\x12\x11\n\tstartedOn\x18\x04 \x01(\x03\x12\x11\n\texpiresOn\x18\x05 \x01(\x03\x12\x11\n\tescalated\x18\x06 \x01(\x08\x12\x14\n\x0c\x63heckedOutBy\x18\x07 \x01(\t\x12\x17\n\x0f\x63\x61nForceCheckIn\x18\x08 \x01(\x08\"\xa9\x02\n\x0fWorkflowProcess\x12\x0f\n\x07\x66lowUid\x18\x01 \x01(\x0c\x12\x0e\n\x06userId\x18\x02 \x01(\x03\x12)\n\x08resource\x18\x03 \x01(\x0b\x32\x17.GraphSync.GraphSyncRef\x12\x11\n\tstartedOn\x18\x04 \x01(\x03\x12\x11\n\texpiresOn\x18\x05 \x01(\x03\x12\x12\n\x06reason\x18\x06 \x01(\x0c\x42\x02\x18\x01\x12\x13\n\x0bmfaVerified\x18\x07 \x01(\x08\x12\x13\n\x0b\x65xternalRef\x18\x08 \x01(\x0c\x12\x0c\n\x04user\x18\t \x01(\t\x12\x45\n\x12workflowParameters\x18\n \x03(\x0b\x32).NotificationCenter.NotificationParameter\x12\x11\n\tescalated\x18\x0b \x01(\x08\"U\n\x10WorkflowApproval\x12\x0e\n\x06userId\x18\x01 \x01(\x03\x12\x0c\n\x04user\x18\x02 \x01(\t\x12\x0f\n\x07\x66lowUid\x18\x03 \x01(\x0c\x12\x12\n\napprovedOn\x18\x04 \x01(\x03\"\xcb\x01\n\x0fWorkflowContext\x12\x30\n\x0eworkflowConfig\x18\x01 \x01(\x0b\x32\x18.Workflow.WorkflowConfig\x12+\n\x08workflow\x18\x02 \x01(\x0b\x32\x19.Workflow.WorkflowProcess\x12-\n\tapprovals\x18\x03 \x03(\x0b\x32\x1a.Workflow.WorkflowApproval\x12*\n\x07\x62locker\x18\x04 \x01(\x0b\x32\x19.Workflow.WorkflowProcess\"u\n\rWorkflowState\x12\x0f\n\x07\x66lowUid\x18\x01 \x01(\x0c\x12)\n\x08resource\x18\x02 \x01(\x0b\x32\x17.GraphSync.GraphSyncRef\x12(\n\x06status\x18\x03 \x01(\x0b\x32\x18.Workflow.WorkflowStatus\"b\n\x15WorkflowAccessRequest\x12)\n\x08resource\x18\x01 \x01(\x0b\x32\x17.GraphSync.GraphSyncRef\x12\x0e\n\x06reason\x18\x02 \x01(\x0c\x12\x0e\n\x06ticket\x18\x03 \x01(\x0c\"O\n\x18WorkflowApprovalOrDenial\x12\x0f\n\x07\x66lowUid\x18\x01 \x01(\x0c\x12\x0c\n\x04\x64\x65ny\x18\x02 \x01(\x08\x12\x14\n\x0c\x64\x65nialReason\x18\x03 \x01(\x0c\"=\n\x0fUserAccessState\x12*\n\tworkflows\x18\x01 \x03(\x0b\x32\x17.Workflow.WorkflowState\"@\n\x10\x41pprovalRequests\x12,\n\tworkflows\x18\x01 \x03(\x0b\x32\x19.Workflow.WorkflowProcess\"4\n\x0eTimeOfDayRange\x12\x11\n\tstartTime\x18\x01 \x01(\x05\x12\x0f\n\x07\x65ndTime\x18\x02 \x01(\x05\"\xa3\x02\n\x12\x41pprovalQueueEntry\x12(\n\x07\x66lowRef\x18\x01 \x01(\x0b\x32\x17.GraphSync.GraphSyncRef\x12,\n\x0b\x61pproverRef\x18\x02 \x01(\x0b\x32\x17.GraphSync.GraphSyncRef\x12)\n\x04kind\x18\x03 \x01(\x0e\x32\x1b.Workflow.ApprovalQueueKind\x12\x12\n\nnotifyAtMs\x18\x04 \x01(\x03\x12\x1c\n\x0frequesterUserId\x18\x05 \x01(\x03H\x00\x88\x01\x01\x12&\n\x19predefinedNotificationUid\x18\x06 \x01(\x0cH\x01\x88\x01\x01\x42\x12\n\x10_requesterUserIdB\x1c\n\x1a_predefinedNotificationUid\"\x80\x01\n\x14TemporalAccessFilter\x12,\n\ntimeRanges\x18\x01 \x03(\x0b\x32\x18.Workflow.TimeOfDayRange\x12(\n\x0b\x61llowedDays\x18\x02 \x03(\x0e\x32\x13.Workflow.DayOfWeek\x12\x10\n\x08timeZone\x18\x03 \x01(\t\"#\n\x0f\x41uthorizedUsers\x12\x10\n\x08username\x18\x01 \x03(\t*[\n\rWorkflowStage\x12\x15\n\x11WS_READY_TO_START\x10\x00\x12\x0e\n\nWS_STARTED\x10\x01\x12\x13\n\x0fWS_NEEDS_ACTION\x10\x02\x12\x0e\n\nWS_WAITING\x10\x03*i\n\x0f\x41\x63\x63\x65ssCondition\x12\x0f\n\x0b\x41\x43_APPROVAL\x10\x00\x12\x0e\n\nAC_CHECKIN\x10\x01\x12\n\n\x06\x41\x43_MFA\x10\x02\x12\x0b\n\x07\x41\x43_TIME\x10\x03\x12\r\n\tAC_REASON\x10\x04\x12\r\n\tAC_TICKET\x10\x05*\x84\x01\n\tDayOfWeek\x12\x1b\n\x17\x44\x41Y_OF_WEEK_UNSPECIFIED\x10\x00\x12\n\n\x06MONDAY\x10\x01\x12\x0b\n\x07TUESDAY\x10\x02\x12\r\n\tWEDNESDAY\x10\x03\x12\x0c\n\x08THURSDAY\x10\x04\x12\n\n\x06\x46RIDAY\x10\x05\x12\x0c\n\x08SATURDAY\x10\x06\x12\n\n\x06SUNDAY\x10\x07*9\n\x11\x41pprovalQueueKind\x12\x10\n\x0c\x41QK_APPROVAL\x10\x00\x12\x12\n\x0e\x41QK_ESCALATION\x10\x01\x42$\n\x18\x63om.keepersecurity.protoB\x08Workflowb\x06proto3') + +_globals = globals() +_builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals) +_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, 'workflow_pb2', _globals) +if not _descriptor._USE_C_DESCRIPTORS: + _globals['DESCRIPTOR']._loaded_options = None + _globals['DESCRIPTOR']._serialized_options = b'\n\030com.keepersecurity.protoB\010Workflow' + _globals['_WORKFLOWPROCESS'].fields_by_name['reason']._loaded_options = None + _globals['_WORKFLOWPROCESS'].fields_by_name['reason']._serialized_options = b'\030\001' + _globals['_WORKFLOWSTAGE']._serialized_start=2423 + _globals['_WORKFLOWSTAGE']._serialized_end=2514 + _globals['_ACCESSCONDITION']._serialized_start=2516 + _globals['_ACCESSCONDITION']._serialized_end=2621 + _globals['_DAYOFWEEK']._serialized_start=2624 + _globals['_DAYOFWEEK']._serialized_end=2756 + _globals['_APPROVALQUEUEKIND']._serialized_start=2758 + _globals['_APPROVALQUEUEKIND']._serialized_end=2815 + _globals['_WORKFLOWAPPROVER']._serialized_start=72 + _globals['_WORKFLOWAPPROVER']._serialized_end=202 + _globals['_WORKFLOWPARAMETERS']._serialized_start=205 + _globals['_WORKFLOWPARAMETERS']._serialized_end=490 + _globals['_WORKFLOWCONFIG']._serialized_start=493 + _globals['_WORKFLOWCONFIG']._serialized_end=625 + _globals['_WORKFLOWSTATUS']._serialized_start=628 + _globals['_WORKFLOWSTATUS']._serialized_end=883 + _globals['_WORKFLOWPROCESS']._serialized_start=886 + _globals['_WORKFLOWPROCESS']._serialized_end=1183 + _globals['_WORKFLOWAPPROVAL']._serialized_start=1185 + _globals['_WORKFLOWAPPROVAL']._serialized_end=1270 + _globals['_WORKFLOWCONTEXT']._serialized_start=1273 + _globals['_WORKFLOWCONTEXT']._serialized_end=1476 + _globals['_WORKFLOWSTATE']._serialized_start=1478 + _globals['_WORKFLOWSTATE']._serialized_end=1595 + _globals['_WORKFLOWACCESSREQUEST']._serialized_start=1597 + _globals['_WORKFLOWACCESSREQUEST']._serialized_end=1695 + _globals['_WORKFLOWAPPROVALORDENIAL']._serialized_start=1697 + _globals['_WORKFLOWAPPROVALORDENIAL']._serialized_end=1776 + _globals['_USERACCESSSTATE']._serialized_start=1778 + _globals['_USERACCESSSTATE']._serialized_end=1839 + _globals['_APPROVALREQUESTS']._serialized_start=1841 + _globals['_APPROVALREQUESTS']._serialized_end=1905 + _globals['_TIMEOFDAYRANGE']._serialized_start=1907 + _globals['_TIMEOFDAYRANGE']._serialized_end=1959 + _globals['_APPROVALQUEUEENTRY']._serialized_start=1962 + _globals['_APPROVALQUEUEENTRY']._serialized_end=2253 + _globals['_TEMPORALACCESSFILTER']._serialized_start=2256 + _globals['_TEMPORALACCESSFILTER']._serialized_end=2384 + _globals['_AUTHORIZEDUSERS']._serialized_start=2386 + _globals['_AUTHORIZEDUSERS']._serialized_end=2421 +# @@protoc_insertion_point(module_scope) diff --git a/keepersdk-package/src/keepersdk/proto/workflow_pb2.pyi b/keepersdk-package/src/keepersdk/proto/workflow_pb2.pyi new file mode 100644 index 00000000..2e56ebc4 --- /dev/null +++ b/keepersdk-package/src/keepersdk/proto/workflow_pb2.pyi @@ -0,0 +1,259 @@ +from . import GraphSync_pb2 as _GraphSync_pb2 +from . import NotificationCenter_pb2 as _NotificationCenter_pb2 +from google.protobuf.internal import containers as _containers +from google.protobuf.internal import enum_type_wrapper as _enum_type_wrapper +from google.protobuf import descriptor as _descriptor +from google.protobuf import message as _message +from typing import ClassVar as _ClassVar, Iterable as _Iterable, Mapping as _Mapping, Optional as _Optional, Union as _Union + +DESCRIPTOR: _descriptor.FileDescriptor + +class WorkflowStage(int, metaclass=_enum_type_wrapper.EnumTypeWrapper): + __slots__ = () + WS_READY_TO_START: _ClassVar[WorkflowStage] + WS_STARTED: _ClassVar[WorkflowStage] + WS_NEEDS_ACTION: _ClassVar[WorkflowStage] + WS_WAITING: _ClassVar[WorkflowStage] + +class AccessCondition(int, metaclass=_enum_type_wrapper.EnumTypeWrapper): + __slots__ = () + AC_APPROVAL: _ClassVar[AccessCondition] + AC_CHECKIN: _ClassVar[AccessCondition] + AC_MFA: _ClassVar[AccessCondition] + AC_TIME: _ClassVar[AccessCondition] + AC_REASON: _ClassVar[AccessCondition] + AC_TICKET: _ClassVar[AccessCondition] + +class DayOfWeek(int, metaclass=_enum_type_wrapper.EnumTypeWrapper): + __slots__ = () + DAY_OF_WEEK_UNSPECIFIED: _ClassVar[DayOfWeek] + MONDAY: _ClassVar[DayOfWeek] + TUESDAY: _ClassVar[DayOfWeek] + WEDNESDAY: _ClassVar[DayOfWeek] + THURSDAY: _ClassVar[DayOfWeek] + FRIDAY: _ClassVar[DayOfWeek] + SATURDAY: _ClassVar[DayOfWeek] + SUNDAY: _ClassVar[DayOfWeek] + +class ApprovalQueueKind(int, metaclass=_enum_type_wrapper.EnumTypeWrapper): + __slots__ = () + AQK_APPROVAL: _ClassVar[ApprovalQueueKind] + AQK_ESCALATION: _ClassVar[ApprovalQueueKind] +WS_READY_TO_START: WorkflowStage +WS_STARTED: WorkflowStage +WS_NEEDS_ACTION: WorkflowStage +WS_WAITING: WorkflowStage +AC_APPROVAL: AccessCondition +AC_CHECKIN: AccessCondition +AC_MFA: AccessCondition +AC_TIME: AccessCondition +AC_REASON: AccessCondition +AC_TICKET: AccessCondition +DAY_OF_WEEK_UNSPECIFIED: DayOfWeek +MONDAY: DayOfWeek +TUESDAY: DayOfWeek +WEDNESDAY: DayOfWeek +THURSDAY: DayOfWeek +FRIDAY: DayOfWeek +SATURDAY: DayOfWeek +SUNDAY: DayOfWeek +AQK_APPROVAL: ApprovalQueueKind +AQK_ESCALATION: ApprovalQueueKind + +class WorkflowApprover(_message.Message): + __slots__ = ("user", "userId", "teamUid", "escalation", "escalationAfterMs") + USER_FIELD_NUMBER: _ClassVar[int] + USERID_FIELD_NUMBER: _ClassVar[int] + TEAMUID_FIELD_NUMBER: _ClassVar[int] + ESCALATION_FIELD_NUMBER: _ClassVar[int] + ESCALATIONAFTERMS_FIELD_NUMBER: _ClassVar[int] + user: str + userId: int + teamUid: bytes + escalation: bool + escalationAfterMs: int + def __init__(self, user: _Optional[str] = ..., userId: _Optional[int] = ..., teamUid: _Optional[bytes] = ..., escalation: bool = ..., escalationAfterMs: _Optional[int] = ...) -> None: ... + +class WorkflowParameters(_message.Message): + __slots__ = ("resource", "approvalsNeeded", "checkoutNeeded", "startAccessOnApproval", "requireReason", "requireTicket", "requireMFA", "accessLength", "allowedTimes") + RESOURCE_FIELD_NUMBER: _ClassVar[int] + APPROVALSNEEDED_FIELD_NUMBER: _ClassVar[int] + CHECKOUTNEEDED_FIELD_NUMBER: _ClassVar[int] + STARTACCESSONAPPROVAL_FIELD_NUMBER: _ClassVar[int] + REQUIREREASON_FIELD_NUMBER: _ClassVar[int] + REQUIRETICKET_FIELD_NUMBER: _ClassVar[int] + REQUIREMFA_FIELD_NUMBER: _ClassVar[int] + ACCESSLENGTH_FIELD_NUMBER: _ClassVar[int] + ALLOWEDTIMES_FIELD_NUMBER: _ClassVar[int] + resource: _GraphSync_pb2.GraphSyncRef + approvalsNeeded: int + checkoutNeeded: bool + startAccessOnApproval: bool + requireReason: bool + requireTicket: bool + requireMFA: bool + accessLength: int + allowedTimes: TemporalAccessFilter + def __init__(self, resource: _Optional[_Union[_GraphSync_pb2.GraphSyncRef, _Mapping]] = ..., approvalsNeeded: _Optional[int] = ..., checkoutNeeded: bool = ..., startAccessOnApproval: bool = ..., requireReason: bool = ..., requireTicket: bool = ..., requireMFA: bool = ..., accessLength: _Optional[int] = ..., allowedTimes: _Optional[_Union[TemporalAccessFilter, _Mapping]] = ...) -> None: ... + +class WorkflowConfig(_message.Message): + __slots__ = ("parameters", "approvers", "createdOn") + PARAMETERS_FIELD_NUMBER: _ClassVar[int] + APPROVERS_FIELD_NUMBER: _ClassVar[int] + CREATEDON_FIELD_NUMBER: _ClassVar[int] + parameters: WorkflowParameters + approvers: _containers.RepeatedCompositeFieldContainer[WorkflowApprover] + createdOn: int + def __init__(self, parameters: _Optional[_Union[WorkflowParameters, _Mapping]] = ..., approvers: _Optional[_Iterable[_Union[WorkflowApprover, _Mapping]]] = ..., createdOn: _Optional[int] = ...) -> None: ... + +class WorkflowStatus(_message.Message): + __slots__ = ("stage", "conditions", "approvedBy", "startedOn", "expiresOn", "escalated", "checkedOutBy", "canForceCheckIn") + STAGE_FIELD_NUMBER: _ClassVar[int] + CONDITIONS_FIELD_NUMBER: _ClassVar[int] + APPROVEDBY_FIELD_NUMBER: _ClassVar[int] + STARTEDON_FIELD_NUMBER: _ClassVar[int] + EXPIRESON_FIELD_NUMBER: _ClassVar[int] + ESCALATED_FIELD_NUMBER: _ClassVar[int] + CHECKEDOUTBY_FIELD_NUMBER: _ClassVar[int] + CANFORCECHECKIN_FIELD_NUMBER: _ClassVar[int] + stage: WorkflowStage + conditions: _containers.RepeatedScalarFieldContainer[AccessCondition] + approvedBy: _containers.RepeatedCompositeFieldContainer[WorkflowApproval] + startedOn: int + expiresOn: int + escalated: bool + checkedOutBy: str + canForceCheckIn: bool + def __init__(self, stage: _Optional[_Union[WorkflowStage, str]] = ..., conditions: _Optional[_Iterable[_Union[AccessCondition, str]]] = ..., approvedBy: _Optional[_Iterable[_Union[WorkflowApproval, _Mapping]]] = ..., startedOn: _Optional[int] = ..., expiresOn: _Optional[int] = ..., escalated: bool = ..., checkedOutBy: _Optional[str] = ..., canForceCheckIn: bool = ...) -> None: ... + +class WorkflowProcess(_message.Message): + __slots__ = ("flowUid", "userId", "resource", "startedOn", "expiresOn", "reason", "mfaVerified", "externalRef", "user", "workflowParameters", "escalated") + FLOWUID_FIELD_NUMBER: _ClassVar[int] + USERID_FIELD_NUMBER: _ClassVar[int] + RESOURCE_FIELD_NUMBER: _ClassVar[int] + STARTEDON_FIELD_NUMBER: _ClassVar[int] + EXPIRESON_FIELD_NUMBER: _ClassVar[int] + REASON_FIELD_NUMBER: _ClassVar[int] + MFAVERIFIED_FIELD_NUMBER: _ClassVar[int] + EXTERNALREF_FIELD_NUMBER: _ClassVar[int] + USER_FIELD_NUMBER: _ClassVar[int] + WORKFLOWPARAMETERS_FIELD_NUMBER: _ClassVar[int] + ESCALATED_FIELD_NUMBER: _ClassVar[int] + flowUid: bytes + userId: int + resource: _GraphSync_pb2.GraphSyncRef + startedOn: int + expiresOn: int + reason: bytes + mfaVerified: bool + externalRef: bytes + user: str + workflowParameters: _containers.RepeatedCompositeFieldContainer[_NotificationCenter_pb2.NotificationParameter] + escalated: bool + def __init__(self, flowUid: _Optional[bytes] = ..., userId: _Optional[int] = ..., resource: _Optional[_Union[_GraphSync_pb2.GraphSyncRef, _Mapping]] = ..., startedOn: _Optional[int] = ..., expiresOn: _Optional[int] = ..., reason: _Optional[bytes] = ..., mfaVerified: bool = ..., externalRef: _Optional[bytes] = ..., user: _Optional[str] = ..., workflowParameters: _Optional[_Iterable[_Union[_NotificationCenter_pb2.NotificationParameter, _Mapping]]] = ..., escalated: bool = ...) -> None: ... + +class WorkflowApproval(_message.Message): + __slots__ = ("userId", "user", "flowUid", "approvedOn") + USERID_FIELD_NUMBER: _ClassVar[int] + USER_FIELD_NUMBER: _ClassVar[int] + FLOWUID_FIELD_NUMBER: _ClassVar[int] + APPROVEDON_FIELD_NUMBER: _ClassVar[int] + userId: int + user: str + flowUid: bytes + approvedOn: int + def __init__(self, userId: _Optional[int] = ..., user: _Optional[str] = ..., flowUid: _Optional[bytes] = ..., approvedOn: _Optional[int] = ...) -> None: ... + +class WorkflowContext(_message.Message): + __slots__ = ("workflowConfig", "workflow", "approvals", "blocker") + WORKFLOWCONFIG_FIELD_NUMBER: _ClassVar[int] + WORKFLOW_FIELD_NUMBER: _ClassVar[int] + APPROVALS_FIELD_NUMBER: _ClassVar[int] + BLOCKER_FIELD_NUMBER: _ClassVar[int] + workflowConfig: WorkflowConfig + workflow: WorkflowProcess + approvals: _containers.RepeatedCompositeFieldContainer[WorkflowApproval] + blocker: WorkflowProcess + def __init__(self, workflowConfig: _Optional[_Union[WorkflowConfig, _Mapping]] = ..., workflow: _Optional[_Union[WorkflowProcess, _Mapping]] = ..., approvals: _Optional[_Iterable[_Union[WorkflowApproval, _Mapping]]] = ..., blocker: _Optional[_Union[WorkflowProcess, _Mapping]] = ...) -> None: ... + +class WorkflowState(_message.Message): + __slots__ = ("flowUid", "resource", "status") + FLOWUID_FIELD_NUMBER: _ClassVar[int] + RESOURCE_FIELD_NUMBER: _ClassVar[int] + STATUS_FIELD_NUMBER: _ClassVar[int] + flowUid: bytes + resource: _GraphSync_pb2.GraphSyncRef + status: WorkflowStatus + def __init__(self, flowUid: _Optional[bytes] = ..., resource: _Optional[_Union[_GraphSync_pb2.GraphSyncRef, _Mapping]] = ..., status: _Optional[_Union[WorkflowStatus, _Mapping]] = ...) -> None: ... + +class WorkflowAccessRequest(_message.Message): + __slots__ = ("resource", "reason", "ticket") + RESOURCE_FIELD_NUMBER: _ClassVar[int] + REASON_FIELD_NUMBER: _ClassVar[int] + TICKET_FIELD_NUMBER: _ClassVar[int] + resource: _GraphSync_pb2.GraphSyncRef + reason: bytes + ticket: bytes + def __init__(self, resource: _Optional[_Union[_GraphSync_pb2.GraphSyncRef, _Mapping]] = ..., reason: _Optional[bytes] = ..., ticket: _Optional[bytes] = ...) -> None: ... + +class WorkflowApprovalOrDenial(_message.Message): + __slots__ = ("flowUid", "deny", "denialReason") + FLOWUID_FIELD_NUMBER: _ClassVar[int] + DENY_FIELD_NUMBER: _ClassVar[int] + DENIALREASON_FIELD_NUMBER: _ClassVar[int] + flowUid: bytes + deny: bool + denialReason: bytes + def __init__(self, flowUid: _Optional[bytes] = ..., deny: bool = ..., denialReason: _Optional[bytes] = ...) -> None: ... + +class UserAccessState(_message.Message): + __slots__ = ("workflows",) + WORKFLOWS_FIELD_NUMBER: _ClassVar[int] + workflows: _containers.RepeatedCompositeFieldContainer[WorkflowState] + def __init__(self, workflows: _Optional[_Iterable[_Union[WorkflowState, _Mapping]]] = ...) -> None: ... + +class ApprovalRequests(_message.Message): + __slots__ = ("workflows",) + WORKFLOWS_FIELD_NUMBER: _ClassVar[int] + workflows: _containers.RepeatedCompositeFieldContainer[WorkflowProcess] + def __init__(self, workflows: _Optional[_Iterable[_Union[WorkflowProcess, _Mapping]]] = ...) -> None: ... + +class TimeOfDayRange(_message.Message): + __slots__ = ("startTime", "endTime") + STARTTIME_FIELD_NUMBER: _ClassVar[int] + ENDTIME_FIELD_NUMBER: _ClassVar[int] + startTime: int + endTime: int + def __init__(self, startTime: _Optional[int] = ..., endTime: _Optional[int] = ...) -> None: ... + +class ApprovalQueueEntry(_message.Message): + __slots__ = ("flowRef", "approverRef", "kind", "notifyAtMs", "requesterUserId", "predefinedNotificationUid") + FLOWREF_FIELD_NUMBER: _ClassVar[int] + APPROVERREF_FIELD_NUMBER: _ClassVar[int] + KIND_FIELD_NUMBER: _ClassVar[int] + NOTIFYATMS_FIELD_NUMBER: _ClassVar[int] + REQUESTERUSERID_FIELD_NUMBER: _ClassVar[int] + PREDEFINEDNOTIFICATIONUID_FIELD_NUMBER: _ClassVar[int] + flowRef: _GraphSync_pb2.GraphSyncRef + approverRef: _GraphSync_pb2.GraphSyncRef + kind: ApprovalQueueKind + notifyAtMs: int + requesterUserId: int + predefinedNotificationUid: bytes + def __init__(self, flowRef: _Optional[_Union[_GraphSync_pb2.GraphSyncRef, _Mapping]] = ..., approverRef: _Optional[_Union[_GraphSync_pb2.GraphSyncRef, _Mapping]] = ..., kind: _Optional[_Union[ApprovalQueueKind, str]] = ..., notifyAtMs: _Optional[int] = ..., requesterUserId: _Optional[int] = ..., predefinedNotificationUid: _Optional[bytes] = ...) -> None: ... + +class TemporalAccessFilter(_message.Message): + __slots__ = ("timeRanges", "allowedDays", "timeZone") + TIMERANGES_FIELD_NUMBER: _ClassVar[int] + ALLOWEDDAYS_FIELD_NUMBER: _ClassVar[int] + TIMEZONE_FIELD_NUMBER: _ClassVar[int] + timeRanges: _containers.RepeatedCompositeFieldContainer[TimeOfDayRange] + allowedDays: _containers.RepeatedScalarFieldContainer[DayOfWeek] + timeZone: str + def __init__(self, timeRanges: _Optional[_Iterable[_Union[TimeOfDayRange, _Mapping]]] = ..., allowedDays: _Optional[_Iterable[_Union[DayOfWeek, str]]] = ..., timeZone: _Optional[str] = ...) -> None: ... + +class AuthorizedUsers(_message.Message): + __slots__ = ("username",) + USERNAME_FIELD_NUMBER: _ClassVar[int] + username: _containers.RepeatedScalarFieldContainer[str] + def __init__(self, username: _Optional[_Iterable[str]] = ...) -> None: ... diff --git a/keepersdk-package/unit_tests/test_workflow.py b/keepersdk-package/unit_tests/test_workflow.py new file mode 100644 index 00000000..c51dd46e --- /dev/null +++ b/keepersdk-package/unit_tests/test_workflow.py @@ -0,0 +1,84 @@ +import os +import unittest + +from keepersdk.errors import KeeperError +from keepersdk.helpers.workflow import WorkflowError, WorkflowFormatter +from keepersdk.proto import workflow_pb2 + + +class WorkflowFormatterTestCase(unittest.TestCase): + def test_parse_duration(self): + self.assertEqual(WorkflowFormatter.parse_duration('1d'), 86_400_000) + self.assertEqual(WorkflowFormatter.parse_duration('2h'), 7_200_000) + self.assertEqual(WorkflowFormatter.parse_duration('30m'), 1_800_000) + self.assertEqual(WorkflowFormatter.parse_duration('15'), 15 * 60_000) + + def test_parse_duration_invalid(self): + with self.assertRaises(WorkflowError): + WorkflowFormatter.parse_duration('0h') + with self.assertRaises(WorkflowError): + WorkflowFormatter.parse_duration('abc') + self.assertIsInstance(WorkflowError('x'), KeeperError) + + def test_format_duration(self): + self.assertEqual(WorkflowFormatter.format_duration(86_400_000), '1 day') + self.assertEqual(WorkflowFormatter.format_duration(172_800_000), '2 days') + self.assertEqual(WorkflowFormatter.format_duration(3_600_000), '1 hour') + self.assertEqual(WorkflowFormatter.format_duration(60_000), '1 minute') + self.assertEqual(WorkflowFormatter.format_duration(5_000), '5 seconds') + + def test_format_stage(self): + self.assertEqual( + WorkflowFormatter.format_stage(workflow_pb2.WS_STARTED), 'Started') + self.assertEqual( + WorkflowFormatter.format_stage(workflow_pb2.WS_WAITING), 'Waiting') + + def test_format_conditions(self): + text = WorkflowFormatter.format_conditions( + [workflow_pb2.AC_APPROVAL, workflow_pb2.AC_MFA]) + self.assertEqual(text, 'Approval Required, MFA Required') + + def test_build_temporal_filter_days(self): + os.environ['TZ'] = 'America/New_York' + temporal = WorkflowFormatter.build_temporal_filter('mon,fri', None) + self.assertIsNotNone(temporal) + self.assertEqual(list(temporal.allowedDays), [workflow_pb2.MONDAY, workflow_pb2.FRIDAY]) + self.assertEqual(temporal.timeZone, 'America/New_York') + + def test_build_temporal_filter_time_range(self): + os.environ['TZ'] = 'Asia/Kolkata' + temporal = WorkflowFormatter.build_temporal_filter(None, '09:00-17:30') + self.assertEqual(len(temporal.timeRanges), 1) + self.assertEqual(temporal.timeRanges[0].startTime, 900) + self.assertEqual(temporal.timeRanges[0].endTime, 1730) + + def test_build_temporal_filter_invalid_day(self): + os.environ['TZ'] = 'UTC' + with self.assertRaises(WorkflowError): + WorkflowFormatter.build_temporal_filter('funday', None) + + def test_build_temporal_filter_invalid_time_range(self): + os.environ['TZ'] = 'UTC' + with self.assertRaises(WorkflowError): + WorkflowFormatter.build_temporal_filter(None, '17:00-09:00') + + def test_build_temporal_filter_invalid_timezone(self): + os.environ['TZ'] = 'garbage/value' + with self.assertRaises(WorkflowError): + WorkflowFormatter.build_temporal_filter(None, '09:00-17:00') + + def test_format_temporal_filter(self): + temporal = workflow_pb2.TemporalAccessFilter() + temporal.allowedDays.append(workflow_pb2.MONDAY) + tr = temporal.timeRanges.add() + tr.startTime = 900 + tr.endTime = 1700 + temporal.timeZone = 'UTC' + formatted = WorkflowFormatter.format_temporal_filter(temporal) + self.assertEqual(formatted['allowed_days'], ['Monday']) + self.assertEqual(formatted['time_ranges'], ['09:00-17:00']) + self.assertEqual(formatted['timezone'], 'UTC') + + +if __name__ == '__main__': + unittest.main() From 6b57e089f3b8e655c2f1e62d6ca951160390ae06 Mon Sep 17 00:00:00 2001 From: adeshmukh-ks Date: Sat, 5 Sep 2026 21:14:09 +0530 Subject: [PATCH 2/4] Enhancement to add parent folder uid and name while fetching folder detail (#231) * Enhancement to add parent folder uid and name while fetching folder detail * Fix import in pam rotation script --- .../sdk_examples/nested_shared_folders/nsf_get.py | 5 ++++- .../src/keepercli/commands/nsf_commands.py | 11 +++++++++-- .../src/keepercli/commands/pam/pam_rotation.py | 1 + .../src/keepersdk/vault/nsf_management.py | 12 ++++++++++++ 4 files changed, 26 insertions(+), 3 deletions(-) diff --git a/examples/sdk_examples/nested_shared_folders/nsf_get.py b/examples/sdk_examples/nested_shared_folders/nsf_get.py index bdaeb0b5..b1db4581 100644 --- a/examples/sdk_examples/nested_shared_folders/nsf_get.py +++ b/examples/sdk_examples/nested_shared_folders/nsf_get.py @@ -514,7 +514,10 @@ def close_vault(vault: vault_online.VaultOnline, keeper_auth_context: keeper_aut def nsf_get(vault: vault_online.VaultOnline) -> None: - """Get NSF record or folder details by UID or title (nsf-get).""" + """Get NSF record or folder details by UID or title (nsf-get). + + For folders, the response includes parent_uid and parent_name (when available). + """ ITEM_UID_OR_TITLE = "" # Record/folder UID or title detail = nsf_management.get_nsf_item(vault, ITEM_UID_OR_TITLE) diff --git a/keepercli-package/src/keepercli/commands/nsf_commands.py b/keepercli-package/src/keepercli/commands/nsf_commands.py index f86932cc..ea322e5f 100644 --- a/keepercli-package/src/keepercli/commands/nsf_commands.py +++ b/keepercli-package/src/keepercli/commands/nsf_commands.py @@ -355,7 +355,11 @@ def _print_folder_detail(detail: Dict[str, Any], verbose: bool) -> None: logger.info('') logger.info('{0:>25s}: {1}'.format('NSF Folder UID', detail.get('nsf_folder_uid', ''))) logger.info('{0:>25s}: {1}'.format('Name', detail.get('name', ''))) - logger.info('{0:>25s}: {1}'.format('Parent', detail.get('parent_uid', ''))) + if detail.get('parent_uid'): + parent_display = detail.get('parent_uid', '') + if detail.get('parent_name'): + parent_display = f"{detail['parent_name']} ({detail['parent_uid']})" + logger.info('{0:>25s}: {1}'.format('Parent', parent_display)) NsfGetCommand._print_folder_access( detail.get('access') or {}, verbose, @@ -368,8 +372,11 @@ def _print_folder_json(detail: Dict[str, Any], verbose: bool) -> None: fo = { 'nsf_folder_uid': detail.get('nsf_folder_uid'), 'name': detail.get('name'), - 'parent_uid': detail.get('parent_uid'), } + if detail.get('parent_uid'): + fo['parent_uid'] = detail.get('parent_uid') + if detail.get('parent_name'): + fo['parent_name'] = detail.get('parent_name') if detail.get('owner_username'): fo['owner'] = detail['owner_username'] access = detail.get('access') or {} diff --git a/keepercli-package/src/keepercli/commands/pam/pam_rotation.py b/keepercli-package/src/keepercli/commands/pam/pam_rotation.py index 0efcb4b9..ec35efb8 100644 --- a/keepercli-package/src/keepercli/commands/pam/pam_rotation.py +++ b/keepercli-package/src/keepercli/commands/pam/pam_rotation.py @@ -18,6 +18,7 @@ vault_utils, record_facades, attachment, + nsf_management ) from keepersdk.proto import pam_pb2, router_pb2 diff --git a/keepersdk-package/src/keepersdk/vault/nsf_management.py b/keepersdk-package/src/keepersdk/vault/nsf_management.py index b1890225..33e0bda5 100644 --- a/keepersdk-package/src/keepersdk/vault/nsf_management.py +++ b/keepersdk-package/src/keepersdk/vault/nsf_management.py @@ -408,6 +408,18 @@ def get_nsf_folder_detail( result['owner_username'] = row.owner_username result['owner_account_uid'] = row.owner_account_uid + if folder.parent_uid: + parent_uid = folder.parent_uid + root_folder_uid = vault.vault_data.root_folder.folder_uid + if parent_uid == root_folder_uid or parent_uid.startswith('AAAAAA'): + result['parent_uid'] = parent_uid + result['parent_name'] = 'root' + else: + result['parent_uid'] = parent_uid + parent_folder = _nsf_view(vault).get_folder(parent_uid) + if parent_folder: + result['parent_name'] = parent_folder.name or parent_uid + if include_access: try: access = get_nsf_folder_access(vault, [folder_uid]) From 17622d3e51285bc9938e4b375dcbee5418c0f322 Mon Sep 17 00:00:00 2001 From: adeshmukh-ks Date: Tue, 8 Sep 2026 18:51:32 +0530 Subject: [PATCH 3/4] Added cache for nsf folder and record access detai (#233) ls and improved batching --- .../src/keepercli/commands/nsf_commands.py | 31 ++ .../src/keepercli/register_commands.py | 1 + .../src/keepersdk/vault/nsf_data.py | 28 +- .../src/keepersdk/vault/nsf_management.py | 276 ++++++++++++------ 4 files changed, 245 insertions(+), 91 deletions(-) diff --git a/keepercli-package/src/keepercli/commands/nsf_commands.py b/keepercli-package/src/keepercli/commands/nsf_commands.py index ea322e5f..14384095 100644 --- a/keepercli-package/src/keepercli/commands/nsf_commands.py +++ b/keepercli-package/src/keepercli/commands/nsf_commands.py @@ -1607,3 +1607,34 @@ def __init__(self): self.register_command(NsfShortcutKeepCommand(), 'keep') self.default_verb = 'list' + +class NsfLoadAccessCacheCommand(base.ArgparseCommand): + + def __init__(self): + parser = argparse.ArgumentParser( + prog='nsf-load-access', + description='Load access details for every NSF folder and record into the in-memory vault cache', + ) + super().__init__(parser) + + def add_arguments_to_parser(parser: argparse.ArgumentParser) -> None: + parser.add_argument('--folders', dest='load_folder', action='store_true', + help='Load access details for folders') + parser.add_argument('--records', dest='load_record', action='store_true', + help='Load access details for records') + + def execute(self, context: KeeperParams, **kwargs): + vault = _require_vault(context) + load_folder = kwargs.get('load_folder', False) + load_record = kwargs.get('load_record', False) + if not load_folder and not load_record: + raise base.CommandError('At least one of --folders or --records is required') + + def _run(): + return nsf_management.load_nsf_access_details(vault) + + loaded = _wrap_nsf('nsf-load-access', _run) + logger.info( + 'Loaded access details for %d folder(s) and %d record(s) into the NSF cache', + loaded.get('folders', 0), loaded.get('records', 0)) + diff --git a/keepercli-package/src/keepercli/register_commands.py b/keepercli-package/src/keepercli/register_commands.py index d41a9152..fc378872 100644 --- a/keepercli-package/src/keepercli/register_commands.py +++ b/keepercli-package/src/keepercli/register_commands.py @@ -92,6 +92,7 @@ def register_commands(commands: base.CliCommands, scopes: Optional[base.CommandS commands.register_command('share-report', share_report.ShareReportCommand(), base.CommandScope.Vault) commands.register_command('nsf-list', nsf_commands.NsfListCommand(), base.CommandScope.Vault) commands.register_command('nsf-get', nsf_commands.NsfGetCommand(), base.CommandScope.Vault) + commands.register_command('nsf-load-access', nsf_commands.NsfLoadAccessCacheCommand(), base.CommandScope.Vault) commands.register_command('nsf-record-add', nsf_commands.NsfRecordAddCommand(), base.CommandScope.Vault) commands.register_command('nsf-record-update', nsf_commands.NsfRecordUpdateCommand(), base.CommandScope.Vault) commands.register_command('nsf-record-details', nsf_commands.NsfRecordDetailsCommand(), base.CommandScope.Vault) diff --git a/keepersdk-package/src/keepersdk/vault/nsf_data.py b/keepersdk-package/src/keepersdk/vault/nsf_data.py index 619beb73..cba50908 100644 --- a/keepersdk-package/src/keepersdk/vault/nsf_data.py +++ b/keepersdk-package/src/keepersdk/vault/nsf_data.py @@ -1,7 +1,7 @@ from __future__ import annotations from dataclasses import dataclass, field -from typing import Dict, Iterable, List, Mapping, Optional, Set +from typing import Any, Dict, Iterable, List, Mapping, Optional, Set from ..authentication import keeper_auth from . import nsf_crypto, nsf_storage_types as nsf @@ -71,6 +71,8 @@ def __init__( self._auth_context = auth_context self._folders: Dict[str, NSFFolderNode] = {} self._records: Dict[str, NSFRecordEntry] = {} + self._folder_access_details: Dict[str, Dict[str, Any]] = {} + self._record_access_details: Dict[str, List[Dict[str, Any]]] = {} if auth_context is not None: self.rebuild_nsf(auth_context) @@ -90,6 +92,30 @@ def get_folder(self, folder_uid: str) -> Optional[NSFFolderNode]: def get_record(self, record_uid: str) -> Optional[NSFRecordEntry]: return self._records.get(record_uid) + def nsf_folder_access_details(self) -> Iterable[Dict[str, Any]]: + yield from self._folder_access_details.values() + + def get_nsf_folder_access_detail(self, folder_uid: str) -> Optional[Dict[str, Any]]: + return self._folder_access_details.get(folder_uid) + + def set_nsf_folder_access_detail(self, folder_uid: str, detail: Dict[str, Any]) -> None: + self._folder_access_details[folder_uid] = detail + + def clear_nsf_folder_access_details(self) -> None: + self._folder_access_details.clear() + + def nsf_record_access_details(self) -> Iterable[List[Dict[str, Any]]]: + yield from self._record_access_details.values() + + def get_nsf_record_access_detail(self, record_uid: str) -> Optional[List[Dict[str, Any]]]: + return self._record_access_details.get(record_uid) + + def set_nsf_record_access_detail(self, record_uid: str, detail: List[Dict[str, Any]]) -> None: + self._record_access_details[record_uid] = detail + + def clear_nsf_record_access_details(self) -> None: + self._record_access_details.clear() + @property def folder_count(self) -> int: return len(self._folders) diff --git a/keepersdk-package/src/keepersdk/vault/nsf_management.py b/keepersdk-package/src/keepersdk/vault/nsf_management.py index 33e0bda5..69cc210b 100644 --- a/keepersdk-package/src/keepersdk/vault/nsf_management.py +++ b/keepersdk-package/src/keepersdk/vault/nsf_management.py @@ -8,7 +8,7 @@ from .. import crypto, utils from ..errors import KeeperApiError from ..proto import folder_pb2, record_endpoints_pb2, record_pb2, remove_pb2, record_details_pb2, folder_access_pb2 -from . import nsf_crypto, nsf_data, nsf_common, sync_down, vault_extensions +from . import nsf_crypto, nsf_data, nsf_common, sync_down, vault_extensions, vault_record from .vault_online import VaultOnline ROOT_FOLDER_UID = 'AAAAAAAAAAAAAAAAAPmtNA' @@ -885,7 +885,7 @@ def update_nsf_record( def update_nsf_typed_record( vault: VaultOnline, - record: 'vault_record.TypedRecord', + record: vault_record.TypedRecord, *, request_sync: bool = True) -> NsfModifyResult: """Update an NSF typed record, including record-link adds/removes for file/script refs. @@ -1033,45 +1033,50 @@ def get_nsf_record_details( return {'data': out_data, 'forbidden_records': forbidden} +_RECORD_ACCESS_CHUNK = 100 + + def get_nsf_record_accesses( vault: VaultOnline, record_uids: Iterable[str]) -> Dict[str, Any]: - """``vault/records/v3/details/access``.""" + """``vault/records/v3/details/access``, chunked at 100 UIDs per request.""" uids = [resolve_nsf_record_uid(vault, u) or u for u in record_uids] uids = [u for u in uids if u] if not uids: raise NsfError('At least one record UID is required') - rq = record_details_pb2.RecordAccessRequest() - for uid in uids: - rq.recordUids.append(utils.base64_url_decode(uid)) - rs = vault.keeper_auth.execute_auth_rest('vault/records/v3/details/access', rq, response_type=record_details_pb2.RecordAccessResponse) - if rs is None: - return {'record_accesses': [], 'forbidden_records': []} - - result = {'record_accesses': [], 'forbidden_records': []} - for ra in rs.recordAccesses: - d = ra.data - ai = ra.accessorInfo - ao = { - 'record_uid': utils.base64_url_encode(d.recordUid), - 'accessor_name': ai.name, - 'access_type': folder_pb2.AccessType.Name(d.accessType) if hasattr(d, 'accessType') else 'UNKNOWN', - 'access_type_uid': utils.base64_url_encode(d.accessTypeUid), - 'owner': getattr(d, 'owner', False), - 'inherited': bool(getattr(d, 'inherited', False)), - 'access_role_type': int(getattr(d, 'accessRoleType', 0) or 0), - } - for flag in ('can_view_title', 'can_edit', 'can_view', 'can_list_access', - 'can_update_access', 'can_delete', 'can_change_ownership', - 'can_request_access', 'can_approve_access', 'denied_access'): - if flag == 'denied_access': - ao[flag] = getattr(d, 'deniedAccess', False) - else: - ao[flag] = getattr(d, flag, False) - result['record_accesses'].append(ao) - for fu in rs.forbiddenRecords: - result['forbidden_records'].append(utils.base64_url_encode(fu)) + result: Dict[str, Any] = {'record_accesses': [], 'forbidden_records': []} + for i in range(0, len(uids), _RECORD_ACCESS_CHUNK): + chunk = uids[i:i + _RECORD_ACCESS_CHUNK] + rq = record_details_pb2.RecordAccessRequest() + for uid in chunk: + rq.recordUids.append(utils.base64_url_decode(uid)) + rs = vault.keeper_auth.execute_auth_rest('vault/records/v3/details/access', rq, response_type=record_details_pb2.RecordAccessResponse) + if rs is None: + continue + + for ra in rs.recordAccesses: + d = ra.data + ai = ra.accessorInfo + ao = { + 'record_uid': utils.base64_url_encode(d.recordUid), + 'accessor_name': ai.name, + 'access_type': folder_pb2.AccessType.Name(d.accessType) if hasattr(d, 'accessType') else 'UNKNOWN', + 'access_type_uid': utils.base64_url_encode(d.accessTypeUid), + 'owner': getattr(d, 'owner', False), + 'inherited': bool(getattr(d, 'inherited', False)), + 'access_role_type': int(getattr(d, 'accessRoleType', 0) or 0), + } + for flag in ('can_view_title', 'can_edit', 'can_view', 'can_list_access', + 'can_update_access', 'can_delete', 'can_change_ownership', + 'can_request_access', 'can_approve_access', 'denied_access'): + if flag == 'denied_access': + ao[flag] = getattr(d, 'deniedAccess', False) + else: + ao[flag] = getattr(d, flag, False) + result['record_accesses'].append(ao) + for fu in rs.forbiddenRecords: + result['forbidden_records'].append(utils.base64_url_encode(fu)) return result @@ -1091,10 +1096,95 @@ def _resolve_uid_to_username(vault: VaultOnline, uid_b64: str) -> Optional[str]: pass +_FOLDER_ACCESS_CHUNK = 100 + + +_FOLDER_ACCESS_MAX_PAGES = 50 +"""Safety cap on continuation-token pages consumed per UID chunk.""" + + +def _parse_folder_accessor(vault: VaultOnline, a: Any) -> Dict[str, Any]: + auid = utils.base64_url_encode(a.accessTypeUid) + at = folder_pb2.AccessType.Name(a.accessType) + rt = folder_pb2.AccessRoleType.Name(a.accessRoleType) + username = None + if at == 'AT_USER': + username = _resolve_uid_to_username(vault, auid) + ai = { + 'accessor_uid': auid, 'access_type': at, 'role': rt, + 'access_role_type': int(a.accessRoleType), + 'inherited': bool(a.inherited), 'hidden': bool(a.hidden), + 'username': username, + 'date_created': a.dateCreated or None, + 'last_modified': a.lastModified or None, + } + if at == 'AT_OWNER': + ai['owner'] = True + if a.HasField('permissions'): + p = a.permissions + ai['permissions'] = { + 'can_add': bool(p.canAdd), 'can_remove': bool(p.canRemove), + 'can_delete': bool(p.canDelete), + 'can_list_access': bool(p.canListAccess), + 'can_update_access': bool(p.canUpdateAccess), + 'can_change_ownership': bool(p.canChangeOwnership), + 'can_edit_records': bool(p.canEditRecords), + 'can_view_records': bool(p.canViewRecords), + 'can_approve_access': bool(p.canApproveAccess), + 'can_request_access': bool(p.canRequestAccess), + 'can_update_setting': bool(p.canUpdateSetting), + 'can_list_records': bool(p.canListRecords), + 'can_list_folders': bool(p.canListFolders), + } + return ai + + +def _fetch_nsf_folder_access_chunk( + vault: VaultOnline, chunk: List[str]) -> Dict[str, Dict[str, Any]]: + """Fetch access details for one <=100-UID chunk, fully draining continuation pages.""" + accumulated: Dict[str, Dict[str, Any]] = {} + continuation_token = None + for _ in range(_FOLDER_ACCESS_MAX_PAGES): + rq = folder_access_pb2.GetFolderAccessRequest() + for uid in chunk: + rq.folderUid.append(utils.base64_url_decode(uid)) + if continuation_token is not None: + rq.continuationToken.lastModified = continuation_token + rs = vault.keeper_auth.execute_auth_rest( + 'vault/folders/v3/access', rq, response_type=folder_access_pb2.GetFolderAccessResponse) + if rs is None: + break + + for fr in rs.folderAccessResults: + fuid = utils.base64_url_encode(fr.folderUid) + if fr.HasField('error'): + err = fr.error + accumulated[fuid] = { + 'folder_uid': fuid, + 'error': {'status': folder_pb2.FolderModifyStatus.Name(err.status), + 'message': err.message}, + 'success': False} + else: + accessors = [_parse_folder_accessor(vault, a) for a in fr.accessors] + entry = accumulated.setdefault( + fuid, {'folder_uid': fuid, 'accessors': [], 'success': True}) + entry['accessors'].extend(accessors) + + if not rs.hasMore or not rs.HasField('continuationToken'): + break + continuation_token = rs.continuationToken.lastModified + return accumulated + + def get_nsf_folder_access( vault: VaultOnline, folder_uids: Iterable[str]) -> Dict[str, Any]: - """``vault/folders/v3/access``.""" + """``vault/folders/v3/access``, chunked at 100 UIDs per request. + + Fully drains continuation-token pagination within each chunk before + moving to the next, so the returned ``results`` are always complete — + there is no caller-facing ``has_more``/``continuation_token`` to track. + """ uids: List[str] = [] for raw in folder_uids: resolved = resolve_nsf_folder_uid(vault, raw) or raw @@ -1103,62 +1193,68 @@ def get_nsf_folder_access( if not uids: raise NsfError('At least one folder UID is required') - rq = folder_access_pb2.GetFolderAccessRequest() - for uid in uids: - rq.folderUid.append(utils.base64_url_decode(uid)) - rs = vault.keeper_auth.execute_auth_rest('vault/folders/v3/access', rq, response_type=folder_access_pb2.GetFolderAccessResponse) - results = [] - for fr in rs.folderAccessResults: - fuid = utils.base64_url_encode(fr.folderUid) - if fr.HasField('error'): - err = fr.error - results.append({ - 'folder_uid': fuid, - 'error': {'status': folder_pb2.FolderModifyStatus.Name(err.status), - 'message': err.message}, - 'success': False}) - else: - accessors = [] - for a in fr.accessors: - auid = utils.base64_url_encode(a.accessTypeUid) - at = folder_pb2.AccessType.Name(a.accessType) - rt = folder_pb2.AccessRoleType.Name(a.accessRoleType) - username = None - if at == 'AT_USER': - username = _resolve_uid_to_username(vault, auid) - ai = { - 'accessor_uid': auid, 'access_type': at, 'role': rt, - 'access_role_type': int(a.accessRoleType), - 'inherited': bool(a.inherited), 'hidden': bool(a.hidden), - 'username': username, - 'date_created': a.dateCreated or None, - 'last_modified': a.lastModified or None, - } - if at == 'AT_OWNER': - ai['owner'] = True - if a.HasField('permissions'): - p = a.permissions - ai['permissions'] = { - 'can_add': bool(p.canAdd), 'can_remove': bool(p.canRemove), - 'can_delete': bool(p.canDelete), - 'can_list_access': bool(p.canListAccess), - 'can_update_access': bool(p.canUpdateAccess), - 'can_change_ownership': bool(p.canChangeOwnership), - 'can_edit_records': bool(p.canEditRecords), - 'can_view_records': bool(p.canViewRecords), - 'can_approve_access': bool(p.canApproveAccess), - 'can_request_access': bool(p.canRequestAccess), - 'can_update_setting': bool(p.canUpdateSetting), - 'can_list_records': bool(p.canListRecords), - 'can_list_folders': bool(p.canListFolders), - } - accessors.append(ai) - results.append({'folder_uid': fuid, 'accessors': accessors, 'success': True}) - - rd = {'results': results, 'has_more': bool(rs.hasMore)} - if rs.HasField('continuationToken'): - rd['continuation_token'] = rs.continuationToken.lastModified - return rd + results: List[Dict[str, Any]] = [] + for i in range(0, len(uids), _FOLDER_ACCESS_CHUNK): + chunk = uids[i:i + _FOLDER_ACCESS_CHUNK] + accumulated = _fetch_nsf_folder_access_chunk(vault, chunk) + results.extend(accumulated.values()) + return {'results': results} + + +def load_nsf_access_details(vault: VaultOnline, load_folder: bool = False, load_record: bool = False) -> Dict[str, int]: + """Fetch access details for every NSF folder/record and cache them on ``vault.nsf_data``. + + Scans all folder and record UIDs currently known to the NSF cache + (``vault.nsf_data``), fetches their access details via + :func:`get_nsf_folder_access` / :func:`get_nsf_record_accesses`, and stores + the results back onto ``vault.nsf_data`` — folders keyed by folder UID (the + full result entry returned by :func:`get_nsf_folder_access`), records keyed + by record UID (the list of accessor rows for that record). + """ + view = _nsf_view(vault) + if not view.folders() and not view.records(): + raise NsfError('Pass at least one of load_folder or load_record to load access details for NSF cache') + folder_uid_list = [f.folder_uid for f in view.folders()] if load_folder else [] + record_uid_list = [r.record_uid for r in view.records()] if load_record else [] + if not folder_uid_list and not record_uid_list: + raise NsfError('NSF cache has no folders or records to load access details for') + + loaded = {'folders': 0, 'records': 0} + + if folder_uid_list: + access = get_nsf_folder_access(vault, folder_uid_list) + for fr in access.get('results') or []: + fuid = fr.get('folder_uid') + if not fuid: + continue + view.set_nsf_folder_access_detail(fuid, fr) + loaded['folders'] += 1 + + if record_uid_list: + access = get_nsf_record_accesses(vault, record_uid_list) + by_uid: Dict[str, List[Dict[str, Any]]] = {} + for ao in access.get('record_accesses') or []: + by_uid.setdefault(ao['record_uid'], []).append(ao) + for uid in record_uid_list: + view.set_nsf_record_access_detail(uid, by_uid.get(uid, [])) + loaded['records'] += 1 + + return loaded + + +def get_nsf_access_details_from_cache(vault: VaultOnline, record_uid: str = None, folder_uid: str = None) -> Dict[str, Any]: + view = _nsf_view(vault) + if record_uid: + record = view.get_record(record_uid) + if not record: + raise NsfError(f'NSF record not found: {record_uid}') + return vault.nsf_data._record_access_details.get(record_uid) or {} + if folder_uid: + folder = view.get_folder(folder_uid) + if not folder: + raise NsfError(f'NSF folder not found: {folder_uid}') + return vault.nsf_data._folder_access_details.get(folder_uid) or {} + raise ValueError("At least one of record_uid or folder_uid is required") def _request_sync(vault: VaultOnline, request_sync: bool) -> None: From 9b2d0ef042ae00222f99720b3bdd8c4e677e867b Mon Sep 17 00:00:00 2001 From: Sergey Kolupaev Date: Wed, 9 Sep 2026 09:59:11 -0700 Subject: [PATCH 4/4] Release 1.2.7 --- keepersdk-package/src/keepersdk/__init__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/keepersdk-package/src/keepersdk/__init__.py b/keepersdk-package/src/keepersdk/__init__.py index dbf644f2..f189b014 100644 --- a/keepersdk-package/src/keepersdk/__init__.py +++ b/keepersdk-package/src/keepersdk/__init__.py @@ -10,6 +10,6 @@ # from . import background -__version__ = '1.2.6' +__version__ = '1.2.7' background.init()