From f885ed01873f0702d6b060c40207892c5ba06784 Mon Sep 17 00:00:00 2001 From: Ivan Dimov <78815270+idimov-keeper@users.noreply.github.com> Date: Thu, 17 Sep 2026 19:44:38 -0500 Subject: [PATCH] Add ephemeral JIT fallback and --credential override for KeeperRDP Proxy tunnels --- .../tunnel/port_forward/tunnel_helpers.py | 49 +++- .../commands/tunnel_and_connections.py | 242 +++++++++++++++--- .../test_pam_tunnel_ephemeral_credential.py | 200 +++++++++++++++ 3 files changed, 450 insertions(+), 41 deletions(-) create mode 100644 unit-tests/pam/test_pam_tunnel_ephemeral_credential.py diff --git a/keepercommander/commands/tunnel/port_forward/tunnel_helpers.py b/keepercommander/commands/tunnel/port_forward/tunnel_helpers.py index e3384c78b..d9f5fb07b 100644 --- a/keepercommander/commands/tunnel/port_forward/tunnel_helpers.py +++ b/keepercommander/commands/tunnel/port_forward/tunnel_helpers.py @@ -177,6 +177,43 @@ def print_above_keeper_prompt(msg): # VERIFY_SSL applies to WebSocket SSL only; HTTP uses params.ssl_verify. VERIFY_SSL = bool(os.environ.get("VERIFY_SSL", "TRUE") == "TRUE") + +def ephemeral_gateway_timeout_ms(is_ephemeral, default=GATEWAY_TIMEOUT): + """Gateway wait timeout (ms) for a tunnel-start offer. + + Ephemeral JIT account creation on the gateway takes 30-90s (mirrors the + connection path's bump in terminal_connection.py) — the normal + ``default`` (30s) is too tight for it. + """ + if not is_ephemeral: + return default + try: + return int(os.environ.get('PAM_GATEWAY_OFFER_TIMEOUT_EPHEMERAL_MS', '120000')) + except (TypeError, ValueError): + return 120000 + + +def credential_override_data_fields(credential_type, credential_data): + """Fields to merge into a tunnel-start offer's encrypted inner payload + for a --credential/-cr override. Only 'userSupplied' is recognized today + (the only value Commander's --credential flag sends); anything else + yields no fields, so plain tunnel starts are unaffected. + """ + if credential_type != 'userSupplied' or not credential_data: + return {} + return { + 'username': credential_data.get('username', ''), + 'password': credential_data.get('password', ''), + } + + +def credential_override_input_fields(credential_type): + """Fields to merge into a tunnel-start offer's outer (unencrypted) + gateway action inputs for a --credential/-cr override.""" + if credential_type != 'userSupplied': + return {} + return {'credentialType': 'userSupplied', 'allowSupplyUser': True} + # ICE candidate buffering - store until SDP answer is received # Global conversation key management for multiple concurrent tunnels @@ -2337,7 +2374,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, two_factor_value=None, kind='start', probe_duration=30, probe_turn_only=False, probe_stun_only=False): + seed, target_host, target_port, socks, trickle_ice=True, record_title=None, allow_supply_host=False, two_factor_value=None, kind='start', probe_duration=30, probe_turn_only=False, probe_stun_only=False, credential_type=None, credential_data=None, is_ephemeral=False): """ Start a tunnel using Rust WebRTC with trickle ICE via HTTP POST and WebSocket responses. @@ -2662,6 +2699,13 @@ def start_rust_tunnel(params, record_uid, gateway_uid, host, port, # used for validators (record-type check, `allowKeeperXxxProxy` # presence check, launch-credential preflight) and to print the # right banner; it does not need to round-trip to the gateway. + # + # credential_type/credential_data are the exception: an explicit + # --credential override DOES need to round-trip, since it's telling + # the gateway to use inline username/password instead of its own + # ephemeral/linked resolution. + data.update(credential_override_data_fields(credential_type, credential_data)) + string_data = json.dumps(data) bytes_data = string_to_bytes(string_data) encrypted_data = tunnel_encrypt(symmetric_key, bytes_data) @@ -2700,6 +2744,7 @@ def start_rust_tunnel(params, record_uid, gateway_uid, host, port, } if two_factor_value: inputs['twoFactorValue'] = two_factor_value + inputs.update(credential_override_input_fields(credential_type)) router_response = router_send_action_to_gateway( params=params, @@ -2710,7 +2755,7 @@ def start_rust_tunnel(params, record_uid, gateway_uid, host, port, ), message_type=pam_pb2.CMT_CONNECT, is_streaming=trickle_ice, - gateway_timeout=GATEWAY_TIMEOUT, + gateway_timeout=ephemeral_gateway_timeout_ms(is_ephemeral), **offer_kwargs ) diff --git a/keepercommander/commands/tunnel_and_connections.py b/keepercommander/commands/tunnel_and_connections.py index 3a530118b..f92c86cc1 100644 --- a/keepercommander/commands/tunnel_and_connections.py +++ b/keepercommander/commands/tunnel_and_connections.py @@ -36,8 +36,10 @@ wait_for_tunnel_connection, create_rust_webrtc_settings, \ print_above_keeper_prompt from .pam.router_helper import get_dag_leafs -from .pam.vault_target import update_pam_record, reload_pam_record_if_nsf_updated +from .pam.vault_target import update_pam_record, reload_pam_record_if_nsf_updated, records_in_folder +from .pam_import.keeper_ai_settings import get_resource_jit_settings from .pam_import.nsf_helpers import sync_down_preserving_nsf_keys +from .pam_launch.terminal_connection import _extract_user_record_credentials from .tunnel_registry import ( PARENT_GRACE_SECONDS, is_pid_alive, @@ -51,7 +53,7 @@ from ..error import CommandError import json from ..params import LAST_RECORD_UID -from ..subfolder import find_folders +from ..subfolder import find_folders, try_resolve_path from ..utils import value_to_boolean from ..constants import get_relay_host, get_router_host, get_keeper_server_hostname @@ -90,6 +92,93 @@ def _coerce_settings_subdicts(entry, *keys): return changed +def _resource_has_ephemeral_jit(params, record_uid): + """True if the resource's jit_settings.createEphemeral is enabled on the DAG. + + Non-fatal on any read failure (unconfigured/unreadable jit_settings just + means the record isn't JIT-configured, which is the normal case). + """ + try: + jit = get_resource_jit_settings(params, record_uid) + except Exception as e: + logging.debug(f"Could not read jit_settings for {record_uid}: {e}") + return False + return bool(jit and jit.get('createEphemeral')) + + +def _classify_proxy_target(record_uid, record_type, protocol, is_keeper_proxy, launch_credential): + """Classify a `pam tunnel start` target into a proxy backend, given the + record's type/protocol and which of --proxy / --credential were passed. + + Pure decision logic (no params/network access) so it's directly + unit-testable. Returns (is_keeperdb_proxy, is_keeperrdp_proxy, error) — + ``error`` is a message to print and abort on, or None to proceed. + """ + if record_type == 'pamDatabase': + is_keeperdb_proxy, is_keeperrdp_proxy = True, False + elif record_type == 'pamMachine' and protocol == 'rdp': + is_keeperdb_proxy, is_keeperrdp_proxy = False, True + else: + detail = f' (protocol="{protocol}")' if record_type == 'pamMachine' else '' + if launch_credential and not is_keeper_proxy: + return False, False, "--credential requires a proxied tunnel; plain tunnels don't authenticate." + return False, False, ( + f'--proxy is supported on pamDatabase or pamMachine+RDP records. ' + f'Record {record_uid} is of type "{record_type}"{detail}.' + ) + + if launch_credential and is_keeperdb_proxy: + return False, False, "--credential is not yet supported for KeeperDB Proxy tunnels." + + return is_keeperdb_proxy, is_keeperrdp_proxy, None + + +def _pam_allow_supply_user(pam_settings_value): + """Read pamSettings.connection.allowSupplyUser — the checkbox that gates + --credential's userSupplied override (and, on the connections side, the + equivalent -cr/--credential flag).""" + if not isinstance(pam_settings_value, dict): + return False + return bool((pam_settings_value.get('connection') or {}).get('allowSupplyUser')) + + +def _resolve_credential_record(params, token): + """Resolve a --credential/-cr token (UID, path, or title) to a record UID. + + Lightweight resolver for `pam tunnel start --credential`: supports UID and + path/title lookup across any record type. Unlike `pam launch`'s resolver, + it has no substring fallback and no interactive multi-match picker — + ambiguous or unmatched input simply returns None so the caller can print a + single clear error. + """ + if not token: + return None + token = token.strip() + + if token in params.record_cache: + return token + + rs = try_resolve_path(params, token) + if rs is not None: + folder, name = rs + if folder is not None and name: + folder_uid = folder.uid or '' + for uid in (records_in_folder(params, folder_uid) or []): + record = vault.KeeperRecord.load(params, uid) + if record and record.title and record.title.lower() == name.lower(): + return uid + + token_lower = token.lower() + matches = [] + for uid in params.record_cache: + record = vault.KeeperRecord.load(params, uid) + if record and record.title and record.title.lower() == token_lower: + matches.append(uid) + if len(matches) == 1: + return matches[0] + return None + + # Group Commands class PAMTunnelCommand(GroupCommand): @@ -574,12 +663,16 @@ def execute(self, params, **kwargs): f'{bcolors.FAIL}--keeper-db-proxy is only supported for pamDatabase records. ' f'Record "{record_name}" is of type "{record_type}".{bcolors.ENDC}') if keeper_db_proxy == 'on' and not tmp_dag.check_if_resource_has_launch_credential(record_uid): - raise CommandError('', - f'{bcolors.FAIL}No Launch Credentials assigned to record "{record_uid}". ' - f'Please assign launch credentials to the record before enabling ' - f'the database proxy.\n' - f'Use: {bcolors.OKBLUE}pam connection edit ' - f'--launch-user (-lu) {bcolors.ENDC}') + if not _resource_has_ephemeral_jit(params, record_uid): + raise CommandError('', + f'{bcolors.FAIL}No Launch Credentials assigned to record "{record_uid}". ' + f'Please assign launch credentials to the record before enabling ' + f'the database proxy.\n' + f'Use: {bcolors.OKBLUE}pam connection edit ' + f'--launch-user (-lu) {bcolors.ENDC}') + logging.debug(f"No linked launch credential found for record {record_uid}; " + f"ephemeral JIT is enabled, proceeding without a static " + f"launch credential") if not pam_settings: pam_settings = vault.TypedField.new_field('pamSettings', {"connection": {}, "portForward": {}}, "") record.custom.append(pam_settings) @@ -642,12 +735,16 @@ def execute(self, params, **kwargs): dirty = True else: if keeper_proxy == 'on' and not tmp_dag.check_if_resource_has_launch_credential(record_uid): - raise CommandError('', - f'{bcolors.FAIL}No Launch Credentials assigned to record "{record_uid}". ' - f'Please assign launch credentials to the record before enabling ' - f'the proxy.\n' - f'Use: {bcolors.OKBLUE}pam connection edit ' - f'--launch-user (-lu) {bcolors.ENDC}') + if not _resource_has_ephemeral_jit(params, record_uid): + raise CommandError('', + f'{bcolors.FAIL}No Launch Credentials assigned to record "{record_uid}". ' + f'Please assign launch credentials to the record before enabling ' + f'the proxy.\n' + f'Use: {bcolors.OKBLUE}pam connection edit ' + f'--launch-user (-lu) {bcolors.ENDC}') + logging.debug(f"No linked launch credential found for record {record_uid}; " + f"ephemeral JIT is enabled, proceeding without a static " + f"launch credential") if not pam_settings: pam_settings = vault.TypedField.new_field('pamSettings', {"connection": {}, "portForward": {}}, "") record.custom.append(pam_settings) @@ -704,6 +801,12 @@ class PAMTunnelStartCommand(Command): help='Activate Keeper Proxy (KeeperDB for pamDatabase, KeeperRDP for ' 'pamMachine + RDP): the gateway substitutes credentials from your ' 'Keeper vault when the local client connects to the tunnel.') + pam_cmd_parser.add_argument('--credential', '-cr', required=False, dest='launch_credential', type=str, + help='Record (UID, path, or title) to use for KeeperRDP Proxy tunnel ' + 'credentials, overriding the resource\'s linked launch credential or ' + 'ephemeral-JIT setting. Requires "Allow users to select credentials ' + 'from their vault" (allowSupplyUser) on the record. Not supported for ' + 'plain (non-proxied) tunnels.') pam_cmd_parser.add_argument('--reason', '-r', required=False, dest='workflow_reason', type=str, help='Justification text for workflow access request. Used when the record\'s ' 'workflow requires a reason; non-interactive equivalent of the inline prompt.') @@ -957,21 +1060,29 @@ def execute(self, params, **kwargs): # presence + launch-credential preflight) and the post-start # banner. is_keeper_proxy = bool(kwargs.get('proxy')) + # --credential/-cr: explicit credential override for a KeeperRDP Proxy + # tunnel. Valid only when the target actually resolves to a KeeperRDP + # Proxy tunnel (allowKeeperRDPProxy on a pamMachine+RDP record) — that's + # the only backend that authenticates against the target, so it's the + # only one an explicit credential can apply to. Detected independent of + # --proxy: the gateway auto-routes to KeeperRDP Proxy from the record's + # own allowKeeperRDPProxy setting regardless of whether --proxy was + # passed on this invocation (same as --target-host/allowSupplyHost + # today), so --credential alone is enough to opt in. + launch_credential = kwargs.get('launch_credential') is_keeperdb_proxy = False is_keeperrdp_proxy = False db_type_for_banner = None - if is_keeper_proxy: + credential_type_override = None + credential_data_override = None + if is_keeper_proxy or launch_credential: record_type = record.record_type protocol = self._resolve_connection_protocol(pam_settings_value) - if record_type == 'pamDatabase': - is_keeperdb_proxy = True - elif record_type == 'pamMachine' and protocol == 'rdp': - is_keeperrdp_proxy = True - else: - detail = f' (protocol="{protocol}")' if record_type == 'pamMachine' else '' - print(f"{bcolors.FAIL}--proxy is supported on pamDatabase or pamMachine+RDP records. " - f"Record {record_uid} is of type \"{record_type}\"{detail}.{bcolors.ENDC}") + is_keeperdb_proxy, is_keeperrdp_proxy, proxy_target_error = _classify_proxy_target( + record_uid, record_type, protocol, is_keeper_proxy, launch_credential) + if proxy_target_error: + print(f"{bcolors.FAIL}{proxy_target_error}{bcolors.ENDC}") return if is_keeperdb_proxy: @@ -992,12 +1103,16 @@ def execute(self, params, **kwargs): _existing_cfg = get_config_uid(params, _est, _ett, record_uid) _proxy_dag = TunnelDAG(params, _est, _ett, _existing_cfg, transmission_key=_tk) if not _proxy_dag.check_if_resource_has_launch_credential(record_uid): - print(f"{bcolors.FAIL}No Launch Credentials assigned to record \"{record_uid}\". " - f"Please assign launch credentials before using --proxy.{bcolors.ENDC}") - print(f"{bcolors.WARNING}Use: " - f"{bcolors.OKBLUE}pam connection edit --launch-user (-lu) " - f"{bcolors.ENDC}") - return + if not _resource_has_ephemeral_jit(params, record_uid): + print(f"{bcolors.FAIL}No Launch Credentials assigned to record \"{record_uid}\". " + f"Please assign launch credentials before using --proxy.{bcolors.ENDC}") + print(f"{bcolors.WARNING}Use: " + f"{bcolors.OKBLUE}pam connection edit --launch-user (-lu) " + f"{bcolors.ENDC}") + return + logging.debug(f"No linked launch credential found for record {record_uid}; " + f"ephemeral JIT is enabled, proceeding without a static " + f"launch credential") db_type_for_banner = self._resolve_database_type(record, pam_settings_value) else: # KeeperRDP Proxy: validate the record flag set via @@ -1021,16 +1136,46 @@ def execute(self, params, **kwargs): # into tunnel_params via _process_user_record, so failing # without one yields the cryptic # `gateway_webrtcaction_missing_rdp_credentials`. - _est, _ett, _tk = get_keeper_tokens(params) - _existing_cfg = get_config_uid(params, _est, _ett, record_uid) - _proxy_dag = TunnelDAG(params, _est, _ett, _existing_cfg, transmission_key=_tk) - if not _proxy_dag.check_if_resource_has_launch_credential(record_uid): - print(f"{bcolors.FAIL}No Launch Credentials assigned to record \"{record_uid}\". " - f"Please assign launch credentials before using --proxy.{bcolors.ENDC}") - print(f"{bcolors.WARNING}Use: " - f"{bcolors.OKBLUE}pam connection edit --launch-user (-lu) " - f"{bcolors.ENDC}") - return + # --credential supplies its own credential and bypasses this + # requirement entirely — independent of the ephemeral fallback + # above (see plan Section 2.1 NB2). + if not launch_credential: + _est, _ett, _tk = get_keeper_tokens(params) + _existing_cfg = get_config_uid(params, _est, _ett, record_uid) + _proxy_dag = TunnelDAG(params, _est, _ett, _existing_cfg, transmission_key=_tk) + if not _proxy_dag.check_if_resource_has_launch_credential(record_uid): + if not _resource_has_ephemeral_jit(params, record_uid): + print(f"{bcolors.FAIL}No Launch Credentials assigned to record \"{record_uid}\". " + f"Please assign launch credentials before using --proxy.{bcolors.ENDC}") + print(f"{bcolors.WARNING}Use: " + f"{bcolors.OKBLUE}pam connection edit --launch-user (-lu) " + f"{bcolors.ENDC}") + return + logging.debug(f"No linked launch credential found for record {record_uid}; " + f"ephemeral JIT is enabled, proceeding without a static " + f"launch credential") + + if launch_credential: + if not _pam_allow_supply_user(pam_settings_value): + print(f"{bcolors.FAIL}--credential requires the \"Allow users to select " + f"credentials from their vault\" option to be enabled on " + f"record \"{record_uid}\".{bcolors.ENDC}") + print(f"{bcolors.WARNING}Enable it with: {bcolors.OKBLUE}record-update " + f"-r {record_uid} pamSettings=$JSON:" + f'{{"connection":{{"allowSupplyUser":true}}}}{bcolors.ENDC}') + return + credential_record_uid = _resolve_credential_record(params, launch_credential) + if not credential_record_uid: + print(f"{bcolors.FAIL}--credential record \"{launch_credential}\" not found.{bcolors.ENDC}") + return + credential_fields = _extract_user_record_credentials(params, credential_record_uid) + credential_type_override = 'userSupplied' + credential_data_override = { + 'username': credential_fields.get('username', ''), + 'password': credential_fields.get('password', ''), + } + logging.debug(f"Using --credential override for tunnel start on record " + f"{record_uid}: credential record={credential_record_uid}") # Get target host and port if allow_supply_host: @@ -1232,6 +1377,16 @@ def execute(self, params, **kwargs): f"WebRTC cleanup is best-effort.{bcolors.ENDC}") return + # Ephemeral-JIT account creation on the gateway takes 30-90s (mirrors + # the connection path's timeout bump in terminal_connection.py) — only + # relevant for proxy tunnels, and only when no explicit --credential + # override is in play (an override skips ephemeral entirely). + is_ephemeral = ( + (is_keeperdb_proxy or is_keeperrdp_proxy) + and not credential_type_override + and _resource_has_ephemeral_jit(params, record_uid) + ) + # When `allow_ephemeral_fallback` is on we're about to attempt # the legacy default port and the pre-probe said it was free. # In the rare TOCTOU window the bind can still race and lose, @@ -1250,6 +1405,9 @@ def execute(self, params, **kwargs): target_host, target_port, socks, trickle_ice, record.title, allow_supply_host=allow_supply_host, two_factor_value=two_factor_value, + credential_type=credential_type_override, + credential_data=credential_data_override, + is_ephemeral=is_ephemeral, ) finally: logging.disable(logging.NOTSET) @@ -1259,6 +1417,9 @@ def execute(self, params, **kwargs): target_host, target_port, socks, trickle_ice, record.title, allow_supply_host=allow_supply_host, two_factor_value=two_factor_value, + credential_type=credential_type_override, + credential_data=credential_data_override, + is_ephemeral=is_ephemeral, ) # No-`--port` legacy default fallback. If Rust failed to bind the @@ -1299,6 +1460,9 @@ def execute(self, params, **kwargs): target_host, target_port, socks, trickle_ice, record.title, allow_supply_host=allow_supply_host, two_factor_value=two_factor_value, + credential_type=credential_type_override, + credential_data=credential_data_override, + is_ephemeral=is_ephemeral, ) if result and result.get("success"): diff --git a/unit-tests/pam/test_pam_tunnel_ephemeral_credential.py b/unit-tests/pam/test_pam_tunnel_ephemeral_credential.py new file mode 100644 index 000000000..496f991ef --- /dev/null +++ b/unit-tests/pam/test_pam_tunnel_ephemeral_credential.py @@ -0,0 +1,200 @@ +""" +Unit tests for the KC (Commander) half of the ephemeral/--credential proxy +tunnel plan: + +- Section 2.1: launch-credential preflight falls back to ephemeral JIT. +- Section 2.2: gateway timeout bump for ephemeral tunnel starts. +- Section 3.2: --credential/-cr proxy-target classification, allowSupplyUser + gate, and record resolution for `pam tunnel start`. + +These test the small, pure/near-pure functions extracted from +tunnel_and_connections.py / tunnel_helpers.py rather than the full CLI +command (which would require mocking KeeperParams/TunnelDAG/gateway network +calls not otherwise exercised by this repo's test suite). +""" + +import unittest +from unittest import mock + +from keepercommander.commands.tunnel_and_connections import ( + _classify_proxy_target, + _pam_allow_supply_user, + _resolve_credential_record, + _resource_has_ephemeral_jit, +) +from keepercommander.commands.tunnel.port_forward.tunnel_helpers import ( + credential_override_data_fields, + credential_override_input_fields, + ephemeral_gateway_timeout_ms, +) + + +class TestClassifyProxyTarget(unittest.TestCase): + def test_database_record_is_keeperdb_proxy(self): + is_db, is_rdp, error = _classify_proxy_target('rec-uid', 'pamDatabase', '', True, None) + self.assertTrue(is_db) + self.assertFalse(is_rdp) + self.assertIsNone(error) + + def test_machine_rdp_record_is_keeperrdp_proxy(self): + is_db, is_rdp, error = _classify_proxy_target('rec-uid', 'pamMachine', 'rdp', True, None) + self.assertFalse(is_db) + self.assertTrue(is_rdp) + self.assertIsNone(error) + + def test_unsupported_type_with_proxy_flag_gives_proxy_message(self): + is_db, is_rdp, error = _classify_proxy_target('rec-uid', 'pamMachine', 'ssh', True, None) + self.assertFalse(is_db) + self.assertFalse(is_rdp) + self.assertIn('--proxy is supported on', error) + + def test_unsupported_type_with_credential_only_gives_plain_tunnel_message(self): + # --credential with no --proxy, on a record that isn't proxy-capable + # at all: distinct message from the --proxy-driven one above. + is_db, is_rdp, error = _classify_proxy_target('rec-uid', 'pamMachine', 'ssh', False, 'some-record') + self.assertFalse(is_db) + self.assertFalse(is_rdp) + self.assertEqual(error, "--credential requires a proxied tunnel; plain tunnels don't authenticate.") + + def test_credential_on_database_record_is_rejected(self): + # KeeperDB Proxy keeps "ephemeral always wins, no override" — reject + # --credential outright instead of accepting a no-op flag. + is_db, is_rdp, error = _classify_proxy_target('rec-uid', 'pamDatabase', '', True, 'some-record') + self.assertFalse(is_db) + self.assertFalse(is_rdp) + self.assertEqual(error, '--credential is not yet supported for KeeperDB Proxy tunnels.') + + def test_credential_on_database_record_rejected_even_without_proxy_flag(self): + is_db, is_rdp, error = _classify_proxy_target('rec-uid', 'pamDatabase', '', False, 'some-record') + self.assertFalse(is_db) + self.assertFalse(is_rdp) + self.assertEqual(error, '--credential is not yet supported for KeeperDB Proxy tunnels.') + + def test_credential_alone_detects_rdp_proxy_without_proxy_flag(self): + # --credential alone (no --proxy) still resolves to KeeperRDP Proxy + # when the record is a pamMachine+RDP record — the gateway auto-routes + # from the record's own settings regardless of the client-side flag. + is_db, is_rdp, error = _classify_proxy_target('rec-uid', 'pamMachine', 'rdp', False, 'some-record') + self.assertFalse(is_db) + self.assertTrue(is_rdp) + self.assertIsNone(error) + + +class TestPamAllowSupplyUser(unittest.TestCase): + def test_true_when_set(self): + self.assertTrue(_pam_allow_supply_user({'connection': {'allowSupplyUser': True}})) + + def test_false_when_absent(self): + self.assertFalse(_pam_allow_supply_user({'connection': {}})) + + def test_false_when_connection_missing(self): + self.assertFalse(_pam_allow_supply_user({})) + + def test_false_when_not_a_dict(self): + self.assertFalse(_pam_allow_supply_user(None)) + + +class TestResourceHasEphemeralJit(unittest.TestCase): + @mock.patch('keepercommander.commands.tunnel_and_connections.get_resource_jit_settings') + def test_true_when_create_ephemeral_set(self, mock_get_jit): + mock_get_jit.return_value = {'createEphemeral': True} + self.assertTrue(_resource_has_ephemeral_jit(object(), 'uid1')) + + @mock.patch('keepercommander.commands.tunnel_and_connections.get_resource_jit_settings') + def test_false_when_create_ephemeral_false(self, mock_get_jit): + mock_get_jit.return_value = {'createEphemeral': False} + self.assertFalse(_resource_has_ephemeral_jit(object(), 'uid1')) + + @mock.patch('keepercommander.commands.tunnel_and_connections.get_resource_jit_settings') + def test_false_when_no_jit_settings(self, mock_get_jit): + mock_get_jit.return_value = None + self.assertFalse(_resource_has_ephemeral_jit(object(), 'uid1')) + + @mock.patch('keepercommander.commands.tunnel_and_connections.get_resource_jit_settings') + def test_false_on_read_error(self, mock_get_jit): + mock_get_jit.side_effect = Exception('DAG unavailable') + self.assertFalse(_resource_has_ephemeral_jit(object(), 'uid1')) + + +class _FakeRecord: + def __init__(self, title): + self.title = title + + +class _FakeParams: + def __init__(self, record_cache): + self.record_cache = record_cache + + +class TestResolveCredentialRecord(unittest.TestCase): + def test_uid_hit_returns_uid_directly(self): + params = _FakeParams({'abc123': {}}) + self.assertEqual(_resolve_credential_record(params, 'abc123'), 'abc123') + + def test_empty_token_returns_none(self): + params = _FakeParams({}) + self.assertIsNone(_resolve_credential_record(params, '')) + self.assertIsNone(_resolve_credential_record(params, None)) + + @mock.patch('keepercommander.commands.tunnel_and_connections.try_resolve_path', return_value=None) + @mock.patch('keepercommander.vault.KeeperRecord.load') + def test_exact_title_match_returns_uid(self, mock_load, mock_resolve_path): + params = _FakeParams({'uid-1': {}, 'uid-2': {}}) + + def load_side_effect(_params, uid): + return {'uid-1': _FakeRecord('My Server'), 'uid-2': _FakeRecord('Other Record')}[uid] + + mock_load.side_effect = load_side_effect + self.assertEqual(_resolve_credential_record(params, 'My Server'), 'uid-1') + + @mock.patch('keepercommander.commands.tunnel_and_connections.try_resolve_path', return_value=None) + @mock.patch('keepercommander.vault.KeeperRecord.load') + def test_ambiguous_title_returns_none(self, mock_load, mock_resolve_path): + params = _FakeParams({'uid-1': {}, 'uid-2': {}}) + mock_load.return_value = _FakeRecord('Duplicate Title') + self.assertIsNone(_resolve_credential_record(params, 'Duplicate Title')) + + @mock.patch('keepercommander.commands.tunnel_and_connections.try_resolve_path', return_value=None) + @mock.patch('keepercommander.vault.KeeperRecord.load', return_value=None) + def test_no_match_returns_none(self, mock_load, mock_resolve_path): + params = _FakeParams({'uid-1': {}}) + self.assertIsNone(_resolve_credential_record(params, 'Nonexistent')) + + +class TestEphemeralGatewayTimeoutMs(unittest.TestCase): + def test_default_when_not_ephemeral(self): + self.assertEqual(ephemeral_gateway_timeout_ms(False, default=30000), 30000) + + def test_bumped_when_ephemeral(self): + self.assertEqual(ephemeral_gateway_timeout_ms(True, default=30000), 120000) + + @mock.patch.dict('os.environ', {'PAM_GATEWAY_OFFER_TIMEOUT_EPHEMERAL_MS': '45000'}) + def test_bump_respects_env_override(self): + self.assertEqual(ephemeral_gateway_timeout_ms(True, default=30000), 45000) + + +class TestCredentialOverridePayloadFields(unittest.TestCase): + def test_data_fields_for_user_supplied(self): + fields = credential_override_data_fields('userSupplied', {'username': 'bob', 'password': 'secret'}) + self.assertEqual(fields, {'username': 'bob', 'password': 'secret'}) + + def test_data_fields_empty_when_not_user_supplied(self): + self.assertEqual(credential_override_data_fields(None, {'username': 'bob', 'password': 'secret'}), {}) + self.assertEqual(credential_override_data_fields('linked', {'username': 'bob', 'password': 'secret'}), {}) + + def test_data_fields_empty_when_no_credential_data(self): + self.assertEqual(credential_override_data_fields('userSupplied', None), {}) + + def test_input_fields_for_user_supplied(self): + self.assertEqual( + credential_override_input_fields('userSupplied'), + {'credentialType': 'userSupplied', 'allowSupplyUser': True}, + ) + + def test_input_fields_empty_when_not_user_supplied(self): + self.assertEqual(credential_override_input_fields(None), {}) + self.assertEqual(credential_override_input_fields('linked'), {}) + + +if __name__ == '__main__': + unittest.main()