Coverage for manila/api/validation/__init__.py: 90%

1# Licensed under the Apache License, Version 2.0 (the "License"); you may 

2# not use this file except in compliance with the License. You may obtain 

3# a copy of the License at 

4# 

5# http://www.apache.org/licenses/LICENSE-2.0 

6# 

7# Unless required by applicable law or agreed to in writing, software 

8# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 

9# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 

10# License for the specific language governing permissions and limitations 

11# under the License. 

12 

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

14 

15import functools 

16import typing as ty 

17 

18from oslo_serialization import jsonutils 

19import webob 

20 

21from manila.api.openstack import api_version_request as api_version 

22from manila.api.openstack import wsgi 

23from manila.api.validation import validators 

24from manila import exception 

25from manila.i18n import _ 

26 

27 

28def validated(cls): 

29 cls._validated = True 

30 return cls 

31 

32 

33def _schema_validator( 

34 schema: ty.Dict[str, ty.Any], 

35 target: ty.Dict[str, ty.Any], 

36 min_version: ty.Optional[str], 

37 max_version: ty.Optional[str], 

38 args: ty.Any, 

39 kwargs: ty.Any, 

40 is_body: bool = True, 

41): 

42 """A helper method to execute JSON Schema Validation. 

43 

44 This method checks the request version whether matches the specified 

45 ``max_version`` and ``min_version``. If the version range matches the 

46 request, we validate ``schema`` against ``target``. A failure will result 

47 in ``ValidationError`` being raised. 

48 

49 :param schema: The JSON Schema schema used to validate the target. 

50 :param target: The target to be validated by the schema. 

51 :param min_version: A string indicating the minimum API version ``schema`` 

52 applies against. 

53 :param max_version: A string indicating the maximum API version ``schema`` 

54 applies against. 

55 :param args: Positional arguments which passed into original method. 

56 :param kwargs: Keyword arguments which passed into original method. 

57 :param is_body: Whether ``target`` is a HTTP request body or not. 

58 :returns: None. 

59 :raises: ``ValidationError`` if validation fails. 

60 """ 

61 min_ver = api_version.APIVersionRequest(min_version) 

62 max_ver = api_version.APIVersionRequest(max_version) 

63 

64 # NOTE: The request object is always the second argument. However, numerous 

65 # unittests pass in the request object via kwargs instead so we handle that 

66 # as well. 

67 # TODO(stephenfin): Fix unit tests so we don't have to to do this 

68 if 'req' in kwargs: 

69 ver = kwargs['req'].api_version_request 

70 else: 

71 ver = args[1].api_version_request 

72 

73 if ver.matches(min_ver, max_ver): 

74 # Only validate against the schema if it lies within 

75 # the version range specified. Note that if both min 

76 # and max are not specified the validator will always 

77 # be run. 

78 schema_validator = validators._SchemaValidator(schema, is_body=is_body) 

79 schema_validator.validate(target) 

80 

81 

82def request_body_schema( 

83 schema: ty.Dict[str, ty.Any], 

84 min_version: ty.Optional[str] = None, 

85 max_version: ty.Optional[str] = None, 

86): 

87 """Register a schema to validate request body. 

88 

89 ``schema`` will be used for validating the request body just before the API 

90 method is executed. 

91 

92 :param schema: The JSON Schema schema used to validate the target. 

93 :param min_version: A string indicating the minimum API version ``schema`` 

94 applies against. 

95 :param max_version: A string indicating the maximum API version ``schema`` 

96 applies against. 

97 """ 

98 

99 def add_validator(func): 

100 @functools.wraps(func) 

101 def wrapper(*args, **kwargs): 

102 _schema_validator( 

103 schema, 

104 kwargs['body'], 

105 min_version, 

106 max_version, 

107 args, 

108 kwargs, 

109 is_body=True, 

110 ) 

111 return func(*args, **kwargs) 

112 

113 wrapper._request_body_schema = schema 

114 

115 return wrapper 

116 

117 return add_validator 

118 

119 

120def request_query_schema( 

121 schema: ty.Dict[str, ty.Any], 

122 min_version: ty.Optional[str] = None, 

123 max_version: ty.Optional[str] = None, 

124): 

125 """Register a schema to validate request query string parameters. 

126 

127 ``schema`` will be used for validating request query strings just before 

128 the API method is executed. 

129 

130 :param schema: The JSON Schema schema used to validate the target. 

131 :param min_version: A string indicating the minimum API version ``schema`` 

132 applies against. 

133 :param max_version: A string indicating the maximum API version ``schema`` 

134 applies against. 

135 """ 

136 

137 def add_validator(func): 

138 @functools.wraps(func) 

139 def wrapper(*args, **kwargs): 

140 # NOTE: The request object is always the second argument. However, 

141 # numerous unittests pass in the request object via kwargs instead 

142 # so we handle that as well. 

143 # TODO(stephenfin): Fix unit tests so we don't have to to do this 

144 if 'req' in kwargs: 144 ↛ 145line 144 didn't jump to line 145 because the condition on line 144 was never true

145 req = kwargs['req'] 

146 else: 

147 req = args[1] 

148 

149 # NOTE: The webob package throws UnicodeError when param cannot be 

150 # decoded. Catch this and raise HTTP 400. 

151 try: 

152 query = req.GET.dict_of_lists() 

153 except UnicodeDecodeError: 

154 msg = _('Query string is not UTF-8 encoded') 

155 raise exception.ValidationError(msg) 

156 

157 _schema_validator( 

158 schema, 

159 query, 

160 min_version, 

161 max_version, 

162 args, 

163 kwargs, 

164 is_body=True, 

165 ) 

166 return func(*args, **kwargs) 

167 

168 wrapper._request_query_schema = schema 

169 

170 return wrapper 

171 

172 return add_validator 

173 

174 

175def response_body_schema( 

176 schema: ty.Dict[str, ty.Any], 

177 min_version: ty.Optional[str] = None, 

178 max_version: ty.Optional[str] = None, 

179): 

180 """Register a schema to validate response body. 

181 

182 ``schema`` will be used for validating the response body just after the API 

183 method is executed. 

184 

185 :param schema: The JSON Schema schema used to validate the target. 

186 :param min_version: A string indicating the minimum API version ``schema`` 

187 applies against. 

188 :param max_version: A string indicating the maximum API version ``schema`` 

189 applies against. 

190 """ 

191 

192 def add_validator(func): 

193 @functools.wraps(func) 

194 def wrapper(*args, **kwargs): 

195 response = func(*args, **kwargs) 

196 

197 # NOTE(stephenfin): If our response is an object, we need to 

198 # serializer and deserialize to convert e.g. date-time to strings 

199 if isinstance(response, wsgi.ResponseObject): 199 ↛ 200line 199 didn't jump to line 200 because the condition on line 199 was never true

200 serializer = wsgi.JSONDictSerializer() 

201 _body = serializer.serialize(response.obj) 

202 # TODO(stephenfin): We should replace all instances of this with 

203 # wsgi.ResponseObject 

204 elif isinstance(response, webob.Response): 

205 _body = response.body 

206 else: 

207 serializer = wsgi.JSONDictSerializer() 

208 _body = serializer.serialize(response) 

209 

210 if _body == b'': 

211 body = None 

212 else: 

213 body = jsonutils.loads(_body) 

214 

215 _schema_validator( 

216 schema, 

217 body, 

218 min_version, 

219 max_version, 

220 args, 

221 kwargs, 

222 is_body=True, 

223 ) 

224 return response 

225 

226 wrapper._response_body_schema = schema 

227 

228 return wrapper 

229 

230 return add_validator