openstack
/
magnetodb
Code Issues Proposed changes

Prevent leaking internal details out via exceptions 

When lower level exceptions occur, return a generic
message in the response to clients to prevent leaking
internal information outside. Added unit tests
for FaultWrapper. Added integration tests to check
responses to API calls when exception occur.

Change-Id: I63b214e427a0d7483b4421398aa9dd3371ffe0a0
Closes-Bug: #1415404
This commit is contained in:
Yogeshwar Shenoy 2015-02-06 14:53:46 +05:30
parent 60de9a3c51
commit 8a3b1947f6
3 changed files with 181 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
magnetodb/common/middleware/fault.py
View File

@ -118,8 +118,11 @@ class FaultWrapper(wsgi.Middleware):
LOG.debug(ex)
return req.get_response(Fault(self._error(ex)))
except Exception as ex:
# some lower lever exception. It is better to know about it
# some lower level exception. It is better to know about it
# so, log the original message
LOG.exception(ex)
# but don't propagate internal details beyond here
ex.args = (u'message="An Internal Error Occurred"',)
return req.get_response(Fault(self._error(ex)))
@classmethod

87
magnetodb/tests/unittests/api/openstack/v1/test_exception_responses.py Normal file
View File

@ -0,0 +1,87 @@
# Copyright 2015 Reliance Jio Infocomm Ltd.
# All Rights Reserved.
#
# 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.
import httplib
import mock
from oslo_serialization import jsonutils as json
from magnetodb.common import exception
from magnetodb.tests.unittests.api.openstack.v1 import test_base_testcase
class ExceptionResponsesTest(test_base_testcase.APITestCase):
"""Test for messages returned when exceptions occur in API calls.
This test uses the list_tables api call, since it is one of the
simpler ones.
"""
def _get_api_call_error(self):
"""Make the API call and return the response object."""
headers = {'Content-Type': 'application/json',
'Accept': 'application/json'}
conn = httplib.HTTPConnection('localhost:8080')
url = '/v1/data/default_tenant/tables'
conn.request("GET", url, headers=headers)
response = conn.getresponse()
json_response = response.read()
response_model = json.loads(json_response)
return (response_model['code'],
response_model['error']['message'])
@mock.patch('magnetodb.storage.list_tables')
def test_validation_error_message(self, mock_list_tables):
"""Test the error message received when a ValidationError occurs. """
expected_message = 'There was some validation error'
mock_list_tables.side_effect = \
exception.ValidationError(expected_message)
(code, message) = self._get_api_call_error()
self.assertEqual(expected_message, message)
self.assertEqual(400, code)
@mock.patch('magnetodb.storage.list_tables')
def test_backend_exception_message(self, mock_list_tables):
"""Test the error message received when a BackendInteractionException
occurs.
"""
expected_message = 'There was some backend interaction exception'
mock_list_tables.side_effect = \
exception.BackendInteractionException(expected_message)
(code, message) = self._get_api_call_error()
self.assertEqual(expected_message, message)
self.assertEqual(500, code)
@mock.patch('magnetodb.storage.list_tables')
def test_low_level_exception_message(self, mock_list_tables):
"""Test the error message received when a low level exception
occurs. In particular, the message should not expose low
level details.
"""
exception_message = 'This message contains low level details'
mock_list_tables.side_effect = Exception(exception_message)
(code, message) = self._get_api_call_error()
self.assertNotEqual(exception_message, message)
self.assertEqual(500, code)

90
magnetodb/tests/unittests/common/middleware/test_fault.py Normal file
View File

@ -0,0 +1,90 @@
# Copyright 2015 Reliance Jio Infocomm Ltd.
# All Rights Reserved.
#
# 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.
import unittest
import mock
from magnetodb.common.middleware import fault
from magnetodb.common import exception
class TestFaultWrapper(unittest.TestCase):
"""Unit tests for FaultWrapper."""
def setUp(self):
self.request = mock.Mock()
self.fault_wrapper = fault.FaultWrapper(None, dict())
def _side_effect_generator(self, return_values):
"""
Helper function to create the appropriate side_effect for Mock
objects used in this file.
In particular, this helper is for governing behavior when the
same mocked method is invoked multiple times. We pop the first
element in the list and take action based on that value.
:param return_values: the list of values governing behavior at
each invocation
:return: the side_effect which maybe passed to a Mock object
"""
def side_effect(*args):
result = return_values.pop(0)
if isinstance(result, Exception):
raise result
if result == 'use args':
return args
return result
return side_effect
def test_process_request_no_exception(self):
"""Test the positive case response. """
expected_response = "Request Succeeded"
self.request.get_response = mock.Mock(return_value=expected_response)
response = self.fault_wrapper.process_request(self.request)
self.assertEqual(expected_response, response)
def test_process_request_validation_error(self):
"""Test the case where a ValidationError exception occurs. """
expected_message = "A validation error occurred"
validation_ex = exception.ValidationError(expected_message)
side_effect = self._side_effect_generator([validation_ex, "use args"])
self.request.get_response = mock.Mock(side_effect=side_effect)
response = self.fault_wrapper.process_request(self.request)
message = response[0].error['error']['message']
self.assertEqual(expected_message, message)
def test_process_request_backend_exception(self):
"""Test the case where BackendInteractionException occurs. """
expected_message = "A backend interaction error occurred"
backend_ex = exception.BackendInteractionException(expected_message)
side_effect = self._side_effect_generator([backend_ex, "use args"])
self.request.get_response = mock.Mock(side_effect=side_effect)
response = self.fault_wrapper.process_request(self.request)
message = response[0].error['error']['message']
self.assertEqual(expected_message, message)
def test_process_request_low_level_exception(self):
"""Test case where a low level exception occurs. """
ex_message = "Low level internal details message"
ex = Exception(ex_message)
side_effect = self._side_effect_generator([ex, "use args"])
self.request.get_response = mock.Mock(side_effect=side_effect)
response = self.fault_wrapper.process_request(self.request)
message = response[0].error['error']['message']
self.assertNotEqual(ex_message, message)