Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion keepercommander/command_categories.py
Original file line number Diff line number Diff line change
Expand Up @@ -116,7 +116,7 @@
'nsf-mkdir', 'nsf-record-add', 'nsf-record-update', 'nsf-rndir', 'nsf-list',
'nsf-share-folder', 'nsf-record-details', 'nsf-share-record',
'nsf-record-permission', 'nsf-transfer-record',
'nsf-ln', 'nsf-rm', 'nsf-rmdir', 'nsf-shortcut', 'nsf-get'
'nsf-ln', 'nsf-rm', 'nsf-rmdir', 'nsf-shortcut', 'nsf-get', 'nsf-move'
},

# Legacy Commands
Expand Down
3 changes: 3 additions & 0 deletions keepercommander/commands/nested_share_folder/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@
NestedShareFolderListCommand,
NestedShareFolderShareCommand,
NestedShareFolderRemoveCommand,
NestedSharedMoveCommander,
)

# Record commands
Expand Down Expand Up @@ -77,6 +78,7 @@ def register_commands(commands):
commands['nsf-rmdir'] = NestedShareFolderRemoveCommand()
commands['nsf-shortcut'] = NestedShareRecordShortcutCommand()
commands['nsf-get'] = NestedShareGetCommand()
commands['nsf-move'] = NestedSharedMoveCommander()


def register_command_info(aliases, command_info):
Expand All @@ -96,3 +98,4 @@ def register_command_info(aliases, command_info):
command_info['nsf-rmdir'] = 'Remove a Nested Share Folder and its contents'
command_info['nsf-shortcut'] = 'Manage Nested Share Record shortcuts'
command_info['nsf-get'] = 'Get details of a Nested Share Record or folder'
command_info['nsf-move'] = 'Move a Nested Share Record or folder to a new location'
26 changes: 22 additions & 4 deletions keepercommander/commands/nested_share_folder/folder_commands.py
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@
nested_share_folder_list_parser,
nested_share_folder_share_parser,
nested_share_folder_rmdir_parser,
nested_share_move_parser
)


Expand Down Expand Up @@ -209,7 +210,7 @@ def execute(self, params, **kwargs):

