Prevent form data from being consumed by Flask

samcli/local/apigw/service.py

@@ -347,7 +347,7 @@ def _construct_event(flask_request, port, binary_types):
         endpoint = PathConverter.convert_path_to_api_gateway(flask_request.endpoint)
         method = flask_request.method
 
-        request_data = flask_request.data
+        request_data = flask_request.get_data()
 
         request_mimetype = flask_request.mimetype
 

tests/functional/local/apigw/test_service.py

@@ -223,6 +223,28 @@ def test_calling_service_with_data_on_path(self):
         self.assertEquals(response.status_code, 200)
         self.assertEquals(response.headers.get('Content-Type'), "application/json")
 
+    def test_calling_service_with_form_data_on_path(self):
+        path = "/something"
+        body = {"key1": "value1"}
+        expected = make_service_response(self.port,
+                                         scheme=self.scheme,
+                                         method="POST",
+                                         resourcePath=path,
+                                         resolvedResourcePath=path,
+                                         pathParameters=None,
+                                         body='key1=value1',
+                                         queryParams=None,
+                                         headers={"Content-Length": "11",
+                                                  "Content-Type": "application/x-www-form-urlencoded"})
+
+        response = requests.post(self.url + path, data=body)
+
+        actual = response.json()
+
+        self.assertEquals(actual, expected)
+        self.assertEquals(response.status_code, 200)
+        self.assertEquals(response.headers.get('Content-Type'), "application/json")
+
     def test_calling_service_with_path_params(self):
         path = '/something/event1'
         expected = make_service_response(self.port,

tests/integration/local/start_api/test_start_api.py

@@ -369,6 +369,21 @@ def test_binary_request(self):
         self.assertEquals(response.headers.get("Content-Type"), "image/gif")
         self.assertEquals(response.content, input_data)
 
+    def test_request_with_form_data(self):
+        """
+        Form-encoded data should be put into the Event to Lambda
+        """
+        response = requests.post(self.url + "/echoeventbody",
+                                 headers={"Content-Type": "application/x-www-form-urlencoded"},
+                                 data='key=value')
+
+        self.assertEquals(response.status_code, 200)
+
+        response_data = response.json()
+
+        self.assertEquals(response_data.get("headers").get("Content-Type"), "application/x-www-form-urlencoded")
+        self.assertEquals(response_data.get("body"), "key=value")
+
     def test_request_with_query_params(self):
         """
         Query params given should be put into the Event to Lambda

tests/integration/testdata/start_api/template.yaml

@@ -53,6 +53,12 @@ Resources:
             Method: GET
             Path: /id/{id}
 
+        EchoEventBodyPath:
+          Type: Api
+          Properties:
+            Method: POST
+            Path: /echoeventbody
+
   ContentTypeSetterFunction:
     Type: AWS::Serverless::Function
     Properties:

tests/unit/local/apigw/test_service.py

@@ -445,7 +445,7 @@ def setUp(self):
         self.request_mock.path = "path"
         self.request_mock.method = "GET"
         self.request_mock.remote_addr = "190.0.0.0"
-        self.request_mock.data = b"DATA!!!!"
+        self.request_mock.get_data.return_value = b"DATA!!!!"
         self.request_mock.args = {"query": ["params"]}
         self.request_mock.headers = {"Content-Type": "application/json", "X-Test": "Value"}
         self.request_mock.view_args = {"path": "params"}
@@ -471,7 +471,7 @@ def test_construct_event_with_data(self):
         self.assertEquals(json.loads(actual_event_str), self.expected_dict)
 
     def test_construct_event_no_data(self):
-        self.request_mock.data = None
+        self.request_mock.get_data.return_value = None
         self.expected_dict["body"] = None
 
         actual_event_str = Service._construct_event(self.request_mock, 3000, binary_types=[])
@@ -484,7 +484,7 @@ def test_construct_event_with_binary_data(self, should_base64_encode_patch):
         binary_body = b"011000100110100101101110011000010111001001111001"  # binary in binary
         base64_body = base64.b64encode(binary_body).decode('utf-8')
 
-        self.request_mock.data = binary_body
+        self.request_mock.get_data.return_value = binary_body
         self.expected_dict["body"] = base64_body
         self.expected_dict["isBase64Encoded"] = True