Permission bug fixes for NSF folders and record sharing

Author: adeshmukh-ks

keepercli-package/src/keepercli/commands/nsf_commands.py

@@ -1279,7 +1279,7 @@ def execute(self, context: KeeperParams, **kwargs):
                 vault, kwargs.get('folder'), action=action, role=role,
                 recursive=kwargs.get('recursive', False), current_user=login))
 
-        if not plan.updates and not plan.creates and not plan.revokes:
+        if not plan.updates and not plan.creates and not plan.revokes and not plan.denies:
             if plan.skipped:
                 logger.warning('No permission changes can be made (see skipped entries).')
             else:
@@ -1309,7 +1309,7 @@ def execute(self, context: KeeperParams, **kwargs):
     def _print_plan(plan) -> None:
         for label, items in (
                 ('SKIP', plan.skipped), ('GRANT/UPDATE', plan.updates + plan.creates),
-                ('REVOKE', plan.revokes)):
+                ('REVOKE', plan.revokes), ('DENY INHERITED', plan.denies)):
             if not items:
                 continue
             logger.info(f'\n{label}:')

keepersdk-package/src/keepersdk/vault/nsf_common.py

@@ -2,7 +2,8 @@
 
 from __future__ import annotations
 
-from typing import Any, Dict, Optional, Tuple
+import json
+from typing import Any, Dict, List, Optional, Tuple
 
 from .. import crypto, utils
 from ..proto import folder_pb2, record_pb2, record_sharing_pb2
@@ -101,18 +102,21 @@ def encrypt_for_team(
         plaintext_key: bytes,
         team_keys,
         *,
+        prefer_aes: bool = False,
         forbid_rsa: bool = False) -> Tuple[bytes, int]:
     aes = getattr(team_keys, 'aes', None)
     ec_bytes = getattr(team_keys, 'ec', None)
     rsa_bytes = getattr(team_keys, 'rsa', None)
+    if prefer_aes and aes:
+        if forbid_rsa:
+            return crypto.encrypt_aes_v2(plaintext_key, aes), folder_pb2.encrypted_by_data_key_gcm
+        return crypto.encrypt_aes_v1(plaintext_key, aes), folder_pb2.encrypted_by_data_key
     if rsa_bytes and not forbid_rsa:
         rsa_key = crypto.load_rsa_public_key(rsa_bytes)
         return crypto.encrypt_rsa(plaintext_key, rsa_key), folder_pb2.encrypted_by_public_key
     if ec_bytes:
         ec_key = crypto.load_ec_public_key(ec_bytes)
         return crypto.encrypt_ec(plaintext_key, ec_key), folder_pb2.encrypted_by_public_key_ecc
-    if aes:
-        return crypto.encrypt_aes_v2(plaintext_key, aes), folder_pb2.encrypted_by_data_key_gcm
     raise ValueError('No public key found for team')
 
 
