Browse Source

Add tools to get keystone auth plugin

With directly provide auth string(with contain a json formate
with auth_type and auth info), we can release context to specific
auth_type and give user the ability to provide other Keystone
(or their own) authentication method (like using
`v3applicationcredential` or others).
The format for `auth` and `auth_type` follows exactly Keystone
plugins like in clouds.yaml file [1].

[1] https://docs.openstack.org/keystoneauth/latest/
plugin-options.html#additional-loaders

Change-Id: Ic4dc2292a82860b9bb54ecb9e3b1a4dc806dab2c
Story: #2002126
Task: #26904
changes/50/580550/12
ricolin 3 years ago
parent
commit
4ee754f359
  1. 64
      heat/common/auth_plugin.py
  2. 9
      heat/common/exception.py
  3. 96
      heat/tests/test_common_auth_plugin.py

64
heat/common/auth_plugin.py

@ -0,0 +1,64 @@
#
# 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 json
from keystoneauth1 import loading as ks_loading
from oslo_log import log as logging
from heat.common import exception
LOG = logging.getLogger(__name__)
def parse_auth_credential_to_dict(cred):
"""Parse credential to dict"""
def validate(cred):
valid_keys = ['auth_type', 'auth']
for k in valid_keys:
if k not in cred:
raise ValueError('Missing key in auth information, the '
'correct format contains %s.' % valid_keys)
try:
_cred = json.loads(cred)
except ValueError as e:
LOG.error('Failed to parse credential with error: %s' % e)
raise ValueError('Failed to parse credential, please check your '
'Stack Credential format.')
validate(_cred)
return _cred
def validate_auth_plugin(auth_plugin, keystone_session):
"""Validate if this auth_plugin is valid to use."""
try:
auth_plugin.get_token(keystone_session)
except Exception as e:
# TODO(ricolin) Add heat document link for plugin information,
# once we generated one.
failure_reason = ("Failed to validate auth_plugin with error %s. "
"Please make sure the credential you provide is "
"correct. Also make sure the it is a valid Keystone "
"auth plugin type and contain in your "
"environment." % e)
raise exception.AuthorizationFailure(failure_reason=failure_reason)
def get_keystone_plugin_loader(auth, keystone_session):
cred = parse_auth_credential_to_dict(auth)
auth_plugin = ks_loading.get_plugin_loader(
cred.get('auth_type')).load_from_options(
**cred.get('auth'))
validate_auth_plugin(auth_plugin, keystone_session)
return auth_plugin

9
heat/common/exception.py

@ -90,7 +90,14 @@ class MissingCredentialError(HeatException):
class AuthorizationFailure(HeatException):
msg_fmt = _("Authorization failed.")
msg_fmt = _("Authorization failed.%(failure_reason)s")
def __init__(self, failure_reason=""):
if failure_reason != "":
# Add a space to make message more readable
failure_reason = " " + failure_reason
super(AuthorizationFailure, self).__init__(
failure_reason=failure_reason)
class NotAuthenticated(HeatException):

96
heat/tests/test_common_auth_plugin.py

@ -0,0 +1,96 @@
#
# 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 json
from keystoneauth1 import loading as ks_loading
from keystoneauth1 import session
import mock
import six
from heat.common import auth_plugin
from heat.common import config
from heat.common import exception
from heat.common import policy
from heat.tests import common
class TestAuthPlugin(common.HeatTestCase):
def setUp(self):
self.credential = (
'{"auth_type": "v3applicationcredential", '
'"auth": {"auth_url": "http://192.168.1.101/identity/v3", '
'"application_credential_id": '
'"9dfa187e5a354484bf9c49a2b674333a", '
'"application_credential_secret": "sec"} }')
self.m_plugin = mock.Mock()
self.m_loader = self.patchobject(
ks_loading, 'get_plugin_loader', return_value=self.m_plugin)
self.patchobject(policy.Enforcer, 'check_is_admin')
self.secret_id = '0eca0615-c330-41aa-b0cb-a2493a770409'
self.session = session.Session(
**config.get_ssl_options('keystone'))
super(TestAuthPlugin, self).setUp()
def _get_keystone_plugin_loader(self):
auth_plugin.get_keystone_plugin_loader(self.credential, self.session)
self.m_plugin.load_from_options.assert_called_once_with(
application_credential_id='9dfa187e5a354484bf9c49a2b674333a',
application_credential_secret='sec',
auth_url='http://192.168.1.101/identity/v3')
self.m_loader.assert_called_once_with('v3applicationcredential')
def test_get_keystone_plugin_loader(self):
self._get_keystone_plugin_loader()
# called in validate_auth_plugin
self.assertEqual(
1, self.m_plugin.load_from_options().get_token.call_count)
def test_get_keystone_plugin_loader_with_no_AuthZ(self):
self.m_plugin.load_from_options().get_token.side_effect = Exception
self.assertRaises(
exception.AuthorizationFailure, self._get_keystone_plugin_loader)
self.assertEqual(
1, self.m_plugin.load_from_options().get_token.call_count)
def test_parse_auth_credential_to_dict(self):
cred_dict = json.loads(self.credential)
self.assertEqual(
cred_dict, auth_plugin.parse_auth_credential_to_dict(
self.credential))
def test_parse_auth_credential_to_dict_with_value_error(self):
credential = (
'{"auth": {"auth_url": "http://192.168.1.101/identity/v3", '
'"application_credential_id": '
'"9dfa187e5a354484bf9c49a2b674333a", '
'"application_credential_secret": "sec"} }')
error = self.assertRaises(
ValueError, auth_plugin.parse_auth_credential_to_dict, credential)
self.assertEqual("Missing key in auth information, the correct "
"format contains [\'auth_type\', \'auth\'].",
six.text_type(error))
def test_parse_auth_credential_to_dict_with_json_error(self):
credential = (
"{'auth_type': v3applicationcredential, "
"'auth': {'auth_url': 'http://192.168.1.101/identity/v3', "
"'application_credential_id': "
"'9dfa187e5a354484bf9c49a2b674333a', "
"'application_credential_secret': 'sec'} }")
error = self.assertRaises(
ValueError, auth_plugin.parse_auth_credential_to_dict, credential)
error_msg = ('Failed to parse credential, please check your Stack '
'Credential format.')
self.assertEqual(error_msg, six.text_type(error))
Loading…
Cancel
Save