Skip to content

Streamline enterprise node, role, and team management bug fix #1773

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
pvagare-ks merged 3 commits into from
Jan 21, 2026
+335 −29
Merged

Streamline enterprise node, role, and team management bug fix #1773

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
c498cf7
initial commit
pvagare-ks Jan 20, 2026
7e35dc2
updated unit test
pvagare-ks Jan 20, 2026
89e66d9
updated status code
pvagare-ks Jan 20, 2026
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
243 changes: 227 additions & 16 deletions keepercommander/commands/enterprise.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -132,6 +132,7 @@ def register_command_info(aliases, command_info):


enterprise_node_parser = argparse.ArgumentParser(prog='enterprise-node', description='Manage an enterprise node')
enterprise_node_parser.add_argument('-f', '--force', dest='force', action='store_true', help='do not prompt for confirmation')
enterprise_node_parser.add_argument('--wipe-out', dest='wipe_out', action='store_true', help='wipe out node content')
enterprise_node_parser.add_argument('--add', dest='add', action='store_true', help='create node')
enterprise_node_parser.add_argument('--parent', dest='parent', action='store', help='Parent Node Name or ID')
Expand Down Expand Up @@ -1285,7 +1286,7 @@ def traverse_to_root(node_id, depth):
if not node.get('parent_id'):
raise CommandError('enterprise-node', 'Cannot wipe out root node')