@@ -320,6 +324,292 @@ def folder_access_role_label(
     return 'unknown'
 
 
+_PERMISSION_CAMEL_KEYS: Dict[str, str] = {
+    'can_update_access': 'canUpdateAccess',
+    'can_update_setting': 'canUpdateSetting',
+    'can_delete': 'canDelete',
+    'can_change_ownership': 'canChangeOwnership',
+    'can_edit': 'canEdit',
+    'can_view': 'canView',
+    'can_list_access': 'canListAccess',
+}
+
+
+def _current_user_account_uid_b64(vault: VaultOnline) -> str:
+    account_uid = vault.keeper_auth.auth_context.account_uid
+    return utils.base64_url_encode(account_uid) if account_uid else ''
+
+
+def _parse_permissions_blob(raw: Any) -> Dict[str, Any]:
+    if isinstance(raw, dict):
+        return raw
+    if isinstance(raw, str) and raw:
+        try:
+            parsed = json.loads(raw)
+            return parsed if isinstance(parsed, dict) else {}
+        except (TypeError, ValueError):
+            return {}
+    return {}
+
+
+def _permission_value(perms: Dict[str, Any], key: str) -> bool:
+    if not perms:
+        return False
+    if key in perms:
+        return bool(perms[key])
+    camel = _PERMISSION_CAMEL_KEYS.get(key)
+    if camel and camel in perms:
+        return bool(perms[camel])
+    return False
+
+
+def _access_type_is_owner(access_type: Any) -> bool:
+    if access_type == folder_pb2.AT_OWNER:
+        return True
+    return isinstance(access_type, str) and access_type == 'AT_OWNER'
+
+
+def is_current_user_nsf_accessor(
+        accessor: Dict[str, Any],
+        vault: VaultOnline,
+        account_uid_b64: str) -> bool:
+    """Return True when *accessor* belongs to the logged-in user."""
+    username = vault.keeper_auth.auth_context.username
+    accessor_username = accessor.get('username') or accessor.get('accessor_name')
+    if accessor_username and username:
+        return accessor_username.casefold() == username.casefold()
+    accessor_uid = accessor.get('access_type_uid') or accessor.get('accessor_uid')
+    return bool(accessor_uid and account_uid_b64 and accessor_uid == account_uid_b64)
+
+
+def _folder_owner_info(vault: VaultOnline, folder_uid: str) -> Tuple[Optional[str], Optional[str]]:
+    view = vault.nsf_data
+    if view is None:
+        return None, None
+    row = view.storage.folders.get_entity(folder_uid)
+    if row is None:
+        return None, None
+    return row.owner_username or None, row.owner_account_uid or None
+
+
+def is_nsf_folder_owner_user(vault: VaultOnline, folder_uid: str) -> bool:
+    """Return True when the logged-in user owns *folder_uid*."""
+    account_uid_b64 = _current_user_account_uid_b64(vault)
+    username = vault.keeper_auth.auth_context.username
+    owner_username, owner_account_uid = _folder_owner_info(vault, folder_uid)
+    if owner_account_uid and account_uid_b64 and owner_account_uid == account_uid_b64:
+        return True
+    if owner_username and username and owner_username.casefold() == username.casefold():
+        return True
+    return False
+
+
+def _folder_accessor_from_storage(fa: Any) -> Dict[str, Any]:
+    return {
+        'access_type_uid': fa.access_type_uid,
+        'access_type': fa.access_type,
+        'permissions': _parse_permissions_blob(fa.permissions_json),
+    }
+
+
+def collect_nsf_folder_accessors(vault: VaultOnline, folder_uid: str) -> List[Dict[str, Any]]:
+    """Folder accessor rows from sync cache, falling back to the access API."""
+    accessors: List[Dict[str, Any]] = []
+    view = vault.nsf_data
+    if view is not None:
+        for fa in view.storage.folder_accesses.get_links_by_subject(folder_uid):
+            accessors.append(_folder_accessor_from_storage(fa))
+    if accessors:
+        return accessors
+    from .nsf_management import get_nsf_folder_access
+    try:
+        info = get_nsf_folder_access(vault, [folder_uid])
+        for result in info.get('results', []):
+            if result.get('success'):
+                accessors.extend(result.get('accessors', []))
+    except Exception:
+        pass
+    return accessors
+
+
+def _record_accessor_from_storage(ra: Any) -> Dict[str, Any]:
+    return {
+        'access_type_uid': ra.access_type_uid,
+        'owner': ra.owner,
+        'inherited': ra.inherited,
+        'denied_access': ra.denied_access,
+        'can_update_access': ra.can_update_access,
+        'can_change_ownership': ra.can_change_ownership,
+        'can_delete': ra.can_delete,
+        'can_edit': ra.can_edit,
+    }
+
+
+def find_record_user_accesses(
+        vault: VaultOnline,
+        record_uid: str,
+        recipient_email: str) -> List[Dict[str, Any]]:
+    """Return non-owner AT_USER accessor rows for *recipient_email* on *record_uid*."""
+    email_cf = recipient_email.casefold()
+    matches: List[Dict[str, Any]] = []
+    for accessor in collect_nsf_record_accessors(vault, record_uid):
+        if accessor.get('owner'):
+            continue
+        access_type = accessor.get('access_type') or 'AT_USER'
+        if access_type not in ('AT_USER', ''):
+            continue
+        accessor_name = accessor.get('accessor_name') or accessor.get('username') or ''
+        if accessor_name.casefold() != email_cf:
+            continue
+        matches.append(accessor)
+    return matches
+
+
+def record_user_has_direct_access(accesses: List[Dict[str, Any]]) -> bool:
+    return any(not accessor.get('inherited') for accessor in accesses)
+
+
+def record_user_has_inherited_access(accesses: List[Dict[str, Any]]) -> bool:
+    return any(accessor.get('inherited') for accessor in accesses)
+
+
+def collect_nsf_record_accessors(vault: VaultOnline, record_uid: str) -> List[Dict[str, Any]]:
+    """Record accessor rows from sync cache, falling back to the access API."""
+    accessors: List[Dict[str, Any]] = []
+    view = vault.nsf_data
+    if view is not None:
+        for ra in view.storage.record_accesses.get_links_by_subject(record_uid):
+            accessors.append(_record_accessor_from_storage(ra))
+    if accessors:
+        return accessors
+    from .nsf_management import get_nsf_record_accesses
+    try:
+        info = get_nsf_record_accesses(vault, [record_uid])
+        accessors.extend(info.get('record_accesses', []))
+    except Exception:
+        pass
+    return accessors
+
+
+def _record_permission_value(accessor: Dict[str, Any], key: str) -> bool:
+    if key in accessor:
+        return bool(accessor[key])
+    return _permission_value(accessor.get('permissions') or {}, key)
+
+
+def require_nsf_folder_permission(
+        vault: VaultOnline,
+        folder_uid: str,
+        permission_key: str,
+        error_message: str) -> None:
+    """Raise ValueError when the current user lacks *permission_key* on a folder."""
+    if is_nsf_folder_owner_user(vault, folder_uid):
+        return
+
+    accessors = collect_nsf_folder_accessors(vault, folder_uid)
+    if not accessors:
+        return
+
+    account_uid_b64 = _current_user_account_uid_b64(vault)
+    owner_username, owner_account_uid = _folder_owner_info(vault, folder_uid)
+    for accessor in accessors:
+        if not is_current_user_nsf_accessor(accessor, vault, account_uid_b64):
+            continue
+        if (accessor.get('owner')
+                or _access_type_is_owner(accessor.get('access_type'))
+                or is_nsf_folder_owner(accessor, owner_username, owner_account_uid)):
+            return
+        perms = accessor.get('permissions') or {}
+        if _permission_value(perms, permission_key):
+            return
+        raise ValueError(error_message)
+
+    raise ValueError(error_message)
+
+
+def require_nsf_folder_share_permission(vault: VaultOnline, folder_uid: str) -> None:
+    """Raise ValueError when the current user cannot share or manage folder access."""
+    require_nsf_folder_permission(
+        vault,
+        folder_uid,
+        'can_update_access',
+        'You do not have permission to share this folder.')
+
+
+def require_nsf_record_permission(
+        vault: VaultOnline,
+        record_uid: str,
+        permission_key: str,
+        error_message: str) -> None:
+    """Raise ValueError when the current user lacks *permission_key* on a record."""
+    accessors = collect_nsf_record_accessors(vault, record_uid)
+    if not accessors:
+        return
+
+    account_uid_b64 = _current_user_account_uid_b64(vault)
+    for accessor in accessors:
+        if not is_current_user_nsf_accessor(accessor, vault, account_uid_b64):
+            continue
+        if accessor.get('owner'):
+            return
+        if _record_permission_value(accessor, permission_key):
+            return
+        raise ValueError(error_message)
+
+    raise ValueError(error_message)
+
+
+def require_nsf_record_share_permission(vault: VaultOnline, record_uid: str) -> None:
+    """Raise ValueError when the current user cannot share or manage record access."""
+    require_nsf_record_permission(
+        vault,
+        record_uid,
+        'can_update_access',
+        'You do not have permission to share this record.')
+
+
+def require_nsf_record_ownership_permission(vault: VaultOnline, record_uid: str) -> None:
+    """Raise ValueError when the current user cannot transfer record ownership."""
+    require_nsf_record_permission(
+        vault,
+        record_uid,
+        'can_change_ownership',
+        'You do not have permission to transfer ownership of this record.')
+
+
+def folder_inherits_parent_permissions(vault: VaultOnline, folder_uid: str) -> bool:
+    """Return True when *folder_uid* has a parent and still inherits its access list."""
+    from .nsf_management import ROOT_FOLDER_UID
+
+    view = vault.nsf_data
+    if view is None:
+        return False
+    node = view.get_folder(folder_uid)
+    parent_uid = node.parent_uid if node else None
+    if not parent_uid:
+        row = view.storage.folders.get_entity(folder_uid)
+        parent_uid = row.parent_uid if row else None
+    if not parent_uid or parent_uid == ROOT_FOLDER_UID:
+        return False
+    row = view.storage.folders.get_entity(folder_uid)
+    if row is None:
+        return True
+    return row.inherit_user_permissions != int(folder_pb2.BOOLEAN_FALSE)
+
+
+def ensure_folder_direct_permissions(
+        vault: VaultOnline,
+        folder_uid: str,
+        *,
+        request_sync: bool = False) -> bool:
+    """Disable parent permission inheritance so folder access changes apply locally."""
+    if not folder_inherits_parent_permissions(vault, folder_uid):
+        return False
+    from .nsf_management import update_nsf_folder
+    update_nsf_folder(vault, folder_uid, inherit_permissions=False, request_sync=request_sync)
+    return True
+
+
 def access_role_label(access: Dict[str, Any]) -> str:
     if access.get('owner'):
         return 'owner'

keepersdk-package/src/keepersdk/vault/nsf_management.py

@@ -786,8 +786,11 @@ def get_nsf_record_accesses(
         }
         for flag in ('can_view_title', 'can_edit', 'can_view', 'can_list_access',
                      'can_update_access', 'can_delete', 'can_change_ownership',
-                     'can_request_access', 'can_approve_access'):
-            ao[flag] = getattr(d, flag, False)
+                     'can_request_access', 'can_approve_access', 'denied_access'):
+            if flag == 'denied_access':
+                ao[flag] = getattr(d, 'deniedAccess', False)
+            else:
+                ao[flag] = getattr(d, flag, False)
         result['record_accesses'].append(ao)
     for fu in rs.forbiddenRecords:
         result['forbidden_records'].append(utils.base64_url_encode(fu))