Release 18.0.12

.github/workflows/test-with-pytest.yml

@@ -9,16 +9,16 @@ jobs:
   test-with-pytest:
     strategy:
       matrix:
-        python-version: ['3.8', '3.14']
+        python-version: ['3.9', '3.14']
 
     runs-on: ubuntu-24.04
 
     steps:
       - name: Checkout branch
-        uses: actions/checkout@v4
+        uses: actions/checkout@v6
 
       - name: Set up Python ${{ matrix.python-version }}
-        uses: actions/setup-python@v4
+        uses: actions/setup-python@v6
         with:
           python-version: ${{ matrix.python-version }}
 

docker-entrypoint.sh

@@ -159,38 +159,31 @@ process_ksm_config() {
 # DEVICE SETUP AND AUTHENTICATION FUNCTIONS
 # =============================================================================
 
-# Setup device registration and persistent login
+# Setup device registration and persistent login.
+# All three steps run in a single Commander process so login + vault sync
+# only happens once instead of three times.
 setup_device() {
     local user="$1"
     local password="$2"
     local server="$3"
-
-    # Step 1: Register device
-    log "Registering device..."
-    if ! python3 keeper.py --user "${user}" --password "${password}" \
-        --server "${server}" this-device register; then
-        log "ERROR: Device registration failed"
-        exit 1
-    fi
-
-    # Step 2: Enable persistent login
-    log "Enabling persistent login..."
-    if ! python3 keeper.py --user "${user}" --password "${password}" \
-        --server "${server}" this-device persistent-login on; then
-        log "ERROR: Persistent login setup failed"
-        exit 1
-    fi
 
-    # Step 3: Set timeout
-    log "Setting device logout timeout to 30 Days..."
+    log "Running device setup (register, persistent login, timeout)..."
+    local setup_script
+    setup_script=$(mktemp /tmp/keeper_setup_XXXXXX.cmd)
+    cat > "${setup_script}" <<CMDS
+this-device register
+this-device persistent-login on
+this-device timeout ${DEVICE_TIMEOUT}
+CMDS
+
     if ! python3 keeper.py --user "${user}" --password "${password}" \
-        --server "${server}" this-device timeout "${DEVICE_TIMEOUT}" \
-        > /dev/null; then
-        log "ERROR: Timeout setup failed"
+        --server "${server}" "${setup_script}"; then
+        log "ERROR: Device setup failed"
+        rm -f "${setup_script}"
         exit 1
     fi
-    
-    log "Device Logout Timeout set successfully"
+
+    rm -f "${setup_script}"
     log "Device setup completed successfully"
 }
 

examples/pam_import_generator.py

@@ -30,10 +30,21 @@
 import csv
 import json
 import os
+import secrets
+import string
 import sys
 from pathlib import Path
 from typing import Any, Dict, List
 
+_ALPHANUM = string.ascii_letters + string.digits
+
+
+def _random_alphanumeric(length: int = 16) -> str:
+    return "".join(secrets.choice(_ALPHANUM) for _ in range(length))
+
+
+_AD_DEMO = _random_alphanumeric()
+_MACHINE_DEMO = _random_alphanumeric()
 
 DEFAULT_IMPORT_TEMPLATE = """{
     "project": "Project1",
@@ -70,7 +81,7 @@
                     "title": "XXX:DomainAdmin",
                     "_comment_login_password": "Must provide valid credentials but delete sensitive data/json after import",
                     "login": "XXX:[email protected]",
-                    "password": "XXX:P4ssw0rd_123",
+                    "password": "XXX:__AD_DEMO__",
                     "rotation_settings": { "rotation": "general", "enabled": "on", "schedule": {"type": "on-demand"}}
                 }]
             },
@@ -106,13 +117,13 @@
                     "_comment_login": "username value from CSV",
                     "login": "xxx:Administrator",
                     "_comment_password": "password value from CSV",
-                    "password": "xxx:P4ssw0rd_123",
+                    "password": "xxx:__MACHINE_DEMO__",
                     "rotation_settings": { "rotation": "general", "enabled": "on", "schedule": {"type": "on-demand"} }
                 }]
             }
         ]
     }
-}"""
+}""".replace("__AD_DEMO__", _AD_DEMO).replace("__MACHINE_DEMO__", _MACHINE_DEMO)
 
 
 def _build_cli() -> argparse.ArgumentParser:

keepercommander/__init__.py

@@ -10,4 +10,4 @@
 # Contact: [email protected]
 #
 
-__version__ = '18.0.11'
+__version__ = '18.0.12'

keepercommander/api.py

@@ -377,8 +377,12 @@ def load_team_keys(params, team_uids):          # type: (KeeperParams, List[str]
                     encrypted_team_key = t.get('encrypted_team_key')
                     if encrypted_team_key:
                         team_key = crypto.decrypt_aes_v2(utils.base64_url_decode(encrypted_team_key), tree_key)
-                        params.key_cache[team_uid] = PublicKeys(aes=team_key)
-                        s.remove(team_uid)
+                        existing = params.key_cache.get(team_uid)
+                        params.key_cache[team_uid] = PublicKeys(
+                            aes=team_key,
+                            rsa=getattr(existing, 'rsa', b'') or b'',
+                            ec=getattr(existing, 'ec', b'') or b'')
+                        # Still fetch asymmetric public keys via team_get_keys below.
                 except Exception as e:
                     logging.debug('Team UID \"%s\": Decrypt key error: %s', team_uid, str(e))
 
@@ -397,30 +401,51 @@ def load_team_keys(params, team_uids):          # type: (KeeperParams, List[str]
         }
         rs = communicate(params, rq)
         if 'keys' in rs:
+            merged = {}  # team_uid -> PublicKeys fields
             for tk in rs['keys']:
+                team_uid = tk.get('team_uid')
+                if not team_uid:
+                    continue
+                if team_uid not in merged:
+                    existing = params.key_cache.get(team_uid)
+                    merged[team_uid] = {
+                        'aes': getattr(existing, 'aes', b'') or b'',
+                        'rsa': getattr(existing, 'rsa', b'') or b'',
+                        'ec': getattr(existing, 'ec', b'') or b'',
+                    }
+                # Read the symmetric/wrapped team key from the 'key' field
                 if 'key' in tk:
-                    team_uid = tk['team_uid']
                     try:
-                        aes = b''
-                        rsa = b''
-                        ec = b''
                         encrypted_key = utils.base64_url_decode(tk['key'])
-                        key_type = tk['type']
+                        key_type = tk.get('type')
                         if key_type == 1:
-                            aes = crypto.decrypt_aes_v1(encrypted_key, params.data_key)
+                            merged[team_uid]['aes'] = crypto.decrypt_aes_v1(encrypted_key, params.data_key)
                         elif key_type == 2:
-                            aes = crypto.decrypt_rsa(encrypted_key, params.rsa_key2)
+                            merged[team_uid]['aes'] = crypto.decrypt_rsa(encrypted_key, params.rsa_key2)
                         elif key_type == 3:
-                            aes = crypto.decrypt_aes_v2(encrypted_key, params.data_key)
+                            merged[team_uid]['aes'] = crypto.decrypt_aes_v2(encrypted_key, params.data_key)
                         elif key_type == 4:
-                            aes = crypto.decrypt_ec(encrypted_key, params.ecc_key)
+                            merged[team_uid]['aes'] = crypto.decrypt_ec(encrypted_key, params.ecc_key)
                         elif key_type == -1:
-                            ec = encrypted_key
+                            merged[team_uid]['ec'] = encrypted_key
                         elif key_type == -3:
-                            rsa = encrypted_key
-                        params.key_cache[team_uid] = PublicKeys(rsa=rsa, aes=aes, ec=ec)
+                            merged[team_uid]['rsa'] = encrypted_key
+                    except Exception as e:
+                        logging.debug(e)
+                # Read the raw team public key from 'team_public_key' field (separate from 'key')
+                if 'team_public_key' in tk:
+                    try:
+                        pub_key_bytes = utils.base64_url_decode(tk['team_public_key'])
+                        pub_key_type = tk.get('team_public_key_type')
+                        if pub_key_type == -3:
+                            merged[team_uid]['rsa'] = pub_key_bytes
+                        elif pub_key_type == -1:
+                            merged[team_uid]['ec'] = pub_key_bytes
                     except Exception as e:
                         logging.debug(e)
+            for team_uid, kd in merged.items():
+                params.key_cache[team_uid] = PublicKeys(
+                    rsa=kd['rsa'], aes=kd['aes'], ec=kd['ec'])
 
 
 def load_available_teams(params):

keepercommander/commands/base.py

@@ -915,6 +915,11 @@ def resolve_single_record(params, record_name):  # type: (KeeperParams, str) ->
         if record_name in params.record_cache:
             return vault.KeeperRecord.load(params, record_name)
 
+        from .pam_import.record_loader import load_pam_record
+        rec = load_pam_record(params, record_name)
+        if rec:
+            return rec
+
         rs = try_resolve_path(params, record_name)
         if rs is None:
             return None

keepercommander/commands/credential_provision.py

@@ -47,11 +47,12 @@
 from .. import api, vault, vault_extensions, crypto, utils, generator
 from ..error import CommandError
 from ..params import KeeperParams
-from ..subfolder import try_resolve_path, get_folder_path
-from ..record_management import add_record_to_folder
+from ..subfolder import try_resolve_path
 from ..record_facades import LoginRecordFacade
 from ..proto import APIRequest_pb2
 from ..commands.folder import FolderMakeCommand
+from ..commands.pam.vault_target import (
+    create_record_in_folder, records_in_folder, resolve_provision_target_folder)
 
 # RecordLink and discoveryrotation require pydantic (Python 3.8+)
 try:
@@ -1313,35 +1314,24 @@ def _check_duplicate_by_username_in_folder(self, config: Dict[str, Any], params:
         username = config['account']['username']
         pam_config_uid = config['account']['pam_config_uid']
 
-        # Get the same folder path that will be used for PAM User creation
         try:
-            gateway_folder_uid, gateway_folder_path = self._get_gateway_application_folder(pam_config_uid, params)
+            folder_uid = self._resolve_pam_user_folder_uid(config, params)
         except Exception as e:
+            logging.warning(
+                f'Could not resolve target folder for duplicate username check '
+                f'(username={username}, PAM config={pam_config_uid}): {e}'
+            )
             return False
 
-        # Determine target folder (same logic as _create_pam_user)
-        user_specified_folder = config.get('vault', {}).get('folder')
-
-        if user_specified_folder:
-            target_folder_path = f"{gateway_folder_path}/{user_specified_folder.strip('/')}"
-        else:
-            department = config['user'].get('department', 'Default')
-            target_folder_path = f"{gateway_folder_path}/PAM Users/{department}"
-
-        # Get folder UID
-        folder_uid = self._get_folder_uid(target_folder_path, params)
-
         if not folder_uid:
             return False
 
-        # Get all records in this folder
-        records_in_folder = params.subfolder_record_cache.get(folder_uid, set())
-
-        if not records_in_folder:
+        records_in_folder_set = records_in_folder(params, folder_uid)
+        if not records_in_folder_set:
             return False
 
         # Search for PAM Users in folder with matching username
-        for record_uid in records_in_folder:
+        for record_uid in records_in_folder_set:
             try:
                 record = vault.KeeperRecord.load(params, record_uid)
 
@@ -1358,7 +1348,7 @@ def _check_duplicate_by_username_in_folder(self, config: Dict[str, Any], params:
                 if facade.login == username:
                     logging.error(f'❌ Duplicate PAM User found (by username in folder):')
                     logging.error(f'   Username: {username}')
-                    logging.error(f'   Folder: {target_folder_path}')
+                    logging.error(f'   Folder UID: {folder_uid}')
                     logging.error(f'   Existing UID: {record_uid}')
                     logging.error(f'   Title: {record.title}')
                     return True
@@ -1481,28 +1471,7 @@ def _create_pam_user(
         username = config['account']['username']
         pam_config_uid = config['account']['pam_config_uid']
 
-        # Get gateway application folder from PAM Config
-        gateway_folder_uid, gateway_folder_path = self._get_gateway_application_folder(pam_config_uid, params)
-
-        # Determine target folder
-        user_specified_folder = config.get('vault', {}).get('folder')
-
-        if user_specified_folder:
-            # Check if the value is a folder UID (exists in folder cache)
-            if user_specified_folder in params.folder_cache:
-                folder_uid = user_specified_folder
-            elif user_specified_folder in params.shared_folder_cache:
-                folder_uid = user_specified_folder
-            else:
-                # Treat as a relative folder path
-                self._validate_folder_path(user_specified_folder)
-                target_folder_path = f"{gateway_folder_path}/{user_specified_folder.strip('/')}"
-                folder_uid = self._ensure_folder_exists(target_folder_path, params)
-        else:
-            # Auto-generate subfolder based on department
-            department = config['user'].get('department', 'Default')
-            target_folder_path = f"{gateway_folder_path}/PAM Users/{department}"
-            folder_uid = self._ensure_folder_exists(target_folder_path, params)
+        folder_uid = self._resolve_pam_user_folder_uid(config, params)
 
         # Create PAM User typed record
         pam_user = vault.TypedRecord()
@@ -1576,7 +1545,7 @@ def _create_pam_user(
         try:
 
             # Create record in vault and add to folder
-            add_record_to_folder(params, pam_user, folder_uid)
+            create_record_in_folder(params, pam_user, folder_uid, command='credential-provision')
 
             # Sync to get the record UID
             api.sync_down(params)
@@ -1619,6 +1588,20 @@ def _validate_folder_path(self, folder_path: str) -> None:
                 f"Example: 'PAM Users/Engineering' (not '/Shared Folders/...')"
             )
 
+    def _resolve_pam_user_folder_uid(self, config: Dict[str, Any], params: KeeperParams) -> str:
+        pam_config_uid = config['account']['pam_config_uid']
+        user_specified_folder = config.get('vault', {}).get('folder')
+        if user_specified_folder:
+            self._validate_folder_path(user_specified_folder)
+        department = config.get('user', {}).get('department', 'Default')
+        return resolve_provision_target_folder(
+            params,
+            pam_config_uid,
+            folder_spec=user_specified_folder,
+            department=department,
+            command='credential-provision',
+        )
+
     def _get_gateway_application_folder(self, pam_config_uid: str, params: KeeperParams) -> Tuple[str, str]:
         """
         Get gateway application folder from PAM Configuration.
