KC-1504, KC-1519, KC-1526: Fix NSF ownership bypass and attacker-shared folder name collisions in mkdir, mv, ln, and record-add --folder

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

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: