openstack
/
keystone
Code Issues Proposed changes

Implement system-scoped tokens 

This commit exposes the necessary bits to expose system-scoped
token authenticate and validation via the API

bp system-scope

Change-Id: I572a8e48953f493d521fd2aa00007df46e562e2e
This commit is contained in:
Lance Bragstad 2017-12-05 16:44:12 +00:00
parent 19a2ccb51e
commit 5d6f4bb1ee
10 changed files with 336 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
keystone/assignment/core.py
View File

@ -1051,12 +1051,15 @@ class Manager(manager.Manager):
# a domain assignment, we might as well kill all the tokens for
# the user, since in the vast majority of cases all the tokens
# for a user will be within one domain anyway, so not worth
# trying to delete tokens for each project in the domain.
# trying to delete tokens for each project in the domain. If the
# assignment is a system assignment, invalidate all tokens from the
# cache. A future patch may optimize this to only remove specific
# system-scoped tokens from the cache.
if 'user_id' in assignment:
if 'project_id' in assignment:
user_and_project_ids.append(
(assignment['user_id'], assignment['project_id']))
elif 'domain_id' in assignment:
elif 'domain_id' or 'system' in assignment:
self._emit_invalidate_user_token_persistence(
assignment['user_id'])
elif 'group_id' in assignment:
@ -1083,7 +1086,7 @@ class Manager(manager.Manager):
for user in users:
user_and_project_ids.append(
(user['id'], assignment['project_id']))
elif 'domain_id' in assignment:
elif 'domain_id' or 'system' in assignment:
for user in users:
self._emit_invalidate_user_token_persistence(
user['id'])

15
keystone/auth/controllers.py
View File

@ -116,7 +116,9 @@ class Auth(controller.V3Controller):
if auth_context.get('access_token_id'):
auth_info.set_scope(None, auth_context['project_id'], None)
self._check_and_set_default_scoping(auth_info, auth_context)
(domain_id, project_id, trust, unscoped) = auth_info.get_scope()
(domain_id, project_id, trust, unscoped, system) = (
auth_info.get_scope()
)
# NOTE(notmorgan): only methods that actually run and succeed will
# be in the auth_context['method_names'] list. Do not blindly take
@ -138,8 +140,9 @@ class Auth(controller.V3Controller):
is_domain = auth_context.get('is_domain')
(token_id, token_data) = self.token_provider_api.issue_token(
auth_context['user_id'], method_names, expires_at=expires_at,
project_id=project_id, is_domain=is_domain,
domain_id=domain_id, auth_context=auth_context, trust=trust,
system=system, project_id=project_id,
is_domain=is_domain, domain_id=domain_id,
auth_context=auth_context, trust=trust,
include_catalog=include_catalog,
parent_audit_id=token_audit_id)
@ -155,10 +158,12 @@ class Auth(controller.V3Controller):
raise exception.Unauthorized(e)
def _check_and_set_default_scoping(self, auth_info, auth_context):
(domain_id, project_id, trust, unscoped) = auth_info.get_scope()
(domain_id, project_id, trust, unscoped, system) = (
auth_info.get_scope()
)
if trust:
project_id = trust['project_id']
if domain_id or project_id or trust:
if system or domain_id or project_id or trust:
# scope is specified
return

59
keystone/auth/core.py
View File

