Support noauth2 auth_strategy

Currently, noauth2 auth_strategy is not supported. If you specify
[DEFAULT]auth_strategy=noauth2 in `/etc/masakari/ masakari.conf`
and start the masakari api service, it gives ImportError.

Added NoAuthMiddleware to support noauth2 auth_strategy.

Closes-Bug: #1836354
Change-Id: I288cd5838b09d3e6c31b6408fff85e6bb5f529a6
This commit is contained in:
Arthur Dayne 2019-07-13 10:37:30 +08:00 committed by Tushar Patil
parent 24f678cd29
commit 3bc725eb7a
3 changed files with 96 additions and 0 deletions

View File

@ -7,6 +7,7 @@ use = call:masakari.api.urlmap:urlmap_factory
[composite:masakari_api_v1]
use = call:masakari.api.auth:pipeline_factory_v1
keystone = cors http_proxy_to_wsgi request_id faultwrap sizelimit authtoken keystonecontext osapi_masakari_app_v1
noauth2 = cors http_proxy_to_wsgi request_id faultwrap sizelimit noauth2 osapi_masakari_app_v1
# filters
[filter:cors]

View File

@ -125,3 +125,45 @@ class MasakariKeystoneContext(wsgi.Middleware):
roles = req.headers.get('X_ROLES', '')
return [r.strip() for r in roles.split(',')]
class NoAuthMiddleware(wsgi.Middleware):
"""Return a fake token if one isn't specified.
noauth2 provides admin privs if 'admin' is provided as the user id.
"""
@webob.dec.wsgify(RequestClass=wsgi.Request)
def __call__(self, req):
user_id = req.headers.get('X_USER', 'admin')
user_id = req.headers.get('X_USER_ID', user_id)
project_name = req.headers.get('X_TENANT_NAME')
user_name = req.headers.get('X_USER_NAME')
req_id = req.environ.get(request_id.ENV_REQUEST_ID)
remote_address = req.remote_addr
if CONF.use_forwarded_for:
remote_address = req.headers.get('X-Forwarded-For', remote_address)
service_catalog = None
if req.headers.get('X_SERVICE_CATALOG') is not None:
try:
catalog_header = req.headers.get('X_SERVICE_CATALOG')
service_catalog = jsonutils.loads(catalog_header)
except ValueError:
raise webob.exc.HTTPInternalServerError(
_('Invalid service catalog json.'))
ctx = context.RequestContext(user_id,
user_name=user_name,
project_name=project_name,
remote_address=remote_address,
service_catalog=service_catalog,
request_id=req_id,
is_admin=True)
req.environ['masakari.context'] = ctx
return self.application

View File

@ -79,6 +79,59 @@ class TestMasakariKeystoneContextMiddleware(test.NoDBTestCase):
self.assertEqual(req_id, self.context.request_id)
class TestNoAuthMiddleware(test.NoDBTestCase):
def setUp(self):
super(TestNoAuthMiddleware, self).setUp()
@webob.dec.wsgify()
def fake_app(req):
self.context = req.environ['masakari.context']
return webob.Response()
self.context = None
self.middleware = masakari.api.auth.NoAuthMiddleware(fake_app)
self.request = webob.Request.blank('/')
self.request.headers['X_SERVICE_CATALOG'] = jsonutils.dumps({})
def test_no_user_or_user_id(self):
response = self.request.get_response(self.middleware)
self.assertEqual(response.status_int, http.OK)
def test_user_id_only(self):
self.request.headers['X_USER_ID'] = 'testuserid'
response = self.request.get_response(self.middleware)
self.assertEqual(response.status_int, http.OK)
self.assertEqual(self.context.user_id, 'testuserid')
def test_user_only(self):
self.request.headers['X_USER'] = 'testuser'
response = self.request.get_response(self.middleware)
self.assertEqual(response.status_int, http.OK)
self.assertEqual(self.context.user_id, 'testuser')
def test_user_id_trumps_user(self):
self.request.headers['X_USER_ID'] = 'testuserid'
self.request.headers['X_USER'] = 'testuser'
response = self.request.get_response(self.middleware)
self.assertEqual(response.status_int, http.OK)
self.assertEqual(self.context.user_id, 'testuserid')
def test_invalid_service_catalog(self):
self.request.headers['X_USER'] = 'testuser'
self.request.headers['X_SERVICE_CATALOG'] = "bad json"
response = self.request.get_response(self.middleware)
self.assertEqual(response.status_int, http.INTERNAL_SERVER_ERROR)
def test_request_id_extracted_from_env(self):
req_id = 'dummy-request-id'
self.request.headers['X_PROJECT_ID'] = 'testtenantid'
self.request.headers['X_USER_ID'] = 'testuserid'
self.request.environ[request_id.ENV_REQUEST_ID] = req_id
self.request.get_response(self.middleware)
self.assertEqual(req_id, self.context.request_id)
class TestKeystoneMiddlewareRoles(test.NoDBTestCase):
def setUp(self):