9266386129

Committed 28 May 2024 08:53AM UTC coverage: 94.147% (-0.03%) from 94.177%

Build # 9266386129

Build Type

push

github

Committed by

web-flow

Commit Message

Split parsing from validation completely (#1934)

During the refactoring for Connexion 3, we deduplicated all
`coerce_type` calls during the refactoring and moved them into the
`uri_parser` so it is done in a single place. When looking into #1931
however, I noticed we are still using the `uri_parser` from more places
than we should.

This PR centralizes the parsing further into the `ConnexionRequest`
class. During validation, we now instantiate a `ConnexionRequest`
instead of a Starlette `Request`, and leverage it instead of calling the
`uri_parser` directly.

I split the parsing and validation in the tests so they can be tested in
isolation.

Run Details

4 of 4 new or added lines in 1 file covered. (100.0%)

1 existing line in 1 file now uncovered.

3635 of 3861 relevant lines covered (94.15%)

4.7 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

96.3

/connexion/validators/parameter.py

import collections
import copy
import logging

from jsonschema import Draft4Validator, ValidationError

from connexion.exceptions import BadRequestProblem, ExtraParameterProblem
from connexion.lifecycle import ConnexionRequest
from connexion.utils import boolean, is_null, is_nullable

logger = logging.getLogger("connexion.validators.parameter")

TYPE_MAP = {"integer": int, "number": float, "boolean": boolean, "object": dict}

try:
    draft4_format_checker = Draft4Validator.FORMAT_CHECKER  # type: ignore
except AttributeError:  # jsonschema < 4.5.0
    from jsonschema import draft4_format_checker


class ParameterValidator:
    def __init__(
        self,
        parameters,
        uri_parser,
        strict_validation=False,
        security_query_params=None,
    ):
        """
        :param parameters: List of request parameter dictionaries
        :param uri_parser: class to use for uri parsing
        :param strict_validation: Flag indicating if parameters not in spec are allowed
        :param security_query_params: List of query parameter names used for security.
            These parameters will be ignored when checking for extra parameters in case of
            strict validation.
        """
        self.parameters = collections.defaultdict(list)
        for p in parameters:
            self.parameters[p["in"]].append(p)

        self.uri_parser = uri_parser
        self.strict_validation = strict_validation
        self.security_query_params = set(security_query_params or [])

    @staticmethod
    def validate_parameter(parameter_type, value, param, param_name=None):
        if is_nullable(param) and is_null(value):
            return

        elif value is not None:
            param = copy.deepcopy(param)
            param = param.get("schema", param)
            try:
                Draft4Validator(param, format_checker=draft4_format_checker).validate(
                    value
                )
            except ValidationError as exception:
                return str(exception)

        elif param.get("required"):
            return "Missing {parameter_type} parameter '{param[name]}'".format(
                **locals()
            )

    @staticmethod
    def validate_parameter_list(request_params, spec_params):
        request_params = set(request_params)
        spec_params = set(spec_params)

        return request_params.difference(spec_params)

    def validate_query_parameter_list(self, request, security_params=None):
        request_params = request.query_params.keys()
        spec_params = [x["name"] for x in self.parameters.get("query", [])]
        spec_params.extend(security_params or [])
        return self.validate_parameter_list(request_params, spec_params)

    def validate_query_parameter(self, param, request):
        """
        Validate a single query parameter (request.args in Flask)

        :type param: dict
        :rtype: str
        """
        val = request.query_params.get(param["name"])
        return self.validate_parameter("query", val, param)

    def validate_path_parameter(self, param, request):
        val = request.path_params.get(param["name"].replace("-", "_"))
        return self.validate_parameter("path", val, param)

    def validate_header_parameter(self, param, request):
        val = request.headers.get(param["name"])
        return self.validate_parameter("header", val, param)

    def validate_cookie_parameter(self, param, request):
        val = request.cookies.get(param["name"])
        return self.validate_parameter("cookie", val, param)

    def validate(self, scope):
        logger.debug("%s validating parameters...", scope.get("path"))

        request = ConnexionRequest(scope, uri_parser=self.uri_parser)
        self.validate_request(request)

    def validate_request(self, request):
        if self.strict_validation:
            query_errors = self.validate_query_parameter_list(
                request, security_params=self.security_query_params
            )

            if query_errors:
                raise ExtraParameterProblem(
                    param_type="query", extra_params=query_errors
                )

        for param in self.parameters.get("query", []):
            error = self.validate_query_parameter(param, request)
            if error:
                raise BadRequestProblem(detail=error)

        for param in self.parameters.get("path", []):
            error = self.validate_path_parameter(param, request)
            if error:
                raise BadRequestProblem(detail=error)

        for param in self.parameters.get("header", []):
            error = self.validate_header_parameter(param, request)
            if error:
                raise BadRequestProblem(detail=error)

        for param in self.parameters.get("cookie", []):
            error = self.validate_cookie_parameter(param, request)
            if error:
                raise BadRequestProblem(detail=error)

1	import collections	5✔
2	import copy	5✔
3	import logging	5✔
4
5	from jsonschema import Draft4Validator, ValidationError	5✔
6
7	from connexion.exceptions import BadRequestProblem, ExtraParameterProblem	5✔
8	from connexion.lifecycle import ConnexionRequest	5✔
9	from connexion.utils import boolean, is_null, is_nullable	5✔
10
11	logger = logging.getLogger("connexion.validators.parameter")	5✔
12
13	TYPE_MAP = {"integer": int, "number": float, "boolean": boolean, "object": dict}	5✔
14
15	try:	5✔
16	draft4_format_checker = Draft4Validator.FORMAT_CHECKER # type: ignore	5✔
17	except AttributeError: # jsonschema < 4.5.0	×
18	from jsonschema import draft4_format_checker	×
19
20
21	class ParameterValidator:	5✔
22	def __init__(	5✔
23	self,
24	parameters,
25	uri_parser,
26	strict_validation=False,
27	security_query_params=None,
28	):
29	"""
30	:param parameters: List of request parameter dictionaries
31	:param uri_parser: class to use for uri parsing
32	:param strict_validation: Flag indicating if parameters not in spec are allowed
33	:param security_query_params: List of query parameter names used for security.
34	These parameters will be ignored when checking for extra parameters in case of
35	strict validation.
36	"""
37	self.parameters = collections.defaultdict(list)	5✔
38	for p in parameters:	5✔
39	self.parameters[p["in"]].append(p)	5✔
40
41	self.uri_parser = uri_parser	5✔
42	self.strict_validation = strict_validation	5✔
43	self.security_query_params = set(security_query_params or [])	5✔
44
45	@staticmethod	5✔
46	def validate_parameter(parameter_type, value, param, param_name=None):	5✔
47	if is_nullable(param) and is_null(value):	5✔
48	return	5✔
49
50	elif value is not None:	5✔
51	param = copy.deepcopy(param)	5✔
52	param = param.get("schema", param)	5✔
53	try:	5✔
54	Draft4Validator(param, format_checker=draft4_format_checker).validate(	5✔
55	value
56	)
57	except ValidationError as exception:	5✔
58	return str(exception)	5✔
59
60	elif param.get("required"):	5✔
61	return "Missing {parameter_type} parameter '{param[name]}'".format(	5✔
62	**locals()
63	)
64
65	@staticmethod	5✔
66	def validate_parameter_list(request_params, spec_params):	5✔
67	request_params = set(request_params)	5✔
68	spec_params = set(spec_params)	5✔
69
70	return request_params.difference(spec_params)	5✔
71
72	def validate_query_parameter_list(self, request, security_params=None):	5✔
73	request_params = request.query_params.keys()	5✔
74	spec_params = [x["name"] for x in self.parameters.get("query", [])]	5✔
75	spec_params.extend(security_params or [])	5✔
76	return self.validate_parameter_list(request_params, spec_params)	5✔
77
78	def validate_query_parameter(self, param, request):	5✔
79	"""
80	Validate a single query parameter (request.args in Flask)
81
82	:type param: dict
83	:rtype: str
84	"""
85	val = request.query_params.get(param["name"])	5✔
86	return self.validate_parameter("query", val, param)	5✔
87
88	def validate_path_parameter(self, param, request):	5✔
89	val = request.path_params.get(param["name"].replace("-", "_"))	5✔
90	return self.validate_parameter("path", val, param)	5✔
91
92	def validate_header_parameter(self, param, request):	5✔
93	val = request.headers.get(param["name"])	5✔
94	return self.validate_parameter("header", val, param)	5✔
95
96	def validate_cookie_parameter(self, param, request):	5✔
97	val = request.cookies.get(param["name"])	5✔
98	return self.validate_parameter("cookie", val, param)	5✔
99
100	def validate(self, scope):	5✔
101	logger.debug("%s validating parameters...", scope.get("path"))	5✔
102
103	request = ConnexionRequest(scope, uri_parser=self.uri_parser)	5✔
104	self.validate_request(request)	5✔
105
106	def validate_request(self, request):	5✔
107	if self.strict_validation:	5✔
108	query_errors = self.validate_query_parameter_list(	5✔
109	request, security_params=self.security_query_params
110	)
111
112	if query_errors:	5✔
113	raise ExtraParameterProblem(	5✔
114	param_type="query", extra_params=query_errors
115	)
116
117	for param in self.parameters.get("query", []):	5✔
118	error = self.validate_query_parameter(param, request)	5✔
119	if error:	5✔
120	raise BadRequestProblem(detail=error)	5✔
121
122	for param in self.parameters.get("path", []):	5✔
123	error = self.validate_path_parameter(param, request)	5✔
124	if error:	5✔
125	raise BadRequestProblem(detail=error)	5✔
126
127	for param in self.parameters.get("header", []):	5✔
128	error = self.validate_header_parameter(param, request)	5✔
129	if error:	5✔
130	raise BadRequestProblem(detail=error)	5✔
131
132	for param in self.parameters.get("cookie", []):	5✔
133	error = self.validate_cookie_parameter(param, request)	5✔
134	if error:	5✔
UNCOV 135	raise BadRequestProblem(detail=error)	×

spec-first / connexion / 9266386129

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous