diff --git a/keepercommander/commands/base.py b/keepercommander/commands/base.py index 572e0f8c0..a882674a0 100644 --- a/keepercommander/commands/base.py +++ b/keepercommander/commands/base.py @@ -29,7 +29,7 @@ from .. import api, crypto, utils, vault, resources, error from ..params import KeeperParams -from ..subfolder import try_resolve_path, BaseFolderNode +from ..subfolder import try_resolve_path, FolderPathAmbiguousError, BaseFolderNode aliases = {} # type: Dict[str, str] commands = {} # type: Dict[str, Command] @@ -1060,7 +1060,11 @@ def resolve_folder(params, folder_name): # type: (KeeperParams, str) -> Optio if folder_name in params.folder_cache: return folder_name else: - rs = try_resolve_path(params, folder_name) + try: + rs = try_resolve_path( + params, folder_name, raise_on_ambiguous=True) + except FolderPathAmbiguousError as e: + raise error.CommandError('', str(e)) if rs is not None: folder, record_name = rs if folder and not record_name: diff --git a/keepercommander/commands/folder.py b/keepercommander/commands/folder.py index ade0054c6..9221e5f57 100644 --- a/keepercommander/commands/folder.py +++ b/keepercommander/commands/folder.py @@ -33,7 +33,8 @@ from ..params import LAST_SHARED_FOLDER_UID, LAST_FOLDER_UID from ..proto import folder_pb2 from ..record import Record -from ..subfolder import BaseFolderNode, SharedFolderNode, UserFolderNode, SharedFolderFolderNode, try_resolve_path, \ +from ..subfolder import BaseFolderNode, SharedFolderNode, UserFolderNode, SharedFolderFolderNode, \ + FolderPathAmbiguousError, try_resolve_path, \ find_folders, get_contained_record_uids, get_contained_folder_uids @@ -408,13 +409,20 @@ def execute(self, params, **kwargs): if folder_name in params.folder_cache or folder_name in params.nested_share_folders: params.current_folder = folder_name else: - rs = try_resolve_path(params, folder_name) + try: + rs = try_resolve_path( + params, folder_name, raise_on_ambiguous=True) + except FolderPathAmbiguousError as e: + raise CommandError('cd', str(e)) if rs is not None: folder, pattern = rs if len(pattern) == 0: params.current_folder = folder.uid else: raise CommandError('cd', 'Folder {0} not found'.format(folder_name)) + else: + raise CommandError( + 'cd', 'Folder {0} not found or is ambiguous'.format(folder_name)) class FolderTreeCommand(Command): @@ -483,7 +491,11 @@ def execute(self, params, **kwargs): if folder_name in params.folder_cache: folder_uid = folder_name else: - rs = try_resolve_path(params, folder_name) + try: + rs = try_resolve_path( + params, folder_name, raise_on_ambiguous=True) + except FolderPathAmbiguousError as e: + raise CommandError('rndir', str(e)) if rs is not None: folder, pattern = rs if len(pattern) == 0: @@ -555,7 +567,11 @@ def execute(self, params, **kwargs): name = kwargs['folder'] if 'folder' in kwargs else None if name: - rs = try_resolve_path(params, name) + try: + rs = try_resolve_path( + params, name, raise_on_ambiguous=True) + except FolderPathAmbiguousError as e: + raise CommandError('mkdir', str(e)) if rs is not None: base_folder, name = rs if len(name) == 0: @@ -700,7 +716,10 @@ def execute(self, params, **kwargs): folders = [] pattern_list = kwargs.get('pattern', []) for pattern in pattern_list: - rs = try_resolve_path(params, pattern) + try: + rs = try_resolve_path(params, pattern, raise_on_ambiguous=True) + except FolderPathAmbiguousError as e: + raise CommandError('rmdir', str(e)) if rs is None: regex_pattern = pattern else: @@ -833,7 +852,11 @@ def execute(self, params, **kwargs): if dst_path in params.folder_cache: dst_folder = params.folder_cache[dst_path] else: - dst = try_resolve_path(params, dst_path) + try: + dst = try_resolve_path( + params, dst_path, raise_on_ambiguous=True) + except FolderPathAmbiguousError as e: + raise CommandError('mv', str(e)) if dst is None: raise CommandError('mv', 'Destination path should be existing folder') dst_folder, name = dst @@ -867,7 +890,11 @@ def execute(self, params, **kwargs): raise CommandError('mv', 'Source and Destination folders are the same') source.append((src_folder, None)) else: - src = try_resolve_path(params, src_path) + try: + src = try_resolve_path( + params, src_path, raise_on_ambiguous=True) + except FolderPathAmbiguousError as e: + raise CommandError('mv', str(e)) if src is None: raise CommandError('mv', 'Source path should be existing record or folder') diff --git a/keepercommander/commands/ksm.py b/keepercommander/commands/ksm.py index a80c5ad05..fe61d484b 100644 --- a/keepercommander/commands/ksm.py +++ b/keepercommander/commands/ksm.py @@ -1641,18 +1641,6 @@ def get_app_record(params, app_name_or_uid): if nsf_rec: return nsf_rec - for rec_cache_val in params.record_cache.values(): - rec = KSMCommand._app_record_from_cache_entry(rec_cache_val) - if not rec: - continue - r_uid = rec.get('record_uid') - try: - r_unencr_dict = json.loads(rec.get('data_unencrypted').decode('utf-8')) - except Exception: - continue - if r_unencr_dict.get('title') == app_name_or_uid or r_uid == app_name_or_uid: - return rec - resolved_uid = resolve_nested_share_record_uid(params, app_name_or_uid) if resolved_uid: if resolved_uid in getattr(params, 'record_cache', {}): diff --git a/keepercommander/commands/nested_share_folder/display_commands.py b/keepercommander/commands/nested_share_folder/display_commands.py index 54df4ef10..4e0cc7063 100644 --- a/keepercommander/commands/nested_share_folder/display_commands.py +++ b/keepercommander/commands/nested_share_folder/display_commands.py @@ -137,42 +137,13 @@ def execute(self, params, **kwargs): @staticmethod def _resolve_as_folder(params, uid): - nsf_folders = getattr(params, 'nested_share_folders', {}) - if uid in nsf_folders: - return uid - lower = uid.lower() - matches = [f for f, o in nsf_folders.items() if o.get('name', '').lower() == lower] - if len(matches) == 1: - return matches[0] - if len(matches) > 1: - from keepercommander.display import dump_report_data - rows = [[f, nsf_folders[f].get('name', '')] for f in matches] - dump_report_data(rows, ['Folder UID', 'Name'], - title='Multiple folders match the name', row_number=True) - return None + resolved = _nsf.resolve_folder_identifier(params, uid) + return resolved if resolved in getattr(params, 'nested_share_folders', {}) else None @staticmethod def _resolve_as_record(params, uid): - nsf_records = getattr(params, 'nested_share_records', {}) - if uid in nsf_records: - return uid - lower = uid.lower() - nsf_data = getattr(params, 'nested_share_record_data', {}) - matches = [] - for ruid in nsf_records: - if ruid in nsf_data and 'data_json' in nsf_data[ruid]: - title = nsf_data[ruid]['data_json'].get('title', '') - if title and lower in title.lower(): - matches.append((ruid, title)) - if len(matches) == 1: - return matches[0][0] - if len(matches) > 1: - from keepercommander.display import dump_report_data - dump_report_data( - sorted(matches, key=lambda x: x[1]), - ['Record UID', 'Title'], - title='Multiple records match the title', row_number=True) - return None + resolved = _nsf.resolve_nested_share_record_uid(params, uid) + return resolved if resolved in getattr(params, 'nested_share_records', {}) else None # ── Record display ──────────────────────────────────────────────── diff --git a/keepercommander/commands/nested_share_folder/folder_commands.py b/keepercommander/commands/nested_share_folder/folder_commands.py index 061bef518..6f1e70561 100644 --- a/keepercommander/commands/nested_share_folder/folder_commands.py +++ b/keepercommander/commands/nested_share_folder/folder_commands.py @@ -21,7 +21,7 @@ from ..base import Command from ...error import CommandError from ... import nested_share_folder as _nsf -from ... import vault_extensions +from ... import utils, vault_extensions from .helpers import ( ROOT_FOLDER_UID, normalize_parent_uid, resolve_folder_uid, parse_expiration, @@ -32,6 +32,7 @@ ensure_nested_share_folder, is_nested_share_folder, is_nested_share_folder_owner_email, owner_share_target_message, ) +from ...nested_share_folder.folder_api import _is_current_account_folder_owner from .parsers import ( nested_share_folder_mkdir_parser, nested_share_folder_update_parser, @@ -134,6 +135,14 @@ def _cache_new_folder(params, folder_uid, name, parent_uid, folder_key=None): 'name': name, 'parent_uid': parent_uid or '', } + # The ownership fields are normally populated by sync. Preserve them + # for folders created in the current session so the ownership-aware + # resolver can still resolve a path before the next sync. + if isinstance(getattr(params, 'user', None), str) and params.user: + entry['owner_username'] = params.user + if getattr(params, 'account_uid_bytes', None): + entry['owner_account_uid'] = utils.base64_url_encode( + params.account_uid_bytes) if folder_key: entry['folder_key_unencrypted'] = folder_key nsf[folder_uid] = entry @@ -155,9 +164,12 @@ def _find_existing_child(params, folder_name, parent_uid): """Find an existing NSF folder named *folder_name* whose parent matches *parent_uid*. ``parent_uid=None`` means "root level". """ + nsf_folders = getattr(params, 'nested_share_folders', {}) name_lower = folder_name.lower() looking_for_root = not parent_uid + owned_matches = [] + unknown_matches = [] for fuid, fobj in nsf_folders.items(): if fobj.get('name', '').lower() != name_lower: continue @@ -167,12 +179,24 @@ def _find_existing_child(params, folder_name, parent_uid): normalized in ('', 'root') or (raw_parent and raw_parent not in nsf_folders) ) - if looking_for_root: - if is_root_child: - return fuid - else: - if raw_parent == parent_uid: - return fuid + parent_matches = is_root_child if looking_for_root else raw_parent == parent_uid + if not parent_matches: + continue + + if _is_current_account_folder_owner(params, fobj): + owned_matches.append(fuid) + elif not (fobj.get('owner_username') or fobj.get('owner_account_uid')): + + unknown_matches.append(fuid) + + matches = owned_matches or unknown_matches + if len(matches) > 1: + raise CommandError( + 'nsf-mkdir', + f'Multiple Nested Share Folders match "{folder_name}". ' + 'Use a folder UID or a more specific path.') + if matches: + return matches[0] return None @@ -677,4 +701,4 @@ def execute(self, params, **kwargs): destination = kwargs.get('dst') if not source or not destination: raise CommandError('nsf-move', 'Both source and destination are required.') - move_nested_share_item(params, source, destination) \ No newline at end of file + move_nested_share_item(params, source, destination) diff --git a/keepercommander/commands/nested_share_folder/record_commands.py b/keepercommander/commands/nested_share_folder/record_commands.py index faf277ce1..101116e1d 100644 --- a/keepercommander/commands/nested_share_folder/record_commands.py +++ b/keepercommander/commands/nested_share_folder/record_commands.py @@ -504,12 +504,11 @@ def _resolve_target(cls, params, target, records, nsf_records, raise CommandError('nsf-shortcut list', f'Record UID {target} does not have shortcuts') return {target} - lower = target.casefold() - for uid in nsf_records: - if cls._record_title(uid, nsf_record_data).casefold() == lower: - if uid not in records: - raise CommandError('nsf-shortcut list', f'Record "{target}" does not have shortcuts') - return {uid} + resolved_record = _nsf.resolve_nested_share_record_uid(params, target) + if resolved_record in nsf_records: + if resolved_record not in records: + raise CommandError('nsf-shortcut list', f'Record "{target}" does not have shortcuts') + return {resolved_record} resolved_folder = _nsf.resolve_folder_identifier(params, target) if resolved_folder: @@ -536,7 +535,7 @@ def execute(self, params, **kwargs): nsf_folders = getattr(params, 'nested_share_folders', {}) nsf_record_data = getattr(params, 'nested_share_record_data', {}) - record_uid = self._resolve_record(target, nsf_records, nsf_record_data) + record_uid = self._resolve_record(params, target, nsf_records, nsf_record_data) keep_folder_uid = self._resolve_keep_folder(params, kwargs.get('folder'), nsf_folders) records = NestedShareRecordShortcutCommand.get_record_shortcuts(params) @@ -581,20 +580,12 @@ def execute(self, params, **kwargs): target, keep_name, len(folders_to_remove)) @staticmethod - def _resolve_record(target, nsf_records, nsf_record_data=None): + def _resolve_record(params, target, nsf_records, nsf_record_data=None): if target in nsf_records: return target - lower = target.casefold() - if nsf_record_data: - for uid in nsf_records: - rd = nsf_record_data.get(uid) or {} - dj = rd.get('data_json') or {} - if (dj.get('title') or '').casefold() == lower: - return uid - - for uid, rec in nsf_records.items(): - if rec.get('title', '').casefold() == lower: - return uid + resolved_record = _nsf.resolve_nested_share_record_uid(params, target) + if resolved_record in nsf_records: + return resolved_record raise CommandError('nsf-shortcut keep', f'Record "{target}" not found in Nested Share Folder') @staticmethod diff --git a/keepercommander/commands/nested_share_folder/sharing_commands.py b/keepercommander/commands/nested_share_folder/sharing_commands.py index 5e66a7545..7fe9b546e 100644 --- a/keepercommander/commands/nested_share_folder/sharing_commands.py +++ b/keepercommander/commands/nested_share_folder/sharing_commands.py @@ -376,12 +376,25 @@ def _resolve_folder(nsf_folders, folder_name, params=None): # friendly cross-type message instead of "not found". ensure_nested_share_folder(params, resolved, 'nsf-record-permission', identifier=folder_name) + # The ownership-aware resolver deliberately returns None for an + # ambiguous or unowned name. Do not fall back to a first-match + # scan, which would reintroduce the shadowing bug. + raise CommandError('nsf-record-permission', + f'Folder "{folder_name}" not found or is ambiguous') + if folder_name in nsf_folders: return folder_name, nsf_folders[folder_name].get('name', folder_name) - lower = folder_name.lower() - for fuid, fobj in nsf_folders.items(): - if fobj.get('name', '').lower() == lower: - return fuid, fobj.get('name', fuid) + matches = [ + (fuid, fobj.get('name', fuid)) + for fuid, fobj in nsf_folders.items() + if fobj.get('name', '').casefold() == folder_name.casefold() + ] + if len(matches) == 1: + return matches[0] + if len(matches) > 1: + raise CommandError('nsf-record-permission', + f'Multiple Nested Share Folders match "{folder_name}". ' + 'Use a folder UID.') raise CommandError('nsf-record-permission', f'Folder "{folder_name}" not found') @staticmethod diff --git a/keepercommander/commands/record_edit.py b/keepercommander/commands/record_edit.py index 4b6bd2da3..a6233cb17 100644 --- a/keepercommander/commands/record_edit.py +++ b/keepercommander/commands/record_edit.py @@ -915,7 +915,10 @@ def execute(self, params, **kwargs): is_self_destruct_link = False # Time-based only, can be used multiple times logging.info('--send-email used without --self-destruct, creating 24 hour time-based share link') - folder_uid = FolderMixin.resolve_folder(params, kwargs.get('folder')) + folder_name = kwargs.get('folder') + folder_uid = FolderMixin.resolve_folder(params, folder_name) + if folder_name and not folder_uid: + raise CommandError('record-add', f'Folder "{folder_name}" cannot be found') self.warnings.clear() title = kwargs.get('title') diff --git a/keepercommander/nested_share_folder/folder_api.py b/keepercommander/nested_share_folder/folder_api.py index 1aadbae34..f96e8d390 100644 --- a/keepercommander/nested_share_folder/folder_api.py +++ b/keepercommander/nested_share_folder/folder_api.py @@ -151,29 +151,168 @@ def folder_access_update_v3(params, folder_access_adds=None, # ══════════════════════════════════════════════════════════════════════════ def resolve_folder_identifier(params, folder_identifier): - if folder_identifier in getattr(params, 'nested_share_folders', {}): + if not isinstance(folder_identifier, str): + return None + + nsf_folders = getattr(params, 'nested_share_folders', {}) + subfolder_cache = getattr(params, 'subfolder_cache', {}) + folder_cache = getattr(params, 'folder_cache', {}) + + # UIDs are unambiguous and are the supported way to select a shared-in + # folder whose name is not owned by the current account. + if folder_identifier in nsf_folders: return folder_identifier - if folder_identifier in getattr(params, 'subfolder_cache', {}): + if folder_identifier in subfolder_cache: return folder_identifier - if folder_identifier in getattr(params, 'folder_cache', {}): + if folder_identifier in folder_cache: return folder_identifier - matching = [uid for uid, obj in getattr(params, 'nested_share_folders', {}).items() - if obj.get('name', '').lower() == folder_identifier.lower()] + # A slash makes the identifier a path, not a folder name. Resolve it as a + # path before considering any flat-name match. This prevents a shared-in + # folder named "Clients/Acme" from shadowing the user's own Clients/Acme + # hierarchy. + if '/' in folder_identifier or '\\' in folder_identifier: + nsf_uid = _resolve_owned_nsf_path(params, folder_identifier) + if nsf_uid: + return nsf_uid + + folder_cache = getattr(params, 'folder_cache', {}) + if isinstance(folder_cache, dict) and folder_cache: + rs = try_resolve_path(params, folder_identifier.replace('\\', '/')) + if rs is not None: + folder, pattern = rs + if (folder and not pattern and + _is_safe_path_folder(params, folder)): + return folder.uid + return None + + # Flat-name matching is intentionally limited to folders owned by the + # current account. Shared-in folders can still be selected by UID. + matching = [ + uid for uid, obj in nsf_folders.items() + if obj.get('name', '').casefold() == folder_identifier.casefold() + and _is_current_account_folder_owner(params, obj) + ] if len(matching) == 1: return matching[0] if len(matching) > 1: logging.warning("Multiple folders match '%s'. Use UID instead.", folder_identifier) return None - rs = try_resolve_path(params, folder_identifier) - if rs is not None: - folder, pattern = rs - if folder and not pattern: - return folder.uid + folder_cache = getattr(params, 'folder_cache', {}) + if isinstance(folder_cache, dict) and folder_cache: + rs = try_resolve_path(params, folder_identifier) + if rs is not None: + folder, pattern = rs + if (folder and not pattern and + _is_safe_path_folder(params, folder)): + return folder.uid return None +def _is_current_account_folder_owner(params, folder_obj): + """Return whether *folder_obj* is owned by the current account. + + Missing ownership metadata is treated as unknown, not as ownership. This + is important because shared-in folders are present in the same local cache + as folders owned by the current account. + """ + owner_username = folder_obj.get('owner_username') + current_username = getattr(params, 'user', None) + if (isinstance(owner_username, str) and owner_username and + isinstance(current_username, str) and current_username and + owner_username.casefold() == current_username.casefold()): + return True + + owner_account_uid = folder_obj.get('owner_account_uid') + account_uid_bytes = getattr(params, 'account_uid_bytes', None) + if owner_account_uid and account_uid_bytes: + try: + return owner_account_uid == utils.base64_url_encode(account_uid_bytes) + except Exception: + pass + return False + + +def _resolve_owned_nsf_path(params, folder_identifier): + """Resolve a path through current-account-owned NSF folders. + + NSF folder names may contain a literal slash, represented by ``//`` in a + path. The parser mirrors ``NestedShareFolderMkdirCommand._parse_path`` but + is kept here to avoid importing command-layer code into the API module. + """ + nsf_folders = getattr(params, 'nested_share_folders', {}) + normalized = folder_identifier.replace('\\', '/') + absolute = normalized.startswith('/') + if absolute: + normalized = normalized[1:] + + sentinel = '\x00' + components = [ + part.replace(sentinel, '/').strip() + for part in normalized.replace('//', sentinel).split('/') + if part.replace(sentinel, '/').strip() + ] + if not components: + return None + + current = None + if not absolute: + candidate = getattr(params, 'current_folder', None) + if candidate in nsf_folders: + current = candidate + + for component in components: + matches = [] + for uid, folder_obj in nsf_folders.items(): + if not _is_current_account_folder_owner(params, folder_obj): + continue + if folder_obj.get('name', '').casefold() != component.casefold(): + continue + + parent_uid = folder_obj.get('parent_uid') or '' + if current: + if parent_uid == current: + matches.append(uid) + else: + # Missing parents and server root sentinels represent the + # Nested Share Folder root. + if not parent_uid or parent_uid not in nsf_folders: + matches.append(uid) + + if len(matches) != 1: + return None + current = matches[0] + + return current + + +def _is_safe_path_folder(params, folder): + """Return whether a legacy path result contains no unowned NSF folder.""" + nsf_folders = getattr(params, 'nested_share_folders', {}) + folder_cache = getattr(params, 'folder_cache', {}) + visited = set() + current = folder + + while current is not None: + folder_uid = getattr(current, 'uid', None) + if not folder_uid or folder_uid in visited: + break + visited.add(folder_uid) + + if (folder_uid in nsf_folders and + not _is_current_account_folder_owner( + params, nsf_folders[folder_uid])): + return False + + parent_uid = getattr(current, 'parent_uid', None) + if not parent_uid or not isinstance(folder_cache, dict): + break + current = folder_cache.get(parent_uid) + + return True + + # ══════════════════════════════════════════════════════════════════════════ # High-level: create / create_batch # ══════════════════════════════════════════════════════════════════════════ diff --git a/keepercommander/nested_share_folder/removal_api.py b/keepercommander/nested_share_folder/removal_api.py index f30e8b3ac..2bd53fdb8 100644 --- a/keepercommander/nested_share_folder/removal_api.py +++ b/keepercommander/nested_share_folder/removal_api.py @@ -101,9 +101,17 @@ def resolve_nested_share_record_uid(params, identifier): 2. Direct UID match in record_cache (vault records) 3. Title match from decrypted nested_share_record_data 4. Title match from record_cache (data_unencrypted) + + When a title has exactly one accessible match and no owned match, the + shared-in record is retained for compatibility with existing shared-record + workflows. Callers that require an owned record should use an explicit UID + or apply an owner-only check before mutating it. """ import json + if not isinstance(identifier, str): + return None + nsf = getattr(params, 'nested_share_records', {}) if identifier in nsf: return identifier @@ -115,14 +123,16 @@ def resolve_nested_share_record_uid(params, identifier): lower = identifier.casefold() nsf_data = getattr(params, 'nested_share_record_data', {}) + nsf_matches = [] for uid in nsf: rd = nsf_data.get(uid, {}) dj = rd.get('data_json', {}) if isinstance(dj, dict): t = dj.get('title', '') if isinstance(t, str) and t.casefold() == lower: - return uid + nsf_matches.append(uid) + classic_matches = [] for uid, rec in rc.items(): if uid in nsf: continue @@ -132,31 +142,71 @@ def resolve_nested_share_record_uid(params, identifier): dj = json.loads(data.decode('utf-8') if isinstance(data, bytes) else data) t = dj.get('title', '') if isinstance(t, str) and t.casefold() == lower: - return uid + classic_matches.append(uid) except Exception: pass + matches = nsf_matches + classic_matches + if not matches: + return None + + owned_matches = [ + uid for uid in matches + if _is_current_account_record_owner( + params, uid, + nsf_data.get(uid, {}) if uid in nsf else rc.get(uid, {}), + rc.get(uid, {}), + ) + ] + if len(owned_matches) == 1: + return owned_matches[0] + if len(owned_matches) > 1: + logging.warning("Multiple owned records match '%s'. Use UID instead.", identifier) + return None + if len(matches) == 1: + return matches[0] + + logging.warning("Multiple records match '%s'. Use UID instead.", identifier) return None +def _is_current_account_record_owner(params, record_uid, nsf_data, record_obj): + """Return whether cached ownership identifies the current account.""" + current_username = getattr(params, 'user', None) + current_account_uid = '' + account_uid_bytes = getattr(params, 'account_uid_bytes', None) + if account_uid_bytes: + try: + from .. import utils + current_account_uid = utils.base64_url_encode(account_uid_bytes) + except Exception: + pass + + for obj in (nsf_data or {}, record_obj or {}, + getattr(params, 'meta_data_cache', {}).get(record_uid, {})): + owner_username = obj.get('user_username') or obj.get('owner_username') + if (isinstance(owner_username, str) and owner_username and + isinstance(current_username, str) and current_username and + owner_username.casefold() == current_username.casefold()): + return True + + owner_uid = obj.get('user_account_uid') or obj.get('owner_account_uid') + if owner_uid and current_account_uid and owner_uid == current_account_uid: + return True + + owner_info = getattr(params, 'record_owner_cache', {}).get(record_uid) + if owner_info and owner_info.owner is True: + return True + return False + + def resolve_nested_share_folder_uid(params, identifier): """Resolve a folder identifier (UID or name) to a Nested Share Folder UID.""" - nsf = getattr(params, 'nested_share_folders', {}) - if identifier in nsf: - return identifier - fc = getattr(params, 'folder_cache', {}) - if identifier in fc: - return identifier - lower = identifier.casefold() - for uid, f in nsf.items(): - name = f.get('name', '') - if isinstance(name, str) and name.casefold() == lower: - return uid - for uid, f in getattr(params, 'subfolder_cache', {}).items(): - name = f.get('name', '') - if isinstance(name, str) and name.casefold() == lower: - return uid - return None + # Keep all NSF folder lookups behind the ownership-aware resolver. This + # helper is used by commands outside the main folder API, so an independent + # name scan here would reintroduce the shared-in shadowing vulnerability. + from .folder_api import resolve_folder_identifier + return resolve_folder_identifier(params, identifier) # ══════════════════════════════════════════════════════════════════════════ diff --git a/keepercommander/subfolder.py b/keepercommander/subfolder.py index 895999716..ad65cfdad 100644 --- a/keepercommander/subfolder.py +++ b/keepercommander/subfolder.py @@ -14,6 +14,15 @@ from .params import KeeperParams +class FolderPathAmbiguousError(ValueError): + """Raised when a path component resolves to more than one folder.""" + + def __init__(self, component): + super().__init__( + f"Multiple folders match path component '{component}'. " + "Use a folder UID instead.") + + def get_folder_path(params, folder_uid, delimiter='/'): uid = folder_uid path = '' @@ -109,7 +118,8 @@ def contained_folders(params, folders, component): return result -def lookup_path(params, folder, components): +def lookup_path(params, folder, components, find_all_matches=False, + raise_on_ambiguous=False): # type: (KeeperParams, BaseFolderNode, List[Optional[str]]) -> Tuple[int, List[Optional[BaseFolderNode]]] """ Lookup path of components within the folder cache. @@ -123,6 +133,11 @@ def lookup_path(params, folder, components): child_folders = contained_folders(params, folders, component) if not child_folders: break + if len(child_folders) > 1: + if raise_on_ambiguous: + raise FolderPathAmbiguousError(component) + if not find_all_matches: + return remainder, None folders = child_folders remainder = index + 1 return remainder, folders @@ -143,7 +158,8 @@ def path_split(params, folder, path_string): return folder, components -def try_resolve_path(params, path, find_all_matches=False): +def try_resolve_path(params, path, find_all_matches=False, + raise_on_ambiguous=False): # type: (KeeperParams, str, bool) -> Tuple[Union[List[BaseFolderNode], BaseFolderNode, None], Optional[str]] """ Look up the final keepercommander.subfolder.UserFolderNode and name of the final component(s). @@ -164,7 +180,12 @@ def try_resolve_path(params, path, find_all_matches=False): folder, components = path_split(params, folder, path) - remainder, folders = lookup_path(params, folder, components) + remainder, folders = lookup_path( + params, folder, components, + find_all_matches=find_all_matches, + raise_on_ambiguous=raise_on_ambiguous) + if folders is None: + return None tail = components[remainder:] diff --git a/unit-tests/test_command_folder.py b/unit-tests/test_command_folder.py index fcfd4beab..238e4eca1 100644 --- a/unit-tests/test_command_folder.py +++ b/unit-tests/test_command_folder.py @@ -115,6 +115,32 @@ def test_remove_non_existing_folder(self): with self.assertRaises(CommandError): cmd.execute(params, folder='Invalid Name') + def test_remove_rejects_ambiguous_folder_path(self): + from keepercommander.subfolder import UserFolderNode + + params = get_synced_params() + first = UserFolderNode() + first.uid = 'a' * 22 + first.name = 'Ambiguous' + first.parent_uid = None + first.subfolders = [] + second = UserFolderNode() + second.uid = 'b' * 22 + second.name = 'Ambiguous' + second.parent_uid = None + second.subfolders = [] + params.folder_cache[first.uid] = first + params.folder_cache[second.uid] = second + params.root_folder.subfolders.extend([first.uid, second.uid]) + + cmd = folder.FolderRemoveCommand() + with mock.patch( + 'keepercommander.enforcement.MasterPasswordReentryEnforcer.check_and_enforce', + return_value=True): + with self.assertRaises(CommandError) as ctx: + cmd.execute(params, pattern=['Ambiguous'], force=True, quiet=True) + self.assertIn('Multiple folders match', str(ctx.exception)) + def test_delete_folders(self): params = get_synced_params() cmd = folder.FolderRemoveCommand() diff --git a/unit-tests/test_command_ksm.py b/unit-tests/test_command_ksm.py index ab0c98716..e126aec68 100644 --- a/unit-tests/test_command_ksm.py +++ b/unit-tests/test_command_ksm.py @@ -160,6 +160,46 @@ def test_get_app_record_by_nsf_path(self, mock_resolve): self.assertIsNotNone(rec) self.assertEqual(rec['record_uid'], app_uid) + def test_get_app_record_rejects_duplicate_classic_titles(self): + params = self._make_params() + first_uid = 'classicApp000000000001' + second_uid = 'classicApp000000000002' + app_data = lambda uid: { + 'record_uid': uid, + 'version': 5, + 'data_unencrypted': b'{"type":"app","title":"Duplicate App"}', + 'record_key_unencrypted': b'app_key', + } + params.record_cache = { + first_uid: app_data(first_uid), + second_uid: app_data(second_uid), + } + params.user = 'victim@example.com' + params.account_uid_bytes = None + params.meta_data_cache = {} + params.record_owner_cache = {} + + self.assertIsNone(KSMCommand.get_app_record(params, 'Duplicate App')) + + def test_get_app_record_rejects_duplicate_nsf_titles(self): + params = self._make_params() + first_uid = 'nsfApp00000000000001' + second_uid = 'nsfApp00000000000002' + params.nested_share_records = { + first_uid: {'version': 5, 'record_key_unencrypted': b'a'}, + second_uid: {'version': 5, 'record_key_unencrypted': b'b'}, + } + params.nested_share_record_data = { + first_uid: {'data_json': {'type': 'app', 'title': 'Duplicate App'}}, + second_uid: {'data_json': {'type': 'app', 'title': 'Duplicate App'}}, + } + params.user = 'victim@example.com' + params.account_uid_bytes = None + params.meta_data_cache = {} + params.record_owner_cache = {} + + self.assertIsNone(KSMCommand.get_app_record(params, 'Duplicate App')) + def test_get_ksm_app_display_info_from_nsf_metadata(self): params = self._make_params() app_uid = 'appUid00000000000001' diff --git a/unit-tests/test_command_record.py b/unit-tests/test_command_record.py index 584f5172d..d16201fc5 100644 --- a/unit-tests/test_command_record.py +++ b/unit-tests/test_command_record.py @@ -97,6 +97,27 @@ def artf(p, r, f): self.assertIsNotNone(field) self.assertEqual(field.get_default_value(str), 'BBB') + @mock.patch('keepercommander.commands.record_edit.FolderMixin.resolve_folder', + return_value=None) + @mock.patch('keepercommander.record_management.add_record_to_folder') + def test_add_command_rejects_unresolvable_folder(self, mock_add, mock_resolve): + params = get_synced_params() + cmd = record_edit.RecordAddCommand() + + with self.assertRaises(CommandError) as ctx: + cmd.execute( + params, + folder='Work', + title='Ambiguous Folder POC', + record_type='login', + fields=[], + force=True, + ) + + self.assertIn('Folder "Work" cannot be found', str(ctx.exception)) + mock_resolve.assert_called_once_with(params, 'Work') + mock_add.assert_not_called() + def _run_add(self, params, **kwargs): """Run record-add with the API mocked; return the TypedRecord that would be saved.""" cmd = record_edit.RecordAddCommand() diff --git a/unit-tests/test_nested_share_folder.py b/unit-tests/test_nested_share_folder.py index 0c3fac318..e62c4ca5f 100644 --- a/unit-tests/test_nested_share_folder.py +++ b/unit-tests/test_nested_share_folder.py @@ -105,6 +105,310 @@ def _make_sharing_status(record_uid, recipient_uid_bytes=None): class TestCommandHelpers(TestCase): + def test_folder_path_prefers_owned_nsf_hierarchy(self): + from keepercommander.nested_share_folder.folder_api import resolve_folder_identifier + + parent_uid, parent = _make_folder(name='Clients') + child_uid, child = _make_folder(name='Acme', parent_uid=parent_uid) + attacker_uid, attacker = _make_folder(name='Clients/Acme') + parent['owner_username'] = 'victim@example.com' + child['owner_username'] = 'victim@example.com' + attacker['owner_username'] = 'attacker@example.com' + params = _make_params( + user='victim@example.com', + nested_share_folders={ + parent_uid: parent, child_uid: child, attacker_uid: attacker, + }, + ) + + self.assertEqual( + resolve_folder_identifier(params, 'Clients/Acme'), child_uid) + + def test_folder_path_does_not_flat_match_shared_in_folder(self): + from keepercommander.nested_share_folder.folder_api import resolve_folder_identifier + + attacker_uid, attacker = _make_folder(name='Clients/Acme') + attacker['owner_username'] = 'attacker@example.com' + params = _make_params( + user='victim@example.com', + nested_share_folders={attacker_uid: attacker}, + ) + + self.assertIsNone(resolve_folder_identifier(params, 'Clients/Acme')) + + def test_folder_path_does_not_use_unowned_nsf_folder_cache(self): + from keepercommander.nested_share_folder.folder_api import resolve_folder_identifier + from keepercommander.subfolder import ( + NestedShareFolderNode, RootFolderNode, + ) + + parent_uid, parent_obj = _make_folder(name='Clients') + child_uid, child_obj = _make_folder(name='Acme', parent_uid=parent_uid) + parent_obj['owner_username'] = 'attacker@example.com' + child_obj['owner_username'] = 'attacker@example.com' + + root = RootFolderNode() + parent = NestedShareFolderNode() + parent.uid = parent_uid + parent.parent_uid = None + parent.name = 'Clients' + parent.subfolders = [child_uid] + child = NestedShareFolderNode() + child.uid = child_uid + child.parent_uid = parent_uid + child.name = 'Acme' + root.subfolders = [parent_uid] + + params = _make_params( + user='victim@example.com', + root_folder=root, + current_folder='', + nested_share_folders={ + parent_uid: parent_obj, + child_uid: child_obj, + }, + folder_cache={ + parent_uid: parent, + child_uid: child, + }, + ) + + self.assertIsNone(resolve_folder_identifier(params, 'Clients/Acme')) + + def test_flat_folder_name_requires_current_account_ownership(self): + from keepercommander.nested_share_folder.folder_api import resolve_folder_identifier + + owned_uid, owned = _make_folder(name='Owned') + shared_uid, shared = _make_folder(name='Shared') + owned['owner_username'] = 'victim@example.com' + shared['owner_username'] = 'attacker@example.com' + params = _make_params( + user='victim@example.com', + nested_share_folders={owned_uid: owned, shared_uid: shared}, + ) + + self.assertEqual(resolve_folder_identifier(params, 'Owned'), owned_uid) + self.assertIsNone(resolve_folder_identifier(params, 'Shared')) + self.assertEqual(resolve_folder_identifier(params, shared_uid), shared_uid) + + def test_flat_folder_name_does_not_fall_back_to_unowned_nsf_cache(self): + from keepercommander.nested_share_folder.folder_api import resolve_folder_identifier + from keepercommander.subfolder import NestedShareFolderNode, RootFolderNode + + shared_uid, shared_obj = _make_folder(name='Shared') + shared_obj['owner_username'] = 'attacker@example.com' + + root = RootFolderNode() + shared_node = NestedShareFolderNode() + shared_node.uid = shared_uid + shared_node.parent_uid = None + shared_node.name = 'Shared' + root.subfolders = [shared_uid] + + params = _make_params( + user='victim@example.com', + root_folder=root, + current_folder='', + nested_share_folders={shared_uid: shared_obj}, + folder_cache={shared_uid: shared_node}, + ) + + self.assertIsNone(resolve_folder_identifier(params, 'Shared')) + + def test_duplicate_owned_nsf_folder_name_is_ambiguous(self): + from keepercommander.nested_share_folder.folder_api import resolve_folder_identifier + + first_uid, first = _make_folder(name='Work') + second_uid, second = _make_folder(name='Work') + first['owner_username'] = 'victim@example.com' + second['owner_username'] = 'victim@example.com' + params = _make_params( + user='victim@example.com', + nested_share_folders={first_uid: first, second_uid: second}, + ) + + self.assertIsNone(resolve_folder_identifier(params, 'Work')) + + def test_nsf_mkdir_child_resolution_ignores_shared_in_folder(self): + from keepercommander.commands.nested_share_folder.folder_commands import ( + NestedShareFolderMkdirCommand, + ) + + owned_uid, owned = _make_folder(name='Work') + owned['owner_username'] = 'victim@example.com' + shared_uid, shared = _make_folder(name='Work') + shared['owner_username'] = 'attacker@example.com' + params = _make_params( + user='victim@example.com', + nested_share_folders={owned_uid: owned, shared_uid: shared}, + ) + + self.assertEqual( + NestedShareFolderMkdirCommand._find_existing_child( + params, 'Work', None), + owned_uid) + + def test_nsf_mkdir_child_resolution_rejects_duplicate_owned_folders(self): + from keepercommander.commands.nested_share_folder.folder_commands import ( + NestedShareFolderMkdirCommand, + ) + + first_uid, first = _make_folder(name='Work') + second_uid, second = _make_folder(name='Work') + first['owner_username'] = 'victim@example.com' + second['owner_username'] = 'victim@example.com' + params = _make_params( + user='victim@example.com', + nested_share_folders={first_uid: first, second_uid: second}, + ) + + with self.assertRaises(CommandError): + NestedShareFolderMkdirCommand._find_existing_child( + params, 'Work', None) + + def test_record_permission_rejects_unowned_folder_fallback(self): + from keepercommander.commands.nested_share_folder.sharing_commands import ( + NestedShareRecordPermissionCommand, + ) + + shared_uid, shared = _make_folder(name='Shared') + shared['owner_username'] = 'attacker@example.com' + params = _make_params( + user='victim@example.com', + nested_share_folders={shared_uid: shared}, + ) + + with self.assertRaises(CommandError): + NestedShareRecordPermissionCommand._resolve_folder( + params.nested_share_folders, 'Shared', params) + + def test_shortcut_record_resolution_rejects_duplicate_titles(self): + from keepercommander.commands.nested_share_folder.record_commands import ( + NestedShareRecordShortcutListCommand, + NestedShareRecordShortcutKeepCommand, + ) + + first_uid, first = _make_record(title='Duplicate Shortcut') + second_uid, second = _make_record(title='Duplicate Shortcut') + params = _make_params( + nested_share_records={first_uid: first, second_uid: second}, + nested_share_record_data={ + first_uid: {'data_json': {'title': 'Duplicate Shortcut'}}, + second_uid: {'data_json': {'title': 'Duplicate Shortcut'}}, + }, + ) + records = { + first_uid: {'folder-a', 'folder-b'}, + second_uid: {'folder-a', 'folder-b'}, + } + + with self.assertRaises(CommandError): + NestedShareRecordShortcutListCommand._resolve_target( + params, 'Duplicate Shortcut', records, + params.nested_share_records, + params.nested_share_record_data, + params.nested_share_folders) + with self.assertRaises(CommandError): + NestedShareRecordShortcutKeepCommand._resolve_record( + params, 'Duplicate Shortcut', + params.nested_share_records, + params.nested_share_record_data) + + def test_record_title_prefers_owned_record(self): + from keepercommander.nested_share_folder.removal_api import ( + resolve_nested_share_record_uid) + from keepercommander.params import RecordOwner + + shared_uid, shared = _make_record(title='Prod DB') + owned_uid, owned = _make_record(title='Prod DB') + params = _make_params( + user='victim@example.com', + nested_share_records={shared_uid: shared}, + nested_share_record_data={ + shared_uid: { + 'data_json': {'title': 'Prod DB'}, + 'user_username': 'attacker@example.com', + }, + }, + record_cache={ + owned_uid: { + 'data_unencrypted': json.dumps({'title': 'Prod DB'}).encode(), + }, + }, + record_owner_cache={ + owned_uid: RecordOwner(True, _ACCOUNT_UID), + }, + ) + + self.assertEqual( + resolve_nested_share_record_uid(params, 'Prod DB'), owned_uid) + + def test_record_title_fails_on_ambiguous_shared_records(self): + from keepercommander.nested_share_folder.removal_api import ( + resolve_nested_share_record_uid) + + first_uid, first = _make_record(title='Duplicate') + second_uid, second = _make_record(title='Duplicate') + params = _make_params( + nested_share_records={first_uid: first, second_uid: second}, + nested_share_record_data={ + first_uid: {'data_json': {'title': 'Duplicate'}}, + second_uid: {'data_json': {'title': 'Duplicate'}}, + }, + ) + + self.assertIsNone( + resolve_nested_share_record_uid(params, 'Duplicate')) + + def test_record_title_fails_on_duplicate_classic_records(self): + from keepercommander.nested_share_folder.removal_api import ( + resolve_nested_share_record_uid) + + first_uid, first = _make_record(title='Duplicate Classic') + second_uid, second = _make_record(title='Duplicate Classic') + first['owner_account_uid'] = 'other-account' + second['owner_account_uid'] = 'other-account' + params = _make_params( + user='victim@example.com', + nested_share_records={}, + record_cache={ + first_uid: { + 'data_unencrypted': json.dumps( + {'title': 'Duplicate Classic'}).encode(), + }, + second_uid: { + 'data_unencrypted': json.dumps( + {'title': 'Duplicate Classic'}).encode(), + }, + }, + meta_data_cache={ + first_uid: {'owner_account_uid': 'other-account'}, + second_uid: {'owner_account_uid': 'other-account'}, + }, + ) + + self.assertIsNone( + resolve_nested_share_record_uid(params, 'Duplicate Classic')) + + def test_record_title_allows_single_shared_record(self): + from keepercommander.nested_share_folder.removal_api import ( + resolve_nested_share_record_uid) + + shared_uid, shared = _make_record(title='Shared Record') + params = _make_params( + nested_share_records={shared_uid: shared}, + nested_share_record_data={ + shared_uid: { + 'data_json': {'title': 'Shared Record'}, + 'user_username': 'owner@example.com', + }, + }, + user='victim@example.com', + ) + + self.assertEqual( + resolve_nested_share_record_uid(params, 'Shared Record'), shared_uid) + def test_parse_expiration_none(self): from keepercommander.commands.nested_share_folder.helpers import parse_expiration self.assertIsNone(parse_expiration(None, None, 'test')) @@ -491,6 +795,24 @@ def test_add_record_rejects_restricted_record_type(self, mock_create): self.assertIn('restricted', str(ctx.exception).lower()) mock_create.assert_not_called() + @patch('keepercommander.commands.nested_share_folder.record_commands._nsf.create_record_v3') + def test_add_record_rejects_shared_in_path_shadow(self, mock_create): + from keepercommander.commands.nested_share_folder import NestedShareRecordAddCommand + + attacker_uid, attacker = _make_folder(name='Clients/Acme') + attacker['owner_username'] = 'attacker@example.com' + params = _make_params( + user='victim@example.com', + nested_share_folders={attacker_uid: attacker}, + record_type_cache={}, + ) + + with self.assertRaises(CommandError): + NestedShareRecordAddCommand().execute( + params, title='Blocked', record_type='general', + folder_uid='Clients/Acme', fields=[], force=True) + mock_create.assert_not_called() + @patch('keepercommander.commands.nested_share_folder.record_commands._nsf.create_record_v3') def test_add_record_rejects_weak_password_without_force(self, mock_create): from keepercommander.commands.nested_share_folder import NestedShareRecordAddCommand @@ -2519,6 +2841,16 @@ def test_get_record_details(self, mock_details): def test_get_record_access(self, mock_accesses): pass + def test_get_does_not_treat_legacy_folder_as_nsf(self): + from keepercommander.commands.nested_share_folder.display_commands import NestedShareGetCommand + + legacy_uid = utils.generate_uid() + params = _make_params(folder_cache={legacy_uid: {'name': 'Legacy'}}) + cmd = NestedShareGetCommand() + + with self.assertRaises(CommandError): + cmd.execute(params, uid=legacy_uid) + @patch('keepercommander.nested_share_folder.get_folder_access_v3') def test_folder_json_uses_team_name_in_team_permissions(self, mock_access): """nsf-get JSON should show team names, not team UIDs, in team_permissions.""" diff --git a/unit-tests/test_subfolder.py b/unit-tests/test_subfolder.py index a54f7fa58..af295422d 100644 --- a/unit-tests/test_subfolder.py +++ b/unit-tests/test_subfolder.py @@ -82,3 +82,30 @@ def test_subfolder_try_resolve_path(input_, expected_folder, expected_final): actual_folder, actual_final = subfolder.try_resolve_path(global_params, input_) assert actual_folder is expected_folder assert actual_final == expected_final + + +def test_subfolder_try_resolve_path_rejects_ambiguous_folder_name(): + first_uid = 'b' * 22 + second_uid = 'c' * 22 + first = BFN(type='user_folder', uid=first_uid, parent_uid=None, + name='Work', subfolders=[]) + second = BFN(type='shared_folder', uid=second_uid, parent_uid=None, + name='Work', subfolders=[]) + root = BFN(type='/', uid=None, parent_uid=None, name='root', + subfolders=[first_uid, second_uid]) + params = Mock() + params.folder_cache = { + first_uid: first, + second_uid: second, + } + params.root_folder = root + params.current_folder = '' + + assert subfolder.try_resolve_path(params, 'Work') is None + with pytest.raises(subfolder.FolderPathAmbiguousError): + subfolder.try_resolve_path(params, 'Work', raise_on_ambiguous=True) + + folders, pattern = subfolder.try_resolve_path( + params, 'Work', find_all_matches=True) + assert pattern == '' + assert {folder.uid for folder in folders} == {first_uid, second_uid}