openstack/blazar
Code Issues Proposed changes

Merge "Use built-in oslo context de/serialization" into stable/xena

Browse Source
This commit is contained in:
Zuul 2022-05-05 17:44:51 +00:00 committed by Gerrit Code Review
commit 2a34a6f6a2
14 changed files with 153 additions and 136 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
blazar/api/context.py
View File

@ -27,18 +27,5 @@ def ctx_from_headers(headers):
except TypeError:
raise exceptions.WrongFormat()
kwargs = {"user_id": headers['X-User-Id'],
"project_id": headers['X-Project-Id'],
"auth_token": headers['X-Auth-Token'],
"service_catalog": service_catalog,
"user_name": headers['X-User-Name'],
"project_name": headers['X-Project-Name'],
"roles": list(
map(str.strip, headers['X-Roles'].split(',')))}
# For v1 only, request_id and global_request_id will be available.
if headers.environ['PATH_INFO'].startswith('/v1'):
kwargs['request_id'] = headers.environ['openstack.request_id']
kwargs['global_request_id'] = headers.environ.get(
'openstack.global_request_id')
return context.BlazarContext(**kwargs)
return context.BlazarContext.from_environ(headers.environ,
service_catalog=service_catalog)

47
blazar/context.py
View File

@ -15,31 +15,29 @@
import threading
from oslo_config import cfg
from oslo_context import context
CONF = cfg.CONF
class BlazarContext(context.RequestContext):
# service_catalog is not by default read from a dict
# when deserializing a context.
FROM_DICT_EXTRA_KEYS = ['service_catalog']
_context_stack = threading.local()
def __init__(self, user_id=None, project_id=None, project_name=None,
service_catalog=None, user_name=None, **kwargs):
def __init__(self, service_catalog=None, **kwargs):
# NOTE(neha-alhat): During serializing/deserializing context object
# over the RPC layer, below extra parameters which are passed by
# `oslo.messaging` are popped as these parameters are not required.
kwargs.pop('client_timeout', None)
kwargs.pop('user_identity', None)
kwargs.pop('project', None)
if user_id:
kwargs['user_id'] = user_id
if project_id:
kwargs['project_id'] = project_id
super(BlazarContext, self).__init__(**kwargs)
self.project_name = project_name
self.user_name = user_name
self.service_catalog = service_catalog or []
if self.is_admin and 'admin' not in self.roles:
@ -65,28 +63,35 @@ class BlazarContext(context.RequestContext):
except (AttributeError, IndexError):
raise RuntimeError("Context isn't available here")
# NOTE(yorik-sar): as long as oslo.rpc requires this
def to_dict(self):
result = super(BlazarContext, self).to_dict()
result['user_id'] = self.user_id
result['user_name'] = self.user_name
result['project_id'] = self.project_id
result['project_name'] = self.project_name
result['service_catalog'] = self.service_catalog
return result
@classmethod
def elevated(cls):
def admin(cls):
try:
ctx = cls.current()
cur = cls.current()
request_id = cur.request_id
global_request_id = cur.global_request_id
service_catalog = cur.service_catalog
except RuntimeError:
ctx = None
return cls(ctx, is_admin=True)
request_id = global_request_id = service_catalog = None
return cls(
user_name=CONF.os_admin_username,
user_domain_name=CONF.os_admin_user_domain_name,
project_name=CONF.os_admin_project_name,
project_domain_name=CONF.os_admin_project_domain_name,
is_admin=True,
service_catalog=service_catalog,
request_id=request_id,
global_request_id=global_request_id
)
def current():
return BlazarContext.current()
def elevated():
return BlazarContext.elevated()
def admin():
return BlazarContext.admin()

2
blazar/policy.py
View File

@ -91,7 +91,7 @@ def enforce(context, action, target, do_raise=True):
init()
credentials = context.to_dict()
credentials = context.to_policy_values()
# Add the exceptions arguments if asked to do a raise
extra = {}

10
blazar/tests/api/__init__.py
View File

