Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/connexion/middleware/request_validation.py: 30%

1""" 

2Validation Middleware. 

3""" 

4 

5import logging 

6import typing as t 

7 

8from starlette.types import ASGIApp, Receive, Scope, Send 

9 

10from connexion import utils 

11from connexion.datastructures import MediaTypeDict 

12from connexion.exceptions import UnsupportedMediaTypeProblem 

13from connexion.middleware.abstract import RoutedAPI, RoutedMiddleware 

14from connexion.operations import AbstractOperation 

15from connexion.validators import VALIDATOR_MAP 

16 

17logger = logging.getLogger("connexion.middleware.validation") 

18 

19 

20class RequestValidationOperation: 

21 def __init__( 

22 self, 

23 next_app: ASGIApp, 

24 *, 

25 operation: AbstractOperation, 

26 strict_validation: bool = False, 

27 validator_map: t.Optional[dict] = None, 

28 ) -> None: 

29 self.next_app = next_app 

30 self._operation = operation 

31 self.strict_validation = strict_validation 

32 self._validator_map = VALIDATOR_MAP.copy() 

33 self._validator_map.update(validator_map or {}) 

34 

35 def extract_content_type( 

36 self, headers: t.List[t.Tuple[bytes, bytes]] 

37 ) -> t.Tuple[str, str]: 

38 """Extract the mime type and encoding from the content type headers. 

39 

40 :param headers: Headers from ASGI scope 

41 

42 :return: A tuple of mime type, encoding 

43 """ 

44 content_type = utils.extract_content_type(headers) 

45 mime_type, encoding = utils.split_content_type(content_type) 

46 if mime_type is None: 

47 # Content-type header is not required. Take a best guess. 

48 try: 

49 mime_type = self._operation.consumes[0] 

50 except IndexError: 

51 mime_type = "application/octet-stream" 

52 if encoding is None: 

53 encoding = "utf-8" 

54 

55 return mime_type, encoding 

56 

57 def validate_mime_type(self, mime_type: str) -> None: 

58 """Validate the mime type against the spec if it defines which mime types are accepted. 

59 

60 :param mime_type: mime type from content type header 

61 """ 

62 if not self._operation.consumes: 

63 return 

64 

65 # Convert to MediaTypeDict to handle media-ranges 

66 media_type_dict = MediaTypeDict( 

67 [(c.lower(), None) for c in self._operation.consumes] 

68 ) 

69 if mime_type.lower() not in media_type_dict: 

70 raise UnsupportedMediaTypeProblem( 

71 detail=f"Invalid Content-type ({mime_type}), " 

72 f"expected {self._operation.consumes}" 

73 ) 

74 

75 @property 

76 def security_query_params(self) -> t.List[str]: 

77 """Get the names of query parameters that are used for security.""" 

78 if not hasattr(self, "_security_query_params"): 

79 security_query_params: t.List[str] = [] 

80 if self._operation.security is None: 

81 self._security_query_params = security_query_params 

82 return self._security_query_params 

83 

84 for security_req in self._operation.security: 

85 for scheme_name in security_req: 

86 security_scheme = self._operation.security_schemes[scheme_name] 

87 

88 if ( 

89 security_scheme["type"] == "apiKey" 

90 and security_scheme["in"] == "query" 

91 ): 

92 # Only query parameters need to be considered for strict_validation 

93 security_query_params.append(security_scheme["name"]) 

94 self._security_query_params = security_query_params 

95 

96 return self._security_query_params 

97 

98 async def __call__(self, scope: Scope, receive: Receive, send: Send): 

99 # Validate parameters & headers 

100 uri_parser_class = self._operation._uri_parser_class 

101 uri_parser = uri_parser_class( 

102 self._operation.parameters, self._operation.body_definition() 

103 ) 

104 parameter_validator_cls = self._validator_map["parameter"] 

105 parameter_validator = parameter_validator_cls( # type: ignore 

106 self._operation.parameters, 

107 uri_parser=uri_parser, 

108 strict_validation=self.strict_validation, 

109 security_query_params=self.security_query_params, 

110 ) 

111 parameter_validator.validate(scope) 

112 

113 # Extract content type 

114 headers = scope["headers"] 

115 mime_type, encoding = self.extract_content_type(headers) 

116 self.validate_mime_type(mime_type) 

117 

118 # Validate body 

119 schema = self._operation.body_schema(mime_type) 

120 if schema: 

121 try: 

122 body_validator = self._validator_map["body"][mime_type] # type: ignore 

123 except KeyError: 

124 logger.info( 

125 f"Skipping validation. No validator registered for content type: " 

126 f"{mime_type}." 

127 ) 

128 else: 

129 validator = body_validator( 

130 schema=schema, 

131 required=self._operation.request_body.get("required", False), 

132 nullable=utils.is_nullable( 

133 self._operation.body_definition(mime_type) 

134 ), 

135 encoding=encoding, 

136 strict_validation=self.strict_validation, 

137 uri_parser=self._operation.uri_parser_class( 

138 self._operation.parameters, self._operation.body_definition() 

139 ), 

140 ) 

141 receive, scope = await validator.wrap_receive(receive, scope=scope) 

142 

143 await self.next_app(scope, receive, send) 

144 

145 

146class RequestValidationAPI(RoutedAPI[RequestValidationOperation]): 

147 """Validation API.""" 

148 

149 def __init__( 

150 self, 

151 *args, 

152 strict_validation=False, 

153 validator_map=None, 

154 uri_parser_class=None, 

155 **kwargs, 

156 ): 

157 super().__init__(*args, **kwargs) 

158 self.validator_map = validator_map 

159 

160 logger.debug("Strict Request Validation: %s", str(strict_validation)) 

161 self.strict_validation = strict_validation 

162 

163 self.uri_parser_class = uri_parser_class 

164 

165 self.add_paths() 

166 

167 def make_operation( 

168 self, operation: AbstractOperation 

169 ) -> RequestValidationOperation: 

170 return RequestValidationOperation( 

171 self.next_app, 

172 operation=operation, 

173 strict_validation=self.strict_validation, 

174 validator_map=self.validator_map, 

175 ) 

176 

177 

178class RequestValidationMiddleware(RoutedMiddleware[RequestValidationAPI]): 

179 """Middleware for validating requests according to the API contract.""" 

180 

181 api_cls = RequestValidationAPI 

182 

183 

184class MissingValidationOperation(Exception): 

185 """Missing validation operation"""