with command_error_handler('nsf-rndir'):
result = _nsf.update_folder_v3(
params=params, folder_uid=folder_arg, folder_name=new_name,
params=params, folder_uid=folder_uid, folder_name=new_name,
color=color,
)
check_result(result, 'nsf-rndir')
Expand Down Expand Up @@ -447,6 +448,7 @@ def _expand_existing(params, folder_uid, folder_arg):
that can be removed, including inherited access.
"""
from keepercommander.proto import folder_pb2
at_owner = int(folder_pb2.AT_OWNER)
at_user = int(folder_pb2.AT_USER)
at_team = int(folder_pb2.AT_TEAM)

Expand All @@ -457,13 +459,14 @@ def _expand_existing(params, folder_uid, folder_arg):
if not fr.get('success'):
continue
for accessor in fr.get('accessors', []):
if accessor.get('access_type') == 'AT_OWNER':
access_type = int(accessor.get('access_type', 0) or 0)
if access_type == at_owner:
continue
if accessor.get('access_type') == 'AT_TEAM':
if access_type == at_team:
team_uid = accessor.get('accessor_uid')
if team_uid:
result.append(('team', team_uid))
elif accessor.get('access_type') == 'AT_USER':
elif 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):
Expand Down Expand Up @@ -657,3 +660,18 @@ def _impact_summary(folder_uid, name, operation, impact, quiet):
for w in impact.get('warnings', []):
lines.append(f" Warning: {w}")
return lines


class NestedSharedMoveCommander(Command):
""" Move a Nested Share Folder or record. """

def get_parser(self):
return nested_share_move_parser

def execute(self, params, **kwargs):
from .helpers import move_nested_share_item
source = kwargs.get('src')
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)
191 changes: 186 additions & 5 deletions keepercommander/commands/nested_share_folder/helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
from typing import Optional

from ...error import CommandError
from ... import nested_share_folder as _nsf

logger = logging.getLogger(__name__)

Expand Down Expand Up @@ -144,6 +145,8 @@ def suppress_exit(self, status=0, message=None):

def normalize_parent_uid(uid):
"""Normalize root folder UIDs to a consistent ``'root'`` or empty string."""
if not isinstance(uid, str):
uid = str(uid) if uid else ''
if uid == ROOT_FOLDER_UID or uid == 'root':
return 'root'
return uid or ''
Expand Down Expand Up @@ -191,7 +194,12 @@ def is_nested_share_folder_owner_email(params, folder_uid, 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()
if not owner_username:
logger.debug('Folder owner_username not found for folder_uid=%s', folder_uid)
return False
if not isinstance(email, str):
email = str(email)
return owner_username.casefold() == email.casefold()


def raise_if_record_share_target_is_owner(params, record_uid, email, cmd_name, *,
Expand Down Expand Up @@ -367,21 +375,179 @@ def add_records(fuid):

visited = set()

# Build parent→children index for efficient tree traversal
children_index = {}
if recursive:
for child_uid, child_obj in nsf_folders.items():
parent = child_obj.get('parent_uid') or ROOT_FOLDER_UID
if parent not in children_index:
children_index[parent] = []
children_index[parent].append(child_uid)

def walk(fuid):
if fuid in visited:
return
visited.add(fuid)
add_records(fuid)
if not recursive:
return
for child_uid, child_obj in nsf_folders.items():
if child_obj.get('parent_uid') == fuid and child_uid not in visited:
for child_uid in children_index.get(fuid, []):
if child_uid not in visited:
walk(child_uid)

walk(folder_uid)
return record_uids


# ═══════════════════════════════════════════════════════════════════════════
# Move
# ═══════════════════════════════════════════════════════════════════════════

MAX_NESTED_SHARE_FOLDER_DEPTH = 5
"""Maximum number of folder levels below the Nested Share Folder root."""


def _folder_depth(params, folder_uid):
"""Return the nesting depth of *folder_uid* (root's direct children = 1)."""
nsf_folders = getattr(params, 'nested_share_folders', {})
depth = 0
cur = folder_uid
visited = set()
while cur and cur != ROOT_FOLDER_UID and cur in nsf_folders and cur not in visited:
visited.add(cur)
depth += 1
cur = nsf_folders[cur].get('parent_uid')
return depth


def _folder_subtree_height(params, folder_uid, _visited=None):
"""Return the number of additional folder levels below *folder_uid*
(0 when it has no sub-folders).
"""
nsf_folders = getattr(params, 'nested_share_folders', {})
visited = _visited if _visited is not None else set()
if folder_uid in visited:
return 0
visited.add(folder_uid)
children = [uid for uid, obj in nsf_folders.items()
if obj.get('parent_uid') == folder_uid]
if not children:
return 0
return 1 + max(_folder_subtree_height(params, c, visited) for c in children)


def validate_folder_move_depth(params, folder_uid, dest_folder_uid, cmd_name):
"""Raise if moving *folder_uid* under *dest_folder_uid* would push any of
its descendants past ``MAX_NESTED_SHARE_FOLDER_DEPTH`` levels.
"""
dest_depth = _folder_depth(params, dest_folder_uid)
subtree_height = _folder_subtree_height(params, folder_uid)
new_depth = dest_depth + 1 + subtree_height
if new_depth > MAX_NESTED_SHARE_FOLDER_DEPTH:
raise CommandError(
cmd_name,
f"Cannot move folder: resulting nesting depth ({new_depth}) would "
f"exceed the maximum of {MAX_NESTED_SHARE_FOLDER_DEPTH} levels.")


def resolve_move_destination_folder(params, destination, cmd_name):
"""Resolve *destination* to an NSF folder UID, or ``ROOT_FOLDER_UID``.

Mirrors the classic ``mv`` command's root-folder detection: an empty
value, the literal ``'root'``, or the NSF root sentinel UID all mean the
Nested Share Folder root.
"""
if not destination or destination.strip().casefold() == 'root':
return ROOT_FOLDER_UID
if destination == ROOT_FOLDER_UID:
return ROOT_FOLDER_UID
resolved = resolve_folder_uid(params, destination)
if not resolved or not is_nested_share_folder(params, resolved):
raise CommandError(
cmd_name,
f"Destination '{destination}' is not a Nested Share Folder or "
f"the root folder.")
return resolved


def _check_move_result(result, cmd_name):
"""Raise ``CommandError`` when a folder/record move result is not a success."""
if result.get('success'):
return
status = result.get('move_result_status')
message = normalize_nsf_user_message(result.get('message'))
if status and status not in ('MOVE_RESULT_STATUS_UNSPECIFIED', ''):
detail = status.replace('_', ' ').title()
raise CommandError(cmd_name, f"{detail}{(': ' + message) if message else ''}")
raise CommandError(cmd_name, message or 'Move failed')


def _move_nested_share_folder(params, folder_uid, dest_folder_uid, cmd_name):

if folder_uid == dest_folder_uid:
raise CommandError(cmd_name, 'Cannot move a folder into itself.')

folder_obj = getattr(params, 'nested_share_folders', {}).get(folder_uid) or {}
current_parent_uid = folder_obj.get('parent_uid') or ROOT_FOLDER_UID

check_folder_remove_permission(params, current_parent_uid, cmd_name)
check_folder_add_permission(params, dest_folder_uid, cmd_name)
validate_folder_move_depth(params, folder_uid, dest_folder_uid, cmd_name)

target = None if dest_folder_uid == ROOT_FOLDER_UID else dest_folder_uid
with command_error_handler(cmd_name):
result = _nsf.move_folder_v3(params, folder_uid, target_parent_uid=target)
_check_move_result(result, cmd_name)
params.sync_data = True


def _move_nested_share_record(params, record_uid, dest_folder_uid, cmd_name):

location = find_folder_location(params, record_uid)
src_folder_uid = (location or {}).get('uid') or ROOT_FOLDER_UID
if src_folder_uid == dest_folder_uid:
raise CommandError(cmd_name, 'Record is already in the destination folder.')

check_folder_remove_permission(params, src_folder_uid, cmd_name)
check_folder_add_permission(params, dest_folder_uid, cmd_name)

source = None if src_folder_uid == ROOT_FOLDER_UID else src_folder_uid
target = None if dest_folder_uid == ROOT_FOLDER_UID else dest_folder_uid
with command_error_handler(cmd_name):
result = _nsf.move_folder_record_v3(
params, record_uid, source_folder_uid=source, target_folder_uid=target)
_check_move_result(result, cmd_name)
params.sync_data = True


def move_nested_share_item(params, source, destination):
"""Move a Nested Share Folder record or folder to a new location.

*source* must resolve to an existing Nested Share Folder record or
folder; legacy (non-NSF) items are rejected. *destination* must resolve
to a Nested Share Folder or the root. Folder moves are capped at
``MAX_NESTED_SHARE_FOLDER_DEPTH`` nesting levels; records may be moved
into a folder at that maximum depth.
"""
from ...nested_share_folder import removal_api as _removal_api

cmd_name = 'nsf-move'
dest_folder_uid = resolve_move_destination_folder(params, destination, cmd_name)

folder_uid = _removal_api.resolve_nested_share_folder_uid(params, source)
if folder_uid and is_nested_share_folder(params, folder_uid):
_move_nested_share_folder(params, folder_uid, dest_folder_uid, cmd_name)
return

record_uid = _removal_api.resolve_nested_share_record_uid(params, source)
if record_uid and is_nested_share_record(params, record_uid):
_move_nested_share_record(params, record_uid, dest_folder_uid, cmd_name)
return

raise CommandError(
cmd_name, f"'{source}' is not a Nested Share Folder record or folder.")


# ═══════════════════════════════════════════════════════════════════════════
# Expiration parsing
# ═══════════════════════════════════════════════════════════════════════════
Expand Down Expand Up @@ -541,6 +707,8 @@ def format_role_display(role):
from ...proto import folder_pb2
try:
role = folder_pb2.AccessRoleType.Name(role)
except ValueError:
return f'UNKNOWN({role})'
except Exception:
return str(role)
if isinstance(role, str):
Expand Down Expand Up @@ -568,9 +736,9 @@ def get_access_role_label(access):
# ═══════════════════════════════════════════════════════════════════════════

def format_timestamp(ms):
"""Format a millisecond epoch timestamp as ``'YYYY-MM-DD HH:MM:SS'``."""
"""Format a millisecond epoch timestamp as ``'YYYY-MM-DD HH:MM:SS'`` (UTC)."""
if ms:
return datetime.datetime.fromtimestamp(ms / 1000).strftime('%Y-%m-%d %H:%M:%S')
return datetime.datetime.fromtimestamp(ms / 1000, tz=timezone.utc).strftime('%Y-%m-%d %H:%M:%S')
return ''


Expand All @@ -596,6 +764,18 @@ def check_folder_delete_permission(params, folder_uid, cmd_name):
'You do not have permission to delete this folder.', cmd_name)