@ -27,6 +27,9 @@ from blazar import tests
PATH_PREFIX = '/v2'
FAKE_PROJECT = '981b767265174c108bc5a61185b748ac'
FAKE_USER = 'b4fdb2fff13545ceb751295096cc18ee'
class APITest(tests.TestCase):
"""Used for unittests tests of Pecan controllers."""
@ -40,11 +43,12 @@ class APITest(tests.TestCase):
def fake_ctx_from_headers(headers):
if not headers:
return context.BlazarContext(
user_id='fake', project_id='fake', roles=['member'])
user_id=FAKE_USER, project_id=FAKE_PROJECT,
roles=['member'])
roles = headers.get('X-Roles', str('member')).split(',')
return context.BlazarContext(
user_id=headers.get('X-User-Id', 'fake'),
project_id=headers.get('X-Project-Id', 'fake'),
user_id=headers.get('X-User-Id', FAKE_USER),
project_id=headers.get('X-Project-Id', FAKE_PROJECT),
auth_token=headers.get('X-Auth-Token', None),
service_catalog=None,
user_name=headers.get('X-User-Name', 'fake'),

30
blazar/tests/api/test_context.py
View File

@ -19,7 +19,6 @@ import webob
from werkzeug import wrappers
from blazar.api import context as api_context
from blazar import context
from blazar import exceptions
from blazar import tests
@ -33,9 +32,10 @@ class ContextTestCase(tests.TestCase):
'X-Project-Id': uuidsentinel.project_id,
'X-Auth-Token': '111-111-111',
'X-User-Name': 'user_name',
'X-User-Domain-Name': 'user_domain_name',
'X-Project-Name': 'project_name',
'X-Project-Domain-Name': 'project_domain_name',
'X-Roles': 'user_name0, user_name1'}
self.context = self.patch(context, 'BlazarContext')
self.catalog = jsonutils.dump_as_bytes({'nova': 'catalog'})
def test_ctx_from_headers_no_catalog(self):
@ -64,19 +64,24 @@ class ContextTestCaseV1(ContextTestCase):
'/v1/leases',
headers=self.fake_headers,
environ_base=environ_base)
api_context.ctx_from_headers(req.headers)
self.context.assert_called_once_with(
context = api_context.ctx_from_headers(req.headers)
expected = dict(
user_id=uuidsentinel.user_id,
roles=['user_name0',
'user_name1'],
project_name='project_name',
project_domain_name='project_domain_name',
auth_token='111-111-111',
service_catalog={'nova': 'catalog'},
project_id=uuidsentinel.project_id,
user_name='user_name',
user_domain_name='user_domain_name',
request_id='req-' + uuidsentinel.reqid,
global_request_id='req-' + uuidsentinel.globalreqid)
global_request_id='req-' + uuidsentinel.globalreqid
)
for k, v in expected.items():
self.assertEqual(getattr(context, k, None), v)
class ContextTestCaseV2(ContextTestCase):
@ -85,14 +90,19 @@ class ContextTestCaseV2(ContextTestCase):
self.fake_headers['X-Service-Catalog'] = self.catalog
req = webob.Request.blank('/v2/leases')
req.headers = self.fake_headers
api_context.ctx_from_headers(req.headers)
self.context.assert_called_once_with(
context = api_context.ctx_from_headers(req.headers)
expected = dict(
user_id=uuidsentinel.user_id,
roles=['user_name0',
'user_name1'],
project_name='project_name',
project_domain_name='project_domain_name',
auth_token='111-111-111',
service_catalog={'nova': 'catalog'},
project_id=uuidsentinel.project_id,
user_name='user_name')
user_name='user_name',
user_domain_name='user_domain_name'
)
for k, v in expected.items():
self.assertEqual(getattr(context, k, None), v)

5
blazar/tests/api/v2/test_leases.py
View File

