diff options
Diffstat (limited to 'openstackclient')
5 files changed, 711 insertions, 0 deletions
diff --git a/openstackclient/identity/common.py b/openstackclient/identity/common.py index e119f660..f36f5f73 100644 --- a/openstackclient/identity/common.py +++ b/openstackclient/identity/common.py @@ -101,6 +101,13 @@ def _get_token_resource(client, resource, parsed_name, parsed_domain=None): # user/project under different domain may has a same name if parsed_domain and parsed_domain not in obj['domain'].values(): return parsed_name + if isinstance(obj, list): + for item in obj: + if item['name'] == parsed_name: + return item['id'] + if item['id'] == parsed_name: + return parsed_name + return parsed_name return obj['id'] if obj['name'] == parsed_name else parsed_name # diaper defense in case parsing the token fails except Exception: # noqa diff --git a/openstackclient/identity/v3/application_credential.py b/openstackclient/identity/v3/application_credential.py new file mode 100644 index 00000000..747fa20e --- /dev/null +++ b/openstackclient/identity/v3/application_credential.py @@ -0,0 +1,220 @@ +# Copyright 2018 SUSE Linux GmbH +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# + +"""Identity v3 Application Credential action implementations""" + +import datetime +import logging + +from osc_lib.command import command +from osc_lib import exceptions +from osc_lib import utils +import six + +from openstackclient.i18n import _ +from openstackclient.identity import common + + +LOG = logging.getLogger(__name__) + + +class CreateApplicationCredential(command.ShowOne): + _description = _("Create new application credential") + + def get_parser(self, prog_name): + parser = super(CreateApplicationCredential, self).get_parser(prog_name) + parser.add_argument( + 'name', + metavar='<name>', + help=_('Name of the application credential'), + ) + parser.add_argument( + '--secret', + metavar='<secret>', + help=_('Secret to use for authentication (if not provided, one' + ' will be generated)'), + ) + parser.add_argument( + '--role', + metavar='<role>', + action='append', + default=[], + help=_('Roles to authorize (name or ID) (repeat option to set' + ' multiple values)'), + ) + parser.add_argument( + '--expiration', + metavar='<expiration>', + help=_('Sets an expiration date for the application credential,' + ' format of YYYY-mm-ddTHH:MM:SS (if not provided, the' + ' application credential will not expire)'), + ) + parser.add_argument( + '--description', + metavar='<description>', + help=_('Application credential description'), + ) + parser.add_argument( + '--unrestricted', + action="store_true", + help=_('Enable application credential to create and delete other' + ' application credentials and trusts (this is potentially' + ' dangerous behavior and is disabled by default)'), + ) + parser.add_argument( + '--restricted', + action="store_true", + help=_('Prohibit application credential from creating and deleting' + ' other application credentials and trusts (this is the' + ' default behavior)'), + ) + return parser + + def take_action(self, parsed_args): + identity_client = self.app.client_manager.identity + + role_ids = [] + for role in parsed_args.role: + # A user can only create an application credential for themself, + # not for another user even as an admin, and only on the project to + # which they are currently scoped with a subset of the role + # assignments they have on that project. Don't bother trying to + # look up roles via keystone, just introspect the token. + role_id = common._get_token_resource(identity_client, "roles", + role) + role_ids.append(role_id) + + expires_at = None + if parsed_args.expiration: + expires_at = datetime.datetime.strptime(parsed_args.expiration, + '%Y-%m-%dT%H:%M:%S') + + if parsed_args.restricted: + unrestricted = False + else: + unrestricted = parsed_args.unrestricted + + app_cred_manager = identity_client.application_credentials + application_credential = app_cred_manager.create( + parsed_args.name, + roles=role_ids, + expires_at=expires_at, + description=parsed_args.description, + secret=parsed_args.secret, + unrestricted=unrestricted, + ) + + application_credential._info.pop('links', None) + + # Format roles into something sensible + roles = application_credential._info.pop('roles') + msg = ' '.join(r['name'] for r in roles) + application_credential._info['roles'] = msg + + return zip(*sorted(six.iteritems(application_credential._info))) + + +class DeleteApplicationCredential(command.Command): + _description = _("Delete application credentials(s)") + + def get_parser(self, prog_name): + parser = super(DeleteApplicationCredential, self).get_parser(prog_name) + parser.add_argument( + 'application_credential', + metavar='<application-credential>', + nargs="+", + help=_('Application credentials(s) to delete (name or ID)'), + ) + return parser + + def take_action(self, parsed_args): + identity_client = self.app.client_manager.identity + + errors = 0 + for ac in parsed_args.application_credential: + try: + app_cred = utils.find_resource( + identity_client.application_credentials, ac) + identity_client.application_credentials.delete(app_cred.id) + except Exception as e: + errors += 1 + LOG.error(_("Failed to delete application credential with " + "name or ID '%(ac)s': %(e)s"), + {'ac': ac, 'e': e}) + + if errors > 0: + total = len(parsed_args.application_credential) + msg = (_("%(errors)s of %(total)s application credentials failed " + "to delete.") % {'errors': errors, 'total': total}) + raise exceptions.CommandError(msg) + + +class ListApplicationCredential(command.Lister): + _description = _("List application credentials") + + def get_parser(self, prog_name): + parser = super(ListApplicationCredential, self).get_parser(prog_name) + parser.add_argument( + '--user', + metavar='<user>', + help=_('User whose application credentials to list (name or ID)'), + ) + common.add_user_domain_option_to_parser(parser) + return parser + + def take_action(self, parsed_args): + identity_client = self.app.client_manager.identity + if parsed_args.user: + user_id = common.find_user(identity_client, + parsed_args.user, + parsed_args.user_domain).id + else: + user_id = None + + columns = ('ID', 'Name', 'Project ID', 'Description', 'Expires At') + data = identity_client.application_credentials.list( + user=user_id) + return (columns, + (utils.get_item_properties( + s, columns, + formatters={}, + ) for s in data)) + + +class ShowApplicationCredential(command.ShowOne): + _description = _("Display application credential details") + + def get_parser(self, prog_name): + parser = super(ShowApplicationCredential, self).get_parser(prog_name) + parser.add_argument( + 'application_credential', + metavar='<application-credential>', + help=_('Application credential to display (name or ID)'), + ) + return parser + + def take_action(self, parsed_args): + identity_client = self.app.client_manager.identity + app_cred = utils.find_resource(identity_client.application_credentials, + parsed_args.application_credential) + + app_cred._info.pop('links', None) + + # Format roles into something sensible + roles = app_cred._info.pop('roles') + msg = ' '.join(r['name'] for r in roles) + app_cred._info['roles'] = msg + + return zip(*sorted(six.iteritems(app_cred._info))) diff --git a/openstackclient/tests/functional/identity/v3/test_application_credential.py b/openstackclient/tests/functional/identity/v3/test_application_credential.py new file mode 100644 index 00000000..daf64607 --- /dev/null +++ b/openstackclient/tests/functional/identity/v3/test_application_credential.py @@ -0,0 +1,143 @@ +# Copyright 2018 SUSE Linux GmbH +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +import datetime + +from tempest.lib.common.utils import data_utils + +from openstackclient.tests.functional.identity.v3 import common + + +class ApplicationCredentialTests(common.IdentityTests): + + APPLICATION_CREDENTIAL_FIELDS = ['id', 'name', 'project_id', + 'description', 'roles', 'expires_at', + 'unrestricted'] + APPLICATION_CREDENTIAL_LIST_HEADERS = ['ID', 'Name', 'Project ID', + 'Description', 'Expires At'] + + def test_application_credential_create(self): + name = data_utils.rand_name('name') + raw_output = self.openstack('application credential create %(name)s' + % {'name': name}) + self.addCleanup( + self.openstack, + 'application credential delete %(name)s' % {'name': name}) + items = self.parse_show(raw_output) + self.assert_show_fields(items, self.APPLICATION_CREDENTIAL_FIELDS) + + def _create_role_assignments(self): + try: + user = self.openstack('configuration show -f value' + ' -c auth.username') + except Exception: + user = self.openstack('configuration show -f value' + ' -c auth.user_id') + try: + user_domain = self.openstack('configuration show -f value' + ' -c auth.user_domain_name') + except Exception: + user_domain = self.openstack('configuration show -f value' + ' -c auth.user_domain_id') + try: + project = self.openstack('configuration show -f value' + ' -c auth.project_name') + except Exception: + project = self.openstack('configuration show -f value' + ' -c auth.project_id') + try: + project_domain = self.openstack('configuration show -f value' + ' -c auth.project_domain_name') + except Exception: + project_domain = self.openstack('configuration show -f value' + ' -c auth.project_domain_id') + role1 = self._create_dummy_role() + role2 = self._create_dummy_role() + for role in role1, role2: + self.openstack('role add' + ' --user %(user)s' + ' --user-domain %(user_domain)s' + ' --project %(project)s' + ' --project-domain %(project_domain)s' + ' %(role)s' + % {'user': user, + 'user_domain': user_domain, + 'project': project, + 'project_domain': project_domain, + 'role': role}) + self.addCleanup(self.openstack, + 'role remove' + ' --user %(user)s' + ' --user-domain %(user_domain)s' + ' --project %(project)s' + ' --project-domain %(project_domain)s' + ' %(role)s' + % {'user': user, + 'user_domain': user_domain, + 'project': project, + 'project_domain': project_domain, + 'role': role}) + return role1, role2 + + def test_application_credential_create_with_options(self): + name = data_utils.rand_name('name') + secret = data_utils.rand_name('secret') + description = data_utils.rand_name('description') + tomorrow = (datetime.datetime.utcnow() + + datetime.timedelta(days=1)).strftime('%Y-%m-%dT%H:%M:%S%z') + role1, role2 = self._create_role_assignments() + raw_output = self.openstack('application credential create %(name)s' + ' --secret %(secret)s' + ' --description %(description)s' + ' --expiration %(tomorrow)s' + ' --role %(role1)s' + ' --role %(role2)s' + ' --unrestricted' + % {'name': name, + 'secret': secret, + 'description': description, + 'tomorrow': tomorrow, + 'role1': role1, + 'role2': role2}) + self.addCleanup( + self.openstack, + 'application credential delete %(name)s' % {'name': name}) + items = self.parse_show(raw_output) + self.assert_show_fields(items, self.APPLICATION_CREDENTIAL_FIELDS) + + def test_application_credential_delete(self): + name = data_utils.rand_name('name') + self.openstack('application credential create %(name)s' + % {'name': name}) + raw_output = self.openstack('application credential delete ' + '%(name)s' % {'name': name}) + self.assertEqual(0, len(raw_output)) + + def test_application_credential_list(self): + raw_output = self.openstack('application credential list') + items = self.parse_listing(raw_output) + self.assert_table_structure( + items, self.APPLICATION_CREDENTIAL_LIST_HEADERS) + + def test_application_credential_show(self): + name = data_utils.rand_name('name') + raw_output = self.openstack('application credential create %(name)s' + % {'name': name}) + self.addCleanup( + self.openstack, + 'application credential delete %(name)s' % {'name': name}) + raw_output = self.openstack('application credential show ' + '%(name)s' % {'name': name}) + items = self.parse_show(raw_output) + self.assert_show_fields(items, self.APPLICATION_CREDENTIAL_FIELDS) diff --git a/openstackclient/tests/unit/identity/v3/fakes.py b/openstackclient/tests/unit/identity/v3/fakes.py index 8ceca3ce..77928792 100644 --- a/openstackclient/tests/unit/identity/v3/fakes.py +++ b/openstackclient/tests/unit/identity/v3/fakes.py @@ -14,6 +14,7 @@ # import copy +import datetime import uuid from keystoneauth1 import access @@ -457,6 +458,34 @@ OAUTH_VERIFIER = { 'oauth_verifier': oauth_verifier_pin } +app_cred_id = 'app-cred-id' +app_cred_name = 'testing_app_cred' +app_cred_role = {"id": role_id, "name": role_name, "domain": None}, +app_cred_description = 'app credential for testing' +app_cred_expires = datetime.datetime(2022, 1, 1, 0, 0) +app_cred_expires_str = app_cred_expires.strftime('%Y-%m-%dT%H:%M:%S%z') +app_cred_secret = 'moresecuresecret' +APP_CRED_BASIC = { + 'id': app_cred_id, + 'name': app_cred_name, + 'project_id': project_id, + 'roles': app_cred_role, + 'description': None, + 'expires_at': None, + 'unrestricted': False, + 'secret': app_cred_secret +} +APP_CRED_OPTIONS = { + 'id': app_cred_id, + 'name': app_cred_name, + 'project_id': project_id, + 'roles': app_cred_role, + 'description': app_cred_description, + 'expires_at': app_cred_expires_str, + 'unrestricted': False, + 'secret': app_cred_secret +} + def fake_auth_ref(fake_token, fake_service=None): """Create an auth_ref using keystoneauth's fixtures""" @@ -544,6 +573,9 @@ class FakeIdentityv3Client(object): self.auth = FakeAuth() self.auth.client = mock.Mock() self.auth.client.resource_class = fakes.FakeResource(None, {}) + self.application_credentials = mock.Mock() + self.application_credentials.resource_class = fakes.FakeResource(None, + {}) class FakeFederationManager(object): diff --git a/openstackclient/tests/unit/identity/v3/test_application_credential.py b/openstackclient/tests/unit/identity/v3/test_application_credential.py new file mode 100644 index 00000000..e7c8ede8 --- /dev/null +++ b/openstackclient/tests/unit/identity/v3/test_application_credential.py @@ -0,0 +1,309 @@ +# Copyright 2018 SUSE Linux GmbH +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# + +import copy + +import mock +from osc_lib import exceptions +from osc_lib import utils + +from openstackclient.identity.v3 import application_credential +from openstackclient.tests.unit import fakes +from openstackclient.tests.unit.identity.v3 import fakes as identity_fakes + + +class TestApplicationCredential(identity_fakes.TestIdentityv3): + + def setUp(self): + super(TestApplicationCredential, self).setUp() + + identity_manager = self.app.client_manager.identity + self.app_creds_mock = identity_manager.application_credentials + self.app_creds_mock.reset_mock() + self.roles_mock = identity_manager.roles + self.roles_mock.reset_mock() + + +class TestApplicationCredentialCreate(TestApplicationCredential): + + def setUp(self): + super(TestApplicationCredentialCreate, self).setUp() + + self.roles_mock.get.return_value = fakes.FakeResource( + None, + copy.deepcopy(identity_fakes.ROLE), + loaded=True, + ) + + # Get the command object to test + self.cmd = application_credential.CreateApplicationCredential( + self.app, None) + + def test_application_credential_create_basic(self): + self.app_creds_mock.create.return_value = fakes.FakeResource( + None, + copy.deepcopy(identity_fakes.APP_CRED_BASIC), + loaded=True, + ) + + name = identity_fakes.app_cred_name + arglist = [ + name + ] + verifylist = [ + ('name', identity_fakes.app_cred_name) + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + # In base command class ShowOne in cliff, abstract method take_action() + # returns a two-part tuple with a tuple of column names and a tuple of + # data to be shown. + columns, data = self.cmd.take_action(parsed_args) + + # Set expected values + kwargs = { + 'secret': None, + 'roles': [], + 'expires_at': None, + 'description': None, + 'unrestricted': False, + } + self.app_creds_mock.create.assert_called_with( + name, + **kwargs + ) + + collist = ('description', 'expires_at', 'id', 'name', 'project_id', + 'roles', 'secret', 'unrestricted') + self.assertEqual(collist, columns) + datalist = ( + None, + None, + identity_fakes.app_cred_id, + identity_fakes.app_cred_name, + identity_fakes.project_id, + identity_fakes.role_name, + identity_fakes.app_cred_secret, + False, + ) + self.assertEqual(datalist, data) + + def test_application_credential_create_with_options(self): + name = identity_fakes.app_cred_name + self.app_creds_mock.create.return_value = fakes.FakeResource( + None, + copy.deepcopy(identity_fakes.APP_CRED_OPTIONS), + loaded=True, + ) + + arglist = [ + name, + '--secret', 'moresecuresecret', + '--role', identity_fakes.role_id, + '--expiration', identity_fakes.app_cred_expires_str, + '--description', 'credential for testing' + ] + verifylist = [ + ('name', identity_fakes.app_cred_name), + ('secret', 'moresecuresecret'), + ('role', [identity_fakes.role_id]), + ('expiration', identity_fakes.app_cred_expires_str), + ('description', 'credential for testing') + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + # In base command class ShowOne in cliff, abstract method take_action() + # returns a two-part tuple with a tuple of column names and a tuple of + # data to be shown. + columns, data = self.cmd.take_action(parsed_args) + + # Set expected values + kwargs = { + 'secret': 'moresecuresecret', + 'roles': [identity_fakes.role_id], + 'expires_at': identity_fakes.app_cred_expires, + 'description': 'credential for testing', + 'unrestricted': False + } + self.app_creds_mock.create.assert_called_with( + name, + **kwargs + ) + + collist = ('description', 'expires_at', 'id', 'name', 'project_id', + 'roles', 'secret', 'unrestricted') + self.assertEqual(collist, columns) + datalist = ( + identity_fakes.app_cred_description, + identity_fakes.app_cred_expires_str, + identity_fakes.app_cred_id, + identity_fakes.app_cred_name, + identity_fakes.project_id, + identity_fakes.role_name, + identity_fakes.app_cred_secret, + False, + ) + self.assertEqual(datalist, data) + + +class TestApplicationCredentialDelete(TestApplicationCredential): + + def setUp(self): + super(TestApplicationCredentialDelete, self).setUp() + + # This is the return value for utils.find_resource() + self.app_creds_mock.get.return_value = fakes.FakeResource( + None, + copy.deepcopy(identity_fakes.APP_CRED_BASIC), + loaded=True, + ) + self.app_creds_mock.delete.return_value = None + + # Get the command object to test + self.cmd = application_credential.DeleteApplicationCredential( + self.app, None) + + def test_application_credential_delete(self): + arglist = [ + identity_fakes.app_cred_id, + ] + verifylist = [ + ('application_credential', [identity_fakes.app_cred_id]) + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + + self.app_creds_mock.delete.assert_called_with( + identity_fakes.app_cred_id, + ) + self.assertIsNone(result) + + @mock.patch.object(utils, 'find_resource') + def test_delete_multi_app_creds_with_exception(self, find_mock): + find_mock.side_effect = [self.app_creds_mock.get.return_value, + exceptions.CommandError] + arglist = [ + identity_fakes.app_cred_id, + 'nonexistent_app_cred', + ] + verifylist = [ + ('application_credential', arglist), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + try: + self.cmd.take_action(parsed_args) + self.fail('CommandError should be raised.') + except exceptions.CommandError as e: + self.assertEqual('1 of 2 application credentials failed to' + ' delete.', str(e)) + + find_mock.assert_any_call(self.app_creds_mock, + identity_fakes.app_cred_id) + find_mock.assert_any_call(self.app_creds_mock, + 'nonexistent_app_cred') + + self.assertEqual(2, find_mock.call_count) + self.app_creds_mock.delete.assert_called_once_with( + identity_fakes.app_cred_id) + + +class TestApplicationCredentialList(TestApplicationCredential): + + def setUp(self): + super(TestApplicationCredentialList, self).setUp() + + self.app_creds_mock.list.return_value = [ + fakes.FakeResource( + None, + copy.deepcopy(identity_fakes.APP_CRED_BASIC), + loaded=True, + ), + ] + + # Get the command object to test + self.cmd = application_credential.ListApplicationCredential(self.app, + None) + + def test_application_credential_list(self): + arglist = [] + verifylist = [] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + # In base command class Lister in cliff, abstract method take_action() + # returns a tuple containing the column names and an iterable + # containing the data to be listed. + columns, data = self.cmd.take_action(parsed_args) + + self.app_creds_mock.list.assert_called_with(user=None) + + collist = ('ID', 'Name', 'Project ID', 'Description', 'Expires At') + self.assertEqual(collist, columns) + datalist = (( + identity_fakes.app_cred_id, + identity_fakes.app_cred_name, + identity_fakes.project_id, + None, + None + ), ) + self.assertEqual(datalist, tuple(data)) + + +class TestApplicationCredentialShow(TestApplicationCredential): + + def setUp(self): + super(TestApplicationCredentialShow, self).setUp() + + self.app_creds_mock.get.return_value = fakes.FakeResource( + None, + copy.deepcopy(identity_fakes.APP_CRED_BASIC), + loaded=True, + ) + + # Get the command object to test + self.cmd = application_credential.ShowApplicationCredential(self.app, + None) + + def test_application_credential_show(self): + arglist = [ + identity_fakes.app_cred_id, + ] + verifylist = [ + ('application_credential', identity_fakes.app_cred_id), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + # In base command class ShowOne in cliff, abstract method take_action() + # returns a two-part tuple with a tuple of column names and a tuple of + # data to be shown. + columns, data = self.cmd.take_action(parsed_args) + + self.app_creds_mock.get.assert_called_with(identity_fakes.app_cred_id) + + collist = ('description', 'expires_at', 'id', 'name', 'project_id', + 'roles', 'secret', 'unrestricted') + self.assertEqual(collist, columns) + datalist = ( + None, + None, + identity_fakes.app_cred_id, + identity_fakes.app_cred_name, + identity_fakes.project_id, + identity_fakes.role_name, + identity_fakes.app_cred_secret, + False, + ) + self.assertEqual(datalist, data) |