openstack
/
keystone
Code Issues Proposed changes

Implements Pluggable V3 Token Provider 

Abstract V3 token provider backend to make token provider pluggable. It enables
deployers to customize token management to add their own capabilities.
Token provider is responsible for issuing, checking, validating, and
revoking tokens. Note the distinction between token 'driver' and 'provider'.
Token 'driver' simply provides token persistence. It does not issue or
interpret tokens.

Token provider is specified by the 'provider' property in the '[token]'
section of the Keystone configuration file.

Partially implemented blueprint pluggable-token-format.

This patch also fixes bug 1186061.

Change-Id: I755fb850765ea99e5237626a2e645e6ceb42a9d3
This commit is contained in:
Guang Yee 2013-06-20 10:06:17 -07:00
parent 24a6f41405
commit c238ace309
28 changed files with 1143 additions and 557 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
doc/source/configuration.rst
View File

@ -74,7 +74,7 @@ following sections:
* ``[s3]`` - Amazon S3 authentication driver configuration.
* ``[identity]`` - identity system driver configuration
* ``[catalog]`` - service catalog driver configuration
* ``[token]`` - token driver configuration
* ``[token]`` - token driver & token provider configuration
* ``[policy]`` - policy system driver configuration for RBAC
* ``[signing]`` - cryptographic signatures for PKI based tokens
* ``[ssl]`` - SSL configuration
@ -148,6 +148,26 @@ invoked, all plugins must succeed in order to for the entire
authentication to be successful. Furthermore, all the plugins invoked must
agree on the ``user_id`` in the ``auth_context``.
Token Provider
--------------
Keystone supports customizable token provider and it is specified in the
``[token]`` section of the configuration file. Keystone provides both UUID and
PKI token providers, with PKI token provider enabled as default. However, users
may register their own token provider by configuring the following property.
* ``provider`` - token provider driver. Defaults to
``keystone.token.providers.pki.Provider``
Note that ``token_format`` in the ``[signing]`` section is deprecated but still
being supported for backward compatibility. Therefore, if ``provider`` is set
to ``keystone.token.providers.pki.Provider``, ``token_format`` must be ``PKI``.
Conversely, if ``provider`` is ``keystone.token.providers.uuid.Provider``,
``token_format`` must be ``UUID``.
For a customized provider, ``token_format`` must not set to ``PKI`` or
``UUID``.
Certificates for PKI
--------------------
@ -163,7 +183,9 @@ private key should only be readable by the system user that will run Keystone.
The values that specify where to read the certificates are under the
``[signing]`` section of the configuration file. The configuration values are:
* ``token_format`` - Determines the algorithm used to generate tokens. Can be either ``UUID`` or ``PKI``. Defaults to ``PKI``
* ``token_format`` - Determines the algorithm used to generate tokens. Can be
either ``UUID`` or ``PKI``. Defaults to ``PKI``. This option must be used in
conjunction with ``provider`` configuration in the ``[token]`` section.
* ``certfile`` - Location of certificate used to verify tokens. Default is ``/etc/keystone/ssl/certs/signing_cert.pem``
* ``keyfile`` - Location of private key used to sign tokens. Default is ``/etc/keystone/ssl/private/signing_key.pem``
* ``ca_certs`` - Location of certificate for the authority that issued the above certificate. Default is ``/etc/keystone/ssl/certs/ca.pem``

4
etc/keystone.conf.sample
View File

@ -119,8 +119,12 @@
# template_file = default_catalog.templates
[token]
# Provides token persistence.
# driver = keystone.token.backends.sql.Token
# Controls the token construction, validation, and revocation operations.
# provider = keystone.token.providers.pki.Provider
# Amount of time a token should remain valid (in seconds)
# expiration = 86400

85
keystone/auth/controllers.py
View File

@ -14,12 +14,11 @@
# License for the specific language governing permissions and limitations
# under the License.
import json
from keystone.auth import token_factory
from keystone.common import cms
from keystone.common import controller
from keystone.common import dependency
from keystone.common import logging
from keystone.common import wsgi
from keystone import config
from keystone import exception
from keystone import identity
@ -190,6 +189,10 @@ class AuthInfo(object):
self._scope_data = (None, None, trust_ref)
def _validate_auth_methods(self):
if 'identity' not in self.auth:
raise exception.ValidationError(attribute='identity',
target='auth')
# make sure auth methods are provided
if 'methods' not in self.auth['identity']:
raise exception.ValidationError(attribute='methods',
@ -267,6 +270,7 @@ class AuthInfo(object):
self._scope_data = (domain_id, project_id, trust)
@dependency.requires('token_provider_api')
class Auth(controller.V3Controller):
def __init__(self, *args, **kw):
super(Auth, self).__init__(*args, **kw)
@ -280,14 +284,22 @@ class Auth(controller.V3Controller):
auth_context = {'extras': {}, 'method_names': []}
self.authenticate(context, auth_info, auth_context)
self._check_and_set_default_scoping(auth_info, auth_context)
(token_id, token_data) = token_factory.create_token(
auth_context, auth_info)
return token_factory.render_token_data_response(
token_id, token_data, created=True)
except exception.SecurityError:
raise
except Exception as e:
LOG.exception(e)
(domain_id, project_id, trust) = auth_info.get_scope()
method_names = auth_info.get_method_names()
method_names += auth_context.get('method_names', [])
# make sure the list is unique
method_names = list(set(method_names))
(token_id, token_data) = self.token_provider_api.issue_token(
user_id=auth_context['user_id'],
method_names=method_names,
expires_at=auth_context.get('expires_at'),
project_id=project_id,
domain_id=domain_id,
auth_context=auth_context,
trust=trust)
return render_token_data_response(token_id, token_data,
created=True)
except exception.TrustNotFound as e:
raise exception.Unauthorized(e)
def _check_and_set_default_scoping(self, auth_info, auth_context):
@ -355,44 +367,41 @@ class Auth(controller.V3Controller):
msg = _('User not found')
raise exception.Unauthorized(msg)
def _get_token_ref(self, context, token_id, belongs_to=None):
token_ref = self.token_api.get_token(token_id)
if cms.is_ans1_token(token_id):
verified_token = cms.cms_verify(cms.token_to_cms(token_id),
CONF.signing.certfile,
CONF.signing.ca_certs)
token_ref = json.loads(verified_token)
if belongs_to:
assert token_ref['project']['id'] == belongs_to
return token_ref
@controller.protected
def check_token(self, context):
try:
token_id = context.get('subject_token_id')
belongs_to = context['query_string'].get('belongsTo')
assert self._get_token_ref(context, token_id, belongs_to)
except Exception as e:
LOG.error(e)
raise exception.Unauthorized(e)
token_id = context.get('subject_token_id')
self.token_provider_api.check_token(token_id)
@controller.protected
def revoke_token(self, context):
token_id = context.get('subject_token_id')
return self.token_controllers_ref.delete_token(context, token_id)
return self.token_provider_api.revoke_token(token_id)
@controller.protected
def validate_token(self, context):
token_id = context.get('subject_token_id')
self.check_token(context)
token_ref = self.token_api.get_token(token_id)
token_data = token_factory.recreate_token_data(
token_ref.get('token_data'),
token_ref['expires'],
token_ref.get('user'),
token_ref.get('tenant'))
return token_factory.render_token_data_response(token_id, token_data)
token_data = self.token_provider_api.validate_token(token_id)
return render_token_data_response(token_id, token_data)
@controller.protected
def revocation_list(self, context, auth=None):
return self.token_controllers_ref.revocation_list(context, auth)
#FIXME(gyee): not sure if it belongs here or keystone.common. Park it here
# for now.
def render_token_data_response(token_id, token_data, created=False):
"""Render token data HTTP response.
Stash token ID into the X-Subject-Token header.
"""
headers = [('X-Subject-Token', token_id)]
if created:
status = (201, 'Created')
else:
status = (200, 'OK')
return wsgi.render_response(body=token_data,
status=status, headers=headers)

12
keystone/auth/plugins/password.py
View File

@ -103,8 +103,14 @@ class Password(auth.AuthMethodHandler):
# FIXME(gyee): identity.authenticate() can use some refactoring since
# all we care is password matches
self.identity_api.authenticate(
user_id=user_info.user_id,
password=user_info.password)
try:
self.identity_api.authenticate(
user_id=user_info.user_id,
password=user_info.password)
except AssertionError:
# authentication failed because of invalid username or password
msg = _('Invalid username or password')
raise exception.Unauthorized(msg)
if 'user_id' not in user_context:
user_context['user_id'] = user_info.user_id

3
keystone/auth/plugins/token.py
View File

@ -46,7 +46,8 @@ class Token(auth.AuthMethodHandler):
token_ref['token_data']['token']['extras'])
user_context['method_names'].extend(
token_ref['token_data']['token']['methods'])
if 'trust' in token_ref['token_data']:
if ('OS-TRUST:trust' in token_ref['token_data']['token'] or
'trust' in token_ref['token_data']['token']):
raise exception.Forbidden()
except AssertionError as e:
LOG.error(e)

