From 67f388ca11e4eb17fb7841630c0b82232932f094 Mon Sep 17 00:00:00 2001 From: alisboav Date: Nov 30 2023 14:45:28 +0000 Subject: [PATCH 1/2] Add command to list users with a particular permission --- diff --git a/cli/koji_cli/commands.py b/cli/koji_cli/commands.py index 69b262d..7785679 100644 --- a/cli/koji_cli/commands.py +++ b/cli/koji_cli/commands.py @@ -2154,6 +2154,29 @@ def anon_handle_list_volumes(goptions, session, args): for volinfo in session.listVolumes(): print(volinfo['name']) +def handle_list_users(goptions, session, args): + "[info] List of users with given a permission" + usage = "usage: %prog list-users [options]" + parser = OptionParser(usage=get_usage_str(usage)) + parser.add_option("--perm", help="List users that have a given permission") + (options, args) = parser.parse_args(args) + if len(args) > 0: + parser.error("This command takes no arguments") + activate_session(session, goptions) + if not session.hasPerm('admin'): + parser.error("This action requires admin privileges") + users = [] + if options.perm: + users_list = session.getPermsUser(options.perm) + if users_list: + for p in users_list: + users.append({'name': p}) + else: + parser.error("No such permission: %s" % options.perm) + else: + parser.error("Please provide a permission with --perm") + for user in users: + print(user['name']) def handle_list_permissions(goptions, session, args): "[info] List user permissions" diff --git a/kojihub/auth.py b/kojihub/auth.py index dc52b81..af5dff6 100644 --- a/kojihub/auth.py +++ b/kojihub/auth.py @@ -790,6 +790,13 @@ def get_user_perms(user_id): result = query.execute() return [r['name'] for r in result] +def get_users_with_perm(perm_id): + query = QueryProcessor(tables=['user_perms'], columns=['name'], + clauses=['active = TRUE', 'perm_id=%(perm_id)s'], + joins=['users ON user_id = users.id'], + values={'perm_id': perm_id}) + result = query.execute() + return [r['name'] for r in result] def get_user_data(user_id): query = QueryProcessor(tables=['users'], columns=['name', 'status', 'usertype'], diff --git a/kojihub/kojihub.py b/kojihub/kojihub.py index 775cb5d..8ab9c37 100644 --- a/kojihub/kojihub.py +++ b/kojihub/kojihub.py @@ -74,7 +74,7 @@ from koji.util import ( multi_fnmatch, safer_move, ) -from .auth import get_user_perms, get_user_groups +from .auth import get_user_perms, get_user_groups, get_users_with_perm from .db import ( # noqa: F401 BulkInsertProcessor, DeleteProcessor, @@ -13390,6 +13390,13 @@ class RootExports(object): user_info = get_user(userID, strict=True) return get_user_perms(user_info['id']) + def getPermsUser(self, perm=None): + """Get a list of users with the given permission. + Options: + - perm: Permission name.""" + perm_id = get_perm_id(perm) + return get_users_with_perm(perm_id) + def getAllPerms(self): """Get a list of all permissions in the system. Returns a list of maps. Each map contains the following keys: diff --git a/tests/test_cli/data/list-commands.txt b/tests/test_cli/data/list-commands.txt index c434b68..6b72fd9 100644 --- a/tests/test_cli/data/list-commands.txt +++ b/tests/test_cli/data/list-commands.txt @@ -115,6 +115,7 @@ info commands: list-targets List the build targets list-tasks Print the list of tasks list-untagged List untagged builds + list-users List users with a given permission list-volumes List storage volumes mock-config Create a mock config repoinfo Print basic information about a repo diff --git a/tests/test_cli/test_list_users.py b/tests/test_cli/test_list_users.py new file mode 100644 index 0000000..b1f205b --- /dev/null +++ b/tests/test_cli/test_list_users.py @@ -0,0 +1,120 @@ +from __future__ import absolute_import +import mock +import six +import unittest + +from koji_cli.commands import handle_list_users +from . import utils + + +class TestListUsers(utils.CliTestCase): + + def setUp(self): + self.error_format = """Usage: %s list-users [options] +(Specify the --help global option for a list of other help options) + +%s: error: {message} +""" % (self.progname, self.progname) + + self.session = mock.MagicMock() + self.activate_session_mock = mock.patch('koji_cli.commands.activate_session').start() + self.options = mock.MagicMock() + self.options.quiet = True + self.users = [{'id': 101, 'name': "user01"}, + {'id': 102, 'name': "user02"}, + {'id': 103, 'name': "user03"}, + {'id': 104, 'name': "user04"}, + {'id': 105, 'name': "user05"} + ] + self.perm='admin' + + + def test_handle_list_users_arg_error(self): + """Test handle_list_users argument error (no argument is required)""" + expected = self.format_error_message("This command takes no arguments") + self.assert_system_exit( + handle_list_users, + self.options, + self.session, + ['arg-1', 'arg-2'], + stderr=expected, + activate_session=None, + exit_code=2 + ) + self.activate_session_mock.assert_not_called() + self.session.getPermsUser.assert_not_called() + + + def test_handle_list_users_notadmin_error(self): + """Test handle_list_users using a non admin user""" + expected = self.format_error_message("This action requires admin privileges") + self.session.hasPerm.return_value = False + self.assert_system_exit( + handle_list_users, + self.options, + self.session, + ['--perm', self.perm], + stderr=expected, + activate_session=None, + exit_code=2 + ) + self.activate_session_mock.assert_called_once() + self.session.getPermsUser.assert_not_called() + + def test_handle_list_users_perm_not_exist(self): + """Test handle_list_users when perm does not exist""" + self.session.getPermsUser.return_value = [] + expected = self.format_error_message("No such permission: notperm") + self.assert_system_exit( + handle_list_users, + self.options, + self.session, + ['--perm', 'notperm'], + stderr=expected, + activate_session=None, + exit_code=2 + ) + self.activate_session_mock.assert_called_once() + self.session.getPermsUser.assert_called_once() + + @mock.patch('sys.stdout', new_callable=six.StringIO) + def test_handle_list_users_no_perm(self, stdout): + """Test handle_list_users when --perm is not setting""" + expected = self.format_error_message("""Please provide a permission with --perm""") + self.assert_system_exit( + handle_list_users, + self.options, + self.session, + ['--'], + stderr=expected, + activate_session=None, + exit_code=2 + ) + self.activate_session_mock.assert_called_once() + self.session.getPermsUser.assert_not_called() + + @mock.patch('sys.stdout', new_callable=six.StringIO) + def test_handle_list_users_perm(self, stdout): + """Test handle_list_users user permissions""" + expected = "user01\nuser02\nuser03\nuser04\nuser05\n" + users = [p['name'] for p in self.users[::1]] + self.session.getPermsUser.return_value = users + handle_list_users(self.options, self.session, ['--perm', self.perm]) + self.assert_console_message(stdout, expected) + self.activate_session_mock.assert_called_once() + self.session.getPermsUser.assert_called_once() + + def test_handle_list_users_help(self): + self.assert_help( + handle_list_users, + """Usage: %s list-users [options] +(Specify the --help global option for a list of other help options) + +Options: + -h, --help show this help message and exit + --perm=PERM List users that have a given permission +""" % self.progname) + + +if __name__ == '__main__': + unittest.main() From fd9b765f0d4757f9f1ff9f794f53b53c5cd8af14 Mon Sep 17 00:00:00 2001 From: Marta Vila Fernandes Date: Dec 05 2023 08:37:13 +0000 Subject: [PATCH 2/2] make the admin check in kojihub --- diff --git a/cli/koji_cli/commands.py b/cli/koji_cli/commands.py index 7785679..7b41ff6 100644 --- a/cli/koji_cli/commands.py +++ b/cli/koji_cli/commands.py @@ -2163,8 +2163,6 @@ def handle_list_users(goptions, session, args): if len(args) > 0: parser.error("This command takes no arguments") activate_session(session, goptions) - if not session.hasPerm('admin'): - parser.error("This action requires admin privileges") users = [] if options.perm: users_list = session.getPermsUser(options.perm) diff --git a/kojihub/kojihub.py b/kojihub/kojihub.py index 8ab9c37..3aa2997 100644 --- a/kojihub/kojihub.py +++ b/kojihub/kojihub.py @@ -13394,7 +13394,11 @@ class RootExports(object): """Get a list of users with the given permission. Options: - perm: Permission name.""" + if not context.session.hasPerm('admin'): + raise koji.ActionNotAllowed("This action requires admin privileges") perm_id = get_perm_id(perm) + if not perm_id: + raise koji.GenericError('Permission %s not available' % perm) return get_users_with_perm(perm_id) def getAllPerms(self): diff --git a/tests/test_cli/test_list_users.py b/tests/test_cli/test_list_users.py index b1f205b..e3cf04b 100644 --- a/tests/test_cli/test_list_users.py +++ b/tests/test_cli/test_list_users.py @@ -45,22 +45,6 @@ class TestListUsers(utils.CliTestCase): self.session.getPermsUser.assert_not_called() - def test_handle_list_users_notadmin_error(self): - """Test handle_list_users using a non admin user""" - expected = self.format_error_message("This action requires admin privileges") - self.session.hasPerm.return_value = False - self.assert_system_exit( - handle_list_users, - self.options, - self.session, - ['--perm', self.perm], - stderr=expected, - activate_session=None, - exit_code=2 - ) - self.activate_session_mock.assert_called_once() - self.session.getPermsUser.assert_not_called() - def test_handle_list_users_perm_not_exist(self): """Test handle_list_users when perm does not exist""" self.session.getPermsUser.return_value = [] @@ -85,7 +69,7 @@ class TestListUsers(utils.CliTestCase): handle_list_users, self.options, self.session, - ['--'], + [], stderr=expected, activate_session=None, exit_code=2 diff --git a/tests/test_hub/test_get_users_with_perm.py b/tests/test_hub/test_get_users_with_perm.py new file mode 100644 index 0000000..627dfef --- /dev/null +++ b/tests/test_hub/test_get_users_with_perm.py @@ -0,0 +1,32 @@ +import mock +import unittest +import koji +import kojihub + + +class TestGetPermsUser(unittest.TestCase): + def setUp(self): + self.context = mock.patch('kojihub.kojihub.context').start() + self.get_perm_id = mock.patch('kojihub.kojihub.get_perm_id').start() + self.get_users_with_perm = mock.patch('kojihub.kojihub.get_users_with_perm').start() + + def tearDown(self): + mock.patch.stopall() + + def test_no_admin(self): + self.context.session.hasPerm.return_value = False + with self.assertRaises(koji.ActionNotAllowed) as ex: + kojihub.RootExports().getPermsUser('admin') + self.assertEqual("This action requires admin privileges", str(ex.exception)) + self.get_users_with_perm.assert_not_called() + + def test_no_perm(self): + self.get_perm_id.return_value = None + with self.assertRaises(koji.GenericError): + kojihub.RootExports().getPermsUser('noperm') + self.get_users_with_perm.assert_not_called() + + def test_normal(self): + self.get_perm_id.return_value = 1 + kojihub.RootExports().getPermsUser('admin') + self.get_users_with_perm.assert_called_once_with(1)