# keystone.api.validation

# Source code for keystone.api.validation

    # 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.

    """API request/response validating middleware."""

    import functools
    import typing as ty

    import flask
    from oslo_log import log
    from oslo_serialization import jsonutils

    from keystone.api.validation import validators
    import keystone.conf
    from keystone import exception

    CONF = keystone.conf.CONF
    LOG = log.getLogger(__name__)

    def validated(cls):
        cls._validated = True
        return cls

    def _schema_validator(
        schema: ty.Dict[str, ty.Any],
        target: ty.Dict[str, ty.Any],
        args: ty.Any,
        kwargs: ty.Any,
        is_body: bool = True,
    ):
        """A helper method to execute JSON Schema Validation.

        This method checks the request version whether matches the specified
        ``max_version`` and ``min_version``. If the version range matches the
        request, we validate ``schema`` against ``target``. A failure will result
        in ``ValidationError`` being raised.

        :param schema: The JSON Schema schema used to validate the target.
        :param target: The target to be validated by the schema.
        :param args: Positional arguments which passed into original method.
        :param kwargs: Keyword arguments which passed into original method.
        :param is_body: Whether ``target`` is a HTTP request body or not.
        :returns: None.
        :raises: ``ValidationError`` if validation fails.
        """
        schema_validator = validators._SchemaValidator(schema, is_body=is_body)
        schema_validator.validate(target)

    def request_body_schema(schema: ty.Optional[ty.Dict[str, ty.Any]] = None):
        """Register a schema to validate request body.

        ``schema`` will be used for validating the request body just before the API
        method is executed.

        :param schema: The JSON Schema schema used to validate the target. If
            empty value is passed no validation will be performed.
        :param min_version: A string indicating the minimum API version ``schema``
            applies against.
        :param max_version: A string indicating the maximum API version ``schema``
            applies against.
        """

        def add_validator(func):
            @functools.wraps(func)
            def wrapper(*args, **kwargs):
                if schema is not None:
                    _schema_validator(
                        schema,
                        flask.request.get_json(silent=True, force=True) or {},
                        args,
                        kwargs,
                        is_body=True,
                    )
                return func(*args, **kwargs)

            wrapper._request_body_schema = schema

            return wrapper

        return add_validator

    def request_query_schema(schema: ty.Optional[ty.Dict[str, ty.Any]] = None):
        """Register a schema to validate request query string parameters.

        ``schema`` will be used for validating request query strings just before
        the API method is executed.

        :param schema: The JSON Schema schema used to validate the target. If
            empty value is passed no validation will be performed.
        """

        def add_validator(func):
            @functools.wraps(func)
            def wrapper(*args, **kwargs):
                if schema is not None:
                    # NOTE: The request object is always the second argument.
                    # However, numerous unittests pass in the request object
                    # via kwargs instead so we handle that as well.
                    # TODO(stephenfin): Fix unit tests so we don't have to do this
                    if "req" in kwargs:
                        req = kwargs["req"]
                    else:
                        req = flask.request.args

                    _schema_validator(schema, req, args, kwargs, is_body=True)
                return func(*args, **kwargs)

            wrapper._request_query_schema = schema

            return wrapper

        return add_validator

    def response_body_schema(schema: ty.Optional[ty.Dict[str, ty.Any]] = None):
        """Register a schema to validate response body.

        ``schema`` will be used for validating the response body just after the API
        method is executed.

        :param schema: The JSON Schema schema used to validate the target. If
            empty value is passed no validation will be performed.
        :param min_version: A string indicating the minimum API version ``schema``
            applies against.
        :param max_version: A string indicating the maximum API version ``schema``
            applies against.
        """

        def add_validator(func):
            @functools.wraps(func)
            def wrapper(*args, **kwargs):
                response = func(*args, **kwargs)

                if CONF.api.response_validation == 'ignore':
                    # don't waste our time checking anything if we're ignoring
                    # schema errors
                    return response

                if schema is None:
                    return response

                # In Flask it is not uncommon that the method return a tuple of
                # body and the status code. In the runtime Keystone only return
                # a body, but some of the used testtools do return a tuple.
                if isinstance(response, tuple):
                    _body = response[0]
                else:
                    _body = response

                # NOTE(stephenfin): If our response is an object, we need to
                # serializer and deserialize to convert e.g. date-time
                # to strings
                _body = jsonutils.dump_as_bytes(_body)

                if _body == b'':
                    body = None
                else:
                    body = jsonutils.loads(_body)

                try:
                    _schema_validator(schema, body, args, kwargs, is_body=True)
                except exception.SchemaValidationError:
                    if CONF.api.response_validation == 'warn':
                        LOG.exception('Schema failed to validate')
                    else:
                        raise

                return response

            wrapper._response_body_schema = schema

            return wrapper

        return add_validator

this page last updated: 2026-08-17 12:45:24

Except where otherwise noted, this document is licensed under Creative Commons Attribution 3.0 License. See all OpenStack Legal Documents.

found an error? report a bug