From 299dd72902f86209572b5bdb494727f5b3eee306 Mon Sep 17 00:00:00 2001
From: sshrushanth-ks <sshrushanth@keepersecurity.com>
Date: Wed, 15 Jul 2026 22:19:56 +0530
Subject: [PATCH 1/3] Reject share commands that target a record or folder
 owner.

Fail early with a clear message for share-record, share-folder,
nsf-share-record, and nsf-share-folder so owners are not granted,
updated, or revoked as if they were share recipients.
---
 .../nested_share_folder/folder_commands.py    |  9 +++
 .../commands/nested_share_folder/helpers.py   | 43 ++++++++++
 .../nested_share_folder/sharing_commands.py   |  4 +
 keepercommander/commands/register.py          | 71 ++++++++++++++++-
 .../nested_share_folder/__init__.py           |  3 +-
 .../nested_share_folder/record_api.py         | 10 +++
 unit-tests/test_command_register.py           | 79 +++++++++++++++++++
 unit-tests/test_nested_share_folder.py        | 40 +++++++++-
 8 files changed, 255 insertions(+), 4 deletions(-)

diff --git a/keepercommander/commands/nested_share_folder/folder_commands.py b/keepercommander/commands/nested_share_folder/folder_commands.py
index eb8be8341..09259fcee 100644
--- a/keepercommander/commands/nested_share_folder/folder_commands.py
+++ b/keepercommander/commands/nested_share_folder/folder_commands.py
@@ -28,6 +28,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,
@@ -437,6 +438,8 @@ 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):
+                                continue
                             result.append(('user', username))
         except Exception as exc:
             logging.debug(
@@ -451,6 +454,8 @@ 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):
+                            continue
                         result.append(('user', username))
                 elif access_type == at_team:
                     team_uid = a.get('access_type_uid')
@@ -465,6 +470,10 @@ def _expand_existing(params, folder_uid, folder_arg):
     @classmethod
     def _apply(cls, params, action, folder_uid, recipient, role, expiration,
                 as_team=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,
diff --git a/keepercommander/commands/nested_share_folder/helpers.py b/keepercommander/commands/nested_share_folder/helpers.py
index b4352a1dd..8e02fa99c 100644
--- a/keepercommander/commands/nested_share_folder/helpers.py
+++ b/keepercommander/commands/nested_share_folder/helpers.py
@@ -164,6 +164,49 @@ 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, *, allow_owner_transfer=False):
+    """Raise CommandError when *email* is the owner of *record_uid*.
+
+    Ownership transfer (``-a owner``) is allowed unless the target already owns
+    the record (``allow_owner_transfer=True`` then raises a no-op-style error).
+    """
+    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__).debug(
+            "Could not resolve owner for record '%s': %s", record_uid, exc)
+        return
+    owner = find_record_owner_username(access_result, record_uid)
+    if not owner or owner.casefold() != email.casefold():
+        return
+    if allow_owner_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."""
diff --git a/keepercommander/commands/nested_share_folder/sharing_commands.py b/keepercommander/commands/nested_share_folder/sharing_commands.py
index d4029fac5..f5e1fcc39 100644
--- a/keepercommander/commands/nested_share_folder/sharing_commands.py
+++ b/keepercommander/commands/nested_share_folder/sharing_commands.py
@@ -32,6 +32,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,
@@ -86,6 +87,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',
+                        allow_owner_transfer=(action == 'owner'))
                     result, effective_action = self._dispatch(
                         params, action, record_uid, email, access_role_type, expiration)
                     self._log_results(result, effective_action, email)
diff --git a/keepercommander/commands/register.py b/keepercommander/commands/register.py
index b8de28b37..0b4c0a6bd 100644
--- a/keepercommander/commands/register.py
+++ b/keepercommander/commands/register.py
@@ -302,6 +302,52 @@ 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."""
+    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.
+    for user in shared_folder.get('users', []):
+        username = (user.get('username') or '')
+        if username.casefold() != email_cf:
+            continue
+        if user.get('owner') is True:
+            return True
+    # Fall back: owner_account_uid + current session user.
+    owner_uid = shared_folder.get('owner_account_uid')
+    if params and owner_uid and params.user and params.user.casefold() == email_cf:
+        if owner_uid == utils.base64_url_encode(params.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 +591,10 @@ 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))
+                            sf_users.update(
+                                (x['username'] for x in sh_fol['users']
+                                 if x['username'] != params.user
+                                 and not _is_shared_folder_owner_email(sh_fol, x['username'], params=params)))
                         if 'teams' in sh_fol:
                             sf_teams.update((x['team_uid'] for x in sh_fol['teams']))
                     if all_records:
