[IMP] endpoint: implement optional request content schema validation

Author: ivantodorovich

endpoint/__manifest__.py

@@ -11,6 +11,9 @@
     "maintainers": ["simahawk"],
     "website": "https://github.com/OCA/web-api",
     "depends": ["endpoint_route_handler", "rpc_helper"],
+    "external_dependencies": {
+        "python": ["jsonschema", "PyYAML"],
+    },
     "data": [
         "data/server_action.xml",
         "security/ir.model.access.csv",

endpoint/exceptions.py

@@ -0,0 +1,26 @@
+# Copyright 2026 Camptocamp SA (https://www.camptocamp.com).
+# License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl).
+
+import json
+
+import werkzeug.exceptions
+import werkzeug.wrappers
+
+
+class RequestValidationError(werkzeug.exceptions.BadRequest):
+    """Bad request raised when the body fails JSON Schema validation.
+
+    Emits ``{"detail": [{"loc", "msg", "type"}, ...]}`` (FastAPI-style)
+    instead of the generic werkzeug HTML body.
+    """
+
+    def __init__(self, detail):
+        super().__init__()
+        self.detail = detail
+
+    def get_response(self, environ=None, scope=None):
+        return werkzeug.wrappers.Response(
+            json.dumps({"detail": self.detail}),
+            status=self.code,
+            mimetype="application/json",
+        )

endpoint/models/endpoint_mixin.py

@@ -2,16 +2,22 @@
 # @author: Simone Orsi <simone.orsi@camptocamp.com>
 # License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl).
 
+import json
 import textwrap
 
+import jsonschema
 import werkzeug
+import yaml
+from lxml import etree
 
 from odoo import api, exceptions, fields, http, models
 from odoo.exceptions import UserError
 from odoo.tools import safe_eval
 
 from odoo.addons.rpc_helper.decorator import disable_rpc
 
+from ..exceptions import RequestValidationError
+
 hashlib = safe_eval.wrap_module(
     __import__("hashlib"),
     [
@@ -48,6 +54,18 @@ class EndpointMixin(models.AbstractModel):
         selection="_selection_exec_mode",
         required=True,
     )
+    request_content_schema = fields.Text(
+        help=(
+            "Optional schema validated against the parsed request body. "
+            "The accepted format depends on 'Request content type':\n"
+            " - application/json: JSON Schema (Draft 2020-12), as YAML or JSON\n"
+            " - application/xml, text/xml: XML Schema (XSD)\n"
+            "If empty, no validation runs."
+        ),
+    )
+    request_content_schema_applicable = fields.Boolean(
+        compute="_compute_request_content_schema_applicable",
+    )
     code_snippet = fields.Text()
     code_snippet_docs = fields.Text(
         compute="_compute_code_snippet_docs",
@@ -80,6 +98,69 @@ def _validate_exec__code(self):
                 )
             )
 
+    def _get_request_content_schema_applicable_for_types(self):
+        """Content types for which ``request_content_schema`` applies."""
+        return ["application/json", "application/xml"]
+
+    @api.depends("request_method", "request_content_type")
+    def _compute_request_content_schema_applicable(self):
+        applicable_types = self._get_request_content_schema_applicable_for_types()
+        for rec in self:
+            rec.request_content_schema_applicable = (
+                rec.request_method in ("POST", "PUT")
+                and rec.request_content_type in applicable_types
+            )
+
+    @api.onchange("request_content_type")
+    def _onchange_request_content_type_clear_schema(self):
+        # The schema format depends on the content type (e.g. JSON Schema vs
+        # XSD), so it cannot survive a content type change.
+        self.request_content_schema = False
+
+    @api.constrains("request_content_schema", "request_content_type")
+    def _check_request_content_schema(self):
+        for rec in self:
+            if not rec.request_content_schema:
+                continue
+            elif rec.request_content_type == "application/json":
+                rec._check_request_content_schema_json()
+            elif rec.request_content_type == "application/xml":
+                rec._check_request_content_schema_xml()
+
+    def _check_request_content_schema_json(self):
+        try:
+            schema = yaml.safe_load(self.request_content_schema)
+        except yaml.YAMLError as exception:
+            raise UserError(
+                self.env._("Invalid YAML/JSON in request content schema: %s", exception)
+            ) from exception
+        try:
+            jsonschema.Draft202012Validator.check_schema(schema)
+        except jsonschema.SchemaError as exception:
+            raise UserError(
+                self.env._(
+                    "Invalid JSON Schema in request content schema: %s",
+                    exception.message,
+                )
+            ) from exception
+
+    def _check_request_content_schema_xml(self):
+        try:
+            schema_doc = etree.fromstring(self.request_content_schema.encode())
+        except etree.XMLSyntaxError as exception:
+            raise UserError(
+                self.env._("Invalid XML in request content schema: %s", exception)
+            ) from exception
+        try:
+            etree.XMLSchema(schema_doc)
+        except etree.XMLSchemaParseError as exception:
+            raise UserError(
+                self.env._(
+                    "Invalid XML Schema (XSD) in request content schema: %s",
+                    exception,
+                )
+            ) from exception
+
     @api.constrains("auth_type")
     def _check_auth(self):
         for rec in self:
@@ -233,6 +314,59 @@ def _validate_request(self, request):
         ):
             self._logger.error("_validate_request: UnsupportedMediaType")
             raise werkzeug.exceptions.UnsupportedMediaType()
