openstack/python-barbicanclient
Code Issues Proposed changes

Add auth plugin for Rackspace Identity

Browse Source 
Change-Id: I2f99a209eec3ee8beb9dd394f43849901c98044a
This commit is contained in:
Douglas Mendizabal
2013-12-17 17:22:25 -06:00
parent 9d67ba5899
commit 4bd35f7bdc
2 changed files with 198 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

102
barbicanclient/common/auth.py
View File

@@ -12,10 +12,13 @@
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import abc
import json
import logging
from keystoneclient.v2_0 import client as ksclient
from keystoneclient import exceptions
import requests
LOG = logging.getLogger(__name__)
@@ -23,16 +26,34 @@ LOG = logging.getLogger(__name__)
class AuthException(Exception):
"""Raised when authorization fails."""
def __init__(self, message):
self.message = message
pass
class KeystoneAuthV2(object):
class AuthPluginBase(object):
"""Base class for Auth plugins."""
__metaclass__ = abc.ABCMeta
@abc.abstractproperty
def auth_token(self):
"""
Returns a valid token to be used in X-Auth-Token header for
api requests.
"""
@abc.abstractproperty
def barbican_url(self):
"""
Returns the barbican endpoint url, including the version.
"""
class KeystoneAuthV2(AuthPluginBase):
def __init__(self, auth_url='', username='', password='',
tenant_name='', tenant_id='', insecure=False, keystone=None):
if not all([auth_url, username, password, tenant_name or tenant_id]):
raise ValueError('Please provide auth_url, username, password,'
' and tenant_id or tenant_name)')
' and tenant_id or tenant_name.')
self._keystone = keystone or ksclient.Client(username=username,
password=password,
tenant_name=tenant_name,
@@ -67,3 +88,76 @@ class KeystoneAuthV2(object):
LOG.error('Barbican endpoint not found in keystone catalog.')
raise AuthException('Barbican endpoint not found.')
return self._barbican_url
class RackspaceAuthV2(AuthPluginBase):
def __init__(self, auth_url='', username='', api_key='', password=''):
if not all([auth_url, username, api_key or password]):
raise ValueError('Please provide auth_url, username, api_key or '
'password.')
self._auth_url = auth_url
self._username = username
self._api_key = api_key
self._password = password
self._auth_token = None
self._barbican_url = None
self.tenant_id = None
self._authenticate()
@property
def auth_token(self):
return self._auth_token
@property
def barbican_url(self):
return self._barbican_url
def _authenticate(self):
auth_url = '{0}/tokens'.format(self._auth_url)
headers = {'Accept': 'application/json',
'Content-Type': 'application/json'}
if self._api_key:
payload = self._authenticate_with_api_key()
else:
payload = self._authenticate_with_password()
r = requests.post(auth_url, data=json.dumps(payload), headers=headers)
try:
r.raise_for_status()
except requests.HTTPError:
msg = 'HTTPError ({0}): Unable to authenticate with Rackspace.'
msg = msg.format(r.status_code)
LOG.error(msg)
raise AuthException(msg)
try:
data = r.json()
except ValueError:
msg = 'Error parsing response from Rackspace Identity.'
LOG.error(msg)
raise AuthException(msg)
else:
#TODO(dmend): get barbican_url from catalog
self._auth_token = data['access']['token']['id']
self.tenant_id = data['access']['token']['tenant']['id']
def _authenticate_with_api_key(self):
return {
'auth': {
'RAX-KSKEY:apiKeyCredentials': {
'username': self._username,
'apiKey': self._api_key
}
}
}
def _authenticate_with_password(self):
return {
'auth': {
'passwordCredentials': {
'username': self._username,
'password': self._password
}
}
}

100
barbicanclient/test/common/test_auth.py
View File

@@ -12,8 +12,10 @@
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import json
import mock
import requests
import unittest2 as unittest
from barbicanclient.common import auth
@@ -44,3 +46,101 @@ class WhenTestingKeystoneAuthentication(unittest.TestCase):
barbican_url = 'https://www.barbican.com'
self.keystone_auth._barbican_url = barbican_url
self.assertEquals(barbican_url, self.keystone_auth.barbican_url)
class WhenTestingRackspaceAuthentication(unittest.TestCase):
def setUp(self):
self._auth_url = 'https://auth.url.com'
self._username = 'username'
self._api_key = 'api_key'
self._auth_token = '078a50dcdc984a639bb287c8d4adf541'
self._tenant_id = '123456'
self._response = requests.Response()
self._response._content = json.dumps({
'access': {
'token': {
'id': self._auth_token,
'expires': '2013-12-19T23:06:17.047Z',
'tenant': {
'id': self._tenant_id,
'name': '123456'
}
}
}
})
patcher = mock.patch('barbicanclient.common.auth.requests.post')
self._mock_post = patcher.start()
self._mock_post.return_value = self._response
self.addCleanup(patcher.stop)
def test_auth_url_username_and_api_key_are_required(self):
with self.assertRaises(ValueError):
identity = auth.RackspaceAuthV2()
with self.assertRaises(ValueError):
identity = auth.RackspaceAuthV2(self._auth_url)
with self.assertRaises(ValueError):
identity = auth.RackspaceAuthV2(self._auth_url,
self._username)
with self.assertRaises(ValueError):
identity = auth.RackspaceAuthV2(self._auth_url,
api_key=self._api_key)
def test_tokens_is_appended_to_auth_url(self):
identity = auth.RackspaceAuthV2(self._auth_url,
self._username,
api_key=self._api_key)
self._mock_post.assert_called_with(
'https://auth.url.com/tokens',
data=mock.ANY,
headers=mock.ANY)
def test_authenticate_with_api_key(self):
with mock.patch(
'barbicanclient.common.auth.RackspaceAuthV2.'
'_authenticate_with_api_key'
) as mock_authenticate:
mock_authenticate.return_value = {}
identity = auth.RackspaceAuthV2(self._auth_url,
self._username,
api_key=self._api_key)
mock_authenticate.assert_called_once_with()
def test_authenticate_with_password(self):
with mock.patch(
'barbicanclient.common.auth.RackspaceAuthV2.'
'_authenticate_with_password'
) as mock_authenticate:
mock_authenticate.return_value = {}
identity = auth.RackspaceAuthV2(self._auth_url,
self._username,
password='password')
mock_authenticate.assert_called_once_with()
def test_auth_exception_thrown_for_bad_status(self):
self._response.status_code = 400
with self.assertRaises(auth.AuthException):
identity = auth.RackspaceAuthV2(self._auth_url,
self._username,
api_key=self._api_key)
def test_error_raised_for_bad_response_from_server(self):
self._response._content = 'Not JSON'
with self.assertRaises(auth.AuthException):
identity = auth.RackspaceAuthV2(self._auth_url,
self._username,
api_key=self._api_key)
def test_auth_token_is_set(self):
identity = auth.RackspaceAuthV2(self._auth_url,
self._username,
api_key=self._api_key)
self.assertEqual(identity.auth_token, self._auth_token)
def test_tenant_id_is_set(self):
identity = auth.RackspaceAuthV2(self._auth_url,
self._username,
api_key=self._api_key)
self.assertEqual(identity.tenant_id, self._tenant_id)