@@ -691,6 +740,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 +1160,20 @@ 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:
+                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)
diff --git a/keepercommander/nested_share_folder/__init__.py b/keepercommander/nested_share_folder/__init__.py
index 8436b6ba1..fc702293a 100644
--- a/keepercommander/nested_share_folder/__init__.py
+++ b/keepercommander/nested_share_folder/__init__.py
@@ -40,7 +40,8 @@
         'create_record_data_v3', 'record_add_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',
         'batch_update_record_shares_v3', 'batch_create_record_shares_v3',
         'batch_unshare_records_v3',
diff --git a/keepercommander/nested_share_folder/record_api.py b/keepercommander/nested_share_folder/record_api.py
index f0bcc9e69..40c4b26e5 100644
--- a/keepercommander/nested_share_folder/record_api.py
+++ b/keepercommander/nested_share_folder/record_api.py
@@ -360,6 +360,16 @@ 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."""
+    for access in access_result.get('record_accesses', []):
+        if access.get('record_uid') != record_uid:
+            continue
+        if access.get('owner'):
+            return access.get('accessor_name') or ''
+    return ''
+
+
 _SHARE_EXPIRATION_NOOP_TOLERANCE_MS = 60_000
 
 
diff --git a/unit-tests/test_command_register.py b/unit-tests/test_command_register.py
index 2f35ea315..8e734d72a 100644
--- a/unit-tests/test_command_register.py
+++ b/unit-tests/test_command_register.py
@@ -68,6 +68,61 @@ 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 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()
+
+        def shared_with_owner(params_, record_uids, is_share_admin):
+            for uid in record_uids:
+                if uid in params_.record_cache:
+                    params_.record_cache[uid]['shares'] = {
+                        'user_permissions': [
+                            {'username': params_.user, 'owner': True},
+                            {'username': 'user2@keepersecurity.com', 'owner': False,
+                             'shareable': False, 'editable': False},
+                        ]
+                    }
+
+        self.record_share_mock = mock.patch('keepercommander.api.get_record_shares').start()
+        self.record_share_mock.side_effect = shared_with_owner
+
+        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()
+
+        def shared_with_owner(params_, record_uids, is_share_admin):
+            for uid in record_uids:
+                if uid in params_.record_cache:
+                    params_.record_cache[uid]['shares'] = {
+                        'user_permissions': [
+                            {'username': params_.user, 'owner': True},
+                        ]
+                    }
+
+        self.record_share_mock = mock.patch('keepercommander.api.get_record_shares').start()
+        self.record_share_mock.side_effect = shared_with_owner
+
+        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 +363,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()
diff --git a/unit-tests/test_nested_share_folder.py b/unit-tests/test_nested_share_folder.py
index f946d8b07..3aeb33f03 100644
--- a/unit-tests/test_nested_share_folder.py
+++ b/unit-tests/test_nested_share_folder.py
@@ -833,7 +833,7 @@ def test_share_record(self, mock_share):
         cmd = NestedShareRecordShareCommand()
         with mock.patch('builtins.print'):
             cmd.execute(_make_params(nested_share_records={ruid: robj}),
-                        record=ruid, email='user@example.com',
+                        record=ruid, email=['user@example.com'],
                         action='grant', role='viewer')
 
     @patch('keepercommander.nested_share_folder.record_api.unshare_record_v3')
@@ -847,9 +847,45 @@ def test_share_record_revoke(self, mock_unshare):
         cmd = NestedShareRecordShareCommand()
         with mock.patch('builtins.print'):
             cmd.execute(_make_params(nested_share_records={ruid: robj}),
-                        record=ruid, email='user@example.com',
+                        record=ruid, email=['user@example.com'],
                         action='revoke')
 
+    @patch('keepercommander.nested_share_folder.record_api.get_record_accesses_v3')
+    @patch('keepercommander.nested_share_folder.record_api.share_record_v3')
+    def test_share_record_rejects_grant_to_owner(self, mock_share, mock_accesses):
+        from keepercommander.commands.nested_share_folder import NestedShareRecordShareCommand
+        ruid, robj = _make_record()
+        owner = 'owner@example.com'
+        mock_accesses.return_value = {
+            'record_accesses': [{
+                'record_uid': ruid,
+                'accessor_name': owner,
+                'owner': True,
+                'access_type': 'AT_USER',
+            }],
+            'forbidden_records': [],
+        }
+        cmd = NestedShareRecordShareCommand()
+        with self.assertRaises(CommandError) as ctx:
+            cmd.execute(_make_params(nested_share_records={ruid: robj}),
+                        record=ruid, email=[owner],
+                        action='grant', role='viewer')
+        self.assertIn('is the owner', str(ctx.exception))
+        mock_share.assert_not_called()
+
+    @patch('keepercommander.nested_share_folder.folder_api.grant_folder_access_v3')
+    def test_share_folder_rejects_grant_to_owner(self, mock_grant):
+        from keepercommander.commands.nested_share_folder import NestedShareFolderShareCommand
+        fuid, fobj = _make_folder()
+        owner = 'owner@example.com'
+        fobj['owner_username'] = owner
+        cmd = NestedShareFolderShareCommand()
+        with self.assertRaises(CommandError) as ctx:
+            cmd.execute(_make_params(nested_share_folders={fuid: fobj}),
+                        folder=[fuid], user=[owner], action='grant', role='viewer')
+        self.assertIn('is the owner', str(ctx.exception))
+        mock_grant.assert_not_called()
+
     @patch('keepercommander.nested_share_folder.folder_api.grant_folder_access_v3')
     def test_share_folder_invite_message_uses_command_prefix(self, mock_grant):
         from keepercommander.commands.nested_share_folder import NestedShareFolderShareCommand

From 73a21b53a6ec7fa82072c6463c75a5a53303c942 Mon Sep 17 00:00:00 2001
From: sshrushanth-ks <sshrushanth@keepersecurity.com>
Date: Wed, 22 Jul 2026 15:23:43 +0530
Subject: [PATCH 2/3] resolved review comments

---
 .../commands/nested_share_folder/helpers.py   | 10 +++-
 keepercommander/commands/register.py          | 53 ++++++++++++++-----
 .../nested_share_folder/record_api.py         | 15 +++---
 unit-tests/test_command_register.py           | 48 +++++++++--------
 4 files changed, 81 insertions(+), 45 deletions(-)

diff --git a/keepercommander/commands/nested_share_folder/helpers.py b/keepercommander/commands/nested_share_folder/helpers.py
index 8e02fa99c..bed478102 100644
--- a/keepercommander/commands/nested_share_folder/helpers.py
+++ b/keepercommander/commands/nested_share_folder/helpers.py
@@ -189,14 +189,20 @@ def raise_if_record_share_target_is_owner(params, record_uid, email, cmd_name, *
 
     Ownership transfer (``-a owner``) is allowed unless the target already owns
     the record (``allow_owner_transfer=True`` then raises a no-op-style error).
+
+    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__).debug(
-            "Could not resolve owner for record '%s': %s", record_uid, exc)
+        logging.getLogger(__name__).warning(
+            "Could not resolve owner for record '%s'; proceeding without "
+            "client-side owner check: %s", record_uid, exc)
         return
     owner = find_record_owner_username(access_result, record_uid)
     if not owner or owner.casefold() != email.casefold():
diff --git a/keepercommander/commands/register.py b/keepercommander/commands/register.py
index 0b4c0a6bd..4c3222e6b 100644
--- a/keepercommander/commands/register.py
+++ b/keepercommander/commands/register.py
@@ -313,7 +313,16 @@ def _owner_share_target_message(email, entity='record'):
 
 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."""