@@ -1636,35 +1619,11 @@ def _get_gateway_application_folder(self, pam_config_uid: str, params: KeeperPar
         Raises:
             ValueError: If PAM Config not found or folder not accessible
         """
-
-        # Load PAM Config record
-        pam_config_record = vault.KeeperRecord.load(params, pam_config_uid)
-        if not pam_config_record:
-            raise ValueError(f"PAM Configuration not found: {pam_config_uid}")
-
-        # Use facade to access folder_uid
-        facade = PamConfigurationRecordFacade()
-        facade.record = pam_config_record
-        facade.load_typed_fields()
-
-        folder_uid = facade.folder_uid
-
-        if not folder_uid:
-            raise ValueError(
-                f"PAM Configuration '{pam_config_record.title}' has no application folder configured.\n"
-                f"Please configure the application folder in the PAM Configuration record."
-            )
-
-        # Get folder path from folder_uid using existing utility
-        if folder_uid not in params.folder_cache:
-            raise ValueError(
-                f"Application folder (UID: {folder_uid}) not found in vault.\n"
-                f"Ensure the folder is shared with you."
-            )
-
-        folder_path = get_folder_path(params, folder_uid)
-
-        return folder_uid, folder_path
+        from ..commands.pam.vault_target import resolve_pam_application_folder
+        try:
+            return resolve_pam_application_folder(params, pam_config_uid, command='credential-provision')
+        except CommandError as exc:
+            raise ValueError(str(exc)) from exc
 
     def _ensure_folder_exists(self, folder_path: str, params: KeeperParams) -> Optional[str]:
         """