# Copyright (C) 2011-2024 RhodeCode GmbH
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License, version 3
# (only), as published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see .
#
# This program is dual-licensed. If you wish to learn more about the
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/
import logging
from rhodecode.api import (
jsonrpc_method, JSONRPCError, JSONRPCForbidden, JSONRPCValidationError)
from rhodecode.api.utils import (
Optional, OAttr, store_update, has_superadmin_permission, get_origin,
get_user_or_error, get_user_group_or_error, get_perm_or_error)
from rhodecode.lib import audit_logger
from rhodecode.lib.auth import HasUserGroupPermissionAnyApi, HasPermissionAnyApi
from rhodecode.lib.exceptions import UserGroupAssignedException
from rhodecode.model.db import Session
from rhodecode.model.permission import PermissionModel
from rhodecode.model.scm import UserGroupList
from rhodecode.model.user_group import UserGroupModel
from rhodecode.model import validation_schema
from rhodecode.model.validation_schema.schemas import user_group_schema
log = logging.getLogger(__name__)
@jsonrpc_method()
def get_user_group(request, apiuser, usergroupid):
"""
Returns the data of an existing user group.
This command can only be run using an |authtoken| with admin rights to
the specified repository.
:param apiuser: This is filled automatically from the |authtoken|.
:type apiuser: AuthUser
:param usergroupid: Set the user group from which to return data.
:type usergroupid: str or int
Example error output:
.. code-block:: bash
{
"error": null,
"id": ,
"result": {
"active": true,
"group_description": "group description",
"group_name": "group name",
"permissions": [
{
"name": "owner-name",
"origin": "owner",
"permission": "usergroup.admin",
"type": "user"
},
{
{
"name": "user name",
"origin": "permission",
"permission": "usergroup.admin",
"type": "user"
},
{
"name": "user group name",
"origin": "permission",
"permission": "usergroup.write",
"type": "user_group"
}
],
"permissions_summary": {
"repositories": {
"aa-root-level-repo-1": "repository.admin"
},
"repositories_groups": {}
},
"owner": "owner name",
"users": [],
"users_group_id": 2
}
}
"""
user_group = get_user_group_or_error(usergroupid)
if not has_superadmin_permission(apiuser):
# check if we have at least read permission for this user group !
_perms = ('usergroup.read', 'usergroup.write', 'usergroup.admin',)
if not HasUserGroupPermissionAnyApi(*_perms)(
user=apiuser, user_group_name=user_group.users_group_name):
raise JSONRPCError('user group `{}` does not exist'.format(
usergroupid))
permissions = []
for _user in user_group.permissions():
user_data = {
'name': _user.username,
'permission': _user.permission,
'origin': get_origin(_user),
'type': "user",
}
permissions.append(user_data)
for _user_group in user_group.permission_user_groups():
user_group_data = {
'name': _user_group.users_group_name,
'permission': _user_group.permission,
'origin': get_origin(_user_group),
'type': "user_group",
}
permissions.append(user_group_data)
data = user_group.get_api_data()
data["permissions"] = permissions
data["permissions_summary"] = UserGroupModel().get_perms_summary(
user_group.users_group_id)
return data
@jsonrpc_method()
def get_user_groups(request, apiuser):
"""
Lists all the existing user groups within RhodeCode.
This command can only be run using an |authtoken| with admin rights to
the specified repository.
This command takes the following options:
:param apiuser: This is filled automatically from the |authtoken|.
:type apiuser: AuthUser
Example error output:
.. code-block:: bash
id :
result : [,...]
error : null
"""
include_secrets = has_superadmin_permission(apiuser)
result = []
_perms = ('usergroup.read', 'usergroup.write', 'usergroup.admin',)
extras = {'user': apiuser}
for user_group in UserGroupList(UserGroupModel().get_all(),
perm_set=_perms, extra_kwargs=extras):
result.append(
user_group.get_api_data(include_secrets=include_secrets))
return result
@jsonrpc_method()
def create_user_group(
request, apiuser, group_name, description=Optional(''),
owner=Optional(OAttr('apiuser')), active=Optional(True),
sync=Optional(None)):
"""
Creates a new user group.
This command can only be run using an |authtoken| with admin rights to
the specified repository.
This command takes the following options:
:param apiuser: This is filled automatically from the |authtoken|.
:type apiuser: AuthUser
:param group_name: Set the name of the new user group.
:type group_name: str
:param description: Give a description of the new user group.
:type description: str
:param owner: Set the owner of the new user group.
If not set, the owner is the |authtoken| user.
:type owner: Optional(str or int)
:param active: Set this group as active.
:type active: Optional(``True`` | ``False``)
:param sync: Set enabled or disabled the automatically sync from
external authentication types like ldap. If User Group will be named like
one from e.g ldap and sync flag is enabled members will be synced automatically.
Sync type when enabled via API is set to `manual_api`
:type sync: Optional(``True`` | ``False``)
Example output:
.. code-block:: bash
id :
result: {
"msg": "created new user group ``",
"user_group":
}
error: null
Example error output:
.. code-block:: bash
id :
result : null
error : {
"user group `` already exist"
or
"failed to create group ``"
}
"""
if not has_superadmin_permission(apiuser):
if not HasPermissionAnyApi('hg.usergroup.create.true')(user=apiuser):
raise JSONRPCForbidden()
if UserGroupModel().get_by_name(group_name):
raise JSONRPCError(f"user group `{group_name}` already exist")
if isinstance(owner, Optional):
owner = apiuser.user_id
owner = get_user_or_error(owner)
active = Optional.extract(active)
description = Optional.extract(description)
sync = Optional.extract(sync)
# set the sync option based on group_data
group_data = None
if sync:
group_data = {
'extern_type': 'manual_api',
'extern_type_set_by': apiuser.username
}
schema = user_group_schema.UserGroupSchema().bind(
# user caller
user=apiuser)
try:
schema_data = schema.deserialize(dict(
user_group_name=group_name,
user_group_description=description,
user_group_owner=owner.username,
user_group_active=active,
))
except validation_schema.Invalid as err:
raise JSONRPCValidationError(colander_exc=err)
try:
user_group = UserGroupModel().create(
name=schema_data['user_group_name'],
description=schema_data['user_group_description'],
owner=owner,
active=schema_data['user_group_active'], group_data=group_data)
Session().flush()
creation_data = user_group.get_api_data()
audit_logger.store_api(
'user_group.create', action_data={'data': creation_data},
user=apiuser)
Session().commit()
affected_user_ids = [apiuser.user_id, owner.user_id]
PermissionModel().trigger_permission_flush(affected_user_ids)
return {
'msg': 'created new user group `%s`' % group_name,
'user_group': creation_data
}
except Exception:
log.exception("Error occurred during creation of user group")
raise JSONRPCError(f'failed to create group `{group_name}`')
@jsonrpc_method()
def update_user_group(request, apiuser, usergroupid, group_name=Optional(''),
description=Optional(''), owner=Optional(None),
active=Optional(True), sync=Optional(None)):
"""
Updates the specified `user group` with the details provided.
This command can only be run using an |authtoken| with admin rights to
the specified repository.
:param apiuser: This is filled automatically from the |authtoken|.
:type apiuser: AuthUser
:param usergroupid: Set the id of the `user group` to update.
:type usergroupid: str or int
:param group_name: Set the new name the `user group`
:type group_name: str
:param description: Give a description for the `user group`
:type description: str
:param owner: Set the owner of the `user group`.
:type owner: Optional(str or int)
:param active: Set the group as active.
:type active: Optional(``True`` | ``False``)
:param sync: Set enabled or disabled the automatically sync from
external authentication types like ldap. If User Group will be named like
one from e.g ldap and sync flag is enabled members will be synced automatically.
Sync type when enabled via API is set to `manual_api`
:type sync: Optional(``True`` | ``False``)
Example output:
.. code-block:: bash
id :
result : {
"msg": 'updated user group ID: ',
"user_group":
}
error : null
Example error output:
.. code-block:: bash
id :
result : null
error : {
"failed to update user group ``"
}
"""
user_group = get_user_group_or_error(usergroupid)
include_secrets = False
if not has_superadmin_permission(apiuser):
# check if we have admin permission for this user group !
_perms = ('usergroup.admin',)
if not HasUserGroupPermissionAnyApi(*_perms)(
user=apiuser, user_group_name=user_group.users_group_name):
raise JSONRPCError(
f'user group `{usergroupid}` does not exist')
else:
include_secrets = True
if not isinstance(owner, Optional):
owner = get_user_or_error(owner)
old_data = user_group.get_api_data()
updates = {}
store_update(updates, group_name, 'users_group_name')
store_update(updates, description, 'user_group_description')
store_update(updates, owner, 'user')
store_update(updates, active, 'users_group_active')
sync = Optional.extract(sync)
group_data = None
if sync is True:
group_data = {
'extern_type': 'manual_api',
'extern_type_set_by': apiuser.username
}
if sync is False:
group_data = user_group.group_data
if group_data and "extern_type" in group_data:
del group_data["extern_type"]
try:
UserGroupModel().update(user_group, updates, group_data=group_data)
audit_logger.store_api(
'user_group.edit', action_data={'old_data': old_data},
user=apiuser)
Session().commit()
return {
'msg': 'updated user group ID:{} {}'.format(
user_group.users_group_id, user_group.users_group_name),
'user_group': user_group.get_api_data(
include_secrets=include_secrets)
}
except Exception:
log.exception("Error occurred during update of user group")
raise JSONRPCError(
f'failed to update user group `{usergroupid}`')
@jsonrpc_method()
def delete_user_group(request, apiuser, usergroupid):
"""
Deletes the specified `user group`.
This command can only be run using an |authtoken| with admin rights to
the specified repository.
This command takes the following options:
:param apiuser: filled automatically from apikey
:type apiuser: AuthUser
:param usergroupid:
:type usergroupid: int
Example output:
.. code-block:: bash
id :
result : {
"msg": "deleted user group ID: "
}
error : null
Example error output:
.. code-block:: bash
id :
result : null
error : {
"failed to delete user group ID: "
or
"RepoGroup assigned to "
}
"""
user_group = get_user_group_or_error(usergroupid)
if not has_superadmin_permission(apiuser):
# check if we have admin permission for this user group !
_perms = ('usergroup.admin',)
if not HasUserGroupPermissionAnyApi(*_perms)(
user=apiuser, user_group_name=user_group.users_group_name):
raise JSONRPCError(
f'user group `{usergroupid}` does not exist')
old_data = user_group.get_api_data()
try:
UserGroupModel().delete(user_group)
audit_logger.store_api(
'user_group.delete', action_data={'old_data': old_data},
user=apiuser)
Session().commit()
return {
'msg': 'deleted user group ID:{} {}'.format(
user_group.users_group_id, user_group.users_group_name),
'user_group': None
}
except UserGroupAssignedException as e:
log.exception("UserGroupAssigned error")
raise JSONRPCError(str(e))
except Exception:
log.exception("Error occurred during deletion of user group")
raise JSONRPCError(
'failed to delete user group ID:%s %s' %(
user_group.users_group_id, user_group.users_group_name))
@jsonrpc_method()
def add_user_to_user_group(request, apiuser, usergroupid, userid):
"""
Adds a user to a `user group`. If the user already exists in the group
this command will return false.
This command can only be run using an |authtoken| with admin rights to
the specified user group.
This command takes the following options:
:param apiuser: This is filled automatically from the |authtoken|.
:type apiuser: AuthUser
:param usergroupid: Set the name of the `user group` to which a
user will be added.
:type usergroupid: int
:param userid: Set the `user_id` of the user to add to the group.
:type userid: int
Example output:
.. code-block:: bash
id :
result : {
"success": True|False # depends on if member is in group
"msg": "added member `` to user group `` |
User is already in that group"
}
error : null
Example error output:
.. code-block:: bash
id :
result : null
error : {
"failed to add member to user group ``"
}
"""
user = get_user_or_error(userid)
user_group = get_user_group_or_error(usergroupid)
if not has_superadmin_permission(apiuser):
# check if we have admin permission for this user group !
_perms = ('usergroup.admin',)
if not HasUserGroupPermissionAnyApi(*_perms)(
user=apiuser, user_group_name=user_group.users_group_name):
raise JSONRPCError('user group `{}` does not exist'.format(
usergroupid))
old_values = user_group.get_api_data()
try:
ugm = UserGroupModel().add_user_to_group(user_group, user)
success = True if ugm is not True else False
msg = 'added member `{}` to user group `{}`'.format(
user.username, user_group.users_group_name
)
msg = msg if success else 'User is already in that group'
if success:
user_data = user.get_api_data()
audit_logger.store_api(
'user_group.edit.member.add',
action_data={'user': user_data, 'old_data': old_values},
user=apiuser)
Session().commit()
return {
'success': success,
'msg': msg
}
except Exception:
log.exception("Error occurred during adding a member to user group")
raise JSONRPCError(
'failed to add member to user group `{}`'.format(
user_group.users_group_name,
)
)
@jsonrpc_method()
def remove_user_from_user_group(request, apiuser, usergroupid, userid):
"""
Removes a user from a user group.
* If the specified user is not in the group, this command will return
`false`.
This command can only be run using an |authtoken| with admin rights to
the specified user group.
:param apiuser: This is filled automatically from the |authtoken|.
:type apiuser: AuthUser
:param usergroupid: Sets the user group name.
:type usergroupid: str or int
:param userid: The user you wish to remove from |RCE|.
:type userid: str or int
Example output:
.. code-block:: bash
id :
result: {
"success": True|False, # depends on if member is in group
"msg": "removed member from user group |
User wasn't in group"
}
error: null
"""
user = get_user_or_error(userid)
user_group = get_user_group_or_error(usergroupid)
if not has_superadmin_permission(apiuser):
# check if we have admin permission for this user group !
_perms = ('usergroup.admin',)
if not HasUserGroupPermissionAnyApi(*_perms)(
user=apiuser, user_group_name=user_group.users_group_name):
raise JSONRPCError(
f'user group `{usergroupid}` does not exist')
old_values = user_group.get_api_data()
try:
success = UserGroupModel().remove_user_from_group(user_group, user)
msg = 'removed member `{}` from user group `{}`'.format(
user.username, user_group.users_group_name
)
msg = msg if success else "User wasn't in group"
if success:
user_data = user.get_api_data()
audit_logger.store_api(
'user_group.edit.member.delete',
action_data={'user': user_data, 'old_data': old_values},
user=apiuser)
Session().commit()
return {'success': success, 'msg': msg}
except Exception:
log.exception("Error occurred during removing an member from user group")
raise JSONRPCError(
'failed to remove member from user group `{}`'.format(
user_group.users_group_name,
)
)
@jsonrpc_method()
def grant_user_permission_to_user_group(
request, apiuser, usergroupid, userid, perm):
"""
Set permissions for a user in a user group.
:param apiuser: This is filled automatically from the |authtoken|.
:type apiuser: AuthUser
:param usergroupid: Set the user group to edit permissions on.
:type usergroupid: str or int
:param userid: Set the user from whom you wish to set permissions.
:type userid: str
:param perm: (usergroup.(none|read|write|admin))
:type perm: str
Example output:
.. code-block:: bash
id :
result : {
"msg": "Granted perm: `` for user: `` in user group: ``",
"success": true
}
error : null
"""
user_group = get_user_group_or_error(usergroupid)
if not has_superadmin_permission(apiuser):
# check if we have admin permission for this user group !
_perms = ('usergroup.admin',)
if not HasUserGroupPermissionAnyApi(*_perms)(
user=apiuser, user_group_name=user_group.users_group_name):
raise JSONRPCError(
f'user group `{usergroupid}` does not exist')
user = get_user_or_error(userid)
perm = get_perm_or_error(perm, prefix='usergroup.')
try:
changes = UserGroupModel().grant_user_permission(
user_group=user_group, user=user, perm=perm)
action_data = {
'added': changes['added'],
'updated': changes['updated'],
'deleted': changes['deleted'],
}
audit_logger.store_api(
'user_group.edit.permissions', action_data=action_data,
user=apiuser)
Session().commit()
PermissionModel().flush_user_permission_caches(changes)
return {
'msg':
'Granted perm: `{}` for user: `{}` in user group: `{}`'.format(
perm.permission_name, user.username,
user_group.users_group_name
),
'success': True
}
except Exception:
log.exception("Error occurred during editing permissions "
"for user in user group")
raise JSONRPCError(
'failed to edit permission for user: '
'`%s` in user group: `%s`' % (
userid, user_group.users_group_name))
@jsonrpc_method()
def revoke_user_permission_from_user_group(
request, apiuser, usergroupid, userid):
"""
Revoke a users permissions in a user group.
:param apiuser: This is filled automatically from the |authtoken|.
:type apiuser: AuthUser
:param usergroupid: Set the user group from which to revoke the user
permissions.
:type: usergroupid: str or int
:param userid: Set the userid of the user whose permissions will be
revoked.
:type userid: str
Example output:
.. code-block:: bash
id :
result : {
"msg": "Revoked perm for user: `` in user group: ``",
"success": true
}
error : null
"""
user_group = get_user_group_or_error(usergroupid)
if not has_superadmin_permission(apiuser):
# check if we have admin permission for this user group !
_perms = ('usergroup.admin',)
if not HasUserGroupPermissionAnyApi(*_perms)(
user=apiuser, user_group_name=user_group.users_group_name):
raise JSONRPCError(
f'user group `{usergroupid}` does not exist')
user = get_user_or_error(userid)
try:
changes = UserGroupModel().revoke_user_permission(
user_group=user_group, user=user)
action_data = {
'added': changes['added'],
'updated': changes['updated'],
'deleted': changes['deleted'],
}
audit_logger.store_api(
'user_group.edit.permissions', action_data=action_data,
user=apiuser)
Session().commit()
PermissionModel().flush_user_permission_caches(changes)
return {
'msg': 'Revoked perm for user: `{}` in user group: `{}`'.format(
user.username, user_group.users_group_name
),
'success': True
}
except Exception:
log.exception("Error occurred during editing permissions "
"for user in user group")
raise JSONRPCError(
'failed to edit permission for user: `%s` in user group: `%s`'
% (userid, user_group.users_group_name))
@jsonrpc_method()
def grant_user_group_permission_to_user_group(
request, apiuser, usergroupid, sourceusergroupid, perm):
"""
Give one user group permissions to another user group.
:param apiuser: This is filled automatically from the |authtoken|.
:type apiuser: AuthUser
:param usergroupid: Set the user group on which to edit permissions.
:type usergroupid: str or int
:param sourceusergroupid: Set the source user group to which
access/permissions will be granted.
:type sourceusergroupid: str or int
:param perm: (usergroup.(none|read|write|admin))
:type perm: str
Example output:
.. code-block:: bash
id :
result : {
"msg": "Granted perm: `` for user group: `` in user group: ``",
"success": true
}
error : null
"""
user_group = get_user_group_or_error(sourceusergroupid)
target_user_group = get_user_group_or_error(usergroupid)
perm = get_perm_or_error(perm, prefix='usergroup.')
if not has_superadmin_permission(apiuser):
# check if we have admin permission for this user group !
_perms = ('usergroup.admin',)
if not HasUserGroupPermissionAnyApi(*_perms)(
user=apiuser,
user_group_name=target_user_group.users_group_name):
raise JSONRPCError(
f'to user group `{usergroupid}` does not exist')
# check if we have at least read permission for source user group !
_perms = ('usergroup.read', 'usergroup.write', 'usergroup.admin',)
if not HasUserGroupPermissionAnyApi(*_perms)(
user=apiuser, user_group_name=user_group.users_group_name):
raise JSONRPCError(
f'user group `{sourceusergroupid}` does not exist')
try:
changes = UserGroupModel().grant_user_group_permission(
target_user_group=target_user_group,
user_group=user_group, perm=perm)
action_data = {
'added': changes['added'],
'updated': changes['updated'],
'deleted': changes['deleted'],
}
audit_logger.store_api(
'user_group.edit.permissions', action_data=action_data,
user=apiuser)
Session().commit()
PermissionModel().flush_user_permission_caches(changes)
return {
'msg': 'Granted perm: `%s` for user group: `%s` '
'in user group: `%s`' % (
perm.permission_name, user_group.users_group_name,
target_user_group.users_group_name
),
'success': True
}
except Exception:
log.exception("Error occurred during editing permissions "
"for user group in user group")
raise JSONRPCError(
'failed to edit permission for user group: `%s` in '
'user group: `%s`' % (
sourceusergroupid, target_user_group.users_group_name
)
)
@jsonrpc_method()
def revoke_user_group_permission_from_user_group(
request, apiuser, usergroupid, sourceusergroupid):
"""
Revoke the permissions that one user group has to another.
:param apiuser: This is filled automatically from the |authtoken|.
:type apiuser: AuthUser
:param usergroupid: Set the user group on which to edit permissions.
:type usergroupid: str or int
:param sourceusergroupid: Set the user group from which permissions
are revoked.
:type sourceusergroupid: str or int
Example output:
.. code-block:: bash
id :
result : {
"msg": "Revoked perm for user group: `` in user group: ``",
"success": true
}
error : null
"""
user_group = get_user_group_or_error(sourceusergroupid)
target_user_group = get_user_group_or_error(usergroupid)
if not has_superadmin_permission(apiuser):
# check if we have admin permission for this user group !
_perms = ('usergroup.admin',)
if not HasUserGroupPermissionAnyApi(*_perms)(
user=apiuser,
user_group_name=target_user_group.users_group_name):
raise JSONRPCError(
f'to user group `{usergroupid}` does not exist')
# check if we have at least read permission
# for the source user group !
_perms = ('usergroup.read', 'usergroup.write', 'usergroup.admin',)
if not HasUserGroupPermissionAnyApi(*_perms)(
user=apiuser, user_group_name=user_group.users_group_name):
raise JSONRPCError(
f'user group `{sourceusergroupid}` does not exist')
try:
changes = UserGroupModel().revoke_user_group_permission(
target_user_group=target_user_group, user_group=user_group)
action_data = {
'added': changes['added'],
'updated': changes['updated'],
'deleted': changes['deleted'],
}
audit_logger.store_api(
'user_group.edit.permissions', action_data=action_data,
user=apiuser)
Session().commit()
PermissionModel().flush_user_permission_caches(changes)
return {
'msg': 'Revoked perm for user group: '
'`%s` in user group: `%s`' % (
user_group.users_group_name,
target_user_group.users_group_name
),
'success': True
}
except Exception:
log.exception("Error occurred during editing permissions "
"for user group in user group")
raise JSONRPCError(
'failed to edit permission for user group: '
'`%s` in user group: `%s`' % (
sourceusergroupid, target_user_group.users_group_name
)
)