KC-1504, KC-1519: Fix Nested Share Folder path resolution ownership bypass, and Prevent duplicate folder names from routing records into attacker-owned shared folders

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:

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')
 

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', {}):

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 ────────────────────────────────────────────────
 

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)
+        move_nested_share_item(params, source, destination)

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

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

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')