def check_folder_add_permission(params, folder_uid, cmd_name):
"""Raise if the current user cannot add items into the folder."""
_check_folder_permission(params, folder_uid, 'can_add',
'You do not have permission to add items to this folder.', cmd_name)


def check_folder_remove_permission(params, folder_uid, cmd_name):
"""Raise if the current user cannot remove items from the folder."""
_check_folder_permission(params, folder_uid, 'can_remove',
'You do not have permission to remove items from this folder.', cmd_name)


def check_record_edit_permission(params, record_uid, cmd_name):
"""Raise if the current user cannot edit the record."""
_check_record_permission(params, record_uid, 'can_edit',
Expand Down Expand Up @@ -658,6 +838,7 @@ def _check_folder_permission(params, folder_uid, permission_key, error_message,
from ...proto import folder_pb2
accesses = getattr(params, 'nested_share_folder_accesses', {}).get(folder_uid, [])
if not accesses:
logger.debug('No accesses found for folder_uid=%s; skipping permission check', folder_uid)
return

current_account_uid = _current_user_account_uid(params)
Expand Down
15 changes: 15 additions & 0 deletions keepercommander/commands/nested_share_folder/parsers.py
Original file line number Diff line number Diff line change
Expand Up @@ -371,3 +371,18 @@ def _make_parser(prog, description):
nested_share_get_parser.add_argument(
'--include-dag', dest='include_dag', action='store_true', default=False,
help='Include DAG/GraphSync information in json output (PAM record types only)')


# ══════════════════════════════════════════════════════════════════════════
# Move parser
# ══════════════════════════════════════════════════════════════════════════

nested_share_move_parser = _make_parser(
'nsf-move',
'Move a Nested Share Record or folder to a new location')
nested_share_move_parser.add_argument(
'src', type=str, help='Source record or folder UID/title/path')
nested_share_move_parser.add_argument(
'dst', type=str,
help="Destination folder UID/title/path, or 'root' for the Nested "
"Share Folder root")
4 changes: 4 additions & 0 deletions keepercommander/nested_share_folder/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,10 @@
'find_nested_share_folders_for_record',
'resolve_nested_share_record_uid', 'resolve_nested_share_folder_uid',
],
'move_api': [
'folder_move_v3', 'move_folder_v3',
'folder_record_move_v3', 'move_folder_record_v3',
],
'acl_cache': [
'warm_for_tree', 'warm_nsf_folder_share_cache', 'warm_nsf_record_share_cache',
'warm_classic_record_shares', 'clear_share_caches', 'ensure_share_caches',
Expand Down
Loading