@ -28,9 +28,8 @@ def fake_lease(**kw):
'end_date': kw.get('end_date', '2014-02-01 13:37'),
'trust_id': kw.get('trust_id',
'35b17138b3644e6aa1318f3099c5be68'),
'user_id': kw.get('user_id', 'efd8780712d24b389c705f5c2ac427ff'),
'project_id': kw.get('project_id',
'bd9431c18d694ad3803a8d4a6b89fd36'),
'user_id': kw.get('user_id', api.FAKE_USER),
'project_id': kw.get('project_id', api.FAKE_PROJECT),
'reservations': kw.get('reservations', [
{
'resource_id': '1234',

59
blazar/tests/test_context.py
View File

@ -13,14 +13,27 @@
# See the License for the specific language governing permissions and
# limitations under the License.
from oslo_config import cfg
from oslo_config import fixture as conf_fixture
from oslo_utils.fixture import uuidsentinel
from blazar import context
from blazar import tests
CONF = cfg.CONF
class TestBlazarContext(tests.TestCase):
def setUp(self):
super(TestBlazarContext, self).setUp()
self.cfg = self.useFixture(conf_fixture.Config(CONF))
self.cfg.config(os_admin_username='fake-admin')
self.cfg.config(os_admin_user_domain_name='fake-admin-domain')
self.cfg.config(os_admin_project_name='fake-admin-project')
self.cfg.config(os_admin_project_domain_name='fake-admin-domain')
def test_to_dict(self):
ctx = context.BlazarContext(
user_id=111, project_id=222,
@ -33,8 +46,6 @@ class TestBlazarContext(tests.TestCase):
'is_admin_project': True,
'project': 222,
'project_domain': None,
'project_id': 222,
'project_name': None,
'read_only': False,
'request_id': 'req-679033b7-1755-4929-bf85-eb3bfaef7e0b',
'resource_uuid': None,
@ -45,15 +56,9 @@ class TestBlazarContext(tests.TestCase):
'tenant': 222,
'user': 111,
'user_domain': None,
'user_id': 111,
'user_identity': '111 222 - - -',
'user_name': None}
'user_identity': '111 222 - - -'}
self.assertEqual(expected, ctx.to_dict())
def test_elevated_empty(self):
ctx = context.BlazarContext.elevated()
self.assertTrue(ctx.is_admin)
def test_service_catalog_default(self):
ctxt = context.BlazarContext(user_id=uuidsentinel.user_id,
project_id=uuidsentinel.project_id)
@ -69,14 +74,30 @@ class TestBlazarContext(tests.TestCase):
service_catalog=None)
self.assertEqual([], ctxt.service_catalog)
def test_blazar_context_elevated(self):
user_context = context.BlazarContext(
user_id=uuidsentinel.user_id,
project_id=uuidsentinel.project_id, is_admin=False)
self.assertFalse(user_context.is_admin)
def test_admin(self):
ctx = context.admin()
self.assertEqual(ctx.user_name, 'fake-admin')
self.assertEqual(ctx.user_domain_name, 'fake-admin-domain')
self.assertEqual(ctx.project_name, 'fake-admin-project')
self.assertEqual(ctx.project_domain_name, 'fake-admin-domain')
self.assertEqual(ctx.is_admin, True)
admin_context = user_context.elevated()
self.assertFalse(user_context.is_admin)
self.assertTrue(admin_context.is_admin)
self.assertNotIn('admin', user_context.roles)
self.assertIn('admin', admin_context.roles)
def test_admin_nested(self):
"""Test that admin properties take priority over current context."""
request_id = 'req-679033b7-1755-4929-bf85-eb3bfaef7e0b'
service_catalog = ['foo']
ctx = context.BlazarContext(
user_name='fake-user', user_domain_name='fake-user-domain',
project_name='fake-project',
project_domain_name='fake-user-domain',
service_catalog=service_catalog, request_id=request_id)
with ctx:
admin_ctx = context.admin()
self.assertEqual(admin_ctx.user_name, 'fake-admin')
self.assertEqual(admin_ctx.user_domain_name, 'fake-admin-domain')
self.assertEqual(admin_ctx.project_name, 'fake-admin-project')
self.assertEqual(admin_ctx.project_domain_name,
'fake-admin-domain')
self.assertEqual(admin_ctx.is_admin, True)
self.assertEqual(admin_ctx.request_id, request_id)
self.assertEqual(admin_ctx.service_catalog, service_catalog)

9
blazar/tests/test_policy.py
View File

@ -52,15 +52,6 @@ class BlazarPolicyTestCase(tests.TestCase):
self.assertRaises(exceptions.PolicyNotAuthorized, policy.enforce,
self.context, action, target)
def test_elevatedpolicy(self):
target = {'user_id': self.context.user_id,
'project_id': self.context.project_id}
action = "blazar:oshosts:get"
self.assertRaises(exceptions.PolicyNotAuthorized, policy.enforce,
self.context, action, target)
elevated_context = self.context.elevated()
self.assertTrue(policy.enforce(elevated_context, action, target))
def test_authorize(self):
@policy.authorize('leases', 'get', ctx=self.context)

71
blazar/tests/utils/openstack/test_keystone.py
View File

@ -38,56 +38,59 @@ class TestCKClient(tests.TestCase):
self.version = '3'
self.username = 'fake_user'
self.user_domain_name = 'fake_user_domain'
self.token = 'fake_token'
self.password = 'fake_pass'
self.tenant_name = 'fake_project'
self.project_name = 'fake_project'
self.project_domain_name = 'fake_project_domain'
self.auth_url = 'fake_url'
self.trust_id = 'fake_trust'
def test_client_from_kwargs(self):
self.ctx.side_effect = RuntimeError
self.keystone.BlazarKeystoneClient(version=self.version,
username=self.username,
password=self.password,
tenant_name=self.tenant_name,
trust_id=self.trust_id,
auth_url=self.auth_url)
self.client.assert_called_once_with(version=self.version,
trust_id=self.trust_id,
username=self.username,
password=self.password,
auth_url=self.auth_url)
self.keystone.BlazarKeystoneClient(
version=self.version,
username=self.username,
password=self.password,
project_name=self.project_name,
trust_id=self.trust_id,
auth_url=self.auth_url)
self.client.assert_called_once_with(
version=self.version,
trust_id=self.trust_id,
username=self.username,
password=self.password,
auth_url=self.auth_url)
def test_client_from_kwargs_and_ctx(self):
self.keystone.BlazarKeystoneClient(version=self.version,
username=self.username,
password=self.password,
tenant_name=self.tenant_name,
auth_url=self.auth_url)
self.client.assert_called_once_with(version=self.version,
tenant_name=self.tenant_name,
endpoint='http://fake.com/',
username=self.username,
password=self.password,
auth_url=self.auth_url,
global_request_id=self.
context.current().
global_request_id)
self.keystone.BlazarKeystoneClient(
version=self.version,
username=self.username,
user_domain_name=self.user_domain_name,
password=self.password,
project_name=self.project_name,
project_domain_name=self.project_domain_name,
auth_url=self.auth_url)
self.client.assert_called_once_with(
version=self.version,
username=self.username,
user_domain_name=self.user_domain_name,
project_name=self.project_name,
project_domain_name=self.project_domain_name,
endpoint='http://fake.com/',
password=self.password,
auth_url=self.auth_url,
global_request_id=self.context.current().global_request_id)
def test_client_from_ctx(self):
self.keystone.BlazarKeystoneClient()
self.client.assert_called_once_with(
version='3',
username=self.ctx().user_name,
user_domain_name=self.ctx().user_domain_name,
token=self.ctx().auth_token,
tenant_name=self.ctx().project_name,
project_name=self.ctx().project_name,
project_domain_name=self.ctx().project_domain_name,
auth_url='http://fake.com/',
endpoint='http://fake.com/',
global_request_id=self.context.current().global_request_id)

8
blazar/tests/utils/test_trusts.py
View File

@ -65,10 +65,8 @@ class TestTrusts(tests.TestCase):
'global_request_id': self.context.current().global_request_id,
'is_admin': False,
'is_admin_project': True,
'project': self.client().tenant_id,
'project': self.client().project_id,
'project_domain': None,
'project_id': self.client().tenant_id,
'project_name': 'admin',
'read_only': False,
'request_id': ctx.request_id,
'resource_uuid': None,
@ -76,10 +74,8 @@ class TestTrusts(tests.TestCase):
'service_catalog': ctx.service_catalog,
'show_deleted': False,
'system_scope': None,
'tenant': self.client().tenant_id,
'user': None,
'user_domain': None,
'user_id': None}
'user_domain': None}
self.assertDictContainsSubset(fake_ctx_dict, ctx.to_dict())
def test_use_trust_auth_dict(self):

