open_api: more request body support

This fixes python2 support of request body and adds support for
base64 format. Multipart request body are not supported at this point.

bt5/erp5_open_api/DocumentTemplateItem/portal_components/document.erp5.OpenAPIService.py

@@ -25,9 +25,13 @@
 #
 ##############################################################################
 
+import base64
+import binascii
 import json
 import typing
+import six
 from six.moves.urllib.parse import unquote
+
 if typing.TYPE_CHECKING:
   from typing import Any, Callable, Optional
   from erp5.component.document.OpenAPITypeInformation import OpenAPIOperation, OpenAPIParameter
@@ -258,10 +262,8 @@ class OpenAPIService(XMLObject):
         parameter,
         parameter.getJSONSchema(),
       )
-    requestBody = self.validateParameter(
-      'request body',
+    requestBody = self.validateRequestBody(
       operation.getRequestBodyValue(request),
-      {},
       operation.getRequestBodyJSONSchema(request),
     )
     if requestBody:
@@ -296,6 +298,37 @@ class OpenAPIService(XMLObject):
             parameter_name=parameter_name, e=e.message), str(e))
     return parameter_value
 
+  security.declareProtected(
+    Permissions.AccessContentsInformation, 'validateRequestBody')
+
+  def validateRequestBody(self, parameter_value, schema):
+    # type: (str, dict) -> Any
+    """Validate the request body raising a ParameterValidationError
+    when the parameter is not valid according to the corresponding schema.
+    """
+    if schema is not None:
+      if schema.get('type') == 'string':
+        if schema.get('format') == 'base64':
+          try:
+            return base64.b64decode(parameter_value)
+          except (binascii.Error, TypeError) as e:
+            if isinstance(e, TypeError):
+              # BBB on python2 this raises a generic type error
+              # but we don't want to ignore potential TypeErrors
+              # on python3 here
+              if six.PY3:
+                raise
+            raise ParameterValidationError(
+              'Error validating request body: {e}'.format(e=str(e)))
+        elif schema.get('format') == 'binary':
+          return parameter_value or b''
+    return self.validateParameter(
+      'request body',
+      parameter_value,
+      {},
+      schema,
+    )
+
   def executeMethod(self, request):
     # type: (HTTPRequest) -> Any
     operation = self.getMatchingOperation(request)

bt5/erp5_open_api/DocumentTemplateItem/portal_components/document.erp5.OpenAPITypeInformation.py

@@ -175,13 +175,15 @@ class OpenAPIOperation(dict):
     # type: (HTTPRequest) -> Optional[dict]
     """Returns the schema for the request body, or None if no `requestBody` defined
     """
-    request_content_type = request.getHeader('content-type')
-    # TODO there might be $ref ?
-    request_body_definition = self.get(
-      'requestBody', {'content': {}})['content'].get(request_content_type)
-    if request_body_definition:
-      return SchemaWithComponents(
-        self._schema, request_body_definition.get('schema', {}))
+    exact_request_content_type = request.getHeader('content-type')
+    wildcard_request_content_type = '%s/*' % ((exact_request_content_type or '').split('/')[0])
+    for request_content_type in exact_request_content_type, wildcard_request_content_type, '*/*':
+      # TODO there might be $ref ?
+      request_body_definition = self.get(
+        'requestBody', {'content': {}})['content'].get(request_content_type)
+      if request_body_definition:
+        return SchemaWithComponents(
+          self._schema, request_body_definition.get('schema', {}))
 
 
 class OpenAPIParameter(dict):

bt5/erp5_open_api/TestTemplateItem/portal_components/test.erp5.testOpenAPIService.py

@@ -25,6 +25,13 @@
 #
 ##############################################################################
 
+import six
+# pylint:disable=no-name-in-module
+if six.PY2:
+  from base64 import encodestring as base64_encodebytes
+else:
+  from base64 import encodebytes as base64_encodebytes
+# pylint:enable=no-name-in-module
 import io
 import json
 import unittest
@@ -1242,3 +1249,72 @@ class TestURLPathWithWebSiteAndVirtualHost(OpenAPIPetStoreTestCase):
         self.connector.getRelativeUrl()
     ))
     self.assertEqual(response.getBody(), b'"ok"')
+
+
+class TestOpenAPIRequestBody(OpenAPITestCase):
+  _type_id = 'Test Open API Request Body'
+
+  _open_api_schema = json.dumps(
+    {
+      'openapi': '3.0.3',
+      'info': {
+        'title': 'TestOpenAPIRequestBody',
+        'version': '0.0.0'
+      },
+      'paths': {
+        '/post': {
+          'post': {
+            'operationId': 'testPostByContentType',
+            'requestBody': {
+              'content': {
+                'image/*': {
+                  'schema': {
+                    'type': 'string',
+                    'format': 'binary',
+                  }
+                },
+                'application/x-base64': {
+                  'schema': {
+                    'type': 'string',
+                    'format': 'base64',
+                  }
+                }
+              }
+            }
+          }
+        }
+      }
+    })
+
+  def test_request_body_content_encoding(self):
+    self.addPythonScript(
+      'TestOpenAPIRequestBody_testPostByContentType',
+      'body=None',
+      'container.REQUEST.RESPONSE.setHeader("Content-Type", "application/octet-stream")\n'
+      'return body',
+    )
+    response = self.publish(
+      self.connector.getPath() + '/post',
+      request_method='POST',
+      stdin=io.BytesIO(b'png file content'),
+      env={"CONTENT_TYPE": 'image/png'})
+    self.assertEqual(response.getBody(), b'png file content')
+    self.assertEqual(response.getStatus(), 200)
+
+    response = self.publish(
+      self.connector.getPath() + '/post',
+      request_method='POST',
+      stdin=io.BytesIO(base64_encodebytes(b'base64 file content')),
+      env={"CONTENT_TYPE": 'application/x-base64'})
+    self.assertEqual(response.getBody(), b'base64 file content')
+    self.assertEqual(response.getStatus(), 200)
+
+    response = self.publish(
+      self.connector.getPath() + '/post',
+      request_method='POST',
+      stdin=io.BytesIO(b'not base64'),
+      env={"CONTENT_TYPE": 'application/x-base64'})
+    self.assertEqual(response.getStatus(), 400)
+    body = json.loads(response.getBody())
+    self.assertEqual(body['type'], 'parameter-validation-error')
+    self.assertIn('Error validating request body:', body['title'])