openstack
/
keystone
Code Issues Proposed changes

Implement system reader & member for policy association 

This change modifies the policies for policy association
API to be more self-service by properly checking for
system scopes. It also includes the test cases.

Subsequent patches will -

 - add functionality for system admin for policy association
 - domains user test coverage for policy association
 - project user test coverage for policy association
 - Removing obsolete policies in policy.v3cloudsample.json file

Change-Id: I0b6ddc961d65301b4b95b0ba1c2515ef4a782d55
Partial-Bug: #1805409
This commit is contained in:
Vishakha Agarwal 2019-08-22 16:25:47 +05:30
parent 0e52753898
commit b831856af3
2 changed files with 353 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

69
keystone/common/policies/policy_association.py
View File

@ -10,6 +10,7 @@
# License for the specific language governing permissions and limitations
# under the License.
from oslo_log import versionutils
from oslo_policy import policy
from keystone.common.policies import base
@ -18,6 +19,39 @@ from keystone.common.policies import base
# System-scoped tokens should be required to manage policy associations to
# existing system-level resources.
deprecated_check_policy_association_for_endpoint = policy.DeprecatedRule(
name=base.IDENTITY % 'check_policy_association_for_endpoint',
check_str=base.RULE_ADMIN_REQUIRED,
)
deprecated_check_policy_association_for_service = policy.DeprecatedRule(
name=base.IDENTITY % 'check_policy_association_for_service',
check_str=base.RULE_ADMIN_REQUIRED,
)
deprecated_check_policy_association_for_region_and_service = policy.DeprecatedRule(
name=base.IDENTITY % 'check_policy_association_for_region_and_service',
check_str=base.RULE_ADMIN_REQUIRED,
)
deprecated_get_policy_for_endpoint = policy.DeprecatedRule(
name=base.IDENTITY % 'get_policy_for_endpoint',
check_str=base.RULE_ADMIN_REQUIRED,
)
deprecated_list_endpoints_for_policy = policy.DeprecatedRule(
name=base.IDENTITY % 'list_endpoints_for_policy',
check_str=base.RULE_ADMIN_REQUIRED,
)
DEPRECATED_REASON = """
As of the Train release, the policy association API now understands default
roles and system-scoped tokens, making the API more granular by default without
compromising security. The new policy defaults account for these changes
automatically. Be sure to take these new defaults into consideration if you are
relying on overrides in your deployment for the policy association API.
"""
policy_association_policies = [
policy.DocumentedRuleDefault(
name=base.IDENTITY % 'create_policy_association_for_endpoint',
@ -29,7 +63,7 @@ policy_association_policies = [
'method': 'PUT'}]),
policy.DocumentedRuleDefault(
name=base.IDENTITY % 'check_policy_association_for_endpoint',
check_str=base.RULE_ADMIN_REQUIRED,
check_str=base.SYSTEM_READER,
scope_types=['system'],
description='Check policy association for endpoint.',
operations=[{'path': ('/v3/policies/{policy_id}/OS-ENDPOINT-POLICY/'
@ -37,7 +71,10 @@ policy_association_policies = [
'method': 'GET'},
{'path': ('/v3/policies/{policy_id}/OS-ENDPOINT-POLICY/'
'endpoints/{endpoint_id}'),
'method': 'HEAD'}]),
'method': 'HEAD'}],
deprecated_rule=deprecated_check_policy_association_for_endpoint,
deprecated_reason=DEPRECATED_REASON,
deprecated_since=versionutils.deprecated.TRAIN),
policy.DocumentedRuleDefault(
name=base.IDENTITY % 'delete_policy_association_for_endpoint',
check_str=base.RULE_ADMIN_REQUIRED,
@ -56,7 +93,7 @@ policy_association_policies = [
'method': 'PUT'}]),
policy.DocumentedRuleDefault(
name=base.IDENTITY % 'check_policy_association_for_service',
check_str=base.RULE_ADMIN_REQUIRED,
check_str=base.SYSTEM_READER,
scope_types=['system'],
description='Check policy association for service.',
operations=[{'path': ('/v3/policies/{policy_id}/OS-ENDPOINT-POLICY/'
@ -64,7 +101,10 @@ policy_association_policies = [
'method': 'GET'},
{'path': ('/v3/policies/{policy_id}/OS-ENDPOINT-POLICY/'
'services/{service_id}'),
'method': 'HEAD'}]),
'method': 'HEAD'}],
deprecated_rule=deprecated_check_policy_association_for_service,
deprecated_reason=DEPRECATED_REASON,
deprecated_since=versionutils.deprecated.TRAIN),
policy.DocumentedRuleDefault(
name=base.IDENTITY % 'delete_policy_association_for_service',
check_str=base.RULE_ADMIN_REQUIRED,
@ -85,7 +125,7 @@ policy_association_policies = [
'method': 'PUT'}]),
policy.DocumentedRuleDefault(
name=base.IDENTITY % 'check_policy_association_for_region_and_service',
check_str=base.RULE_ADMIN_REQUIRED,
check_str=base.SYSTEM_READER,
scope_types=['system'],
description='Check policy association for region and service.',
operations=[{'path': ('/v3/policies/{policy_id}/OS-ENDPOINT-POLICY/'
@ -93,7 +133,10 @@ policy_association_policies = [
'method': 'GET'},
{'path': ('/v3/policies/{policy_id}/OS-ENDPOINT-POLICY/'
'services/{service_id}/regions/{region_id}'),
'method': 'HEAD'}]),
'method': 'HEAD'}],
deprecated_rule=deprecated_check_policy_association_for_region_and_service,
deprecated_reason=DEPRECATED_REASON,
deprecated_since=versionutils.deprecated.TRAIN),
policy.DocumentedRuleDefault(
name=base.IDENTITY % (
'delete_policy_association_for_region_and_service'),
@ -105,7 +148,7 @@ policy_association_policies = [
'method': 'DELETE'}]),
policy.DocumentedRuleDefault(
name=base.IDENTITY % 'get_policy_for_endpoint',
check_str=base.RULE_ADMIN_REQUIRED,
check_str=base.SYSTEM_READER,
scope_types=['system'],
description='Get policy for endpoint.',
operations=[{'path': ('/v3/endpoints/{endpoint_id}/OS-ENDPOINT-POLICY/'
@ -113,15 +156,21 @@ policy_association_policies = [
'method': 'GET'},
{'path': ('/v3/endpoints/{endpoint_id}/OS-ENDPOINT-POLICY/'
'policy'),
'method': 'HEAD'}]),
'method': 'HEAD'}],
deprecated_rule=deprecated_get_policy_for_endpoint,
deprecated_reason=DEPRECATED_REASON,
deprecated_since=versionutils.deprecated.TRAIN),
policy.DocumentedRuleDefault(
name=base.IDENTITY % 'list_endpoints_for_policy',
check_str=base.RULE_ADMIN_REQUIRED,
check_str=base.SYSTEM_READER,
scope_types=['system'],
description='List endpoints for policy.',
operations=[{'path': ('/v3/policies/{policy_id}/OS-ENDPOINT-POLICY/'
'endpoints'),
'method': 'GET'}])
'method': 'GET'}],
deprecated_rule=deprecated_list_endpoints_for_policy,
deprecated_reason=DEPRECATED_REASON,
deprecated_since=versionutils.deprecated.TRAIN)
]

