KC-1361: Reject share commands that target a record or folder owner

keepercommander/commands/nested_share_folder/folder_commands.py

@@ -30,6 +30,7 @@
     check_folder_edit_permission, check_folder_share_permission, check_folder_delete_permission,
     classify_share_recipient,
     ensure_nested_share_folder, is_nested_share_folder,
+    is_nested_share_folder_owner_email, owner_share_target_message,
 )
 from .parsers import (
     nested_share_folder_mkdir_parser,
@@ -465,6 +466,11 @@ def _expand_existing(params, folder_uid, folder_arg):
                     elif accessor.get('access_type') == 'AT_USER':
                         username = accessor.get('username')
                         if username and username != params.user:
+                            if is_nested_share_folder_owner_email(params, folder_uid, username):
+                                logging.info(
+                                    "nsf-share-folder: skipping owner '%s' for "
+                                    "folder '%s'", username, folder_arg)
+                                continue
                             result.append(('user', username))
         except Exception as exc:
             logging.debug(
@@ -479,6 +485,11 @@ def _expand_existing(params, folder_uid, folder_arg):
                 if access_type == at_user:
                     username = a.get('username')
                     if username and username != params.user:
+                        if is_nested_share_folder_owner_email(params, folder_uid, username):
+                            logging.info(
+                                "nsf-share-folder: skipping owner '%s' for "
+                                "folder '%s'", username, folder_arg)
+                            continue
                         result.append(('user', username))
                 elif access_type == at_team:
                     team_uid = a.get('access_type_uid')
@@ -493,6 +504,10 @@ def _expand_existing(params, folder_uid, folder_arg):
     @classmethod
     def _apply(cls, params, action, folder_uid, recipient, role, expiration,
                 as_team=False, rotate_on_expiration=False):
+        if not as_team and is_nested_share_folder_owner_email(params, folder_uid, recipient):
+            raise CommandError(
+                'nsf-share-folder',
+                owner_share_target_message(recipient, entity='folder'))
         api_name, verb = cls._ACTIONS[action]
         api_func = getattr(_nsf, api_name)
         kw = dict(params=params, folder_uid=folder_uid, user_uid=recipient,

keepercommander/commands/nested_share_folder/helpers.py

@@ -174,6 +174,57 @@ def resolve_folder_uid(params, identifier):
     "only on Nested Share Folders."
 )
 
+_OWNER_SHARE_TARGET_MSG = (
+    "'{email}' is the owner of this {entity} and already has full access. "
+    "Share permissions cannot be granted, changed, or revoked for the owner."
+)
+
+
+def owner_share_target_message(email, entity='record'):
+    """User-facing message when a share command targets the owner."""
+    return _OWNER_SHARE_TARGET_MSG.format(email=email, entity=entity)
+
+
+def is_nested_share_folder_owner_email(params, folder_uid, email):
+    """True when *email* matches the NSF folder owner from sync-down."""
+    if not folder_uid or not email:
+        return False
+    fobj = getattr(params, 'nested_share_folders', {}).get(folder_uid) or {}
+    owner_username = fobj.get('owner_username') or ''
+    return bool(owner_username) and owner_username.casefold() == email.casefold()
+
+
+def raise_if_record_share_target_is_owner(params, record_uid, email, cmd_name, *,
+                                          is_ownership_transfer=False):
+    """Raise CommandError when *email* is the owner of *record_uid*.
+
+    When *is_ownership_transfer* is True (``-a owner``), raising means the
+    target already owns the record so the transfer would be a no-op. When False,
+    grant/revoke/update against the owner is rejected.
+
+    If owner lookup via ``get_record_accesses_v3`` fails, we log a warning and
+    return without raising. Blocking the share on a transient access-API error
+    would reject legitimate grants to non-owners; the server still rejects
+    invalid owner-targeted shares if we miss the check client-side.
+    """
+    from ...nested_share_folder.record_api import (
+        get_record_accesses_v3, find_record_owner_username)
+    try:
+        access_result = get_record_accesses_v3(params, [record_uid])
+    except Exception as exc:
+        logging.getLogger(__name__).warning(
+            "Could not resolve owner for record '%s'; proceeding without "
+            "client-side owner check: %s", record_uid, exc)
+        return
+    owner_username = find_record_owner_username(access_result, record_uid)
+    if not owner_username or owner_username.casefold() != email.casefold():
+        return
+    if is_ownership_transfer:
+        raise CommandError(
+            cmd_name,
+            f"'{email}' already owns this record. Ownership transfer is a no-op.")
+    raise CommandError(cmd_name, owner_share_target_message(email, entity='record'))
+
 
 def is_nested_share_record(params, record_uid):
     """Return True when *record_uid* is a Nested Share Folder (v3) record."""

keepercommander/commands/nested_share_folder/sharing_commands.py

@@ -34,6 +34,7 @@
     check_record_share_permission,
     collect_records_in_folder,
     ensure_nested_share_folder, ensure_nested_share_record,
+    raise_if_record_share_target_is_owner,
 )
 from .parsers import (
     nested_share_record_share_parser,
@@ -102,6 +103,9 @@ def execute(self, params, **kwargs):
         with command_error_handler('nsf-share-record'):
             for email in emails:
                 for record_uid in record_uids:
+                    raise_if_record_share_target_is_owner(
+                        params, record_uid, email, 'nsf-share-record',
+                        is_ownership_transfer=(action == 'owner'))
                     result, effective_action = self._dispatch(
                         params, action, record_uid, email, access_role_type, expiration,
                         rotate_on_expiration)

keepercommander/commands/register.py

@@ -302,6 +302,65 @@ def _folder_user_lookup(shared_folder, email):
     return None
 
 
+def _owner_share_target_message(email, entity='record'):
+    # type: (str, str) -> str
+    """User-facing message when a share command targets the owner."""
+    return (
+        f"'{email}' is the owner of this {entity} and already has full access. "
+        f"Share permissions cannot be granted, changed, or revoked for the owner."
+    )
+
+
+def _is_shared_folder_owner_email(shared_folder, email, params=None):
+    # type: (dict, str, Optional[KeeperParams]) -> bool
+    """True when *email* matches the classic shared folder owner from sync-down.
+
+    Resolution order:
+      1. ``owner_username`` from sync-down (primary; present for most folders)
+      2. ``users[].owner`` flag when sync includes it on the user row
+      3. ``owner_account_uid`` matched to the current session user — only used
+         when username/flag are absent. If sync omits ``owner_account_uid`` or
+         the session has no ``account_uid_bytes``, this path is skipped and we
+         return False rather than guessing.
+    """
+    if not shared_folder or not email:
+        return False
+    email_cf = email.casefold()
+    owner_username = shared_folder.get('owner_username') or ''
+    if owner_username and owner_username.casefold() == email_cf:
+        return True
+    # Some sync payloads also put an owner flag on the user row.
+    users = shared_folder.get('users', [])
+    if any(
+        user.get('owner') is True
+        and (user.get('username') or '').casefold() == email_cf
+        for user in users
+    ):
+        return True
+    # Last resort: only when the target is the logged-in user and sync exposes
+    # owner_account_uid. Skipped when either field is missing.
+    owner_uid = shared_folder.get('owner_account_uid')
+    account_uid_bytes = getattr(params, 'account_uid_bytes', None) if params else None
+    if (params and owner_uid and account_uid_bytes
+            and params.user and params.user.casefold() == email_cf):
+        if owner_uid == utils.base64_url_encode(account_uid_bytes):
+            return True
+    return False
+
+
+def _find_record_owner_username(existing_shares, params=None, record_uid=None):
+    # type: (dict, Optional[KeeperParams], Optional[str]) -> Optional[str]
+    """Return the record owner username from share rows or local owner cache."""
+    for username, perm in (existing_shares or {}).items():
+        if perm and perm.get('owner'):
+            return username
+    if params is not None and record_uid and params.user:
+        owner_info = getattr(params, 'record_owner_cache', {}).get(record_uid)
+        if owner_info and owner_info.owner:
+            return params.user
+    return None
+
+
 def format_share_expiration_ms(expiration_ms):
     # type: (int) -> str
     """Format a share expiration timestamp (milliseconds) for log output."""
@@ -545,7 +604,18 @@ def prep_rq(recs, users, curr_sf):
                 if all_users or all_records:
                     if all_users:
                         if 'users' in sh_fol:
-                            sf_users.update((x['username'] for x in sh_fol['users'] if x['username'] != params.user))
+                            for x in sh_fol['users']:
+                                username = x.get('username')
+                                if not username or username == params.user:
+                                    continue
+                                if _is_shared_folder_owner_email(
+                                        sh_fol, username, params=params):
+                                    logging.debug(
+                                        "share-folder: skipping owner '%s' in "
+                                        "bulk user update for folder '%s'",
+                                        username, sf_uid)
+                                    continue
+                                sf_users.add(username)
                         if 'teams' in sh_fol:
                             sf_teams.update((x['team_uid'] for x in sh_fol['teams']))
                     if all_records:
@@ -691,6 +761,13 @@ def apply_share_expiration(target):
                 current_user = _folder_user_lookup(curr_sf, email)
                 if current_user:
                     email = current_user['username']
+                if _is_shared_folder_owner_email(curr_sf, email, params=params):
+                    # Allow the existing owner self-removal / succession flow in
+                    # _confirm_folder_user_removals; reject grant/update of owner perms.
+                    if action == 'grant':
+                        raise CommandError(
+                            'share-folder',
+                            _owner_share_target_message(email, entity='shared folder'))
                 uo = folder_pb2.SharedFolderUpdateUser()
                 uo.username = email
                 apply_share_expiration(uo)
@@ -1104,7 +1181,24 @@ def apply_share_expiration(ro):   # type: (record_pb2.SharedRecord) -> None
                     pass
 
             record_path = api.resolve_record_share_path(params, record_uid)
+            owner_username = _find_record_owner_username(
+                existing_shares, params=params, record_uid=record_uid)
             for email in all_users:
+                # Precedence: owner rejection runs before existing-share add/update
+                # logic below. Granting/revoking the owner is never valid, even when
+                # they also appear in existing_shares; "already shared" handling only
+                # applies to non-owners.
+                if owner_username and email.casefold() == owner_username.casefold():
+                    # Ownership transfer (-a owner) is allowed only when the target
+                    # is not already the owner. Grant/revoke must never target owner.
+                    if action == 'owner':
+                        raise CommandError(
+                            'share-record',
+                            f"'{email}' already owns this record. Ownership transfer is a no-op.")
+                    raise CommandError(
+                        'share-record',
+                        _owner_share_target_message(email, entity='record'))
+
                 ro = record_pb2.SharedRecord()
                 ro.toUsername = email
                 ro.recordUid = utils.base64_url_decode(record_uid)

keepercommander/nested_share_folder/__init__.py

@@ -43,7 +43,8 @@
         'create_record_data_v3', 'record_add_v3', 'record_add_pam_configuration_v3', 'record_update_v3',
         'create_record_v3', 'update_record_v3', 'create_records_batch_v3',
         'get_record_details_v3', 'get_record_accesses_v3',
-        'find_direct_user_share_access', 'is_record_share_update_noop',
+        'find_direct_user_share_access', 'find_record_owner_username',
+        'is_record_share_update_noop',
         'share_record_v3', 'update_record_share_v3', 'unshare_record_v3',
         'share_record_to_application_v3', 'update_record_share_to_application_v3',
         'unshare_record_from_application_v3',

keepercommander/nested_share_folder/record_api.py

@@ -380,6 +380,17 @@ def find_direct_user_share_access(access_result, record_uid, email):
     return None
 
 
+def find_record_owner_username(access_result, record_uid):
+    """Return the owner username for *record_uid* from a get_record_accesses_v3 result."""
+    return next(
+        (
+            access.get('accessor_name') or ''
+            for access in access_result.get('record_accesses', [])
+            if access.get('record_uid') == record_uid and access.get('owner')
+        ),
+        '',
+    )
+
 _SHARE_EXPIRATION_NOOP_TOLERANCE_MS = 60_000
 
 

unit-tests/test_command_register.py

@@ -68,6 +68,65 @@ def shared(params, record_uids, is_share_admin):
         cmd.execute(params, email=['user2@keepersecurity.com'], action='revoke', record=record_uid)
         self.assertEqual(len(TestRegister.expected_commands), 0)
 
+    def _shared_with_owner_side_effect(self, owner_username=None, extra_users=None):
+        """Build a get_record_shares side_effect that marks *owner_username* as owner.
+
+        Defaults to the session user. *extra_users* is an optional list of
+        non-owner emails included in user_permissions.
+        """
+        extras = list(extra_users or [])
+
+        def shared_with_owner(params_, record_uids, is_share_admin):
+            owner = owner_username if owner_username is not None else params_.user
+            for uid in record_uids:
+                if uid not in params_.record_cache:
+                    continue
+                perms = [{'username': owner, 'owner': True}]
+                for email in extras:
+                    perms.append({
+                        'username': email, 'owner': False,
+                        'shareable': False, 'editable': False,
+                    })
+                params_.record_cache[uid]['shares'] = {'user_permissions': perms}
+
+        return shared_with_owner
+
+    def test_share_record_rejects_grant_to_owner(self):
+        """Granting share permissions to the record owner must fail client-side."""
+        params = get_synced_params()
+        record_uid = next(iter([x['record_uid'] for x in params.meta_data_cache.values() if x['can_share']]))
+        cmd = register.ShareRecordCommand()
+
+        self.record_share_mock = mock.patch('keepercommander.api.get_record_shares').start()
+        self.record_share_mock.side_effect = self._shared_with_owner_side_effect(
+            extra_users=['user2@keepersecurity.com'])
+
+        with self.assertRaises(CommandError) as ctx:
+            cmd.prep_request(params, dict(
+                email=[params.user],
+                action='grant',
+                record=record_uid,
+            ))
+        self.assertIn('is the owner', str(ctx.exception))
+        self.assertIn('already has full access', str(ctx.exception))
+
+    def test_share_record_rejects_owner_transfer_to_self(self):
+        """Ownership transfer to the current owner is a no-op and must be rejected."""
+        params = get_synced_params()
+        record_uid = next(iter([x['record_uid'] for x in params.meta_data_cache.values() if x['can_share']]))
+        cmd = register.ShareRecordCommand()
+
+        self.record_share_mock = mock.patch('keepercommander.api.get_record_shares').start()
+        self.record_share_mock.side_effect = self._shared_with_owner_side_effect()
+
+        with self.assertRaises(CommandError) as ctx:
+            cmd.prep_request(params, dict(
+                email=[params.user],
+                action='owner',
+                record=record_uid,
+            ))
+        self.assertIn('already owns', str(ctx.exception))
+
     @contextmanager
     def _make_record_rotation_eligible(self, params, target_uid):
         """Present the record as a pamUser with rotation configured (ROE-eligible)."""
@@ -308,6 +367,30 @@ def test_share_folder(self):
         cmd.execute(params, action='remove', user=['user2@keepersecurity.com'], folder=shared_folder_uid)
         self.assertEqual(len(TestRegister.expected_commands), 0)
 
+    def test_share_folder_rejects_grant_to_owner(self):
+        """Granting folder permissions to the folder owner must fail client-side."""
+        params = get_synced_params()
+        shared_folder_uid = next(iter(params.shared_folder_cache.keys()))
+        owner = params.user
+        curr_sf = dict(params.shared_folder_cache[shared_folder_uid])
+        curr_sf['owner_username'] = owner
+        curr_sf['users'] = [
+            {'username': owner, 'manage_records': True, 'manage_users': True},
+            {'username': 'user2@keepersecurity.com', 'manage_records': True, 'manage_users': True},
+        ]
+
+        with self.assertRaises(CommandError) as ctx:
+            register.ShareFolderCommand.prepare_request(
+                params,
+                kwargs={'action': 'grant', 'manage_records': 'on', 'manage_users': 'off'},
+                curr_sf=curr_sf,
+                users=[owner],
+                teams=[],
+                rec_uids=[],
+            )
+        self.assertIn('is the owner', str(ctx.exception))
+        self.assertIn('shared folder', str(ctx.exception))
+
     def test_share_folder_prepare_request_sets_rotate_on_expiration(self):
         """Folder-wide expiration/ROE applies to user/team protos, not record protos."""
         params = get_synced_params()