openstack/octavia
Code Issues Proposed changes

Fixes API list handling of unscoped tokens

Browse Source 
The API list methods were not handling unscoped tokens correctly.
If the API is using the admin_or_owner-policy.yaml policy override file,
and a user used an unscoped token, the API will list objects for all
projects. This patch corrects that issue.
If you are using the default policies, the API handles unscoped tokens
correctly.

Change-Id: I88e64fd5e8a4c709f735be85b85139dbb52e4acd
This commit is contained in:
Michael Johnson 2020-09-21 15:18:11 -07:00
parent e45ecc2111
commit 9453701fb4
10 changed files with 314 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
octavia/api/v2/controllers/base.py
View File

@ -232,6 +232,10 @@ class BaseController(pecan_rest.RestController):
if project_id is None:
project_id = context.project_id
# If we still don't know who it is, reject it.
if project_id is None:
raise exceptions.PolicyForbidden()
# Check authorization to list objects under this project
self._auth_validate_action(context, project_id,
constants.RBAC_GET_ALL)

2
octavia/tests/functional/api/v2/test_flavors.py
View File

@ -292,7 +292,6 @@ class TestFlavors(base.BaseAPITest):
flavor2 = self.create_flavor('name2', 'description', self.fp.get('id'),
True)
self.assertTrue(uuidutils.is_uuid_like(flavor2.get('id')))
response = self.get(self.FLAVORS_PATH)
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.api_settings.get('auth_strategy')
@ -316,6 +315,7 @@ class TestFlavors(base.BaseAPITest):
with mock.patch(
"oslo_context.context.RequestContext.to_policy_values",
return_value=override_credentials):
response = self.get(self.FLAVORS_PATH)
api_list = response.json.get(self.root_tag_list)
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
self.assertEqual(2, len(api_list))

52
octavia/tests/functional/api/v2/test_health_monitor.py
View File

