openstack
/
oslo.messaging
Code Issues Proposed changes

Merge "Allow dispatcher to restrict endpoint methods."

This commit is contained in:
Jenkins 2016-08-31 10:16:12 +00:00 committed by Gerrit Code Review
parent d8c6bf7b7c d3a8f280eb
commit 4723011f20
6 changed files with 227 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
doc/source/server.rst
View File

@ -8,6 +8,14 @@ Server
.. autofunction:: get_rpc_server
.. autoclass:: RPCAccessPolicyBase
.. autoclass:: LegacyRPCAccessPolicy
.. autoclass:: DefaultRPCAccessPolicy
.. autoclass:: ExplicitRPCAccessPolicy
.. autoclass:: RPCDispatcher
.. autoclass:: MessageHandlingServer
@ -15,6 +23,8 @@ Server
.. autofunction:: expected_exceptions
.. autofunction:: expose
.. autoexception:: ExpectedException
.. autofunction:: get_local_context

5
oslo_messaging/rpc/__init__.py
View File

@ -18,6 +18,10 @@ __all__ = [
'ExpectedException',
'NoSuchMethod',
'RPCClient',
'RPCAccessPolicyBase',
'LegacyRPCAccessPolicy',
'DefaultRPCAccessPolicy',
'ExplicitRPCAccessPolicy',
'RPCDispatcher',
'RPCDispatcherError',
'RPCVersionCapError',
@ -25,6 +29,7 @@ __all__ = [
'UnsupportedVersion',
'expected_exceptions',
'get_rpc_server',
'expose'
]
from .client import *

85
oslo_messaging/rpc/dispatcher.py
View File

@ -18,17 +18,24 @@
__all__ = [
'NoSuchMethod',
'RPCAccessPolicyBase',
'LegacyRPCAccessPolicy',
'DefaultRPCAccessPolicy',
'ExplicitRPCAccessPolicy',
'RPCDispatcher',
'RPCDispatcherError',
'UnsupportedVersion',
'ExpectedException',
]
from abc import ABCMeta
from abc import abstractmethod
import logging
import sys
import six
from debtcollector.updating import updated_kwarg_default_value
from oslo_messaging import _utils as utils
from oslo_messaging import dispatcher
from oslo_messaging import serializer as msg_serializer
@ -74,6 +81,52 @@ class UnsupportedVersion(RPCDispatcherError):
self.method = method
@six.add_metaclass(ABCMeta)
class RPCAccessPolicyBase(object):
"""Determines which endpoint methods may be invoked via RPC"""
@abstractmethod
def is_allowed(self, endpoint, method):
"""Applies an access policy to the rpc method
:param endpoint: the instance of a rpc endpoint
:param method: the method of the endpoint
:return: True if the method may be invoked via RPC, else False.
"""
class LegacyRPCAccessPolicy(RPCAccessPolicyBase):
"""The legacy access policy allows RPC access to all callable endpoint
methods including private methods (methods prefixed by '_')
"""
def is_allowed(self, endpoint, method):
return True
class DefaultRPCAccessPolicy(RPCAccessPolicyBase):
"""The default access policy prevents RPC calls to private methods
(methods prefixed by '_')
.. note::
LegacyRPCAdapterPolicy currently needs to be the default while we have
projects that rely on exposing private methods.
"""
def is_allowed(self, endpoint, method):
return not method.startswith('_')
class ExplicitRPCAccessPolicy(RPCAccessPolicyBase):
"""Policy which requires decorated endpoint methods to allow dispatch"""
def is_allowed(self, endpoint, method):
if hasattr(endpoint, method):
return hasattr(getattr(endpoint, method), 'exposed')
return False
class RPCDispatcher(dispatcher.DispatcherBase):
"""A message dispatcher which understands RPC messages.
@ -86,13 +139,24 @@ class RPCDispatcher(dispatcher.DispatcherBase):
in the message and matches those against a list of available endpoints.
Endpoints may have a target attribute describing the namespace and version
of the methods exposed by that object. All public methods on an endpoint
object are remotely invokable by clients.
of the methods exposed by that object.
The RPCDispatcher may have an access_policy attribute which determines
which of the endpoint methods are to be dispatched.
The default access_policy dispatches all public methods
on an endpoint object.
"""
def __init__(self, endpoints, serializer):
@updated_kwarg_default_value('access_policy', None, DefaultRPCAccessPolicy,
message='access_policy defaults to '
'LegacyRPCAccessPolicy which '
'exposes private methods. Explicitly '
'set access_policy to '
'DefaultRPCAccessPolicy or '
'ExplicitRPCAccessPolicy.',
version='?')
def __init__(self, endpoints, serializer, access_policy=None):
"""Construct a rpc server dispatcher.
:param endpoints: list of endpoint objects for dispatching to
@ -102,6 +166,16 @@ class RPCDispatcher(dispatcher.DispatcherBase):
self.endpoints = endpoints
self.serializer = serializer or msg_serializer.NoOpSerializer()
self._default_target = msg_target.Target()
if access_policy is not None:
if issubclass(access_policy, RPCAccessPolicyBase):
self.access_policy = access_policy()
else:
raise TypeError('access_policy must be a subclass of '
'RPCAccessPolicyBase')
else:
# TODO(pvinci): Change to DefaultRPCAccessPolicy when setting to
# DefaultRCPAccessPolicy no longer breaks in tempest tests.
self.access_policy = LegacyRPCAccessPolicy()
@staticmethod
def _is_namespace(target, namespace):
@ -147,7 +221,8 @@ class RPCDispatcher(dispatcher.DispatcherBase):
continue
if hasattr(endpoint, method):
return self._do_dispatch(endpoint, method, ctxt, args)
if self.access_policy.is_allowed(endpoint, method):
return self._do_dispatch(endpoint, method, ctxt, args)
found_compatible = True

31
oslo_messaging/rpc/server.py
View File

@ -100,6 +100,7 @@ to - primitive types.
__all__ = [
'get_rpc_server',
'expected_exceptions',
'expose'
]
import logging
@ -156,7 +157,7 @@ class RPCServer(msg_server.MessageHandlingServer):
def get_rpc_server(transport, target, endpoints,
executor='blocking', serializer=None):
executor='blocking', serializer=None, access_policy=None):
"""Construct an RPC server.
The executor parameter controls how incoming messages will be received and
@ -177,8 +178,12 @@ def get_rpc_server(transport, target, endpoints,
:type executor: str
:param serializer: an optional entity serializer
:type serializer: Serializer
:param access_policy: an optional access policy.
Defaults to LegacyRPCAccessPolicy
:type access_policy: RPCAccessPolicyBase
"""
dispatcher = rpc_dispatcher.RPCDispatcher(endpoints, serializer)
dispatcher = rpc_dispatcher.RPCDispatcher(endpoints, serializer,
access_policy)
return RPCServer(transport, target, dispatcher, executor)
@ -207,3 +212,25 @@ def expected_exceptions(*exceptions):
raise rpc_dispatcher.ExpectedException()
return inner
return outer
def expose(func):
"""Decorator for RPC endpoint methods that are exposed to the RPC client.
If the dispatcher's access_policy is set to ExplicitRPCAccessPolicy then
endpoint methods need to be explicitly exposed.::
# foo() cannot be invoked by an RPC client
def foo(self):
pass
# bar() can be invoked by an RPC client
@rpc.expose
def bar(self):
pass
"""
func.exposed = True
return func

96
oslo_messaging/tests/rpc/test_dispatcher.py
View File

@ -1,4 +1,3 @@
# Copyright 2013 Red Hat, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
@ -16,6 +15,7 @@
import testscenarios
import oslo_messaging
from oslo_messaging import rpc
from oslo_messaging import serializer as msg_serializer
from oslo_messaging.tests import utils as test_utils
from six.moves import mock
@ -24,92 +24,161 @@ load_tests = testscenarios.load_tests_apply_scenarios
class _FakeEndpoint(object):
def __init__(self, target=None):
self.target = target
def foo(self, ctxt, **kwargs):
pass
@rpc.expose
def bar(self, ctxt, **kwargs):
pass
def _foobar(self, ctxt, **kwargs):
pass
class TestDispatcher(test_utils.BaseTestCase):
scenarios = [
('no_endpoints',
dict(endpoints=[],
access_policy=None,
dispatch_to=None,
ctxt={}, msg=dict(method='foo'),
exposed_methods=['foo', 'bar', '_foobar'],
success=False, ex=oslo_messaging.UnsupportedVersion)),
('default_target',
dict(endpoints=[{}],
access_policy=None,
dispatch_to=dict(endpoint=0, method='foo'),
ctxt={}, msg=dict(method='foo'),
exposed_methods=['foo', 'bar', '_foobar'],
success=True, ex=None)),
('default_target_ctxt_and_args',
dict(endpoints=[{}],
access_policy=oslo_messaging.LegacyRPCAccessPolicy,
dispatch_to=dict(endpoint=0, method='bar'),
ctxt=dict(user='bob'), msg=dict(method='bar',
args=dict(blaa=True)),
exposed_methods=['foo', 'bar', '_foobar'],
success=True, ex=None)),
('default_target_namespace',
dict(endpoints=[{}],
access_policy=oslo_messaging.LegacyRPCAccessPolicy,
dispatch_to=dict(endpoint=0, method='foo'),
ctxt={}, msg=dict(method='foo', namespace=None),
exposed_methods=['foo', 'bar', '_foobar'],
success=True, ex=None)),
('default_target_version',
dict(endpoints=[{}],
access_policy=oslo_messaging.DefaultRPCAccessPolicy,
dispatch_to=dict(endpoint=0, method='foo'),
ctxt={}, msg=dict(method='foo', version='1.0'),
exposed_methods=['foo', 'bar'],
success=True, ex=None)),
('default_target_no_such_method',
dict(endpoints=[{}],
access_policy=oslo_messaging.DefaultRPCAccessPolicy,
dispatch_to=None,
ctxt={}, msg=dict(method='foobar'),
exposed_methods=['foo', 'bar'],
success=False, ex=oslo_messaging.NoSuchMethod)),
('namespace',
dict(endpoints=[{}, dict(namespace='testns')],
access_policy=oslo_messaging.DefaultRPCAccessPolicy,
dispatch_to=dict(endpoint=1, method='foo'),
ctxt={}, msg=dict(method='foo', namespace='testns'),
exposed_methods=['foo', 'bar'],
success=True, ex=None)),
('namespace_mismatch',
dict(endpoints=[{}, dict(namespace='testns')],
access_policy=oslo_messaging.DefaultRPCAccessPolicy,
dispatch_to=None,
ctxt={}, msg=dict(method='foo', namespace='nstest'),
exposed_methods=['foo', 'bar'],
success=False, ex=oslo_messaging.UnsupportedVersion)),
('version',
dict(endpoints=[dict(version='1.5'), dict(version='3.4')],
access_policy=oslo_messaging.DefaultRPCAccessPolicy,
dispatch_to=dict(endpoint=1, method='foo'),
ctxt={}, msg=dict(method='foo', version='3.2'),
exposed_methods=['foo', 'bar'],
success=True, ex=None)),
('version_mismatch',
dict(endpoints=[dict(version='1.5'), dict(version='3.0')],
access_policy=oslo_messaging.DefaultRPCAccessPolicy,
dispatch_to=None,
ctxt={}, msg=dict(method='foo', version='3.2'),
exposed_methods=['foo', 'bar'],
success=False, ex=oslo_messaging.UnsupportedVersion)),
('message_in_null_namespace_with_multiple_namespaces',
dict(endpoints=[dict(namespace='testns',
legacy_namespaces=[None])],
access_policy=oslo_messaging.DefaultRPCAccessPolicy,
dispatch_to=dict(endpoint=0, method='foo'),
ctxt={}, msg=dict(method='foo', namespace=None),
exposed_methods=['foo', 'bar'],
success=True, ex=None)),
('message_in_wrong_namespace_with_multiple_namespaces',
dict(endpoints=[dict(namespace='testns',
legacy_namespaces=['second', None])],
access_policy=oslo_messaging.DefaultRPCAccessPolicy,
dispatch_to=None,
ctxt={}, msg=dict(method='foo', namespace='wrong'),
exposed_methods=['foo', 'bar'],
success=False, ex=oslo_messaging.UnsupportedVersion)),
('message_with_endpoint_no_private_and_public_method',
dict(endpoints=[dict(namespace='testns',
legacy_namespaces=['second', None])],
access_policy=oslo_messaging.DefaultRPCAccessPolicy,
dispatch_to=dict(endpoint=0, method='foo'),
ctxt={}, msg=dict(method='foo', namespace='testns'),
exposed_methods=['foo', 'bar'],
success=True, ex=None)),
('message_with_endpoint_no_private_and_private_method',
dict(endpoints=[dict(namespace='testns',
legacy_namespaces=['second', None], )],
access_policy=oslo_messaging.DefaultRPCAccessPolicy,
dispatch_to=dict(endpoint=0, method='_foobar'),
ctxt={}, msg=dict(method='_foobar', namespace='testns'),
exposed_methods=['foo', 'bar'],
success=False, ex=oslo_messaging.NoSuchMethod)),
('message_with_endpoint_explicitly_exposed_without_exposed_method',
dict(endpoints=[dict(namespace='testns',
legacy_namespaces=['second', None], )],
access_policy=oslo_messaging.ExplicitRPCAccessPolicy,
dispatch_to=dict(endpoint=0, method='foo'),
ctxt={}, msg=dict(method='foo', namespace='testns'),
exposed_methods=['bar'],
success=False, ex=oslo_messaging.NoSuchMethod)),
('message_with_endpoint_explicitly_exposed_with_exposed_method',
dict(endpoints=[dict(namespace='testns',
legacy_namespaces=['second', None], )],
access_policy=oslo_messaging.ExplicitRPCAccessPolicy,
dispatch_to=dict(endpoint=0, method='bar'),
ctxt={}, msg=dict(method='bar', namespace='testns'),
exposed_methods=['bar'],
success=True, ex=None)),
]
def test_dispatcher(self):
endpoints = [mock.Mock(spec=_FakeEndpoint,
target=oslo_messaging.Target(**e))
for e in self.endpoints]
def _set_endpoint_mock_properties(endpoint):
endpoint.foo = mock.Mock(spec=dir(_FakeEndpoint.foo))
# mock doesn't pick up the decorated method.
endpoint.bar = mock.Mock(spec=dir(_FakeEndpoint.bar))
endpoint.bar.exposed = mock.PropertyMock(return_value=True)
endpoint._foobar = mock.Mock(spec=dir(_FakeEndpoint._foobar))
return endpoint
endpoints = [_set_endpoint_mock_properties(mock.Mock(
spec=_FakeEndpoint, target=oslo_messaging.Target(**e)))
for e in self.endpoints]
serializer = None
dispatcher = oslo_messaging.RPCDispatcher(endpoints, serializer)
dispatcher = oslo_messaging.RPCDispatcher(endpoints, serializer,
self.access_policy)
incoming = mock.Mock(ctxt=self.ctxt, message=self.msg)
@ -130,22 +199,23 @@ class TestDispatcher(test_utils.BaseTestCase):
self.assertEqual(self.msg.get('method'), ex.method)
else:
self.assertTrue(self.success,
"Not expected success of operation durung testing")
"Unexpected success of operation during testing")
self.assertIsNotNone(res)
for n, endpoint in enumerate(endpoints):
for method_name in ['foo', 'bar']:
for method_name in self.exposed_methods:
method = getattr(endpoint, method_name)
if self.dispatch_to and n == self.dispatch_to['endpoint'] and \
method_name == self.dispatch_to['method']:
method_name == self.dispatch_to['method'] and \
method_name in self.exposed_methods:
method.assert_called_once_with(
self.ctxt, **self.msg.get('args', {}))
else:
self.assertEqual(0, method.call_count)
self.assertEqual(0, method.call_count,
'method: {}'.format(method))
class TestSerializer(test_utils.BaseTestCase):
scenarios = [
('no_args_or_retval',
dict(ctxt={}, dctxt={}, args={}, retval=None)),
@ -174,7 +244,7 @@ class TestSerializer(test_utils.BaseTestCase):
for arg in self.args:
serializer.deserialize_entity(self.dctxt, arg).AndReturn('d' + arg)
serializer.serialize_entity(self.dctxt, self.retval).\
serializer.serialize_entity(self.dctxt, self.retval). \
AndReturn('s' + self.retval if self.retval else None)
self.mox.ReplayAll()

20
oslo_messaging/tests/rpc/test_server.py
View File

@ -21,6 +21,7 @@ import testscenarios
import mock
import oslo_messaging
from oslo_messaging import rpc
from oslo_messaging.rpc import server as rpc_server_module
from oslo_messaging import server as server_module
from oslo_messaging.tests import utils as test_utils
@ -861,3 +862,22 @@ class TestServerLocking(test_utils.BaseTestCase):
# We timed out. Ensure we didn't log anything.
self.assertFalse(mock_log.warning.called)
class TestRPCExposeDecorator(test_utils.BaseTestCase):
def foo(self):
pass
@rpc.expose
def bar(self):
"""bar docstring"""
pass
def test_undecorated(self):
self.assertRaises(AttributeError, lambda: self.foo.exposed)
def test_decorated(self):
self.assertEqual(True, self.bar.exposed)
self.assertEqual("""bar docstring""", self.bar.__doc__)
self.assertEqual('bar', self.bar.__name__)