8
blazar/utils/openstack/keystone.py
View File

@ -101,7 +101,9 @@ class BlazarKeystoneClient(object):
kwargs.setdefault('version', cfg.CONF.keystone_client_version)
if ctx is not None:
kwargs.setdefault('username', ctx.user_name)
kwargs.setdefault('tenant_name', ctx.project_name)
kwargs.setdefault('user_domain_name', ctx.user_domain_name)
kwargs.setdefault('project_name', ctx.project_name)
kwargs.setdefault('project_domain_name', ctx.project_domain_name)
kwargs.setdefault('global_request_id', ctx.global_request_id)
if not kwargs.get('auth_url'):
kwargs['auth_url'] = base.url_for(
@ -121,8 +123,8 @@ class BlazarKeystoneClient(object):
# NOTE(dbelova): we need this checking to support current
# keystoneclient: token can only be scoped now to either
# a trust or project, not both.
if kwargs.get('trust_id') and kwargs.get('tenant_name'):
kwargs.pop('tenant_name')
if kwargs.get('trust_id') and kwargs.get('project_name'):
kwargs.pop('project_name')
try:
# NOTE(n.s.): we shall remove this try: except: clause when

2
blazar/utils/service.py
View File

@ -74,7 +74,7 @@ class ContextEndpointHandler(object):
method = getattr(self.__endpoint, name)
def run_method(__ctx, **kwargs):
with context.BlazarContext(**__ctx):
with context.BlazarContext.from_dict(__ctx):
return method(**kwargs)
return run_method

16
blazar/utils/trusts.py
View File

@ -29,9 +29,8 @@ def create_trust():
client = keystone.BlazarKeystoneClient()
trustee_id = keystone.BlazarKeystoneClient(
username=CONF.os_admin_username,
password=CONF.os_admin_password,
tenant_name=CONF.os_admin_project_name).user_id
ctx=context.admin()).user_id
ctx = context.current()
trust = client.trusts.create(trustor_user=ctx.user_id,
@ -53,31 +52,26 @@ def create_ctx_from_trust(trust_id):
"""Return context built from given trust."""
ctx = context.current()
ctx = context.BlazarContext(
user_name=CONF.os_admin_username,
project_name=CONF.os_admin_project_name,
request_id=ctx.request_id,
global_request_id=ctx.global_request_id
)
auth_url = "%s://%s:%s" % (CONF.os_auth_protocol,
base.get_os_auth_host(CONF),
CONF.os_auth_port)
if CONF.os_auth_prefix:
auth_url += "/%s" % CONF.os_auth_prefix
client = keystone.BlazarKeystoneClient(
password=CONF.os_admin_password,
trust_id=trust_id,
auth_url=auth_url,
ctx=ctx,
ctx=context.admin(),
)
# use 'with ctx' statement in the place you need context from trust
return context.BlazarContext(
user_name=ctx.user_name,
project_name=ctx.project_name,
user_domain_name=ctx.user_domain_name,
auth_token=client.auth_token,
service_catalog=client.service_catalog.catalog['catalog'],
project_id=client.tenant_id,
project_id=client.project_id,
request_id=ctx.request_id,
global_request_id=ctx.global_request_id
)

5
releasenotes/notes/bug-1881162-ebe012fcc7176594.yaml Normal file
View File

@ -0,0 +1,5 @@
---
fixes:
- |
Allows users of multiple Keystone domains to create leases; previously only users
and projects in the default domain could use Blazar.