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..db5eea53a 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, contained_folders, lookup_path, path_split, try_resolve_path, \ find_folders, get_contained_record_uids, get_contained_folder_uids @@ -259,7 +260,11 @@ def execute(self, params, **kwargs): folder = params.folder_cache[params.current_folder] if params.current_folder in params.folder_cache else params.root_folder pattern = kwargs['pattern'] if 'pattern' in kwargs else None if pattern: - rs = try_resolve_path(params, kwargs['pattern']) + try: + rs = try_resolve_path( + params, kwargs['pattern'], raise_on_ambiguous=True) + except FolderPathAmbiguousError as e: + raise CommandError('ls', str(e)) if rs is not None: folder, pattern = rs @@ -408,13 +413,19 @@ 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'.format(folder_name)) class FolderTreeCommand(Command): @@ -438,7 +449,12 @@ def execute(self, params, **kwargs): params, folder, verbose=verbose, show_records=records, shares=shares, nsf_shares=nsf_shares, hide_shares_key=hide_key, title=title, fmt=fmt)) else: - folders, pattern = try_resolve_path(params, folder_name, find_all_matches=True) + try: + folders, pattern = try_resolve_path( + params, folder_name, find_all_matches=True, + raise_on_ambiguous=True) + except FolderPathAmbiguousError as e: + raise CommandError('tree', str(e)) if not pattern: for idx, folder in enumerate(folders): trees.append(formatted_tree( @@ -483,7 +499,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,12 +575,10 @@ def execute(self, params, **kwargs): name = kwargs['folder'] if 'folder' in kwargs else None if name: - rs = try_resolve_path(params, name) - if rs is not None: - base_folder, name = rs - if len(name) == 0: - logging.warning('mkdir: Folder "%s" already exists', kwargs['folder']) - return + base_folder, name, already_exists = self._resolve_parent_for_creation(params, name) + if already_exists: + logging.warning('mkdir: Folder "%s" already exists', kwargs['folder']) + return shared_folder = kwargs['shared_folder'] if 'shared_folder' in kwargs else None user_folder = kwargs['user_folder'] if 'user_folder' in kwargs else None @@ -666,6 +684,96 @@ def execute(self, params, **kwargs): logging.info(json.dumps(response_data)) return folder_uid + @staticmethod + def _resolve_parent_for_creation(params, folder_path): + """Resolve only the parent of a new folder path. + + A duplicate name is valid at the root level, so the final component + must not be resolved as an existing folder in that case. Every parent + component still has to resolve to exactly one folder. + """ + base_folder = ( + params.folder_cache[params.current_folder] + if params.current_folder in params.folder_cache + else params.root_folder + ) + base_folder, components = path_split(params, base_folder, folder_path) + components = [component for component in components if component != ''] + if not components: + raise CommandError('mkdir', 'Folder name is required') + + leaf = components[-1] + parent_components = components[:-1] + if parent_components: + try: + remainder, parents = lookup_path( + params, base_folder, parent_components, + raise_on_ambiguous=True) + except FolderPathAmbiguousError as e: + raise CommandError('mkdir', str(e)) + if remainder != len(parent_components) or not parents: + parent_path = '/'.join( + component.replace('/', '//') for component in parent_components) + raise CommandError('mkdir', f'Parent folder "{parent_path}" not found') + base_folder = parents[0] + + existing_children = contained_folders(params, [base_folder], leaf) + if len(existing_children) > 1: + raise CommandError( + 'mkdir', + f'Multiple folders match path component "{leaf}". ' + 'Use a folder UID instead.') + if len(existing_children) == 1: + return base_folder, leaf, True + else: + existing_children = contained_folders(params, [base_folder], leaf) + if leaf in params.folder_cache: + return base_folder, leaf, True + if (len(existing_children) > 1 + and base_folder.type != BaseFolderNode.RootFolderType): + raise CommandError( + 'mkdir', + f'Multiple folders match path component "{leaf}". ' + 'Use a folder UID instead.') + if existing_children: + # Root-level duplicate names are allowed. A folder-relative + # leaf, however, must remain unique under its parent. + if base_folder.type != BaseFolderNode.RootFolderType: + return base_folder, leaf, True + if any(FolderMakeCommand._is_owned_root_folder(params, f) + for f in existing_children): + return base_folder, leaf, True + + return base_folder, leaf, False + + @staticmethod + def _is_owned_root_folder(params, folder): + """Return whether a root folder belongs to the current account.""" + if folder.type == BaseFolderNode.UserFolderType: + return True + if folder.type != BaseFolderNode.SharedFolderType: + return False + + shared_folder = getattr(params, 'shared_folder_cache', {}).get(folder.uid, {}) + owner_username = shared_folder.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_uid = shared_folder.get('owner_account_uid') + account_uid_bytes = getattr(params, 'account_uid_bytes', None) + if owner_uid and account_uid_bytes: + return owner_uid == utils.base64_url_encode(account_uid_bytes) + + return any( + user.get('owner') is True + and (user.get('username') or '').casefold() + == (current_username or '').casefold() + for user in shared_folder.get('users', []) + ) + def get_folder_path(params, uid): path = '' @@ -700,7 +808,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 +944,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 +982,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..c3172f194 100644 --- a/keepercommander/commands/ksm.py +++ b/keepercommander/commands/ksm.py @@ -34,7 +34,8 @@ from .nested_share_folder.helpers import ( is_nested_share_folder, is_nested_share_record, load_record_metadata, resolve_folder_uid) from ..nested_share_folder.removal_api import ( - resolve_nested_share_folder_uid, resolve_nested_share_record_uid) + resolve_nested_share_folder_uid, resolve_nested_share_record_uid, + _is_current_account_record_owner) from .. import api, utils, crypto, vault from ..params import KeeperParams from ..display import bcolors @@ -1627,6 +1628,50 @@ def _normalize_nsf_app_record(params, record_uid): 'source': 'nested_share_folder', } + @staticmethod + def _app_title(data): + return data.get('title', '') if isinstance(data, dict) else '' + + @staticmethod + def _app_record_candidates(params, title): + """Return only app records matching *title*, before ownership ranking.""" + candidates = [] + seen = set() + title_cf = title.casefold() + + for uid, rec in getattr(params, 'record_cache', {}).items(): + app = KSMCommand._app_record_from_cache_entry(rec) + if not app or uid in seen: + continue + try: + raw_data = rec.get('data_unencrypted', b'') + if isinstance(raw_data, bytes): + raw_data = raw_data.decode('utf-8') + data = json.loads(raw_data) + except (AttributeError, TypeError, ValueError): + continue + if KSMCommand._app_title(data).casefold() == title_cf: + candidates.append((uid, app, rec, None)) + seen.add(uid) + + for uid in getattr(params, 'nested_share_records', {}): + if uid in seen: + continue + app = KSMCommand._normalize_nsf_app_record(params, uid) + if not app: + continue + data = getattr(params, 'nested_share_record_data', {}).get(uid, {}).get( + 'data_json', {}) + if KSMCommand._app_title(data).casefold() == title_cf: + candidates.append(( + uid, app, + getattr(params, 'nested_share_record_data', {}).get(uid, {}), + getattr(params, 'record_cache', {}).get(uid, {}), + )) + seen.add(uid) + + return candidates + @staticmethod def get_app_record(params, app_name_or_uid): if not app_name_or_uid: @@ -1641,17 +1686,26 @@ 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 + app_candidates = KSMCommand._app_record_candidates(params, app_name_or_uid) + if app_candidates: + owned = [ + candidate for candidate in app_candidates + if _is_current_account_record_owner( + params, candidate[0], candidate[2], candidate[3]) + ] + if len(owned) == 1: + return owned[0][1] + if len(owned) > 1: + logging.warning( + "Multiple owned Secrets Manager applications match '%s'. " + "Use the application UID instead.", app_name_or_uid) + return None + if len(app_candidates) == 1: + return app_candidates[0][1] + logging.warning( + "Multiple Secrets Manager applications match '%s'. " + "Use the application UID instead.", app_name_or_uid) + return None resolved_uid = resolve_nested_share_record_uid(params, app_name_or_uid) if resolved_uid: diff --git a/keepercommander/commands/nested_share_folder/display_commands.py b/keepercommander/commands/nested_share_folder/display_commands.py index 54df4ef10..44b55b32d 100644 --- a/keepercommander/commands/nested_share_folder/display_commands.py +++ b/keepercommander/commands/nested_share_folder/display_commands.py @@ -131,48 +131,22 @@ def execute(self, params, **kwargs): show_inherited=show_inherited, show_denied=show_denied) return - raise CommandError('nsf-get', f'Cannot find any Nested Share Folder object with UID or title: {uid}') + raise CommandError( + 'nsf-get', + f'Cannot resolve "{uid}" by name. Use the object UID; run nsf-list ' + 'to find folder UIDs. Shared-in and ambiguous folder names require UID lookup.') # ── Resolution ──────────────────────────────────────────────────── @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..025da04b4 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, @@ -73,7 +74,17 @@ def execute(self, params, **kwargs): for idx, segment in enumerate(segments): is_leaf = (idx == last_idx) - existing_uid = self._resolve_segment(params, segment, parent_uid) + # Duplicate names are allowed at the NSF root. For child paths, + # every parent and existing child must still resolve uniquely. + allow_new_root_leaf = ( + is_leaf and not parent_uid + and not is_nested_share_folder(params, segment) + ) + existing_uid = ( + self._find_owned_root_child(params, segment) + if allow_new_root_leaf + else self._resolve_segment(params, segment, parent_uid) + ) if existing_uid: if is_leaf: logging.warning('nsf-mkdir: Folder "%s" already exists', segment) @@ -134,6 +145,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 @@ -156,23 +175,69 @@ def _find_existing_child(params, folder_name, parent_uid): *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 + candidate_uids = [] for fuid, fobj in nsf_folders.items(): - if fobj.get('name', '').lower() != name_lower: - continue raw_parent = fobj.get('parent_uid') or '' normalized = normalize_parent_uid(raw_parent) is_root_child = ( 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 parent_matches: + candidate_uids.append(fuid) + + exact_uids = [ + fuid for fuid in candidate_uids + if nsf_folders[fuid].get('name', '') == folder_name + ] + matching_uids = exact_uids or [ + fuid for fuid in candidate_uids + if nsf_folders[fuid].get('name', '').casefold() + == folder_name.casefold() + ] + + owned_matches = [] + for fuid in matching_uids: + fobj = nsf_folders[fuid] + if _is_current_account_folder_owner(params, fobj): + owned_matches.append(fuid) + + # A unique shared-in folder is a valid parent for team workflows. + # When names collide, prefer the current account's folder and fail + # closed if ownership cannot disambiguate the candidates. + matches = owned_matches if owned_matches else matching_uids + 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 + + @staticmethod + def _find_owned_root_child(params, folder_name): + """Find an owned root folder without treating shared-in names as existing.""" + nsf_folders = getattr(params, 'nested_share_folders', {}) + exact = [ + uid for uid, fobj in nsf_folders.items() + if not (fobj.get('parent_uid') or '') + and fobj.get('name', '') == folder_name + and _is_current_account_folder_owner(params, fobj) + ] + if exact: + return exact[0] + + folded = folder_name.casefold() + for uid, fobj in nsf_folders.items(): + parent_uid = fobj.get('parent_uid') or '' + is_root = (not parent_uid or normalize_parent_uid(parent_uid) == 'root' + or parent_uid not in nsf_folders) + if (is_root and fobj.get('name', '').casefold() == folded + and _is_current_account_folder_owner(params, fobj)): + return uid return None @@ -677,4 +742,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..b09276136 100644 --- a/keepercommander/nested_share_folder/folder_api.py +++ b/keepercommander/nested_share_folder/folder_api.py @@ -150,30 +150,176 @@ def folder_access_update_v3(params, folder_access_adds=None, # Resolution # ══════════════════════════════════════════════════════════════════════════ +_AMBIGUOUS_RESOLUTION = object() + + 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 always unambiguous and remain the preferred way to select a + # shared-in folder when names collide. + 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()] - if len(matching) == 1: - return matching[0] - if len(matching) > 1: - logging.warning("Multiple folders match '%s'. Use UID instead.", folder_identifier) + # 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_nsf_path(params, folder_identifier) + if nsf_uid is _AMBIGUOUS_RESOLUTION: + return None + 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: + return folder.uid 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 + matching = _matching_nsf_folder_uids(nsf_folders, folder_identifier) + selected = _select_nsf_folder_uid(params, nsf_folders, matching) + if selected is _AMBIGUOUS_RESOLUTION: + logging.warning("Multiple folders match '%s'. Use UID instead.", folder_identifier) + return None + if selected: + return selected + + 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: + 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. A + unique unknown/shared-in folder can still be selected by name; ownership + only decides which candidate wins when names collide. + """ + 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 _matching_nsf_folder_uids(nsf_folders, folder_name): + """Return exact-case matches, or case-insensitive matches when exact is absent.""" + exact = [ + uid for uid, folder_obj in nsf_folders.items() + if folder_obj.get('name', '') == folder_name + ] + if exact: + return exact + folded = folder_name.casefold() + return [ + uid for uid, folder_obj in nsf_folders.items() + if folder_obj.get('name', '').casefold() == folded + ] + + +def _select_nsf_folder_uid(params, nsf_folders, matching): + """Prefer one owned match; otherwise allow one shared-in match.""" + if not matching: + return None + owned = [ + uid for uid in matching + if _is_current_account_folder_owner(params, nsf_folders[uid]) + ] + if len(owned) == 1: + return owned[0] + if len(owned) > 1: + return _AMBIGUOUS_RESOLUTION + return matching[0] if len(matching) == 1 else _AMBIGUOUS_RESOLUTION + + +def _resolve_nsf_path(params, folder_identifier): + """Resolve an NSF path with exact-case and ownership-aware matching. + + 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: + candidates = [] + for uid, folder_obj in nsf_folders.items(): + parent_uid = folder_obj.get('parent_uid') or '' + if current: + if parent_uid == current: + candidates.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: + candidates.append(uid) + + candidate_folders = {uid: nsf_folders[uid] for uid in candidates} + matching = _matching_nsf_folder_uids(candidate_folders, component) + selected = _select_nsf_folder_uid(params, nsf_folders, matching) + if selected is _AMBIGUOUS_RESOLUTION: + return _AMBIGUOUS_RESOLUTION + if not selected: + return None + current = selected + + return current + + +def _is_safe_path_folder(params, folder): + """Retained for callers that need a path-safety predicate.""" + 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..c5e4c88a0 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,85 @@ 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 + + # The cache entries passed by callers are normally dictionaries, but some + # sync paths expose a collection of metadata entries. Normalize both + # forms before looking up ownership so a malformed/legacy cache cannot + # turn title resolution into an AttributeError. + metadata_cache = getattr(params, 'meta_data_cache', {}) + metadata_entry = (metadata_cache.get(record_uid, {}) + if isinstance(metadata_cache, dict) else {}) + + objects = [] + for value in (nsf_data, record_obj, metadata_entry): + if isinstance(value, dict): + objects.append(value) + elif isinstance(value, (list, tuple, set)): + objects.extend(item for item in value if isinstance(item, dict)) + + for obj in objects: + 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..b50ca4bf0 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 = '' @@ -104,12 +113,29 @@ def contained_folders(params, folders, component): elif component in get_folder_ids(): result = [get_folder_by_id(component)] else: - result = [get_folder_by_id(uid) for f in folders for uid in f.subfolders if get_folder_by_id(uid).name == component] + children = [ + get_folder_by_id(uid) + for f in folders + for uid in f.subfolders + if (get_folder_by_id(uid) is not None + and (get_folder_by_id(uid).name or '') == component) + ] + if not children: + component_casefold = component.casefold() + children = [ + get_folder_by_id(uid) + for f in folders + for uid in f.subfolders + if (get_folder_by_id(uid) is not None + and (get_folder_by_id(uid).name or '').casefold() == component_casefold) + ] + result = children 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 +149,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 +174,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 +196,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..5f77f97bb 100644 --- a/unit-tests/test_command_folder.py +++ b/unit-tests/test_command_folder.py @@ -59,6 +59,24 @@ def test_tree(self): with self.assertRaises(CommandError): cmd.execute(params, folder='Invalid') + def test_tree_surfaces_ambiguous_folder_path(self): + from keepercommander.subfolder import UserFolderNode + + params = get_synced_params() + first = UserFolderNode() + first.uid = 'e' * 22 + first.name = 'Ambiguous Tree' + second = UserFolderNode() + second.uid = 'f' * 22 + second.name = 'Ambiguous Tree' + params.folder_cache[first.uid] = first + params.folder_cache[second.uid] = second + params.root_folder.subfolders.extend([first.uid, second.uid]) + + with self.assertRaises(CommandError) as ctx: + folder.FolderTreeCommand().execute(params, folder='Ambiguous Tree') + self.assertIn('Multiple folders match', str(ctx.exception)) + def test_make_folder(self): params = get_synced_params() cmd = folder.FolderMakeCommand() @@ -108,6 +126,79 @@ def is_shared_folder_folder(rq): cmd.execute(params, folder='New SubFolder') self.assertTrue(KeeperApiHelper.is_expect_empty()) + def test_make_folder_allows_duplicate_root_name(self): + from keepercommander.subfolder import SharedFolderNode + + params = get_synced_params() + existing = SharedFolderNode() + existing.uid = 'c' * 22 + existing.name = 'Duplicate Root' + existing.parent_uid = None + existing.subfolders = [] + second = SharedFolderNode() + second.uid = 'h' * 22 + second.name = 'Duplicate Root' + second.parent_uid = None + second.subfolders = [] + params.folder_cache[existing.uid] = existing + params.folder_cache[second.uid] = second + params.root_folder.subfolders.extend([existing.uid, second.uid]) + + def is_user_folder(rq): + self.assertEqual(rq['command'], 'folder_add') + self.assertEqual(rq['folder_type'], 'user_folder') + + KeeperApiHelper.communicate_expect([is_user_folder]) + folder.FolderMakeCommand().execute( + params, user_folder=True, folder='Duplicate Root') + self.assertTrue(KeeperApiHelper.is_expect_empty()) + + def test_make_folder_rejects_duplicate_parent_name(self): + from keepercommander.subfolder import UserFolderNode + + params = get_synced_params() + first = UserFolderNode() + first.uid = 'd' * 22 + first.name = 'Duplicate Parent' + first.parent_uid = None + first.subfolders = [] + second = UserFolderNode() + second.uid = 'e' * 22 + second.name = 'Duplicate Parent' + 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]) + + with self.assertRaises(CommandError): + folder.FolderMakeCommand().execute( + params, user_folder=True, folder='Duplicate Parent/Child') + + def test_make_folder_keeps_child_name_unique_under_current_folder(self): + from keepercommander.subfolder import UserFolderNode + + params = get_synced_params() + parent = UserFolderNode() + parent.uid = 'f' * 22 + parent.name = 'Parent' + parent.parent_uid = None + parent.subfolders = [] + child = UserFolderNode() + child.uid = 'g' * 22 + child.name = 'Child' + child.parent_uid = parent.uid + child.subfolders = [] + parent.subfolders.append(child.uid) + params.folder_cache[parent.uid] = parent + params.folder_cache[child.uid] = child + params.root_folder.subfolders.append(parent.uid) + params.current_folder = parent.uid + + with self.assertLogs(level=logging.WARNING): + folder.FolderMakeCommand().execute( + params, user_folder=True, folder='Child') + def test_remove_non_existing_folder(self): params = get_synced_params() cmd = folder.FolderRemoveCommand() @@ -115,6 +206,50 @@ 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_list_surfaces_ambiguous_folder_path(self): + from keepercommander.subfolder import UserFolderNode + + params = get_synced_params() + first = UserFolderNode() + first.uid = 'c' * 22 + first.name = 'Ambiguous List' + second = UserFolderNode() + second.uid = 'd' * 22 + second.name = 'Ambiguous List' + params.folder_cache[first.uid] = first + params.folder_cache[second.uid] = second + params.root_folder.subfolders.extend([first.uid, second.uid]) + + with self.assertRaises(CommandError) as ctx: + folder.FolderListCommand().execute(params, pattern='Ambiguous List') + 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..14cdd22a5 100644 --- a/unit-tests/test_command_ksm.py +++ b/unit-tests/test_command_ksm.py @@ -160,6 +160,95 @@ 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_app_record_ignores_non_app_title_collisions(self): + params = self._make_params() + app_uid = 'appUid00000000000001' + login_uid = 'loginUid0000000000001' + params.record_cache = { + app_uid: { + 'record_uid': app_uid, + 'version': 5, + 'data_unencrypted': b'{"type":"app","title":"Prod App"}', + 'record_key_unencrypted': b'app_key', + }, + login_uid: { + 'record_uid': login_uid, + 'version': 3, + 'data_unencrypted': b'{"type":"login","title":"Prod App"}', + }, + } + + result = KSMCommand.get_app_record(params, 'Prod App') + self.assertIsNotNone(result) + self.assertEqual(result['record_uid'], app_uid) + + def test_get_app_record_allows_unique_shared_app_title(self): + params = self._make_params() + app_uid = 'appUid00000000000001' + login_uid = 'loginUid0000000000001' + params.user = 'victim@example.com' + params.record_cache = { + app_uid: { + 'record_uid': app_uid, + 'version': 5, + 'data_unencrypted': b'{"type":"app","title":"Prod App"}', + 'record_key_unencrypted': b'app_key', + }, + login_uid: { + 'record_uid': login_uid, + 'version': 3, + 'data_unencrypted': b'{"type":"login","title":"Prod App"}', + }, + } + params.meta_data_cache = { + app_uid: {'owner_username': 'attacker@example.com'}, + login_uid: {'owner_username': 'victim@example.com'}, + } + + result = KSMCommand.get_app_record(params, 'Prod App') + self.assertIsNotNone(result) + self.assertEqual(result['record_uid'], app_uid) + 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..294436201 100644 --- a/unit-tests/test_nested_share_folder.py +++ b/unit-tests/test_nested_share_folder.py @@ -105,6 +105,344 @@ 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_allows_unique_shared_in_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.assertEqual( + resolve_folder_identifier(params, 'Clients/Acme'), + child_uid) + + 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.assertEqual(resolve_folder_identifier(params, 'Shared'), shared_uid) + self.assertEqual(resolve_folder_identifier(params, shared_uid), shared_uid) + + def test_flat_folder_name_allows_unique_shared_in_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.assertEqual(resolve_folder_identifier(params, 'Shared'), shared_uid) + + 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_resolves_unique_shared_in_parent(self): + from keepercommander.commands.nested_share_folder.folder_commands import ( + NestedShareFolderMkdirCommand, + ) + + shared_uid, shared = _make_folder(name='Team Work') + shared['owner_username'] = 'attacker@example.com' + params = _make_params( + user='victim@example.com', + nested_share_folders={shared_uid: shared}, + ) + + self.assertEqual( + NestedShareFolderMkdirCommand._find_existing_child( + params, 'Team Work', None), + shared_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_resolves_unique_shared_in_folder(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}, + ) + + self.assertEqual( + NestedShareRecordPermissionCommand._resolve_folder( + params.nested_share_folders, 'Shared', params), + (shared_uid, 'Shared')) + + 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_record_owner_check_handles_collection_metadata(self): + from keepercommander.nested_share_folder.removal_api import ( + _is_current_account_record_owner) + + params = _make_params(user='victim@example.com') + self.assertTrue(_is_current_account_record_owner( + params, 'record-uid', + [{'owner_username': 'victim@example.com'}], + {'data_unencrypted': b''})) + self.assertFalse(_is_current_account_record_owner( + params, 'record-uid', + ['invalid metadata'], + {'data_unencrypted': b''})) + def test_parse_expiration_none(self): from keepercommander.commands.nested_share_folder.helpers import parse_expiration self.assertIsNone(parse_expiration(None, None, 'test')) @@ -307,6 +645,62 @@ def test_mkdir(self, mock_create): cmd.execute(_make_params(), folder='NewFolder') mock_create.assert_called_once() + @patch('keepercommander.commands.nested_share_folder.folder_commands._nsf.create_folder_v3') + def test_mkdir_allows_duplicate_root_name(self, mock_create): + from keepercommander.commands.nested_share_folder import NestedShareFolderMkdirCommand + + existing_uid, existing = _make_folder(name='Duplicate Root') + second_uid, second = _make_folder(name='Duplicate Root') + mock_create.return_value = { + 'folder_uid': utils.generate_uid(), 'status': 'SUCCESS', + 'message': '', 'success': True, + } + params = _make_params( + nested_share_folders={ + existing_uid: existing, + second_uid: second, + }) + + NestedShareFolderMkdirCommand().execute(params, folder='Duplicate Root') + + mock_create.assert_called_once() + self.assertIsNone(mock_create.call_args.kwargs['parent_uid']) + + @patch('keepercommander.commands.nested_share_folder.folder_commands._nsf.create_folder_v3') + def test_mkdir_is_idempotent_for_owned_root_name(self, mock_create): + from keepercommander.commands.nested_share_folder import NestedShareFolderMkdirCommand + + existing_uid, existing = _make_folder(name='Owned Root') + existing['owner_username'] = 'victim@example.com' + params = _make_params( + user='victim@example.com', + nested_share_folders={existing_uid: existing}) + + result = NestedShareFolderMkdirCommand().execute( + params, folder='Owned Root') + + self.assertEqual(result, existing_uid) + mock_create.assert_not_called() + + @patch('keepercommander.commands.nested_share_folder.folder_commands._nsf.create_folder_v3') + def test_mkdir_rejects_duplicate_parent_name(self, mock_create): + from keepercommander.commands.nested_share_folder import NestedShareFolderMkdirCommand + + first_uid, first = _make_folder(name='Duplicate Parent') + second_uid, second = _make_folder(name='Duplicate Parent') + params = _make_params( + user='victim@example.com', + nested_share_folders={ + first_uid: dict(first, owner_username='victim@example.com'), + second_uid: dict(second, owner_username='victim@example.com'), + }, + ) + + with self.assertRaises(CommandError): + NestedShareFolderMkdirCommand().execute( + params, folder='Duplicate Parent/Child') + mock_create.assert_not_called() + @patch('keepercommander.commands.nested_share_folder.folder_commands._nsf.create_folder_v3') def test_mkdir_resolves_parent_uid_in_path(self, mock_create): """Parent path segments that are NSF UIDs resolve to the folder, not a name.""" @@ -491,6 +885,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 +2931,27 @@ 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) + + def test_get_unresolved_name_explains_uid_lookup(self): + from keepercommander.commands.nested_share_folder.display_commands import NestedShareGetCommand + + params = _make_params(user='victim@example.com') + cmd = NestedShareGetCommand() + + with self.assertRaises(CommandError) as ctx: + cmd.execute(params, uid='Clients//Acme1') + self.assertIn('Shared-in and ambiguous folder names require UID lookup', + str(ctx.exception)) + @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..af5881247 100644 --- a/unit-tests/test_subfolder.py +++ b/unit-tests/test_subfolder.py @@ -82,3 +82,58 @@ 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} + + +def test_subfolder_try_resolve_path_rejects_case_variant_folder_shadowing(): + 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 = '' + + exact_folder, exact_tail = subfolder.try_resolve_path(params, 'Work') + assert exact_folder is first + assert exact_tail == '' + lower_folder, lower_tail = subfolder.try_resolve_path(params, 'work') + assert lower_folder is second + assert lower_tail == '' + assert subfolder.try_resolve_path(params, 'WORK') is None + with pytest.raises(subfolder.FolderPathAmbiguousError): + subfolder.try_resolve_path(params, 'WORK', raise_on_ambiguous=True)