@ -143,12 +143,14 @@ class AuthInfo(provider_api.ProviderAPIMixin, object):
def __init__(self, auth=None):
self.auth = auth
self._scope_data = (None, None, None, None)
# self._scope_data is (domain_id, project_id, trust_ref, unscoped)
# project scope: (None, project_id, None, None)
# domain scope: (domain_id, None, None, None)
# trust scope: (None, None, trust_ref, None)
# unscoped: (None, None, None, 'unscoped')
self._scope_data = (None, None, None, None, None)
# self._scope_data is
# (domain_id, project_id, trust_ref, unscoped, system)
# project scope: (None, project_id, None, None, None)
# domain scope: (domain_id, None, None, None, None)
# trust scope: (None, None, trust_ref, None, None)
# unscoped: (None, None, None, 'unscoped', None)
# system: (None, None, None, None, 'all')
def _assert_project_is_enabled(self, project_ref):
# ensure the project is enabled
@ -234,19 +236,22 @@ class AuthInfo(provider_api.ProviderAPIMixin, object):
if sum(['project' in self.auth['scope'],
'domain' in self.auth['scope'],
'unscoped' in self.auth['scope'],
'system' in self.auth['scope'],
'OS-TRUST:trust' in self.auth['scope']]) != 1:
raise exception.ValidationError(
attribute='project, domain, OS-TRUST:trust or unscoped',
target='scope')
msg = 'system, project, domain, OS-TRUST:trust or unscoped'
raise exception.ValidationError(attribute=msg, target='scope')
if 'system' in self.auth['scope']:
self._scope_data = (None, None, None, None, 'all')
return
if 'unscoped' in self.auth['scope']:
self._scope_data = (None, None, None, 'unscoped')
self._scope_data = (None, None, None, 'unscoped', None)
return
if 'project' in self.auth['scope']:
project_ref = self._lookup_project(self.auth['scope']['project'])
self._scope_data = (None, project_ref['id'], None, None)
self._scope_data = (None, project_ref['id'], None, None, None)
elif 'domain' in self.auth['scope']:
domain_ref = self._lookup_domain(self.auth['scope']['domain'])
self._scope_data = (domain_ref['id'], None, None, None)
self._scope_data = (domain_ref['id'], None, None, None, None)
elif 'OS-TRUST:trust' in self.auth['scope']:
if not CONF.trust.enabled:
raise exception.Forbidden('Trusts are disabled.')
@ -256,9 +261,12 @@ class AuthInfo(provider_api.ProviderAPIMixin, object):
if trust_ref.get('project_id') is not None:
project_ref = self._lookup_project(
{'id': trust_ref['project_id']})
self._scope_data = (None, project_ref['id'], trust_ref, None)
self._scope_data = (
None, project_ref['id'], trust_ref, None, None
)
else:
self._scope_data = (None, None, trust_ref, None)
self._scope_data = (None, None, trust_ref, None, None)
def _validate_auth_methods(self):
# make sure all the method data/payload are provided
@ -323,22 +331,25 @@ class AuthInfo(provider_api.ProviderAPIMixin, object):
Verify and return the scoping information.
:returns: (domain_id, project_id, trust_ref, unscoped).
If scope to a project, (None, project_id, None, None)
:returns: (domain_id, project_id, trust_ref, unscoped, system).
If scope to a project, (None, project_id, None, None, None)
will be returned.
If scoped to a domain, (domain_id, None, None, None)
If scoped to a domain, (domain_id, None, None, None, None)
will be returned.
If scoped to a trust, (None, project_id, trust_ref, None),
If scoped to a trust,
(None, project_id, trust_ref, None, None),
Will be returned, where the project_id comes from the
trust definition.
If unscoped, (None, None, None, 'unscoped') will be
If unscoped, (None, None, None, 'unscoped', None) will be
returned.
If system_scoped, (None, None, None, None, 'all') will be
returned.
"""
return self._scope_data
def set_scope(self, domain_id=None, project_id=None, trust=None,
unscoped=None):
unscoped=None, system=None):
"""Set scope information."""
if domain_id and project_id:
msg = _('Scoping to both domain and project is not allowed')
@ -349,7 +360,13 @@ class AuthInfo(provider_api.ProviderAPIMixin, object):
if project_id and trust:
msg = _('Scoping to both project and trust is not allowed')
raise ValueError(msg)
self._scope_data = (domain_id, project_id, trust, unscoped)
if system and project_id:
msg = _('Scoping to both project and system is not allowed')
raise ValueError(msg)
if system and domain_id:
msg = _('Scoping to both domain and system is not allowed')
raise ValueError(msg)
self._scope_data = (domain_id, project_id, trust, unscoped, system)
class UserMFARulesValidator(provider_api.ProviderAPIMixin, object):

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

@ -68,6 +68,13 @@ def token_authenticate(request, token_ref):
# state in Keystone. To do so is to invite elevation of
# privilege attacks
project_scoped = 'project' in request.json_body['auth'].get(
'scope', {}
)
domain_scoped = 'domain' in request.json_body['auth'].get(
'scope', {}
)
if token_ref.oauth_scoped:
raise exception.ForbiddenAction(
action=_(
@ -78,6 +85,13 @@ def token_authenticate(request, token_ref):
action=_(
'Using trust-scoped token to create another token. '
'Create a new trust-scoped token instead'))
elif token_ref.system_scoped and (project_scoped or domain_scoped):
raise exception.ForbiddenAction(
action=_(
'Using a system-scoped token to create a project-scoped '
'or domain-scoped token is not allowed.'
)
)
if not CONF.token.allow_rescope_scoped_token:
# Do not allow conversion from scoped tokens.

11
keystone/auth/schema.py
View File

@ -10,6 +10,9 @@
# License for the specific language governing permissions and limitations
# under the License.
from keystone.common.validation import parameter_types
token_issue = {
'type': 'object',
'properties': {
@ -86,8 +89,14 @@ token_issue = {
'type': 'object',
'properties': {
'id': {'type': 'string', },
},
}
},
'system': {
'type': 'object',
'properties': {
'all': parameter_types.boolean
}
}
},
},
},

2
keystone/middleware/auth.py
View File

@ -58,7 +58,7 @@ class AuthContextMiddleware(provider_api.ProviderAPIMixin,
"""
tokenless_helper = tokenless_auth.TokenlessAuthHelper(request.environ)
(domain_id, project_id, trust_ref, unscoped) = (
(domain_id, project_id, trust_ref, unscoped, system) = (
tokenless_helper.get_scope())
user_ref = tokenless_helper.get_mapped_user(
project_id,

6
keystone/models/token_model.py
View File

@ -176,6 +176,10 @@ class KeystoneToken(dict):
return self['is_domain']
return False
@property
def system_scoped(self):
return 'system' in self
@property
def project_scoped(self):
return 'project' in self
@ -186,7 +190,7 @@ class KeystoneToken(dict):
@property
def scoped(self):
return self.project_scoped or self.domain_scoped
return self.project_scoped or self.domain_scoped or self.system_scoped
@property
def is_admin_project(self):

4
keystone/tests/common/auth.py
View File

@ -19,7 +19,9 @@ class AuthTestMixin(object):
project_domain_name=None, domain_id=None,
domain_name=None, trust_id=None, unscoped=None):
scope_data = {}
if unscoped:
if system:
scope_data['system'] = {'all': True}
elif unscoped:
scope_data['unscoped'] = {}
elif system:
scope_data['system'] = {'all': True}

220
keystone/tests/unit/test_v3_auth.py
View File

@ -622,6 +622,226 @@ class TokenAPITests(object):
expected_status=http_client.NOT_FOUND
)
def test_create_system_token_with_user_id(self):
path = '/system/users/%(user_id)s/roles/%(role_id)s' % {
'user_id': self.user['id'],
'role_id': self.role_id
}
self.put(path=path)
auth_request_body = self.build_authentication_request(
user_id=self.user['id'],
password=self.user['password'],
system=True
)
response = self.v3_create_token(auth_request_body)
self.assertValidSystemScopedTokenResponse(response)
def test_create_system_token_with_username(self):
path = '/system/users/%(user_id)s/roles/%(role_id)s' % {
'user_id': self.user['id'],
'role_id': self.role_id
}
self.put(path=path)
auth_request_body = self.build_authentication_request(
username=self.user['name'],
password=self.user['password'],
user_domain_id=self.domain['id'],
system=True
)
response = self.v3_create_token(auth_request_body)
self.assertValidSystemScopedTokenResponse(response)
def test_create_system_token_fails_without_system_assignment(self):
auth_request_body = self.build_authentication_request(
user_id=self.user['id'],
password=self.user['password'],
system=True
)
self.v3_create_token(
auth_request_body,
expected_status=http_client.UNAUTHORIZED
)
def test_system_token_is_invalid_after_disabling_user(self):
path = '/system/users/%(user_id)s/roles/%(role_id)s' % {
'user_id': self.user['id'],
'role_id': self.role_id
}
self.put(path=path)
auth_request_body = self.build_authentication_request(
username=self.user['name'],
password=self.user['password'],
user_domain_id=self.domain['id'],
system=True
)
response = self.v3_create_token(auth_request_body)
self.assertValidSystemScopedTokenResponse(response)
token = response.headers.get('X-Subject-Token')
self._validate_token(token)
# NOTE(lbragstad): This would make a good test for groups, but
# apparently it's not possible to disable a group.
user_ref = {
'user': {
'enabled': False
}
}
self.patch(
'/users/%(user_id)s' % {'user_id': self.user['id']},
body=user_ref
)
self.admin_request(
path='/v3/auth/tokens',
headers={'X-Auth-Token': token,
'X-Subject-Token': token},
method='GET',
expected_status=http_client.UNAUTHORIZED
)
self.admin_request(
path='/v3/auth/tokens',
headers={'X-Auth-Token': token,
'X-Subject-Token': token},
method='HEAD',
expected_status=http_client.UNAUTHORIZED
)
def test_create_system_token_via_system_group_assignment(self):
ref = {
'group': unit.new_group_ref(
domain_id=CONF.identity.default_domain_id
)
}
group = self.post('/groups', body=ref).json_body['group']
path = '/system/groups/%(group_id)s/roles/%(role_id)s' % {
'group_id': group['id'],
'role_id': self.role_id
}
self.put(path=path)
path = '/groups/%(group_id)s/users/%(user_id)s' % {
'group_id': group['id'],
'user_id': self.user['id']
}
self.put(path=path)
auth_request_body = self.build_authentication_request(
user_id=self.user['id'],
password=self.user['password'],
system=True
)
response = self.v3_create_token(auth_request_body)
self.assertValidSystemScopedTokenResponse(response)
token = response.headers.get('X-Subject-Token')
self._validate_token(token)
def test_revoke_system_token(self):
path = '/system/users/%(user_id)s/roles/%(role_id)s' % {
'user_id': self.user['id'],
'role_id': self.role_id
}
self.put(path=path)
auth_request_body = self.build_authentication_request(
username=self.user['name'],
password=self.user['password'],
user_domain_id=self.domain['id'],
system=True
)
response = self.v3_create_token(auth_request_body)
self.assertValidSystemScopedTokenResponse(response)
token = response.headers.get('X-Subject-Token')
self._validate_token(token)
self._revoke_token(token)
self._validate_token(token, expected_status=http_client.NOT_FOUND)
def test_system_token_is_invalid_after_deleting_system_role(self):
ref = {'role': unit.new_role_ref()}
system_role = self.post('/roles', body=ref).json_body['role']
path = '/system/users/%(user_id)s/roles/%(role_id)s' % {
'user_id': self.user['id'],
'role_id': system_role['id']
}
self.put(path=path)
auth_request_body = self.build_authentication_request(
username=self.user['name'],
password=self.user['password'],
user_domain_id=self.domain['id'],
system=True
)
response = self.v3_create_token(auth_request_body)
self.assertValidSystemScopedTokenResponse(response)
token = response.headers.get('X-Subject-Token')
self._validate_token(token)
self.delete('/roles/%(role_id)s' % {'role_id': system_role['id']})
self._validate_token(token, expected_status=http_client.NOT_FOUND)
def test_rescoping_a_system_token_for_a_project_token_fails(self):
ref = {'role': unit.new_role_ref()}
system_role = self.post('/roles', body=ref).json_body['role']
path = '/system/users/%(user_id)s/roles/%(role_id)s' % {
'user_id': self.user['id'],
'role_id': system_role['id']
}
self.put(path=path)
auth_request_body = self.build_authentication_request(
username=self.user['name'],
password=self.user['password'],
user_domain_id=self.domain['id'],
system=True
)
response = self.v3_create_token(auth_request_body)
self.assertValidSystemScopedTokenResponse(response)
system_token = response.headers.get('X-Subject-Token')
auth_request_body = self.build_authentication_request(
token=system_token, project_id=self.project_id
)
self.v3_create_token(
auth_request_body, expected_status=http_client.FORBIDDEN
)
def test_rescoping_a_system_token_for_a_domain_token_fails(self):
ref = {'role': unit.new_role_ref()}
system_role = self.post('/roles', body=ref).json_body['role']
path = '/system/users/%(user_id)s/roles/%(role_id)s' % {
'user_id': self.user['id'],
'role_id': system_role['id']
}
self.put(path=path)
auth_request_body = self.build_authentication_request(
username=self.user['name'],
password=self.user['password'],
user_domain_id=self.domain['id'],
system=True
)
response = self.v3_create_token(auth_request_body)
self.assertValidSystemScopedTokenResponse(response)
system_token = response.headers.get('X-Subject-Token')
auth_request_body = self.build_authentication_request(
token=system_token, domain_id=CONF.identity.default_domain_id
)
self.v3_create_token(
auth_request_body, expected_status=http_client.FORBIDDEN
)
def test_create_domain_token_scoped_with_domain_id_and_user_id(self):
# grant the user a role on the domain
path = '/domains/%s/users/%s/roles/%s' % (

34
keystone/token/providers/common.py
View File

@ -16,6 +16,7 @@ from __future__ import absolute_import
import base64
import datetime
import itertools
import uuid
from oslo_log import log
@ -164,8 +165,25 @@ class V3TokenDataHelper(provider_api.ProviderAPIMixin, object):
project['name'] == admin_project_name and
project['domain']['name'] == admin_project_domain_name)
def _get_roles_for_user(self, user_id, domain_id, project_id):
def _get_roles_for_user(self, user_id, system, domain_id, project_id):
roles = []
if system:
group_ids = [
group['id'] for
group in PROVIDERS.identity_api.list_groups_for_user(user_id)
]
group_roles = []
for group_id in group_ids:
roles = PROVIDERS.assignment_api.list_system_grants_for_group(
group_id
)
for role in roles:
group_roles.append(role)
user_roles = PROVIDERS.assignment_api.list_system_grants_for_user(
user_id
)
return itertools.chain(group_roles, user_roles)
if domain_id:
roles = PROVIDERS.assignment_api.get_roles_for_user_and_domain(
user_id, domain_id)
@ -176,8 +194,8 @@ class V3TokenDataHelper(provider_api.ProviderAPIMixin, object):
def populate_roles_for_federated_user(self, token_data, group_ids,
project_id=None, domain_id=None,
user_id=None):
"""Populate roles basing on provided groups and project/domain.
user_id=None, system=None):
"""Populate roles basing on provided groups and assignments.
Used for federated users with dynamically assigned groups.
This method does not return anything, yet it modifies token_data in
@ -188,6 +206,7 @@ class V3TokenDataHelper(provider_api.ProviderAPIMixin, object):
:param project_id: project ID to scope to
:param domain_id: domain ID to scope to
:param user_id: user ID
:param system: system scope if applicable
:raises keystone.exception.Unauthorized: when no roles were found
@ -214,8 +233,9 @@ class V3TokenDataHelper(provider_api.ProviderAPIMixin, object):
roles = PROVIDERS.assignment_api.get_roles_for_groups(
group_ids, project_id, domain_id
)
roles = roles + self._get_roles_for_user(user_id, domain_id,
project_id)
roles = roles + self._get_roles_for_user(
user_id, system, domain_id, project_id
)
# NOTE(lbragstad): Remove duplicate role references from a list of
# roles. It is often suggested that this be done with:
@ -363,6 +383,7 @@ class V3TokenDataHelper(provider_api.ProviderAPIMixin, object):
_('Trustee has no delegated roles.'))
else:
for role in self._get_roles_for_user(token_user_id,
system,
token_domain_id,
token_project_id):
filtered_roles.append({'id': role['id'],
@ -565,6 +586,7 @@ class BaseProvider(provider_api.ProviderAPIMixin, base.Provider):
}
}
# FIXME(lbragstad): This will have to account for system-scoping, too.
if project_id or domain_id:
self.v3_token_data_helper.populate_roles_for_federated_user(
token_data, group_ids, project_id, domain_id, user_id)
@ -588,6 +610,8 @@ class BaseProvider(provider_api.ProviderAPIMixin, base.Provider):
expires_at = token_data['token']['expires_at']
audit_ids = token_data['token'].get('audit_ids')
system = token_data['token'].get('system', {}).get('all')
if system:
system = 'all'
domain_id = token_data['token'].get('domain', {}).get('id')
project_id = token_data['token'].get('project', {}).get('id')
access_token = None