diff --git a/keepercommander/commands/pam_import/cyberark_import.py b/keepercommander/commands/pam_import/cyberark_import.py
index 494ca5c41..243aaa920 100644
--- a/keepercommander/commands/pam_import/cyberark_import.py
+++ b/keepercommander/commands/pam_import/cyberark_import.py
@@ -47,6 +47,7 @@
exclude_system_safes,
resolve_account_dependents,
resolve_linked_accounts,
+ PAM_SERVICE_ADD_TYPES,
pick_admin_credentials,
pick_launch_credentials,
detect_dual_account,
@@ -705,7 +706,7 @@ def _collect_dependents(self, account: dict, record: dict,
f" • {_esc(dep.get('service_name', '') or '?')} "
f"({_esc(dep.get('raw_type', '') or 'unknown')}) "
f"on {_esc(dep.get('machine_address', '') or '?')} "
- f"→ pam action service {_esc(mapped_to)}"))
+ f"→ pam action service add --type {_esc(mapped_to)}"))
def _apply_folder_paths(self, record: dict, safe_name: str,
folder_mapper: SafeFolderMapper):
@@ -1504,17 +1505,7 @@ def _apply_service_dependent_mappings(
self, mapped: MappedImportResult, project_result: dict,
unmapped_items: list[dict],
) -> Optional[dict]:
- """Replay CyberArk dependents as KeeperPAM service-account mappings.
-
- Iterates over ``mapped.dependents`` collected during the mapping phase
- and invokes ``PAMActionServiceAddCommand`` once per (machine, user, type)
- tuple that resolves to imported records. Categories with no Keeper
- equivalent, missing host machines, and non-Windows OS hosts are all
- skipped silently and accounted for in the returned summary so the
- import report can surface them.
-
- Returns a summary dict, or ``None`` when nothing to do.
- """
+ """Replay CyberArk dependents as KeeperPAM service-account mappings."""
opts = self.options
if opts.skip_dependents or not mapped.dependents:
return None
@@ -1595,18 +1586,33 @@ def _apply_service_dependent_mappings(
add_cmd = PAMActionServiceAddCommand()
for dep in mapped.dependents:
+ # Match pam action service add --type {service,task,iis_pool}
service_type = dep.get("service_type")
- if service_type not in ("service", "task", "iis"):
+ if service_type == "iis":
+ service_type = "iis_pool"
+ if service_type not in PAM_SERVICE_ADD_TYPES:
summary["skipped_unsupported"] += 1
continue
+ service_name = (dep.get("service_name") or "").strip()
+ if not service_name:
+ summary["skipped_other"] += 1
+ summary["details"].append({
+ "service": "",
+ "host": dep.get("machine_address", ""),
+ "type": dep.get("raw_type", ""),
+ "reason": "pam action service add requires --name "
+ "(Windows service / task / IIS pool name)",
+ })
+ continue
+
machine_record = self._find_machine_record(
dep.get("machine_address", ""), machine_index,
)
if machine_record is None:
summary["skipped_missing_machine"] += 1
summary["details"].append({
- "service": dep.get("service_name", ""),
+ "service": service_name,
"host": dep.get("machine_address", ""),
"type": dep.get("raw_type", ""),
"reason": "no PAM Machine record imported for this host",
@@ -1617,18 +1623,20 @@ def _apply_service_dependent_mappings(
summary["skipped_non_windows"] += 1
unmapped_items.append({
"category": "CyberArk dependent",
- "item": (f"{dep.get('service_name') or dep.get('raw_type')} "
+ "item": (f"{service_name or dep.get('raw_type')} "
f"on {dep.get('machine_address')}"),
"action": "Host is not Windows — Keeper PAM can only rotate "
"Windows service / task / IIS credentials",
})
continue
- user_record = user_index.get(dep.get("master_user_title", ""))
+ user_record = user_index.get(
+ (dep.get("master_user_title") or "").casefold(),
+ )
if user_record is None:
summary["skipped_missing_user"] += 1
summary["details"].append({
- "service": dep.get("service_name", ""),
+ "service": service_name,
"host": dep.get("machine_address", ""),
"type": dep.get("raw_type", ""),
"reason": "PAM User record not found in vault after import",
@@ -1636,27 +1644,35 @@ def _apply_service_dependent_mappings(
continue
try:
+ # execute() reads argparse dest names, not CLI flag names:
+ # --type → service_type, --name → name, --machine-uid → machine_uid
add_cmd.execute(
self.params,
gateway=gateway_uid,
configuration_uid=gateway_context.configuration.record_uid,
machine_uid=machine_record.record_uid,
user_uid=user_record.record_uid,
- type=service_type,
+ service_type=service_type,
+ name=service_name,
)
summary["added"] += 1
except Exception as e: # noqa: BLE001 — never block reporting
summary["skipped_other"] += 1
+ err_text = str(e).strip() or type(e).__name__
logging.warning(
- "Failed to register %s mapping for %s on %s: %s",
- service_type, dep.get("service_name", "?"),
+ "Failed to register %s mapping for %s on %s: %s: %s",
+ service_type, service_name,
dep.get("machine_address", "?"), type(e).__name__,
+ err_text,
)
summary["details"].append({
- "service": dep.get("service_name", ""),
+ "service": service_name,
"host": dep.get("machine_address", ""),
"type": dep.get("raw_type", ""),
- "reason": f"pam action service add failed: {type(e).__name__}",
+ "reason": (
+ f"pam action service add failed: {type(e).__name__}: "
+ f"{err_text.splitlines()[0]}"
+ ),
})
return summary
@@ -1836,6 +1852,17 @@ def _attach_report_files(self, notes_text: str, report_config_uid: str,
pass
+# Unicode dashes that users commonly paste instead of ASCII '-'.
+_DASH_TRANSLATE = str.maketrans({
+ '\u2010': '-', # hyphen
+ '\u2011': '-', # non-breaking hyphen
+ '\u2012': '-', # figure dash
+ '\u2013': '-', # en dash
+ '\u2014': '-', # em dash
+ '\u2212': '-', # minus
+})
+
+
class CyberArkPAMImportCommand(Command):
parser = argparse.ArgumentParser(
prog="pam project cyberark-import",
@@ -2202,12 +2229,38 @@ def _list_safes_detailed(safes: list[dict], system_excluded: int):
print()
@staticmethod
- def _interactive_safe_picker(safes: list[dict]) -> Optional[str]:
- """Show safes and let user select which to import.
+ def _parse_index_selection(choice: str, count: int) -> list[int]:
+ """Parse a selection string into 0-based indices."""
+ selected: list[int] = []
+ seen: set[int] = set()
+ if count <= 0:
+ return selected
+ token_re = re.compile(r'^(\d+)(?:-(\d+))?$')
+ for part in choice.split(','):
+ part = part.strip().translate(_DASH_TRANSLATE)
+ part = re.sub(r'\s+', '', part)
+ if not part:
+ continue
+ m = token_re.match(part)
+ if not m:
+ continue
+ start = int(m.group(1))
+ end = int(m.group(2)) if m.group(2) is not None else start
+ lo, hi = min(start, end), max(start, end)
+ lo = max(1, lo)
+ hi = min(count, hi)
+ if lo > hi:
+ continue
+ for num in range(lo, hi + 1):
+ idx = num - 1
+ if idx not in seen:
+ selected.append(idx)
+ seen.add(idx)
+ return selected
- Returns comma-separated safe names for apply_safe_filter,
- or None to import all.
- """
+ @staticmethod
+ def _interactive_safe_picker(safes: list[dict]) -> Optional[str]:
+ """Show safes and let user select which to import."""
print(f'\n{bcolors.OKBLUE}CyberArk Safes Found:{bcolors.ENDC}')
print('─' * 50)
numbered = []
@@ -2219,25 +2272,25 @@ def _interactive_safe_picker(safes: list[dict]) -> Optional[str]:
print()
try:
- choice = input(f' Select safes (comma-separated numbers, or A for all) [A]: ').strip()
+ choice = input(
+ ' Select safes (numbers/ranges e.g. 1-4,6,8-10, or A for all) [A]: '
+ ).strip()
except EOFError:
return None
if not choice or choice.upper() == 'A':
return None
- selected = []
- for part in choice.split(','):
- part = part.strip()
- try:
- idx = int(part) - 1
- if 0 <= idx < len(numbered):
- selected.append(numbered[idx])
- except ValueError:
- continue
+ selected = [
+ numbered[idx]
+ for idx in CyberArkPAMImportCommand._parse_index_selection(choice, len(numbered))
+ ]
if not selected:
- return None
+ print(
+ f'{bcolors.FAIL}No valid indexes in {choice!r}. Import cancelled.{bcolors.ENDC}'
+ )
+ return ''
logging.warning('Selected safes: %s', ', '.join(selected))
return ','.join(selected)
diff --git a/keepercommander/importer/commands.py b/keepercommander/importer/commands.py
index fef8a3b83..d2a19398a 100644
--- a/keepercommander/importer/commands.py
+++ b/keepercommander/importer/commands.py
@@ -85,6 +85,8 @@ def register_command_info(aliases, command_info):
help='Display skipped records')
import_parser.add_argument('--secret-ids', dest='secret_ids', action='store',
help='Comma separated list of secret IDs to fetch (Thycotic)')
+import_parser.add_argument('--target-node', '--node', dest='target_node', action='store',
+ help='node name or ID for CyberArk-provisioned users, teams, and roles (default: root node)')
import_parser.add_argument(
'name', type=str,
help='file name (json, csv , keepass, 1password), account name (lastpass), or URL (ManageEngine, Thycotic). '
@@ -285,6 +287,10 @@ def execute(self, params, **kwargs):
logging.warning(f'Record type "{record_type}" not found.')
return
+ if kwargs.get('target_node') and import_format != 'cyberark':
+ logging.warning('--target-node/--node is only used with --format=cyberark; ignoring')
+ kwargs['target_node'] = None
+
logging.info('Processing... please wait.')
imp_exp._import(params, import_format, import_name, manage_users=manage_users, manage_records=manage_records,
can_edit=can_edit, can_share=can_share, **kwargs)
diff --git a/keepercommander/importer/cyberark/cyberark.py b/keepercommander/importer/cyberark/cyberark.py
index 2620d17ce..4b03d6a09 100644
--- a/keepercommander/importer/cyberark/cyberark.py
+++ b/keepercommander/importer/cyberark/cyberark.py
@@ -22,6 +22,7 @@
from ... import api, crypto, utils
from ...commands.enterprise_common import EnterpriseCommand
from ...constants import EMAIL_PATTERN
+from .pam.ui import _esc
from ..importer import (
BaseDownloadMembership,
BaseImporter,
@@ -978,6 +979,20 @@ def _do_import_inner(self, filename, **kwargs):
return
pvwa_host, authorization_token, query_params = auth
+ params = kwargs.get("params")
+ will_teams = environ.get("_CYBERARK_SKIP_TEAMS", "").lower() not in ("1", "true", "yes")
+ will_create_users = environ.get("_CYBERARK_SKIP_CREATE_USERS", "").lower() not in ("1", "true", "yes")
+ will_print_users = environ.get("_CYBERARK_SKIP_USERS_LIST", "").lower() not in ("1", "true", "yes")
+ target_node = kwargs.get("target_node")
+ if target_node is not None:
+ target_node = str(target_node).strip() or None
+
+ provision_node_id = None
+ if will_teams:
+ provision_node_id = self._resolve_provisioning_node_id(params, target_node)
+ if target_node and provision_node_id is None:
+ return
+
safes = self._resolve_safes(pvwa_host, authorization_token)
if not safes:
return
@@ -1013,11 +1028,6 @@ def _do_import_inner(self, filename, **kwargs):
continue
safe_accounts[safe] = accounts
- params = kwargs.get("params")
- will_teams = environ.get("_CYBERARK_SKIP_TEAMS", "").lower() not in ("1", "true", "yes")
- will_create_users = environ.get("_CYBERARK_SKIP_CREATE_USERS", "").lower() not in ("1", "true", "yes")
- will_print_users = environ.get("_CYBERARK_SKIP_USERS_LIST", "").lower() not in ("1", "true", "yes")
-
# Gather the CyberArk identities (groups + users) that will become Keeper
# teams, roles and users so they can be previewed before the import. The
# fetched users are reused after the import (no second fetch).
@@ -1090,6 +1100,21 @@ def _do_import_inner(self, filename, **kwargs):
summary_lines.append(f" - {len(group_names)} user group(s) as Keeper teams and roles")
if eligible_users:
summary_lines.append(f" - {len(eligible_users)} user(s) provisioned as Keeper users")
+ if will_teams and provision_node_id is not None:
+ if target_node:
+ summary_lines.append(
+ f' - provision teams, roles, and users into node {_esc(target_node)} '
+ f'(id {provision_node_id})'
+ )
+ else:
+ summary_lines.append(
+ f' - provision teams, roles, and users into the default root node '
+ f'(id {provision_node_id})'
+ )
+ elif will_teams:
+ summary_lines.append(
+ ' - teams/roles/users will be skipped (no provisioning node)'
+ )
if not self._confirm_import(pvwa_host, summary="\n".join(summary_lines)):
print_formatted_text(HTML("\nImport cancelled by user"))
return
@@ -1210,15 +1235,88 @@ def _do_import_inner(self, filename, **kwargs):
# Import CyberArk User Groups as Keeper Enterprise Teams + Roles, then optionally
# create Keeper users (using their real CyberArk business emails) and
# assign them to the matching Keeper Roles.
- if will_teams:
+ if will_teams and provision_node_id is not None:
self.import_user_groups(
pvwa_host, authorization_token, params,
cyberark_users=cyberark_users,
+ target_node=target_node,
+ node_id=provision_node_id,
)
print_formatted_text(HTML("\nImport completed"))
- def import_user_groups(self, pvwa_host, authorization_token, params, cyberark_users=None):
+ def _resolve_provisioning_node_id(self, params, target_node=None):
+ """Resolve the enterprise node for CyberArk teams/roles/users.
+
+ If ``target_node`` is set (name or numeric ID), resolve it via
+ ``EnterpriseCommand.resolve_nodes``. Otherwise fall back to the first
+ user-root node (same default as ``enterprise-user --add`` / ``enterprise-team --add``).
+
+ Returns the node id, or ``None`` if resolution fails (errors are printed).
+ """
+ if target_node is not None:
+ target_node = str(target_node).strip() or None
+
+ if not params or not getattr(params, "enterprise", None):
+ if params is None:
+ msg = (
+ "Cannot create Keeper Teams: Keeper session is not "
+ "available to the importer (no params)."
+ )
+ else:
+ msg = (
+ "Cannot create Keeper Teams/users: the logged-in account "
+ "is not an enterprise admin (no enterprise data loaded)."
+ )
+ print_formatted_text(HTML(msg))
+ return None
+
+ if target_node:
+ try:
+ nodes = list(EnterpriseCommand.resolve_nodes(params, target_node))
+ except (KeyError, TypeError):
+ nodes = []
+ if len(nodes) == 0:
+ print_formatted_text(
+ HTML(
+ f"Cannot provision into node: "
+ f'node "{_esc(target_node)}" was not found.'
+ )
+ )
+ return None
+ if len(nodes) > 1:
+ print_formatted_text(
+ HTML(
+ f"Cannot provision into node: "
+ f'more than one node matches "{_esc(target_node)}". '
+ f"Use the numeric node ID."
+ )
+ )
+ return None
+ return nodes[0]["node_id"]
+
+ # Default: first user-root node (loads managed nodes if needed), then
+ # the first tree root (parent_id unset/0).
+ try:
+ root_nodes = list(EnterpriseCommand.get_user_root_nodes(params))
+ except Exception as e:
+ logging.debug("Failed to load user root nodes: %s", e)
+ root_nodes = []
+ if root_nodes:
+ return root_nodes[0]
+ for n in params.enterprise.get("nodes", []) or []:
+ if not n.get("parent_id"):
+ return n["node_id"]
+ print_formatted_text(
+ HTML(
+ "Cannot create Keeper Teams/users: no root node found in the "
+ "enterprise tree."
+ )
+ )
+ return None
+
+ def import_user_groups(self, pvwa_host, authorization_token, params, cyberark_users=None,
+ target_node=None, node_id=None):
"""Fetch CyberArk User Groups and create them as Keeper Enterprise Teams.
This mirrors the ``enterprise-team --add`` command flow: for each
@@ -1304,25 +1402,18 @@ def import_user_groups(self, pvwa_host, authorization_token, params, cyberark_us
existing_team_names.add(team["name"].lower())
# Determine the target node id (same default as enterprise-team --add):
- # the first user-root node when no --node was specified.
- node_id = None
- for nid in params.enterprise.get("user_root_nodes", []) or []:
- node_id = nid
- break
+ # --target-node when specified, otherwise the first user-root node.
if node_id is None:
- # Fall back to the first node in the tree (root has parent_id=0)
- for n in params.enterprise.get("nodes", []) or []:
- if not n.get("parent_id"):
- node_id = n["node_id"]
- break
+ node_id = self._resolve_provisioning_node_id(params, target_node)
if node_id is None:
+ return
+ if target_node:
print_formatted_text(
HTML(
- "Cannot create Keeper Teams: no root node found in the "
- "enterprise tree."
+ f"Provisioning teams, roles, and users into node "
+ f"{_esc(target_node)} (id {node_id})"
)
)
- return
print_formatted_text(
HTML(f"Importing {len(groups)} user groups as Keeper Teams (members not provisioned):\n"),
@@ -1743,24 +1834,10 @@ def _create_keeper_users_and_assign_roles(self, groups, cyberark_users, params,
if uname:
existing_user_by_email[uname] = u
- # Determine the target node (root node) for new invitations.
- invite_node_id = None
- for nid in params.enterprise.get("user_root_nodes", []) or []:
- invite_node_id = nid
- break
- if invite_node_id is None:
- for n in params.enterprise.get("nodes", []) or []:
- if not n.get("parent_id"):
- invite_node_id = n["node_id"]
- break
- if invite_node_id is None:
- print_formatted_text(
- HTML(
- "\nCannot invite Keeper users: no root node found in "
- "the enterprise tree."
- )
- )
+ # Caller already resolved --target-node (or the default root).
+ if node_id is None:
return
+ invite_node_id = node_id
tree_key = params.enterprise.get("unencrypted_tree_key")
if not tree_key:
diff --git a/keepercommander/importer/cyberark/pam/__init__.py b/keepercommander/importer/cyberark/pam/__init__.py
index c34884751..21bbfdf4e 100644
--- a/keepercommander/importer/cyberark/pam/__init__.py
+++ b/keepercommander/importer/cyberark/pam/__init__.py
@@ -39,6 +39,7 @@
validate_import_data,
)
from .dependents import (
+ PAM_SERVICE_ADD_TYPES,
_normalize_dependent_type,
resolve_account_dependents,
)
@@ -91,6 +92,7 @@
"RecordDecision",
"MAX_FETCH_RECORDS",
"MAX_SAFE_NAME_LENGTH",
+ "PAM_SERVICE_ADD_TYPES",
"PermissionMapper",
"RecordKind",
"SafeFolderMapper",
diff --git a/keepercommander/importer/cyberark/pam/dependents.py b/keepercommander/importer/cyberark/pam/dependents.py
index 1b03242f2..7259308f2 100644
--- a/keepercommander/importer/cyberark/pam/dependents.py
+++ b/keepercommander/importer/cyberark/pam/dependents.py
@@ -12,9 +12,9 @@
# scheduled tasks and IIS application pools running on remote hosts. The
# ``/Accounts/{id}/Dependents`` endpoint returns one entry per (host, service,
# type) tuple. KeeperPAM models the same relationship via
-# ``pam action service add`` (machine-uid + user-uid + type), so the importer
-# collects dependents during the mapping phase and replays them as service
-# mappings after the vault import succeeds.
+# ``pam action service add`` (--machine-uid, --user-uid, --type, --name), so
+# the importer collects dependents during the mapping phase and replays them
+# as service mappings after the vault import succeeds.
from __future__ import annotations
@@ -30,11 +30,14 @@
from .client import CyberArkPVWAClient
-# CyberArk dependent ``type`` / ``platformId`` values → Keeper service-mapping
-# verbs accepted by ``PAMActionServiceAddCommand`` (--type service|task|iis).
-# Keys are matched case-insensitively after stripping non-alphanumerics so
-# spellings like ``Windows Service``, ``Win32Service``, ``WinService``, and
-# the Privilege Cloud ``SchedTask`` platformId all resolve.
+# Exact ``pam action service add --type`` choices.
+PAM_SERVICE_ADD_TYPES = frozenset({"service", "task", "iis_pool"})
+
+# CyberArk dependent ``type`` / ``platformId`` values → Keeper ``--type``
+# values (service | task | iis_pool). Keys are matched case-insensitively
+# after stripping non-alphanumerics so spellings like ``Windows Service``,
+# ``Win32Service``, ``WinService``, and Privilege Cloud ``SchedTask`` /
+# ``IISAppPool`` all resolve.
_DEPENDENT_TYPE_ALIASES: Dict[str, str] = {
# Windows services
"windowsservice": "service",
@@ -47,11 +50,13 @@
"windowsscheduledtask": "task",
"schedtask": "task",
"task": "task",
- # IIS application pools
- "iisapppool": "iis",
- "iisapplicationpool": "iis",
- "iisapppools": "iis",
- "iis": "iis",
+ # IIS application pools → Keeper --type iis_pool
+ "iisapppool": "iis_pool",
+ "iisapplicationpool": "iis_pool",
+ "iisapppools": "iis_pool",
+ "iispool": "iis_pool",
+ "iis_pool": "iis_pool",
+ "iis": "iis_pool",
}
@@ -78,10 +83,11 @@ def resolve_account_dependents(client: 'CyberArkPVWAClient',
* ``machine_address`` — host where the service runs (used to find the
Keeper PAM Machine record).
- * ``service_type`` — Keeper verb (service|task|iis) or ``None`` for
- unsupported categories.
+ * ``service_type`` — ``pam action service add --type`` value
+ (service|task|iis_pool) or ``None`` for unsupported categories.
* ``raw_type`` — original CyberArk ``Type`` string (kept for reporting).
- * ``service_name`` — informational, surfaced in the report only.
+ * ``service_name`` — ``pam action service add --name`` (Windows
+ service / scheduled-task / IIS pool name).
* ``master_user_title`` — Keeper title of the pamUser record that holds
the rotated credential (i.e. the user the service runs as).
* ``master_account_id`` / ``master_account_name`` — CyberArk source IDs
diff --git a/keepercommander/importer/imp_exp.py b/keepercommander/importer/imp_exp.py
index 1acb9ed0b..6dcc2cc96 100644
--- a/keepercommander/importer/imp_exp.py
+++ b/keepercommander/importer/imp_exp.py
@@ -750,6 +750,7 @@ def _import(params, file_format, filename, **kwargs):
dry_run = kwargs.get('dry_run') is True
show_skipped = kwargs.get('show_skipped') is True
secret_ids = kwargs.get('secret_ids')
+ target_node = kwargs.get('target_node')
import_into = kwargs.get('import_into') or ''
if import_into:
@@ -771,7 +772,8 @@ def _import(params, file_format, filename, **kwargs):
classic_shared = shared and not use_nsf
for x in importer.execute(filename, params=params, users_only=import_users, filter_folder=filter_folder,
- old_domain=old_domain, new_domain=new_domain, tmpdir=tmpdir, secret_ids=secret_ids, dry_run=dry_run):
+ old_domain=old_domain, new_domain=new_domain, tmpdir=tmpdir, secret_ids=secret_ids,
+ dry_run=dry_run, target_node=target_node):
if isinstance(x, ImportRecord):
if filter_folder and not importer.support_folder_filter():
if not x.folders:
diff --git a/tests/test_cyberark_pam_import.py b/tests/test_cyberark_pam_import.py
index 619e7e646..382870cd9 100644
--- a/tests/test_cyberark_pam_import.py
+++ b/tests/test_cyberark_pam_import.py
@@ -1453,13 +1453,53 @@ def test_select_specific(self):
result = CyberArkPAMImportCommand._interactive_safe_picker(safes)
assert result == "Alpha,Gamma"
+ def test_select_range(self):
+ from keepercommander.commands.pam_import.cyberark_import import CyberArkPAMImportCommand
+ from unittest.mock import patch
+ safes = [{"safeName": f"S{i}"} for i in range(1, 6)]
+ with patch("builtins.input", return_value="2-4"):
+ result = CyberArkPAMImportCommand._interactive_safe_picker(safes)
+ assert result == "S2,S3,S4"
+
+ def test_select_mixed_ranges_and_indexes(self):
+ from keepercommander.commands.pam_import.cyberark_import import CyberArkPAMImportCommand
+ from unittest.mock import patch
+ safes = [{"safeName": f"S{i}"} for i in range(1, 21)]
+ with patch("builtins.input", return_value="1,2,3,6-9,11,14-18"):
+ result = CyberArkPAMImportCommand._interactive_safe_picker(safes)
+ assert result == "S1,S2,S3,S6,S7,S8,S9,S11,S14,S15,S16,S17,S18"
+
+ def test_select_reversed_range(self):
+ from keepercommander.commands.pam_import.cyberark_import import CyberArkPAMImportCommand
+ from unittest.mock import patch
+ safes = [{"safeName": f"S{i}"} for i in range(1, 6)]
+ with patch("builtins.input", return_value="4-1"):
+ result = CyberArkPAMImportCommand._interactive_safe_picker(safes)
+ assert result == "S1,S2,S3,S4"
+
+ def test_select_deduplicates(self):
+ from keepercommander.commands.pam_import.cyberark_import import CyberArkPAMImportCommand
+ from unittest.mock import patch
+ safes = [{"safeName": f"S{i}"} for i in range(1, 6)]
+ with patch("builtins.input", return_value="1,1-3,2"):
+ result = CyberArkPAMImportCommand._interactive_safe_picker(safes)
+ assert result == "S1,S2,S3"
+
def test_select_invalid_input(self):
from keepercommander.commands.pam_import.cyberark_import import CyberArkPAMImportCommand
from unittest.mock import patch
safes = [{"safeName": "Safe1"}]
with patch("builtins.input", return_value="abc"):
result = CyberArkPAMImportCommand._interactive_safe_picker(safes)
- assert result is None # invalid → all
+ assert result == "" # invalid non-empty → abort, do not import all
+
+ def test_select_invalid_range_syntax_aborts(self):
+ from keepercommander.commands.pam_import.cyberark_import import CyberArkPAMImportCommand
+ from unittest.mock import patch
+ safes = [{"safeName": f"S{i}"} for i in range(1, 6)]
+ with patch("builtins.input", return_value="1..4"):
+ result = CyberArkPAMImportCommand._interactive_safe_picker(safes)
+ assert result == ""
def test_eof_returns_none(self):
from keepercommander.commands.pam_import.cyberark_import import CyberArkPAMImportCommand
@@ -1470,6 +1510,176 @@ def test_eof_returns_none(self):
assert result is None
+class TestParseIndexSelection:
+ """Unit tests for _parse_index_selection."""
+
+ def test_single_indexes(self):
+ from keepercommander.commands.pam_import.cyberark_import import CyberArkPAMImportCommand
+ assert CyberArkPAMImportCommand._parse_index_selection("1,3", 5) == [0, 2]
+
+ def test_range(self):
+ from keepercommander.commands.pam_import.cyberark_import import CyberArkPAMImportCommand
+ assert CyberArkPAMImportCommand._parse_index_selection("1-4", 10) == [0, 1, 2, 3]
+
+ def test_mixed(self):
+ from keepercommander.commands.pam_import.cyberark_import import CyberArkPAMImportCommand
+ assert CyberArkPAMImportCommand._parse_index_selection("1,2,3,6-9,11,14-18", 20) == [
+ 0, 1, 2, 5, 6, 7, 8, 10, 13, 14, 15, 16, 17
+ ]
+
+ def test_out_of_range_skipped(self):
+ from keepercommander.commands.pam_import.cyberark_import import CyberArkPAMImportCommand
+ assert CyberArkPAMImportCommand._parse_index_selection("1,99,2-3,50-60", 5) == [0, 1, 2]
+
+ def test_whitespace_tolerant(self):
+ from keepercommander.commands.pam_import.cyberark_import import CyberArkPAMImportCommand
+ assert CyberArkPAMImportCommand._parse_index_selection(" 1 , 3 - 5 , 7 ", 10) == [0, 2, 3, 4, 6]
+
+ def test_clamps_huge_range(self):
+ from keepercommander.commands.pam_import.cyberark_import import CyberArkPAMImportCommand
+ assert CyberArkPAMImportCommand._parse_index_selection("1-1000000", 5) == [0, 1, 2, 3, 4]
+
+ def test_unicode_en_dash(self):
+ from keepercommander.commands.pam_import.cyberark_import import CyberArkPAMImportCommand
+ assert CyberArkPAMImportCommand._parse_index_selection("2–4", 6) == [1, 2, 3]
+
+ def test_rejects_malformed_tokens(self):
+ from keepercommander.commands.pam_import.cyberark_import import CyberArkPAMImportCommand
+ assert CyberArkPAMImportCommand._parse_index_selection("1-2-3", 10) == []
+ assert CyberArkPAMImportCommand._parse_index_selection("1--5", 10) == []
+ assert CyberArkPAMImportCommand._parse_index_selection("1..4", 10) == []
+ assert CyberArkPAMImportCommand._parse_index_selection("1:4", 10) == []
+
+ def test_empty_tokens_ignored(self):
+ from keepercommander.commands.pam_import.cyberark_import import CyberArkPAMImportCommand
+ assert CyberArkPAMImportCommand._parse_index_selection("1,,3", 5) == [0, 2]
+
+
+def _enterprise_params(nodes=None, user_root_nodes=None, enterprise_name="Acme"):
+ params = MagicMock()
+ params.enterprise = {
+ "nodes": nodes if nodes is not None else [
+ {"node_id": 1, "data": {}},
+ {"node_id": 2, "parent_id": 1, "data": {"displayname": "Engineering"}},
+ {"node_id": 3, "parent_id": 1, "data": {"displayname": "R&D"}},
+ {"node_id": 4, "parent_id": 1, "data": {"displayname": "Dup"}},
+ {"node_id": 5, "parent_id": 1, "data": {"displayname": "Dup"}},
+ ],
+ "user_root_nodes": user_root_nodes if user_root_nodes is not None else [1],
+ "enterprise_name": enterprise_name,
+ }
+ return params
+
+
+class TestResolveProvisioningNodeId:
+ """Tests for CyberArkImporter._resolve_provisioning_node_id."""
+
+ def _importer(self):
+ from keepercommander.importer.cyberark.cyberark import CyberArkImporter
+ return CyberArkImporter()
+
+ @patch("keepercommander.importer.cyberark.cyberark.print_formatted_text")
+ def test_resolves_by_name(self, _print):
+ importer = self._importer()
+ assert importer._resolve_provisioning_node_id(_enterprise_params(), "Engineering") == 2
+
+ @patch("keepercommander.importer.cyberark.cyberark.print_formatted_text")
+ def test_resolves_by_numeric_id(self, _print):
+ importer = self._importer()
+ assert importer._resolve_provisioning_node_id(_enterprise_params(), "3") == 3
+
+ @patch("keepercommander.importer.cyberark.cyberark.print_formatted_text")
+ def test_strips_whitespace(self, _print):
+ importer = self._importer()
+ assert importer._resolve_provisioning_node_id(_enterprise_params(), " Engineering ") == 2
+
+ @patch("keepercommander.importer.cyberark.cyberark.print_formatted_text")
+ def test_ampersand_name_found(self, _print):
+ importer = self._importer()
+ assert importer._resolve_provisioning_node_id(_enterprise_params(), "R&D") == 3
+
+ @patch("keepercommander.importer.cyberark.cyberark.print_formatted_text")
+ def test_ampersand_name_not_found_does_not_raise(self, _print):
+ importer = self._importer()
+ assert importer._resolve_provisioning_node_id(_enterprise_params(), "No&Such") is None
+
+ @patch("keepercommander.importer.cyberark.cyberark.print_formatted_text")
+ def test_not_found(self, _print):
+ importer = self._importer()
+ assert importer._resolve_provisioning_node_id(_enterprise_params(), "Missing") is None
+
+ @patch("keepercommander.importer.cyberark.cyberark.print_formatted_text")
+ def test_ambiguous_name(self, _print):
+ importer = self._importer()
+ assert importer._resolve_provisioning_node_id(_enterprise_params(), "Dup") is None
+
+ @patch("keepercommander.importer.cyberark.cyberark.print_formatted_text")
+ def test_default_user_root_node(self, _print):
+ importer = self._importer()
+ assert importer._resolve_provisioning_node_id(
+ _enterprise_params(user_root_nodes=[99]), None
+ ) == 99
+
+ @patch("keepercommander.importer.cyberark.cyberark.print_formatted_text")
+ def test_default_falls_back_to_tree_root(self, _print):
+ importer = self._importer()
+ assert importer._resolve_provisioning_node_id(
+ _enterprise_params(user_root_nodes=[]), None
+ ) == 1
+
+ @patch("keepercommander.importer.cyberark.cyberark.print_formatted_text")
+ def test_no_enterprise(self, _print):
+ importer = self._importer()
+ params = MagicMock()
+ params.enterprise = None
+ assert importer._resolve_provisioning_node_id(params, "Engineering") is None
+
+ @patch("keepercommander.importer.cyberark.cyberark.print_formatted_text")
+ def test_no_params(self, _print):
+ importer = self._importer()
+ assert importer._resolve_provisioning_node_id(None, "Engineering") is None
+
+ def test_success_html_escapes_ampersand(self):
+ from prompt_toolkit import HTML
+ from keepercommander.importer.cyberark.pam.ui import _esc
+ HTML(
+ f"Provisioning teams, roles, and users into node "
+ f"{_esc('R&D')} (id 3)"
+ )
+
+
+class TestImportTargetNodeArgparse:
+ """--target-node / --node on the vault import parser."""
+
+ def test_target_node_flag(self):
+ from keepercommander.importer.commands import import_parser
+ ns = import_parser.parse_args(["--format", "cyberark", "--target-node", "Eng", "https://pvwa"])
+ assert ns.target_node == "Eng"
+
+ def test_node_alias(self):
+ from keepercommander.importer.commands import import_parser
+ ns = import_parser.parse_args(["--format", "cyberark", "--node", "Eng", "https://pvwa"])
+ assert ns.target_node == "Eng"
+
+ def test_ignored_for_non_cyberark_format(self):
+ from keepercommander.importer.commands import RecordImportCommand
+ cmd = RecordImportCommand()
+ params = MagicMock()
+ params.enforcements = None
+ with patch("keepercommander.importer.commands.imp_exp._import") as mock_import:
+ cmd.execute(params, format="json", name="vault.json", target_node="Eng")
+ assert mock_import.call_args.kwargs.get("target_node") is None
+
+ def test_kept_for_cyberark_format(self):
+ from keepercommander.importer.commands import RecordImportCommand
+ cmd = RecordImportCommand()
+ params = MagicMock()
+ params.enforcements = None
+ with patch("keepercommander.importer.commands.imp_exp._import") as mock_import:
+ cmd.execute(params, format="cyberark", name="https://pvwa", target_node="Eng")
+ assert mock_import.call_args.kwargs.get("target_node") == "Eng"
+
+
class TestListSafesDetailed:
"""Tests for _list_safes_detailed."""