Skip to content
Merged
2 changes: 1 addition & 1 deletion keepercommander/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,4 +10,4 @@
# Contact: commander@keepersecurity.com
#

__version__ = '18.1.4'
__version__ = '18.1.5'
5 changes: 3 additions & 2 deletions keepercommander/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -845,9 +845,10 @@ def execute_router(params, endpoint, request, *, rs_type=None):

payload = request.SerializeToString() if request else None
rs_bytes = execute_router_rest(params, endpoint, payload)
if rs_type and rs_bytes:
if rs_type and rs_bytes is not None:
response = rs_type()
response.ParseFromString(rs_bytes)
if rs_bytes:
response.ParseFromString(rs_bytes)
if logger.level <= logging.DEBUG:
js = google.protobuf.json_format.MessageToJson(response)
logger.debug('>>> [ROUTER RS] \"%s\": %s', endpoint, js)
Expand Down
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")
Loading
Loading