Implementation of PAM Workflow Commands

keepercommander/cli.py

@@ -126,6 +126,7 @@ def clean_description(desc):
         ('pam rotation', 'Manage Rotations'),
         ('pam split', 'Split credentials from legacy PAM Machine'),
         ('pam tunnel', 'Manage Tunnels'),
+        ('pam workflow', 'Manage PAM Workflows'),
     ]
     domain_subcommands = [
         ('domain list (dl)', 'List all reserved domains for the enterprise'),

keepercommander/commands/discoveryrotation.py

@@ -76,6 +76,7 @@
 from .pam_debug.vertex import PAMDebugVertexCommand
 from .pam_import.commands import PAMProjectCommand
 from .pam_launch.launch import PAMLaunchCommand
+from .workflow import PAMWorkflowCommand
 from .pam_service.list import PAMActionServiceListCommand
 from .pam_service.add import PAMActionServiceAddCommand
 from .pam_service.remove import PAMActionServiceRemoveCommand
@@ -186,6 +187,7 @@ def __init__(self):
         self.register_command('rbi', PAMRbiCommand(), 'Manage Remote Browser Isolation', 'b')
         self.register_command('project', PAMProjectCommand(), 'PAM Project Import/Export', 'p')
         self.register_command('launch', PAMLaunchCommand(), 'Launch a connection to a PAM resource', 'l')
+        self.register_command('workflow', PAMWorkflowCommand(), 'Manage PAM Workflows', 'w')
 
 
 class PAMGatewayCommand(GroupCommand):

keepercommander/commands/pam_launch/launch.py

@@ -284,6 +284,17 @@ def execute(self, params: KeeperParams, **kwargs):
 
             logging.debug(f"Found record: {record_uid}")
 
+            # Workflow access check and 2FA prompt
+            try:
+                from ..workflow import check_workflow_and_prompt_2fa
+                should_proceed, two_factor_value = check_workflow_and_prompt_2fa(params, record_uid)
+                if not should_proceed:
+                    return
+                if two_factor_value:
+                    kwargs['two_factor_value'] = two_factor_value
+            except ImportError:
+                pass
+
             # Validate --user and --host parameters against allowSupply flags
             # Note: cmdline options override record data when provided
             # launch_credential_uid = kwargs.get('launch_credential_uid')

keepercommander/commands/pam_launch/terminal_connection.py

@@ -1328,6 +1328,11 @@ def _open_terminal_webrtc_tunnel(params: KeeperParams,
                 logging.debug("Using userSupplied credential type - user will provide credentials")
             # else: no credentialType - gateway uses pamMachine credentials directly (backward compatible)
 
+            # Add 2FA value if workflow requires MFA
+            two_factor_value = kwargs.get('two_factor_value')
+            if two_factor_value:
+                inputs['twoFactorValue'] = two_factor_value
+
             # Router token is no longer extracted from cookies (removed in commit 338a9fda)
             # Router affinity is now handled server-side
 

keepercommander/commands/tunnel/port_forward/tunnel_helpers.py

@@ -1966,7 +1966,7 @@ def cleanup(self):
         logging.debug("TunnelSignalHandler cleaned up")
 
 def start_rust_tunnel(params, record_uid, gateway_uid, host, port,
-                      seed, target_host, target_port, socks, trickle_ice=True, record_title=None, allow_supply_host=False):
+                      seed, target_host, target_port, socks, trickle_ice=True, record_title=None, allow_supply_host=False, two_factor_value=None):
     """
     Start a tunnel using Rust WebRTC with trickle ICE via HTTP POST and WebSocket responses.
 
@@ -2284,23 +2284,29 @@ def start_rust_tunnel(params, record_uid, gateway_uid, host, port,
                     }
                 if trickle_ice and http_session is not None:
                     offer_kwargs["http_session"] = http_session
+
+                # Build tunnel inputs
+                inputs = {
+                    "recordUid": record_uid,
+                    "tubeId": commander_tube_id,
+                    'kind': 'start',
+                    'base64Nonce': base64_nonce,
+                    'conversationType': 'tunnel',
+                    "data": encrypted_data,
+                    "trickleICE": trickle_ice,
+                }
+                if two_factor_value:
+                    inputs['twoFactorValue'] = two_factor_value
+
                 router_response = router_send_action_to_gateway(
                     params=params,
                     destination_gateway_uid_str=gateway_uid,
                     gateway_action=GatewayActionWebRTCSession(
                         conversation_id = conversation_id_original,
-                        inputs={
-                            "recordUid": record_uid,
-                            "tubeId": commander_tube_id,
-                            'kind': 'start',
-                            'base64Nonce': base64_nonce,
-                            'conversationType': 'tunnel',
-                            "data": encrypted_data,
-                            "trickleICE": trickle_ice,
-                        }
+                        inputs=inputs
                     ),
                     message_type=pam_pb2.CMT_CONNECT,
-                    is_streaming=trickle_ice,  # Streaming only for trickle ICE
+                    is_streaming=trickle_ice,
                     gateway_timeout=GATEWAY_TIMEOUT,
                     **offer_kwargs
                 )

keepercommander/commands/tunnel_and_connections.py

@@ -411,6 +411,11 @@ def execute(self, params, **kwargs):
                 if _remove_tunneling_override_port and pam_settings.value[0]['portForward'].get('port'):
                     pam_settings.value[0]['portForward'].pop('port')
                     dirty = True
+            # Persist the record changes (new pamSettings field or port modifications)
+            if dirty:
+                record_management.update_record(params, record)
+                api.sync_down(params)
+                dirty = False
             if not tmp_dag.is_tunneling_config_set_up(record_uid):
                 print(f"{bcolors.FAIL}No PAM Configuration UID set. This must be set for tunneling to work. "
                       f"This can be done by running "
@@ -538,6 +543,16 @@ def execute(self, params, **kwargs):
             print(f"{bcolors.FAIL}Record {record_uid} not found.{bcolors.ENDC}")
             return
 
+        # Workflow access check and 2FA prompt
+        two_factor_value = None
+        try:
+            from .workflow import check_workflow_and_prompt_2fa
+            should_proceed, two_factor_value = check_workflow_and_prompt_2fa(params, record_uid)
+            if not should_proceed:
+                return
+        except ImportError:
+            pass
+
         # Validate PAM settings
         pam_settings = record.get_typed_field('pamSettings')
         if not pam_settings:
@@ -633,7 +648,7 @@ def execute(self, params, **kwargs):
 
         # Use Rust WebRTC implementation with configurable trickle ICE
         trickle_ice = not no_trickle_ice
-        result = start_rust_tunnel(params, record_uid, gateway_uid, host, port, seed, target_host, target_port, socks, trickle_ice, record.title, allow_supply_host=allow_supply_host)
+        result = start_rust_tunnel(params, record_uid, gateway_uid, host, port, seed, target_host, target_port, socks, trickle_ice, record.title, allow_supply_host=allow_supply_host, two_factor_value=two_factor_value)
 
         if result and result.get("success"):
             # The helper will show endpoint table when local socket is actually listening

keepercommander/commands/workflow/__init__.py

@@ -0,0 +1,15 @@
+#  _  __
+# | |/ /___ ___ _ __  ___ _ _ ®
+# | ' </ -_) -_) '_ \/ -_) '_|
+# |_|\_\___\___| .__/\___|_|
+#              |_|
+#
+# Keeper Commander
+# Copyright 2026 Keeper Security Inc.
+# Contact: ops@keepersecurity.com
+#
+
+__all__ = ['PAMWorkflowCommand', 'check_workflow_access', 'check_workflow_and_prompt_2fa']
+
+from .registry import PAMWorkflowCommand
+from .mfa import check_workflow_access, check_workflow_and_prompt_2fa

keepercommander/commands/workflow/approver_commands.py

@@ -0,0 +1,211 @@
+#  _  __
+# | |/ /___ ___ _ __  ___ _ _ ®
+# | ' </ -_) -_) '_ \/ -_) '_|
+# |_|\_\___\___| .__/\___|_|
+#              |_|
+#
+# Keeper Commander
+# Copyright 2026 Keeper Security Inc.
+# Contact: ops@keepersecurity.com
+#
+
+import argparse
+import json
+from datetime import datetime
+
+from ..base import Command, dump_report_data
+from ..pam.router_helper import _post_request_to_router
+from ...display import bcolors
+from ...error import CommandError
+from ...params import KeeperParams
+from ...proto import workflow_pb2
+from ... import utils
+
+from .helpers import RecordResolver, WorkflowFormatter
+
+
+class WorkflowGetApprovalRequestsCommand(Command):
+    parser = argparse.ArgumentParser(
+        prog='pam workflow pending',
+        description='Get pending approval requests',
+    )
+    parser.add_argument('--format', dest='format', action='store',
+                        choices=['table', 'json'], default='table', help='Output format')
+
+    def get_parser(self):
+        return WorkflowGetApprovalRequestsCommand.parser
+
+    def execute(self, params: KeeperParams, **kwargs):
+        try:
+            response = _post_request_to_router(
+                params, 'get_approval_requests',
+                rs_type=workflow_pb2.ApprovalRequests,
+            )
+
+            if not response or not response.workflows:
+                if kwargs.get('format') == 'json':
+                    print(json.dumps({'requests': []}, indent=2))
+                else:
+                    print(f"\n{bcolors.WARNING}No approval requests{bcolors.ENDC}\n")
+                return
+
+            wf_data = [
+                (wf, self._resolve_status(params, wf))
+                for wf in response.workflows
+            ]
+
+            if kwargs.get('format') == 'json':
+                self._print_json(params, wf_data)
+            else:
+                self._print_table(params, wf_data)
+
+        except Exception as e:
+            raise CommandError('', f'Failed to get approval requests: {str(e)}')
+
+    @staticmethod
+    def _resolve_status(params, wf):
+        if wf.startedOn:
+            return 'Approved'
+        try:
+            st = workflow_pb2.WorkflowState()
+            st.flowUid = wf.flowUid
+            ws = _post_request_to_router(
+                params, 'get_workflow_state',
+                rq_proto=st, rs_type=workflow_pb2.WorkflowState,
+            )
+            if ws and ws.status and ws.status.stage in (
+                workflow_pb2.WS_READY_TO_START, workflow_pb2.WS_STARTED,
+            ):
+                return 'Approved'
+        except Exception:
+            pass
+        return 'Pending'
+
+    @staticmethod
+    def _print_json(params, wf_data):
+        result = {
+            'requests': [
+                {
+                    'flow_uid': utils.base64_url_encode(wf.flowUid),
+                    'status': status,
+                    'requested_by': RecordResolver.resolve_user(params, wf.userId),
+                    'record_uid': utils.base64_url_encode(wf.resource.value),
+                    'record_name': RecordResolver.resolve_name(params, wf.resource),
+                    'started_on': wf.startedOn or None,
+                    'expires_on': wf.expiresOn or None,
+                    'duration': (
+                        WorkflowFormatter.format_duration(wf.expiresOn - wf.startedOn)
+                        if wf.expiresOn and wf.startedOn else None
+                    ),
+                    'reason': wf.reason.decode('utf-8') if wf.reason else None,
+                    'external_ref': wf.externalRef.decode('utf-8') if wf.externalRef else None,
+                }
+                for wf, status in wf_data
+            ],
+        }
+        print(json.dumps(result, indent=2))
+
+    @staticmethod
+    def _print_table(params, wf_data):
+        rows = []
+        for wf, status in wf_data:
+            record_uid = utils.base64_url_encode(wf.resource.value) if wf.resource.value else ''
+            record_name = RecordResolver.resolve_name(params, wf.resource)
+            flow_uid = utils.base64_url_encode(wf.flowUid)
+            requested_by = RecordResolver.resolve_user(params, wf.userId)
+            started = (
+                datetime.fromtimestamp(wf.startedOn / 1000).strftime('%Y-%m-%d %H:%M:%S')
+                if wf.startedOn else ''
+            )
+            expires = (
+                datetime.fromtimestamp(wf.expiresOn / 1000).strftime('%Y-%m-%d %H:%M:%S')
+                if wf.expiresOn else ''
+            )
+            duration = (
+                WorkflowFormatter.format_duration(wf.expiresOn - wf.startedOn)
+                if wf.expiresOn and wf.startedOn else ''
+            )
+            rows.append([status, record_name, record_uid, flow_uid, requested_by, started, expires, duration])
+
+        headers = ['Status', 'Record Name', 'Record UID', 'Flow UID', 'Requested By', 'Started', 'Expires', 'Duration']
+        print()
+        dump_report_data(rows, headers=headers, sort_by=0)
+        print()
+
+
+class WorkflowApproveCommand(Command):
+    parser = argparse.ArgumentParser(
+        prog='pam workflow approve',
+        description='Approve a workflow access request',
+    )
+    parser.add_argument('flow_uid', help='Flow UID of the workflow to approve')
+    parser.add_argument('--format', dest='format', action='store',
+                        choices=['table', 'json'], default='table', help='Output format')
+
+    def get_parser(self):
+        return WorkflowApproveCommand.parser
+
+    def execute(self, params: KeeperParams, **kwargs):
+        flow_uid = kwargs.get('flow_uid')
+        flow_uid_bytes = utils.base64_url_decode(flow_uid)
+
+        approval = workflow_pb2.WorkflowApprovalOrDenial()
+        approval.flowUid = flow_uid_bytes
+        approval.deny = False
+
+        try:
+            _post_request_to_router(params, 'approve_workflow_access', rq_proto=approval)
+
+            if kwargs.get('format') == 'json':
+                result = {'status': 'success', 'flow_uid': flow_uid, 'action': 'approved'}
+                print(json.dumps(result, indent=2))
+            else:
+                print(f"\n{bcolors.OKGREEN}✓ Access request approved{bcolors.ENDC}\n")
+                print(f"Flow UID: {flow_uid}")
+                print()
+
+        except Exception as e:
+            raise CommandError('', f'Failed to approve request: {str(e)}')
+
+
+class WorkflowDenyCommand(Command):
+    parser = argparse.ArgumentParser(
+        prog='pam workflow deny',
+        description='Deny a workflow access request',
+    )
+    parser.add_argument('flow_uid', help='Flow UID of the workflow to deny')
+    parser.add_argument('-r', '--reason', help='Reason for denial')
+    parser.add_argument('--format', dest='format', action='store',
+                        choices=['table', 'json'], default='table', help='Output format')
+
+    def get_parser(self):
+        return WorkflowDenyCommand.parser
+
+    def execute(self, params: KeeperParams, **kwargs):
+        flow_uid = kwargs.get('flow_uid')
+        reason = kwargs.get('reason') or ''
+        flow_uid_bytes = utils.base64_url_decode(flow_uid)
+
+        denial = workflow_pb2.WorkflowApprovalOrDenial()
+        denial.flowUid = flow_uid_bytes
+        denial.deny = True
+        if reason:
+            denial.denialReason = reason
+
+        try:
+            _post_request_to_router(params, 'deny_workflow_access', rq_proto=denial)
+
+            if kwargs.get('format') == 'json':
+                result = {'status': 'success', 'flow_uid': flow_uid, 'action': 'denied'}
+                if reason:
+                    result['reason'] = reason
+                print(json.dumps(result, indent=2))
+            else:
+                print(f"\n{bcolors.WARNING}Access request denied{bcolors.ENDC}\n")
+                print(f"Flow UID: {flow_uid}")
+                if reason:
+                    print(f"Reason: {reason}")
+                print()
+
+        except Exception as e:
+            raise CommandError('', f'Failed to deny request: {str(e)}')