368
keystone/auth/token_factory.py
View File

@ -1,368 +0,0 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2013 OpenStack LLC
#
# 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.
"""Token Factory"""
import json
import sys
import uuid
import webob
from keystone import catalog
from keystone.common import cms
from keystone.common import environment
from keystone.common import logging
from keystone.common import utils
from keystone import config
from keystone import exception
from keystone import identity
from keystone.openstack.common import jsonutils
from keystone.openstack.common import timeutils
from keystone import token as token_module
from keystone import trust
CONF = config.CONF
LOG = logging.getLogger(__name__)
class TokenDataHelper(object):
"""Token data helper."""
def __init__(self):
self.identity_api = identity.Manager()
self.catalog_api = catalog.Manager()
self.trust_api = trust.Manager()
def _get_filtered_domain(self, domain_id):
domain_ref = self.identity_api.get_domain(domain_id)
return {'id': domain_ref['id'], 'name': domain_ref['name']}
def _populate_scope(self, token_data, domain_id, project_id):
if 'domain' in token_data or 'project' in token_data:
return
if domain_id:
token_data['domain'] = self._get_filtered_domain(domain_id)
if project_id:
project_ref = self.identity_api.get_project(project_id)
filtered_project = {
'id': project_ref['id'],
'name': project_ref['name']}
filtered_project['domain'] = self._get_filtered_domain(
project_ref['domain_id'])
token_data['project'] = filtered_project
def _get_project_roles_for_user(self, user_id, project_id):
roles = self.identity_api.get_roles_for_user_and_project(
user_id, project_id)
roles_ref = []
for role_id in roles:
role_ref = self.identity_api.get_role(role_id)
role_ref.setdefault('project_id', project_id)
roles_ref.append(role_ref)
# user have no project roles, therefore access denied
if len(roles_ref) == 0:
msg = _('User have no access to project')
LOG.debug(msg)
raise exception.Unauthorized(msg)
return roles_ref
def _get_domain_roles_for_user(self, user_id, domain_id):
roles = self.identity_api.get_roles_for_user_and_domain(
user_id, domain_id)
roles_ref = []
for role_id in roles:
role_ref = self.identity_api.get_role(role_id)
role_ref.setdefault('domain_id', domain_id)
roles_ref.append(role_ref)
# user have no domain roles, therefore access denied
if len(roles_ref) == 0:
msg = _('User have no access to domain')
LOG.debug(msg)
raise exception.Unauthorized(msg)
return roles_ref
def _get_roles_for_user(self, user_id, domain_id, project_id):
roles = []
if domain_id:
roles = self._get_domain_roles_for_user(user_id, domain_id)
if project_id:
roles = self._get_project_roles_for_user(user_id, project_id)
return roles
def _populate_user(self, token_data, user_id, domain_id, project_id,
trust):
if 'user' in token_data:
return
user_ref = self.identity_api.get_user(user_id)
if CONF.trust.enabled and trust:
trustor_user_ref = self.identity_api.get_user(
trust['trustor_user_id'])
if not trustor_user_ref['enabled']:
raise exception.Forbidden()
if trust['impersonation']:
user_ref = trustor_user_ref
token_data['OS-TRUST:trust'] = (
{
'id': trust['id'],
'trustor_user': {'id': trust['trustor_user_id']},
'trustee_user': {'id': trust['trustee_user_id']},
'impersonation': trust['impersonation']
})
filtered_user = {
'id': user_ref['id'],
'name': user_ref['name'],
'domain': self._get_filtered_domain(user_ref['domain_id'])}
token_data['user'] = filtered_user
def _populate_roles(self, token_data, user_id, domain_id, project_id,
trust):
if 'roles' in token_data:
return
if CONF.trust.enabled and trust:
token_user_id = trust['trustor_user_id']
token_project_id = trust['project_id']
#trusts do not support domains yet
token_domain_id = None
else:
token_user_id = user_id
token_project_id = project_id
token_domain_id = domain_id
if token_domain_id or token_project_id:
roles = self._get_roles_for_user(token_user_id,
token_domain_id,
token_project_id)
filtered_roles = []
if CONF.trust.enabled and trust:
for trust_role in trust['roles']:
match_roles = [x for x in roles
if x['id'] == trust_role['id']]
if match_roles:
filtered_roles.append(match_roles[0])
else:
raise exception.Forbidden()
else:
for role in roles:
filtered_roles.append({'id': role['id'],
'name': role['name']})
token_data['roles'] = filtered_roles
def _populate_service_catalog(self, token_data, user_id,
domain_id, project_id, trust):
if 'catalog' in token_data:
return
if CONF.trust.enabled and trust:
user_id = trust['trustor_user_id']
if project_id or domain_id:
try:
service_catalog = self.catalog_api.get_v3_catalog(
user_id, project_id)
# TODO(ayoung): KVS backend needs a sample implementation
except exception.NotImplemented:
service_catalog = {}
# TODO(gyee): v3 service catalog is not quite completed yet
# TODO(ayoung): Enforce Endpoints for trust
token_data['catalog'] = service_catalog
def _populate_token(self, token_data, expires=None, trust=None):
if not expires:
expires = token_module.default_expire_time()
if not isinstance(expires, basestring):
expires = timeutils.isotime(expires, subsecond=True)
token_data['expires_at'] = expires
token_data['issued_at'] = timeutils.isotime(subsecond=True)
def get_token_data(self, user_id, method_names, extras,
domain_id=None, project_id=None, expires=None,
trust=None, token=None):
token_data = {'methods': method_names,
'extras': extras}
# We've probably already written these to the token
for x in ('roles', 'user', 'catalog', 'project', 'domain'):
if token and x in token:
token_data[x] = token[x]
if CONF.trust.enabled and trust:
if user_id != trust['trustee_user_id']:
raise exception.Forbidden()
self._populate_scope(token_data, domain_id, project_id)
self._populate_user(token_data, user_id, domain_id, project_id, trust)
self._populate_roles(token_data, user_id, domain_id, project_id, trust)
self._populate_service_catalog(token_data, user_id, domain_id,
project_id, trust)
self._populate_token(token_data, expires, trust)
return {'token': token_data}
def recreate_token_data(token_data=None, expires=None,
user_ref=None, project_ref=None):
"""Recreate token from an existing token.
Repopulate the ephemeral data and return the new token data.
"""
new_expires = expires
project_id = None
user_id = None
domain_id = None
methods = ['password', 'token']
extras = {}
# NOTE(termie): Let's get some things straight here, because this code
# is wrong but tested as such:
# token_data, if it exists, is going to look like:
# {'token': ... the actual token data + a superfluous extras field ...}
# this data is actually stored in the database in the 'extras' column and
# then deserialized and added to the token_ref, that already has the
# the 'expires', 'user_id', and 'id' columns from the db.
# the 'user' and 'tenant' fields are being added to the
# token_ref due to being deserialized from the 'extras' column
#
# So, how this all looks in the db:
# id = some_id
# user_id = some_user_id
# expires = some_expiration
# extras = {'user': {'id': some_used_id},
# 'tenant': {'id': some_tenant_id},
# 'token_data': 'token': {'domain': {'id': some_domain_id},
# 'project': {'id': some_project_id},
# 'domain': {'id': some_domain_id},
# 'user': {'id': some_user_id},
# 'roles': [{'id': some_role_id}, ...],
# 'catalog': ...,
# 'expires_at': some_expiry_time,
# 'issued_at': now(),
# 'methods': ['password', 'token'],
# 'extras': { ... empty? ...}
#
# TODO(termie): reduce stored token complexity, bug filed at:
# https://bugs.launchpad.net/keystone/+bug/1159990
if token_data:
# peel the outer layer so its easier to operate
token = token_data['token']
domain_id = (token['domain']['id'] if 'domain' in token
else None)
project_id = (token['project']['id'] if 'project' in token
else None)
if not new_expires:
# support Grizzly-3 to Grizzly-RC1 transition
# tokens issued in G3 has 'expires' instead of 'expires_at'
new_expires = token.get('expires_at',
token.get('expires'))
user_id = token['user']['id']
methods = token['methods']
extras = token['extras']
else:
token = None
project_id = project_ref['id'] if project_ref else None
user_id = user_ref['id']
token_data_helper = TokenDataHelper()
return token_data_helper.get_token_data(user_id,
methods,
extras,
domain_id,
project_id,
new_expires,
token=token)
def create_token(auth_context, auth_info):
token_data_helper = TokenDataHelper()
(domain_id, project_id, trust) = auth_info.get_scope()
method_names = list(set(auth_info.get_method_names() +
auth_context.get('method_names', [])))
token_data = token_data_helper.get_token_data(
auth_context['user_id'],
method_names,
auth_context['extras'],
domain_id,
project_id,
auth_context.get('expires_at', None),
trust)
if CONF.signing.token_format == 'UUID':
token_id = uuid.uuid4().hex
elif CONF.signing.token_format == 'PKI':
try:
token_id = cms.cms_sign_token(json.dumps(token_data),
CONF.signing.certfile,
CONF.signing.keyfile)
except environment.subprocess.CalledProcessError:
raise exception.UnexpectedError(_(
'Unable to sign token.'))
else:
raise exception.UnexpectedError(_(
'Invalid value for token_format: %s.'
' Allowed values are PKI or UUID.') %
CONF.signing.token_format)
token_api = token_module.Manager()
try:
expiry = token_data['token']['expires_at']
if isinstance(expiry, basestring):
expiry = timeutils.normalize_time(timeutils.parse_isotime(expiry))
role_ids = []
if 'project' in token_data['token']:
# project-scoped token, fill in the v2 token data
# all we care are the role IDs
role_ids = [role['id'] for role in token_data['token']['roles']]
metadata_ref = {'roles': role_ids}
data = dict(key=token_id,
id=token_id,
expires=expiry,
user=token_data['token']['user'],
tenant=token_data['token'].get('project'),
metadata=metadata_ref,
token_data=token_data,
trust_id=trust['id'] if trust else None)
token_api.create_token(token_id, data)
except Exception:
exc_info = sys.exc_info()
# an identical token may have been created already.
# if so, return the token_data as it is also identical
try:
token_api.get_token(token_id)
except exception.TokenNotFound:
raise exc_info[0], exc_info[1], exc_info[2]
return (token_id, token_data)
def render_token_data_response(token_id, token_data, created=False):
"""Render token data HTTP response.
Stash token ID into the X-Auth-Token header.
"""
headers = [('X-Subject-Token', token_id)]
headers.append(('Vary', 'X-Auth-Token'))
headers.append(('Content-Type', 'application/json'))
if created:
status = (201, 'Created')
else:
status = (200, 'OK')
body = jsonutils.dumps(token_data, cls=utils.SmarterEncoder)
return webob.Response(body=body,
status='%s %s' % status,
headerlist=headers)