answer = user_choice(
answer = 'y' if kwargs.get('force') else user_choice(
bcolors.FAIL + bcolors.BOLD + '\nALERT!\n' + bcolors.ENDC +
'This action cannot be undone.\n\n' +
'Do you want to proceed with deletion?', 'yn', 'n')
Expand All @@ -1309,14 +1310,15 @@ def traverse_to_root(node_id, depth):
roles = [x for x in params.enterprise['roles'] if x['node_id'] in nodes]
role_set = set([x['role_id'] for x in managed_nodes])
role_set = role_set.union([x['role_id'] for x in roles])
for ru in params.enterprise['role_users']:
if ru['role_id'] in role_set:
rq = {
'command': 'role_user_remove',
'role_id': ru['role_id'],
'enterprise_user_id': ru['enterprise_user_id']
}
request_batch.append(rq)
if 'role_users' in params.enterprise:
for ru in params.enterprise['role_users']:
if ru['role_id'] in role_set:
rq = {
'command': 'role_user_remove',
'role_id': ru['role_id'],
'enterprise_user_id': ru['enterprise_user_id']
}
request_batch.append(rq)
for mn in managed_nodes:
rq = {
'command': 'role_managed_node_remove',
Expand Down Expand Up @@ -1356,6 +1358,12 @@ def traverse_to_root(node_id, depth):
'node_id': node_id
}
request_batch.append(rq)

# Check if there's anything to wipe out
if not request_batch:
node_name = node.get('data', {}).get('displayname') or str(node['node_id'])
logging.info('Node \'%s\' is empty. Nothing to wipe out.', node_name)
return
elif parent_id or kwargs.get('displayname'):
display_name = kwargs.get('displayname')
def is_in_chain(node_id, parent_id):
Expand Down Expand Up @@ -2157,17 +2165,62 @@ def execute(self, params, **kwargs):
raise CommandError('enterprise-user', 'No root nodes were detected. Specify --node parameter')
node_id = root_nodes[0]

# Collect role_ids for newly created roles
new_role_ids = []
for role_name in role_names:
data = json.dumps({ "displayname": role_name }).encode('utf-8')
role_id = self.get_enterprise_id(params)
new_role_ids.append(role_id)
rq = {
"command": "role_add",
"role_id": self.get_enterprise_id(params),
"role_id": role_id,
"node_id": node_id,
"encrypted_data": utils.base64_url_encode(crypto.encrypt_aes_v1(data, tree_key)),
"visible_below": (kwargs.get('visible_below') == 'on') or False,
"new_user_inherit": (kwargs.get('new_user') == 'on') or False
}
request_batch.append(rq)

if kwargs.get('add_admin') and new_role_ids:
skip_display = True
node_lookup = {}
if 'nodes' in params.enterprise:
for node in params.enterprise['nodes']:
node_lookup[str(node['node_id'])] = node
if node.get('parent_id'):
node_name = node['data'].get('displayname')
else:
node_name = params.enterprise['enterprise_name']
node_name = node_name.lower()
value = node_lookup.get(node_name)
if value is None:
value = node
elif type(value) == list:
value.append(node)
else:
value = [value, node]
node_lookup[node_name] = value

admin_nodes = {}
for admin_node_name in kwargs.get('add_admin'):
value = node_lookup.get(admin_node_name.lower())
if value is None:
logging.warning('Node %s could not be resolved', admin_node_name)
elif isinstance(value, dict):
admin_nodes[value['node_id']] = value['data'].get('displayname') or params.enterprise['enterprise_name']
elif isinstance(value, list):
logging.warning('Node name \'%s\' is not unique. Use Node ID. Skipping', admin_node_name)

for role_id in new_role_ids:
for admin_node_id, admin_node_display_name in admin_nodes.items():
rq = {
"command": "role_managed_node_add",
"role_id": role_id,
"managed_node_id": admin_node_id,
"cascade_node_management": (kwargs.get('cascade') == 'on') or False,
"tree_keys": []
}
request_batch.append(rq)
else:
for role_name in role_names:
logging.warning('Role %s is not found: Skipping', role_name)
Expand Down Expand Up @@ -3057,6 +3110,88 @@ class EnterpriseTeamCommand(EnterpriseCommand):
def get_parser(self):
return enterprise_team_parser

@staticmethod
def _resolve_users(params, user_list):
"""Resolve user names/IDs to user objects. Returns dict {user_id: user_node}"""
users = {}
for u in user_list:
uname = u.lower()
user_node = None
if 'users' in params.enterprise:
for user in params.enterprise['users']:
if uname in {str(user['enterprise_user_id']), user['username'].lower()}:
user_node = user
break
if user_node:
users[user_node['enterprise_user_id']] = user_node
else:
logging.warning('User %s could not be resolved', u)
return users

@staticmethod
def _create_add_user_request(params, team_uid, team_key, user, hsf_flag):
"""Create a request to add a user to a team with proper encryption"""
user_id = user['enterprise_user_id']
username = user['username']

api.load_user_public_keys(params, [username], False)
user_keys = params.key_cache.get(username)

if not user_keys:
logging.warning('Cannot get user %s public key', username)
return None

rq = {
'command': 'team_enterprise_user_add',
'team_uid': team_uid,
'enterprise_user_id': user_id,
'user_type': 2 if hsf_flag == 'on' else 1 if hsf_flag else 0,
}

if params.forbid_rsa:
if user_keys.ec:
ec_key = crypto.load_ec_public_key(user_keys.ec)
encrypted_team_key = crypto.encrypt_ec(team_key, ec_key)
rq['team_key'] = utils.base64_url_encode(encrypted_team_key)
rq['team_key_type'] = 'encrypted_by_public_key_ecc'
else:
logging.warning('User %s does not have EC key', username)
return None
else:
if user_keys.rsa:
rsa_key = crypto.load_rsa_public_key(user_keys.rsa)
encrypted_team_key = crypto.encrypt_rsa(team_key, rsa_key)
rq['team_key'] = utils.base64_url_encode(encrypted_team_key)
rq['team_key_type'] = 'encrypted_by_public_key'
else:
logging.warning('User %s does not have RSA key', username)
return None

return rq

@staticmethod
def _resolve_roles(params, role_list):
"""Resolve role names/IDs to role objects. Returns dict {role_id: role_name} excluding admin roles"""
role_changes = {}
for role in role_list:
role_node = next((
r for r in params.enterprise['roles']
if role in (str(r['role_id']), r['data'].get('displayname'))
), None)
if role_node:
# Check if role has administrative permissions
is_managed_role = any(
mn['role_id'] == role_node['role_id']
for mn in params.enterprise.get('managed_nodes', [])
)
if is_managed_role:
logging.warning('Teams cannot be assigned to roles with administrative permissions.')
else:
role_changes[role_node['role_id']] = role_node['data'].get('displayname')
else:
logging.warning('Role %s cannot be resolved', role)
return role_changes

def execute(self, params, **kwargs):
if (kwargs.get('add') or kwargs.get('approve')) and kwargs.get('remove'):
raise CommandError('enterprise-team', "'add'/'approve' and 'delete' commands are mutually exclusive.")
Expand Down Expand Up @@ -3110,7 +3245,8 @@ def execute(self, params, **kwargs):
matched_teams = list(matched.values())
request_batch = []
non_batch_update_msgs = []
has_warnings = False
has_warnings = False
new_team_roles = None

if kwargs.get('add') or kwargs.get('approve'):
queue = []
Expand All @@ -3133,6 +3269,8 @@ def execute(self, params, **kwargs):
raise CommandError('enterprise-user', 'No root nodes were detected. Specify --node parameter')
node_id = root_nodes[0]

new_teams = {} # {team_uid: (team_name, team_key, is_new)}

for item in queue:
is_new_team = type(item) == str
team_name = item if is_new_team else item['name']
Expand Down Expand Up @@ -3164,6 +3302,35 @@ def execute(self, params, **kwargs):
rq['private_key'] = utils.base64_url_encode(encrypted_rsa_private_key)

request_batch.append(rq)

if is_new_team:
new_teams[team_uid] = (team_name, team_key, True)

if kwargs.get('add_user') and new_teams:
skip_display = True
users = self._resolve_users(params, kwargs.get('add_user'))
if not users:
has_warnings = True

hsf = kwargs.get('hide_shared_folders') or ''
for team_uid, (team_name, team_key, is_new) in new_teams.items():
for user_id, user in users.items():
if user['status'] == 'active':
rq = self._create_add_user_request(params, team_uid, team_key, user, hsf)
if rq:
request_batch.append(rq)
else:
request_batch.append({
'command': 'team_queue_user',
'team_uid': team_uid,
'enterprise_user_id': user_id
})

# Role additions for new teams will be handled after team creation
new_team_roles = None
if kwargs.get('add_role') and new_teams:
skip_display = True
new_team_roles = (new_teams, kwargs.get('add_role'))
else:
for team_name in team_names:
logging.warning('\'%s\' team is not found: Skipping', team_name)
Expand Down Expand Up @@ -3280,11 +3447,31 @@ def execute(self, params, **kwargs):
'enterprise_user_id': user_id
}
else:
rq = {
'command': 'team_enterprise_user_remove',
'team_uid': team['team_uid'],
'enterprise_user_id': user_id
}
is_member = False
username = user['username']
team_name = team['name']