+        self._validate_request_content(request)
+
+    def _validate_request_content(self, request):
+        if not (self.request_content_schema and self.request_content_schema_applicable):
+            return
+        if self.request_content_type == "application/json":
+            self._validate_request_content_json(request)
+        elif self.request_content_type == "application/xml":
+            self._validate_request_content_xml(request)
+
+    def _validate_request_content_json(self, request):
+        try:
+            body = request.get_json_data()
+        except json.JSONDecodeError as exception:
+            self._logger.error("Invalid JSON body: %s", exception)
+            raise RequestValidationError(
+                [{"loc": ["body"], "msg": str(exception), "type": "json_invalid"}]
+            ) from exception
+        schema = yaml.safe_load(self.request_content_schema)
+        errors = list(jsonschema.Draft202012Validator(schema).iter_errors(body))
+        if errors:
+            self._logger.error("Schema validation failed (%d errors)", len(errors))
+            raise RequestValidationError(
+                [
+                    {
+                        "loc": ["body", *err.absolute_path],
+                        "msg": err.message,
+                        "type": err.validator,
+                    }
+                    for err in errors
+                ]
+            )
+
+    def _validate_request_content_xml(self, request):
+        body = request.httprequest.get_data()
+        try:
+            doc = etree.fromstring(body)
+        except etree.XMLSyntaxError as exception:
+            self._logger.error("Invalid XML body: %s", exception)
+            raise RequestValidationError(
+                [{"loc": ["body"], "msg": str(exception), "type": "xml_invalid"}]
+            ) from exception
+        schema = etree.XMLSchema(etree.fromstring(self.request_content_schema.encode()))
+        if not schema.validate(doc):
+            self._logger.error(
+                "Schema validation failed (%d errors)", len(schema.error_log)
+            )
+            raise RequestValidationError(
+                [
+                    {"loc": ["body"], "msg": err.message, "type": "xml_schema"}
+                    for err in schema.error_log
+                ]
+            )
 
     def _get_handler(self):
         try:

endpoint/tests/__init__.py

@@ -1,2 +1,3 @@
 from . import test_endpoint
+from . import test_endpoint_content_schema_validation
 from . import test_endpoint_controller

endpoint/tests/common.py

@@ -101,6 +101,42 @@ def _setup_demo_records(env):
             ),
         }
     )
+    endpoints += env["endpoint.endpoint"].create(
+        {
+            "name": "Demo Endpoint 8",
+            "route": "/demo/schema",
+            "request_method": "POST",
+            "request_content_type": "application/json",
+            "auth_type": "public",
+            "exec_as_user_id": demo_user.id,
+            "exec_mode": "code",
+            "code_snippet": 'result = {"payload": {"ok": True}}',
+            "request_content_schema": (
+                "type: object\n"
+                "required: [data]\n"
+                "properties:\n"
+                "  data:\n"
+                "    type: array\n"
+            ),
+        }
+    )
+    endpoints += env["endpoint.endpoint"].create(
+        {
+            "name": "Demo Endpoint 9",
+            "route": "/demo/schema-xml",
+            "request_method": "POST",
+            "request_content_type": "application/xml",
+            "auth_type": "public",
+            "exec_as_user_id": demo_user.id,
+            "exec_mode": "code",
+            "code_snippet": 'result = {"payload": {"ok": True}}',
+            "request_content_schema": (
+                '<xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema">'
+                '<xs:element name="greeting" type="xs:string"/>'
+                "</xs:schema>"
+            ),
+        }
+    )
     return endpoints