6
keystone/common/config.py
View File

@ -399,3 +399,9 @@ def configure():
# PasteDeploy config file
register_str('config_file', group='paste_deploy', default=None)
# token provider
register_str(
'provider',
group='token',
default='keystone.token.providers.pki.Provider')

7
keystone/common/sql/core.py
View File

@ -80,13 +80,6 @@ def initialize_decorator(init):
v = str(v)
if column.type.length and \
column.type.length < len(v):
#if signing.token_format == 'PKI', the id will
#store it's public key which is very long.
if config.CONF.signing.token_format == 'PKI' and \
self.__tablename__ == 'token' and \
k == 'id':
continue
raise exception.StringLengthExceeded(
string=v, type=k, length=column.type.length)

3
keystone/service.py
View File

@ -41,7 +41,8 @@ DRIVERS = dict(
identity_api=identity.Manager(),
policy_api=policy.Manager(),
token_api=token.Manager(),
trust_api=trust.Manager())
trust_api=trust.Manager(),
token_provider_api=token.provider.Manager())
@logging.fail_gracefully

4
keystone/test.py
View File

@ -252,8 +252,8 @@ class TestCase(NoModule, unittest.TestCase):
def load_backends(self):
"""Initializes each manager and assigns them to an attribute."""
for manager in [assignment, catalog, credential,
identity, policy, token, trust]:
for manager in [assignment, catalog, credential, identity, policy,
token, trust]:
manager_name = '%s_api' % manager.__name__.split('.')[-1]
setattr(self, manager_name, manager.Manager())

1
keystone/token/__init__.py
View File

@ -17,4 +17,5 @@
from keystone.token import controllers
from keystone.token.core import *
from keystone.token import provider
from keystone.token import routers

3
keystone/token/backends/kvs.py
View File

@ -26,7 +26,6 @@ class Token(kvs.Base, token.Driver):
# Public interface
def get_token(self, token_id):
token_id = token.unique_id(token_id)
try:
ref = self.db.get('token-%s' % token_id)
except exception.NotFound:
@ -41,7 +40,6 @@ class Token(kvs.Base, token.Driver):
raise exception.TokenNotFound(token_id=token_id)
def create_token(self, token_id, data):
token_id = token.unique_id(token_id)
data_copy = copy.deepcopy(data)
data_copy['id'] = token_id
if not data_copy.get('expires'):
@ -52,7 +50,6 @@ class Token(kvs.Base, token.Driver):
return copy.deepcopy(data_copy)
def delete_token(self, token_id):
token_id = token.unique_id(token_id)
try:
token_ref = self.get_token(token_id)
self.db.delete('token-%s' % token_id)

12
keystone/token/backends/memcache.py
View File

@ -65,7 +65,7 @@ class Token(token.Driver):
def get_token(self, token_id):
if token_id is None:
raise exception.TokenNotFound(token_id='')
ptk = self._prefix_token_id(token.unique_id(token_id))
ptk = self._prefix_token_id(token_id)
token_ref = self.client.get(ptk)
if token_ref is None:
raise exception.TokenNotFound(token_id=token_id)
@ -74,7 +74,7 @@ class Token(token.Driver):
def create_token(self, token_id, data):
data_copy = copy.deepcopy(data)
ptk = self._prefix_token_id(token.unique_id(token_id))
ptk = self._prefix_token_id(token_id)
if not data_copy.get('expires'):
data_copy['expires'] = token.default_expire_time()
if not data_copy.get('user_id'):
@ -118,7 +118,7 @@ class Token(token.Driver):
if record is not None:
token_list = jsonutils.loads('[%s]' % record)
for token_i in token_list:
ptk = self._prefix_token_id(token.unique_id(token_i))
ptk = self._prefix_token_id(token_i)
token_ref = self.client.get(ptk)
if not token_ref:
# skip tokens that do not exist in memcache
@ -174,8 +174,8 @@ class Token(token.Driver):
def delete_token(self, token_id):
# Test for existence
data = self.get_token(token.unique_id(token_id))
ptk = self._prefix_token_id(token.unique_id(token_id))
data = self.get_token(token_id)
ptk = self._prefix_token_id(token_id)
result = self.client.delete(ptk)
self._add_to_revocation_list(data)
return result
@ -186,7 +186,7 @@ class Token(token.Driver):
user_record = self.client.get(user_key) or ""
token_list = jsonutils.loads('[%s]' % user_record)
for token_id in token_list:
ptk = self._prefix_token_id(token.unique_id(token_id))
ptk = self._prefix_token_id(token_id)
token_ref = self.client.get(ptk)
if token_ref:
if tenant_id is not None:

6
keystone/token/backends/sql.py
View File

