validate.py (9663B)
1 #Embedded file name: C:\BuildAgent\work\20d73f1b24eed06e\eve\release\V24.01\carbon\common\lib\wsgiref\validate.py 2 __all__ = ['validator'] 3 import re 4 import sys 5 from types import DictType, StringType, TupleType, ListType 6 import warnings 7 header_re = re.compile('^[a-zA-Z][a-zA-Z0-9\\-_]*$') 8 bad_header_value_re = re.compile('[\\000-\\037]') 9 10 class WSGIWarning(Warning): 11 pass 12 13 14 def assert_(cond, *args): 15 if not cond: 16 raise AssertionError(*args) 17 18 19 def validator(application): 20 21 def lint_app(*args, **kw): 22 assert_(len(args) == 2, 'Two arguments required') 23 assert_(not kw, 'No keyword arguments allowed') 24 environ, start_response = args 25 check_environ(environ) 26 start_response_started = [] 27 28 def start_response_wrapper(*args, **kw): 29 assert_(len(args) == 2 or len(args) == 3, 'Invalid number of arguments: %s' % (args,)) 30 assert_(not kw, 'No keyword arguments allowed') 31 status = args[0] 32 headers = args[1] 33 if len(args) == 3: 34 exc_info = args[2] 35 else: 36 exc_info = None 37 check_status(status) 38 check_headers(headers) 39 check_content_type(status, headers) 40 check_exc_info(exc_info) 41 start_response_started.append(None) 42 return WriteWrapper(start_response(*args)) 43 44 environ['wsgi.input'] = InputWrapper(environ['wsgi.input']) 45 environ['wsgi.errors'] = ErrorWrapper(environ['wsgi.errors']) 46 iterator = application(environ, start_response_wrapper) 47 assert_(iterator is not None and iterator != False, 'The application must return an iterator, if only an empty list') 48 check_iterator(iterator) 49 return IteratorWrapper(iterator, start_response_started) 50 51 return lint_app 52 53 54 class InputWrapper: 55 56 def __init__(self, wsgi_input): 57 self.input = wsgi_input 58 59 def read(self, *args): 60 assert_(len(args) <= 1) 61 v = self.input.read(*args) 62 assert_(type(v) is type('')) 63 return v 64 65 def readline(self): 66 v = self.input.readline() 67 assert_(type(v) is type('')) 68 return v 69 70 def readlines(self, *args): 71 assert_(len(args) <= 1) 72 lines = self.input.readlines(*args) 73 assert_(type(lines) is type([])) 74 for line in lines: 75 assert_(type(line) is type('')) 76 77 return lines 78 79 def __iter__(self): 80 while 1: 81 line = self.readline() 82 if not line: 83 return 84 yield line 85 86 def close(self): 87 assert_(0, 'input.close() must not be called') 88 89 90 class ErrorWrapper: 91 92 def __init__(self, wsgi_errors): 93 self.errors = wsgi_errors 94 95 def write(self, s): 96 assert_(type(s) is type('')) 97 self.errors.write(s) 98 99 def flush(self): 100 self.errors.flush() 101 102 def writelines(self, seq): 103 for line in seq: 104 self.write(line) 105 106 def close(self): 107 assert_(0, 'errors.close() must not be called') 108 109 110 class WriteWrapper: 111 112 def __init__(self, wsgi_writer): 113 self.writer = wsgi_writer 114 115 def __call__(self, s): 116 assert_(type(s) is type('')) 117 self.writer(s) 118 119 120 class PartialIteratorWrapper: 121 122 def __init__(self, wsgi_iterator): 123 self.iterator = wsgi_iterator 124 125 def __iter__(self): 126 return IteratorWrapper(self.iterator, None) 127 128 129 class IteratorWrapper: 130 131 def __init__(self, wsgi_iterator, check_start_response): 132 self.original_iterator = wsgi_iterator 133 self.iterator = iter(wsgi_iterator) 134 self.closed = False 135 self.check_start_response = check_start_response 136 137 def __iter__(self): 138 return self 139 140 def next(self): 141 assert_(not self.closed, 'Iterator read after closed') 142 v = self.iterator.next() 143 if self.check_start_response is not None: 144 assert_(self.check_start_response, 'The application returns and we started iterating over its body, but start_response has not yet been called') 145 self.check_start_response = None 146 return v 147 148 def close(self): 149 self.closed = True 150 if hasattr(self.original_iterator, 'close'): 151 self.original_iterator.close() 152 153 def __del__(self): 154 if not self.closed: 155 sys.stderr.write('Iterator garbage collected without being closed') 156 assert_(self.closed, 'Iterator garbage collected without being closed') 157 158 159 def check_environ(environ): 160 assert_(type(environ) is DictType, 'Environment is not of the right type: %r (environment: %r)' % (type(environ), environ)) 161 for key in ['REQUEST_METHOD', 162 'SERVER_NAME', 163 'SERVER_PORT', 164 'wsgi.version', 165 'wsgi.input', 166 'wsgi.errors', 167 'wsgi.multithread', 168 'wsgi.multiprocess', 169 'wsgi.run_once']: 170 assert_(key in environ, 'Environment missing required key: %r' % (key,)) 171 172 for key in ['HTTP_CONTENT_TYPE', 'HTTP_CONTENT_LENGTH']: 173 assert_(key not in environ, 'Environment should not have the key: %s (use %s instead)' % (key, key[5:])) 174 175 if 'QUERY_STRING' not in environ: 176 warnings.warn('QUERY_STRING is not in the WSGI environment; the cgi module will use sys.argv when this variable is missing, so application errors are more likely', WSGIWarning) 177 for key in environ.keys(): 178 if '.' in key: 179 continue 180 assert_(type(environ[key]) is StringType, 'Environmental variable %s is not a string: %r (value: %r)' % (key, type(environ[key]), environ[key])) 181 182 assert_(type(environ['wsgi.version']) is TupleType, 'wsgi.version should be a tuple (%r)' % (environ['wsgi.version'],)) 183 assert_(environ['wsgi.url_scheme'] in ('http', 'https'), 'wsgi.url_scheme unknown: %r' % environ['wsgi.url_scheme']) 184 check_input(environ['wsgi.input']) 185 check_errors(environ['wsgi.errors']) 186 if environ['REQUEST_METHOD'] not in ('GET', 'HEAD', 'POST', 'OPTIONS', 'PUT', 'DELETE', 'TRACE'): 187 warnings.warn('Unknown REQUEST_METHOD: %r' % environ['REQUEST_METHOD'], WSGIWarning) 188 assert_(not environ.get('SCRIPT_NAME') or environ['SCRIPT_NAME'].startswith('/'), "SCRIPT_NAME doesn't start with /: %r" % environ['SCRIPT_NAME']) 189 assert_(not environ.get('PATH_INFO') or environ['PATH_INFO'].startswith('/'), "PATH_INFO doesn't start with /: %r" % environ['PATH_INFO']) 190 if environ.get('CONTENT_LENGTH'): 191 assert_(int(environ['CONTENT_LENGTH']) >= 0, 'Invalid CONTENT_LENGTH: %r' % environ['CONTENT_LENGTH']) 192 if not environ.get('SCRIPT_NAME'): 193 assert_('PATH_INFO' in environ, "One of SCRIPT_NAME or PATH_INFO are required (PATH_INFO should at least be '/' if SCRIPT_NAME is empty)") 194 assert_(environ.get('SCRIPT_NAME') != '/', "SCRIPT_NAME cannot be '/'; it should instead be '', and PATH_INFO should be '/'") 195 196 197 def check_input(wsgi_input): 198 for attr in ['read', 199 'readline', 200 'readlines', 201 '__iter__']: 202 assert_(hasattr(wsgi_input, attr), "wsgi.input (%r) doesn't have the attribute %s" % (wsgi_input, attr)) 203 204 205 def check_errors(wsgi_errors): 206 for attr in ['flush', 'write', 'writelines']: 207 assert_(hasattr(wsgi_errors, attr), "wsgi.errors (%r) doesn't have the attribute %s" % (wsgi_errors, attr)) 208 209 210 def check_status(status): 211 assert_(type(status) is StringType, 'Status must be a string (not %r)' % status) 212 status_code = status.split(None, 1)[0] 213 assert_(len(status_code) == 3, 'Status codes must be three characters: %r' % status_code) 214 status_int = int(status_code) 215 assert_(status_int >= 100, 'Status code is invalid: %r' % status_int) 216 if len(status) < 4 or status[3] != ' ': 217 warnings.warn('The status string (%r) should be a three-digit integer followed by a single space and a status explanation' % status, WSGIWarning) 218 219 220 def check_headers(headers): 221 assert_(type(headers) is ListType, 'Headers (%r) must be of type list: %r' % (headers, type(headers))) 222 header_names = {} 223 for item in headers: 224 assert_(type(item) is TupleType, 'Individual headers (%r) must be of type tuple: %r' % (item, type(item))) 225 assert_(len(item) == 2) 226 name, value = item 227 assert_(name.lower() != 'status', 'The Status header cannot be used; it conflicts with CGI script, and HTTP status is not given through headers (value: %r).' % value) 228 header_names[name.lower()] = None 229 assert_('\n' not in name and ':' not in name, "Header names may not contain ':' or '\\n': %r" % name) 230 assert_(header_re.search(name), 'Bad header name: %r' % name) 231 assert_(not name.endswith('-') and not name.endswith('_'), "Names may not end in '-' or '_': %r" % name) 232 if bad_header_value_re.search(value): 233 assert_(0, 'Bad header value: %r (bad char: %r)' % (value, bad_header_value_re.search(value).group(0))) 234 235 236 def check_content_type(status, headers): 237 code = int(status.split(None, 1)[0]) 238 NO_MESSAGE_BODY = (204, 304) 239 for name, value in headers: 240 if name.lower() == 'content-type': 241 if code not in NO_MESSAGE_BODY: 242 return 243 assert_(0, 'Content-Type header found in a %s response, which must not return content.' % code) 244 245 if code not in NO_MESSAGE_BODY: 246 assert_(0, 'No Content-Type header found in headers (%s)' % headers) 247 248 249 def check_exc_info(exc_info): 250 assert_(exc_info is None or type(exc_info) is type(()), 'exc_info (%r) is not a tuple: %r' % (exc_info, type(exc_info))) 251 252 253 def check_iterator(iterator): 254 assert_(not isinstance(iterator, str), 'You should not return a string as your application iterator, instead return a single-item list containing that string.')