diff --git a/openstackclient/common/clientmanager.py b/openstackclient/common/clientmanager.py index ec005dc092..9e4b9c937f 100644 --- a/openstackclient/common/clientmanager.py +++ b/openstackclient/common/clientmanager.py @@ -15,20 +15,13 @@ """Manage access to the clients, including authenticating when needed.""" -import copy import logging import pkg_resources import sys from keystoneauth1.loading import base from osc_lib.api import auth -from osc_lib import exceptions -from oslo_utils import strutils -import requests -import six - -from openstackclient.common import session as osc_session -from openstackclient.identity import client as identity_client +from osc_lib import clientmanager LOG = logging.getLogger(__name__) @@ -109,267 +102,44 @@ def build_auth_params(auth_plugin_name, cmd_options): return (auth_plugin_loader, auth_params) -class ClientCache(object): - """Descriptor class for caching created client handles.""" +class ClientManager(clientmanager.ClientManager): + """Manages access to API clients, including authentication - def __init__(self, factory): - self.factory = factory - self._handle = None - - def __get__(self, instance, owner): - # Tell the ClientManager to login to keystone - if self._handle is None: - try: - self._handle = self.factory(instance) - except AttributeError as err: - # Make sure the failure propagates. Otherwise, the plugin just - # quietly isn't there. - new_err = exceptions.PluginAttributeError(err) - six.reraise(new_err.__class__, new_err, sys.exc_info()[2]) - return self._handle - - -class ClientManager(object): - """Manages access to API clients, including authentication.""" + Wrap osc_lib's ClientManager to maintain compatibility for the existing + plugin V2 interface. Some currently private attributes become public + in osc-lib so we need to maintain a transition period. + """ # A simple incrementing version for the plugin to know what is available PLUGIN_INTERFACE_VERSION = "2" - identity = ClientCache(identity_client.make_client) - - def __getattr__(self, name): - # this is for the auth-related parameters. - if name in ['_' + o.replace('-', '_') - for o in auth.OPTIONS_LIST]: - return self._auth_params[name[1:]] - - raise AttributeError(name) - def __init__( self, cli_options=None, api_version=None, - verify=True, pw_func=None, ): - """Set up a ClientManager - - :param cli_options: - Options collected from the command-line, environment, or wherever - :param api_version: - Dict of API versions: key is API name, value is the version - :param verify: - TLS certificate verification; may be a boolean to enable or disable - server certificate verification, or a filename of a CA certificate - bundle to be used in verification (implies True) - :param pw_func: - Callback function for asking the user for a password. The function - takes an optional string for the prompt ('Password: ' on None) and - returns a string containing the password - """ - - self._cli_options = cli_options - self._api_version = api_version - self._pw_callback = pw_func - self._url = self._cli_options.auth.get('url') - self._region_name = self._cli_options.region_name - self._interface = self._cli_options.interface - - self.timing = self._cli_options.timing - - self._auth_ref = None - self.session = None - - # verify is the Requests-compatible form - self._verify = verify - # also store in the form used by the legacy client libs - self._cacert = None - if isinstance(verify, bool): - self._insecure = not verify - else: - self._cacert = verify - self._insecure = False - - # Set up client certificate and key - # NOTE(cbrandily): This converts client certificate/key to requests - # cert argument: None (no client certificate), a path - # to client certificate or a tuple with client - # certificate/key paths. - self._cert = self._cli_options.cert - if self._cert and self._cli_options.key: - self._cert = self._cert, self._cli_options.key - - # Get logging from root logger - root_logger = logging.getLogger('') - LOG.setLevel(root_logger.getEffectiveLevel()) - - # NOTE(gyee): use this flag to indicate whether auth setup has already - # been completed. If so, do not perform auth setup again. The reason - # we need this flag is that we want to be able to perform auth setup - # outside of auth_ref as auth_ref itself is a property. We can not - # retrofit auth_ref to optionally skip scope check. Some operations - # do not require a scoped token. In those cases, we call setup_auth - # prior to dereferrencing auth_ref. - self._auth_setup_completed = False - - def _set_default_scope_options(self): - # TODO(mordred): This is a usability improvement that's broadly useful - # We should port it back up into os-client-config. - default_domain = self._cli_options.default_domain - - # NOTE(hieulq): If USER_DOMAIN_NAME, USER_DOMAIN_ID, PROJECT_DOMAIN_ID - # or PROJECT_DOMAIN_NAME is present and API_VERSION is 2.0, then - # ignore all domain related configs. - if (self._api_version.get('identity') == '2.0' and - self.auth_plugin_name.endswith('password')): - domain_props = ['project_domain_name', 'project_domain_id', - 'user_domain_name', 'user_domain_id'] - for prop in domain_props: - if self._auth_params.pop(prop, None) is not None: - LOG.warning("Ignoring domain related configs " + - prop + " because identity API version is 2.0") - return - - # NOTE(aloga): The scope parameters below only apply to v3 and v3 - # related auth plugins, so we stop the parameter checking if v2 is - # being used. - if (self._api_version.get('identity') != '3' or - self.auth_plugin_name.startswith('v2')): - return - - # NOTE(stevemar): If PROJECT_DOMAIN_ID or PROJECT_DOMAIN_NAME is - # present, then do not change the behaviour. Otherwise, set the - # PROJECT_DOMAIN_ID to 'OS_DEFAULT_DOMAIN' for better usability. - if ('project_domain_id' in self._auth_params and - not self._auth_params.get('project_domain_id') and - not self._auth_params.get('project_domain_name')): - self._auth_params['project_domain_id'] = default_domain - - # NOTE(stevemar): If USER_DOMAIN_ID or USER_DOMAIN_NAME is present, - # then do not change the behaviour. Otherwise, set the - # USER_DOMAIN_ID to 'OS_DEFAULT_DOMAIN' for better usability. - if ('user_domain_id' in self._auth_params and - not self._auth_params.get('user_domain_id') and - not self._auth_params.get('user_domain_name')): - self._auth_params['user_domain_id'] = default_domain - - def setup_auth(self): - """Set up authentication - - This is deferred until authentication is actually attempted because - it gets in the way of things that do not require auth. - """ - - if self._auth_setup_completed: - return - - # If no auth type is named by the user, select one based on - # the supplied options - self.auth_plugin_name = select_auth_plugin(self._cli_options) - - # Basic option checking to avoid unhelpful error messages - auth.check_valid_authentication_options(self._cli_options, - self.auth_plugin_name) - - # Horrible hack alert...must handle prompt for null password if - # password auth is requested. - if (self.auth_plugin_name.endswith('password') and - not self._cli_options.auth.get('password')): - self._cli_options.auth['password'] = self._pw_callback() - - (auth_plugin, self._auth_params) = build_auth_params( - self.auth_plugin_name, - self._cli_options, + super(ClientManager, self).__init__( + cli_options=cli_options, + api_version=api_version, + pw_func=pw_func, ) - self._set_default_scope_options() - - # For compatibility until all clients can be updated - if 'project_name' in self._auth_params: - self._project_name = self._auth_params['project_name'] - elif 'tenant_name' in self._auth_params: - self._project_name = self._auth_params['tenant_name'] - - LOG.info('Using auth plugin: %s', self.auth_plugin_name) - LOG.debug('Using parameters %s', - strutils.mask_password(self._auth_params)) - self.auth = auth_plugin.load_from_options(**self._auth_params) - # needed by SAML authentication - request_session = requests.session() - self.session = osc_session.TimingSession( - auth=self.auth, - session=request_session, - verify=self._verify, - cert=self._cert, - user_agent=USER_AGENT, - ) - - self._auth_setup_completed = True - - def validate_scope(self): - if self._auth_ref.project_id is not None: - # We already have a project scope. - return - if self._auth_ref.domain_id is not None: - # We already have a domain scope. - return - - # We do not have a scoped token (and the user's default project scope - # was not implied), so the client needs to be explicitly configured - # with a scope. - auth.check_valid_authorization_options(self._cli_options, - self.auth_plugin_name) - - @property - def auth_ref(self): - """Dereference will trigger an auth if it hasn't already""" - if not self._auth_ref: - self.setup_auth() - LOG.debug("Get auth_ref") - self._auth_ref = self.auth.get_auth_ref(self.session) - return self._auth_ref + # TODO(dtroyer): For compatibility; mark this for removal when plugin + # interface v2 is removed + self._region_name = self.region_name + self._interface = self.interface + self._cacert = self.cacert + self._insecure = not self.verify def is_network_endpoint_enabled(self): """Check if the network endpoint is enabled""" - # Trigger authentication necessary to determine if the network - # endpoint is enabled. - if self.auth_ref: - service_catalog = self.auth_ref.service_catalog - else: - service_catalog = None - # Assume that the network endpoint is enabled. - network_endpoint_enabled = True - if service_catalog: - if 'network' in service_catalog.get_endpoints(): - LOG.debug("Network endpoint in service catalog") - else: - LOG.debug("No network endpoint in service catalog") - network_endpoint_enabled = False - else: - LOG.debug("No service catalog, assuming network endpoint enabled") - return network_endpoint_enabled - def get_endpoint_for_service_type(self, service_type, region_name=None, - interface='public'): - """Return the endpoint URL for the service type.""" - if not interface: - interface = 'public' - # See if we are using password flow auth, i.e. we have a - # service catalog to select endpoints from - if self.auth_ref: - endpoint = self.auth_ref.service_catalog.url_for( - service_type=service_type, - region_name=region_name, - interface=interface, - ) - else: - # Get the passed endpoint directly from the auth plugin - endpoint = self.auth.get_endpoint(self.session, - interface=interface) - return endpoint - - def get_configuration(self): - return copy.deepcopy(self._cli_options.config) + # NOTE(dtroyer): is_service_available() can also return None if + # there is no Service Catalog, callers here are + # not expecting that so fold None into True to + # use Network API by default + return self.is_service_available('network') is not False # Plugin Support @@ -391,7 +161,7 @@ def get_plugin_modules(group): setattr( ClientManager, module.API_NAME, - ClientCache( + clientmanager.ClientCache( getattr(sys.modules[ep.module_name], 'make_client', None) ), ) diff --git a/openstackclient/shell.py b/openstackclient/shell.py index ed729e537c..b4f2df43b2 100644 --- a/openstackclient/shell.py +++ b/openstackclient/shell.py @@ -26,6 +26,7 @@ from cliff import app from cliff import command from cliff import complete from cliff import help +from osc_lib.cli import client_config as cloud_config from osc_lib.command import timing from osc_lib import exceptions as exc from osc_lib import logs @@ -38,8 +39,6 @@ from openstackclient.common import clientmanager from openstackclient.common import commandmanager from openstackclient.i18n import _ -from os_client_config import config as cloud_config - osprofiler_profiler = importutils.try_import("osprofiler.profiler") @@ -309,6 +308,9 @@ class OpenStackShell(app.App): tenant_id = getattr(self.options, 'tenant_id', None) tenant_name = getattr(self.options, 'tenant_name', None) + # Save default domain + self.default_domain = self.options.default_domain + # handle some v2/v3 authentication inconsistencies by just acting like # both the project and tenant information are both present. This can # go away if we stop registering all the argparse options together. @@ -325,7 +327,7 @@ class OpenStackShell(app.App): # Ignore the default value of interface. Only if it is set later # will it be used. try: - cc = cloud_config.OpenStackConfig( + cc = cloud_config.OSC_Config( override_defaults={ 'interface': None, 'auth_type': auth_type, @@ -368,9 +370,6 @@ class OpenStackShell(app.App): if self.verify and self.cloud.cacert: self.verify = self.cloud.cacert - # Save default domain - self.default_domain = self.options.default_domain - # Loop through extensions to get API versions for mod in clientmanager.PLUGIN_MODULES: default_version = getattr(mod, 'DEFAULT_API_VERSION', None) @@ -429,7 +428,6 @@ class OpenStackShell(app.App): self.client_manager = clientmanager.ClientManager( cli_options=self.cloud, - verify=self.verify, api_version=self.api_version, pw_func=prompt_for_password, ) diff --git a/openstackclient/tests/common/test_clientmanager.py b/openstackclient/tests/common/test_clientmanager.py index 117c718429..625e175a1d 100644 --- a/openstackclient/tests/common/test_clientmanager.py +++ b/openstackclient/tests/common/test_clientmanager.py @@ -13,111 +13,34 @@ # under the License. # -import json as jsonutils -import mock +import copy -from keystoneauth1.access import service_catalog -from keystoneauth1.identity import v2 as auth_v2 from keystoneauth1 import token_endpoint -from osc_lib.api import auth -from osc_lib import exceptions as exc -from requests_mock.contrib import fixture +from osc_lib.tests import utils as osc_lib_test_utils from openstackclient.common import clientmanager from openstackclient.tests import fakes -from openstackclient.tests import utils -API_VERSION = {"identity": "2.0"} -AUTH_REF = {'version': 'v2.0'} -AUTH_REF.update(fakes.TEST_RESPONSE_DICT['access']) -SERVICE_CATALOG = service_catalog.ServiceCatalogV2(AUTH_REF) +class TestClientManager(osc_lib_test_utils.TestClientManager): - -# This is deferred in api.auth but we need it here... -auth.get_options_list() - - -class Container(object): - attr = clientmanager.ClientCache(lambda x: object()) - buggy_attr = clientmanager.ClientCache(lambda x: x.foo) - - def __init__(self): - pass - - -class FakeOptions(object): - - def __init__(self, **kwargs): - for option in auth.OPTIONS_LIST: - setattr(self, option.replace('-', '_'), None) - self.auth_type = None - self.identity_api_version = '2.0' - self.timing = None - self.region_name = None - self.interface = None - self.url = None - self.auth = {} - self.cert = None - self.key = None - self.default_domain = 'default' - self.__dict__.update(kwargs) - - -class TestClientCache(utils.TestCase): - - def test_singleton(self): - # NOTE(dtroyer): Verify that the ClientCache descriptor only invokes - # the factory one time and always returns the same value after that. - c = Container() - self.assertEqual(c.attr, c.attr) - - def test_attribute_error_propagates(self): - c = Container() - err = self.assertRaises(exc.PluginAttributeError, - getattr, c, 'buggy_attr') - self.assertNotIsInstance(err, AttributeError) - self.assertEqual("'Container' object has no attribute 'foo'", str(err)) - - -class TestClientManager(utils.TestCase): - - def setUp(self): - super(TestClientManager, self).setUp() - self.mock = mock.Mock() - self.requests = self.useFixture(fixture.Fixture()) - # fake v2password token retrieval - self.stub_auth(json=fakes.TEST_RESPONSE_DICT) - # fake token and token_endpoint retrieval - self.stub_auth(json=fakes.TEST_RESPONSE_DICT, - url='/'.join([fakes.AUTH_URL, 'v2.0/tokens'])) - # fake v3password token retrieval - self.stub_auth(json=fakes.TEST_RESPONSE_DICT_V3, - url='/'.join([fakes.AUTH_URL, 'auth/tokens'])) - # fake password version endpoint discovery - self.stub_auth(json=fakes.TEST_VERSIONS, - url=fakes.AUTH_URL, - verb='GET') + def _clientmanager_class(self): + """Allow subclasses to override the ClientManager class""" + return clientmanager.ClientManager def test_client_manager_token_endpoint(self): - - client_manager = clientmanager.ClientManager( - cli_options=FakeOptions( - auth_type='token_endpoint', - auth=dict( - token=fakes.AUTH_TOKEN, - url=fakes.AUTH_URL, - ), - ), - api_version=API_VERSION, - verify=True + token_auth = { + 'url': fakes.AUTH_URL, + 'token': fakes.AUTH_TOKEN, + } + client_manager = self._make_clientmanager( + auth_args=token_auth, + auth_plugin_name='token_endpoint', ) - client_manager.setup_auth() - client_manager.auth_ref self.assertEqual( fakes.AUTH_URL, - client_manager._url, + client_manager._cli_options.config['auth']['url'], ) self.assertEqual( fakes.AUTH_TOKEN, @@ -127,256 +50,20 @@ class TestClientManager(utils.TestCase): client_manager.auth, token_endpoint.Token, ) - self.assertFalse(client_manager._insecure) - self.assertTrue(client_manager._verify) - self.assertTrue(client_manager.is_network_endpoint_enabled()) - - def test_client_manager_token(self): - - client_manager = clientmanager.ClientManager( - cli_options=FakeOptions( - auth=dict( - token=fakes.AUTH_TOKEN, - auth_url=fakes.AUTH_URL, - ), - auth_type='v2token', - interface=fakes.INTERFACE, - region_name=fakes.REGION_NAME, - ), - api_version=API_VERSION, - verify=True - ) - client_manager.setup_auth() - client_manager.auth_ref - - self.assertEqual( - fakes.AUTH_URL, - client_manager._auth_url, - ) - self.assertIsInstance( - client_manager.auth, - auth_v2.Token, - ) - self.assertEqual( - fakes.INTERFACE, - client_manager._interface, - ) - self.assertEqual( - fakes.REGION_NAME, - client_manager._region_name, - ) - self.assertFalse(client_manager._insecure) - self.assertTrue(client_manager._verify) - self.assertTrue(client_manager.is_network_endpoint_enabled()) - - def test_client_manager_password(self): - - client_manager = clientmanager.ClientManager( - cli_options=FakeOptions( - auth=dict( - auth_url=fakes.AUTH_URL, - username=fakes.USERNAME, - password=fakes.PASSWORD, - project_name=fakes.PROJECT_NAME, - ), - ), - api_version=API_VERSION, - verify=False, - ) - client_manager.setup_auth() - client_manager.auth_ref - - self.assertEqual( - fakes.AUTH_URL, - client_manager._auth_url, - ) - self.assertEqual( - fakes.USERNAME, - client_manager._username, - ) - self.assertEqual( - fakes.PASSWORD, - client_manager._password, - ) - self.assertIsInstance( - client_manager.auth, - auth_v2.Password, - ) - self.assertTrue(client_manager._insecure) - self.assertFalse(client_manager._verify) - # These need to stick around until the old-style clients are gone - self.assertEqual( - AUTH_REF.pop('version'), - client_manager.auth_ref.version, - ) - self.assertEqual( - fakes.to_unicode_dict(AUTH_REF), - client_manager.auth_ref._data['access'], - ) - self.assertEqual( - dir(SERVICE_CATALOG), - dir(client_manager.auth_ref.service_catalog), - ) self.assertTrue(client_manager.is_network_endpoint_enabled()) def test_client_manager_network_endpoint_disabled(self): - - client_manager = clientmanager.ClientManager( - cli_options=FakeOptions( - auth=dict( - auth_url=fakes.AUTH_URL, - username=fakes.USERNAME, - password=fakes.PASSWORD, - project_name=fakes.PROJECT_NAME, - ), - auth_type='v3password', - ), - api_version={"identity": "3"}, - verify=False, + auth_args = copy.deepcopy(self.default_password_auth) + auth_args.update({ + 'user_domain_name': 'default', + 'project_domain_name': 'default', + }) + # v3 fake doesn't have network endpoint + client_manager = self._make_clientmanager( + auth_args=auth_args, + identity_api_version='3', + auth_plugin_name='v3password', ) - client_manager.setup_auth() - client_manager.auth_ref - # v3 fake doesn't have network endpoint. + self.assertFalse(client_manager.is_service_available('network')) self.assertFalse(client_manager.is_network_endpoint_enabled()) - - def stub_auth(self, json=None, url=None, verb=None, **kwargs): - subject_token = fakes.AUTH_TOKEN - base_url = fakes.AUTH_URL - if json: - text = jsonutils.dumps(json) - headers = {'X-Subject-Token': subject_token, - 'Content-Type': 'application/json'} - if not url: - url = '/'.join([base_url, 'tokens']) - url = url.replace("/?", "?") - if not verb: - verb = 'POST' - self.requests.register_uri(verb, - url, - headers=headers, - text=text) - - def test_client_manager_password_verify_ca(self): - - client_manager = clientmanager.ClientManager( - cli_options=FakeOptions( - auth=dict( - auth_url=fakes.AUTH_URL, - username=fakes.USERNAME, - password=fakes.PASSWORD, - project_name=fakes.PROJECT_NAME, - ), - auth_type='v2password', - ), - api_version=API_VERSION, - verify='cafile', - ) - client_manager.setup_auth() - client_manager.auth_ref - - self.assertFalse(client_manager._insecure) - self.assertTrue(client_manager._verify) - self.assertEqual('cafile', client_manager._cacert) - self.assertTrue(client_manager.is_network_endpoint_enabled()) - - def test_client_manager_password_no_cert(self): - client_manager = clientmanager.ClientManager( - cli_options=FakeOptions()) - self.assertIsNone(client_manager._cert) - - def test_client_manager_password_client_cert(self): - client_manager = clientmanager.ClientManager( - cli_options=FakeOptions(cert='cert')) - self.assertEqual('cert', client_manager._cert) - - def test_client_manager_password_client_cert_and_key(self): - client_manager = clientmanager.ClientManager( - cli_options=FakeOptions(cert='cert', key='key')) - self.assertEqual(('cert', 'key'), client_manager._cert) - - def _select_auth_plugin(self, auth_params, api_version, auth_plugin_name): - auth_params['auth_type'] = auth_plugin_name - auth_params['identity_api_version'] = api_version - - client_manager = clientmanager.ClientManager( - cli_options=FakeOptions(**auth_params), - api_version={"identity": api_version}, - verify=True - ) - client_manager.setup_auth() - client_manager.auth_ref - - self.assertEqual( - auth_plugin_name, - client_manager.auth_plugin_name, - ) - - def test_client_manager_select_auth_plugin(self): - # test token auth - params = dict( - auth=dict( - auth_url=fakes.AUTH_URL, - token=fakes.AUTH_TOKEN, - ), - ) - self._select_auth_plugin(params, '2.0', 'v2token') - self._select_auth_plugin(params, '3', 'v3token') - self._select_auth_plugin(params, 'XXX', 'token') - # test token/endpoint auth - params = dict( - auth_plugin='token_endpoint', - auth=dict( - url='test', - token=fakes.AUTH_TOKEN, - ), - ) - self._select_auth_plugin(params, 'XXX', 'token_endpoint') - # test password auth - params = dict( - auth=dict( - auth_url=fakes.AUTH_URL, - username=fakes.USERNAME, - password=fakes.PASSWORD, - project_name=fakes.PROJECT_NAME, - ), - ) - self._select_auth_plugin(params, '2.0', 'v2password') - self._select_auth_plugin(params, '3', 'v3password') - self._select_auth_plugin(params, 'XXX', 'password') - - def test_client_manager_select_auth_plugin_failure(self): - client_manager = clientmanager.ClientManager( - cli_options=FakeOptions(os_auth_plugin=''), - api_version=API_VERSION, - verify=True, - ) - self.assertRaises( - exc.CommandError, - client_manager.setup_auth, - ) - - @mock.patch('osc_lib.api.auth.check_valid_authentication_options') - def test_client_manager_auth_setup_once(self, check_authn_options_func): - client_manager = clientmanager.ClientManager( - cli_options=FakeOptions( - auth=dict( - auth_url=fakes.AUTH_URL, - username=fakes.USERNAME, - password=fakes.PASSWORD, - project_name=fakes.PROJECT_NAME, - ), - ), - api_version=API_VERSION, - verify=False, - ) - self.assertFalse(client_manager._auth_setup_completed) - client_manager.setup_auth() - self.assertTrue(check_authn_options_func.called) - self.assertTrue(client_manager._auth_setup_completed) - - # now make sure we don't do auth setup the second time around - # by checking whether check_valid_auth_options() gets called again - check_authn_options_func.reset_mock() - client_manager.auth_ref - check_authn_options_func.assert_not_called()