294
keystone/tests/unit/protection/v3/test_policy_association.py Normal file
View File

@ -0,0 +1,294 @@
# 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 six.moves import http_client
from keystone.common import provider_api
import keystone.conf
from keystone.tests.common import auth as common_auth
from keystone.tests import unit
from keystone.tests.unit import base_classes
from keystone.tests.unit import ksfixtures
CONF = keystone.conf.CONF
PROVIDERS = provider_api.ProviderAPIs
class _SystemUserPoliciesAssociationTests(object):
"""Common default functionality for all system users."""
def test_user_can_check_policy_association_for_endpoint(self):
policy = unit.new_policy_ref()
policy = PROVIDERS.policy_api.create_policy(policy['id'], policy)
service = PROVIDERS.catalog_api.create_service(
uuid.uuid4().hex, unit.new_service_ref()
)
endpoint = unit.new_endpoint_ref(service['id'], region_id=None)
endpoint = PROVIDERS.catalog_api.create_endpoint(
endpoint['id'], endpoint
)
PROVIDERS.endpoint_policy_api.create_policy_association(
policy['id'], endpoint['id'])
with self.test_client() as c:
c.get('/v3/policies/%s/OS-ENDPOINT-POLICY/endpoints/%s'
% (policy['id'], endpoint['id']),
headers=self.headers,
expected_status_code=http_client.NO_CONTENT)
def test_user_can_check_policy_association_for_service(self):
policy = unit.new_policy_ref()
policy = PROVIDERS.policy_api.create_policy(policy['id'], policy)
service = PROVIDERS.catalog_api.create_service(
uuid.uuid4().hex, unit.new_service_ref()
)
PROVIDERS.endpoint_policy_api.create_policy_association(
policy['id'], service_id=service['id'])
with self.test_client() as c:
c.get('/v3/policies/%s/OS-ENDPOINT-POLICY/services/%s'
% (policy['id'], service['id']),
headers=self.headers,
expected_status_code=http_client.NO_CONTENT)
def test_user_can_check_policy_association_for_region_and_service(self):
policy = unit.new_policy_ref()
policy = PROVIDERS.policy_api.create_policy(policy['id'], policy)
service = PROVIDERS.catalog_api.create_service(
uuid.uuid4().hex, unit.new_service_ref()
)
region = PROVIDERS.catalog_api.create_region(unit.new_region_ref())
PROVIDERS.endpoint_policy_api.create_policy_association(
policy['id'], service_id=service['id'], region_id=region['id']
)
with self.test_client() as c:
c.get('/v3/policies/%s/OS-ENDPOINT-POLICY/services/%s/regions/%s'
% (policy['id'], service['id'], region['id']),
headers=self.headers,
expected_status_code=http_client.NO_CONTENT)
def test_user_can_get_policy_for_endpoint(self):
policy = unit.new_policy_ref()
policy = PROVIDERS.policy_api.create_policy(policy['id'], policy)
service = PROVIDERS.catalog_api.create_service(
uuid.uuid4().hex, unit.new_service_ref()
)
endpoint = unit.new_endpoint_ref(service['id'], region_id=None)
endpoint = PROVIDERS.catalog_api.create_endpoint(
endpoint['id'], endpoint
)
PROVIDERS.endpoint_policy_api.create_policy_association(
policy['id'], endpoint['id']
)
with self.test_client() as c:
c.get('/v3/endpoints/%s/OS-ENDPOINT-POLICY/policy'
% (endpoint['id']),
headers=self.headers)
def test_user_list_endpoints_for_policy(self):
policy = unit.new_policy_ref()
policy = PROVIDERS.policy_api.create_policy(policy['id'], policy)
service = PROVIDERS.catalog_api.create_service(
uuid.uuid4().hex, unit.new_service_ref()
)
endpoint = unit.new_endpoint_ref(service['id'], region_id=None)
endpoint = PROVIDERS.catalog_api.create_endpoint(
endpoint['id'], endpoint
)
PROVIDERS.endpoint_policy_api.create_policy_association(
policy['id'], endpoint['id']
)
with self.test_client() as c:
r = c.get('/v3/endpoints/%s/OS-ENDPOINT-POLICY/policy'
% (endpoint['id']),
headers=self.headers)
self.assertIn(policy['id'], r.json['policy']['id'])
class _SystemReaderAndMemberPoliciesAssociationTests(object):
def test_user_cannot_create_policy_association_for_endpoint(self):
policy = unit.new_policy_ref()
policy = PROVIDERS.policy_api.create_policy(policy['id'], policy)
service = PROVIDERS.catalog_api.create_service(
uuid.uuid4().hex, unit.new_service_ref()
)
endpoint = unit.new_endpoint_ref(service['id'], region_id=None)
endpoint = PROVIDERS.catalog_api.create_endpoint(
endpoint['id'], endpoint
)
with self.test_client() as c:
c.put(
'/v3/policies/%s/OS-ENDPOINT-POLICY/endpoints/%s'
% (policy['id'], endpoint['id']),
headers=self.headers,
expected_status_code=http_client.FORBIDDEN
)
def test_user_cannot_delete_policy_association_for_endpoint(self):
policy = unit.new_policy_ref()
policy = PROVIDERS.policy_api.create_policy(policy['id'], policy)
service = PROVIDERS.catalog_api.create_service(
uuid.uuid4().hex, unit.new_service_ref()
)
endpoint = unit.new_endpoint_ref(service['id'], region_id=None)
endpoint = PROVIDERS.catalog_api.create_endpoint(
endpoint['id'], endpoint
)
with self.test_client() as c:
c.delete(
'/v3/policies/%s/OS-ENDPOINT-POLICY/endpoints/%s'
% (policy['id'], endpoint['id']),
headers=self.headers,
expected_status_code=http_client.FORBIDDEN
)
def test_user_cannot_create_policy_association_for_service(self):
policy = unit.new_policy_ref()
policy = PROVIDERS.policy_api.create_policy(policy['id'], policy)
service = PROVIDERS.catalog_api.create_service(
uuid.uuid4().hex, unit.new_service_ref()
)
with self.test_client() as c:
c.put(
'/v3/policies/%s/OS-ENDPOINT-POLICY/services/%s'
% (policy['id'], service['id']),
headers=self.headers,
expected_status_code=http_client.FORBIDDEN
)
def test_user_cannot_delete_policy_association_for_service(self):
policy = unit.new_policy_ref()
policy = PROVIDERS.policy_api.create_policy(policy['id'], policy)
service = PROVIDERS.catalog_api.create_service(
uuid.uuid4().hex, unit.new_service_ref()
)
with self.test_client() as c:
c.delete(
'/v3/policies/%s/OS-ENDPOINT-POLICY/services/%s'
% (policy['id'], service['id']),
headers=self.headers,
expected_status_code=http_client.FORBIDDEN
)
def test_user_cannot_create_policy_association_for_region_and_service(self):
policy = unit.new_policy_ref()
policy = PROVIDERS.policy_api.create_policy(policy['id'], policy)
service = PROVIDERS.catalog_api.create_service(
uuid.uuid4().hex, unit.new_service_ref()
)
region = PROVIDERS.catalog_api.create_region(unit.new_region_ref())
with self.test_client() as c:
c.put(
'/v3/policies/%s/OS-ENDPOINT-POLICY/services/%s/regions/%s'
% (policy['id'], service['id'], region['id']),
headers=self.headers,
expected_status_code=http_client.FORBIDDEN
)
def test_user_cannot_delete_policy_association_for_region_and_service(self):
policy = unit.new_policy_ref()
policy = PROVIDERS.policy_api.create_policy(policy['id'], policy)
service = PROVIDERS.catalog_api.create_service(
uuid.uuid4().hex, unit.new_service_ref()
)
region = PROVIDERS.catalog_api.create_region(unit.new_region_ref())
with self.test_client() as c:
c.delete(
'/v3/policies/%s/OS-ENDPOINT-POLICY/services/%s/regions/%s'
% (policy['id'], service['id'], region['id']),
headers=self.headers,
expected_status_code=http_client.FORBIDDEN
)
class SystemReaderTests(base_classes.TestCaseWithBootstrap,
common_auth.AuthTestMixin,
_SystemUserPoliciesAssociationTests,
_SystemReaderAndMemberPoliciesAssociationTests):
def setUp(self):
super(SystemReaderTests, self).setUp()
self.loadapp()
self.useFixture(ksfixtures.Policy(self.config_fixture))
self.config_fixture.config(group='oslo_policy', enforce_scope=True)
system_reader = unit.new_user_ref(
domain_id=CONF.identity.default_domain_id
)
self.user_id = PROVIDERS.identity_api.create_user(
system_reader
)['id']
PROVIDERS.assignment_api.create_system_grant_for_user(
self.user_id, self.bootstrapper.reader_role_id
)
auth = self.build_authentication_request(
user_id=self.user_id, password=system_reader['password'],
system=True
)
# Grab a token using the persona we're testing and prepare headers
# for requests we'll be making in the tests.
with self.test_client() as c:
r = c.post('/v3/auth/tokens', json=auth)
self.token_id = r.headers['X-Subject-Token']
self.headers = {'X-Auth-Token': self.token_id}
class SystemMemberTests(base_classes.TestCaseWithBootstrap,
common_auth.AuthTestMixin,
_SystemUserPoliciesAssociationTests,
_SystemReaderAndMemberPoliciesAssociationTests):
def setUp(self):
super(SystemMemberTests, self).setUp()
self.loadapp()
self.useFixture(ksfixtures.Policy(self.config_fixture))
self.config_fixture.config(group='oslo_policy', enforce_scope=True)
system_member = unit.new_user_ref(
domain_id=CONF.identity.default_domain_id
)
self.user_id = PROVIDERS.identity_api.create_user(
system_member
)['id']
PROVIDERS.assignment_api.create_system_grant_for_user(
self.user_id, self.bootstrapper.member_role_id
)
auth = self.build_authentication_request(
user_id=self.user_id, password=system_member['password'],
system=True
)
# Grab a token using the persona we're testing and prepare headers
# for requests we'll be making in the tests.
with self.test_client() as c:
r = c.post('/v3/auth/tokens', json=auth)
self.token_id = r.headers['X-Subject-Token']
self.headers = {'X-Auth-Token': self.token_id}