+    """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()
@@ -321,16 +330,20 @@ def _is_shared_folder_owner_email(shared_folder, email, params=None):
     if owner_username and owner_username.casefold() == email_cf:
         return True
     # Some sync payloads also put an owner flag on the user row.
-    for user in shared_folder.get('users', []):
-        username = (user.get('username') or '')
-        if username.casefold() != email_cf:
-            continue
-        if user.get('owner') is True:
-            return True
-    # Fall back: owner_account_uid + current session user.
+    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')
-    if params and owner_uid and params.user and params.user.casefold() == email_cf:
-        if owner_uid == utils.base64_url_encode(params.account_uid_bytes):
+    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
 
@@ -591,10 +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
-                                 and not _is_shared_folder_owner_email(sh_fol, x['username'], params=params)))
+                            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:
@@ -1163,6 +1184,10 @@ def apply_share_expiration(ro):   # type: (record_pb2.SharedRecord) -> None
             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.
diff --git a/keepercommander/nested_share_folder/record_api.py b/keepercommander/nested_share_folder/record_api.py
index 40c4b26e5..8c4bb371b 100644
--- a/keepercommander/nested_share_folder/record_api.py
+++ b/keepercommander/nested_share_folder/record_api.py
@@ -362,13 +362,14 @@ def find_direct_user_share_access(access_result, record_uid, email):
 
 def find_record_owner_username(access_result, record_uid):
     """Return the owner username for *record_uid* from a get_record_accesses_v3 result."""
-    for access in access_result.get('record_accesses', []):
-        if access.get('record_uid') != record_uid:
-            continue
-        if access.get('owner'):
-            return access.get('accessor_name') or ''
-    return ''
-
+    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
 
diff --git a/unit-tests/test_command_register.py b/unit-tests/test_command_register.py
index 8e734d72a..c5a8b9d88 100644
--- a/unit-tests/test_command_register.py
+++ b/unit-tests/test_command_register.py
@@ -68,25 +68,38 @@ 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()
 
-        def shared_with_owner(params_, record_uids, is_share_admin):
-            for uid in record_uids:
-                if uid in params_.record_cache:
-                    params_.record_cache[uid]['shares'] = {
-                        'user_permissions': [
-                            {'username': params_.user, 'owner': True},
-                            {'username': 'user2@keepersecurity.com', 'owner': False,
-                             'shareable': False, 'editable': False},
-                        ]
-                    }
-
         self.record_share_mock = mock.patch('keepercommander.api.get_record_shares').start()
-        self.record_share_mock.side_effect = shared_with_owner
+        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(
@@ -103,17 +116,8 @@ def test_share_record_rejects_owner_transfer_to_self(self):
         record_uid = next(iter([x['record_uid'] for x in params.meta_data_cache.values() if x['can_share']]))
         cmd = register.ShareRecordCommand()
 
-        def shared_with_owner(params_, record_uids, is_share_admin):
-            for uid in record_uids:
-                if uid in params_.record_cache:
-                    params_.record_cache[uid]['shares'] = {
-                        'user_permissions': [
-                            {'username': params_.user, 'owner': True},
-                        ]
-                    }
-
         self.record_share_mock = mock.patch('keepercommander.api.get_record_shares').start()
-        self.record_share_mock.side_effect = shared_with_owner
+        self.record_share_mock.side_effect = self._shared_with_owner_side_effect()
 
         with self.assertRaises(CommandError) as ctx:
             cmd.prep_request(params, dict(

From 87293d61b427102a9a02e38b65318caa04cb403f Mon Sep 17 00:00:00 2001
From: sshrushanth-ks <sshrushanth@keepersecurity.com>
Date: Thu, 23 Jul 2026 11:50:46 +0530
Subject: [PATCH 3/3] addressed logging and variable naming comments

---
 .../nested_share_folder/folder_commands.py         |  6 ++++++
 .../commands/nested_share_folder/helpers.py        | 14 ++++++++------
 .../nested_share_folder/sharing_commands.py        |  2 +-
 3 files changed, 15 insertions(+), 7 deletions(-)

diff --git a/keepercommander/commands/nested_share_folder/folder_commands.py b/keepercommander/commands/nested_share_folder/folder_commands.py
index 09259fcee..a5f76f32f 100644
--- a/keepercommander/commands/nested_share_folder/folder_commands.py
+++ b/keepercommander/commands/nested_share_folder/folder_commands.py
@@ -439,6 +439,9 @@ def _expand_existing(params, folder_uid, folder_arg):
                         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:
@@ -455,6 +458,9 @@ def _expand_existing(params, folder_uid, folder_arg):
                     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:
diff --git a/keepercommander/commands/nested_share_folder/helpers.py b/keepercommander/commands/nested_share_folder/helpers.py
index bed478102..ff1ed942b 100644
--- a/keepercommander/commands/nested_share_folder/helpers.py
+++ b/keepercommander/commands/nested_share_folder/helpers.py
@@ -184,11 +184,13 @@ def is_nested_share_folder_owner_email(params, folder_uid, email):
     return bool(owner_username) and owner_username.casefold() == email.casefold()
 
 
-def raise_if_record_share_target_is_owner(params, record_uid, email, cmd_name, *, allow_owner_transfer=False):
+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*.
 
-    Ownership transfer (``-a owner``) is allowed unless the target already owns
-    the record (``allow_owner_transfer=True`` then raises a no-op-style error).
+    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
@@ -204,10 +206,10 @@ def raise_if_record_share_target_is_owner(params, record_uid, email, cmd_name, *
             "Could not resolve owner for record '%s'; proceeding without "
             "client-side owner check: %s", record_uid, exc)
         return
-    owner = find_record_owner_username(access_result, record_uid)
-    if not owner or owner.casefold() != email.casefold():
+    owner_username = find_record_owner_username(access_result, record_uid)
+    if not owner_username or owner_username.casefold() != email.casefold():
         return
-    if allow_owner_transfer:
+    if is_ownership_transfer:
         raise CommandError(
             cmd_name,
             f"'{email}' already owns this record. Ownership transfer is a no-op.")
diff --git a/keepercommander/commands/nested_share_folder/sharing_commands.py b/keepercommander/commands/nested_share_folder/sharing_commands.py
index f5e1fcc39..62f78da0f 100644
--- a/keepercommander/commands/nested_share_folder/sharing_commands.py
+++ b/keepercommander/commands/nested_share_folder/sharing_commands.py
@@ -89,7 +89,7 @@ def execute(self, params, **kwargs):
                 for record_uid in record_uids:
                     raise_if_record_share_target_is_owner(
                         params, record_uid, email, 'nsf-share-record',
-                        allow_owner_transfer=(action == 'owner'))
+                        is_ownership_transfer=(action == 'owner'))
                     result, effective_action = self._dispatch(
                         params, action, record_uid, email, access_role_type, expiration)
                     self._log_results(result, effective_action, email)