openstack/octavia
Code Issues Proposed changes

Add RBAC enforcement to Octavia v2 API

Browse Source 
This patch adds policies and enforcement to the Octavia v2 API for
load balancers and listeners.  Child patches will add the rest of the API.

In this patch I also correct some improper functional tests.

Change-Id: Id8a2d15c117c54bd45fc8bb76bf71aff1b3c8fe9
Closes-Bug: #1690481
This commit is contained in:
Michael Johnson 2017-06-09 18:53:18 -07:00 committed by Nir Magnezi
parent e5ac4a0426
commit 0ce46fe8d0
22 changed files with 1267 additions and 67 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
devstack/plugin.sh
View File

@ -457,6 +457,10 @@ function octavia_start {
run_process $OCTAVIA_HOUSEKEEPER "$OCTAVIA_HOUSEKEEPER_BINARY $OCTAVIA_HOUSEKEEPER_ARGS"
run_process $OCTAVIA_HEALTHMANAGER "$OCTAVIA_HEALTHMANAGER_BINARY $OCTAVIA_HEALTHMANAGER_ARGS"
if [ $OCTAVIA_NODE == 'main' ] || [ $OCTAVIA_NODE == 'standalone' ] ; then
add_load-balancer_roles
fi
}
function octavia_stop {
@ -516,6 +520,14 @@ function octavia_cleanup {
sudo rm -rf $NOVA_STATE_PATH $NOVA_AUTH_CACHE_DIR
}
function add_load-balancer_roles {
openstack role create load-balancer_observer
openstack role create load-balancer_global_observer
openstack role create load-balancer_member
openstack role create load-balancer_admin
openstack role add --user demo --project demo load-balancer_member
}
# check for service enabled
if is_service_enabled $OCTAVIA; then
if [ $OCTAVIA_NODE == 'main' ] || [ $OCTAVIA_NODE == 'standalone' ] ; then # main-ha node stuff only

9
doc/source/conf.py
View File

@ -41,7 +41,8 @@ extensions = ['sphinx.ext.autodoc',
'sphinxcontrib.nwdiag',
'sphinx.ext.graphviz',
'oslosphinx',
'oslo_config.sphinxext'
'oslo_config.sphinxext',
'oslo_policy.sphinxpolicygen'
]
todo_include_todos = True
@ -129,6 +130,8 @@ html_theme = 'nature'
# pixels large.
#html_favicon = None
html_static_path = ['_static']
# If not '', a 'Last updated on:' timestamp is inserted at every page bottom,
# using the given strftime format.
git_cmd = ["git", "log", "--pretty=format:'%ad, commit %h'", "--date=local",
@ -278,3 +281,7 @@ epub_copyright = u'2014, OpenStack Octavia Team'
# Allow duplicate toc entries.
#epub_tocdup = True
# RBAC sample policy file generation
policy_generator_config_file = '../../etc/policy/octavia-policy-generator.conf'
sample_policy_basename = '_static/octavia'

1
doc/source/index.rst
View File

@ -44,6 +44,7 @@ For operators
guides/dev-quick-start.rst
guides/operator-maintenance.rst
main/configref.rst
main/policy.rst
main/Anchor.rst
devref/apache-httpd.rst

72
doc/source/main/policy.rst Normal file
View File

@ -0,0 +1,72 @@
================
Octavia Policies
================
The default policy is to not allow access unless the auth_strategy is 'noauth'.
Users must be a member of one of the following roles to have access to
the load-balancer API:
.. glossary::
role:load-balancer_observer
User has access to load-balancer read-only APIs
role:load-balancer_global_observer
User has access to load-balancer read-only APIs including resources
owned by others.
role:load-balancer_member
User has access to load-balancer read and write APIs
role:load-balancer_admin
User is considered an admin for all load-balnacer APIs including
resources owned by others.
role:admin
User is admin to all APIs
.. note::
'is_admin:True' is a policy rule that takes into account the
auth_strategy == noauth configuration setting.
It is equivalent to 'rule:context_is_admin or {auth_strategy == noauth}'
if that would be valid syntax.
Sample File Generation
----------------------
To generate a sample policy.yaml file from the Octavia defaults, run the
oslo policy generation script::
oslopolicy-sample-generator
--config-file etc/policy/octavia-policy-generator.conf
--output-file policy.yaml.sample
Merged File Generation
----------------------
This will output a policy file which includes all registered policy defaults
and all policies configured with a policy file. This file shows the effective
policy in use by the project::
oslopolicy-policy-generator
--config-file etc/policy/octavia-policy-generator.conf
This tool uses the output_file path from the config-file.
List Redundant Configurations
-----------------------------
This will output a list of matches for policy rules that are defined in a
configuration file where the rule does not differ from a registered default
rule. These are rules that can be removed from the policy file with no change
in effective policy::
oslopolicy-list-redundant
--config-file etc/policy/octavia-policy-generator.conf
Default Octavia Policies
------------------------
.. literalinclude:: ../_static/octavia.policy.yaml.sample

14
etc/policy/README.rst Normal file
View File

@ -0,0 +1,14 @@
===========================
Octavia Sample Policy Files
===========================
The sample policy.json files described here can be copied into
/etc/octavia/policy.json to override the default RBAC policy for Octavia.
admin_or_owner-policy.json
--------------------------
This policy file disables the requirement for load-balancer service users to
have one of the load-balancer:* roles. It provides a similar policy to
legacy OpenStack policies where any user or admin has access to load-balancer
resources that they own. Users with the admin role has access to all
load-balancer resources, whether they own them or not.

8
etc/policy/admin_or_owner-policy.json Normal file
View File

@ -0,0 +1,8 @@
{
"context_is_admin": "role:admin or role:load-balancer_admin",
"admin_or_owner": "is_admin:True or project_id:%(project_id)s",
"load-balancer:read": "rule:admin_or_owner",
"load-balancer:read-global": "is_admin:True",
"load-balancer:write": "rule:admin_or_owner"
}

3
etc/policy/octavia-policy-generator.conf Normal file
View File

@ -0,0 +1,3 @@
[DEFAULT]
output_file = etc/policy.yaml.sample
namespace = octavia

54
octavia/api/v2/controllers/listener.py
View File

@ -60,6 +60,13 @@ class ListenersController(base.BaseController):
"""Gets a single listener's details."""
context = pecan.request.context.get('octavia_context')
db_listener = self._get_db_listener(context.session, id)
# Check that the user is authorized to show this listener
action = '{rbac_obj}{action}'.format(
rbac_obj=constants.RBAC_LISTENER, action='get_one')
target = {'project_id': db_listener.project_id}
context.policy.authorize(action, target)
result = self._convert_db_to_type(db_listener,
listener_types.ListenerResponse)
return listener_types.ListenerRootResponse(listener=result)
@ -70,17 +77,31 @@ class ListenersController(base.BaseController):
"""Lists all listeners."""
pcontext = pecan.request.context
context = pcontext.get('octavia_context')
if context.is_admin or CONF.auth_strategy == constants.NOAUTH:
if project_id:
project_id = {'project_id': project_id}
else:
project_id = {}
# Check that the user is authorized to list lbs under all projects
action = '{rbac_obj}{action}'.format(
rbac_obj=constants.RBAC_LISTENER, action='get_all-global')
target = {'project_id': project_id}
if not context.policy.authorize(action, target, do_raise=False):
# Not a global observer or admin
if project_id is None:
project_id = context.project_id
# Check that the user is authorized to list lbs under this project
action = '{rbac_obj}{action}'.format(
rbac_obj=constants.RBAC_LISTENER, action='get_all')
target = {'project_id': project_id}
context.policy.authorize(action, target)
if project_id is None:
query_filter = {}
else:
project_id = {'project_id': context.project_id}
query_filter = {'project_id': project_id}
db_listeners, links = self.repositories.listener.get_all(
context.session, show_deleted=False,
pagination_helper=pcontext.get(constants.PAGINATION_HELPER),
**project_id)
**query_filter)
result = self._convert_db_to_type(
db_listeners, [listener_types.ListenerResponse])
return listener_types.ListenersRootResponse(
@ -190,6 +211,12 @@ class ListenersController(base.BaseController):
listener.project_id = self._get_lb_project_id(
context.session, load_balancer_id)
# Check that the user is authorized to create under this project
action = '{rbac_obj}{action}'.format(
rbac_obj=constants.RBAC_LISTENER, action='post')
target = {'project_id': listener.project_id}
context.policy.authorize(action, target)
lock_session = db_api.get_session(autocommit=False)
if self.repositories.check_quota_met(
context.session,
@ -258,6 +285,12 @@ class ListenersController(base.BaseController):
db_listener = self._get_db_listener(context.session, id)
load_balancer_id = db_listener.load_balancer_id
# Check that the user is authorized to update this listener
action = '{rbac_obj}{action}'.format(
rbac_obj=constants.RBAC_LISTENER, action='put')
target = {'project_id': db_listener.project_id}
context.policy.authorize(action, target)
# TODO(rm_work): Do we need something like this? What do we do on an
# empty body for a PUT?
if not listener:
@ -293,6 +326,13 @@ class ListenersController(base.BaseController):
context = pecan.request.context.get('octavia_context')
db_listener = self._get_db_listener(context.session, id)
load_balancer_id = db_listener.load_balancer_id
# Check that the user is authorized to delete this listener
action = '{rbac_obj}{action}'.format(
rbac_obj=constants.RBAC_LISTENER, action='delete')
target = {'project_id': db_listener.project_id}
context.policy.authorize(action, target)
self._test_lb_and_listener_statuses(
context.session, load_balancer_id,
id=id, listener_status=constants.PENDING_DELETE)

64
octavia/api/v2/controllers/load_balancer.py
View File

@ -51,6 +51,13 @@ class LoadBalancersController(base.BaseController):
"""Gets a single load balancer's details."""
context = pecan.request.context.get('octavia_context')
load_balancer = self._get_db_lb(context.session, id)
# Check that the user is authorized to show this lb
action = '{rbac_obj}{action}'.format(
rbac_obj=constants.RBAC_LOADBALANCER, action='get_one')
target = {'project_id': load_balancer.project_id}
context.policy.authorize(action, target)
result = self._convert_db_to_type(
load_balancer, lb_types.LoadBalancerResponse)
return lb_types.LoadBalancerRootResponse(loadbalancer=result)
@ -61,17 +68,31 @@ class LoadBalancersController(base.BaseController):
"""Lists all load balancers."""
pcontext = pecan.request.context
context = pcontext.get('octavia_context')
if context.is_admin or CONF.auth_strategy == constants.NOAUTH:
if project_id:
project_id = {'project_id': project_id}
else:
project_id = {}
# Check that the user is authorized to list lbs under all projects
action = '{rbac_obj}{action}'.format(
rbac_obj=constants.RBAC_LOADBALANCER, action='get_all-global')
target = {'project_id': project_id}
if not context.policy.authorize(action, target, do_raise=False):
# Not a global observer or admin
if project_id is None:
project_id = context.project_id
# Check that the user is authorized to list lbs under this project
action = '{rbac_obj}{action}'.format(
rbac_obj=constants.RBAC_LOADBALANCER, action='get_all')
target = {'project_id': project_id}
context.policy.authorize(action, target)
if project_id is None:
query_filter = {}
else:
project_id = {'project_id': context.project_id}
query_filter = {'project_id': project_id}
load_balancers, links = self.repositories.load_balancer.get_all(
context.session, show_deleted=False,
pagination_helper=pcontext.get(constants.PAGINATION_HELPER),
**project_id)
**query_filter)
result = self._convert_db_to_type(
load_balancers, [lb_types.LoadBalancerResponse])
return lb_types.LoadBalancersRootResponse(
@ -161,15 +182,18 @@ class LoadBalancersController(base.BaseController):
load_balancer = load_balancer.loadbalancer
context = pecan.request.context.get('octavia_context')
project_id = context.project_id
if context.is_admin or CONF.auth_strategy == constants.NOAUTH:
if load_balancer.project_id:
project_id = load_balancer.project_id
if not load_balancer.project_id and context.project_id:
load_balancer.project_id = context.project_id
if not project_id:
if not load_balancer.project_id:
raise exceptions.ValidationException(detail=_(
"Missing project ID in request where one is required."))
load_balancer.project_id = project_id
# Check that the user is authorized to create under this project
action = '{rbac_obj}{action}'.format(
rbac_obj=constants.RBAC_LOADBALANCER, action='post')
target = {'project_id': load_balancer.project_id}
context.policy.authorize(action, target)
self._validate_vip_request_object(load_balancer)
@ -339,6 +363,13 @@ class LoadBalancersController(base.BaseController):
load_balancer = load_balancer.loadbalancer
context = pecan.request.context.get('octavia_context')
db_lb = self._get_db_lb(context.session, id)
# Check that the user is authorized to update this lb
action = '{rbac_obj}{action}'.format(
rbac_obj=constants.RBAC_LOADBALANCER, action='put')
target = {'project_id': db_lb.project_id}
context.policy.authorize(action, target)
self._test_lb_status(context.session, id)
try:
LOG.info("Sending updated Load Balancer %s to the handler", id)
@ -357,6 +388,13 @@ class LoadBalancersController(base.BaseController):
context = pecan.request.context.get('octavia_context')
cascade = strutils.bool_from_string(cascade)
db_lb = self._get_db_lb(context.session, id)
# Check that the user is authorized to delete this lb
action = '{rbac_obj}{action}'.format(
rbac_obj=constants.RBAC_LOADBALANCER, action='delete')
target = {'project_id': db_lb.project_id}
context.policy.authorize(action, target)
with db_api.get_lock_session() as lock_session:
self._test_lb_status(lock_session, id,
lb_status=constants.PENDING_DELETE)

9
octavia/common/constants.py
View File

@ -421,3 +421,12 @@ ALLOWED_SORT_DIR = (ASC, DESC)
DEFAULT_SORT_DIR = ASC
DEFAULT_SORT_KEYS = ['created_at', 'id']
DEFAULT_PAGE_SIZE = 1000
# RBAC
LOADBALANCER_API = 'os_load-balancer_api'
RULE_API_READ = 'rule:load-balancer:read'
RULE_API_READ_GLOBAL = 'rule:load-balancer:read-global'
RULE_API_WRITE = 'rule:load-balancer:write'
RULE_ANY = '@'
RBAC_LOADBALANCER = '{}:loadbalancer:'.format(LOADBALANCER_API)
RBAC_LISTENER = '{}:listener:'.format(LOADBALANCER_API)

11
octavia/common/context.py
View File

@ -12,25 +12,32 @@
# License for the specific language governing permissions and limitations
# under the License.
from oslo_config import cfg
from oslo_context import context as common_context
from octavia.common import constants
from octavia.common import policy
from octavia.db import api as db_api
CONF = cfg.CONF
class Context(common_context.RequestContext):
_session = None
def __init__(self, user=None, project_id=None, is_admin=False, **kwargs):
def __init__(self, user_id=None, project_id=None, **kwargs):
if project_id:
kwargs['tenant'] = project_id
super(Context, self).__init__(is_admin=is_admin, **kwargs)
super(Context, self).__init__(**kwargs)
self.policy = policy.Policy(self)
self.is_admin = (self.policy.check_is_admin() or
CONF.auth_strategy == constants.NOAUTH)
@property
def session(self):
if self._session is None:

12
octavia/common/policy.py
View File

@ -18,6 +18,7 @@ from oslo_config import cfg
from oslo_policy import policy as oslo_policy
from oslo_utils import excutils
from octavia.common import config
from octavia.common import exceptions
from octavia import policies
@ -57,7 +58,6 @@ class Policy(oslo_policy.Enforcer):
def authorize(self, action, target, do_raise=True, exc=None):
"""Verifies that the action is valid on the target in this context.
:param context: nova context
:param action: string representing the action to be checked
this should be colon separated for clarity.
i.e. ``compute:create_instance``,
@ -84,6 +84,10 @@ class Policy(oslo_policy.Enforcer):
do_raise is False.
"""
credentials = self.context.to_policy_values()
# Inject is_admin into the credentials to allow override via
# config auth_strategy = constants.NOAUTH
credentials['is_admin'] = self.context.is_admin
if not exc:
exc = exceptions.NotAuthorized
@ -127,3 +131,9 @@ class IsAdminCheck(oslo_policy.Check):
"""Determine whether is_admin matches the requested value."""
return creds['is_admin'] == self.expected
# This is used for the oslopolicy-policy-generator tool
def get_no_context_enforcer():
config.init([])
return Policy(None)

4
octavia/policies/__init__.py
View File

@ -14,9 +14,13 @@
import itertools
from octavia.policies import base
from octavia.policies import listener
from octavia.policies import loadbalancer
def list_rules():
return itertools.chain(
base.list_rules(),
loadbalancer.list_rules(),
listener.list_rules(),
)

54
octavia/policies/base.py
View File

@ -13,10 +13,56 @@
from oslo_policy import policy
rules = [
policy.RuleDefault('context_is_admin', 'role:admin'),
policy.RuleDefault('admin_or_owner',
'is_admin:True or project_id:%(project_id)s'),
policy.RuleDefault('admin_api', 'is_admin:True'),
# The default is to not allow access unless the auth_strategy is 'noauth'.
# Users must be a member of one of the following roles to have access to
# the load-balancer API:
#
# role:load-balancer_observer
# User has access to load-balancer read-only APIs
# role:load-balancer_global_observer
# User has access to load-balancer read-only APIs including resources
# owned by others.
# role:load-balancer_member
# User has access to load-balancer read and write APIs
# role:load-balancer_admin
# User is considered an admin for all load-balnacer APIs including
# resources owned by others.
# role:admin
# User is admin to all APIs
policy.RuleDefault('context_is_admin',
'role:admin or role:load-balancer_admin'),
# Note: 'is_admin:True' is a policy rule that takes into account the
# auth_strategy == noauth configuration setting.
# It is equivalent to 'rule:context_is_admin or {auth_strategy == noauth}'
policy.RuleDefault('load-balancer:owner', 'project_id:%(project_id)s'),
# API access roles
policy.RuleDefault('load-balancer:observer_and_owner',
'role:load-balancer_observer and '
'rule:load-balancer:owner'),
policy.RuleDefault('load-balancer:global_observer',
'role:load-balancer_global_observer'),
policy.RuleDefault('load-balancer:member_and_owner',
'role:load-balancer_member and '
'rule:load-balancer:owner'),
# API access methods
policy.RuleDefault('load-balancer:read',
'rule:load-balancer:observer_and_owner or '
'rule:load-balancer:global_observer or '
'rule:load-balancer:member_and_owner or is_admin:True'),
policy.RuleDefault('load-balancer:read-global',
'rule:load-balancer:global_observer or '
'is_admin:True'),
policy.RuleDefault('load-balancer:write',
'rule:load-balancer:member_and_owner or is_admin:True'),
]

75
octavia/policies/listener.py Normal file
View File

@ -0,0 +1,75 @@
# Copyright 2017 Rackspace, US Inc.
# 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.
from octavia.common import constants
from oslo_policy import policy
rules = [
policy.DocumentedRuleDefault(
'{rbac_obj}{action}'.format(rbac_obj=constants.RBAC_LISTENER,
action='get_all'),
constants.RULE_API_READ,
"List Listeners",
[{'method': 'GET', 'path': '/v2.0/lbaas/listeners'}]
),
policy.DocumentedRuleDefault(
'{rbac_obj}{action}'.format(rbac_obj=constants.RBAC_LISTENER,
action='get_all-global'),
constants.RULE_API_READ_GLOBAL,
"List Listeners including resources owned by others",
[{'method': 'GET', 'path': '/v2.0/lbaas/listeners'}]
),
policy.DocumentedRuleDefault(
'{rbac_obj}{action}'.format(rbac_obj=constants.RBAC_LISTENER,
action='post'),
constants.RULE_API_WRITE,
"Create a Listener",
[{'method': 'POST', 'path': '/v2.0/lbaas/listeners'}]
),
policy.DocumentedRuleDefault(
'{rbac_obj}{action}'.format(rbac_obj=constants.RBAC_LISTENER,
action='get_one'),
constants.RULE_API_READ,
"Show Listener details",
[{'method': 'GET',
'path': '/v2.0/lbaas/listeners/{listener_id}'}]
),
policy.DocumentedRuleDefault(
'{rbac_obj}{action}'.format(rbac_obj=constants.RBAC_LISTENER,
action='put'),
constants.RULE_API_WRITE,
"Update a Listener",
[{'method': 'PUT',
'path': '/v2.0/lbaas/listeners/{listener_id}'}]
),
policy.DocumentedRuleDefault(
'{rbac_obj}{action}'.format(rbac_obj=constants.RBAC_LISTENER,
action='delete'),
constants.RULE_API_WRITE,
"Remove a Listener",
[{'method': 'DELETE',
'path': '/v2.0/lbaas/listeners/{listener_id}'}]
),
policy.DocumentedRuleDefault(
'{rbac_obj}{action}'.format(rbac_obj=constants.RBAC_LISTENER,
action='get_stats'),
constants.RULE_API_READ,
"Show Listener statistics",
[{'method': 'GET',
'path': '/v2.0/lbaas/listeners/{listener_id}/stats'}]
),
]
def list_rules():
return rules

83
octavia/policies/loadbalancer.py Normal file
View File

@ -0,0 +1,83 @@
# Copyright 2017 Rackspace, US Inc.
# 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.
from octavia.common import constants
from oslo_policy import policy
rules = [
policy.DocumentedRuleDefault(
'{rbac_obj}{action}'.format(rbac_obj=constants.RBAC_LOADBALANCER,
action='get_all'),
constants.RULE_API_READ,
"List Load Balancers",
[{'method': 'GET', 'path': '/v2.0/lbaas/loadbalancers'}]
),
policy.DocumentedRuleDefault(
'{rbac_obj}{action}'.format(rbac_obj=constants.RBAC_LOADBALANCER,
action='get_all-global'),
constants.RULE_API_READ_GLOBAL,
"List Load Balancers including resources owned by others",
[{'method': 'GET', 'path': '/v2.0/lbaas/loadbalancers'}]
),
policy.DocumentedRuleDefault(
'{rbac_obj}{action}'.format(rbac_obj=constants.RBAC_LOADBALANCER,
action='post'),
constants.RULE_API_WRITE,
"Create a Load Balancer",
[{'method': 'POST', 'path': '/v2.0/lbaas/loadbalancers'}]
),
policy.DocumentedRuleDefault(
'{rbac_obj}{action}'.format(rbac_obj=constants.RBAC_LOADBALANCER,
action='get_one'),
constants.RULE_API_READ,
"Show Load Balancer details",
[{'method': 'GET',
'path': '/v2.0/lbaas/loadbalancers/{loadbalancer_id}'}]
),
policy.DocumentedRuleDefault(
'{rbac_obj}{action}'.format(rbac_obj=constants.RBAC_LOADBALANCER,
action='put'),
constants.RULE_API_WRITE,
"Update a Load Balancer",
[{'method': 'PUT',
'path': '/v2.0/lbaas/loadbalancers/{loadbalancer_id}'}]
),
policy.DocumentedRuleDefault(
'{rbac_obj}{action}'.format(rbac_obj=constants.RBAC_LOADBALANCER,
action='delete'),
constants.RULE_API_WRITE,
"Remove a Load Balancer",
[{'method': 'DELETE',
'path': '/v2.0/lbaas/loadbalancers/{loadbalancer_id}'}]
),
policy.DocumentedRuleDefault(
'{rbac_obj}{action}'.format(rbac_obj=constants.RBAC_LOADBALANCER,
action='get_stats'),
constants.RULE_API_READ,
"Show Load Balancer statistics",
[{'method': 'GET',
'path': '/v2.0/lbaas/loadbalancers/{loadbalancer_id}/stats'}]
),
policy.DocumentedRuleDefault(
'{rbac_obj}{action}'.format(rbac_obj=constants.RBAC_LOADBALANCER,
action='get_status'),
constants.RULE_API_READ,
"Show Load Balancer status",
[{'method': 'GET',
'path': '/v2.0/lbaas/loadbalancers/{loadbalancer_id}/status'}]
),
]
def list_rules():
return rules

20
octavia/tests/functional/api/v2/base.py
View File

@ -63,6 +63,9 @@ class BaseAPITest(base_db_test.OctaviaDBTestBase):
QUOTA_PATH = QUOTAS_PATH + '/{project_id}'
QUOTA_DEFAULT_PATH = QUOTAS_PATH + '/{project_id}/default'
NOT_AUTHORIZED_BODY = {'debuginfo': None, 'faultcode': 'Client',
'faultstring': 'Not authorized.'}
def setUp(self):
super(BaseAPITest, self).setUp()
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
@ -353,6 +356,7 @@ class BaseAPITest(base_db_test.OctaviaDBTestBase):
explicit_status = True if status is not None else False
if not explicit_status:
status = constants.ACTIVE
if status == constants.DELETED:
op_status = constants.OFFLINE
elif status == constants.ACTIVE:
@ -371,16 +375,6 @@ class BaseAPITest(base_db_test.OctaviaDBTestBase):
provisioning_status=provisioning_status,
operating_status=operating_status)
def assert_final_lb_statuses(self, lb_id, delete=False):
expected_prov_status = constants.ACTIVE
expected_op_status = constants.ONLINE
if delete:
expected_prov_status = constants.DELETED
expected_op_status = constants.OFFLINE
self.set_lb_status(lb_id, status=expected_prov_status)
self.assert_correct_lb_status(expected_prov_status, expected_op_status,
lb_id)
def assert_final_listener_statuses(self, lb_id, listener_id, delete=False):
expected_prov_status = constants.ACTIVE
expected_op_status = constants.ONLINE
@ -392,8 +386,8 @@ class BaseAPITest(base_db_test.OctaviaDBTestBase):
expected_op_status,
listener_id)
def assert_correct_lb_status(self, provisioning_status, operating_status,
lb_id):
def assert_correct_lb_status(self, lb_id,
operating_status, provisioning_status):
api_lb = self.get(
self.LB_PATH.format(lb_id=lb_id)).json.get('loadbalancer')
self.assertEqual(provisioning_status,
@ -473,7 +467,7 @@ class BaseAPITest(base_db_test.OctaviaDBTestBase):
l7rule_op_status=constants.ONLINE,
hm_op_status=constants.ONLINE):
if lb_id:
self.assert_correct_lb_status(lb_prov_status, lb_op_status, lb_id)
self.assert_correct_lb_status(lb_id, lb_op_status, lb_prov_status)
if listener_id:
self.assert_correct_listener_status(
listener_prov_status, listener_op_status, listener_id)

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

@ -15,6 +15,8 @@
import mock
from oslo_config import cfg
from oslo_config import fixture as oslo_fixture
from oslo_utils import uuidutils
from octavia.common import constants
@ -33,6 +35,7 @@ class TestListener(base.BaseAPITest):
super(TestListener, self).setUp()
self.lb = self.create_load_balancer(uuidutils.generate_uuid())
self.lb_id = self.lb.get('loadbalancer').get('id')
self.project_id = self.lb.get('loadbalancer').get('project_id')
self.set_lb_status(self.lb_id)
self.listener_path = self.LISTENERS_PATH + '/{listener_id}'
self.pool = self.create_pool(
@ -87,8 +90,24 @@ class TestListener(base.BaseAPITest):
self.conf.config(auth_strategy=constants.KEYSTONE)
with mock.patch.object(octavia.common.context.Context, 'project_id',
listener3['project_id']):
listeners = self.get(
self.LISTENERS_PATH).json.get(self.root_tag_list)
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': self.project_id}
with mock.patch(
"oslo_context.context.RequestContext.to_policy_values",
return_value=override_credentials):
listeners = self.get(
self.LISTENERS_PATH).json.get(self.root_tag_list)
self.conf.config(auth_strategy=auth_strategy)
self.assertEqual(1, len(listeners))
@ -97,6 +116,80 @@ class TestListener(base.BaseAPITest):
self.assertIn((listener3.get('id'), listener3.get('protocol_port')),
listener_id_ports)
def test_get_all_non_admin_global_observer(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).get(self.root_tag)
self.set_lb_status(lb1_id)
listener2 = self.create_listener(
constants.PROTOCOL_HTTP, 81, lb1_id).get(self.root_tag)
self.set_lb_status(lb1_id)
listener3 = self.create_listener(
constants.PROTOCOL_HTTP, 82, lb1_id).get(self.root_tag)
self.set_lb_status(lb1_id)
auth_strategy = self.conf.conf.get('auth_strategy')
self.conf.config(auth_strategy=constants.KEYSTONE)
with mock.patch.object(octavia.common.context.Context, 'project_id',
self.project_id):
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_global_observer'],
'user_id': None,
'is_admin': False,
'service_user_domain_id': None,
'project_domain_id': None,
'service_roles': [],
'project_id': self.project_id}
with mock.patch(
"oslo_context.context.RequestContext.to_policy_values",
return_value=override_credentials):
listeners = self.get(self.LISTENERS_PATH)
listeners = listeners.json.get(self.root_tag_list)
self.conf.config(auth_strategy=auth_strategy)
self.assertEqual(3, len(listeners))
listener_id_ports = [(l.get('id'), l.get('protocol_port'))
for l in listeners]
self.assertIn((listener1.get('id'), listener1.get('protocol_port')),
listener_id_ports)
self.assertIn((listener2.get('id'), listener2.get('protocol_port')),
listener_id_ports)
self.assertIn((listener3.get('id'), listener3.get('protocol_port')),
listener_id_ports)
def test_get_all_not_authorized(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.get('auth_strategy')
self.conf.config(auth_strategy=constants.KEYSTONE)
with mock.patch.object(octavia.common.context.Context, 'project_id',
uuidutils.generate_uuid()):
listeners = self.get(self.LISTENERS_PATH, status=401).json
self.conf.config(auth_strategy=auth_strategy)
self.assertEqual(self.NOT_AUTHORIZED_BODY, listeners)
def test_get_all_by_project_id(self):
project1_id = uuidutils.generate_uuid()
project2_id = uuidutils.generate_uuid()
@ -218,6 +311,52 @@ class TestListener(base.BaseAPITest):
api_listener = response.json.get(self.root_tag)
self.assertEqual(listener, api_listener)
def test_get_authorized(self):
listener = self.create_listener(
constants.PROTOCOL_HTTP, 80, self.lb_id).get(self.root_tag)
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.get('auth_strategy')
self.conf.config(auth_strategy=constants.TESTING)
with mock.patch.object(octavia.common.context.Context, 'project_id',
self.project_id):
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': self.project_id}
with mock.patch(
"oslo_context.context.RequestContext.to_policy_values",
return_value=override_credentials):
response = self.get(self.listener_path.format(
listener_id=listener['id']))
api_listener = response.json.get(self.root_tag)
self.assertEqual(listener, api_listener)
self.conf.config(auth_strategy=auth_strategy)
def test_get_not_authorized(self):
listener = self.create_listener(
constants.PROTOCOL_HTTP, 80, self.lb_id).get(self.root_tag)
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.get('auth_strategy')
self.conf.config(auth_strategy=constants.TESTING)
with mock.patch.object(octavia.common.context.Context, 'project_id',
uuidutils.generate_uuid()):
response = self.get(self.listener_path.format(
listener_id=listener['id']), status=401)
self.conf.config(auth_strategy=auth_strategy)
self.assertEqual(self.NOT_AUTHORIZED_BODY, response.json)
def test_get_hides_deleted(self):
api_listener = self.create_listener(
constants.PROTOCOL_HTTP, 80, self.lb_id).get(self.root_tag)
@ -268,9 +407,8 @@ class TestListener(base.BaseAPITest):
self.assertIsNotNone(listener_api.pop('created_at'))
self.assertIsNone(listener_api.pop('updated_at'))
self.assertNotEqual(lb_listener, listener_api)
self.assert_correct_lb_status(constants.PENDING_UPDATE,
constants.ONLINE, self.lb_id)
self.assert_final_lb_statuses(self.lb_id)
self.assert_correct_lb_status(self.lb_id, constants.ONLINE,
constants.PENDING_UPDATE)
self.assert_final_listener_statuses(self.lb_id, listener_api.get('id'))
def test_create_duplicate_fails(self):
@ -353,9 +491,8 @@ class TestListener(base.BaseAPITest):
self.assertIsNotNone(listener_api.pop('created_at'))
self.assertIsNone(listener_api.pop('updated_at'))
self.assertNotEqual(lb_listener, listener_api)
self.assert_correct_lb_status(constants.PENDING_UPDATE,
constants.ONLINE, self.lb_id)
self.assert_final_lb_statuses(self.lb_id)
self.assert_correct_lb_status(self.lb_id, constants.ONLINE,
constants.PENDING_UPDATE)
self.assert_final_listener_statuses(self.lb_id, listener_api['id'])
def test_create_over_quota(self):
@ -378,6 +515,94 @@ class TestListener(base.BaseAPITest):
listener_prov_status=constants.ERROR,
listener_op_status=constants.OFFLINE)
def test_create_authorized(self, **optionals):
sni1 = uuidutils.generate_uuid()
sni2 = uuidutils.generate_uuid()
lb_listener = {'name': 'listener1', 'default_pool_id': None,
'description': 'desc1',
'admin_state_up': False,
'protocol': constants.PROTOCOL_HTTP,
'protocol_port': 80, 'connection_limit': 10,
'default_tls_container_ref': uuidutils.generate_uuid(),
'sni_container_refs': [sni1, sni2],
'insert_headers': {},
'project_id': self.project_id,
'loadbalancer_id': self.lb_id}
lb_listener.update(optionals)
body = self._build_body(lb_listener)
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.get('auth_strategy')
self.conf.config(auth_strategy=constants.TESTING)
with mock.patch.object(octavia.common.context.Context, 'project_id',
self.project_id):
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': self.project_id}
with mock.patch(
"oslo_context.context.RequestContext.to_policy_values",
return_value=override_credentials):
response = self.post(self.LISTENERS_PATH, body)
self.conf.config(auth_strategy=auth_strategy)
listener_api = response.json['listener']
extra_expects = {'provisioning_status': constants.PENDING_CREATE,
'operating_status': constants.OFFLINE}
lb_listener.update(extra_expects)
self.assertTrue(uuidutils.is_uuid_like(listener_api.get('id')))
for key, value in optionals.items():
self.assertEqual(value, lb_listener.get(key))
lb_listener['id'] = listener_api.get('id')
lb_listener.pop('sni_container_refs')
sni_ex = [sni1, sni2]
sni_resp = listener_api.pop('sni_container_refs')
self.assertEqual(2, len(sni_resp))
for sni in sni_resp:
self.assertIn(sni, sni_ex)
self.assertIsNotNone(listener_api.pop('created_at'))
self.assertIsNone(listener_api.pop('updated_at'))
self.assertNotEqual(lb_listener, listener_api)
self.assert_correct_lb_status(self.lb_id, constants.ONLINE,
constants.PENDING_UPDATE)
self.assert_final_listener_statuses(self.lb_id, listener_api.get('id'))
def test_create_not_authorized(self, **optionals):
sni1 = uuidutils.generate_uuid()
sni2 = uuidutils.generate_uuid()
lb_listener = {'name': 'listener1', 'default_pool_id': None,
'description': 'desc1',
'admin_state_up': False,
'protocol': constants.PROTOCOL_HTTP,
'protocol_port': 80, 'connection_limit': 10,
'default_tls_container_ref': uuidutils.generate_uuid(),
'sni_container_refs': [sni1, sni2],
'insert_headers': {},
'project_id': self.project_id,
'loadbalancer_id': self.lb_id}
lb_listener.update(optionals)
body = self._build_body(lb_listener)
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.get('auth_strategy')
self.conf.config(auth_strategy=constants.TESTING)
with mock.patch.object(octavia.common.context.Context, 'project_id',
uuidutils.generate_uuid()):
response = self.post(self.LISTENERS_PATH, body, status=401)
self.conf.config(auth_strategy=auth_strategy)
self.assertEqual(self.NOT_AUTHORIZED_BODY, response.json)
def test_update_with_bad_handler(self):
api_listener = self.create_listener(
constants.PROTOCOL_HTTP, 80,
@ -437,8 +662,8 @@ class TestListener(base.BaseAPITest):
self.assertEqual(listener['created_at'], api_listener['created_at'])
self.assertNotEqual(listener['updated_at'], api_listener['updated_at'])
self.assertNotEqual(listener, api_listener)
self.assert_correct_lb_status(constants.PENDING_UPDATE,
constants.ONLINE, self.lb_id)
self.assert_correct_lb_status(self.lb_id, constants.ONLINE,
constants.PENDING_UPDATE)
self.assert_final_listener_statuses(self.lb_id,
api_listener['id'])
@ -460,11 +685,90 @@ class TestListener(base.BaseAPITest):
listener_path = self.LISTENER_PATH.format(
listener_id=listener['listener']['id'])
self.put(listener_path, body, status=404)
self.assert_correct_lb_status(constants.ACTIVE, constants.ONLINE,
self.lb_id)
self.assert_correct_lb_status(self.lb_id, constants.ONLINE,
constants.ACTIVE)
self.assert_final_listener_statuses(self.lb_id,
listener['listener']['id'])
def test_update_authorized(self):
tls_uuid = uuidutils.generate_uuid()
listener = self.create_listener(
constants.PROTOCOL_TCP, 80, self.lb_id,
name='listener1', description='desc1',
admin_state_up=False, connection_limit=10,
default_tls_container_ref=tls_uuid,
default_pool_id=None).get(self.root_tag)
self.set_lb_status(self.lb_id)
new_listener = {'name': 'listener2', 'admin_state_up': True,
'default_pool_id': self.pool_id}
body = self._build_body(new_listener)
listener_path = self.LISTENER_PATH.format(
listener_id=listener['id'])
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.get('auth_strategy')
self.conf.config(auth_strategy=constants.TESTING)
with mock.patch.object(octavia.common.context.Context, 'project_id',
self.project_id):
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': self.project_id}
with mock.patch(
"oslo_context.context.RequestContext.to_policy_values",
return_value=override_credentials):
api_listener = self.put(listener_path, body)
api_listener = api_listener.json.get(self.root_tag)
self.conf.config(auth_strategy=auth_strategy)
update_expect = {'name': 'listener2', 'admin_state_up': True,
'default_pool_id': self.pool_id,
'provisioning_status': constants.PENDING_UPDATE,
'operating_status': constants.ONLINE}
listener.update(update_expect)
self.assertEqual(listener['created_at'], api_listener['created_at'])
self.assertNotEqual(listener['updated_at'], api_listener['updated_at'])
self.assertNotEqual(listener, api_listener)
self.assert_correct_lb_status(self.lb_id, constants.ONLINE,
constants.PENDING_UPDATE)
self.assert_final_listener_statuses(self.lb_id,
api_listener['id'])
def test_update_not_authorized(self):
tls_uuid = uuidutils.generate_uuid()
listener = self.create_listener(
constants.PROTOCOL_TCP, 80, self.lb_id,
name='listener1', description='desc1',
admin_state_up=False, connection_limit=10,
default_tls_container_ref=tls_uuid,
default_pool_id=None).get(self.root_tag)
self.set_lb_status(self.lb_id)
new_listener = {'name': 'listener2', 'admin_state_up': True,
'default_pool_id': self.pool_id}
body = self._build_body(new_listener)
listener_path = self.LISTENER_PATH.format(
listener_id=listener['id'])
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.get('auth_strategy')
self.conf.config(auth_strategy=constants.TESTING)
with mock.patch.object(octavia.common.context.Context, 'project_id',
uuidutils.generate_uuid()):
api_listener = self.put(listener_path, body, status=401)
self.conf.config(auth_strategy=auth_strategy)
self.assertEqual(self.NOT_AUTHORIZED_BODY, api_listener.json)
self.assert_correct_lb_status(self.lb_id, constants.ONLINE,
constants.ACTIVE)
def test_create_listeners_same_port(self):
listener1 = self.create_listener(constants.PROTOCOL_TCP, 80,
self.lb_id)
@ -495,12 +799,77 @@ class TestListener(base.BaseAPITest):
self.assertIsNone(listener['listener'].pop('updated_at'))
self.assertIsNotNone(api_listener.pop('updated_at'))
self.assertNotEqual(listener, api_listener)
self.assert_correct_lb_status(constants.PENDING_UPDATE,
constants.ONLINE, self.lb_id)
self.assert_final_lb_statuses(self.lb_id)
self.assert_correct_lb_status(self.lb_id, constants.ONLINE,
constants.PENDING_UPDATE)
self.assert_final_listener_statuses(self.lb_id, api_listener['id'],
delete=True)
def test_delete_authorized(self):
listener = self.create_listener(constants.PROTOCOL_HTTP, 80,
self.lb_id)
self.set_lb_status(self.lb_id)
listener_path = self.LISTENER_PATH.format(
listener_id=listener['listener']['id'])
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.get('auth_strategy')
self.conf.config(auth_strategy=constants.TESTING)
with mock.patch.object(octavia.common.context.Context, 'project_id',
self.project_id):
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': self.project_id}
with mock.patch(
"oslo_context.context.RequestContext.to_policy_values",
return_value=override_credentials):
self.delete(listener_path)
self.conf.config(auth_strategy=auth_strategy)
response = self.get(listener_path)
api_listener = response.json['listener']
expected = {'name': None, 'default_pool_id': None,
'description': None, 'admin_state_up': True,
'operating_status': constants.ONLINE,
'provisioning_status': constants.PENDING_DELETE,
'connection_limit': None}
listener['listener'].update(expected)
self.assertIsNone(listener['listener'].pop('updated_at'))
self.assertIsNotNone(api_listener.pop('updated_at'))
self.assertNotEqual(listener, api_listener)
self.assert_correct_lb_status(self.lb_id, constants.ONLINE,
constants.PENDING_UPDATE)
self.assert_final_listener_statuses(self.lb_id, api_listener['id'],
delete=True)
def test_delete_not_authorized(self):
listener = self.create_listener(constants.PROTOCOL_HTTP, 80,
self.lb_id)
self.set_lb_status(self.lb_id)
listener_path = self.LISTENER_PATH.format(
listener_id=listener['listener']['id'])
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.get('auth_strategy')
self.conf.config(auth_strategy=constants.TESTING)
with mock.patch.object(octavia.common.context.Context, 'project_id',
uuidutils.generate_uuid()):
self.delete(listener_path, status=401)
self.conf.config(auth_strategy=auth_strategy)
self.assert_correct_lb_status(self.lb_id, constants.ONLINE,
constants.ACTIVE)
def test_delete_bad_listener_id(self):
listener_path = self.LISTENER_PATH.format(listener_id='SEAN-CONNERY')
self.delete(listener_path, status=404)

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

@ -15,6 +15,8 @@
import copy
import mock
from oslo_config import cfg
from oslo_config import fixture as oslo_fixture
from oslo_utils import uuidutils
from octavia.common import constants
@ -51,7 +53,6 @@ class TestLoadBalancer(base.BaseAPITest):
self.assertIsNone(resp.get('updated_at'))
for key, value in optionals.items():
self.assertEqual(value, req.get(key))
self.assert_final_lb_statuses(resp.get('id'))
def test_empty_list(self):
response = self.get(self.LBS_PATH)
@ -324,6 +325,77 @@ class TestLoadBalancer(base.BaseAPITest):
api_lb = self.test_create(project_id=project_id)
self.assertEqual(project_id, api_lb.get('project_id'))
def test_create_no_project_id(self, **optionals):
lb_json = {'name': 'test1',
'vip_subnet_id': uuidutils.generate_uuid()
}
lb_json.update(optionals)
body = self._build_body(lb_json)
self.post(self.LBS_PATH, body, status=400)
def test_create_context_project_id(self, **optionals):
lb_json = {'name': 'test1',
'vip_subnet_id': uuidutils.generate_uuid()
}
lb_json.update(optionals)
body = self._build_body(lb_json)
with mock.patch.object(octavia.common.context.Context, 'project_id',
self.project_id):
response = self.post(self.LBS_PATH, body)
api_lb = response.json.get(self.root_tag)
self._assert_request_matches_response(lb_json, api_lb)
def test_create_authorized(self, **optionals):
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.get('auth_strategy')
self.conf.config(auth_strategy=constants.TESTING)
project_id = uuidutils.generate_uuid()
lb_json = {'name': 'test1',
'vip_subnet_id': uuidutils.generate_uuid(),
'project_id': project_id
}
lb_json.update(optionals)
body = self._build_body(lb_json)
with mock.patch.object(octavia.common.context.Context, 'project_id',
project_id):
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': project_id}
with mock.patch(
"oslo_context.context.RequestContext.to_policy_values",
return_value=override_credentials):
response = self.post(self.LBS_PATH, body)
api_lb = response.json.get(self.root_tag)
self.conf.config(auth_strategy=auth_strategy)
self._assert_request_matches_response(lb_json, api_lb)
def test_create_not_authorized(self, **optionals):
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.get('auth_strategy')
self.conf.config(auth_strategy=constants.TESTING)
lb_json = {'name': 'test1',
'vip_subnet_id': uuidutils.generate_uuid(),
'project_id': uuidutils.generate_uuid()
}
lb_json.update(optionals)
body = self._build_body(lb_json)
with mock.patch.object(octavia.common.context.Context, 'project_id',
uuidutils.generate_uuid()):
response = self.post(self.LBS_PATH, body, status=401)
api_lb = response.json
self.conf.config(auth_strategy=auth_strategy)
self.assertEqual(self.NOT_AUTHORIZED_BODY, api_lb)
def test_get_all_admin(self):
project_id = uuidutils.generate_uuid()
lb1 = self.create_load_balancer(uuidutils.generate_uuid(),
@ -357,7 +429,23 @@ class TestLoadBalancer(base.BaseAPITest):
self.conf.config(auth_strategy=constants.KEYSTONE)
with mock.patch.object(octavia.common.context.Context, 'project_id',
self.project_id):
response = self.get(self.LBS_PATH)
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': self.project_id}
with mock.patch(
"oslo_context.context.RequestContext.to_policy_values",
return_value=override_credentials):
response = self.get(self.LBS_PATH)
self.conf.config(auth_strategy=auth_strategy)
lbs = response.json.get(self.root_tag_list)
@ -365,6 +453,67 @@ 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_non_admin_global_observer(self):
project_id = uuidutils.generate_uuid()
lb1 = self.create_load_balancer(uuidutils.generate_uuid(),
name='lb1', project_id=project_id)
lb2 = 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)
lb1 = lb1.get(self.root_tag)
lb2 = lb2.get(self.root_tag)
lb3 = lb3.get(self.root_tag)
auth_strategy = self.conf.conf.get('auth_strategy')
self.conf.config(auth_strategy=constants.KEYSTONE)
with mock.patch.object(octavia.common.context.Context, 'project_id',
self.project_id):
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_global_observer'],
'user_id': None,
'is_admin': False,
'service_user_domain_id': None,
'project_domain_id': None,
'service_roles': [],
'project_id': self.project_id}
with mock.patch(
"oslo_context.context.RequestContext.to_policy_values",
return_value=override_credentials):
response = self.get(self.LBS_PATH)
self.conf.config(auth_strategy=auth_strategy)
lbs = response.json.get(self.root_tag_list)
self.assertEqual(3, len(lbs))
lb_id_names = [(lb.get('id'), lb.get('name')) for lb in lbs]
self.assertIn((lb1.get('id'), lb1.get('name')), lb_id_names)
self.assertIn((lb2.get('id'), lb2.get('name')), lb_id_names)
self.assertIn((lb3.get('id'), lb3.get('name')), lb_id_names)
def test_get_all_not_authorized(self):
project_id = uuidutils.generate_uuid()
self.create_load_balancer(uuidutils.generate_uuid(),
name='lb1', project_id=self.project_id)
self.create_load_balancer(uuidutils.generate_uuid(),
name='lb2', project_id=project_id)
self.create_load_balancer(uuidutils.generate_uuid(),
name='lb3', project_id=project_id)
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.get('auth_strategy')
self.conf.config(auth_strategy=constants.TESTING)
LB_PROJECT_PATH = '{}?project_id={}'.format(self.LBS_PATH, project_id)
with mock.patch.object(octavia.common.context.Context, 'project_id',
self.project_id):
response = self.get(LB_PROJECT_PATH, status=401)
api_lb = response.json
self.conf.config(auth_strategy=auth_strategy)
self.assertEqual(self.NOT_AUTHORIZED_BODY, api_lb)
def test_get_all_by_project_id(self):
project1_id = uuidutils.generate_uuid()
project2_id = uuidutils.generate_uuid()
@ -512,6 +661,98 @@ class TestLoadBalancer(base.BaseAPITest):
path = self.LB_PATH.format(lb_id='SEAN-CONNERY')
self.get(path, status=404)
def test_get_authorized(self):
project_id = uuidutils.generate_uuid()
subnet = network_models.Subnet(id=uuidutils.generate_uuid())
network = network_models.Network(id=uuidutils.generate_uuid(),
subnets=[subnet])
port = network_models.Port(id=uuidutils.generate_uuid(),
network_id=network.id)
with mock.patch(
"octavia.network.drivers.noop_driver.driver.NoopManager"
".get_network") as mock_get_network, mock.patch(
"octavia.network.drivers.noop_driver.driver.NoopManager"
".get_port") as mock_get_port:
mock_get_network.return_value = network
mock_get_port.return_value = port
lb = self.create_load_balancer(subnet.id,
vip_address='10.0.0.1',
vip_network_id=network.id,
vip_port_id=port.id,
name='lb1',
project_id=project_id,
description='desc1',
admin_state_up=False)
lb_dict = lb.get(self.root_tag)
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.get('auth_strategy')
self.conf.config(auth_strategy=constants.TESTING)
with mock.patch.object(octavia.common.context.Context, 'project_id',
project_id):
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': project_id}
with mock.patch(
"oslo_context.context.RequestContext.to_policy_values",
return_value=override_credentials):
response = self.get(self.LB_PATH.format(
lb_id=lb_dict.get('id'))).json.get(self.root_tag)
self.assertEqual('lb1', response.get('name'))
self.assertEqual(project_id, response.get('project_id'))
self.assertEqual('desc1', response.get('description'))
self.assertFalse(response.get('admin_state_up'))
self.assertEqual('10.0.0.1', response.get('vip_address'))
self.assertEqual(subnet.id, response.get('vip_subnet_id'))
self.assertEqual(network.id, response.get('vip_network_id'))
self.assertEqual(port.id, response.get('vip_port_id'))
self.conf.config(auth_strategy=auth_strategy)
def test_get_not_authorized(self):
project_id = uuidutils.generate_uuid()
subnet = network_models.Subnet(id=uuidutils.generate_uuid())
network = network_models.Network(id=uuidutils.generate_uuid(),
subnets=[subnet])
port = network_models.Port(id=uuidutils.generate_uuid(),
network_id=network.id)
with mock.patch(
"octavia.network.drivers.noop_driver.driver.NoopManager"
".get_network") as mock_get_network, mock.patch(
"octavia.network.drivers.noop_driver.driver.NoopManager"
".get_port") as mock_get_port:
mock_get_network.return_value = network
mock_get_port.return_value = port
lb = self.create_load_balancer(subnet.id,
vip_address='10.0.0.1',
vip_network_id=network.id,
vip_port_id=port.id,
name='lb1',
project_id=project_id,
description='desc1',
admin_state_up=False)
lb_dict = lb.get(self.root_tag)
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.get('auth_strategy')
self.conf.config(auth_strategy=constants.TESTING)
with mock.patch.object(octavia.common.context.Context, 'project_id',
uuidutils.generate_uuid()):
response = self.get(self.LB_PATH.format(lb_id=lb_dict.get('id')),
status=401)
api_lb = response.json
self.conf.config(auth_strategy=auth_strategy)
self.assertEqual(self.NOT_AUTHORIZED_BODY, api_lb)
def test_create_over_quota(self):
self.start_quota_mock(data_models.LoadBalancer)
lb_json = {'name': 'test1',
@ -538,11 +779,10 @@ class TestLoadBalancer(base.BaseAPITest):
self.assertEqual(project_id, api_lb.get('project_id'))
self.assertEqual('desc1', api_lb.get('description'))
self.assertFalse(api_lb.get('admin_state_up'))
self.assertEqual(lb.get('operational_status'),
api_lb.get('operational_status'))
self.assertIsNotNone(api_lb.get('created_at'))
self.assertIsNotNone(api_lb.get('updated_at'))
self.assert_final_lb_statuses(api_lb.get('id'))
self.assert_correct_lb_status(api_lb.get('id'), constants.ONLINE,
constants.PENDING_UPDATE)
def test_update_with_vip(self):
project_id = uuidutils.generate_uuid()
@ -573,6 +813,76 @@ class TestLoadBalancer(base.BaseAPITest):
self.put(self.LB_PATH.format(lb_id=lb_dict.get('id')),
lb_json, status=409)
def test_update_authorized(self):
project_id = uuidutils.generate_uuid()
lb = self.create_load_balancer(uuidutils.generate_uuid(),
name='lb1',
project_id=project_id,
description='desc1',
admin_state_up=False)
lb_dict = lb.get(self.root_tag)
lb_json = self._build_body({'name': 'lb2'})
lb = self.set_lb_status(lb_dict.get('id'))
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.get('auth_strategy')
self.conf.config(auth_strategy=constants.TESTING)
with mock.patch.object(octavia.common.context.Context, 'project_id',
project_id):
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': project_id}
with mock.patch(
"oslo_context.context.RequestContext.to_policy_values",
return_value=override_credentials):
response = self.put(
self.LB_PATH.format(lb_id=lb_dict.get('id')), lb_json)
api_lb = response.json.get(self.root_tag)
self.conf.config(auth_strategy=auth_strategy)
self.assertIsNotNone(api_lb.get('vip_subnet_id'))
self.assertEqual('lb1', api_lb.get('name'))
self.assertEqual(project_id, api_lb.get('project_id'))
self.assertEqual('desc1', api_lb.get('description'))
self.assertFalse(api_lb.get('admin_state_up'))
self.assertIsNotNone(api_lb.get('created_at'))
self.assertIsNotNone(api_lb.get('updated_at'))
self.assert_correct_lb_status(api_lb.get('id'), constants.ONLINE,
constants.PENDING_UPDATE)
def test_update_not_authorized(self):
project_id = uuidutils.generate_uuid()
lb = self.create_load_balancer(uuidutils.generate_uuid(),
name='lb1',
project_id=project_id,
description='desc1',
admin_state_up=False)
lb_dict = lb.get(self.root_tag)
lb_json = self._build_body({'name': 'lb2'})
lb = self.set_lb_status(lb_dict.get('id'))
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.get('auth_strategy')
self.conf.config(auth_strategy=constants.TESTING)
with mock.patch.object(octavia.common.context.Context, 'project_id',
uuidutils.generate_uuid()):
response = self.put(self.LB_PATH.format(lb_id=lb_dict.get('id')),
lb_json, status=401)
api_lb = response.json
self.conf.config(auth_strategy=auth_strategy)
self.assertEqual(self.NOT_AUTHORIZED_BODY, api_lb)
self.assert_correct_lb_status(lb_dict.get('id'), constants.ONLINE,
constants.ACTIVE)
def test_delete_pending_create(self):
project_id = uuidutils.generate_uuid()
lb = self.create_load_balancer(uuidutils.generate_uuid(),
@ -650,8 +960,7 @@ class TestLoadBalancer(base.BaseAPITest):
def test_delete(self):
project_id = uuidutils.generate_uuid()
lb = self.create_load_balancer(uuidutils.generate_uuid(),
name='lb1',
project_id=project_id,
name='lb1', project_id=project_id,
description='desc1',
admin_state_up=False)
lb_dict = lb.get(self.root_tag)
@ -663,9 +972,78 @@ class TestLoadBalancer(base.BaseAPITest):
self.assertEqual('desc1', api_lb.get('description'))
self.assertEqual(project_id, api_lb.get('project_id'))
self.assertFalse(api_lb.get('admin_state_up'))
self.assert_correct_lb_status(api_lb.get('id'), constants.ONLINE,
constants.PENDING_DELETE)
def test_delete_authorized(self):
project_id = uuidutils.generate_uuid()
lb = self.create_load_balancer(uuidutils.generate_uuid(),
name='lb1',
project_id=project_id,
description='desc1',
admin_state_up=False)
lb_dict = lb.get(self.root_tag)
lb = self.set_lb_status(lb_dict.get('id'))
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.get('auth_strategy')
self.conf.config(auth_strategy=constants.TESTING)
with mock.patch.object(octavia.common.context.Context, 'project_id',
project_id):
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': project_id}
with mock.patch(
"oslo_context.context.RequestContext.to_policy_values",
return_value=override_credentials):
self.delete(self.LB_PATH.format(lb_id=lb_dict.get('id')))
self.conf.config(auth_strategy=auth_strategy)
response = self.get(self.LB_PATH.format(lb_id=lb_dict.get('id')))
api_lb = response.json.get(self.root_tag)
self.assertEqual('lb1', api_lb.get('name'))
self.assertEqual('desc1', api_lb.get('description'))
self.assertEqual(project_id, api_lb.get('project_id'))
self.assertFalse(api_lb.get('admin_state_up'))
self.assertEqual(lb.get('operational_status'),
api_lb.get('operational_status'))
self.assert_final_lb_statuses(api_lb.get('id'), delete=True)
self.assert_correct_lb_status(api_lb.get('id'), constants.ONLINE,
constants.PENDING_DELETE)
def test_delete_not_authorized(self):
project_id = uuidutils.generate_uuid()
lb = self.create_load_balancer(uuidutils.generate_uuid(),
name='lb1',
project_id=project_id,
description='desc1',
admin_state_up=False)
lb_dict = lb.get(self.root_tag)
lb = self.set_lb_status(lb_dict.get('id'))
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.get('auth_strategy')
self.conf.config(auth_strategy=constants.TESTING)
with mock.patch.object(octavia.common.context.Context, 'project_id',
uuidutils.generate_uuid()):
self.delete(self.LB_PATH.format(lb_id=lb_dict.get('id')),
status=401)
self.conf.config(auth_strategy=auth_strategy)
response = self.get(self.LB_PATH.format(lb_id=lb_dict.get('id')))
api_lb = response.json.get(self.root_tag)
self.assertEqual('lb1', api_lb.get('name'))
self.assertEqual('desc1', api_lb.get('description'))
self.assertEqual(project_id, api_lb.get('project_id'))
self.assertFalse(api_lb.get('admin_state_up'))
self.assert_correct_lb_status(api_lb.get('id'), constants.ONLINE,
constants.ACTIVE)
def test_delete_fails_with_pool(self):
project_id = uuidutils.generate_uuid()

23
octavia/tests/unit/common/test_policy.py
View File

@ -40,14 +40,15 @@ class PolicyFileTestCase(base.TestCase):
self.conf.load_raw_values(
group='oslo_policy', policy_file=tmp.name)
tmp.write('{"example:test": ""}')
tmp.flush()
self.context = context.Context('fake', 'fake')
rule = oslo_policy.RuleDefault('example:test', "")
self.context.policy.register_defaults([rule])
action = "example:test"
tmp.write('{"example:test": ""}')
tmp.flush()
self.context.policy.authorize(action, self.target)
tmp.seek(0)
@ -158,11 +159,21 @@ class PolicyTestCase(base.TestCase):
self.assertTrue(self.context.policy.check_is_admin())
def test_get_enforcer(self):
self.assertTrue(isinstance(policy.get_no_context_enforcer(),
oslo_policy.Enforcer))
class IsAdminCheckTestCase(base.TestCase):
def setUp(self):
super(IsAdminCheckTestCase, self).setUp()
self.conf = self.useFixture(oslo_fixture.Config())
# diltram: this one must be removed after fixing issue in oslo.config
# https://bugs.launchpad.net/oslo.config/+bug/1645868
self.conf.conf.__call__(args=[])
self.context = context.Context('fake', 'fake')
def test_init_true(self):
@ -200,6 +211,12 @@ class AdminRolePolicyTestCase(base.TestCase):
def setUp(self):
super(AdminRolePolicyTestCase, self).setUp()
self.conf = self.useFixture(oslo_fixture.Config())
# diltram: this one must be removed after fixing issue in oslo.config
# https://bugs.launchpad.net/oslo.config/+bug/1645868
self.conf.conf.__call__(args=[])
self.context = context.Context('fake', 'fake', roles=['member'])
self.actions = self.context.policy.get_rules().keys()
self.target = {}
@ -211,5 +228,5 @@ class AdminRolePolicyTestCase(base.TestCase):
"""
for action in self.actions:
self.assertRaises(
oslo_policy.PolicyNotAuthorized, self.context.policy.authorize,
exceptions.NotAuthorized, self.context.policy.authorize,
action, self.target)

4
setup.cfg
View File

@ -89,6 +89,10 @@ oslo.config.opts =
octavia = octavia.opts:list_opts
tempest.test_plugins =
octavia = octavia.tests.tempest.plugin:OctaviaTempestPlugin
oslo.policy.policies =
octavia = octavia.policies:list_rules
oslo.policy.enforcer =
octavia = octavia.common.policy:get_no_context_enforcer
[compile_catalog]
directory = octavia/locale

9
tox.ini
View File

@ -90,7 +90,16 @@ commands =
--namespace oslo.messaging \
--namespace keystonemiddleware.auth_token
[testenv:genpolicy]
whitelist_externals = mkdir
commands =
mkdir -p etc/octavia
oslopolicy-sample-generator \
--config-file etc/policy/octavia-policy-generator.conf \
--output-file etc/octavia/json.policy.sample
[testenv:specs]
whitelist_externals = rm
commands =
find . -type f -name "*.pyc" -delete
rm -f .testrepository/times.dbm