# Check in active team members
if 'team_users' in params.enterprise:
is_member = any(1 for t in params.enterprise['team_users']
if t['team_uid'] == team_uid and t['enterprise_user_id'] == user_id)

# Check in queued team members
if not is_member and 'queued_team_users' in params.enterprise:
for qtu in params.enterprise['queued_team_users']:
if qtu['team_uid'] == team_uid and user_id in qtu.get('users', []):
is_member = True
break

if is_member:
rq = {
'command': 'team_enterprise_user_remove',
'team_uid': team['team_uid'],
'enterprise_user_id': user_id
}
else:
logging.warning('User %s is not a member of team \'%s\'', username, team_name)
has_warnings = True
if rq:
request_batch.append(rq)
elif node_id or kwargs.get('name') or kwargs.get('restrict_edit') or kwargs.get('restrict_share') or kwargs.get('restrict_view'):
Expand Down Expand Up @@ -3333,11 +3520,35 @@ def execute(self, params, **kwargs):
else:
logging.warning('\'%s\' %s team failed to %s user %s: %s', team_name, 'queued' if command == 'team_queue_user' else '',
'delete' if command == 'team_enterprise_user_remove' else 'add', user_name, rs['message'])
elif command in {'role_team_add', 'role_team_remove'}:
role_id = rq.get('role_id')
role_name = next((r['data'].get('displayname') for r in params.enterprise.get('roles', [])
if r['role_id'] == role_id), str(role_id))
action = 'assign' if command == 'role_team_add' else 'remove'
if rs['result'] == 'success':
logging.info('\'%s\' role %sed to team \'%s\'', role_name, action, team_name)
else:
logging.warning('Failed to %s role \'%s\' to/from team \'%s\': %s', action, role_name, team_name, rs['message'])

if request_batch or len(non_batch_update_msgs) > 0:
for update_msg in non_batch_update_msgs:
logging.info(update_msg)
api.query_enterprise(params)

# Handle role additions for newly created teams (must be done after team exists)
if new_team_roles:
new_teams_dict, role_list = new_team_roles
# Fetch updated team data to get proper team objects
created_teams = []
for team_uid in new_teams_dict.keys():
team_data = next((t for t in params.enterprise.get('teams', []) if t['team_uid'] == team_uid), None)
if team_data:
created_teams.append(team_data)

if created_teams:
role_msgs = self.change_team_roles(params, created_teams, role_list, None)
for msg in role_msgs:
logging.info(msg)
elif not has_warnings:
for team in matched_teams:
print('\n')
Expand Down
14 changes: 11 additions & 3 deletions keepercommander/service/util/command_util.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -148,9 +148,17 @@ def execute(cls, command: str) -> Tuple[Any, int]:
# Always let the parser handle the response (including empty responses and logs)
response = parse_keeper_response(command, response, log_output)

status_code = 200
if isinstance(response, dict) and response.get("status") == "error":
status_code = 400
if isinstance(response, dict):
if 'status_code' in response:
status_code = response['status_code']
elif response.get("status") == "error":
status_code = 400
elif response.get("status") == "warning":
status_code = 400
else:
status_code = 200
else:
status_code = 200

response = CommandExecutor.encrypt_response(response)
logger.debug(f"Command executed successfully")
Expand Down
Loading