From 272d1afc619ed9c10427546047836550c0255e41 Mon Sep 17 00:00:00 2001 From: Ivan Dimov <78815270+idimov-keeper@users.noreply.github.com> Date: Tue, 21 Jul 2026 15:57:15 -0500 Subject: [PATCH] Fix NSF folder titles for team-shared folders. Decrypt team-wrapped folderAccesses keys and run NSF decrypt after team keys so sharees see real names instead of Unnamed. Co-authored-by: Cursor --- keepercommander/nested_share_folder/sync.py | 184 ++++++++++++---- keepercommander/sync_down.py | 8 +- unit-tests/test_nsf_folder_key_decrypt.py | 222 ++++++++++++++++++++ 3 files changed, 367 insertions(+), 47 deletions(-) create mode 100644 unit-tests/test_nsf_folder_key_decrypt.py diff --git a/keepercommander/nested_share_folder/sync.py b/keepercommander/nested_share_folder/sync.py index 32992276d..996012fa6 100644 --- a/keepercommander/nested_share_folder/sync.py +++ b/keepercommander/nested_share_folder/sync.py @@ -668,73 +668,169 @@ def _try_decrypt_with_user_keys(enc_key, params): return None +def _try_decrypt_with_typed_key(encrypted_key, key_type, aes_key=None, rsa_key=None, ecc_key=None): + """Decrypt *encrypted_key* using the algorithm indicated by *key_type*. + + Returns plaintext bytes or None. Mirrors Web Vault ``decryptFolderKeyByType``. + """ + try: + if key_type == folder_pb2.encrypted_by_data_key_gcm: + if aes_key is not None: + return crypto.decrypt_aes_v2(encrypted_key, aes_key) + elif key_type == folder_pb2.encrypted_by_data_key: + if aes_key is not None: + return crypto.decrypt_aes_v1(encrypted_key, aes_key) + elif key_type == folder_pb2.encrypted_by_public_key: + if rsa_key is not None: + return crypto.decrypt_rsa(encrypted_key, rsa_key) + elif key_type == folder_pb2.encrypted_by_public_key_ecc: + if ecc_key is not None: + return crypto.decrypt_ec(encrypted_key, ecc_key) + except Exception: + return None + return None + + +def _team_decrypt_material(params, team_uid): + """Return (aes_key, rsa_private_key, ecc_private_key) for a decrypted team, or Nones.""" + team_cache = getattr(params, 'team_cache', None) or {} + team = team_cache.get(team_uid) + if not team or 'team_key_unencrypted' not in team: + return None, None, None + + aes_key = team['team_key_unencrypted'] + rsa_key = None + ecc_key = None + if 'team_private_key_unencrypted' in team: + try: + rsa_key = crypto.load_rsa_private_key(team['team_private_key_unencrypted']) + except Exception: + pass + if 'team_ec_private_key_unencrypted' in team: + try: + ecc_key = crypto.load_ec_private_key(team['team_ec_private_key_unencrypted']) + except Exception: + pass + return aes_key, rsa_key, ecc_key + + +def _try_decrypt_from_folder_access(folder_uid, params): + """Unwrap a folder key from folderAccesses (user or team), mirroring Web Vault. + + Team-shared NSF folders store the sharee's copy of the folder key in + folderAccesses encrypted with the team AES/RSA/ECC key. Direct user shares + use the recipient's data key or public key. + """ + accesses = getattr(params, 'nested_share_folder_accesses', {}).get(folder_uid) or [] + team_cache = getattr(params, 'team_cache', None) or {} + + for fa in accesses: + if 'folder_key' not in fa: + continue + + encrypted_key = fa['folder_key']['encrypted_key'] + key_type = fa['folder_key']['encrypted_key_type'] + access_type = fa.get('access_type') + access_uid = fa.get('access_type_uid') + + use_team = ( + access_type == folder_pb2.AT_TEAM + or (access_uid and access_uid in team_cache) + ) + + try: + folder_key = None + if use_team and access_uid: + team_aes, team_rsa, team_ecc = _team_decrypt_material(params, access_uid) + folder_key = _try_decrypt_with_typed_key( + encrypted_key, key_type, + aes_key=team_aes, rsa_key=team_rsa, ecc_key=team_ecc, + ) + # Algorithm fallbacks (typed decrypt may disagree with wire type) + if not folder_key and team_aes is not None: + folder_key = _try_decrypt_symmetric(encrypted_key, team_aes) + if not folder_key and team_rsa is not None: + try: + folder_key = crypto.decrypt_rsa(encrypted_key, team_rsa) + except Exception: + pass + if not folder_key and team_ecc is not None: + try: + folder_key = crypto.decrypt_ec(encrypted_key, team_ecc) + except Exception: + pass + + if not folder_key: + folder_key = _try_decrypt_with_typed_key( + encrypted_key, key_type, + aes_key=params.data_key, + rsa_key=params.rsa_key2, + ecc_key=params.ecc_key, + ) + if not folder_key: + folder_key = _try_decrypt_with_user_keys(encrypted_key, params) + + if folder_key and len(folder_key) == 32: + return folder_key + except Exception as e: + logging.debug( + 'Failed to decrypt folder key for %s from access data: %s', folder_uid, e + ) + return None + + def _decrypt_nested_share_folder_keys(params): - """Decrypt Nested Share Folder folder and record keys.""" + """Decrypt Nested Share Folder folder and record keys. + + Mirrors Web Vault ``process-keeper-drive-folders``: + - ENCRYPTED_BY_USER_KEY → user keys, then folderAccesses + - ENCRYPTED_BY_PARENT_KEY → parent folder key, then folderAccesses + - ENCRYPTED_BY_TEAM_KEY → folderAccesses (team-wrapped keys) + """ newly_decrypted = True - + while newly_decrypted: newly_decrypted = False - + for folder_uid, folder_obj in params.nested_share_folders.items(): if 'folder_key_unencrypted' in folder_obj: continue folder_key = None + force_accesses = False if folder_uid in params.nested_share_folder_keys: for fk in params.nested_share_folder_keys[folder_uid]: enc_key = fk['encrypted_key'] + key_type = fk['key_type'] try: - if fk['key_type'] == folder_pb2.ENCRYPTED_BY_USER_KEY: - # FolderKeyEncryptionType only tells us the KEY SOURCE (user vs parent), - # not the encryption algorithm. Try all algorithms in likelihood order: - # AES-256-GCM (60 B) — modern default - # AES-256-CBC (48 B) — legacy - # RSA-2048 (256 B) — shared folder re-encrypted for this user - # ECC — EC-based key wrap + if key_type == folder_pb2.ENCRYPTED_BY_TEAM_KEY: + # Key lives in folderAccesses encrypted with the team key. + force_accesses = True + break + if key_type == folder_pb2.ENCRYPTED_BY_USER_KEY: + # FolderKeyEncryptionType is KEY SOURCE, not algorithm. folder_key = _try_decrypt_with_user_keys(enc_key, params) if folder_key: break - elif fk['key_type'] == folder_pb2.ENCRYPTED_BY_PARENT_KEY: - parent_uid = folder_obj.get('parent_uid') + force_accesses = True + elif key_type == folder_pb2.ENCRYPTED_BY_PARENT_KEY: + parent_uid = fk.get('parent_uid') or folder_obj.get('parent_uid') if parent_uid and parent_uid in params.nested_share_folders: parent_folder = params.nested_share_folders[parent_uid] if 'folder_key_unencrypted' in parent_folder: - parent_key = parent_folder['folder_key_unencrypted'] - folder_key = _try_decrypt_symmetric(enc_key, parent_key) + folder_key = _try_decrypt_symmetric( + enc_key, parent_folder['folder_key_unencrypted'] + ) if folder_key: break + # Sharees often get PARENT_KEY metadata without the owner parent. + force_accesses = True except Exception as e: - logging.debug(f"Failed to decrypt folder key for {folder_uid}: {e}") - - # Fallback: try from folder access data (EncryptedDataKey — has explicit algorithm) - if not folder_key and folder_uid in params.nested_share_folder_accesses: - for fa in params.nested_share_folder_accesses[folder_uid]: - if 'folder_key' not in fa: - continue + logging.debug('Failed to decrypt folder key for %s: %s', folder_uid, e) - try: - encrypted_key = fa['folder_key']['encrypted_key'] - key_type = fa['folder_key']['encrypted_key_type'] - - if key_type == folder_pb2.encrypted_by_data_key_gcm: - folder_key = crypto.decrypt_aes_v2(encrypted_key, params.data_key) - elif key_type == folder_pb2.encrypted_by_data_key: - folder_key = crypto.decrypt_aes_v1(encrypted_key, params.data_key) - elif key_type == folder_pb2.encrypted_by_public_key: - if params.rsa_key2: - folder_key = crypto.decrypt_rsa(encrypted_key, params.rsa_key2) - elif key_type == folder_pb2.encrypted_by_public_key_ecc: - if params.ecc_key: - folder_key = crypto.decrypt_ec(encrypted_key, params.ecc_key) - else: - # Unknown type — try all user keys as a last resort - folder_key = _try_decrypt_with_user_keys(encrypted_key, params) - - if folder_key: - break - except Exception as e: - logging.debug(f"Failed to decrypt folder key for {folder_uid} from access data: {e}") + if not folder_key and (force_accesses or folder_uid in params.nested_share_folder_accesses): + folder_key = _try_decrypt_from_folder_access(folder_uid, params) if folder_key: folder_obj['folder_key_unencrypted'] = folder_key @@ -748,7 +844,7 @@ def _decrypt_nested_share_folder_keys(params): if 'color' in data_json: folder_obj['color'] = data_json['color'] except Exception as e: - logging.debug(f"Failed to decrypt folder data for {folder_uid}: {e}") + logging.debug('Failed to decrypt folder data for %s: %s', folder_uid, e) _decrypt_nested_share_record_keys(params) diff --git a/keepercommander/sync_down.py b/keepercommander/sync_down.py index ad6759e1c..45fd3a510 100644 --- a/keepercommander/sync_down.py +++ b/keepercommander/sync_down.py @@ -611,9 +611,6 @@ def convert_user_folder_shared_folder(ufsf): params.revision = revision - if nsf_enabled: - nested_share_folder_sync.process(params, nsf_acc) - for sf in params.shared_folder_cache.values(): owner = sf.get('owner_username') if not owner: @@ -763,6 +760,11 @@ def convert_user_folder_shared_folder(ufsf): del params.team_cache[team_uid] to_delete.clear() + # NSF folder keys may be wrapped with team keys (ENCRYPTED_BY_TEAM_KEY / + # folderAccesses). Decrypt only after team keys are available. + if nsf_enabled: + nested_share_folder_sync.process(params, nsf_acc) + logging.debug('Decrypting shared folder keys') for shared_folder_uid, shared_folder in params.shared_folder_cache.items(): if 'shared_folder_key_unencrypted' not in shared_folder and 'shared_folder_key' in shared_folder: diff --git a/unit-tests/test_nsf_folder_key_decrypt.py b/unit-tests/test_nsf_folder_key_decrypt.py new file mode 100644 index 000000000..3226cf34d --- /dev/null +++ b/unit-tests/test_nsf_folder_key_decrypt.py @@ -0,0 +1,222 @@ +"""Unit tests for NSF folder key / name decrypt (shared + team paths).""" + +import json +from unittest import TestCase +from unittest.mock import Mock + +from keepercommander import crypto, utils +from keepercommander.nested_share_folder import sync as nsf_sync +from keepercommander.proto import folder_pb2 + + +def _make_params(**overrides): + p = Mock() + p.data_key = utils.generate_aes_key() + p.rsa_key2 = None + p.ecc_key = None + p.team_cache = {} + p.nested_share_folders = {} + p.nested_share_folder_keys = {} + p.nested_share_folder_accesses = {} + p.nested_share_records = {} + p.nested_share_record_data = {} + p.nested_share_record_keys = {} + p.nested_share_folder_records = {} + p.nested_share_record_links = {} + for k, v in overrides.items(): + setattr(p, k, v) + return p + + +def _encrypted_folder(name, folder_key): + """Return (folder_uid, folder_obj, folder_key) with AES-GCM encrypted data.""" + folder_uid = utils.generate_uid() + data = crypto.encrypt_aes_v2(json.dumps({'name': name}).encode('utf-8'), folder_key) + folder_obj = { + 'folder_uid': folder_uid, + 'parent_uid': None, + 'data': data, + } + return folder_uid, folder_obj, folder_key + + +class TestNsfFolderKeyDecrypt(TestCase): + + def test_user_key_owner_path_decrypts_name(self): + params = _make_params() + folder_key = utils.generate_aes_key() + folder_uid, folder_obj, _ = _encrypted_folder('Owner Folder', folder_key) + params.nested_share_folders[folder_uid] = folder_obj + params.nested_share_folder_keys[folder_uid] = [{ + 'folder_uid': folder_uid, + 'parent_uid': None, + 'encrypted_key': crypto.encrypt_aes_v2(folder_key, params.data_key), + 'key_type': folder_pb2.ENCRYPTED_BY_USER_KEY, + }] + + nsf_sync._decrypt_nested_share_folder_keys(params) + + self.assertEqual(folder_obj['name'], 'Owner Folder') + self.assertEqual(folder_obj['folder_key_unencrypted'], folder_key) + + def test_team_key_via_folder_access_decrypts_name(self): + """ENCRYPTED_BY_TEAM_KEY + AT_TEAM access wrapped with team AES key.""" + team_uid = utils.generate_uid() + team_aes = utils.generate_aes_key() + folder_key = utils.generate_aes_key() + params = _make_params(team_cache={ + team_uid: { + 'team_uid': team_uid, + 'team_key_unencrypted': team_aes, + }, + }) + folder_uid, folder_obj, _ = _encrypted_folder('Team Shared NSF', folder_key) + params.nested_share_folders[folder_uid] = folder_obj + params.nested_share_folder_keys[folder_uid] = [{ + 'folder_uid': folder_uid, + 'parent_uid': None, + 'encrypted_key': b'', # unused for TEAM_KEY + 'key_type': folder_pb2.ENCRYPTED_BY_TEAM_KEY, + }] + params.nested_share_folder_accesses[folder_uid] = [{ + 'folder_uid': folder_uid, + 'access_type_uid': team_uid, + 'access_type': folder_pb2.AT_TEAM, + 'folder_key': { + 'encrypted_key': crypto.encrypt_aes_v2(folder_key, team_aes), + 'encrypted_key_type': folder_pb2.encrypted_by_data_key_gcm, + }, + }] + + nsf_sync._decrypt_nested_share_folder_keys(params) + + self.assertEqual(folder_obj['name'], 'Team Shared NSF') + self.assertEqual(folder_obj['folder_key_unencrypted'], folder_key) + + def test_team_key_fails_without_decrypted_team_key(self): + """Ordering dependency: encrypted-only team_cache entry cannot unwrap.""" + team_uid = utils.generate_uid() + team_aes = utils.generate_aes_key() + folder_key = utils.generate_aes_key() + params = _make_params(team_cache={ + team_uid: { + 'team_uid': team_uid, + # team_key present but not yet decrypted + 'team_key': utils.base64_url_encode( + crypto.encrypt_aes_v2(team_aes, utils.generate_aes_key()) + ), + 'team_key_type': 1, + }, + }) + folder_uid, folder_obj, _ = _encrypted_folder('Hidden Name', folder_key) + params.nested_share_folders[folder_uid] = folder_obj + params.nested_share_folder_keys[folder_uid] = [{ + 'folder_uid': folder_uid, + 'parent_uid': None, + 'encrypted_key': b'', + 'key_type': folder_pb2.ENCRYPTED_BY_TEAM_KEY, + }] + params.nested_share_folder_accesses[folder_uid] = [{ + 'folder_uid': folder_uid, + 'access_type_uid': team_uid, + 'access_type': folder_pb2.AT_TEAM, + 'folder_key': { + 'encrypted_key': crypto.encrypt_aes_v2(folder_key, team_aes), + 'encrypted_key_type': folder_pb2.encrypted_by_data_key_gcm, + }, + }] + + nsf_sync._decrypt_nested_share_folder_keys(params) + + self.assertNotIn('folder_key_unencrypted', folder_obj) + self.assertNotIn('name', folder_obj) + + def test_parent_key_without_parent_falls_back_to_team_access(self): + """Sharee gets PARENT_KEY metadata but no parent folder; key is in accesses.""" + team_uid = utils.generate_uid() + team_aes = utils.generate_aes_key() + folder_key = utils.generate_aes_key() + parent_uid = utils.generate_uid() + params = _make_params(team_cache={ + team_uid: { + 'team_uid': team_uid, + 'team_key_unencrypted': team_aes, + }, + }) + folder_uid, folder_obj, _ = _encrypted_folder('Child Shared', folder_key) + folder_obj['parent_uid'] = parent_uid + params.nested_share_folders[folder_uid] = folder_obj + # Parent not in nested_share_folders (outside sharee hierarchy) + params.nested_share_folder_keys[folder_uid] = [{ + 'folder_uid': folder_uid, + 'parent_uid': parent_uid, + 'encrypted_key': crypto.encrypt_aes_v2(folder_key, utils.generate_aes_key()), + 'key_type': folder_pb2.ENCRYPTED_BY_PARENT_KEY, + }] + params.nested_share_folder_accesses[folder_uid] = [{ + 'folder_uid': folder_uid, + 'access_type_uid': team_uid, + 'access_type': folder_pb2.AT_TEAM, + 'folder_key': { + 'encrypted_key': crypto.encrypt_aes_v2(folder_key, team_aes), + 'encrypted_key_type': folder_pb2.encrypted_by_data_key_gcm, + }, + }] + + nsf_sync._decrypt_nested_share_folder_keys(params) + + self.assertEqual(folder_obj['name'], 'Child Shared') + + def test_parent_key_with_parent_unwraps_chain(self): + params = _make_params() + parent_key = utils.generate_aes_key() + child_key = utils.generate_aes_key() + parent_uid, parent_obj, _ = _encrypted_folder('Parent', parent_key) + child_uid, child_obj, _ = _encrypted_folder('Child', child_key) + child_obj['parent_uid'] = parent_uid + params.nested_share_folders[parent_uid] = parent_obj + params.nested_share_folders[child_uid] = child_obj + params.nested_share_folder_keys[parent_uid] = [{ + 'folder_uid': parent_uid, + 'parent_uid': None, + 'encrypted_key': crypto.encrypt_aes_v2(parent_key, params.data_key), + 'key_type': folder_pb2.ENCRYPTED_BY_USER_KEY, + }] + params.nested_share_folder_keys[child_uid] = [{ + 'folder_uid': child_uid, + 'parent_uid': parent_uid, + 'encrypted_key': crypto.encrypt_aes_v2(child_key, parent_key), + 'key_type': folder_pb2.ENCRYPTED_BY_PARENT_KEY, + }] + + nsf_sync._decrypt_nested_share_folder_keys(params) + + self.assertEqual(parent_obj['name'], 'Parent') + self.assertEqual(child_obj['name'], 'Child') + + def test_user_access_fallback_when_user_key_fails(self): + """USER_KEY wrap fails; folderAccesses has user-wrapped key.""" + params = _make_params() + folder_key = utils.generate_aes_key() + folder_uid, folder_obj, _ = _encrypted_folder('Access Shared', folder_key) + params.nested_share_folders[folder_uid] = folder_obj + params.nested_share_folder_keys[folder_uid] = [{ + 'folder_uid': folder_uid, + 'parent_uid': None, + # Wrapped with a key the user does not have + 'encrypted_key': crypto.encrypt_aes_v2(folder_key, utils.generate_aes_key()), + 'key_type': folder_pb2.ENCRYPTED_BY_USER_KEY, + }] + params.nested_share_folder_accesses[folder_uid] = [{ + 'folder_uid': folder_uid, + 'access_type_uid': utils.generate_uid(), + 'access_type': folder_pb2.AT_USER, + 'folder_key': { + 'encrypted_key': crypto.encrypt_aes_v2(folder_key, params.data_key), + 'encrypted_key_type': folder_pb2.encrypted_by_data_key_gcm, + }, + }] + + nsf_sync._decrypt_nested_share_folder_keys(params) + + self.assertEqual(folder_obj['name'], 'Access Shared')