@ -306,6 +306,58 @@ class TestHealthMonitor(base.BaseAPITest):
hm_id_protocols = [(hm.get('id'), hm.get('type')) for hm in hms]
self.assertIn((hm3.get('id'), hm3.get('type')), hm_id_protocols)
def test_get_all_unscoped_token(self):
project_id = uuidutils.generate_uuid()
lb1 = self.create_load_balancer(uuidutils.generate_uuid(), name='lb1',
project_id=project_id)
lb1_id = lb1.get('loadbalancer').get('id')
self.set_lb_status(lb1_id)
pool1 = self.create_pool(
lb1_id, constants.PROTOCOL_HTTP,
constants.LB_ALGORITHM_ROUND_ROBIN).get('pool')
self.set_lb_status(lb1_id)
pool2 = self.create_pool(
lb1_id, constants.PROTOCOL_HTTPS,
constants.LB_ALGORITHM_ROUND_ROBIN).get('pool')
self.set_lb_status(lb1_id)
self.create_health_monitor(
pool1.get('id'), constants.HEALTH_MONITOR_HTTP,
1, 1, 1, 1).get(self.root_tag)
self.set_lb_status(lb1_id)
self.create_health_monitor(
pool2.get('id'), constants.HEALTH_MONITOR_PING,
1, 1, 1, 1).get(self.root_tag)
self.set_lb_status(lb1_id)
self.create_health_monitor(
self.pool_id, constants.HEALTH_MONITOR_TCP,
1, 1, 1, 1).get(self.root_tag)
self.set_lb_status(self.lb_id)
auth_strategy = self.conf.conf.api_settings.get('auth_strategy')
self.conf.config(group='api_settings',
auth_strategy=constants.KEYSTONE)
with mock.patch.object(octavia.common.context.Context, 'project_id',
None):
override_credentials = {
'service_user_id': None,
'user_domain_id': None,
'is_admin_project': True,
'service_project_domain_id': None,
'service_project_id': None,
'roles': ['load-balancer_member'],
'user_id': None,
'is_admin': False,
'service_user_domain_id': None,
'project_domain_id': None,
'service_roles': [],
'project_id': None}
with mock.patch(
"oslo_context.context.RequestContext.to_policy_values",
return_value=override_credentials):
result = self.get(self.HMS_PATH, status=403).json
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
self.assertEqual(self.NOT_AUTHORIZED_BODY, result)
def test_get_all_non_admin_global_observer(self):
project_id = uuidutils.generate_uuid()
lb1 = self.create_load_balancer(uuidutils.generate_uuid(), name='lb1',

52
octavia/tests/functional/api/v2/test_l7policy.py
View File

@ -224,6 +224,58 @@ class TestL7Policy(base.BaseAPITest):
self.assertIn((api_l7p_c.get('id'), api_l7p_c.get('action')),
policy_id_actions)
def test_get_all_unscoped_token(self):
project_id = uuidutils.generate_uuid()
lb1 = self.create_load_balancer(uuidutils.generate_uuid(), name='lb1',
project_id=project_id)
lb1_id = lb1.get('loadbalancer').get('id')
self.set_lb_status(lb1_id)
listener1 = self.create_listener(constants.PROTOCOL_HTTP, 80,
lb1_id)
listener1_id = listener1.get('listener').get('id')
self.set_lb_status(lb1_id)
pool1 = self.create_pool(lb1_id, constants.PROTOCOL_HTTP,
constants.LB_ALGORITHM_ROUND_ROBIN)
pool1_id = pool1.get('pool').get('id')
self.set_lb_status(lb1_id)
self.create_l7policy(
listener1_id,
constants.L7POLICY_ACTION_REJECT).get(self.root_tag)
self.set_lb_status(lb1_id)
self.create_l7policy(
listener1_id, constants.L7POLICY_ACTION_REDIRECT_TO_POOL,
position=2, redirect_pool_id=pool1_id).get(self.root_tag)
self.set_lb_status(lb1_id)
self.create_l7policy(
self.listener_id, constants.L7POLICY_ACTION_REDIRECT_TO_URL,
redirect_url='http://localhost/').get(self.root_tag)
self.set_lb_status(lb1_id)
auth_strategy = self.conf.conf.api_settings.get('auth_strategy')
self.conf.config(group='api_settings',
auth_strategy=constants.KEYSTONE)
with mock.patch.object(octavia.common.context.Context, 'project_id',
None):
override_credentials = {
'service_user_id': None,
'user_domain_id': None,
'is_admin_project': True,
'service_project_domain_id': None,
'service_project_id': None,
'roles': ['load-balancer_member'],
'user_id': None,
'is_admin': False,
'service_user_domain_id': None,
'project_domain_id': None,
'service_roles': [],
'project_id': None}
with mock.patch(
"oslo_context.context.RequestContext.to_policy_values",
return_value=override_credentials):
result = self.get(self.L7POLICIES_PATH, status=403).json
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
self.assertEqual(self.NOT_AUTHORIZED_BODY, result)
def test_get_all_non_admin_global_observer(self):
project_id = uuidutils.generate_uuid()
lb1 = self.create_load_balancer(uuidutils.generate_uuid(), name='lb1',

36
octavia/tests/functional/api/v2/test_l7rule.py
View File

@ -194,6 +194,42 @@ class TestL7Rule(base.BaseAPITest):
self.assertIn((api_l7r_b.get('id'), api_l7r_b.get('type')),
rule_id_types)
def test_get_all_unscoped_token(self):
self.create_l7rule(
self.l7policy_id, constants.L7RULE_TYPE_PATH,
constants.L7RULE_COMPARE_TYPE_STARTS_WITH,
'/api').get(self.root_tag)
self.set_lb_status(self.lb_id)
self.create_l7rule(
self.l7policy_id, constants.L7RULE_TYPE_COOKIE,
constants.L7RULE_COMPARE_TYPE_CONTAINS, 'some-value',
key='some-cookie').get(self.root_tag)
self.set_lb_status(self.lb_id)
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.api_settings.get('auth_strategy')
self.conf.config(group='api_settings', auth_strategy=constants.TESTING)
with mock.patch.object(octavia.common.context.Context, 'project_id',
None):
override_credentials = {
'service_user_id': None,
'user_domain_id': None,
'is_admin_project': True,
'service_project_domain_id': None,
'service_project_id': None,
'roles': ['load-balancer_member'],
'user_id': None,
'is_admin': False,
'service_user_domain_id': None,
'project_domain_id': None,
'service_roles': [],
'project_id': None}
with mock.patch(
"oslo_context.context.RequestContext.to_policy_values",
return_value=override_credentials):
result = self.get(self.l7rules_path, status=403).json
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
self.assertEqual(self.NOT_AUTHORIZED_BODY, result)
def test_get_all_not_authorized(self):
self.create_l7rule(
self.l7policy_id, constants.L7RULE_TYPE_PATH,

41
octavia/tests/functional/api/v2/test_listener.py
View File

@ -131,6 +131,47 @@ class TestListener(base.BaseAPITest):
self.assertIn((listener3.get('id'), listener3.get('protocol_port')),
listener_id_ports)
def test_get_all_unscoped_token(self):
project_id = uuidutils.generate_uuid()
lb1 = self.create_load_balancer(uuidutils.generate_uuid(),
name='lb1', project_id=project_id)
lb1_id = lb1.get('loadbalancer').get('id')
self.set_lb_status(lb1_id)
self.create_listener(constants.PROTOCOL_HTTP, 80,
lb1_id)
self.set_lb_status(lb1_id)
self.create_listener(constants.PROTOCOL_HTTP, 81,
lb1_id)
self.set_lb_status(lb1_id)
self.create_listener(constants.PROTOCOL_HTTP, 82,
self.lb_id).get(self.root_tag)
self.set_lb_status(self.lb_id)
auth_strategy = self.conf.conf.api_settings.get('auth_strategy')
self.conf.config(group='api_settings',
auth_strategy=constants.KEYSTONE)
with mock.patch.object(octavia.common.context.Context, 'project_id',
None):
override_credentials = {
'service_user_id': None,
'user_domain_id': None,
'is_admin_project': True,
'service_project_domain_id': None,
'service_project_id': None,
'roles': ['load-balancer_member'],
'user_id': None,
'is_admin': False,
'service_user_domain_id': None,
'project_domain_id': None,
'service_roles': [],
'project_id': None}
with mock.patch(
"oslo_context.context.RequestContext.to_policy_values",
return_value=override_credentials):
result = self.get(self.LISTENERS_PATH, status=403).json
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
self.assertEqual(self.NOT_AUTHORIZED_BODY, result)
def test_get_all_non_admin_global_observer(self):
project_id = uuidutils.generate_uuid()
lb1 = self.create_load_balancer(uuidutils.generate_uuid(),

35
octavia/tests/functional/api/v2/test_load_balancer.py
View File

@ -1194,6 +1194,41 @@ class TestLoadBalancer(base.BaseAPITest):
lb_id_names = [(lb.get('id'), lb.get('name')) for lb in lbs]
self.assertIn((lb3.get('id'), lb3.get('name')), lb_id_names)
def test_get_all_unscoped_token(self):
project_id = uuidutils.generate_uuid()
self.create_load_balancer(uuidutils.generate_uuid(),
name='lb1', project_id=project_id)
self.create_load_balancer(uuidutils.generate_uuid(),
name='lb2', project_id=project_id)
lb3 = self.create_load_balancer(uuidutils.generate_uuid(),
name='lb3', project_id=self.project_id)
lb3 = lb3.get(self.root_tag)
auth_strategy = self.conf.conf.api_settings.get('auth_strategy')
self.conf.config(group='api_settings',
auth_strategy=constants.KEYSTONE)
with mock.patch.object(octavia.common.context.Context, 'project_id',
None):
override_credentials = {
'service_user_id': None,
'user_domain_id': None,
'is_admin_project': True,
'service_project_domain_id': None,
'service_project_id': None,
'roles': ['load-balancer_member'],
'user_id': None,
'is_admin': False,
'service_user_domain_id': None,
'project_domain_id': None,
'service_roles': [],
'project_id': None}
with mock.patch(
"oslo_context.context.RequestContext.to_policy_values",
return_value=override_credentials):
result = self.get(self.LBS_PATH, status=403).json
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
self.assertEqual(self.NOT_AUTHORIZED_BODY, result)
def test_get_all_non_admin_global_observer(self):
project_id = uuidutils.generate_uuid()
lb1 = self.create_load_balancer(uuidutils.generate_uuid(),

39
octavia/tests/functional/api/v2/test_member.py
View File

@ -212,6 +212,45 @@ class TestMember(base.BaseAPITest):
for m in [api_m_1, api_m_2]:
self.assertIn(m, response)
def test_get_all_unscoped_token(self):
api_m_1 = self.create_member(
self.pool_id, '192.0.2.1', 80).get(self.root_tag)
self.set_lb_status(self.lb_id)
api_m_2 = self.create_member(
self.pool_id, '192.0.2.2', 80).get(self.root_tag)
self.set_lb_status(self.lb_id)
# Original objects didn't have the updated operating/provisioning
# status that exists in the DB.
for m in [api_m_1, api_m_2]:
m['operating_status'] = constants.ONLINE
m['provisioning_status'] = constants.ACTIVE
m.pop('updated_at')
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.api_settings.get('auth_strategy')
self.conf.config(group='api_settings', auth_strategy=constants.TESTING)
with mock.patch.object(octavia.common.context.Context, 'project_id',
None):
override_credentials = {
'service_user_id': None,
'user_domain_id': None,
'is_admin_project': True,
'service_project_domain_id': None,
'service_project_id': None,
'roles': ['load-balancer_member'],
'user_id': None,
'is_admin': False,
'service_user_domain_id': None,
'project_domain_id': None,
'service_roles': [],
'project_id': None}
with mock.patch(
"oslo_context.context.RequestContext.to_policy_values",
return_value=override_credentials):
result = self.get(self.members_path, status=403).json
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
self.assertEqual(self.NOT_AUTHORIZED_BODY, result)
def test_get_all_not_authorized(self):
api_m_1 = self.create_member(
self.pool_id, '192.0.2.1', 80).get(self.root_tag)

44
octavia/tests/functional/api/v2/test_pool.py
View File

@ -263,6 +263,50 @@ class TestPool(base.BaseAPITest):
self.assertIn((pool3.get('id'), pool3.get('protocol')),
pool_id_protocols)
def test_get_all_unscoped_token(self):
project_id = uuidutils.generate_uuid()
lb1 = self.create_load_balancer(uuidutils.generate_uuid(), name='lb1',
project_id=project_id)
lb1_id = lb1.get('loadbalancer').get('id')
self.set_lb_status(lb1_id)
self.create_pool(
lb1_id, constants.PROTOCOL_HTTP,
constants.LB_ALGORITHM_ROUND_ROBIN).get(self.root_tag)
self.set_lb_status(lb1_id)
self.create_pool(
lb1_id, constants.PROTOCOL_HTTPS,
constants.LB_ALGORITHM_ROUND_ROBIN).get(self.root_tag)
self.set_lb_status(lb1_id)
self.create_pool(
self.lb_id, constants.PROTOCOL_TCP,
constants.LB_ALGORITHM_ROUND_ROBIN).get(self.root_tag)
self.set_lb_status(self.lb_id)
auth_strategy = self.conf.conf.api_settings.get('auth_strategy')
self.conf.config(group='api_settings',
auth_strategy=constants.KEYSTONE)
with mock.patch.object(octavia.common.context.Context, 'project_id',
None):
override_credentials = {
'service_user_id': None,
'user_domain_id': None,
'is_admin_project': True,
'service_project_domain_id': None,
'service_project_id': None,
'roles': ['load-balancer_member'],
'user_id': None,
'is_admin': False,
'service_user_domain_id': None,
'project_domain_id': None,
'service_roles': [],
'project_id': None}
with mock.patch(
"oslo_context.context.RequestContext.to_policy_values",
return_value=override_credentials):
result = self.get(self.POOLS_PATH, status=403).json
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
self.assertEqual(self.NOT_AUTHORIZED_BODY, result)
def test_get_all_non_admin_global_observer(self):
project_id = uuidutils.generate_uuid()
lb1 = self.create_load_balancer(uuidutils.generate_uuid(), name='lb1',

10
releasenotes/notes/get-all-unscoped-token-61da95856bc662e0.yaml Normal file
View File

@ -0,0 +1,10 @@
---
security:
- |
If you are using the admin_or_owner-policy.yaml policy override file
you should upgrade your API processes to include the unscoped token fix.
The default policies are not affected by this issue.
fixes:
- |
Fixes an issue when using the admin_or_owner-policy.yaml policy override
file and unscoped tokens.