Merge pull request #29 from dave-shawley/handle-bad-content-type

Handle Invalid Content-Type headers

docs/history.rst

@@ -1,6 +1,11 @@
 Version History
 ===============
 
+Next Release
+------------
+- Return a "400 Bad Request" when an invalid Content-Type header is received
+  instead of failing with an internal server error
+
 `3.0.3`_ (14 Sep 2020)
 ----------------------
 - Import from collections.abc instead of collections (thanks @nullsvm)

sprockets/mixins/mediatype/content.py

@@ -319,9 +319,14 @@ class ContentMixin:
         """
         if self._request_body is None:
             settings = get_settings(self.application, force_instance=True)
-            content_type_header = headers.parse_content_type(
-                self.request.headers.get('Content-Type',
-                                         settings.default_content_type))
+            content_type = self.request.headers.get(
+                'Content-Type', settings.default_content_type)
+
+            try:
+                content_type_header = headers.parse_content_type(content_type)
+            except ValueError:
+                raise web.HTTPError(400, 'failed to parse content type %s',
+                                    content_type)
             content_type = '/'.join([content_type_header.content_type,
                                      content_type_header.content_subtype])
             if content_type_header.content_suffix is not None:

tests.py

@@ -114,9 +114,13 @@ class SendResponseTests(testing.AsyncHTTPTestCase):
 
 
 class GetRequestBodyTests(testing.AsyncHTTPTestCase):
+    def setUp(self):
+        self.app = None
+        super().setUp()
 
     def get_app(self):
-        return examples.make_application(debug=True)
+        self.app = examples.make_application(debug=True)
+        return self.app
 
     def test_that_request_with_unhandled_type_results_in_415(self):
         response = self.fetch(
@@ -157,6 +161,13 @@ class GetRequestBodyTests(testing.AsyncHTTPTestCase):
         self.assertEqual(response.code, 200)
         self.assertEqual(json.loads(response.body.decode()), body)
 
+    def test_that_invalid_content_types_result_in_bad_request(self):
+        content.set_default_content_type(self.app, None, None)
+        response = self.fetch(
+            '/', method='POST', body='{"hi":"there"}',
+            headers={'Content-Type': 'application-json'})
+        self.assertEqual(response.code, 400)
+
 
 class JSONTranscoderTests(unittest.TestCase):