@ -41,7 +41,7 @@ class Token(sql.Base, token.Driver):
if token_id is None:
raise exception.TokenNotFound(token_id=token_id)
session = self.get_session()
token_ref = session.query(TokenModel).get(token.unique_id(token_id))
token_ref = session.query(TokenModel).get(token_id)
now = datetime.datetime.utcnow()
if not token_ref or not token_ref.valid:
raise exception.TokenNotFound(token_id=token_id)
@ -59,7 +59,6 @@ class Token(sql.Base, token.Driver):
data_copy['user_id'] = data_copy['user']['id']
token_ref = TokenModel.from_dict(data_copy)
token_ref.id = token.unique_id(token_id)
token_ref.valid = True
session = self.get_session()
with session.begin():
@ -69,9 +68,8 @@ class Token(sql.Base, token.Driver):
def delete_token(self, token_id):
session = self.get_session()
key = token.unique_id(token_id)
with session.begin():
token_ref = session.query(TokenModel).get(key)
token_ref = session.query(TokenModel).get(token_id)
if not token_ref or not token_ref.valid:
raise exception.TokenNotFound(token_id=token_id)
token_ref.valid = False

37
keystone/token/core.py
View File

@ -16,6 +16,7 @@
"""Main entry point into the Token service."""
import copy
import datetime
from keystone.common import cms
@ -32,19 +33,6 @@ config.register_int('expiration', group='token', default=86400)
LOG = logging.getLogger(__name__)
def unique_id(token_id):
"""Return a unique ID for a token.
The returned value is useful as the primary key of a database table,
memcache store, or other lookup table.
:returns: Given a PKI token, returns it's hashed value. Otherwise, returns
the passed-in value (such as a UUID token ID or an existing
hash).
"""
return cms.cms_hash_token(token_id)
def default_expire_time():
"""Determine when a fresh token should expire.
@ -114,6 +102,29 @@ class Manager(manager.Manager):
def __init__(self):
super(Manager, self).__init__(CONF.token.driver)
def _unique_id(self, token_id):
"""Return a unique ID for a token.
The returned value is useful as the primary key of a database table,
memcache store, or other lookup table.
:returns: Given a PKI token, returns it's hashed value. Otherwise,
returns the passed-in value (such as a UUID token ID or an
existing hash).
"""
return cms.cms_hash_token(token_id)
def get_token(self, token_id):
return self.driver.get_token(self._unique_id(token_id))
def create_token(self, token_id, data):
data_copy = copy.deepcopy(data)
data_copy['id'] = self._unique_id(token_id)
return self.driver.create_token(self._unique_id(token_id), data_copy)
def delete_token(self, token_id):
return self.driver.delete_token(self._unique_id(token_id))
class Driver(object):
"""Interface description for a Token driver."""

136
keystone/token/provider.py Normal file
View File

@ -0,0 +1,136 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2012 OpenStack LLC
#
# 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.
"""Token provider interface."""
from keystone.common import dependency
from keystone.common import logging
from keystone.common import manager
from keystone import config
from keystone import exception
CONF = config.CONF
LOG = logging.getLogger(__name__)
# supported token versions
V2 = 'v2.0'
V3 = 'v3.0'
class UnsupportedTokenVersionException(Exception):
"""Token version is unrecognizable or unsupported."""
pass
@dependency.provider('token_provider_api')
class Manager(manager.Manager):
"""Default pivot point for the token provider backend.
See :mod:`keystone.common.manager.Manager` for more details on how this
dynamically calls the backend.
"""
def __init__(self):
# FIXME(gyee): we are deprecating CONF.signing.token_format. This code
# is to ensure the token provider configuration agrees with
# CONF.signing.token_format.
if ((CONF.signing.token_format == 'PKI' and
not CONF.token.provider.endswith('.pki.Provider')) or
(CONF.signing.token_format == 'UUID' and
not CONF.token.provider.endswith('uuid.Provider'))):
raise ValueError('token_format conflicts with token provider')
super(Manager, self).__init__(CONF.token.provider)
class Provider(object):
"""Interface description for a Token provider."""
def get_token_version(self, token_data):
"""Return the version of the given token data.
If the given token data is unrecognizable,
UnsupportedTokenVersionException is raised.
"""
raise exception.NotImplemented()
def issue_token(self, version='v3.0', **kwargs):
"""Issue a V3 token.
For V3 tokens, 'user_id', 'method_names', must present in kwargs.
Optionally, kwargs may contain 'expires_at' for rescope tokens;
'project_id' for project-scoped token; 'domain_id' for
domain-scoped token; and 'auth_context' from the authentication
plugins.
:param context: request context
:type context: dictionary
:param version: version of the token to be issued
:type version: string
:param kwargs: information needed for token creation. Parameters
may be different depending on token version.
:type kwargs: dictionary
:returns: (token_id, token_data)
"""
raise exception.NotImplemented()
def revoke_token(self, token_id):
"""Revoke a given token.
:param token_id: identity of the token
:type token_id: string
:returns: None.
"""
raise exception.NotImplemented()
def validate_token(self, token_id, belongs_to=None, version='v3.0'):
"""Validate the given token and return the token data.
Must raise Unauthorized exception if unable to validate token.
:param token_id: identity of the token
:type token_id: string
:param belongs_to: identity of the scoped project to validate
:type belongs_to: string
:param version: version of the token to be validated
:type version: string
:returns: token data
:raises: keystone.exception.Unauthorized
"""
raise exception.NotImplemented()
def check_token(self, token_id, belongs_to=None, version='v3.0'):
"""Check the validity of the given V3 token.
Must raise Unauthorized exception if unable to check token.
:param token_id: identity of the token
:type token_id: string
:param belongs_to: identity of the scoped project to validate
:type belongs_to: string
:param version: version of the token to check
:type version: string
:returns: None
:raises: keystone.exception.Unauthorized
"""

0
keystone/token/providers/__init__.py Normal file
View File

44
keystone/token/providers/pki.py Normal file
View File

@ -0,0 +1,44 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2013 OpenStack LLC
#
# 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.
"""Keystone PKI Token Provider"""
import json
from keystone.common import cms
from keystone.common import environment
from keystone.common import logging
from keystone import config
from keystone import exception
from keystone.token.providers import uuid
CONF = config.CONF
LOG = logging.getLogger(__name__)
class Provider(uuid.Provider):
def _get_token_id(self, token_data):
try:
token_id = cms.cms_sign_token(json.dumps(token_data),
CONF.signing.certfile,
CONF.signing.keyfile)
return token_id
except environment.subprocess.CalledProcessError:
LOG.exception('Unable to sign token')
raise exception.UnexpectedError(_(
'Unable to sign token.'))

349
keystone/token/providers/uuid.py Normal file
View File

@ -0,0 +1,349 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2013 OpenStack LLC
#
# 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.
"""Keystone UUID Token Provider"""
from __future__ import absolute_import
import sys
import uuid
from keystone.common import dependency
from keystone.common import logging
from keystone import config
from keystone import exception
from keystone.openstack.common import timeutils
from keystone import token
from keystone.token import provider as token_provider
from keystone import trust
LOG = logging.getLogger(__name__)
CONF = config.CONF
DEFAULT_DOMAIN_ID = CONF.identity.default_domain_id
@dependency.requires('catalog_api', 'identity_api')
class V3TokenDataHelper(object):
"""Token data helper."""
def __init__(self):
if CONF.trust.enabled:
self.trust_api = trust.Manager()
def _get_filtered_domain(self, domain_id):
domain_ref = self.identity_api.get_domain(domain_id)
return {'id': domain_ref['id'], 'name': domain_ref['name']}
def _get_filtered_project(self, project_id):
project_ref = self.identity_api.get_project(project_id)
filtered_project = {
'id': project_ref['id'],
'name': project_ref['name']}
filtered_project['domain'] = self._get_filtered_domain(
project_ref['domain_id'])
return filtered_project
def _populate_scope(self, token_data, domain_id, project_id):
if 'domain' in token_data or 'project' in token_data:
# scope already exist, no need to populate it again
return
if domain_id:
token_data['domain'] = self._get_filtered_domain(domain_id)
if project_id:
token_data['project'] = self._get_filtered_project(project_id)
def _get_roles_for_user(self, user_id, domain_id, project_id):
roles = []
if domain_id:
roles = self.identity_api.get_roles_for_user_and_domain(
user_id, domain_id)
if project_id:
roles = self.identity_api.get_roles_for_user_and_project(
user_id, project_id)
return [self.identity_api.get_role(role_id) for role_id in roles]
def _populate_user(self, token_data, user_id, domain_id, project_id,
trust):
if 'user' in token_data:
# no need to repopulate user if it already exists
return
user_ref = self.identity_api.get_user(user_id)
if CONF.trust.enabled and trust and 'OS-TRUST:trust' not in token_data:
trustor_user_ref = (self.identity_api.get_user(
trust['trustor_user_id']))
if not trustor_user_ref['enabled']:
raise exception.Forbidden(_('Trustor is disabled.'))
if trust['impersonation']:
user_ref = trustor_user_ref
token_data['OS-TRUST:trust'] = (
{
'id': trust['id'],
'trustor_user': {'id': trust['trustor_user_id']},
'trustee_user': {'id': trust['trustee_user_id']},
'impersonation': trust['impersonation']
})
filtered_user = {
'id': user_ref['id'],
'name': user_ref['name'],
'domain': self._get_filtered_domain(user_ref['domain_id'])}
token_data['user'] = filtered_user
def _populate_roles(self, token_data, user_id, domain_id, project_id,
trust):
if 'roles' in token_data:
# no need to repopulate roles
return
if CONF.trust.enabled and trust:
token_user_id = trust['trustor_user_id']
token_project_id = trust['project_id']
#trusts do not support domains yet
token_domain_id = None
else:
token_user_id = user_id
token_project_id = project_id
token_domain_id = domain_id
if token_domain_id or token_project_id:
roles = self._get_roles_for_user(token_user_id,
token_domain_id,
token_project_id)
filtered_roles = []
if CONF.trust.enabled and trust:
for trust_role in trust['roles']:
match_roles = [x for x in roles
if x['id'] == trust_role['id']]
if match_roles:
filtered_roles.append(match_roles[0])
else:
raise exception.Forbidden(
_('Trustee have no delegated roles.'))
else:
for role in roles:
filtered_roles.append({'id': role['id'],
'name': role['name']})
# user has no project or domain roles, therefore access denied
if not filtered_roles:
if token_project_id:
msg = _('User %(user_id)s have no access '
'to project %(project_id)s') % {
'user_id': user_id,
'project_id': token_project_id}
else:
msg = _('User %(user_id)s have no access '
'to domain %(domain_id)s') % {
'user_id': user_id,
'domain_id': token_domain_id}
LOG.debug(msg)
raise exception.Unauthorized(msg)
token_data['roles'] = filtered_roles
def _populate_service_catalog(self, token_data, user_id,
domain_id, project_id, trust):
if 'catalog' in token_data:
# no need to repopulate service catalog
return
if CONF.trust.enabled and trust:
user_id = trust['trustor_user_id']
if project_id or domain_id:
try:
service_catalog = self.catalog_api.get_v3_catalog(
user_id, project_id)
# TODO(ayoung): KVS backend needs a sample implementation
except exception.NotImplemented:
service_catalog = {}
# TODO(gyee): v3 service catalog is not quite completed yet
# TODO(ayoung): Enforce Endpoints for trust
token_data['catalog'] = service_catalog
def _populate_token_dates(self, token_data, expires=None, trust=None):
if not expires:
expires = token.default_expire_time()
if not isinstance(expires, basestring):
expires = timeutils.isotime(expires, subsecond=True)
token_data['expires_at'] = expires
token_data['issued_at'] = timeutils.isotime(subsecond=True)
def get_token_data(self, user_id, method_names, extras,
domain_id=None, project_id=None, expires=None,
trust=None, token=None):
token_data = {'methods': method_names,
'extras': extras}
# We've probably already written these to the token
if token:
for x in ('roles', 'user', 'catalog', 'project', 'domain'):
if x in token:
token_data[x] = token[x]
if CONF.trust.enabled and trust:
if user_id != trust['trustee_user_id']:
raise exception.Forbidden(_('User is not a trustee.'))
self._populate_scope(token_data, domain_id, project_id)
self._populate_user(token_data, user_id, domain_id, project_id, trust)
self._populate_roles(token_data, user_id, domain_id, project_id, trust)
self._populate_service_catalog(token_data, user_id, domain_id,
project_id, trust)
self._populate_token_dates(token_data, expires=expires, trust=trust)
return {'token': token_data}
@dependency.requires('token_api', 'identity_api')
class Provider(token_provider.Provider):
def __init__(self, *args, **kwargs):
super(Provider, self).__init__(*args, **kwargs)
if CONF.trust.enabled:
self.trust_api = trust.Manager()
self.v3_token_data_helper = V3TokenDataHelper()
def get_token_version(self, token_data):
if token_data and isinstance(token_data, dict):
if 'access' in token_data:
return token_provider.V2
if 'token' in token_data and 'methods' in token_data['token']:
return token_provider.V3
raise token_provider.UnsupportedTokenVersionException()
def _get_token_id(self, token_data):
return uuid.uuid4().hex
def _issue_v3_token(self, **kwargs):
user_id = kwargs.get('user_id')
method_names = kwargs.get('method_names')
expires_at = kwargs.get('expires_at')
project_id = kwargs.get('project_id')
domain_id = kwargs.get('domain_id')
auth_context = kwargs.get('auth_context')
trust = kwargs.get('trust')
metadata_ref = kwargs.get('metadata_ref')
# for V2, trust is stashed in metadata_ref
if (CONF.trust.enabled and not trust and metadata_ref and
'trust_id' in metadata_ref):
trust = self.trust_api.get_trust(metadata_ref['trust_id'])
token_data = self.v3_token_data_helper.get_token_data(
user_id,
method_names,
auth_context.get('extras') if auth_context else None,
domain_id=domain_id,
project_id=project_id,
expires=expires_at,
trust=trust)
token_id = self._get_token_id(token_data)
try:
expiry = token_data['token']['expires_at']
if isinstance(expiry, basestring):
expiry = timeutils.normalize_time(
timeutils.parse_isotime(expiry))
# FIXME(gyee): is there really a need to store roles in metadata?
role_ids = []
metadata_ref = kwargs.get('metadata_ref', {})
if 'project' in token_data['token']:
# project-scoped token, fill in the v2 token data
# all we care are the role IDs
role_ids = [r['id'] for r in token_data['token']['roles']]
metadata_ref = {'roles': role_ids}
if trust:
metadata_ref.setdefault('trust_id', trust['id'])
metadata_ref.setdefault('trustee_user_id',
trust['trustee_user_id'])
data = dict(key=token_id,
id=token_id,
expires=expiry,
user=token_data['token']['user'],
tenant=token_data['token'].get('project'),
metadata=metadata_ref,
token_data=token_data,
trust_id=trust['id'] if trust else None)
self.token_api.create_token(token_id, data)
except Exception:
exc_info = sys.exc_info()
# an identical token may have been created already.
# if so, return the token_data as it is also identical
try:
self.token_api.get_token(token_id)
except exception.TokenNotFound:
raise exc_info[0], exc_info[1], exc_info[2]
return (token_id, token_data)
def issue_token(self, version='v3.0', **kwargs):
if version == token_provider.V3:
return self._issue_v3_token(**kwargs)
raise token_provider.UnsupportedTokenVersionException
def _verify_token(self, token_id, belongs_to=None):
"""Verify the given token and return the token_ref."""
token_ref = self.token_api.get_token(token_id=token_id)
assert token_ref
if belongs_to:
assert token_ref['tenant']['id'] == belongs_to
return token_ref
def revoke_token(self, token_id):
self.token_api.delete_token(token_id=token_id)
def _validate_v3_token(self, token_id):
token_ref = self._verify_token(token_id)
# FIXME(gyee): performance or correctness? Should we return the
# cached token or reconstruct it? Obviously if we are going with
# the cached token, any role, project, or domain name changes
# will not be reflected. One may argue that with PKI tokens,
# we are essentially doing cached token validation anyway.
# Lets go with the cached token strategy. Since token
# management layer is now pluggable, one can always provide
# their own implementation to suit their needs.
token_data = token_ref.get('token_data')
if not token_data or 'token' not in token_data:
# token ref is created by V2 API
project_id = None
project_ref = token_ref.get('tenant')
if project_ref:
project_id = project_ref['id']
token_data = self.v3_token_data_helper.get_token_data(
token_ref['user']['id'],
['password', 'token'],
{},
project_id=project_id,
expires=token_ref['expires'])
return token_data
def validate_token(self, token_id, belongs_to=None,
version='v3.0'):
try:
if version == token_provider.V3:
return self._validate_v3_token(token_id)
raise token_provider.UnsupportedTokenVersionException()
except exception.TokenNotFound as e:
LOG.exception(_('Failed to verify token'))
raise exception.Unauthorized(e)
def check_token(self, token_id, belongs_to=None,
version='v3.0', **kwargs):
try:
if version == token_provider.V3:
self._verify_token(token_id)
else:
raise token_provider.UnsupportedTokenVersionException()
except exception.TokenNotFound as e:
LOG.exception(_('Failed to verify token'))
raise exception.Unauthorized(e)

14
tests/test_auth.py
View File

@ -68,6 +68,10 @@ class AuthTest(test.TestCase):
self.load_backends()
self.load_fixtures(default_fixtures)
# need to register the token provider first because auth controller
# depends on it
token.provider.Manager()
self.controller = token.controllers.Auth()
def assertEqualTokens(self, a, b):
@ -653,12 +657,12 @@ class AuthWithTrust(AuthTest):
def test_v3_trust_token_get_token_fails(self):
auth_response = self.fetch_v3_token_from_trust()
trust_token = auth_response.headers['X-Subject-Token']
v3_token_data = {
"methods": ["token"],
"token": {"id": trust_token}
}
v3_token_data = {'identity': {
'methods': ['token'],
'token': {'id': trust_token}
}}
self.assertRaises(
exception.Unauthorized,
exception.Forbidden,
self.auth_v3_controller.authenticate_for_token,
{}, v3_token_data)

6
tests/test_auth_plugin.py
View File

@ -20,6 +20,7 @@ from keystone import test
from keystone import auth
from keystone import exception
from keystone import token
# for testing purposes only
@ -49,6 +50,11 @@ class TestAuthPlugin(test.TestCase):
test.testsdir('test_auth_plugin.conf')])
self.load_backends()
auth.controllers.AUTH_METHODS[METHOD_NAME] = SimpleChallengeResponse()
# need to register the token provider first because auth controller
# depends on it
token.provider.Manager()
self.api = auth.controllers.Auth()
def test_unsupported_auth_method(self):

2
tests/test_backend_memcache.py
View File

@ -164,7 +164,7 @@ class MemcacheToken(test.TestCase, test_backend.TokenTests):
user_token_list = jsonutils.loads('[%s]' % user_record)
self.assertEquals(len(user_token_list), 2)
expired_token_ptk = self.token_api.driver._prefix_token_id(
token.unique_id(expired_token_id))
expired_token_id)
expired_token = self.token_api.driver.client.get(expired_token_ptk)
expired_token['expires'] = (timeutils.utcnow() - expire_delta)
self.token_api.driver.client.set(expired_token_ptk, expired_token)

1
tests/test_cert_setup.py
View File

@ -61,7 +61,6 @@ class CertSetupTestCase(test.TestCase):
self.controller = token.controllers.Auth()
def test_can_handle_missing_certs(self):
self.opt_in_group('signing', token_format='PKI')
self.opt_in_group('signing', certfile='invalid')
user = {
'id': 'fake1',

5
tests/test_pki_token_provider.conf Normal file
View File

@ -0,0 +1,5 @@
[signing]
token_format = PKI
[token]
provider = keystone.token.providers.pki.Provider

396
tests/test_token_provider.py Normal file
View File

@ -0,0 +1,396 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2013 OpenStack LLC
#
# 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 uuid
from keystone import test
from keystone import token
SAMPLE_V2_TOKEN = {
"access": {
"trust": {
"id": "abc123",
"trustee_user_id": "123456"
},
"serviceCatalog": [
{
"endpoints": [
{
"adminURL": "http://localhost:8774/v1.1/01257",
"id": "51934fe63a5b4ac0a32664f64eb462c3",
"internalURL": "http://localhost:8774/v1.1/01257",
"publicURL": "http://localhost:8774/v1.1/01257",
"region": "RegionOne"
}
],
"endpoints_links": [],
"name": "nova",
"type": "compute"
},
{
"endpoints": [
{
"adminURL": "http://localhost:9292",
"id": "aaa17a539e364297a7845d67c7c7cc4b",
"internalURL": "http://localhost:9292",
"publicURL": "http://localhost:9292",
"region": "RegionOne"
}
],
"endpoints_links": [],
"name": "glance",
"type": "image"
},
{
"endpoints": [
{
"adminURL": "http://localhost:8776/v1/01257",
"id": "077d82df25304abeac2294004441db5a",
"internalURL": "http://localhost:8776/v1/01257",
"publicURL": "http://localhost:8776/v1/01257",
"region": "RegionOne"
}
],
"endpoints_links": [],
"name": "volume",
"type": "volume"
},
{
"endpoints": [
{
"adminURL": "http://localhost:8773/services/Admin",
"id": "b06997fd08414903ad458836efaa9067",
"internalURL": "http://localhost:8773/services/Cloud",
"publicURL": "http://localhost:8773/services/Cloud",
"region": "RegionOne"
}
],
"endpoints_links": [],
"name": "ec2",
"type": "ec2"
},
{
"endpoints": [
{
"adminURL": "http://localhost:8888/v1",
"id": "7bd0c643e05a4a2ab40902b2fa0dd4e6",
"internalURL": "http://localhost:8888/v1/AUTH_01257",
"publicURL": "http://localhost:8888/v1/AUTH_01257",
"region": "RegionOne"
}
],
"endpoints_links": [],
"name": "swift",
"type": "object-store"
},
{
"endpoints": [
{
"adminURL": "http://localhost:35357/v2.0",
"id": "02850c5d1d094887bdc46e81e1e15dc7",
"internalURL": "http://localhost:5000/v2.0",
"publicURL": "http://localhost:5000/v2.0",
"region": "RegionOne"
}
],
"endpoints_links": [],
"name": "keystone",
"type": "identity"
}
],
"token": {
"expires": "2013-05-22T00:02:43.941430Z",
"id": "ce4fc2d36eea4cc9a36e666ac2f1029a",
"issued_at": "2013-05-21T00:02:43.941473Z",
"tenant": {
"enabled": True,
"id": "01257",
"name": "service"
}
},
"user": {
"id": "f19ddbe2c53c46f189fe66d0a7a9c9ce",
"name": "nova",
"roles": [
{
"name": "_member_"
},
{
"name": "admin"
}
],
"roles_links": [],
"username": "nova"
}
}
}
SAMPLE_V3_TOKEN = {
"token": {
"catalog": [
{
"endpoints": [
{
"id": "02850c5d1d094887bdc46e81e1e15dc7",
"interface": "admin",
"region": "RegionOne",
"url": "http://localhost:35357/v2.0"
},
{
"id": "446e244b75034a9ab4b0811e82d0b7c8",
"interface": "internal",
"region": "RegionOne",
"url": "http://localhost:5000/v2.0"
},
{
"id": "47fa3d9f499240abb5dfcf2668f168cd",
"interface": "public",
"region": "RegionOne",
"url": "http://localhost:5000/v2.0"
}
],
"id": "26d7541715a44a4d9adad96f9872b633",
"type": "identity",
},
{
"endpoints": [
{
"id": "aaa17a539e364297a7845d67c7c7cc4b",
"interface": "admin",
"region": "RegionOne",
"url": "http://localhost:9292"
},
{
"id": "4fa9620e42394cb1974736dce0856c71",
"interface": "internal",
"region": "RegionOne",
"url": "http://localhost:9292"
},
{
"id": "9673687f9bc441d88dec37942bfd603b",
"interface": "public",
"region": "RegionOne",
"url": "http://localhost:9292"
}
],
"id": "d27a41843f4e4b0e8cf6dac4082deb0d",
"type": "image",
},
{
"endpoints": [
{
"id": "7bd0c643e05a4a2ab40902b2fa0dd4e6",
"interface": "admin",
"region": "RegionOne",
"url": "http://localhost:8888/v1"
},
{
"id": "43bef154594d4ccb8e49014d20624e1d",
"interface": "internal",
"region": "RegionOne",
"url": "http://localhost:8888/v1/AUTH_01257"
},
{
"id": "e63b5f5d7aa3493690189d0ff843b9b3",
"interface": "public",
"region": "RegionOne",
"url": "http://localhost:8888/v1/AUTH_01257"
}
],
"id": "a669e152f1104810a4b6701aade721bb",
"type": "object-store",
},
{
"endpoints": [
{
"id": "51934fe63a5b4ac0a32664f64eb462c3",
"interface": "admin",
"region": "RegionOne",
"url": "http://localhost:8774/v1.1/01257"
},
{
"id": "869b535eea0d42e483ae9da0d868ebad",
"interface": "internal",
"region": "RegionOne",
"url": "http://localhost:8774/v1.1/01257"
},
{
"id": "93583824c18f4263a2245ca432b132a6",
"interface": "public",
"region": "RegionOne",
"url": "http://localhost:8774/v1.1/01257"
}
],
"id": "7f32cc2af6c9476e82d75f80e8b3bbb8",
"type": "compute",
},
{
"endpoints": [
{
"id": "b06997fd08414903ad458836efaa9067",
"interface": "admin",
"region": "RegionOne",
"url": "http://localhost:8773/services/Admin"
},
{
"id": "411f7de7c9a8484c9b46c254fb2676e2",
"interface": "internal",
"region": "RegionOne",
"url": "http://localhost:8773/services/Cloud"
},
{
"id": "f21c93f3da014785854b4126d0109c49",
"interface": "public",
"region": "RegionOne",
"url": "http://localhost:8773/services/Cloud"
}
],
"id": "b08c9c7d4ef543eba5eeb766f72e5aa1",
"type": "ec2",
},
{
"endpoints": [
{
"id": "077d82df25304abeac2294004441db5a",
"interface": "admin",
"region": "RegionOne",
"url": "http://localhost:8776/v1/01257"
},
{
"id": "875bf282362c40219665278b4fd11467",
"interface": "internal",
"region": "RegionOne",
"url": "http://localhost:8776/v1/01257"
},
{
"id": "cd229aa6df0640dc858a8026eb7e640c",
"interface": "public",
"region": "RegionOne",
"url": "http://localhost:8776/v1/01257"
}
],
"id": "5db21b82617f4a95816064736a7bec22",
"type": "volume",
}
],
"expires_at": "2013-05-22T00:02:43.941430Z",
"issued_at": "2013-05-21T00:02:43.941473Z",
"methods": [
"password"
],
"project": {
"domain": {
"id": "default",
"name": "Default"
},
"id": "01257",
"name": "service"
},
"roles": [
{
"id": "9fe2ff9ee4384b1894a90878d3e92bab",
"name": "_member_"
},
{
"id": "53bff13443bd4450b97f978881d47b18",
"name": "admin"
}
],
"user": {
"domain": {
"id": "default",
"name": "Default"
},
"id": "f19ddbe2c53c46f189fe66d0a7a9c9ce",
"name": "nova"
},
"OS-TRUST:trust": {
"id": "abc123",
"trustee_user_id": "123456",
"trustor_user_id": "333333",
"impersonation": False
}
}
}
class TestTokenProvider(test.TestCase):
def setUp(self):
super(TestTokenProvider, self).setUp()
self.load_backends()
self.token_provider_api = token.provider.Manager()
def test_get_token_version(self):
self.assertEqual(
token.provider.V2,
self.token_provider_api.get_token_version(SAMPLE_V2_TOKEN))
self.assertEqual(
token.provider.V3,
self.token_provider_api.get_token_version(SAMPLE_V3_TOKEN))
self.assertRaises(token.provider.UnsupportedTokenVersionException,
self.token_provider_api.get_token_version,
'bogus')
def test_issue_token(self):
self.assertRaises(token.provider.UnsupportedTokenVersionException,
self.token_provider_api.issue_token,
'bogus_version')
def test_validate_token(self):
self.assertRaises(token.provider.UnsupportedTokenVersionException,
self.token_provider_api.validate_token,
uuid.uuid4().hex,
None,
'bogus_version')
def test_token_format_provider_mismatch(self):
self.opt_in_group('signing', token_format='UUID')
self.opt_in_group('token',
provider='keystone.token.providers.pki.Provider')
try:
token.provider.Manager()
raise Exception(
'expecting ValueError on token provider misconfiguration')
except ValueError:
pass
self.opt_in_group('signing', token_format='PKI')
self.opt_in_group('token',
provider='keystone.token.providers.uuid.Provider')
try:
token.provider.Manager()
raise Exception(
'expecting ValueError on token provider misconfiguration')
except ValueError:
pass
# should be OK as token_format and provider aligns
self.opt_in_group('signing', token_format='PKI')
self.opt_in_group('token',
provider='keystone.token.providers.pki.Provider')
token.provider.Manager()
self.opt_in_group('signing', token_format='UUID')
self.opt_in_group('token',
provider='keystone.token.providers.uuid.Provider')
token.provider.Manager()
# custom provider should be OK too
self.opt_in_group('signing', token_format='CUSTOM')
self.opt_in_group('token',
provider='keystone.token.providers.pki.Provider')
token.provider.Manager()

5
tests/test_uuid_token_provider.conf Normal file
View File

@ -0,0 +1,5 @@
[signing]
token_format = UUID
[token]
provider = keystone.token.providers.uuid.Provider

18
tests/test_v3.py
View File

@ -5,6 +5,7 @@ from lxml import etree
import webtest
from keystone import test
from keystone import token
from keystone import auth
from keystone.common import serializer
@ -22,6 +23,15 @@ TIME_FORMAT = '%Y-%m-%dT%H:%M:%S.%fZ'
class RestfulTestCase(test_content_types.RestfulTestCase):
_config_file_list = [test.etcdir('keystone.conf.sample'),
test.testsdir('test_overrides.conf'),
test.testsdir('backend_sql.conf'),
test.testsdir('backend_sql_disk.conf')]
#override this to sepcify the complete list of configuration files
def config_files(self):
return self._config_file_list
def setUp(self, load_sample_data=True):
"""Setup for v3 Restful Test Cases.
@ -30,15 +40,13 @@ class RestfulTestCase(test_content_types.RestfulTestCase):
load_sample_data should be set to false.
"""
self.config([
test.etcdir('keystone.conf.sample'),
test.testsdir('test_overrides.conf'),
test.testsdir('backend_sql.conf'),
test.testsdir('backend_sql_disk.conf')])
self.config(self.config_files())
test.setup_test_database()
self.load_backends()
self.token_provider_api = token.provider.Manager()
self.public_app = webtest.TestApp(
self.loadapp('keystone', name='main'))
self.admin_app = webtest.TestApp(

147
tests/test_v3_auth.py
View File

@ -21,6 +21,7 @@ from keystone import auth
from keystone.common import cms
from keystone import config
from keystone import exception
from keystone import test
import test_v3
@ -96,9 +97,14 @@ class TestAuthInfo(test_v3.RestfulTestCase):
method_name)
class TestTokenAPIs(test_v3.RestfulTestCase):
class TestPKITokenAPIs(test_v3.RestfulTestCase):
def config_files(self):
conf_files = super(TestPKITokenAPIs, self).config_files()
conf_files.append(test.testsdir('test_pki_token_provider.conf'))
return conf_files
def setUp(self):
super(TestTokenAPIs, self).setUp()
super(TestPKITokenAPIs, self).setUp()
auth_data = self.build_authentication_request(
username=self.user['name'],
user_domain_id=self.domain_id,
@ -111,8 +117,7 @@ class TestTokenAPIs(test_v3.RestfulTestCase):
def test_default_fixture_scope_token(self):
self.assertIsNotNone(self.get_scoped_token())
def test_v3_pki_token_id(self):
self.opt_in_group('signing', token_format='PKI')
def test_v3_token_id(self):
auth_data = self.build_authentication_request(
user_id=self.user['id'],
password=self.user['password'])
@ -120,13 +125,19 @@ class TestTokenAPIs(test_v3.RestfulTestCase):
token_data = resp.result
token_id = resp.headers.get('X-Subject-Token')
self.assertIn('expires_at', token_data['token'])
token_signed = cms.cms_sign_token(json.dumps(token_data),
CONF.signing.certfile,
CONF.signing.keyfile)
self.assertEqual(token_signed, token_id)
expected_token_id = cms.cms_sign_token(json.dumps(token_data),
CONF.signing.certfile,
CONF.signing.keyfile)
self.assertEqual(expected_token_id, token_id)
# should be able to validate hash PKI token as well
hash_token_id = cms.cms_hash_token(token_id)
headers = {'X-Subject-Token': hash_token_id}
resp = self.get('/auth/tokens', headers=headers)
expected_token_data = resp.result
self.assertDictEqual(expected_token_data, token_data)
def test_v3_v2_intermix_non_default_domain_failed(self):
self.opt_in_group('signing', token_format='UUID')
auth_data = self.build_authentication_request(
user_id=self.user['id'],
password=self.user['password'])
@ -141,7 +152,6 @@ class TestTokenAPIs(test_v3.RestfulTestCase):
expected_status=401)
def test_v3_v2_intermix_domain_scoped_token_failed(self):
self.opt_in_group('signing', token_format='UUID')
# grant the domain role to user
path = '/domains/%s/users/%s/roles/%s' % (
self.domain['id'], self.user['id'], self.role['id'])
@ -175,8 +185,7 @@ class TestTokenAPIs(test_v3.RestfulTestCase):
method='GET',
expected_status=401)
def test_v3_v2_unscoped_uuid_token_intermix(self):
self.opt_in_group('signing', token_format='UUID')
def test_v3_v2_unscoped_token_intermix(self):
auth_data = self.build_authentication_request(
user_id=self.default_domain_user['id'],
password=self.default_domain_user['password'])
@ -197,32 +206,9 @@ class TestTokenAPIs(test_v3.RestfulTestCase):
self.assertIn(v2_token['access']['token']['expires'][:-1],
token_data['token']['expires_at'])
def test_v3_v2_unscoped_pki_token_intermix(self):
self.opt_in_group('signing', token_format='PKI')
auth_data = self.build_authentication_request(
user_id=self.default_domain_user['id'],
password=self.default_domain_user['password'])
resp = self.post('/auth/tokens', body=auth_data)
token_data = resp.result
token = resp.headers.get('X-Subject-Token')
# now validate the v3 token with v2 API
path = '/v2.0/tokens/%s' % (token)
resp = self.admin_request(path=path,
token='ADMIN',
method='GET')
v2_token = resp.result
self.assertEqual(v2_token['access']['user']['id'],
token_data['token']['user']['id'])
# v2 token time has not fraction of second precision so
# just need to make sure the non fraction part agrees
self.assertIn(v2_token['access']['token']['expires'][:-1],
token_data['token']['expires_at'])
def test_v3_v2_uuid_token_intermix(self):
def test_v3_v2_token_intermix(self):
# FIXME(gyee): PKI tokens are not interchangeable because token
# data is baked into the token itself.
self.opt_in_group('signing', token_format='UUID')
auth_data = self.build_authentication_request(
user_id=self.default_domain_user['id'],
password=self.default_domain_user['password'],
@ -246,10 +232,7 @@ class TestTokenAPIs(test_v3.RestfulTestCase):
self.assertEqual(v2_token['access']['user']['roles'][0]['id'],
token_data['token']['roles'][0]['id'])
def test_v3_v2_pki_token_intermix(self):
# FIXME(gyee): PKI tokens are not interchangeable because token
# data is baked into the token itself.
self.opt_in_group('signing', token_format='PKI')
def test_v3_v2_hashed_pki_token_intermix(self):
auth_data = self.build_authentication_request(
user_id=self.default_domain_user['id'],
password=self.default_domain_user['password'],
@ -258,7 +241,8 @@ class TestTokenAPIs(test_v3.RestfulTestCase):
token_data = resp.result
token = resp.headers.get('X-Subject-Token')
# now validate the v3 token with v2 API
# should be able to validate a hash PKI token in v2 too
token = cms.cms_hash_token(token)
path = '/v2.0/tokens/%s' % (token)
resp = self.admin_request(path=path,
token='ADMIN',
@ -268,13 +252,12 @@ class TestTokenAPIs(test_v3.RestfulTestCase):
token_data['token']['user']['id'])
# v2 token time has not fraction of second precision so
# just need to make sure the non fraction part agrees
self.assertIn(v2_token['access']['token']['expires'][-1],
self.assertIn(v2_token['access']['token']['expires'][:-1],
token_data['token']['expires_at'])
self.assertEqual(v2_token['access']['user']['roles'][0]['id'],
token_data['token']['roles'][0]['id'])
def test_v2_v3_unscoped_uuid_token_intermix(self):
self.opt_in_group('signing', token_format='UUID')
def test_v2_v3_unscoped_token_intermix(self):
body = {
'auth': {
'passwordCredentials': {
@ -297,59 +280,7 @@ class TestTokenAPIs(test_v3.RestfulTestCase):
self.assertIn(v2_token_data['access']['token']['expires'][-1],
token_data['token']['expires_at'])
def test_v2_v3_unscoped_pki_token_intermix(self):
self.opt_in_group('signing', token_format='PKI')
body = {
'auth': {
'passwordCredentials': {
'userId': self.user['id'],
'password': self.user['password']
}
}}
resp = self.admin_request(path='/v2.0/tokens',
method='POST',
body=body)
v2_token_data = resp.result
v2_token = v2_token_data['access']['token']['id']
headers = {'X-Subject-Token': v2_token}
resp = self.get('/auth/tokens', headers=headers)
token_data = resp.result
self.assertEqual(v2_token_data['access']['user']['id'],
token_data['token']['user']['id'])
# v2 token time has not fraction of second precision so
# just need to make sure the non fraction part agrees
self.assertIn(v2_token_data['access']['token']['expires'][-1],
token_data['token']['expires_at'])
def test_v2_v3_uuid_token_intermix(self):
self.opt_in_group('signing', token_format='UUID')
body = {
'auth': {
'passwordCredentials': {
'userId': self.user['id'],
'password': self.user['password']
},
'tenantId': self.project['id']
}}
resp = self.admin_request(path='/v2.0/tokens',
method='POST',
body=body)
v2_token_data = resp.result
v2_token = v2_token_data['access']['token']['id']
headers = {'X-Subject-Token': v2_token}
resp = self.get('/auth/tokens', headers=headers)
token_data = resp.result
self.assertEqual(v2_token_data['access']['user']['id'],
token_data['token']['user']['id'])
# v2 token time has not fraction of second precision so
# just need to make sure the non fraction part agrees
self.assertIn(v2_token_data['access']['token']['expires'][-1],
token_data['token']['expires_at'])
self.assertEqual(v2_token_data['access']['user']['roles'][0]['name'],
token_data['token']['roles'][0]['name'])
def test_v2_v3_pki_token_intermix(self):
self.opt_in_group('signing', token_format='PKI')
def test_v2_v3_token_intermix(self):
body = {
'auth': {
'passwordCredentials': {
@ -402,6 +333,28 @@ class TestTokenAPIs(test_v3.RestfulTestCase):
self.assertIn('signed', r.result)
class TestUUIDTokenAPIs(TestPKITokenAPIs):
def config_files(self):
conf_files = super(TestUUIDTokenAPIs, self).config_files()
conf_files.append(test.testsdir('test_uuid_token_provider.conf'))
return conf_files
def test_v3_token_id(self):
auth_data = self.build_authentication_request(
user_id=self.user['id'],
password=self.user['password'])
resp = self.post('/auth/tokens', body=auth_data)
token_data = resp.result
token_id = resp.headers.get('X-Subject-Token')
self.assertIn('expires_at', token_data['token'])
self.assertFalse(cms.is_ans1_token(token_id))
def test_v3_v2_hashed_pki_token_intermix(self):
# this test is only applicable for PKI tokens
# skipping it for UUID tokens
pass
class TestTokenRevoking(test_v3.RestfulTestCase):
"""Test token revocation on the v3 Identity API."""